From 3d105bcac3517691cefb893382a54633204f8229 Mon Sep 17 00:00:00 2001 From: Yash Date: Sat, 5 Sep 2026 12:12:00 +0000 Subject: [PATCH] fix(harness): track session input delivery [Agent Map 04/15] --- .changeset/ordinary-session-input.md | 9 + .../src/core/ingest-credentials.test.ts | 32 +- .../harness/src/core/ingest-credentials.ts | 45 +- .../harness/src/core/session-manager.test.ts | 868 +++++++++++++++++- packages/harness/src/core/session-manager.ts | 812 ++++++++++++++-- packages/harness/src/core/task-manager.ts | 2 +- packages/harness/src/server/index.ts | 16 +- packages/harness/src/server/ingest.test.ts | 217 ++++- packages/harness/src/server/ingest.ts | 72 +- .../harness/src/server/terminal-ws.test.ts | 109 +++ packages/harness/src/server/terminal-ws.ts | 46 +- 11 files changed, 2074 insertions(+), 154 deletions(-) create mode 100644 .changeset/ordinary-session-input.md create mode 100644 packages/harness/src/server/terminal-ws.test.ts diff --git a/.changeset/ordinary-session-input.md b/.changeset/ordinary-session-input.md new file mode 100644 index 000000000..53c3b0705 --- /dev/null +++ b/.changeset/ordinary-session-input.md @@ -0,0 +1,9 @@ +--- +"@sapiom/harness": minor +--- + +Track ordinary session input delivery and runtime ownership so partial input, preemption, stale ingest events, and shutdown are handled consistently. + +Background input now yields to terminal keystrokes received while its durable pre-write hook is pending. A submission displaced before writing compensates its durable claim so it can be recovered safely. + +**Breaking for embedders** (minor while the package is pre-1.0): `SessionManager.write()` can throw an isolation error with code `SESSION_INPUT_ISOLATION_REQUIRED` when prior partial input cannot be cleared. Callers forwarding terminal bytes should handle this failure and keep the terminal available for a later retry instead of assuming every call returns a boolean. diff --git a/packages/harness/src/core/ingest-credentials.test.ts b/packages/harness/src/core/ingest-credentials.test.ts index 22c90166f..e974ffb3f 100644 --- a/packages/harness/src/core/ingest-credentials.test.ts +++ b/packages/harness/src/core/ingest-credentials.test.ts @@ -5,26 +5,38 @@ import { IngestCredentialRegistry } from "./ingest-credentials.js"; describe("IngestCredentialRegistry", () => { it("binds an opaque credential to exactly one session", () => { const tokens = ["token-a", "token-b"]; - const registry = new IngestCredentialRegistry(() => tokens.shift()!); + const epochs = ["epoch-a", "epoch-b"]; + const registry = new IngestCredentialRegistry( + () => tokens.shift()!, + () => epochs.shift()!, + ); const first = registry.issue("session-a"); const second = registry.issue("session-b"); - expect(registry.authenticate("session-a", first)).toBe(true); - expect(registry.authenticate("session-b", second)).toBe(true); - expect(registry.authenticate("session-b", first)).toBe(false); - expect(registry.authenticate("session-a", second)).toBe(false); - expect(registry.authenticate("unknown", first)).toBe(false); + expect(first).toEqual({ token: "token-a", runtimeEpoch: "epoch-a" }); + expect(second).toEqual({ token: "token-b", runtimeEpoch: "epoch-b" }); + expect(registry.authenticate("session-a", first.token)).toBe("epoch-a"); + expect(registry.authenticate("session-b", second.token)).toBe("epoch-b"); + expect(registry.authenticate("session-b", first.token)).toBeNull(); + expect(registry.authenticate("session-a", second.token)).toBeNull(); + expect(registry.authenticate("unknown", first.token)).toBeNull(); }); it("rotates on a new launch and revokes on terminal cleanup", () => { const tokens = ["old-token", "new-token"]; - const registry = new IngestCredentialRegistry(() => tokens.shift()!); + const epochs = ["old-epoch", "new-epoch"]; + const registry = new IngestCredentialRegistry( + () => tokens.shift()!, + () => epochs.shift()!, + ); const oldToken = registry.issue("session-a"); const newToken = registry.issue("session-a"); - expect(registry.authenticate("session-a", oldToken)).toBe(false); - expect(registry.authenticate("session-a", newToken)).toBe(true); + expect(registry.authenticate("session-a", oldToken.token)).toBeNull(); + expect(registry.authenticate("session-a", newToken.token)).toBe( + "new-epoch", + ); registry.revoke("session-a"); - expect(registry.authenticate("session-a", newToken)).toBe(false); + expect(registry.authenticate("session-a", newToken.token)).toBeNull(); }); }); diff --git a/packages/harness/src/core/ingest-credentials.ts b/packages/harness/src/core/ingest-credentials.ts index 9cb3ee164..fd90458f2 100644 --- a/packages/harness/src/core/ingest-credentials.ts +++ b/packages/harness/src/core/ingest-credentials.ts @@ -1,12 +1,19 @@ import { createHash, randomBytes, timingSafeEqual } from "node:crypto"; export interface IngestCredentialProvider { - /** Issues a new opaque capability and invalidates any prior one for this id. */ - issue(sessionId: string): string; - authenticate(sessionId: string, token: string): boolean; + /** Issues a new opaque capability/runtime epoch and invalidates any prior one. */ + issue(sessionId: string): IssuedIngestCredential; + /** Returns the server-owned runtime epoch for this exact capability. */ + authenticate(sessionId: string, token: string): string | null; revoke(sessionId: string): void; } +export interface IssuedIngestCredential { + token: string; + /** Opaque process-local identity for the PTY generation receiving `token`. */ + runtimeEpoch: string; +} + const EMPTY_DIGEST = Buffer.alloc(32); function digest(token: string): Buffer { @@ -20,31 +27,43 @@ function digest(token: string): Buffer { * the registry intentionally does not persist. */ export class IngestCredentialRegistry implements IngestCredentialProvider { - private readonly digests = new Map(); + private readonly credentials = new Map< + string, + { digest: Buffer; runtimeEpoch: string } + >(); constructor( private readonly generateToken: () => string = () => randomBytes(32).toString("base64url"), + private readonly generateRuntimeEpoch: () => string = () => + randomBytes(16).toString("base64url"), ) {} - issue(sessionId: string): string { + issue(sessionId: string): IssuedIngestCredential { if (!sessionId) throw new Error("ingest credential requires a session id"); const token = this.generateToken(); if (!token || token.length > 512) { throw new Error("invalid generated ingest credential"); } - this.digests.set(sessionId, digest(token)); - return token; + const runtimeEpoch = this.generateRuntimeEpoch(); + if (!runtimeEpoch || runtimeEpoch.length > 128) { + throw new Error("invalid generated ingest runtime epoch"); + } + this.credentials.set(sessionId, { digest: digest(token), runtimeEpoch }); + return { token, runtimeEpoch }; } - authenticate(sessionId: string, token: string): boolean { - if (!sessionId || !token || token.length > 512) return false; - const expected = this.digests.get(sessionId); - const matches = timingSafeEqual(expected ?? EMPTY_DIGEST, digest(token)); - return expected !== undefined && matches; + authenticate(sessionId: string, token: string): string | null { + if (!sessionId || !token || token.length > 512) return null; + const expected = this.credentials.get(sessionId); + const matches = timingSafeEqual( + expected?.digest ?? EMPTY_DIGEST, + digest(token), + ); + return expected !== undefined && matches ? expected.runtimeEpoch : null; } revoke(sessionId: string): void { - this.digests.delete(sessionId); + this.credentials.delete(sessionId); } } diff --git a/packages/harness/src/core/session-manager.test.ts b/packages/harness/src/core/session-manager.test.ts index fe1565754..2d60a031d 100644 --- a/packages/harness/src/core/session-manager.test.ts +++ b/packages/harness/src/core/session-manager.test.ts @@ -14,6 +14,9 @@ import { CodexAdapter } from "./adapters/codex.js"; import { ExternalHarnessError, SessionNotResumeableError } from "./errors.js"; import { SessionInputGuardRejectedError, + SessionBackgroundInputPreemptedError, + SessionInputIsolationError, + SessionManagerClosingError, ProjectSessionScopeUnavailableError, SessionManager, sanitizeExitTail, @@ -80,6 +83,18 @@ function createFakeAdapter(overrides: Partial = {}): HarnessAdap }; } +function deferred(): { + promise: Promise; + resolve: (value: T) => void; +} { + let resolve!: (value: T) => void; + const promise = new Promise((resolvePromise) => { + resolve = resolvePromise; + }); + return { promise, resolve }; +} + + describe("SessionManager", () => { let dir: string; let sessionsPath: string; @@ -103,6 +118,9 @@ describe("SessionManager", () => { opts: { adapter?: HarnessAdapter; spawnPty?: PtySpawnFn; + onRuntimeEpochTransition?: SessionManagerOptions["onRuntimeEpochTransition"]; + onTerminalInput?: SessionManagerOptions["onTerminalInput"]; + loadSpawnPty?: SessionManagerOptions["loadSpawnPty"]; buildLaunchOpts?: SessionManagerOptions["buildLaunchOpts"]; resolveAgentMapIdentity?: SessionManagerOptions["resolveAgentMapIdentity"]; onAgentMapSessionExit?: SessionManagerOptions["onAgentMapSessionExit"]; @@ -122,7 +140,7 @@ describe("SessionManager", () => { ) { const adapter = opts.adapter ?? createFakeAdapter(); const spawns: ReturnType[] = []; - const spawnPty: PtySpawnFn = + const spawnPty: PtySpawnFn | undefined = opts.loadSpawnPty ? undefined : opts.spawnPty ?? ((file, args) => { const fake = createFakePty(opts.fakePid); @@ -139,6 +157,9 @@ describe("SessionManager", () => { new IngestCredentialRegistry(() => "boot-token"), sessionsPath, spawnPty, + loadSpawnPty: opts.loadSpawnPty, + onTerminalInput: opts.onTerminalInput, + onRuntimeEpochTransition: opts.onRuntimeEpochTransition, buildLaunchOpts: opts.buildLaunchOpts, resolveAgentMapIdentity: opts.resolveAgentMapIdentity, onAgentMapSessionExit: opts.onAgentMapSessionExit, @@ -2843,10 +2864,10 @@ describe("SessionManager", () => { }); it("issues and revokes a capability for the exact spawned session", async () => { - const issue = vi.fn((id: string) => `token-for:${id}`); + const issue = vi.fn((id: string) => ({ token: `token-for:${id}`, runtimeEpoch: `epoch-for:${id}` })); const revoke = vi.fn(); const { manager, spawns } = makeManager({ - ingestCredentials: { issue, revoke, authenticate: () => false }, + ingestCredentials: { issue, revoke, authenticate: () => null }, }); const session = await manager.create({ cwd: "/tmp/proj", @@ -3203,4 +3224,845 @@ describe("SessionManager", () => { expect((caught as ExternalHarnessError).code).toBe("HARNESS_EXTERNAL"); }); }); + + + it("reports exact input write phases and kills only an exact runtime", async () => { + const { manager, spawns } = makeManager(); + const session = await manager.create({ + cwd: "/tmp/proj", + harness: "claude-code", + }); + const runtime = manager.getRuntimeEpoch(session.id)!; + manager.setReady(session.id, runtime); + + const submitted = await manager.submitInputTracked( + session.id, + "Implement the scoped task", + ); + expect(submitted).toEqual({ accepted: true, phase: "enter-written" }); + expect(await manager.killIfRuntime(session.id, "foreign-runtime")).toBe( + false, + ); + expect(spawns[0]!.pty.kill).not.toHaveBeenCalled(); + + spawns[0]!.pty.write.mockImplementationOnce(() => { + throw new Error("ambiguous write"); + }); + const ambiguous = await manager.submitInputTracked( + session.id, + "Retry-sensitive task", + ); + expect(ambiguous).toMatchObject({ + accepted: false, + phase: "text-staged", + error: expect.any(Error), + }); + }); + + + it("closes PTY admission before shutdown and rejects creates and resumes", async () => { + let releaseLaunchOptions!: () => void; + const launchOptionsReady = new Promise((resolve) => { + releaseLaunchOptions = resolve; + }); + const spawnPty = vi.fn(() => { + return createFakePty().pty as unknown as ReturnType; + }); + const { manager } = makeManager({ + spawnPty, + buildLaunchOpts: async () => { + await launchOptionsReady; + return { prompt: "not-used-by-interactive-launch" }; + }, + }); + + const creating = manager.create({ + cwd: "/tmp/proj", + harness: "claude-code", + }); + manager.beginShutdown(); + releaseLaunchOptions(); + + await expect(creating).rejects.toBeInstanceOf(SessionManagerClosingError); + expect(spawnPty).not.toHaveBeenCalled(); + expect(manager.list()).toEqual([ + expect.objectContaining({ status: "exited" }), + ]); + await expect( + manager.create({ cwd: "/tmp/second", harness: "claude-code" }), + ).rejects.toBeInstanceOf(SessionManagerClosingError); + await expect(manager.resume(manager.list()[0]!.id)).rejects.toBeInstanceOf( + SessionManagerClosingError, + ); + }); + + + it("classifies raw input on a recognized trust screen without blocking the user's bytes", async () => { + const onTerminalInput = vi.fn(); + const { manager, spawns } = makeManager({ + onTerminalInput, + adapter: createFakeAdapter({ + detectBlockingPrompt: (output) => output.includes("Do you trust"), + }), + }); + const session = await manager.create({ + cwd: "/tmp/proj", + harness: "claude-code", + }); + spawns[0]?.emitData("Do you trust the files in this folder?\r\n"); + + expect(manager.write(session.id, "y\r")).toBe(true); + expect(onTerminalInput).toHaveBeenCalledWith(session.id, { + blockingPrompt: true, + runtimeEpoch: manager.getRuntimeEpoch(session.id), + }); + expect(spawns[0]?.pty.write).toHaveBeenCalledWith("y\r"); + }); + + + it("claims the starting lifecycle before asynchronous resume config is regenerated", async () => { + const resumeConfig = deferred(); + let buildCount = 0; + const buildLaunchOpts = vi.fn(async () => { + buildCount += 1; + if (buildCount === 2) await resumeConfig.promise; + return {}; + }); + const { manager, adapter, spawns } = makeManager({ buildLaunchOpts }); + const session = await manager.create({ + cwd: "/tmp/proj", + harness: "claude-code", + }); + await manager.setAgentSessionId(session.id, "agent-uuid-1"); + spawns[0]?.emitExit(0); + await manager.flush(); + + const statuses: HarnessSession["status"][] = []; + const unsubscribe = manager.onStatusChange((updated) => { + if (updated.id === session.id) statuses.push(updated.status); + }); + const resumed = manager.resume(session.id); + await vi.waitFor(() => expect(buildCount).toBe(2)); + expect(manager.get(session.id)?.status).toBe("starting"); + + // Bootstrap exit bookkeeping can finish after kill() resolves. Its metadata + // update must observe the claimed resume lifecycle, so server cleanup does + // not remove the config currently being regenerated. + await manager.setProjectBootstrapMetadata(session.id, { + projectId: "project-1", + userId: "user-1", + targetSessionId: session.id, + bootstrap: { status: "skipped", reason: "user-proceeded" }, + queuedInputIds: [], + }); + expect(statuses.at(-1)).toBe("starting"); + expect(adapter.resume).not.toHaveBeenCalled(); + + resumeConfig.resolve(); + await resumed; + unsubscribe(); + expect(manager.get(session.id)?.status).toBe("running"); + }); + + + it("revalidates create scope after the lazy PTY loader settles and before admission", async () => { + const loader = deferred(); + const loadSpawnPty = vi.fn(() => loader.promise); + const spawnPty = vi.fn( + () => createFakePty().pty as unknown as ReturnType, + ); + let inScope = true; + const resolveAgentMapIdentity = vi.fn(async (sessionId: string) => + inScope + ? { projectId: "project-1", userId: "user-1", sessionId } + : undefined, + ); + const onAgentMapSessionExit = vi.fn(); + const { manager, adapter } = makeManager({ + loadSpawnPty, + resolveAgentMapIdentity, + onAgentMapSessionExit, + }); + + const creating = manager.create({ + cwd: "/tmp/proj", + harness: "claude-code", + }); + await vi.waitFor(() => expect(loadSpawnPty).toHaveBeenCalledOnce()); + expect(resolveAgentMapIdentity).toHaveBeenCalledOnce(); + inScope = false; + loader.resolve(spawnPty); + + await expect(creating).rejects.toBeInstanceOf( + ProjectSessionScopeUnavailableError, + ); + expect(resolveAgentMapIdentity).toHaveBeenCalledTimes(2); + expect(adapter.launch).toHaveBeenCalledOnce(); + expect(spawnPty).not.toHaveBeenCalled(); + expect(manager.list()).toEqual([ + expect.objectContaining({ status: "exited" }), + ]); + expect(onAgentMapSessionExit).toHaveBeenCalledWith(manager.list()[0]!.id); + }); + + + it("revalidates resume scope after the lazy PTY loader settles and before admission", async () => { + const resumeLoader = deferred(); + const fakePty = createFakePty(); + const spawnPty = vi.fn( + () => fakePty.pty as unknown as ReturnType, + ); + let loadCount = 0; + const loadSpawnPty = vi.fn(async () => { + loadCount += 1; + return loadCount === 1 ? spawnPty : resumeLoader.promise; + }); + let inScope = true; + const resolveAgentMapIdentity = vi.fn(async (sessionId: string) => + inScope + ? { projectId: "project-1", userId: "user-1", sessionId } + : undefined, + ); + const onAgentMapSessionExit = vi.fn(); + const { manager, adapter } = makeManager({ + loadSpawnPty, + resolveAgentMapIdentity, + onAgentMapSessionExit, + }); + const session = await manager.create({ + cwd: "/tmp/proj", + harness: "claude-code", + }); + await manager.setAgentSessionId(session.id, "provider-project-session"); + fakePty.emitExit(0); + await manager.flush(); + + const resuming = manager.resume(session.id); + await vi.waitFor(() => expect(loadSpawnPty).toHaveBeenCalledTimes(2)); + expect(resolveAgentMapIdentity).toHaveBeenCalledTimes(4); + inScope = false; + resumeLoader.resolve(spawnPty); + + await expect(resuming).rejects.toBeInstanceOf( + ProjectSessionScopeUnavailableError, + ); + expect(resolveAgentMapIdentity).toHaveBeenCalledTimes(5); + expect(adapter.resume).toHaveBeenCalledOnce(); + expect(spawnPty).toHaveBeenCalledOnce(); + expect(manager.get(session.id)).toMatchObject({ + id: session.id, + status: "exited", + agentMapIdentity: { + projectId: "project-1", + userId: "user-1", + sessionId: session.id, + }, + }); + expect(onAgentMapSessionExit).toHaveBeenCalledWith(session.id); + }); + + + it("rotates the authoritative runtime epoch and rejects stale lifecycle signals after resume", async () => { + const tokens = ["token-a", "token-b"]; + const epochs = ["epoch-a", "epoch-b"]; + const { manager, spawns } = makeManager({ + ingestCredentials: new IngestCredentialRegistry( + () => tokens.shift()!, + () => epochs.shift()!, + ), + }); + const session = await manager.create({ + cwd: "/tmp/proj", + harness: "claude-code", + }); + const firstEpoch = manager.getRuntimeEpoch(session.id); + expect(firstEpoch).toBe("epoch-a"); + expect( + await manager.setAgentSessionId( + session.id, + "provider-shared", + "startup", + firstEpoch!, + ), + ).toBe(true); + + spawns[0]!.emitExit(0); + await manager.flush(); + expect(manager.isCurrentRuntimeEpoch(session.id, firstEpoch!)).toBe(false); + expect(manager.acceptsIngestRuntimeEpoch(session.id, firstEpoch!)).toBe( + true, + ); + manager.setReady(session.id, firstEpoch!); + expect(manager.get(session.id)?.ready).toBe(false); + + await manager.resume(session.id); + const secondEpoch = manager.getRuntimeEpoch(session.id); + expect(secondEpoch).toBe("epoch-b"); + expect(secondEpoch).not.toBe(firstEpoch); + expect(manager.isCurrentRuntimeEpoch(session.id, firstEpoch!)).toBe(false); + expect(manager.acceptsIngestRuntimeEpoch(session.id, firstEpoch!)).toBe( + false, + ); + expect(manager.isCurrentRuntimeEpoch(session.id, secondEpoch!)).toBe(true); + expect(manager.acceptsIngestRuntimeEpoch(session.id, secondEpoch!)).toBe( + true, + ); + + manager.setReady(session.id, firstEpoch!); + expect(manager.get(session.id)?.ready).toBe(false); + expect( + await manager.setAgentSessionId( + session.id, + "provider-shared", + "resume", + firstEpoch!, + ), + ).toBe(false); + + manager.setReady(session.id, secondEpoch!); + expect(manager.get(session.id)?.ready).toBe(true); + expect( + await manager.setAgentSessionId( + session.id, + "provider-shared", + "resume", + secondEpoch!, + ), + ).toBe(true); + }); + + + it("fences transcript identity state to the exact live runtime epoch", async () => { + const { manager } = makeManager({ + adapter: createFakeAdapter({ eventSource: "transcript-tail" }), + ingestCredentials: new IngestCredentialRegistry( + () => "token-a", + () => "epoch-a", + ), + }); + const session = await manager.create({ + cwd: "/tmp/proj", + harness: "claude-code", + }); + + expect(manager.getAdapterIdentityState(session.id, "epoch-a")).toBe("pending"); + expect(manager.setAdapterIdentityState(session.id, "stale-epoch", "ready")).toBe(false); + expect(manager.setAdapterIdentityState(session.id, "epoch-a", "ready")).toBe(true); + expect(manager.getAdapterIdentityState(session.id, "epoch-a")).toBe("ready"); + }); + + + it("does not publish a PTY when the runtime epoch transition fails", async () => { + const spawnPty = vi.fn(() => { + return createFakePty().pty as unknown as ReturnType; + }); + const transition = vi.fn( + async (_session: HarnessSession, _runtimeEpoch: string | null) => { + throw new Error("durable epoch retirement failed"); + }, + ); + const { manager } = makeManager({ + spawnPty, + onRuntimeEpochTransition: transition, + }); + + await expect( + manager.create({ cwd: "/tmp/proj", harness: "claude-code" }), + ).rejects.toThrow("durable epoch retirement failed"); + + expect(transition).toHaveBeenCalledOnce(); + expect(transition.mock.calls[0]?.[1]).toEqual(expect.any(String)); + expect(spawnPty).not.toHaveBeenCalled(); + expect(manager.list()).toEqual([ + expect.objectContaining({ status: "exited", ready: false }), + ]); + expect(manager.isLive(manager.list()[0]!.id)).toBe(false); + }); + + describe("tracked runtime input", () => { + beforeEach(() => { vi.useFakeTimers(); }); + afterEach(() => { vi.useRealTimers(); }); + + + + it("lets durable API input cancel only a staged background turn", async () => { + const onTerminalInput = vi.fn(); + const { manager, spawns } = makeManager({ onTerminalInput }); + const session = await manager.create({ + cwd: "/tmp/proj", + harness: "claude-code", + }); + manager.setReady(session.id); + + const background = manager.submitInput( + session.id, + "automatic map bootstrap", + true, + undefined, + true, + ); + expect(spawns[0]?.pty.write).toHaveBeenCalledWith( + "automatic map bootstrap", + ); + + expect(manager.preemptBackgroundInput(session.id)).toBe(true); + expect(manager.preemptBackgroundInput(session.id)).toBe(false); + expect(spawns[0]?.pty.write).toHaveBeenNthCalledWith(2, "\x15"); + expect(onTerminalInput).not.toHaveBeenCalled(); + + const assertion = expect(background).rejects.toMatchObject({ + code: "SESSION_BACKGROUND_INPUT_PREEMPTED", + staged: true, + }); + await vi.advanceTimersByTimeAsync(300); + await assertion; + expect(spawns[0]?.pty.write).not.toHaveBeenCalledWith("\r"); + }); + + + + it("waits for the durable pre-write hook before crossing the PTY boundary", async () => { + const { manager, spawns } = makeManager(); + const session = await manager.create({ + cwd: "/tmp/proj", + harness: "claude-code", + }); + manager.setReady(session.id); + const phase = deferred(); + const beforeFirstWrite = vi.fn(() => phase.promise); + + const submitting = manager.submitInput( + session.id, + "durable turn", + true, + undefined, + false, + { beforeFirstWrite, canWriteNow: () => true }, + ); + await Promise.resolve(); + expect(beforeFirstWrite).toHaveBeenCalledOnce(); + expect(spawns[0]?.pty.write).not.toHaveBeenCalled(); + + phase.resolve(); + await vi.advanceTimersByTimeAsync(300); + await expect(submitting).resolves.toBe(true); + expect(spawns[0]?.pty.write).toHaveBeenNthCalledWith(1, "durable turn"); + expect(spawns[0]?.pty.write).toHaveBeenNthCalledWith(2, "\r"); + }); + + + + it("preserves user keystrokes received during a deferred pre-write hook", async () => { + const { manager, spawns } = makeManager(); + const session = await manager.create({ + cwd: "/tmp/proj", + harness: "claude-code", + }); + manager.setReady(session.id); + const phase = deferred(); + const beforeFirstWrite = vi.fn(() => phase.promise); + const onNotSubmitted = vi.fn(async () => {}); + const background = manager.submitInputTracked(session.id, "background turn", { + lifecycle: { beforeFirstWrite, onNotSubmitted }, + }); + expect(beforeFirstWrite).toHaveBeenCalledOnce(); + + manager.write(session.id, "user draft"); + phase.resolve(); + await vi.advanceTimersByTimeAsync(300); + + expect(await background).toMatchObject({ + accepted: false, + phase: "not-written", + error: { code: "SESSION_BACKGROUND_INPUT_PREEMPTED", staged: false }, + }); + expect(onNotSubmitted).toHaveBeenCalledOnce(); + expect(spawns[0]?.pty.write.mock.calls).toEqual([["user draft"]]); + }); + + it("compensates a durable claim when racing staged input wins during its pre-write hook", async () => { + const { manager, spawns } = makeManager(); + const session = await manager.create({ + cwd: "/tmp/proj", + harness: "claude-code", + }); + manager.setReady(session.id); + const phase = deferred(); + const onNotSubmitted = vi.fn(async () => {}); + const background = manager.submitInputTracked(session.id, "background turn", { + lifecycle: { beforeFirstWrite: () => phase.promise, onNotSubmitted }, + }); + const foreground = manager.submitInput(session.id, "foreground turn"); + phase.resolve(); + const result = await background; + await vi.advanceTimersByTimeAsync(300); + await foreground; + + expect(result).toMatchObject({ + accepted: false, + phase: "not-written", + error: { code: "SESSION_BACKGROUND_INPUT_PREEMPTED", staged: false }, + }); + expect(onNotSubmitted).toHaveBeenCalledOnce(); + expect(spawns[0]?.pty.write.mock.calls).toEqual([["foreground turn"], ["\r"]]); + }); + + it("records positive not-submitted evidence when the PTY rejects text before Enter", async () => { + const { manager, spawns } = makeManager(); + const session = await manager.create({ + cwd: "/tmp/proj", + harness: "claude-code", + }); + manager.setReady(session.id); + const beforeFirstWrite = vi.fn(async () => {}); + const onNotSubmitted = vi.fn(async () => {}); + spawns[0]?.pty.write.mockImplementationOnce(() => { + throw new Error("PTY rejected text"); + }); + + await expect( + manager.submitInput(session.id, "durable turn", true, undefined, true, { + beforeFirstWrite, + onNotSubmitted, + }), + ).rejects.toThrow("PTY rejected text"); + + expect(beforeFirstWrite).toHaveBeenCalledOnce(); + expect(onNotSubmitted).toHaveBeenCalledOnce(); + expect(spawns[0]?.pty.write).toHaveBeenNthCalledWith(1, "durable turn"); + expect(spawns[0]?.pty.write).toHaveBeenNthCalledWith(2, "\x15"); + expect(spawns[0]?.pty.write).not.toHaveBeenCalledWith("\r"); + }); + + + + it("withholds not-submitted evidence when a partial text line cannot be cleared", async () => { + const { manager, spawns } = makeManager(); + const session = await manager.create({ + cwd: "/tmp/proj", + harness: "claude-code", + }); + manager.setReady(session.id); + const onNotSubmitted = vi.fn(async () => {}); + spawns[0]?.pty.write + .mockImplementationOnce(() => { + throw new Error("PTY rejected text after a possible prefix"); + }) + .mockImplementationOnce(() => { + throw new Error("PTY also rejected line cleanup"); + }); + + await expect( + manager.submitInput(session.id, "durable turn", true, undefined, true, { + onNotSubmitted, + }), + ).rejects.toThrow("PTY rejected text after a possible prefix"); + + expect(spawns[0]?.pty.write).toHaveBeenNthCalledWith(1, "durable turn"); + expect(spawns[0]?.pty.write).toHaveBeenNthCalledWith(2, "\x15"); + expect(spawns[0]?.pty.write).not.toHaveBeenCalledWith("\r"); + expect(onNotSubmitted).not.toHaveBeenCalled(); + }); + + + + it("closes a partial bracketed paste before proving the composer line was cleared", async () => { + const { manager, spawns } = makeManager(); + const session = await manager.create({ + cwd: "/tmp/proj", + harness: "claude-code", + }); + manager.setReady(session.id); + spawns[0]?.emitData("\x1b[?2004h"); + const onNotSubmitted = vi.fn(async () => {}); + spawns[0]?.pty.write.mockImplementationOnce(() => { + throw new Error("partial bracketed paste"); + }); + + await expect( + manager.submitInput(session.id, "line one\nline two", true, undefined, true, { + onNotSubmitted, + }), + ).rejects.toThrow("partial bracketed paste"); + + expect(spawns[0]?.pty.write).toHaveBeenNthCalledWith( + 1, + "\x1b[200~line one\nline two\x1b[201~", + ); + expect(spawns[0]?.pty.write).toHaveBeenNthCalledWith(2, "\x1b[201~"); + expect(spawns[0]?.pty.write).toHaveBeenNthCalledWith(3, "\x15"); + expect(spawns[0]?.pty.write).not.toHaveBeenCalledWith("\r"); + expect(onNotSubmitted).toHaveBeenCalledOnce(); + }); + + + + it("withholds retry proof when a partial bracketed paste cannot be closed", async () => { + const { manager, spawns } = makeManager(); + const session = await manager.create({ + cwd: "/tmp/proj", + harness: "claude-code", + }); + manager.setReady(session.id); + spawns[0]?.emitData("\x1b[?2004h"); + const onNotSubmitted = vi.fn(async () => {}); + spawns[0]?.pty.write + .mockImplementationOnce(() => { + throw new Error("partial bracketed paste"); + }) + .mockImplementationOnce(() => { + throw new Error("paste closer rejected"); + }); + + await expect( + manager.submitInput(session.id, "unsafe paste", true, undefined, true, { + onNotSubmitted, + }), + ).rejects.toThrow("partial bracketed paste"); + + expect(spawns[0]?.pty.write).toHaveBeenNthCalledWith(2, "\x1b[201~"); + expect(spawns[0]?.pty.write).toHaveBeenNthCalledWith(3, "\x15"); + expect(onNotSubmitted).not.toHaveBeenCalled(); + }); + + + + it("withholds retry proof when line cleanup fails after closing a partial paste", async () => { + const { manager, spawns } = makeManager(); + const session = await manager.create({ + cwd: "/tmp/proj", + harness: "claude-code", + }); + manager.setReady(session.id); + spawns[0]?.emitData("\x1b[?2004h"); + const onNotSubmitted = vi.fn(async () => {}); + spawns[0]?.pty.write + .mockImplementationOnce(() => { + throw new Error("partial bracketed paste"); + }) + .mockImplementationOnce(() => {}) + .mockImplementationOnce(() => { + throw new Error("line cleanup rejected"); + }); + + await expect( + manager.submitInput(session.id, "unsafe paste", true, undefined, true, { + onNotSubmitted, + }), + ).rejects.toThrow("partial bracketed paste"); + + expect(spawns[0]?.pty.write).toHaveBeenNthCalledWith(2, "\x1b[201~"); + expect(spawns[0]?.pty.write).toHaveBeenNthCalledWith(3, "\x15"); + expect(onNotSubmitted).not.toHaveBeenCalled(); + }); + + + + it("blocks later raw input on a poisoned composer until a reset succeeds", async () => { + const { manager, spawns } = makeManager(); + const session = await manager.create({ + cwd: "/tmp/proj", + harness: "claude-code", + }); + manager.setReady(session.id); + let call = 0; + spawns[0]?.pty.write.mockImplementation(() => { + call += 1; + if (call <= 3) throw new Error(`injected write failure ${call}`); + }); + + await expect( + manager.submitInput(session.id, "partial A", true, undefined, true), + ).rejects.toThrow("injected write failure 1"); + expect(() => manager.write(session.id, "must not append to A")).toThrow( + SessionInputIsolationError, + ); + expect(spawns[0]?.pty.write).not.toHaveBeenCalledWith( + "must not append to A", + ); + + expect(manager.write(session.id, "safe after reset")).toBe(true); + expect(spawns[0]?.pty.write).toHaveBeenNthCalledWith(4, "\x15"); + expect(spawns[0]?.pty.write).toHaveBeenNthCalledWith( + 5, + "safe after reset", + ); + }); + + + + it("uses Ctrl-C only as a fallback after a complete staged paste", async () => { + const { manager, spawns } = makeManager(); + const session = await manager.create({ + cwd: "/tmp/proj", + harness: "claude-code", + }); + manager.setReady(session.id); + spawns[0]?.emitData("\x1b[?2004h"); + const onNotSubmitted = vi.fn(async () => {}); + const background = manager.submitInput( + session.id, + "complete staged paste", + true, + undefined, + true, + { onNotSubmitted }, + ); + const rejected = expect(background).rejects.toBeInstanceOf( + SessionBackgroundInputPreemptedError, + ); + spawns[0]?.pty.write.mockImplementationOnce(() => { + throw new Error("Ctrl-U rejected"); + }); + + expect(manager.preemptBackgroundInput(session.id)).toBe(true); + expect(spawns[0]?.pty.write).toHaveBeenNthCalledWith(2, "\x15"); + expect(spawns[0]?.pty.write).toHaveBeenNthCalledWith(3, "\x03"); + await vi.advanceTimersByTimeAsync(300); + await rejected; + expect(onNotSubmitted).toHaveBeenCalledOnce(); + }); + + + + it("poisons arbitrary submit:false text after a partial write", async () => { + const { manager, spawns } = makeManager(); + const session = await manager.create({ + cwd: "/tmp/proj", + harness: "claude-code", + }); + manager.setReady(session.id); + spawns[0]?.pty.write.mockImplementationOnce(() => { + throw new Error("partial draft"); + }); + + await expect( + manager.submitInput(session.id, "multi-byte draft", false), + ).rejects.toThrow("partial draft"); + expect(manager.write(session.id, "new raw input")).toBe(true); + expect(spawns[0]?.pty.write).toHaveBeenNthCalledWith(2, "\x15"); + expect(spawns[0]?.pty.write).toHaveBeenNthCalledWith(3, "new raw input"); + }); + + + + it("does not claim not-submitted proof when the Enter write is ambiguous", async () => { + const { manager, spawns } = makeManager(); + const session = await manager.create({ + cwd: "/tmp/proj", + harness: "claude-code", + }); + manager.setReady(session.id); + const onNotSubmitted = vi.fn(async () => {}); + spawns[0]?.pty.write + .mockImplementationOnce(() => {}) + .mockImplementationOnce(() => { + throw new Error("PTY Enter outcome unknown"); + }); + + const submission = manager.submitInput( + session.id, + "durable turn", + true, + undefined, + true, + { onNotSubmitted }, + ); + const rejected = expect(submission).rejects.toThrow( + "PTY Enter outcome unknown", + ); + await vi.advanceTimersByTimeAsync(300); + await rejected; + + expect(spawns[0]?.pty.write).toHaveBeenNthCalledWith(1, "durable turn"); + expect(spawns[0]?.pty.write).toHaveBeenNthCalledWith(2, "\r"); + expect(onNotSubmitted).not.toHaveBeenCalled(); + }); + + + + it("clears staged text and never writes Enter when shutdown wins the final admission boundary", async () => { + const { manager, spawns } = makeManager(); + const session = await manager.create({ + cwd: "/tmp/proj", + harness: "claude-code", + }); + manager.setReady(session.id); + const finalAuthorization = deferred(); + const canWrite = vi + .fn<() => boolean | Promise>() + .mockReturnValueOnce(true) + .mockImplementationOnce(() => finalAuthorization.promise); + + const submitting = manager.submitInput( + session.id, + "must not submit after shutdown", + true, + canWrite, + false, + { canWriteNow: () => true }, + ); + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); + expect(spawns[0]?.pty.write).toHaveBeenCalledWith( + "must not submit after shutdown", + ); + await vi.advanceTimersByTimeAsync(300); + manager.beginShutdown(); + finalAuthorization.resolve(true); + + await expect(submitting).rejects.toMatchObject({ staged: true }); + expect(spawns[0]?.pty.write).toHaveBeenLastCalledWith("\x15"); + expect(spawns[0]?.pty.write).not.toHaveBeenCalledWith("\r"); + }); + + + + it("lets raw user input preempt a staged background turn without combining either prompt", async () => { + const onTerminalInput = vi.fn(); + const { manager, spawns } = makeManager({ onTerminalInput }); + const session = await manager.create({ + cwd: "/tmp/proj", + harness: "claude-code", + }); + manager.setReady(session.id); + + const background = manager.submitInput( + session.id, + "automatic map bootstrap", + true, + undefined, + true, + ); + expect(spawns[0]?.pty.write).toHaveBeenCalledTimes(1); + expect(spawns[0]?.pty.write).toHaveBeenCalledWith( + "automatic map bootstrap", + ); + + expect(manager.write(session.id, "implement the API now\r")).toBe(true); + expect(onTerminalInput).toHaveBeenCalledWith(session.id, { + blockingPrompt: false, + runtimeEpoch: manager.getRuntimeEpoch(session.id), + }); + expect(spawns[0]?.pty.write).toHaveBeenNthCalledWith(2, "\x15"); + expect(spawns[0]?.pty.write).toHaveBeenNthCalledWith( + 3, + "implement the API now\r", + ); + + const assertion = expect(background).rejects.toMatchObject({ + code: "SESSION_BACKGROUND_INPUT_PREEMPTED", + staged: true, + }); + await vi.advanceTimersByTimeAsync(300); + await assertion; + + expect(spawns[0]?.pty.write).toHaveBeenCalledTimes(3); + expect(spawns[0]?.pty.write).not.toHaveBeenCalledWith("\r"); + await expect(background).rejects.toBeInstanceOf( + SessionBackgroundInputPreemptedError, + ); + }); +}); }); diff --git a/packages/harness/src/core/session-manager.ts b/packages/harness/src/core/session-manager.ts index 2a4b4463e..2598bf17c 100644 --- a/packages/harness/src/core/session-manager.ts +++ b/packages/harness/src/core/session-manager.ts @@ -23,6 +23,7 @@ import { } from "../shared/types.js"; import type { PlannerSessionMetadata, + ProjectBootstrapMetadata, ProjectAgentSession, } from "../shared/agent-map.js"; import { migratePersistedProjectIdentity } from "./project-session-legacy-migration.js"; @@ -46,7 +47,7 @@ import { UnknownSessionError, } from "./errors.js"; import { listHarnessAdapters } from "./adapters/registry.js"; -import type { IngestCredentialProvider } from "./ingest-credentials.js"; +import type { IngestCredentialProvider, IssuedIngestCredential } from "./ingest-credentials.js"; export { AdapterNotFoundError, @@ -305,7 +306,10 @@ export function sanitizeExitTail(raw: string): string | null { return trimmed.slice(-EXIT_TAIL_BYTES); } -export type SessionStatusListener = (session: HarnessSession) => void; +export type SessionStatusListener = ( + session: HarnessSession, + context: SessionStatusContext, +) => void; export type SessionDataListener = (chunk: string) => void; /** See `onActivity()`. */ export type SessionActivityListener = (harnessSessionId: string) => void; @@ -337,6 +341,23 @@ export type LaunchOptsBuilder = ( const defaultBuildLaunchOpts: LaunchOptsBuilder = () => ({}); export interface SessionManagerOptions { + /** + * Serializes coordinator ownership before a PTY generation is published. + * `null` retracts a prepared epoch when pre-publication setup fails. + */ + onRuntimeEpochTransition?: ( + session: HarnessSession, + runtimeEpoch: string | null, + ) => Promise | void; + + /** Synchronous notification before a real terminal write crosses the PTY. */ + onTerminalInput?: (sessionId: string, context: TerminalInputContext) => void; + + /** Async node-pty loader seam used only when `spawnPty` is absent. Production + * uses the module loader; tests use this to prove authorization is checked + * after that final await and before PTY admission. */ + loadSpawnPty?: () => Promise; + adapters: Partial>; /** Base URL the harness server is reachable at, e.g. http://127.0.0.1:4100. */ ingestUrl: string; @@ -436,6 +457,9 @@ export interface TrustedSessionResumeOptions { } interface PtyHandle { + /** Server-owned identity for this exact live PTY generation. */ + runtimeEpoch: string; + pty: IPty; buffer: string; /** Latest content-bearing terminal repaint used for current-screen checks. @@ -513,10 +537,321 @@ interface PtyHandle { killed: boolean; } +export type AdapterIdentityState = + | "not-required" + | "pending" + | "ready" + | "ambiguous" + | "unavailable"; + + +export interface SessionStatusContext { + /** Exact live/retiring PTY generation, or null before a PTY exists. */ + runtimeEpoch: string | null; +} + + +export interface TerminalInputContext { + /** Server-owned identity of the exact PTY receiving these bytes. */ + runtimeEpoch: string; + /** The current adapter screen is a recognized trust/login/setup blocker. */ + blockingPrompt: boolean; +} + + +export type TrackedSessionInputResult = Readonly<{ + accepted: boolean; + phase: SessionInputWritePhase; + error?: unknown; +}>; + + +export type SessionInputWritePhase = + | "not-written" + | "text-staged" + | "enter-written"; + + +export interface SessionInputWriteLifecycle { + /** Durable transition that must commit before the first PTY byte. */ + beforeFirstWrite?: () => Promise; + /** Synchronous final admission fence checked immediately before each write. */ + canWriteNow?: () => boolean; + /** Durable positive evidence that the writer returned before attempting + * Enter. Errors at the Enter write are intentionally excluded. */ + onNotSubmitted?: () => Promise; + /** Synchronous byte-boundary observation for durable delivery recovery. */ + onWritePhase?: (phase: SessionInputWritePhase) => void; +} + + +/** A prior partial PTY write could not be safely removed from the composer. */ +export class SessionInputIsolationError extends Error { + readonly code = "SESSION_INPUT_ISOLATION_REQUIRED"; + + constructor() { + super("session input is blocked until the terminal composer is reset"); + this.name = "SessionInputIsolationError"; + } +} + + +export class SessionManagerClosingError extends Error { + readonly code = "SESSION_MANAGER_CLOSING"; + + constructor() { + super("session manager is shutting down"); + this.name = "SessionManagerClosingError"; + } +} + + +/** A real terminal write preempted a lower-priority background injection. */ +export class SessionBackgroundInputPreemptedError extends Error { + readonly code = "SESSION_BACKGROUND_INPUT_PREEMPTED"; + + constructor(readonly staged: boolean) { + super("background session input was preempted by user input"); + this.name = "SessionBackgroundInputPreemptedError"; + } +} + + export class SessionManager { + + /** + * Internal tracked variant for retry-safe coordinator delivery. It never + * turns an ambiguous write exception into zero-byte proof: callers receive + * the furthest phase observed at the exact PTY boundary. + */ + async submitInputTracked( + id: string, + text: string, + options: Readonly<{ + canWrite?: () => boolean | Promise; + lifecycle?: Omit; + background?: boolean; + }> = {}, + ): Promise { + let phase: SessionInputWritePhase = "not-written"; + try { + const accepted = await this.submitInput( + id, + text, + true, + options.canWrite, + options.background ?? true, + { + ...options.lifecycle, + onWritePhase: (next) => { + phase = next; + }, + }, + ); + return { accepted, phase }; + } catch (error) { + return { accepted: false, phase, error }; + } + } + + + /** + * Cancel only a lower-priority server-owned background submission. Unlike + * write(), this does not forward bytes or preempt an ordinary user/API + * submission. It is safe to call before staging begins; the coordinator's + * submit guard covers that side of the race. + */ + preemptBackgroundInput(id: string): boolean { + const staged = this.stagedInputs.get(id); + if (!staged?.background || staged.preempted) return false; + staged.preempted = true; + if (staged.textWritten) { + if (this.abandonStagedLine(staged.handle)) { + staged.lineCleared = true; + } + } + return true; + } + + + /** Abandon a fully staged line. Ctrl-C is a safe fallback here because a + * successful full bracketed-paste write already carried its closing marker. */ + private abandonStagedLine(handle: PtyHandle): boolean { + try { + handle.pty.write("\x15"); + this.observeTrustedTerminalInput(handle, "\x15"); + return true; + } catch { + try { + handle.pty.write("\x03"); + this.observeTrustedTerminalInput(handle, "\x03"); + return true; + } catch { + this.markComposerUnsafe(handle, false); + return false; + } + } + } + + + /** + * Recover only a previously poisoned composer, before any new user or + * server-owned text is written. Closing bracketed paste (when required) and + * abandoning the line are both non-submitting operations. Failure leaves + * the handle poisoned and no caller payload is forwarded. + */ + private resetUnsafeComposer(handle: PtyHandle): boolean { + const unsafe = this.unsafeComposers.get(handle); + if (!unsafe) return true; + if (this.closing) return false; + let pasteMayBeOpen = unsafe.pasteMayBeOpen; + if (pasteMayBeOpen) { + try { + handle.pty.write(BRACKETED_PASTE_END); + this.observeTrustedTerminalInput(handle, BRACKETED_PASTE_END); + pasteMayBeOpen = false; + } catch { + return false; + } + } + try { + handle.pty.write("\x15"); + this.observeTrustedTerminalInput(handle, "\x15"); + } catch { + this.unsafeComposers.set(handle, { pasteMayBeOpen }); + return false; + } + this.unsafeComposers.delete(handle); + return true; + } + + + private markComposerUnsafe( + handle: PtyHandle, + pasteMayBeOpen: boolean, + ): void { + const current = this.unsafeComposers.get(handle); + this.unsafeComposers.set(handle, { + pasteMayBeOpen: current?.pasteMayBeOpen === true || pasteMayBeOpen, + }); + } + + + /** Close admission before a server shutdown snapshots live PTYs. */ + beginShutdown(): void { + this.closing = true; + } + + + /** Kill only the exact PTY generation a losing coordinator created. */ + killIfRuntime(id: string, runtimeEpoch: string): Promise { + if (this.ptys.get(id)?.runtimeEpoch !== runtimeEpoch) + return Promise.resolve(false); + return this.kill(id); + } + + + /** + * Fail-closed admission for already-authenticated ingest work. A terminal + * event may finish after its PTY retires, but a replacement handle always + * takes ownership immediately and rejects every earlier epoch. + */ + acceptsIngestRuntimeEpoch(id: string, runtimeEpoch: string): boolean { + const live = this.ptys.get(id); + if (live) return live.runtimeEpoch === runtimeEpoch; + return ( + this.sessions.get(id)?.status === "exited" && + this.retiredRuntimeEpochs.get(id) === runtimeEpoch + ); + } + + + /** True only for the exact PTY generation that is live right now. */ + isCurrentRuntimeEpoch(id: string, runtimeEpoch: string): boolean { + const live = this.ptys.get(id); + return live?.runtimeEpoch === runtimeEpoch; + } + + + /** Server-only acknowledgement from an adapter-owned identity broker. */ + setAdapterIdentityState( + id: string, + runtimeEpoch: string, + state: Exclude, + ): boolean { + const current = this.adapterIdentityStates.get(id); + if (!current || current.runtimeEpoch !== runtimeEpoch || + !this.isCurrentRuntimeEpoch(id, runtimeEpoch)) return false; + current.state = state; + return true; + } + + + /** Exact-runtime adapter identity used by trusted background delivery. */ + getAdapterIdentityState(id: string, runtimeEpoch: string): AdapterIdentityState { + const state = this.adapterIdentityStates.get(id); + if (!state || state.runtimeEpoch !== runtimeEpoch) return "pending"; + return state.state; + } + + + /** Opaque identity for the exact live PTY generation behind `id`. */ + getRuntimeEpoch(id: string): string | null { + return this.ptys.get(id)?.runtimeEpoch ?? null; + } + + private closing = false; + + /** Handle-local poison from a partial write whose composer cleanup could not + * be proven. A replacement PTY is clean by construction; the same handle + * must complete this reset before any later text or Enter is allowed. */ + private readonly unsafeComposers = new WeakMap< + PtyHandle, + { pasteMayBeOpen: boolean } + >(); + + /** One text→Enter transaction may be staged per session. */ + private readonly stagedInputs = new Map< + string, + { + handle: PtyHandle; + background: boolean; + preempted: boolean; + textWritten: boolean; + lineCleared: boolean; + } + >(); + + /** Monotonic raw-input observations used to preempt background injection. */ + private readonly terminalInputEpochs = new Map(); + + /** Adapter-owned correlation for transcript-backed runtimes. This is kept + * separate from terminal readiness: a TUI can be interactive before its + * exact vendor transcript has been identified. */ + private readonly adapterIdentityStates = new Map< + string, + { runtimeEpoch: string; state: AdapterIdentityState } + >(); + + /** Last cleanly retired PTY generation. It may finish already-admitted ingest + * work while the session is exited, but loses immediately to a replacement. */ + private readonly retiredRuntimeEpochs = new Map(); + + private readonly onRuntimeEpochTransition: SessionManagerOptions["onRuntimeEpochTransition"]; + + private readonly onTerminalInput: ( + sessionId: string, + context: TerminalInputContext, + ) => void; + + private readonly loadSpawnPty: () => Promise; + + private readonly issueIngestCredential: ( + sessionId: string, + ) => IssuedIngestCredential; + private readonly adapters: Partial>; private readonly ingestUrl: string; - private readonly issueIngestToken: (sessionId: string) => string; private readonly revokeIngestToken: (sessionId: string) => void; private readonly collectorUrl: string | undefined; private readonly sessionsPath: string; @@ -569,7 +904,7 @@ export class SessionManager { constructor(options: SessionManagerOptions) { this.adapters = options.adapters; this.ingestUrl = options.ingestUrl; - this.issueIngestToken = (sessionId) => + this.issueIngestCredential = (sessionId) => options.ingestCredentials.issue(sessionId); this.revokeIngestToken = (sessionId) => options.ingestCredentials.revoke(sessionId); @@ -577,6 +912,9 @@ export class SessionManager { this.sessionsPath = expandHome(options.sessionsPath ?? HARNESS_PATHS.sessions); this.agentSessionOwnersPath = `${this.sessionsPath}.agent-session-owners.json`; this.spawnPty = options.spawnPty; + this.loadSpawnPty = options.loadSpawnPty ?? loadDefaultSpawn; + this.onTerminalInput = options.onTerminalInput ?? (() => {}); + this.onRuntimeEpochTransition = options.onRuntimeEpochTransition; this.buildLaunchOpts = options.buildLaunchOpts ?? defaultBuildLaunchOpts; this.resolveAgentMapIdentity = options.resolveAgentMapIdentity; this.onAgentMapSessionExit = options.onAgentMapSessionExit; @@ -718,6 +1056,7 @@ export class SessionManager { req: CreateSessionRequest, trusted: TrustedSessionCreateOptions = {}, ): Promise { + if (this.closing) throw new SessionManagerClosingError(); const id = this.generateId(); const adapter = this.getAdapter(req.harness); const planning = trusted.planning?.(id); @@ -787,8 +1126,9 @@ export class SessionManager { // "running" — it must never show a bare empty iframe because nothing's // been written to .sapiom/canvas/index.html yet. await this.ensureCanvasTemplate(session.cwd); - await this.revalidateAgentMapIdentity(id, session.cwd, agentMapIdentity); - await this.spawn(session, spec); + await this.spawn(session, spec, () => + this.revalidateAgentMapIdentity(id, session.cwd, agentMapIdentity), + ); } catch (err) { // The first persist may itself be the failure, so reconciliation is // best-effort: always repair the in-memory record to "exited", attempt @@ -863,6 +1203,7 @@ export class SessionManager { id: string, trusted: TrustedSessionResumeOptions = {}, ): Promise { + if (this.closing) throw new SessionManagerClosingError(); const session = this.sessions.get(id); if (!session) throw new UnknownSessionError(id); if (this.rejectedProjectSessionMetadata.has(id)) throw new ProjectSessionScopeUnavailableError(id); @@ -903,41 +1244,56 @@ export class SessionManager { if (agentMapIdentity) session.agentMapIdentity = structuredClone(agentMapIdentity); else delete session.agentMapIdentity; - const launchContext = - trusted.promptAppendix || agentMapIdentity - ? { - ...(trusted.promptAppendix - ? { promptAppendix: trusted.promptAppendix } - : {}), - ...(agentMapIdentity ? { agentMapIdentity } : {}), - resume: true as const, - } - : undefined; - const opts: LaunchOpts = { - harnessSessionId: id, - cwd: session.cwd, - ...(await (launchContext - ? this.buildLaunchOpts(id, session, launchContext) - : this.buildLaunchOpts(id, session))), - }; + // Claim the pre-PTY resume window before generated launch state is built. + // Exit observers may finish asynchronous bookkeeping after kill() resolves; + // they must see this lifecycle as starting, not schedule cleanup against + // files that the resumed process is currently regenerating. + const lastActiveBeforeResume = session.lastActiveAt; + const statusBeforeResume = session.status; + const exitCodeBeforeResume = session.exitCode; + session.status = "starting"; + session.exitCode = null; + session.lastActiveAt = this.now(); + let opts: LaunchOpts; let spec: SpawnSpec; try { + const launchContext = + trusted.promptAppendix || agentMapIdentity + ? { + ...(trusted.promptAppendix + ? { promptAppendix: trusted.promptAppendix } + : {}), + ...(agentMapIdentity ? { agentMapIdentity } : {}), + resume: true as const, + } + : undefined; + opts = { + harnessSessionId: id, + cwd: session.cwd, + ...(await (launchContext + ? this.buildLaunchOpts(id, session, launchContext) + : this.buildLaunchOpts(id, session))), + }; spec = adapter.resume(session.agentSessionId, opts); } catch (error) { + // Resume preparation may rotate project capabilities or write generated + // launch state before the process exists. No starting state was exposed + // or persisted yet, so restore the exact prior record while releasing + // any prepared authority. + session.status = statusBeforeResume; + session.exitCode = exitCodeBeforeResume; + session.lastActiveAt = lastActiveBeforeResume; await Promise.resolve(this.onAgentMapSessionExit?.(id)).catch(() => {}); throw error; } - // Kept so the failure path below can put it back: `lastActiveAt` is - // stamped here only to keep sweepDeadSessions() from reaping this record + // The prior value is kept so the failure path below can put it back: + // `lastActiveAt` is stamped only to keep sweepDeadSessions() from reaping + // this record // during the pre-pty window (it reaps non-exited records with no pty once // they're older than the grace period). If the resume never produces a // pty, that stamp is not activity and must not survive — otherwise a // session idle since last night reports "Ran for 6h 25m" purely because // someone clicked Resume. - const lastActiveBeforeResume = session.lastActiveAt; - session.status = "starting"; - session.exitCode = null; - session.lastActiveAt = this.now(); try { await this.persist(); this.emitStatus(session); @@ -951,8 +1307,14 @@ export class SessionManager { // — a session from before the canvas kit existed, or one whose canvas // file was somehow deleted, still gets a live pane on resume. await this.ensureCanvasTemplate(session.cwd); - await this.revalidateAgentMapIdentity(id, session.cwd, agentMapIdentity); - await this.spawn(session, spec); + await this.spawn(session, spec, () => + this.revalidateAgentMapIdentity( + session.id, + session.cwd, + agentMapIdentity, + ), + ); + } catch (err) { // Same best-effort reconciliation as create(): the first persist can be // the failure, and a failed repair must not replace that original error. @@ -1090,12 +1452,60 @@ export class SessionManager { } } + /** + * Forward terminal bytes without submitting a separate prompt. + * @throws SessionInputIsolationError when prior partial input cannot be cleared. + */ write(id: string, data: string): boolean { const handle = this.ptys.get(id); if (!handle) return false; - handle.pty.write(data); - this.observeTrustedTerminalInput(handle, data); + if (!this.resetUnsafeComposer(handle)) { + throw new SessionInputIsolationError(); + } const session = this.sessions.get(id); + if (session && session.status !== "exited") { + try { + const adapter = this.adapters[session.harness]; + const blockingPrompt = Boolean( + adapter?.detectBlockingPrompt && + (this.hasCurrentBlockingPrompt(adapter, handle) || + this.hasRetainedBlockingPrompt(adapter, handle)), + ); + this.onTerminalInput(id, { + runtimeEpoch: handle.runtimeEpoch, + blockingPrompt, + }); + } catch { + // Input priority is local correctness; lifecycle telemetry/persistence + // callbacks are best effort and cannot block a person's terminal. + } + } + this.terminalInputEpochs.set( + id, + (this.terminalInputEpochs.get(id) ?? 0) + 1, + ); + const staged = this.stagedInputs.get(id); + if (staged?.handle === handle && !staged.preempted) { + staged.preempted = true; + if (staged.textWritten) { + // Remove the server-staged line before forwarding the person's bytes, + // so the two inputs can never be submitted as one corrupted prompt. + if (!this.abandonStagedLine(handle)) { + throw new SessionInputIsolationError(); + } + staged.lineCleared = true; + } + } + try { + handle.pty.write(data); + } catch (error) { + // Raw terminal data can also be reported failed after staging a prefix. + // Fence the exact handle before any later user or server-owned write; + // bracketed-paste input requires its closing marker during recovery. + this.markComposerUnsafe(handle, data.includes(BRACKETED_PASTE_START)); + throw error; + } + this.observeTrustedTerminalInput(handle, data); if (session) { session.lastActiveAt = this.now(); void this.persist(); @@ -1123,6 +1533,8 @@ export class SessionManager { text: string, submit = true, canWrite?: () => boolean | Promise, + background = false, + lifecycle?: SessionInputWriteLifecycle, ): Promise { const remainsAuthorized = async (): Promise => { if (!canWrite) return true; @@ -1134,6 +1546,7 @@ export class SessionManager { }; const session = this.sessions.get(id); if (!session) return false; + const initialTerminalInputEpoch = this.terminalInputEpochs.get(id) ?? 0; // An external-harness session (e.g. conductor) never has a pty — surfacing // HARNESS_EXTERNAL here gives a 409 "managed by the X app" instead of a @@ -1141,7 +1554,8 @@ export class SessionManager { const handle = this.ptys.get(id); if (!handle) { const info = listHarnessAdapters().find((a) => a.id === session.harness); - if (info?.mode === "external") throw new ExternalHarnessError(session.harness, info.label); + if (info?.mode === "external") + throw new ExternalHarnessError(session.harness, info.label); return false; } if (!this.isReadyEnough(session, handle)) { @@ -1158,12 +1572,64 @@ export class SessionManager { if (canWrite && !(await remainsAuthorized())) { throw new SessionInputGuardRejectedError(false); } + if ( + background && + (this.terminalInputEpochs.get(id) ?? 0) !== initialTerminalInputEpoch + ) { + throw new SessionBackgroundInputPreemptedError(false); + } + if (!this.resetUnsafeComposer(handle)) { + // No byte from this logical submission has been written, but the prior + // unknown composer cannot yet be declared clean. Let durable callers + // keep their request queued; never append it to residual text. + await lifecycle?.onNotSubmitted?.().catch(() => {}); + throw new SessionInputIsolationError(); + } + if (lifecycle?.beforeFirstWrite) { + try { + await lifecycle.beforeFirstWrite(); + } catch (error) { + await lifecycle.onNotSubmitted?.().catch(() => {}); + throw error; + } + } + if ( + background && + (this.terminalInputEpochs.get(id) ?? 0) !== initialTerminalInputEpoch + ) { + // The durable claim may yield while a person types into this composer. + // Retire that claim without appending background text or pressing Enter. + await lifecycle?.onNotSubmitted?.().catch(() => {}); + throw new SessionBackgroundInputPreemptedError(false); + } + if (this.closing || (lifecycle?.canWriteNow && !lifecycle.canWriteNow())) { + await lifecycle?.onNotSubmitted?.().catch(() => {}); + throw new SessionInputGuardRejectedError(false); + } if (!submit) { - handle.pty.write(text); + try { + lifecycle?.onWritePhase?.("text-staged"); + handle.pty.write(text); + } catch (error) { + // submit:false is public arbitrary draft text, not a single control + // byte. A partial failure can therefore leave composer content even + // though no Enter was requested. + this.markComposerUnsafe(handle, false); + throw error; + } this.observeTrustedSubmittedText(handle, text); } else if (text.length === 0) { - handle.pty.write("\r"); + try { + lifecycle?.onWritePhase?.("text-staged"); + handle.pty.write("\r"); + lifecycle?.onWritePhase?.("enter-written"); + } catch (error) { + // Enter may have crossed or may have left an existing draft intact. + // Either way, require a proven reset before another submission. + this.markComposerUnsafe(handle, false); + throw error; + } this.observeTrustedTerminalInput(handle, "\r"); } else { // Bracketed when the app supports it: newlines in the prompt (the canvas @@ -1186,27 +1652,103 @@ export class SessionManager { ? handle.bracketedPaste.enabled : this.platform === "win32" && (this.adapters[session.harness]?.assumesBracketedPaste ?? false); - handle.pty.write(paste ? wrapPaste(text) : text); - // Observe the server-owned plaintext rather than the bracketed-paste - // transport wrapper. Embedded newlines invalidate the line, so prompt - // text containing `/clear` cannot impersonate an exact slash command. - this.observeTrustedSubmittedText(handle, text); - await sleep(SUBMIT_DELAY_MS); - // The pty may have been killed/replaced while we were waiting. - if (this.ptys.get(id) !== handle) return false; - // A project/account can change during the deliberate text→Enter delay. - // Do not submit the staged text under stale authority. - if (canWrite && !(await remainsAuthorized())) { - // Text was staged but not submitted. Clear the composer before - // releasing control so a later keypress cannot submit project-scoped - // content into the now-stale planner. Ctrl-U is a local line-clear, - // not an Enter/submission gesture. - handle.pty.write("\x15"); - this.observeTrustedTerminalInput(handle, "\x15"); - throw new SessionInputGuardRejectedError(true); + const staged = { + handle, + background, + preempted: false, + textWritten: false, + lineCleared: false, + }; + if (this.stagedInputs.has(id)) { + await lifecycle?.onNotSubmitted?.().catch(() => {}); + throw new SessionBackgroundInputPreemptedError(false); + } + this.stagedInputs.set(id, staged); + let enterAttempted = false; + try { + try { + lifecycle?.onWritePhase?.("text-staged"); + handle.pty.write(paste ? wrapPaste(text) : text); + } catch (error) { + // A PTY can report a text-write failure after staging a prefix. Clear + // that possible partial line before claiming the logical submission + // is safe to retry. A partial bracketed paste can leave the terminal + // inside paste mode, where Ctrl-U is merely pasted content, so close + // that mode first. Every required cleanup write must succeed before + // exposing positive not-submitted evidence. + let pasteClosed = !paste; + try { + if (paste) { + handle.pty.write(BRACKETED_PASTE_END); + this.observeTrustedTerminalInput(handle, BRACKETED_PASTE_END); + pasteClosed = true; + } + } catch { + // Still attempt a non-submitting line clear below, but do not call + // it proof when the terminal may remain in bracketed-paste mode. + } + try { + handle.pty.write("\x15"); + this.observeTrustedTerminalInput(handle, "\x15"); + staged.lineCleared = pasteClosed; + } catch { + // The coordinator retains its dispatch intent and fails closed. + } + if (!staged.lineCleared) { + this.markComposerUnsafe(handle, !pasteClosed); + } + throw error; + } + staged.textWritten = true; + // Observe the server-owned plaintext rather than the bracketed-paste + // transport wrapper. Embedded newlines invalidate the line, so prompt + // text containing `/clear` cannot impersonate an exact slash command. + this.observeTrustedSubmittedText(handle, text); + await sleep(SUBMIT_DELAY_MS); + if (staged.preempted) { + throw new SessionBackgroundInputPreemptedError(true); + } + // The pty may have been killed/replaced while we were waiting. + if (this.ptys.get(id) !== handle) return false; + // A project/account can change during the deliberate text→Enter delay. + // Do not submit the staged text under stale authority. + if (canWrite && !(await remainsAuthorized())) { + // Text was staged but not submitted. Clear the composer before + // releasing control so a later keypress cannot submit project-scoped + // content into the now-stale session. Ctrl-U is a local line-clear, + // not an Enter/submission gesture. + staged.lineCleared = this.abandonStagedLine(handle); + throw new SessionInputGuardRejectedError(true); + } + // `await remainsAuthorized()` necessarily yields. Shutdown or a + // coordinator generation change can win in that gap, so this final + // synchronous fence is the last operation before Enter. + if ( + this.closing || + (lifecycle?.canWriteNow && !lifecycle.canWriteNow()) + ) { + staged.lineCleared = this.abandonStagedLine(handle); + throw new SessionInputGuardRejectedError(true); + } + enterAttempted = true; + handle.pty.write("\r"); + lifecycle?.onWritePhase?.("enter-written"); + this.observeTrustedTerminalInput(handle, "\r"); + } catch (error) { + if (enterAttempted) { + // A failed Enter write is ambiguous: it may have submitted A or may + // have left A's full staged line in place. Fence the handle so B can + // never append until a non-submitting reset succeeds. + this.markComposerUnsafe(handle, false); + } else if (staged.lineCleared) { + await lifecycle?.onNotSubmitted?.().catch(() => {}); + } + throw error; + } finally { + if (this.stagedInputs.get(id) === staged) { + this.stagedInputs.delete(id); + } } - handle.pty.write("\r"); - this.observeTrustedTerminalInput(handle, "\r"); } session.lastActiveAt = this.now(); @@ -1386,9 +1928,10 @@ export class SessionManager { id: string, agentSessionId: string, source?: unknown, + runtimeEpoch?: string, ): Promise { return this.serializeAgentSessionIdentity(() => - this.setAgentSessionIdLocked(id, agentSessionId, source), + this.setAgentSessionIdLocked(id, agentSessionId, source, runtimeEpoch), ); } @@ -1414,19 +1957,23 @@ export class SessionManager { id: string, agentSessionId: string, source?: unknown, + runtimeEpoch?: string, ): Promise { const session = this.sessions.get(id); if (!session) return false; const handle = this.ptys.get(id); - const transitionSource = source === "clear" || source === "resume" ? source : null; + if (runtimeEpoch !== undefined && handle?.runtimeEpoch !== runtimeEpoch) { + return false; + } + const transitionSource = + source === "clear" || source === "resume" ? source : null; let authorization = handle?.agentSessionRotation ?? null; if (handle && authorization && authorization.expiresAt <= Date.now()) { handle.agentSessionRotation = null; authorization = null; } const matchesAuthorization = - transitionSource !== null && - authorization?.source === transitionSource; + transitionSource !== null && authorization?.source === transitionSource; // A matching clear/resume SessionStart consumes the user gesture even // when this is the first vendor id, Claude keeps the same id, or the @@ -1451,7 +1998,8 @@ export class SessionManager { // pointers. A fresh session can therefore never reclaim A's old id and // merge its events/transcript with A after a restart. if (ownerId !== undefined && ownerId !== id) return false; - if (ownerId === undefined) await this.reserveAgentSessionIdentity(digest, id); + if (ownerId === undefined) + await this.reserveAgentSessionIdentity(digest, id); const candidate = { ...session, agentSessionId }; let releaseFence: () => void = () => {}; @@ -1634,15 +2182,36 @@ export class SessionManager { * translated for Codex), or from an adapter-declared readiness fallback. * Idempotent; a session that's exited or already ready is a silent no-op. */ - setReady(id: string): void { + setReady(id: string, runtimeEpoch?: string): void { const session = this.sessions.get(id); - if (!session || session.ready) return; + if ( + !session || + session.status === "exited" || + session.ready || + (runtimeEpoch !== undefined && + this.ptys.get(id)?.runtimeEpoch !== runtimeEpoch) + ) + return; session.ready = true; void this.persist(); this.emitStatus(session); } /** Persist a coordinator-owned metadata projection before exposing it. */ + + + /** Persist the neutral project-bootstrap projection before exposing it. */ + async setProjectBootstrapMetadata( + id: string, + metadata: ProjectBootstrapMetadata, + ): Promise { + const session = this.sessions.get(id); + if (!session) throw new UnknownSessionError(id); + session.projectBootstrap = structuredClone(metadata); + await this.persist(); + this.emitStatus(session); + } + async setPlanningMetadata( id: string, metadata: PlannerSessionMetadata, @@ -1869,9 +2438,21 @@ export class SessionManager { this.emitStatus(session); } - private async spawn(session: HarnessSession, spec: SpawnSpec): Promise { + private async spawn( + session: HarnessSession, + spec: SpawnSpec, + revalidateAdmission?: () => Promise, + ): Promise { + if (this.closing) throw new SessionManagerClosingError(); const adapter = this.getAdapter(session.harness); - const spawnFn = this.spawnPty ?? (await loadDefaultSpawn()); + const spawnFn = this.spawnPty ?? (await this.loadSpawnPty()); + // Loading node-pty is lazy and asynchronous. Revalidate the project + // principal only after that final setup await; a binding or authenticated + // user can change while the module loads. The closing check follows the + // authorization await and then admission remains synchronous, so + // beginShutdown/killAll cannot miss a newly admitted process either. + await revalidateAdmission?.(); + if (this.closing) throw new SessionManagerClosingError(); const env: Record = {}; for (const [key, value] of Object.entries(process.env)) { if (value !== undefined) env[key] = value; @@ -1892,7 +2473,12 @@ export class SessionManager { env.TERM = "xterm-256color"; env.COLORTERM = "truecolor"; env[ENV.ingestUrl] = `${this.ingestUrl.replace(/\/$/, "")}/ingest`; - env[ENV.ingestToken] = this.issueIngestToken(session.id); + const ingestCredential = this.issueIngestCredential(session.id); + this.adapterIdentityStates.set(session.id, { + runtimeEpoch: ingestCredential.runtimeEpoch, + state: adapter.eventSource === "transcript-tail" ? "pending" : "not-required", + }); + env[ENV.ingestToken] = ingestCredential.token; env[ENV.sessionId] = session.id; if (this.collectorUrl) env[ENV.collectorUrl] = this.collectorUrl; @@ -1904,17 +2490,45 @@ export class SessionManager { // No-op on POSIX. const target = resolveSpawnTarget(spec.command, spec.args); - // A throw here — spawnFn itself, or loadDefaultSpawn() above (a broken - // node-pty prebuild surfaces there, not at import time) — propagates to - // create()/resume(), which own reconciling the session record to - // "exited" for every pre-pty failure, not just this one. - const pty: IPty = spawnFn(target.command, target.args, { - name: "xterm-256color", - cols: DEFAULT_COLS, - rows: DEFAULT_ROWS, - cwd: spec.cwd, - env, - }); + let epochTransitioned = false; + let pty: IPty; + try { + // The coordinator must durably retire every owner from the prior PTY + // generation before this replacement becomes observable. Its epoch is + // server-generated beside the private ingest capability and never comes + // from a hook, model payload, or browser request. + await this.onRuntimeEpochTransition?.( + { ...session }, + ingestCredential.runtimeEpoch, + ); + epochTransitioned = true; + // Epoch transition is fallible and may wait on durable state. Revalidate + // project/user scope and shutdown admission once more after that await. + await revalidateAdmission?.(); + if (this.closing) throw new SessionManagerClosingError(); + // A throw here — spawnFn itself, or loadDefaultSpawn() above (a broken + // node-pty prebuild surfaces there, not at import time) — propagates to + // create()/resume(), which own reconciling the session record to + // "exited" for every pre-pty failure, not just this one. + pty = spawnFn(target.command, target.args, { + name: "xterm-256color", + cols: DEFAULT_COLS, + rows: DEFAULT_ROWS, + cwd: spec.cwd, + env, + }); + } catch (error) { + this.revokeIngestToken(session.id); + const identityState = this.adapterIdentityStates.get(session.id); + if (identityState?.runtimeEpoch === ingestCredential.runtimeEpoch) + this.adapterIdentityStates.delete(session.id); + if (epochTransitioned) { + await Promise.resolve( + this.onRuntimeEpochTransition?.({ ...session }, null), + ).catch(() => {}); + } + throw error; + } const emitter = new EventEmitter(); emitter.setMaxListeners(0); @@ -1924,6 +2538,7 @@ export class SessionManager { }); const handle: PtyHandle = { pty, + runtimeEpoch: ingestCredential.runtimeEpoch, buffer: "", readinessBuffer: "", readinessHistory: "", @@ -1945,6 +2560,7 @@ export class SessionManager { resolveExited, killed: false, }; + this.retiredRuntimeEpochs.delete(session.id); this.ptys.set(session.id, handle); session.status = "running"; @@ -2059,7 +2675,11 @@ export class SessionManager { * silent no-op rather than double-transitioning or clobbering a newer * session/handle that's since taken its place (e.g. a resume). */ - private markExited(id: string, handle: PtyHandle, exitCode: number | null): void { + private markExited( + id: string, + handle: PtyHandle, + exitCode: number | null, + ): void { if (this.ptys.get(id) !== handle) return; // Preserve the tail of output BEFORE the handle (and its buffer) is dropped // — this is the only chance to keep the agent's own error line. Worth it @@ -2076,13 +2696,20 @@ export class SessionManager { ? sanitizeExitTail(handle.buffer) : null; this.ptys.delete(id); + const identityState = this.adapterIdentityStates.get(id); + if (identityState?.runtimeEpoch === handle.runtimeEpoch) + this.adapterIdentityStates.delete(id); + this.retiredRuntimeEpochs.set(id, handle.runtimeEpoch); this.lastActivityBroadcast.delete(id); // Resolve after the pty map is cleaned up. transitionExited runs // synchronously to set status before any awaiting continuation resumes. handle.resolveExited(); const session = this.sessions.get(id); if (!session) return; - void this.transitionExited(session, exitCode, { exitTail }); + void this.transitionExited(session, exitCode, { + exitTail, + runtimeEpoch: handle.runtimeEpoch, + }); } /** @@ -2096,11 +2723,21 @@ export class SessionManager { private transitionExited( session: HarnessSession, exitCode: number | null, - { stampLastActive = true, exitTail = null }: { stampLastActive?: boolean; exitTail?: string | null } = {}, + { + stampLastActive = true, + exitTail = null, + runtimeEpoch = null, + }: { + stampLastActive?: boolean; + exitTail?: string | null; + runtimeEpoch?: string | null; + } = {}, ): Promise { this.revokeIngestToken(session.id); try { - void Promise.resolve(this.onAgentMapSessionExit?.(session.id)).catch(() => {}); + void Promise.resolve(this.onAgentMapSessionExit?.(session.id)).catch( + () => {}, + ); } catch { // Capability cleanup never delays durable session reconciliation. } @@ -2116,12 +2753,19 @@ export class SessionManager { // what made an untouched session's duration grow on every failed Resume. if (stampLastActive) session.lastActiveAt = this.now(); const persisted = this.persist(); - this.emitStatus(session); + this.emitStatus(session, runtimeEpoch); return persisted; } - private emitStatus(session: HarnessSession): void { - this.statusEmitter.emit("status", { ...session }); + private emitStatus( + session: HarnessSession, + runtimeEpoch = this.getRuntimeEpoch(session.id), + ): void { + this.statusEmitter.emit( + "status", + { ...session }, + { runtimeEpoch } satisfies SessionStatusContext, + ); } private agentSessionIdentityDigest(agentSessionId: string): string { diff --git a/packages/harness/src/core/task-manager.ts b/packages/harness/src/core/task-manager.ts index 413ac3d0d..0aa103640 100644 --- a/packages/harness/src/core/task-manager.ts +++ b/packages/harness/src/core/task-manager.ts @@ -183,7 +183,7 @@ export class TaskManager { this.adapters = options.adapters; this.ingestUrl = options.ingestUrl; this.issueIngestToken = (sessionId) => - options.ingestCredentials.issue(sessionId); + options.ingestCredentials.issue(sessionId).token; this.revokeIngestToken = (sessionId) => options.ingestCredentials.revoke(sessionId); this.collectorUrl = options.collectorUrl; diff --git a/packages/harness/src/server/index.ts b/packages/harness/src/server/index.ts index 61d9d06c7..1fde5b7d9 100644 --- a/packages/harness/src/server/index.ts +++ b/packages/harness/src/server/index.ts @@ -3557,9 +3557,11 @@ export const startServer = async ( const ingestDeps: IngestDeps = { authenticate: (sessionId, token) => ingestCredentials.authenticate(sessionId, token), + isCurrentRuntime: (sessionId, runtimeEpoch) => + sessionManager.acceptsIngestRuntimeEpoch(sessionId, runtimeEpoch), normalize: normalizeHookEvent, resolveSession: resolveIngestSession, - onAgentSessionResolved: (harnessSessionId, agentSessionId, source) => { + onAgentSessionResolved: (harnessSessionId, agentSessionId, source, runtimeEpoch) => { // Record the agent session id — used by session-manager for resume // (agentSessionId feeds the --resume flag) and by the codex tailer for // exact-match rollout discovery. @@ -3567,10 +3569,11 @@ export const startServer = async ( harnessSessionId, agentSessionId, source, + runtimeEpoch, ); }, - onSessionReady: (harnessSessionId) => { - sessionManager.setReady(harnessSessionId); + onSessionReady: (harnessSessionId, runtimeEpoch) => { + sessionManager.setReady(harnessSessionId, runtimeEpoch); }, store: eventStore, batcher, @@ -3665,6 +3668,8 @@ export const startServer = async ( const session = sessionManager.get(harnessSessionId); if (!session) return; + const runtimeEpoch = sessionManager.getRuntimeEpoch(harnessSessionId); + if (runtimeEpoch === null) return; const rolloutPath = await discoverCodexRolloutPath(session); if (!rolloutPath) { console.error( @@ -3675,7 +3680,7 @@ export const startServer = async ( // 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.get(harnessSessionId)?.status !== "running") return; + if (!sessionManager.isCurrentRuntimeEpoch(harnessSessionId, runtimeEpoch)) return; const tailer = tailCodexRollout({ rolloutPath, @@ -3692,7 +3697,7 @@ export const startServer = async ( harnessSessionId, payload, }; - void processIngest(body, ingestDeps, seqCounter).catch( + void processIngest(body, ingestDeps, seqCounter, runtimeEpoch).catch( (err: unknown) => { console.error("[harness] codex tailer ingest error:", err); }, @@ -3840,6 +3845,7 @@ export const startServer = async ( // itself is bounded (KILL_ESCALATION_MS + KILL_ESCALATION_CONFIRM_MS // = 2500ms); the outer timeout here is a final safety net above that. const SHUTDOWN_KILL_TIMEOUT_MS = 5_000; + sessionManager.beginShutdown(); const killsSettled = Promise.all([ sessionManager.killAll(), taskManager.killAll(), diff --git a/packages/harness/src/server/ingest.test.ts b/packages/harness/src/server/ingest.test.ts index 6d6e765c1..00316e040 100644 --- a/packages/harness/src/server/ingest.test.ts +++ b/packages/harness/src/server/ingest.test.ts @@ -22,6 +22,7 @@ import { } from "./ingest.js"; const INGEST_TOKEN = "test-token"; +const RUNTIME_EPOCH = "runtime-epoch-1"; function postIngest(baseUrl: string, body: unknown, token = INGEST_TOKEN) { return fetch(`${baseUrl}/ingest`, { @@ -82,7 +83,11 @@ describe("createIngestRouter", () => { const deps: IngestDeps = { authenticate: (sessionId, token) => - sessionId === "session-1" && token === INGEST_TOKEN, + sessionId === "session-1" && token === INGEST_TOKEN + ? RUNTIME_EPOCH + : null, + isCurrentRuntime: (_sessionId, runtimeEpoch) => + runtimeEpoch === RUNTIME_EPOCH, normalize: normalizeHookEvent, resolveSession: (harnessSessionId) => sessions.get(harnessSessionId), onAgentSessionResolved: (harnessSessionId, agentSessionId, source) => { @@ -124,7 +129,11 @@ describe("createIngestRouter", () => { }); it("rejects requests without a valid bearer token", async () => { - const res = await postIngest(baseUrl, { hookEvent: "SessionStart" }, "wrong-token"); + const res = await postIngest( + baseUrl, + { hookEvent: "SessionStart" }, + "wrong-token", + ); expect(res.status).toBe(401); expect(stored).toHaveLength(0); }); @@ -155,8 +164,11 @@ describe("createIngestRouter", () => { it("binds a valid bearer token to the body session id", async () => { start({ authenticate: (sessionId, token) => - (sessionId === "session-1" && token === INGEST_TOKEN) || - (sessionId === "session-2" && token === "token-2"), + sessionId === "session-1" && token === INGEST_TOKEN + ? RUNTIME_EPOCH + : sessionId === "session-2" && token === "token-2" + ? RUNTIME_EPOCH + : null, }); sessions.set("session-2", { harness: "claude-code", @@ -207,6 +219,63 @@ describe("createIngestRouter", () => { expect(enqueued).toHaveLength(1); }); + it("keeps the authenticated runtime epoch across asynchronous HTTP processing", async () => { + const enrichmentEntered = deferred(); + const enrichmentCommit = deferred(); + let activeEpoch = "epoch-a"; + const persistedEpochs: string[] = []; + start({ + authenticate: (_sessionId, token) => + token === "token-a" + ? "epoch-a" + : token === "token-b" + ? "epoch-b" + : null, + isCurrentRuntime: (_sessionId, runtimeEpoch) => + runtimeEpoch === activeEpoch, + enrichFromTranscript: async (event) => { + enrichmentEntered.resolve(); + await enrichmentCommit.promise; + return event; + }, + onEventPersisted: (_event, runtimeEpoch) => + persistedEpochs.push(runtimeEpoch), + }); + + const acceptedOld = await postIngest( + baseUrl, + { + hookEvent: "Stop", + harnessSessionId: "session-1", + payload: { last_assistant_message: "old reply" }, + }, + "token-a", + ); + expect(acceptedOld.status).toBe(200); + await enrichmentEntered.promise; + + activeEpoch = "epoch-b"; + enrichmentCommit.resolve(); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(stored).toEqual([]); + expect(enqueued).toEqual([]); + + const acceptedCurrent = await postIngest( + baseUrl, + { + hookEvent: "UserPromptSubmit", + harnessSessionId: "session-1", + // Payload/model input cannot select the trusted server-side epoch. + payload: { prompt: "current prompt", runtimeEpoch: "epoch-a" }, + }, + "token-b", + ); + expect(acceptedCurrent.status).toBe(200); + await vi.waitFor(() => expect(stored).toHaveLength(1)); + expect(stored[0]?.payload.prompt).toBe("current prompt"); + expect(persistedEpochs).toEqual(["epoch-b"]); + }); + it("notifies transcript consumers only after the event is durably appended", async () => { await new Promise((resolve) => server.close(() => resolve())); const appendEntered = deferred(); @@ -274,7 +343,8 @@ describe("createIngestRouter", () => { const decorated: AnalyticsEvent[] = []; const batched: AnalyticsEvent[] = []; const deps: IngestDeps = { - authenticate: () => true, + authenticate: () => RUNTIME_EPOCH, + isCurrentRuntime: () => true, normalize: normalizeHookEvent, resolveSession: () => context, onAgentSessionResolved: async (_harnessId, agentSessionId) => { @@ -299,6 +369,7 @@ describe("createIngestRouter", () => { }, deps, seqCounter, + RUNTIME_EPOCH, ); await identityEntered.promise; @@ -312,6 +383,7 @@ describe("createIngestRouter", () => { }, { ...deps }, seqCounter, + RUNTIME_EPOCH, ); await Promise.resolve(); await Promise.resolve(); @@ -335,6 +407,108 @@ describe("createIngestRouter", () => { expect(batched).toHaveLength(2); }); + it("does not let a stale SessionStart mark a replacement runtime ready", async () => { + const identityEntered = deferred(); + const identityCommit = deferred(); + let activeEpoch = "epoch-a"; + const ready: Array<{ sessionId: string; runtimeEpoch: string }> = []; + const context: IngestSessionContext = { + harness: "claude-code", + userId: "user-1", + tenantId: "tenant-1", + machineId: "machine-1", + agentSessionId: "provider-shared", + }; + const deps: IngestDeps = { + authenticate: () => activeEpoch, + isCurrentRuntime: (_sessionId, runtimeEpoch) => + runtimeEpoch === activeEpoch, + normalize: normalizeHookEvent, + resolveSession: () => context, + onAgentSessionResolved: async () => { + identityEntered.resolve(); + await identityCommit.promise; + return true; + }, + onSessionReady: (sessionId, runtimeEpoch) => + ready.push({ sessionId, runtimeEpoch }), + store: { append: async () => {} }, + batcher: { enqueue: () => {} }, + }; + const processing = processIngest( + { + hookEvent: "SessionStart", + harnessSessionId: "session-replaced", + payload: { session_id: "provider-shared", source: "resume" }, + }, + deps, + createSeqCounter(), + "epoch-a", + ); + await identityEntered.promise; + + activeEpoch = "epoch-b"; + identityCommit.resolve(); + await processing; + + expect(ready).toEqual([]); + }); + + it("does not let a SessionEnd from a retired runtime reset the replacement sequence", async () => { + const appendEntered = deferred(); + const appendCommit = deferred(); + const resets: string[] = []; + let activeEpoch = "epoch-a"; + const context: IngestSessionContext = { + harness: "claude-code", + userId: "user-1", + tenantId: "tenant-1", + machineId: "machine-1", + agentSessionId: "provider-shared", + }; + const deps: IngestDeps = { + authenticate: () => activeEpoch, + isCurrentRuntime: (_sessionId, runtimeEpoch) => + runtimeEpoch === activeEpoch, + normalize: normalizeHookEvent, + resolveSession: () => context, + onAgentSessionResolved: () => true, + store: { + append: async () => { + appendEntered.resolve(); + await appendCommit.promise; + }, + }, + batcher: { enqueue: () => {} }, + }; + let nextSeq = 0; + const seqCounter: ReturnType = { + next: () => ++nextSeq, + reset: (sessionId: string) => { + resets.push(sessionId); + nextSeq = 0; + }, + }; + const processing = processIngest( + { + hookEvent: "SessionEnd", + harnessSessionId: "session-replaced", + payload: { reason: "old process exited" }, + }, + deps, + seqCounter, + "epoch-a", + ); + await appendEntered.promise; + + activeEpoch = "epoch-b"; + appendCommit.resolve(); + await processing; + + expect(resets).toEqual([]); + expect(seqCounter.next("session-replaced")).toBe(2); + }); + it("lets a follower re-resolve the prior pin after a SessionStart commit failure", async () => { const identityCommit = deferred(); const identityEntered = deferred(); @@ -348,7 +522,8 @@ describe("createIngestRouter", () => { const appended: AnalyticsEvent[] = []; const decorated: AnalyticsEvent[] = []; const deps: IngestDeps = { - authenticate: () => true, + authenticate: () => RUNTIME_EPOCH, + isCurrentRuntime: () => true, normalize: normalizeHookEvent, resolveSession: () => context, onAgentSessionResolved: async () => { @@ -372,6 +547,7 @@ describe("createIngestRouter", () => { }, deps, seqCounter, + RUNTIME_EPOCH, ); await identityEntered.promise; const followerProcessing = processIngest( @@ -382,13 +558,16 @@ describe("createIngestRouter", () => { }, deps, seqCounter, + RUNTIME_EPOCH, ); await Promise.resolve(); expect(appended).toEqual([]); expect(decorated).toEqual([]); identityCommit.reject(new Error("sessions registry unavailable")); - await expect(startProcessing).rejects.toThrow("sessions registry unavailable"); + await expect(startProcessing).rejects.toThrow( + "sessions registry unavailable", + ); await followerProcessing; expect(context.agentSessionId).toBe("agent-prior"); @@ -409,7 +588,9 @@ describe("createIngestRouter", () => { start({ authenticate: (sessionId, token) => (sessionId === "session-1" || sessionId === "session-2") && - token === INGEST_TOKEN, + token === INGEST_TOKEN + ? RUNTIME_EPOCH + : null, store: eventStore, }); sessions.get("session-1")!.agentSessionId = "agent-a"; @@ -477,7 +658,9 @@ describe("createIngestRouter", () => { it("ignores a SessionStart whose vendor identity conflicts with the pinned session", async () => { const ready: string[] = []; - start({ onSessionReady: (harnessSessionId) => ready.push(harnessSessionId) }); + start({ + onSessionReady: (harnessSessionId) => ready.push(harnessSessionId), + }); sessions.get("session-1")!.agentSessionId = "agent-pinned"; const res = await postIngest(baseUrl, { @@ -524,7 +707,9 @@ describe("createIngestRouter", () => { it("calls onSessionReady on session.start — the readiness signal SessionManager gates programmatic input on", async () => { const ready: string[] = []; - start({ onSessionReady: (harnessSessionId) => ready.push(harnessSessionId) }); + start({ + onSessionReady: (harnessSessionId) => ready.push(harnessSessionId), + }); const res = await postIngest(baseUrl, { hookEvent: "SessionStart", @@ -538,7 +723,9 @@ describe("createIngestRouter", () => { it("does not call onSessionReady for events other than SessionStart", async () => { const ready: string[] = []; - start({ onSessionReady: (harnessSessionId) => ready.push(harnessSessionId) }); + start({ + onSessionReady: (harnessSessionId) => ready.push(harnessSessionId), + }); await postIngest(baseUrl, { hookEvent: "UserPromptSubmit", @@ -620,11 +807,11 @@ describe("createIngestRouter", () => { start({ decorateEvent: (event) => ({ ...event, - payload: { ...event.payload, plannerOrigin: "infrastructure" }, + payload: { ...event.payload, projectBootstrapOrigin: "infrastructure" }, }), projectTelemetryEvent: (event) => ({ ...event, - payload: { planner: true, origin: event.payload.plannerOrigin }, + payload: { bootstrap: true, origin: event.payload.projectBootstrapOrigin }, }), }); @@ -637,10 +824,10 @@ describe("createIngestRouter", () => { await vi.waitFor(() => expect(stored).toHaveLength(1)); expect(stored[0].payload).toMatchObject({ prompt: "private control prompt", - plannerOrigin: "infrastructure", + projectBootstrapOrigin: "infrastructure", }); expect(enqueued[0].payload).toEqual({ - planner: true, + bootstrap: true, origin: "infrastructure", }); expect(JSON.stringify(enqueued[0])).not.toContain("private control prompt"); diff --git a/packages/harness/src/server/ingest.ts b/packages/harness/src/server/ingest.ts index 382b46f9e..2bbdcf7a3 100644 --- a/packages/harness/src/server/ingest.ts +++ b/packages/harness/src/server/ingest.ts @@ -25,8 +25,14 @@ export interface IngestSessionContext { } export interface IngestDeps { - /** Authenticate the bearer capability against the body session id. */ - authenticate: (harnessSessionId: string, token: string) => boolean; + /** Authenticate and return the server-owned epoch bound to this capability. */ + authenticate: (harnessSessionId: string, token: string) => string | null; + /** + * True while this exact PTY generation may finish ingest work. A terminal + * event already admitted before exit may complete until a replacement PTY + * takes ownership; the epoch still comes only from trusted server state. + */ + isCurrentRuntime: (harnessSessionId: string, runtimeEpoch: string) => boolean; /** Raw hook payload -> AnalyticsEvent, or null to skip (e.g. PreToolUse). */ normalize: ( hookEvent: string, @@ -41,6 +47,7 @@ export interface IngestDeps { harnessSessionId: string, agentSessionId: string, source: unknown, + runtimeEpoch: string, ) => boolean | Promise; /** * Called once a SessionStart(-equivalent) event is actually processed for @@ -50,7 +57,7 @@ export interface IngestDeps { * event), kept separate since "ready" and "agent session id known" are * conceptually distinct even though they happen to co-occur today. */ - onSessionReady?: (harnessSessionId: string) => void; + onSessionReady?: (harnessSessionId: string, runtimeEpoch: string) => void; store: { append(event: AnalyticsEvent): Promise }; batcher: { enqueue(event: AnalyticsEvent): void }; /** Optional transcript backfill for turn.completed / session.end. */ @@ -61,11 +68,17 @@ export interface IngestDeps { /** Called for every successfully normalized event (after any transcript * enrichment), before it's persisted — e.g. to feed a tool.call event's * command/output text to dev-server port detection. */ - onNormalizedEvent?: (event: AnalyticsEvent) => void; + onNormalizedEvent?: (event: AnalyticsEvent, runtimeEpoch: string) => void; /** Local-only annotation (for example planner control-turn correlation). */ - decorateEvent?: (event: AnalyticsEvent) => AnalyticsEvent; + decorateEvent?: ( + event: AnalyticsEvent, + runtimeEpoch: string, + ) => AnalyticsEvent; /** Content-free projection used only for remote product telemetry. */ - projectTelemetryEvent?: (event: AnalyticsEvent) => AnalyticsEvent; + projectTelemetryEvent?: ( + event: AnalyticsEvent, + runtimeEpoch: string, + ) => AnalyticsEvent; /** * Called for every event AFTER it has been persisted to the local store — * the seam for consumers that need to read the store back and see this event @@ -76,7 +89,7 @@ export interface IngestDeps { * Synchronous and best-effort, like the other hooks here: whatever it starts * is the consumer's to detach, and it must not throw. */ - onEventPersisted?: (event: AnalyticsEvent) => void; + onEventPersisted?: (event: AnalyticsEvent, runtimeEpoch: string) => void; /** * Called for every raw hook event BEFORE normalization — fired even for * hook events that don't produce an analytics event. A UI-transport-only @@ -84,7 +97,12 @@ export interface IngestDeps { * consumers that need to observe raw hook activity. Currently unused; * retained as an extension point. */ - onRawHookEvent?: (hookEvent: string, harnessSessionId: string, payload: Record) => void; + onRawHookEvent?: ( + hookEvent: string, + harnessSessionId: string, + payload: Record, + runtimeEpoch: string, + ) => void; onError?: (err: unknown) => void; /** Injectable for tests; defaults to a fresh per-router counter. */ seqCounter?: SeqCounter; @@ -131,6 +149,7 @@ export async function processIngest( body: IngestRequestBody, deps: IngestDeps, seqCounter: SeqCounter, + runtimeEpoch: string, ): Promise { const hookEvent = body.hookEvent; const harnessSessionId = body.harnessSessionId; @@ -144,7 +163,7 @@ export async function processIngest( const prior = queues.get(harnessSessionId) ?? Promise.resolve(); const next = prior .catch(() => {}) - .then(() => processIngestNow(body, deps, seqCounter)); + .then(() => processIngestNow(body, deps, seqCounter, runtimeEpoch)); queues.set(harnessSessionId, next); try { await next; @@ -157,10 +176,12 @@ async function processIngestNow( body: IngestRequestBody, deps: IngestDeps, seqCounter: SeqCounter, + runtimeEpoch: string, ): Promise { const hookEvent = body.hookEvent!; const harnessSessionId = body.harnessSessionId!; + if (!deps.isCurrentRuntime(harnessSessionId, runtimeEpoch)) return; const session = deps.resolveSession(harnessSessionId); if (!session) return; @@ -169,7 +190,12 @@ async function processIngestNow( // Fire before normalization so a consumer can observe raw hook events, // including ones that produce no analytics event. UI-transport-only seam; // no consumer today. - deps.onRawHookEvent?.(hookEvent, harnessSessionId, hookPayload); + deps.onRawHookEvent?.( + hookEvent, + harnessSessionId, + hookPayload, + runtimeEpoch, + ); const event = deps.normalize(hookEvent, hookPayload, { userId: session.userId, @@ -189,6 +215,7 @@ async function processIngestNow( typeof hookPayload.transcript_path === "string" ? hookPayload.transcript_path : undefined; finalEvent = await deps.enrichFromTranscript(event, transcriptPath); } + if (!deps.isCurrentRuntime(harnessSessionId, runtimeEpoch)) return; if (hookEvent === "SessionStart") { if ( @@ -197,6 +224,7 @@ async function processIngestNow( harnessSessionId, finalEvent.agentSessionId, finalEvent.payload.source, + runtimeEpoch, )) ) { // The bearer capability authenticates the harness session, not an @@ -205,7 +233,8 @@ async function processIngestNow( // local/remote event history under the pinned identity. return; } - deps.onSessionReady?.(harnessSessionId); + if (!deps.isCurrentRuntime(harnessSessionId, runtimeEpoch)) return; + deps.onSessionReady?.(harnessSessionId, runtimeEpoch); } else { // Only SessionStart may propose or rotate a vendor identity through the // authority check above. Every other hook's `payload.session_id` is @@ -219,11 +248,16 @@ async function processIngestNow( }; } - finalEvent = deps.decorateEvent?.(finalEvent) ?? finalEvent; - deps.onNormalizedEvent?.(finalEvent); + if (!deps.isCurrentRuntime(harnessSessionId, runtimeEpoch)) return; + finalEvent = deps.decorateEvent?.(finalEvent, runtimeEpoch) ?? finalEvent; + if (!deps.isCurrentRuntime(harnessSessionId, runtimeEpoch)) return; + deps.onNormalizedEvent?.(finalEvent, runtimeEpoch); await deps.store.append(finalEvent); - deps.onEventPersisted?.(finalEvent); - deps.batcher.enqueue(deps.projectTelemetryEvent?.(finalEvent) ?? finalEvent); + if (!deps.isCurrentRuntime(harnessSessionId, runtimeEpoch)) return; + deps.onEventPersisted?.(finalEvent, runtimeEpoch); + deps.batcher.enqueue( + deps.projectTelemetryEvent?.(finalEvent, runtimeEpoch) ?? finalEvent, + ); if (hookEvent === "SessionEnd") { seqCounter.reset(harnessSessionId); @@ -255,10 +289,14 @@ export function createIngestRouter( : {}; const harnessSessionId = typeof body.harnessSessionId === "string" ? body.harnessSessionId : ""; + const runtimeEpoch = + token !== null && harnessSessionId !== "" + ? deps.authenticate(harnessSessionId, token) + : null; if ( token === null || harnessSessionId === "" || - !deps.authenticate(harnessSessionId, token) + runtimeEpoch === null ) { res.status(401).json({ ok: false }); return; @@ -268,7 +306,7 @@ export function createIngestRouter( // agent's hook pipeline. Processing happens after the response is sent. res.status(200).json({ ok: true }); - void processIngest(body, deps, seqCounter).catch((err) => { + void processIngest(body, deps, seqCounter, runtimeEpoch).catch((err) => { deps.onError?.(err); }); }); diff --git a/packages/harness/src/server/terminal-ws.test.ts b/packages/harness/src/server/terminal-ws.test.ts new file mode 100644 index 000000000..88d2102c0 --- /dev/null +++ b/packages/harness/src/server/terminal-ws.test.ts @@ -0,0 +1,109 @@ +import { EventEmitter } from "node:events"; +import type { IncomingMessage } from "node:http"; +import { describe, expect, it, vi } from "vitest"; +import type { WebSocket } from "ws"; + +import type { SessionManager } from "../core/session-manager.js"; +import { SessionInputIsolationError } from "../core/session-manager.js"; +import { createTerminalWebSocketHandler } from "./terminal-ws.js"; + +const BOOT_TOKEN = "boot-token-123"; + +function createFakeWs() { + const emitter = new EventEmitter(); + const ws = { + readyState: 1, + OPEN: 1, + send: vi.fn(), + close: vi.fn(), + on: (event: string, cb: (...args: unknown[]) => void) => + emitter.on(event, cb), + }; + return { ws: ws as unknown as WebSocket, emitter }; +} + +function createSessionManager( + write: (sessionId: string, text: string) => boolean, +) { + const detach = vi.fn(); + const manager = { + get: vi.fn(() => ({ id: "session-1" })), + attach: vi.fn(() => detach), + resize: vi.fn(), + write: vi.fn(write), + } as unknown as SessionManager; + return { manager, detach }; +} + +describe("createTerminalWebSocketHandler", () => { + it("relays raw input and resize control messages", () => { + const { manager } = createSessionManager(() => true); + const { ws, emitter } = createFakeWs(); + + createTerminalWebSocketHandler(manager, BOOT_TOKEN)( + ws, + {} as IncomingMessage, + new URLSearchParams({ session: "session-1", token: BOOT_TOKEN }), + ); + + emitter.emit("message", Buffer.from("hello"), true); + emitter.emit( + "message", + Buffer.from(JSON.stringify({ type: "resize", cols: 120, rows: 40 })), + false, + ); + + expect(manager.write).toHaveBeenCalledWith("session-1", "hello"); + expect(manager.resize).toHaveBeenCalledWith("session-1", 120, 40); + }); + + it("closes with a fixed content-free reason when composer isolation blocks input", () => { + const { manager, detach } = createSessionManager(() => { + throw new SessionInputIsolationError(); + }); + const { ws, emitter } = createFakeWs(); + + createTerminalWebSocketHandler(manager, BOOT_TOKEN)( + ws, + {} as IncomingMessage, + new URLSearchParams({ session: "session-1", token: BOOT_TOKEN }), + ); + + expect(() => + emitter.emit("message", Buffer.from("private input"), true), + ).not.toThrow(); + expect(ws.close).toHaveBeenCalledWith(1011, "terminal input unavailable"); + expect(JSON.stringify(vi.mocked(ws.close).mock.calls)).not.toContain( + "private input", + ); + expect(detach).toHaveBeenCalledTimes(1); + + emitter.emit("message", Buffer.from("later input"), true); + emitter.emit("close"); + emitter.emit("error", new Error("ignored socket error")); + + expect(manager.write).toHaveBeenCalledTimes(1); + expect(detach).toHaveBeenCalledTimes(1); + }); + + it("does not expose an unexpected PTY error in the close frame", () => { + const { manager } = createSessionManager(() => { + throw new Error("provider payload and path must remain private"); + }); + const { ws, emitter } = createFakeWs(); + + createTerminalWebSocketHandler(manager, BOOT_TOKEN)( + ws, + {} as IncomingMessage, + new URLSearchParams({ session: "session-1", token: BOOT_TOKEN }), + ); + + expect(() => + emitter.emit("message", Buffer.from("private input"), true), + ).not.toThrow(); + expect(ws.close).toHaveBeenCalledWith(1011, "terminal input unavailable"); + expect(JSON.stringify(vi.mocked(ws.close).mock.calls)).not.toContain( + "provider payload", + ); + }); +}); diff --git a/packages/harness/src/server/terminal-ws.ts b/packages/harness/src/server/terminal-ws.ts index 5feba65cd..5db61348e 100644 --- a/packages/harness/src/server/terminal-ws.ts +++ b/packages/harness/src/server/terminal-ws.ts @@ -10,11 +10,18 @@ import type { SessionManager } from "../core/session-manager.js"; import type { TerminalControlMessage } from "../shared/types.js"; import { timingSafeEqualString } from "./auth.js"; +const TERMINAL_INPUT_FAILURE_CODE = 1011; +const TERMINAL_INPUT_FAILURE_REASON = "terminal input unavailable"; + function parseControlMessage(text: string): TerminalControlMessage | undefined { if (!text.startsWith("{")) return undefined; try { const parsed = JSON.parse(text) as Partial; - if (parsed.type === "resize" && typeof parsed.cols === "number" && typeof parsed.rows === "number") { + if ( + parsed.type === "resize" && + typeof parsed.cols === "number" && + typeof parsed.rows === "number" + ) { return { type: "resize", cols: parsed.cols, rows: parsed.rows }; } } catch { @@ -23,8 +30,15 @@ function parseControlMessage(text: string): TerminalControlMessage | undefined { return undefined; } -export function createTerminalWebSocketHandler(sessionManager: SessionManager, bootToken: string) { - return (ws: WebSocket, _req: IncomingMessage, params: URLSearchParams): void => { +export function createTerminalWebSocketHandler( + sessionManager: SessionManager, + bootToken: string, +) { + return ( + ws: WebSocket, + _req: IncomingMessage, + params: URLSearchParams, + ): void => { const sessionId = params.get("session"); const token = params.get("token") ?? ""; @@ -49,7 +63,16 @@ export function createTerminalWebSocketHandler(sessionManager: SessionManager, b return; } + let detached = false; + let inputClosed = false; + const detachOnce = (): void => { + if (detached) return; + detached = true; + detach(); + }; + ws.on("message", (data, isBinary) => { + if (inputClosed) return; const text = data.toString("utf8"); if (!isBinary) { const control = parseControlMessage(text); @@ -58,10 +81,21 @@ export function createTerminalWebSocketHandler(sessionManager: SessionManager, b return; } } - sessionManager.write(sessionId, text); + try { + sessionManager.write(sessionId, text); + } catch { + // A partial PTY write can deliberately fence the composer until it is + // reset. Never let that synchronous safety failure escape EventEmitter + // as an uncaught exception, and never expose input or provider details + // in the close frame. The client may reconnect and retry once the + // handle's non-submitting reset succeeds. + inputClosed = true; + detachOnce(); + ws.close(TERMINAL_INPUT_FAILURE_CODE, TERMINAL_INPUT_FAILURE_REASON); + } }); - ws.on("close", () => detach()); - ws.on("error", () => detach()); + ws.on("close", detachOnce); + ws.on("error", detachOnce); }; }