perf(run-logs): skip durable log restore for a proven-new run (#17367)

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<string, Buffer>();
+  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<Run>) {
+  return {
+    update: () => ({
+      set: (values: Partial<Run>) => ({
+        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<Run>;
+
+async function beginAttempt(store: ReturnType<typeof createDurableRunLogStore>, row: Partial<Run>) {
+  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]);
+  });
+});

server/src/services/heartbeat/run-output.ts

@@ -26,6 +26,21 @@ export interface HeartbeatRunOutputInput {
   getCurrentUserRedactionOptions: () => Promise<CurrentUserRedactionOptions>;
 }
 
+/**
+ * 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<Run, "logStore" | "logRef" | "logBytes" | "lastOutputSeq">,
+): 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

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" } });

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<RunLogHandle>;
+  begin(input: RunLogBeginInput): Promise<RunLogHandle>;
   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) {