From 58fb5cfe70f892f1301b6e4c222311b12b333bcc Mon Sep 17 00:00:00 2001 From: Yash Date: Sat, 5 Sep 2026 12:12:00 +0000 Subject: [PATCH] feat(harness): persist delegation ownership [Agent Map 12/15] --- .changeset/durable-delegation-state.md | 5 + .../core/subsession-coordinator-store.test.ts | 1263 ++++++++++ .../src/core/subsession-coordinator-store.ts | 2171 +++++++++++++++++ packages/harness/src/index.ts | 41 + .../subsession-delegation-codec.test.ts | 272 +++ .../src/shared/subsession-delegation-codec.ts | 358 +++ .../src/shared/subsession-delegation.ts | 244 ++ 7 files changed, 4354 insertions(+) create mode 100644 .changeset/durable-delegation-state.md create mode 100644 packages/harness/src/core/subsession-coordinator-store.test.ts create mode 100644 packages/harness/src/core/subsession-coordinator-store.ts create mode 100644 packages/harness/src/shared/subsession-delegation-codec.test.ts create mode 100644 packages/harness/src/shared/subsession-delegation-codec.ts create mode 100644 packages/harness/src/shared/subsession-delegation.ts diff --git a/.changeset/durable-delegation-state.md b/.changeset/durable-delegation-state.md new file mode 100644 index 00000000..71437e0c --- /dev/null +++ b/.changeset/durable-delegation-state.md @@ -0,0 +1,5 @@ +--- +"@sapiom/harness": minor +--- + +Publish delegation request/result contracts, lifecycle record types, bounded limits and canonical codec/digest helpers. Add durable reservations, exact child bindings, request receipts and release history in preparation for tool activation. Retain unfinished private cleanup proof until cleanup completes, clear completed spawn claims on lifecycle transitions, and use locale-independent release-key ordering for durable replay. diff --git a/packages/harness/src/core/subsession-coordinator-store.test.ts b/packages/harness/src/core/subsession-coordinator-store.test.ts new file mode 100644 index 00000000..0c4aca1e --- /dev/null +++ b/packages/harness/src/core/subsession-coordinator-store.test.ts @@ -0,0 +1,1263 @@ +import * as fs from "node:fs/promises"; +import * as os from "node:os"; +import * as path from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; + +import type { ProjectAgentSession } from "../shared/agent-map.js"; +import type { SubsessionBindingId } from "../shared/subsession-delegation.js"; +import { + SubsessionCoordinatorStore, + SubsessionCoordinatorStoreError, +} from "./subsession-coordinator-store.js"; + +const projectId = "project_00000000-0000-4000-8000-000000000001"; +const identity: ProjectAgentSession = { + projectId, + userId: "user-1", + sessionId: "parent-session-1", +}; +const target = { + harness: "codex" as const, + projectRoot: "/project/root", + ownerId: "coordinator-1", +}; + +const delegate = ( + requestKey = "request-1", + delegations: Array<{ + delegationKey: string; + outcome: string; + kickoffContext?: string; + }> = [{ delegationKey: "research", outcome: "Collect evidence" }], +) => ({ + schemaVersion: 1, + requestKey, + operation: { kind: "delegate", delegations }, +}); +const release = ( + requestKey = "release-1", + delegationKeys = ["research"], +) => ({ + schemaVersion: 1, + requestKey, + operation: { kind: "release", delegationKeys }, +}); +const releaseDormant = (requestKey = "release-dormant-1", limit = 16) => ({ + schemaVersion: 1, + requestKey, + operation: { kind: "release-dormant", limit }, +}); +describe("SubsessionCoordinatorStore", () => { + const roots: string[] = []; + + afterEach(async () => { + await Promise.all( + roots.splice(0).map((root) => + fs.rm(root, { recursive: true, force: true }), + ), + ); + }); + + async function fixture() { + const root = await fs.mkdtemp( + path.join(os.tmpdir(), "subsession-coordinator-store-"), + ); + roots.push(root); + return root; + } + + it("converges concurrent instances on one receipt, binding, and reserved real session id", async () => { + const root = await fixture(); + const firstEvent = vi.fn(); + const secondEvent = vi.fn(); + const first = new SubsessionCoordinatorStore(root, { + onEvent: firstEvent, + }); + const second = new SubsessionCoordinatorStore(root, { + onEvent: secondEvent, + }); + + const [left, right] = await Promise.all([ + first.reserveDelegations(identity, delegate(), target), + second.reserveDelegations(identity, delegate(), target), + ]); + const restarted = await new SubsessionCoordinatorStore(root).read(projectId); + + expect(left.bindings).toEqual(right.bindings); + expect([left.replayed, right.replayed].sort()).toEqual([false, true]); + expect(restarted.requestReceipts).toHaveLength(1); + expect(restarted.bindings).toHaveLength(1); + expect(restarted.bindings[0]!.sessionId).toMatch( + /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/u, + ); + const file = path.join(root, "projects", projectId, "subsessions.json"); + expect((await fs.stat(file)).mode & 0o777).toBe(0o600); + expect( + firstEvent.mock.calls.filter( + ([event]) => event.name === "subsession.binding_reserved", + ).length + + secondEvent.mock.calls.filter( + ([event]) => event.name === "subsession.binding_reserved", + ).length, + ).toBe(1); + }); + + it("rejects changed request and binding keys without changing the original", async () => { + const root = await fixture(); + const store = new SubsessionCoordinatorStore(root); + const original = await store.reserveDelegations( + identity, + delegate(), + target, + ); + + await expect( + store.reserveDelegations( + identity, + delegate("request-1", [ + { delegationKey: "research", outcome: "Different task" }, + ]), + target, + ), + ).rejects.toMatchObject({ code: "request_key_reused" }); + await expect( + store.reserveDelegations( + identity, + delegate("request-2", [ + { delegationKey: "research", outcome: "Different task" }, + ]), + target, + ), + ).rejects.toMatchObject({ code: "delegation_key_reused" }); + + const aggregate = await store.read(projectId); + expect(aggregate.requestReceipts).toHaveLength(1); + expect(aggregate.bindings).toEqual(original.bindings); + }); + + it("reserves idempotent releases only for the trusted parent binding", async () => { + const root = await fixture(); + const store = new SubsessionCoordinatorStore(root); + const binding = ( + await store.reserveDelegations(identity, delegate(), target) + ).bindings[0]!; + + const first = await store.reserveReleases(identity, release()); + const replay = await store.reserveReleases(identity, release()); + expect(first).toMatchObject({ + replayed: false, + bindings: [{ state: "bound", binding: { bindingId: binding.bindingId } }], + }); + expect(replay).toEqual({ ...first, replayed: true }); + + const foreign = { ...identity, sessionId: "manual-session" }; + await expect( + store.reserveReleases(foreign, release("foreign-release")), + ).resolves.toMatchObject({ + replayed: false, + bindings: [{ state: "absent", delegationKey: "research" }], + }); + expect((await store.read(projectId)).requestReceipts).toHaveLength(3); + }); + + it("reserves known and unknown release keys independently and replays both", async () => { + const root = await fixture(); + const store = new SubsessionCoordinatorStore(root); + const binding = ( + await store.reserveDelegations(identity, delegate(), target) + ).bindings[0]!; + const request = release("mixed-release", ["missing", "research"]); + + const first = await store.reserveReleases(identity, request); + const replay = await store.reserveReleases(identity, request); + + expect(first.bindings).toEqual([ + { state: "absent", delegationKey: "missing" }, + { state: "bound", binding }, + ]); + expect(replay).toEqual({ ...first, replayed: true }); + expect((await store.read(projectId)).requestReceipts.at(-1)).toMatchObject({ + operation: "release", + bindingIds: [binding.bindingId], + }); + }); + + it("retains a released binding tombstone while an active release receipt references it", async () => { + const root = await fixture(); + const store = new SubsessionCoordinatorStore(root, { + receiptRetentionLimit: 2, + historyTombstoneLimit: 1, + }); + const first = ( + await store.reserveDelegations(identity, delegate("request-1"), target) + ).bindings[0]!; + await store.closeBinding(identity, first.bindingId, first.sessionId); + const second = ( + await store.reserveDelegations( + identity, + delegate("request-2", [ + { delegationKey: "publisher", outcome: "Publish evidence" }, + ]), + target, + ) + ).bindings[0]!; + await store.reserveDelegations( + identity, + delegate("request-3", [ + { delegationKey: "writer", outcome: "Write evidence" }, + ]), + target, + ); + const releaseRequest = release("release-first"); + const released = await store.reserveReleases(identity, releaseRequest); + expect(released.bindings[0]).toMatchObject({ + state: "released", + binding: { bindingId: first.bindingId }, + }); + + await store.closeBinding(identity, second.bindingId, second.sessionId); + await store.reserveDelegations( + identity, + delegate("request-4", [ + { delegationKey: "editor", outcome: "Edit evidence" }, + ]), + target, + ); + + const aggregate = await store.read(projectId); + expect(aggregate.bindingTombstones).toContainEqual( + expect.objectContaining({ bindingId: first.bindingId }), + ); + expect(await store.reserveReleases(identity, releaseRequest)).toMatchObject({ + replayed: true, + bindings: [ + { state: "released", binding: { bindingId: first.bindingId } }, + ], + }); + }); + + it("refreshes child context with an idempotent receipt and a new delivery epoch", async () => { + const root = await fixture(); + const store = new SubsessionCoordinatorStore(root); + const reserved = await store.reserveDelegations( + identity, + delegate(), + target, + ); + const binding = reserved.bindings[0]!; + const request = { + schemaVersion: 1, + requestKey: "refresh-1", + operation: { + kind: "refresh-focused-context", + target: { kind: "child", delegationKey: "research" }, + expectedContextEpoch: binding.contextEpoch, + expectedContextDigest: binding.contextDigest, + focus: null, + }, + } as const; + + const first = await store.refreshFocusedContext(identity, request); + const replay = await store.refreshFocusedContext(identity, request); + + expect(first.replayed).toBe(false); + expect(replay.replayed).toBe(true); + expect(first.binding.contextEpoch).toBe(2); + expect(first.binding.deliveries).toHaveLength(1); + expect(first.binding.deliveries[0]!.contextEpoch).toBe(2); + expect(replay.binding).toEqual(first.binding); + await expect( + store.refreshFocusedContext(identity, { + ...request, + operation: { ...request.operation, expectedContextEpoch: 7 }, + }), + ).rejects.toMatchObject({ code: "request_key_reused" }); + }); + + it("reuses a compatible binding across request keys and reserves a batch atomically", async () => { + const root = await fixture(); + const store = new SubsessionCoordinatorStore(root); + const first = await store.reserveDelegations(identity, delegate(), target); + const second = await store.reserveDelegations( + identity, + delegate("request-2"), + target, + ); + expect(second.replayed).toBe(false); + expect(second.bindings[0]!.bindingId).toBe(first.bindings[0]!.bindingId); + + await expect( + store.reserveDelegations( + identity, + delegate("request-3", [ + { delegationKey: "publisher", outcome: "Publish evidence" }, + { delegationKey: "research", outcome: "Changed task" }, + ]), + target, + ), + ).rejects.toMatchObject({ code: "delegation_key_reused" }); + const aggregate = await store.read(projectId); + expect(aggregate.bindings.map(({ delegationKey }) => delegationKey)).toEqual([ + "research", + ]); + expect(aggregate.requestReceipts).toHaveLength(2); + }); + + it.each(["write", "file-sync", "rename", "directory-sync"] as const)( + "exposes only complete state when %s fails", + async (failedStep) => { + const root = await fixture(); + let fail = false; + const store = new SubsessionCoordinatorStore(root, { + beforePersistStep: (step) => { + if (fail && step === failedStep) throw new Error("injected failure"); + }, + }); + await store.read(projectId); + fail = true; + await expect( + store.reserveDelegations(identity, delegate(), target), + ).rejects.toMatchObject({ code: "storage_unavailable" }); + + const restarted = await new SubsessionCoordinatorStore(root).read( + projectId, + ); + expect(restarted.bindings.length).toBe( + failedStep === "directory-sync" ? 1 : 0, + ); + expect(restarted.requestReceipts.length).toBe(restarted.bindings.length); + }, + ); + + it("allows one spawn claimant and requires inspection before expired takeover", async () => { + const root = await fixture(); + let now = new Date("2026-09-04T12:00:00.000Z"); + const options = { + now: () => now, + claimTtlMs: 1_000, + }; + const first = new SubsessionCoordinatorStore(root, options); + const second = new SubsessionCoordinatorStore(root, options); + const reserved = await first.reserveDelegations( + identity, + delegate(), + target, + ); + const binding = reserved.bindings[0]!; + + const [left, right] = await Promise.all([ + first.claimSpawn(identity, binding.bindingId, { + ownerId: "coordinator-1", + expectedLifecycleEpoch: 1, + expectedSpawnEpoch: 0, + }), + second.claimSpawn(identity, binding.bindingId, { + ownerId: "coordinator-2", + expectedLifecycleEpoch: 1, + expectedSpawnEpoch: 0, + }), + ]); + const winner = [left, right].find((result) => result.claimed)!; + const loser = [left, right].find((result) => !result.claimed)!; + expect(loser).toMatchObject({ claimed: false, reason: "active" }); + + now = new Date("2026-09-04T12:00:02.000Z"); + const observed = await second.claimSpawn(identity, binding.bindingId, { + ownerId: "coordinator-2", + expectedLifecycleEpoch: winner.binding.lifecycleEpoch, + expectedSpawnEpoch: winner.binding.spawnEpoch, + }); + expect(observed).toMatchObject({ + claimed: false, + reason: "expired-requires-inspection", + }); + if (!winner.claimed || !winner.binding.spawnClaim) + throw new Error("missing winning claim"); + const takeover = await second.takeoverExpiredSpawnClaim( + identity, + binding.bindingId, + { + ownerId: "coordinator-2", + expiredClaimId: winner.binding.spawnClaim.claimId, + expectedLifecycleEpoch: winner.binding.lifecycleEpoch, + expectedSpawnEpoch: winner.binding.spawnEpoch, + }, + ); + expect(takeover.claimed).toBe(true); + expect(takeover.binding.spawnEpoch).toBe(2); + }); + + it("fences stale spawn callbacks and only releases a claim with zero-process proof", async () => { + const root = await fixture(); + const store = new SubsessionCoordinatorStore(root); + const binding = ( + await store.reserveDelegations(identity, delegate(), target) + ).bindings[0]!; + const claim = await store.claimSpawn(identity, binding.bindingId, { + ownerId: "coordinator-1", + expectedLifecycleEpoch: 1, + expectedSpawnEpoch: 0, + }); + if (!claim.claimed || !claim.binding.spawnClaim) + throw new Error("claim was not acquired"); + + await expect( + store.attachSpawnedRuntime(identity, binding.bindingId, { + claimId: "claim_stale", + spawnEpoch: claim.binding.spawnEpoch, + runtimeToken: "runtime-stale", + incarnation: 1, + }), + ).rejects.toMatchObject({ code: "claim_conflict" }); + const released = await store.releaseUnspawnedClaim( + identity, + binding.bindingId, + { + claimId: claim.binding.spawnClaim.claimId, + spawnEpoch: claim.binding.spawnEpoch, + proof: "no-process-created", + }, + ); + expect(released).toMatchObject({ + sessionState: "reserved", + spawnClaim: null, + runtime: null, + }); + }); + + it("persists a failed spawn and clears the departed spawn claim", async () => { + const root = await fixture(); + const store = new SubsessionCoordinatorStore(root); + const binding = (await store.reserveDelegations(identity, delegate(), target)).bindings[0]!; + const claimed = await store.claimSpawn(identity, binding.bindingId, { + ownerId: "coordinator-1", expectedLifecycleEpoch: binding.lifecycleEpoch, expectedSpawnEpoch: binding.spawnEpoch, + }); + expect(claimed.claimed).toBe(true); + const failed = await store.transitionSession(identity, binding.bindingId, { + expectedLifecycleEpoch: claimed.binding.lifecycleEpoch, expectedSpawnEpoch: claimed.binding.spawnEpoch, + expectedRuntimeToken: null, state: "failed", + error: { code: "session_create_failed", retryable: true, recovery: "retry" }, + }); + expect(failed).toMatchObject({ sessionState: "failed", spawnClaim: null, runtime: null, + lastError: { code: "session_create_failed", retryable: true, recovery: "retry" } }); + const restarted = await new SubsessionCoordinatorStore(root).read(projectId); + expect(restarted.bindings[0]).toEqual(failed); + }); + + it("persists one kickoff sender and never blindly retries uncertain delivery", async () => { + const root = await fixture(); + const store = new SubsessionCoordinatorStore(root); + const binding = ( + await store.reserveDelegations(identity, delegate(), target) + ).bindings[0]!; + const spawn = await store.claimSpawn(identity, binding.bindingId, { + ownerId: "coordinator-1", + expectedLifecycleEpoch: 1, + expectedSpawnEpoch: 0, + }); + if (!spawn.claimed || !spawn.binding.spawnClaim) + throw new Error("spawn claim was not acquired"); + const starting = await store.attachSpawnedRuntime( + identity, + binding.bindingId, + { + claimId: spawn.binding.spawnClaim.claimId, + spawnEpoch: spawn.binding.spawnEpoch, + runtimeToken: "runtime-1", + incarnation: 1, + }, + ); + const ready = await store.transitionSession(identity, binding.bindingId, { + expectedLifecycleEpoch: starting.lifecycleEpoch, + expectedSpawnEpoch: starting.spawnEpoch, + expectedRuntimeToken: "runtime-1", + state: "ready", + }); + + const [left, right] = await Promise.all([ + store.claimKickoff(identity, binding.bindingId, { + ownerId: "sender-1", + expectedLifecycleEpoch: ready.lifecycleEpoch, + expectedSpawnEpoch: ready.spawnEpoch, + expectedContextEpoch: ready.contextEpoch, + eventWatermark: "event-10", + }), + new SubsessionCoordinatorStore(root).claimKickoff( + identity, + binding.bindingId, + { + ownerId: "sender-2", + expectedLifecycleEpoch: ready.lifecycleEpoch, + expectedSpawnEpoch: ready.spawnEpoch, + expectedContextEpoch: ready.contextEpoch, + eventWatermark: "event-10", + }, + ), + ]); + const winner = [left, right].find((result) => result.claimed)!; + expect([left, right].filter((result) => result.claimed)).toHaveLength(1); + if (!winner.claimed) throw new Error("kickoff claim was not acquired"); + const delivery = winner.binding.deliveries[0]!; + if (!delivery.claim) throw new Error("kickoff claim was not persisted"); + + const uncertain = await store.recordKickoffWrite( + identity, + binding.bindingId, + { + contextEpoch: delivery.contextEpoch, + deliveryId: delivery.deliveryId, + inputId: delivery.inputId, + claimId: delivery.claim.claimId, + phase: "text-staged", + }, + ); + expect(uncertain.deliveries[0]!.state).toBe("uncertain"); + await expect( + store.claimKickoff(identity, binding.bindingId, { + ownerId: "sender-3", + expectedLifecycleEpoch: uncertain.lifecycleEpoch, + expectedSpawnEpoch: uncertain.spawnEpoch, + expectedContextEpoch: uncertain.contextEpoch, + eventWatermark: "event-10", + }), + ).resolves.toMatchObject({ claimed: false, reason: "terminal" }); + + const acknowledged = await store.acknowledgeKickoff( + identity, + binding.bindingId, + { + contextEpoch: delivery.contextEpoch, + deliveryId: delivery.deliveryId, + inputId: delivery.inputId, + eventWatermark: "event-10", + }, + ); + expect(acknowledged.deliveries[0]!.state).toBe("acknowledged"); + await expect( + store.acknowledgeKickoff(identity, binding.bindingId, { + contextEpoch: delivery.contextEpoch, + deliveryId: delivery.deliveryId, + inputId: "input_foreign", + eventWatermark: "event-10", + }), + ).rejects.toBeInstanceOf(SubsessionCoordinatorStoreError); + }); + + it("queues a refresh while the prior delivery awaits acknowledgement", async () => { + const root = await fixture(); + const store = new SubsessionCoordinatorStore(root); + const binding = ( + await store.reserveDelegations(identity, delegate(), target) + ).bindings[0]!; + const spawn = await store.claimSpawn(identity, binding.bindingId, { + ownerId: "coordinator-1", + expectedLifecycleEpoch: binding.lifecycleEpoch, + expectedSpawnEpoch: binding.spawnEpoch, + }); + if (!spawn.claimed || !spawn.binding.spawnClaim) + throw new Error("spawn claim was not acquired"); + const starting = await store.attachSpawnedRuntime( + identity, + binding.bindingId, + { + claimId: spawn.binding.spawnClaim.claimId, + spawnEpoch: spawn.binding.spawnEpoch, + runtimeToken: "runtime-refresh", + incarnation: 1, + }, + ); + const ready = await store.transitionSession(identity, binding.bindingId, { + expectedLifecycleEpoch: starting.lifecycleEpoch, + expectedSpawnEpoch: starting.spawnEpoch, + expectedRuntimeToken: "runtime-refresh", + state: "ready", + }); + const claimed = await store.claimKickoff(identity, binding.bindingId, { + ownerId: "sender-1", + expectedLifecycleEpoch: ready.lifecycleEpoch, + expectedSpawnEpoch: ready.spawnEpoch, + expectedContextEpoch: ready.contextEpoch, + eventWatermark: "event-10", + }); + if (!claimed.claimed || !claimed.binding.deliveries[0]!.claim) + throw new Error("kickoff claim was not acquired"); + const submitted = await store.recordKickoffWrite( + identity, + binding.bindingId, + { + contextEpoch: claimed.binding.contextEpoch, + deliveryId: claimed.binding.deliveries[0]!.deliveryId, + inputId: claimed.binding.deliveries[0]!.inputId, + claimId: claimed.binding.deliveries[0]!.claim!.claimId, + phase: "enter-written", + }, + ); + + const refreshed = await store.refreshFocusedContext(identity, { + schemaVersion: 1, + requestKey: "refresh-while-awaiting-ack", + operation: { + kind: "refresh-focused-context", + target: { kind: "child", delegationKey: "research" }, + expectedContextEpoch: submitted.contextEpoch, + expectedContextDigest: submitted.contextDigest, + focus: null, + }, + }); + + expect(refreshed.binding.deliveries).toHaveLength(2); + expect(refreshed.binding.deliveries.map(({ state }) => state)).toEqual([ + "submitted-unacknowledged", + "pending", + ]); + }); + + it("scopes mutations to the trusted parent and never adopts a foreign binding", async () => { + const root = await fixture(); + const store = new SubsessionCoordinatorStore(root); + const binding = ( + await store.reserveDelegations(identity, delegate(), target) + ).bindings[0]!; + const foreign: ProjectAgentSession = { + ...identity, + sessionId: "manual-session-with-no-binding", + }; + await expect( + store.claimSpawn(foreign, binding.bindingId as SubsessionBindingId, { + ownerId: "coordinator-foreign", + expectedLifecycleEpoch: 1, + expectedSpawnEpoch: 0, + }), + ).rejects.toMatchObject({ code: "binding_scope_mismatch" }); + expect((await store.read(projectId)).bindings[0]).toEqual(binding); + }); + + it("bounds nested delegation depth and concurrently live coordinator sessions", async () => { + const root = await fixture(); + const store = new SubsessionCoordinatorStore(root, { + maxDelegationDepth: 2, + liveSessionLimit: 2, + }); + const first = ( + await store.reserveDelegations(identity, delegate(), target) + ).bindings[0]!; + const child: ProjectAgentSession = { + ...identity, + sessionId: first.sessionId, + }; + const second = ( + await store.reserveDelegations( + child, + delegate("nested-1", [ + { delegationKey: "nested", outcome: "Nested task" }, + ]), + target, + ) + ).bindings[0]!; + + expect(first).toMatchObject({ parentBindingId: null, delegationDepth: 1 }); + expect(second).toMatchObject({ + parentBindingId: first.bindingId, + delegationDepth: 2, + }); + await expect( + store.reserveDelegations( + { ...identity, sessionId: second.sessionId }, + delegate("too-deep", [ + { delegationKey: "third", outcome: "Too deep" }, + ]), + target, + ), + ).rejects.toMatchObject({ code: "delegation_depth_exceeded" }); + await expect( + store.reserveDelegations( + identity, + delegate("over-live-limit", [ + { delegationKey: "publisher", outcome: "Publish evidence" }, + ]), + target, + ), + ).rejects.toMatchObject({ code: "live_session_limit_reached" }); + expect((await store.read(projectId)).bindings).toHaveLength(2); + }); + + it("expires receipts into tombstones and reclaims closed bindings without reopening their keys", async () => { + const root = await fixture(); + const store = new SubsessionCoordinatorStore(root, { + receiptRetentionLimit: 1, + }); + const first = ( + await store.reserveDelegations(identity, delegate(), target) + ).bindings[0]!; + await store.closeBinding(identity, first.bindingId, first.sessionId); + await store.reserveDelegations( + identity, + delegate("request-2", [ + { delegationKey: "publisher", outcome: "Publish evidence" }, + ]), + target, + ); + + const aggregate = await store.read(projectId); + expect(aggregate.requestReceipts.map(({ requestKey }) => requestKey)).toEqual([ + "request-2", + ]); + expect(aggregate.requestTombstones).toContainEqual( + expect.objectContaining({ requestKey: "request-1" }), + ); + expect(aggregate.bindings.map(({ delegationKey }) => delegationKey)).toEqual([ + "publisher", + ]); + expect(aggregate.bindingTombstones).toContainEqual( + expect.objectContaining({ + bindingId: first.bindingId, + delegationKey: "research", + sessionId: first.sessionId, + }), + ); + await expect( + store.closeOwnedBinding({ + projectId, + parentSessionId: identity.sessionId, + bindingId: first.bindingId, + sessionId: first.sessionId, + }), + ).resolves.toBeUndefined(); + await expect( + store.closeOwnedBinding({ + projectId, + parentSessionId: identity.sessionId, + bindingId: first.bindingId, + sessionId: "foreign-session", + }), + ).rejects.toMatchObject({ code: "binding_not_found" }); + await expect( + store.reserveDelegations(identity, delegate(), target), + ).rejects.toMatchObject({ code: "request_key_expired" }); + await expect( + store.reserveDelegations( + identity, + delegate("request-3", [ + { delegationKey: "research", outcome: "Collect evidence" }, + ]), + target, + ), + ).rejects.toMatchObject({ code: "session_closed" }); + }); + + it("expires oldest key and ownership tombstones instead of dead-ending the project", async () => { + const root = await fixture(); + const store = new SubsessionCoordinatorStore(root, { + receiptRetentionLimit: 1, + historyTombstoneLimit: 1, + }); + const first = ( + await store.reserveDelegations(identity, delegate("request-1"), target) + ).bindings[0]!; + await store.closeBinding(identity, first.bindingId, first.sessionId); + const second = ( + await store.reserveDelegations( + identity, + delegate("request-2", [ + { delegationKey: "publisher", outcome: "Publish evidence" }, + ]), + target, + ) + ).bindings[0]!; + await store.closeBinding(identity, second.bindingId, second.sessionId); + await expect( + store.reserveDelegations(identity, delegate("request-1"), target), + ).rejects.toMatchObject({ code: "request_key_expired" }); + await store.reserveDelegations( + identity, + delegate("request-3", [ + { delegationKey: "writer", outcome: "Write evidence" }, + ]), + target, + ); + + const aggregate = await store.read(projectId); + expect(aggregate.requestTombstones).toHaveLength(1); + expect(aggregate.requestTombstones[0]!.requestKey).toBe("request-2"); + expect(aggregate.bindingTombstones).toHaveLength(1); + expect(aggregate.bindingTombstones[0]!.bindingId).toBe(second.bindingId); + }); + + it.each(["exited", "failed"] as const)( + "keeps a %s binding resumable and charges capacity only when it is re-referenced", + async (sessionState) => { + const root = await fixture(); + const store = new SubsessionCoordinatorStore(root, { + receiptRetentionLimit: 1, + liveSessionLimit: 2, + }); + const first = ( + await store.reserveDelegations(identity, delegate("request-1"), target) + ).bindings[0]!; + const claim = await store.claimSpawn(identity, first.bindingId, { + ownerId: "coordinator-1", + expectedLifecycleEpoch: first.lifecycleEpoch, + expectedSpawnEpoch: first.spawnEpoch, + }); + if (!claim.claimed || !claim.binding.spawnClaim) + throw new Error("spawn claim was not acquired"); + const starting = await store.attachSpawnedRuntime( + identity, + first.bindingId, + { + claimId: claim.binding.spawnClaim.claimId, + spawnEpoch: claim.binding.spawnEpoch, + runtimeToken: "runtime-exited", + incarnation: 1, + }, + ); + await store.transitionSession(identity, first.bindingId, { + expectedLifecycleEpoch: starting.lifecycleEpoch, + expectedSpawnEpoch: starting.spawnEpoch, + expectedRuntimeToken: "runtime-exited", + state: sessionState, + }); + + await store.reserveDelegations( + identity, + delegate("request-2", [ + { delegationKey: "publisher", outcome: "Publish evidence" }, + ]), + target, + ); + const replay = await store.reserveDelegations( + identity, + delegate("request-3"), + target, + ); + + const aggregate = await store.read(projectId); + expect(replay.bindings[0]).toMatchObject({ + bindingId: first.bindingId, + sessionId: first.sessionId, + sessionState: "spawn-claimed", + }); + expect(aggregate.bindings).toContainEqual(replay.bindings[0]); + expect(aggregate.bindingTombstones).not.toContainEqual( + expect.objectContaining({ bindingId: first.bindingId }), + ); + await expect( + store.reserveDelegations( + identity, + delegate("request-4", [ + { delegationKey: "writer", outcome: "Write evidence" }, + ]), + target, + ), + ).rejects.toMatchObject({ code: "live_session_limit_reached" }); + expect((await store.read(projectId)).bindings).toHaveLength(2); + await store.reserveReleases( + identity, + release(`release-${sessionState}`), + ); + const closed = await store.closeBinding( + identity, + first.bindingId, + first.sessionId, + ); + expect(closed.sessionState).toBe("closed"); + }, + ); + + it("lets a new parent delegate after an old parent's descendants become dormant", async () => { + const root = await fixture(); + const store = new SubsessionCoordinatorStore(root, { liveSessionLimit: 2 }); + const dormant = ( + await store.reserveDelegations(identity, delegate("old-request"), target) + ).bindings[0]!; + const claim = await store.claimSpawn(identity, dormant.bindingId, { + ownerId: "coordinator-1", + expectedLifecycleEpoch: dormant.lifecycleEpoch, + expectedSpawnEpoch: dormant.spawnEpoch, + }); + if (!claim.claimed || !claim.binding.spawnClaim) + throw new Error("spawn claim was not acquired"); + const starting = await store.attachSpawnedRuntime( + identity, + dormant.bindingId, + { + claimId: claim.binding.spawnClaim.claimId, + spawnEpoch: claim.binding.spawnEpoch, + runtimeToken: "runtime-dormant", + incarnation: 1, + }, + ); + await store.transitionSession(identity, dormant.bindingId, { + expectedLifecycleEpoch: starting.lifecycleEpoch, + expectedSpawnEpoch: starting.spawnEpoch, + expectedRuntimeToken: "runtime-dormant", + state: "exited", + }); + + const newParent = { ...identity, sessionId: "parent-session-2" }; + const active = await store.reserveDelegations( + newParent, + delegate("new-request", [ + { delegationKey: "publisher", outcome: "Publish evidence" }, + { delegationKey: "writer", outcome: "Write evidence" }, + ]), + target, + ); + + expect(active.bindings).toHaveLength(2); + expect((await store.read(projectId)).bindings).toContainEqual( + expect.objectContaining({ + bindingId: dormant.bindingId, + sessionState: "exited", + }), + ); + await expect( + store.reserveDelegations( + newParent, + delegate("new-request-2", [ + { delegationKey: "editor", outcome: "Edit evidence" }, + ]), + target, + ), + ).rejects.toMatchObject({ code: "live_session_limit_reached" }); + }); + + it("fences a dormant resume racing a new parent's active reservation", async () => { + const root = await fixture(); + const firstStore = new SubsessionCoordinatorStore(root, { + liveSessionLimit: 1, + }); + const secondStore = new SubsessionCoordinatorStore(root, { + liveSessionLimit: 1, + }); + const dormant = ( + await firstStore.reserveDelegations( + identity, + delegate("old-request"), + target, + ) + ).bindings[0]!; + const claim = await firstStore.claimSpawn(identity, dormant.bindingId, { + ownerId: "coordinator-1", + expectedLifecycleEpoch: dormant.lifecycleEpoch, + expectedSpawnEpoch: dormant.spawnEpoch, + }); + if (!claim.claimed || !claim.binding.spawnClaim) + throw new Error("spawn claim was not acquired"); + const starting = await firstStore.attachSpawnedRuntime( + identity, + dormant.bindingId, + { + claimId: claim.binding.spawnClaim.claimId, + spawnEpoch: claim.binding.spawnEpoch, + runtimeToken: "runtime-dormant-race", + incarnation: 1, + }, + ); + await firstStore.transitionSession(identity, dormant.bindingId, { + expectedLifecycleEpoch: starting.lifecycleEpoch, + expectedSpawnEpoch: starting.spawnEpoch, + expectedRuntimeToken: "runtime-dormant-race", + state: "exited", + }); + + const newParent = { ...identity, sessionId: "parent-session-2" }; + const results = await Promise.allSettled([ + firstStore.reserveDelegations( + identity, + delegate("resume-request"), + target, + ), + secondStore.reserveDelegations( + newParent, + delegate("new-request", [ + { delegationKey: "publisher", outcome: "Publish evidence" }, + ]), + { ...target, ownerId: "coordinator-2" }, + ), + ]); + + expect(results.filter(({ status }) => status === "fulfilled")).toHaveLength(1); + expect(results.filter(({ status }) => status === "rejected")).toHaveLength(1); + expect(results.find(({ status }) => status === "rejected")).toMatchObject({ + reason: { code: "live_session_limit_reached" }, + }); + const aggregate = await firstStore.read(projectId); + expect( + aggregate.bindings.filter(({ sessionState }) => + [ + "reserved", + "spawn-claimed", + "starting", + "awaiting-ready", + "ready", + ].includes(sessionState), + ), + ).toHaveLength(1); + expect(aggregate.bindings).toContainEqual( + expect.objectContaining({ bindingId: dormant.bindingId }), + ); + }); + + it("reclaims a bounded dormant binding at history capacity and preserves replay", async () => { + const root = await fixture(); + const store = new SubsessionCoordinatorStore(root, { + bindingLimit: 2, + liveSessionLimit: 2, + }); + const dormant = ( + await store.reserveDelegations( + identity, + delegate("old-request", [ + { delegationKey: "research", outcome: "Collect evidence" }, + { delegationKey: "publisher", outcome: "Publish evidence" }, + ]), + target, + ) + ).bindings; + for (const [index, binding] of dormant.entries()) { + const claim = await store.claimSpawn(identity, binding.bindingId, { + ownerId: "coordinator-1", + expectedLifecycleEpoch: binding.lifecycleEpoch, + expectedSpawnEpoch: binding.spawnEpoch, + }); + if (!claim.claimed || !claim.binding.spawnClaim) + throw new Error("spawn claim was not acquired"); + const starting = await store.attachSpawnedRuntime( + identity, + binding.bindingId, + { + claimId: claim.binding.spawnClaim.claimId, + spawnEpoch: claim.binding.spawnEpoch, + runtimeToken: `runtime-history-${index}`, + incarnation: 1, + }, + ); + await store.transitionSession(identity, binding.bindingId, { + expectedLifecycleEpoch: starting.lifecycleEpoch, + expectedSpawnEpoch: starting.spawnEpoch, + expectedRuntimeToken: `runtime-history-${index}`, + state: index === 0 ? "exited" : "failed", + }); + } + + const newParent = { ...identity, sessionId: "parent-session-2" }; + await expect( + store.reserveDelegations( + newParent, + delegate("new-request", [ + { delegationKey: "writer", outcome: "Write evidence" }, + ]), + target, + ), + ).rejects.toMatchObject({ code: "history_quota_exceeded" }); + + const request = releaseDormant("sweep-at-cap", 1); + const reserved = await store.reserveDormantReleases( + newParent, + request, + [dormant[0]!.bindingId], + ); + expect(reserved.bindings).toEqual([ + { + state: "evicted", + binding: expect.objectContaining({ + bindingId: dormant[0]!.bindingId, + disposition: "dormant-evicted", + }), + }, + ]); + await expect( + store.reserveDelegations( + identity, + delegate("old-request", [ + { delegationKey: "research", outcome: "Collect evidence" }, + { delegationKey: "publisher", outcome: "Publish evidence" }, + ]), + target, + ), + ).rejects.toMatchObject({ code: "request_key_expired" }); + const replay = await store.reserveDormantReleases( + newParent, + request, + [], + ); + expect(replay).toMatchObject({ + replayed: true, + bindings: [ + { + state: "released", + binding: { bindingId: dormant[0]!.bindingId }, + }, + ], + }); + + const created = await store.reserveDelegations( + newParent, + delegate("new-request", [ + { delegationKey: "writer", outcome: "Write evidence" }, + ]), + target, + ); + expect(created.bindings[0]).toMatchObject({ delegationKey: "writer" }); + expect((await store.read(projectId)).bindings).toContainEqual( + expect.objectContaining({ + bindingId: dormant[1]!.bindingId, + sessionState: "failed", + }), + ); + }); + + it("atomically excludes an active child from project-wide dormant release", async () => { + const root = await fixture(); + const store = new SubsessionCoordinatorStore(root); + const binding = ( + await store.reserveDelegations( + identity, + delegate("parent-race", [ + { delegationKey: "research", outcome: "Collect evidence" }, + ]), + target, + ) + ).bindings[0]!; + const reserved = await store.reserveDormantReleases( + { ...identity, sessionId: "new-parent" }, + releaseDormant("active-child", 1), + [binding.bindingId], + ); + + expect(reserved.bindings).toEqual([]); + expect((await store.read(projectId)).bindings).toContainEqual( + expect.objectContaining({ + bindingId: binding.bindingId, + sessionState: "reserved", + }), + ); + }); + + it("retains unfinished dormant cleanup beyond receipt and tombstone retention", async () => { + const root = await fixture(); + const options = { receiptRetentionLimit: 1, historyTombstoneLimit: 1 }; + const store = new SubsessionCoordinatorStore(root, options); + const reserved = await store.reserveDelegations(identity, delegate("cleanup-history", [ + { delegationKey: "research", outcome: "Collect evidence" }, + { delegationKey: "publisher", outcome: "Publish evidence" }, + ]), target); + for (const binding of reserved.bindings) { + await store.transitionSession(identity, binding.bindingId, { + expectedLifecycleEpoch: binding.lifecycleEpoch, + expectedSpawnEpoch: binding.spawnEpoch, + expectedRuntimeToken: null, + state: "failed", + }); + } + const [first, second] = reserved.bindings; + await store.reserveDormantReleases(identity, releaseDormant("cleanup-first", 1), [first!.bindingId]); + await store.reserveDormantReleases(identity, releaseDormant("cleanup-second", 1), [second!.bindingId]); + + const restarted = new SubsessionCoordinatorStore(root, options); + expect((await restarted.read(projectId)).bindingTombstones).toHaveLength(2); + await expect(restarted.reserveDormantReleases(identity, releaseDormant("cleanup-first", 1), [])) + .rejects.toMatchObject({ code: "request_key_expired" }); + const retry = await restarted.reserveDormantReleases(identity, + releaseDormant("cleanup-retry", 1), [first!.bindingId]); + expect(retry.bindings).toMatchObject([{ + state: "released", binding: { bindingId: first!.bindingId }, + }]); + await restarted.reserveDormantReleases(identity, releaseDormant("advance-cleanup", 1), []); + await restarted.completeDormantReleaseCleanup(identity, first!.bindingId, first!.sessionId); + const remaining = (await restarted.read(projectId)).bindingTombstones; + expect(remaining).toHaveLength(1); + expect(remaining[0]?.bindingId).toBe(second!.bindingId); + }); + + it("reclaims released capacity so a sixty-fifth delegation can be reserved", async () => { + const root = await fixture(); + const store = new SubsessionCoordinatorStore(root, { + receiptRetentionLimit: 1, + }); + const bindings = []; + for (let batch = 0; batch < 4; batch += 1) { + const reserved = await store.reserveDelegations( + identity, + delegate( + `capacity-${batch}`, + Array.from({ length: 16 }, (_, index) => ({ + delegationKey: `child-${batch * 16 + index + 1}`, + outcome: `Task ${batch * 16 + index + 1}`, + })), + ), + target, + ); + bindings.push(...reserved.bindings); + } + await expect( + store.reserveDelegations( + identity, + delegate("capacity-65", [ + { delegationKey: "child-65", outcome: "Task 65" }, + ]), + target, + ), + ).rejects.toMatchObject({ code: "live_session_limit_reached" }); + + const released = await store.reserveReleases( + identity, + release("release-capacity", ["child-1"]), + ); + expect(released.bindings[0]).toMatchObject({ + state: "bound", + binding: { bindingId: bindings[0]!.bindingId }, + }); + await store.closeBinding( + identity, + bindings[0]!.bindingId, + bindings[0]!.sessionId, + ); + const sixtyFifth = await store.reserveDelegations( + identity, + delegate("capacity-65", [ + { delegationKey: "child-65", outcome: "Task 65" }, + ]), + target, + ); + + expect(sixtyFifth.bindings[0]!.delegationKey).toBe("child-65"); + const aggregate = await store.read(projectId); + expect(aggregate.bindings).toHaveLength(64); + expect(aggregate.bindingTombstones).toContainEqual( + expect.objectContaining({ bindingId: bindings[0]!.bindingId }), + ); + }); + + it("prunes proven terminal deliveries so long-lived focused refresh stays writable", async () => { + const root = await fixture(); + const store = new SubsessionCoordinatorStore(root, { + receiptRetentionLimit: 1, + historyTombstoneLimit: 2, + }); + let binding = ( + await store.reserveDelegations(identity, delegate(), target) + ).bindings[0]!; + + for (let index = 1; index <= 70; index += 1) { + binding = ( + await store.refreshFocusedContext(identity, { + schemaVersion: 1, + requestKey: `refresh-${index}`, + operation: { + kind: "refresh-focused-context", + target: { kind: "child", delegationKey: "research" }, + expectedContextEpoch: binding.contextEpoch, + expectedContextDigest: binding.contextDigest, + focus: null, + }, + }) + ).binding; + } + + expect(binding.contextEpoch).toBe(71); + expect(binding.deliveries).toHaveLength(1); + expect(binding.deliveries[0]!.contextEpoch).toBe(71); + }); +}); diff --git a/packages/harness/src/core/subsession-coordinator-store.ts b/packages/harness/src/core/subsession-coordinator-store.ts new file mode 100644 index 00000000..688a9f00 --- /dev/null +++ b/packages/harness/src/core/subsession-coordinator-store.ts @@ -0,0 +1,2171 @@ +import { randomUUID } from "node:crypto"; +import * as fs from "node:fs/promises"; +import * as path from "node:path"; + +import type { ProjectAgentSession, StudioProjectId } from "../shared/agent-map.js"; +import { canonicalDigest } from "../shared/agent-map-canonical.js"; +import { + hasAgentMapControlCharacter, + parseProjectAgentActorRef, +} from "../shared/agent-map-codec.js"; +import type { HarnessKind } from "../shared/types.js"; +import { + computeCanonicalDelegationBindingDigest, + computeCanonicalDelegationRequestDigest, + computeSubsessionContextDigest, + parseProjectSubsessionRequest, +} from "../shared/subsession-delegation-codec.js"; +import { + PROJECT_SUBSESSION_CLAIM_TTL_MS, + PROJECT_SUBSESSION_LIVE_SESSION_LIMIT, + PROJECT_SUBSESSION_MAX_DEPTH, + PROJECT_SUBSESSION_SCHEMA_VERSION, + SUBSESSION_COORDINATOR_STORAGE_SCHEMA_VERSION, + type CanonicalDelegationRequestDigest, + type DelegatedSessionState, + type DelegationError, + type ProjectSubsessionRequest, + type SubsessionBindingId, + type SubsessionBindingRecord, + type SubsessionClaim, + type SubsessionKickoffDelivery, + type SubsessionProjectionDigest, +} from "../shared/subsession-delegation.js"; +import { DurableFileLock } from "./durable-file-lock.js"; +import { isStudioProjectId } from "./studio-project-catalog.js"; + +export const SUBSESSION_COORDINATOR_BINDING_LIMIT = 8_192; +export const SUBSESSION_COORDINATOR_RECEIPT_LIMIT = 8_192; +export const SUBSESSION_COORDINATOR_RECEIPT_RETENTION_LIMIT = 256; +export const SUBSESSION_COORDINATOR_DELIVERY_LIMIT = 64; + +export type SubsessionCoordinatorStoreErrorCode = + | "malformed_state" + | "unsupported_schema" + | "storage_unavailable" + | "capacity_exceeded" + | "history_quota_exceeded" + | "live_session_limit_reached" + | "delegation_depth_exceeded" + | "request_key_reused" + | "request_key_expired" + | "delegation_key_reused" + | "binding_not_found" + | "binding_scope_mismatch" + | "lifecycle_conflict" + | "claim_conflict" + | "session_closed"; + +export class SubsessionCoordinatorStoreError extends Error { + constructor( + readonly code: SubsessionCoordinatorStoreErrorCode, + readonly schemaVersion?: number, + ) { + super( + code === "storage_unavailable" + ? "Subsession coordinator storage is unavailable" + : code === "unsupported_schema" + ? "Subsession coordinator state uses an unsupported schema" + : "Subsession coordinator operation was rejected", + ); + this.name = "SubsessionCoordinatorStoreError"; + } +} + +export type SubsessionCoordinatorRequestReceipt = Readonly<{ + parentSessionId: string; + requestKey: string; + requestDigest: CanonicalDelegationRequestDigest; + operation: ProjectSubsessionRequest["operation"]["kind"]; + bindingIds: readonly SubsessionBindingId[]; + createdAt: string; +}>; + +export type SubsessionCoordinatorRequestTombstone = + SubsessionCoordinatorRequestReceipt; + +export type SubsessionCoordinatorBindingTombstone = Readonly<{ + bindingId: SubsessionBindingId; + parentSessionId: string; + parentBindingId: SubsessionBindingId | null; + delegationDepth: number; + delegationKey: string; + bindingDigest: string; + sessionId: string; + disposition: "terminal" | "dormant-evicted"; + /** Absent on older records: retain dormant eviction proof until verified. */ + cleanupComplete?: true; + closedAt: string; +}>; + +export type SubsessionCoordinatorAggregate = Readonly<{ + schemaVersion: typeof SUBSESSION_COORDINATOR_STORAGE_SCHEMA_VERSION; + recordVersion: number; + projectId: StudioProjectId; + requestReceipts: readonly SubsessionCoordinatorRequestReceipt[]; + requestTombstones: readonly SubsessionCoordinatorRequestTombstone[]; + bindingTombstones: readonly SubsessionCoordinatorBindingTombstone[]; + bindings: readonly SubsessionBindingRecord[]; + createdAt: string; + updatedAt: string; + aggregateDigest: string; +}>; + +export interface SubsessionCoordinatorStoreEvent { + name: + | "subsession.store_initialized" + | "subsession.binding_reserved" + | "subsession.duplicate_prevented" + | "subsession.spawn_claimed" + | "subsession.kickoff_claimed" + | "subsession.kickoff_uncertain"; + projectId: StudioProjectId; + count?: number; +} + +export interface ReservedDelegations { + replayed: boolean; + requestDigest: CanonicalDelegationRequestDigest; + bindings: readonly SubsessionBindingRecord[]; +} + +export type ReleasableSubsessionBinding = + | Readonly<{ + state: "bound"; + binding: SubsessionBindingRecord; + }> + | Readonly<{ + /** This request atomically committed the dormant eviction. */ + state: "evicted"; + binding: SubsessionCoordinatorBindingTombstone; + }> + | Readonly<{ + state: "released"; + binding: SubsessionCoordinatorBindingTombstone; + }> + | Readonly<{ + state: "absent"; + delegationKey: string; + }>; + +export interface ReservedReleases { + replayed: boolean; + requestDigest: CanonicalDelegationRequestDigest; + bindings: readonly ReleasableSubsessionBinding[]; +} + +export type SpawnClaimResult = + | Readonly<{ claimed: true; binding: SubsessionBindingRecord }> + | Readonly<{ + claimed: false; + reason: "active" | "expired-requires-inspection"; + binding: SubsessionBindingRecord; + }>; + +export type KickoffClaimResult = + | Readonly<{ claimed: true; binding: SubsessionBindingRecord }> + | Readonly<{ + claimed: false; + reason: "already-claimed" | "expired-requires-reconciliation" | "terminal"; + binding: SubsessionBindingRecord; + }>; + +export type FocusedContextRefreshResult = Readonly<{ + replayed: boolean; + requestDigest: CanonicalDelegationRequestDigest; + binding: SubsessionBindingRecord; +}>; + +type ShallowMutable = { -readonly [K in keyof T]: T[K] }; +type MutableClaim = ShallowMutable; +type MutableDelivery = Omit< + ShallowMutable, + "claim" +> & { claim: MutableClaim | null }; +type MutableRuntime = ShallowMutable< + NonNullable +>; +type MutableBinding = Omit< + ShallowMutable, + "spawnClaim" | "runtime" | "deliveries" | "lastError" +> & { + spawnClaim: MutableClaim | null; + runtime: MutableRuntime | null; + deliveries: MutableDelivery[]; + lastError: DelegationError | null; +}; +type MutableAggregate = Omit< + ShallowMutable, + "requestReceipts" | "requestTombstones" | "bindingTombstones" | "bindings" +> & { + requestReceipts: SubsessionCoordinatorRequestReceipt[]; + requestTombstones: SubsessionCoordinatorRequestTombstone[]; + bindingTombstones: SubsessionCoordinatorBindingTombstone[]; + bindings: MutableBinding[]; +}; + +const storageError = () => + new SubsessionCoordinatorStoreError("storage_unavailable"); + +const isRecord = (value: unknown): value is Record => + typeof value === "object" && value !== null && !Array.isArray(value); + +const exact = (value: Record, keys: readonly string[]) => { + const actual = Object.keys(value).sort(); + const expected = [...keys].sort(); + return ( + actual.length === expected.length && + actual.every((key, index) => key === expected[index]) + ); +}; + +const timestamp = (value: unknown): value is string => { + if (typeof value !== "string") return false; + try { + return new Date(value).toISOString() === value; + } catch { + return false; + } +}; + +const digest = (value: unknown): value is string => + typeof value === "string" && /^sha256:[0-9a-f]{64}$/u.test(value); + +const identifier = (value: unknown, prefix?: string): value is string => + typeof value === "string" && + value.length > 0 && + value.length <= 256 && + !hasAgentMapControlCharacter(value) && + (prefix === undefined || value.startsWith(`${prefix}_`)); + +const parseClaim = (value: unknown): SubsessionClaim | null => { + if (value === null) return null; + if ( + !isRecord(value) || + !exact(value, ["claimId", "ownerId", "claimedAt", "expiresAt"]) || + !identifier(value.claimId) || + !identifier(value.ownerId) || + !timestamp(value.claimedAt) || + !timestamp(value.expiresAt) || + value.expiresAt <= value.claimedAt + ) { + throw new SubsessionCoordinatorStoreError("malformed_state"); + } + return structuredClone(value) as unknown as SubsessionClaim; +}; + +const parseDelivery = (value: unknown): SubsessionKickoffDelivery => { + if ( + !isRecord(value) || + !exact(value, [ + "contextEpoch", + "deliveryId", + "inputId", + "eventWatermark", + "state", + "attempt", + "claim", + "submittedAt", + "acknowledgedAt", + ]) || + !Number.isSafeInteger(value.contextEpoch) || + (value.contextEpoch as number) < 1 || + !identifier(value.deliveryId) || + !identifier(value.inputId) || + (value.eventWatermark !== null && !identifier(value.eventWatermark)) || + ![ + "pending", + "claimed", + "submitted-unacknowledged", + "acknowledged", + "uncertain", + ].includes(String(value.state)) || + !Number.isSafeInteger(value.attempt) || + (value.attempt as number) < 0 || + (value.submittedAt !== null && !timestamp(value.submittedAt)) || + (value.acknowledgedAt !== null && !timestamp(value.acknowledgedAt)) + ) { + throw new SubsessionCoordinatorStoreError("malformed_state"); + } + const claim = parseClaim(value.claim); + if ( + (value.state === "claimed") !== (claim !== null) || + (value.state === "acknowledged") !== (value.acknowledgedAt !== null) + ) { + throw new SubsessionCoordinatorStoreError("malformed_state"); + } + return { ...structuredClone(value), claim } as unknown as SubsessionKickoffDelivery; +}; + +const parseBoundedError = (value: unknown): DelegationError | null => { + if (value === null) return null; + if (!isRecord(value)) + throw new SubsessionCoordinatorStoreError("malformed_state"); + const allowed = ["code", "retryable", "recovery", "issues"]; + if ( + !Object.keys(value).every((key) => allowed.includes(key)) || + !exact( + Object.fromEntries( + Object.entries(value).filter(([, entry]) => entry !== undefined), + ), + value.issues === undefined + ? ["code", "retryable", "recovery"] + : allowed, + ) || + !identifier(value.code) || + typeof value.retryable !== "boolean" || + !identifier(value.recovery) || + (value.issues !== undefined && + (!Array.isArray(value.issues) || + value.issues.length > 32 || + !value.issues.every( + (issue) => + isRecord(issue) && + exact(issue, ["path", "code"]) && + identifier(issue.path) && + identifier(issue.code), + ))) + ) { + throw new SubsessionCoordinatorStoreError("malformed_state"); + } + return structuredClone(value) as unknown as DelegationError; +}; + +function parseBinding( + value: unknown, + projectId: StudioProjectId, +): SubsessionBindingRecord { + if ( + !isRecord(value) || + !exact(value, [ + "bindingId", + "projectId", + "parentSessionId", + "parentBindingId", + "delegationDepth", + "delegationKey", + "bindingDigest", + "outcome", + "kickoffContext", + "initialFocus", + "sessionId", + "harness", + "projectRoot", + "lifecycleEpoch", + "spawnEpoch", + "contextEpoch", + "contextDigest", + "contextState", + "currentFocus", + "projectionDigest", + "sessionState", + "spawnClaim", + "runtime", + "deliveries", + "lastError", + "createdAt", + "updatedAt", + ]) || + value.projectId !== projectId || + !identifier(value.bindingId, "binding") || + !identifier(value.parentSessionId) || + (value.parentBindingId !== null && + !identifier(value.parentBindingId, "binding")) || + !Number.isSafeInteger(value.delegationDepth) || + (value.delegationDepth as number) < 1 || + (value.delegationDepth as number) > PROJECT_SUBSESSION_MAX_DEPTH || + !identifier(value.sessionId) || + !["claude-code", "codex"].includes(String(value.harness)) || + typeof value.projectRoot !== "string" || + !path.isAbsolute(value.projectRoot) || + !Number.isSafeInteger(value.lifecycleEpoch) || + (value.lifecycleEpoch as number) < 1 || + !Number.isSafeInteger(value.spawnEpoch) || + (value.spawnEpoch as number) < 0 || + !Number.isSafeInteger(value.contextEpoch) || + (value.contextEpoch as number) < 1 || + !digest(value.bindingDigest) || + !digest(value.contextDigest) || + (value.projectionDigest !== null && !digest(value.projectionDigest)) || + !["none", "current", "stale", "refreshing"].includes( + String(value.contextState), + ) || + ![ + "reserved", + "spawn-claimed", + "starting", + "awaiting-ready", + "ready", + "exited", + "failed", + "closed", + ].includes(String(value.sessionState)) || + !timestamp(value.createdAt) || + !timestamp(value.updatedAt) || + !Array.isArray(value.deliveries) || + value.deliveries.length < 1 || + value.deliveries.length > SUBSESSION_COORDINATOR_DELIVERY_LIMIT + ) { + throw new SubsessionCoordinatorStoreError("malformed_state"); + } + + const parsedRequest = parseProjectSubsessionRequest( + { + schemaVersion: PROJECT_SUBSESSION_SCHEMA_VERSION, + requestKey: "persistence-check", + operation: { + kind: "delegate", + delegations: [ + { + delegationKey: value.delegationKey, + outcome: value.outcome, + ...(value.kickoffContext === null + ? {} + : { kickoffContext: value.kickoffContext }), + ...(value.initialFocus === null + ? {} + : { focus: value.initialFocus }), + }, + ], + }, + }, + projectId, + ); + if (parsedRequest.operation.kind !== "delegate") + throw new SubsessionCoordinatorStoreError("malformed_state"); + const delegation = parsedRequest.operation.delegations[0]!; + const currentFocus = + value.currentFocus === null + ? null + : (() => { + const parsed = parseProjectSubsessionRequest( + { + schemaVersion: PROJECT_SUBSESSION_SCHEMA_VERSION, + requestKey: "context-check", + operation: { + kind: "delegate", + delegations: [ + { + delegationKey: "context-check", + outcome: "Context integrity check", + focus: value.currentFocus, + }, + ], + }, + }, + projectId, + ); + if (parsed.operation.kind !== "delegate") + throw new SubsessionCoordinatorStoreError("malformed_state"); + return parsed.operation.delegations[0]!.focus!; + })(); + if ( + computeCanonicalDelegationBindingDigest(delegation) !== + value.bindingDigest || + computeSubsessionContextDigest(currentFocus) !== value.contextDigest + ) { + throw new SubsessionCoordinatorStoreError("malformed_state"); + } + const spawnClaim = parseClaim(value.spawnClaim); + if ((value.sessionState === "spawn-claimed") !== (spawnClaim !== null)) + throw new SubsessionCoordinatorStoreError("malformed_state"); + let runtime: SubsessionBindingRecord["runtime"] = null; + if (value.runtime !== null) { + if ( + !isRecord(value.runtime) || + !exact(value.runtime, ["runtimeToken", "incarnation", "spawnEpoch"]) || + !identifier(value.runtime.runtimeToken) || + !Number.isSafeInteger(value.runtime.incarnation) || + (value.runtime.incarnation as number) < 1 || + value.runtime.spawnEpoch !== value.spawnEpoch + ) { + throw new SubsessionCoordinatorStoreError("malformed_state"); + } + runtime = structuredClone(value.runtime) as SubsessionBindingRecord["runtime"]; + } + const deliveries = value.deliveries.map(parseDelivery); + if ( + new Set(deliveries.map(({ contextEpoch }) => contextEpoch)).size !== + deliveries.length || + deliveries.at(-1)?.contextEpoch !== value.contextEpoch + ) { + throw new SubsessionCoordinatorStoreError("malformed_state"); + } + return { + ...structuredClone(value), + initialFocus: delegation.focus ?? null, + currentFocus, + spawnClaim, + runtime, + deliveries, + lastError: parseBoundedError(value.lastError), + } as unknown as SubsessionBindingRecord; +} + +const aggregateDigest = ( + value: Omit | SubsessionCoordinatorAggregate, +) => + canonicalDigest( + "sapiom.project-subsession.aggregate.v1", + Object.fromEntries( + Object.entries(value).filter(([key]) => key !== "aggregateDigest"), + ), + ); + +function parseReceipt( + value: unknown, +): SubsessionCoordinatorRequestReceipt { + if ( + !isRecord(value) || + !exact(value, [ + "parentSessionId", + "requestKey", + "requestDigest", + "operation", + "bindingIds", + "createdAt", + ]) || + !identifier(value.parentSessionId) || + !identifier(value.requestKey) || + !digest(value.requestDigest) || + ![ + "delegate", + "refresh-focused-context", + "release", + "release-dormant", + ].includes( + String(value.operation), + ) || + !Array.isArray(value.bindingIds) || + value.bindingIds.length > 16 || + !value.bindingIds.every((entry) => identifier(entry, "binding")) || + new Set(value.bindingIds).size !== value.bindingIds.length || + !timestamp(value.createdAt) + ) { + throw new SubsessionCoordinatorStoreError("malformed_state"); + } + return structuredClone(value) as unknown as SubsessionCoordinatorRequestReceipt; +} + +function parseBindingTombstone( + value: unknown, +): SubsessionCoordinatorBindingTombstone { + if ( + !isRecord(value) || + !exact(value, [ + "bindingId", + "parentSessionId", + "parentBindingId", + "delegationDepth", + "delegationKey", + "bindingDigest", + "sessionId", + "disposition", + "closedAt", + ...("cleanupComplete" in value ? ["cleanupComplete"] : []), + ]) || + !identifier(value.bindingId, "binding") || + !identifier(value.parentSessionId) || + (value.parentBindingId !== null && + !identifier(value.parentBindingId, "binding")) || + !Number.isSafeInteger(value.delegationDepth) || + (value.delegationDepth as number) < 1 || + (value.delegationDepth as number) > PROJECT_SUBSESSION_MAX_DEPTH || + !identifier(value.delegationKey) || + !digest(value.bindingDigest) || + !identifier(value.sessionId) || + !["terminal", "dormant-evicted"].includes(String(value.disposition)) || + ("cleanupComplete" in value && + (value.cleanupComplete !== true || value.disposition !== "dormant-evicted")) || + !timestamp(value.closedAt) + ) { + throw new SubsessionCoordinatorStoreError("malformed_state"); + } + return structuredClone(value) as unknown as SubsessionCoordinatorBindingTombstone; +} + +export function parseSubsessionCoordinatorAggregate( + value: unknown, + expectedProjectId: StudioProjectId, +): SubsessionCoordinatorAggregate { + if (!isRecord(value)) + throw new SubsessionCoordinatorStoreError("malformed_state"); + if ( + value.schemaVersion !== SUBSESSION_COORDINATOR_STORAGE_SCHEMA_VERSION + ) { + throw new SubsessionCoordinatorStoreError( + "unsupported_schema", + typeof value.schemaVersion === "number" ? value.schemaVersion : undefined, + ); + } + if ( + !exact(value, [ + "schemaVersion", + "recordVersion", + "projectId", + "requestReceipts", + "requestTombstones", + "bindingTombstones", + "bindings", + "createdAt", + "updatedAt", + "aggregateDigest", + ]) || + value.projectId !== expectedProjectId || + !Number.isSafeInteger(value.recordVersion) || + (value.recordVersion as number) < 1 || + !Array.isArray(value.requestReceipts) || + value.requestReceipts.length > SUBSESSION_COORDINATOR_RECEIPT_LIMIT || + !Array.isArray(value.requestTombstones) || + value.requestTombstones.length > SUBSESSION_COORDINATOR_RECEIPT_LIMIT || + !Array.isArray(value.bindingTombstones) || + value.bindingTombstones.length > SUBSESSION_COORDINATOR_BINDING_LIMIT || + !Array.isArray(value.bindings) || + value.bindings.length > SUBSESSION_COORDINATOR_BINDING_LIMIT || + !timestamp(value.createdAt) || + !timestamp(value.updatedAt) || + !digest(value.aggregateDigest) || + aggregateDigest(value as unknown as SubsessionCoordinatorAggregate) !== + value.aggregateDigest + ) { + throw new SubsessionCoordinatorStoreError("malformed_state"); + } + const requestReceipts = value.requestReceipts.map(parseReceipt); + const requestTombstones = value.requestTombstones.map(parseReceipt); + const bindingTombstones = value.bindingTombstones.map(parseBindingTombstone); + const bindings = value.bindings.map((entry) => + parseBinding(entry, expectedProjectId), + ); + const requestKeys = [...requestReceipts, ...requestTombstones].map( + ({ parentSessionId, requestKey }) => `${parentSessionId}\0${requestKey}`, + ); + const bindingKeys = bindings.map( + ({ parentSessionId, delegationKey }) => + `${parentSessionId}\0${delegationKey}`, + ); + const allBindings = [...bindings, ...bindingTombstones]; + const terminalBindingKeys = [ + ...bindingKeys, + ...bindingTombstones.map( + ({ disposition, parentSessionId, delegationKey }) => + disposition === "terminal" + ? `${parentSessionId}\0${delegationKey}` + : null, + ), + ].filter((key): key is string => key !== null); + if ( + new Set(requestKeys).size !== requestKeys.length || + new Set(terminalBindingKeys).size !== terminalBindingKeys.length || + new Set(allBindings.map(({ bindingId }) => bindingId)).size !== + allBindings.length || + new Set(allBindings.map(({ sessionId }) => sessionId)).size !== + allBindings.length + ) { + throw new SubsessionCoordinatorStoreError("malformed_state"); + } + if ( + requestReceipts.some(({ bindingIds: ids }) => + ids.some( + (bindingId) => + !bindings.some((binding) => binding.bindingId === bindingId) && + !bindingTombstones.some( + (binding) => binding.bindingId === bindingId, + ), + ), + ) + ) { + throw new SubsessionCoordinatorStoreError("malformed_state"); + } + const bindingsById = new Map( + allBindings.map((binding) => [binding.bindingId, binding]), + ); + if ( + allBindings.some((binding) => { + if (binding.parentBindingId === null) + return binding.delegationDepth !== 1; + const parent = bindingsById.get(binding.parentBindingId); + return parent !== undefined && + binding.delegationDepth !== parent.delegationDepth + 1; + }) + ) { + throw new SubsessionCoordinatorStoreError("malformed_state"); + } + return { + ...structuredClone(value), + requestReceipts, + requestTombstones, + bindingTombstones, + bindings, + } as unknown as SubsessionCoordinatorAggregate; +} + +const transitions: Readonly> = { + reserved: ["spawn-claimed", "failed", "closed"], + "spawn-claimed": ["reserved", "starting", "failed", "closed"], + starting: ["awaiting-ready", "ready", "exited", "failed", "closed"], + "awaiting-ready": ["ready", "exited", "failed", "closed"], + ready: ["exited", "failed", "closed"], + exited: ["spawn-claimed", "failed", "closed"], + failed: ["spawn-claimed", "closed"], + closed: [], +}; + +export class SubsessionCoordinatorStore { + private readonly queues = new Map>(); + + constructor( + private readonly agentMapRoot: string, + private readonly options: { + now?: () => Date; + generateId?: () => string; + generateSessionId?: () => string; + claimTtlMs?: number; + receiptRetentionLimit?: number; + historyTombstoneLimit?: number; + bindingLimit?: number; + liveSessionLimit?: number; + maxDelegationDepth?: number; + onEvent?: (event: SubsessionCoordinatorStoreEvent) => void | Promise; + beforePersistStep?: ( + step: "write" | "file-sync" | "rename" | "directory-sync", + ) => void | Promise; + } = {}, + ) {} + + private filePath(projectId: StudioProjectId): string { + return path.join( + this.agentMapRoot, + "projects", + projectId, + "subsessions.json", + ); + } + + private now(): string { + return (this.options.now?.() ?? new Date()).toISOString(); + } + + private id(): string { + return (this.options.generateId ?? randomUUID)(); + } + + private bindingLimit(): number { + return Math.max( + 1, + Math.min( + this.options.bindingLimit ?? SUBSESSION_COORDINATOR_BINDING_LIMIT, + SUBSESSION_COORDINATOR_BINDING_LIMIT, + ), + ); + } + + private compactTerminalHistory(aggregate: MutableAggregate): void { + const historyLimit = Math.max( + 1, + Math.min( + this.options.historyTombstoneLimit ?? SUBSESSION_COORDINATOR_RECEIPT_LIMIT, + SUBSESSION_COORDINATOR_RECEIPT_LIMIT, + ), + ); + const retention = Math.max( + 1, + Math.min( + this.options.receiptRetentionLimit ?? + SUBSESSION_COORDINATOR_RECEIPT_RETENTION_LIMIT, + SUBSESSION_COORDINATOR_RECEIPT_LIMIT, + ), + ); + const expiring = Math.max(0, aggregate.requestReceipts.length - retention); + for (let count = 0; count < expiring; count += 1) { + const expired = aggregate.requestReceipts.shift(); + if (expired) { + aggregate.requestTombstones.push({ + parentSessionId: expired.parentSessionId, + requestKey: expired.requestKey, + requestDigest: expired.requestDigest, + operation: expired.operation, + bindingIds: expired.bindingIds, + createdAt: expired.createdAt, + }); + } + } + if (aggregate.requestTombstones.length > historyLimit) { + aggregate.requestTombstones.splice( + 0, + aggregate.requestTombstones.length - historyLimit, + ); + } + + const referenced = new Set( + aggregate.requestReceipts.flatMap(({ bindingIds }) => bindingIds), + ); + const reclaimable = aggregate.bindings.filter( + (binding) => + binding.sessionState === "closed" && !referenced.has(binding.bindingId), + ); + for (const binding of reclaimable) { + aggregate.bindingTombstones.push({ + bindingId: binding.bindingId, + parentSessionId: binding.parentSessionId, + parentBindingId: binding.parentBindingId, + delegationDepth: binding.delegationDepth, + delegationKey: binding.delegationKey, + bindingDigest: binding.bindingDigest, + sessionId: binding.sessionId, + disposition: "terminal", + closedAt: binding.updatedAt, + }); + } + if (aggregate.bindingTombstones.length > historyLimit) { + let remaining = aggregate.bindingTombstones.length - historyLimit; + aggregate.bindingTombstones = aggregate.bindingTombstones.filter( + ({ bindingId, disposition, cleanupComplete }) => { + if (remaining === 0 || referenced.has(bindingId) || + (disposition === "dormant-evicted" && !cleanupComplete)) return true; + remaining -= 1; + return false; + }, + ); + } + if (aggregate.bindingTombstones.length > SUBSESSION_COORDINATOR_BINDING_LIMIT) + throw new SubsessionCoordinatorStoreError("history_quota_exceeded"); + if (reclaimable.length > 0) { + const reclaimed = new Set(reclaimable.map(({ bindingId }) => bindingId)); + aggregate.bindings = aggregate.bindings.filter( + ({ bindingId }) => !reclaimed.has(bindingId), + ); + } + } + + private emit(event: SubsessionCoordinatorStoreEvent): void { + try { + void Promise.resolve(this.options.onEvent?.(event)).catch(() => {}); + } catch { + // Content-free observability cannot alter durable state. + } + } + + private initial(projectId: StudioProjectId): SubsessionCoordinatorAggregate { + const now = this.now(); + const initial = { + schemaVersion: SUBSESSION_COORDINATOR_STORAGE_SCHEMA_VERSION, + recordVersion: 1, + projectId, + requestReceipts: [], + requestTombstones: [], + bindingTombstones: [], + bindings: [], + createdAt: now, + updatedAt: now, + } as const; + return { ...initial, aggregateDigest: aggregateDigest(initial) }; + } + + private async readDisk(projectId: StudioProjectId): Promise<{ + aggregate: SubsessionCoordinatorAggregate; + created: boolean; + }> { + try { + const decoded = JSON.parse( + await fs.readFile(this.filePath(projectId), "utf8"), + ) as unknown; + return { + aggregate: parseSubsessionCoordinatorAggregate(decoded, projectId), + created: false, + }; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") + return { aggregate: this.initial(projectId), created: true }; + if ( + error instanceof SubsessionCoordinatorStoreError && + error.code !== "storage_unavailable" + ) { + throw error; + } + if (error instanceof SyntaxError) + throw new SubsessionCoordinatorStoreError("malformed_state"); + throw storageError(); + } + } + + private async persist( + projectId: StudioProjectId, + aggregate: SubsessionCoordinatorAggregate, + ): Promise { + const file = this.filePath(projectId); + const directory = path.dirname(file); + const temporary = `${file}.tmp-${process.pid}-${randomUUID()}`; + let handle: fs.FileHandle | undefined; + try { + await fs.mkdir(directory, { recursive: true, mode: 0o700 }); + handle = await fs.open(temporary, "wx", 0o600); + await this.options.beforePersistStep?.("write"); + await handle.writeFile(`${JSON.stringify(aggregate, null, 2)}\n`, "utf8"); + await this.options.beforePersistStep?.("file-sync"); + await handle.sync(); + await handle.close(); + handle = undefined; + await this.options.beforePersistStep?.("rename"); + await fs.rename(temporary, file); + await fs.chmod(file, 0o600); + const directoryHandle = await fs.open(directory, "r"); + try { + await this.options.beforePersistStep?.("directory-sync"); + await directoryHandle.sync(); + } finally { + await directoryHandle.close(); + } + } catch { + throw storageError(); + } finally { + await handle?.close().catch(() => {}); + await fs.rm(temporary, { force: true }).catch(() => {}); + } + } + + private enqueue( + projectId: StudioProjectId, + operation: () => Promise, + ): Promise { + const previous = this.queues.get(projectId) ?? Promise.resolve(); + const result = previous.then(operation, operation); + const tail = result.then( + () => undefined, + () => undefined, + ); + this.queues.set(projectId, tail); + void tail.finally(() => { + if (this.queues.get(projectId) === tail) this.queues.delete(projectId); + }); + return result; + } + + private async transact( + projectId: StudioProjectId, + operation: ( + aggregate: MutableAggregate, + ) => Promise<{ value: T; next?: MutableAggregate }>, + ): Promise { + if (!isStudioProjectId(projectId)) + throw new SubsessionCoordinatorStoreError("malformed_state"); + return this.enqueue(projectId, async () => { + const release = await new DurableFileLock(this.filePath(projectId), { + storageError, + }).acquire(); + try { + const loaded = await this.readDisk(projectId); + const outcome = await operation( + structuredClone(loaded.aggregate) as unknown as MutableAggregate, + ); + if (loaded.created || outcome.next) { + const candidate = outcome.next ?? + (structuredClone(loaded.aggregate) as unknown as MutableAggregate); + const sealed = parseSubsessionCoordinatorAggregate( + { + ...candidate, + aggregateDigest: aggregateDigest(candidate), + }, + projectId, + ); + await this.persist(projectId, sealed); + } + if (loaded.created) + this.emit({ name: "subsession.store_initialized", projectId }); + return structuredClone(outcome.value); + } finally { + await release(); + } + }); + } + + read(projectId: StudioProjectId): Promise { + return this.transact(projectId, async (aggregate) => ({ value: aggregate })); + } + + readBinding( + identity: ProjectAgentSession, + selector: Readonly< + | { kind: "binding-id"; bindingId: SubsessionBindingId } + | { kind: "child"; delegationKey: string } + | { kind: "self" } + >, + ): Promise { + return this.transact(identity.projectId, async (aggregate) => { + const binding = + selector.kind === "binding-id" + ? aggregate.bindings.find( + (entry) => + entry.bindingId === selector.bindingId && + entry.parentSessionId === identity.sessionId, + ) + : selector.kind === "child" + ? aggregate.bindings.find( + (entry) => + entry.parentSessionId === identity.sessionId && + entry.delegationKey === selector.delegationKey, + ) + : aggregate.bindings.find( + (entry) => entry.sessionId === identity.sessionId, + ); + if (!binding) + throw new SubsessionCoordinatorStoreError("binding_not_found"); + if (binding.projectId !== identity.projectId) + throw new SubsessionCoordinatorStoreError("binding_scope_mismatch"); + return { value: binding }; + }); + } + + setFocusedContextState( + identity: ProjectAgentSession, + bindingId: SubsessionBindingId, + request: Readonly<{ + expectedContextEpoch: number; + expectedContextDigest: string; + state: "none" | "current" | "stale"; + projectionDigest: SubsessionProjectionDigest | null; + }>, + ): Promise { + return this.transact(identity.projectId, async (aggregate) => { + const binding = this.scopedBinding(aggregate, identity, bindingId); + if ( + binding.contextEpoch !== request.expectedContextEpoch || + binding.contextDigest !== request.expectedContextDigest + ) { + throw new SubsessionCoordinatorStoreError("lifecycle_conflict"); + } + if ( + binding.contextState === request.state && + binding.projectionDigest === request.projectionDigest + ) { + return { value: binding }; + } + const now = this.now(); + binding.contextState = request.state; + binding.projectionDigest = request.projectionDigest; + binding.updatedAt = now; + aggregate.recordVersion += 1; + aggregate.updatedAt = now; + return { value: binding, next: aggregate }; + }); + } + + closeBinding( + identity: ProjectAgentSession, + bindingId: SubsessionBindingId, + expectedSessionId: string, + ): Promise { + return this.transact(identity.projectId, async (aggregate) => { + const binding = this.scopedBinding(aggregate, identity, bindingId); + if (binding.sessionId !== expectedSessionId) + throw new SubsessionCoordinatorStoreError("binding_scope_mismatch"); + if (binding.sessionState === "closed") return { value: binding }; + const now = this.now(); + binding.sessionState = "closed"; + binding.lifecycleEpoch += 1; + binding.spawnClaim = null; + binding.runtime = null; + binding.updatedAt = now; + this.compactTerminalHistory(aggregate); + aggregate.recordVersion += 1; + aggregate.updatedAt = now; + return { value: binding, next: aggregate }; + }); + } + + /** Compacts an exact durably closed binding while release receipts retain replay. */ + finalizeReleasedBinding( + identity: ProjectAgentSession, + bindingId: SubsessionBindingId, + expectedSessionId: string, + ): Promise { + return this.transact(identity.projectId, async (aggregate) => { + const existing = aggregate.bindingTombstones.find( + (entry) => entry.bindingId === bindingId, + ); + if (existing) { + if ( + existing.parentSessionId !== identity.sessionId || + existing.sessionId !== expectedSessionId + ) { + throw new SubsessionCoordinatorStoreError("binding_scope_mismatch"); + } + return { value: existing }; + } + const binding = this.scopedBinding(aggregate, identity, bindingId); + if (binding.sessionId !== expectedSessionId) + throw new SubsessionCoordinatorStoreError("binding_scope_mismatch"); + if (binding.sessionState !== "closed") + throw new SubsessionCoordinatorStoreError("lifecycle_conflict"); + const tombstone: SubsessionCoordinatorBindingTombstone = { + bindingId: binding.bindingId, + parentSessionId: binding.parentSessionId, + parentBindingId: binding.parentBindingId, + delegationDepth: binding.delegationDepth, + delegationKey: binding.delegationKey, + bindingDigest: binding.bindingDigest, + sessionId: binding.sessionId, + disposition: "terminal", + closedAt: binding.updatedAt, + }; + aggregate.bindings = aggregate.bindings.filter( + (entry) => entry.bindingId !== bindingId, + ); + aggregate.bindingTombstones.push(tombstone); + this.compactTerminalHistory(aggregate); + aggregate.recordVersion += 1; + aggregate.updatedAt = this.now(); + return { value: tombstone, next: aggregate }; + }); + } + + reserveReleases( + identity: ProjectAgentSession, + rawRequest: unknown, + ): Promise { + const request = parseProjectSubsessionRequest(rawRequest, identity.projectId); + if (request.operation.kind !== "release") + throw new SubsessionCoordinatorStoreError("malformed_state"); + const operation = request.operation; + const requestDigest = computeCanonicalDelegationRequestDigest(request); + return this.transact(identity.projectId, async (aggregate) => { + const sameRequest = ( + receipt: Pick, + ) => + receipt.parentSessionId === identity.sessionId && + receipt.requestKey === request.requestKey; + const resolve = (bindingId: SubsessionBindingId): ReleasableSubsessionBinding => { + const binding = aggregate.bindings.find( + (entry) => entry.bindingId === bindingId, + ); + if (binding) { + if (binding.parentSessionId !== identity.sessionId) + throw new SubsessionCoordinatorStoreError("binding_scope_mismatch"); + return { state: "bound", binding }; + } + const released = aggregate.bindingTombstones.find( + (entry) => entry.bindingId === bindingId, + ); + if (!released || released.parentSessionId !== identity.sessionId) + throw new SubsessionCoordinatorStoreError("malformed_state"); + return { state: "released", binding: released }; + }; + const previous = aggregate.requestReceipts.find(sameRequest); + if (previous) { + if ( + previous.requestDigest !== requestDigest || + previous.operation !== "release" + ) { + throw new SubsessionCoordinatorStoreError("request_key_reused"); + } + const resolved = previous.bindingIds.map(resolve); + return { + value: { + replayed: true, + requestDigest, + bindings: operation.delegationKeys.map( + (delegationKey) => + resolved.find( + (entry) => + entry.state !== "absent" && + entry.binding.delegationKey === delegationKey, + ) ?? { state: "absent", delegationKey }, + ), + }, + }; + } + if (aggregate.requestTombstones.some(sameRequest)) + throw new SubsessionCoordinatorStoreError("request_key_expired"); + const bindings = operation.delegationKeys.map( + (delegationKey): ReleasableSubsessionBinding => { + const binding = aggregate.bindings.find( + (entry) => + entry.parentSessionId === identity.sessionId && + entry.delegationKey === delegationKey, + ); + if (binding) return { state: "bound", binding }; + const released = aggregate.bindingTombstones.find( + (entry) => + entry.parentSessionId === identity.sessionId && + entry.delegationKey === delegationKey, + ); + if (released) return { state: "released", binding: released }; + return { state: "absent", delegationKey }; + }, + ); + const now = this.now(); + aggregate.requestReceipts.push({ + parentSessionId: identity.sessionId, + requestKey: request.requestKey, + requestDigest, + operation: "release", + bindingIds: bindings.flatMap((entry) => + entry.state === "absent" ? [] : [entry.binding.bindingId], + ), + createdAt: now, + }); + this.compactTerminalHistory(aggregate); + aggregate.recordVersion += 1; + aggregate.updatedAt = now; + return { + value: { replayed: false, requestDigest, bindings }, + next: aggregate, + }; + }); + } + + /** + * Reserves an explicit project-scoped cleanup of dormant coordinator-owned + * bindings. Candidate IDs are selected by the trusted coordinator and never + * accepted from the public request. The transaction rechecks the child state; + * parent liveness is intentionally irrelevant to this explicit project-wide + * destructive operation. + */ + reserveDormantReleases( + identity: ProjectAgentSession, + rawRequest: unknown, + candidateBindingIds: readonly SubsessionBindingId[], + ): Promise { + const request = parseProjectSubsessionRequest(rawRequest, identity.projectId); + if (request.operation.kind !== "release-dormant") + throw new SubsessionCoordinatorStoreError("malformed_state"); + if ( + candidateBindingIds.length > request.operation.limit || + new Set(candidateBindingIds).size !== candidateBindingIds.length || + !candidateBindingIds.every((bindingId) => + identifier(bindingId, "binding"), + ) + ) { + throw new SubsessionCoordinatorStoreError("malformed_state"); + } + const requestDigest = computeCanonicalDelegationRequestDigest(request); + return this.transact(identity.projectId, async (aggregate) => { + const sameRequest = ( + receipt: Pick, + ) => + receipt.parentSessionId === identity.sessionId && + receipt.requestKey === request.requestKey; + const resolve = ( + bindingId: SubsessionBindingId, + ): ReleasableSubsessionBinding => { + const binding = aggregate.bindings.find( + (entry) => entry.bindingId === bindingId, + ); + if (binding) return { state: "bound", binding }; + const released = aggregate.bindingTombstones.find( + (entry) => entry.bindingId === bindingId, + ); + if (released) return { state: "released", binding: released }; + throw new SubsessionCoordinatorStoreError("malformed_state"); + }; + const previous = aggregate.requestReceipts.find(sameRequest); + if (previous) { + if ( + previous.requestDigest !== requestDigest || + previous.operation !== "release-dormant" + ) { + throw new SubsessionCoordinatorStoreError("request_key_reused"); + } + return { + value: { + replayed: true, + requestDigest, + bindings: previous.bindingIds.map(resolve), + }, + }; + } + if (aggregate.requestTombstones.some(sameRequest)) + throw new SubsessionCoordinatorStoreError("request_key_expired"); + this.compactTerminalHistory(aggregate); + const now = this.now(); + const bindings: ReleasableSubsessionBinding[] = []; + const evictedBindingIds = new Set(); + for (const bindingId of candidateBindingIds) { + const binding = aggregate.bindings.find( + (entry) => entry.bindingId === bindingId, + ); + if (binding) { + if (["exited", "failed"].includes(binding.sessionState)) { + // The explicit destructive boundary and request receipt commit in + // the same transaction. A concurrent resume must lose this fence + // before any exact private ownership marker is removed. + const tombstone: SubsessionCoordinatorBindingTombstone = { + bindingId: binding.bindingId, + parentSessionId: binding.parentSessionId, + parentBindingId: binding.parentBindingId, + delegationDepth: binding.delegationDepth, + delegationKey: binding.delegationKey, + bindingDigest: binding.bindingDigest, + sessionId: binding.sessionId, + disposition: "dormant-evicted", + closedAt: now, + }; + aggregate.bindingTombstones.push(tombstone); + evictedBindingIds.add(binding.bindingId); + bindings.push({ state: "evicted", binding: tombstone }); + } + continue; + } + const released = aggregate.bindingTombstones.find( + (entry) => entry.bindingId === bindingId, + ); + if (released) bindings.push({ state: "released", binding: released }); + } + if (evictedBindingIds.size > 0) { + const retainedReceipts: SubsessionCoordinatorRequestReceipt[] = []; + for (const receipt of aggregate.requestReceipts) { + if ( + receipt.operation !== "release-dormant" && + receipt.bindingIds.some((bindingId) => + evictedBindingIds.has(bindingId), + ) + ) { + aggregate.requestTombstones.push(receipt); + } else { + retainedReceipts.push(receipt); + } + } + aggregate.requestReceipts = retainedReceipts; + aggregate.bindings = aggregate.bindings.filter( + ({ bindingId }) => !evictedBindingIds.has(bindingId), + ); + } + aggregate.requestReceipts.push({ + parentSessionId: identity.sessionId, + requestKey: request.requestKey, + requestDigest, + operation: "release-dormant", + bindingIds: bindings.flatMap((entry) => + entry.state === "absent" ? [] : [entry.binding.bindingId], + ), + createdAt: now, + }); + this.compactTerminalHistory(aggregate); + aggregate.recordVersion += 1; + aggregate.updatedAt = now; + return { + value: { replayed: false, requestDigest, bindings }, + next: aggregate, + }; + }); + } + + /** Release eviction proof only after the exact private session close succeeds. */ + completeDormantReleaseCleanup( + identity: ProjectAgentSession, + bindingId: SubsessionBindingId, + expectedSessionId: string, + ): Promise { + return this.transact(identity.projectId, async (aggregate) => { + const index = aggregate.bindingTombstones.findIndex( + (entry) => entry.bindingId === bindingId, + ); + const binding = aggregate.bindingTombstones[index]; + if (!binding) return { value: undefined }; + if (binding.sessionId !== expectedSessionId || binding.disposition !== "dormant-evicted") + throw new SubsessionCoordinatorStoreError("binding_scope_mismatch"); + if (binding.cleanupComplete) return { value: undefined }; + aggregate.bindingTombstones[index] = { ...binding, cleanupComplete: true }; + this.compactTerminalHistory(aggregate); + aggregate.recordVersion += 1; + aggregate.updatedAt = this.now(); + return { value: undefined, next: aggregate }; + }); + } + + /** Server-only bridge from SessionManager's private two-sided marker. */ + closeOwnedBinding(marker: Readonly<{ + projectId: StudioProjectId; + parentSessionId: string; + bindingId: string; + sessionId: string; + }>): Promise { + return this.transact(marker.projectId, async (aggregate) => { + const binding = aggregate.bindings.find( + ({ bindingId }) => bindingId === marker.bindingId, + ); + if (!binding) { + const tombstone = aggregate.bindingTombstones.find( + ({ bindingId }) => bindingId === marker.bindingId, + ); + if ( + tombstone?.parentSessionId === marker.parentSessionId && + tombstone.sessionId === marker.sessionId + ) { + return { value: undefined }; + } + throw new SubsessionCoordinatorStoreError("binding_not_found"); + } + if ( + binding.parentSessionId !== marker.parentSessionId || + binding.sessionId !== marker.sessionId + ) { + throw new SubsessionCoordinatorStoreError("binding_scope_mismatch"); + } + if (binding.sessionState === "closed") return { value: undefined }; + const now = this.now(); + binding.sessionState = "closed"; + binding.lifecycleEpoch += 1; + binding.spawnClaim = null; + binding.runtime = null; + binding.updatedAt = now; + this.compactTerminalHistory(aggregate); + aggregate.recordVersion += 1; + aggregate.updatedAt = now; + return { value: undefined, next: aggregate }; + }); + } + + refreshFocusedContext( + identity: ProjectAgentSession, + rawRequest: unknown, + ): Promise { + const request = parseProjectSubsessionRequest(rawRequest, identity.projectId); + if (request.operation.kind !== "refresh-focused-context") + throw new SubsessionCoordinatorStoreError("malformed_state"); + const operation = request.operation; + const targetSelector = operation.target; + const requestDigest = computeCanonicalDelegationRequestDigest(request); + return this.transact(identity.projectId, async (aggregate) => { + const sameRequest = ( + receipt: Pick, + ) => + receipt.parentSessionId === identity.sessionId && + receipt.requestKey === request.requestKey; + const previous = aggregate.requestReceipts.find(sameRequest); + if (previous) { + if ( + previous.requestDigest !== requestDigest || + previous.operation !== "refresh-focused-context" || + previous.bindingIds.length !== 1 + ) { + throw new SubsessionCoordinatorStoreError("request_key_reused"); + } + const binding = aggregate.bindings.find( + ({ bindingId }) => bindingId === previous.bindingIds[0], + ); + if (!binding && aggregate.bindingTombstones.some( + ({ bindingId }) => bindingId === previous.bindingIds[0], + )) { + throw new SubsessionCoordinatorStoreError("session_closed"); + } + if (!binding) + throw new SubsessionCoordinatorStoreError("malformed_state"); + return { value: { replayed: true, requestDigest, binding } }; + } + if (aggregate.requestTombstones.some(sameRequest)) + throw new SubsessionCoordinatorStoreError("request_key_expired"); + const target = + targetSelector.kind === "self" + ? aggregate.bindings.find( + ({ sessionId }) => sessionId === identity.sessionId, + ) + : aggregate.bindings.find( + ({ parentSessionId, delegationKey }) => + parentSessionId === identity.sessionId && + delegationKey === targetSelector.delegationKey, + ); + if (!target) + throw new SubsessionCoordinatorStoreError("binding_not_found"); + if ( + target.contextEpoch !== operation.expectedContextEpoch || + target.contextDigest !== operation.expectedContextDigest + ) { + throw new SubsessionCoordinatorStoreError("lifecycle_conflict"); + } + const currentDelivery = target.deliveries.find( + ({ contextEpoch }) => contextEpoch === target.contextEpoch, + ); + if (currentDelivery?.state === "uncertain") + throw new SubsessionCoordinatorStoreError("claim_conflict"); + target.deliveries = target.deliveries.filter(({ state }) => + ["claimed", "submitted-unacknowledged", "uncertain"].includes(state), + ); + if (target.deliveries.length >= SUBSESSION_COORDINATOR_DELIVERY_LIMIT) + throw new SubsessionCoordinatorStoreError("history_quota_exceeded"); + + const now = this.now(); + target.contextEpoch += 1; + target.contextDigest = computeSubsessionContextDigest( + operation.focus, + ); + target.contextState = + operation.focus === null ? "none" : "refreshing"; + target.currentFocus = operation.focus; + target.projectionDigest = null; + target.deliveries.push({ + contextEpoch: target.contextEpoch, + deliveryId: `delivery_${this.id()}`, + inputId: `input_${this.id()}`, + eventWatermark: null, + state: "pending", + attempt: 0, + claim: null, + submittedAt: null, + acknowledgedAt: null, + }); + target.updatedAt = now; + aggregate.requestReceipts.push({ + parentSessionId: identity.sessionId, + requestKey: request.requestKey, + requestDigest, + operation: "refresh-focused-context", + bindingIds: [target.bindingId], + createdAt: now, + }); + this.compactTerminalHistory(aggregate); + aggregate.recordVersion += 1; + aggregate.updatedAt = now; + return { + value: { replayed: false, requestDigest, binding: target }, + next: aggregate, + }; + }); + } + + async reserveDelegations( + identity: ProjectAgentSession, + rawRequest: unknown, + target: Readonly<{ + harness: HarnessKind; + projectRoot: string; + ownerId: string; + }>, + ): Promise { + parseProjectAgentActorRef({ + userId: identity.userId, + sessionId: identity.sessionId, + }); + const request = parseProjectSubsessionRequest(rawRequest, identity.projectId); + if (request.operation.kind !== "delegate") + throw new SubsessionCoordinatorStoreError("malformed_state"); + if ( + !["claude-code", "codex"].includes(target.harness) || + !path.isAbsolute(target.projectRoot) || + target.projectRoot.includes("\0") || + !identifier(target.ownerId) + ) { + throw new SubsessionCoordinatorStoreError("malformed_state"); + } + const requestDigest = computeCanonicalDelegationRequestDigest(request); + const operation = request.operation; + return this.transact(identity.projectId, async (aggregate) => { + const now = this.now(); + const activeCount = () => + aggregate.bindings.filter(({ sessionState }) => + [ + "reserved", + "spawn-claimed", + "starting", + "awaiting-ready", + "ready", + ].includes(sessionState), + ).length; + const activateDormant = (binding: MutableBinding): void => { + binding.spawnEpoch += 1; + binding.lifecycleEpoch += 1; + binding.spawnClaim = this.claim(now, target.ownerId) as MutableClaim; + binding.sessionState = "spawn-claimed"; + binding.updatedAt = now; + }; + const sameRequest = ( + receipt: Pick, + ): boolean => + receipt.parentSessionId === identity.sessionId && + receipt.requestKey === request.requestKey; + const previous = aggregate.requestReceipts.find(sameRequest); + if (previous) { + if ( + previous.requestDigest !== requestDigest || + previous.operation !== "delegate" + ) { + throw new SubsessionCoordinatorStoreError("request_key_reused"); + } + const bindings = previous.bindingIds.map((bindingId) => { + const binding = aggregate.bindings.find( + (entry) => entry.bindingId === bindingId, + ); + if (!binding && aggregate.bindingTombstones.some( + (entry) => entry.bindingId === bindingId, + )) { + throw new SubsessionCoordinatorStoreError("session_closed"); + } + if (!binding) + throw new SubsessionCoordinatorStoreError("malformed_state"); + return binding; + }); + const dormant = bindings.filter(({ sessionState }) => + ["exited", "failed"].includes(sessionState), + ); + if ( + activeCount() + dormant.length > + (this.options.liveSessionLimit ?? + PROJECT_SUBSESSION_LIVE_SESSION_LIMIT) + ) { + throw new SubsessionCoordinatorStoreError( + "live_session_limit_reached", + ); + } + for (const binding of dormant) activateDormant(binding); + this.emit({ + name: "subsession.duplicate_prevented", + projectId: identity.projectId, + count: bindings.length, + }); + if (dormant.length === 0) + return { value: { replayed: true, requestDigest, bindings } }; + aggregate.recordVersion += 1; + aggregate.updatedAt = now; + return { + value: { replayed: true, requestDigest, bindings }, + next: aggregate, + }; + } + if (aggregate.requestTombstones.some(sameRequest)) + throw new SubsessionCoordinatorStoreError("request_key_expired"); + this.compactTerminalHistory(aggregate); + const bindings: SubsessionBindingRecord[] = []; + let created = 0; + const live = activeCount(); + const parentBinding = aggregate.bindings.find( + ({ sessionId }) => sessionId === identity.sessionId, + ); + const delegationDepth = (parentBinding?.delegationDepth ?? 0) + 1; + if ( + delegationDepth > + Math.min( + this.options.maxDelegationDepth ?? PROJECT_SUBSESSION_MAX_DEPTH, + PROJECT_SUBSESSION_MAX_DEPTH, + ) + ) { + throw new SubsessionCoordinatorStoreError("delegation_depth_exceeded"); + } + let additionalLive = 0; + for (const delegation of operation.delegations) { + const bindingDigest = + computeCanonicalDelegationBindingDigest(delegation); + const existing = aggregate.bindings.find( + (entry) => + entry.parentSessionId === identity.sessionId && + entry.delegationKey === delegation.delegationKey, + ); + if (existing) { + if (existing.bindingDigest !== bindingDigest) + throw new SubsessionCoordinatorStoreError( + "delegation_key_reused", + ); + if (existing.sessionState === "closed") + throw new SubsessionCoordinatorStoreError("session_closed"); + if (["exited", "failed"].includes(existing.sessionState)) { + additionalLive += 1; + activateDormant(existing); + } + bindings.push(existing); + continue; + } + const terminal = aggregate.bindingTombstones.find( + (entry) => + entry.parentSessionId === identity.sessionId && + entry.delegationKey === delegation.delegationKey && + entry.disposition === "terminal", + ); + if (terminal) { + if (terminal.bindingDigest !== bindingDigest) + throw new SubsessionCoordinatorStoreError("delegation_key_reused"); + throw new SubsessionCoordinatorStoreError("session_closed"); + } + if (aggregate.bindings.length >= this.bindingLimit()) { + throw new SubsessionCoordinatorStoreError("history_quota_exceeded"); + } + additionalLive += 1; + const contextFocus = delegation.focus ?? null; + const contextEpoch = 1; + const binding: SubsessionBindingRecord = { + bindingId: `binding_${this.id()}` as SubsessionBindingId, + projectId: identity.projectId, + parentSessionId: identity.sessionId, + parentBindingId: parentBinding?.bindingId ?? null, + delegationDepth, + delegationKey: delegation.delegationKey, + bindingDigest, + outcome: delegation.outcome, + kickoffContext: delegation.kickoffContext ?? null, + initialFocus: contextFocus, + sessionId: (this.options.generateSessionId ?? randomUUID)(), + harness: target.harness, + projectRoot: target.projectRoot, + lifecycleEpoch: 1, + spawnEpoch: 0, + contextEpoch, + contextDigest: computeSubsessionContextDigest(contextFocus), + contextState: contextFocus === null ? "none" : "current", + currentFocus: contextFocus, + projectionDigest: null, + sessionState: "reserved", + spawnClaim: null, + runtime: null, + deliveries: [ + { + contextEpoch, + deliveryId: `delivery_${this.id()}`, + inputId: `input_${this.id()}`, + eventWatermark: null, + state: "pending", + attempt: 0, + claim: null, + submittedAt: null, + acknowledgedAt: null, + }, + ], + lastError: null, + createdAt: now, + updatedAt: now, + }; + aggregate.bindings.push(binding as unknown as MutableBinding); + bindings.push(binding); + created += 1; + } + if ( + live + additionalLive > + (this.options.liveSessionLimit ?? PROJECT_SUBSESSION_LIVE_SESSION_LIMIT) + ) { + throw new SubsessionCoordinatorStoreError("live_session_limit_reached"); + } + const receipt: SubsessionCoordinatorRequestReceipt = { + parentSessionId: identity.sessionId, + requestKey: request.requestKey, + requestDigest, + operation: "delegate", + bindingIds: bindings.map(({ bindingId }) => bindingId), + createdAt: now, + }; + aggregate.requestReceipts.push(receipt); + this.compactTerminalHistory(aggregate); + aggregate.recordVersion += 1; + aggregate.updatedAt = now; + this.emit({ + name: "subsession.binding_reserved", + projectId: identity.projectId, + count: created, + }); + return { + value: { replayed: false, requestDigest, bindings }, + next: aggregate, + }; + }); + } + + private scopedBinding( + aggregate: MutableAggregate, + identity: ProjectAgentSession, + bindingId: SubsessionBindingId, + ): MutableBinding { + const binding = aggregate.bindings.find( + (entry) => entry.bindingId === bindingId, + ); + if (!binding) + throw new SubsessionCoordinatorStoreError("binding_not_found"); + if ( + binding.projectId !== identity.projectId || + binding.parentSessionId !== identity.sessionId + ) { + throw new SubsessionCoordinatorStoreError("binding_scope_mismatch"); + } + return binding; + } + + private claim(now: string, ownerId: string): SubsessionClaim { + return { + claimId: `claim_${this.id()}`, + ownerId, + claimedAt: now, + expiresAt: new Date( + new Date(now).getTime() + + (this.options.claimTtlMs ?? PROJECT_SUBSESSION_CLAIM_TTL_MS), + ).toISOString(), + }; + } + + claimSpawn( + identity: ProjectAgentSession, + bindingId: SubsessionBindingId, + request: Readonly<{ + ownerId: string; + expectedLifecycleEpoch: number; + expectedSpawnEpoch: number; + }>, + ): Promise { + return this.transact(identity.projectId, async (aggregate) => { + const binding = this.scopedBinding(aggregate, identity, bindingId); + if (binding.sessionState === "closed") + throw new SubsessionCoordinatorStoreError("session_closed"); + if (binding.spawnClaim) { + return { + value: { + claimed: false, + reason: + binding.spawnClaim.expiresAt <= this.now() + ? "expired-requires-inspection" + : "active", + binding, + }, + }; + } + if ( + binding.lifecycleEpoch !== request.expectedLifecycleEpoch || + binding.spawnEpoch !== request.expectedSpawnEpoch || + !["reserved", "exited", "failed"].includes(binding.sessionState) || + binding.runtime !== null + ) { + throw new SubsessionCoordinatorStoreError("lifecycle_conflict"); + } + const now = this.now(); + binding.spawnEpoch += 1; + binding.lifecycleEpoch += 1; + binding.spawnClaim = this.claim(now, request.ownerId) as MutableClaim; + binding.sessionState = "spawn-claimed"; + binding.updatedAt = now; + aggregate.recordVersion += 1; + aggregate.updatedAt = now; + this.emit({ + name: "subsession.spawn_claimed", + projectId: identity.projectId, + }); + return { + value: { claimed: true, binding }, + next: aggregate, + }; + }); + } + + takeoverExpiredSpawnClaim( + identity: ProjectAgentSession, + bindingId: SubsessionBindingId, + request: Readonly<{ + ownerId: string; + expiredClaimId: string; + expectedLifecycleEpoch: number; + expectedSpawnEpoch: number; + }>, + ): Promise { + return this.transact(identity.projectId, async (aggregate) => { + const binding = this.scopedBinding(aggregate, identity, bindingId); + const now = this.now(); + if ( + !binding.spawnClaim || + binding.spawnClaim.claimId !== request.expiredClaimId || + binding.spawnClaim.expiresAt > now || + binding.lifecycleEpoch !== request.expectedLifecycleEpoch || + binding.spawnEpoch !== request.expectedSpawnEpoch || + binding.sessionState !== "spawn-claimed" || + binding.runtime !== null + ) { + throw new SubsessionCoordinatorStoreError("claim_conflict"); + } + binding.spawnEpoch += 1; + binding.lifecycleEpoch += 1; + binding.spawnClaim = this.claim(now, request.ownerId) as MutableClaim; + binding.updatedAt = now; + aggregate.recordVersion += 1; + aggregate.updatedAt = now; + return { + value: { claimed: true, binding }, + next: aggregate, + }; + }); + } + + releaseUnspawnedClaim( + identity: ProjectAgentSession, + bindingId: SubsessionBindingId, + request: Readonly<{ + claimId: string; + spawnEpoch: number; + proof: "no-process-created"; + }>, + ): Promise { + return this.transact(identity.projectId, async (aggregate) => { + const binding = this.scopedBinding(aggregate, identity, bindingId); + if ( + binding.spawnClaim?.claimId !== request.claimId || + binding.spawnEpoch !== request.spawnEpoch || + binding.sessionState !== "spawn-claimed" || + binding.runtime !== null + ) { + throw new SubsessionCoordinatorStoreError("claim_conflict"); + } + const now = this.now(); + binding.spawnClaim = null; + binding.sessionState = "reserved"; + binding.lifecycleEpoch += 1; + binding.updatedAt = now; + aggregate.recordVersion += 1; + aggregate.updatedAt = now; + return { value: binding, next: aggregate }; + }); + } + + attachSpawnedRuntime( + identity: ProjectAgentSession, + bindingId: SubsessionBindingId, + request: Readonly<{ + claimId: string; + spawnEpoch: number; + runtimeToken: string; + incarnation: number; + }>, + ): Promise { + return this.transact(identity.projectId, async (aggregate) => { + const binding = this.scopedBinding(aggregate, identity, bindingId); + if ( + binding.spawnClaim?.claimId !== request.claimId || + binding.spawnEpoch !== request.spawnEpoch || + binding.sessionState !== "spawn-claimed" || + binding.runtime !== null || + !identifier(request.runtimeToken) || + !Number.isSafeInteger(request.incarnation) || + request.incarnation < 1 + ) { + throw new SubsessionCoordinatorStoreError("claim_conflict"); + } + const now = this.now(); + binding.runtime = { + runtimeToken: request.runtimeToken, + incarnation: request.incarnation, + spawnEpoch: request.spawnEpoch, + }; + binding.spawnClaim = null; + binding.sessionState = "starting"; + binding.lifecycleEpoch += 1; + binding.updatedAt = now; + aggregate.recordVersion += 1; + aggregate.updatedAt = now; + return { value: binding, next: aggregate }; + }); + } + + transitionSession( + identity: ProjectAgentSession, + bindingId: SubsessionBindingId, + request: Readonly<{ + expectedLifecycleEpoch: number; + expectedSpawnEpoch: number; + expectedRuntimeToken: string | null; + state: DelegatedSessionState; + error?: DelegationError | null; + }>, + ): Promise { + return this.transact(identity.projectId, async (aggregate) => { + const binding = this.scopedBinding(aggregate, identity, bindingId); + if ( + binding.lifecycleEpoch !== request.expectedLifecycleEpoch || + binding.spawnEpoch !== request.expectedSpawnEpoch || + (binding.runtime?.runtimeToken ?? null) !== request.expectedRuntimeToken || + !transitions[binding.sessionState].includes(request.state) + ) { + throw new SubsessionCoordinatorStoreError("lifecycle_conflict"); + } + const now = this.now(); + binding.sessionState = request.state; + if (request.state !== "spawn-claimed") binding.spawnClaim = null; + binding.lifecycleEpoch += 1; + if (["exited", "failed", "closed"].includes(request.state)) + binding.runtime = null; + binding.lastError = request.error ?? null; + binding.updatedAt = now; + aggregate.recordVersion += 1; + aggregate.updatedAt = now; + return { value: binding, next: aggregate }; + }); + } + + claimKickoff( + identity: ProjectAgentSession, + bindingId: SubsessionBindingId, + request: Readonly<{ + ownerId: string; + expectedLifecycleEpoch: number; + expectedSpawnEpoch: number; + expectedContextEpoch: number; + eventWatermark: string; + }>, + ): Promise { + return this.transact(identity.projectId, async (aggregate) => { + const binding = this.scopedBinding(aggregate, identity, bindingId); + const delivery = binding.deliveries.find( + (entry) => entry.contextEpoch === request.expectedContextEpoch, + ); + if (!delivery) + throw new SubsessionCoordinatorStoreError("lifecycle_conflict"); + if ( + binding.lifecycleEpoch !== request.expectedLifecycleEpoch || + binding.spawnEpoch !== request.expectedSpawnEpoch || + binding.contextEpoch !== request.expectedContextEpoch || + binding.sessionState !== "ready" || + !binding.runtime + ) { + throw new SubsessionCoordinatorStoreError("lifecycle_conflict"); + } + if (delivery.state !== "pending") { + const expired = + delivery.state === "claimed" && + delivery.claim !== null && + delivery.claim.expiresAt <= this.now(); + return { + value: { + claimed: false, + reason: expired + ? "expired-requires-reconciliation" + : delivery.state === "claimed" + ? "already-claimed" + : "terminal", + binding, + }, + }; + } + const now = this.now(); + delivery.state = "claimed"; + delivery.attempt += 1; + delivery.claim = this.claim(now, request.ownerId) as MutableClaim; + delivery.eventWatermark = request.eventWatermark; + binding.updatedAt = now; + aggregate.recordVersion += 1; + aggregate.updatedAt = now; + this.emit({ + name: "subsession.kickoff_claimed", + projectId: identity.projectId, + }); + return { + value: { claimed: true, binding }, + next: aggregate, + }; + }); + } + + recordKickoffWrite( + identity: ProjectAgentSession, + bindingId: SubsessionBindingId, + request: Readonly<{ + contextEpoch: number; + deliveryId: string; + inputId: string; + claimId: string; + phase: "not-written" | "text-staged" | "enter-written"; + }>, + ): Promise { + return this.transact(identity.projectId, async (aggregate) => { + const binding = this.scopedBinding(aggregate, identity, bindingId); + const delivery = binding.deliveries.find( + (entry) => entry.contextEpoch === request.contextEpoch, + ); + if ( + !delivery || + delivery.deliveryId !== request.deliveryId || + delivery.inputId !== request.inputId || + delivery.state !== "claimed" || + delivery.claim?.claimId !== request.claimId + ) { + throw new SubsessionCoordinatorStoreError("claim_conflict"); + } + const now = this.now(); + delivery.claim = null; + if (request.phase === "not-written") { + delivery.state = "pending"; + delivery.eventWatermark = null; + } else if (request.phase === "enter-written") { + delivery.state = "submitted-unacknowledged"; + delivery.submittedAt = now; + } else { + delivery.state = "uncertain"; + } + binding.updatedAt = now; + aggregate.recordVersion += 1; + aggregate.updatedAt = now; + if (request.phase === "text-staged") + this.emit({ + name: "subsession.kickoff_uncertain", + projectId: identity.projectId, + }); + return { value: binding, next: aggregate }; + }); + } + + markKickoffUncertain( + identity: ProjectAgentSession, + bindingId: SubsessionBindingId, + request: Readonly<{ + contextEpoch: number; + deliveryId: string; + inputId: string; + }>, + ): Promise { + return this.transact(identity.projectId, async (aggregate) => { + const binding = this.scopedBinding(aggregate, identity, bindingId); + const delivery = binding.deliveries.find( + (entry) => entry.contextEpoch === request.contextEpoch, + ); + if ( + !delivery || + delivery.deliveryId !== request.deliveryId || + delivery.inputId !== request.inputId || + !["claimed", "submitted-unacknowledged", "uncertain"].includes( + delivery.state, + ) + ) { + throw new SubsessionCoordinatorStoreError("claim_conflict"); + } + if (delivery.state === "uncertain") return { value: binding }; + const now = this.now(); + delivery.state = "uncertain"; + delivery.claim = null; + binding.updatedAt = now; + aggregate.recordVersion += 1; + aggregate.updatedAt = now; + this.emit({ + name: "subsession.kickoff_uncertain", + projectId: identity.projectId, + }); + return { value: binding, next: aggregate }; + }); + } + + acknowledgeKickoff( + identity: ProjectAgentSession, + bindingId: SubsessionBindingId, + request: Readonly<{ + contextEpoch: number; + deliveryId: string; + inputId: string; + eventWatermark: string; + }>, + ): Promise { + return this.transact(identity.projectId, async (aggregate) => { + const binding = this.scopedBinding(aggregate, identity, bindingId); + const delivery = binding.deliveries.find( + (entry) => entry.contextEpoch === request.contextEpoch, + ); + if ( + !delivery || + delivery.deliveryId !== request.deliveryId || + delivery.inputId !== request.inputId || + delivery.eventWatermark !== request.eventWatermark || + ![ + "claimed", + "submitted-unacknowledged", + "uncertain", + "acknowledged", + ].includes(delivery.state) + ) { + throw new SubsessionCoordinatorStoreError("claim_conflict"); + } + if (delivery.state === "acknowledged") return { value: binding }; + const now = this.now(); + delivery.state = "acknowledged"; + delivery.claim = null; + delivery.acknowledgedAt = now; + binding.updatedAt = now; + aggregate.recordVersion += 1; + aggregate.updatedAt = now; + return { value: binding, next: aggregate }; + }); + } +} diff --git a/packages/harness/src/index.ts b/packages/harness/src/index.ts index 97be98ac..f9ea50bc 100644 --- a/packages/harness/src/index.ts +++ b/packages/harness/src/index.ts @@ -235,6 +235,47 @@ export type { FocusedSessionContextResult, } from "./core/focused-session-context.js"; +export { + PROJECT_SUBSESSION_CLAIM_TTL_MS, + PROJECT_SUBSESSION_DELEGATION_LIMIT, + PROJECT_SUBSESSION_KEY_BYTES, + PROJECT_SUBSESSION_KICKOFF_CONTEXT_BYTES, + PROJECT_SUBSESSION_OUTCOME_BYTES, + PROJECT_SUBSESSION_REQUEST_BYTES, + PROJECT_SUBSESSION_SCHEMA_VERSION, + SUBSESSION_COORDINATOR_STORAGE_SCHEMA_VERSION, +} from "./shared/subsession-delegation.js"; +export type { + CanonicalDelegationBindingDigest, + CanonicalDelegationRequestDigest, + DelegatedContextState, + DelegatedKickoffState, + DelegatedSessionState, + DelegationError, + DelegationErrorCode, + DelegationFocusRef, + DelegationItemOutcome, + DelegationItemResult, + DelegationRecovery, + ProjectSubsessionDelegation, + ProjectSubsessionRequest, + ProjectSubsessionResult, + SubsessionBindingId, + SubsessionBindingRecord, + SubsessionClaim, + SubsessionContextDigest, + SubsessionKickoffDelivery, + SubsessionProjectionDigest, + SubsessionRuntimeBinding, +} from "./shared/subsession-delegation.js"; +export { + computeCanonicalDelegationBindingDigest, + computeCanonicalDelegationRequestDigest, + computeSubsessionContextDigest, + parseProjectSubsessionRequest, + SubsessionDelegationValidationError, +} from "./shared/subsession-delegation-codec.js"; + export { PROJECT_AGENT_PROMPT_APPENDIX, projectAgentPromptAppendix, diff --git a/packages/harness/src/shared/subsession-delegation-codec.test.ts b/packages/harness/src/shared/subsession-delegation-codec.test.ts new file mode 100644 index 00000000..e8a26d25 --- /dev/null +++ b/packages/harness/src/shared/subsession-delegation-codec.test.ts @@ -0,0 +1,272 @@ +import { describe, expect, it } from "vitest"; + +import { + computeCanonicalDelegationBindingDigest, + computeCanonicalDelegationRequestDigest, + parseProjectSubsessionRequest, + SubsessionDelegationValidationError, +} from "./subsession-delegation-codec.js"; + +const projectId = "project_00000000-0000-4000-8000-000000000001"; +const map = { + projectId, + versionId: "mapv_018f0000-0000-7000-8000-000000000001", + contentDigest: `sha256:${"1".repeat(64)}`, +}; +const plan = { + projectId, + planId: "plan_018f0000-0000-7000-8000-000000000002", + versionId: "planv_018f0000-0000-7000-8000-000000000003", + semanticDigest: `sha256:${"2".repeat(64)}`, +}; + +const request = (delegations: unknown[]) => ({ + schemaVersion: 1, + requestKey: "request-1", + operation: { kind: "delegate", delegations }, +}); + +describe("subsession delegation codec", () => { + it("normalizes text and canonicalizes batch order before hashing", () => { + const left = parseProjectSubsessionRequest( + request([ + { + delegationKey: "publisher", + outcome: "Publish e\u0301vidence\r\nwithout changing scope", + focus: { + kind: "assignment", + map, + plan, + assignmentId: "work_018f0000-0000-7000-8000-000000000004", + }, + }, + { delegationKey: "research", outcome: "Collect evidence" }, + ]), + projectId, + ); + const right = parseProjectSubsessionRequest( + request([ + { delegationKey: "research", outcome: "Collect evidence" }, + { + delegationKey: "publisher", + outcome: "Publish évidence\nwithout changing scope", + focus: { + kind: "assignment", + map, + plan, + assignmentId: "work_018f0000-0000-7000-8000-000000000004", + }, + }, + ]), + projectId, + ); + + expect(left).toEqual(right); + expect(computeCanonicalDelegationRequestDigest(left)).toBe( + computeCanonicalDelegationRequestDigest(right), + ); + expect(left.operation.kind).toBe("delegate"); + if (left.operation.kind === "delegate") { + expect(left.operation.delegations.map((entry) => entry.delegationKey)).toEqual([ + "publisher", + "research", + ]); + } + }); + + it("canonicalizes release keys by code point independently of locale collation", () => { + const release = (delegationKeys: string[]) => parseProjectSubsessionRequest({ + schemaVersion: 1, requestKey: "release-order", operation: { kind: "release", delegationKeys }, + }, projectId); + const parsed = release(["ab", "a-c"]); + expect(parsed.operation).toEqual({ kind: "release", delegationKeys: ["a-c", "ab"] }); + expect(computeCanonicalDelegationRequestDigest(parsed)).toBe( + computeCanonicalDelegationRequestDigest(release(["a-c", "ab"])), + ); + }); + + it("separates request identity from immutable binding content", () => { + const first = parseProjectSubsessionRequest( + request([{ delegationKey: "research", outcome: "Collect evidence" }]), + projectId, + ); + const second = parseProjectSubsessionRequest( + { + ...request([ + { delegationKey: "research", outcome: "Collect evidence" }, + ]), + requestKey: "request-2", + }, + projectId, + ); + expect(computeCanonicalDelegationRequestDigest(first)).not.toBe( + computeCanonicalDelegationRequestDigest(second), + ); + if (first.operation.kind !== "delegate" || second.operation.kind !== "delegate") + throw new Error("unexpected operation"); + expect( + computeCanonicalDelegationBindingDigest(first.operation.delegations[0]!), + ).toBe( + computeCanonicalDelegationBindingDigest(second.operation.delegations[0]!), + ); + }); + + it.each([ + ["empty batch", request([]), "capacity_exceeded"], + [ + "duplicate keys", + request([ + { delegationKey: "same", outcome: "First" }, + { delegationKey: "same", outcome: "Second" }, + ]), + "invalid_request", + ], + [ + "separator in key", + request([{ delegationKey: "parent/child", outcome: "Do work" }]), + "invalid_request", + ], + [ + "oversized outcome", + request([{ delegationKey: "large", outcome: "x".repeat(4_097) }]), + "invalid_request", + ], + [ + "unsupported schema", + { ...request([{ delegationKey: "one", outcome: "Do work" }]), schemaVersion: 2 }, + "unsupported_schema", + ], + ])("rejects %s before side effects", (_name, input, code) => { + expect(() => parseProjectSubsessionRequest(input, projectId)).toThrowError( + expect.objectContaining({ code }), + ); + }); + + it("rejects exact focus from another project", () => { + let error: unknown; + try { + parseProjectSubsessionRequest( + request([ + { + delegationKey: "foreign", + outcome: "Do work", + focus: { + kind: "map-node", + map: { ...map, projectId: "project_foreign" }, + plan: null, + nodeId: "node_018f0000-0000-7000-8000-000000000005", + }, + }, + ]), + projectId, + ); + } catch (failure) { + error = failure; + } + expect(error).toBeInstanceOf(SubsessionDelegationValidationError); + expect(error).toMatchObject({ + code: "invalid_request", + issues: [{ code: "invalid_or_cross_project_focus" }], + }); + }); + + it("accepts an exact self refresh without granting arbitrary session selection", () => { + expect( + parseProjectSubsessionRequest( + { + schemaVersion: 1, + requestKey: "refresh-1", + operation: { + kind: "refresh-focused-context", + target: { kind: "self" }, + expectedContextEpoch: 2, + expectedContextDigest: `sha256:${"3".repeat(64)}`, + focus: null, + }, + }, + projectId, + ), + ).toMatchObject({ + operation: { + target: { kind: "self" }, + expectedContextEpoch: 2, + }, + }); + }); + + it("canonicalizes a bounded release without accepting session ids", () => { + const release = parseProjectSubsessionRequest( + { + schemaVersion: 1, + requestKey: "release-1", + operation: { + kind: "release", + delegationKeys: ["writer", "research"], + }, + }, + projectId, + ); + + expect(release.operation).toEqual({ + kind: "release", + delegationKeys: ["research", "writer"], + }); + let duplicateError: unknown; + try { + parseProjectSubsessionRequest( + { + schemaVersion: 1, + requestKey: "release-2", + operation: { + kind: "release", + delegationKeys: ["research", "research"], + }, + }, + projectId, + ); + } catch (error) { + duplicateError = error; + } + expect(duplicateError).toMatchObject({ + code: "invalid_request", + issues: [{ code: "duplicate_delegation_key" }], + }); + }); + + it("accepts only a bounded server-selected dormant release", () => { + expect( + parseProjectSubsessionRequest( + { + schemaVersion: 1, + requestKey: "release-dormant-1", + operation: { kind: "release-dormant", limit: 16 }, + }, + projectId, + ).operation, + ).toEqual({ kind: "release-dormant", limit: 16 }); + expect(() => + parseProjectSubsessionRequest( + { + schemaVersion: 1, + requestKey: "release-dormant-2", + operation: { + kind: "release-dormant", + limit: 1, + sessionIds: ["manual-session"], + }, + }, + projectId, + ), + ).toThrowError(SubsessionDelegationValidationError); + expect(() => + parseProjectSubsessionRequest( + { + schemaVersion: 1, + requestKey: "release-dormant-3", + operation: { kind: "release-dormant", limit: 17 }, + }, + projectId, + ), + ).toThrowError(SubsessionDelegationValidationError); + }); +}); diff --git a/packages/harness/src/shared/subsession-delegation-codec.ts b/packages/harness/src/shared/subsession-delegation-codec.ts new file mode 100644 index 00000000..1cefdb00 --- /dev/null +++ b/packages/harness/src/shared/subsession-delegation-codec.ts @@ -0,0 +1,358 @@ +import { Buffer } from "node:buffer"; + +import { + AGENT_MAP_UUID_V7_PATTERN, + hasAgentMapControlCharacter, +} from "./agent-map-codec.js"; +import { canonicalDigest, canonicalJson } from "./agent-map-canonical.js"; +import { + parseAgentBriefVersionRef, + parseAgentMapVersionRef, + parseProjectBuildPlanVersionRef, +} from "./build-plan-codec.js"; +import { + PROJECT_SUBSESSION_DELEGATION_LIMIT, + PROJECT_SUBSESSION_KEY_BYTES, + PROJECT_SUBSESSION_KICKOFF_CONTEXT_BYTES, + PROJECT_SUBSESSION_OUTCOME_BYTES, + PROJECT_SUBSESSION_REQUEST_BYTES, + PROJECT_SUBSESSION_SCHEMA_VERSION, + type CanonicalDelegationBindingDigest, + type CanonicalDelegationRequestDigest, + type DelegationFocusRef, + type ProjectSubsessionDelegation, + type ProjectSubsessionRequest, + type SubsessionContextDigest, +} from "./subsession-delegation.js"; + +export interface SubsessionDelegationValidationIssue { + path: string; + code: string; +} + +export class SubsessionDelegationValidationError extends Error { + readonly code: "invalid_request" | "unsupported_schema" | "capacity_exceeded"; + readonly issues: readonly SubsessionDelegationValidationIssue[]; + + constructor( + code: "invalid_request" | "unsupported_schema" | "capacity_exceeded", + issues: readonly SubsessionDelegationValidationIssue[], + ) { + super("Project subsession request is invalid"); + this.name = "SubsessionDelegationValidationError"; + this.code = code; + this.issues = issues.slice(0, 32).map(({ path, code: issueCode }) => ({ + path: path.slice(0, 256), + code: issueCode.slice(0, 128), + })); + } +} + +const isRecord = (value: unknown): value is Record => + typeof value === "object" && value !== null && !Array.isArray(value); + +const hasExactKeys = ( + value: Record, + required: readonly string[], + optional: readonly string[] = [], +): boolean => { + const keys = Object.keys(value); + const allowed = new Set([...required, ...optional]); + return required.every((key) => keys.includes(key)) && + keys.every((key) => allowed.has(key)); +}; + +const normalizeText = (value: string): string => + value.normalize("NFC").replace(/\r\n?/gu, "\n"); + +const byteLength = (value: string): number => Buffer.byteLength(value, "utf8"); + +const isKey = (value: unknown): value is string => + typeof value === "string" && + byteLength(value) >= 1 && + byteLength(value) <= PROJECT_SUBSESSION_KEY_BYTES && + /^[A-Za-z0-9._-]+$/u.test(value); + +const isPromptText = ( + value: unknown, + maximumBytes: number, +): value is string => + typeof value === "string" && + value.trim().length > 0 && + byteLength(normalizeText(value)) <= maximumBytes && + ![...value].some((character) => { + const point = character.codePointAt(0) ?? 0; + return hasAgentMapControlCharacter(character) && + point !== 0x09 && point !== 0x0a && point !== 0x0d; + }); + +const id = (value: unknown, prefix: string): value is string => + typeof value === "string" && + new RegExp(`^${prefix}_${AGENT_MAP_UUID_V7_PATTERN}$`, "u").test(value); + +const digest = (value: unknown): value is SubsessionContextDigest => + typeof value === "string" && /^sha256:[0-9a-f]{64}$/u.test(value); + +function invalid(path: string, code: string): never { + throw new SubsessionDelegationValidationError("invalid_request", [ + { path, code }, + ]); +} + +function parseFocus( + value: unknown, + expectedProjectId: string, + path: string, +): DelegationFocusRef { + if (!isRecord(value) || typeof value.kind !== "string") + return invalid(path, "invalid_focus"); + try { + if ( + value.kind === "assignment" && + hasExactKeys(value, ["kind", "map", "plan", "assignmentId"]) && + id(value.assignmentId, "work") + ) { + return { + kind: "assignment", + map: parseAgentMapVersionRef(value.map, expectedProjectId), + plan: parseProjectBuildPlanVersionRef(value.plan, expectedProjectId), + assignmentId: value.assignmentId, + } as DelegationFocusRef; + } + if ( + value.kind === "map-node" && + hasExactKeys(value, ["kind", "map", "plan", "nodeId"]) && + id(value.nodeId, "node") + ) { + return { + kind: "map-node", + map: parseAgentMapVersionRef(value.map, expectedProjectId), + plan: + value.plan === null + ? null + : parseProjectBuildPlanVersionRef(value.plan, expectedProjectId), + nodeId: value.nodeId, + } as DelegationFocusRef; + } + if ( + value.kind === "brief" && + hasExactKeys(value, ["kind", "brief"]) + ) { + return { + kind: "brief", + brief: parseAgentBriefVersionRef(value.brief, expectedProjectId), + }; + } + } catch { + // Collapse codec detail into the bounded public issue below. + } + return invalid(path, "invalid_or_cross_project_focus"); +} + +function parseDelegation( + value: unknown, + expectedProjectId: string, + index: number, +): ProjectSubsessionDelegation { + const path = `operation.delegations[${index}]`; + if ( + !isRecord(value) || + !hasExactKeys(value, ["delegationKey", "outcome"], [ + "kickoffContext", + "focus", + ]) || + !isKey(value.delegationKey) || + !isPromptText(value.outcome, PROJECT_SUBSESSION_OUTCOME_BYTES) || + (value.kickoffContext !== undefined && + !isPromptText( + value.kickoffContext, + PROJECT_SUBSESSION_KICKOFF_CONTEXT_BYTES, + )) + ) { + return invalid(path, "invalid_delegation"); + } + return { + delegationKey: normalizeText(value.delegationKey), + outcome: normalizeText(value.outcome), + ...(value.kickoffContext === undefined + ? {} + : { kickoffContext: normalizeText(value.kickoffContext) }), + ...(value.focus === undefined + ? {} + : { focus: parseFocus(value.focus, expectedProjectId, `${path}.focus`) }), + }; +} + +export function parseProjectSubsessionRequest( + value: unknown, + expectedProjectId: string, +): ProjectSubsessionRequest { + if (!isRecord(value)) invalid("$", "expected_object"); + if (value.schemaVersion !== PROJECT_SUBSESSION_SCHEMA_VERSION) { + throw new SubsessionDelegationValidationError("unsupported_schema", [ + { path: "schemaVersion", code: "unsupported_schema" }, + ]); + } + if ( + !hasExactKeys(value, ["schemaVersion", "requestKey", "operation"]) || + !isKey(value.requestKey) || + !isRecord(value.operation) || + typeof value.operation.kind !== "string" + ) { + return invalid("$", "invalid_envelope"); + } + + let operation: ProjectSubsessionRequest["operation"]; + if ( + value.operation.kind === "delegate" && + hasExactKeys(value.operation, ["kind", "delegations"]) && + Array.isArray(value.operation.delegations) + ) { + if ( + value.operation.delegations.length < 1 || + value.operation.delegations.length > PROJECT_SUBSESSION_DELEGATION_LIMIT + ) { + throw new SubsessionDelegationValidationError("capacity_exceeded", [ + { path: "operation.delegations", code: "delegation_count" }, + ]); + } + const delegations = value.operation.delegations + .map((entry, index) => parseDelegation(entry, expectedProjectId, index)) + .sort((left, right) => + left.delegationKey < right.delegationKey + ? -1 + : left.delegationKey > right.delegationKey + ? 1 + : 0, + ); + if ( + new Set(delegations.map(({ delegationKey }) => delegationKey)).size !== + delegations.length + ) { + return invalid("operation.delegations", "duplicate_delegation_key"); + } + operation = { kind: "delegate", delegations }; + } else if ( + value.operation.kind === "refresh-focused-context" && + hasExactKeys(value.operation, [ + "kind", + "target", + "expectedContextEpoch", + "expectedContextDigest", + "focus", + ]) && + isRecord(value.operation.target) && + Number.isSafeInteger(value.operation.expectedContextEpoch) && + (value.operation.expectedContextEpoch as number) > 0 && + digest(value.operation.expectedContextDigest) + ) { + let target: Extract< + ProjectSubsessionRequest["operation"], + { kind: "refresh-focused-context" } + >["target"]; + if ( + value.operation.target.kind === "self" && + hasExactKeys(value.operation.target, ["kind"]) + ) { + target = { kind: "self" }; + } else if ( + value.operation.target.kind === "child" && + hasExactKeys(value.operation.target, ["kind", "delegationKey"]) && + isKey(value.operation.target.delegationKey) + ) { + target = { + kind: "child", + delegationKey: normalizeText(value.operation.target.delegationKey), + }; + } else { + return invalid("operation.target", "invalid_target"); + } + operation = { + kind: "refresh-focused-context", + target, + expectedContextEpoch: value.operation.expectedContextEpoch as number, + expectedContextDigest: value.operation.expectedContextDigest, + focus: + value.operation.focus === null + ? null + : parseFocus( + value.operation.focus, + expectedProjectId, + "operation.focus", + ), + }; + } else if ( + value.operation.kind === "release" && + hasExactKeys(value.operation, ["kind", "delegationKeys"]) && + Array.isArray(value.operation.delegationKeys) + ) { + if ( + value.operation.delegationKeys.length < 1 || + value.operation.delegationKeys.length > PROJECT_SUBSESSION_DELEGATION_LIMIT + ) { + throw new SubsessionDelegationValidationError("capacity_exceeded", [ + { path: "operation.delegationKeys", code: "delegation_count" }, + ]); + } + if (!value.operation.delegationKeys.every(isKey)) + return invalid("operation.delegationKeys", "invalid_delegation_key"); + const delegationKeys = value.operation.delegationKeys + .map(normalizeText) + .sort((left, right) => left < right ? -1 : left > right ? 1 : 0); + if (new Set(delegationKeys).size !== delegationKeys.length) + return invalid("operation.delegationKeys", "duplicate_delegation_key"); + operation = { kind: "release", delegationKeys }; + } else if ( + value.operation.kind === "release-dormant" && + hasExactKeys(value.operation, ["kind", "limit"]) && + Number.isSafeInteger(value.operation.limit) && + (value.operation.limit as number) >= 1 && + (value.operation.limit as number) <= PROJECT_SUBSESSION_DELEGATION_LIMIT + ) { + operation = { + kind: "release-dormant", + limit: value.operation.limit as number, + }; + } else { + return invalid("operation", "invalid_operation"); + } + + const parsed: ProjectSubsessionRequest = { + schemaVersion: PROJECT_SUBSESSION_SCHEMA_VERSION, + requestKey: normalizeText(value.requestKey), + operation, + }; + if (byteLength(canonicalJson(parsed)) > PROJECT_SUBSESSION_REQUEST_BYTES) { + throw new SubsessionDelegationValidationError("capacity_exceeded", [ + { path: "$", code: "request_bytes" }, + ]); + } + return parsed; +} + +export function computeCanonicalDelegationRequestDigest( + request: ProjectSubsessionRequest, +): CanonicalDelegationRequestDigest { + return canonicalDigest( + "sapiom.project-subsession.request.v1", + request, + ) as CanonicalDelegationRequestDigest; +} + +export function computeCanonicalDelegationBindingDigest( + delegation: ProjectSubsessionDelegation, +): CanonicalDelegationBindingDigest { + return canonicalDigest( + "sapiom.project-subsession.binding.v1", + delegation, + ) as CanonicalDelegationBindingDigest; +} + +export function computeSubsessionContextDigest( + focus: DelegationFocusRef | null, +): SubsessionContextDigest { + return canonicalDigest( + "sapiom.project-subsession.context.v1", + focus, + ) as SubsessionContextDigest; +} diff --git a/packages/harness/src/shared/subsession-delegation.ts b/packages/harness/src/shared/subsession-delegation.ts new file mode 100644 index 00000000..ae281cb4 --- /dev/null +++ b/packages/harness/src/shared/subsession-delegation.ts @@ -0,0 +1,244 @@ +import type { + AgentMapVersionRef, + PlanNodeId, + StudioProjectId, +} from "./agent-map.js"; +import type { + AgentBriefVersionRef, + PlanningAssignmentId, + ProjectBuildPlanVersionRef, +} from "./build-plan.js"; + +export const PROJECT_SUBSESSION_SCHEMA_VERSION = 1 as const; +export const SUBSESSION_COORDINATOR_STORAGE_SCHEMA_VERSION = 1 as const; + +export const PROJECT_SUBSESSION_DELEGATION_LIMIT = 16; +export const PROJECT_SUBSESSION_KEY_BYTES = 128; +export const PROJECT_SUBSESSION_OUTCOME_BYTES = 4 * 1_024; +export const PROJECT_SUBSESSION_KICKOFF_CONTEXT_BYTES = 16 * 1_024; +export const PROJECT_SUBSESSION_REQUEST_BYTES = 64 * 1_024; +export const PROJECT_SUBSESSION_CLAIM_TTL_MS = 120_000; +export const PROJECT_SUBSESSION_MAX_DEPTH = 4; +export const PROJECT_SUBSESSION_LIVE_SESSION_LIMIT = 64; + +type Brand = string & { readonly __brand: TBrand }; + +export type CanonicalDelegationRequestDigest = + Brand<"CanonicalDelegationRequestDigest">; +export type CanonicalDelegationBindingDigest = + Brand<"CanonicalDelegationBindingDigest">; +export type SubsessionBindingId = Brand<"SubsessionBindingId">; +export type SubsessionContextDigest = Brand<"SubsessionContextDigest">; +export type SubsessionProjectionDigest = Brand<"SubsessionProjectionDigest">; + +export type DelegationFocusRef = + | Readonly<{ + kind: "assignment"; + map: AgentMapVersionRef; + plan: ProjectBuildPlanVersionRef; + assignmentId: PlanningAssignmentId; + }> + | Readonly<{ + kind: "map-node"; + map: AgentMapVersionRef; + plan: ProjectBuildPlanVersionRef | null; + nodeId: PlanNodeId; + }> + | Readonly<{ + kind: "brief"; + brief: AgentBriefVersionRef; + }>; + +export type ProjectSubsessionDelegation = Readonly<{ + delegationKey: string; + outcome: string; + kickoffContext?: string; + focus?: DelegationFocusRef; +}>; + +export type ProjectSubsessionRequest = Readonly<{ + schemaVersion: typeof PROJECT_SUBSESSION_SCHEMA_VERSION; + requestKey: string; + operation: + | Readonly<{ + kind: "delegate"; + delegations: readonly ProjectSubsessionDelegation[]; + }> + | Readonly<{ + kind: "refresh-focused-context"; + target: + | Readonly<{ kind: "self" }> + | Readonly<{ kind: "child"; delegationKey: string }>; + expectedContextEpoch: number; + expectedContextDigest: SubsessionContextDigest; + focus: DelegationFocusRef | null; + }> + | Readonly<{ + kind: "release"; + delegationKeys: readonly string[]; + }> + | Readonly<{ + /** + * Explicitly releases at most `limit` dormant coordinator bindings in + * the current project. Selection is server-side; callers never provide + * session IDs. + */ + kind: "release-dormant"; + limit: number; + }>; +}>; + +export type DelegationErrorCode = + | "invalid_capability" + | "expired_capability" + | "revoked_capability" + | "capability_scope_mismatch" + | "invalid_request" + | "unsupported_schema" + | "capacity_exceeded" + | "request_key_reused" + | "request_key_expired" + | "storage_unavailable" + | "internal_error" + | "delegation_key_reused" + | "context_not_found" + | "context_stale" + | "context_refresh_conflict" + | "binding_session_mismatch" + | "session_incompatible" + | "session_unreachable" + | "session_closed" + | "adapter_unavailable" + | "adapter_identity_ambiguous" + | "session_create_failed" + | "session_restart_failed" + | "readiness_timeout" + | "kickoff_failed"; + +export type DelegationRecovery = + | "none" + | "correct" + | "retry" + | "reread" + | "refresh_context" + | "inspect_session" + | "new_request_key" + | "new_delegation_key" + | "release_dormant" + | "reduce_request"; + +export type DelegationError = Readonly<{ + code: DelegationErrorCode; + retryable: boolean; + recovery: DelegationRecovery; + issues?: readonly Readonly<{ path: string; code: string }>[]; +}>; + +export type DelegatedSessionState = + | "reserved" + | "spawn-claimed" + | "starting" + | "awaiting-ready" + | "ready" + | "exited" + | "failed" + | "closed"; + +export type DelegatedContextState = + | "none" + | "current" + | "stale" + | "refreshing"; + +export type DelegatedKickoffState = + | "pending" + | "claimed" + | "submitted-unacknowledged" + | "acknowledged" + | "uncertain"; + +export type DelegationItemOutcome = + | "created" + | "reused" + | "already-running" + | "released" + | "failed"; + +export type DelegationItemResult = Readonly<{ + delegationKey: string; + bindingId: SubsessionBindingId | null; + sessionId: string | null; + outcome: DelegationItemOutcome; + sessionState: DelegatedSessionState; + contextState: DelegatedContextState; + kickoffState: DelegatedKickoffState; + error?: DelegationError; +}>; + +export type ProjectSubsessionResult = Readonly<{ + schemaVersion: typeof PROJECT_SUBSESSION_SCHEMA_VERSION; + requestKey: string; + requestDigest: CanonicalDelegationRequestDigest; + replayed: boolean; + results: readonly DelegationItemResult[]; +}>; + +export type SubsessionClaim = Readonly<{ + claimId: string; + ownerId: string; + claimedAt: string; + expiresAt: string; +}>; + +export type SubsessionRuntimeBinding = Readonly<{ + runtimeToken: string; + incarnation: number; + spawnEpoch: number; +}>; + +export type SubsessionKickoffDelivery = Readonly<{ + contextEpoch: number; + deliveryId: string; + inputId: string; + eventWatermark: string | null; + state: DelegatedKickoffState; + attempt: number; + claim: SubsessionClaim | null; + submittedAt: string | null; + acknowledgedAt: string | null; +}>; + +/** + * Durable coordinator ownership. The private SessionManager marker added by + * the runtime slice must match project, parent, binding, session, and + * incarnation before this record authorizes any session mutation. + */ +export type SubsessionBindingRecord = Readonly<{ + bindingId: SubsessionBindingId; + projectId: StudioProjectId; + parentSessionId: string; + parentBindingId: SubsessionBindingId | null; + delegationDepth: number; + delegationKey: string; + bindingDigest: CanonicalDelegationBindingDigest; + outcome: string; + kickoffContext: string | null; + initialFocus: DelegationFocusRef | null; + sessionId: string; + harness: "claude-code" | "codex"; + projectRoot: string; + lifecycleEpoch: number; + spawnEpoch: number; + contextEpoch: number; + contextDigest: SubsessionContextDigest; + contextState: DelegatedContextState; + currentFocus: DelegationFocusRef | null; + projectionDigest: SubsessionProjectionDigest | null; + sessionState: DelegatedSessionState; + spawnClaim: SubsessionClaim | null; + runtime: SubsessionRuntimeBinding | null; + deliveries: readonly SubsessionKickoffDelivery[]; + lastError: DelegationError | null; + createdAt: string; + updatedAt: string; +}>;