diff --git a/packages/paperclip-runner/docs/durable-recovery.md b/packages/paperclip-runner/docs/durable-recovery.md index 1e8bc2acf22..ed505bc7712 100644 --- a/packages/paperclip-runner/docs/durable-recovery.md +++ b/packages/paperclip-runner/docs/durable-recovery.md @@ -292,6 +292,7 @@ model or rely on an LLM choosing the desired timing. | Execution and cleanup both fail | Original execution error retained; cleanup error attached | `native-session-runtime.test.ts` | | Retry in same process or after local log loss | Earlier attempts survive locally and in the durable mirror | Server `run-log-store.test.ts` | | Two restores overlap and one appends before the other publishes | Complete prefix is published without replacement; both attempts survive | Server `run-log-store.test.ts` | +| New run whose row has no log reference, output sequence, or log size | No durable-mirror request; a later attempt with a recorded log reference still restores after local loss | Server `run-log-store.test.ts`, `heartbeat/run-output.test.ts` | ## Residual local trust and revocation window diff --git a/server/src/services/heartbeat/run-output.test.ts b/server/src/services/heartbeat/run-output.test.ts new file mode 100644 index 00000000000..bcfb7c3c2bc --- /dev/null +++ b/server/src/services/heartbeat/run-output.test.ts @@ -0,0 +1,123 @@ +import { promises as fs } from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { Readable } from "node:stream"; +import type { Db, heartbeatRuns } from "@paperclipai/db"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { createDurableRunLogStore } from "../run-log-store.js"; +import type { StorageProvider } from "../../storage/types.js"; +import { createHeartbeatRunOutput, isNewRunLog } from "./run-output.js"; + +vi.mock("../live-events.js", () => ({ publishLiveEvent: vi.fn() })); + +type Run = typeof heartbeatRuns.$inferSelect; + +function createMemoryProvider() { + const objects = new Map(); + const calls = { head: 0, get: 0 }; + const provider: StorageProvider = { + id: "s3", + async putObject(input) { + const chunks: Buffer[] = []; + for await (const chunk of Readable.from(input.body)) chunks.push(Buffer.from(chunk)); + objects.set(input.objectKey, Buffer.concat(chunks)); + }, + async getObject(input) { + calls.get++; + const buf = objects.get(input.objectKey)!; + return { stream: Readable.from(buf), contentLength: buf.length }; + }, + async headObject(input) { + calls.head++; + const buf = objects.get(input.objectKey); + return buf ? { exists: true, contentLength: buf.length } : { exists: false }; + }, + async deleteObject(input) { + objects.delete(input.objectKey); + }, + }; + return { provider, objects, calls }; +} + +// The recorder persists the handle with update().set().where(); capture it the +// way the run row would carry it into a later attempt. +function createRunRowDb(row: Partial) { + return { + update: () => ({ + set: (values: Partial) => ({ + where: async () => { Object.assign(row, values); }, + }), + }), + } as unknown as Db; +} + +const newRunRow = { + id: "run-1", companyId: "company-1", agentId: "agent-1", status: "running", + logStore: null, logRef: null, logBytes: null, lastOutputSeq: 0, lastOutputAt: null, +} as Partial; + +async function beginAttempt(store: ReturnType, row: Partial) { + const output = createHeartbeatRunOutput(createRunRowDb(row), { + run: row as Run, + runId: row.id!, + issueId: null, + runLogStore: store, + identityRedactor: { chunk: (_stream, chunk) => chunk, redact: (value) => value }, + secretKeys: new Set(), + appendRunEvent: vi.fn(), + getCurrentUserRedactionOptions: async () => ({ enabled: false }), + }); + const { onLog } = await output.begin({ ...row } as Run); + return { output, onLog }; +} + +let baseDir: string; +beforeEach(async () => { + baseDir = await fs.mkdtemp(path.join(os.tmpdir(), "run-output-test-")); +}); +afterEach(async () => { + await fs.rm(baseDir, { recursive: true, force: true }); +}); + +describe("heartbeat run output log recovery", () => { + it("treats only a run row with no recorded log or output as new", () => { + expect(isNewRunLog({ logStore: null, logRef: null, logBytes: null, lastOutputSeq: 0 })).toBe(true); + expect(isNewRunLog({ logStore: "local_file", logRef: "c/a/r.ndjson", logBytes: null, lastOutputSeq: 0 })).toBe(false); + expect(isNewRunLog({ logStore: null, logRef: "c/a/r.ndjson", logBytes: null, lastOutputSeq: 0 })).toBe(false); + expect(isNewRunLog({ logStore: null, logRef: null, logBytes: null, lastOutputSeq: 3 })).toBe(false); + expect(isNewRunLog({ logStore: null, logRef: null, logBytes: 120, lastOutputSeq: 0 })).toBe(false); + }); + + it("makes no remote log recovery request for a new run", async () => { + const { provider, calls } = createMemoryProvider(); + const store = createDurableRunLogStore({ basePath: baseDir, s3: { provider } }); + const row = { ...newRunRow }; + const { onLog } = await beginAttempt(store, row); + await onLog("stdout", "hello\n"); + expect(calls).toEqual({ head: 0, get: 0 }); + expect(row.logRef).toBe(path.join("company-1", "agent-1", "run-1.ndjson")); + }); + + it("restores an existing attempt's durable log after local state is lost", async () => { + const { provider, objects, calls } = createMemoryProvider(); + const row = { ...newRunRow }; + const firstStore = createDurableRunLogStore({ basePath: baseDir, s3: { provider } }); + const first = await beginAttempt(firstStore, row); + await first.onLog("stdout", "attempt one\n"); + await firstStore.finalize(first.output.handle!); + expect(calls).toEqual({ head: 0, get: 0 }); + + // Process restart on a new pod: local logs and in-memory state are gone, + // but the run row still records the earlier attempt's log. + await fs.rm(baseDir, { recursive: true, force: true }); + const restartedStore = createDurableRunLogStore({ basePath: baseDir, s3: { provider } }); + const second = await beginAttempt(restartedStore, row); + expect(calls).toEqual({ head: 1, get: 1 }); + await second.onLog("stdout", "attempt two\n"); + await restartedStore.finalize(second.output.handle!); + + const lines = objects.get(row.logRef!)!.toString().trim().split("\n").map((line) => JSON.parse(line)); + expect(lines.map((line) => line.chunk)).toEqual(["attempt one\n", "attempt two\n"]); + expect(lines.map((line) => line.seq)).toEqual([1, 2]); + }); +}); diff --git a/server/src/services/heartbeat/run-output.ts b/server/src/services/heartbeat/run-output.ts index 1b1091b9052..2b285ef8297 100644 --- a/server/src/services/heartbeat/run-output.ts +++ b/server/src/services/heartbeat/run-output.ts @@ -26,6 +26,21 @@ export interface HeartbeatRunOutputInput { getCurrentUserRedactionOptions: () => Promise; } +/** + * True only when the persisted run row proves no earlier attempt began its log. + * The first begin() records logStore/logRef before any output can be appended + * or mirrored to object storage; output sequence and final size are recorded + * after that. A row with none of them has no durable log prefix to restore. + */ +export function isNewRunLog( + run: Pick, +): boolean { + return run.logStore == null + && run.logRef == null + && run.logBytes == null + && Number(run.lastOutputSeq ?? 0) === 0; +} + /** One recorder per execution attempt; the log store itself stays service-scoped. */ export function createHeartbeatRunOutput(db: Db, input: HeartbeatRunOutputInput) { const { run, runId, issueId, runLogStore, identityRedactor, secretKeys, appendRunEvent, getCurrentUserRedactionOptions } = input; @@ -74,6 +89,7 @@ export function createHeartbeatRunOutput(db: Db, input: HeartbeatRunOutputInput) companyId: run.companyId, agentId: run.agentId, runId, + newRun: isNewRunLog(run), }); await db diff --git a/server/src/services/run-log-store.test.ts b/server/src/services/run-log-store.test.ts index 75b7efda358..c4ad5900fea 100644 --- a/server/src/services/run-log-store.test.ts +++ b/server/src/services/run-log-store.test.ts @@ -175,6 +175,24 @@ describe("createDurableRunLogStore", () => { } finally { spy.mockRestore(); } }); + it("skips the durable restore request only for a proven-new run", async () => { + const { provider, objects, calls } = createMemoryProvider(); + const store = createDurableRunLogStore({ basePath: baseDir, s3: { provider } }); + const fresh = await store.begin({ ...begin, newRun: true }); + expect(calls).toMatchObject({ head: 0, get: 0 }); + await store.append(fresh, { stream: "stdout", chunk: "first attempt", ts: "t1" }); + await store.finalize(fresh); + + // Lose local state, then begin again without the proof: the prefix restores. + await fs.rm(baseDir, { recursive: true, force: true }); + const recovered = await store.begin(begin); + expect(calls).toMatchObject({ head: 1, get: 1 }); + await store.append(recovered, { stream: "stdout", chunk: "second attempt", ts: "t2" }); + await store.finalize(recovered); + expect(objects.get(fresh.logRef)!.toString().trim().split("\n").map((line) => JSON.parse(line).chunk)) + .toEqual(["first attempt", "second attempt"]); + }); + it("uploads the complete log to S3 on finalize", async () => { const { provider, objects, calls } = createMemoryProvider(); const store = createDurableRunLogStore({ basePath: baseDir, s3: { provider, keyPrefix: "run-logs" } }); diff --git a/server/src/services/run-log-store.ts b/server/src/services/run-log-store.ts index 85a51ba1e0b..7e26b15261e 100644 --- a/server/src/services/run-log-store.ts +++ b/server/src/services/run-log-store.ts @@ -35,8 +35,18 @@ export interface RunLogFinalizeSummary { compressed: boolean; } +export interface RunLogBeginInput { + companyId: string; + agentId: string; + runId: string; + // Set only when persisted run lifecycle state proves no earlier attempt + // began this log, so no durable prefix can exist to restore. Omitted means + // the run may be a retry or recovery and the durable mirror is checked. + newRun?: boolean; +} + export interface RunLogStore { - begin(input: { companyId: string; agentId: string; runId: string }): Promise; + begin(input: RunLogBeginInput): Promise; append( handle: RunLogHandle, event: { stream: "stdout" | "stderr" | "system"; chunk: string; ts: string; seq?: number }, @@ -300,11 +310,12 @@ export function createDurableRunLogStore(options: DurableRunLogStoreOptions): Ru await retireInflightMirror(relPath); // Retries share a run ID. Never truncate their earlier evidence. On a // new pod, restore the durable prefix before opening another attempt. + // A proven-new run has no prefix, so skip the remote round trip. const exists = await fs.stat(absPath).then(() => true, (error) => { if (error.code === "ENOENT") return false; throw error; }); - if (!exists && s3) { + if (!exists && s3 && input.newRun !== true) { const key = s3Key(relPath); const head = await s3.provider.headObject({ objectKey: key }); if (head.exists) {