fix(harness): preserve selected coding agent for template launches

.changeset/archive-before-retention.md

@@ -0,0 +1,5 @@
+---
+"@sapiom/harness": patch
+---
+
+Limit archive backfill to 200 conversations per maintenance pass. Keep source events while work remains or archiving fails, and retry at the next scheduled cleanup.

.changeset/template-harness-selection.md

@@ -0,0 +1,5 @@
+---
+"@sapiom/harness": patch
+---
+
+Preserve the selected coding agent when launching a template from the new-session composer or template gallery, including bundled starters. Codex selections no longer start Claude Code sessions.

packages/harness/README.md

@@ -21,6 +21,8 @@ system prompt, in whatever project directory you choose.
   Agent Studio only configures it. The `+` beside a project starts a session at
   that project root; the tab-strip `+` starts a sibling session. Sessions have
   resumable chat history.
+- **Templates** — quick starts, the template gallery, and bundled starters use
+  your selected coding agent.
 - **Agents rail** — agent projects (`sapiom.json`) discovered and
   tracked, with one-click local test run, deploy, production run, and
   open-in-Sapiom actions. How that discovery is rooted and bounded, how a

packages/harness/src/core/record-archive.test.ts

@@ -354,43 +354,46 @@ describe("backfillSessionRecords", () => {
     const read = new Set<string>();
 
     const archived = await backfillSessionRecords({
-      conversationIds: async () => ["sess-live", "sess-done", "sess-missing"],
+      conversationIds: async () => ["sess-live", "sess-missing", "sess-done"],
       readFromEvents: async (id) => {
         read.add(id);
         return record({ harnessSessionId: id, mergedSessionIds: [id], agentSessionId: null });
       },
       archive,
       isLiveSession: (id) => id === "sess-live",
+      maxRecords: 1,
     });
 
-    expect(archived).toEqual(["sess-missing"]);
+    expect(archived).toEqual({ archived: ["sess-missing"], complete: true });
     // A live session is never even folded — the point is not to store a
     // half-finished record over the one its exit will write.
     expect([...read]).toEqual(["sess-missing"]);
     expect(await archive.has("sess-live")).toBe(false);
   });
 
