chore: auto compact chat

This commit is contained in:
Rotimi Best
2026-05-21 19:46:17 +01:00
parent 4e4119d597
commit 30e5f9cd54
15 changed files with 1247 additions and 122 deletions
@@ -0,0 +1,85 @@
import { and, eq } from 'drizzle-orm';
import * as schema from '@db/schema';
import { db } from '@db/drizzle';
export interface ChatModelContextInput {
conversationId: string;
courseId: string;
userId: string;
modelSummary?: string;
compactedThroughMessageId?: string | null;
sourceIds?: string[];
}
export async function getChatModelContext(conversationId: string, userId: string) {
try {
const [row] = await db
.select()
.from(schema.aiChatModelContext)
.where(
and(eq(schema.aiChatModelContext.conversationId, conversationId), eq(schema.aiChatModelContext.userId, userId))
)
.limit(1);
return row ?? null;
} catch (error) {
console.error('getChatModelContext error:', error);
throw new Error('Failed to fetch chat model context');
}
}
export async function upsertChatModelContext(input: ChatModelContextInput) {
try {
const now = new Date().toISOString();
const [row] = await db
.insert(schema.aiChatModelContext)
.values({
conversationId: input.conversationId,
courseId: input.courseId,
userId: input.userId,
modelSummary: input.modelSummary ?? '',
compactedThroughMessageId: input.compactedThroughMessageId ?? null,
sourceIds: input.sourceIds ?? [],
updatedAt: now
})
.onConflictDoUpdate({
target: schema.aiChatModelContext.conversationId,
set: {
courseId: input.courseId,
userId: input.userId,
modelSummary: input.modelSummary ?? '',
compactedThroughMessageId: input.compactedThroughMessageId ?? null,
sourceIds: input.sourceIds ?? [],
updatedAt: now
}
})
.returning();
return row;
} catch (error) {
console.error('upsertChatModelContext error:', error);
throw new Error('Failed to save chat model context');
}
}
export async function appendConversationSourceIds(conversationId: string, userId: string, sourceIds: string[]) {
try {
const existing = await getChatModelContext(conversationId, userId);
if (!existing) return null;
const mergedSourceIds = Array.from(new Set([...(existing.sourceIds ?? []), ...sourceIds]));
const [row] = await db
.update(schema.aiChatModelContext)
.set({ sourceIds: mergedSourceIds, updatedAt: new Date().toISOString() })
.where(
and(eq(schema.aiChatModelContext.conversationId, conversationId), eq(schema.aiChatModelContext.userId, userId))
)
.returning();
return row ?? null;
} catch (error) {
console.error('appendConversationSourceIds error:', error);
throw new Error('Failed to update chat model context sources');
}
}
+2
View File
@@ -1,6 +1,8 @@
export * from './chat-history';
export * from './chat-document';
export * from './chat-resource';
export * from './chat-model-context';
export * from './run-state';
export * from './credit-purchase';
export * from './token-usage';
export * from './usage-aggregations';
+303
View File
@@ -0,0 +1,303 @@
import { and, asc, desc, eq } from 'drizzle-orm';
import * as schema from '@db/schema';
import { db } from '@db/drizzle';
import type {
TAiAgentRun,
TAiAgentRunEvent,
TAiAgentRunStatus,
TAiAgentRunStep,
TNewAiAgentRun,
TNewAiAgentRunEvent,
TNewAiAgentRunStep
} from '@db/types';
export interface AgentRunInstruction {
id: string;
text: string;
createdBy: string;
createdAt: string;
}
export interface CreateAgentRunInput {
orgId: string;
courseId: string;
conversationId?: string | null;
userId: string;
phase?: string;
approvedPlan?: Record<string, unknown> | null;
executionCursor?: Record<string, unknown>;
sourceIds?: string[];
modelSummary?: string;
}
export interface UpdateAgentRunInput {
runId: string;
userId: string;
status?: TAiAgentRunStatus;
phase?: string;
progressPercent?: number;
currentStepKey?: string | null;
approvedPlan?: Record<string, unknown> | null;
executionCursor?: Record<string, unknown>;
sourceIds?: string[];
modelSummary?: string;
queuedInstructions?: AgentRunInstruction[];
lastError?: { code: string; message: string; details?: unknown } | null;
attempt?: number;
workerId?: string | null;
lockedAt?: string | null;
cancelRequestedAt?: string | null;
startedAt?: string | null;
finishedAt?: string | null;
}
export interface UpsertAgentRunStepInput {
runId: string;
stepKey: string;
stepType: string;
status?: TAiAgentRunStatus;
inputHash?: string | null;
input?: Record<string, unknown> | null;
output?: Record<string, unknown> | null;
outputIds?: string[];
idempotencyKey?: string | null;
error?: { code: string; message: string; details?: unknown } | null;
attempt?: number;
startedAt?: string | null;
finishedAt?: string | null;
}
export interface CreateAgentRunEventInput {
runId: string;
eventType: string;
message?: string | null;
payload?: Record<string, unknown> | null;
}
function hasInputKey<T extends object>(input: T, key: keyof T): boolean {
return Object.prototype.hasOwnProperty.call(input, key);
}
export async function createAgentRun(input: CreateAgentRunInput): Promise<TAiAgentRun> {
try {
const values: TNewAiAgentRun = {
orgId: input.orgId,
courseId: input.courseId,
conversationId: input.conversationId ?? null,
userId: input.userId,
phase: input.phase ?? 'planning',
approvedPlan: input.approvedPlan ?? null,
executionCursor: input.executionCursor ?? {},
sourceIds: input.sourceIds ?? [],
modelSummary: input.modelSummary ?? '',
queuedInstructions: []
};
const [row] = await db.insert(schema.aiAgentRun).values(values).returning();
if (!row) {
throw new Error('Insert returned no row');
}
return row;
} catch (error) {
console.error('createAgentRun error:', error);
throw new Error('Failed to create agent run');
}
}
export async function listAgentRuns(courseId: string, userId: string): Promise<TAiAgentRun[]> {
try {
return await db
.select()
.from(schema.aiAgentRun)
.where(and(eq(schema.aiAgentRun.courseId, courseId), eq(schema.aiAgentRun.userId, userId)))
.orderBy(desc(schema.aiAgentRun.updatedAt));
} catch (error) {
console.error('listAgentRuns error:', error);
throw new Error('Failed to list agent runs');
}
}
export async function getAgentRun(runId: string, userId: string): Promise<TAiAgentRun | null> {
try {
const [row] = await db
.select()
.from(schema.aiAgentRun)
.where(and(eq(schema.aiAgentRun.id, runId), eq(schema.aiAgentRun.userId, userId)))
.limit(1);
return row ?? null;
} catch (error) {
console.error('getAgentRun error:', error);
throw new Error('Failed to get agent run');
}
}
export async function getAgentRunById(runId: string): Promise<TAiAgentRun | null> {
try {
const [row] = await db.select().from(schema.aiAgentRun).where(eq(schema.aiAgentRun.id, runId)).limit(1);
return row ?? null;
} catch (error) {
console.error('getAgentRunById error:', error);
throw new Error('Failed to get agent run by id');
}
}
export async function updateAgentRun(input: UpdateAgentRunInput): Promise<TAiAgentRun | null> {
try {
const updateData: Partial<TNewAiAgentRun> = {
updatedAt: new Date().toISOString()
};
if (hasInputKey(input, 'status')) updateData.status = input.status;
if (hasInputKey(input, 'phase')) updateData.phase = input.phase;
if (hasInputKey(input, 'progressPercent')) updateData.progressPercent = input.progressPercent;
if (hasInputKey(input, 'currentStepKey')) updateData.currentStepKey = input.currentStepKey;
if (hasInputKey(input, 'approvedPlan')) updateData.approvedPlan = input.approvedPlan;
if (hasInputKey(input, 'executionCursor')) updateData.executionCursor = input.executionCursor;
if (hasInputKey(input, 'sourceIds')) updateData.sourceIds = input.sourceIds;
if (hasInputKey(input, 'modelSummary')) updateData.modelSummary = input.modelSummary;
if (hasInputKey(input, 'queuedInstructions')) updateData.queuedInstructions = input.queuedInstructions;
if (hasInputKey(input, 'lastError')) updateData.lastError = input.lastError;
if (hasInputKey(input, 'attempt')) updateData.attempt = input.attempt;
if (hasInputKey(input, 'workerId')) updateData.workerId = input.workerId;
if (hasInputKey(input, 'lockedAt')) updateData.lockedAt = input.lockedAt;
if (hasInputKey(input, 'cancelRequestedAt')) updateData.cancelRequestedAt = input.cancelRequestedAt;
if (hasInputKey(input, 'startedAt')) updateData.startedAt = input.startedAt;
if (hasInputKey(input, 'finishedAt')) updateData.finishedAt = input.finishedAt;
const [row] = await db
.update(schema.aiAgentRun)
.set(updateData)
.where(and(eq(schema.aiAgentRun.id, input.runId), eq(schema.aiAgentRun.userId, input.userId)))
.returning();
return row ?? null;
} catch (error) {
console.error('updateAgentRun error:', error);
throw new Error('Failed to update agent run');
}
}
export async function appendAgentRunInstruction(
runId: string,
userId: string,
instruction: AgentRunInstruction
): Promise<TAiAgentRun | null> {
try {
const existingRun = await getAgentRun(runId, userId);
if (!existingRun) return null;
return await updateAgentRun({
runId,
userId,
queuedInstructions: [...(existingRun.queuedInstructions ?? []), instruction]
});
} catch (error) {
console.error('appendAgentRunInstruction error:', error);
throw new Error('Failed to append agent run instruction');
}
}
export async function upsertAgentRunStep(input: UpsertAgentRunStepInput): Promise<TAiAgentRunStep> {
try {
const now = new Date().toISOString();
const values: TNewAiAgentRunStep = {
runId: input.runId,
stepKey: input.stepKey,
stepType: input.stepType,
status: input.status ?? 'queued',
inputHash: input.inputHash ?? null,
input: input.input ?? null,
output: input.output ?? null,
outputIds: input.outputIds ?? [],
idempotencyKey: input.idempotencyKey ?? null,
error: input.error ?? null,
attempt: input.attempt ?? 0,
startedAt: input.startedAt ?? null,
finishedAt: input.finishedAt ?? null
};
const [row] = await db
.insert(schema.aiAgentRunStep)
.values(values)
.onConflictDoUpdate({
target: [schema.aiAgentRunStep.runId, schema.aiAgentRunStep.stepKey],
set: {
stepType: input.stepType,
status: input.status ?? 'queued',
inputHash: input.inputHash ?? null,
input: input.input ?? null,
output: input.output ?? null,
outputIds: input.outputIds ?? [],
idempotencyKey: input.idempotencyKey ?? null,
error: input.error ?? null,
attempt: input.attempt ?? 0,
startedAt: input.startedAt ?? null,
finishedAt: input.finishedAt ?? null,
updatedAt: now
}
})
.returning();
if (!row) {
throw new Error('Upsert returned no row');
}
return row;
} catch (error) {
console.error('upsertAgentRunStep error:', error);
throw new Error('Failed to upsert agent run step');
}
}
export async function listAgentRunSteps(runId: string): Promise<TAiAgentRunStep[]> {
try {
return await db
.select()
.from(schema.aiAgentRunStep)
.where(eq(schema.aiAgentRunStep.runId, runId))
.orderBy(asc(schema.aiAgentRunStep.createdAt));
} catch (error) {
console.error('listAgentRunSteps error:', error);
throw new Error('Failed to list agent run steps');
}
}
export async function createAgentRunEvent(input: CreateAgentRunEventInput): Promise<TAiAgentRunEvent> {
try {
const values: TNewAiAgentRunEvent = {
runId: input.runId,
eventType: input.eventType,
message: input.message ?? null,
payload: input.payload ?? null
};
const [row] = await db.insert(schema.aiAgentRunEvent).values(values).returning();
if (!row) {
throw new Error('Insert returned no row');
}
return row;
} catch (error) {
console.error('createAgentRunEvent error:', error);
throw new Error('Failed to create agent run event');
}
}
export async function listAgentRunEvents(runId: string, limit = 100): Promise<TAiAgentRunEvent[]> {
try {
return await db
.select()
.from(schema.aiAgentRunEvent)
.where(eq(schema.aiAgentRunEvent.runId, runId))
.orderBy(asc(schema.aiAgentRunEvent.createdAt))
.limit(limit);
} catch (error) {
console.error('listAgentRunEvents error:', error);
throw new Error('Failed to list agent run events');
}
}
+163
View File
@@ -30,6 +30,15 @@ export const courseImportSourceType = pgEnum('COURSE_IMPORT_SOURCE_TYPE', ['prom
export const courseImportDraftStatus = pgEnum('COURSE_IMPORT_DRAFT_STATUS', ['DRAFT', 'PUBLISHED', 'ARCHIVED']);
export const organizationApiKeyType = pgEnum('ORGANIZATION_API_KEY_TYPE', ['mcp', 'api', 'zapier']);
export const automationUsageCategory = pgEnum('AUTOMATION_USAGE_CATEGORY', ['read', 'write', 'publish']);
export const aiAgentRunStatus = pgEnum('AI_AGENT_RUN_STATUS', [
'queued',
'running',
'waiting_for_input',
'paused',
'completed',
'failed',
'canceled'
]);
export const widgetStatus = pgEnum('WIDGET_STATUS', ['DRAFT', 'PUBLISHED', 'ARCHIVED']);
export const widgetLayoutType = pgEnum('WIDGET_LAYOUT_TYPE', [
'card_grid',
@@ -3084,6 +3093,38 @@ export const aiChatConversation = pgTable(
]
);
export const aiChatModelContext = pgTable(
'ai_chat_model_context',
{
conversationId: uuid('conversation_id').primaryKey(),
courseId: uuid('course_id').notNull(),
userId: uuid('user_id').notNull(),
modelSummary: text('model_summary').notNull().default(''),
compactedThroughMessageId: text('compacted_through_message_id'),
sourceIds: jsonb('source_ids').default([]).notNull().$type<string[]>(),
createdAt: timestamp('created_at', { withTimezone: true, mode: 'string' }).defaultNow().notNull(),
updatedAt: timestamp('updated_at', { withTimezone: true, mode: 'string' }).defaultNow().notNull()
},
(table) => [
foreignKey({
columns: [table.conversationId],
foreignColumns: [aiChatConversation.id],
name: 'ai_chat_model_context_conversation_id_fkey'
}).onDelete('cascade'),
foreignKey({
columns: [table.courseId],
foreignColumns: [course.id],
name: 'ai_chat_model_context_course_id_fkey'
}),
foreignKey({
columns: [table.userId],
foreignColumns: [profile.id],
name: 'ai_chat_model_context_user_id_fkey'
}).onDelete('cascade'),
index('idx_ai_chat_model_context_course_user').on(table.courseId, table.userId)
]
);
export const aiChatDocument = pgTable(
'ai_chat_document',
{
@@ -3109,6 +3150,128 @@ export const aiChatDocument = pgTable(
]
);
export const aiAgentRun = pgTable(
'ai_agent_run',
{
id: uuid()
.default(sql`gen_random_uuid()`)
.primaryKey()
.notNull(),
orgId: uuid('org_id').notNull(),
courseId: uuid('course_id').notNull(),
conversationId: uuid('conversation_id'),
userId: uuid('user_id').notNull(),
status: aiAgentRunStatus('status').default('queued').notNull(),
phase: varchar({ length: 64 }).default('planning').notNull(),
progressPercent: integer('progress_percent').default(0).notNull(),
currentStepKey: varchar('current_step_key', { length: 128 }),
approvedPlan: jsonb('approved_plan').$type<Record<string, unknown> | null>(),
executionCursor: jsonb('execution_cursor').default({}).notNull().$type<Record<string, unknown>>(),
sourceIds: jsonb('source_ids').default([]).notNull().$type<string[]>(),
modelSummary: text('model_summary').notNull().default(''),
queuedInstructions: jsonb('queued_instructions')
.default([])
.notNull()
.$type<Array<{ id: string; text: string; createdBy: string; createdAt: string }>>(),
lastError: jsonb('last_error').$type<{ code: string; message: string; details?: unknown } | null>(),
attempt: integer().default(0).notNull(),
maxAttempts: integer('max_attempts').default(3).notNull(),
workerId: text('worker_id'),
lockedAt: timestamp('locked_at', { withTimezone: true, mode: 'string' }),
cancelRequestedAt: timestamp('cancel_requested_at', { withTimezone: true, mode: 'string' }),
startedAt: timestamp('started_at', { withTimezone: true, mode: 'string' }),
finishedAt: timestamp('finished_at', { withTimezone: true, mode: 'string' }),
createdAt: timestamp('created_at', { withTimezone: true, mode: 'string' }).defaultNow().notNull(),
updatedAt: timestamp('updated_at', { withTimezone: true, mode: 'string' }).defaultNow().notNull()
},
(table) => [
foreignKey({
columns: [table.orgId],
foreignColumns: [organization.id],
name: 'ai_agent_run_org_id_fkey'
}).onDelete('cascade'),
foreignKey({
columns: [table.courseId],
foreignColumns: [course.id],
name: 'ai_agent_run_course_id_fkey'
}).onDelete('cascade'),
foreignKey({
columns: [table.conversationId],
foreignColumns: [aiChatConversation.id],
name: 'ai_agent_run_conversation_id_fkey'
}).onDelete('set null'),
foreignKey({
columns: [table.userId],
foreignColumns: [profile.id],
name: 'ai_agent_run_user_id_fkey'
}).onDelete('cascade'),
index('idx_ai_agent_run_course_user').on(table.courseId, table.userId),
index('idx_ai_agent_run_org_status').on(table.orgId, table.status),
index('idx_ai_agent_run_conversation').on(table.conversationId),
index('idx_ai_agent_run_updated_at').on(table.updatedAt)
]
);
export const aiAgentRunStep = pgTable(
'ai_agent_run_step',
{
id: uuid()
.default(sql`gen_random_uuid()`)
.primaryKey()
.notNull(),
runId: uuid('run_id').notNull(),
stepKey: varchar('step_key', { length: 128 }).notNull(),
stepType: varchar('step_type', { length: 64 }).notNull(),
status: aiAgentRunStatus('status').default('queued').notNull(),
inputHash: text('input_hash'),
input: jsonb().$type<Record<string, unknown> | null>(),
output: jsonb().$type<Record<string, unknown> | null>(),
outputIds: jsonb('output_ids').default([]).notNull().$type<string[]>(),
idempotencyKey: text('idempotency_key'),
error: jsonb().$type<{ code: string; message: string; details?: unknown } | null>(),
attempt: integer().default(0).notNull(),
startedAt: timestamp('started_at', { withTimezone: true, mode: 'string' }),
finishedAt: timestamp('finished_at', { withTimezone: true, mode: 'string' }),
createdAt: timestamp('created_at', { withTimezone: true, mode: 'string' }).defaultNow().notNull(),
updatedAt: timestamp('updated_at', { withTimezone: true, mode: 'string' }).defaultNow().notNull()
},
(table) => [
foreignKey({
columns: [table.runId],
foreignColumns: [aiAgentRun.id],
name: 'ai_agent_run_step_run_id_fkey'
}).onDelete('cascade'),
unique('ai_agent_run_step_run_step_unique').on(table.runId, table.stepKey),
index('idx_ai_agent_run_step_run_status').on(table.runId, table.status)
]
);
export const aiAgentRunEvent = pgTable(
'ai_agent_run_event',
{
id: bigint({ mode: 'number' }).primaryKey().generatedByDefaultAsIdentity({
name: 'ai_agent_run_event_id_seq',
startWith: 1,
increment: 1,
minValue: 1,
cache: 1
}),
runId: uuid('run_id').notNull(),
eventType: varchar('event_type', { length: 64 }).notNull(),
message: text(),
payload: jsonb().$type<Record<string, unknown> | null>(),
createdAt: timestamp('created_at', { withTimezone: true, mode: 'string' }).defaultNow().notNull()
},
(table) => [
foreignKey({
columns: [table.runId],
foreignColumns: [aiAgentRun.id],
name: 'ai_agent_run_event_run_id_fkey'
}).onDelete('cascade'),
index('idx_ai_agent_run_event_run_created').on(table.runId, table.createdAt)
]
);
// ─── BullMQ Job State (Postgres source of truth) ─────────────────────────────
// User-visible status for queued/long-running domain runs. BullMQ holds the
// operational job; these tables hold the entity the dashboard polls and
+12
View File
@@ -206,6 +206,7 @@ export type TSsoProvider = (typeof schema.organizationSsoProviderType.enumValues
export type TOrganizationApiKeyType = (typeof schema.organizationApiKeyType.enumValues)[number];
export type TAutomationUsageCategory = (typeof schema.automationUsageCategory.enumValues)[number];
export type TAiAgentRunStatus = (typeof schema.aiAgentRunStatus.enumValues)[number];
export type TCourseImportSourceType = (typeof schema.courseImportSourceType.enumValues)[number];
@@ -214,6 +215,17 @@ export type TWidgetStatus = (typeof schema.widgetStatus.enumValues)[number];
export type TWidgetLayoutType = (typeof schema.widgetLayoutType.enumValues)[number];
export type TWidgetSelectionMode = (typeof schema.widgetSelectionMode.enumValues)[number];
// ─── AI agent durable runs ───────────────────────────────────────────────────
export type TAiAgentRun = typeof schema.aiAgentRun.$inferSelect;
export type TNewAiAgentRun = typeof schema.aiAgentRun.$inferInsert;
export type TAiAgentRunStep = typeof schema.aiAgentRunStep.$inferSelect;
export type TNewAiAgentRunStep = typeof schema.aiAgentRunStep.$inferInsert;
export type TAiAgentRunEvent = typeof schema.aiAgentRunEvent.$inferSelect;
export type TNewAiAgentRunEvent = typeof schema.aiAgentRunEvent.$inferInsert;
// ─── Job state ───────────────────────────────────────────────────────────────
export type TJobStatus = (typeof schema.jobStatus.enumValues)[number];