feat(harness): manage owned child sessions [Agent Map 13/15]

.changeset/owned-child-session-lifecycle.md

@@ -0,0 +1,7 @@
+---
+"@sapiom/harness": minor
+---
+
+Add trusted child-session creation, recovery and closure with exact binding checks, plus exclusive Codex rollout attribution for simultaneous runtimes. Failed launches retain retryable ownership, and finished discovery releases pending runtime registrations.
+
+The SessionManager returned by startServer exposes the owned-session lifecycle methods. Callers can handle the public SubsessionBindingMismatchError when an operation does not match its coordinator binding and SubsessionFreshRestartForbiddenError when a fresh restart would overwrite a recorded or explicitly closed conversation. The close() operation must be awaited because durable closure bookkeeping can reject.

packages/harness/src/core/collector/codex-rollout-broker.test.ts

@@ -0,0 +1,183 @@
+import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { afterEach, describe, expect, it, vi } from "vitest";
+
+import { CodexRolloutBroker } from "./codex-rollout-broker.js";
+import * as codexTailer from "./codex-tailer.js";
+
+const meta = (id: string, cwd: string, timestamp: string) =>
+  `${JSON.stringify({ type: "session_meta", payload: { id, cwd, timestamp } })}\n`;
+
+describe("CodexRolloutBroker", () => {
+  const roots: string[] = [];
+
+  afterEach(async () => {
+    vi.restoreAllMocks();
+    await Promise.all(
+      roots.splice(0).map((root) => rm(root, { recursive: true, force: true })),
+    );
+  });
+
+  async function fixture() {
+    const home = await mkdtemp(join(tmpdir(), "codex-rollout-broker-"));
+    const cwd = join(home, "project");
+    const sessions = join(home, ".codex", "sessions", "2026", "09", "04");
+    await Promise.all([mkdir(cwd), mkdir(sessions, { recursive: true })]);
+    roots.push(home);
+    return { home, cwd, sessions };
+  }
+
+  it("uniquely assigns concurrent same-root rollouts by process epoch", async () => {
+    const { home, cwd, sessions } = await fixture();
+    const firstTime = Date.parse("2026-09-04T10:00:00.000Z");
+    const secondTime = Date.parse("2026-09-04T10:00:01.000Z");
+    const firstPath = join(sessions, "rollout-first.jsonl");
+    const secondPath = join(sessions, "rollout-second.jsonl");
+    await writeFile(
+      firstPath,
+      meta("agent-first", cwd, "2026-09-04T10:00:00.500Z"),
+    );
+    await writeFile(
+      secondPath,
+      meta("agent-second", cwd, "2026-09-04T10:00:01.500Z"),
+    );
+    const broker = new CodexRolloutBroker(home);
+    broker.register({
+      sessionId: "first",
+      runtimeEpoch: "runtime-1",
+      cwd,
+      sinceMs: firstTime,
+    });
+    broker.register({
+      sessionId: "second",
+      runtimeEpoch: "runtime-2",
+      cwd,
+      sinceMs: secondTime,
+    });
+
+    await expect(
+      broker.claimFresh({
+        sessionId: "first",
+        runtimeEpoch: "runtime-1",
+        cwd,
+        sinceMs: firstTime,
+      }),
+    ).resolves.toEqual({ outcome: "claimed", path: firstPath });
+    await expect(
+      broker.claimFresh({
+        sessionId: "second",
+        runtimeEpoch: "runtime-2",
+        cwd,
+        sinceMs: secondTime,
+      }),
+    ).resolves.toEqual({ outcome: "claimed", path: secondPath });
+  });
+
+  it("fails closed when same-root process epochs cannot distinguish candidates", async () => {
+    const { home, cwd, sessions } = await fixture();
+    const sinceMs = Date.parse("2026-09-04T10:00:00.000Z");
+    await writeFile(
+      join(sessions, "rollout-a.jsonl"),
+      meta("agent-a", cwd, "2026-09-04T10:00:01.000Z"),
+    );
+    await writeFile(
+      join(sessions, "rollout-b.jsonl"),
+      meta("agent-b", cwd, "2026-09-04T10:00:02.000Z"),
+    );
+    const broker = new CodexRolloutBroker(home);
+    broker.register({
+      sessionId: "first",
+      runtimeEpoch: "runtime-1",
+      cwd,
+      sinceMs,
+    });
+    broker.register({
+      sessionId: "second",
+      runtimeEpoch: "runtime-2",
+      cwd,
+      sinceMs,
+    });
+
+    await expect(
+      broker.claimFresh({
+        sessionId: "first",
+        runtimeEpoch: "runtime-1",
+        cwd,
+        sinceMs,
+      }),
+    ).resolves.toEqual({ outcome: "ambiguous", path: null });
+    await expect(
+      broker.claimFresh({
+        sessionId: "second",
+        runtimeEpoch: "runtime-2",
+        cwd,
+        sinceMs,
+      }),
+    ).resolves.toEqual({ outcome: "ambiguous", path: null });
+  });
+
+  it("does not assign a fresh rollout after its runtime was released during discovery", async () => {
+    const { home, cwd } = await fixture();
+    const candidate = {
+      path: "/fake/next-rollout.jsonl",
+      agentSessionId: "agent-next",
+      timestampMs: Date.now(),
+      mtimeMs: Date.now(),
+    };
+    let finishDiscovery!: (candidates: typeof candidate[]) => void;
+    const finder = vi.spyOn(codexTailer, "findRolloutCandidates")
+      .mockImplementationOnce(() => new Promise((resolve) => { finishDiscovery = resolve; }))
+      .mockResolvedValue([candidate]);
+    const broker = new CodexRolloutBroker(home);
+    const pending = broker.claimFresh({
+      sessionId: "retired-session",
+      runtimeEpoch: "retired-runtime",
+      cwd,
+      sinceMs: 0,
+    });
+    await vi.waitFor(() => expect(finder).toHaveBeenCalledOnce());
+    broker.releaseSession("retired-session");
+    finishDiscovery([candidate]);
+    await expect(pending).resolves.toEqual({ outcome: "pending", path: null });
+    await expect(broker.claimFresh({
+      sessionId: "next-session",
+      runtimeEpoch: "next-runtime",
+      cwd,
+      sinceMs: 0,
+    })).resolves.toEqual({ outcome: "claimed", path: candidate.path });
+  });
+
+  it("allows only the same Harness session to reclaim an exact rollout on resume", async () => {
+    const { home, cwd, sessions } = await fixture();
+    const rolloutPath = join(sessions, "rollout-resume.jsonl");
+    await writeFile(
+      rolloutPath,
+      meta("agent-resume", cwd, "2026-09-04T10:00:01.000Z"),
+    );
+    const broker = new CodexRolloutBroker(home);
+    const base = { cwd, sinceMs: 0, agentSessionId: "agent-resume" };
+    await expect(
+      broker.claimExact({
+        ...base,
+        sessionId: "owner",
+        runtimeEpoch: "runtime-1",
+      }),
+    ).resolves.toEqual({ outcome: "claimed", path: rolloutPath });
+    broker.release("owner", "runtime-1");
+    await expect(
+      broker.claimExact({
+        ...base,
+        sessionId: "owner",
+        runtimeEpoch: "runtime-2",
+      }),
+    ).resolves.toEqual({ outcome: "claimed", path: rolloutPath });
+    await expect(
+      broker.claimExact({
+        ...base,
+        sessionId: "foreign",
+        runtimeEpoch: "runtime-3",
+      }),
+    ).resolves.toEqual({ outcome: "pending", path: null });
+  });
+});

