mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-28 12:04:41 +01:00
perf(server): narrow native session resume database reads (#17934)
<!-- entire-trail-link-start --> https://entire.com/gh/paperclipai/-/trails/94/changes/paperclip/94 <!-- entire-trail-link-end --> ## Thinking Path > - Paperclip manages AI agents and their work. > - Native agents resume prior provider sessions from saved run checkpoints. > - Resume selection must respect every newer progress barrier. > - The prior query returned large context and profile JSON for up to 32 candidates. > - Most candidates need only process, status, binding, and checkpoint-presence checks. > - This PR selects the same source in one SQL snapshot and returns only its required payloads. > - This reduces database response bytes while preserving session authority and saved task context. ## Linked Issues or Issue Description **What existing behavior does this improve?** Native task/session resume selection in `findNativeSessionResumeRun`. **Current behavior** At baseline `bf3ef12f3a78d0b15a32ad57f7e16520fbf6cd5e`, a resume lookup reads complete context and profile JSON for up to 32 runs, then reads provider-event evidence. A synthetic 15-run history returns about 4.44 MB even though one checkpoint is selected. **Proposed behavior** A bounded CTE finds the first checkpoint or progress barrier. The same SQL statement returns the selected source's native input, checkpoint, tool fingerprint, wake envelope, canonical task brief, and process identity. All fallback barriers and the 32-run limit stay in place. **Reason and benefit** Rejected candidates no longer transfer saved prompts and execution inputs to Node. A single snapshot also prevents a gap between candidate selection and payload hydration. **Breaking changes** None to public APIs or stored history. This private service returns the fields consumed by runtime selection and checkpoint rebinding. It keeps complete required nested values and distinguishes missing brief fields from explicit JSON nulls. **Additional context** Checked `ROADMAP.md`. This is a targeted improvement to existing resume behavior. Public searches for native session resume projections found no duplicate. Related work includes #6573 (durable agent sessions) and #17011 (managed Claude resume opt-in); neither changes this database read. Provider-refusal recovery in #17676 is a separate concern and is unchanged. ## What Changed - Select the newest progress barrier within the existing exact company/agent/task/session and 32-run window in one SQL snapshot. - Return three required profile fields, the full wake envelope, and the saved task's ID, title, and description only for the selected source. - Preserve invalid non-null checkpoint barriers, provider evidence, process authority, terminal-status rules, timestamp precision, and exact string binding checks. - Add real PostgreSQL regression coverage and a reproducible baseline/treatment benchmark. Document the read and benchmark command. ## Verification - `pnpm -r typecheck` passed. `pnpm build` passed. The final server `tsc --noEmit` also passed. The later benchmark-only relocation passed `pnpm check:module-boundaries` and the real PostgreSQL proof; runtime and regression-test source are unchanged. - Related suites: `native-session-resume-projection.test.ts`, `native-session-resume.test.ts`, `heartbeat/retries.test.ts`, and `heartbeat-retry-scheduling.test.ts`: 78 passed and 54 skipped across 3 passed files and 1 skipped file. Embedded PostgreSQL/platform gates caused the skips. Resume orchestration and the 10 new SQL tests ran against isolated PostgreSQL 17.11. The final 10-test projection rerun passed. - New SQL cases cover all seven provider evidence kinds, each process authority field, all terminal statuses, wrong bindings and runtime, invalid checkpoints, null/scalar/array/object JSON, missing versus null brief fields, the 32-row bound, ordering ties, future barriers, submillisecond timestamps, and a change committed before the one selection statement. - `pnpm test:run` was attempted. The host's embedded PostgreSQL startup failed after five attempts. The broad run recorded failures/skips in embedded-database suites and was stopped after that concrete host limitation (exit 143). It is not a local full-suite pass. The full final-head CI matrix passes on `dcde4e3dac2173321d7be2406e7775af8411826a`: 55 successful checks, one neutral check, and two optional Storybook skips. Greptile completed at 5/5 on that exact head, with zero unresolved findings. The initial benchmark-location policy failure was fixed before this final run. - Reproduce the benchmark with `pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-native-session-resume-reads.ts --baseline-ref=bf3ef12f3a78d0b15a32ad57f7e16520fbf6cd5e`. The optional `PAPERCLIP_SESSION_RESUME_BENCHMARK_DATABASE_URL` must name an isolated database ending in `_test`. - The harness imports the pinned baseline module, invokes both actual services, captures complete generated SQL and parameters, compares every retained consumer field, and captures `EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON)`. Its incompressible wide fixture keeps a complete 96 KiB native input in the selected result. Raw PostgreSQL text DataRow frames are measured exactly; RowDescription, protocol command metadata, TLS, and TCP are excluded. | Synthetic local case | Baseline DataRow bytes | Treatment DataRow bytes | Reduction | | --- | ---: | ---: | ---: | | 15 runs, one selected checkpoint | 4,441,143 | 104,599 | 97.64% | | 15 unused bootstraps | 4,435,590 | 0 | 100% | | One small selected source | 6,395 | 6,319 | 1.19% | The committed-head benchmark uses two warmups and seven alternating samples per variant. Local round-trip medians and ranges were: | Synthetic local case | Baseline median (range), ms | Treatment median (range), ms | | --- | ---: | ---: | | Wide history | 20.32 (18.63–24.37) | 1.63 (1.40–2.21) | | Unused history | 18.52 (17.63–20.35) | 1.20 (1.13–1.56) | | Small selected source | 1.05 (0.95–1.70) | 0.78 (0.73–1.30) | The small case has overlapping ranges and a 1.19% byte change, so it is neutral. `EXPLAIN` server execution time was higher for the treatment (wide: 0.402 ms versus 0.028 ms across the two baseline statements). `EXPLAIN` does not serialize result rows; the new JSON presence checks do extra server work while avoiding large result transfer. These are local synthetic round-trip measurements, not production latency claims. The final proof ran on `dcde4e3dac2173321d7be2406e7775af8411826a` with no runtime source diff. ## Risks - Selection policy now has a SQL form. Regression tests preserve the pure selector's barrier semantics, including invalid non-null checkpoints and JavaScript Date precision. - Checkpoint-presence checks can still detoast candidate profiles inside PostgreSQL. The change reduces returned bytes; it does not eliminate all database JSON work. - The selected checkpoint, native input, and wake envelope can still be large because consumers require them in full. No new payload cap is introduced. - There is no migration, cache, background polling, ownership lock change, deployment, or merge in this PR. ## Model Used OpenAI GPT-6 through Codex. The host did not expose an exact runtime model ID or context-window size. The agent used reasoning, code execution, real PostgreSQL, and GitHub tools. ## 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:
DottaandPaperclip authored and GitHub committed
1 parent
3d07e06be7
commit
dc8cbdb819
4 files changed
+429
-32
No files matched your search
@@ -49,6 +49,18 @@ and leaves all existing cases passive. Source identities are not grants.
|
||||
|
||||
### Task query performance
|
||||
|
||||
Native session resume selects its newest progress barrier within the existing
|
||||
32-run task/session window in one SQL snapshot. Only the chosen checkpoint source
|
||||
returns its saved native input, checkpoint, tool fingerprint, wake envelope, and
|
||||
canonical task brief. Unused bootstrap candidates do not return full context or
|
||||
profiles. Wrong runtime, nonterminal status, process identity, provider events,
|
||||
invalid but non-null checkpoints, and future checkpoint barriers retain their
|
||||
existing selection rules. Stored run payloads and history reads stay complete.
|
||||
Run the synthetic PostgreSQL comparison with a pinned pre-change revision:
|
||||
`pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-native-session-resume-reads.ts --baseline-ref=<commit>`.
|
||||
An optional `PAPERCLIP_SESSION_RESUME_BENCHMARK_DATABASE_URL` must name a dedicated
|
||||
test database ending in `_test`; otherwise the harness starts disposable PostgreSQL.
|
||||
|
||||
Agent request authentication uses a compact run-identity read. It omits saved
|
||||
execution payloads, results, and log excerpts from the query response. The read
|
||||
still locks the task before the run and reconciles pending steering from durable
|
||||
|
||||
@@ -0,0 +1,127 @@
|
||||
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 { findNativeSessionResumeRun } from "../server/src/services/native-runtime/native-session-resume.js";
|
||||
import { resolveExternalTestDatabaseUrl } from "../server/src/__tests__/helpers/external-test-database.js";
|
||||
|
||||
type Query = { sql: string; params: unknown[] };
|
||||
const root = fileURLToPath(new URL("../", import.meta.url));
|
||||
const baselineRef = process.argv.find(arg => arg.startsWith("--baseline-ref="))?.slice(15);
|
||||
assert.ok(baselineRef, "Pass a pinned pre-fix --baseline-ref");
|
||||
const baselineRevision = execFileSync("git", ["rev-parse", "--verify", `${baselineRef}^{commit}`], { cwd: root, encoding: "utf8" }).trim();
|
||||
const sourcePath = "server/src/services/native-runtime/native-session-resume.ts";
|
||||
const treatmentRevision = execFileSync("git", ["rev-parse", "HEAD"], { cwd: root, encoding: "utf8" }).trim();
|
||||
const treatmentDiff = execFileSync("git", ["diff", "HEAD", "--", sourcePath], { cwd: root, encoding: "utf8" });
|
||||
const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_SESSION_RESUME_BENCHMARK_DATABASE_URL");
|
||||
const database = externalUrl ? { connectionString: externalUrl, cleanup: async () => {} } : await startEmbeddedPostgresTestDatabase("native-session-resume-benchmark-");
|
||||
const db = createDb(database.connectionString);
|
||||
const baselineDir = await fs.mkdtemp(path.join(root, "server/scripts/.session-resume-baseline-"));
|
||||
const companyIds: string[] = [];
|
||||
try {
|
||||
if (externalUrl) await applyPendingMigrations(externalUrl);
|
||||
const moduleUrl = new URL(`../${sourcePath}`, import.meta.url);
|
||||
const source = execFileSync("git", ["show", `${baselineRevision}:${sourcePath}`], { cwd: root, encoding: "utf8" });
|
||||
await fs.writeFile(path.join(baselineDir, "resume.ts"), source.replace(/from "(\.{1,2}\/[^\"]+)"/g,
|
||||
(_match, specifier: string) => `from ${JSON.stringify(fileURLToPath(new URL(specifier, moduleUrl)))}`));
|
||||
const baseline = await import(pathToFileURL(path.join(baselineDir, "resume.ts")).href) as { findNativeSessionResumeRun: typeof findNativeSessionResumeRun };
|
||||
async function capture(invoke: typeof findNativeSessionResumeRun, input: Parameters<typeof findNativeSessionResumeRun>[1]) {
|
||||
const queries: Query[] = [];
|
||||
const wrap = (target: Db): Db => new Proxy(target, { get(target, key) {
|
||||
if (key === "with") return (...args: Parameters<Db["with"]>) => wrap(target.with(...args) as Db);
|
||||
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 query = from(...args), execute = query.execute.bind(query);
|
||||
query.execute = async (...executeArgs: Parameters<typeof execute>) => {
|
||||
queries.push(query.toSQL());
|
||||
return execute(...executeArgs);
|
||||
};
|
||||
return query;
|
||||
}) as typeof builder.from;
|
||||
return builder;
|
||||
};
|
||||
}});
|
||||
return { result: await invoke(wrap(db), input), queries };
|
||||
}
|
||||
const median = (values: number[]) => [...values].sort((a, b) => a - b)[Math.floor(values.length / 2)];
|
||||
const measurements = [];
|
||||
for (const scenario of ["wide_history", "unused_history", "small_selected"] as const) {
|
||||
const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), normalizedSessionId = randomUUID();
|
||||
companyIds.push(companyId);
|
||||
await db.insert(companies).values({ id: companyId, name: "Synthetic resume benchmark", issuePrefix: `B${companyId.replaceAll("-", "")}` });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Synthetic runner", adapterType: "paperclip_runner" });
|
||||
await db.insert(issues).values({ id: issueId, companyId, title: "Retain canonical task" });
|
||||
const runCount = scenario === "small_selected" ? 1 : 15;
|
||||
const prompt = randomBytes(scenario === "small_selected" ? 16 : 144 * 1024).toString("base64");
|
||||
const profilePadding = randomBytes(scenario === "small_selected" ? 16 : 72 * 1024).toString("base64");
|
||||
const neededInput = { task: { title: "Prior task", description: "Canonical description", prompt: profilePadding }, runtimeContext: { instructions: "Keep complete input" } };
|
||||
const checkpoint = { sessionId: normalizedSessionId, state: randomBytes(4096).toString("base64") };
|
||||
await db.insert(heartbeatRuns).values(Array.from({ length: runCount }, (_, n) => ({
|
||||
id: randomUUID(), companyId, agentId, nativeIssueId: issueId, nativeSessionId: normalizedSessionId,
|
||||
runtimeMode: "native", status: "failed", createdAt: new Date(Date.UTC(2026, 8, 8, 0, 0, n)),
|
||||
runnerProfileJson: { padding: "Unused profile metadata", nativeExecutionInput: neededInput, nativeToolContractFingerprint: "fixture",
|
||||
...(scenario !== "unused_history" && n === 0 ? { sessionCheckpoint: checkpoint } : {}) },
|
||||
contextSnapshot: { prompt, paperclipWake: { externalChatProvider: "slack", githubReviewConfigurations: [{ key: "review" }] },
|
||||
paperclipIssue: { id: issueId, title: "Prior canonical title", description: "Prior canonical description" } },
|
||||
})));
|
||||
await db.$client.unsafe("ANALYZE heartbeat_runs");
|
||||
const input = { companyId, agentId, issueId, normalizedSessionId, currentRunId: randomUUID(), beforeCreatedAt: new Date("2026-09-09T00:00:00Z") };
|
||||
const before = await capture(baseline.findNativeSessionResumeRun, input), after = await capture(findNativeSessionResumeRun, input);
|
||||
const contract = (run: typeof after.result) => run && ({ ...run,
|
||||
runnerProfileJson: { nativeExecutionInput: run.runnerProfileJson?.nativeExecutionInput ?? null,
|
||||
nativeToolContractFingerprint: run.runnerProfileJson?.nativeToolContractFingerprint ?? null,
|
||||
sessionCheckpoint: run.runnerProfileJson?.sessionCheckpoint ?? null },
|
||||
contextSnapshot: { paperclipWake: run.contextSnapshot?.paperclipWake ?? null, paperclipIssue: run.contextSnapshot?.paperclipIssue ?? null } });
|
||||
assert.deepEqual(contract(after.result), contract(before.result));
|
||||
const variants = [before.queries, after.queries], samples: number[][] = [[], []], dataRowBytes = [0, 0], resultRows = [0, 0];
|
||||
for (let iteration = 0; iteration < 9; iteration++) {
|
||||
for (const variant of iteration % 2 ? [1, 0] : [0, 1]) {
|
||||
const start = performance.now();
|
||||
let bytes = 0, count = 0;
|
||||
for (const query of variants[variant]) {
|
||||
const rows = await db.$client.unsafe(query.sql, query.params as never[]).raw();
|
||||
count += rows.length;
|
||||
bytes += rows.reduce((total, row) => total + 7 + row.reduce((size: number, value: Buffer | null) => size + 4 + (value?.length ?? 0), 0), 0);
|
||||
}
|
||||
if (iteration >= 2) samples[variant].push(performance.now() - start);
|
||||
dataRowBytes[variant] = bytes; resultRows[variant] = count;
|
||||
}
|
||||
}
|
||||
const plans = [];
|
||||
for (const queries of variants) {
|
||||
const variantPlans = [];
|
||||
for (const query of queries) {
|
||||
const [plan] = await db.$client.unsafe(`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${query.sql}`, query.params as never[]);
|
||||
variantPlans.push(plan["QUERY PLAN"]);
|
||||
}
|
||||
plans.push(variantPlans);
|
||||
}
|
||||
if (!process.argv.includes("--baseline-only") && scenario !== "small_selected") assert.ok(dataRowBytes[1] < dataRowBytes[0] / 10);
|
||||
measurements.push({ scenario, runCount, equalConsumerFields: true, baselineQueryCount: before.queries.length, treatmentQueryCount: after.queries.length,
|
||||
dataRowBytes, resultRows, baselineMedianMs: median(samples[0]), treatmentMedianMs: median(samples[1]), samples, queries: variants, plans });
|
||||
}
|
||||
console.log(JSON.stringify({ baselineRevision, treatmentRevision, treatmentDiff,
|
||||
acceptance: "Equal selected consumer fields; wide and unused histories remove at least 90% of DataRow bytes; no timing threshold",
|
||||
byteBoundary: "Exact text-format PostgreSQL DataRow frames (7 bytes framing + 4 bytes length per column + raw field bytes); excludes RowDescription, command metadata, TLS and TCP",
|
||||
postgres: await db.$client.unsafe("select version()"), measurements }, null, 2));
|
||||
} finally {
|
||||
try {
|
||||
for (const companyId of companyIds) {
|
||||
await db.$client.unsafe("DELETE FROM heartbeat_runs WHERE company_id = $1::uuid", [companyId]);
|
||||
await db.$client.unsafe("DELETE FROM issues WHERE company_id = $1::uuid", [companyId]);
|
||||
await db.$client.unsafe("DELETE FROM agents WHERE company_id = $1::uuid", [companyId]);
|
||||
await db.$client.unsafe("DELETE FROM companies WHERE id = $1::uuid", [companyId]);
|
||||
}
|
||||
} finally {
|
||||
try { await db.$client.end({ timeout: 0 }); }
|
||||
finally {
|
||||
try { await database.cleanup(); }
|
||||
finally { await fs.rm(baselineDir, { recursive: true, force: true }); }
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,203 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { agents, companies, createDb, heartbeatRunEvents, heartbeatRuns, issues, type Db } from "@paperclipai/db";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
|
||||
import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js";
|
||||
import { findNativeSessionResumeRun } from "./native-session-resume.js";
|
||||
|
||||
const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_TEST_DATABASE_URL");
|
||||
const supported = Boolean(externalUrl) || (await getEmbeddedPostgresTestSupport()).supported;
|
||||
|
||||
(supported ? describe : describe.skip)("native session resume SQL projection", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: Db;
|
||||
const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), normalizedSessionId = randomUUID();
|
||||
const currentRunId = randomUUID();
|
||||
const at = (second: number) => new Date(Date.UTC(2026, 8, 8, 0, 0, second));
|
||||
const input = { companyId, agentId, issueId, normalizedSessionId, currentRunId, beforeCreatedAt: at(30) };
|
||||
const checkpoint = { sessionId: "provider-session", identity: { runId: "prior" }, complete: [false, 0, "", null] };
|
||||
const nativeExecutionInput = { binding: { issueId }, task: { prompt: "Complete native input", description: "Description" }, runtimeContext: { nested: [1, 2] } };
|
||||
const paperclipWake = { externalChatProvider: "slack", githubReviewConfigurations: [{ key: "configuration", nested: { retained: true } }], other: ["preserved"] };
|
||||
const paperclipIssue = { id: issueId, title: "Saved canonical title", description: "Saved canonical description" };
|
||||
const profile = { nativeExecutionInput, sessionCheckpoint: checkpoint, nativeToolContractFingerprint: "fingerprint", unrelated: "Omit" };
|
||||
const run = (second: number, overrides: Partial<typeof heartbeatRuns.$inferInsert> = {}) => ({
|
||||
id: randomUUID(), companyId, agentId, nativeIssueId: issueId, nativeSessionId: normalizedSessionId,
|
||||
runtimeMode: "native", status: "failed", createdAt: at(second), runnerProfileJson: {}, ...overrides,
|
||||
});
|
||||
const select = () => findNativeSessionResumeRun(db, input);
|
||||
const source = () => run(0, { runnerInstanceId: randomUUID(), processPid: 17, processGroupId: 19, processStartedAt: at(0),
|
||||
runnerProfileJson: profile, contextSnapshot: { paperclipWake, paperclipIssue: { ...paperclipIssue, unrelated: "Omit" }, prompt: "Omit" } });
|
||||
const writeProfile = (id: string, value: unknown) => db.execute(sql`update ${heartbeatRuns}
|
||||
set runner_profile_json = ${JSON.stringify(value)}::jsonb where ${heartbeatRuns.id} = ${id}`);
|
||||
|
||||
beforeAll(async () => {
|
||||
database = externalUrl ? { connectionString: externalUrl, cleanup: async () => {} } : await startEmbeddedPostgresTestDatabase("native-resume-projection-");
|
||||
db = createDb(database.connectionString);
|
||||
await db.insert(companies).values({ id: companyId, name: "Resume projection", issuePrefix: `N${companyId.replaceAll("-", "")}` });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Runner" });
|
||||
await db.insert(issues).values({ id: issueId, companyId, title: "Resume task" });
|
||||
});
|
||||
afterEach(async () => {
|
||||
await db.delete(heartbeatRunEvents).where(eq(heartbeatRunEvents.companyId, companyId));
|
||||
await db.delete(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId));
|
||||
});
|
||||
afterAll(async () => {
|
||||
if (!db) return;
|
||||
try {
|
||||
await db.delete(issues).where(eq(issues.companyId, companyId));
|
||||
await db.delete(agents).where(eq(agents.companyId, companyId));
|
||||
await db.delete(companies).where(eq(companies.id, companyId));
|
||||
} finally {
|
||||
try { await db.$client.end({ timeout: 0 }); }
|
||||
finally { await database.cleanup(); }
|
||||
}
|
||||
});
|
||||
|
||||
it("retains full required input, checkpoint, wake, canonical brief and process identity only for the selected source", async () => {
|
||||
const saved = source();
|
||||
await db.insert(heartbeatRuns).values([saved, run(1), run(2)]);
|
||||
const result = await select();
|
||||
expect(result).toEqual({ ...saved, runnerProfileJson: { nativeExecutionInput, sessionCheckpoint: checkpoint, nativeToolContractFingerprint: "fingerprint" },
|
||||
contextSnapshot: { paperclipWake, paperclipIssue } });
|
||||
const stored = await db.select({ profile: heartbeatRuns.runnerProfileJson, context: heartbeatRuns.contextSnapshot }).from(heartbeatRuns).where(eq(heartbeatRuns.id, saved.id));
|
||||
expect(stored[0].profile).toEqual(profile);
|
||||
expect(stored[0].context).toEqual(saved.contextSnapshot);
|
||||
for (const status of ["succeeded", "interrupted", "cancelled", "timed_out"]) {
|
||||
await db.update(heartbeatRuns).set({ status }).where(eq(heartbeatRuns.id, saved.id));
|
||||
expect((await select())?.id).toBe(saved.id);
|
||||
}
|
||||
});
|
||||
|
||||
it("treats every non-null checkpoint as a barrier, including invalid false, zero, empty string, array and object values", async () => {
|
||||
const saved = source(), newer = run(1);
|
||||
await db.insert(heartbeatRuns).values([saved, newer]);
|
||||
for (const value of [false, 0, "", "invalid", [], {}]) {
|
||||
await writeProfile(newer.id, { sessionCheckpoint: value });
|
||||
const selected = await select();
|
||||
expect(selected?.id).toBe(newer.id);
|
||||
expect(selected?.runnerProfileJson.sessionCheckpoint).toEqual(value);
|
||||
}
|
||||
});
|
||||
|
||||
it("skips only unused SQL-null, JSON-null, scalar, array and checkpoint-null profiles", async () => {
|
||||
const saved = source(), newer = run(1, { runnerProfileJson: null });
|
||||
await db.insert(heartbeatRuns).values([saved, newer]);
|
||||
expect((await select())?.id).toBe(saved.id);
|
||||
for (const value of [null, false, 0, "sessionCheckpoint", [], ["sessionCheckpoint"], {}, { sessionCheckpoint: null }]) {
|
||||
await writeProfile(newer.id, value);
|
||||
expect((await select())?.id).toBe(saved.id);
|
||||
}
|
||||
});
|
||||
|
||||
it("preserves malformed selected context values and missing brief fields", async () => {
|
||||
const saved = source();
|
||||
await db.insert(heartbeatRuns).values(saved);
|
||||
for (const value of [null, false, 0, "invalid", [], {}, { paperclipWake: [], paperclipIssue: { id: issueId, title: false } },
|
||||
{ paperclipIssue: { id: issueId, title: "Valid title without description" } },
|
||||
{ paperclipIssue: { id: issueId, title: "Valid title", description: null } }]) {
|
||||
await db.execute(sql`update ${heartbeatRuns} set context_snapshot = ${JSON.stringify(value)}::jsonb where ${heartbeatRuns.id} = ${saved.id}`);
|
||||
const result = await select();
|
||||
const object = typeof value === "object" && value !== null && !Array.isArray(value) ? value : {};
|
||||
expect(result?.contextSnapshot.paperclipWake).toEqual("paperclipWake" in object ? object.paperclipWake : null);
|
||||
expect(result?.contextSnapshot.paperclipIssue).toEqual("paperclipIssue" in object ? object.paperclipIssue : null);
|
||||
}
|
||||
});
|
||||
|
||||
it("stops at wrong runtime, nonterminal status and each process authority field", async () => {
|
||||
await db.insert(heartbeatRuns).values(source());
|
||||
for (const override of [{ runtimeMode: "legacy" }, { status: "running" }, { processPid: 0 }, { processGroupId: 0 }, { processStartedAt: at(0) }]) {
|
||||
const newer = run(1, override);
|
||||
await db.insert(heartbeatRuns).values(newer);
|
||||
expect(await select()).toBeNull();
|
||||
await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, newer.id));
|
||||
}
|
||||
});
|
||||
|
||||
it("checks every provider evidence event and ignores unrelated event kinds", async () => {
|
||||
const saved = source(), newer = run(1);
|
||||
await db.insert(heartbeatRuns).values([saved, newer]);
|
||||
for (const eventType of ["harness.ready", "session.started", "session.resumed", "session.updated", "turn.started", "provider.event", "provider.rpc_result", "log"]) {
|
||||
await db.insert(heartbeatRunEvents).values({ companyId, agentId, runId: newer.id, seq: 1, eventType });
|
||||
if (eventType === "log") expect((await select())?.id).toBe(saved.id);
|
||||
else expect(await select()).toBeNull();
|
||||
await db.delete(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, newer.id));
|
||||
}
|
||||
});
|
||||
|
||||
it("excludes current and mismatched agent, company, task and session rows before applying barriers", async () => {
|
||||
const saved = source(), otherCompany = randomUUID(), otherAgent = randomUUID(), otherIssue = randomUUID();
|
||||
await db.insert(companies).values({ id: otherCompany, name: "Other", issuePrefix: `D${otherCompany.replaceAll("-", "")}` });
|
||||
await db.insert(agents).values({ id: otherAgent, companyId: otherCompany, name: "Other" });
|
||||
await db.insert(issues).values({ id: otherIssue, companyId, title: "Other" });
|
||||
try {
|
||||
await db.insert(heartbeatRuns).values([saved, run(1, { id: currentRunId, status: "running" }),
|
||||
run(2, { companyId: otherCompany, status: "running" }), run(3, { agentId: otherAgent, status: "running" }),
|
||||
run(4, { nativeIssueId: otherIssue, status: "running" }), run(5, { nativeSessionId: randomUUID(), status: "running" })]);
|
||||
expect((await select())?.id).toBe(saved.id);
|
||||
} finally {
|
||||
await db.delete(heartbeatRuns).where(eq(heartbeatRuns.companyId, otherCompany));
|
||||
await db.delete(heartbeatRuns).where(eq(heartbeatRuns.agentId, otherAgent));
|
||||
await db.delete(issues).where(eq(issues.id, otherIssue));
|
||||
await db.delete(agents).where(eq(agents.id, otherAgent));
|
||||
await db.delete(companies).where(eq(companies.id, otherCompany));
|
||||
}
|
||||
expect(await findNativeSessionResumeRun(db, { ...input, normalizedSessionId: "invalid" })).toBeNull();
|
||||
for (const key of ["companyId", "agentId", "issueId", "normalizedSessionId"] as const) {
|
||||
expect(await findNativeSessionResumeRun(db, { ...input, [key]: input[key].toUpperCase() })).toBeNull();
|
||||
}
|
||||
});
|
||||
|
||||
it("retains the 32-row bound and deterministic created-at/id ordering", async () => {
|
||||
const saved = source(), unused = Array.from({ length: 32 }, () => run(1));
|
||||
await db.insert(heartbeatRuns).values([saved, ...unused]);
|
||||
expect(await select()).toBeNull();
|
||||
await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, unused[0].id));
|
||||
expect((await select())?.id).toBe(saved.id);
|
||||
const tied = [run(2, { runnerProfileJson: { sessionCheckpoint: false } }), run(2, { status: "running" })].sort((a, b) => b.id.localeCompare(a.id));
|
||||
await db.insert(heartbeatRuns).values(tied);
|
||||
if (tied[0].status === "running") expect(await select()).toBeNull();
|
||||
else expect((await select())?.id).toBe(tied[0].id);
|
||||
});
|
||||
|
||||
it("does not skip future checkpoint barriers and preserves Date millisecond comparison", async () => {
|
||||
const saved = source(), future = run(31);
|
||||
await db.insert(heartbeatRuns).values([saved, future]);
|
||||
expect((await select())?.id).toBe(saved.id);
|
||||
await writeProfile(future.id, { sessionCheckpoint: false });
|
||||
expect(await select()).toBeNull();
|
||||
await db.execute(sql`update ${heartbeatRuns} set created_at = '2026-09-08T00:00:30.000999Z'::timestamptz where ${heartbeatRuns.id} = ${future.id}`);
|
||||
expect((await select())?.id).toBe(future.id);
|
||||
});
|
||||
|
||||
it("sees a newly committed barrier and changed source binding in its single selection snapshot", async () => {
|
||||
const saved = source();
|
||||
await db.insert(heartbeatRuns).values(saved);
|
||||
let reads = 0;
|
||||
const changing = new Proxy(db, { get(target, key) {
|
||||
if (key !== "with") return Reflect.get(target, key);
|
||||
return (...ctes: Parameters<Db["with"]>) => {
|
||||
const builder = target.with(...ctes), original = builder.select.bind(builder);
|
||||
builder.select = ((...fields: Parameters<typeof original>) => {
|
||||
const selection = original(...fields), from = selection.from.bind(selection);
|
||||
selection.from = ((...args: Parameters<typeof from>) => {
|
||||
const query = from(...args), execute = query.execute.bind(query);
|
||||
query.execute = async (...executeArgs: Parameters<typeof execute>) => {
|
||||
reads++;
|
||||
await db.insert(heartbeatRuns).values(run(1, { runnerProfileJson: { sessionCheckpoint: 0 } }));
|
||||
await db.update(heartbeatRuns).set({ nativeSessionId: randomUUID() }).where(and(eq(heartbeatRuns.id, saved.id), eq(heartbeatRuns.companyId, companyId)));
|
||||
return execute(...executeArgs);
|
||||
};
|
||||
return query;
|
||||
}) as typeof selection.from;
|
||||
return selection;
|
||||
}) as typeof builder.select;
|
||||
return builder;
|
||||
};
|
||||
}});
|
||||
const selected = await findNativeSessionResumeRun(changing, input);
|
||||
expect(reads).toBe(1);
|
||||
expect(selected?.id).not.toBe(saved.id);
|
||||
expect(selected?.runnerProfileJson.sessionCheckpoint).toBe(0);
|
||||
});
|
||||
});
|
||||
@@ -1,6 +1,6 @@
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { isDeepStrictEqual } from "node:util";
|
||||
import { and, desc, eq, inArray, ne } from "drizzle-orm";
|
||||
import { and, desc, eq, exists, inArray, isNotNull, ne, not, or, sql } from "drizzle-orm";
|
||||
import { heartbeatRunEvents, heartbeatRuns, type Db } from "@paperclipai/db";
|
||||
import type {
|
||||
NativeExecutionInput,
|
||||
@@ -186,6 +186,16 @@ export function selectNativeSessionResumeRun<
|
||||
return null;
|
||||
}
|
||||
|
||||
const NATIVE_SESSION_PROVIDER_EVENT_TYPES = [
|
||||
"harness.ready",
|
||||
"session.started",
|
||||
"session.resumed",
|
||||
"session.updated",
|
||||
"turn.started",
|
||||
"provider.event",
|
||||
"provider.rpc_result",
|
||||
];
|
||||
|
||||
export async function nativeSessionProviderEvidence(
|
||||
db: Pick<Db, "select">,
|
||||
runIds: string[],
|
||||
@@ -197,15 +207,7 @@ export async function nativeSessionProviderEvidence(
|
||||
.where(
|
||||
and(
|
||||
inArray(heartbeatRunEvents.runId, runIds),
|
||||
inArray(heartbeatRunEvents.eventType, [
|
||||
"harness.ready",
|
||||
"session.started",
|
||||
"session.resumed",
|
||||
"session.updated",
|
||||
"turn.started",
|
||||
"provider.event",
|
||||
"provider.rpc_result",
|
||||
]),
|
||||
inArray(heartbeatRunEvents.eventType, NATIVE_SESSION_PROVIDER_EVENT_TYPES),
|
||||
),
|
||||
)
|
||||
.groupBy(heartbeatRunEvents.runId);
|
||||
@@ -224,7 +226,50 @@ export async function findNativeSessionResumeRun(
|
||||
},
|
||||
) {
|
||||
if (!isNativeSessionId(input.normalizedSessionId)) return null;
|
||||
const runs = await db
|
||||
const candidates = db.$with("native_resume_candidates").as(db
|
||||
.select({
|
||||
id: heartbeatRuns.id,
|
||||
runtimeMode: heartbeatRuns.runtimeMode,
|
||||
status: heartbeatRuns.status,
|
||||
createdAt: heartbeatRuns.createdAt,
|
||||
processPid: heartbeatRuns.processPid,
|
||||
processGroupId: heartbeatRuns.processGroupId,
|
||||
processStartedAt: heartbeatRuns.processStartedAt,
|
||||
hasCheckpoint: sql<boolean>`coalesce(${heartbeatRuns.runnerProfileJson}->'sessionCheckpoint' <> 'null'::jsonb, false)`.as("has_checkpoint"),
|
||||
})
|
||||
.from(heartbeatRuns)
|
||||
.where(
|
||||
and(
|
||||
ne(heartbeatRuns.id, input.currentRunId),
|
||||
eq(heartbeatRuns.companyId, input.companyId),
|
||||
eq(heartbeatRuns.agentId, input.agentId),
|
||||
eq(heartbeatRuns.nativeIssueId, input.issueId),
|
||||
eq(heartbeatRuns.nativeSessionId, input.normalizedSessionId),
|
||||
),
|
||||
)
|
||||
.orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id))
|
||||
.limit(32));
|
||||
const terminalStatuses = [...LEGACY_RETRY_SOURCE_TERMINAL_STATUSES];
|
||||
const barrier = db.$with("native_resume_barrier").as(db
|
||||
.select()
|
||||
.from(candidates)
|
||||
.where(or(
|
||||
ne(candidates.runtimeMode, "native"),
|
||||
not(inArray(candidates.status, terminalStatuses)),
|
||||
candidates.hasCheckpoint,
|
||||
isNotNull(candidates.processPid),
|
||||
isNotNull(candidates.processGroupId),
|
||||
isNotNull(candidates.processStartedAt),
|
||||
exists(db.select({ runId: heartbeatRunEvents.runId })
|
||||
.from(heartbeatRunEvents)
|
||||
.where(and(
|
||||
eq(heartbeatRunEvents.runId, candidates.id),
|
||||
inArray(heartbeatRunEvents.eventType, NATIVE_SESSION_PROVIDER_EVENT_TYPES),
|
||||
))),
|
||||
))
|
||||
.orderBy(desc(candidates.createdAt), desc(candidates.id))
|
||||
.limit(1));
|
||||
const [run] = await db.with(candidates, barrier)
|
||||
.select({
|
||||
id: heartbeatRuns.id,
|
||||
companyId: heartbeatRuns.companyId,
|
||||
@@ -238,29 +283,39 @@ export async function findNativeSessionResumeRun(
|
||||
processPid: heartbeatRuns.processPid,
|
||||
processGroupId: heartbeatRuns.processGroupId,
|
||||
processStartedAt: heartbeatRuns.processStartedAt,
|
||||
runnerProfileJson: heartbeatRuns.runnerProfileJson,
|
||||
contextSnapshot: heartbeatRuns.contextSnapshot,
|
||||
runnerProfileJson: sql<{
|
||||
nativeToolContractFingerprint: unknown;
|
||||
nativeExecutionInput: unknown;
|
||||
sessionCheckpoint: unknown;
|
||||
}>`jsonb_build_object(
|
||||
'nativeToolContractFingerprint', ${heartbeatRuns.runnerProfileJson}->'nativeToolContractFingerprint',
|
||||
'nativeExecutionInput', ${heartbeatRuns.runnerProfileJson}->'nativeExecutionInput',
|
||||
'sessionCheckpoint', ${heartbeatRuns.runnerProfileJson}->'sessionCheckpoint'
|
||||
)`,
|
||||
contextSnapshot: sql<{
|
||||
paperclipWake: unknown;
|
||||
paperclipIssue: unknown;
|
||||
}>`jsonb_build_object(
|
||||
'paperclipWake', ${heartbeatRuns.contextSnapshot}->'paperclipWake',
|
||||
'paperclipIssue', case when jsonb_typeof(${heartbeatRuns.contextSnapshot}->'paperclipIssue') = 'object'
|
||||
then (select coalesce(jsonb_object_agg(brief.key, brief.value), '{}'::jsonb)
|
||||
from jsonb_each(${heartbeatRuns.contextSnapshot}->'paperclipIssue') as brief
|
||||
where brief.key in ('id', 'title', 'description')) else ${heartbeatRuns.contextSnapshot}->'paperclipIssue' end
|
||||
)`,
|
||||
})
|
||||
.from(heartbeatRuns)
|
||||
.where(
|
||||
and(
|
||||
ne(heartbeatRuns.id, input.currentRunId),
|
||||
eq(heartbeatRuns.companyId, input.companyId),
|
||||
eq(heartbeatRuns.agentId, input.agentId),
|
||||
eq(heartbeatRuns.nativeIssueId, input.issueId),
|
||||
eq(heartbeatRuns.nativeSessionId, input.normalizedSessionId),
|
||||
),
|
||||
)
|
||||
.orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id))
|
||||
.limit(32);
|
||||
return selectNativeSessionResumeRun({
|
||||
...input,
|
||||
runs,
|
||||
providerEvidenceRunIds: await nativeSessionProviderEvidence(
|
||||
db,
|
||||
runs.map((run) => run.id),
|
||||
),
|
||||
});
|
||||
.innerJoin(barrier, eq(heartbeatRuns.id, barrier.id))
|
||||
.where(and(
|
||||
eq(barrier.runtimeMode, "native"),
|
||||
inArray(barrier.status, terminalStatuses),
|
||||
eq(barrier.hasCheckpoint, true),
|
||||
sql`${heartbeatRuns.companyId}::text = ${input.companyId}`,
|
||||
sql`${heartbeatRuns.agentId}::text = ${input.agentId}`,
|
||||
sql`${heartbeatRuns.nativeIssueId}::text = ${input.issueId}`,
|
||||
sql`${heartbeatRuns.nativeSessionId}::text = ${input.normalizedSessionId}`,
|
||||
sql`date_trunc('milliseconds', ${barrier.createdAt}) <= ${input.beforeCreatedAt.toISOString()}::timestamptz`,
|
||||
));
|
||||
return run ?? null;
|
||||
}
|
||||
|
||||
/** A preassigned session id may rotate only before immutable native admission. */
|
||||
|
||||
Reference in new issue
Block a user