diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 4853284df55..6f7b2b181cc 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -1888,6 +1888,17 @@ TLS and protocol metadata. To use an isolated external database, set `PAPERCLIP_CONTINUATION_BENCHMARK_DATABASE_URL` to a disposable database ending in `_test`. The harness applies migrations there and adds synthetic fixtures. +Continuation history reads select completed tool receipts and the typed +conversation/workspace recovery fields used to choose the prior terminal run. +Every completed receipt keeps its full result. Receipt order, company/agent/task +scope, and the complete retained history stay unchanged. Saved results remain +available to detail, recovery, and execution paths. +Use `pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-continuation-history.ts --baseline-ref=` +to compare generated SQL, complete continuation envelopes, and DataRow bytes. +The benchmark includes a large required-receipt control. Its external database +option is `PAPERCLIP_CONTINUATION_HISTORY_BENCHMARK_DATABASE_URL`, with the same +disposable `_test` database restriction. + Retry scheduling is in `server/src/services/heartbeat/retries.ts`. It owns bounded retry schedules, connection and workspace contention deferrals, shared-workspace holder checks, due retry promotion, and retry-now requests. `createHeartbeatRetries` diff --git a/scripts/benchmark-continuation-history.ts b/scripts/benchmark-continuation-history.ts new file mode 100644 index 00000000000..c26a118c589 --- /dev/null +++ b/scripts/benchmark-continuation-history.ts @@ -0,0 +1,120 @@ +import assert from "node:assert/strict"; +import { execFileSync } from "node:child_process"; +import { randomBytes, randomUUID } from "node:crypto"; +import { promises as fs } from "node:fs"; +import path from "node:path"; +import { fileURLToPath, pathToFileURL } from "node:url"; +import { agents, applyPendingMigrations, companies, createDb, heartbeatRuns, issues, startEmbeddedPostgresTestDatabase, type Db } from "../packages/db/src/index.js"; +import { buildExecutionContinuation } from "../server/src/services/execution-continuation.js"; +import { hasConversationContinuationPolicy } from "../server/src/services/conversation-continuation.js"; +import { resolveExternalTestDatabaseUrl } from "../server/src/__tests__/helpers/external-test-database.js"; + +const repoRoot = fileURLToPath(new URL("../", import.meta.url)); +const git = (...args: string[]) => execFileSync("git", args, { cwd: repoRoot, encoding: "utf8" }); +const baselineRef = process.argv.find(arg => arg.startsWith("--baseline-ref="))?.slice(15); +assert.ok(baselineRef, "Pass a pinned pre-fix --baseline-ref"); +const baselineRevision = git("rev-parse", "--verify", `${baselineRef}^{commit}`).trim(); +const baselineOnly = process.argv.includes("--baseline-only"); +const keepAlive = setInterval(() => {}, 1000); +let database: Awaited> | undefined; +let db: Db | undefined; +let baselineDir: string | undefined; +try { + const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_CONTINUATION_HISTORY_BENCHMARK_DATABASE_URL"); + if (externalUrl) { + await applyPendingMigrations(externalUrl); + database = { connectionString: externalUrl, cleanup: async () => {} }; + } else database = await startEmbeddedPostgresTestDatabase("continuation-history-benchmark-"); + db = createDb(database.connectionString); + const original = new URL("../server/src/services/execution-continuation.ts", import.meta.url); + baselineDir = await fs.mkdtemp(path.join(repoRoot, "server/scripts/.history-baseline-")); + const source = git("show", `${baselineRevision}:server/src/services/execution-continuation.ts`); + const baselinePath = path.join(baselineDir, "continuation.ts"); + await fs.writeFile(baselinePath, source.replace(/(from |import\()"(\.{1,2}\/[^\"]+)"/g, + (_match, prefix: string, specifier: string) => `${prefix}${JSON.stringify(fileURLToPath(new URL(specifier, original)))}`)); + const baseline = await import(pathToFileURL(baselinePath).href) as { buildExecutionContinuation: typeof buildExecutionContinuation }; + const object = (value: unknown): Record => value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; + const receipts = (value: unknown) => Object.entries(object(object(value).apiToolReceipts)).flatMap(([id, entry]) => { + const receipt = object(entry); + return receipt.state === "completed" && typeof receipt.operationId === "string" ? [{ id, operationId: receipt.operationId, result: receipt.result }] : []; + }); + const results = []; + for (const receiptHeavy of [false, true]) { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Synthetic history", issuePrefix: `B${companyId.replaceAll("-", "")}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Fixture", adapterType: "paperclip_runner" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Continue", status: "in_progress", assigneeAgentId: agentId }); + const runCount = receiptHeavy ? 1 : 25; + const requiredResult = receiptHeavy ? randomBytes(1536 * 1024).toString("base64") : { receipt: "required output" }; + for (let index = 0; index < runCount; index++) await db.insert(heartbeatRuns).values({ companyId, agentId, + contextSnapshot: { issueId }, status: "failed", errorCode: "adapter_failed", runtimeMode: "legacy", + createdAt: new Date(Date.UTC(2020, 0, 1, 0, 0, index)), + resultJson: { nativeResult: { output: receiptHeavy ? "small" : randomBytes(144 * 1024).toString("base64") }, + conversationContinuation: "continue_conversation_v1", + apiToolReceipts: { saved: { state: "completed", operationId: "save_document", result: requiredResult, + request: receiptHeavy ? "small" : randomBytes(24 * 1024).toString("base64") }, pending: { state: "started", operationId: "unfinished", result: "not a completed action" } } }, + }); + await db.$client.unsafe("ANALYZE"); + async function capture(invoke: typeof buildExecutionContinuation) { + let query: { sql: string; params: unknown[] } | undefined; + let history: Array> | undefined; + 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 selection = from(...args), execute = selection.execute.bind(selection); + if (args[0] === heartbeatRuns) selection.execute = async (...executeArgs: Parameters) => { + const statement = selection.toSQL(), rows = await execute(...executeArgs); + if (statement.sql.includes('order by "heartbeat_runs"."created_at" asc')) { query = statement; history = rows; } + return rows; + }; + return selection; + }) as typeof builder.from; + return builder; + }; + }}); + const envelope = await invoke({ db: recorded, companyId, issueId, agentId, context: {}, summary: null, exposeLowTrustRaw: false }); + assert.ok(query); assert.ok(history); + return { query, history, envelope }; + } + const before = await capture(baseline.buildExecutionContinuation); + const after = await capture(baselineOnly ? baseline.buildExecutionContinuation : buildExecutionContinuation); + assert.deepEqual(after.envelope, before.envelope); + assert.equal(after.history.length, before.history.length); + for (let index = 0; index < before.history.length; index++) { + const oldRow = before.history[index], newRow = after.history[index]; + for (const [key, value] of Object.entries(oldRow)) if (key !== "result") assert.deepEqual(newRow[key], value); + assert.deepEqual(receipts(newRow.result), receipts(oldRow.result)); + assert.equal(hasConversationContinuationPolicy(newRow.result as Record), hasConversationContinuationPolicy(oldRow.result as Record)); + } + const queries = [before.query, after.query], bytes: number[] = [], samples: number[][] = [[], []]; + for (let iteration = 0; iteration < 9; iteration++) for (const variant of iteration % 2 ? [1, 0] : [0, 1]) { + const query = queries[variant], start = performance.now(); + const rows = await db.$client.unsafe(query.sql, query.params as never[]).raw(); + if (iteration >= 2) samples[variant].push(performance.now() - start); + bytes[variant] = rows.reduce((sum, row) => sum + 7 + row.reduce((n: number, value: Buffer | null) => n + 4 + (value?.length ?? 0), 0), 0); + } + if (!baselineOnly && !receiptHeavy) assert.ok(bytes[1] < bytes[0] / 10, "Wide unused-result fixture must remove more than 90% of DataRow bytes"); + if (receiptHeavy) { + assert.ok(after.envelope.completedActions); + assert.deepEqual(after.envelope.completedActions[0].result, requiredResult); + } + const plans = []; + for (const query of queries) { + const [plan] = await db.$client.unsafe(`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${query.sql}`, query.params as never[]); + plans.push(plan["QUERY PLAN"]); + } + const median = (values: number[]) => [...values].sort((a, b) => a - b)[Math.floor(values.length / 2)]; + results.push({ fixture: receiptHeavy ? "required-receipt-result" : "wide-unused-output", runCount, rows: after.history.length, + equalEnvelope: true, equalRetainedColumnsAndReceipts: true, baselineDataRowBytes: bytes[0], treatmentDataRowBytes: bytes[1], + reduction: 1 - bytes[1] / bytes[0], baselineMedianMs: median(samples[0]), treatmentMedianMs: median(samples[1]), samples, queries, plans }); + } + console.log(JSON.stringify({ baselineRevision, treatmentRevision: git("rev-parse", "HEAD").trim(), + treatmentDiff: git("diff", "HEAD", "--", "server/src/services/execution-continuation.ts"), baselineOnly, + byteBoundary: "Exact text-format PostgreSQL DataRow frames, excluding metadata, TLS and TCP", postgres: await db.$client.unsafe("select version()"), results }, null, 2)); +} finally { + try { await db?.$client.end({ timeout: 0 }); await database?.cleanup(); + if (baselineDir) await fs.rm(baselineDir, { recursive: true, force: true }); + } finally { clearInterval(keepAlive); } +} diff --git a/server/src/services/execution-continuation-history-read.test.ts b/server/src/services/execution-continuation-history-read.test.ts new file mode 100644 index 00000000000..78b288849dc --- /dev/null +++ b/server/src/services/execution-continuation-history-read.test.ts @@ -0,0 +1,128 @@ +import assert from "node:assert/strict"; +import { randomBytes, randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { agents, applyPendingMigrations, companies, createDb, heartbeatRuns, issues, type Db } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "../__tests__/helpers/external-test-database.js"; +import { buildExecutionContinuation } from "./execution-continuation.js"; + +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_EXECUTION_CONTINUATION_TEST_DATABASE_URL"); +const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); +describe.skipIf(!support.supported)("continuation history PostgreSQL projection", () => { + let database: Awaited>; + let db: Db; + beforeAll(async () => { + if (externalUrl) { + await applyPendingMigrations(externalUrl); + database = { connectionString: externalUrl, cleanup: async () => {} }; + } else database = await startEmbeddedPostgresTestDatabase("continuation-history-read-"); + db = createDb(database.connectionString); + }, 60_000); + afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); + + async function fixture(result: unknown) { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "History", issuePrefix: `H${companyId.replaceAll("-", "")}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Executor", adapterType: "paperclip_runner" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Continue", status: "in_progress", assigneeAgentId: agentId }); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "failed", errorCode: "adapter_failed", + contextSnapshot: { issueId }, createdAt: new Date("2020-01-01T00:00:00Z"), resultJson: result as Record | null }).returning(); + return { companyId, agentId, issueId, run }; + } + + async function invoke(f: Awaited>, fullResult: boolean, context: Record = {}) { + const reads: Array>> = []; + const recorded = new Proxy(db, { get(target, key) { + if (key !== "select") return Reflect.get(target, key); + return (...fields: Parameters) => { + const historyRead = Object.hasOwn(fields[0] ?? {}, "retryOfRunId"); + const builder = historyRead && fullResult + ? target.select({ ...fields[0], result: heartbeatRuns.resultJson }) : target.select(...fields); + const from = builder.from.bind(builder); + builder.from = ((...args: Parameters) => { + const query = from(...args), execute = query.execute.bind(query); + if (historyRead) query.execute = async (...executeArgs: Parameters) => { + const rows = await execute(...executeArgs); reads.push(rows); return rows; + }; + return query; + }) as typeof builder.from; + return builder; + }; + }}); + const result = await buildExecutionContinuation({ db: recorded, companyId: f.companyId, agentId: f.agentId, + issueId: f.issueId, context, summary: null, exposeLowTrustRaw: false }); + return { result, reads }; + } + + async function compare(f: Awaited>, context: Record = {}) { + const before = await invoke(f, true, context), after = await invoke(f, false, context); + expect(after.result).toEqual(before.result); + expect(after.reads).toHaveLength(1); + expect(after.reads[0].map(({ result: _result, ...fields }) => fields)).toEqual(before.reads[0].map(({ result: _result, ...fields }) => fields)); + return { before, after }; + } + + it("keeps every completed receipt, its full result, missing values and receipt order", async () => { + const payload = randomBytes(128 * 1024).toString("base64"); + const result = { nativeResult: { output: payload }, conversationContinuation: "continue_conversation_v1", apiToolReceipts: { + "10": { state: "completed", operationId: "ten", result: null, unusedRequest: payload }, + "2": { state: "completed", operationId: "two", result: { nested: [payload, false, null] } }, + missing: { state: "completed", operationId: "" }, + pending: { state: "started", operationId: "pending", result: payload }, + invalid: { state: "completed", operationId: 42, result: payload }, + array: ["completed", "operation"], + scalar: "completed", number: 42, null: null, + } }; + const f = await fixture(result), { before, after } = await compare(f); + assert.ok(after.result.completedActions); + expect(after.result.completedActions.map(receipt => receipt.receiptId)).toEqual(["2", "10", "missing"]); + expect(after.result.completedActions[0].result).toEqual(result.apiToolReceipts["2"].result); + expect(after.result.completedActions[1].result).toBeNull(); + expect(after.result.completedActions[2].result).toBeUndefined(); + expect(Buffer.byteLength(JSON.stringify(after.reads))).toBeLessThan(Buffer.byteLength(JSON.stringify(before.reads)) / 2); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)))[0].resultJson).toEqual(result); + }); + + const recovery = { schema: "paperclip.workspace-restore-recovery.v1" }; + it.each([ + { value: null, interrupted: false }, { value: [], interrupted: false }, { value: "scalar", interrupted: false }, { value: 42, interrupted: false }, + { value: {}, interrupted: false }, { value: { conversationContinuation: "continue_conversation_v1", apiToolReceipts: [1, 2] }, interrupted: true }, + { value: { conversationContinuation: 42, apiToolReceipts: "malformed" }, interrupted: false }, + ...["restore_permission_denied", "restore_lock_timeout", "restore_unsafe_archive", "restore_failed"].map(workspaceRestoreFailure => ({ + value: { conversationContinuation: "continue_conversation_v1", workspaceRestoreFailure, workspaceRestoreRecovery: recovery }, interrupted: false, + })), + { value: { conversationContinuation: "continue_conversation_v1", workspaceRestoreFailure: "restore_unsafe_archive" }, interrupted: false }, + { value: { conversationContinuation: "continue_conversation_v1", workspaceRestoreFailure: "custom", workspaceRestoreRecovery: recovery }, interrupted: true }, + ...[null, [], "schema", { schema: 42 }, { schema: "other" }].map(workspaceRestoreRecovery => ({ + value: { conversationContinuation: "continue_conversation_v1", workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery }, interrupted: true, + })), + ])("keeps continuation/recovery decisions for typed legacy JSON $value", async ({ value, interrupted }) => { + const f = await fixture(value), { after } = await compare(f); + expect(after.result.interruptedRunId).toBe(interrupted ? f.run.id : undefined); + expect(after.result.completedActions).toEqual([]); + }); + + it("keeps full ordered history and source-agent scope without cross-company or cross-task receipts", async () => { + const completed = (operationId: string) => ({ apiToolReceipts: { saved: { state: "completed", operationId, result: { operationId } } } }); + const f = await fixture({ ...completed("first"), conversationContinuation: "continue_conversation_v1" }); + const otherAgentId = randomUUID(); + await db.insert(agents).values({ id: otherAgentId, companyId: f.companyId, name: "Previous owner", adapterType: "paperclip_runner" }); + const [source] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: otherAgentId, status: "succeeded", + contextSnapshot: { issueId: f.issueId }, resultJson: completed("other-agent"), createdAt: new Date("2020-01-01T00:01:00Z") }).returning(); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "cancelled", errorCode: "execution_reconciliation_required", + contextSnapshot: { issueId: f.issueId }, resultJson: completed("cancelled"), createdAt: new Date("2020-01-01T00:02:00Z") }); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "failed", + contextSnapshot: { issueId: randomUUID() }, resultJson: completed("foreign-task") }); + const foreign = await fixture(completed("foreign-company")); + await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: f.issueId } }).where(eq(heartbeatRuns.id, foreign.run.id)); + const ordinary = await compare(f); + assert.ok(ordinary.after.result.completedActions); + expect(ordinary.after.result.completedActions.map(receipt => receipt.operationId)).toEqual(["first", "cancelled"]); + expect(ordinary.after.result.interruptedRunId).toBe(f.run.id); + const resumed = await compare(f, { previousRunId: source.id }); + assert.ok(resumed.after.result.completedActions); + expect(resumed.after.result.completedActions.map(receipt => receipt.operationId)).toEqual(["first", "other-agent", "cancelled"]); + expect(resumed.after.result.interruptedRunId).toBeUndefined(); + }); +}); diff --git a/server/src/services/execution-continuation.test.ts b/server/src/services/execution-continuation.test.ts index 834bf515ee7..fc9e88795ad 100644 --- a/server/src/services/execution-continuation.test.ts +++ b/server/src/services/execution-continuation.test.ts @@ -3,6 +3,7 @@ import { eq } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { agents, + applyPendingMigrations, companies, createDb, heartbeatRuns, @@ -15,6 +16,7 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, } from "../__tests__/helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "../__tests__/helpers/external-test-database.js"; import { StaleExecutionContinuationError, buildExecutionContinuation, currentContinuationOrigins, projectHumanInteractionResponse } from "./execution-continuation.js"; const expectStaleContinuation = async ( @@ -29,7 +31,8 @@ const expectMissingContinuationContext = async (run: () => Promise) => await expect(result).rejects.toThrow("continuation_source_context_missing"); await expect(result).rejects.not.toBeInstanceOf(StaleExecutionContinuationError); }; -const support = await getEmbeddedPostgresTestSupport(); +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_EXECUTION_CONTINUATION_TEST_DATABASE_URL"); +const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); (support.supported ? describe : describe.skip)( "authorized continuation context", () => { @@ -44,13 +47,16 @@ const support = await getEmbeddedPostgresTestSupport(); laterId = randomUUID(), interactionId = randomUUID(); beforeAll(async () => { - database = await startEmbeddedPostgresTestDatabase( + if (externalUrl) { + await applyPendingMigrations(externalUrl); + database = { connectionString: externalUrl, cleanup: async () => {} }; + } else database = await startEmbeddedPostgresTestDatabase( "paperclip-continuation-context-", ); db = createDb(database.connectionString); await db .insert(companies) - .values({ id: companyId, name: "Continuation", issuePrefix: "CTX" }); + .values({ id: companyId, name: "Continuation", issuePrefix: `CTX${companyId.replaceAll("-", "")}` }); await db .insert(agents) .values({ diff --git a/server/src/services/execution-continuation.ts b/server/src/services/execution-continuation.ts index c44ce6ee325..378e7f1382c 100644 --- a/server/src/services/execution-continuation.ts +++ b/server/src/services/execution-continuation.ts @@ -31,6 +31,28 @@ const object = (v: unknown): Record => : {}; const string = (v: unknown) => typeof v === "string" && v.length > 0 ? v : null; +const historyResult = sql | null>`case + when jsonb_typeof(${heartbeatRuns.resultJson}) = 'object' then ( + select coalesce(jsonb_object_agg(field.key, case + when field.key = 'apiToolReceipts' and jsonb_typeof(field.value) = 'object' then ( + select coalesce(jsonb_object_agg(receipt.key, ( + select jsonb_object_agg(detail.key, detail.value) + from jsonb_each(receipt.value) as detail(key, value) + where detail.key in ('state', 'operationId', 'result') + )), '{}'::jsonb) + from jsonb_each(field.value) as receipt(key, value) + where receipt.value->'state' = '"completed"'::jsonb + and jsonb_typeof(receipt.value->'operationId') = 'string' + ) + when field.key = 'workspaceRestoreRecovery' and jsonb_typeof(field.value) = 'object' then ( + select coalesce(jsonb_object_agg(detail.key, detail.value), '{}'::jsonb) + from jsonb_each(field.value) as detail(key, value) + where detail.key = 'schema' + ) + else field.value end), '{}'::jsonb) + from jsonb_each(${heartbeatRuns.resultJson}) as field(key, value) + where field.key in ('apiToolReceipts', 'conversationContinuation', 'workspaceRestoreFailure', 'workspaceRestoreRecovery') + ) else ${heartbeatRuns.resultJson} end`; export function continuationOriginCommentIds(context: unknown): string[] { const c = object(context); const prior = object(c.executionContinuation); @@ -335,7 +357,7 @@ export async function buildExecutionContinuation(input: { ? { kind: "description" as const, id: issue.id, revision: hashObjectiveSource(issue.description) } : { kind: "title" as const, id: issue.id, revision: hashObjectiveSource(issue.title) }; const priorRuns = await db - .select({ id: heartbeatRuns.id, result: heartbeatRuns.resultJson, status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode, runtimeMode: heartbeatRuns.runtimeMode, retryOfRunId: heartbeatRuns.retryOfRunId }) + .select({ id: heartbeatRuns.id, result: historyResult, status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode, runtimeMode: heartbeatRuns.runtimeMode, retryOfRunId: heartbeatRuns.retryOfRunId }) .from(heartbeatRuns) .where( and(