diff --git a/doc/DATABASE.md b/doc/DATABASE.md index 56e9736a66d..1dc73cbf9d5 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -113,6 +113,14 @@ Identity task locks and activation receipts project typed legacy task keys with the native and first-class issue IDs. These projections preserve task binding precedence and validation without transferring prompts or output. +Recovery's latest-run and legacy-continuation eligibility reads select structured +result fields for error families, +workspace/configuration receipts, cancellation attribution, retry deadlines, and +conversation finalization. The `adapter_failed`, `provider_quota`, and +`configuration_incomplete` classifiers retain full results because they inspect +arbitrary provider error text. Stored results and execution/detail reads remain +complete; no recovery decision is cached. + To measure authentication query responses on disposable synthetic data: ```sh diff --git a/server/src/services/recovery/run-result-read.test.ts b/server/src/services/recovery/run-result-read.test.ts new file mode 100644 index 00000000000..7f937c36f0a --- /dev/null +++ b/server/src/services/recovery/run-result-read.test.ts @@ -0,0 +1,107 @@ +import { randomBytes, randomUUID } from "node:crypto"; +import { agents, companies, createDb, heartbeatRuns, type Db } from "@paperclipai/db"; +import { eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { isAiAuthenticationBlocked } from "../ai-auth-failure.js"; +import { recoveryRunResultColumn } from "./run-result-read.js"; +import { classifyAdapterFailureForRecovery, classifyContinuationFailure, recoveryService } from "./service.js"; + +const support = await getEmbeddedPostgresTestSupport(); +describe.skipIf(!support.supported)("recovery result PostgreSQL projection", () => { + let database: Awaited>; + let db: ReturnType; + let companyId: string; + let agentId: string; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("recovery-result-read-"); + db = createDb(database.connectionString); + const [company] = await db.insert(companies).values({ name: "Recovery fixture", issuePrefix: "RR" }).returning(); + const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Recovery agent", adapterType: "process" }).returning(); + companyId = company.id; + agentId = agent.id; + }, 60_000); + afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); + + async function read(resultJson: unknown, errorCode: string | null = null) { + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "succeeded", errorCode, + resultJson: resultJson as Record | null }).returning(); + const [selected] = await db.select({ resultJson: recoveryRunResultColumn }).from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)); + return { run, selected }; + } + + it("keeps typed recovery receipts without returning execution output or modifying history", async () => { + const evidence = { errorFamily: "transient_upstream", workspaceValidation: { fingerprint: "workspace" }, + configurationIncomplete: { reason: "workspace_unavailable", fingerprint: "configuration" }, + cancelledByActorType: "board", retryNotBefore: "2099-01-01T00:00:00Z", transientRetryNotBefore: 123, + providerQuotaRetryNotBefore: null, finalizationReasonCode: "conversation_turn_finished" }; + const payload = randomBytes(512 * 1024).toString("base64"); + const result = { ...evidence, summary: payload, output: payload, providerResponse: { output: payload } }; + const { run, selected } = await read(result); + expect(selected.resultJson).toEqual(evidence); + expect(Buffer.byteLength(JSON.stringify(selected))).toBeLessThan(1024); + expect(classifyContinuationFailure({ ...run, resultJson: selected.resultJson })).toEqual(classifyContinuationFailure(run)); + const [stored] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)); + expect(stored).toEqual(run); + }); + + it.each(["adapter_failed", "provider_quota", "configuration_incomplete"])("retains arbitrary provider evidence for %s classification", async errorCode => { + const now = new Date("2020-01-01T00:00:00Z"); + const result = { providerSpecificField: { nestedResponse: "Rate limit exceeded. Try again after 2099-01-01T00:00:00Z" }, + configurationIncomplete: { reason: "ai_connection_unavailable" }, + output: randomBytes(64 * 1024).toString("base64") }; + const { run, selected } = await read(result, errorCode); + expect(selected.resultJson).toEqual(result); + const projected = { ...run, livenessState: "blocked", resultJson: selected.resultJson }; + const complete = { ...run, livenessState: "blocked" }; + expect(classifyAdapterFailureForRecovery(projected, now)).toEqual(classifyAdapterFailureForRecovery(complete, now)); + expect(classifyContinuationFailure(projected)).toEqual(classifyContinuationFailure(complete)); + expect(isAiAuthenticationBlocked(projected)).toBe(isAiAuthenticationBlocked(complete)); + }); + + it("checks legacy eligibility and repair ancestry without returning large result bodies", async () => { + const result = { output: randomBytes(512 * 1024).toString("base64") }; + const { run: source } = await read(result); + const { run: queued } = await read(result); + await db.update(heartbeatRuns).set({ status: "queued", contextSnapshot: { + issueId: randomUUID(), dispositionRepairSourceRunId: source.id, + legacyDispositionEpisode: { id: source.id, attempt: 1, maxAttempts: 2 }, + } }).where(eq(heartbeatRuns.id, queued.id)); + const reads: unknown[][] = []; + const recorded = new Proxy(db, { get(target, key) { + if (key !== "select") return Reflect.get(target, key); + return (...fields: Parameters) => { + const builder = target.select(...fields), from = builder.from.bind(builder); + builder.from = ((...args: Parameters) => { + const query = from(...args); + if (args[0] === heartbeatRuns) { + const execute = query.execute.bind(query); + query.execute = async (...executeArgs: Parameters) => { + const rows = await execute(...executeArgs); reads.push(rows); return rows; + }; + } + return query; + }) as typeof builder.from; + return builder; + }; + }}); + const service = recoveryService(recorded, { enqueueWakeup: async () => { throw new Error("unexpected wake"); } }); + expect(await service.reconcileLegacyContinuation(source.id)).toBe("skipped"); + expect(await service.legacyRepairDispatchBlock(queued.id)).toBe("missing_issue"); + expect(reads).toHaveLength(3); + expect(Buffer.byteLength(JSON.stringify(reads))).toBeLessThan(16 * 1024); + const [stored] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, source.id)); + expect(stored).toEqual(source); + const [storedQueued] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, queued.id)); + expect(storedQueued.resultJson).toEqual(result); + }); + + it.each([null, [], ["unexpected"], "scalar", 42, false, {}, { errorFamily: false, cancelledByActorType: 3, + workspaceValidation: [], configurationIncomplete: null, retryNotBefore: { invalid: true }, finalizationReasonCode: "" }] + .map(result => ({ result })))("preserves structured evidence types for $result", async ({ result }) => { + const { run, selected } = await read(result); + const expected = result && typeof result === "object" && !Array.isArray(result) ? result : result === null ? null : {}; + expect(selected.resultJson).toEqual(expected); + expect(classifyContinuationFailure({ ...run, resultJson: selected.resultJson })).toEqual(classifyContinuationFailure(run)); + }); +}); diff --git a/server/src/services/recovery/run-result-read.ts b/server/src/services/recovery/run-result-read.ts new file mode 100644 index 00000000000..3f2b0ddecdd --- /dev/null +++ b/server/src/services/recovery/run-result-read.ts @@ -0,0 +1,25 @@ +import { heartbeatRuns } from "@paperclipai/db"; +import { sql } from "drizzle-orm"; + +const recoveryResultKeys = [ + "errorFamily", "workspaceValidation", "configurationIncomplete", + "cancelledByActorType", "retryNotBefore", "transientRetryNotBefore", + "providerQuotaRetryNotBefore", "finalizationReasonCode", +] as const; + +/** These three classifiers inspect arbitrary provider error text. Other + * recovery decisions consume the structured fields below, not model output. + * Keep the full stored result and all execution/detail reads unchanged. */ +export const recoveryRunResultColumn = sql | null>` + case + when ${heartbeatRuns.resultJson} is null then null + when ${heartbeatRuns.errorCode} in ('adapter_failed', 'provider_quota', 'configuration_incomplete') + then ${heartbeatRuns.resultJson} + else ( + select coalesce(jsonb_object_agg(field.key, field.value), '{}'::jsonb) + from jsonb_each(case when jsonb_typeof(${heartbeatRuns.resultJson}) = 'object' + then ${heartbeatRuns.resultJson} else '{}'::jsonb end) as field(key, value) + where field.key in (${sql.join(recoveryResultKeys.map(key => sql`${key}`), sql`, `)}) + ) + end +`; diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index e517b69ec69..e72811ec8d0 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -1,4 +1,5 @@ import { recoveryRunContextColumn, recoveryRunVersionColumn } from "./run-context-read.js"; +import { recoveryRunResultColumn } from "./run-result-read.js"; import { isAgentAwaitingSetup } from "../../modules/agent-lifecycle/index.js"; import { onboardingFirstTaskAwaitingUserSql } from "./onboarding-first-task-wait.js"; import { isAiAuthenticationBlocked } from "../ai-auth-failure.js"; @@ -235,6 +236,11 @@ type ResolvedDependencyWakeBackstopOptions = { source?: ResolvedDependencyWakeBackstopSource; }; +const recoveryRunFields = { + ...getTableColumns(heartbeatRuns), + resultJson: recoveryRunResultColumn, +}; + type LatestIssueRun = | (Pick< typeof heartbeatRuns.$inferSelect, @@ -983,7 +989,7 @@ export function recoveryService( contextSnapshot: recoveryRunContextColumn, rowVersion: recoveryRunVersionColumn, livenessState: heartbeatRuns.livenessState, - resultJson: heartbeatRuns.resultJson, + resultJson: recoveryRunResultColumn, startedAt: heartbeatRuns.startedAt, createdAt: heartbeatRuns.createdAt, }) @@ -1014,7 +1020,7 @@ export function recoveryService( contextSnapshot: recoveryRunContextColumn, rowVersion: recoveryRunVersionColumn, livenessState: heartbeatRuns.livenessState, - resultJson: heartbeatRuns.resultJson, + resultJson: recoveryRunResultColumn, startedAt: heartbeatRuns.startedAt, createdAt: heartbeatRuns.createdAt, }) @@ -1915,7 +1921,7 @@ export function recoveryService( contextSnapshot: recoveryRunContextColumn, rowVersion: recoveryRunVersionColumn, livenessState: heartbeatRuns.livenessState, - resultJson: heartbeatRuns.resultJson, + resultJson: recoveryRunResultColumn, startedAt: heartbeatRuns.startedAt, createdAt: heartbeatRuns.createdAt, }) @@ -3786,7 +3792,7 @@ export function recoveryService( episode: LegacyDispositionEpisode, dispatchRunId?: string, ) { - const [run] = await db.select().from(heartbeatRuns).where(and( + const [run] = await db.select(recoveryRunFields).from(heartbeatRuns).where(and( eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, issue.companyId), )).limit(1); if (!run) return { kind: "skip" as const, reason: "missing_run" }; @@ -3825,7 +3831,7 @@ export function recoveryService( } async function legacyRepairDispatchBlock(runId: string): Promise { - const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).limit(1); + const [run] = await db.select(recoveryRunFields).from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).limit(1); const context = parseObject(run?.contextSnapshot); if (!run || !parseObject(context.legacyDispositionEpisode).id) return null; if (!["queued", "running", "scheduled_retry"].includes(run.status)) return "repair_not_active"; @@ -3835,7 +3841,7 @@ export function recoveryService( const sourceId = readNonEmptyString(context.dispositionRepairSourceRunId) ?? readNonEmptyString(context.retryOfRunId); const issueId = readNonEmptyString(context.issueId); if (!sourceId || !issueId || episode.attempt < 1 || episode.attempt > episode.maxAttempts) return "invalid_repair_binding"; - const [source] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, sourceId), eq(heartbeatRuns.companyId, run.companyId))).limit(1); + const [source] = await db.select(recoveryRunFields).from(heartbeatRuns).where(and(eq(heartbeatRuns.id, sourceId), eq(heartbeatRuns.companyId, run.companyId))).limit(1); if (!source || source.agentId !== run.agentId || legacyDispositionEpisode(source).id !== episode.id) return "invalid_repair_source"; const [issue] = await db.select().from(issues).where(and(eq(issues.id, issueId), eq(issues.companyId, run.companyId))).limit(1); if (!issue) return "missing_issue"; @@ -3847,7 +3853,7 @@ export function recoveryService( } async function reconcileLegacyContinuation(runId: string) { - const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).limit(1); + const [run] = await db.select(recoveryRunFields).from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).limit(1); if (!run || run.runtimeMode === "native" || run.status !== "succeeded") return "skipped" as const; const context = parseObject(run.contextSnapshot); const issueId = readNonEmptyString(context.issueId) ?? readNonEmptyString(context.taskId);