diff --git a/.server-changes/dashboard-agent-maintenance-move.md b/.server-changes/dashboard-agent-maintenance-move.md new file mode 100644 index 0000000000..db0a9c96b2 --- /dev/null +++ b/.server-changes/dashboard-agent-maintenance-move.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: improvement +--- + +Routine cleanup of old dashboard agent data now runs on its own schedule. diff --git a/apps/webapp/app/services/dashboardAgentChatRetention.server.ts b/apps/webapp/app/services/dashboardAgentChatRetention.server.ts index 79fa3bff9f..29841b8ed8 100644 --- a/apps/webapp/app/services/dashboardAgentChatRetention.server.ts +++ b/apps/webapp/app/services/dashboardAgentChatRetention.server.ts @@ -1,71 +1,14 @@ /** - * Retention for soft-deleted chats. A deleted chat is kept for a grace window and then - * hard-deleted with all its child rows; one bounded statement per run, oldest first. - * Also the eventual purge behind organization deletion, which soft-deletes the org's - * chats so this same sweep removes them. + * The purge behind organization deletion: it soft-deletes the org's chats, and retention + * hard-deletes them once the window passes. */ -import { - hardDeleteChatsSoftDeletedBefore, - softDeleteChatsForOrganization, -} from "@internal/dashboard-agent-db"; +import { softDeleteChatsForOrganization } from "@internal/dashboard-agent-db"; import { dashboardAgentDb } from "~/services/dashboardAgentDb.server"; -import { logger } from "~/services/logger.server"; /** - * How long a soft-deleted chat is kept before it and its children are hard-deleted. - * Long enough that an accidental delete can still be investigated; org deletion soft- - * deletes the org's chats, so those are removed the same way once the window passes. - */ -export const CHAT_SOFT_DELETE_RETENTION_MS = 30 * 24 * 60 * 60 * 1000; - -/** Per-run cap. Retention is one bounded statement, not a row-at-a-time loop. */ -const RETENTION_BATCH_LIMIT = 500; - -export type ChatRetentionResult = { - /** Soft-deleted chats past the retention window dropped this run. */ - purged: number; - failed: number; -}; - -export type ChatRetentionDeps = { - now?: () => Date; - limit?: number; - /** Hard-delete chats soft-deleted before `before`. Returns how many went. */ - purge?: (params: { before: Date; limit: number }) => Promise; -}; - -export async function sweepDashboardAgentSoftDeletedChats( - deps: ChatRetentionDeps = {} -): Promise { - const now = deps.now?.() ?? new Date(); - const limit = deps.limit ?? RETENTION_BATCH_LIMIT; - const purge = - deps.purge ?? ((params) => hardDeleteChatsSoftDeletedBefore(dashboardAgentDb, params)); - - const result: ChatRetentionResult = { purged: 0, failed: 0 }; - - try { - result.purged = await purge({ - before: new Date(now.getTime() - CHAT_SOFT_DELETE_RETENTION_MS), - limit, - }); - } catch (error) { - result.failed++; - logger.error("Dashboard agent chat retention failed", { error }); - } - - if (result.failed > 0) { - throw new Error("The dashboard agent chat retention pass failed"); - } - - return result; -} - -/** - * Soft-delete every chat belonging to a deleted organization. The retention sweep above - * hard-deletes them once the window passes, so the org-deletion request never runs a - * cross-database hard delete. + * Soft-delete every chat belonging to a deleted organization. Retention hard-deletes them + * once the window passes, so the org-deletion request never runs a cross-database hard delete. */ export async function purgeDashboardAgentChatsForOrganization(params: { organizationId: string; diff --git a/apps/webapp/app/services/dashboardAgentEvalRetention.server.ts b/apps/webapp/app/services/dashboardAgentEvalRetention.server.ts deleted file mode 100644 index 5548842013..0000000000 --- a/apps/webapp/app/services/dashboardAgentEvalRetention.server.ts +++ /dev/null @@ -1,58 +0,0 @@ -/** - * Retention for the agent's judged-turn rows. The table is append-only quality data with - * no reader, so it can't be left to grow forever; one bounded statement per run, oldest - * first. Runs whether or not the agent is configured — rows outlive the agent project. - */ - -import { deleteTurnEvalsOlderThan } from "@internal/dashboard-agent-db"; -import { dashboardAgentDb } from "~/services/dashboardAgentDb.server"; -import { logger } from "~/services/logger.server"; - -/** - * How long a judged turn is kept. Nothing reads the table today, and the rows carry the - * user's question next to the agent's answer, so the period is the shortest one that still - * lets a month of product review (capability and docs gaps) be aggregated. - */ -export const TURN_EVAL_RETENTION_MS = 30 * 24 * 60 * 60 * 1000; - -/** Per-run cap. Retention is one statement, not a row-at-a-time loop. */ -const RETENTION_BATCH_LIMIT = 500; - -export type TurnEvalRetentionResult = { - /** Rows past the retention period dropped this run. */ - purged: number; - failed: number; -}; - -export type TurnEvalRetentionDeps = { - now?: () => Date; - limit?: number; - /** Drop rows created before `before`. Returns how many went. */ - purge?: (params: { before: Date; limit: number }) => Promise; -}; - -export async function sweepDashboardAgentTurnEvals( - deps: TurnEvalRetentionDeps = {} -): Promise { - const now = deps.now?.() ?? new Date(); - const limit = deps.limit ?? RETENTION_BATCH_LIMIT; - const purge = deps.purge ?? ((params) => deleteTurnEvalsOlderThan(dashboardAgentDb, params)); - - const result: TurnEvalRetentionResult = { purged: 0, failed: 0 }; - - try { - result.purged = await purge({ - before: new Date(now.getTime() - TURN_EVAL_RETENTION_MS), - limit, - }); - } catch (error) { - result.failed++; - logger.error("Dashboard agent turn-eval retention failed", { error }); - } - - if (result.failed > 0) { - throw new Error("The dashboard agent turn-eval retention pass failed"); - } - - return result; -} diff --git a/apps/webapp/app/services/dashboardAgentWatchSweep.server.ts b/apps/webapp/app/services/dashboardAgentWatchSweep.server.ts index 4535885441..c975e3d279 100644 --- a/apps/webapp/app/services/dashboardAgentWatchSweep.server.ts +++ b/apps/webapp/app/services/dashboardAgentWatchSweep.server.ts @@ -6,8 +6,6 @@ import { cancelWatch, claimWatchAlertDispatch, - deleteTerminalWatchesOlderThan, - deleteWatchSubmissionsOlderThan, listExpiredActiveWatches, listWatchBatchGroupsToArm, listWatchesAwaitingDelivery, @@ -57,12 +55,6 @@ export const WATCH_DELIVERY_GRACE_MS = 5 * 60 * 1000; /** Per-run cap for each half of the sweep. Oldest first, so the rest land next run. */ const SWEEP_BATCH_LIMIT = 100; -/** How long a terminal watch is kept. Its outcome also lives in the chat transcript. */ -export const WATCH_RETENTION_MS = 7 * 24 * 60 * 60 * 1000; - -/** Higher than the other caps: retention is one statement, not a row-at-a-time loop. */ -const RETENTION_BATCH_LIMIT = 500; - /** * How many rows one sweep handles at once. An incident expires a whole group together, and a * bound is what stops one slow tenant spending the entire visibility window. @@ -91,10 +83,6 @@ export type WatchSweepResult = { redelivered: number; /** Decided but not handed over, with no agent project. They stay owed. */ deliveryDeferred: number; - /** Long-terminal rows dropped by retention. */ - purged: number; - /** Ledger rows dropped by retention. */ - purgedSubmissions: number; failed: number; }; @@ -113,10 +101,6 @@ export type WatchSweepDeps = { deliver?: (watch: Watch) => Promise; /** Gates the delivery half only. Finalization never depends on it. */ configured?: () => boolean; - /** Drop terminal rows older than `before`. Returns how many went. */ - purgeTerminal?: (params: { before: Date; limit: number }) => Promise; - /** Drop submission-ledger rows older than `before`. */ - purgeSubmissions?: (params: { before: Date; limit: number }) => Promise; /** How many rows are handled at once. */ concurrency?: number; }; @@ -332,11 +316,6 @@ export async function sweepDashboardAgentWatches( const listAwaitingDelivery = deps.listAwaitingDelivery ?? ((params) => listWatchesAwaitingDelivery(dashboardAgentDb, params)); - const purgeTerminal = - deps.purgeTerminal ?? ((params) => deleteTerminalWatchesOlderThan(dashboardAgentDb, params)); - const purgeSubmissions = - deps.purgeSubmissions ?? - ((params) => deleteWatchSubmissionsOlderThan(dashboardAgentDb, params)); const result: WatchSweepResult = { overdue: 0, @@ -347,8 +326,6 @@ export async function sweepDashboardAgentWatches( undelivered: 0, redelivered: 0, deliveryDeferred: 0, - purged: 0, - purgedSubmissions: 0, failed: 0, }; @@ -435,18 +412,6 @@ export async function sweepDashboardAgentWatches( } } - // Retention runs last, over rows both halves are finished with. Its own try/catch so a - // lost retention pass can't mask the other failures. - try { - const before = new Date(now.getTime() - WATCH_RETENTION_MS); - result.purged = await purgeTerminal({ before, limit: RETENTION_BATCH_LIMIT }); - // The ledger's rows age out on the same window: past it no client is still retrying. - result.purgedSubmissions = await purgeSubmissions({ before, limit: RETENTION_BATCH_LIMIT }); - } catch (error) { - result.failed++; - logger.error("Dashboard agent watch sweep: failed to purge terminal watches", { error }); - } - if (result.failed > 0) { throw new Error(`The dashboard agent watch sweep failed on ${result.failed} watches`); } diff --git a/apps/webapp/app/v3/commonWorker.server.ts b/apps/webapp/app/v3/commonWorker.server.ts index 2203b9d744..5736458b12 100644 --- a/apps/webapp/app/v3/commonWorker.server.ts +++ b/apps/webapp/app/v3/commonWorker.server.ts @@ -11,12 +11,7 @@ import { runAttioUserSync, runAttioWorkspaceSync, } from "~/services/attio.server"; -import { - purgeDashboardAgentChatsForOrganization, - sweepDashboardAgentSoftDeletedChats, -} from "~/services/dashboardAgentChatRetention.server"; -import { sweepDashboardAgentTurnEvals } from "~/services/dashboardAgentEvalRetention.server"; -import { sweepDashboardAgentInvestigations } from "~/services/dashboardAgentInvestigationSweep.server"; +import { purgeDashboardAgentChatsForOrganization } from "~/services/dashboardAgentChatRetention.server"; import { rearmDashboardAgentWatchBatches, sweepDashboardAgentWatches, @@ -47,7 +42,7 @@ function initializeWorker() { logger.debug(`👨‍🏭 Initializing common worker at host ${env.COMMON_WORKER_REDIS_HOST}`); - // Only schedule the agent maintenance cron where the agent is actually set up. Otherwise + // Only schedule the agent watch cron where the agent is actually set up. Otherwise // its sweeps hit a missing schema and drip a dead-letter entry every run. const dashboardAgentConfigured = env.DASHBOARD_AGENT_ENABLED === "1" || Boolean(env.DASHBOARD_AGENT_DATABASE_URL); @@ -161,16 +156,15 @@ function initializeWorker() { maxAttempts: 5, }, }, - // Stuck investigation cards and turn-eval retention. + // @deprecated, moved to the dashboard agent project; remove once the queue drains. "dashboardAgent.maintenance": { schema: CronSchema, - visibilityTimeoutMs: 60_000 * 5, - ...(dashboardAgentConfigured ? { cron: "*/5 * * * *", jitterInMs: 30_000 } : {}), + visibilityTimeoutMs: 60_000, retry: { maxAttempts: 1, }, }, - // The watch backstops: expiry, wake redelivery, retention and dead batch chains. + // The watch backstops: expiry, wake redelivery and dead batch chains. "dashboardAgent.watchMaintenance": { schema: CronSchema, visibilityTimeoutMs: 60_000 * 5, @@ -179,7 +173,7 @@ function initializeWorker() { maxAttempts: 1, }, }, - // Soft-deletes a deleted organization's chats; the maintenance sweep purges them. + // Soft-deletes a deleted organization's chats; retention hard-deletes them later. "dashboardAgent.purgeOrganization": { schema: z.object({ organizationId: z.string(), @@ -247,48 +241,15 @@ function initializeWorker() { const service = new BulkActionService(); await service.process(payload.bulkActionId); }, - "dashboardAgent.maintenance": async () => { - // Each backstop runs independently; the first failure is rethrown at the end. - let failure: unknown; - - try { - const investigations = await sweepDashboardAgentInvestigations(); - if (investigations.stale > 0) { - logger.debug("Dashboard agent investigation sweep", investigations); - } - } catch (error) { - failure ??= error; - } - - // Retention on the judged-turn rows. Independent of the agent being configured. - try { - const evals = await sweepDashboardAgentTurnEvals(); - if (evals.purged > 0) { - logger.debug("Dashboard agent turn-eval retention", evals); - } - } catch (error) { - failure ??= error; - } - - // Hard-delete chats soft-deleted past the retention window, with their children. - try { - const chats = await sweepDashboardAgentSoftDeletedChats(); - if (chats.purged > 0) { - logger.debug("Dashboard agent chat retention", chats); - } - } catch (error) { - failure ??= error; - } - - if (failure) throw failure; - }, + // @deprecated, moved to the dashboard agent project; remove once the queue drains. + "dashboardAgent.maintenance": async () => {}, "dashboardAgent.watchMaintenance": async () => { // Each backstop runs independently; the first failure is rethrown at the end. let failure: unknown; try { const watches = await sweepDashboardAgentWatches(); - if (watches.overdue > 0 || watches.undelivered > 0 || watches.purged > 0) { + if (watches.overdue > 0 || watches.undelivered > 0) { logger.debug("Dashboard agent watch sweep", watches); } } catch (error) { diff --git a/apps/webapp/test/dashboardAgentChatRetention.test.ts b/apps/webapp/test/dashboardAgentChatRetention.test.ts index 57c9941e6c..9e5d1fc158 100644 --- a/apps/webapp/test/dashboardAgentChatRetention.test.ts +++ b/apps/webapp/test/dashboardAgentChatRetention.test.ts @@ -4,10 +4,9 @@ import { type DashboardAgentDb, type DashboardAgentDbClient, } from "@internal/dashboard-agent-db"; +import { applyDashboardAgentMigrations } from "@internal/dashboard-agent-db/testing"; import { postgresTest } from "@internal/testcontainers"; import type { PrismaClient } from "@trigger.dev/database"; -import { readdirSync, readFileSync } from "node:fs"; -import path from "node:path"; import { afterEach, describe, expect, vi } from "vitest"; const ctx = vi.hoisted(() => ({ @@ -20,32 +19,15 @@ vi.mock("~/services/dashboardAgentDb.server", () => ({ }, })); -const { sweepDashboardAgentSoftDeletedChats, purgeDashboardAgentChatsForOrganization } = +const { purgeDashboardAgentChatsForOrganization } = await import("~/services/dashboardAgentChatRetention.server"); -/** Replays every migration in order, so a new migration can't leave the suite on a stale schema. */ -async function applyAgentSchema(prisma: PrismaClient) { - const folder = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle"); - const migrations = readdirSync(folder) - .filter((file) => file.endsWith(".sql")) - .sort(); - for (const name of migrations) { - const sql = readFileSync(path.join(folder, name), "utf8"); - for (const statement of sql.split("--> statement-breakpoint")) { - const trimmed = statement.trim(); - if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed); - } - } -} - let agentDbClient: DashboardAgentDbClient | undefined; -let prismaForRaw: PrismaClient | undefined; async function boot(prisma: PrismaClient, connectionUri: string) { - await applyAgentSchema(prisma); + await applyDashboardAgentMigrations((statement) => prisma.$executeRawUnsafe(statement)); agentDbClient = createDashboardAgentDb(connectionUri, { max: 2 }); ctx.agentDb = agentDbClient.db; - prismaForRaw = prisma; } afterEach(async () => { @@ -56,115 +38,7 @@ afterEach(async () => { const ORG = "org_ret"; const USER = "user_ret"; -/** One row in every chatId-keyed table, so a delete that misses one leaves a leak. */ -async function seedChatWithChildren(id: string) { - await createChat(ctx.agentDb, { id, organizationId: ORG, userId: USER }); - const raw = prismaForRaw!; - await raw.$executeRawUnsafe( - `insert into trigger_dashboard_agent.chat_messages (chat_id, message_id, position, role, message) - values ($1, $1 || '-m', 1, 'user', '{}'::jsonb)`, - id - ); - await raw.$executeRawUnsafe( - `insert into trigger_dashboard_agent.chat_sessions (chat_id, public_access_token) values ($1, 'pat')`, - id - ); - await raw.$executeRawUnsafe( - `insert into trigger_dashboard_agent.chat_turn_evals (chat_id, turn, organization_id, user_id) - values ($1, 0, $2, $3)`, - id, - ORG, - USER - ); - await raw.$executeRawUnsafe( - `insert into trigger_dashboard_agent.investigations (id, chat_id, project_ref, environment_ref, state) - values ($1 || '-inv', $1, 'proj', 'env', '{"outcome":"in_progress"}'::jsonb)`, - id - ); - await raw.$executeRawUnsafe( - `insert into trigger_dashboard_agent.watches - (id, chat_id, identity, spec, organization_id, project_id, environment_id, user_id, expires_at) - values ($1 || '-w', $1, 'ident', '{}'::jsonb, $2, 'proj', 'env', $3, now() + interval '1 day')`, - id, - ORG, - USER - ); - await raw.$executeRawUnsafe( - `insert into trigger_dashboard_agent.watch_submissions - (chat_id, client_request_id, organization_id, user_id, project_id, environment_id, draft_hash, draft) - values ($1, 'req', $2, $3, 'proj', 'env', 'hash', '{}'::jsonb)`, - id, - ORG, - USER - ); -} - -async function setDeletedAtDaysAgo(id: string, days: number) { - await prismaForRaw!.$executeRawUnsafe( - `update trigger_dashboard_agent.chats set deleted_at = now() - ($2 || ' days')::interval where id = $1`, - id, - String(days) - ); -} - -const CHILD_TABLES = [ - "chat_messages", - "chat_sessions", - "chat_turn_evals", - "investigations", - "watches", - "watch_submissions", -]; - -async function rowCounts(id: string): Promise> { - const counts: Record = {}; - const chat = await prismaForRaw!.$queryRawUnsafe<{ n: bigint }[]>( - `select count(*)::int as n from trigger_dashboard_agent.chats where id = $1`, - id - ); - counts.chats = Number(chat[0]!.n); - for (const table of CHILD_TABLES) { - const rows = await prismaForRaw!.$queryRawUnsafe<{ n: number }[]>( - `select count(*)::int as n from trigger_dashboard_agent.${table} where chat_id = $1`, - id - ); - counts[table] = Number(rows[0]!.n); - } - return counts; -} - -describe("the dashboard agent chat retention sweep", () => { - postgresTest( - "hard-deletes only chats soft-deleted past the window, with every child row", - async ({ prisma, postgresContainer }) => { - await boot(prisma, postgresContainer.getConnectionUri()); - - await seedChatWithChildren("chat_old"); - await setDeletedAtDaysAgo("chat_old", 40); // past the 30d window - - await seedChatWithChildren("chat_recent"); - await setDeletedAtDaysAgo("chat_recent", 1); // inside the window - - await seedChatWithChildren("chat_live"); // never deleted - - const result = await sweepDashboardAgentSoftDeletedChats(); - expect(result).toEqual({ purged: 1, failed: 0 }); - - const old = await rowCounts("chat_old"); - for (const table of ["chats", ...CHILD_TABLES]) { - expect(old[table], `${table} should be empty for the purged chat`).toBe(0); - } - - const recent = await rowCounts("chat_recent"); - const live = await rowCounts("chat_live"); - for (const table of ["chats", ...CHILD_TABLES]) { - expect(recent[table], `${table} kept for the in-window chat`).toBe(1); - expect(live[table], `${table} kept for the live chat`).toBe(1); - } - }, - 30_000 - ); - +describe("the dashboard agent chat org purge", () => { postgresTest( "org purge soft-deletes the org's chats and leaves other orgs alone", async ({ prisma, postgresContainer }) => { diff --git a/apps/webapp/test/dashboardAgentDurableResume.test.ts b/apps/webapp/test/dashboardAgentDurableResume.test.ts index 8fd75d52e0..20a3575457 100644 --- a/apps/webapp/test/dashboardAgentDurableResume.test.ts +++ b/apps/webapp/test/dashboardAgentDurableResume.test.ts @@ -9,10 +9,9 @@ import { type DashboardAgentDb, type DashboardAgentDbClient, } from "@internal/dashboard-agent-db"; +import { applyDashboardAgentMigrations } from "@internal/dashboard-agent-db/testing"; import { postgresTest } from "@internal/testcontainers"; import type { PrismaClient } from "@trigger.dev/database"; -import { readdirSync, readFileSync } from "node:fs"; -import path from "node:path"; import { afterEach, describe, expect } from "vitest"; /** @@ -35,25 +34,11 @@ import { afterEach, describe, expect } from "vitest"; let agentDb: DashboardAgentDb; let agentDbClient: DashboardAgentDbClient | undefined; -const MIGRATIONS = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle"); - -async function applyAgentSchema(prisma: PrismaClient) { - for (const name of readdirSync(MIGRATIONS) - .filter((file) => file.endsWith(".sql")) - .sort()) { - const sql = readFileSync(path.join(MIGRATIONS, name), "utf8"); - for (const statement of sql.split("--> statement-breakpoint")) { - const trimmed = statement.trim(); - if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed); - } - } -} - const ORG = "org_resume"; const USER = "user_resume"; async function boot(prisma: PrismaClient, connectionUri: string, chatId: string) { - await applyAgentSchema(prisma); + await applyDashboardAgentMigrations((statement) => prisma.$executeRawUnsafe(statement)); agentDbClient = createDashboardAgentDb(connectionUri, { max: 4 }); agentDb = agentDbClient.db; await createChat(agentDb, { id: chatId, organizationId: ORG, userId: USER }); diff --git a/apps/webapp/test/dashboardAgentEvalRetention.test.ts b/apps/webapp/test/dashboardAgentEvalRetention.test.ts deleted file mode 100644 index 4a36876e5a..0000000000 --- a/apps/webapp/test/dashboardAgentEvalRetention.test.ts +++ /dev/null @@ -1,135 +0,0 @@ -import { - createDashboardAgentDb, - insertTurnEval, - type DashboardAgentDb, - type DashboardAgentDbClient, -} from "@internal/dashboard-agent-db"; -import { postgresTest } from "@internal/testcontainers"; -import type { PrismaClient } from "@trigger.dev/database"; -import { readdirSync, readFileSync } from "node:fs"; -import path from "node:path"; -import { afterEach, describe, expect, vi } from "vitest"; - -const ctx = vi.hoisted(() => ({ - agentDb: undefined as unknown as DashboardAgentDb, -})); - -vi.mock("~/services/dashboardAgentDb.server", () => ({ - get dashboardAgentDb() { - return ctx.agentDb; - }, -})); - -const { sweepDashboardAgentTurnEvals, TURN_EVAL_RETENTION_MS } = - await import("~/services/dashboardAgentEvalRetention.server"); - -/** Replays every migration in order, so a new migration can't leave the suite on a stale schema. */ -async function applyAgentSchema(prisma: PrismaClient) { - const folder = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle"); - const migrations = readdirSync(folder) - .filter((file) => file.endsWith(".sql")) - .sort(); - for (const name of migrations) { - const sql = readFileSync(path.join(folder, name), "utf8"); - for (const statement of sql.split("--> statement-breakpoint")) { - const trimmed = statement.trim(); - if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed); - } - } -} - -let agentDbClient: DashboardAgentDbClient | undefined; -let prismaForRaw: PrismaClient | undefined; - -async function boot(prisma: PrismaClient, connectionUri: string) { - await applyAgentSchema(prisma); - agentDbClient = createDashboardAgentDb(connectionUri, { max: 2 }); - ctx.agentDb = agentDbClient.db; - prismaForRaw = prisma; -} - -afterEach(async () => { - await agentDbClient?.close(); - agentDbClient = undefined; -}); - -async function seedEval(args: { chatId: string; turn: number; ageMs: number }) { - await insertTurnEval(ctx.agentDb, { - chatId: args.chatId, - turn: args.turn, - organizationId: "org_retention", - userId: "user_retention", - summary: "the user asked about a failed run", - }); - await prismaForRaw!.$executeRawUnsafe( - `update trigger_dashboard_agent.chat_turn_evals - set created_at = now() - ($3 || ' seconds')::interval - where chat_id = $1 and turn = $2`, - args.chatId, - args.turn, - String(args.ageMs / 1000) - ); -} - -async function remaining(): Promise> { - return prismaForRaw!.$queryRawUnsafe( - `select chat_id, turn from trigger_dashboard_agent.chat_turn_evals order by chat_id, turn` - ); -} - -describe("dashboard agent turn-eval retention", () => { - postgresTest( - "drops only rows past the retention period", - async ({ prisma, postgresContainer }) => { - await boot(prisma, postgresContainer.getConnectionUri()); - - await seedEval({ chatId: "chat_old", turn: 0, ageMs: TURN_EVAL_RETENTION_MS + 60_000 }); - await seedEval({ chatId: "chat_edge", turn: 0, ageMs: TURN_EVAL_RETENTION_MS - 60_000 }); - await seedEval({ chatId: "chat_new", turn: 0, ageMs: 0 }); - - const result = await sweepDashboardAgentTurnEvals(); - expect(result).toEqual({ purged: 1, failed: 0 }); - - const rows = await remaining(); - expect(rows.map((row) => row.chat_id)).toEqual(["chat_edge", "chat_new"]); - } - ); - - postgresTest( - "stops at the batch cap and drains on the next run", - async ({ prisma, postgresContainer }) => { - await boot(prisma, postgresContainer.getConnectionUri()); - - for (let turn = 0; turn < 5; turn++) { - await seedEval({ - chatId: "chat_backlog", - turn, - // Oldest first, so the cap takes a deterministic slice. - ageMs: TURN_EVAL_RETENTION_MS + 60_000 + (5 - turn) * 1_000, - }); - } - - const first = await sweepDashboardAgentTurnEvals({ limit: 2 }); - expect(first).toEqual({ purged: 2, failed: 0 }); - expect(await remaining()).toHaveLength(3); - - const second = await sweepDashboardAgentTurnEvals({ limit: 2 }); - expect(second).toEqual({ purged: 2, failed: 0 }); - - const third = await sweepDashboardAgentTurnEvals({ limit: 2 }); - expect(third).toEqual({ purged: 1, failed: 0 }); - expect(await remaining()).toHaveLength(0); - } - ); - - postgresTest( - "keeps everything when nothing is old enough", - async ({ prisma, postgresContainer }) => { - await boot(prisma, postgresContainer.getConnectionUri()); - await seedEval({ chatId: "chat_fresh", turn: 0, ageMs: 0 }); - - expect(await sweepDashboardAgentTurnEvals()).toEqual({ purged: 0, failed: 0 }); - expect(await remaining()).toHaveLength(1); - } - ); -}); diff --git a/apps/webapp/test/dashboardAgentInvestigationPoison.test.ts b/apps/webapp/test/dashboardAgentInvestigationPoison.test.ts deleted file mode 100644 index 52b87f35c1..0000000000 --- a/apps/webapp/test/dashboardAgentInvestigationPoison.test.ts +++ /dev/null @@ -1,166 +0,0 @@ -import { - createChat, - createDashboardAgentDb, - getInvestigation, - settleInvestigationAndCloseCard, - upsertInvestigationRevision, - type DashboardAgentDb, - type DashboardAgentDbClient, -} from "@internal/dashboard-agent-db"; -import { - investigationStateSchema, - type InvestigationState, -} from "@internal/dashboard-agent-contracts"; -import { postgresTest } from "@internal/testcontainers"; -import type { PrismaClient } from "@trigger.dev/database"; -import { readdirSync, readFileSync } from "node:fs"; -import path from "node:path"; -import { afterEach, describe, expect, vi } from "vitest"; - -const ctx = vi.hoisted(() => ({ - agentDb: undefined as unknown as DashboardAgentDb, -})); - -vi.mock("~/services/dashboardAgentDb.server", () => ({ - get dashboardAgentDb() { - return ctx.agentDb; - }, -})); - -const { sweepDashboardAgentInvestigations, INVESTIGATION_STALE_MS, MAX_SWEEP_ATTEMPTS } = - await import("~/services/dashboardAgentInvestigationSweep.server"); - -async function applyAgentSchema(prisma: PrismaClient) { - const folder = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle"); - const migrations = readdirSync(folder) - .filter((file) => file.endsWith(".sql")) - .sort(); - for (const name of migrations) { - const sql = readFileSync(path.join(folder, name), "utf8"); - for (const statement of sql.split("--> statement-breakpoint")) { - const trimmed = statement.trim(); - if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed); - } - } -} - -let agentDbClient: DashboardAgentDbClient | undefined; -let prismaForRaw: PrismaClient | undefined; - -async function boot(prisma: PrismaClient, connectionUri: string) { - await applyAgentSchema(prisma); - agentDbClient = createDashboardAgentDb(connectionUri, { max: 2 }); - ctx.agentDb = agentDbClient.db; - prismaForRaw = prisma; -} - -afterEach(async () => { - await agentDbClient?.close(); - agentDbClient = undefined; -}); - -const ORG = "org_poison"; -const USER = "user_poison"; - -function openState(): InvestigationState { - return investigationStateSchema.parse({ - outcome: "in_progress", - severity: "warn", - confidence: "medium", - title: "a stuck card", - headline: "Still checking.", - progress: "Reading spans", - checkNext: [], - hypotheses: [], - evidence: [], - }); -} - -async function seedInvestigation(chatId: string, ageMs: number): Promise { - await createChat(ctx.agentDb, { id: chatId, organizationId: ORG, userId: USER }); - const created = await upsertInvestigationRevision(ctx.agentDb, { - chatId, - projectRef: "proj", - environmentRef: "env", - state: openState(), - }); - if (!created.ok) throw new Error("fixture investigation not created"); - await prismaForRaw!.$executeRawUnsafe( - `update trigger_dashboard_agent.investigations - set updated_at = now() - ($2 || ' milliseconds')::interval where id = $1`, - created.id, - String(ageMs) - ); - return created.id; -} - -async function outcomeOf(id: string): Promise { - const row = await getInvestigation(ctx.agentDb, { id }); - return row ? (row.state as { outcome?: string }).outcome : undefined; -} - -const STALE_AGE_MS = INVESTIGATION_STALE_MS + 60_000; -const OLDER_AGE_MS = STALE_AGE_MS + 60_000; - -describe("the investigation sweep with a poison row", () => { - postgresTest( - "a row that always fails to settle cannot pin the head and starve a newer row", - async ({ prisma, postgresContainer }) => { - await boot(prisma, postgresContainer.getConnectionUri()); - - // Poison sorts first (older `updated_at`); renderable is newer. - const poisonId = await seedInvestigation("chat_poison", OLDER_AGE_MS); - const renderableId = await seedInvestigation("chat_ok", STALE_AGE_MS); - - // Only the poison row's settle throws; the renderable one goes through the real path. - const settleAndClose = (params: { id: string; chatId: string; note: string }) => { - if (params.id === poisonId) throw new Error("state isn't renderable"); - return settleInvestigationAndCloseCard(ctx.agentDb, params); - }; - - // limit 1 forces head contention: without backoff the poison row would win every run. - // A failed run throws so the job retries, but the attempt is recorded before it does. - await expect( - sweepDashboardAgentInvestigations({ limit: 1, settleAndClose }) - ).rejects.toThrow(); - expect(await outcomeOf(poisonId)).toBe("in_progress"); - expect(await outcomeOf(renderableId)).toBe("in_progress"); - - // Next run: the poison row now sorts behind the never-attempted renderable one, - // so the newer row is picked and settled despite the poison row still being stale. - const second = await sweepDashboardAgentInvestigations({ limit: 1, settleAndClose }); - expect(second).toMatchObject({ stale: 1, settled: 1, failed: 0 }); - expect(await outcomeOf(renderableId)).toBe("inconclusive"); - expect(await outcomeOf(poisonId)).toBe("in_progress"); - }, - 30_000 - ); - - postgresTest( - "after the attempt cap the poison row is abandoned and leaves the queue", - async ({ prisma, postgresContainer }) => { - await boot(prisma, postgresContainer.getConnectionUri()); - const poisonId = await seedInvestigation("chat_poison", STALE_AGE_MS); - - const settleAndClose = () => { - throw new Error("state isn't renderable"); - }; - - // The first MAX_SWEEP_ATTEMPTS-1 runs record a failed attempt and throw; the row stays stale. - for (let i = 1; i < MAX_SWEEP_ATTEMPTS; i++) { - await expect(sweepDashboardAgentInvestigations({ settleAndClose })).rejects.toThrow(); - expect(await outcomeOf(poisonId)).toBe("in_progress"); - } - - // The capped run force-settles the row without the render path, so it leaves the queue. - const capped = await sweepDashboardAgentInvestigations({ settleAndClose }); - expect(capped).toMatchObject({ stale: 1, abandoned: 1, failed: 0 }); - expect(await outcomeOf(poisonId)).toBe("inconclusive"); - - // Nothing stale remains, so the poison row is no longer swept. - const after = await sweepDashboardAgentInvestigations({ settleAndClose }); - expect(after).toMatchObject({ stale: 0 }); - }, - 30_000 - ); -}); diff --git a/apps/webapp/test/dashboardAgentInvestigationSettlementCard.test.ts b/apps/webapp/test/dashboardAgentInvestigationSettlementCard.test.ts new file mode 100644 index 0000000000..c7759876b3 --- /dev/null +++ b/apps/webapp/test/dashboardAgentInvestigationSettlementCard.test.ts @@ -0,0 +1,83 @@ +import { + forceSettledInvestigationState, + investigationStateSchema, + VIEW_BLOCK_VERSION, + type InvestigationState, +} from "@internal/dashboard-agent-contracts"; +import { + investigationSettlementMessage, + investigationSettlementMessageId, +} from "@internal/dashboard-agent-db"; +import { describe, expect, it } from "vitest"; +import { liveInvestigation } from "~/components/dashboard-agent/progress-line"; + +/** + * The card the sweep appends is what stops the panel's spinner: the panel resolves the + * investigation from the chat's own `render_view` parts, never from the settled row. + */ + +const INVESTIGATION_ID = "inv_settlement_card"; + +function openState(): InvestigationState { + return investigationStateSchema.parse({ + outcome: "in_progress", + severity: "warn", + confidence: "medium", + title: "send-order-receipt keeps failing", + headline: "Checking whether the failures share a payload.", + progress: "Reading the run's spans", + hypotheses: [ + { + id: "h1", + statement: "The new payload dropped a field the task reads.", + verdict: "testing", + evidence: [], + }, + ], + evidence: [], + }); +} + +function openCardMessage(state: InvestigationState) { + return { + id: "msg_open", + role: "assistant", + parts: [ + { + type: "tool-render_view", + toolCallId: "tc_open", + state: "output-available", + input: { blocks: [{ type: "investigation", investigation: state }] }, + output: { + blocks: [ + { + type: "investigation", + investigation: state, + id: INVESTIGATION_ID, + revision: 0, + version: VIEW_BLOCK_VERSION, + }, + ], + }, + }, + ], + }; +} + +describe("the investigation settlement card", () => { + it("ends the panel's spinner once appended to the transcript", () => { + const open = openState(); + const message = investigationSettlementMessage({ + investigationId: INVESTIGATION_ID, + revision: 1, + state: forceSettledInvestigationState(open), + }); + + expect(message).not.toBeNull(); + expect(message!.id).toBe(investigationSettlementMessageId(INVESTIGATION_ID, 1)); + + const transcript = [openCardMessage(open)]; + expect(liveInvestigation(transcript)).toEqual({ progress: "Reading the run's spans" }); + expect(liveInvestigation([...transcript, message as never])).toBeNull(); + }); +}); diff --git a/apps/webapp/test/dashboardAgentInvestigationSweep.test.ts b/apps/webapp/test/dashboardAgentInvestigationSweep.test.ts deleted file mode 100644 index 546ef41d4b..0000000000 --- a/apps/webapp/test/dashboardAgentInvestigationSweep.test.ts +++ /dev/null @@ -1,395 +0,0 @@ -import { - createChat, - createDashboardAgentDb, - getChatMessages, - getInvestigation, - investigationSettlementMessageId, - listStaleOpenInvestigations, - settleInvestigationAndCloseCard, - upsertInvestigationRevision, - type DashboardAgentDb, - type DashboardAgentDbClient, -} from "@internal/dashboard-agent-db"; -import { - investigationStateSchema, - UNSETTLED_INVESTIGATION_NOTE, - type InvestigationState, -} from "@internal/dashboard-agent-contracts"; -import { postgresTest } from "@internal/testcontainers"; -import type { PrismaClient } from "@trigger.dev/database"; -import { readdirSync, readFileSync } from "node:fs"; -import path from "node:path"; -import { afterEach, describe, expect, vi } from "vitest"; - -const ctx = vi.hoisted(() => ({ - agentDb: undefined as unknown as DashboardAgentDb, -})); - -vi.mock("~/services/dashboardAgentDb.server", () => ({ - get dashboardAgentDb() { - return ctx.agentDb; - }, -})); - -const { sweepDashboardAgentInvestigations, INVESTIGATION_STALE_MS } = - await import("~/services/dashboardAgentInvestigationSweep.server"); - -/** Replays every migration in order, so a new migration can't leave the suite on a stale schema. */ -async function applyAgentSchema(prisma: PrismaClient) { - const folder = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle"); - const migrations = readdirSync(folder) - .filter((file) => file.endsWith(".sql")) - .sort(); - for (const name of migrations) { - const sql = readFileSync(path.join(folder, name), "utf8"); - for (const statement of sql.split("--> statement-breakpoint")) { - const trimmed = statement.trim(); - if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed); - } - } -} - -let agentDbClient: DashboardAgentDbClient | undefined; -let prismaForRaw: PrismaClient | undefined; - -async function boot(prisma: PrismaClient, connectionUri: string) { - await applyAgentSchema(prisma); - agentDbClient = createDashboardAgentDb(connectionUri, { max: 2 }); - ctx.agentDb = agentDbClient.db; - prismaForRaw = prisma; -} - -afterEach(async () => { - await agentDbClient?.close(); - agentDbClient = undefined; -}); - -const PROJECT_REF = "proj_sweep"; -const ENV_REF = "env_sweep"; - -async function seedChat(id: string, options: { deleted?: boolean } = {}) { - await createChat(ctx.agentDb, { - id, - organizationId: "org_sweep", - userId: "user_sweep", - }); - if (options.deleted) { - await prismaForRaw!.$executeRawUnsafe( - `update trigger_dashboard_agent.chats set deleted_at = now() where id = $1`, - id - ); - } -} - -function openState(overrides: Partial = {}): InvestigationState { - return investigationStateSchema.parse({ - outcome: "in_progress", - severity: "warn", - confidence: "medium", - title: "send-order-receipt keeps failing", - headline: "Checking whether the failures share a payload.", - progress: "Reading the run's spans", - checkNext: [], - hypotheses: [ - { - id: "h1", - statement: "The new payload dropped a field the task reads.", - verdict: "testing", - evidence: [], - }, - ], - evidence: [], - ...overrides, - }); -} - -async function seedInvestigation(args: { - chatId: string; - state: InvestigationState; - ageMs?: number; -}): Promise { - const created = await upsertInvestigationRevision(ctx.agentDb, { - chatId: args.chatId, - projectRef: PROJECT_REF, - environmentRef: ENV_REF, - state: args.state, - }); - if (!created.ok) throw new Error("the fixture investigation wasn't created"); - - if (args.ageMs !== undefined) { - await prismaForRaw!.$executeRawUnsafe( - `update trigger_dashboard_agent.investigations - set updated_at = now() - ($2 || ' milliseconds')::interval where id = $1`, - created.id, - String(args.ageMs) - ); - } - return created.id; -} - -/** Comfortably past the grace window. */ -const STALE_AGE_MS = INVESTIGATION_STALE_MS + 60_000; - -describe("the dashboard agent investigation sweep", () => { - postgresTest( - "settles a card left in_progress, keeping what was established", - async ({ prisma, postgresContainer }) => { - await boot(prisma, postgresContainer.getConnectionUri()); - await seedChat("chat_stale"); - const id = await seedInvestigation({ - chatId: "chat_stale", - state: openState(), - ageMs: STALE_AGE_MS, - }); - - const result = await sweepDashboardAgentInvestigations(); - expect(result).toMatchObject({ stale: 1, settled: 1, alreadySettled: 0, failed: 0 }); - - const row = await getInvestigation(ctx.agentDb, { id }); - const state = investigationStateSchema.parse(row?.state); - expect(state.outcome).toBe("inconclusive"); - expect(state.confidence).toBe("low"); - expect(state.headline).toBe( - `Checking whether the failures share a payload. ${UNSETTLED_INVESTIGATION_NOTE}` - ); - expect(state.progress).toBeUndefined(); - expect(state.remediation).toBeUndefined(); - expect(state.hypotheses).toHaveLength(1); - expect(state.title).toBe("send-order-receipt keeps failing"); - expect(row?.revision).toBe(1); - - // The settled row is invisible on its own: the panel resolves the card from the - // transcript, so the closing revision has to be in the chat too. - const messages = (await getChatMessages(ctx.agentDb, { - chatId: "chat_stale", - userId: "user_sweep", - organizationId: "org_sweep", - })) as { id: string; parts: Record[] }[] | null; - expect(messages?.map((message) => message.id)).toEqual([ - investigationSettlementMessageId(id, 1), - ]); - const block = messages![0]!.parts[0]!.output.blocks[0]; - expect(block).toMatchObject({ type: "investigation", id, revision: 1 }); - expect(block.investigation.outcome).toBe("inconclusive"); - expect(result.closed).toBe(1); - - // A second run can't stack a second card: the settle is a no-op and the append - // is deduped on the same message id. - expect(await sweepDashboardAgentInvestigations()).toMatchObject({ stale: 0, settled: 0 }); - expect( - ( - (await getChatMessages(ctx.agentDb, { - chatId: "chat_stale", - userId: "user_sweep", - organizationId: "org_sweep", - })) as unknown[] - ).length - ).toBe(1); - }, - // This one pays the container boot and the schema replay, and now asserts the - // transcript on top. - 30_000 - ); - - postgresTest( - "drops a fix a concluded card was carrying", - async ({ prisma, postgresContainer }) => { - await boot(prisma, postgresContainer.getConnectionUri()); - await seedChat("chat_fix"); - // An inconclusive card may not offer a fix, so the settle strips remediation. - const id = await seedInvestigation({ - chatId: "chat_fix", - state: { ...openState(), remediation: "Raise the timeout." } as InvestigationState, - ageMs: STALE_AGE_MS, - }); - - await sweepDashboardAgentInvestigations(); - - const row = await getInvestigation(ctx.agentDb, { id }); - expect(investigationStateSchema.parse(row?.state).remediation).toBeUndefined(); - } - ); - - postgresTest( - "leaves a fresh in_progress card alone — a live turn is never swept", - async ({ prisma, postgresContainer }) => { - await boot(prisma, postgresContainer.getConnectionUri()); - await seedChat("chat_fresh"); - const id = await seedInvestigation({ chatId: "chat_fresh", state: openState() }); - - expect(await sweepDashboardAgentInvestigations()).toMatchObject({ stale: 0, settled: 0 }); - const row = await getInvestigation(ctx.agentDb, { id }); - expect(investigationStateSchema.parse(row?.state).outcome).toBe("in_progress"); - expect(row?.revision).toBe(0); - } - ); - - postgresTest( - "leaves a card that already has an answer alone", - async ({ prisma, postgresContainer }) => { - await boot(prisma, postgresContainer.getConnectionUri()); - await seedChat("chat_done"); - const id = await seedInvestigation({ - chatId: "chat_done", - state: openState({ - outcome: "inconclusive", - progress: undefined, - headline: "Not established: the failures span two versions.", - }), - ageMs: STALE_AGE_MS, - }); - - expect(await sweepDashboardAgentInvestigations()).toMatchObject({ stale: 0, settled: 0 }); - const row = await getInvestigation(ctx.agentDb, { id }); - expect(row?.revision).toBe(0); - expect(investigationStateSchema.parse(row?.state).headline).toBe( - "Not established: the failures span two versions." - ); - } - ); - - postgresTest("skips a card in a deleted chat", async ({ prisma, postgresContainer }) => { - await boot(prisma, postgresContainer.getConnectionUri()); - await seedChat("chat_gone", { deleted: true }); - await seedInvestigation({ - chatId: "chat_gone", - state: openState(), - ageMs: STALE_AGE_MS, - }); - - expect(await sweepDashboardAgentInvestigations()).toMatchObject({ stale: 0, settled: 0 }); - }); - - postgresTest( - "a turn that concludes the card first wins: the settle is a no-op, not an error", - async ({ prisma, postgresContainer }) => { - await boot(prisma, postgresContainer.getConnectionUri()); - await seedChat("chat_race"); - const id = await seedInvestigation({ - chatId: "chat_race", - state: openState(), - ageMs: STALE_AGE_MS, - }); - - const stale = await listStaleOpenInvestigations(ctx.agentDb, { - olderThan: new Date(), - limit: 10, - }); - expect(stale.map((row) => row.id)).toEqual([id]); - - const concluded = await upsertInvestigationRevision(ctx.agentDb, { - id, - chatId: "chat_race", - projectRef: PROJECT_REF, - environmentRef: ENV_REF, - state: openState({ - outcome: "concluded", - confidence: "high", - progress: undefined, - headline: "receipt.ts:42 reads a field the new payload no longer carries.", - remediation: "Guard the dereference and backfill the field.", - }), - }); - expect(concluded.ok).toBe(true); - - const result = await sweepDashboardAgentInvestigations({ listStale: async () => stale }); - expect(result).toMatchObject({ - stale: 1, - settled: 0, - closed: 0, - alreadySettled: 1, - failed: 0, - }); - - const row = await getInvestigation(ctx.agentDb, { id }); - const state = investigationStateSchema.parse(row?.state); - expect(state.outcome).toBe("concluded"); - expect(state.headline).not.toContain(UNSETTLED_INVESTIGATION_NOTE); - expect(row?.revision).toBe(1); - } - ); - - /** - * The failure window. Settling the row and delivering its card used to be two - * operations: once the row was terminal, a failed append left a card reading - * `in_progress` that nothing would ever repair, because this sweep only selects - * `in_progress` rows. They must land together or not at all. - */ - postgresTest( - "a card that can't be delivered leaves the row in_progress, so the next run retries it", - async ({ prisma, postgresContainer }) => { - await boot(prisma, postgresContainer.getConnectionUri()); - await seedChat("chat_undeliverable"); - // A state the settle can merge but no card can be rendered from, so the delivery - // half genuinely fails against a real database. - const id = await seedInvestigation({ - chatId: "chat_undeliverable", - state: { outcome: "in_progress" } as unknown as InvestigationState, - ageMs: STALE_AGE_MS, - }); - - await expect(sweepDashboardAgentInvestigations()).rejects.toThrow( - /failed on 1 investigations/ - ); - - // The settle rolled back with the card: no half-applied terminal row. - const row = await getInvestigation(ctx.agentDb, { id }); - expect(row?.revision).toBe(0); - expect((row?.state as { outcome?: string }).outcome).toBe("in_progress"); - expect( - await getChatMessages(ctx.agentDb, { - chatId: "chat_undeliverable", - userId: "user_sweep", - organizationId: "org_sweep", - }) - ).toEqual([]); - - // And it is still in the selection, so the sweep keeps trying rather than - // leaving a permanent spinner behind. - const stale = await listStaleOpenInvestigations(ctx.agentDb, { - olderThan: new Date(), - limit: 10, - }); - expect(stale.map((candidate) => candidate.id)).toEqual([id]); - }, - 30_000 - ); - - postgresTest( - "one failing row doesn't cost the batch, and the run throws so the job retries", - async ({ prisma, postgresContainer }) => { - await boot(prisma, postgresContainer.getConnectionUri()); - await seedChat("chat_batch"); - const first = await seedInvestigation({ - chatId: "chat_batch", - state: openState(), - ageMs: STALE_AGE_MS + 60_000, - }); - const second = await seedInvestigation({ - chatId: "chat_batch", - state: openState(), - ageMs: STALE_AGE_MS, - }); - - const attempted: string[] = []; - await expect( - sweepDashboardAgentInvestigations({ - settleAndClose: async (params) => { - attempted.push(params.id); - if (params.id === first) throw new Error("the settle failed"); - return settleInvestigationAndCloseCard(ctx.agentDb, params); - }, - }) - ).rejects.toThrow(/failed on 1 investigations/); - - expect(attempted).toEqual([first, second]); - expect( - investigationStateSchema.parse((await getInvestigation(ctx.agentDb, { id: second }))?.state) - .outcome - ).toBe("inconclusive"); - const stuck = await getInvestigation(ctx.agentDb, { id: first }); - expect(stuck?.revision).toBe(0); - expect(investigationStateSchema.parse(stuck?.state).outcome).toBe("in_progress"); - } - ); -}); diff --git a/apps/webapp/test/dashboardAgentInvestigationSweepCard.test.ts b/apps/webapp/test/dashboardAgentInvestigationSweepCard.test.ts deleted file mode 100644 index 7f0a321fd7..0000000000 --- a/apps/webapp/test/dashboardAgentInvestigationSweepCard.test.ts +++ /dev/null @@ -1,188 +0,0 @@ -import { - investigationSettlementMessage, - investigationSettlementMessageId, - type Investigation, - type InvestigationCardMessage, -} from "@internal/dashboard-agent-db"; -import { - forceSettledInvestigationState, - investigationStateSchema, - UNSETTLED_INVESTIGATION_NOTE, - VIEW_BLOCK_VERSION, - type InvestigationState, -} from "@internal/dashboard-agent-contracts"; -import { describe, expect, it, vi } from "vitest"; -import { liveInvestigation } from "~/components/dashboard-agent/progress-line"; - -// The sweep's own datastore is never reached here: every write is injected. The -// connection is only stubbed so importing the service doesn't open a pool. -vi.mock("~/services/dashboardAgentDb.server", () => ({ dashboardAgentDb: undefined })); - -import { sweepDashboardAgentInvestigations } from "~/services/dashboardAgentInvestigationSweep.server"; - -/** - * The transcript half of the sweep, without a container: settling the row is invisible - * to the panel, which resolves a card from the chat's own `render_view` parts. - * - * The database half — that the closing card actually lands in `chat_messages` — is - * covered by `dashboardAgentInvestigationSweep.test.ts`, which needs Postgres. - */ - -const CHAT_ID = "chat_sweep_card"; -const INVESTIGATION_ID = "inv_sweep_card"; - -function openState(): InvestigationState { - return investigationStateSchema.parse({ - outcome: "in_progress", - severity: "warn", - confidence: "medium", - title: "send-order-receipt keeps failing", - headline: "Checking whether the failures share a payload.", - progress: "Reading the run's spans", - hypotheses: [ - { - id: "h1", - statement: "The new payload dropped a field the task reads.", - verdict: "testing", - evidence: [], - }, - ], - evidence: [], - }); -} - -/** The card the agent left behind, as the transcript holds it. */ -function openCardMessage(state: InvestigationState) { - return { - id: "msg_open", - role: "assistant", - parts: [ - { - type: "tool-render_view", - toolCallId: "tc_open", - state: "output-available", - input: { blocks: [{ type: "investigation", investigation: state }] }, - output: { - blocks: [ - { - type: "investigation", - investigation: state, - id: INVESTIGATION_ID, - revision: 0, - version: VIEW_BLOCK_VERSION, - }, - ], - }, - }, - ], - }; -} - -function staleRow(state: InvestigationState): Investigation { - return { - id: INVESTIGATION_ID, - chatId: CHAT_ID, - projectRef: "proj_sweep", - environmentRef: "env_sweep", - revision: 0, - state, - createdAt: new Date(), - updatedAt: new Date(), - } as Investigation; -} - -/** - * Stands in for the datastore: the conditional settle (`in_progress` only, mirroring - * `forceSettledInvestigationState`) and the id-deduped append, as one operation — - * which is what the real query is, so a half-applied settle can't exist. - */ -function fakeStore(initial: InvestigationState) { - const row = { revision: 0, state: initial }; - const appended: InvestigationCardMessage[] = []; - return { - row, - appended, - settleAndClose: async (params: { id: string; chatId: string; note: string }) => { - if (investigationStateSchema.parse(row.state).outcome !== "in_progress") return null; - const state = forceSettledInvestigationState(investigationStateSchema.parse(row.state)); - const revision = row.revision + 1; - - const message = investigationSettlementMessage({ - investigationId: params.id, - revision, - state, - }); - if (!message) throw new Error("the closing card didn't validate"); - - row.state = state; - row.revision = revision; - const closed = !appended.some((existing) => existing.id === message.id); - if (closed) appended.push(message); - return { settled: { id: params.id, revision, state }, closed }; - }, - }; -} - -describe("the dashboard agent investigation sweep's closing card", () => { - it("appends the terminal card to the chat, so the panel stops spinning", async () => { - const open = openState(); - const store = fakeStore(open); - - const result = await sweepDashboardAgentInvestigations({ - listStale: async () => [staleRow(open)], - settleAndClose: store.settleAndClose, - }); - - expect(store.appended).toHaveLength(1); - expect(result).toMatchObject({ stale: 1, settled: 1, closed: 1, alreadySettled: 0, failed: 0 }); - - const message = store.appended[0]!; - expect(message.id).toBe(investigationSettlementMessageId(INVESTIGATION_ID, 1)); - expect(message.role).toBe("assistant"); - - const part = message.parts[0] as { - type: string; - state: string; - output: { blocks: Record[] }; - }; - expect(part.type).toBe("tool-render_view"); - expect(part.state).toBe("output-available"); - expect(part.output.blocks[0]).toMatchObject({ - type: "investigation", - id: INVESTIGATION_ID, - revision: 1, - version: VIEW_BLOCK_VERSION, - }); - const settled = part.output.blocks[0]!.investigation; - expect(settled.outcome).toBe("inconclusive"); - expect(settled.confidence).toBe("low"); - expect(settled.progress).toBeUndefined(); - expect(settled.headline).toContain(UNSETTLED_INVESTIGATION_NOTE); - // What was checked survives: the card closes honestly, it doesn't get blanked. - expect(settled.hypotheses).toHaveLength(1); - - // The panel, over the transcript a refresh loads: the spinner was there, and the - // appended revision is what ends it. - const transcript = [openCardMessage(open)]; - expect(liveInvestigation(transcript)).toEqual({ progress: "Reading the run's spans" }); - expect(liveInvestigation([...transcript, message as never])).toBeNull(); - }); - - it("a retried run neither duplicates the card nor opens a second investigation", async () => { - const open = openState(); - const store = fakeStore(open); - const deps = { - listStale: async () => [staleRow(open)], - settleAndClose: store.settleAndClose, - }; - - await sweepDashboardAgentInvestigations(deps); - const second = await sweepDashboardAgentInvestigations(deps); - - expect(store.appended.map((message) => message.id)).toEqual([ - investigationSettlementMessageId(INVESTIGATION_ID, 1), - ]); - expect(second).toMatchObject({ stale: 1, settled: 0, closed: 0, alreadySettled: 1, failed: 0 }); - expect(store.row.revision).toBe(1); - }); -}); diff --git a/apps/webapp/test/dashboardAgentMessageSurrogate.test.ts b/apps/webapp/test/dashboardAgentMessageSurrogate.test.ts index fc4a325a08..9dedbc5300 100644 --- a/apps/webapp/test/dashboardAgentMessageSurrogate.test.ts +++ b/apps/webapp/test/dashboardAgentMessageSurrogate.test.ts @@ -5,10 +5,9 @@ import { type DashboardAgentDb, type DashboardAgentDbClient, } from "@internal/dashboard-agent-db"; +import { applyDashboardAgentMigrations } from "@internal/dashboard-agent-db/testing"; import { postgresTest } from "@internal/testcontainers"; import type { PrismaClient } from "@trigger.dev/database"; -import { readdirSync, readFileSync } from "node:fs"; -import path from "node:path"; import { afterEach, describe, expect, test } from "vitest"; import { wellFormMessageText } from "~/services/dashboardAgentMessageText.server"; @@ -17,20 +16,6 @@ import { wellFormMessageText } from "~/services/dashboardAgentMessageText.server let agentDb: DashboardAgentDb; let agentDbClient: DashboardAgentDbClient | undefined; -const MIGRATIONS = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle"); - -async function applyAgentSchema(prisma: PrismaClient) { - for (const name of readdirSync(MIGRATIONS) - .filter((file) => file.endsWith(".sql")) - .sort()) { - const sql = readFileSync(path.join(MIGRATIONS, name), "utf8"); - for (const statement of sql.split("--> statement-breakpoint")) { - const trimmed = statement.trim(); - if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed); - } - } -} - const ORG_ID = "org_surrogate"; const USER_ID = "user_surrogate"; const CHAT_ID = "chat_surrogate"; @@ -44,7 +29,7 @@ function userMessage(id: string) { } async function boot(prisma: PrismaClient, connectionUri: string) { - await applyAgentSchema(prisma); + await applyDashboardAgentMigrations((statement) => prisma.$executeRawUnsafe(statement)); agentDbClient = createDashboardAgentDb(connectionUri, { max: 4 }); agentDb = agentDbClient.db; await createChat(agentDb, { id: CHAT_ID, organizationId: ORG_ID, userId: USER_ID }); diff --git a/apps/webapp/test/dashboardAgentQueriesTenantIsolation.test.ts b/apps/webapp/test/dashboardAgentQueriesTenantIsolation.test.ts index 8fccb8c715..e9d37914d0 100644 --- a/apps/webapp/test/dashboardAgentQueriesTenantIsolation.test.ts +++ b/apps/webapp/test/dashboardAgentQueriesTenantIsolation.test.ts @@ -13,31 +13,15 @@ import { type DashboardAgentDb, type DashboardAgentDbClient, } from "@internal/dashboard-agent-db"; +import { applyDashboardAgentMigrations } from "@internal/dashboard-agent-db/testing"; import { postgresTest } from "@internal/testcontainers"; import type { PrismaClient } from "@trigger.dev/database"; -import { readdirSync, readFileSync } from "node:fs"; -import path from "node:path"; import { afterEach, describe, expect } from "vitest"; -/** Replays every migration in order, so a new migration can't leave the suite on a stale schema. */ -async function applyAgentSchema(prisma: PrismaClient) { - const folder = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle"); - const migrations = readdirSync(folder) - .filter((file) => file.endsWith(".sql")) - .sort(); - for (const name of migrations) { - const sql = readFileSync(path.join(folder, name), "utf8"); - for (const statement of sql.split("--> statement-breakpoint")) { - const trimmed = statement.trim(); - if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed); - } - } -} - let agentDbClient: DashboardAgentDbClient | undefined; async function boot(prisma: PrismaClient, connectionUri: string): Promise { - await applyAgentSchema(prisma); + await applyDashboardAgentMigrations((statement) => prisma.$executeRawUnsafe(statement)); agentDbClient = createDashboardAgentDb(connectionUri, { max: 2 }); return agentDbClient.db; } diff --git a/apps/webapp/test/dashboardAgentQuota.test.ts b/apps/webapp/test/dashboardAgentQuota.test.ts index 34841e97eb..656c9b6527 100644 --- a/apps/webapp/test/dashboardAgentQuota.test.ts +++ b/apps/webapp/test/dashboardAgentQuota.test.ts @@ -7,10 +7,9 @@ import { type DashboardAgentDb, type DashboardAgentDbClient, } from "@internal/dashboard-agent-db"; +import { applyDashboardAgentMigrations } from "@internal/dashboard-agent-db/testing"; import { postgresTest } from "@internal/testcontainers"; import type { PrismaClient } from "@trigger.dev/database"; -import { readdirSync, readFileSync } from "node:fs"; -import path from "node:path"; import { afterEach, describe, expect, it } from "vitest"; import { agentTurnCountsAgainstQuota, @@ -27,27 +26,13 @@ import { limitValueAllowingZero } from "~/services/platform.v3.server"; * absent (self-hosted) or the counter read throws. */ -const MIGRATIONS = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle"); - -async function applyAgentSchema(prisma: PrismaClient) { - for (const name of readdirSync(MIGRATIONS) - .filter((file) => file.endsWith(".sql")) - .sort()) { - const sql = readFileSync(path.join(MIGRATIONS, name), "utf8"); - for (const statement of sql.split("--> statement-breakpoint")) { - const trimmed = statement.trim(); - if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed); - } - } -} - const ORG = "org_quota"; const USER = "user_quota"; let agentDbClient: DashboardAgentDbClient | undefined; async function boot(prisma: PrismaClient, connectionUri: string): Promise { - await applyAgentSchema(prisma); + await applyDashboardAgentMigrations((statement) => prisma.$executeRawUnsafe(statement)); agentDbClient = createDashboardAgentDb(connectionUri, { max: 4 }); return agentDbClient.db; } diff --git a/apps/webapp/test/dashboardAgentSurrogatePersist.test.ts b/apps/webapp/test/dashboardAgentSurrogatePersist.test.ts index 49fdfa6418..846465689e 100644 --- a/apps/webapp/test/dashboardAgentSurrogatePersist.test.ts +++ b/apps/webapp/test/dashboardAgentSurrogatePersist.test.ts @@ -8,10 +8,9 @@ import { type DashboardAgentDb, type DashboardAgentDbClient, } from "@internal/dashboard-agent-db"; +import { applyDashboardAgentMigrations } from "@internal/dashboard-agent-db/testing"; import { postgresTest } from "@internal/testcontainers"; import type { PrismaClient } from "@trigger.dev/database"; -import { readdirSync, readFileSync } from "node:fs"; -import path from "node:path"; import { afterEach, describe, expect } from "vitest"; /** @@ -23,20 +22,6 @@ import { afterEach, describe, expect } from "vitest"; let agentDb: DashboardAgentDb; let agentDbClient: DashboardAgentDbClient | undefined; -const MIGRATIONS = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle"); - -async function applyAgentSchema(prisma: PrismaClient) { - for (const name of readdirSync(MIGRATIONS) - .filter((file) => file.endsWith(".sql")) - .sort()) { - const sql = readFileSync(path.join(MIGRATIONS, name), "utf8"); - for (const statement of sql.split("--> statement-breakpoint")) { - const trimmed = statement.trim(); - if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed); - } - } -} - const ORG = "org_surrogate"; const USER = "user_surrogate"; @@ -46,7 +31,7 @@ afterEach(async () => { }); async function boot(prisma: PrismaClient, connectionUri: string, chatId: string) { - await applyAgentSchema(prisma); + await applyDashboardAgentMigrations((statement) => prisma.$executeRawUnsafe(statement)); agentDbClient = createDashboardAgentDb(connectionUri, { max: 4 }); agentDb = agentDbClient.db; await createChat(agentDb, { id: chatId, organizationId: ORG, userId: USER }); diff --git a/apps/webapp/test/dashboardAgentTenantIsolation.test.ts b/apps/webapp/test/dashboardAgentTenantIsolation.test.ts index 3aeb1bc16e..8bbc773c8f 100644 --- a/apps/webapp/test/dashboardAgentTenantIsolation.test.ts +++ b/apps/webapp/test/dashboardAgentTenantIsolation.test.ts @@ -11,10 +11,9 @@ import { type DashboardAgentDb, type DashboardAgentDbClient, } from "@internal/dashboard-agent-db"; +import { applyDashboardAgentMigrations } from "@internal/dashboard-agent-db/testing"; import { postgresTest } from "@internal/testcontainers"; import type { PrismaClient } from "@trigger.dev/database"; -import { readdirSync, readFileSync } from "node:fs"; -import path from "node:path"; import { afterEach, describe, expect } from "vitest"; /** @@ -34,21 +33,6 @@ import { afterEach, describe, expect } from "vitest"; let agentDb: DashboardAgentDb; let agentDbClient: DashboardAgentDbClient | undefined; -const MIGRATIONS = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle"); - -/** Replays every migration in order, so a new migration can't leave the suite on a stale schema. */ -async function applyAgentSchema(prisma: PrismaClient) { - for (const name of readdirSync(MIGRATIONS) - .filter((file) => file.endsWith(".sql")) - .sort()) { - const sql = readFileSync(path.join(MIGRATIONS, name), "utf8"); - for (const statement of sql.split("--> statement-breakpoint")) { - const trimmed = statement.trim(); - if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed); - } - } -} - // Org A owns the chat. Org B and a same-org other user are the foreign tenants. const ORG_A = "org_a"; const USER_A = "user_a"; @@ -57,7 +41,7 @@ const USER_B = "user_b"; const CHAT = "chat_owned_by_a"; async function boot(prisma: PrismaClient, connectionUri: string) { - await applyAgentSchema(prisma); + await applyDashboardAgentMigrations((statement) => prisma.$executeRawUnsafe(statement)); agentDbClient = createDashboardAgentDb(connectionUri, { max: 4 }); agentDb = agentDbClient.db; } diff --git a/apps/webapp/test/dashboardAgentTranscriptStore.test.ts b/apps/webapp/test/dashboardAgentTranscriptStore.test.ts index 525b889c06..1f3dbd157b 100644 --- a/apps/webapp/test/dashboardAgentTranscriptStore.test.ts +++ b/apps/webapp/test/dashboardAgentTranscriptStore.test.ts @@ -17,14 +17,13 @@ import { type DashboardAgentDb, type DashboardAgentDbClient, } from "@internal/dashboard-agent-db"; +import { applyDashboardAgentMigrations } from "@internal/dashboard-agent-db/testing"; import { investigationStateSchema, type InvestigationState, } from "@internal/dashboard-agent-contracts"; import { postgresTest } from "@internal/testcontainers"; import type { PrismaClient } from "@trigger.dev/database"; -import { readdirSync, readFileSync } from "node:fs"; -import path from "node:path"; import { afterEach, describe, expect } from "vitest"; /** @@ -39,28 +38,13 @@ import { afterEach, describe, expect } from "vitest"; let agentDb: DashboardAgentDb; let agentDbClient: DashboardAgentDbClient | undefined; -const MIGRATIONS = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle"); - -/** Replays every migration in order, so a new migration can't leave the suite on a stale schema. */ -async function applyAgentSchema(prisma: PrismaClient) { - for (const name of readdirSync(MIGRATIONS) - .filter((file) => file.endsWith(".sql")) - .sort()) { - const sql = readFileSync(path.join(MIGRATIONS, name), "utf8"); - for (const statement of sql.split("--> statement-breakpoint")) { - const trimmed = statement.trim(); - if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed); - } - } -} - const ORG_ID = "org_store"; const USER_ID = "user_store"; const PROJECT_REF = "proj_store"; const ENV_REF = "env_store"; async function boot(prisma: PrismaClient, connectionUri: string, chatId?: string) { - await applyAgentSchema(prisma); + await applyDashboardAgentMigrations((statement) => prisma.$executeRawUnsafe(statement)); agentDbClient = createDashboardAgentDb(connectionUri, { max: 4 }); agentDb = agentDbClient.db; if (chatId) await createChat(agentDb, { id: chatId, organizationId: ORG_ID, userId: USER_ID }); diff --git a/apps/webapp/test/dashboardAgentUnreadWorkScope.test.ts b/apps/webapp/test/dashboardAgentUnreadWorkScope.test.ts index 82f8716fdf..2606216f1b 100644 --- a/apps/webapp/test/dashboardAgentUnreadWorkScope.test.ts +++ b/apps/webapp/test/dashboardAgentUnreadWorkScope.test.ts @@ -10,31 +10,16 @@ import { type DashboardAgentDb, type DashboardAgentDbClient, } from "@internal/dashboard-agent-db"; +import { applyDashboardAgentMigrations } from "@internal/dashboard-agent-db/testing"; import { postgresTest } from "@internal/testcontainers"; import type { PrismaClient } from "@trigger.dev/database"; -import { readdirSync, readFileSync } from "node:fs"; -import path from "node:path"; import { afterEach, describe, expect } from "vitest"; -const DRIZZLE = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle"); - const SCOPE = { organizationId: "org_1", userId: "user_1" }; const MESSAGED_AT = new Date("2026-02-01T00:00:00.000Z"); const READ_AFTER = new Date("2026-03-01T00:00:00.000Z"); -async function applyAgentSchema(prisma: PrismaClient) { - for (const name of readdirSync(DRIZZLE) - .filter((file) => file.endsWith(".sql")) - .sort()) { - const sql = readFileSync(path.join(DRIZZLE, name), "utf8"); - for (const statement of sql.split("--> statement-breakpoint")) { - const trimmed = statement.trim(); - if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed); - } - } -} - async function seedChat(prisma: PrismaClient, id: string, readAt: Date | null) { await prisma.$executeRawUnsafe( `insert into "trigger_dashboard_agent"."chats" @@ -59,7 +44,7 @@ describe("the chat the panel has on screen", () => { postgresTest( "is left out of the work count, and only it", async ({ prisma, postgresContainer }) => { - await applyAgentSchema(prisma); + await applyDashboardAgentMigrations((statement) => prisma.$executeRawUnsafe(statement)); agentDbClient = createDashboardAgentDb(postgresContainer.getConnectionUri(), { max: 2 }); const agentDb: DashboardAgentDb = agentDbClient.db; diff --git a/apps/webapp/test/dashboardAgentWakeActivity.test.ts b/apps/webapp/test/dashboardAgentWakeActivity.test.ts index 891610f615..ec741e2add 100644 --- a/apps/webapp/test/dashboardAgentWakeActivity.test.ts +++ b/apps/webapp/test/dashboardAgentWakeActivity.test.ts @@ -7,10 +7,8 @@ import { type DashboardAgentDb, type DashboardAgentDbClient, } from "@internal/dashboard-agent-db"; +import { applyDashboardAgentMigrations } from "@internal/dashboard-agent-db/testing"; import { postgresTest } from "@internal/testcontainers"; -import type { PrismaClient } from "@trigger.dev/database"; -import { readdirSync, readFileSync } from "node:fs"; -import path from "node:path"; import { afterEach, describe, expect } from "vitest"; // What the environment layout loader hands the browser. An active watch has to be part of it: @@ -19,19 +17,6 @@ import { afterEach, describe, expect } from "vitest"; let agentDbClient: DashboardAgentDbClient | undefined; let agentDb: DashboardAgentDb; -async function applyAgentSchema(prisma: PrismaClient) { - const folder = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle"); - for (const name of readdirSync(folder) - .filter((file) => file.endsWith(".sql")) - .sort()) { - const sql = readFileSync(path.join(folder, name), "utf8"); - for (const statement of sql.split("--> statement-breakpoint")) { - const trimmed = statement.trim(); - if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed); - } - } -} - const SCOPE = { organizationId: "org_1", userId: "user_1" }; async function seedWatch(): Promise { @@ -58,7 +43,7 @@ describe("the page load's wake activity", () => { postgresTest( "reports an active watch that has never woken anyone", async ({ prisma, postgresContainer }) => { - await applyAgentSchema(prisma); + await applyDashboardAgentMigrations((statement) => prisma.$executeRawUnsafe(statement)); agentDbClient = createDashboardAgentDb(postgresContainer.getConnectionUri(), { max: 2 }); agentDb = agentDbClient.db; diff --git a/apps/webapp/test/dashboardAgentWatchAlertEnvironments.test.ts b/apps/webapp/test/dashboardAgentWatchAlertEnvironments.test.ts index df94c5e819..f2204afdc2 100644 --- a/apps/webapp/test/dashboardAgentWatchAlertEnvironments.test.ts +++ b/apps/webapp/test/dashboardAgentWatchAlertEnvironments.test.ts @@ -10,10 +10,9 @@ import { type DashboardAgentDb, type DashboardAgentDbClient, } from "@internal/dashboard-agent-db"; +import { applyDashboardAgentMigrations } from "@internal/dashboard-agent-db/testing"; import { postgresTest } from "@internal/testcontainers"; import type { PrismaClient, RuntimeEnvironment } from "@trigger.dev/database"; -import { readdirSync, readFileSync } from "node:fs"; -import path from "node:path"; import { afterEach, describe, expect, vi } from "vitest"; const ctx = vi.hoisted(() => ({ @@ -68,28 +67,13 @@ const { alertsWorker } = await import("~/v3/alertsWorker.server"); const enqueue = alertsWorker.enqueue as unknown as ReturnType; -/** Replays every migration in order, so a new migration can't leave this file on a stale schema. */ -async function applyAgentSchema(prisma: PrismaClient) { - const folder = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle"); - const migrations = readdirSync(folder) - .filter((file) => file.endsWith(".sql")) - .sort(); - for (const name of migrations) { - const sql = readFileSync(path.join(folder, name), "utf8"); - for (const statement of sql.split("--> statement-breakpoint")) { - const trimmed = statement.trim(); - if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed); - } - } -} - let agentDbClient: DashboardAgentDbClient | undefined; async function boot(prisma: PrismaClient, connectionUri: string) { ctx.prisma = prisma; ctx.actor = undefined; enqueue.mockClear(); - await applyAgentSchema(prisma); + await applyDashboardAgentMigrations((statement) => prisma.$executeRawUnsafe(statement)); agentDbClient = createDashboardAgentDb(connectionUri, { max: 4 }); ctx.agentDb = agentDbClient.db; } diff --git a/apps/webapp/test/dashboardAgentWatchAlertOwnerScope.test.ts b/apps/webapp/test/dashboardAgentWatchAlertOwnerScope.test.ts index 6331a07fcc..072a2d01ee 100644 --- a/apps/webapp/test/dashboardAgentWatchAlertOwnerScope.test.ts +++ b/apps/webapp/test/dashboardAgentWatchAlertOwnerScope.test.ts @@ -10,12 +10,11 @@ import { type DashboardAgentDb, type DashboardAgentDbClient, } from "@internal/dashboard-agent-db"; +import { applyDashboardAgentMigrations } from "@internal/dashboard-agent-db/testing"; import type { WatchSpec } from "@internal/dashboard-agent-contracts"; import { postgresTest } from "@internal/testcontainers"; import type { PrismaClient } from "@trigger.dev/database"; import type * as SdkModule from "@trigger.dev/sdk"; -import { readdirSync, readFileSync } from "node:fs"; -import path from "node:path"; import { afterEach, describe, expect, vi } from "vitest"; import type { WatchCheckDeps, WatchRunRow } from "~/services/dashboardAgentWatchChecks"; import type * as WatchChecksModule from "~/services/dashboardAgentWatchChecks.server"; @@ -83,27 +82,12 @@ const { action: watchesAction } = await import("~/routes/api.v1.dashboard-agent. const { subscribeUserToWatchAlerts, DASHBOARD_AGENT_WATCH_ALERT_TYPE } = await import("~/services/dashboardAgentWatchAlerts.server"); -/** Replays every migration in order, so a new migration can't leave this file on a stale schema. */ -async function applyAgentSchema(prisma: PrismaClient) { - const folder = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle"); - const migrations = readdirSync(folder) - .filter((file) => file.endsWith(".sql")) - .sort(); - for (const name of migrations) { - const sql = readFileSync(path.join(folder, name), "utf8"); - for (const statement of sql.split("--> statement-breakpoint")) { - const trimmed = statement.trim(); - if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed); - } - } -} - let agentDbClient: DashboardAgentDbClient | undefined; async function boot(prisma: PrismaClient, connectionUri: string) { ctx.prisma = prisma; ctx.actor = undefined; - await applyAgentSchema(prisma); + await applyDashboardAgentMigrations((statement) => prisma.$executeRawUnsafe(statement)); agentDbClient = createDashboardAgentDb(connectionUri, { max: 4 }); ctx.agentDb = agentDbClient.db; } diff --git a/apps/webapp/test/dashboardAgentWatchBatchFairness.test.ts b/apps/webapp/test/dashboardAgentWatchBatchFairness.test.ts index 1a1670860a..e620da69b7 100644 --- a/apps/webapp/test/dashboardAgentWatchBatchFairness.test.ts +++ b/apps/webapp/test/dashboardAgentWatchBatchFairness.test.ts @@ -7,10 +7,8 @@ import { type DashboardAgentDb, type DashboardAgentDbClient, } from "@internal/dashboard-agent-db"; +import { applyDashboardAgentMigrations } from "@internal/dashboard-agent-db/testing"; import { postgresTest } from "@internal/testcontainers"; -import type { PrismaClient } from "@trigger.dev/database"; -import { readdirSync, readFileSync } from "node:fs"; -import path from "node:path"; import { afterEach, describe, expect } from "vitest"; // A group larger than the batch cap must rotate: the same prefix winning every tick would @@ -19,19 +17,6 @@ import { afterEach, describe, expect } from "vitest"; let agentDbClient: DashboardAgentDbClient | undefined; let agentDb: DashboardAgentDb; -async function applyAgentSchema(prisma: PrismaClient) { - const folder = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle"); - for (const name of readdirSync(folder) - .filter((file) => file.endsWith(".sql")) - .sort()) { - const sql = readFileSync(path.join(folder, name), "utf8"); - for (const statement of sql.split("--> statement-breakpoint")) { - const trimmed = statement.trim(); - if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed); - } - } -} - const ENVIRONMENT_ID = "env_batch_fairness"; const CADENCE = 5; @@ -88,7 +73,7 @@ describe("the batch group's fairness", () => { postgresTest( "checks every watch of an over-cap group within a bounded number of ticks", async ({ prisma, postgresContainer }) => { - await applyAgentSchema(prisma); + await applyDashboardAgentMigrations((statement) => prisma.$executeRawUnsafe(statement)); agentDbClient = createDashboardAgentDb(postgresContainer.getConnectionUri(), { max: 4 }); agentDb = agentDbClient.db; @@ -110,7 +95,7 @@ describe("the batch group's fairness", () => { postgresTest( "a watch whose window closes within a cadence is never deferred by the cap", async ({ prisma, postgresContainer }) => { - await applyAgentSchema(prisma); + await applyDashboardAgentMigrations((statement) => prisma.$executeRawUnsafe(statement)); agentDbClient = createDashboardAgentDb(postgresContainer.getConnectionUri(), { max: 4 }); agentDb = agentDbClient.db; diff --git a/apps/webapp/test/dashboardAgentWatchBatchRecording.test.ts b/apps/webapp/test/dashboardAgentWatchBatchRecording.test.ts index 925fc9d95b..acc2ee784c 100644 --- a/apps/webapp/test/dashboardAgentWatchBatchRecording.test.ts +++ b/apps/webapp/test/dashboardAgentWatchBatchRecording.test.ts @@ -9,11 +9,10 @@ import { type DashboardAgentDb, type DashboardAgentDbClient, } from "@internal/dashboard-agent-db"; +import { applyDashboardAgentMigrations } from "@internal/dashboard-agent-db/testing"; import type { WatchSpec } from "@internal/dashboard-agent-contracts"; import { postgresTest } from "@internal/testcontainers"; import type { PrismaClient } from "@trigger.dev/database"; -import { readdirSync, readFileSync } from "node:fs"; -import path from "node:path"; import { afterEach, describe, expect, vi } from "vitest"; import type { WatchCheckDeps } from "~/services/dashboardAgentWatchChecks"; @@ -48,25 +47,11 @@ process.env.SESSION_SECRET = "test-session-secret-for-watch-batch"; const { isDue, runWatchBatchCheck } = await import("~/services/dashboardAgentWatchBatch.server"); const { previousCheckFacts } = await import("~/services/dashboardAgentWatchChecks"); -/** Replays every migration in order, so a new migration can't leave the suite on a stale schema. */ -async function applyAgentSchema(prisma: PrismaClient) { - const folder = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle"); - for (const name of readdirSync(folder) - .filter((file) => file.endsWith(".sql")) - .sort()) { - const sql = readFileSync(path.join(folder, name), "utf8"); - for (const statement of sql.split("--> statement-breakpoint")) { - const trimmed = statement.trim(); - if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed); - } - } -} - let agentDbClient: DashboardAgentDbClient | undefined; async function boot(prisma: PrismaClient, connectionUri: string) { ctx.prisma = prisma; - await applyAgentSchema(prisma); + await applyDashboardAgentMigrations((statement) => prisma.$executeRawUnsafe(statement)); agentDbClient = createDashboardAgentDb(connectionUri, { max: 4 }); ctx.agentDb = agentDbClient.db; } diff --git a/apps/webapp/test/dashboardAgentWatchCardAtomicity.test.ts b/apps/webapp/test/dashboardAgentWatchCardAtomicity.test.ts index 00eae78926..c490dc2f08 100644 --- a/apps/webapp/test/dashboardAgentWatchCardAtomicity.test.ts +++ b/apps/webapp/test/dashboardAgentWatchCardAtomicity.test.ts @@ -10,6 +10,7 @@ import { type DashboardAgentDb, type DashboardAgentDbClient, } from "@internal/dashboard-agent-db"; +import { applyDashboardAgentMigrations } from "@internal/dashboard-agent-db/testing"; import { forceSettledInvestigationState, investigationStateSchema, @@ -17,8 +18,6 @@ import { } from "@internal/dashboard-agent-contracts"; import { postgresTest } from "@internal/testcontainers"; import type { PrismaClient } from "@trigger.dev/database"; -import { readdirSync, readFileSync } from "node:fs"; -import path from "node:path"; import { afterEach, describe, expect } from "vitest"; /** @@ -33,20 +32,6 @@ import { afterEach, describe, expect } from "vitest"; let agentDb: DashboardAgentDb; let agentDbClient: DashboardAgentDbClient | undefined; -/** Replays every migration in order, so a new migration can't leave the suite on a stale schema. */ -async function applyAgentSchema(prisma: PrismaClient) { - const folder = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle"); - for (const name of readdirSync(folder) - .filter((file) => file.endsWith(".sql")) - .sort()) { - const sql = readFileSync(path.join(folder, name), "utf8"); - for (const statement of sql.split("--> statement-breakpoint")) { - const trimmed = statement.trim(); - if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed); - } - } -} - const ORG_ID = "org_watch_card"; const USER_ID = "user_watch_card"; const PROJECT_REF = "proj_watch_card"; @@ -54,7 +39,7 @@ const ENV_REF = "env_watch_card"; const MESSAGE_ID = "investigate:watch:watch_1:fired:investigate:settled"; async function boot(prisma: PrismaClient, connectionUri: string, chatId: string) { - await applyAgentSchema(prisma); + await applyDashboardAgentMigrations((statement) => prisma.$executeRawUnsafe(statement)); agentDbClient = createDashboardAgentDb(connectionUri, { max: 2 }); agentDb = agentDbClient.db; await createChat(agentDb, { id: chatId, organizationId: ORG_ID, userId: USER_ID }); diff --git a/apps/webapp/test/dashboardAgentWatchCardRequestId.test.ts b/apps/webapp/test/dashboardAgentWatchCardRequestId.test.ts index 52ee1d8c2a..a9407d02e4 100644 --- a/apps/webapp/test/dashboardAgentWatchCardRequestId.test.ts +++ b/apps/webapp/test/dashboardAgentWatchCardRequestId.test.ts @@ -3,10 +3,9 @@ import { type DashboardAgentDb, type DashboardAgentDbClient, } from "@internal/dashboard-agent-db"; +import { applyDashboardAgentMigrations } from "@internal/dashboard-agent-db/testing"; import { postgresTest } from "@internal/testcontainers"; import type { PrismaClient } from "@trigger.dev/database"; -import { readdirSync, readFileSync } from "node:fs"; -import path from "node:path"; import { afterEach, describe, expect, vi } from "vitest"; // The submit's idempotency key identifies one card submission. A per-condition fallback would @@ -46,20 +45,6 @@ process.env.SESSION_SECRET = "test-session-secret-for-watch-card-request-id"; const { action } = await import("~/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.dashboard-agent"); -/** Replays every migration in order, so a new migration can't leave the suite on a stale schema. */ -async function applyAgentSchema(prisma: PrismaClient) { - const folder = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle"); - for (const name of readdirSync(folder) - .filter((file) => file.endsWith(".sql")) - .sort()) { - const sql = readFileSync(path.join(folder, name), "utf8"); - for (const statement of sql.split("--> statement-breakpoint")) { - const trimmed = statement.trim(); - if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed); - } - } -} - let agentDbClient: DashboardAgentDbClient | undefined; async function seed(prisma: PrismaClient) { @@ -133,7 +118,7 @@ describe("the watch card submit's request id", () => { "refuses a submit with no clientRequestId, and creates nothing", async ({ prisma, postgresContainer }) => { ctx.prisma = prisma; - await applyAgentSchema(prisma); + await applyDashboardAgentMigrations((statement) => prisma.$executeRawUnsafe(statement)); agentDbClient = createDashboardAgentDb(postgresContainer.getConnectionUri(), { max: 4 }); ctx.agentDb = agentDbClient.db; diff --git a/apps/webapp/test/dashboardAgentWatchErrorFingerprint.test.ts b/apps/webapp/test/dashboardAgentWatchErrorFingerprint.test.ts index 95e8eb839d..6d4e1e3879 100644 --- a/apps/webapp/test/dashboardAgentWatchErrorFingerprint.test.ts +++ b/apps/webapp/test/dashboardAgentWatchErrorFingerprint.test.ts @@ -11,11 +11,10 @@ import { type DashboardAgentDb, type DashboardAgentDbClient, } from "@internal/dashboard-agent-db"; +import { applyDashboardAgentMigrations } from "@internal/dashboard-agent-db/testing"; import type { WatchSpec } from "@internal/dashboard-agent-contracts"; import { postgresTest } from "@internal/testcontainers"; import type { PrismaClient } from "@trigger.dev/database"; -import { readdirSync, readFileSync } from "node:fs"; -import path from "node:path"; import { afterEach, describe, expect, vi } from "vitest"; import type { WatchCheckDeps } from "~/services/dashboardAgentWatchChecks"; @@ -44,25 +43,11 @@ vi.mock("~/v3/canAccessDashboardAgent.server", () => ({ const { createDashboardAgentWatch } = await import("~/services/dashboardAgentWatches.server"); -/** Replays every migration in order, so a new migration can't leave this file on a stale schema. */ -async function applyAgentSchema(prisma: PrismaClient) { - const folder = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle"); - for (const name of readdirSync(folder) - .filter((file) => file.endsWith(".sql")) - .sort()) { - const sql = readFileSync(path.join(folder, name), "utf8"); - for (const statement of sql.split("--> statement-breakpoint")) { - const trimmed = statement.trim(); - if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed); - } - } -} - let agentDbClient: DashboardAgentDbClient | undefined; async function boot(prisma: PrismaClient, connectionUri: string) { ctx.prisma = prisma; - await applyAgentSchema(prisma); + await applyDashboardAgentMigrations((statement) => prisma.$executeRawUnsafe(statement)); agentDbClient = createDashboardAgentDb(connectionUri, { max: 4 }); ctx.agentDb = agentDbClient.db; } diff --git a/apps/webapp/test/dashboardAgentWatchLimitStatus.test.ts b/apps/webapp/test/dashboardAgentWatchLimitStatus.test.ts index a2ad2f3fd5..031ba25e11 100644 --- a/apps/webapp/test/dashboardAgentWatchLimitStatus.test.ts +++ b/apps/webapp/test/dashboardAgentWatchLimitStatus.test.ts @@ -3,10 +3,9 @@ import { type DashboardAgentDb, type DashboardAgentDbClient, } from "@internal/dashboard-agent-db"; +import { applyDashboardAgentMigrations } from "@internal/dashboard-agent-db/testing"; import { postgresTest } from "@internal/testcontainers"; import type { PrismaClient } from "@trigger.dev/database"; -import { readdirSync, readFileSync } from "node:fs"; -import path from "node:path"; import { afterEach, describe, expect, vi } from "vitest"; import type * as WatchLimitsModule from "~/services/dashboardAgentWatchLimits.server"; @@ -61,20 +60,6 @@ process.env.DASHBOARD_AGENT_SECRET_KEY = "test-dashboard-agent-secret"; const { action } = await import("~/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.dashboard-agent"); -/** Replays every migration in order, so a new migration can't leave the suite on a stale schema. */ -async function applyAgentSchema(prisma: PrismaClient) { - const folder = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle"); - for (const name of readdirSync(folder) - .filter((file) => file.endsWith(".sql")) - .sort()) { - const sql = readFileSync(path.join(folder, name), "utf8"); - for (const statement of sql.split("--> statement-breakpoint")) { - const trimmed = statement.trim(); - if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed); - } - } -} - let agentDbClient: DashboardAgentDbClient | undefined; async function seed(prisma: PrismaClient) { @@ -143,7 +128,7 @@ describe("the watch card submit's status for a plan-limit refusal", () => { "answers 409, not 500, when the window is longer than the plan allows", async ({ prisma, postgresContainer }) => { ctx.prisma = prisma; - await applyAgentSchema(prisma); + await applyDashboardAgentMigrations((statement) => prisma.$executeRawUnsafe(statement)); agentDbClient = createDashboardAgentDb(postgresContainer.getConnectionUri(), { max: 4 }); ctx.agentDb = agentDbClient.db; diff --git a/apps/webapp/test/dashboardAgentWatchLimits.test.ts b/apps/webapp/test/dashboardAgentWatchLimits.test.ts index 7bd8f4ee22..32446aa712 100644 --- a/apps/webapp/test/dashboardAgentWatchLimits.test.ts +++ b/apps/webapp/test/dashboardAgentWatchLimits.test.ts @@ -6,12 +6,11 @@ import { type DashboardAgentDb, type DashboardAgentDbClient, } from "@internal/dashboard-agent-db"; +import { applyDashboardAgentMigrations } from "@internal/dashboard-agent-db/testing"; import type { WatchSpec } from "@internal/dashboard-agent-contracts"; import type * as TriggerSdk from "@trigger.dev/sdk"; import { postgresTest } from "@internal/testcontainers"; import type { PrismaClient } from "@trigger.dev/database"; -import { readdirSync, readFileSync } from "node:fs"; -import path from "node:path"; import { afterEach, describe, expect, it, vi } from "vitest"; import type { WatchCheckDeps, WatchRunRow } from "~/services/dashboardAgentWatchChecks"; import type { WatchPlanLimits } from "~/services/dashboardAgentWatchLimits.server"; @@ -54,25 +53,11 @@ const { effectiveWatchMaxHours, resolveWatchPlanLimits, watchLimitHint, UNLIMITE await import("~/services/dashboardAgentWatchLimits.server"); const { limitValueAllowingZero } = await import("~/services/platform.v3.server"); -async function applyAgentSchema(prisma: PrismaClient) { - const folder = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle"); - const migrations = readdirSync(folder) - .filter((file) => file.endsWith(".sql")) - .sort(); - for (const name of migrations) { - const sql = readFileSync(path.join(folder, name), "utf8"); - for (const statement of sql.split("--> statement-breakpoint")) { - const trimmed = statement.trim(); - if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed); - } - } -} - let agentDbClient: DashboardAgentDbClient | undefined; async function boot(prisma: PrismaClient, connectionUri: string) { ctx.prisma = prisma; - await applyAgentSchema(prisma); + await applyDashboardAgentMigrations((statement) => prisma.$executeRawUnsafe(statement)); agentDbClient = createDashboardAgentDb(connectionUri, { max: 4 }); ctx.agentDb = agentDbClient.db; } diff --git a/apps/webapp/test/dashboardAgentWatchQueueName.test.ts b/apps/webapp/test/dashboardAgentWatchQueueName.test.ts index 714975b17a..ecee4d3ee1 100644 --- a/apps/webapp/test/dashboardAgentWatchQueueName.test.ts +++ b/apps/webapp/test/dashboardAgentWatchQueueName.test.ts @@ -12,11 +12,10 @@ import { type DashboardAgentDb, type DashboardAgentDbClient, } from "@internal/dashboard-agent-db"; +import { applyDashboardAgentMigrations } from "@internal/dashboard-agent-db/testing"; import type { WatchSpec } from "@internal/dashboard-agent-contracts"; import { postgresTest } from "@internal/testcontainers"; import type { PrismaClient } from "@trigger.dev/database"; -import { readdirSync, readFileSync } from "node:fs"; -import path from "node:path"; import { afterEach, describe, expect, vi } from "vitest"; import type { WatchCheckDeps } from "~/services/dashboardAgentWatchChecks"; @@ -51,25 +50,11 @@ const { storedQueueName } = await import("~/components/queues/queue-name"); const { queueWatchRecommendation } = await import("~/components/dashboard-agent/watch-recommendations"); -/** Replays every migration in order, so a new migration can't leave this file on a stale schema. */ -async function applyAgentSchema(prisma: PrismaClient) { - const folder = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle"); - for (const name of readdirSync(folder) - .filter((file) => file.endsWith(".sql")) - .sort()) { - const sql = readFileSync(path.join(folder, name), "utf8"); - for (const statement of sql.split("--> statement-breakpoint")) { - const trimmed = statement.trim(); - if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed); - } - } -} - let agentDbClient: DashboardAgentDbClient | undefined; async function boot(prisma: PrismaClient, connectionUri: string) { ctx.prisma = prisma; - await applyAgentSchema(prisma); + await applyDashboardAgentMigrations((statement) => prisma.$executeRawUnsafe(statement)); agentDbClient = createDashboardAgentDb(connectionUri, { max: 4 }); ctx.agentDb = agentDbClient.db; } diff --git a/apps/webapp/test/dashboardAgentWatchSweepAlertOnce.test.ts b/apps/webapp/test/dashboardAgentWatchSweepAlertOnce.test.ts index e85ab277c7..cdf3c10341 100644 --- a/apps/webapp/test/dashboardAgentWatchSweepAlertOnce.test.ts +++ b/apps/webapp/test/dashboardAgentWatchSweepAlertOnce.test.ts @@ -12,11 +12,10 @@ import { type DashboardAgentDbClient, type Watch, } from "@internal/dashboard-agent-db"; +import { applyDashboardAgentMigrations } from "@internal/dashboard-agent-db/testing"; import type { WatchSpec } from "@internal/dashboard-agent-contracts"; import { postgresTest } from "@internal/testcontainers"; import type { PrismaClient } from "@trigger.dev/database"; -import { readdirSync, readFileSync } from "node:fs"; -import path from "node:path"; import { afterEach, beforeEach, describe, expect, vi } from "vitest"; import type { WatchCheckDeps } from "~/services/dashboardAgentWatchChecks"; @@ -54,25 +53,11 @@ const { alertsWorker } = await import("~/v3/alertsWorker.server"); const enqueue = alertsWorker.enqueue as unknown as ReturnType; -/** Replays every migration in order, so a new migration can't leave this file on a stale schema. */ -async function applyAgentSchema(prisma: PrismaClient) { - const folder = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle"); - for (const name of readdirSync(folder) - .filter((file) => file.endsWith(".sql")) - .sort()) { - const sql = readFileSync(path.join(folder, name), "utf8"); - for (const statement of sql.split("--> statement-breakpoint")) { - const trimmed = statement.trim(); - if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed); - } - } -} - let agentDbClient: DashboardAgentDbClient | undefined; async function boot(prisma: PrismaClient, connectionUri: string) { ctx.prisma = prisma; - await applyAgentSchema(prisma); + await applyDashboardAgentMigrations((statement) => prisma.$executeRawUnsafe(statement)); agentDbClient = createDashboardAgentDb(connectionUri, { max: 4 }); ctx.agentDb = agentDbClient.db; } diff --git a/apps/webapp/test/dashboardAgentWatchSweepBoundary.test.ts b/apps/webapp/test/dashboardAgentWatchSweepBoundary.test.ts index bfb97ba2a1..58729220e0 100644 --- a/apps/webapp/test/dashboardAgentWatchSweepBoundary.test.ts +++ b/apps/webapp/test/dashboardAgentWatchSweepBoundary.test.ts @@ -8,11 +8,10 @@ import { type DashboardAgentDbClient, type Watch, } from "@internal/dashboard-agent-db"; +import { applyDashboardAgentMigrations } from "@internal/dashboard-agent-db/testing"; import type { WatchSpec } from "@internal/dashboard-agent-contracts"; import { postgresTest } from "@internal/testcontainers"; import type { PrismaClient } from "@trigger.dev/database"; -import { readdirSync, readFileSync } from "node:fs"; -import path from "node:path"; import { afterEach, describe, expect, vi } from "vitest"; import type { WatchCheckDeps } from "~/services/dashboardAgentWatchChecks"; @@ -51,25 +50,11 @@ process.env.SESSION_SECRET = "test-session-secret-for-watch-sweep"; const { sweepDashboardAgentWatches } = await import("~/services/dashboardAgentWatchSweep.server"); -/** Replays every migration in order, so a new migration can't leave the suite on a stale schema. */ -async function applyAgentSchema(prisma: PrismaClient) { - const folder = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle"); - for (const name of readdirSync(folder) - .filter((file) => file.endsWith(".sql")) - .sort()) { - const sql = readFileSync(path.join(folder, name), "utf8"); - for (const statement of sql.split("--> statement-breakpoint")) { - const trimmed = statement.trim(); - if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed); - } - } -} - let agentDbClient: DashboardAgentDbClient | undefined; async function boot(prisma: PrismaClient, connectionUri: string) { ctx.prisma = prisma; - await applyAgentSchema(prisma); + await applyDashboardAgentMigrations((statement) => prisma.$executeRawUnsafe(statement)); agentDbClient = createDashboardAgentDb(connectionUri, { max: 4 }); ctx.agentDb = agentDbClient.db; } diff --git a/apps/webapp/test/dashboardAgentWatchTenancy.test.ts b/apps/webapp/test/dashboardAgentWatchTenancy.test.ts index de39103a58..ac9f766235 100644 --- a/apps/webapp/test/dashboardAgentWatchTenancy.test.ts +++ b/apps/webapp/test/dashboardAgentWatchTenancy.test.ts @@ -13,11 +13,10 @@ import { type DashboardAgentDb, type DashboardAgentDbClient, } from "@internal/dashboard-agent-db"; +import { applyDashboardAgentMigrations } from "@internal/dashboard-agent-db/testing"; import type { WatchDraft, WatchSpec } from "@internal/dashboard-agent-contracts"; import { postgresTest } from "@internal/testcontainers"; import type { PrismaClient } from "@trigger.dev/database"; -import { readdirSync, readFileSync } from "node:fs"; -import path from "node:path"; import { afterEach, beforeEach, describe, expect, vi } from "vitest"; import type { WatchCheckDeps, WatchRunRow } from "~/services/dashboardAgentWatchChecks"; @@ -76,27 +75,12 @@ const { alertsWorker } = await import("~/v3/alertsWorker.server"); const enqueue = alertsWorker.enqueue as unknown as ReturnType; -/** Replays every migration in order, so a new migration can't leave this file on a stale schema. */ -async function applyAgentSchema(prisma: PrismaClient) { - const folder = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle"); - const migrations = readdirSync(folder) - .filter((file) => file.endsWith(".sql")) - .sort(); - for (const name of migrations) { - const sql = readFileSync(path.join(folder, name), "utf8"); - for (const statement of sql.split("--> statement-breakpoint")) { - const trimmed = statement.trim(); - if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed); - } - } -} - let agentDbClient: DashboardAgentDbClient | undefined; async function boot(prisma: PrismaClient, connectionUri: string) { ctx.prisma = prisma; ctx.actor = undefined; - await applyAgentSchema(prisma); + await applyDashboardAgentMigrations((statement) => prisma.$executeRawUnsafe(statement)); agentDbClient = createDashboardAgentDb(connectionUri, { max: 4 }); ctx.agentDb = agentDbClient.db; } diff --git a/apps/webapp/test/dashboardAgentWatches.test.ts b/apps/webapp/test/dashboardAgentWatches.test.ts index 524d03c579..7cfccf0a2f 100644 --- a/apps/webapp/test/dashboardAgentWatches.test.ts +++ b/apps/webapp/test/dashboardAgentWatches.test.ts @@ -30,11 +30,10 @@ import { type DashboardAgentDbClient, type Watch, } from "@internal/dashboard-agent-db"; +import { applyDashboardAgentMigrations } from "@internal/dashboard-agent-db/testing"; import type { WatchDraft, WatchSpec } from "@internal/dashboard-agent-contracts"; import { postgresTest } from "@internal/testcontainers"; import type { PrismaClient } from "@trigger.dev/database"; -import { readdirSync, readFileSync } from "node:fs"; -import path from "node:path"; import { afterEach, beforeEach, describe, expect, vi } from "vitest"; import { previousCheckFacts, @@ -144,26 +143,11 @@ const { findProjectBySlug } = await import("~/models/project.server"); const { DASHBOARD_AGENT_WATCH_ALERT_TYPE, subscribeUserToWatchAlerts } = await import("~/services/dashboardAgentWatchAlerts.server"); -/** Replays every migration in order, so a new migration can't leave the suite on a stale schema. */ -async function applyAgentSchema(prisma: PrismaClient) { - const folder = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle"); - const migrations = readdirSync(folder) - .filter((file) => file.endsWith(".sql")) - .sort(); - for (const name of migrations) { - const sql = readFileSync(path.join(folder, name), "utf8"); - for (const statement of sql.split("--> statement-breakpoint")) { - const trimmed = statement.trim(); - if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed); - } - } -} - let agentDbClient: DashboardAgentDbClient | undefined; async function boot(prisma: PrismaClient, connectionUri: string) { ctx.prisma = prisma; - await applyAgentSchema(prisma); + await applyDashboardAgentMigrations((statement) => prisma.$executeRawUnsafe(statement)); // A pool, not a single connection: the concurrent-create test needs the advisory lock to span connections. agentDbClient = createDashboardAgentDb(connectionUri, { max: 8 }); ctx.agentDb = agentDbClient.db; @@ -1307,80 +1291,6 @@ describe("the watch sweep", () => { ).toMatchObject({ overdue: 1, expired: 1 }); } ); - - postgresTest( - "retention drops long-terminal rows and nothing else", - async ({ prisma, postgresContainer }) => { - await boot(prisma, postgresContainer.getConnectionUri()); - const seeded = await seed(prisma, "sweep"); - await seedChat(seeded); - await seedChat(seeded, "chat_2"); - await seedChat(seeded, "chat_3"); - - const old = await overdueWatch(seeded, "chat_1"); - const recent = await overdueWatch(seeded, "chat_2"); - await sweepDashboardAgentWatches(sweepDeps({ seeded, delivered: [] })); - await ctx.prisma.$executeRawUnsafe( - `update trigger_dashboard_agent.watches - set delivery_status = 'delivered', delivered_at = now() - where id in ($1, $2)`, - old, - recent - ); - - const active = await create({ seeded, chatId: "chat_3" }); - if (!active.ok || !active.watching) throw new Error("expected an active watch"); - - // Backdate every timestamp the age is measured from, so `greatest(...)` really is old. - await ctx.prisma.$executeRawUnsafe( - `update trigger_dashboard_agent.watches - set created_at = now() - interval '30 days', - fired_at = now() - interval '30 days', - last_checked_at = now() - interval '30 days', - delivered_at = now() - interval '30 days' - where id = $1`, - old - ); - await ctx.prisma.$executeRawUnsafe( - `update trigger_dashboard_agent.watches set created_at = now() - interval '30 days' where id = $1`, - active.watchId - ); - - expect(await sweepDashboardAgentWatches(sweepDeps({ seeded, delivered: [] }))).toMatchObject({ - purged: 1, - failed: 0, - }); - expect(await getWatch(ctx.agentDb, { id: old })).toBeNull(); - expect(await getWatch(ctx.agentDb, { id: recent })).not.toBeNull(); - expect(await getWatch(ctx.agentDb, { id: active.watchId })).not.toBeNull(); - } - ); - - postgresTest( - "retention never takes a row whose wake is still owed", - async ({ prisma, postgresContainer }) => { - await boot(prisma, postgresContainer.getConnectionUri()); - const seeded = await seed(prisma, "sweep"); - await seedChat(seeded); - const watchId = await overdueWatch(seeded); - - await sweepDashboardAgentWatches(sweepDeps({ seeded, delivered: [] })); - await ctx.prisma.$executeRawUnsafe( - `update trigger_dashboard_agent.watches - set delivery_status = 'pending', - created_at = now() - interval '30 days', - fired_at = now() - interval '30 days', - last_checked_at = now() - interval '30 days' - where id = $1`, - watchId - ); - - expect(await sweepDashboardAgentWatches(sweepDeps({ seeded, delivered: [] }))).toMatchObject({ - purged: 0, - }); - expect(await getWatch(ctx.agentDb, { id: watchId })).not.toBeNull(); - } - ); }); describe("the tick claim", () => { diff --git a/internal-packages/dashboard-agent-db/package.json b/internal-packages/dashboard-agent-db/package.json index 24c4215ce9..55e6b6760e 100644 --- a/internal-packages/dashboard-agent-db/package.json +++ b/internal-packages/dashboard-agent-db/package.json @@ -5,6 +5,10 @@ "main": "./src/index.ts", "types": "./src/index.ts", "type": "module", + "exports": { + ".": "./src/index.ts", + "./testing": "./src/testing.ts" + }, "dependencies": { "@internal/dashboard-agent-contracts": "workspace:*", "drizzle-orm": "^0.45.0", diff --git a/internal-packages/dashboard-agent-db/src/testing.ts b/internal-packages/dashboard-agent-db/src/testing.ts new file mode 100644 index 0000000000..21f949ba56 --- /dev/null +++ b/internal-packages/dashboard-agent-db/src/testing.ts @@ -0,0 +1,21 @@ +import { readdir, readFile } from "node:fs/promises"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; + +/** Runs one statement. Lets a suite replay with whatever client it already has. */ +export type MigrationExecutor = (statement: string) => Promise; + +const MIGRATIONS_DIR = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "../drizzle"); + +/** Replays every migration in order, so a new migration can't leave a suite on a stale schema. */ +export async function applyDashboardAgentMigrations(execute: MigrationExecutor): Promise { + const files = (await readdir(MIGRATIONS_DIR)).filter((file) => file.endsWith(".sql")).sort(); + + for (const file of files) { + const migration = await readFile(path.join(MIGRATIONS_DIR, file), "utf8"); + for (const statement of migration.split("--> statement-breakpoint")) { + const trimmed = statement.trim(); + if (trimmed.length > 0) await execute(trimmed); + } + } +} diff --git a/internal-packages/dashboard-agent/GUIDEBOOK.md b/internal-packages/dashboard-agent/GUIDEBOOK.md index 8595fb970f..acae52d364 100644 --- a/internal-packages/dashboard-agent/GUIDEBOOK.md +++ b/internal-packages/dashboard-agent/GUIDEBOOK.md @@ -358,7 +358,7 @@ The order is load-bearing. | Window | 30 min, 1, 2, 6, 12 or 24 hours | The ceiling is 24 hours and the schema enforces both. A watch schedules its own -next check — there is no shared cron — so the first answer lands one cadence +next check — no cron polls for due watches — so the first answer lands one cadence after you confirm the card. `expiresAt` is creation time plus the window. Due watches of one `(environment, cadence)` group can instead be checked @@ -380,8 +380,16 @@ A sweep (`dashboardAgentWatchSweep.server.ts`) is the backstop: - a row still active **2 minutes** past `expiresAt` is finalized; - a resolved row whose wake is still owed **5 minutes** later is redelivered — the sweep can't tell whether the user was already told, so delivery is - id-deduped rather than conditional; -- terminal rows are kept **7 days**; the outcome also lives in the transcript. + id-deduped rather than conditional. + +The agent project runs two scheduled tasks of its own: `dashboard-agent-investigation-sweep` +every 5 minutes settles cards left `in_progress` past **30 minutes** +(`investigation-sweep.ts`), and `dashboard-agent-maintenance` at 03:00 UTC daily is +retention (`maintenance.ts`) — judged turns and soft-deleted chats past **30 +days**, terminal watches and their submission ledger past **7 days** — a purged +watch outcome still lives in the transcript. Retention connects with +`DASHBOARD_AGENT_DATABASE_URL`, falling back to `DATABASE_URL`, like the watch +and sweep tasks, and skips only when neither is set. The sweep re-authorizes each row before reading anything, and carries the previous check's facts into the final evaluation, so a stall streak survives the diff --git a/internal-packages/dashboard-agent/README.md b/internal-packages/dashboard-agent/README.md index b59580137c..5314ec3a10 100644 --- a/internal-packages/dashboard-agent/README.md +++ b/internal-packages/dashboard-agent/README.md @@ -32,7 +32,8 @@ TRIGGER_DASHBOARD_AGENT_PROJECT_REF= pnpm run deploy # trigger d ``` Runtime env the deployed task needs: `DASHBOARD_AGENT_DATABASE_URL` (the agent -datastore) and `OBJECT_STORE_*` (chat.agent's built-in conversation snapshot). +datastore, falling back to `DATABASE_URL` when the store lives in the main +database) and `OBJECT_STORE_*` (chat.agent's built-in conversation snapshot). ## Consumed by the webapp diff --git a/internal-packages/dashboard-agent/package.json b/internal-packages/dashboard-agent/package.json index 2d89391e26..1808d60e77 100644 --- a/internal-packages/dashboard-agent/package.json +++ b/internal-packages/dashboard-agent/package.json @@ -21,6 +21,7 @@ }, "devDependencies": { "@ai-sdk/provider": "3.0.8", + "@internal/testcontainers": "workspace:*", "@trigger.dev/build": "workspace:*", "trigger.dev": "workspace:*", "vitest": "4.1.7" diff --git a/internal-packages/dashboard-agent/src/investigation-sweep.test.ts b/internal-packages/dashboard-agent/src/investigation-sweep.test.ts new file mode 100644 index 0000000000..40e5da5d62 --- /dev/null +++ b/internal-packages/dashboard-agent/src/investigation-sweep.test.ts @@ -0,0 +1,587 @@ +import { + forceSettledInvestigationState, + investigationStateSchema, + UNSETTLED_INVESTIGATION_NOTE, + VIEW_BLOCK_VERSION, + type InvestigationState, +} from "@internal/dashboard-agent-contracts"; +import { + createChat, + createDashboardAgentDb, + getChatMessages, + getInvestigation, + investigationSettlementMessage, + investigationSettlementMessageId, + listStaleOpenInvestigations, + settleInvestigationAndCloseCard, + upsertInvestigationRevision, + type DashboardAgentDb, + type DashboardAgentDbClient, + type Investigation, + type InvestigationCardMessage, +} from "@internal/dashboard-agent-db"; +import { applyDashboardAgentMigrations } from "@internal/dashboard-agent-db/testing"; +import { postgresTest } from "@internal/testcontainers"; +import { afterEach, describe, expect, it } from "vitest"; +import { + INVESTIGATION_STALE_MS, + MAX_SWEEP_ATTEMPTS, + sweepDashboardAgentInvestigations, +} from "./investigation-sweep"; +import { watchConnectionString } from "./watch-task-adapters"; + +const ORG = "org_sweep"; +const USER = "user_sweep"; +const PROJECT_REF = "proj_sweep"; +const ENV_REF = "env_sweep"; + +/** Comfortably past the grace window. */ +const STALE_AGE_MS = INVESTIGATION_STALE_MS + 60_000; + +let client: DashboardAgentDbClient | undefined; + +async function boot(connectionUri: string): Promise { + client = createDashboardAgentDb(connectionUri, { max: 2 }); + await applyDashboardAgentMigrations((statement) => client!.sql.unsafe(statement)); + return client.db; +} + +function raw(statement: string, params: unknown[] = []) { + return client!.sql.unsafe(statement, params as never[]); +} + +afterEach(async () => { + await client?.close(); + client = undefined; +}); + +function openState(overrides: Partial = {}): InvestigationState { + return investigationStateSchema.parse({ + outcome: "in_progress", + severity: "warn", + confidence: "medium", + title: "send-order-receipt keeps failing", + headline: "Checking whether the failures share a payload.", + progress: "Reading the run's spans", + checkNext: [], + hypotheses: [ + { + id: "h1", + statement: "The new payload dropped a field the task reads.", + verdict: "testing", + evidence: [], + }, + ], + evidence: [], + ...overrides, + }); +} + +async function seedChat(db: DashboardAgentDb, id: string, options: { deleted?: boolean } = {}) { + await createChat(db, { id, organizationId: ORG, userId: USER }); + if (options.deleted) { + await raw(`update trigger_dashboard_agent.chats set deleted_at = now() where id = $1`, [id]); + } +} + +async function seedInvestigation( + db: DashboardAgentDb, + args: { chatId: string; state: InvestigationState; ageMs?: number } +): Promise { + const created = await upsertInvestigationRevision(db, { + chatId: args.chatId, + projectRef: PROJECT_REF, + environmentRef: ENV_REF, + state: args.state, + }); + if (!created.ok) throw new Error("the fixture investigation wasn't created"); + + if (args.ageMs !== undefined) { + await raw( + `update trigger_dashboard_agent.investigations + set updated_at = now() - ($2 || ' milliseconds')::interval where id = $1`, + [created.id, String(args.ageMs)] + ); + } + return created.id; +} + +async function outcomeOf(db: DashboardAgentDb, id: string): Promise { + const row = await getInvestigation(db, { id }); + return row ? (row.state as { outcome?: string }).outcome : undefined; +} + +function messagesOf(db: DashboardAgentDb, chatId: string) { + return getChatMessages(db, { chatId, userId: USER, organizationId: ORG }) as Promise< + { id: string; parts: Record[] }[] | null + >; +} + +describe("the dashboard agent investigation sweep", () => { + postgresTest( + "settles a card left in_progress, keeping what was established", + async ({ postgresContainer }) => { + const db = await boot(postgresContainer.getConnectionUri()); + await seedChat(db, "chat_stale"); + const id = await seedInvestigation(db, { + chatId: "chat_stale", + state: openState(), + ageMs: STALE_AGE_MS, + }); + + const result = await sweepDashboardAgentInvestigations(db); + expect(result).toMatchObject({ + stale: 1, + settled: 1, + closed: 1, + alreadySettled: 0, + failed: 0, + }); + + const row = await getInvestigation(db, { id }); + const state = investigationStateSchema.parse(row?.state); + expect(state.outcome).toBe("inconclusive"); + expect(state.confidence).toBe("low"); + expect(state.headline).toBe( + `Checking whether the failures share a payload. ${UNSETTLED_INVESTIGATION_NOTE}` + ); + expect(state.progress).toBeUndefined(); + expect(state.remediation).toBeUndefined(); + expect(state.hypotheses).toHaveLength(1); + expect(state.title).toBe("send-order-receipt keeps failing"); + expect(row?.revision).toBe(1); + + // The settled row is invisible on its own: the panel resolves the card from the + // transcript, so the closing revision has to be in the chat too. + const messages = await messagesOf(db, "chat_stale"); + expect(messages?.map((message) => message.id)).toEqual([ + investigationSettlementMessageId(id, 1), + ]); + const block = messages![0]!.parts[0]!.output.blocks[0]; + expect(block).toMatchObject({ type: "investigation", id, revision: 1 }); + expect(block.investigation.outcome).toBe("inconclusive"); + + // A second run can't stack a second card: the settle is a no-op and the append + // is deduped on the same message id. + expect(await sweepDashboardAgentInvestigations(db)).toMatchObject({ stale: 0, settled: 0 }); + expect((await messagesOf(db, "chat_stale"))?.length).toBe(1); + }, + 60_000 + ); + + postgresTest( + "drops a fix a concluded card was carrying", + async ({ postgresContainer }) => { + const db = await boot(postgresContainer.getConnectionUri()); + await seedChat(db, "chat_fix"); + // An inconclusive card may not offer a fix, so the settle strips remediation. + const id = await seedInvestigation(db, { + chatId: "chat_fix", + state: { ...openState(), remediation: "Raise the timeout." } as InvestigationState, + ageMs: STALE_AGE_MS, + }); + + await sweepDashboardAgentInvestigations(db); + + const row = await getInvestigation(db, { id }); + expect(investigationStateSchema.parse(row?.state).remediation).toBeUndefined(); + }, + 60_000 + ); + + postgresTest( + "leaves a fresh in_progress card alone — a live turn is never swept", + async ({ postgresContainer }) => { + const db = await boot(postgresContainer.getConnectionUri()); + await seedChat(db, "chat_fresh"); + const id = await seedInvestigation(db, { chatId: "chat_fresh", state: openState() }); + + expect(await sweepDashboardAgentInvestigations(db)).toMatchObject({ stale: 0, settled: 0 }); + const row = await getInvestigation(db, { id }); + expect(investigationStateSchema.parse(row?.state).outcome).toBe("in_progress"); + expect(row?.revision).toBe(0); + }, + 60_000 + ); + + postgresTest( + "leaves a card that already has an answer alone", + async ({ postgresContainer }) => { + const db = await boot(postgresContainer.getConnectionUri()); + await seedChat(db, "chat_done"); + const id = await seedInvestigation(db, { + chatId: "chat_done", + state: openState({ + outcome: "inconclusive", + progress: undefined, + headline: "Not established: the failures span two versions.", + }), + ageMs: STALE_AGE_MS, + }); + + expect(await sweepDashboardAgentInvestigations(db)).toMatchObject({ stale: 0, settled: 0 }); + const row = await getInvestigation(db, { id }); + expect(row?.revision).toBe(0); + expect(investigationStateSchema.parse(row?.state).headline).toBe( + "Not established: the failures span two versions." + ); + }, + 60_000 + ); + + postgresTest( + "skips a card in a deleted chat", + async ({ postgresContainer }) => { + const db = await boot(postgresContainer.getConnectionUri()); + await seedChat(db, "chat_gone", { deleted: true }); + await seedInvestigation(db, { + chatId: "chat_gone", + state: openState(), + ageMs: STALE_AGE_MS, + }); + + expect(await sweepDashboardAgentInvestigations(db)).toMatchObject({ stale: 0, settled: 0 }); + }, + 60_000 + ); + + postgresTest( + "a turn that concludes the card first wins: the settle is a no-op, not an error", + async ({ postgresContainer }) => { + const db = await boot(postgresContainer.getConnectionUri()); + await seedChat(db, "chat_race"); + const id = await seedInvestigation(db, { + chatId: "chat_race", + state: openState(), + ageMs: STALE_AGE_MS, + }); + + const stale = await listStaleOpenInvestigations(db, { olderThan: new Date(), limit: 10 }); + expect(stale.map((row) => row.id)).toEqual([id]); + + const concluded = await upsertInvestigationRevision(db, { + id, + chatId: "chat_race", + projectRef: PROJECT_REF, + environmentRef: ENV_REF, + state: openState({ + outcome: "concluded", + confidence: "high", + progress: undefined, + headline: "receipt.ts:42 reads a field the new payload no longer carries.", + remediation: "Guard the dereference and backfill the field.", + }), + }); + expect(concluded.ok).toBe(true); + + const result = await sweepDashboardAgentInvestigations(db, { listStale: async () => stale }); + expect(result).toMatchObject({ + stale: 1, + settled: 0, + closed: 0, + alreadySettled: 1, + failed: 0, + }); + + const row = await getInvestigation(db, { id }); + const state = investigationStateSchema.parse(row?.state); + expect(state.outcome).toBe("concluded"); + expect(state.headline).not.toContain(UNSETTLED_INVESTIGATION_NOTE); + expect(row?.revision).toBe(1); + }, + 60_000 + ); + + /** + * The failure window. Settling the row and delivering its card used to be two + * operations: once the row was terminal, a failed append left a card reading + * `in_progress` that nothing would ever repair, because this sweep only selects + * `in_progress` rows. They must land together or not at all. + */ + postgresTest( + "a card that can't be delivered leaves the row in_progress, so the next run retries it", + async ({ postgresContainer }) => { + const db = await boot(postgresContainer.getConnectionUri()); + await seedChat(db, "chat_undeliverable"); + // A state the settle can merge but no card can be rendered from, so the delivery + // half genuinely fails against a real database. + const id = await seedInvestigation(db, { + chatId: "chat_undeliverable", + state: { outcome: "in_progress" } as unknown as InvestigationState, + ageMs: STALE_AGE_MS, + }); + + await expect(sweepDashboardAgentInvestigations(db)).rejects.toThrow( + /failed on 1 investigations/ + ); + + // The settle rolled back with the card: no half-applied terminal row. + const row = await getInvestigation(db, { id }); + expect(row?.revision).toBe(0); + expect((row!.state as { outcome?: string }).outcome).toBe("in_progress"); + expect(await messagesOf(db, "chat_undeliverable")).toEqual([]); + + // And it is still in the selection, so the sweep keeps trying rather than + // leaving a permanent spinner behind. + const stale = await listStaleOpenInvestigations(db, { olderThan: new Date(), limit: 10 }); + expect(stale.map((candidate) => candidate.id)).toEqual([id]); + }, + 60_000 + ); + + postgresTest( + "one failing row doesn't cost the batch, and the run throws so the job retries", + async ({ postgresContainer }) => { + const db = await boot(postgresContainer.getConnectionUri()); + await seedChat(db, "chat_batch"); + const first = await seedInvestigation(db, { + chatId: "chat_batch", + state: openState(), + ageMs: STALE_AGE_MS + 60_000, + }); + const second = await seedInvestigation(db, { + chatId: "chat_batch", + state: openState(), + ageMs: STALE_AGE_MS, + }); + + const attempted: string[] = []; + await expect( + sweepDashboardAgentInvestigations(db, { + settleAndClose: async (params) => { + attempted.push(params.id); + if (params.id === first) throw new Error("the settle failed"); + return settleInvestigationAndCloseCard(db, params); + }, + }) + ).rejects.toThrow(/failed on 1 investigations/); + + expect(attempted).toEqual([first, second]); + expect(await outcomeOf(db, second)).toBe("inconclusive"); + const stuck = await getInvestigation(db, { id: first }); + expect(stuck?.revision).toBe(0); + expect(await outcomeOf(db, first)).toBe("in_progress"); + }, + 60_000 + ); +}); + +describe("the investigation sweep with a poison row", () => { + postgresTest( + "a row that always fails to settle cannot pin the head and starve a newer row", + async ({ postgresContainer }) => { + const db = await boot(postgresContainer.getConnectionUri()); + + // Poison sorts first (older `updated_at`); renderable is newer. + await seedChat(db, "chat_poison"); + const poisonId = await seedInvestigation(db, { + chatId: "chat_poison", + state: openState(), + ageMs: STALE_AGE_MS + 60_000, + }); + await seedChat(db, "chat_ok"); + const renderableId = await seedInvestigation(db, { + chatId: "chat_ok", + state: openState(), + ageMs: STALE_AGE_MS, + }); + + // Only the poison row's settle throws; the renderable one goes through the real path. + const settleAndClose = (params: { id: string; chatId: string; note: string }) => { + if (params.id === poisonId) throw new Error("state isn't renderable"); + return settleInvestigationAndCloseCard(db, params); + }; + + // limit 1 forces head contention: without backoff the poison row would win every run. + // A failed run throws so the job retries, but the attempt is recorded before it does. + await expect( + sweepDashboardAgentInvestigations(db, { limit: 1, settleAndClose }) + ).rejects.toThrow(); + expect(await outcomeOf(db, poisonId)).toBe("in_progress"); + expect(await outcomeOf(db, renderableId)).toBe("in_progress"); + + // Next run: the poison row now sorts behind the never-attempted renderable one, + // so the newer row is picked and settled despite the poison row still being stale. + const second = await sweepDashboardAgentInvestigations(db, { limit: 1, settleAndClose }); + expect(second).toMatchObject({ stale: 1, settled: 1, failed: 0 }); + expect(await outcomeOf(db, renderableId)).toBe("inconclusive"); + expect(await outcomeOf(db, poisonId)).toBe("in_progress"); + }, + 60_000 + ); + + postgresTest( + "after the attempt cap the poison row is abandoned and leaves the queue", + async ({ postgresContainer }) => { + const db = await boot(postgresContainer.getConnectionUri()); + await seedChat(db, "chat_poison"); + const poisonId = await seedInvestigation(db, { + chatId: "chat_poison", + state: openState(), + ageMs: STALE_AGE_MS, + }); + + const settleAndClose = () => { + throw new Error("state isn't renderable"); + }; + + // The first MAX_SWEEP_ATTEMPTS-1 runs record a failed attempt and throw; the row stays stale. + for (let i = 1; i < MAX_SWEEP_ATTEMPTS; i++) { + await expect(sweepDashboardAgentInvestigations(db, { settleAndClose })).rejects.toThrow(); + expect(await outcomeOf(db, poisonId)).toBe("in_progress"); + } + + // The capped run force-settles the row without the render path, so it leaves the queue. + const capped = await sweepDashboardAgentInvestigations(db, { settleAndClose }); + expect(capped).toMatchObject({ stale: 1, abandoned: 1, failed: 0 }); + expect(await outcomeOf(db, poisonId)).toBe("inconclusive"); + + // Nothing stale remains, so the poison row is no longer swept. + expect(await sweepDashboardAgentInvestigations(db, { settleAndClose })).toMatchObject({ + stale: 0, + }); + }, + 60_000 + ); +}); + +/** + * The closing card's shape, without a container: every write is injected, so the db is + * never reached. + */ +describe("the investigation sweep's closing card", () => { + const CHAT_ID = "chat_sweep_card"; + const INVESTIGATION_ID = "inv_sweep_card"; + const noDb = undefined as unknown as DashboardAgentDb; + + function staleRow(state: InvestigationState): Investigation { + return { + id: INVESTIGATION_ID, + chatId: CHAT_ID, + projectRef: PROJECT_REF, + environmentRef: ENV_REF, + revision: 0, + state, + createdAt: new Date(), + updatedAt: new Date(), + } as Investigation; + } + + /** + * Stands in for the datastore: the conditional settle (`in_progress` only) and the + * id-deduped append, as one operation — which is what the real query is, so a + * half-applied settle can't exist. + */ + function fakeStore(initial: InvestigationState) { + const row = { revision: 0, state: initial }; + const appended: InvestigationCardMessage[] = []; + return { + row, + appended, + settleAndClose: async (params: { id: string; chatId: string; note: string }) => { + if (investigationStateSchema.parse(row.state).outcome !== "in_progress") return null; + const state = forceSettledInvestigationState(investigationStateSchema.parse(row.state)); + const revision = row.revision + 1; + + const message = investigationSettlementMessage({ + investigationId: params.id, + revision, + state, + }); + if (!message) throw new Error("the closing card didn't validate"); + + row.state = state; + row.revision = revision; + const closed = !appended.some((existing) => existing.id === message.id); + if (closed) appended.push(message); + return { settled: { id: params.id, revision, state }, closed }; + }, + }; + } + + it("appends the terminal card to the chat, so the panel stops spinning", async () => { + const open = openState(); + const store = fakeStore(open); + + const result = await sweepDashboardAgentInvestigations(noDb, { + listStale: async () => [staleRow(open)], + settleAndClose: store.settleAndClose, + }); + + expect(store.appended).toHaveLength(1); + expect(result).toMatchObject({ stale: 1, settled: 1, closed: 1, alreadySettled: 0, failed: 0 }); + + const message = store.appended[0]!; + expect(message.id).toBe(investigationSettlementMessageId(INVESTIGATION_ID, 1)); + expect(message.role).toBe("assistant"); + + const part = message.parts[0] as { + type: string; + state: string; + output: { blocks: Record[] }; + }; + expect(part.type).toBe("tool-render_view"); + expect(part.state).toBe("output-available"); + expect(part.output.blocks[0]).toMatchObject({ + type: "investigation", + id: INVESTIGATION_ID, + revision: 1, + version: VIEW_BLOCK_VERSION, + }); + const settled = part.output.blocks[0]!.investigation; + expect(settled.outcome).toBe("inconclusive"); + expect(settled.confidence).toBe("low"); + expect(settled.progress).toBeUndefined(); + expect(settled.headline).toContain(UNSETTLED_INVESTIGATION_NOTE); + // What was checked survives: the card closes honestly, it doesn't get blanked. + expect(settled.hypotheses).toHaveLength(1); + }); + + it("a retried run neither duplicates the card nor opens a second investigation", async () => { + const open = openState(); + const store = fakeStore(open); + const deps = { + listStale: async () => [staleRow(open)], + settleAndClose: store.settleAndClose, + }; + + await sweepDashboardAgentInvestigations(noDb, deps); + const second = await sweepDashboardAgentInvestigations(noDb, deps); + + expect(store.appended.map((message) => message.id)).toEqual([ + investigationSettlementMessageId(INVESTIGATION_ID, 1), + ]); + expect(second).toMatchObject({ stale: 1, settled: 0, closed: 0, alreadySettled: 1, failed: 0 }); + expect(store.row.revision).toBe(1); + }); +}); + +describe("the investigation sweep database guard", () => { + it("skips rather than throwing when neither url is set", () => { + expect(watchConnectionString({} as NodeJS.ProcessEnv)).toBeUndefined(); + }); + + it("falls back to the main database url", () => { + expect(watchConnectionString({ DATABASE_URL: "postgres://main" } as NodeJS.ProcessEnv)).toBe( + "postgres://main" + ); + }); + + it("uses the agent's own url when it is set", () => { + expect( + watchConnectionString({ + DASHBOARD_AGENT_DATABASE_URL: "postgres://agent", + DATABASE_URL: "postgres://main", + } as NodeJS.ProcessEnv) + ).toBe("postgres://agent"); + }); + + it("treats an empty dedicated url as unset", () => { + expect( + watchConnectionString({ + DASHBOARD_AGENT_DATABASE_URL: "", + DATABASE_URL: "postgres://main", + } as NodeJS.ProcessEnv) + ).toBe("postgres://main"); + }); +}); diff --git a/apps/webapp/app/services/dashboardAgentInvestigationSweep.server.ts b/internal-packages/dashboard-agent/src/investigation-sweep.ts similarity index 78% rename from apps/webapp/app/services/dashboardAgentInvestigationSweep.server.ts rename to internal-packages/dashboard-agent/src/investigation-sweep.ts index 38853ea0dd..75de120a5e 100644 --- a/apps/webapp/app/services/dashboardAgentInvestigationSweep.server.ts +++ b/internal-packages/dashboard-agent/src/investigation-sweep.ts @@ -1,20 +1,22 @@ -/** - * The investigation backstop, for cards left `in_progress`. They settle as `inconclusive`, - * conditional on the row still being `in_progress`, so a concluding turn wins the race. - */ - +import { UNSETTLED_INVESTIGATION_NOTE } from "@internal/dashboard-agent-contracts"; import { listStaleOpenInvestigations, recordInvestigationSweepAttempt, settleInvestigationAndCloseCard, settleInvestigationAsInconclusive, + type DashboardAgentDb, type Investigation, type SettledInvestigation, type SettledInvestigationCard, } from "@internal/dashboard-agent-db"; -import { UNSETTLED_INVESTIGATION_NOTE } from "@internal/dashboard-agent-contracts"; -import { dashboardAgentDb } from "~/services/dashboardAgentDb.server"; -import { logger } from "~/services/logger.server"; +import { logger, schedules } from "@trigger.dev/sdk"; +import { serializeError } from "./serialize-error"; +import { getWatchDb, watchConnectionString } from "./watch-task-adapters"; + +/** + * The investigation backstop, for cards left `in_progress`. They settle as `inconclusive`, + * conditional on the row still being `in_progress`, so a concluding turn wins the race. + */ /** * How long a card may sit `in_progress` before the sweep settles it. Must outlast the @@ -69,18 +71,18 @@ export type InvestigationSweepDeps = { * and the run throws at the end if any failed so the job is retried. */ export async function sweepDashboardAgentInvestigations( + db: DashboardAgentDb, deps: InvestigationSweepDeps = {} ): Promise { const now = deps.now?.() ?? new Date(); const limit = deps.limit ?? SWEEP_BATCH_LIMIT; - const listStale = - deps.listStale ?? ((params) => listStaleOpenInvestigations(dashboardAgentDb, params)); + const listStale = deps.listStale ?? ((params) => listStaleOpenInvestigations(db, params)); const settleAndClose = - deps.settleAndClose ?? ((params) => settleInvestigationAndCloseCard(dashboardAgentDb, params)); + deps.settleAndClose ?? ((params) => settleInvestigationAndCloseCard(db, params)); const recordAttempt = - deps.recordAttempt ?? ((params) => recordInvestigationSweepAttempt(dashboardAgentDb, params)); + deps.recordAttempt ?? ((params) => recordInvestigationSweepAttempt(db, params)); const forceAbandon = - deps.forceAbandon ?? ((params) => settleInvestigationAsInconclusive(dashboardAgentDb, params)); + deps.forceAbandon ?? ((params) => settleInvestigationAsInconclusive(db, params)); const result: InvestigationSweepResult = { stale: 0, @@ -121,10 +123,10 @@ export async function sweepDashboardAgentInvestigations( try { attempts = await recordAttempt({ id: investigation.id }); } catch (recordError) { - logger.error("Dashboard agent investigation sweep: failed to record a sweep attempt", { + logger.error("dashboard-agent investigation sweep: failed to record a sweep attempt", { investigationId: investigation.id, chatId: investigation.chatId, - error: recordError, + error: serializeError(recordError), }); } @@ -135,7 +137,7 @@ export async function sweepDashboardAgentInvestigations( await forceAbandon({ id: investigation.id, note: UNSETTLED_INVESTIGATION_NOTE }); result.abandoned++; logger.warn( - "Dashboard agent investigation sweep: abandoned a card past the attempt cap", + "dashboard-agent investigation sweep: abandoned a card past the attempt cap", { investigationId: investigation.id, chatId: investigation.chatId, @@ -144,20 +146,20 @@ export async function sweepDashboardAgentInvestigations( ); continue; } catch (abandonError) { - logger.error("Dashboard agent investigation sweep: failed to abandon a poison card", { + logger.error("dashboard-agent investigation sweep: failed to abandon a poison card", { investigationId: investigation.id, chatId: investigation.chatId, - error: abandonError, + error: serializeError(abandonError), }); } } result.failed++; - logger.error("Dashboard agent investigation sweep: failed to settle an investigation", { + logger.error("dashboard-agent investigation sweep: failed to settle an investigation", { investigationId: investigation.id, chatId: investigation.chatId, attempts, - error, + error: serializeError(error), }); } } @@ -170,3 +172,33 @@ export async function sweepDashboardAgentInvestigations( return result; } + +const EMPTY_SWEEP_RESULT: InvestigationSweepResult = { + stale: 0, + settled: 0, + closed: 0, + alreadySettled: 0, + abandoned: 0, + failed: 0, +}; + +export const dashboardAgentInvestigationSweep = schedules.task({ + id: "dashboard-agent-investigation-sweep", + cron: "*/5 * * * *", + retry: { maxAttempts: 3 }, + run: async (): Promise => { + if (!watchConnectionString()) { + logger.warn( + "dashboard-agent investigation sweep skipped: no DASHBOARD_AGENT_DATABASE_URL or DATABASE_URL" + ); + return { ...EMPTY_SWEEP_RESULT }; + } + + const { db } = getWatchDb(); + const result = await sweepDashboardAgentInvestigations(db); + + logger.info("dashboard-agent investigations swept", result); + + return result; + }, +}); diff --git a/internal-packages/dashboard-agent/src/maintenance.test.ts b/internal-packages/dashboard-agent/src/maintenance.test.ts new file mode 100644 index 0000000000..3340cac3a2 --- /dev/null +++ b/internal-packages/dashboard-agent/src/maintenance.test.ts @@ -0,0 +1,320 @@ +import { + createChat, + createDashboardAgentDb, + insertTurnEval, + type DashboardAgentDb, + type DashboardAgentDbClient, +} from "@internal/dashboard-agent-db"; +import { applyDashboardAgentMigrations } from "@internal/dashboard-agent-db/testing"; +import { postgresTest } from "@internal/testcontainers"; +import { afterEach, describe, expect } from "vitest"; +import { runDashboardAgentRetention, TURN_EVAL_RETENTION_MS } from "./maintenance"; + +const ORG = "org_retention"; +const USER = "user_retention"; + +let client: DashboardAgentDbClient | undefined; + +async function boot(connectionUri: string): Promise { + client = createDashboardAgentDb(connectionUri, { max: 2 }); + await applyDashboardAgentMigrations((statement) => client!.sql.unsafe(statement)); + return client.db; +} + +function raw(statement: string, params: unknown[] = []) { + return client!.sql.unsafe(statement, params as never[]); +} + +afterEach(async () => { + await client?.close(); + client = undefined; +}); + +async function seedTurnEval( + db: DashboardAgentDb, + args: { chatId: string; turn: number; ageMs: number } +) { + await insertTurnEval(db, { + chatId: args.chatId, + turn: args.turn, + organizationId: ORG, + userId: USER, + summary: "the user asked about a failed run", + }); + await raw( + `update trigger_dashboard_agent.chat_turn_evals + set created_at = now() - ($3 || ' seconds')::interval + where chat_id = $1 and turn = $2`, + [args.chatId, args.turn, String(args.ageMs / 1000)] + ); +} + +async function seedTerminalWatch(id: string, chatId: string, ageDays: number) { + await raw( + `insert into trigger_dashboard_agent.watches + (id, chat_id, identity, spec, organization_id, project_id, environment_id, user_id, + status, delivery_status, expires_at, created_at) + values ($1, $2, $1, '{}'::jsonb, $3, 'proj', 'env', $4, + 'expired', 'delivered', now(), now() - ($5 || ' days')::interval)`, + [id, chatId, ORG, USER, String(ageDays)] + ); +} + +async function seedSubmission(chatId: string, requestId: string, ageDays: number) { + await raw( + `insert into trigger_dashboard_agent.watch_submissions + (chat_id, client_request_id, organization_id, user_id, project_id, environment_id, + draft_hash, draft, created_at) + values ($1, $2, $3, $4, 'proj', 'env', 'hash', '{}'::jsonb, now() - ($5 || ' days')::interval)`, + [chatId, requestId, ORG, USER, String(ageDays)] + ); +} + +/** One row in every chatId-keyed table, so a delete that misses one leaves a leak. */ +async function seedChatWithChildren(db: DashboardAgentDb, id: string) { + await createChat(db, { id, organizationId: ORG, userId: USER }); + await raw( + `insert into trigger_dashboard_agent.chat_messages (chat_id, message_id, position, role, message) + values ($1, $1 || '-m', 1, 'user', '{}'::jsonb)`, + [id] + ); + await raw( + `insert into trigger_dashboard_agent.chat_sessions (chat_id, public_access_token) values ($1, 'pat')`, + [id] + ); + await raw( + `insert into trigger_dashboard_agent.chat_turn_evals (chat_id, turn, organization_id, user_id) + values ($1, 0, $2, $3)`, + [id, ORG, USER] + ); + await raw( + `insert into trigger_dashboard_agent.investigations (id, chat_id, project_ref, environment_ref, state) + values ($1 || '-inv', $1, 'proj', 'env', '{"outcome":"in_progress"}'::jsonb)`, + [id] + ); + await raw( + `insert into trigger_dashboard_agent.watches + (id, chat_id, identity, spec, organization_id, project_id, environment_id, user_id, expires_at) + values ($1 || '-w', $1, 'ident', '{}'::jsonb, $2, 'proj', 'env', $3, now() + interval '1 day')`, + [id, ORG, USER] + ); + await raw( + `insert into trigger_dashboard_agent.watch_submissions + (chat_id, client_request_id, organization_id, user_id, project_id, environment_id, draft_hash, draft) + values ($1, 'req', $2, $3, 'proj', 'env', 'hash', '{}'::jsonb)`, + [id, ORG, USER] + ); +} + +const CHILD_TABLES = [ + "chat_messages", + "chat_sessions", + "chat_turn_evals", + "investigations", + "watches", + "watch_submissions", +]; + +async function rowCounts(id: string): Promise> { + const counts: Record = {}; + for (const table of ["chats", ...CHILD_TABLES]) { + const column = table === "chats" ? "id" : "chat_id"; + const rows = await raw( + `select count(*)::int as n from trigger_dashboard_agent.${table} where ${column} = $1`, + [id] + ); + counts[table] = Number((rows as unknown as { n: number }[])[0]!.n); + } + return counts; +} + +async function count(table: string): Promise { + const rows = await raw(`select count(*)::int as n from trigger_dashboard_agent.${table}`); + return Number((rows as unknown as { n: number }[])[0]!.n); +} + +describe("the dashboard agent retention pass", () => { + postgresTest( + "drops turn evals, soft-deleted chats and finished watches past their windows", + async ({ postgresContainer }) => { + const db = await boot(postgresContainer.getConnectionUri()); + + await createChat(db, { id: "chat_evals", organizationId: ORG, userId: USER }); + await seedTurnEval(db, { + chatId: "chat_evals", + turn: 0, + ageMs: TURN_EVAL_RETENTION_MS + 60_000, + }); + await seedTurnEval(db, { chatId: "chat_evals", turn: 1, ageMs: 0 }); + + await createChat(db, { id: "chat_gone", organizationId: ORG, userId: USER }); + await createChat(db, { id: "chat_kept", organizationId: ORG, userId: USER }); + await raw( + `update trigger_dashboard_agent.chats set deleted_at = now() - interval '40 days' where id = 'chat_gone'` + ); + await raw( + `update trigger_dashboard_agent.chats set deleted_at = now() - interval '1 day' where id = 'chat_kept'` + ); + + await createChat(db, { id: "chat_watches", organizationId: ORG, userId: USER }); + await seedTerminalWatch("watch_old", "chat_watches", 10); + await seedTerminalWatch("watch_new", "chat_watches", 1); + await seedSubmission("chat_watches", "req_old", 10); + await seedSubmission("chat_watches", "req_new", 1); + + expect(await runDashboardAgentRetention(db)).toEqual({ + turnEvals: 1, + chats: 1, + watches: 1, + watchSubmissions: 1, + }); + + expect(await count("chat_turn_evals")).toBe(1); + const chats = await raw(`select id from trigger_dashboard_agent.chats order by id`); + expect((chats as unknown as { id: string }[]).map((row) => row.id)).toEqual([ + "chat_evals", + "chat_kept", + "chat_watches", + ]); + const watches = await raw(`select id from trigger_dashboard_agent.watches order by id`); + expect((watches as unknown as { id: string }[]).map((row) => row.id)).toEqual(["watch_new"]); + const submissions = await raw( + `select client_request_id from trigger_dashboard_agent.watch_submissions` + ); + expect( + (submissions as unknown as { client_request_id: string }[]).map( + (row) => row.client_request_id + ) + ).toEqual(["req_new"]); + }, + 60_000 + ); + + postgresTest( + "a purged chat takes every chatId-keyed child row with it", + async ({ postgresContainer }) => { + const db = await boot(postgresContainer.getConnectionUri()); + + await seedChatWithChildren(db, "chat_cascade_old"); + await seedChatWithChildren(db, "chat_cascade_new"); + await raw( + `update trigger_dashboard_agent.chats set deleted_at = now() - interval '40 days' where id = 'chat_cascade_old'` + ); + await raw( + `update trigger_dashboard_agent.chats set deleted_at = now() - interval '1 day' where id = 'chat_cascade_new'` + ); + + expect(await runDashboardAgentRetention(db)).toMatchObject({ chats: 1 }); + + const purged = await rowCounts("chat_cascade_old"); + const kept = await rowCounts("chat_cascade_new"); + for (const table of ["chats", ...CHILD_TABLES]) { + expect(purged[table], `${table} should be empty for the purged chat`).toBe(0); + expect(kept[table], `${table} kept for the in-window chat`).toBe(1); + } + }, + 60_000 + ); + + postgresTest( + "an active watch is never purged, however old it is", + async ({ postgresContainer }) => { + const db = await boot(postgresContainer.getConnectionUri()); + await createChat(db, { id: "chat_active", organizationId: ORG, userId: USER }); + await raw( + `insert into trigger_dashboard_agent.watches + (id, chat_id, identity, spec, organization_id, project_id, environment_id, user_id, expires_at, created_at) + values ('watch_active', 'chat_active', 'ident', '{}'::jsonb, $1, 'proj', 'env', $2, + now() + interval '1 day', now() - interval '30 days')`, + [ORG, USER] + ); + + expect(await runDashboardAgentRetention(db)).toMatchObject({ watches: 0 }); + expect(await count("watches")).toBe(1); + }, + 60_000 + ); + + postgresTest( + "drains a backlog over several batches instead of leaving it for tomorrow", + async ({ postgresContainer }) => { + const db = await boot(postgresContainer.getConnectionUri()); + await createChat(db, { id: "chat_backlog", organizationId: ORG, userId: USER }); + for (let turn = 0; turn < 5; turn++) { + await seedTurnEval(db, { + chatId: "chat_backlog", + turn, + ageMs: TURN_EVAL_RETENTION_MS + 60_000, + }); + } + await seedTerminalWatch("watch_a", "chat_backlog", 10); + await seedTerminalWatch("watch_b", "chat_backlog", 10); + await seedSubmission("chat_backlog", "req_a", 10); + await seedSubmission("chat_backlog", "req_b", 10); + + expect(await runDashboardAgentRetention(db, { limit: 1 })).toEqual({ + turnEvals: 5, + chats: 0, + watches: 2, + watchSubmissions: 2, + }); + expect(await count("chat_turn_evals")).toBe(0); + expect(await count("watches")).toBe(0); + expect(await count("watch_submissions")).toBe(0); + }, + 60_000 + ); + + postgresTest( + "the batch cap bounds one run, and the rest goes next run", + async ({ postgresContainer }) => { + const db = await boot(postgresContainer.getConnectionUri()); + await createChat(db, { id: "chat_capped", organizationId: ORG, userId: USER }); + for (let turn = 0; turn < 5; turn++) { + await seedTurnEval(db, { + chatId: "chat_capped", + turn, + ageMs: TURN_EVAL_RETENTION_MS + 60_000, + }); + } + + expect(await runDashboardAgentRetention(db, { limit: 2, maxBatches: 2 })).toMatchObject({ + turnEvals: 4, + }); + expect(await count("chat_turn_evals")).toBe(1); + + expect(await runDashboardAgentRetention(db, { limit: 2, maxBatches: 2 })).toMatchObject({ + turnEvals: 1, + }); + expect(await count("chat_turn_evals")).toBe(0); + }, + 60_000 + ); + + postgresTest( + "one failing pass doesn't cost the others, and the run throws so the job retries", + async ({ postgresContainer }) => { + const db = await boot(postgresContainer.getConnectionUri()); + await createChat(db, { id: "chat_failing", organizationId: ORG, userId: USER }); + await seedTurnEval(db, { + chatId: "chat_failing", + turn: 0, + ageMs: TURN_EVAL_RETENTION_MS + 60_000, + }); + await seedTerminalWatch("watch_after", "chat_failing", 10); + + await expect( + runDashboardAgentRetention(db, { + purgeChats: async () => { + throw new Error("the chat purge failed"); + }, + }) + ).rejects.toThrow(/chats/); + + // The passes either side of the failing one still ran. + expect(await count("chat_turn_evals")).toBe(0); + expect(await count("watches")).toBe(0); + }, + 60_000 + ); +}); diff --git a/internal-packages/dashboard-agent/src/maintenance.ts b/internal-packages/dashboard-agent/src/maintenance.ts new file mode 100644 index 0000000000..380de3bf4f --- /dev/null +++ b/internal-packages/dashboard-agent/src/maintenance.ts @@ -0,0 +1,136 @@ +import { + deleteTerminalWatchesOlderThan, + deleteTurnEvalsOlderThan, + deleteWatchSubmissionsOlderThan, + hardDeleteChatsSoftDeletedBefore, + type DashboardAgentDb, +} from "@internal/dashboard-agent-db"; +import { logger, schedules } from "@trigger.dev/sdk"; +import { serializeError } from "./serialize-error"; +import { getWatchDb, watchConnectionString } from "./watch-task-adapters"; + +/** + * Retention for the agent's own datastore: judged turns, soft-deleted chats, and the + * finished watch rows. One daily pass, oldest first, bounded per statement. + */ + +/** + * How long a judged turn is kept. Nothing reads the table today, and the rows carry the + * user's question next to the agent's answer, so the period is the shortest one that still + * lets a month of product review be aggregated. + */ +export const TURN_EVAL_RETENTION_MS = 30 * 24 * 60 * 60 * 1000; + +/** + * How long a soft-deleted chat is kept before it and its children are hard-deleted. Long + * enough that an accidental delete can still be investigated; organization deletion soft- + * deletes the org's chats, so those are removed the same way once the window passes. + */ +const CHAT_SOFT_DELETE_RETENTION_MS = 30 * 24 * 60 * 60 * 1000; + +/** How long a terminal watch and its submission ledger are kept. */ +const WATCH_RETENTION_MS = 7 * 24 * 60 * 60 * 1000; + +/** Per-statement cap. */ +const RETENTION_BATCH_LIMIT = 500; + +/** Cap on the statements one pass may run, so a huge backlog can't run forever. */ +const MAX_RETENTION_BATCHES = 200; + +export type RetentionResult = { + turnEvals: number; + chats: number; + watches: number; + watchSubmissions: number; +}; + +type Purge = (params: { before: Date; limit: number }) => Promise; + +export type RetentionDeps = { + now?: () => Date; + limit?: number; + maxBatches?: number; + purgeTurnEvals?: Purge; + purgeChats?: Purge; + purgeWatches?: Purge; + purgeWatchSubmissions?: Purge; +}; + +/** + * Runs every retention pass. Each is bounded and independent: a failing one is reported at + * the end so it can't mask the others, and the throw retries the run. + */ +export async function runDashboardAgentRetention( + db: DashboardAgentDb, + deps: RetentionDeps = {} +): Promise { + const now = deps.now?.() ?? new Date(); + const limit = deps.limit ?? RETENTION_BATCH_LIMIT; + const maxBatches = deps.maxBatches ?? MAX_RETENTION_BATCHES; + + const purgeTurnEvals = deps.purgeTurnEvals ?? ((params) => deleteTurnEvalsOlderThan(db, params)); + const purgeChats = deps.purgeChats ?? ((params) => hardDeleteChatsSoftDeletedBefore(db, params)); + const purgeWatches = + deps.purgeWatches ?? ((params) => deleteTerminalWatchesOlderThan(db, params)); + const purgeWatchSubmissions = + deps.purgeWatchSubmissions ?? ((params) => deleteWatchSubmissionsOlderThan(db, params)); + + const result: RetentionResult = { turnEvals: 0, chats: 0, watches: 0, watchSubmissions: 0 }; + const failed: string[] = []; + + const cutoff = (retentionMs: number) => new Date(now.getTime() - retentionMs); + + // Drains rather than deleting one batch a day: the statement is capped, so a backlog + // needs several of them. + async function drain(name: string, purge: Purge, before: Date): Promise { + let total = 0; + try { + for (let batch = 0; batch < maxBatches; batch++) { + const deleted = await purge({ before, limit }); + total += deleted; + if (deleted < limit) break; + if (batch === maxBatches - 1) { + logger.warn(`dashboard-agent retention hit the batch cap: ${name}`, { total, before }); + } + } + } catch (error) { + failed.push(name); + logger.error(`dashboard-agent retention failed: ${name}`, { error: serializeError(error) }); + } + return total; + } + + result.turnEvals = await drain("turn evals", purgeTurnEvals, cutoff(TURN_EVAL_RETENTION_MS)); + result.chats = await drain("chats", purgeChats, cutoff(CHAT_SOFT_DELETE_RETENTION_MS)); + + const watchesBefore = cutoff(WATCH_RETENTION_MS); + result.watches = await drain("watches", purgeWatches, watchesBefore); + // The ledger's rows age out on the same window: past it no client is still retrying. + result.watchSubmissions = await drain("watch submissions", purgeWatchSubmissions, watchesBefore); + + if (failed.length > 0) { + throw new Error(`The dashboard agent retention pass failed: ${failed.join(", ")}`); + } + + return result; +} + +export const dashboardAgentMaintenance = schedules.task({ + id: "dashboard-agent-maintenance", + cron: "0 3 * * *", + retry: { maxAttempts: 3 }, + run: async (): Promise => { + if (!watchConnectionString()) { + logger.warn( + "dashboard-agent maintenance skipped: no DASHBOARD_AGENT_DATABASE_URL or DATABASE_URL" + ); + return undefined; + } + + const { db } = getWatchDb(); + + const result = await runDashboardAgentRetention(db); + logger.info("dashboard-agent retention swept", result); + return result; + }, +}); diff --git a/internal-packages/dashboard-agent/src/serialize-error.ts b/internal-packages/dashboard-agent/src/serialize-error.ts new file mode 100644 index 0000000000..7c21700796 --- /dev/null +++ b/internal-packages/dashboard-agent/src/serialize-error.ts @@ -0,0 +1,4 @@ +/** A raw Error serializes to `{}` in structured logs, so log its fields. */ +export function serializeError(error: unknown): { message: string; stack?: string } | string { + return error instanceof Error ? { message: error.message, stack: error.stack } : String(error); +} diff --git a/internal-packages/dashboard-agent/src/watch-task-adapters.ts b/internal-packages/dashboard-agent/src/watch-task-adapters.ts index 45222b6fc4..5add3203fe 100644 --- a/internal-packages/dashboard-agent/src/watch-task-adapters.ts +++ b/internal-packages/dashboard-agent/src/watch-task-adapters.ts @@ -17,14 +17,20 @@ import type { WatchBatchCheckResponse, WatchBatchTickPayload } from "./watch-bat /** What the two watch tasks plug into the lifecycle: the db, the wake append, the callbacks. */ +/** The url the watch, sweep and retention tasks connect with, so an unwired deployment can skip instead of throw. */ +export function watchConnectionString(env: NodeJS.ProcessEnv = process.env): string | undefined { + // `||`, so an empty dedicated url falls back instead of connecting to "". + return env.DASHBOARD_AGENT_DATABASE_URL || env.DATABASE_URL; +} + // One connection pool per worker process. let dbClient: DashboardAgentDbClient | undefined; export function getWatchDb(): DashboardAgentDbClient { if (!dbClient) { - const connectionString = process.env.DASHBOARD_AGENT_DATABASE_URL ?? process.env.DATABASE_URL; + const connectionString = watchConnectionString(); if (!connectionString) { throw new Error( - "DASHBOARD_AGENT_DATABASE_URL (or DATABASE_URL) must be set for the watch task" + "DASHBOARD_AGENT_DATABASE_URL (or DATABASE_URL) must be set for the watch and sweep tasks" ); } dbClient = createDashboardAgentDb(connectionString, { max: 2 }); diff --git a/internal-packages/dashboard-agent/vitest.config.ts b/internal-packages/dashboard-agent/vitest.config.ts index 04b861ccf2..6b4efb4cfd 100644 --- a/internal-packages/dashboard-agent/vitest.config.ts +++ b/internal-packages/dashboard-agent/vitest.config.ts @@ -1,11 +1,15 @@ +import { DurationShardingSequencer } from "@internal/testcontainers/sequencer"; import { defineConfig } from "vitest/config"; export default defineConfig({ test: { + sequence: { sequencer: DurationShardingSequencer }, include: ["src/**/*.test.ts"], environment: "node", - testTimeout: 20000, - hookTimeout: 20000, + isolate: true, + fileParallelism: false, + testTimeout: 120_000, + hookTimeout: 120_000, }, esbuild: { target: "node18", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index e0a261a608..f698fa90ca 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1040,6 +1040,9 @@ importers: '@ai-sdk/provider': specifier: 3.0.8 version: 3.0.8 + '@internal/testcontainers': + specifier: workspace:* + version: link:../testcontainers '@trigger.dev/build': specifier: workspace:* version: link:../../packages/build @@ -28560,7 +28563,7 @@ snapshots: node-abi@3.89.0: dependencies: - semver: 7.8.5 + semver: 7.8.1 optional: true node-abort-controller@3.1.1: {}