-  it("stops at its cap and reports what it left behind", async () => {
-    const capped: number[] = [];
-    const archived = await backfillSessionRecords({
+  it("stops at its cap and finishes the remaining work on later passes", async () => {
+    const options = {
       conversationIds: async () => ["a", "b", "c", "d"],
-      readFromEvents: async (id) => record({ harnessSessionId: id, mergedSessionIds: [id], agentSessionId: null }),
+      readFromEvents: async (id: string) => record({ harnessSessionId: id, mergedSessionIds: [id], agentSessionId: null }),
       archive,
       maxRecords: 2,
-      onCapped: (remaining) => capped.push(remaining),
-    });
+    };
 
-    expect(archived).toEqual(["a", "b"]);
-    expect(capped).toEqual([2]);
+    expect(await backfillSessionRecords(options)).toEqual({ archived: ["a", "b"], complete: false });
+    expect(await archive.has("c")).toBe(false);
+    expect(await backfillSessionRecords(options)).toEqual({ archived: ["c", "d"], complete: true });
+    expect(await backfillSessionRecords(options)).toEqual({ archived: [], complete: true });
   });
 
-  it("skips a conversation the fold has nothing for, without failing the pass", async () => {
+  it.each(["missing", "empty"])("skips a %s conversation without failing the pass", async (kind) => {
     const archived = await backfillSessionRecords({
       conversationIds: async () => ["gone", "here"],
       readFromEvents: async (id) =>
-        id === "gone" ? null : record({ harnessSessionId: id, mergedSessionIds: [id], agentSessionId: null }),
+        id === "gone"
+          ? (kind === "missing" ? null : record({ turns: [], turnCount: 0 }))
+          : record({ harnessSessionId: id, mergedSessionIds: [id], agentSessionId: null }),
       archive,
     });
-    expect(archived).toEqual(["here"]);
+    expect(archived).toEqual({ archived: ["here"], complete: true });
   });
 });

packages/harness/src/core/record-archive.ts

@@ -529,15 +529,10 @@ export interface BackfillOptions {
   isLiveSession?: (harnessSessionId: string) => boolean;
   /** Ceiling on how many conversations one pass archives. */
   maxRecords?: number;
-  /** Called with the number of eligible conversations left unarchived when the
-   *  cap cut the pass short — a bounded pass must say what it didn't do. */
-  onCapped?: (remaining: number) => void;
 }
 
-/** Default ceiling for one backfill pass. High enough to cover a typical
- *  install's whole history on the first boot after this shipped, low enough
- *  that a pathological log doesn't turn boot into a write storm. Whatever is
- *  left is archived by the next boot's pass. */
+/** Limit startup disk writes. Any remainder waits for a later maintenance
+ *  pass; retention must preserve the source events until backfill completes. */
 export const RECORDS_BACKFILL_MAX = 200;
 
 /**
@@ -550,25 +545,24 @@ export const RECORDS_BACKFILL_MAX = 200;
  * on. Idempotent: a conversation already archived is skipped, so the steady
  * state after the first pass is "nothing to do".
  *
- * Never throws. Returns the ids it archived.
+ * Returns the ids it archived and whether the pass finished. An incomplete
+ * pass or a read/write failure must prevent retention from deleting sources.
  */
-export async function backfillSessionRecords(options: BackfillOptions): Promise<string[]> {
+export async function backfillSessionRecords(
+  options: BackfillOptions,
+): Promise<{ archived: string[]; complete: boolean }> {
   const maxRecords = options.maxRecords ?? RECORDS_BACKFILL_MAX;
   const ids = await options.conversationIds();
   const archived: string[] = [];
-  let remaining = 0;
   for (const id of ids) {
     if (options.isLiveSession?.(id)) continue;
     if (await options.archive.has(id)) continue;
-    if (archived.length >= maxRecords) {
-      remaining += 1;
-      continue;
-    }
+    if (archived.length >= maxRecords) return { archived, complete: false };
     const record = await options.readFromEvents(id);
-    if (!record) continue;
+    if (!record || record.turns.length === 0) continue;
     const written = await options.archive.write(record);
-    if (written) archived.push(id);
+    if (!written) throw new Error(`Could not archive conversation ${id}; keeping source events`);
+    archived.push(id);
   }
-  if (remaining > 0) options.onCapped?.(remaining);
-  return archived;
+  return { archived, complete: true };
 }

packages/harness/src/server/index.ts

@@ -3024,58 +3024,44 @@ 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);
+  // Bound archive work per pass and run passes one at a time. Remaining work
+  // or a read/write failure skips cleanup; the next timer tick retries the
+  // conversations that are not yet archived.
+  let recordMaintenance = Promise.resolve();
+  // Session-exit sweeps can evict archives between passes. Remember completed
+  // writes until event cleanup succeeds, so a large backfill makes progress.
+  const archivedDuringBackfill = new Set<string>();
+  const runRecordMaintenance = (): Promise<void> => {
+    recordMaintenance = recordMaintenance.then(async () => {
+      const { archived, complete } = await backfillSessionRecords({
+        conversationIds: async () => (await sessionRecordReader.conversationIds())
+          .filter((id) => !archivedDuringBackfill.has(id)),
+        readFromEvents: (id) => sessionRecordReader.readFromEvents(id),
+        archive: recordArchive,
+        isLiveSession: (id) => {
+          const session = sessionManager.get(id);
+          return session !== undefined && session.status !== "exited";
+        },
       });
+      for (const id of archived) archivedDuringBackfill.add(id);
+      if (!complete) {
+        console.warn("[harness] archive backfill reached its limit; keeping source events until the next pass");
+        return;
+      }
+      await recordArchive.sweep();
+      // The exclusive queue also protects retention's read/filter/rename from
+      // concurrent event appends.
+      await eventStore.runExclusive(() => sweepNdjson(eventStorePath));
+      archivedDuringBackfill.clear();
+    }).catch((err: unknown) => {
+      console.error("[harness] session record maintenance failed:", err);
+    });
+    return recordMaintenance;
   };
-  runNdjsonSweep();
-  const ndjsonRetentionTimer = setInterval(
-    runNdjsonSweep,
-    NDJSON_RETENTION_SWEEP_MS,
-  );
+  void runRecordMaintenance();
+  const ndjsonRetentionTimer = setInterval(runRecordMaintenance, 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.
-  //
-  // 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
-  // anyway.
-  const recordBackfill = backfillSessionRecords({
-    conversationIds: () => sessionRecordReader.conversationIds(),
-    readFromEvents: (id) => sessionRecordReader.readFromEvents(id),
-    archive: recordArchive,
-    isLiveSession: (harnessSessionId) => {
-      const session = sessionManager.get(harnessSessionId);
-      return session !== undefined && session.status !== "exited";
-    },
-    onCapped: (remaining) => {
-      console.error(
-        `[harness] session record backfill hit its per-boot cap; ${remaining} conversation(s) left for the next boot`,
-      );
-    },
-  })
-    .then(() => recordArchive.sweep())
-    .catch((err: unknown) => {
-      console.error("[harness] session record backfill failed:", err);
-    });
-
   const harnessVersion = readVersion();
   const batcher = createHarnessEmitter({
     telemetryOptIn: options.telemetryOptIn,
@@ -4663,7 +4649,7 @@ export const startServer = async (
         await registrationClosing;
         await settle(() => sessionManager.flush());
         await settle(async () => {
-          await recordBackfill;
+          await recordMaintenance;
           while (pendingRecordArchives.size > 0) {
             await Promise.all([...pendingRecordArchives]);
           }

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

@@ -15,11 +15,13 @@
  *     disappearing at its 30-day mark.
  */
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
-import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
+import { mkdir, mkdtemp, readFile, readdir, rm, writeFile } from "node:fs/promises";
 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 });
   });
 
@@ -229,14 +232,15 @@ describe("session record archive wiring", () => {
     );
   }, 20_000);
 
-  it("archives conversations that ended before the archive existed", async () => {
+  it.each([1, 200, 201, 401])("archives historical conversations before cleanup (count=%i)", async (count) => {
     // 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",
@@ -260,16 +264,96 @@ describe("session record archive wiring", () => {
         payload: { assistantText: "done" },
       },
     ];
-    await writeFile(eventStorePath, events.map((e) => `${JSON.stringify(e)}\n`).join(""), "utf8");
+    const history = Array.from({ length: count }, (_, index) => events.map((event) => ({
+      ...event,
+      eventId: `${event.eventId}-${index}`,
+      seq: event.seq + index * events.length,
+      ts: new Date(Date.parse(event.ts) + index).toISOString(),
+      harnessSessionId: index === 0 ? event.harnessSessionId : `${event.harnessSessionId}-${index}`,
+      agentSessionId: index === 0 ? event.agentSessionId : `${event.agentSessionId}-${index}`,
+    }))).flat();
+    await writeFile(eventStorePath, history.map((event) => `${JSON.stringify(event)}\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");
+    const intervals = vi.spyOn(globalThis, "setInterval");
+    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");
-      },
-      { timeout: 10_000, interval: 100 },
-    );
+    await backfillSpy.mock.results[0].value;
+    if (count > 200) {
+      expect((await readdir(recordsRoot)).filter((file) => file.endsWith(".json"))).toHaveLength(200);
+      expect(await readArchived("sess-legacy")).toBeNull();
+      expect(sweepSpy).not.toHaveBeenCalled();
+      expect(await readFile(eventStorePath, "utf8")).toContain("sess-legacy");
+      // Finish the remaining work on the next scheduled pass, not at boot.
+      const tick = intervals.mock.calls.find(([, ms]) => ms === 6 * 60 * 60 * 1_000)?.[0];
+      expect(tick).toBeTypeOf("function");
+      for (let pass = 1; pass * 200 < count; pass += 1) {
+        // Session-exit archive sweeps can evict completed files between passes.
+        // Force that eviction to prove a large backfill still makes progress.
+        if (count > 400) {
+          await recordArchive.createRecordArchive({ root: recordsRoot, maxTotalBytes: 0 }).sweep();
+        }
+        expect(sweepSpy).not.toHaveBeenCalled();
+        expect(await readFile(eventStorePath, "utf8")).toContain("sess-legacy");
+        await (tick as () => Promise<void>)();
+      }
+      expect(sweepSpy).toHaveBeenCalledOnce();
+    }
+
+    await vi.waitFor(async () => {
+      expect(await readFile(eventStorePath, "utf8")).not.toContain("sess-legacy");
+    }, { timeout: 10_000, interval: 100 });
+    // The oldest conversation must survive even when it needs a later pass.
+    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);
+
+  it.each(["write", "read"])("keeps source events when archive %s fails", async (failure) => {
+    const prompt = "keep this conversation until it is archived";
+    const event: AnalyticsEvent = {
+      eventId: "evt-unarchived", seq: 1,
+      ts: new Date(Date.now() - retention.DEFAULT_MAX_AGE_MS - 60_000).toISOString(),
+      userId: null, tenantId: null, machineId: "machine-1",
+      harnessSessionId: "sess-unarchived", agentSessionId: "agent-unarchived",
+      harness: "claude-code", type: "prompt.submitted", payload: { prompt },
+    };
+    await writeFile(eventStorePath, `${JSON.stringify(event)}\n`, "utf8");
+    const backfill = recordArchive.backfillSessionRecords;
+    const backfillSpy = vi.spyOn(recordArchive, "backfillSessionRecords");
+    const intervals = vi.spyOn(globalThis, "setInterval");
+    if (failure === "write") await writeFile(recordsRoot, "blocks archive directory creation");
+    else backfillSpy.mockRejectedValue(new Error("archive source unavailable"));
+    const sweepSpy = vi.spyOn(retention, "sweepNdjson");
+    server = await boot();
+    const tick = intervals.mock.calls.find(([, ms]) => ms === 6 * 60 * 60 * 1_000)?.[0];
+    expect(tick).toBeTypeOf("function");
+    const runMaintenance = tick as () => Promise<void>;
+    // Another scheduled attempt must also preserve the source while it fails.
+    await runMaintenance();
+    expect(sweepSpy).not.toHaveBeenCalled();
+    expect(await readFile(eventStorePath, "utf8")).toContain(prompt);
+    if (failure === "write") await rm(recordsRoot);
+    else backfillSpy.mockImplementation(backfill);
+    await runMaintenance();
+    expect(await readFile(eventStorePath, "utf8")).not.toContain(prompt);
+    expect((await fetchRecord("agent-unarchived")).body?.turns[0].prompt).toBe(prompt);
+  });
 });