diff --git a/.changeset/writable-project-subsessions.md b/.changeset/writable-project-subsessions.md new file mode 100644 index 00000000..e8793995 --- /dev/null +++ b/.changeset/writable-project-subsessions.md @@ -0,0 +1,9 @@ +--- +"@sapiom/harness": minor +--- + +Add `project_subsession_delegate` on Studio's project MCP endpoint, with delegate, focused-context refresh, release and dormant-release operations. Delegation supports 16 children per batch, four nesting levels and 64 active or explicitly re-referenced coordinator-owned sessions per project. + +Readiness waits share a 30-second batch budget and return partial retry results with durable session identities. Treat uncertain kickoff delivery as terminal until acknowledged, and use a new request key when canonical content changes. Dormant cleanup remains recoverable after an earlier request expires. + +**Breaking:** `AnalyticsEventType` includes new `subsession.*` events. Update exhaustive event consumers to handle the added values. diff --git a/packages/harness/README.md b/packages/harness/README.md index 196e786b..14df5ca5 100644 --- a/packages/harness/README.md +++ b/packages/harness/README.md @@ -158,13 +158,25 @@ renews its inactivity lease, while session exit, resume rotation, signed-in principal changes, and server shutdown revoke it. Consumers should not copy, persist, log, or reuse the capability outside the launched session. -Every trusted project session receives the same three project-wide tools: +Every trusted project session receives the same nine project-wide tools: - `agent_map_read` reads the current confirmed workspace and shared proposal. - `agent_map_validate` validates one complete operation batch without mutating shared state or allocating permanent IDs. - `agent_map_propose` atomically and idempotently applies one validated batch to the shared Proposed map. +- `build_plan_read` reads the current plan or an exact historical version. +- `build_plan_validate` previews a plan replacement without writing it. +- `build_plan_apply` commits an idempotent plan replacement. +- `build_plan_rebase` reconciles the current plan against an exact map version. +- `build_plan_brief_refresh` refreshes canonical or focused context from exact sources. +- `project_subsession_delegate` creates or reuses writable child sessions, + refreshes focused context, releases owned children, or reclaims dormant bindings. + +Delegation accepts up to 16 children per batch, four nesting levels and 64 active +or explicitly re-referenced coordinator-owned sessions per project. Readiness +waits share a 30-second batch budget; partial `readiness_timeout` results can be +retried explicitly with the same request key and durable session identities. HTTP contracts that need more than a type to use are written up under `docs/`: diff --git a/packages/harness/docs/shared-build-plan.md b/packages/harness/docs/shared-build-plan.md index e750d5c5..6f1a8342 100644 --- a/packages/harness/docs/shared-build-plan.md +++ b/packages/harness/docs/shared-build-plan.md @@ -103,3 +103,89 @@ prefix; existing `brief-planv_*` caller IDs remain valid. Automatic failure resu use the same recovery advice as explicit refresh: correct input, reread sources, use a new explicit refresh request, retry transient storage, or request manual intervention for permanent limits. + +## Writable project subsessions + +Every ordinary project session discovers `project_subsession_delegate` beside +the shared map, plan, and brief tools. The operation creates, reuses, or +releases one to sixteen ordinary writable sessions. Each child receives the same common +project-agent prompt, coding capabilities, project tools, and delegation tool, +so nested delegation follows the same path. An exact assignment, map node, or +brief may focus the child, but focus never changes its tools or authority. +Delegation is bounded to four levels and 64 active or explicitly re-referenced +coordinator-owned sessions per project. Dormant exited or failed bindings retain +their exact resume identity without holding an active slot until they are +re-referenced. A parent can idempotently release its own child bindings by +delegation key to close their real Harness sessions and recover capacity; it +cannot name arbitrary session IDs or release another parent's or a manual +session. Unknown or expired keys converge as already released without exposing +a session identity. + +Callers provide both a request key and a delegation key. Identity is scoped by +the private session capability to the trusted project and parent session. +Identical retries converge on the same durable binding and real Harness session +ID until an explicit project-wide dormant eviction; changing canonical request +or binding content under an existing key otherwise fails explicitly. All binding +IDs and session IDs for a bounded batch are reserved in one durable transaction +before the first process is spawned. +Older request receipts compact into bounded key tombstones. User-closed +bindings compact into bounded ownership tombstones once no retained receipt +references them; an explicit release finalizes immediately to the same +tombstone while its receipt retains deterministic replay. Exited and failed +bindings remain available for the coordinator's ordinary resume and recovery +paths until explicitly released. Once the durable coordinator close succeeds, +SessionManager prunes the exact private ownership marker and close tombstone; a +failed final cleanup retains that proof for the next idempotent retry. The +oldest tombstones expire as the retention window advances, so routine +delegation, release, and focused-context refreshes cannot permanently exhaust a +project. +Proven acknowledged or unsent delivery epochs are likewise pruned when a newer +focused-context delivery replaces them; ambiguous delivery evidence is retained. + +If exited or failed bindings fill durable binding history, any current project +session may explicitly invoke the bounded `release-dormant` operation. The +coordinator selects at most sixteen eligible records inside the +capability-derived project; the request accepts no session IDs and never selects +active bindings or manual sessions. Parent liveness is intentionally irrelevant: +this explicit project-wide destructive operation relinquishes dormant delegation +resume identity even when the original parent is active. It retains the ordinary +Harness conversation/session history, but compacts the coordinator binding and +ends automatic resume through that binding. The sweep remains idempotent and +restart-safe, while prior request receipts referencing an evicted binding become +bounded expiry tombstones. Retrying one of those keys returns +`request_key_expired` with `new_request_key`; a fresh request key may atomically +create one new binding/session for the same delegation key. Durable-history +capacity errors expose the explicit `release_dormant` recovery code; an +all-active live cap continues to require session inspection instead of suggesting +an inapplicable dormant cleanup. A bounded private-marker cleanup error may +accompany an already-`released` result because eviction is durable first. Exact +cleanup proof remains available until the private close completes. After the +original sweep receipt expires, a fresh bounded `release-dormant` request also +retries unfinished cleanup without changing the release outcome or emitting a +second release event. + +The coordinator shares a 30-second readiness wait budget across a whole delegation +batch and both readiness and adapter-identity phases. On exhaustion it returns +partial `readiness_timeout` / `retry` results, retaining completed children and +reserved IDs for the rest. It stops processing later items and starts no background +continuation; an explicit retry of the same request reconciles the same bindings. +The budget bounds readiness waiting; in-flight durable writes and process creation +finish before their result is reported. Trusted tests or hosts can lower the budget +with `batchWaitTimeoutMs`, but cannot increase it beyond 30 seconds. + +The coordinator waits for canonical adapter readiness and exact transcript +identity, then uses fenced spawn and delivery epochs to submit one kickoff. +Delivery states distinguish pending, claimed, submitted without acknowledgement, +acknowledged, and uncertain. An uncertain delivery is never resent blindly. +Exact focused references are checked before delivery, and stale context returns +an explicit refresh path without closing the session or changing writability. + +Coordinator recovery starts from its own two-sided private binding marker. It +does not infer ownership from cwd, title, assignment, map membership, or process +similarity, and it never adopts, renames, resumes, closes, or removes an +unrelated manual session. Tabs remain projections of ordinary live sessions, +deduplicated by the real session ID and exact server-derived project identity. + +Delegation telemetry contains only event names, project/session identifiers, +and bounded error codes. Task text, kickoff context, focused prose, source, +paths, secrets, credentials, and raw adapter output are excluded. diff --git a/packages/harness/src/core/subsession-coordinator.test.ts b/packages/harness/src/core/subsession-coordinator.test.ts new file mode 100644 index 00000000..05d6dca4 --- /dev/null +++ b/packages/harness/src/core/subsession-coordinator.test.ts @@ -0,0 +1,1269 @@ +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 { AgentMapGraph, AgentMapVersion, AgentMapVersionId, PlanNodeId } from "../shared/agent-map.js"; +import { + computeAgentMapVersionRecordDigest, + computeGraphContentDigest, +} from "../shared/agent-map-canonical.js"; +import type { + BuildPlanAssignmentIntent, + ProjectBuildPlanId, + ProjectBuildPlanVersion, + ProjectBuildPlanVersionId, +} from "../shared/build-plan.js"; +import type { + AnalyticsEvent, + HarnessAdapter, + SpawnSpec, +} from "../shared/types.js"; +import type { BuildPlanStore } from "./build-plan-store.js"; +import { + computeBuildPlanRecordDigest, + computeBuildPlanSemanticDigest, +} from "./build-plan-canonicalization.js"; +import { compileCanonicalWorkstreamBriefs } from "./agent-brief-compiler.js"; +import { createEmptyProjectPlanningAggregate } from "./agent-map-aggregate-migration.js"; +import type { EventReader } from "./collector/store.js"; +import { SubsessionBindingMismatchError } from "./errors.js"; +import { IngestCredentialRegistry } from "./ingest-credentials.js"; +import { SessionManager, type PtySpawnFn } from "./session-manager.js"; +import { + SubsessionCoordinator, + SubsessionCoordinatorError, + type SubsessionCoordinatorEvent, +} from "./subsession-coordinator.js"; +import { + SubsessionCoordinatorStore, + SubsessionCoordinatorStoreError, +} from "./subsession-coordinator-store.js"; + +const projectId = "project_00000000-0000-4000-8000-000000000001"; +const parentId = "parent-session-1"; +const identity: ProjectAgentSession = { + projectId, + userId: "user-1", + sessionId: parentId, +}; +const plannedAgentId = "node_018f0000-0000-7000-8000-000000000010" as PlanNodeId; +const assignmentId = "work_018f0000-0000-7000-8000-000000000020" as BuildPlanAssignmentIntent["id"]; + +function focusedPlanningAggregate() { + const graph: AgentMapGraph = { + nodes: [{ id: plannedAgentId, kind: "agent", name: "Research", purpose: "Rank stocks", + ownerAgentId: null, contractRefs: ["ResearchReport"] }], + relationships: [], + }; + const contentDigest = computeGraphContentDigest(graph); + const mapBase = { + schemaVersion: 1 as const, projectId, + versionId: "mapv_018f0000-0000-7000-8000-000000000001" as AgentMapVersionId, + version: 1, parentVersionId: null, changeKind: "created" as const, + restoredFromVersionId: null, graph, contentDigest, + authoredBy: { userId: "user-1", sessionId: parentId }, + createdAt: "2026-09-04T00:00:00.000Z", + origin: { kind: "request" as const, requestDigest: `sha256:${"1".repeat(64)}`, + operationIds: [], touchKeys: [] }, + }; + const map: AgentMapVersion = { ...mapBase, recordDigest: computeAgentMapVersionRecordDigest(mapBase) }; + const content = { + outcome: "Publish ranked stocks", nonGoals: [], milestones: [], sequenceGates: [], + sharedConstraints: [], repositoryIntents: [], integrationCriteria: [], acceptanceCriteria: [], + decisions: [], unresolvedDecisions: [], risks: [], + assignments: [{ id: assignmentId, plannedAgentId, briefId: null, + mission: "Rank ten stocks", scope: ["Research"], nonGoals: [], dependencies: [] }], + }; + const semanticDigest = computeBuildPlanSemanticDigest(content); + const planBase = { + schemaVersion: 1 as const, projectId, + planId: "plan_018f0000-0000-7000-8000-000000000001" as ProjectBuildPlanId, + versionId: "planv_018f0000-0000-7000-8000-000000000001" as ProjectBuildPlanVersionId, + version: 1, parentVersionId: null, changeKind: "created" as const, + restoredFromVersionId: null, + map: { projectId, versionId: map.versionId, contentDigest: map.contentDigest }, + content, semanticDigest, authoredBy: { userId: "user-1", sessionId: parentId }, + createdAt: "2026-09-04T00:00:01.000Z", + origin: { kind: "request" as const, requestDigest: `sha256:${"2".repeat(64)}`, + operationIds: [], touchKeys: [] }, + }; + const plan: ProjectBuildPlanVersion = { + ...planBase, + recordDigest: computeBuildPlanRecordDigest(planBase), + }; + const brief = compileCanonicalWorkstreamBriefs({ + projectId, map, plan, mapHistory: [map], planHistory: [plan], previousBriefs: [], + }).briefs[0]!.brief; + const aggregate = createEmptyProjectPlanningAggregate(projectId, "2026-09-04T00:00:00.000Z"); + aggregate.mapVersions.push(map); + aggregate.buildPlanVersions.push(plan); + aggregate.briefVersionsById[brief.briefId] = [brief]; + aggregate.current.map = { + projectId, versionId: map.versionId, contentDigest: map.contentDigest, + }; + aggregate.current.buildPlan = { + projectId, planId: plan.planId, versionId: plan.versionId, + semanticDigest: plan.semanticDigest, + }; + aggregate.current.briefsByScope[brief.scopeKey] = { + scopeKey: brief.scopeKey, focusScope: brief.focusScope, briefId: brief.briefId, + status: "active", + version: { projectId, briefId: brief.briefId, versionId: brief.versionId, + semanticDigest: brief.semanticDigest }, + }; + return { aggregate, brief }; +} + +function adapter( + resumable = false, + eventSource: HarnessAdapter["eventSource"] = "hooks", +): HarnessAdapter { + const spec = (cwd: string): SpawnSpec => ({ + command: "fake-claude", + args: [], + env: {}, + cwd, + }); + return { + id: "claude-code", + eventSource, + launch: ({ cwd }) => spec(cwd), + resume: (_id, { cwd }) => spec(cwd), + doctor: async () => [], + listPastSessions: async () => [], + canResume: async () => resumable, + }; +} + +function fakePty() { + const data: Array<(chunk: string) => void> = []; + const exits: Array<(event: { exitCode: number }) => void> = []; + const writes: string[] = []; + return { + pty: { + write: vi.fn((value: string) => writes.push(String(value))), + resize: vi.fn(), + kill: vi.fn(), + onData: (listener: (chunk: string) => void) => { + data.push(listener); + return { dispose: () => {} }; + }, + onExit: (listener: (event: { exitCode: number }) => void) => { + exits.push(listener); + return { dispose: () => {} }; + }, + } as unknown as ReturnType, + writes, + emitExit: (exitCode = 0) => + exits.forEach((listener) => listener({ exitCode })), + }; +} + +describe("SubsessionCoordinator", () => { + const roots: string[] = []; + const managers: SessionManager[] = []; + + afterEach(async () => { + vi.useRealTimers(); + await Promise.all(managers.splice(0).map((manager) => manager.flush())); + await Promise.all( + roots.splice(0).map((root) => + fs.rm(root, { recursive: true, force: true }), + ), + ); + }); + + async function fixture( + resumable = false, + childIdentityState?: "ready" | "ambiguous", + managerOptions: Pick< + ConstructorParameters[0], + "writeSubsessionBindingRegistry" + > = {}, + storeOptions: NonNullable< + ConstructorParameters[1] + > = {}, + ) { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "subsession-service-")); + roots.push(root); + const spawned: ReturnType[] = []; + const launchContexts: Array[0]["buildLaunchOpts"]>>[2]> = []; + const spawnPty = vi.fn(() => { + const spawnedPty = fakePty(); + spawned.push(spawnedPty); + return spawnedPty.pty; + }); + const closeStore: { current?: SubsessionCoordinatorStore } = {}; + const manager = new SessionManager({ + adapters: { + "claude-code": adapter( + resumable, + childIdentityState ? "transcript-tail" : "hooks", + ), + }, + ingestUrl: "http://127.0.0.1:4100/ingest", + ingestCredentials: new IngestCredentialRegistry(), + sessionsPath: path.join(root, "sessions.json"), + spawnPty, + buildLaunchOpts: (_sessionId, _request, context) => { + launchContexts.push(context); + return {}; + }, + onSubsessionUserClosed: async (marker) => { + await closeStore.current?.closeOwnedBinding(marker); + }, + resolveAgentMapIdentity: async (_sessionId, _cwd, persisted) => persisted, + ...managerOptions, + }); + managers.push(manager); + await manager.init(); + await manager.create( + { cwd: root, harness: "claude-code" }, + { + agentMapIdentity: (sessionId) => ({ ...identity, sessionId }), + }, + ); + const parent = manager.list()[0]!; + expect(parent.id).not.toBe(parentId); + // The caller capability identity is server-derived from its real Harness + // session ID, so use that exact value for the fixture. + const caller = { ...identity, sessionId: parent.id }; + manager.setReady(parent.id, manager.getRuntimeEpoch(parent.id)!); + const unsubscribe = manager.onStatusChange((session, context) => { + if ( + session.id !== parent.id && + session.status === "running" && + !session.ready && + context.runtimeEpoch + ) { + manager.setReady(session.id, context.runtimeEpoch); + if (childIdentityState) + manager.setAdapterIdentityState( + session.id, + context.runtimeEpoch, + childIdentityState, + ); + } + }); + const events: AnalyticsEvent[] = []; + const recordedTurnSessionIds = new Set(); + const eventReader: EventReader = { + async *read(filter) { + const ids = filter?.harnessSessionId; + const accepted = new Set( + typeof ids === "string" ? [ids] : ids ?? [], + ); + for (const event of events) { + if (accepted.size > 0 && !accepted.has(event.harnessSessionId)) + continue; + if (filter?.types && !filter.types.includes(event.type)) continue; + yield event; + } + }, + index: async () => ({ + bySession: new Map( + [...recordedTurnSessionIds].map((sessionId) => [ + sessionId, + { + harnessSessionId: sessionId, + spans: [], + eventCount: 1, + turnCount: 1, + agentSessionIds: [], + harness: "claude-code" as const, + firstTs: null, + lastTs: null, + }, + ]), + ), + byAgentSession: new Map(), + }), + }; + const store = new SubsessionCoordinatorStore( + path.join(root, "agent-map"), + storeOptions, + ); + closeStore.current = store; + const telemetry: SubsessionCoordinatorEvent[] = []; + const planningStore = { + read: vi.fn(async () => { + throw new Error("no focused context expected"); + }), + } as unknown as BuildPlanStore; + const newCoordinator = (ownerId?: string, waitOptions: { + readinessTimeoutMs?: number; batchWaitTimeoutMs?: number; + } = {}) => new SubsessionCoordinator({ + store, sessionManager: manager, planningStore, eventReader, + readinessTimeoutMs: 500, + ...waitOptions, + onEvent: (event) => { + telemetry.push(event); + }, + ...(ownerId ? { ownerId } : {}), + }); + const coordinator = newCoordinator(); + return { + root, + manager, + caller, + store, + coordinator, + newCoordinator, + planningStore, + events, + recordedTurnSessionIds, + spawnPty, + spawned, + launchContexts, + telemetry, + unsubscribe, + }; + } + + const request = { + schemaVersion: 1, + requestKey: "request-1", + operation: { + kind: "delegate", + delegations: [ + { + delegationKey: "research", + outcome: "Implement the research slice", + kickoffContext: "Run the focused tests.", + }, + ], + }, + } as const; + const releaseRequest = { + schemaVersion: 1, + requestKey: "release-1", + operation: { + kind: "release", + delegationKeys: ["research"], + }, + } as const; + const dormantReleaseRequest = { + schemaVersion: 1, + requestKey: "release-dormant-1", + operation: { kind: "release-dormant", limit: 1 }, + } as const; + + it("creates one ordinary writable child and reuses it on retry", async () => { + const { coordinator, caller, manager, spawnPty, telemetry, unsubscribe } = + await fixture(); + const first = await coordinator.execute(caller, request); + const replay = await coordinator.execute(caller, request); + unsubscribe(); + + expect(first.results[0]).toMatchObject({ + outcome: "created", + sessionState: "ready", + contextState: "none", + kickoffState: "submitted-unacknowledged", + }); + expect(replay.results[0]).toMatchObject({ + outcome: "reused", + sessionId: first.results[0]!.sessionId, + }); + expect(manager.list()).toHaveLength(2); + expect(spawnPty).toHaveBeenCalledTimes(2); + const child = manager.get(first.results[0]!.sessionId!); + expect(child?.agentMapIdentity).toEqual({ + projectId, + userId: caller.userId, + sessionId: first.results[0]!.sessionId, + }); + expect(telemetry).toEqual(expect.arrayContaining([ + expect.objectContaining({ name: "subsession.requested", projectId }), + expect.objectContaining({ name: "subsession.created", projectId }), + expect.objectContaining({ name: "subsession.ready", projectId }), + expect.objectContaining({ name: "subsession.kickoff_submitted", projectId }), + ])); + expect(JSON.stringify(telemetry)).not.toContain("Implement the research slice"); + expect(JSON.stringify(telemetry)).not.toContain("Run the focused tests"); + }); + + it("idempotently releases and closes the exact real child session", async () => { + const { coordinator, caller, manager, store, spawned, telemetry, unsubscribe } = + await fixture(); + const created = await coordinator.execute(caller, request); + const childId = created.results[0]!.sessionId!; + + const releasing = coordinator.execute(caller, releaseRequest); + await vi.waitFor(() => expect(spawned[1]!.pty.kill).toHaveBeenCalledTimes(1)); + spawned[1]!.emitExit(0); + const released = await releasing; + const replay = await coordinator.execute(caller, releaseRequest); + const aggregate = await store.read(projectId); + unsubscribe(); + + expect(released).toMatchObject({ + replayed: false, + results: [{ + delegationKey: "research", + sessionId: childId, + outcome: "released", + sessionState: "closed", + }], + }); + expect(replay).toMatchObject({ + replayed: true, + results: [{ sessionId: childId, outcome: "released" }], + }); + expect(manager.get(childId)).toMatchObject({ status: "exited" }); + expect(manager.getSubsessionBinding(childId)).toBeNull(); + expect(aggregate.bindingTombstones[0]).toMatchObject({ + sessionId: childId, + }); + expect(telemetry).toContainEqual( + expect.objectContaining({ + name: "subsession.released", + projectId, + sessionId: childId, + }), + ); + }); + + it("finishes private binding cleanup when a release is retried after a partial failure", async () => { + let writeCount = 0; + let failCleanup = true; + const writeSubsessionBindingRegistry = vi.fn( + async (file: string, serialized: string) => { + writeCount += 1; + if (failCleanup && writeCount === 3) + throw new Error("injected cleanup persistence failure"); + await fs.writeFile(file, serialized, "utf8"); + }, + ); + const { coordinator, caller, manager, spawned, unsubscribe } = await fixture( + false, + undefined, + { writeSubsessionBindingRegistry }, + ); + const created = await coordinator.execute(caller, request); + const childId = created.results[0]!.sessionId!; + + const releasing = coordinator.execute(caller, releaseRequest); + await vi.waitFor(() => expect(spawned[1]!.pty.kill).toHaveBeenCalledTimes(1)); + spawned[1]!.emitExit(0); + const partial = await releasing; + expect(partial.results[0]).toMatchObject({ + sessionId: childId, + outcome: "failed", + }); + expect(manager.getSubsessionBinding(childId)).not.toBeNull(); + + failCleanup = false; + const retried = await coordinator.execute(caller, releaseRequest); + unsubscribe(); + + expect(retried).toMatchObject({ + replayed: true, + results: [{ sessionId: childId, outcome: "released" }], + }); + expect(manager.getSubsessionBinding(childId)).toBeNull(); + expect(writeSubsessionBindingRegistry).toHaveBeenCalledTimes(5); + }); + + it("bounds a not-ready batch and resumes the same durable children only on an explicit retry", async () => { + const { manager, caller, store, newCoordinator, spawnPty, spawned, unsubscribe } = await fixture(); + unsubscribe(); + const batch = { ...request, operation: { kind: "delegate", delegations: ["alpha", "beta", "gamma"].map((delegationKey) => ({ + delegationKey, outcome: `Implement ${delegationKey}`, + })) } }; + const limited = newCoordinator("limited-wait", { readinessTimeoutMs: 250, batchWaitTimeoutMs: 100 }); + const first = await limited.execute(caller, batch); + expect(first.results).toHaveLength(3); + expect(first.results.map(({ error }) => error?.code)).toEqual(["readiness_timeout", "readiness_timeout", "readiness_timeout"]); + expect(first.results.map(({ sessionState }) => sessionState)).toEqual(["awaiting-ready", "reserved", "reserved"]); + expect(spawnPty).toHaveBeenCalledTimes(2); + const beforeRetry = await store.read(projectId); + expect(beforeRetry.bindings.map(({ sessionId }) => sessionId).sort()).toEqual(first.results.map(({ sessionId }) => sessionId).sort()); + expect(beforeRetry.bindings.every(({ spawnClaim }) => spawnClaim === null)).toBe(true); + const child = first.results[0]!; + if (!child.sessionId) throw new Error("missing reserved child session ID"); + manager.setReady(child.sessionId, manager.getRuntimeEpoch(child.sessionId)!); + await new Promise((resolve) => setTimeout(resolve, 150)); + expect(spawnPty).toHaveBeenCalledTimes(2); + expect(spawned[1]!.writes).toEqual([]); + + const stop = manager.onStatusChange((session, context) => { + if (session.id !== caller.sessionId && session.status === "running" && !session.ready && context.runtimeEpoch) + manager.setReady(session.id, context.runtimeEpoch); + }); + try { + const retried = await newCoordinator("explicit-retry", { batchWaitTimeoutMs: 5_000 }).execute(caller, batch); + expect(retried.replayed).toBe(true); + expect(retried.results.every(({ error }) => error === undefined)).toBe(true); + expect(retried.results.map(({ sessionId }) => sessionId)).toEqual(first.results.map(({ sessionId }) => sessionId)); + expect(spawnPty).toHaveBeenCalledTimes(4); + } finally { stop(); } + }); + + it("shares one batch wait deadline across readiness and adapter identity", async () => { + const { manager, caller, newCoordinator, spawnPty, unsubscribe } = await fixture(false, "ready"); + unsubscribe(); + const timers: ReturnType[] = []; + const stop = manager.onStatusChange((session, context) => { + if (session.id !== caller.sessionId && session.status === "running" && !session.ready && context.runtimeEpoch) + timers.push(setTimeout(() => manager.setReady(session.id, context.runtimeEpoch!), 50)); + }); + try { + const coordinator = newCoordinator("identity-wait", { readinessTimeoutMs: 250, batchWaitTimeoutMs: 100 }); + const result = await coordinator.execute(caller, request); + expect(result.results[0]).toMatchObject({ sessionState: "awaiting-ready", + error: { code: "readiness_timeout", retryable: true, recovery: "retry" } }); + expect(spawnPty).toHaveBeenCalledTimes(2); + } finally { stop(); timers.forEach(clearTimeout); } + }); + + it("releases known children in a mixed batch and treats unknown keys as already released", async () => { + const { coordinator, caller, spawned, unsubscribe } = await fixture(); + const created = await coordinator.execute(caller, request); + const childId = created.results[0]!.sessionId!; + const mixed = { + schemaVersion: 1, + requestKey: "release-mixed", + operation: { + kind: "release", + delegationKeys: ["missing", "research"], + }, + } as const; + + const releasing = coordinator.execute(caller, mixed); + await vi.waitFor(() => expect(spawned[1]!.pty.kill).toHaveBeenCalledTimes(1)); + spawned[1]!.emitExit(0); + const released = await releasing; + const replay = await coordinator.execute(caller, mixed); + unsubscribe(); + + expect(released.results).toEqual([ + expect.objectContaining({ + delegationKey: "missing", + bindingId: null, + sessionId: null, + outcome: "released", + }), + expect.objectContaining({ + delegationKey: "research", + sessionId: childId, + outcome: "released", + }), + ]); + expect(replay).toMatchObject({ + replayed: true, + results: [ + { delegationKey: "missing", outcome: "released" }, + { delegationKey: "research", sessionId: childId, outcome: "released" }, + ], + }); + }); + + it("expires the old request and lets an active parent recreate a sibling-evicted dormant child", async () => { + const { + coordinator, + caller, + manager, + store, + spawned, + spawnPty, + telemetry, + unsubscribe, + } = + await fixture(false, undefined, {}, { bindingLimit: 1 }); + const created = await coordinator.execute(caller, request); + const childId = created.results[0]!.sessionId!; + spawned[1]!.emitExit(0); + await manager.flush(); + const binding = (await store.read(projectId)).bindings[0]!; + await store.transitionSession(caller, binding.bindingId, { + expectedLifecycleEpoch: binding.lifecycleEpoch, + expectedSpawnEpoch: binding.spawnEpoch, + expectedRuntimeToken: binding.runtime?.runtimeToken ?? null, + state: "exited", + }); + + const manual = await manager.create({ + cwd: "/tmp/manual-dormant-session", + harness: "claude-code", + }); + spawned[2]!.emitExit(0); + await manager.flush(); + const nextParent = await manager.create( + { cwd: manager.get(caller.sessionId)!.cwd, harness: "claude-code" }, + { + agentMapIdentity: (sessionId) => ({ + projectId, + userId: caller.userId, + sessionId, + }), + }, + ); + manager.setReady(nextParent.id, manager.getRuntimeEpoch(nextParent.id)!); + const nextCaller = nextParent.agentMapIdentity!; + await expect( + coordinator.execute(nextCaller, { + ...request, + requestKey: "blocked-before-dormant-release", + operation: { + ...request.operation, + delegations: [{ + delegationKey: "writer", + outcome: "Write evidence", + }], + }, + }), + ).rejects.toMatchObject({ + detail: { + code: "capacity_exceeded", + retryable: false, + recovery: "release_dormant", + }, + }); + + const released = await coordinator.execute( + nextCaller, + dormantReleaseRequest, + ); + const replay = await coordinator.execute(nextCaller, dormantReleaseRequest); + expect(released.results).toEqual([ + expect.objectContaining({ + delegationKey: "research", + sessionId: childId, + outcome: "released", + }), + ]); + expect(replay).toMatchObject({ + replayed: true, + results: [{ sessionId: childId, outcome: "released" }], + }); + expect(manager.get(childId)).toMatchObject({ status: "exited" }); + expect(manager.getSubsessionBinding(childId)).toBeNull(); + expect(manager.get(caller.sessionId)?.status).not.toBe("exited"); + expect(manager.get(manual.id)).toMatchObject({ status: "exited" }); + expect(manager.getSubsessionBinding(manual.id)).toBeNull(); + expect(telemetry).toContainEqual( + expect.objectContaining({ + name: "subsession.released", + projectId, + sessionId: childId, + }), + ); + await expect( + coordinator.execute( + { ...nextCaller, projectId: "project_foreign" }, + { ...dormantReleaseRequest, requestKey: "foreign-sweep" }, + ), + ).rejects.toMatchObject({ + detail: { code: "capability_scope_mismatch" }, + }); + + await expect(coordinator.execute(caller, request)).rejects.toMatchObject({ + detail: { + code: "request_key_expired", + retryable: false, + recovery: "new_request_key", + }, + }); + const recreatedRequest = { + ...request, + requestKey: "after-dormant-release", + } as const; + const next = await coordinator.execute(caller, recreatedRequest); + const nextReplay = await coordinator.execute(caller, recreatedRequest); + const activeSweep = await coordinator.execute(nextCaller, { + ...dormantReleaseRequest, + requestKey: "active-and-manual-exclusion", + }); + unsubscribe(); + + expect(next.results[0]).toMatchObject({ + delegationKey: "research", + outcome: "created", + }); + expect(next.results[0]!.sessionId).not.toBe(childId); + expect(nextReplay).toMatchObject({ + replayed: true, + results: [{ + delegationKey: "research", + sessionId: next.results[0]!.sessionId, + outcome: "reused", + }], + }); + expect(activeSweep.results).toEqual([]); + expect(spawnPty).toHaveBeenCalledTimes(5); + expect(manager.get(caller.sessionId)?.status).not.toBe("exited"); + expect(manager.get(next.results[0]!.sessionId!)?.status).not.toBe("exited"); + expect(manager.get(manual.id)).toMatchObject({ status: "exited" }); + expect((await store.read(projectId)).bindings).toHaveLength(1); + }); + + it("reports a committed dormant eviction truthfully when private cleanup must retry", async () => { + const { coordinator, caller, manager, store, spawned, telemetry, unsubscribe } = + await fixture(); + const created = await coordinator.execute(caller, request); + const childId = created.results[0]!.sessionId!; + spawned[1]!.emitExit(0); + await manager.flush(); + const binding = (await store.read(projectId)).bindings[0]!; + await store.transitionSession(caller, binding.bindingId, { + expectedLifecycleEpoch: binding.lifecycleEpoch, + expectedSpawnEpoch: binding.spawnEpoch, + expectedRuntimeToken: binding.runtime?.runtimeToken ?? null, + state: "exited", + }); + const closeBound = vi + .spyOn(manager, "closeBound") + .mockRejectedValueOnce(new SubsessionBindingMismatchError()); + + const released = await coordinator.execute(caller, dormantReleaseRequest); + expect(released.results[0]).toMatchObject({ + delegationKey: "research", + sessionId: childId, + outcome: "released", + sessionState: "closed", + error: { + code: "binding_session_mismatch", + retryable: false, + recovery: "inspect_session", + }, + }); + expect((await store.read(projectId)).bindingTombstones[0]).toMatchObject({ + sessionId: childId, + disposition: "dormant-evicted", + }); + expect(manager.getSubsessionBinding(childId)).not.toBeNull(); + + closeBound.mockRestore(); + const replay = await coordinator.execute(caller, dormantReleaseRequest); + unsubscribe(); + + expect(replay).toMatchObject({ + replayed: true, + results: [{ sessionId: childId, outcome: "released" }], + }); + expect(manager.getSubsessionBinding(childId)).toBeNull(); + expect( + telemetry.filter( + (event) => + event.name === "subsession.released" && event.sessionId === childId, + ), + ).toHaveLength(1); + }); + + it("recovers unfinished dormant cleanup under a new key after its receipt expires", async () => { + const { coordinator, newCoordinator, caller, manager, store, spawned, telemetry, unsubscribe } = + await fixture(false, undefined, {}, { + receiptRetentionLimit: 1, + historyTombstoneLimit: 1, + }); + const created = await coordinator.execute(caller, request); + const childId = created.results[0]!.sessionId!; + const binding = (await store.read(projectId)).bindings[0]!; + // A failed coordinator binding can still have a live process whose exact + // private close must finish after the durable eviction has committed. + await store.transitionSession(caller, binding.bindingId, { + expectedLifecycleEpoch: binding.lifecycleEpoch, + expectedSpawnEpoch: binding.spawnEpoch, + expectedRuntimeToken: binding.runtime?.runtimeToken ?? null, + state: "failed", + }); + vi.spyOn(manager, "closeBound") + .mockRejectedValueOnce(new SubsessionBindingMismatchError()); + const partial = await coordinator.execute(caller, dormantReleaseRequest); + expect(partial.results[0]).toMatchObject({ + sessionId: childId, + outcome: "released", + error: { code: "binding_session_mismatch" }, + }); + + await store.reserveDormantReleases(caller, { + ...dormantReleaseRequest, + requestKey: "advance-cleanup-receipts", + }, []); + await expect(coordinator.execute(caller, dormantReleaseRequest)) + .rejects.toMatchObject({ detail: { code: "request_key_expired" } }); + expect(manager.get(childId)?.status).toBe("running"); + expect(manager.getSubsessionBinding(childId)).not.toBeNull(); + + const recovering = newCoordinator("restarted-cleanup-owner").execute(caller, { + ...dormantReleaseRequest, + requestKey: "retry-unfinished-cleanup", + }); + await vi.waitFor(() => expect(spawned[1]!.pty.kill).toHaveBeenCalledTimes(1)); + spawned[1]!.emitExit(0); + const recovered = await recovering; + unsubscribe(); + + expect(recovered).toMatchObject({ + replayed: false, + results: [{ sessionId: childId, outcome: "released" }], + }); + expect(recovered.results[0]?.error).toBeUndefined(); + expect(manager.getSubsessionBinding(childId)).toBeNull(); + expect(telemetry.filter((event) => + event.name === "subsession.released" && event.sessionId === childId, + )).toHaveLength(1); + }); + + it.each(["exited", "failed"] as const)( + "releases an already-%s child without spawning or resuming it", + async (terminalState) => { + const { + coordinator, + caller, + manager, + store, + spawned, + spawnPty, + unsubscribe, + } = await fixture(); + const created = await coordinator.execute(caller, request); + const childId = created.results[0]!.sessionId!; + spawned[1]!.emitExit(0); + await manager.flush(); + const binding = (await store.read(projectId)).bindings[0]!; + await store.transitionSession(caller, binding.bindingId, { + expectedLifecycleEpoch: binding.lifecycleEpoch, + expectedSpawnEpoch: binding.spawnEpoch, + expectedRuntimeToken: binding.runtime?.runtimeToken ?? null, + state: terminalState, + }); + + const released = await coordinator.execute(caller, releaseRequest); + unsubscribe(); + + expect(released.results[0]).toMatchObject({ + sessionId: childId, + outcome: "released", + sessionState: "closed", + }); + expect(spawnPty).toHaveBeenCalledTimes(2); + }, + ); + + it("fails closed instead of releasing or killing a manual session", async () => { + const { coordinator, caller, manager, store, spawned, unsubscribe } = + await fixture(); + const manual = await manager.create({ + cwd: "/tmp/manual-project-session", + harness: "claude-code", + }); + const reserved = await store.reserveDelegations(caller, request, { + harness: "claude-code", + projectRoot: "/tmp/delegated-project-session", + ownerId: "coordinator-test", + }); + const release = await store.reserveReleases(caller, releaseRequest); + vi.spyOn(store, "reserveReleases").mockResolvedValueOnce({ + ...release, + bindings: [{ + state: "bound", + binding: { ...reserved.bindings[0]!, sessionId: manual.id }, + }], + }); + + const result = await coordinator.execute(caller, releaseRequest); + unsubscribe(); + + expect(result.results[0]).toMatchObject({ + outcome: "failed", + error: { + code: "binding_session_mismatch", + retryable: false, + }, + }); + expect(manager.get(manual.id)).toMatchObject({ status: "running" }); + expect(spawned[1]!.pty.kill).not.toHaveBeenCalled(); + expect((await store.read(projectId)).bindings[0]).toMatchObject({ + sessionState: "reserved", + }); + }); + + it("converges independent coordinator instances on one child process", async () => { + const { coordinator, newCoordinator, caller, manager, spawnPty, unsubscribe } = + await fixture(); + const other = newCoordinator("other-coordinator"); + const [first, second] = await Promise.all([ + coordinator.execute(caller, request), + other.execute(caller, request), + ]); + unsubscribe(); + + expect(first.results[0]!.sessionId).toBe(second.results[0]!.sessionId); + expect([first.results[0]!.outcome, second.results[0]!.outcome]).toEqual( + expect.arrayContaining(["created", "already-running"]), + ); + expect(manager.list()).toHaveLength(2); + expect(spawnPty).toHaveBeenCalledTimes(2); + }); + + it("atomically renews an expired self-owned spawn claim across coordinators", async () => { + const { + newCoordinator, + caller, + manager, + store, + spawnPty, + unsubscribe, + } = await fixture(false, undefined, {}, { claimTtlMs: 500 }); + const parent = manager.get(caller.sessionId)!; + const binding = ( + await store.reserveDelegations(caller, request, { + harness: parent.harness, + projectRoot: parent.cwd, + ownerId: "coordinator-self", + }) + ).bindings[0]!; + const original = await store.claimSpawn(caller, binding.bindingId, { + ownerId: "coordinator-self", + expectedLifecycleEpoch: binding.lifecycleEpoch, + expectedSpawnEpoch: binding.spawnEpoch, + }); + if (!original.claimed) throw new Error("spawn claim was not acquired"); + await new Promise((resolve) => setTimeout(resolve, 550)); + + const self = newCoordinator("coordinator-self"); + const other = newCoordinator("coordinator-other"); + const [first, second] = await Promise.all([ + self.execute(caller, request), + other.execute(caller, request), + ]); + const aggregate = await store.read(projectId); + unsubscribe(); + + expect(first.results[0]!.sessionId).toBe(second.results[0]!.sessionId); + expect(spawnPty).toHaveBeenCalledTimes(2); + expect(aggregate.bindings[0]).toMatchObject({ + bindingId: binding.bindingId, + spawnEpoch: 2, + sessionState: "ready", + }); + }); + + it("acknowledges only the exact persisted kickoff marker", async () => { + const { coordinator, caller, manager, store, spawned, unsubscribe } = + await fixture(); + const result = await coordinator.execute(caller, request); + const sessionId = result.results[0]!.sessionId!; + const prompt = spawned[1]!.writes + .find((value) => value.includes("sapiom-project-delegation"))!; + const event: AnalyticsEvent = { + eventId: "event-1", + seq: 1, + ts: new Date().toISOString(), + userId: caller.userId, + tenantId: null, + machineId: "machine-1", + harnessSessionId: sessionId, + agentSessionId: null, + harness: "claude-code", + type: "prompt.submitted", + payload: { prompt }, + }; + await coordinator.onEventPersisted( + event, + manager.getRuntimeEpoch(sessionId)!, + ); + const aggregate = await store.read(projectId); + unsubscribe(); + + expect(aggregate.bindings[0]!.deliveries[0]!.state).toBe("acknowledged"); + }); + + it("never fresh-restarts an exited child after kickoff delivery becomes uncertain", async () => { + const { coordinator, caller, manager, store, spawned, spawnPty, unsubscribe } = + await fixture(); + const first = await coordinator.execute(caller, request); + const childId = first.results[0]!.sessionId!; + spawned[1]!.emitExit(1); + await manager.flush(); + + const retried = await coordinator.execute(caller, request); + const aggregate = await store.read(projectId); + unsubscribe(); + + expect(retried.results[0]).toMatchObject({ + outcome: "failed", + sessionId: childId, + kickoffState: "uncertain", + error: { code: "session_unreachable", retryable: false }, + }); + expect(spawnPty).toHaveBeenCalledTimes(2); + expect(aggregate.bindings[0]!.deliveries[0]!.state).toBe("uncertain"); + }); + + it("resumes an exited coordinator-owned vendor conversation under the same Harness id", async () => { + const { coordinator, caller, manager, spawned, spawnPty, unsubscribe } = + await fixture(true); + const first = await coordinator.execute(caller, request); + const childId = first.results[0]!.sessionId!; + const runtime = manager.getRuntimeEpoch(childId)!; + await manager.setAgentSessionId(childId, "agent-child-1", "startup", runtime); + spawned[1]!.emitExit(0); + await manager.flush(); + + const retried = await coordinator.execute(caller, request); + unsubscribe(); + + expect(retried.results[0]).toMatchObject({ + outcome: "reused", + sessionId: childId, + sessionState: "ready", + kickoffState: "uncertain", + }); + expect(manager.list().filter(({ id }) => id === childId)).toHaveLength(1); + expect(spawnPty).toHaveBeenCalledTimes(3); + }); + + it("fails closed when the caller identity is not its trusted session scope", async () => { + const { coordinator, caller, manager, unsubscribe } = await fixture(); + await expect( + coordinator.execute({ ...caller, projectId: "project_00000000-0000-4000-8000-000000000099" }, request), + ).rejects.toEqual( + expect.objectContaining>({ + detail: expect.objectContaining({ code: "capability_scope_mismatch" }), + }), + ); + unsubscribe(); + expect(manager.list()).toHaveLength(1); + }); + + it("preserves bounded codec codes and issues for callers", async () => { + const { coordinator, caller, unsubscribe } = await fixture(); + await expect( + coordinator.execute(caller, { + schemaVersion: 2, + requestKey: "unsupported", + operation: { kind: "delegate", delegations: [] }, + }), + ).rejects.toMatchObject({ + detail: { + code: "unsupported_schema", + retryable: false, + recovery: "correct", + issues: [{ path: "schemaVersion", code: "unsupported_schema" }], + }, + }); + await expect( + coordinator.execute(caller, { + schemaVersion: 1, + requestKey: "utf8-overflow", + operation: { + kind: "delegate", + delegations: [ + { + delegationKey: "research", + outcome: "界".repeat(2_000), + }, + ], + }, + }), + ).rejects.toMatchObject({ + detail: { + code: "invalid_request", + retryable: false, + recovery: "correct", + issues: [ + { + path: "operation.delegations[0]", + code: "invalid_delegation", + }, + ], + }, + }); + unsubscribe(); + }); + + it("directs dormant history exhaustion to bounded dormant release", async () => { + const { coordinator, caller, store, unsubscribe } = await fixture(); + vi.spyOn(store, "reserveDelegations").mockRejectedValueOnce( + new SubsessionCoordinatorStoreError("history_quota_exceeded"), + ); + await expect(coordinator.execute(caller, request)).rejects.toMatchObject({ + detail: { + code: "capacity_exceeded", + retryable: false, + recovery: "release_dormant", + }, + }); + unsubscribe(); + }); + + it("directs genuinely live-session capacity exhaustion to session inspection", async () => { + const { coordinator, caller, store, unsubscribe } = await fixture(); + vi.spyOn(store, "reserveDelegations").mockRejectedValueOnce( + new SubsessionCoordinatorStoreError("live_session_limit_reached"), + ); + await expect(coordinator.execute(caller, request)).rejects.toMatchObject({ + detail: { + code: "capacity_exceeded", + retryable: false, + recovery: "inspect_session", + }, + }); + unsubscribe(); + }); + + it("requires a fresh request key after its bounded receipt window expires", async () => { + const { coordinator, caller, store, unsubscribe } = await fixture(); + vi.spyOn(store, "reserveDelegations").mockRejectedValueOnce( + new SubsessionCoordinatorStoreError("request_key_expired"), + ); + await expect(coordinator.execute(caller, request)).rejects.toMatchObject({ + detail: { + code: "request_key_expired", + retryable: false, + recovery: "new_request_key", + }, + }); + unsubscribe(); + }); + + it("writes no kickoff when adapter identity correlation is ambiguous", async () => { + const { coordinator, caller, spawned, unsubscribe } = await fixture( + false, + "ambiguous", + ); + const result = await coordinator.execute(caller, request); + unsubscribe(); + + expect(result.results[0]).toMatchObject({ + outcome: "failed", + sessionState: "awaiting-ready", + kickoffState: "pending", + error: { + code: "adapter_identity_ambiguous", + retryable: false, + recovery: "inspect_session", + }, + }); + expect(spawned[1]!.writes).toEqual([]); + }); + + it("reports recorded-turn fresh restart rejection as terminal", async () => { + const { + coordinator, + caller, + manager, + recordedTurnSessionIds, + spawned, + unsubscribe, + } = await fixture(false, "ambiguous"); + const first = await coordinator.execute(caller, request); + const childId = first.results[0]!.sessionId!; + recordedTurnSessionIds.add(childId); + spawned[1]!.emitExit(1); + await manager.flush(); + + const retried = await coordinator.execute(caller, request); + unsubscribe(); + + expect(retried.results[0]).toMatchObject({ + outcome: "failed", + sessionId: childId, + kickoffState: "pending", + error: { + code: "session_restart_failed", + retryable: false, + recovery: "inspect_session", + }, + }); + }); + + it("removes a user-closed child from live coordinator capacity", async () => { + const { coordinator, caller, manager, store, spawned, unsubscribe } = + await fixture(); + const created = await coordinator.execute(caller, request); + const childId = created.results[0]!.sessionId!; + + const closing = manager.close(childId); + spawned[1]!.emitExit(0); + await closing; + const aggregate = await store.read(projectId); + unsubscribe(); + + expect(aggregate.bindings[0]).toMatchObject({ + sessionId: childId, + sessionState: "closed", + runtime: null, + }); + }); + + it("delivers one exact brief overlay and surfaces later staleness without restricting the child", async () => { + const { aggregate, brief } = focusedPlanningAggregate(); + const { + coordinator, + caller, + manager, + planningStore, + spawned, + launchContexts, + unsubscribe, + } = await fixture(); + vi.mocked(planningStore.read).mockResolvedValue(aggregate); + const focusedRequest = { + schemaVersion: 1, + requestKey: "focused-request", + operation: { + kind: "delegate", + delegations: [{ + delegationKey: "focused-research", + outcome: "Implement the focused research slice", + focus: { + kind: "brief", + brief: { + projectId, + briefId: brief.briefId, + versionId: brief.versionId, + semanticDigest: brief.semanticDigest, + }, + }, + }], + }, + } as const; + + const first = await coordinator.execute(caller, focusedRequest); + const replay = await coordinator.execute(caller, focusedRequest); + const childId = first.results[0]!.sessionId!; + const kickoffWrites = spawned[1]!.writes.filter((value) => + value.includes("sapiom-project-delegation"), + ); + expect(first.results[0]).toMatchObject({ + outcome: "created", + contextState: "current", + sessionState: "ready", + }); + expect(replay.results[0]).toMatchObject({ + outcome: "reused", + sessionId: childId, + }); + expect(kickoffWrites).toHaveLength(1); + expect(launchContexts[1]?.focusedContext).toContain("focused-project-context"); + expect(launchContexts[1]?.focusedContext).toContain(brief.versionId); + expect(kickoffWrites[0]).not.toContain("focused-project-context"); + + aggregate.current.briefsByScope[brief.scopeKey] = { + ...aggregate.current.briefsByScope[brief.scopeKey]!, + status: "retired", + }; + const stale = await coordinator.execute(caller, focusedRequest); + unsubscribe(); + + expect(stale.results[0]).toMatchObject({ + outcome: "failed", + sessionId: childId, + contextState: "stale", + error: { code: "context_stale", recovery: "refresh_context" }, + }); + expect(manager.get(childId)).toMatchObject({ + status: "running", + agentMapIdentity: { projectId, sessionId: childId }, + }); + expect(spawned[1]!.writes.filter((value) => + value.includes("sapiom-project-delegation"), + )).toHaveLength(1); + }); +}); diff --git a/packages/harness/src/core/subsession-coordinator.ts b/packages/harness/src/core/subsession-coordinator.ts new file mode 100644 index 00000000..6048a9f8 --- /dev/null +++ b/packages/harness/src/core/subsession-coordinator.ts @@ -0,0 +1,1494 @@ +import { randomUUID } from "node:crypto"; + +import type { + ProjectAgentSession, + StudioProjectId, +} from "../shared/agent-map.js"; +import { canonicalDigest } from "../shared/agent-map-canonical.js"; +import { + agentMapVersionRefsEqual, + projectBuildPlanVersionRefsEqual, + type AgentBriefVersion, +} from "../shared/build-plan.js"; +import { + parseProjectSubsessionRequest, + SubsessionDelegationValidationError, +} from "../shared/subsession-delegation-codec.js"; +import type { + DelegationError, + DelegationFocusRef, + DelegationItemOutcome, + DelegationItemResult, + ProjectSubsessionRequest, + ProjectSubsessionResult, + SubsessionBindingRecord, + SubsessionProjectionDigest, +} from "../shared/subsession-delegation.js"; +import type { AnalyticsEvent, HarnessSession } from "../shared/types.js"; +import type { BuildPlanStore } from "./build-plan-store.js"; +import type { EventReader } from "./collector/store.js"; +import { + serializeFocusedSessionContext, + type FocusedSessionContextProjection, +} from "./focused-session-context.js"; +import { + SessionNotReadyError, + SubsessionBindingMismatchError, + SubsessionFreshRestartForbiddenError, +} from "./errors.js"; +import type { + SessionManager, + TrustedSubsessionBindingMarker, +} from "./session-manager.js"; +import { + SubsessionCoordinatorStore, + SubsessionCoordinatorStoreError, + type ReservedReleases, +} from "./subsession-coordinator-store.js"; + +const DEFAULT_READINESS_TIMEOUT_MS = 30_000; +const DEFAULT_BATCH_WAIT_TIMEOUT_MS = 30_000; +const ADAPTER_IDENTITY_POLL_MS = 25; +const KICKOFF_MARKER = //u; + +export interface SubsessionCoordinatorEvent { + name: + | "subsession.requested" + | "subsession.created" + | "subsession.reused" + | "subsession.released" + | "subsession.ready" + | "subsession.failed" + | "subsession.kickoff_submitted" + | "subsession.kickoff_acknowledged" + | "subsession.kickoff_uncertain" + | "subsession.context_stale" + | "subsession.manual_session_protected"; + projectId: StudioProjectId; + sessionId?: string; + code?: DelegationError["code"]; +} + +export class SubsessionCoordinatorError extends Error { + constructor(readonly detail: DelegationError) { + super(detail.code); + this.name = "SubsessionCoordinatorError"; + } +} + +export interface SubsessionCoordinatorOptions { + store: SubsessionCoordinatorStore; + sessionManager: SessionManager; + planningStore: BuildPlanStore; + eventReader: EventReader; + ownerId?: string; + readinessTimeoutMs?: number; + /** Shared readiness/adapter wait budget for one delegation batch. */ + batchWaitTimeoutMs?: number; + onEvent?: (event: SubsessionCoordinatorEvent) => void | Promise; +} + +type ResolvedFocus = Readonly<{ + state: "none" | "current" | "stale"; + projection: FocusedSessionContextProjection | null; + projectionDigest: SubsessionProjectionDigest | null; +}>; + +type DelegateRequest = Omit & + Readonly<{ + operation: Extract< + ProjectSubsessionRequest["operation"], + { kind: "delegate" } + >; + }>; +type RefreshRequest = Omit & + Readonly<{ + operation: Extract< + ProjectSubsessionRequest["operation"], + { kind: "refresh-focused-context" } + >; + }>; +type ReleaseRequest = Omit & + Readonly<{ + operation: Extract< + ProjectSubsessionRequest["operation"], + { kind: "release" } + >; + }>; +type DormantReleaseRequest = Omit & + Readonly<{ + operation: Extract< + ProjectSubsessionRequest["operation"], + { kind: "release-dormant" } + >; + }>; + +const error = ( + code: DelegationError["code"], + retryable: boolean, + recovery: DelegationError["recovery"], + issues?: DelegationError["issues"], +): DelegationError => ({ + code, + retryable, + recovery, + ...(issues === undefined ? {} : { issues }), +}); + +const currentDelivery = (binding: SubsessionBindingRecord) => + binding.deliveries.find( + ({ contextEpoch }) => contextEpoch === binding.contextEpoch, + ) ?? binding.deliveries.at(-1)!; + +const bindingIdentity = ( + caller: ProjectAgentSession, + binding: SubsessionBindingRecord, +): ProjectAgentSession => ({ + projectId: binding.projectId, + userId: caller.userId, + sessionId: binding.parentSessionId, +}); + +const markerFor = ( + binding: SubsessionBindingRecord, + incarnation: number, +): TrustedSubsessionBindingMarker => ({ + projectId: binding.projectId, + parentSessionId: binding.parentSessionId, + bindingId: binding.bindingId, + sessionId: binding.sessionId, + incarnation, + spawnEpoch: binding.spawnEpoch, +}); + +const refsMatchBrief = ( + brief: AgentBriefVersion, + focus: Extract, +) => + brief.projectId === focus.brief.projectId && + brief.briefId === focus.brief.briefId && + brief.versionId === focus.brief.versionId && + brief.semanticDigest === focus.brief.semanticDigest; + +export class SubsessionCoordinator { + private readonly ownerId: string; + private readonly readinessTimeoutMs: number; + private readonly batchWaitTimeoutMs: number; + + constructor(private readonly options: SubsessionCoordinatorOptions) { + this.ownerId = options.ownerId ?? `coordinator_${randomUUID()}`; + this.readinessTimeoutMs = + options.readinessTimeoutMs ?? DEFAULT_READINESS_TIMEOUT_MS; + this.batchWaitTimeoutMs = options.batchWaitTimeoutMs ?? DEFAULT_BATCH_WAIT_TIMEOUT_MS; + if (!Number.isSafeInteger(this.batchWaitTimeoutMs) || this.batchWaitTimeoutMs < 1 || + this.batchWaitTimeoutMs > DEFAULT_BATCH_WAIT_TIMEOUT_MS) + throw new RangeError("batchWaitTimeoutMs must be between 1 and 30000"); + } + + private emit(event: SubsessionCoordinatorEvent): void { + try { + void Promise.resolve(this.options.onEvent?.(event)).catch(() => {}); + } catch { + // Content-free telemetry cannot change delegation behavior. + } + } + + async execute( + identity: ProjectAgentSession, + rawRequest: unknown, + ): Promise { + let request: ProjectSubsessionRequest; + try { + request = parseProjectSubsessionRequest(rawRequest, identity.projectId); + } catch (cause) { + if (cause instanceof SubsessionDelegationValidationError) { + throw new SubsessionCoordinatorError( + error( + cause.code, + false, + cause.code === "capacity_exceeded" ? "reduce_request" : "correct", + cause.issues, + ), + ); + } + throw new SubsessionCoordinatorError( + error("invalid_request", false, "correct"), + ); + } + this.assertCaller(identity); + this.emit({ name: "subsession.requested", projectId: identity.projectId }); + if (request.operation.kind === "refresh-focused-context") + return this.refresh(identity, request as RefreshRequest); + if (request.operation.kind === "release") + return this.release(identity, request as ReleaseRequest); + if (request.operation.kind === "release-dormant") + return this.releaseDormant(identity, request as DormantReleaseRequest); + return this.delegate(identity, request as DelegateRequest); + } + + private assertCaller(identity: ProjectAgentSession): HarnessSession { + const caller = this.options.sessionManager.get(identity.sessionId); + if ( + !caller?.agentMapIdentity || + caller.agentMapIdentity.projectId !== identity.projectId || + caller.agentMapIdentity.userId !== identity.userId || + caller.agentMapIdentity.sessionId !== identity.sessionId + ) { + throw new SubsessionCoordinatorError( + error("capability_scope_mismatch", false, "none"), + ); + } + return caller; + } + + private async delegate( + identity: ProjectAgentSession, + request: DelegateRequest, + ): Promise { + const waitDeadline = Date.now() + this.batchWaitTimeoutMs; + const caller = this.assertCaller(identity); + let reserved; + try { + reserved = await this.options.store.reserveDelegations(identity, request, { + harness: caller.harness, + projectRoot: caller.cwd, + ownerId: this.ownerId, + }); + } catch (cause) { + throw this.wholeCallError(cause); + } + const results: DelegationItemResult[] = []; + for (const binding of reserved.bindings) { + results.push(Date.now() >= waitDeadline + ? this.failedResult(binding, error("readiness_timeout", true, "retry")) + : await this.reconcileBinding(identity, binding, reserved.replayed, waitDeadline)); + } + return { + schemaVersion: 1, + requestKey: request.requestKey, + requestDigest: reserved.requestDigest, + replayed: reserved.replayed, + results: results.sort((left, right) => + left.delegationKey.localeCompare(right.delegationKey), + ), + }; + } + + private async refresh( + identity: ProjectAgentSession, + request: RefreshRequest, + ): Promise { + let refreshed; + try { + refreshed = await this.options.store.refreshFocusedContext(identity, request); + } catch (cause) { + throw this.wholeCallError(cause, true); + } + const result = await this.reconcileBinding( + identity, + refreshed.binding, + refreshed.replayed, + ); + return { + schemaVersion: 1, + requestKey: request.requestKey, + requestDigest: refreshed.requestDigest, + replayed: refreshed.replayed, + results: [result], + }; + } + + private async release( + identity: ProjectAgentSession, + request: ReleaseRequest, + ): Promise { + let reserved; + try { + reserved = await this.options.store.reserveReleases(identity, request); + } catch (cause) { + throw this.wholeCallError(cause); + } + return this.releaseReserved(identity, request.requestKey, reserved); + } + + private async releaseDormant( + identity: ProjectAgentSession, + request: DormantReleaseRequest, + ): Promise { + let reserved; + try { + const aggregate = await this.options.store.read(identity.projectId); + const pendingCleanup = aggregate.bindingTombstones + .filter(({ disposition, cleanupComplete }) => + disposition === "dormant-evicted" && !cleanupComplete, + ); + const candidates = [ + ...pendingCleanup.map((binding) => ({ ...binding, updatedAt: binding.closedAt })), + ...aggregate.bindings.filter(({ sessionState }) => + ["exited", "failed"].includes(sessionState), + ), + ]; + const candidateBindingIds = candidates + .sort((left, right) => + left.updatedAt.localeCompare(right.updatedAt) || + left.bindingId.localeCompare(right.bindingId), + ) + .slice(0, request.operation.limit) + .map(({ bindingId }) => bindingId); + reserved = await this.options.store.reserveDormantReleases( + identity, + request, + candidateBindingIds, + ); + } catch (cause) { + throw this.wholeCallError(cause); + } + return this.releaseReserved(identity, request.requestKey, reserved); + } + + private async releaseReserved( + identity: ProjectAgentSession, + requestKey: string, + reserved: ReservedReleases, + ): Promise { + const results: DelegationItemResult[] = []; + for (const target of reserved.bindings) { + if (target.state === "absent") { + results.push({ + delegationKey: target.delegationKey, + bindingId: null, + sessionId: null, + outcome: "released", + sessionState: "closed", + contextState: "none", + kickoffState: "pending", + }); + continue; + } + if (target.state === "released" || target.state === "evicted") { + const binding = target.binding; + const newlyEvicted = target.state === "evicted"; + if (newlyEvicted) { + this.emit({ + name: "subsession.released", + projectId: identity.projectId, + sessionId: binding.sessionId, + }); + } + try { + const privateMarker = + this.options.sessionManager.getSubsessionBinding(binding.sessionId); + if (privateMarker) { + if ( + privateMarker.projectId !== identity.projectId || + privateMarker.parentSessionId !== binding.parentSessionId || + privateMarker.bindingId !== binding.bindingId || + privateMarker.sessionId !== binding.sessionId + ) { + throw error( + "binding_session_mismatch", + false, + "inspect_session", + ); + } + await this.options.sessionManager.closeBound(privateMarker); + } + if (binding.disposition === "dormant-evicted") + await this.options.store.completeDormantReleaseCleanup( + identity, + binding.bindingId, + binding.sessionId, + ); + results.push(this.releasedResult(binding)); + } catch (cause) { + const detail = this.itemError(cause); + this.emit({ + name: + detail.code === "binding_session_mismatch" + ? "subsession.manual_session_protected" + : "subsession.failed", + projectId: identity.projectId, + sessionId: binding.sessionId, + code: detail.code, + }); + results.push({ + ...this.releasedResult(binding), + error: detail, + }); + } + continue; + } + const binding = target.binding; + const scopedIdentity = bindingIdentity(identity, binding); + try { + const privateMarker = this.options.sessionManager.getSubsessionBinding( + binding.sessionId, + ); + const session = this.options.sessionManager.get(binding.sessionId); + if (privateMarker) { + const expected = markerFor( + binding, + binding.runtime?.incarnation ?? privateMarker.incarnation, + ); + if (!this.options.sessionManager.matchesSubsessionBinding(expected)) + throw error( + "binding_session_mismatch", + false, + "inspect_session", + ); + await this.options.sessionManager.closeBound(expected); + } else if (binding.sessionState === "closed") { + // The durable coordinator close won before private marker cleanup. + } else if ( + session || + binding.runtime !== null || + !["reserved", "spawn-claimed"].includes(binding.sessionState) + ) { + throw error( + "binding_session_mismatch", + false, + "inspect_session", + ); + } + await this.options.store.closeBinding( + scopedIdentity, + binding.bindingId, + binding.sessionId, + ); + const closed = await this.options.store.finalizeReleasedBinding( + scopedIdentity, + binding.bindingId, + binding.sessionId, + ); + this.emit({ + name: "subsession.released", + projectId: binding.projectId, + sessionId: binding.sessionId, + }); + results.push(this.releasedResult(closed)); + } catch (cause) { + const detail = this.itemError(cause); + this.emit({ + name: + detail.code === "binding_session_mismatch" + ? "subsession.manual_session_protected" + : "subsession.failed", + projectId: binding.projectId, + sessionId: binding.sessionId, + code: detail.code, + }); + results.push(this.failedResult(binding, detail)); + } + } + return { + schemaVersion: 1, + requestKey, + requestDigest: reserved.requestDigest, + replayed: reserved.replayed, + results: results.sort((left, right) => + left.delegationKey.localeCompare(right.delegationKey), + ), + }; + } + + private async reconcileBinding( + caller: ProjectAgentSession, + initial: SubsessionBindingRecord, + replayed: boolean, + waitDeadline = Date.now() + this.batchWaitTimeoutMs, + ): Promise { + const identity = bindingIdentity(caller, initial); + let binding = initial; + try { + binding = await this.reconcileHistoricalDelivery(identity, binding); + const focused = await this.resolveFocus(binding); + binding = await this.options.store.setFocusedContextState( + identity, + binding.bindingId, + { + expectedContextEpoch: binding.contextEpoch, + expectedContextDigest: binding.contextDigest, + state: focused.state, + projectionDigest: focused.projectionDigest, + }, + ); + if (focused.state === "stale") { + this.emit({ + name: "subsession.context_stale", + projectId: binding.projectId, + sessionId: binding.sessionId, + code: "context_stale", + }); + return this.failedResult( + binding, + error("context_stale", false, "refresh_context"), + ); + } + if (Date.now() >= waitDeadline) throw new SessionNotReadyError(binding.sessionId); + const ensured = await this.ensureSession(caller, binding, focused.projection, waitDeadline); + binding = ensured.binding; + if (binding.sessionState !== "ready") + return this.result(binding, "already-running"); + binding = await this.deliver(identity, binding, focused.projection); + const outcome: DelegationItemOutcome = ensured.created + ? "created" + : replayed || ensured.reused + ? "reused" + : "already-running"; + this.emit({ + name: ensured.created ? "subsession.created" : "subsession.reused", + projectId: binding.projectId, + sessionId: binding.sessionId, + }); + return this.result(binding, outcome); + } catch (cause) { + binding = + (await this.options.store + .readBinding(identity, { + kind: "binding-id", + bindingId: binding.bindingId, + }) + .catch(() => null)) ?? binding; + const detail = this.itemError(cause); + this.emit({ + name: + detail.code === "binding_session_mismatch" + ? "subsession.manual_session_protected" + : "subsession.failed", + projectId: binding.projectId, + sessionId: binding.sessionId, + code: detail.code, + }); + return this.failedResult(binding, detail); + } + } + + private async resolveFocus( + binding: SubsessionBindingRecord, + ): Promise { + const focus = binding.currentFocus; + if (!focus) + return { state: "none", projection: null, projectionDigest: null }; + const aggregate = await this.options.planningStore.read(binding.projectId); + const mapRef = focus.kind === "brief" ? null : focus.map; + const map = aggregate.mapVersions.find( + ({ versionId }) => versionId === (mapRef?.versionId ?? ""), + ); + if (focus.kind === "brief") { + const brief = Object.values(aggregate.briefVersionsById) + .flat() + .find((candidate) => refsMatchBrief(candidate, focus)); + if (!brief) throw error("context_not_found", false, "reread"); + const exactMap = aggregate.mapVersions.find( + ({ versionId }) => versionId === brief.map.versionId, + ); + const exactPlan = aggregate.buildPlanVersions.find( + ({ versionId }) => versionId === brief.plan.versionId, + ); + if (!exactMap || !exactPlan) + throw error("context_not_found", false, "reread"); + const pointer = Object.values(aggregate.current.briefsByScope).find( + ({ briefId }) => briefId === brief.briefId, + ); + const stale = + !aggregate.current.map || + !aggregate.current.buildPlan || + !agentMapVersionRefsEqual(aggregate.current.map, brief.map) || + !projectBuildPlanVersionRefsEqual( + aggregate.current.buildPlan, + brief.plan, + ) || + !pointer || + pointer.status !== "active" || + pointer.version.versionId !== brief.versionId || + pointer.version.semanticDigest !== brief.semanticDigest; + if (stale) + return { state: "stale", projection: null, projectionDigest: null }; + const projection = serializeFocusedSessionContext({ + map: exactMap, + plan: exactPlan, + brief, + }); + if (!projection.ok) + throw error("context_not_found", false, "reread"); + return { + state: "current", + projection: projection.projection, + projectionDigest: canonicalDigest( + "sapiom.subsession.focused-projection.v1", + projection.projection, + ) as SubsessionProjectionDigest, + }; + } + if ( + !map || + !agentMapVersionRefsEqual( + { + projectId: map.projectId, + versionId: map.versionId, + contentDigest: map.contentDigest, + }, + focus.map, + ) + ) { + throw error("context_not_found", false, "reread"); + } + const currentMap = aggregate.current.map; + let stale = !currentMap || !agentMapVersionRefsEqual(currentMap, focus.map); + if (focus.kind === "assignment") { + const plan = aggregate.buildPlanVersions.find( + ({ versionId }) => versionId === focus.plan.versionId, + ); + if ( + !plan || + !projectBuildPlanVersionRefsEqual( + { + projectId: plan.projectId, + planId: plan.planId, + versionId: plan.versionId, + semanticDigest: plan.semanticDigest, + }, + focus.plan, + ) || + !plan.content.assignments.some(({ id }) => id === focus.assignmentId) + ) { + throw error("context_not_found", false, "reread"); + } + stale = + stale || + !aggregate.current.buildPlan || + !projectBuildPlanVersionRefsEqual( + aggregate.current.buildPlan, + focus.plan, + ); + } else { + if (!map.graph.nodes.some(({ id }) => id === focus.nodeId)) + throw error("context_not_found", false, "reread"); + if (focus.plan) { + const plan = aggregate.buildPlanVersions.find( + ({ versionId }) => versionId === focus.plan!.versionId, + ); + if ( + !plan || + !projectBuildPlanVersionRefsEqual( + { + projectId: plan.projectId, + planId: plan.planId, + versionId: plan.versionId, + semanticDigest: plan.semanticDigest, + }, + focus.plan, + ) + ) { + throw error("context_not_found", false, "reread"); + } + stale = + stale || + !aggregate.current.buildPlan || + !projectBuildPlanVersionRefsEqual( + aggregate.current.buildPlan, + focus.plan, + ); + } + } + return { + state: stale ? "stale" : "current", + projection: null, + projectionDigest: null, + }; + } + + private async ensureSession( + caller: ProjectAgentSession, + initial: SubsessionBindingRecord, + projection: FocusedSessionContextProjection | null, + waitDeadline: number, + ): Promise<{ + binding: SubsessionBindingRecord; + created: boolean; + reused: boolean; + }> { + const identity = bindingIdentity(caller, initial); + let binding = await this.options.store.readBinding(identity, { + kind: "binding-id", + bindingId: initial.bindingId, + }); + const existing = this.options.sessionManager.get(binding.sessionId); + const privateMarker = + this.options.sessionManager.getSubsessionBinding(binding.sessionId); + if (existing || privateMarker) { + const incarnation = binding.runtime?.incarnation ?? privateMarker?.incarnation ?? 1; + const expected = markerFor(binding, incarnation); + if ( + !privateMarker || + !this.options.sessionManager.matchesSubsessionBinding(expected) + ) { + throw new SubsessionBindingMismatchError(); + } + if (!existing) { + if ( + !binding.spawnClaim || + binding.spawnClaim.expiresAt > new Date().toISOString() + ) { + return { binding, created: false, reused: true }; + } + const session = await this.options.sessionManager.createReserved( + binding.sessionId, + { cwd: binding.projectRoot, harness: binding.harness }, + expected, + { + agentMapIdentity: (sessionId) => ({ + projectId: binding.projectId, + userId: caller.userId, + sessionId, + }), + initialTitle: this.title(binding.outcome), + ...(projection ? { focusedContext: () => projection } : {}), + }, + ); + const runtimeToken = this.options.sessionManager.getRuntimeEpoch( + session.id, + ); + if (!runtimeToken) + throw error("session_create_failed", true, "retry"); + binding = await this.options.store.attachSpawnedRuntime( + identity, + binding.bindingId, + { + claimId: binding.spawnClaim.claimId, + spawnEpoch: binding.spawnEpoch, + runtimeToken, + incarnation: expected.incarnation, + }, + ); + binding = await this.advanceToReady(identity, binding, runtimeToken, waitDeadline); + return { binding, created: false, reused: true }; + } + if (this.options.sessionManager.wasSubsessionClosedByUser(expected)) { + await this.options.store.closeBinding( + identity, + binding.bindingId, + binding.sessionId, + ); + throw error("session_closed", false, "inspect_session"); + } + if (this.options.sessionManager.isLive(binding.sessionId)) { + const runtimeToken = + this.options.sessionManager.getRuntimeEpoch(binding.sessionId)!; + if (binding.runtime?.runtimeToken === runtimeToken) { + binding = await this.advanceToReady(identity, binding, runtimeToken, waitDeadline); + return { binding, created: false, reused: true }; + } + if (binding.spawnClaim) { + if (binding.spawnClaim.expiresAt > new Date().toISOString()) + return { binding, created: false, reused: true }; + binding = await this.options.store.attachSpawnedRuntime( + identity, + binding.bindingId, + { + claimId: binding.spawnClaim.claimId, + spawnEpoch: binding.spawnEpoch, + runtimeToken, + incarnation: privateMarker.incarnation, + }, + ); + binding = await this.advanceToReady(identity, binding, runtimeToken, waitDeadline); + return { binding, created: false, reused: true }; + } + throw error("binding_session_mismatch", false, "inspect_session"); + } + if (existing.status === "exited") { + binding = await this.recoverExitedSession( + caller, + identity, + binding, + privateMarker, + projection, + waitDeadline, + ); + return { binding, created: false, reused: true }; + } + throw error("session_unreachable", true, "inspect_session"); + } + + let claim = + binding.sessionState === "spawn-claimed" && + binding.spawnClaim?.ownerId === this.ownerId && + binding.spawnClaim.expiresAt > new Date().toISOString() + ? { claimed: true as const, binding } + : await this.options.store.claimSpawn( + identity, + binding.bindingId, + { + ownerId: this.ownerId, + expectedLifecycleEpoch: binding.lifecycleEpoch, + expectedSpawnEpoch: binding.spawnEpoch, + }, + ); + if (!claim.claimed) { + if (claim.reason !== "expired-requires-inspection") + return { binding: claim.binding, created: false, reused: true }; + const expired = claim.binding.spawnClaim; + if (!expired) + throw error("session_unreachable", true, "inspect_session"); + claim = await this.options.store.takeoverExpiredSpawnClaim( + identity, + claim.binding.bindingId, + { + ownerId: this.ownerId, + expiredClaimId: expired.claimId, + expectedLifecycleEpoch: claim.binding.lifecycleEpoch, + expectedSpawnEpoch: claim.binding.spawnEpoch, + }, + ); + if (!claim.claimed) + return { binding: claim.binding, created: false, reused: true }; + } + binding = claim.binding; + const spawnClaim = binding.spawnClaim!; + const marker = markerFor(binding, 1); + let session: HarnessSession; + try { + session = await this.options.sessionManager.createReserved( + binding.sessionId, + { cwd: binding.projectRoot, harness: binding.harness }, + marker, + { + agentMapIdentity: (sessionId) => ({ + projectId: binding.projectId, + userId: caller.userId, + sessionId, + }), + initialTitle: this.title(binding.outcome), + ...(projection + ? { focusedContext: () => projection } + : {}), + }, + ); + } catch (cause) { + // A missing session row is positive proof that createWithId never + // reached its first durable row/process side effect. Every later failure + // keeps the claim fenced for exact inspection instead of guessing. + if (!this.options.sessionManager.get(binding.sessionId)) { + await this.options.store + .releaseUnspawnedClaim(identity, binding.bindingId, { + claimId: spawnClaim.claimId, + spawnEpoch: binding.spawnEpoch, + proof: "no-process-created", + }) + .catch(() => {}); + } + throw cause; + } + const runtimeToken = this.options.sessionManager.getRuntimeEpoch(session.id); + if (!runtimeToken) + throw error("session_create_failed", true, "retry"); + binding = await this.options.store.attachSpawnedRuntime( + identity, + binding.bindingId, + { + claimId: spawnClaim.claimId, + spawnEpoch: binding.spawnEpoch, + runtimeToken, + incarnation: marker.incarnation, + }, + ); + binding = await this.advanceToReady(identity, binding, runtimeToken, waitDeadline); + return { binding, created: true, reused: false }; + } + + private async recoverExitedSession( + caller: ProjectAgentSession, + identity: ProjectAgentSession, + initial: SubsessionBindingRecord, + currentMarker: TrustedSubsessionBindingMarker, + projection: FocusedSessionContextProjection | null, + waitDeadline: number, + ): Promise { + let binding = initial; + if ( + binding.runtime && + ["starting", "awaiting-ready", "ready"].includes(binding.sessionState) + ) { + binding = await this.options.store.transitionSession( + identity, + binding.bindingId, + { + expectedLifecycleEpoch: binding.lifecycleEpoch, + expectedSpawnEpoch: binding.spawnEpoch, + expectedRuntimeToken: binding.runtime.runtimeToken, + state: "exited", + }, + ); + } + const resumable = await this.options.sessionManager.canResumeSession( + binding.sessionId, + ); + if (!resumable && currentDelivery(binding).state !== "pending") { + throw error("session_unreachable", false, "inspect_session"); + } + let claim; + if (binding.sessionState === "spawn-claimed" && binding.spawnClaim) { + if ( + binding.spawnClaim.ownerId === this.ownerId && + binding.spawnClaim.expiresAt > new Date().toISOString() + ) { + claim = { claimed: true as const, binding }; + } else { + if (binding.spawnClaim.expiresAt > new Date().toISOString()) return binding; + claim = await this.options.store.takeoverExpiredSpawnClaim( + identity, + binding.bindingId, + { + ownerId: this.ownerId, + expiredClaimId: binding.spawnClaim.claimId, + expectedLifecycleEpoch: binding.lifecycleEpoch, + expectedSpawnEpoch: binding.spawnEpoch, + }, + ); + } + } else { + claim = await this.options.store.claimSpawn( + identity, + binding.bindingId, + { + ownerId: this.ownerId, + expectedLifecycleEpoch: binding.lifecycleEpoch, + expectedSpawnEpoch: binding.spawnEpoch, + }, + ); + } + if (!claim.claimed) return claim.binding; + binding = claim.binding; + const spawnClaim = binding.spawnClaim!; + const nextMarker = markerFor(binding, currentMarker.incarnation + 1); + const trusted = projection ? { focusedContext: projection } : {}; + if (resumable) { + await this.options.sessionManager.resumeBound( + binding.sessionId, + currentMarker, + nextMarker, + trusted, + ); + } else { + const index = await this.options.eventReader.index(); + const hasRecordedTurns = async () => + (index.bySession.get(binding.sessionId)?.turnCount ?? 0) > 0; + await this.options.sessionManager.restartFreshBound( + binding.sessionId, + currentMarker, + nextMarker, + { + agentMapIdentity: (sessionId) => ({ + projectId: binding.projectId, + userId: caller.userId, + sessionId, + }), + initialTitle: this.title(binding.outcome), + ...(projection ? { focusedContext: () => projection } : {}), + }, + hasRecordedTurns, + ); + } + const runtimeToken = this.options.sessionManager.getRuntimeEpoch( + binding.sessionId, + ); + if (!runtimeToken) + throw error("session_restart_failed", true, "retry"); + binding = await this.options.store.attachSpawnedRuntime( + identity, + binding.bindingId, + { + claimId: spawnClaim.claimId, + spawnEpoch: binding.spawnEpoch, + runtimeToken, + incarnation: nextMarker.incarnation, + }, + ); + return this.advanceToReady(identity, binding, runtimeToken, waitDeadline); + } + + private async advanceToReady( + identity: ProjectAgentSession, + initial: SubsessionBindingRecord, + runtimeToken: string, + waitDeadline: number, + ): Promise { + let binding = initial; + if (binding.sessionState === "starting") { + binding = await this.options.store.transitionSession( + identity, + binding.bindingId, + { + expectedLifecycleEpoch: binding.lifecycleEpoch, + expectedSpawnEpoch: binding.spawnEpoch, + expectedRuntimeToken: runtimeToken, + state: "awaiting-ready", + }, + ); + } + if (binding.sessionState === "awaiting-ready") { + const ready = await this.waitForReady(binding.sessionId, runtimeToken, waitDeadline); + if (!ready) throw new SessionNotReadyError(binding.sessionId); + await this.waitForAdapterIdentity(binding.sessionId, runtimeToken, waitDeadline); + binding = await this.options.store.transitionSession( + identity, + binding.bindingId, + { + expectedLifecycleEpoch: binding.lifecycleEpoch, + expectedSpawnEpoch: binding.spawnEpoch, + expectedRuntimeToken: runtimeToken, + state: "ready", + }, + ); + } + if (binding.sessionState === "ready") + this.emit({ + name: "subsession.ready", + projectId: binding.projectId, + sessionId: binding.sessionId, + }); + return binding; + } + + private async waitForAdapterIdentity( + sessionId: string, + runtimeToken: string, + waitDeadline: number, + ): Promise { + const deadline = Math.min(waitDeadline, Date.now() + this.readinessTimeoutMs); + for (;;) { + if (!this.options.sessionManager.isCurrentRuntimeEpoch(sessionId, runtimeToken)) + throw error("session_unreachable", true, "inspect_session"); + const state = this.options.sessionManager.getAdapterIdentityState( + sessionId, + runtimeToken, + ); + if (state === "ready" || state === "not-required") return; + if (state === "ambiguous") + throw error("adapter_identity_ambiguous", false, "inspect_session"); + if (state === "unavailable") + throw error("adapter_unavailable", true, "retry"); + if (Date.now() >= deadline) + throw error(Date.now() >= waitDeadline ? "readiness_timeout" : "adapter_unavailable", true, "retry"); + await new Promise((resolve) => setTimeout(resolve, Math.min(ADAPTER_IDENTITY_POLL_MS, deadline - Date.now()))); + } + } + + private waitForReady(sessionId: string, runtimeToken: string, waitDeadline: number): Promise { + if ( + this.options.sessionManager.get(sessionId)?.ready && + this.options.sessionManager.isCurrentRuntimeEpoch(sessionId, runtimeToken) + ) { + return Promise.resolve(true); + } + return new Promise((resolve) => { + let settled = false; + const finish = (value: boolean) => { + if (settled) return; + settled = true; + clearTimeout(timer); + unsubscribe(); + resolve(value); + }; + const unsubscribe = this.options.sessionManager.onStatusChange( + (session, context) => { + if (session.id !== sessionId) return; + if (context.runtimeEpoch !== runtimeToken || session.status === "exited") + finish(false); + else if (session.ready) finish(true); + }, + ); + const timer = setTimeout( + () => finish(false), + Math.max(0, Math.min(this.readinessTimeoutMs, waitDeadline - Date.now())), + ); + }); + } + + private async deliver( + identity: ProjectAgentSession, + initial: SubsessionBindingRecord, + projection: FocusedSessionContextProjection | null, + ): Promise { + let binding = await this.reconcileHistoricalDelivery(identity, initial); + let delivery = currentDelivery(binding); + if (delivery.state !== "pending") return binding; + const watermark = await this.latestEventId(binding.sessionId); + const claimed = await this.options.store.claimKickoff( + identity, + binding.bindingId, + { + ownerId: this.ownerId, + expectedLifecycleEpoch: binding.lifecycleEpoch, + expectedSpawnEpoch: binding.spawnEpoch, + expectedContextEpoch: binding.contextEpoch, + eventWatermark: watermark, + }, + ); + if (!claimed.claimed) { + if (claimed.reason === "expired-requires-reconciliation") + return this.options.store.markKickoffUncertain( + identity, + binding.bindingId, + { + contextEpoch: binding.contextEpoch, + deliveryId: delivery.deliveryId, + inputId: delivery.inputId, + }, + ); + return claimed.binding; + } + binding = claimed.binding; + delivery = currentDelivery(binding); + const runtimeToken = binding.runtime?.runtimeToken; + if (!runtimeToken) + throw error("session_unreachable", true, "inspect_session"); + const prompt = this.kickoffPrompt(binding, delivery, projection); + const tracked = await this.options.sessionManager.submitInputTracked( + binding.sessionId, + prompt, + { + background: true, + canWrite: async () => { + const current = await this.options.store.readBinding(identity, { + kind: "binding-id", + bindingId: binding.bindingId, + }); + return ( + current.lifecycleEpoch === binding.lifecycleEpoch && + current.spawnEpoch === binding.spawnEpoch && + current.contextEpoch === binding.contextEpoch && + current.contextDigest === binding.contextDigest && + current.runtime?.runtimeToken === runtimeToken && + this.options.sessionManager.isCurrentRuntimeEpoch( + binding.sessionId, + runtimeToken, + ) + ); + }, + }, + ); + const after = await this.options.store.readBinding(identity, { + kind: "binding-id", + bindingId: binding.bindingId, + }); + if (currentDelivery(after).state === "acknowledged") return after; + binding = await this.options.store.recordKickoffWrite( + identity, + binding.bindingId, + { + contextEpoch: binding.contextEpoch, + deliveryId: delivery.deliveryId, + inputId: delivery.inputId, + claimId: delivery.claim!.claimId, + phase: tracked.phase, + }, + ); + const state = currentDelivery(binding).state; + this.emit({ + name: + state === "uncertain" + ? "subsession.kickoff_uncertain" + : "subsession.kickoff_submitted", + projectId: binding.projectId, + sessionId: binding.sessionId, + }); + return binding; + } + + private kickoffPrompt( + binding: SubsessionBindingRecord, + delivery: ReturnType, + projection: FocusedSessionContextProjection | null, + ): string { + const marker = ``; + if (binding.contextEpoch > 1) { + return [ + "Focused project context refresh for your existing delegated task.", + ...(projection + ? [projection] + : ["The optional focused overlay is now cleared or reference-only. Read current shared project state through the common tools when needed."]), + "This context update does not change your tools, writable policy, or authority.", + marker, + ].join("\n\n"); + } + return [ + "You are an ordinary writable project session delegated by another project session.", + `Outcome: ${binding.outcome}`, + ...(binding.kickoffContext + ? [`Bounded kickoff context:\n${binding.kickoffContext}`] + : []), + "Plan or implement directly as appropriate. Keep shared project state current and delegate further when useful.", + marker, + ].join("\n\n"); + } + + async onEventPersisted( + event: AnalyticsEvent, + runtimeToken: string, + ): Promise { + if (event.type !== "prompt.submitted") return; + const prompt = + typeof event.payload.prompt === "string" ? event.payload.prompt : ""; + const match = KICKOFF_MARKER.exec(prompt); + if (!match) return; + const session = this.options.sessionManager.get(event.harnessSessionId); + if (!session?.agentMapIdentity) return; + const aggregate = await this.options.store.read( + session.agentMapIdentity.projectId, + ); + const binding = aggregate.bindings.find( + ({ sessionId, bindingId }) => + sessionId === event.harnessSessionId && bindingId === match[1], + ); + if ( + !binding || + binding.runtime?.runtimeToken !== runtimeToken || + !this.options.sessionManager.isCurrentRuntimeEpoch( + binding.sessionId, + runtimeToken, + ) + ) { + return; + } + const delivery = binding.deliveries.find( + ({ deliveryId, inputId, contextEpoch }) => + deliveryId === match[2] && + inputId === match[3] && + contextEpoch === Number(match[4]), + ); + if ( + !delivery || + binding.spawnEpoch !== Number(match[5]) || + !delivery.eventWatermark + ) { + return; + } + const identity: ProjectAgentSession = { + projectId: binding.projectId, + userId: session.agentMapIdentity.userId, + sessionId: binding.parentSessionId, + }; + await this.options.store.acknowledgeKickoff( + identity, + binding.bindingId, + { + contextEpoch: binding.contextEpoch, + deliveryId: delivery.deliveryId, + inputId: delivery.inputId, + eventWatermark: delivery.eventWatermark, + }, + ); + this.emit({ + name: "subsession.kickoff_acknowledged", + projectId: binding.projectId, + sessionId: binding.sessionId, + }); + } + + private async reconcileHistoricalDelivery( + identity: ProjectAgentSession, + initial: SubsessionBindingRecord, + ): Promise { + const delivery = currentDelivery(initial); + if ( + !["claimed", "submitted-unacknowledged", "uncertain"].includes( + delivery.state, + ) + ) { + return initial; + } + for await (const event of this.options.eventReader.read({ + harnessSessionId: initial.sessionId, + types: ["prompt.submitted"], + })) { + const prompt = + typeof event.payload.prompt === "string" ? event.payload.prompt : ""; + const match = KICKOFF_MARKER.exec(prompt); + if ( + match?.[1] === initial.bindingId && + match[2] === delivery.deliveryId && + match[3] === delivery.inputId && + Number(match[4]) === initial.contextEpoch && + Number(match[5]) === initial.spawnEpoch && + delivery.eventWatermark + ) { + return this.options.store.acknowledgeKickoff( + identity, + initial.bindingId, + { + contextEpoch: initial.contextEpoch, + deliveryId: delivery.deliveryId, + inputId: delivery.inputId, + eventWatermark: delivery.eventWatermark, + }, + ); + } + } + if ( + delivery.state === "submitted-unacknowledged" || + delivery.state === "uncertain" || + (delivery.state === "claimed" && + delivery.claim !== null && + delivery.claim.expiresAt <= new Date().toISOString()) + ) { + return this.options.store.markKickoffUncertain( + identity, + initial.bindingId, + { + contextEpoch: initial.contextEpoch, + deliveryId: delivery.deliveryId, + inputId: delivery.inputId, + }, + ); + } + return initial; + } + + private async latestEventId(sessionId: string): Promise { + let latest = "event_none"; + for await (const event of this.options.eventReader.read({ + harnessSessionId: sessionId, + })) { + latest = event.eventId; + } + return latest; + } + + private title(outcome: string): string { + const first = outcome.split("\n", 1)[0]?.trim() || "Delegated task"; + return [...first].slice(0, 80).join(""); + } + + private result( + binding: SubsessionBindingRecord, + outcome: DelegationItemOutcome, + ): DelegationItemResult { + return { + delegationKey: binding.delegationKey, + bindingId: binding.bindingId, + sessionId: binding.sessionId, + outcome, + sessionState: binding.sessionState, + contextState: binding.contextState, + kickoffState: currentDelivery(binding).state, + }; + } + + private releasedResult( + binding: Pick< + SubsessionBindingRecord, + "delegationKey" | "bindingId" | "sessionId" + >, + ): DelegationItemResult { + return { + delegationKey: binding.delegationKey, + bindingId: binding.bindingId, + sessionId: binding.sessionId, + outcome: "released", + sessionState: "closed", + contextState: "none", + kickoffState: "pending", + }; + } + + private failedResult( + binding: SubsessionBindingRecord, + detail: DelegationError, + ): DelegationItemResult { + return { ...this.result(binding, "failed"), error: detail }; + } + + private wholeCallError(cause: unknown, refresh = false): SubsessionCoordinatorError { + if (cause instanceof SubsessionCoordinatorError) return cause; + if (cause instanceof SubsessionCoordinatorStoreError) { + if ( + cause.code === "request_key_reused" + ) + return new SubsessionCoordinatorError( + error("request_key_reused", false, "new_request_key"), + ); + if (cause.code === "request_key_expired") + return new SubsessionCoordinatorError( + error("request_key_expired", false, "new_request_key"), + ); + if (cause.code === "delegation_key_reused") + return new SubsessionCoordinatorError( + error("delegation_key_reused", false, "new_delegation_key"), + ); + if (cause.code === "capacity_exceeded") + return new SubsessionCoordinatorError( + error("capacity_exceeded", false, "reduce_request"), + ); + if (cause.code === "live_session_limit_reached") + return new SubsessionCoordinatorError( + error("capacity_exceeded", false, "inspect_session"), + ); + if (cause.code === "history_quota_exceeded") + return new SubsessionCoordinatorError( + error("capacity_exceeded", false, "release_dormant"), + ); + if (cause.code === "delegation_depth_exceeded") { + return new SubsessionCoordinatorError( + error("capacity_exceeded", false, "none"), + ); + } + if (cause.code === "session_closed") + return new SubsessionCoordinatorError( + error("session_closed", false, "inspect_session"), + ); + if (cause.code === "storage_unavailable") + return new SubsessionCoordinatorError( + error("storage_unavailable", true, "retry"), + ); + if (refresh && cause.code === "binding_not_found") + return new SubsessionCoordinatorError( + error("context_not_found", false, "reread"), + ); + if (cause.code === "binding_not_found") + return new SubsessionCoordinatorError( + error("session_closed", false, "inspect_session"), + ); + if (refresh && cause.code === "lifecycle_conflict") + return new SubsessionCoordinatorError( + error("context_refresh_conflict", false, "reread"), + ); + if (refresh && cause.code === "claim_conflict") + return new SubsessionCoordinatorError( + error("kickoff_failed", false, "inspect_session"), + ); + } + return new SubsessionCoordinatorError( + error("internal_error", true, "retry"), + ); + } + + private itemError(cause: unknown): DelegationError { + if ( + typeof cause === "object" && + cause !== null && + "code" in cause && + typeof cause.code === "string" && + "retryable" in cause && + "recovery" in cause + ) { + return cause as DelegationError; + } + if (cause instanceof SubsessionBindingMismatchError) + return error("binding_session_mismatch", false, "inspect_session"); + if (cause instanceof SubsessionFreshRestartForbiddenError) + return error("session_restart_failed", false, "inspect_session"); + if (cause instanceof SessionNotReadyError) + return error("readiness_timeout", true, "retry"); + if (cause instanceof SubsessionCoordinatorStoreError) { + if (cause.code === "storage_unavailable") + return error("storage_unavailable", true, "retry"); + if (cause.code === "session_closed") + return error("session_closed", false, "inspect_session"); + if (["lifecycle_conflict", "claim_conflict"].includes(cause.code)) + return error("session_unreachable", true, "inspect_session"); + } + return error("session_create_failed", true, "retry"); + } +} diff --git a/packages/harness/src/server/agent-map-mcp-tools.ts b/packages/harness/src/server/agent-map-mcp-tools.ts index 16692601..731c84cd 100644 --- a/packages/harness/src/server/agent-map-mcp-tools.ts +++ b/packages/harness/src/server/agent-map-mcp-tools.ts @@ -14,6 +14,10 @@ import { AgentBriefAppendQuotaError, AgentMapWorkspaceStoreError } from "../core import { AgentMapAggregateError } from "../core/agent-map-aggregate-migration.js"; import { AgentBriefService, AgentBriefServiceError } from "../core/agent-brief-service.js"; import { BuildPlanService, BuildPlanServiceError } from "../core/build-plan-service.js"; +import { + SubsessionCoordinator, + SubsessionCoordinatorError, +} from "../core/subsession-coordinator.js"; import { agentBriefRefreshRequestSchema, buildPlanApplyRequestSchema, @@ -56,10 +60,71 @@ const batchSchema = z }) .strict(); +const versionId = z.string().min(1).max(128); +const digest = z.string().regex(/^sha256:[0-9a-f]{64}$/u); +const mapVersionRefSchema = z.object({ + projectId: versionId, + versionId, + contentDigest: digest, +}).strict(); +const planVersionRefSchema = z.object({ + projectId: versionId, + planId: versionId, + versionId, + semanticDigest: digest, +}).strict(); +const briefVersionRefSchema = z.object({ + projectId: versionId, + briefId: versionId, + versionId, + semanticDigest: digest, +}).strict(); +const delegationFocusSchema = z.discriminatedUnion("kind", [ + z.object({ kind: z.literal("assignment"), map: mapVersionRefSchema, + plan: planVersionRefSchema, assignmentId: versionId }).strict(), + z.object({ kind: z.literal("map-node"), map: mapVersionRefSchema, + plan: planVersionRefSchema.nullable(), nodeId: versionId }).strict(), + z.object({ kind: z.literal("brief"), brief: briefVersionRefSchema }).strict(), +]); +const delegationKey = z.string().min(1).max(128).regex(/^[A-Za-z0-9._-]+$/u); +const projectSubsessionRequestSchema = z.object({ + schemaVersion: z.literal(1), + requestKey: delegationKey, + operation: z.discriminatedUnion("kind", [ + z.object({ + kind: z.literal("delegate"), + delegations: z.array(z.object({ + delegationKey, + outcome: z.string().min(1).max(4_096), + kickoffContext: z.string().min(1).max(16_384).optional(), + focus: delegationFocusSchema.optional(), + }).strict()).min(1).max(16), + }).strict(), + z.object({ + kind: z.literal("refresh-focused-context"), + target: z.discriminatedUnion("kind", [ + z.object({ kind: z.literal("self") }).strict(), + z.object({ kind: z.literal("child"), delegationKey }).strict(), + ]), + expectedContextEpoch: z.number().int().positive(), + expectedContextDigest: digest, + focus: delegationFocusSchema.nullable(), + }).strict(), + z.object({ + kind: z.literal("release"), + delegationKeys: z.array(delegationKey).min(1).max(16), + }).strict(), + z.object({ + kind: z.literal("release-dormant"), + limit: z.number().int().min(1).max(16), + }).strict(), + ]), +}).strict(); + export interface AgentMapToolEvent { tool: "agent_map_read" | "agent_map_validate" | "agent_map_propose" | "build_plan_read" | "build_plan_validate" | "build_plan_apply" | "build_plan_rebase" | - "build_plan_brief_refresh"; + "build_plan_brief_refresh" | "project_subsession_delegate"; outcome: "ok" | "error"; errorCode?: string; latencyMs: number; @@ -109,6 +174,8 @@ function errorResult(error: unknown) { ? "new_request" : error.code === "source_mismatch" ? "reread" : error.code === "malformed_input" ? "correct" : error.code === "quota_exceeded" ? "manual_intervention" : "retry" } + : error instanceof SubsessionCoordinatorError + ? error.detail : error instanceof AgentMapWorkspaceStoreError || error instanceof AgentMapAggregateError ? { code: error.code, recovery: error.code === "storage_unavailable" ? "retry" : "manual_intervention" } : { code: "internal_error", recovery: "retry" }; @@ -140,6 +207,7 @@ export function createAgentMapToolServer( service: AgentMapProposalService, buildPlanService: BuildPlanService, agentBriefService: AgentBriefService, + subsessionCoordinator: SubsessionCoordinator, options: AgentMapMcpToolsOptions = {}, ): McpServer { const server = new McpServer({ @@ -317,5 +385,18 @@ export function createAgentMapToolServer( }), ); + server.registerTool( + "project_subsession_delegate", + { + description: "Create, reuse, or release a bounded batch of ordinary writable project subsessions, reclaim a bounded project-wide set of coordinator-owned dormant bindings, or refresh exact focused context, using caller-owned idempotency keys.", + inputSchema: projectSubsessionRequestSchema, + annotations: { readOnlyHint: false, destructiveHint: true, openWorldHint: false }, + }, + async (request) => instrument("project_subsession_delegate", async () => { + const result = await subsessionCoordinator.execute(identity, request); + return toolResult(result, `Delegation reconciled ${result.results.length} project subsession${result.results.length === 1 ? "" : "s"}.`); + }), + ); + return server; } diff --git a/packages/harness/src/server/agent-map-mcp-wiring.test.ts b/packages/harness/src/server/agent-map-mcp-wiring.test.ts index e70274d0..91134577 100644 --- a/packages/harness/src/server/agent-map-mcp-wiring.test.ts +++ b/packages/harness/src/server/agent-map-mcp-wiring.test.ts @@ -117,6 +117,7 @@ it("uses the actual ephemeral port and revokes private MCP launch authority on e "build_plan_read", "build_plan_rebase", "build_plan_validate", + "project_subsession_delegate", ]); const snapshot = await client.callTool({ name: "agent_map_read", @@ -128,6 +129,101 @@ it("uses the actual ephemeral port and revokes private MCP launch authority on e project: { projectId: session.agentMapIdentity!.projectId }, proposal: null, }); + const stopReadyBridge = server.sessionManager.onStatusChange( + (candidate, context) => { + if ( + candidate.id !== session.id && + candidate.status === "running" && + !candidate.ready && + context.runtimeEpoch + ) { + server!.sessionManager.setReady(candidate.id, context.runtimeEpoch); + } + }, + ); + const delegationArguments = { + schemaVersion: 1, + requestKey: "wiring-delegation", + operation: { + kind: "delegate", + delegations: [{ + delegationKey: "child", + outcome: "Implement the focused child task", + }], + }, + }; + const delegated = await client.callTool({ + name: "project_subsession_delegate", + arguments: delegationArguments, + }); + const childMcp = launchOpts?.agentMapMcp; + expect(childMcp).toBeDefined(); + const childClient = new Client({ name: "nested-delegation-test", version: "1" }); + await childClient.connect(new StreamableHTTPClientTransport(new URL(childMcp!.url), { + requestInit: { headers: { Authorization: `Bearer ${childMcp!.bearerToken}` } }, + })); + expect((await childClient.listTools()).tools.map(({ name }) => name)).toContain( + "project_subsession_delegate", + ); + const nested = await childClient.callTool({ + name: "project_subsession_delegate", + arguments: { + schemaVersion: 1, + requestKey: "nested-request", + operation: { + kind: "delegate", + delegations: [{ + delegationKey: "grandchild", + outcome: "Implement the nested task", + }], + }, + }, + }); + expect(nested.structuredContent).toMatchObject({ + results: [{ outcome: "created", sessionState: "ready" }], + }); + await childClient.close(); + const retried = await client.callTool({ + name: "project_subsession_delegate", + arguments: delegationArguments, + }); + const released = await client.callTool({ + name: "project_subsession_delegate", + arguments: { + schemaVersion: 1, + requestKey: "release-child", + operation: { kind: "release", delegationKeys: ["child"] }, + }, + }); + stopReadyBridge(); + expect(delegated.isError).not.toBe(true); + expect(delegated.structuredContent).toMatchObject({ + requestKey: "wiring-delegation", + results: [{ outcome: "created", sessionState: "ready" }], + }); + expect(retried.structuredContent).toMatchObject({ + replayed: true, + results: [{ + outcome: "reused", + sessionId: (delegated.structuredContent as { results: Array<{ sessionId: string }> }).results[0]!.sessionId, + }], + }); + expect(released.structuredContent).toMatchObject({ + results: [{ outcome: "released", sessionState: "closed" }], + }); + expect(server.sessionManager.isLive( + (delegated.structuredContent as { results: Array<{ sessionId: string }> }) + .results[0]!.sessionId, + )).toBe(false); + expect(server.sessionManager.getSubsessionBinding( + (delegated.structuredContent as { results: Array<{ sessionId: string }> }) + .results[0]!.sessionId, + )).toBeNull(); + expect(server.sessionManager.list().filter(({ id }) => + id === (nested.structuredContent as { results: Array<{ sessionId: string }> }) + .results[0]!.sessionId, + )).toHaveLength(1); + expect(server.sessionManager.list()).toHaveLength(3); await client.close(); await server.sessionManager.kill(session.id); @@ -370,6 +466,7 @@ it("gives every signed-out project session the same coding prompt and Agent Map "build_plan_read", "build_plan_rebase", "build_plan_validate", + "project_subsession_delegate", ]); const proposalEvents: BusMessage[] = []; diff --git a/packages/harness/src/server/agent-map-mcp.test.ts b/packages/harness/src/server/agent-map-mcp.test.ts index b39afb72..0a5fcb2d 100644 --- a/packages/harness/src/server/agent-map-mcp.test.ts +++ b/packages/harness/src/server/agent-map-mcp.test.ts @@ -16,6 +16,7 @@ import { AgentMapWorkspaceStore, AgentMapWorkspaceStoreError, AgentBriefAppendQu import { BuildPlanService } from "../core/build-plan-service.js"; import { BuildPlanStore } from "../core/build-plan-store.js"; import { AgentBriefService, AgentBriefServiceError } from "../core/agent-brief-service.js"; +import type { SubsessionCoordinator } from "../core/subsession-coordinator.js"; import { createAgentMapMcpRouter, type AgentMapMcpRouterOptions, @@ -60,8 +61,17 @@ async function fixture( const buildPlanService = new BuildPlanService(new BuildPlanStore(workspaceStore)); const briefStore = new BuildPlanStore(workspaceStore); const agentBriefService = createAgentBriefService?.(briefStore) ?? new AgentBriefService(briefStore); + const subsessionCoordinator = { + execute: vi.fn(async (_identity, request: { requestKey: string }) => ({ + schemaVersion: 1 as const, + requestKey: request.requestKey, + requestDigest: `sha256:${"0".repeat(64)}`, + replayed: false, + results: [], + })), + } as unknown as SubsessionCoordinator; const mcp = createAgentMapMcpRouter({ capabilities, service, buildPlanService, - agentBriefService, ...routerOptions }); + agentBriefService, subsessionCoordinator, ...routerOptions }); const app = express(); app.use(express.json()); app.use(mcp.router); @@ -108,6 +118,7 @@ describe("Agent Map Streamable HTTP MCP", () => { "build_plan_read", "build_plan_rebase", "build_plan_validate", + "project_subsession_delegate", ]); const nonStrict = tools.tools.filter((tool) => !(tool.inputSchema.additionalProperties === false || (Array.isArray(tool.inputSchema.anyOf) && tool.inputSchema.anyOf.every((variant) => @@ -121,6 +132,40 @@ describe("Agent Map Streamable HTTP MCP", () => { isError: true, structuredContent: { code: "malformed_input", recovery: "correct" }, }); + await expect(client.callTool({ + name: "project_subsession_delegate", + arguments: { + schemaVersion: 1, + requestKey: "delegate-one", + operation: { + kind: "delegate", + delegations: [{ + delegationKey: "focused-task", + outcome: "Implement the focused task", + }], + }, + }, + })).resolves.toMatchObject({ + structuredContent: { + schemaVersion: 1, + requestKey: "delegate-one", + replayed: false, + }, + }); + await expect(client.callTool({ + name: "project_subsession_delegate", + arguments: { + schemaVersion: 1, + requestKey: "release-dormant-one", + operation: { kind: "release-dormant", limit: 1 }, + }, + })).resolves.toMatchObject({ + structuredContent: { + schemaVersion: 1, + requestKey: "release-dormant-one", + results: [], + }, + }); const validate = tools.tools.find( ({ name }) => name === "agent_map_validate", )!; diff --git a/packages/harness/src/server/agent-map-mcp.ts b/packages/harness/src/server/agent-map-mcp.ts index 8bda0f4f..00a36b33 100644 --- a/packages/harness/src/server/agent-map-mcp.ts +++ b/packages/harness/src/server/agent-map-mcp.ts @@ -15,6 +15,7 @@ import { import type { AgentMapProposalService } from "../core/agent-map-proposal-service.js"; import type { BuildPlanService } from "../core/build-plan-service.js"; import type { AgentBriefService } from "../core/agent-brief-service.js"; +import type { SubsessionCoordinator } from "../core/subsession-coordinator.js"; import { createAgentMapToolServer, type AgentMapMcpToolsOptions } from "./agent-map-mcp-tools.js"; interface BoundTransport { @@ -30,6 +31,7 @@ export interface AgentMapMcpRouterOptions service: AgentMapProposalService; buildPlanService: BuildPlanService; agentBriefService: AgentBriefService; + subsessionCoordinator: SubsessionCoordinator; readSnapshotFor?: (identity: ResolvedAgentMapCapability["identity"]) => Promise; maxSessions?: number; now?: () => number; @@ -156,7 +158,7 @@ export function createAgentMapMcpRouter(options: AgentMapMcpRouterOptions): Agen if (sessionId) sessions.delete(sessionId); }; const server = createToolServer(capability.identity, options.service, options.buildPlanService, - options.agentBriefService, { + options.agentBriefService, options.subsessionCoordinator, { onEvent: options.onEvent, ...(options.readSnapshotFor ? { diff --git a/packages/harness/src/server/index.ts b/packages/harness/src/server/index.ts index cd9b36ec..f69b8b4c 100644 --- a/packages/harness/src/server/index.ts +++ b/packages/harness/src/server/index.ts @@ -7,6 +7,8 @@ * src/shared/types.ts for the full protocol contract. */ +import { SubsessionCoordinatorStore, type SubsessionCoordinatorStoreEvent } from "../core/subsession-coordinator-store.js"; +import { SubsessionCoordinator, type SubsessionCoordinatorEvent } from "../core/subsession-coordinator.js"; import { CodexRolloutBroker } from "../core/collector/codex-rollout-broker.js"; import { AgentBriefService } from "../core/agent-brief-service.js"; import { BuildPlanStore } from "../core/build-plan-store.js"; @@ -1378,6 +1380,14 @@ export const startServer = async ( // any not-yet-scheduled project IDs in memory so a retry can converge on the // same project instead of creating another one after a transient failure. const projectsAwaitingBootstrapSchedule = new Set(); + const closeCoordinatorOwnedSubsession: { + current?: (marker: { + projectId: string; + parentSessionId: string; + bindingId: string; + sessionId: string; + }) => Promise; + } = {}; const scheduleBootstrapProjects = async ( projectIds: Iterable, userId: string, @@ -1410,6 +1420,9 @@ export const startServer = async ( ingestCredentials, collectorUrl: options.collectorUrl, sessionsPath: options.sessionsPath ?? statePaths.sessions, + onSubsessionUserClosed: async (marker) => { + await closeCoordinatorOwnedSubsession.current?.(marker); + }, buildLaunchOpts, resolveAgentMapIdentity: async (sessionId, cwd, persisted) => { const userId = localProjectPrincipal(projectUserId, machineId); @@ -3156,6 +3169,49 @@ export const startServer = async ( }, }, ); + const emitSubsessionEvent = ( + event: SubsessionCoordinatorEvent | SubsessionCoordinatorStoreEvent, + ): void => { + const eventSessionId = + "sessionId" in event && event.sessionId + ? event.sessionId + : `subsession-${event.projectId}`; + const analyticsEvent: AnalyticsEvent = { + eventId: randomUUID(), + seq: seqCounter.next(eventSessionId), + ts: new Date().toISOString(), + userId: identity?.userId ?? null, + tenantId: identity?.tenantId ?? null, + machineId, + harnessSessionId: eventSessionId, + agentSessionId: null, + harness: sessionManager.get(eventSessionId)?.harness ?? "claude-code", + type: event.name, + payload: { + project_id: event.projectId, + ...("count" in event && event.count !== undefined + ? { count: Math.max(0, Math.min(16, event.count)) } + : {}), + ...("code" in event && event.code ? { error_code: event.code } : {}), + }, + }; + void eventStore.append(analyticsEvent).catch(() => {}); + batcher.enqueue(analyticsEvent); + }; + const subsessionCoordinatorStore = new SubsessionCoordinatorStore( + statePaths.agentMap, + { onEvent: emitSubsessionEvent }, + ); + closeCoordinatorOwnedSubsession.current = async (marker) => { + await subsessionCoordinatorStore.closeOwnedBinding(marker); + }; + const subsessionCoordinator = new SubsessionCoordinator({ + store: subsessionCoordinatorStore, + sessionManager, + planningStore: buildPlanStore, + eventReader: eventStore, + onEvent: emitSubsessionEvent, + }); emitAgentMapCapabilityEvent = (event) => { const analyticsEvent: AnalyticsEvent = { eventId: randomUUID(), @@ -3181,6 +3237,7 @@ export const startServer = async ( service: agentMapProposalService, buildPlanService, agentBriefService, + subsessionCoordinator, readSnapshotFor: async ({ projectId }) => { const project = await studioProjectCatalog.resolve(projectId); if (!project) throw new AgentMapMcpProjectUnavailableError(); @@ -4123,6 +4180,9 @@ export const startServer = async ( void projectBootstrap!.onEventPersisted(event, runtimeEpoch).catch(() => { console.error("[harness] project bootstrap completion failed"); }); + void subsessionCoordinator.onEventPersisted(event, runtimeEpoch).catch(() => { + console.error("[harness] subsession acknowledgement failed"); + }); const recordChanged = sessionRecordChangedMessage(event); if (recordChanged) bus.publish(recordChanged); // The normal end of a session: the SessionEnd hook's event is in the diff --git a/packages/harness/src/shared/agent-map-legacy-migration.test.ts b/packages/harness/src/shared/agent-map-legacy-migration.test.ts index 2bda2d4f..eb0c81de 100644 --- a/packages/harness/src/shared/agent-map-legacy-migration.test.ts +++ b/packages/harness/src/shared/agent-map-legacy-migration.test.ts @@ -47,6 +47,7 @@ describe("deployed E2 actor migration isolation", () => { "agent-map-proposal-service.ts", "agent-map-version.ts", "build-plan-service.ts", + "subsession-coordinator.ts", ]) { await expect(readFile(join(core, live), "utf8")).resolves.not.toContain( "parseLegacyE2ProposalActor", diff --git a/packages/harness/src/shared/types.ts b/packages/harness/src/shared/types.ts index 36a40805..dfe17eed 100644 --- a/packages/harness/src/shared/types.ts +++ b/packages/harness/src/shared/types.ts @@ -832,6 +832,22 @@ export type AnalyticsEventType = | "agent_map.capability" | "build_plan.operation" | "agent_brief.refresh" + | "subsession.store_initialized" + | "subsession.binding_reserved" + | "subsession.duplicate_prevented" + | "subsession.spawn_claimed" + | "subsession.requested" + | "subsession.created" + | "subsession.reused" + | "subsession.released" + | "subsession.ready" + | "subsession.failed" + | "subsession.kickoff_claimed" + | "subsession.kickoff_submitted" + | "subsession.kickoff_acknowledged" + | "subsession.kickoff_uncertain" + | "subsession.context_stale" + | "subsession.manual_session_protected" | "project_agent.identity_migrated" | "project_agent.identity_rejected" | "project_bootstrap.scheduled" diff --git a/packages/harness/web/e2e/subsession-delegation.spec.ts b/packages/harness/web/e2e/subsession-delegation.spec.ts new file mode 100644 index 00000000..89d909bf --- /dev/null +++ b/packages/harness/web/e2e/subsession-delegation.spec.ts @@ -0,0 +1,69 @@ +import { expect, test } from "@playwright/test"; + +test("delegation retries project one ordinary tab per real session without touching manual tabs", async ({ + page, +}) => { + await page.goto("/?seed=0&mockStudioProjects=present"); + await expect(page.locator(".rail-workflows")).toBeVisible(); + await page.getByTestId("project-select-acme-app").click(); + await expect(page.getByTestId("agent-map-frame")).toBeVisible(); + const projectId = "project_00000000-0000-4000-8000-000000000001"; + const tabs = page.getByRole("tablist", { name: "Sessions" }).getByRole("tab"); + const manualTabCount = await tabs.count(); + + await page.evaluate((selectedProjectId) => { + const publish = ( + window as unknown as { + __HARNESS_TEST__?: { + publish?: (message: Record) => void; + }; + } + ).__HARNESS_TEST__?.publish; + const session = (id: string, title: string) => ({ + id, + agentSessionId: null, + harness: "claude-code" as const, + cwd: "/Users/demo/acme-app", + boundWorkflowPath: null, + title, + status: "running" as const, + exitCode: null, + ready: true, + createdAt: "2026-09-04T10:00:00.000Z", + lastActiveAt: "2026-09-04T10:00:00.000Z", + agentMapIdentity: { + projectId: selectedProjectId, + userId: "user_mock", + sessionId: id, + }, + }); + publish?.({ + type: "session.status", + session: session("sess-delegated", "Focused research"), + }); + // A coordinator replay/status refresh carries the same real Harness ID. + publish?.({ + type: "session.status", + session: session("sess-delegated", "Focused research"), + }); + }, projectId); + + await expect(tabs).toHaveCount(manualTabCount + 1); + await expect(page.getByTestId("session-tab-sess-delegated")).toHaveCount(1); + await expect(page.getByTestId("session-tab-sess-boot")).toHaveCount(1); + + await page + .getByTestId("session-tab-sess-delegated") + .getByRole("tab") + .click(); + await expect(page.getByTestId("session-context")).toHaveAttribute( + "data-session-id", + "sess-delegated", + ); + await expect(page.getByTestId("agent-map-frame")).toHaveCount(0); + await expect(page.locator(".harness-terminal")).toBeVisible(); + + await page.getByTestId("project-select-acme-app").click(); + await expect(page.getByTestId("agent-map-frame")).toBeVisible(); + await expect(page.getByTestId("session-tab-sess-boot")).toHaveCount(1); +});