diff --git a/.changeset/owned-child-session-lifecycle.md b/.changeset/owned-child-session-lifecycle.md new file mode 100644 index 000000000..4210a9d27 --- /dev/null +++ b/.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. diff --git a/packages/harness/src/core/collector/codex-rollout-broker.test.ts b/packages/harness/src/core/collector/codex-rollout-broker.test.ts new file mode 100644 index 000000000..1cd11daf4 --- /dev/null +++ b/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 }); + }); +}); diff --git a/packages/harness/src/core/collector/codex-rollout-broker.ts b/packages/harness/src/core/collector/codex-rollout-broker.ts new file mode 100644 index 000000000..98e0d1163 --- /dev/null +++ b/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(); + private readonly assignments = new Map(); + private readonly claimedPaths = new Map(); + private queue: Promise = 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 { + 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 { + 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(); + 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(operation: () => Promise): Promise { + const result = this.queue.catch(() => {}).then(operation); + this.queue = result.then( + () => undefined, + () => undefined, + ); + return result; + } +} diff --git a/packages/harness/src/core/collector/codex-tailer.ts b/packages/harness/src/core/collector/codex-tailer.ts index b319b1e3c..a2d4a771f 100644 --- a/packages/harness/src/core/collector/codex-tailer.ts +++ b/packages/harness/src/core/collector/codex-tailer.ts @@ -269,6 +269,15 @@ export interface FindRolloutFileOptions { agentSessionId?: string; /** Overridable for tests. */ homeDir?: string; + /** Exact paths already owned by another live Harness runtime. */ + excludePaths?: ReadonlySet; +} + +export interface CodexRolloutCandidate { + path: string; + agentSessionId: string | null; + timestampMs: number | null; + mtimeMs: number; } interface RolloutSessionMeta { @@ -342,6 +351,14 @@ async function collectRolloutFiles(dir: string, depth = 0): Promise { * exact rollout path in advance (it's a timestamp+UUID Codex generates itself). */ export async function findRolloutFile(options: FindRolloutFileOptions): Promise { + const candidates = await findRolloutCandidates(options); + return candidates.at(-1)?.path ?? null; +} + +/** Returns every compatible rollout in deterministic chronological order. */ +export async function findRolloutCandidates( + options: FindRolloutFileOptions, +): Promise { const homeDir = options.homeDir ?? homedir(); const root = join(homeDir, ".codex", "sessions"); const files = await collectRolloutFiles(root); @@ -356,21 +373,28 @@ export async function findRolloutFile(options: FindRolloutFileOptions): Promise< // exited, or a test double that never touches the real filesystem). const resolvedCwd = await realpath(options.cwd).catch(() => options.cwd); - let best: { path: string; mtimeMs: number } | null = null; + const candidates: CodexRolloutCandidate[] = []; for (const filePath of files) { + if (options.excludePaths?.has(filePath)) continue; const meta = await readSessionMetaHead(filePath); if (!meta || meta.cwd !== resolvedCwd) continue; if (options.agentSessionId !== undefined) { if (meta.id !== options.agentSessionId) continue; - return filePath; + const fileStat = await stat(filePath).catch(() => null); + if (fileStat) + candidates.push({ path: filePath, agentSessionId: meta.id, + timestampMs: meta.timestampMs, mtimeMs: fileStat.mtimeMs }); + continue; } if (options.sinceMs !== undefined && meta.timestampMs !== null && meta.timestampMs < options.sinceMs) continue; const fileStat = await stat(filePath).catch(() => null); if (!fileStat) continue; - if (!best || fileStat.mtimeMs > best.mtimeMs) best = { path: filePath, mtimeMs: fileStat.mtimeMs }; + candidates.push({ path: filePath, agentSessionId: meta.id, + timestampMs: meta.timestampMs, mtimeMs: fileStat.mtimeMs }); } - return best?.path ?? null; + return candidates.sort((left, right) => + left.mtimeMs - right.mtimeMs || left.path.localeCompare(right.path)); } diff --git a/packages/harness/src/core/errors.ts b/packages/harness/src/core/errors.ts index abfddbb62..695721528 100644 --- a/packages/harness/src/core/errors.ts +++ b/packages/harness/src/core/errors.ts @@ -96,6 +96,30 @@ export class AgentSessionIdentityReservedError extends HarnessError { } } +/** + * A server-owned reserved session ID did not carry the exact private + * coordinator marker. Manual sessions can never satisfy this check by + * matching cwd, title, assignment, or any other public field. + */ +export class SubsessionBindingMismatchError extends HarnessError { + constructor() { + super( + "SUBSESSION_BINDING_MISMATCH", + "The reserved subsession is not owned by this coordinator binding", + ); + } +} + +/** A same-ID fresh start lacked one of its required zero-turn proofs. */ +export class SubsessionFreshRestartForbiddenError extends HarnessError { + constructor() { + super( + "SUBSESSION_FRESH_RESTART_FORBIDDEN", + "The reserved subsession cannot be restarted as a fresh conversation", + ); + } +} + /** * Thrown when an operation requires a harness adapter that has not been * registered. Maps to HTTP 400. diff --git a/packages/harness/src/core/session-manager.test.ts b/packages/harness/src/core/session-manager.test.ts index b534a65d4..2f2d26fa7 100644 --- a/packages/harness/src/core/session-manager.test.ts +++ b/packages/harness/src/core/session-manager.test.ts @@ -18,6 +18,9 @@ import { SessionBackgroundInputPreemptedError, SessionInputIsolationError, SessionManagerClosingError, + SubsessionBindingMismatchError, + SubsessionFreshRestartForbiddenError, + type TrustedSubsessionBindingMarker, ProjectSessionScopeUnavailableError, ProjectBootstrapClaimUnavailableError, SessionManager, @@ -129,6 +132,7 @@ describe("SessionManager", () => { onProjectAgentIdentityMigration?: SessionManagerOptions["onProjectAgentIdentityMigration"]; onProjectBootstrapSession?: SessionManagerOptions["onProjectBootstrapSession"]; onRuntimeEpochTransition?: SessionManagerOptions["onRuntimeEpochTransition"]; + onSubsessionUserClosed?: SessionManagerOptions["onSubsessionUserClosed"]; writeWorkspaceContext?: SessionManagerOptions["writeWorkspaceContext"]; prepareWorkspaceContext?: SessionManagerOptions["prepareWorkspaceContext"]; ensureCanvasTemplate?: SessionManagerOptions["ensureCanvasTemplate"]; @@ -137,6 +141,7 @@ describe("SessionManager", () => { ingestCredentials?: SessionManagerOptions["ingestCredentials"]; writeSessionRegistry?: SessionManagerOptions["writeSessionRegistry"]; writeAgentSessionOwnerRegistry?: SessionManagerOptions["writeAgentSessionOwnerRegistry"]; + writeSubsessionBindingRegistry?: SessionManagerOptions["writeSubsessionBindingRegistry"]; /** Pid given to every fake pty this manager spawns — see createFakePty(). */ fakePid?: number; } = {}, @@ -170,7 +175,7 @@ describe("SessionManager", () => { onProjectAgentIdentityMigration: opts.onProjectAgentIdentityMigration, onProjectBootstrapSession: opts.onProjectBootstrapSession, onRuntimeEpochTransition: opts.onRuntimeEpochTransition, - + onSubsessionUserClosed: opts.onSubsessionUserClosed, writeWorkspaceContext: opts.writeWorkspaceContext, prepareWorkspaceContext: opts.prepareWorkspaceContext, ensureCanvasTemplate: opts.ensureCanvasTemplate, @@ -178,7 +183,7 @@ describe("SessionManager", () => { platform: opts.platform, writeSessionRegistry: opts.writeSessionRegistry, writeAgentSessionOwnerRegistry: opts.writeAgentSessionOwnerRegistry, - + writeSubsessionBindingRegistry: opts.writeSubsessionBindingRegistry, }); managers.push(manager); return { manager, adapter, spawns }; @@ -204,6 +209,414 @@ describe("SessionManager", () => { expect(manager.list()).toHaveLength(1); }); + const marker = ( + sessionId: string, + incarnation = 1, + spawnEpoch = 1, + ): TrustedSubsessionBindingMarker => ({ + projectId: "project_00000000-0000-4000-8000-000000000001", + parentSessionId: "parent-session-1", + bindingId: "binding-1", + sessionId, + incarnation, + spawnEpoch, + }); + + const delegatedCreate = (sessionId: string) => ({ + cwd: "/tmp/proj", + harness: "claude-code" as const, + trusted: { + agentMapIdentity: () => ({ + projectId: "project_00000000-0000-4000-8000-000000000001", + userId: "user-1", + sessionId, + }), + initialTitle: "Collect evidence", + }, + }); + + it("creates a reserved writable session only with its exact private binding", async () => { + const { manager, adapter } = makeManager(); + const sessionId = "00000000-0000-4000-8000-000000000111"; + const input = delegatedCreate(sessionId); + const session = await manager.createReserved( + sessionId, + { cwd: input.cwd, harness: input.harness }, + marker(sessionId), + input.trusted, + ); + + expect(session).toMatchObject({ + id: sessionId, + status: "running", + title: "Collect evidence", + ready: false, + agentMapIdentity: { + projectId: marker(sessionId).projectId, + sessionId, + }, + }); + expect(adapter.launch).toHaveBeenCalledTimes(1); + expect(manager.matchesSubsessionBinding(marker(sessionId))).toBe(true); + expect(await readFile(sessionsPath, "utf8")).not.toContain("binding-1"); + const sidecar = `${sessionsPath}.subsession-bindings.json`; + expect(JSON.parse(await readFile(sidecar, "utf8"))).toMatchObject({ + version: 1, + markers: { [sessionId]: marker(sessionId) }, + closedSessionIds: [], + }); + expect((await stat(sidecar)).mode & 0o777).toBe(0o600); + + const { manager: restartedManager } = makeManager(); + await restartedManager.init(); + expect(restartedManager.getSubsessionBinding(sessionId)).toEqual( + marker(sessionId), + ); + + await expect( + manager.createReserved( + sessionId, + { cwd: input.cwd, harness: input.harness }, + { ...marker(sessionId), bindingId: "foreign-binding" }, + input.trusted, + ), + ).rejects.toBeInstanceOf(SubsessionBindingMismatchError); + expect(adapter.launch).toHaveBeenCalledTimes(1); + }); + + it("never adopts a manual row merely because its reserved id matches", async () => { + const { manager } = makeManager(); + const manual = await manager.create({ + cwd: "/tmp/proj", + harness: "claude-code", + }); + const input = delegatedCreate(manual.id); + await expect( + manager.createReserved( + manual.id, + { cwd: input.cwd, harness: input.harness }, + marker(manual.id), + input.trusted, + ), + ).rejects.toBeInstanceOf(SubsessionBindingMismatchError); + expect(manager.getSubsessionBinding(manual.id)).toBeNull(); + }); + + it("fresh-restarts an exact zero-turn bound row under the same Harness id", async () => { + const { manager, adapter, spawns } = makeManager(); + const sessionId = "00000000-0000-4000-8000-000000000112"; + const input = delegatedCreate(sessionId); + await manager.createReserved( + sessionId, + { cwd: input.cwd, harness: input.harness }, + marker(sessionId), + input.trusted, + ); + spawns[0]!.emitExit(1); + await manager.flush(); + + const restarted = await manager.restartFreshBound( + sessionId, + marker(sessionId), + marker(sessionId, 2, 2), + input.trusted, + async () => false, + ); + expect(restarted).toMatchObject({ id: sessionId, status: "running" }); + expect(manager.list().filter(({ id }) => id === sessionId)).toHaveLength(1); + expect(manager.getSubsessionBinding(sessionId)).toEqual( + marker(sessionId, 2, 2), + ); + expect(adapter.launch).toHaveBeenCalledTimes(2); + }); + + it("resumes an exact coordinator-owned conversation under an advanced marker", async () => { + const { manager, adapter, spawns } = makeManager(); + const sessionId = "00000000-0000-4000-8000-000000000115"; + const input = delegatedCreate(sessionId); + await manager.createReserved( + sessionId, + { cwd: input.cwd, harness: input.harness }, + marker(sessionId), + input.trusted, + ); + const firstRuntime = manager.getRuntimeEpoch(sessionId)!; + await manager.setAgentSessionId( + sessionId, + "agent-session-1", + "startup", + firstRuntime, + ); + spawns[0]!.emitExit(0); + await manager.flush(); + + const resumed = await manager.resumeBound( + sessionId, + marker(sessionId), + marker(sessionId, 2, 2), + ); + + expect(resumed).toMatchObject({ id: sessionId, status: "running" }); + expect(manager.getSubsessionBinding(sessionId)).toEqual( + marker(sessionId, 2, 2), + ); + expect(adapter.resume).toHaveBeenCalledWith( + "agent-session-1", + expect.objectContaining({ harnessSessionId: sessionId }), + ); + expect(manager.list().filter(({ id }) => id === sessionId)).toHaveLength(1); + }); + + it("retries the exact bound resume after the advanced marker outlives a spawn failure", async () => { + const initial = createFakePty(); + const resumedPty = createFakePty(); + const spawnPty = vi.fn() + .mockReturnValueOnce(initial.pty as unknown as ReturnType) + .mockImplementationOnce(() => { throw new Error("resume spawn failed"); }) + .mockReturnValue(resumedPty.pty as unknown as ReturnType); + const { manager, adapter } = makeManager({ spawnPty }); + const sessionId = "00000000-0000-4000-8000-000000000116"; + const input = delegatedCreate(sessionId); + const expected = marker(sessionId); + const next = marker(sessionId, 2, 2); + await manager.createReserved( + sessionId, + { cwd: input.cwd, harness: input.harness }, + expected, + input.trusted, + ); + await manager.setAgentSessionId( + sessionId, + "agent-resume-retry", + "startup", + manager.getRuntimeEpoch(sessionId)!, + ); + initial.emitExit(0); + await manager.flush(); + + await expect(manager.resumeBound(sessionId, expected, next)).rejects.toThrow( + "resume spawn failed", + ); + expect(manager.get(sessionId)?.status).toBe("exited"); + expect(manager.getSubsessionBinding(sessionId)).toEqual(next); + expect(JSON.parse(await readFile(`${sessionsPath}.subsession-bindings.json`, "utf8"))) + .toMatchObject({ markers: { [sessionId]: next } }); + + // Already-next recovery still requires this exact project, parent, binding, + // and next incarnation; it cannot adopt a foreign coordinator's marker. + await expect(manager.resumeBound( + sessionId, + { ...expected, bindingId: "foreign-binding" }, + next, + )).rejects.toBeInstanceOf(SubsessionBindingMismatchError); + await expect(manager.resumeBound(sessionId, expected, marker(sessionId, 3, 3))) + .rejects.toBeInstanceOf(SubsessionBindingMismatchError); + expect(spawnPty).toHaveBeenCalledTimes(2); + + await expect(manager.resumeBound(sessionId, expected, next)).resolves.toMatchObject({ + id: sessionId, + status: "running", + agentSessionId: "agent-resume-retry", + }); + expect(adapter.resume).toHaveBeenCalledTimes(2); + expect(spawnPty).toHaveBeenCalledTimes(3); + expect(manager.getSubsessionBinding(sessionId)).toEqual(next); + expect(manager.list().filter(({ id }) => id === sessionId)).toHaveLength(1); + resumedPty.emitExit(0); + await manager.flush(); + }); + + it("refuses a fresh bound restart when any recorded turn exists", async () => { + const { manager, spawns } = makeManager(); + const sessionId = "00000000-0000-4000-8000-000000000113"; + const input = delegatedCreate(sessionId); + await manager.createReserved( + sessionId, + { cwd: input.cwd, harness: input.harness }, + marker(sessionId), + input.trusted, + ); + spawns[0]!.emitExit(1); + await manager.flush(); + await expect( + manager.restartFreshBound( + sessionId, + marker(sessionId), + marker(sessionId, 2, 2), + input.trusted, + async () => true, + ), + ).rejects.toBeInstanceOf(SubsessionFreshRestartForbiddenError); + expect(manager.getSubsessionBinding(sessionId)).toEqual(marker(sessionId)); + }); + + it("persists an explicit delegated-session close and forbids automatic resurrection", async () => { + const { manager, spawns } = makeManager(); + const sessionId = "00000000-0000-4000-8000-000000000114"; + const input = delegatedCreate(sessionId); + await manager.createReserved( + sessionId, + { cwd: input.cwd, harness: input.harness }, + marker(sessionId), + input.trusted, + ); + const closing = manager.close(sessionId); + spawns[0]!.emitExit(0); + await closing; + expect(manager.wasSubsessionClosedByUser(marker(sessionId))).toBe(true); + await manager.flush(); + await expect( + manager.restartFreshBound( + sessionId, + marker(sessionId), + marker(sessionId, 2, 2), + input.trusted, + async () => false, + ), + ).rejects.toBeInstanceOf(SubsessionFreshRestartForbiddenError); + expect( + JSON.parse( + await readFile(`${sessionsPath}.subsession-bindings.json`, "utf8"), + ), + ).toMatchObject({ closedSessionIds: [sessionId] }); + }); + + it("terminates a delegated PTY even when its user-close tombstone cannot persist", async () => { + let failCloseWrite = false; + const writeSubsessionBindingRegistry = vi.fn(async () => { + if (failCloseWrite) throw new Error("injected close persistence failure"); + }); + const onSubsessionUserClosed = vi.fn(async () => {}); + const { manager, spawns } = makeManager({ + writeSubsessionBindingRegistry, + onSubsessionUserClosed, + }); + const sessionId = "00000000-0000-4000-8000-000000000115"; + const input = delegatedCreate(sessionId); + await manager.createReserved( + sessionId, + { cwd: input.cwd, harness: input.harness }, + marker(sessionId), + input.trusted, + ); + failCloseWrite = true; + + const closing = manager.close(sessionId); + expect(spawns[0]!.pty.kill).toHaveBeenCalledTimes(1); + spawns[0]!.emitExit(0); + await expect(closing).rejects.toThrow("injected close persistence failure"); + expect(manager.get(sessionId)).toMatchObject({ status: "exited" }); + expect(manager.wasSubsessionClosedByUser(marker(sessionId))).toBe(true); + expect(onSubsessionUserClosed).toHaveBeenCalledWith(marker(sessionId)); + + failCloseWrite = false; + await expect(manager.close(sessionId)).resolves.toBe(false); + expect(writeSubsessionBindingRegistry).toHaveBeenCalledTimes(4); + expect(onSubsessionUserClosed).toHaveBeenCalledTimes(2); + }); + + it("prunes durably closed binding proof across release churn and restart", async () => { + const onSubsessionUserClosed = vi.fn(async () => {}); + const { manager, spawns } = makeManager({ onSubsessionUserClosed }); + + for (let index = 0; index < 70; index += 1) { + const sessionId = `00000000-0000-4000-8000-${String(index).padStart(12, "0")}`; + const input = delegatedCreate(sessionId); + await manager.createReserved( + sessionId, + { cwd: input.cwd, harness: input.harness }, + marker(sessionId), + input.trusted, + ); + const closing = manager.closeBound(marker(sessionId)); + spawns[index]!.emitExit(0); + await closing; + expect(manager.getSubsessionBinding(sessionId)).toBeNull(); + } + + expect(onSubsessionUserClosed).toHaveBeenCalledTimes(70); + expect( + JSON.parse( + await readFile(`${sessionsPath}.subsession-bindings.json`, "utf8"), + ), + ).toEqual({ version: 1, markers: {}, closedSessionIds: [] }); + const { manager: restarted } = makeManager({ onSubsessionUserClosed }); + await restarted.init(); + expect(restarted.getSubsessionBinding( + "00000000-0000-4000-8000-000000000000", + )).toBeNull(); + }); + + it("retains exact binding proof when final cleanup fails and prunes it after restart", async () => { + let writeCount = 0; + const writeSubsessionBindingRegistry = vi.fn( + async (file: string, serialized: string) => { + writeCount += 1; + if (writeCount === 3) + throw new Error("injected cleanup persistence failure"); + await writeFile(file, serialized, "utf8"); + }, + ); + const onSubsessionUserClosed = vi.fn(async () => {}); + const { manager, spawns } = makeManager({ + writeSubsessionBindingRegistry, + onSubsessionUserClosed, + }); + const sessionId = "00000000-0000-4000-8000-000000000117"; + const input = delegatedCreate(sessionId); + await manager.createReserved( + sessionId, + { cwd: input.cwd, harness: input.harness }, + marker(sessionId), + input.trusted, + ); + + const closing = manager.closeBound(marker(sessionId)); + spawns[0]!.emitExit(0); + await expect(closing).rejects.toThrow("injected cleanup persistence failure"); + expect(manager.getSubsessionBinding(sessionId)).toEqual(marker(sessionId)); + expect(manager.wasSubsessionClosedByUser(marker(sessionId))).toBe(true); + + const { manager: restarted } = makeManager({ onSubsessionUserClosed }); + await restarted.init(); + expect(restarted.getSubsessionBinding(sessionId)).toEqual(marker(sessionId)); + await expect(restarted.closeBound(marker(sessionId))).resolves.toBe(false); + expect(restarted.getSubsessionBinding(sessionId)).toBeNull(); + expect(writeSubsessionBindingRegistry).toHaveBeenCalledTimes(3); + expect(onSubsessionUserClosed).toHaveBeenCalledTimes(2); + expect( + JSON.parse( + await readFile(`${sessionsPath}.subsession-bindings.json`, "utf8"), + ), + ).toEqual({ version: 1, markers: {}, closedSessionIds: [] }); + }); + + it("closes only an exact coordinator-owned binding through the trusted path", async () => { + const { manager, spawns } = makeManager(); + const sessionId = "00000000-0000-4000-8000-000000000116"; + const input = delegatedCreate(sessionId); + await manager.createReserved( + sessionId, + { cwd: input.cwd, harness: input.harness }, + marker(sessionId), + input.trusted, + ); + + await expect( + manager.closeBound({ ...marker(sessionId), bindingId: "binding_foreign" }), + ).rejects.toBeInstanceOf(SubsessionBindingMismatchError); + expect(spawns[0]!.pty.kill).not.toHaveBeenCalled(); + + const closing = manager.closeBound(marker(sessionId)); + await vi.waitFor(() => + expect(spawns[0]!.pty.kill).toHaveBeenCalledTimes(1), + ); + spawns[0]!.emitExit(0); + await expect(closing).resolves.toBe(true); + await expect(manager.closeBound(marker(sessionId))).resolves.toBe(false); + expect(manager.wasSubsessionClosedByUser(marker(sessionId))).toBe(true); + }); + it("persists sessions to disk and reconciles non-exited sessions to exited on reload", async () => { const { manager } = makeManager(); const session = await manager.create({ diff --git a/packages/harness/src/core/session-manager.ts b/packages/harness/src/core/session-manager.ts index 3e1da372c..c6643075e 100644 --- a/packages/harness/src/core/session-manager.ts +++ b/packages/harness/src/core/session-manager.ts @@ -11,8 +11,10 @@ import { EventEmitter } from "node:events"; import { chmod, mkdir, + open, readFile, rename, + rm, writeFile, } from "node:fs/promises"; import { basename, dirname, join, resolve, sep } from "node:path"; @@ -53,6 +55,8 @@ import { SessionAlreadyLiveError, SessionNotReadyError, SessionNotResumeableError, + SubsessionBindingMismatchError, + SubsessionFreshRestartForbiddenError, UnknownSessionError, } from "./errors.js"; import { listHarnessAdapters } from "./adapters/registry.js"; @@ -68,6 +72,8 @@ export { SessionAlreadyLiveError, SessionNotReadyError, SessionNotResumeableError, + SubsessionBindingMismatchError, + SubsessionFreshRestartForbiddenError, UnknownSessionError, } from "./errors.js"; @@ -100,6 +106,49 @@ function sameProjectAgent( ); } +function parseTrustedSubsessionBindingMarker( + value: unknown, + expectedSessionId?: string, +): TrustedSubsessionBindingMarker | null { + if ( + !isRecord(value) || + Object.keys(value).sort().join(",") !== + "bindingId,incarnation,parentSessionId,projectId,sessionId,spawnEpoch" || + ![value.projectId, value.parentSessionId, value.bindingId, value.sessionId].every( + (entry) => + typeof entry === "string" && + entry.length > 0 && + entry.length <= 256 && + ![...entry].some((character) => { + const point = character.codePointAt(0) ?? 0; + return point <= 0x1f || point === 0x7f; + }), + ) || + (expectedSessionId !== undefined && value.sessionId !== expectedSessionId) || + !Number.isSafeInteger(value.incarnation) || + (value.incarnation as number) < 1 || + !Number.isSafeInteger(value.spawnEpoch) || + (value.spawnEpoch as number) < 1 + ) { + return null; + } + return structuredClone(value) as TrustedSubsessionBindingMarker; +} + +function sameSubsessionBinding( + left: TrustedSubsessionBindingMarker, + right: TrustedSubsessionBindingMarker, +): boolean { + return ( + left.projectId === right.projectId && + left.parentSessionId === right.parentSessionId && + left.bindingId === right.bindingId && + left.sessionId === right.sessionId && + left.incarnation === right.incarnation && + left.spawnEpoch === right.spawnEpoch + ); +} + // node-pty is a native module. Load it lazily so a missing/broken prebuild on // an unsupported platform surfaces as a spawn-time error instead of crashing // the whole server at import time. @@ -266,6 +315,9 @@ const BRACKETED_PASTE_END = "\x1b[201~"; const AGENT_SESSION_OWNER_FILE_VERSION = 1; const AGENT_SESSION_OWNER_MAX_ENTRIES = 50_000; const AGENT_SESSION_OWNER_MAX_BYTES = 4 * 1024 * 1024; +const SUBSESSION_BINDING_FILE_VERSION = 1; +const SUBSESSION_BINDING_MAX_ENTRIES = 8_192; +const SUBSESSION_BINDING_MAX_BYTES = 2 * 1024 * 1024; /** See `recordActivity()`: minimum gap between two `onActivity` broadcasts * for the same session — pty.onData fires per chunk (often many times a * second for a busy TUI), but the SPA's busy indicator only needs "this @@ -419,6 +471,10 @@ export interface SessionManagerOptions { session: HarnessSession, runtimeEpoch: string | null, ) => Promise | void; + /** Mirrors an explicit user close into the coordinator-owned aggregate. */ + onSubsessionUserClosed?: ( + marker: TrustedSubsessionBindingMarker, + ) => Promise | void; /** Revokes launch capabilities/transports after every exit path. */ onAgentMapSessionExit?: (sessionId: string) => void | Promise; now?: () => string; @@ -431,6 +487,11 @@ export interface SessionManagerOptions { file: string, serialized: string, ) => Promise; + /** Fault-injection seam for the private coordinator ownership sidecar. */ + writeSubsessionBindingRegistry?: ( + file: string, + serialized: string, + ) => Promise; /** * Writes HARNESS_CONTEXT_FILE for a session — the caller (server/index.ts's * `writeSessionContext`) owns resolving the session's `boundWorkflowPath` @@ -499,6 +560,11 @@ export interface TrustedSessionResumeOptions { promptAppendix?: string; /** Optional output of serializeFocusedSessionContext; valid only for a project-agent session. */ focusedContext?: FocusedSessionContextProjection; + /** Private two-sided coordinator transition, never accepted by REST. */ + subsessionBindingTransition?: Readonly<{ + expected: TrustedSubsessionBindingMarker; + next: TrustedSubsessionBindingMarker; + }>; } interface PtyHandle { @@ -609,6 +675,16 @@ export type TrackedSessionInputResult = Readonly<{ error?: unknown; }>; +/** Server-private half of a coordinator/session ownership proof. */ +export type TrustedSubsessionBindingMarker = Readonly<{ + projectId: string; + parentSessionId: string; + bindingId: string; + sessionId: string; + incarnation: number; + spawnEpoch: number; +}>; + export type SessionInputWritePhase = | "not-written" @@ -882,6 +958,7 @@ export class SessionManager { private readonly retiredRuntimeEpochs = new Map(); private readonly onRuntimeEpochTransition: SessionManagerOptions["onRuntimeEpochTransition"]; + private readonly onSubsessionUserClosed: SessionManagerOptions["onSubsessionUserClosed"]; private readonly onTerminalInput: ( sessionId: string, @@ -903,6 +980,8 @@ export class SessionManager { * accepted by a HarnessSession. Keeping this outside sessions.json avoids * leaking historical aliases through the browser DTO. */ private readonly agentSessionOwnersPath: string; + /** Never projected through REST; public session fields are not ownership. */ + private readonly subsessionBindingsPath: string; private readonly spawnPty: PtySpawnFn | undefined; private readonly buildLaunchOpts: LaunchOptsBuilder; private readonly resolveAgentMapIdentity: SessionManagerOptions["resolveAgentMapIdentity"]; @@ -917,6 +996,9 @@ export class SessionManager { private readonly writeAgentSessionOwnerRegistry: | ((file: string, serialized: string) => Promise) | undefined; + private readonly writeSubsessionBindingRegistry: + | ((file: string, serialized: string) => Promise) + | undefined; private readonly writeWorkspaceContext: ( session: HarnessSession, ) => Promise; @@ -940,11 +1022,18 @@ export class SessionManager { * state only after the candidate was published or rejected. */ private sessionRegistryIdentityFence: Promise | null = null; private readonly agentSessionOwners = new Map(); + private readonly subsessionBindings = new Map< + string, + TrustedSubsessionBindingMarker + >(); + private readonly userClosedSubsessions = new Set(); /** Serializes the full authorize -> reserve -> pointer commit transition. * A file-level atomic rename alone is insufficient when two starts race the * in-memory ownership check before either write begins. */ private agentSessionIdentityQueue: Promise = Promise.resolve(); private agentSessionOwnerWriteSeq = 0; + private subsessionBindingWriteSeq = 0; + private subsessionBindingQueue: Promise = Promise.resolve(); private initialized = false; constructor(options: SessionManagerOptions) { @@ -959,6 +1048,7 @@ export class SessionManager { options.sessionsPath ?? HARNESS_PATHS.sessions, ); this.agentSessionOwnersPath = `${this.sessionsPath}.agent-session-owners.json`; + this.subsessionBindingsPath = `${this.sessionsPath}.subsession-bindings.json`; this.spawnPty = options.spawnPty; this.loadSpawnPty = options.loadSpawnPty ?? loadDefaultSpawn; this.buildLaunchOpts = options.buildLaunchOpts ?? defaultBuildLaunchOpts; @@ -970,11 +1060,14 @@ export class SessionManager { options.onProjectAgentIdentityMigration; this.onProjectBootstrapSession = options.onProjectBootstrapSession; this.onRuntimeEpochTransition = options.onRuntimeEpochTransition; + this.onSubsessionUserClosed = options.onSubsessionUserClosed; this.now = options.now ?? (() => new Date().toISOString()); this.generateId = options.generateId ?? randomUUID; this.writeSessionRegistry = options.writeSessionRegistry; this.writeAgentSessionOwnerRegistry = options.writeAgentSessionOwnerRegistry; + this.writeSubsessionBindingRegistry = + options.writeSubsessionBindingRegistry; this.writeWorkspaceContext = options.writeWorkspaceContext ?? (async () => {}); this.prepareWorkspaceContext = @@ -1056,6 +1149,7 @@ export class SessionManager { this.sessions.set(session.id, session); } dirty = (await this.loadAgentSessionOwners(persisted)) || dirty; + await this.loadSubsessionBindings(); if (dirty) await this.persist(); } @@ -1187,6 +1281,34 @@ export class SessionManager { if (this.rejectedProjectSessionMetadata.has(id)) { throw new ProjectSessionScopeUnavailableError(id); } + const bindingTransition = trusted.subsessionBindingTransition; + if (bindingTransition) { + const expected = parseTrustedSubsessionBindingMarker( + bindingTransition.expected, + id, + ); + const next = parseTrustedSubsessionBindingMarker( + bindingTransition.next, + id, + ); + const current = this.subsessionBindings.get(id); + if ( + !expected || + !next || + !current || + (!sameSubsessionBinding(current, expected) && + !sameSubsessionBinding(current, next)) || + next.projectId !== expected.projectId || + next.parentSessionId !== expected.parentSessionId || + next.bindingId !== expected.bindingId || + next.sessionId !== expected.sessionId || + next.incarnation !== expected.incarnation + 1 || + next.spawnEpoch <= expected.spawnEpoch || + this.userClosedSubsessions.has(id) + ) { + throw new SubsessionBindingMismatchError(); + } + } const adapter = this.getAdapter(session.harness); // Pre-flight against the agent's OWN store before touching the record. // Holding an agentSessionId only means our SessionStart hook fired once; @@ -1205,6 +1327,20 @@ export class SessionManager { `Sessions that ended before their first prompt are never written to the coding agent's history, so there is nothing to resume — start a new session in this directory instead.`, ); } + if (bindingTransition) { + const current = this.subsessionBindings.get(id)!; + // A failed spawn may leave the exact next marker durably committed. + // Retrying that same transition must not require the old marker again. + if (!sameSubsessionBinding(current, bindingTransition.next)) { + this.subsessionBindings.set(id, bindingTransition.next); + try { + await this.persistSubsessionBindings(); + } catch (error) { + this.subsessionBindings.set(id, current); + throw error; + } + } + } const trustedIdentity = session.agentMapIdentity; const agentMapIdentity = this.resolveAgentMapIdentity ? await this.resolveAgentMapIdentity(id, session.cwd, trustedIdentity) @@ -1309,6 +1445,96 @@ export class SessionManager { return session; } + /** Server-only same-ID resume fenced by the coordinator's private marker. */ + resumeBound( + id: string, + expected: TrustedSubsessionBindingMarker, + next: TrustedSubsessionBindingMarker, + trusted: Omit = {}, + ): Promise { + return this.resume(id, { + ...trusted, + subsessionBindingTransition: { expected, next }, + }); + } + + /** + * Close the session and durably record a user-closed delegated binding. + * Termination starts before storage writes, so a persistence failure cannot + * leave its PTY running. Failed closure bookkeeping retains a tombstone that + * prevents automatic recovery and can be retried by a later close. + * + * Await this operation and handle rejection: binding persistence and the + * coordinator callback can fail, and their completion has no time bound. + * On success, returns kill()'s result: whether a live or stale session was + * transitioned to exited. + */ + async close(id: string): Promise { + const binding = this.subsessionBindings.get(id); + if (binding) { + this.userClosedSubsessions.add(id); + } + // Start termination before persistence so a sidecar fsync failure cannot + // leave a delegated PTY running after the user closes its tab. Keep the + // in-memory tombstone on failure and let a later close retry persistence. + const termination = this.kill(id); + let persistenceError: unknown; + let coordinatorCloseRecorded = false; + if (binding) { + try { + await this.persistSubsessionBindings(); + } catch (error) { + persistenceError = error; + } + try { + if (this.onSubsessionUserClosed) { + await this.onSubsessionUserClosed(binding); + coordinatorCloseRecorded = true; + } + } catch (error) { + persistenceError ??= error; + } + } + const killed = await termination; + if (binding && persistenceError === undefined && coordinatorCloseRecorded) { + const current = this.subsessionBindings.get(id); + if (current && sameSubsessionBinding(current, binding)) { + this.subsessionBindings.delete(id); + this.userClosedSubsessions.delete(id); + try { + await this.persistSubsessionBindings(); + } catch (error) { + this.subsessionBindings.set(id, binding); + this.userClosedSubsessions.add(id); + persistenceError = error; + } + } + } + if (persistenceError !== undefined) throw persistenceError; + return killed; + } + + /** Close only when the caller proves the exact coordinator-owned binding. */ + async closeBound(expected: TrustedSubsessionBindingMarker): Promise { + const parsed = parseTrustedSubsessionBindingMarker( + expected, + expected.sessionId, + ); + if (!parsed) throw new SubsessionBindingMismatchError(); + const operation = async (): Promise => { + const current = this.subsessionBindings.get(parsed.sessionId); + if (!current || !sameSubsessionBinding(current, parsed)) + throw new SubsessionBindingMismatchError(); + return this.close(parsed.sessionId); + }; + const next = this.subsessionBindingQueue.catch(() => {}).then(operation); + this.subsessionBindingQueue = next.then( + () => undefined, + () => undefined, + ); + return next; + } + /** * Signals the session's pty to exit and returns a Promise that resolves * once the process is **actually gone** — not fire-and-forget. @@ -2389,6 +2615,7 @@ export class SessionManager { await Promise.all([...this.projectCreateQueues.values()]); } await this.agentSessionIdentityQueue; + await this.subsessionBindingQueue; await this.writeQueue; } @@ -2888,6 +3115,99 @@ export class SessionManager { await rename(tmpPath, this.agentSessionOwnersPath); } + private async loadSubsessionBindings(): Promise { + let decoded: unknown; + try { + const raw = await readFile(this.subsessionBindingsPath, "utf8"); + if (Buffer.byteLength(raw, "utf8") > SUBSESSION_BINDING_MAX_BYTES) + throw new Error("subsession binding registry exceeds its size limit"); + decoded = JSON.parse(raw) as unknown; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return; + throw error; + } + if ( + !isRecord(decoded) || + Object.keys(decoded).sort().join(",") !== + "closedSessionIds,markers,version" || + decoded.version !== SUBSESSION_BINDING_FILE_VERSION || + !isRecord(decoded.markers) || + !Array.isArray(decoded.closedSessionIds) || + decoded.closedSessionIds.length > SUBSESSION_BINDING_MAX_ENTRIES || + !decoded.closedSessionIds.every( + (sessionId) => typeof sessionId === "string", + ) + ) { + throw new Error("subsession binding registry is malformed"); + } + const entries = Object.entries(decoded.markers); + if (entries.length > SUBSESSION_BINDING_MAX_ENTRIES) + throw new Error("subsession binding registry exceeds its entry limit"); + const bindingIds = new Set(); + for (const [sessionId, value] of entries) { + const marker = parseTrustedSubsessionBindingMarker(value, sessionId); + if (!marker || bindingIds.has(marker.bindingId)) + throw new Error("subsession binding registry is malformed"); + bindingIds.add(marker.bindingId); + this.subsessionBindings.set(sessionId, marker); + } + for (const sessionId of decoded.closedSessionIds) { + if (!this.subsessionBindings.has(sessionId)) + throw new Error("subsession binding registry is malformed"); + this.userClosedSubsessions.add(sessionId); + } + } + + private async persistSubsessionBindings(): Promise { + const markers = Object.fromEntries( + [...this.subsessionBindings.entries()] + .sort(([left], [right]) => left.localeCompare(right)) + .map(([sessionId, marker]) => [sessionId, marker]), + ); + const serialized = `${JSON.stringify( + { + version: SUBSESSION_BINDING_FILE_VERSION, + markers, + closedSessionIds: [...this.userClosedSubsessions].sort(), + }, + null, + 2, + )}\n`; + if (Buffer.byteLength(serialized, "utf8") > SUBSESSION_BINDING_MAX_BYTES) + throw new Error("subsession binding registry exceeds its size limit"); + if (this.writeSubsessionBindingRegistry) { + await this.writeSubsessionBindingRegistry( + this.subsessionBindingsPath, + serialized, + ); + return; + } + const directory = dirname(this.subsessionBindingsPath); + await mkdir(directory, { recursive: true }); + const temporary = `${this.subsessionBindingsPath}.tmp-${process.pid}-${ + this.subsessionBindingWriteSeq++ + }`; + let handle: Awaited> | undefined; + try { + handle = await open(temporary, "wx", 0o600); + await handle.writeFile(serialized, "utf8"); + await handle.sync(); + await handle.close(); + handle = undefined; + await rename(temporary, this.subsessionBindingsPath); + await chmod(this.subsessionBindingsPath, 0o600); + const directoryHandle = await open(directory, "r"); + try { + await directoryHandle.sync(); + } finally { + await directoryHandle.close(); + } + } finally { + await handle?.close().catch(() => {}); + await rm(temporary, { force: true }).catch(() => {}); + } + } + private persistIdentityCandidate(candidate: HarnessSession): Promise { const current = this.list(); const index = current.findIndex((session) => session.id === candidate.id); @@ -2994,13 +3314,98 @@ export class SessionManager { return [...this.pendingCreates.values()]; } + /** + * Server-only reserved-ID create. The private marker is committed before a + * session row or process can exist, closing the row-before-binding crash + * window while preserving the ordinary writable create path. + */ + async createReserved( + reservedSessionId: string, + req: CreateSessionRequest, + markerInput: TrustedSubsessionBindingMarker, + trusted: TrustedSessionCreateOptions, + ): Promise { + const marker = parseTrustedSubsessionBindingMarker( + markerInput, + reservedSessionId, + ); + if (!marker) throw new SubsessionBindingMismatchError(); + const operation = async (): Promise => { + const existingMarker = this.subsessionBindings.get(reservedSessionId); + const existingSession = this.sessions.get(reservedSessionId); + if (existingMarker) { + if (!sameSubsessionBinding(existingMarker, marker)) + throw new SubsessionBindingMismatchError(); + if (this.userClosedSubsessions.has(reservedSessionId)) + throw new SubsessionFreshRestartForbiddenError(); + if (existingSession) return existingSession; + } else { + if (existingSession) throw new SubsessionBindingMismatchError(); + this.subsessionBindings.set(reservedSessionId, marker); + try { + await this.persistSubsessionBindings(); + } catch (error) { + if (this.subsessionBindings.get(reservedSessionId) === marker) + this.subsessionBindings.delete(reservedSessionId); + throw error; + } + } + return this.createWithId(reservedSessionId, req, trusted, marker); + }; + const next = this.subsessionBindingQueue.catch(() => {}).then(operation); + this.subsessionBindingQueue = next.then( + () => undefined, + () => undefined, + ); + return next; + } + + getSubsessionBinding( + sessionId: string, + ): TrustedSubsessionBindingMarker | null { + const marker = this.subsessionBindings.get(sessionId); + return marker ? structuredClone(marker) : null; + } + + matchesSubsessionBinding( + expected: TrustedSubsessionBindingMarker, + ): boolean { + const parsed = parseTrustedSubsessionBindingMarker( + expected, + expected.sessionId, + ); + const current = parsed + ? this.subsessionBindings.get(parsed.sessionId) + : undefined; + return Boolean(parsed && current && sameSubsessionBinding(current, parsed)); + } + + wasSubsessionClosedByUser( + expected: TrustedSubsessionBindingMarker, + ): boolean { + return ( + this.matchesSubsessionBinding(expected) && + this.userClosedSubsessions.has(expected.sessionId) + ); + } private async createWithId( id: string, req: CreateSessionRequest, trusted: TrustedSessionCreateOptions, + expectedSubsessionBinding?: TrustedSubsessionBindingMarker, ): Promise { if (this.closing) throw new SessionManagerClosingError(); + const marker = this.subsessionBindings.get(id); + if ( + (marker !== undefined || expectedSubsessionBinding !== undefined) && + (!marker || + !expectedSubsessionBinding || + !sameSubsessionBinding(marker, expectedSubsessionBinding)) + ) { + throw new SubsessionBindingMismatchError(); + } + if (this.sessions.has(id)) throw new SubsessionBindingMismatchError(); const adapter = this.getAdapter(req.harness); const trustedIdentity = trusted.agentMapIdentity?.(id); const agentMapIdentity = this.resolveAgentMapIdentity @@ -3142,6 +3547,135 @@ export class SessionManager { this.pendingCreates.delete(id); } } + + /** + * Narrow recovery for an exact coordinator-owned row that exited before its + * first turn and has no resumable vendor conversation. The Harness ID stays + * fixed; the private marker advances before a fresh PTY can be admitted. + */ + async restartFreshBound( + id: string, + expected: TrustedSubsessionBindingMarker, + nextInput: TrustedSubsessionBindingMarker, + trusted: TrustedSessionCreateOptions, + hasRecordedTurns: (sessionId: string) => Promise, + ): Promise { + if (this.closing) throw new SessionManagerClosingError(); + const currentExpected = parseTrustedSubsessionBindingMarker(expected, id); + const next = parseTrustedSubsessionBindingMarker(nextInput, id); + const current = this.subsessionBindings.get(id); + const session = this.sessions.get(id); + if ( + !currentExpected || + !next || + !current || + !session || + (current.projectId !== currentExpected.projectId || + current.parentSessionId !== currentExpected.parentSessionId || + current.bindingId !== currentExpected.bindingId || + current.sessionId !== currentExpected.sessionId) || + next.projectId !== currentExpected.projectId || + next.parentSessionId !== currentExpected.parentSessionId || + next.bindingId !== currentExpected.bindingId || + next.sessionId !== currentExpected.sessionId || + next.incarnation !== currentExpected.incarnation + 1 || + next.spawnEpoch <= currentExpected.spawnEpoch || + this.ptys.has(id) || + session.status !== "exited" + ) { + throw new SubsessionBindingMismatchError(); + } + if (this.userClosedSubsessions.has(id)) + throw new SubsessionFreshRestartForbiddenError(); + // A retry may observe the already-advanced marker after the sidecar write + // committed but before the fresh process existed. + if ( + !sameSubsessionBinding(current, currentExpected) && + !sameSubsessionBinding(current, next) + ) { + throw new SubsessionBindingMismatchError(); + } + const adapter = this.getAdapter(session.harness); + if ( + (session.agentSessionId !== null && + (await adapter.canResume(session.agentSessionId, session.cwd))) || + (await hasRecordedTurns(id)) + ) { + throw new SubsessionFreshRestartForbiddenError(); + } + + if (!sameSubsessionBinding(current, next)) { + this.subsessionBindings.set(id, next); + try { + await this.persistSubsessionBindings(); + } catch (error) { + this.subsessionBindings.set(id, current); + throw error; + } + } + + const trustedIdentity = trusted.agentMapIdentity?.(id); + const agentMapIdentity = this.resolveAgentMapIdentity + ? await this.resolveAgentMapIdentity(id, session.cwd, trustedIdentity) + : trustedIdentity; + if ( + !agentMapIdentity || + agentMapIdentity.projectId !== next.projectId || + agentMapIdentity.sessionId !== id + ) { + throw new ProjectSessionScopeUnavailableError(id); + } + + const lastActiveBeforeRestart = session.lastActiveAt; + session.status = "starting"; + session.exitCode = null; + session.exitTail = null; + session.agentSessionId = null; + session.agentMapIdentity = structuredClone(agentMapIdentity); + session.lastActiveAt = this.now(); + let spec: SpawnSpec; + try { + const promptAppendix = trusted.promptAppendix?.(id); + const focusedContext = trusted.focusedContext?.(id); + const sessionStartSystemMessage = + trusted.sessionStartSystemMessage?.(id); + const context = { + ...(promptAppendix ? { promptAppendix } : {}), + ...(focusedContext ? { focusedContext } : {}), + ...(sessionStartSystemMessage + ? { sessionStartSystemMessage } + : {}), + agentMapIdentity, + }; + const opts: LaunchOpts = { + harnessSessionId: id, + cwd: session.cwd, + ...(await this.buildLaunchOpts(id, session, context)), + }; + spec = adapter.launch(opts); + } catch (error) { + session.status = "exited"; + session.lastActiveAt = lastActiveBeforeRestart; + await Promise.resolve(this.onAgentMapSessionExit?.(id)).catch(() => {}); + throw error; + } + try { + await this.persist(); + this.emitStatus(session); + await this.writeWorkspaceContext(session); + await this.ensureCanvasTemplate(session.cwd); + await this.spawn(session, spec, () => + this.revalidateAgentMapIdentity(id, session.cwd, agentMapIdentity), + ); + return session; + } catch (error) { + session.lastActiveAt = lastActiveBeforeRestart; + await this.transitionExited(session, null, { + stampLastActive: false, + }).catch(() => {}); + throw error; + } + } } @@ -3155,3 +3689,7 @@ export class ProjectBootstrapClaimUnavailableError extends Error { this.name = "ProjectBootstrapClaimUnavailableError"; } } + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} diff --git a/packages/harness/src/index.ts b/packages/harness/src/index.ts index f9ea50bc8..39180dd40 100644 --- a/packages/harness/src/index.ts +++ b/packages/harness/src/index.ts @@ -49,6 +49,8 @@ export { SessionNotReadyError, SessionNotResumeableError, SessionAlreadyLiveError, + SubsessionBindingMismatchError, + SubsessionFreshRestartForbiddenError, AdapterNotFoundError, ExternalHarnessError, SpawnTargetError, diff --git a/packages/harness/src/server/codex-tailer-wiring.test.ts b/packages/harness/src/server/codex-tailer-wiring.test.ts index feed12d06..30e847b0c 100644 --- a/packages/harness/src/server/codex-tailer-wiring.test.ts +++ b/packages/harness/src/server/codex-tailer-wiring.test.ts @@ -13,11 +13,12 @@ import { join } from "node:path"; vi.mock("../core/collector/codex-tailer.js", () => ({ tailCodexRollout: vi.fn(), - findRolloutFile: vi.fn(), + findRolloutCandidates: vi.fn(), })); -import { tailCodexRollout, findRolloutFile, type CodexEventListener } from "../core/collector/codex-tailer.js"; +import { tailCodexRollout, findRolloutCandidates, type CodexEventListener } from "../core/collector/codex-tailer.js"; import { startServer, type HarnessServer } from "./index.js"; +import { CodexRolloutBroker } from "../core/collector/codex-rollout-broker.js"; import type { HarnessAdapter, LaunchOpts, SpawnSpec } from "../shared/types.js"; type FakeTailerHandle = { stop: ReturnType; emitSessionEnd: ReturnType }; @@ -62,16 +63,24 @@ describe("codex tailer lifecycle wiring", () => { let server: HarnessServer | undefined; let fakeHandle: FakeTailerHandle; let lastTailerOnEvent: CodexEventListener | undefined; + let tailerEvents: CodexEventListener[]; beforeEach(async () => { dir = await mkdtemp(join(tmpdir(), "harness-codex-wiring-")); fakeHandle = { stop: vi.fn(), emitSessionEnd: vi.fn() }; + tailerEvents = []; vi.mocked(tailCodexRollout).mockReset().mockImplementation((opts) => { lastTailerOnEvent = opts.onEvent; + tailerEvents.push(opts.onEvent); return fakeHandle; }); - vi.mocked(findRolloutFile).mockReset().mockResolvedValue("/fake/rollout/path.jsonl"); + vi.mocked(findRolloutCandidates).mockReset().mockResolvedValue([{ + path: "/fake/rollout/path.jsonl", + agentSessionId: "agent-fixture", + timestampMs: Date.now(), + mtimeMs: Date.now(), + }]); }); afterEach(async () => { @@ -85,6 +94,7 @@ describe("codex tailer lifecycle wiring", () => { await server?.close(); await server?.sessionManager.flush(); server = undefined; + vi.restoreAllMocks(); await rm(dir, { recursive: true, force: true }); }); @@ -103,16 +113,16 @@ describe("codex tailer lifecycle wiring", () => { expect(session.status).toBe("running"); await vi.waitFor(() => { - expect(findRolloutFile).toHaveBeenCalled(); + expect(findRolloutCandidates).toHaveBeenCalled(); expect(tailCodexRollout).toHaveBeenCalledWith( expect.objectContaining({ rolloutPath: "/fake/rollout/path.jsonl" }), ); }); - // findRolloutFile should have been asked for this session's cwd, and (a + // Rollout discovery should have been asked for this session's cwd, and (a // fresh launch has no agentSessionId yet) bounded by sinceMs rather than // an exact id. - expect(findRolloutFile).toHaveBeenCalledWith( + expect(findRolloutCandidates).toHaveBeenCalledWith( expect.objectContaining({ cwd, sinceMs: expect.any(Number) }), ); @@ -140,6 +150,89 @@ describe("codex tailer lifecycle wiring", () => { // too, the outer test timeout could fire first and mask it entirely. }, 15_000); + it("releases a timed-out discovery so the next same-root runtime owns its rollout", async () => { + const cwd = join(dir, "project"); + const register = vi.spyOn(CodexRolloutBroker.prototype, "register"); + server = await startServer({ + port: 0, + bootToken: "test-token", + telemetryOptIn: false, + autoCreateSession: false, + adapters: { codex: fakeCodexAdapter() }, + stateRoot: dir, + }); + // Advance only the discovery deadline; keep real timer scheduling and PTY + // cleanup so this proves timeout handling without waiting fifteen seconds. + const realNow = Date.now.bind(Date); + let discoveryAdvance = 0; + vi.spyOn(Date, "now").mockImplementation(() => realNow() + discoveryAdvance); + vi.mocked(findRolloutCandidates).mockImplementation(async () => { + discoveryAdvance += 16_000; + return []; + }); + const session = await server.sessionManager.create({ cwd, harness: "codex" }); + const runtimeEpoch = server.sessionManager.getRuntimeEpoch(session.id)!; + await vi.waitFor(() => expect(server!.sessionManager.getAdapterIdentityState( + session.id, + runtimeEpoch, + )).toBe("unavailable")); + vi.mocked(Date.now).mockRestore(); + expect(tailCodexRollout).not.toHaveBeenCalled(); + vi.mocked(findRolloutCandidates).mockResolvedValue([{ + path: "/fake/rollout/path.jsonl", + agentSessionId: "agent-next", + timestampMs: Date.now(), + mtimeMs: Date.now(), + }]); + + const broker = register.mock.contexts[0] as CodexRolloutBroker; + // UUIDs sort before this later id. A leaked pending registration would + // receive the singleton candidate and strand this new runtime. + await expect(broker.claimFresh({ + sessionId: "zz-next-session", + runtimeEpoch: "next-runtime", + cwd, + sinceMs: Date.now(), + })).resolves.toEqual({ outcome: "claimed", path: "/fake/rollout/path.jsonl" }); + await server.sessionManager.kill(session.id); + }, 15_000); + + it("releases every pending epoch on an epochless final exit", async () => { + const cwd = join(dir, "project"); + const register = vi.spyOn(CodexRolloutBroker.prototype, "register"); + server = await startServer({ + port: 0, + bootToken: "test-token", + telemetryOptIn: false, + autoCreateSession: false, + adapters: { codex: fakeCodexAdapter() }, + stateRoot: dir, + }); + const session = await server.sessionManager.create({ cwd, harness: "codex" }); + await vi.waitFor(() => expect(tailCodexRollout).toHaveBeenCalled()); + await server.sessionManager.kill(session.id); + const broker = register.mock.contexts[0] as CodexRolloutBroker; + broker.register({ sessionId: session.id, runtimeEpoch: "orphaned-epoch", cwd, sinceMs: 0 }); + // A stale non-exited registry row has no live PTY. Its close path emits an + // exited status with runtimeEpoch=null and must retire all prior epochs. + session.status = "running"; + expect(server.sessionManager.getRuntimeEpoch(session.id)).toBeNull(); + await server.sessionManager.kill(session.id); + expect(session.status).toBe("exited"); + vi.mocked(findRolloutCandidates).mockResolvedValue([{ + path: "/fake/rollout/next.jsonl", + agentSessionId: "agent-next", + timestampMs: Date.now(), + mtimeMs: Date.now(), + }]); + await expect(broker.claimFresh({ + sessionId: "zz-next-session", + runtimeEpoch: "next-runtime", + cwd, + sinceMs: Date.now(), + })).resolves.toEqual({ outcome: "claimed", path: "/fake/rollout/next.jsonl" }); + }, 15_000); + it("resumes by exact agentSessionId rather than cwd+sinceMs when one is already known", async () => { const cwd = join(dir, "project"); server = await startServer({ @@ -162,7 +255,7 @@ describe("codex tailer lifecycle wiring", () => { const resumed = await server.sessionManager.resume(historical.id); await vi.waitFor(() => { - expect(findRolloutFile).toHaveBeenCalledWith( + expect(findRolloutCandidates).toHaveBeenCalledWith( expect.objectContaining({ cwd, agentSessionId: "agent-resumed" }), ); }); @@ -176,6 +269,60 @@ describe("codex tailer lifecycle wiring", () => { // See the first test's comment: the outer test timeout needs the same bump. }, 15_000); + it("rejects a late event from the stopped tailer after the same session resumes", async () => { + const cwd = join(dir, "project"); + server = await startServer({ + port: 0, + bootToken: "test-token", + telemetryOptIn: false, + autoCreateSession: false, + adapters: { codex: fakeCodexAdapter() }, + stateRoot: dir, + }); + + const session = await server.sessionManager.create({ cwd, harness: "codex" }); + await vi.waitFor(() => expect(tailerEvents).toHaveLength(1)); + const firstRuntimeEvent = tailerEvents[0]!; + firstRuntimeEvent("SessionStart", { + source: "codex", + cwd, + session_id: "provider-shared", + }); + await vi.waitFor(() => { + expect(server!.sessionManager.get(session.id)?.agentSessionId).toBe( + "provider-shared", + ); + expect(server!.sessionManager.get(session.id)?.ready).toBe(true); + }); + + await server.sessionManager.kill(session.id); + await vi.waitFor(() => + expect(server!.sessionManager.get(session.id)?.status).toBe("exited"), + ); + await server.sessionManager.resume(session.id); + await vi.waitFor(() => expect(tailerEvents).toHaveLength(2)); + expect(server.sessionManager.get(session.id)?.ready).toBe(false); + + firstRuntimeEvent("SessionStart", { + source: "codex", + cwd, + session_id: "provider-shared", + }); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(server.sessionManager.get(session.id)?.ready).toBe(false); + + tailerEvents[1]!("SessionStart", { + source: "codex", + cwd, + session_id: "provider-shared", + }); + await vi.waitFor(() => + expect(server!.sessionManager.get(session.id)?.ready).toBe(true), + ); + + await server.sessionManager.kill(session.id); + }, 15_000); + it("stops the tailer and does not start a new one for a non-codex session", async () => { server = await startServer({ port: 0, @@ -206,7 +353,7 @@ describe("codex tailer lifecycle wiring", () => { // Give any (incorrect) codex wiring a chance to fire before asserting it didn't. await new Promise((resolve) => setTimeout(resolve, 50)); - expect(findRolloutFile).not.toHaveBeenCalled(); + expect(findRolloutCandidates).not.toHaveBeenCalled(); expect(tailCodexRollout).not.toHaveBeenCalled(); // See the first test's comment: wait for the real spawned process to diff --git a/packages/harness/src/server/index.ts b/packages/harness/src/server/index.ts index 6564e5fb2..cd9b36ecf 100644 --- a/packages/harness/src/server/index.ts +++ b/packages/harness/src/server/index.ts @@ -7,6 +7,7 @@ * src/shared/types.ts for the full protocol contract. */ +import { CodexRolloutBroker } from "../core/collector/codex-rollout-broker.js"; import { AgentBriefService } from "../core/agent-brief-service.js"; import { BuildPlanStore } from "../core/build-plan-store.js"; import { BuildPlanService } from "../core/build-plan-service.js"; @@ -97,7 +98,7 @@ import { migrateHarnessIdentity } from "../core/collector/identity-migration.js" import { normalizeHookEvent } from "../core/collector/normalizer.js"; import { enrichTurnCompleted } from "../core/collector/transcript.js"; import { createSeqCounter } from "../core/collector/seq.js"; -import { findRolloutFile, tailCodexRollout, type CodexTailerHandle } from "../core/collector/codex-tailer.js"; +import { tailCodexRollout, type CodexTailerHandle } from "../core/collector/codex-tailer.js"; import { getOrCreateMachineId } from "../cli/machine-id.js"; import { loadSettings, pruneDeadRecentDirs } from "../cli/settings.js"; import type { HarnessIdentity } from "../cli/auth.js"; @@ -694,6 +695,7 @@ export const startServer = async ( organizationName: identity?.organizationName ?? null, }); const statePaths = resolveStatePaths(options.stateRoot); + const codexRolloutBroker = new CodexRolloutBroker(options.codexHomeDir); const projectBootstrapOutbox = new ProjectBootstrapOutbox( join(statePaths.projectBootstrap, "project-outbox"), ); @@ -1505,6 +1507,18 @@ export const startServer = async ( onTerminalInput: (sessionId, context) => projectBootstrap?.onTerminalInput(sessionId, context), onRuntimeEpochTransition: async (session, runtimeEpoch) => { + if (adapters[session.harness]?.eventSource === "transcript-tail") { + if (runtimeEpoch) { + codexRolloutBroker.register({ + sessionId: session.id, + runtimeEpoch, + cwd: session.cwd, + sinceMs: Date.now(), + }); + } else { + codexRolloutBroker.releaseSession(session.id); + } + } if (!session.projectBootstrap) return; if (!projectBootstrap) { throw new Error("project bootstrap coordinator unavailable"); @@ -4137,25 +4151,34 @@ export const startServer = async ( async function discoverCodexRolloutPath( session: HarnessSession, - ): Promise { + runtimeEpoch: string, + ): Promise<{ path: string | null; ambiguous: boolean }> { const deadline = Date.now() + CODEX_ROLLOUT_DISCOVERY_TIMEOUT_MS; const sinceMs = Date.parse(session.createdAt); + let ambiguous = false; for (;;) { - const found = await findRolloutFile( - session.agentSessionId - ? { - cwd: session.cwd, - agentSessionId: session.agentSessionId, - homeDir: options.codexHomeDir, - } - : { - cwd: session.cwd, - sinceMs: Number.isNaN(sinceMs) ? undefined : sinceMs, - homeDir: options.codexHomeDir, - }, - ); - if (found) return found; - if (Date.now() >= deadline) return null; + // An exited or replaced runtime must not re-register itself on the next + // discovery poll after lifecycle cleanup released its ownership. + if (!sessionManager.isCurrentRuntimeEpoch(session.id, runtimeEpoch)) + return { path: null, ambiguous }; + const claim = session.agentSessionId + ? await codexRolloutBroker.claimExact({ + sessionId: session.id, + runtimeEpoch, + cwd: session.cwd, + sinceMs: Number.isNaN(sinceMs) ? Date.now() : sinceMs, + agentSessionId: session.agentSessionId, + }) + : await codexRolloutBroker.claimFresh({ + sessionId: session.id, + runtimeEpoch, + cwd: session.cwd, + sinceMs: Number.isNaN(sinceMs) ? Date.now() : sinceMs, + }); + if (claim.outcome === "claimed") + return { path: claim.path, ambiguous: false }; + if (claim.outcome === "ambiguous") ambiguous = true; + if (Date.now() >= deadline) return { path: null, ambiguous }; await new Promise((resolve) => setTimeout(resolve, CODEX_ROLLOUT_DISCOVERY_POLL_MS), ); @@ -4165,21 +4188,35 @@ export const startServer = async ( async function startCodexTailerFor(harnessSessionId: string): Promise { if (codexTailers.has(harnessSessionId)) return; const session = sessionManager.get(harnessSessionId); - if (!session) return; - const runtimeEpoch = sessionManager.getRuntimeEpoch(harnessSessionId); - if (runtimeEpoch === null) return; - const rolloutPath = await discoverCodexRolloutPath(session); + if (!session || runtimeEpoch === null) return; + + const discovery = await discoverCodexRolloutPath(session, runtimeEpoch); + const rolloutPath = discovery.path; if (!rolloutPath) { + codexRolloutBroker.release(harnessSessionId, runtimeEpoch); + if (!sessionManager.isCurrentRuntimeEpoch(harnessSessionId, runtimeEpoch)) + return; + sessionManager.setAdapterIdentityState( + harnessSessionId, + runtimeEpoch, + discovery.ambiguous ? "ambiguous" : "unavailable", + ); console.error( - `[harness] codex tailer: no rollout file found for session ${harnessSessionId} (cwd=${session.cwd}) within ${CODEX_ROLLOUT_DISCOVERY_TIMEOUT_MS}ms`, + `[harness] codex tailer: rollout identity ${discovery.ambiguous ? "ambiguous" : "unavailable"} for session ${harnessSessionId} within ${CODEX_ROLLOUT_DISCOVERY_TIMEOUT_MS}ms`, ); return; } // The session may have exited (or already started another tailer via a // status-change re-entry) while discovery was polling. if (codexTailers.has(harnessSessionId)) return; - if (!sessionManager.isCurrentRuntimeEpoch(harnessSessionId, runtimeEpoch)) return; + if ( + sessionManager.get(harnessSessionId)?.status !== "running" || + !sessionManager.isCurrentRuntimeEpoch(harnessSessionId, runtimeEpoch) + ) { + codexRolloutBroker.release(harnessSessionId, runtimeEpoch); + return; + } const tailer = tailCodexRollout({ rolloutPath, @@ -4206,9 +4243,10 @@ export const startServer = async ( console.error("[harness] codex tailer parse error:", err), }); codexTailers.set(harnessSessionId, tailer); + sessionManager.setAdapterIdentityState(harnessSessionId, runtimeEpoch, "ready"); } - sessionManager.onStatusChange((session) => { + sessionManager.onStatusChange((session, context) => { // The codex tailer is only needed for harnesses whose analytics come // from the rollout file (eventSource: "transcript-tail"). Harnesses with // eventSource: "hooks" (claude-code) drive the same pipeline via real @@ -4218,9 +4256,18 @@ export const startServer = async ( if (adapters[session.harness]?.eventSource !== "transcript-tail") return; if (session.status === "running") { startCodexTailerFor(session.id).catch((err: unknown) => { + if (context.runtimeEpoch) { + codexRolloutBroker.release(session.id, context.runtimeEpoch); + sessionManager.setAdapterIdentityState( + session.id, + context.runtimeEpoch, + "unavailable", + ); + } console.error("[harness] codex tailer startup failed:", err); }); } else if (session.status === "exited") { + codexRolloutBroker.releaseSession(session.id); const tailer = codexTailers.get(session.id); if (tailer) { tailer.emitSessionEnd( diff --git a/packages/harness/src/server/rest.ts b/packages/harness/src/server/rest.ts index ac63ef739..acc5be74a 100644 --- a/packages/harness/src/server/rest.ts +++ b/packages/harness/src/server/rest.ts @@ -909,14 +909,18 @@ export function createRestRouter(options: RestRouterOptions): Router { } }); - router.delete("/sessions/:id", (req, res) => { + router.delete("/sessions/:id", async (req, res, next) => { const existed = sessionManager.get(req.params.id) !== undefined; if (!existed) { res.status(404).json({ error: "session not found" }); return; } - void sessionManager.kill(req.params.id); - res.json({ ok: true }); + try { + await sessionManager.close(req.params.id); + res.json({ ok: true }); + } catch (error) { + next(error); + } }); router.post("/sessions/:id/input", async (req, res, next) => {