fix(harness): archive history before event retention

.changeset/archive-before-retention.md

@@ -0,0 +1,5 @@
+---
+"@sapiom/harness": patch
+---
+
+Archive historical conversations before event retention can remove their source events during startup.

packages/harness/src/server/index.ts

@@ -2971,35 +2971,14 @@ export const startServer = async (
     };
   });
 
-  // Boot-time retention sweep: keeps events.ndjson within the 50 MB / 30-day
-  // caps even on long-lived installs. Runs through the store's exclusive queue
-  // so the sweep's read→filter→rename window never races a concurrent append.
-  // Fire-and-forget — a slow FS is no reason to delay server startup.
-  const runNdjsonSweep = (): void => {
-    void eventStore
-      .runExclusive(() => sweepNdjson(eventStorePath))
-      .catch((err: unknown) => {
-        console.error("[harness] events.ndjson retention sweep failed:", err);
-      });
-  };
-  runNdjsonSweep();
-  const ndjsonRetentionTimer = setInterval(
-    runNdjsonSweep,
-    NDJSON_RETENTION_SWEEP_MS,
-  );
-  ndjsonRetentionTimer.unref?.();
-
   // One boot-time pass that archives conversations the log still holds but the
   // archive doesn't, then sweeps the archive's own caps. This is what covers the
   // two cases archiving-at-exit can't: a harness that was force-killed (no exit
   // transition, no session.end), and every session that ended before this
   // existed — whose history would otherwise vanish at its 30-day mark.
   //
-  // It races the ndjson sweep queued above, and deliberately doesn't wait for
-  // it: reads run outside the store's exclusive queue by design (see store.ts),
-  // and either order is correct here — win the race and the record is archived
-  // from bytes retention was about to delete, lose it and the record is archived
-  // from what survived. Both beat not archiving it.
+  // Retention must wait for this pass: reads run outside the store's exclusive
+  // queue, so a sweep could otherwise delete old events before we archive them.
   //
   // Fire-and-forget: boot must not wait on it. The cost is one full index build
   // (~130 ms against a 50 MB log), which the first history open would have paid
@@ -3023,6 +3002,25 @@ export const startServer = async (
       console.error("[harness] session record backfill failed:", err);
     });
 
+  // Boot-time retention sweep: keeps events.ndjson within the 50 MB / 30-day
+  // caps even on long-lived installs. Runs through the store's exclusive queue
+  // so the sweep's read→filter→rename window never races a concurrent append.
+  // Wait for backfill before every sweep, including a timer tick during a slow
+  // boot pass. Server startup stays independent of both maintenance tasks.
+  const runNdjsonSweep = (): void => {
+    void recordBackfill
+      .then(() => eventStore.runExclusive(() => sweepNdjson(eventStorePath)))
+      .catch((err: unknown) => {
+        console.error("[harness] events.ndjson retention sweep failed:", err);
+      });
+  };
+  runNdjsonSweep();
+  const ndjsonRetentionTimer = setInterval(
+    runNdjsonSweep,
+    NDJSON_RETENTION_SWEEP_MS,
+  );
+  ndjsonRetentionTimer.unref?.();
+
   const harnessVersion = readVersion();
   const batcher = createHarnessEmitter({
     telemetryOptIn: options.telemetryOptIn,

packages/harness/src/server/record-archive-wiring.test.ts

@@ -20,6 +20,8 @@ import { tmpdir } from "node:os";
 import { join } from "node:path";
 
 import { startServer, type HarnessServer } from "./index.js";
+import * as recordArchive from "../core/record-archive.js";
+import * as retention from "../core/collector/store-retention.js";
 import type {
   AnalyticsEvent,
   HarnessAdapter,
@@ -79,6 +81,7 @@ describe("session record archive wiring", () => {
     await server?.close();
     await server?.sessionManager.flush();
     server = undefined;
+    vi.restoreAllMocks();
     await rm(dir, { recursive: true, force: true });
   });
 
@@ -232,11 +235,12 @@ describe("session record archive wiring", () => {
   it("archives conversations that ended before the archive existed", async () => {
     // An events file from an install that predates this feature: a complete
     // conversation, no archive, and no registry entry for it.
+    const expiredAt = Date.now() - retention.DEFAULT_MAX_AGE_MS - 60_000;
     const events: AnalyticsEvent[] = [
       {
         eventId: "evt-1",
         seq: 1,
-        ts: "2026-06-01T10:00:00.000Z",
+        ts: new Date(expiredAt).toISOString(),
         userId: null,
         tenantId: null,
         machineId: "machine-1",
@@ -249,7 +253,7 @@ describe("session record archive wiring", () => {
       {
         eventId: "evt-2",
         seq: 2,
-        ts: "2026-06-01T10:00:01.000Z",
+        ts: new Date(expiredAt + 1000).toISOString(),
         userId: null,
         tenantId: null,
         machineId: "machine-1",
@@ -262,14 +266,39 @@ describe("session record archive wiring", () => {
     ];
     await writeFile(eventStorePath, events.map((e) => `${JSON.stringify(e)}\n`).join(""), "utf8");
 
-    server = await boot();
+    // Hold the backfill to prove retention cannot delete its source events,
+    // even when startup finishes before the archive work does.
+    let releaseBackfill!: () => void;
+    const backfillGate = new Promise<void>((resolve) => { releaseBackfill = resolve; });
+    const backfill = recordArchive.backfillSessionRecords;
+    const backfillSpy = vi.spyOn(recordArchive, "backfillSessionRecords")
+      .mockImplementation(async (options) => {
+        await backfillGate;
+        return backfill(options);
+      });
+    const sweepSpy = vi.spyOn(retention, "sweepNdjson");
+    try {
+      server = await boot();
+      expect(backfillSpy).toHaveBeenCalledOnce();
+      expect(sweepSpy).not.toHaveBeenCalled();
+    } finally {
+      releaseBackfill();
+    }
 
     await vi.waitFor(
       async () => {
         const archived = await readArchived("sess-legacy");
         expect(archived?.turns[0].prompt).toBe("from before the archive existed");
+        expect(archived?.turns[0].assistantText).toBe("done");
       },
       { timeout: 10_000, interval: 100 },
     );
+    await vi.waitFor(async () => {
+      expect(await readFile(eventStorePath, "utf8")).not.toContain("sess-legacy");
+    });
+    const archived = await fetchRecord("agent-legacy");
+    expect(archived.status).toBe(200);
+    expect(archived.body?.turns[0].prompt).toBe("from before the archive existed");
+    expect(archived.body?.turns[0].assistantText).toBe("done");
   }, 20_000);
 });