packages/harness/src/core/collector/codex-rollout-broker.ts

@@ -0,0 +1,156 @@
+import {
+  findRolloutCandidates,
+  type CodexRolloutCandidate,
+} from "./codex-tailer.js";
+
+export type CodexRolloutClaimResult =
+  | Readonly<{ outcome: "claimed"; path: string }>
+  | Readonly<{ outcome: "pending" | "ambiguous"; path: null }>;
+
+type PendingRuntime = Readonly<{
+  sessionId: string;
+  runtimeEpoch: string;
+  cwd: string;
+  sinceMs: number;
+}>;
+
+const runtimeKey = (sessionId: string, runtimeEpoch: string) =>
+  `${sessionId}\0${runtimeEpoch}`;
+
+/**
+ * Process-epoch rollout ownership for fresh Codex sessions. A path is claimed
+ * at most once. Singleton elimination across every same-root pending launch
+ * handles the common A={a,b}, B={b} race without guessing; an unresolved
+ * many-to-many match remains explicitly ambiguous.
+ */
+export class CodexRolloutBroker {
+  private readonly pending = new Map<string, PendingRuntime>();
+  private readonly assignments = new Map<string, string>();
+  private readonly claimedPaths = new Map<string, string>();
+  private queue: Promise<void> = Promise.resolve();
+
+  constructor(private readonly homeDir?: string) {}
+
+  register(input: PendingRuntime): void {
+    const key = runtimeKey(input.sessionId, input.runtimeEpoch);
+    if (this.assignments.has(key) || this.pending.has(key)) return;
+    this.pending.set(key, { ...input });
+  }
+
+  release(sessionId: string, runtimeEpoch: string): void {
+    const key = runtimeKey(sessionId, runtimeEpoch);
+    this.pending.delete(key);
+    const assigned = this.assignments.get(key);
+    this.assignments.delete(key);
+    // Keep the path tombstone. A rollout is never adopted by another Harness
+    // session, though an exact resume of the same session may reclaim it.
+    void assigned;
+  }
+
+  releaseSession(sessionId: string): void {
+    for (const [key, pending] of this.pending) {
+      if (pending.sessionId === sessionId) this.pending.delete(key);
+    }
+    for (const key of this.assignments.keys()) {
+      if (key.startsWith(`${sessionId}\0`)) this.assignments.delete(key);
+    }
+  }
+
+  async claimExact(
+    input: PendingRuntime & { agentSessionId: string },
+  ): Promise<CodexRolloutClaimResult> {
+    return this.serialized(async () => {
+      const key = runtimeKey(input.sessionId, input.runtimeEpoch);
+      const assigned = this.assignments.get(key);
+      if (assigned) return { outcome: "claimed", path: assigned } as const;
+      const candidates = await findRolloutCandidates({
+        cwd: input.cwd,
+        agentSessionId: input.agentSessionId,
+        homeDir: this.homeDir,
+      });
+      const candidate = candidates.find(({ path }) => {
+        const owner = this.claimedPaths.get(path);
+        return !owner || owner.startsWith(`${input.sessionId}\0`);
+      });
+      if (!candidate) return { outcome: "pending", path: null } as const;
+      this.assign(key, candidate.path, input.sessionId);
+      return { outcome: "claimed", path: candidate.path } as const;
+    });
+  }
+
+  async claimFresh(input: PendingRuntime): Promise<CodexRolloutClaimResult> {
+    this.register(input);
+    return this.serialized(async () => {
+      const key = runtimeKey(input.sessionId, input.runtimeEpoch);
+      const assigned = this.assignments.get(key);
+      if (assigned) return { outcome: "claimed", path: assigned } as const;
+
+      const group = [...this.pending.entries()].filter(
+        ([, candidate]) => candidate.cwd === input.cwd,
+      );
+      const possibilities = new Map<string, CodexRolloutCandidate[]>();
+      for (const [candidateKey, pending] of group) {
+        possibilities.set(
+          candidateKey,
+          await findRolloutCandidates({
+            cwd: pending.cwd,
+            sinceMs: pending.sinceMs,
+            homeDir: this.homeDir,
+            excludePaths: new Set(this.claimedPaths.keys()),
+          }),
+        );
+      }
+
+      // Discovery awaits filesystem I/O. A runtime released during that wait
+      // must not receive a path or leave a tombstone for a later live session.
+      for (const candidateKey of possibilities.keys()) {
+        if (!this.pending.has(candidateKey)) possibilities.delete(candidateKey);
+      }
+
+      let changed = true;
+      while (changed) {
+        changed = false;
+        const singles = [...possibilities.entries()]
+          .filter(([, candidates]) => candidates.length === 1)
+          .sort(([left], [right]) => left.localeCompare(right));
+        for (const [candidateKey, [candidate]] of singles) {
+          if (!candidate || this.claimedPaths.has(candidate.path)) continue;
+          this.assign(candidateKey, candidate.path);
+          possibilities.delete(candidateKey);
+          for (const remaining of possibilities.values()) {
+            const index = remaining.findIndex(
+              ({ path }) => path === candidate.path,
+            );
+            if (index >= 0) remaining.splice(index, 1);
+          }
+          changed = true;
+        }
+      }
+
+      const resolved = this.assignments.get(key);
+      if (resolved) return { outcome: "claimed", path: resolved } as const;
+      const remaining = possibilities.get(key) ?? [];
+      return {
+        outcome: remaining.length > 1 ? "ambiguous" : "pending",
+        path: null,
+      } as const;
+    });
+  }
+
+  private assign(key: string, path: string, resumableSessionId?: string): void {
+    const owner = this.claimedPaths.get(path);
+    if (owner && !owner.startsWith(`${resumableSessionId ?? ""}\0`)) return;
+    this.assignments.set(key, path);
+    this.claimedPaths.set(path, key);
+    this.pending.delete(key);
+  }
+
+  private serialized<T>(operation: () => Promise<T>): Promise<T> {
+    const result = this.queue.catch(() => {}).then(operation);
+    this.queue = result.then(
+      () => undefined,
+      () => undefined,
+    );
+    return result;
+  }
+}