mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-28 12:04:41 +01:00
perf(server): project continuation history result fields (#17930)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - A continuation carries completed tool receipts so the agent can avoid repeat actions. > - It also checks saved conversation and workspace recovery fields before choosing a prior terminal run. > - The history query transfers every result field for every matching run. > - Native output and unused receipt request fields can make this read large. > - This pull request selects the fields those consumers use and keeps each completed receipt result in full. > - The benefit is less database transfer with the same history, recovery decisions, and completed-action evidence. ## Linked Issues or Issue Description **What existing behavior does this improve?** The history read in `buildExecutionContinuation`. **Current behavior** The query returns complete `result_json` values for all matching runs. Consumers read completed receipt state, operation ID and result, plus conversation and workspace recovery controls. Large native output and receipt request fields cross the connection without being used. **Proposed behavior** Select only those result fields in PostgreSQL. Keep typed policy and recovery values. Keep all eligible completed receipts and their complete result values. Retain missing versus null receipt results, run and receipt order, company/task/agent scope, and the source-run exception. Keep the full retained history without a new limit. **Reason and benefit** The synthetic wide-output fixture removes 99.879% of this query's DataRow bytes. A required 2 MiB receipt-result control saves only 0.0074%, because that complete result is needed. Actual savings depend on result contents; the wide-output result is not a forecast for all deployments. **Breaking changes** None intended. No schema, API, migration, workflow, history limit, or receipt truncation changes. **Additional context** Related: #17481 projects a separate recovery read, and #17896 narrows explicit continuation admission. This PR covers the continuation history query and has no dependency on the other byte reduction PRs. ## What Changed - Project completed tool receipts to their state, operation ID and complete result. Omit fields that the continuation does not read. - Project typed conversation/workspace controls and the recovery schema. Keep the shared policy helper unchanged. - Add real PostgreSQL comparisons for complete continuation envelopes, ordering, scope, legacy JSON, recovery flags, and receipt result preservation. - Add external isolated database support to the existing continuation tests and document a reproducible query benchmark. ## Verification - Pinned baseline: `3cd3af5d75f200185ed4d9703a3145594a489b13`. - PostgreSQL 17.11 with synthetic local data. The harness captures SQL from the real baseline and treatment services. Imported helpers remain unchanged. - Command: `pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-continuation-history.ts --baseline-ref=3cd3af5d75f200185ed4d9703a3145594a489b13`. Set `PAPERCLIP_CONTINUATION_HISTORY_BENCHMARK_DATABASE_URL` to a disposable database ending in `_test` to use an external server. The default uses disposable embedded PostgreSQL. - The baseline was measured before the production edit with `--baseline-only`. The treatment compares complete envelopes, every retained scalar column, completed receipts, and shared policy decisions. The wide unused-output fixture must remove more than 90% of DataRow bytes or the benchmark fails. - Wide unused-output fixture: 25 runs with independent incompressible 192 KiB native output and 32 KiB unused receipt request fields. DataRow bytes: 5,744,950 to 6,925 (99.879% less). - Required-result control: one complete 2 MiB receipt result. DataRow bytes: 2,097,556 to 2,097,401 (0.0074% less). The whole result is checked for equality. - Exact byte boundary: PostgreSQL text-format DataRow frames, including field length prefixes. Excludes RowDescription, command metadata, TLS, TCP, and other service queries. - Seven measured query round trips per variant follow two warmups and alternate order. Final-head wide-output median: 32.86 to 4.41 ms (ranges 26.24–35.54 and 3.62–5.21 ms). Required-result median: 17.01 to 20.60 ms (ranges 16.30–32.89 and 18.85–22.48 ms). The required-result control shows projection overhead and overlapping timing ranges; it has no latency improvement claim. These are local measurements, not deployed latency results. - `PAPERCLIP_EXECUTION_CONTINUATION_TEST_DATABASE_URL=<isolated_test_database> pnpm exec vitest run server/src/services/execution-continuation.test.ts server/src/services/execution-continuation-history-read.test.ts --maxWorkers=1`: 55 tests pass in PostgreSQL 17. - Cases cover numeric receipt key order, empty operation IDs, malformed receipt entries, missing versus null results, complete large results, malformed root JSON, all closed workspace restore failure codes, invalid recovery schemas, cancelled admission rows, and company/task/agent/source-run boundaries. - `pnpm -r typecheck` and `pnpm build` pass on final head `d665d9f5c47b237c81b9cc80fdec1d39ad90754f`. The final-head focused PostgreSQL run passes all 55 tests. The final benchmark records that SHA, an empty source diff, and passing equivalence/byte assertions. The first typecheck found optional array accesses in new test assertions; explicit required-array assertions fixed them. Local full `pnpm test:run` is not claimed because the host embedded PostgreSQL runtime is unavailable; the full test matrix runs in CI. Exact-head Greptile review completed success at 5/5 with no actionable findings or unresolved threads. The full final-head CI matrix passes. All 58 current-head checks are terminal: 54 success, two neutral, and two optional Storybook skips. One original server shard ended with a canceled test step and no assertion failure; its single targeted retry passed. No product change was needed for that interruption. ## Risks - Future continuation consumers must add any newly required result field to this projection. Existing shared policy semantics stay in the helper. - Completed receipt results can still be large and can dominate transfer. This PR preserves them without a cap. The receipt-heavy control adds JSON projection work for only 155 bytes of savings, so it can be slower despite equal output. - JSON malformed-value and property-presence rules affect action evidence and recovery. Real PostgreSQL comparisons check those decisions and complete output. - Production transfer and latency improvements need a fresh deployed capture. Nothing was merged or deployed by this work. ## Model Used OpenAI GPT-6 through Codex. The exact runtime model ID and context window are not exposed by this session. The work used repository tools, code execution, and real PostgreSQL validation. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
77295faebc
commit
56bc470ab2
5 files changed
+291
-4
No files matched your search
@@ -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=<pre-fix-commit>`
|
||||
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`
|
||||
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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<string, unknown> => value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {};
|
||||
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<Record<string, unknown>> | undefined;
|
||||
const recorded = new Proxy(db!, { get(target, key) {
|
||||
if (key !== "select") return Reflect.get(target, key);
|
||||
return (...fields: Parameters<Db["select"]>) => {
|
||||
const builder = target.select(...fields), from = builder.from.bind(builder);
|
||||
builder.from = ((...args: Parameters<typeof from>) => {
|
||||
const selection = from(...args), execute = selection.execute.bind(selection);
|
||||
if (args[0] === heartbeatRuns) selection.execute = async (...executeArgs: Parameters<typeof execute>) => {
|
||||
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<string, unknown>), hasConversationContinuationPolicy(oldRow.result as Record<string, unknown>));
|
||||
}
|
||||
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); }
|
||||
}
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
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<string, unknown> | null }).returning();
|
||||
return { companyId, agentId, issueId, run };
|
||||
}
|
||||
|
||||
async function invoke(f: Awaited<ReturnType<typeof fixture>>, fullResult: boolean, context: Record<string, unknown> = {}) {
|
||||
const reads: Array<Array<Record<string, unknown>>> = [];
|
||||
const recorded = new Proxy(db, { get(target, key) {
|
||||
if (key !== "select") return Reflect.get(target, key);
|
||||
return (...fields: Parameters<Db["select"]>) => {
|
||||
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<typeof from>) => {
|
||||
const query = from(...args), execute = query.execute.bind(query);
|
||||
if (historyRead) query.execute = async (...executeArgs: Parameters<typeof execute>) => {
|
||||
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<ReturnType<typeof fixture>>, context: Record<string, unknown> = {}) {
|
||||
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();
|
||||
});
|
||||
});
|
||||
@@ -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<unknown>) =>
|
||||
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({
|
||||
|
||||
@@ -31,6 +31,28 @@ const object = (v: unknown): Record<string, unknown> =>
|
||||
: {};
|
||||
const string = (v: unknown) =>
|
||||
typeof v === "string" && v.length > 0 ? v : null;
|
||||
const historyResult = sql<Record<string, unknown> | 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(
|
||||
|
||||
Reference in new issue
Block a user