diff --git a/doc/DATABASE.md b/doc/DATABASE.md index 24e06d59830..bf9f4a03333 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -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=`. +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 diff --git a/scripts/benchmark-native-session-resume-reads.ts b/scripts/benchmark-native-session-resume-reads.ts new file mode 100644 index 00000000000..df302f1ba71 --- /dev/null +++ b/scripts/benchmark-native-session-resume-reads.ts @@ -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[1]) { + const queries: Query[] = []; + const wrap = (target: Db): Db => new Proxy(target, { get(target, key) { + if (key === "with") return (...args: Parameters) => wrap(target.with(...args) as Db); + if (key !== "select") return Reflect.get(target, key); + return (...fields: Parameters) => { + const builder = target.select(...fields), from = builder.from.bind(builder); + builder.from = ((...args: Parameters) => { + const query = from(...args), execute = query.execute.bind(query); + query.execute = async (...executeArgs: Parameters) => { + 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 }); } + } + } +} diff --git a/server/src/services/native-runtime/native-session-resume-projection.test.ts b/server/src/services/native-runtime/native-session-resume-projection.test.ts new file mode 100644 index 00000000000..08242f2a690 --- /dev/null +++ b/server/src/services/native-runtime/native-session-resume-projection.test.ts @@ -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>; + 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 = {}) => ({ + 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) => { + const builder = target.with(...ctes), original = builder.select.bind(builder); + builder.select = ((...fields: Parameters) => { + const selection = original(...fields), from = selection.from.bind(selection); + selection.from = ((...args: Parameters) => { + const query = from(...args), execute = query.execute.bind(query); + query.execute = async (...executeArgs: Parameters) => { + 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); + }); +}); diff --git a/server/src/services/native-runtime/native-session-resume.ts b/server/src/services/native-runtime/native-session-resume.ts index a271601ce0b..25463b68c78 100644 --- a/server/src/services/native-runtime/native-session-resume.ts +++ b/server/src/services/native-runtime/native-session-resume.ts @@ -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, 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`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. */