diff --git a/.changeset/shared-proposals-persist.md b/.changeset/shared-proposals-persist.md new file mode 100644 index 00000000..bf8f707c --- /dev/null +++ b/.changeset/shared-proposals-persist.md @@ -0,0 +1,5 @@ +--- +"@sapiom/harness": minor +--- + +Prepare crash-atomic, project-wide Agent Map proposal persistence for the SAP-3060 transport, with attributed operation history, bounded session-scoped idempotency receipts, and history-derived stale-write rebasing. Exact results are retained for a bounded retry window, while older same-session request IDs cannot apply twice. Agent Map workspace reads now return a coherent versioned workspace-and-proposal snapshot, and the named browser-safe contracts required by the accepted-delta bus payload are exported from the package entry point. diff --git a/packages/harness/package.json b/packages/harness/package.json index 877b6e7c..1662b4be 100644 --- a/packages/harness/package.json +++ b/packages/harness/package.json @@ -73,6 +73,7 @@ "node-pty": "^1.1.0", "open": "^10.1.0", "typescript": "~5.9.3", + "uuid": "^10.0.0", "ws": "^8.18.0", "zod": "^3.25.0" }, @@ -84,6 +85,7 @@ "@types/node": "^20.11.30", "@types/react": "^19.0.0", "@types/react-dom": "^19.0.0", + "@types/uuid": "^10.0.0", "@types/ws": "^8.5.10", "@vitejs/plugin-react": "^4.3.4", "@xterm/addon-fit": "^0.10.0", diff --git a/packages/harness/src/core/agent-map-proposal-schema.ts b/packages/harness/src/core/agent-map-proposal-schema.ts index 17953213..3e7740ea 100644 --- a/packages/harness/src/core/agent-map-proposal-schema.ts +++ b/packages/harness/src/core/agent-map-proposal-schema.ts @@ -14,27 +14,21 @@ import { type ProposalValidationIssue, type ProposalValidationResult, } from "../shared/agent-map.js"; - -const UUID_V7 = - "[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}"; -const hasControlCharacter = (value: string): boolean => - [...value].some((character) => { - const codePoint = character.codePointAt(0) ?? 0; - return codePoint <= 0x1f || codePoint === 0x7f; - }); +import { + AGENT_MAP_UUID_V7_PATTERN, + isAgentMapBoundedText, +} from "../shared/agent-map-codec.js"; const boundedText = (maximum: number, allowEmpty = false) => z .string() .max(maximum) - .refine((value) => (allowEmpty ? true : value.length > 0)) - .refine((value) => value.trim() === value) - .refine((value) => !hasControlCharacter(value)); + .refine((value) => isAgentMapBoundedText(value, maximum, allowEmpty)); const opaqueId = (prefix: string) => z .string() - .regex(new RegExp(`^${prefix}_${UUID_V7}$`, "u")) + .regex(new RegExp(`^${prefix}_${AGENT_MAP_UUID_V7_PATTERN}$`, "u")) .transform((value) => value as T); export const planNodeIdSchema = opaqueId("node"); diff --git a/packages/harness/src/core/agent-map-proposal-service.test.ts b/packages/harness/src/core/agent-map-proposal-service.test.ts new file mode 100644 index 00000000..fbfb9398 --- /dev/null +++ b/packages/harness/src/core/agent-map-proposal-service.test.ts @@ -0,0 +1,533 @@ +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 { + DraftRef, + MapProposalId, + PlanNodeId, + PlanRelationshipId, + PlanningSessionIdentity, + ProposalBatchRequest, + ProposalOperationId, +} from "../shared/agent-map.js"; +import { + AgentMapProposalService, + AgentMapProposalValidationError, + type AgentMapPermanentIdAllocator, +} from "./agent-map-proposal-service.js"; +import { AgentMapWorkspaceStore } from "./agent-map-workspace-store.js"; + +const projectId = "project_00000000-0000-4000-8000-000000000001"; + +class Ids implements AgentMapPermanentIdAllocator { + private value = 1; + private next(prefix: string) { + return `${prefix}_00000000-0000-7000-8000-${String(this.value++).padStart(12, "0")}`; + } + allocateNodeId = () => this.next("node") as PlanNodeId; + allocateRelationshipId = () => this.next("rel") as PlanRelationshipId; + allocateProposalId = () => this.next("proposal") as MapProposalId; + allocateOperationId = () => this.next("operation") as ProposalOperationId; +} + +const identity = (sessionId: string): PlanningSessionIdentity => ({ + projectId, + userId: "user-1", + sessionId, + role: "map-planner", +}); + +const addNode = ( + requestId: string, + expectedVersion: number, + proposalId: MapProposalId | null, + draftRef = requestId, +): ProposalBatchRequest => ({ + schemaVersion: 1, + proposalId, + expectedVersion, + requestId, + operations: [ + { + kind: "add-node", + draftRef: draftRef as DraftRef, + node: { + kind: "agent", + name: draftRef, + purpose: "Research", + ownerAgent: null, + contractRefs: [], + }, + }, + ], +}); + +describe("AgentMapProposalService", () => { + const roots: string[] = []; + afterEach(async () => + Promise.all( + roots + .splice(0) + .map((root) => fs.rm(root, { recursive: true, force: true })), + ), + ); + + async function fixture(receiptRetentionLimit?: number) { + const root = await fs.mkdtemp( + path.join(os.tmpdir(), "agent-map-proposal-"), + ); + roots.push(root); + const accepted = vi.fn(); + const outcomes = vi.fn(); + return { + root, + accepted, + outcomes, + service: new AgentMapProposalService(new AgentMapWorkspaceStore(root), { + allocator: new Ids(), + now: () => new Date("2026-09-02T12:00:00.000Z"), + onAccepted: accepted, + onOutcome: outcomes, + ...(receiptRetentionLimit === undefined + ? {} + : { receiptRetentionLimit }), + }), + }; + } + + it("atomically persists one attributed proposal and survives restart", async () => { + const { root, service, accepted } = await fixture(); + const result = await service.propose( + identity("session-1"), + addNode("request-1", 0, null), + ); + const restarted = new AgentMapProposalService( + new AgentMapWorkspaceStore(root), + ); + const snapshot = await restarted.read(projectId); + + expect(snapshot.workspace).toMatchObject({ + recordVersion: 2, + activeProposalId: result.proposalId, + }); + expect(snapshot.proposal).toMatchObject({ + id: result.proposalId, + version: 1, + }); + expect(snapshot.proposal?.history[0]?.actor).toEqual({ + userId: "user-1", + sessionId: "session-1", + role: "map-planner", + assignment: null, + }); + expect(accepted).toHaveBeenCalledOnce(); + }); + + it("returns the durable original receipt without duplicating or rebroadcasting", async () => { + const { service, accepted, outcomes } = await fixture(); + const request = addNode("request-1", 0, null); + if (request.operations[0]?.kind === "add-node") + request.operations[0].node.contractRefs = ["z-contract", "a-contract"]; + const first = await service.propose(identity("session-1"), request); + const reordered = structuredClone(request); + if (reordered.operations[0]?.kind === "add-node") + reordered.operations[0].node.contractRefs.reverse(); + await expect( + service.propose(identity("session-1"), reordered), + ).resolves.toEqual(first); + expect((await service.read(projectId)).proposal?.history).toHaveLength(1); + expect(accepted).toHaveBeenCalledOnce(); + expect(outcomes.mock.calls.map(([event]) => event.name)).toEqual([ + "agent_map.proposal.accepted", + "agent_map.proposal.replayed", + ]); + expect(Object.keys(outcomes.mock.calls[0]![0]).sort()).toEqual([ + "latencyMs", + "name", + "operationCount", + "projectId", + "role", + "sessionId", + ]); + }); + + it("bounds compact receipts and fails closed after exact replay retention", async () => { + const { root, service, accepted } = await fixture(1); + const firstRequest = addNode("request-1", 0, null); + const first = await service.propose(identity("session-1"), firstRequest); + const secondRequest = addNode("request-2", 1, first.proposalId); + const second = await service.propose(identity("session-1"), secondRequest); + const aggregate = await new AgentMapWorkspaceStore(root).readAggregate( + projectId, + ); + + expect(aggregate.receipts).toEqual([ + expect.objectContaining({ + sessionId: "session-1", + requestId: "request-2", + version: 2, + }), + ]); + expect(JSON.stringify(aggregate.receipts)).not.toContain('"delta"'); + expect(JSON.stringify(aggregate.receipts)).not.toContain('"touchSet"'); + await expect( + service.propose(identity("session-1"), firstRequest), + ).rejects.toMatchObject({ + conflict: { code: "request_id_expired", recovery: "new_request" }, + }); + await expect( + service.propose(identity("session-1"), secondRequest), + ).resolves.toEqual(second); + expect(accepted).toHaveBeenCalledTimes(2); + expect((await service.read(projectId)).proposal?.version).toBe(2); + }); + + it("rejects actor identities that the durable codec cannot read", async () => { + const { service } = await fixture(); + await expect( + service.propose( + identity("session\u007f1"), + addNode("request-1", 0, null), + ), + ).rejects.toBeInstanceOf(AgentMapProposalValidationError); + expect(await service.read(projectId)).toMatchObject({ proposal: null }); + }); + + it("rejects changed reuse of a session request ID", async () => { + const { service } = await fixture(1); + await service.propose(identity("session-1"), addNode("request-1", 0, null)); + await expect( + service.propose( + identity("session-1"), + addNode("request-1", 0, null, "different"), + ), + ).rejects.toMatchObject({ + conflict: { code: "request_id_reused", recovery: "new_request" }, + }); + }); + + it("rebases disjoint stale additions and rejects overlapping stale edits", async () => { + const { service } = await fixture(1); + const first = await service.propose( + identity("session-1"), + addNode("request-1", 0, null), + ); + const second = await service.propose( + identity("session-1"), + addNode("request-2", 1, first.proposalId), + ); + await service.propose( + identity("session-2"), + addNode("request-3", 1, first.proposalId), + ); + const nodeId = second.allocatedNodeIds["request-2" as DraftRef]!; + const edit = (session: string, name: string): ProposalBatchRequest => ({ + schemaVersion: 1, + proposalId: first.proposalId, + expectedVersion: 3, + requestId: session, + operations: [{ kind: "update-node", nodeId, changes: { name } }], + }); + await service.propose(identity("session-1"), edit("edit-1", "One")); + await expect( + service.validate(identity("session-2"), edit("edit-2", "Two")), + ).rejects.toMatchObject({ conflict: { code: "stale_version" } }); + await expect( + service.propose(identity("session-2"), edit("edit-2", "Two")), + ).rejects.toMatchObject({ + conflict: { + code: "stale_version", + currentVersion: 4, + affectedNodeIds: [nodeId], + }, + }); + expect((await service.read(projectId)).proposal?.nodes).toHaveLength(3); + }); + + it("derives stale conflicts from history after the conflicting receipt is pruned", async () => { + const { service } = await fixture(1); + const first = await service.propose( + identity("session-1"), + addNode("request-1", 0, null), + ); + const nodeId = first.allocatedNodeIds["request-1" as DraftRef]!; + await service.propose(identity("session-1"), { + schemaVersion: 1, + proposalId: first.proposalId, + expectedVersion: 1, + requestId: "edit", + operations: [ + { kind: "update-node", nodeId, changes: { name: "Changed" } }, + ], + }); + await service.propose( + identity("session-1"), + addNode("disjoint", 2, first.proposalId), + ); + + await expect( + service.propose(identity("session-2"), { + schemaVersion: 1, + proposalId: first.proposalId, + expectedVersion: 1, + requestId: "stale-edit", + operations: [ + { kind: "update-node", nodeId, changes: { purpose: "Stale" } }, + ], + }), + ).rejects.toMatchObject({ + conflict: { code: "stale_version", affectedNodeIds: [nodeId] }, + }); + }); + + it("commits nothing when validation fails", async () => { + const { service, accepted } = await fixture(); + const request = addNode("request-1", 0, null); + request.operations.push(structuredClone(request.operations[0]!)); + await expect( + service.propose(identity("session-1"), request), + ).rejects.toBeInstanceOf(AgentMapProposalValidationError); + expect(await service.read(projectId)).toMatchObject({ + proposal: null, + workspace: { recordVersion: 1 }, + }); + expect(accepted).not.toHaveBeenCalled(); + }); + + it("selects one first writer across independent service instances", async () => { + const { root } = await fixture(); + const left = new AgentMapProposalService(new AgentMapWorkspaceStore(root), { + allocator: new Ids(), + }); + const right = new AgentMapProposalService( + new AgentMapWorkspaceStore(root), + { + allocator: new Ids(), + }, + ); + const outcomes = await Promise.allSettled([ + left.propose(identity("session-left"), addNode("left", 0, null)), + right.propose(identity("session-right"), addNode("right", 0, null)), + ]); + expect( + outcomes.filter((outcome) => outcome.status === "fulfilled"), + ).toHaveLength(1); + expect( + outcomes.filter((outcome) => outcome.status === "rejected"), + ).toHaveLength(1); + expect((await left.read(projectId)).proposal).toMatchObject({ + version: 1, + nodes: [expect.any(Object)], + }); + }); + + it("uses the same write path for planner, assigned, and unplanned builders", async () => { + const { service } = await fixture(); + const first = await service.propose( + identity("planner"), + addNode("planner", 0, null), + ); + const assigned: PlanningSessionIdentity = { + projectId, + userId: "user-1", + sessionId: "assigned", + role: "agent-builder", + assignment: { kind: "planned", agentId: "planned-agent" }, + }; + await service.propose(assigned, addNode("assigned", 1, first.proposalId)); + const unplanned: PlanningSessionIdentity = { + projectId, + userId: "user-1", + sessionId: "unplanned", + role: "agent-builder", + assignment: { kind: "unplanned" }, + }; + await service.propose(unplanned, addNode("unplanned", 2, first.proposalId)); + expect( + (await service.read(projectId)).proposal?.history.map( + ({ actor }) => actor, + ), + ).toEqual([ + { + userId: "user-1", + sessionId: "planner", + role: "map-planner", + assignment: null, + }, + { + userId: "user-1", + sessionId: "assigned", + role: "agent-builder", + assignment: { kind: "planned", agentId: "planned-agent" }, + }, + { + userId: "user-1", + sessionId: "unplanned", + role: "agent-builder", + assignment: { kind: "unplanned" }, + }, + ]); + }); + + it("rejects semantic-edge and delete/update stale conflicts", async () => { + const { service } = await fixture(); + const initial = await service.propose(identity("planner"), { + schemaVersion: 1, + proposalId: null, + expectedVersion: 0, + requestId: "initial", + operations: [ + { + kind: "add-node", + draftRef: "a" as DraftRef, + node: { + kind: "agent", + name: "A", + purpose: "A", + ownerAgent: null, + contractRefs: [], + }, + }, + { + kind: "add-node", + draftRef: "b" as DraftRef, + node: { + kind: "agent", + name: "B", + purpose: "B", + ownerAgent: null, + contractRefs: [], + }, + }, + ], + }); + const relationship = (requestId: string): ProposalBatchRequest => ({ + schemaVersion: 1, + proposalId: initial.proposalId, + expectedVersion: 1, + requestId, + operations: [ + { + kind: "add-relationship", + draftRef: requestId as DraftRef, + relationship: { + from: { nodeId: initial.allocatedNodeIds["a" as DraftRef]! }, + to: { nodeId: initial.allocatedNodeIds["b" as DraftRef]! }, + kind: "invokes", + executionMode: "synchronous", + contractRef: null, + description: requestId, + }, + }, + ], + }); + await service.propose(identity("one"), relationship("edge-one")); + await expect( + service.propose(identity("two"), relationship("edge-two")), + ).rejects.toMatchObject({ conflict: { code: "stale_version" } }); + + const current = (await service.read(projectId)).proposal!; + const nodeId = initial.allocatedNodeIds["a" as DraftRef]!; + await service.propose(identity("one"), { + schemaVersion: 1, + proposalId: initial.proposalId, + expectedVersion: current.version, + requestId: "delete", + operations: [ + { + kind: "remove-relationship", + relationshipId: current.relationships[0]!.id, + }, + { kind: "remove-node", nodeId }, + ], + }); + await expect( + service.propose(identity("two"), { + schemaVersion: 1, + proposalId: initial.proposalId, + expectedVersion: current.version, + requestId: "stale-update", + operations: [ + { kind: "update-node", nodeId, changes: { name: "Changed" } }, + ], + }), + ).rejects.toMatchObject({ + conflict: { code: "stale_version", affectedNodeIds: [nodeId] }, + }); + }); + + it("does not advance durable state when allocation fails", async () => { + const root = await fs.mkdtemp( + path.join(os.tmpdir(), "agent-map-proposal-"), + ); + roots.push(root); + const allocator = new Ids(); + allocator.allocateNodeId = vi.fn(() => { + throw new Error("allocator unavailable"); + }); + const service = new AgentMapProposalService( + new AgentMapWorkspaceStore(root), + { allocator }, + ); + await expect( + service.propose(identity("session-1"), addNode("request-1", 0, null)), + ).rejects.toThrow("allocator unavailable"); + expect(await service.read(projectId)).toMatchObject({ + proposal: null, + workspace: { recordVersion: 1 }, + }); + }); + + it("rejects an injected allocator collision with existing proposal state", async () => { + const { root, service } = await fixture(); + const first = await service.propose( + identity("session-1"), + addNode("request-1", 0, null), + ); + const existingNodeId = first.allocatedNodeIds["request-1" as DraftRef]!; + const allocator = new Ids(); + allocator.allocateNodeId = () => existingNodeId; + const colliding = new AgentMapProposalService( + new AgentMapWorkspaceStore(root), + { allocator }, + ); + await expect( + colliding.propose( + identity("session-2"), + addNode("request-2", 1, first.proposalId), + ), + ).rejects.toThrow("duplicate node ID"); + expect((await service.read(projectId)).proposal?.version).toBe(1); + }); + + it("fails closed when a confirmed base revision cannot be supplied", async () => { + const root = await fs.mkdtemp( + path.join(os.tmpdir(), "agent-map-proposal-"), + ); + roots.push(root); + const file = path.join(root, "projects", projectId, "workspace.json"); + await fs.mkdir(path.dirname(file), { recursive: true }); + await fs.writeFile( + file, + `${JSON.stringify({ + projectId, + schemaVersion: 1, + recordVersion: 2, + confirmedRevisionId: "revision-1", + activeProposalId: null, + projectBuildPlanId: null, + createdAt: "2026-09-02T12:00:00.000Z", + updatedAt: "2026-09-02T12:00:00.000Z", + })}\n`, + ); + const service = new AgentMapProposalService( + new AgentMapWorkspaceStore(root), + ); + await expect( + service.propose(identity("session-1"), addNode("request-1", 0, null)), + ).rejects.toMatchObject({ code: "validation_failed" }); + expect(await service.read(projectId)).toMatchObject({ proposal: null }); + }); +}); diff --git a/packages/harness/src/core/agent-map-proposal-service.ts b/packages/harness/src/core/agent-map-proposal-service.ts new file mode 100644 index 00000000..10308a73 --- /dev/null +++ b/packages/harness/src/core/agent-map-proposal-service.ts @@ -0,0 +1,726 @@ +import { createHash } from "node:crypto"; +import { v7 as uuidv7 } from "uuid"; + +import { + AGENT_MAP_PROPOSAL_SCHEMA_VERSION, + type AcceptedProposalDelta, + type AgentMapGraph, + type MapChangeProposal, + type MapOperation, + type MapProposalId, + type PlanNodeId, + type PlanRelationshipId, + type PlanningSessionIdentity, + type ProposalActor, + type ProposalBatchRequest, + type ProposalBatchResult, + type ProposalConflict, + type ProposalOperationId, + type ProposalValidationIssue, + type StudioProjectId, +} from "../shared/agent-map.js"; +import { parseProposalActor } from "../shared/agent-map-codec.js"; +import { parseProposalBatchRequest } from "./agent-map-proposal-schema.js"; +import { + canonicalizeAgentMapGraph, + derivePersistedMapOperationTouchSet, + materializeValidatedMapBatch, + proposalTouchSetsOverlap, + validateMapOperationBatch, + type AgentMapIdAllocator, + type ProposalTouchSet, +} from "./agent-map-proposal-validator.js"; +import { + AgentMapWorkspaceStore, + AgentMapWorkspaceStoreError, + type AgentMapProposalReceipt, + type AgentMapProjectAggregate, +} from "./agent-map-workspace-store.js"; + +export const AGENT_MAP_PROPOSAL_RECEIPT_RETENTION_LIMIT = 256; + +export class AgentMapProposalValidationError extends Error { + readonly code = "validation_failed" as const; + constructor( + readonly issues: ProposalValidationIssue[], + readonly currentVersion: number, + ) { + super("Agent Map proposal batch is invalid"); + this.name = "AgentMapProposalValidationError"; + } +} + +export class AgentMapProposalConflictError extends Error { + constructor(readonly conflict: ProposalConflict) { + super( + conflict.code === "request_id_reused" + ? "Proposal request ID was reused" + : conflict.code === "request_id_expired" + ? "Proposal request result is no longer retained" + : "Agent Map proposal changed", + ); + this.name = "AgentMapProposalConflictError"; + } +} + +export class AgentMapProposalProjectError extends Error { + readonly code = "cross_project" as const; + constructor() { + super("Proposal identity does not belong to this project"); + this.name = "AgentMapProposalProjectError"; + } +} + +export interface AgentMapPermanentIdAllocator extends AgentMapIdAllocator { + allocateProposalId(): MapProposalId; + allocateOperationId(): ProposalOperationId; +} + +export class UuidV7AgentMapIdAllocator implements AgentMapPermanentIdAllocator { + allocateNodeId = (): PlanNodeId => `node_${uuidv7()}` as PlanNodeId; + allocateRelationshipId = (): PlanRelationshipId => + `rel_${uuidv7()}` as PlanRelationshipId; + allocateProposalId = (): MapProposalId => + `proposal_${uuidv7()}` as MapProposalId; + allocateOperationId = (): ProposalOperationId => + `operation_${uuidv7()}` as ProposalOperationId; +} + +export interface AgentMapProposalServiceOptions { + allocator?: AgentMapPermanentIdAllocator; + now?: () => Date; + readBaseRevision?: ( + projectId: StudioProjectId, + revisionId: string, + ) => Promise; + onAccepted?: (delta: AcceptedProposalDelta) => void | Promise; + onOutcome?: (event: { + name: + | "agent_map.proposal.accepted" + | "agent_map.proposal.replayed" + | "agent_map.proposal.validation_failed" + | "agent_map.proposal.conflict" + | "agent_map.proposal.storage_failed"; + projectId: StudioProjectId; + sessionId: string; + role: PlanningSessionIdentity["role"]; + operationCount: number; + latencyMs: number; + }) => void | Promise; + /** Test seam; production receipts stay bounded by the exported hard limit. */ + receiptRetentionLimit?: number; +} + +const actorFor = (identity: PlanningSessionIdentity): ProposalActor => { + try { + return parseProposalActor({ + userId: identity.userId, + sessionId: identity.sessionId, + role: identity.role, + assignment: + identity.role === "agent-builder" + ? structuredClone(identity.assignment) + : null, + }); + } catch { + throw new AgentMapProposalValidationError( + [ + { + code: "malformed_input", + operationIndex: null, + path: ["identity"], + recovery: "retry", + }, + ], + 0, + ); + } +}; + +function canonicalRequest(request: ProposalBatchRequest): ProposalBatchRequest { + return { + ...request, + operations: request.operations.map((operation) => { + if (operation.kind === "add-node") + return { + ...operation, + node: { + ...operation.node, + contractRefs: [...operation.node.contractRefs].sort(), + }, + }; + if (operation.kind === "update-node") + return { + ...operation, + changes: { + ...operation.changes, + ...(operation.changes.contractRefs + ? { contractRefs: [...operation.changes.contractRefs].sort() } + : {}), + }, + }; + return operation; + }), + }; +} + +const requestDigest = (request: ProposalBatchRequest): string => + createHash("sha256") + .update(JSON.stringify(canonicalRequest(request))) + .digest("hex"); + +function applyOperations( + graph: AgentMapGraph, + operations: readonly MapOperation[], +): AgentMapGraph { + const nodes = new Map( + graph.nodes.map((node) => [node.id, structuredClone(node)]), + ); + const relationships = new Map( + graph.relationships.map((relationship) => [ + relationship.id, + structuredClone(relationship), + ]), + ); + for (const operation of operations) { + switch (operation.kind) { + case "add-node": + nodes.set(operation.node.id, structuredClone(operation.node)); + break; + case "update-node": { + const node = nodes.get(operation.nodeId); + if (node) + nodes.set(operation.nodeId, { + ...node, + ...structuredClone(operation.changes), + }); + break; + } + case "remove-node": + nodes.delete(operation.nodeId); + break; + case "add-relationship": + relationships.set( + operation.relationship.id, + structuredClone(operation.relationship), + ); + break; + case "update-relationship": { + const relationship = relationships.get(operation.relationshipId); + if (relationship) + relationships.set(operation.relationshipId, { + ...relationship, + ...structuredClone(operation.changes), + }); + break; + } + case "remove-relationship": + relationships.delete(operation.relationshipId); + break; + } + } + return canonicalizeAgentMapGraph({ + nodes: [...nodes.values()], + relationships: [...relationships.values()], + }); +} + +function affectedFromTouchSets( + left: ProposalTouchSet, + right: ProposalTouchSet, +): Pick { + const entities = new Set(right.entityKeys); + return { + affectedNodeIds: left.entityKeys + .filter((key) => key.startsWith("node:") && entities.has(key)) + .map((key) => key.slice(5) as PlanNodeId), + affectedRelationshipIds: left.entityKeys + .filter((key) => key.startsWith("relationship:") && entities.has(key)) + .map((key) => key.slice(13) as PlanRelationshipId), + }; +} + +/** Transport-neutral authority for the one shared active proposal per project. */ +export class AgentMapProposalService { + private readonly allocator: AgentMapPermanentIdAllocator; + private readonly now: () => Date; + private readonly receiptRetentionLimit: number; + + constructor( + private readonly store: AgentMapWorkspaceStore, + private readonly options: AgentMapProposalServiceOptions = {}, + ) { + this.allocator = options.allocator ?? new UuidV7AgentMapIdAllocator(); + this.now = options.now ?? (() => new Date()); + const requestedLimit = + options.receiptRetentionLimit ?? + AGENT_MAP_PROPOSAL_RECEIPT_RETENTION_LIMIT; + if (!Number.isSafeInteger(requestedLimit) || requestedLimit < 1) + throw new RangeError("receiptRetentionLimit must be a positive integer"); + this.receiptRetentionLimit = Math.min( + requestedLimit, + AGENT_MAP_PROPOSAL_RECEIPT_RETENTION_LIMIT, + ); + } + + read(projectId: StudioProjectId) { + return this.store.readSnapshot(projectId); + } + + private async baseGraph( + aggregate: AgentMapProjectAggregate, + ): Promise { + const revisionId = aggregate.workspace.confirmedRevisionId; + if (revisionId === null) return { nodes: [], relationships: [] }; + const graph = await this.options.readBaseRevision?.( + aggregate.workspace.projectId, + revisionId, + ); + if (!graph) + throw new AgentMapProposalValidationError( + [ + { + code: "unknown_reference", + operationIndex: null, + path: ["baseRevisionId"], + recovery: "reread", + }, + ], + aggregate.proposal?.version ?? 0, + ); + return canonicalizeAgentMapGraph(graph); + } + + private graphAt( + base: AgentMapGraph, + proposal: MapChangeProposal | null, + version: number, + ): AgentMapGraph { + if (!proposal || version === 0) return base; + const operations: MapOperation[] = []; + for (const record of proposal.history) { + if (record.acceptedVersion > version) break; + operations.push(record.operation); + } + return applyOperations(base, operations); + } + + /** History is authoritative; receipt retention cannot change stale conflicts. */ + private touchSetAfter( + readGraph: AgentMapGraph, + proposal: MapChangeProposal | null, + expectedVersion: number, + ): ProposalTouchSet { + const entities = new Set(); + const semantics = new Set(); + if (!proposal || expectedVersion >= proposal.version) + return { entityKeys: [], semanticRelationshipKeys: [] }; + let graph = readGraph; + let version = -1; + let operations: MapOperation[] = []; + const applyBatch = () => { + if (operations.length === 0) return; + const next = applyOperations(graph, operations); + const touchSet = derivePersistedMapOperationTouchSet( + graph, + operations, + next, + ); + touchSet.entityKeys.forEach((key) => entities.add(key)); + touchSet.semanticRelationshipKeys.forEach((key) => semantics.add(key)); + graph = next; + }; + for (const record of proposal.history) { + if (record.acceptedVersion <= expectedVersion) continue; + if (version !== -1 && record.acceptedVersion !== version) { + applyBatch(); + operations = []; + } + version = record.acceptedVersion; + operations.push(record.operation); + } + applyBatch(); + return { + entityKeys: [...entities].sort(), + semanticRelationshipKeys: [...semantics].sort(), + }; + } + + private resultForReceipt( + proposal: MapChangeProposal, + receipt: AgentMapProposalReceipt, + ): ProposalBatchResult { + const records = proposal.history.filter( + ({ acceptedVersion }) => acceptedVersion === receipt.version, + ); + const first = records[0]!; + const operationIds = records.map(({ id }) => id); + return { + schemaVersion: AGENT_MAP_PROPOSAL_SCHEMA_VERSION, + proposalId: proposal.id, + version: receipt.version, + operationIds, + allocatedNodeIds: structuredClone(receipt.allocatedNodeIds), + allocatedRelationshipIds: structuredClone( + receipt.allocatedRelationshipIds, + ), + delta: { + schemaVersion: AGENT_MAP_PROPOSAL_SCHEMA_VERSION, + projectId: proposal.projectId, + proposalId: proposal.id, + fromVersion: receipt.version - 1, + version: receipt.version, + operationIds, + operations: records.map(({ operation }) => structuredClone(operation)), + actor: structuredClone(first.actor), + acceptedAt: first.acceptedAt, + }, + }; + } + + async validate(identity: PlanningSessionIdentity, input: unknown) { + actorFor(identity); + const parsed = parseProposalBatchRequest(input); + if (!parsed.ok) throw new AgentMapProposalValidationError(parsed.issues, 0); + const aggregate = await this.store.readAggregate(identity.projectId); + const currentVersion = aggregate.proposal?.version ?? 0; + this.assertProposalPointer(aggregate, parsed.value, currentVersion); + if (parsed.value.expectedVersion > currentVersion) + throw this.stale(currentVersion); + const base = await this.baseGraph(aggregate); + const readGraph = this.graphAt( + base, + aggregate.proposal, + parsed.value.expectedVersion, + ); + const atRead = validateMapOperationBatch(readGraph, parsed.value); + if (!atRead.ok) + throw new AgentMapProposalValidationError(atRead.issues, currentVersion); + if (parsed.value.expectedVersion < currentVersion) { + const prior = this.touchSetAfter( + readGraph, + aggregate.proposal, + parsed.value.expectedVersion, + ); + if (proposalTouchSetsOverlap(atRead.value.touchSet, prior)) + throw new AgentMapProposalConflictError({ + code: "stale_version", + currentVersion, + ...affectedFromTouchSets(atRead.value.touchSet, prior), + recovery: "reread", + }); + } + const currentGraph = aggregate.proposal + ? { + nodes: aggregate.proposal.nodes, + relationships: aggregate.proposal.relationships, + } + : base; + const validated = validateMapOperationBatch(currentGraph, parsed.value); + if (!validated.ok) { + if (parsed.value.expectedVersion < currentVersion) + throw this.stale(currentVersion); + throw new AgentMapProposalValidationError( + validated.issues, + currentVersion, + ); + } + return { + schemaVersion: 1 as const, + valid: true as const, + currentVersion, + touchSet: validated.value.touchSet, + }; + } + + async propose( + identity: PlanningSessionIdentity, + input: unknown, + ): Promise { + const startedAt = Date.now(); + const actor = actorFor(identity); + const parsed = parseProposalBatchRequest(input); + if (!parsed.ok) { + this.emitOutcome( + identity, + "agent_map.proposal.validation_failed", + 0, + startedAt, + ); + throw new AgentMapProposalValidationError(parsed.issues, 0); + } + const request = parsed.value; + let acceptedDelta: AcceptedProposalDelta | null = null; + let replayed = false; + let result: ProposalBatchResult; + try { + result = await this.store.transact( + identity.projectId, + async (aggregate) => { + if (aggregate.workspace.projectId !== identity.projectId) + throw new AgentMapProposalProjectError(); + const currentVersion = aggregate.proposal?.version ?? 0; + const digest = requestDigest(request); + const receipt = aggregate.receipts.find( + (candidate) => + candidate.sessionId === identity.sessionId && + candidate.requestId === request.requestId, + ); + if (receipt) { + if (receipt.requestDigest !== digest) + throw new AgentMapProposalConflictError({ + code: "request_id_reused", + currentVersion, + affectedNodeIds: [], + affectedRelationshipIds: [], + recovery: "new_request", + }); + replayed = true; + return { + value: this.resultForReceipt(aggregate.proposal!, receipt), + }; + } + if ( + aggregate.proposal?.history.some( + (record) => + record.actor.sessionId === identity.sessionId && + record.requestId === request.requestId, + ) + ) + // Exact results retain draftRef allocations only for the bounded + // retry window. History remains a permanent, compact tombstone: + // an older retry fails closed instead of applying twice. + throw new AgentMapProposalConflictError({ + code: "request_id_expired", + currentVersion, + affectedNodeIds: [], + affectedRelationshipIds: [], + recovery: "new_request", + }); + this.assertProposalPointer(aggregate, request, currentVersion); + if (request.expectedVersion > currentVersion) + throw this.stale(currentVersion); + + const base = await this.baseGraph(aggregate); + const readGraph = this.graphAt( + base, + aggregate.proposal, + request.expectedVersion, + ); + const atRead = validateMapOperationBatch(readGraph, request); + if (!atRead.ok) + throw new AgentMapProposalValidationError( + atRead.issues, + currentVersion, + ); + + if (request.expectedVersion < currentVersion) { + const prior = this.touchSetAfter( + readGraph, + aggregate.proposal, + request.expectedVersion, + ); + if (proposalTouchSetsOverlap(atRead.value.touchSet, prior)) + throw new AgentMapProposalConflictError({ + code: "stale_version", + currentVersion, + ...affectedFromTouchSets(atRead.value.touchSet, prior), + recovery: "reread", + }); + } + const currentGraph = aggregate.proposal + ? { + nodes: aggregate.proposal.nodes, + relationships: aggregate.proposal.relationships, + } + : base; + const rebased = validateMapOperationBatch(currentGraph, request); + if (!rebased.ok) { + if (request.expectedVersion < currentVersion) + throw this.stale(currentVersion); + throw new AgentMapProposalValidationError( + rebased.issues, + currentVersion, + ); + } + const materialized = materializeValidatedMapBatch( + rebased.value, + this.allocator, + ); + const proposalId = + aggregate.proposal?.id ?? this.allocator.allocateProposalId(); + const version = currentVersion + 1; + const operationIds = materialized.operations.map(() => + this.allocator.allocateOperationId(), + ); + const ids = [ + ...(aggregate.proposal ? [] : [proposalId]), + ...operationIds, + ...Object.values(materialized.allocatedNodeIds), + ...Object.values(materialized.allocatedRelationshipIds), + ]; + const existingIds = new Set([ + ...(aggregate.proposal ? [aggregate.proposal.id] : []), + ...(aggregate.proposal?.nodes.map(({ id }) => id) ?? []), + ...(aggregate.proposal?.relationships.map(({ id }) => id) ?? []), + ...(aggregate.proposal?.history.map(({ id }) => id) ?? []), + ]); + if ( + new Set(ids).size !== ids.length || + ids.some((id) => existingIds.has(id)) + ) + throw new AgentMapProposalValidationError( + [ + { + code: "malformed_input", + operationIndex: null, + path: ["allocator"], + recovery: "retry", + }, + ], + currentVersion, + ); + const acceptedAt = this.now().toISOString(); + const delta: AcceptedProposalDelta = { + schemaVersion: AGENT_MAP_PROPOSAL_SCHEMA_VERSION, + projectId: identity.projectId, + proposalId, + fromVersion: currentVersion, + version, + operationIds, + operations: materialized.operations, + actor, + acceptedAt, + }; + const batchResult: ProposalBatchResult = { + schemaVersion: AGENT_MAP_PROPOSAL_SCHEMA_VERSION, + proposalId, + version, + operationIds, + allocatedNodeIds: materialized.allocatedNodeIds, + allocatedRelationshipIds: materialized.allocatedRelationshipIds, + delta, + }; + const proposal: MapChangeProposal = { + schemaVersion: AGENT_MAP_PROPOSAL_SCHEMA_VERSION, + id: proposalId, + projectId: identity.projectId, + baseRevisionId: aggregate.workspace.confirmedRevisionId, + version, + nodes: materialized.graph.nodes, + relationships: materialized.graph.relationships, + history: [ + ...(aggregate.proposal?.history ?? []), + ...materialized.operations.map((operation, index) => ({ + id: operationIds[index]!, + requestId: request.requestId, + acceptedVersion: version, + operation, + actor, + acceptedAt, + })), + ], + createdAt: aggregate.proposal?.createdAt ?? acceptedAt, + updatedAt: acceptedAt, + }; + const next: AgentMapProjectAggregate = { + ...aggregate, + workspace: { + ...aggregate.workspace, + recordVersion: aggregate.workspace.recordVersion + 1, + activeProposalId: proposalId, + updatedAt: acceptedAt, + }, + proposal, + receipts: [ + ...aggregate.receipts, + { + sessionId: identity.sessionId, + requestId: request.requestId, + requestDigest: digest, + version, + allocatedNodeIds: materialized.allocatedNodeIds, + allocatedRelationshipIds: materialized.allocatedRelationshipIds, + }, + ].slice(-this.receiptRetentionLimit), + }; + acceptedDelta = delta; + return { value: batchResult, next }; + }, + ); + } catch (error) { + this.emitOutcome( + identity, + error instanceof AgentMapProposalConflictError + ? "agent_map.proposal.conflict" + : error instanceof AgentMapWorkspaceStoreError + ? "agent_map.proposal.storage_failed" + : "agent_map.proposal.validation_failed", + request.operations.length, + startedAt, + ); + throw error; + } + if (acceptedDelta) { + try { + await this.options.onAccepted?.(acceptedDelta); + } catch { + // Durable state is authoritative; subscribers recover by refetching. + } + } + this.emitOutcome( + identity, + replayed ? "agent_map.proposal.replayed" : "agent_map.proposal.accepted", + request.operations.length, + startedAt, + ); + return result; + } + + private emitOutcome( + identity: PlanningSessionIdentity, + name: Parameters< + NonNullable + >[0]["name"], + operationCount: number, + startedAt: number, + ): void { + try { + void Promise.resolve( + this.options.onOutcome?.({ + name, + projectId: identity.projectId, + sessionId: identity.sessionId, + role: identity.role, + operationCount, + latencyMs: Math.max(0, Date.now() - startedAt), + }), + ).catch(() => {}); + } catch { + // Content-free observability cannot change proposal semantics. + } + } + + private assertProposalPointer( + aggregate: AgentMapProjectAggregate, + request: ProposalBatchRequest, + currentVersion: number, + ): void { + const active = aggregate.proposal?.id ?? null; + if ( + request.proposalId !== active || + (active === null && request.expectedVersion !== 0) + ) + throw this.stale(currentVersion); + } + + private stale(currentVersion: number) { + return new AgentMapProposalConflictError({ + code: "stale_version", + currentVersion, + affectedNodeIds: [], + affectedRelationshipIds: [], + recovery: "reread", + }); + } +} diff --git a/packages/harness/src/core/agent-map-proposal-validator.ts b/packages/harness/src/core/agent-map-proposal-validator.ts index 6b8498f2..b7f5b6a1 100644 --- a/packages/harness/src/core/agent-map-proposal-validator.ts +++ b/packages/harness/src/core/agent-map-proposal-validator.ts @@ -344,7 +344,8 @@ export function proposalTouchSetsOverlap( ); } -function deriveMaterializedTouchSet( +/** Derive conservative conflict keys from already materialized operations. */ +export function derivePersistedMapOperationTouchSet( current: AgentMapGraph, operations: readonly MapOperation[], prospective: AgentMapGraph, @@ -940,6 +941,10 @@ export function materializeValidatedMapBatch( graph, allocatedNodeIds, allocatedRelationshipIds, - touchSet: deriveMaterializedTouchSet(validated.current, operations, graph), + touchSet: derivePersistedMapOperationTouchSet( + validated.current, + operations, + graph, + ), }; } diff --git a/packages/harness/src/core/agent-map-workspace-store.test.ts b/packages/harness/src/core/agent-map-workspace-store.test.ts index 8cd79b67..becef8a1 100644 --- a/packages/harness/src/core/agent-map-workspace-store.test.ts +++ b/packages/harness/src/core/agent-map-workspace-store.test.ts @@ -105,6 +105,88 @@ describe("AgentMapWorkspaceStore", () => { }); }); + it("migrates the exact E1 record into the aggregate without changing its public projection", async () => { + const root = await fixture(); + const workspacePath = path.join( + root, + "projects", + projectId, + "workspace.json", + ); + const workspace = { + projectId, + schemaVersion: 1, + recordVersion: 1, + confirmedRevisionId: null, + activeProposalId: null, + projectBuildPlanId: null, + createdAt: "2026-09-01T12:00:00.000Z", + updatedAt: "2026-09-01T12:00:00.000Z", + }; + await fs.mkdir(path.dirname(workspacePath), { recursive: true }); + await fs.writeFile(workspacePath, `${JSON.stringify(workspace)}\n`); + + await expect( + new AgentMapWorkspaceStore(root).readOrCreate(projectId), + ).resolves.toEqual(workspace); + expect(JSON.parse(await fs.readFile(workspacePath, "utf8"))).toEqual({ + storageSchemaVersion: 1, + workspace, + proposal: null, + receipts: [], + }); + }); + + it("rejects future aggregate schemas without rewriting them", async () => { + const root = await fixture(); + const workspacePath = path.join( + root, + "projects", + projectId, + "workspace.json", + ); + const raw = `${JSON.stringify({ storageSchemaVersion: 99, workspace: {}, proposal: null, receipts: [] })}\n`; + await fs.mkdir(path.dirname(workspacePath), { recursive: true }); + await fs.writeFile(workspacePath, raw); + await expect( + new AgentMapWorkspaceStore(root).readOrCreate(projectId), + ).rejects.toMatchObject({ + code: "unsupported_schema", + schemaVersion: 99, + }); + expect(await fs.readFile(workspacePath, "utf8")).toBe(raw); + }); + + it.each(["write", "file-sync", "rename", "directory-sync"] as const)( + "does not expose a partial aggregate when %s fails", + async (failedStep) => { + const root = await fixture(); + let fail = false; + const store = new AgentMapWorkspaceStore(root, { + beforePersistStep: (step) => { + if (fail && step === failedStep) throw new Error("injected failure"); + }, + }); + await store.readOrCreate(projectId); + fail = true; + await expect( + store.transact(projectId, async (aggregate) => ({ + value: undefined, + next: { + ...aggregate, + workspace: { ...aggregate.workspace, recordVersion: 2 }, + }, + })), + ).rejects.toMatchObject({ code: "storage_unavailable" }); + const restarted = await new AgentMapWorkspaceStore(root).readOrCreate( + projectId, + ); + expect(restarted.recordVersion).toBe( + failedStep === "directory-sync" ? 2 : 1, + ); + }, + ); + it.each([ ["malformed JSON", "{", "malformed_state"], [ diff --git a/packages/harness/src/core/agent-map-workspace-store.ts b/packages/harness/src/core/agent-map-workspace-store.ts index 64acf8bf..ef78d0e0 100644 --- a/packages/harness/src/core/agent-map-workspace-store.ts +++ b/packages/harness/src/core/agent-map-workspace-store.ts @@ -7,15 +7,35 @@ import { AGENT_MAP_WORKSPACE_SCHEMA_VERSION, type AgentMapErrorCode, type AgentMapWorkspaceState, + type MapChangeProposal, type StudioProjectId, } from "../shared/agent-map.js"; +import { + parseAgentMapProposalReceipt, + parseMapChangeProposal, + type PersistedAgentMapProposalReceipt, +} from "../shared/agent-map-codec.js"; +import { DurableFileLock } from "./durable-file-lock.js"; import { isStudioProjectId } from "./studio-project-catalog.js"; +export const AGENT_MAP_AGGREGATE_STORAGE_SCHEMA_VERSION = 1; + +export type AgentMapProposalReceipt = PersistedAgentMapProposalReceipt; + +export interface AgentMapProjectAggregate { + storageSchemaVersion: typeof AGENT_MAP_AGGREGATE_STORAGE_SCHEMA_VERSION; + workspace: AgentMapWorkspaceState; + proposal: MapChangeProposal | null; + receipts: AgentMapProposalReceipt[]; +} + +export interface AgentMapStoreSnapshot { + workspace: AgentMapWorkspaceState; + proposal: MapChangeProposal | null; +} + export type AgentMapWorkspaceStoreEvent = - | { - name: "agent_map.workspace_initialized"; - projectId: StudioProjectId; - } + | { name: "agent_map.workspace_initialized"; projectId: StudioProjectId } | { name: "agent_map.workspace_read_failed"; projectId: StudioProjectId; @@ -39,72 +59,58 @@ export class AgentMapWorkspaceStoreError extends Error { } } -function isRecord(value: unknown): value is Record { - return typeof value === "object" && value !== null && !Array.isArray(value); -} +const isRecord = (value: unknown): value is Record => + typeof value === "object" && value !== null && !Array.isArray(value); -function hasExactKeys( +const hasExactKeys = ( value: Record, keys: readonly string[], -): boolean { +) => { const actual = Object.keys(value).sort(); const expected = [...keys].sort(); return ( actual.length === expected.length && actual.every((key, index) => key === expected[index]) ); -} - -function hasControlCharacter(value: string): boolean { - return [...value].some((character) => { - const code = character.codePointAt(0)!; - return code <= 0x1f || (code >= 0x7f && code <= 0x9f); - }); -} - -function isOpaqueId(value: unknown): value is string { - return ( - typeof value === "string" && - value !== "" && - value === value.trim() && - !hasControlCharacter(value) && - !value.includes("/") && - !value.includes("\\") && - !value.includes(":") - ); -} +}; -function isNullableOpaqueId(value: unknown): value is string | null { - return value === null || isOpaqueId(value); -} - -function isTimestamp(value: unknown): value is string { +const isTimestamp = (value: unknown): value is string => { if (typeof value !== "string") return false; try { return new Date(value).toISOString() === value; } catch { return false; } -} +}; + +const isOpaqueId = (value: unknown): value is string => + typeof value === "string" && + value.length > 0 && + value === value.trim() && + !value.includes("/") && + !value.includes("\\") && + !value.includes(":") && + ![...value].some((character) => { + const codePoint = character.codePointAt(0) ?? 0; + return codePoint <= 0x1f || codePoint === 0x7f; + }); + +const nullableOpaqueId = (value: unknown): value is string | null => + value === null || isOpaqueId(value); export function parseAgentMapWorkspaceState( value: unknown, expectedProjectId: StudioProjectId, ): AgentMapWorkspaceState { - const readableSchemaVersion = - isRecord(value) && - Number.isSafeInteger(value.schemaVersion) && - (value.schemaVersion as number) >= 0 + const schemaVersion = + isRecord(value) && Number.isSafeInteger(value.schemaVersion) ? (value.schemaVersion as number) : undefined; if ( - readableSchemaVersion !== undefined && - readableSchemaVersion > AGENT_MAP_WORKSPACE_SCHEMA_VERSION + schemaVersion !== undefined && + schemaVersion > AGENT_MAP_WORKSPACE_SCHEMA_VERSION ) { - throw new AgentMapWorkspaceStoreError( - "unsupported_schema", - readableSchemaVersion, - ); + throw new AgentMapWorkspaceStoreError("unsupported_schema", schemaVersion); } if ( !isRecord(value) || @@ -120,60 +126,133 @@ export function parseAgentMapWorkspaceState( ]) || value.projectId !== expectedProjectId || !isStudioProjectId(value.projectId) || - !Number.isSafeInteger(value.schemaVersion) || + value.schemaVersion !== AGENT_MAP_WORKSPACE_SCHEMA_VERSION || !Number.isSafeInteger(value.recordVersion) || (value.recordVersion as number) < 1 || - !isNullableOpaqueId(value.confirmedRevisionId) || - !isNullableOpaqueId(value.activeProposalId) || - !isNullableOpaqueId(value.projectBuildPlanId) || + !nullableOpaqueId(value.confirmedRevisionId) || + !nullableOpaqueId(value.activeProposalId) || + !nullableOpaqueId(value.projectBuildPlanId) || !isTimestamp(value.createdAt) || !isTimestamp(value.updatedAt) - ) { + ) + throw new AgentMapWorkspaceStoreError("malformed_state", schemaVersion); + return value as unknown as AgentMapWorkspaceState; +} + +const storageError = () => + new AgentMapWorkspaceStoreError("storage_unavailable"); + +function parseAggregate( + value: unknown, + projectId: StudioProjectId, +): AgentMapProjectAggregate { + if ( + isRecord(value) && + Number.isSafeInteger(value.storageSchemaVersion) && + (value.storageSchemaVersion as number) > + AGENT_MAP_AGGREGATE_STORAGE_SCHEMA_VERSION + ) throw new AgentMapWorkspaceStoreError( - "malformed_state", - readableSchemaVersion, + "unsupported_schema", + value.storageSchemaVersion as number, ); + if ( + !isRecord(value) || + !hasExactKeys(value, [ + "storageSchemaVersion", + "workspace", + "proposal", + "receipts", + ]) || + value.storageSchemaVersion !== AGENT_MAP_AGGREGATE_STORAGE_SCHEMA_VERSION || + !Array.isArray(value.receipts) + ) + throw new AgentMapWorkspaceStoreError("malformed_state"); + const workspace = parseAgentMapWorkspaceState(value.workspace, projectId); + let proposal: MapChangeProposal | null = null; + if ((value.proposal === null) !== (workspace.activeProposalId === null)) + throw new AgentMapWorkspaceStoreError("malformed_state"); + if (value.proposal !== null && workspace.activeProposalId !== null) { + try { + proposal = parseMapChangeProposal( + value.proposal, + projectId, + workspace.activeProposalId, + ); + } catch { + throw new AgentMapWorkspaceStoreError("malformed_state"); + } } - if (value.schemaVersion !== AGENT_MAP_WORKSPACE_SCHEMA_VERSION) { - throw new AgentMapWorkspaceStoreError( - (value.schemaVersion as number) > AGENT_MAP_WORKSPACE_SCHEMA_VERSION - ? "unsupported_schema" - : "malformed_state", - value.schemaVersion as number, + const receipts: AgentMapProposalReceipt[] = []; + for (const receipt of value.receipts) { + let parsed: AgentMapProposalReceipt; + try { + parsed = parseAgentMapProposalReceipt(receipt); + } catch { + throw new AgentMapWorkspaceStoreError("malformed_state"); + } + const records = + proposal?.history.filter( + ({ acceptedVersion }) => acceptedVersion === parsed.version, + ) ?? []; + const actor = records[0]?.actor; + const acceptedAt = records[0]?.acceptedAt; + const allocatedNodeIds = records.flatMap(({ operation }) => + operation.kind === "add-node" ? [operation.node.id] : [], ); + const allocatedRelationshipIds = records.flatMap(({ operation }) => + operation.kind === "add-relationship" ? [operation.relationship.id] : [], + ); + if ( + proposal === null || + parsed.version > proposal.version || + records.length === 0 || + records.some( + (record) => + record.requestId !== parsed.requestId || + record.actor.sessionId !== parsed.sessionId || + JSON.stringify(record.actor) !== JSON.stringify(actor) || + record.acceptedAt !== acceptedAt, + ) || + JSON.stringify(Object.values(parsed.allocatedNodeIds).sort()) !== + JSON.stringify(allocatedNodeIds.sort()) || + JSON.stringify(Object.values(parsed.allocatedRelationshipIds).sort()) !== + JSON.stringify(allocatedRelationshipIds.sort()) + ) + throw new AgentMapWorkspaceStoreError("malformed_state"); + receipts.push(parsed); } - return { - projectId: value.projectId, - schemaVersion: value.schemaVersion as number, - recordVersion: value.recordVersion as number, - confirmedRevisionId: value.confirmedRevisionId, - activeProposalId: value.activeProposalId, - projectBuildPlanId: value.projectBuildPlanId, - createdAt: value.createdAt, - updatedAt: value.updatedAt, - }; -} - -function storageError(): AgentMapWorkspaceStoreError { - return new AgentMapWorkspaceStoreError("storage_unavailable"); + if ( + new Set( + receipts.map(({ sessionId, requestId }) => `${sessionId}\0${requestId}`), + ).size !== receipts.length + ) + throw new AgentMapWorkspaceStoreError("malformed_state"); + return structuredClone({ + storageSchemaVersion: AGENT_MAP_AGGREGATE_STORAGE_SCHEMA_VERSION, + workspace, + proposal, + receipts, + }) as AgentMapProjectAggregate; } -/** Lazy, restart-safe owner of each project's empty Agent Map workspace. */ +/** Crash-atomic owner of workspace, active proposal, history, and private receipts. */ export class AgentMapWorkspaceStore { - private readonly reads = new Map< - StudioProjectId, - Promise - >(); + private readonly queues = new Map>(); constructor( private readonly agentMapRoot: string, private readonly options: { now?: () => Date; onEvent?: (event: AgentMapWorkspaceStoreEvent) => void | Promise; + /** Deterministic crash-boundary seam for storage fault tests. */ + beforePersistStep?: ( + step: "write" | "file-sync" | "rename" | "directory-sync", + ) => void | Promise; } = {}, ) {} - private workspacePath(projectId: StudioProjectId): string { + private workspacePath(projectId: StudioProjectId) { return path.join( this.agentMapRoot, "projects", @@ -186,123 +265,194 @@ export class AgentMapWorkspaceStore { try { void Promise.resolve(this.options.onEvent?.(event)).catch(() => {}); } catch { - // Observability is best-effort and cannot change durable state semantics. + // Observability cannot change durable state semantics. } } - private async read( - projectId: StudioProjectId, - ): Promise { - const workspacePath = this.workspacePath(projectId); - let raw: string; + private initial(projectId: StudioProjectId): AgentMapProjectAggregate { + const timestamp = (this.options.now?.() ?? new Date()).toISOString(); + return { + storageSchemaVersion: AGENT_MAP_AGGREGATE_STORAGE_SCHEMA_VERSION, + workspace: { + projectId, + schemaVersion: AGENT_MAP_WORKSPACE_SCHEMA_VERSION, + recordVersion: AGENT_MAP_INITIAL_RECORD_VERSION, + confirmedRevisionId: null, + activeProposalId: null, + projectBuildPlanId: null, + createdAt: timestamp, + updatedAt: timestamp, + }, + proposal: null, + receipts: [], + }; + } + + private async readDisk(projectId: StudioProjectId): Promise<{ + aggregate: AgentMapProjectAggregate; + needsWrite: boolean; + created: boolean; + }> { + const file = this.workspacePath(projectId); + let decoded: unknown; try { - raw = await fs.readFile(workspacePath, "utf8"); + decoded = JSON.parse(await fs.readFile(file, "utf8")) as unknown; } catch (error) { - if ((error as NodeJS.ErrnoException).code === "ENOENT") { - return this.create(projectId, workspacePath); - } + if ((error as NodeJS.ErrnoException).code === "ENOENT") + return { + aggregate: this.initial(projectId), + needsWrite: true, + created: true, + }; + if (error instanceof SyntaxError) + throw new AgentMapWorkspaceStoreError("malformed_state"); throw storageError(); } - let decoded: unknown; + // Exact E1 record: migrate under the same lock and atomic rename. try { - decoded = JSON.parse(raw) as unknown; - } catch { - throw new AgentMapWorkspaceStoreError("malformed_state"); + const workspace = parseAgentMapWorkspaceState(decoded, projectId); + return { + aggregate: { + storageSchemaVersion: 1, + workspace, + proposal: null, + receipts: [], + }, + needsWrite: true, + created: false, + }; + } catch (error) { + if (isRecord(decoded) && "storageSchemaVersion" in decoded) { + return { + aggregate: parseAggregate(decoded, projectId), + needsWrite: false, + created: false, + }; + } + throw error; } - return parseAgentMapWorkspaceState(decoded, projectId); } - private async create( + private async persist( projectId: StudioProjectId, - workspacePath: string, - ): Promise { - const timestamp = (this.options.now?.() ?? new Date()).toISOString(); - const workspace: AgentMapWorkspaceState = { - projectId, - schemaVersion: AGENT_MAP_WORKSPACE_SCHEMA_VERSION, - recordVersion: AGENT_MAP_INITIAL_RECORD_VERSION, - confirmedRevisionId: null, - activeProposalId: null, - projectBuildPlanId: null, - createdAt: timestamp, - updatedAt: timestamp, - }; - const directory = path.dirname(workspacePath); - const temporary = `${workspacePath}.tmp-${process.pid}-${randomUUID()}`; + aggregate: AgentMapProjectAggregate, + ): Promise { + const file = this.workspacePath(projectId); + const directory = path.dirname(file); + const temporary = `${file}.tmp-${process.pid}-${randomUUID()}`; + let handle: fs.FileHandle | undefined; try { await fs.mkdir(directory, { recursive: true }); - await fs.writeFile( - temporary, - `${JSON.stringify(workspace, null, 2)}\n`, - "utf8", - ); - // `rename()` replaces an existing target on POSIX, so it cannot select - // one winner across two Studio processes (or even two store instances). - // The temporary file is already complete; linking it into the final name - // is an atomic, no-clobber commit. An EEXIST loser reads and returns the - // winner instead of publishing its divergent timestamp or event. - await fs.link(temporary, workspacePath); - } catch (error) { - if ((error as NodeJS.ErrnoException).code === "EEXIST") { - return this.readExisting(projectId, workspacePath); + handle = await fs.open(temporary, "wx", 0o600); + await this.options.beforePersistStep?.("write"); + await handle.writeFile(`${JSON.stringify(aggregate, null, 2)}\n`, "utf8"); + await this.options.beforePersistStep?.("file-sync"); + await handle.sync(); + await handle.close(); + handle = undefined; + await this.options.beforePersistStep?.("rename"); + await fs.rename(temporary, file); + const directoryHandle = await fs.open(directory, "r"); + try { + await this.options.beforePersistStep?.("directory-sync"); + await directoryHandle.sync(); + } finally { + await directoryHandle.close(); } + } catch { throw storageError(); } finally { + await handle?.close().catch(() => {}); await fs.rm(temporary, { force: true }).catch(() => {}); } - this.emit({ name: "agent_map.workspace_initialized", projectId }); - return workspace; } - private async readExisting( + private enqueue( projectId: StudioProjectId, - workspacePath: string, - ): Promise { - let raw: string; - try { - raw = await fs.readFile(workspacePath, "utf8"); - } catch { - throw storageError(); - } - let decoded: unknown; - try { - decoded = JSON.parse(raw) as unknown; - } catch { + operation: () => Promise, + ): Promise { + const previous = this.queues.get(projectId) ?? Promise.resolve(); + const result = previous.then(operation, operation); + const tail = result.then( + () => undefined, + () => undefined, + ); + this.queues.set(projectId, tail); + void tail.finally(() => { + if (this.queues.get(projectId) === tail) this.queues.delete(projectId); + }); + return result; + } + + private async locked( + projectId: StudioProjectId, + operation: ( + aggregate: AgentMapProjectAggregate, + ) => Promise<{ value: T; next?: AgentMapProjectAggregate }>, + ): Promise { + if (!isStudioProjectId(projectId)) throw new AgentMapWorkspaceStoreError("malformed_state"); + return this.enqueue(projectId, async () => { + const release = await new DurableFileLock(this.workspacePath(projectId), { + storageError, + }).acquire(); + try { + const loaded = await this.readDisk(projectId); + const outcome = await operation(structuredClone(loaded.aggregate)); + if (loaded.needsWrite || outcome.next) { + const next = outcome.next + ? parseAggregate(outcome.next, projectId) + : loaded.aggregate; + await this.persist(projectId, next); + } + if (loaded.created) + this.emit({ name: "agent_map.workspace_initialized", projectId }); + return structuredClone(outcome.value); + } finally { + await release(); + } + }); + } + + async readAggregate( + projectId: StudioProjectId, + ): Promise { + try { + return await this.locked(projectId, async (aggregate) => ({ + value: aggregate, + })); + } catch (error) { + const bounded = + error instanceof AgentMapWorkspaceStoreError ? error : storageError(); + this.emit({ + name: "agent_map.workspace_read_failed", + projectId, + ...(bounded.schemaVersion === undefined + ? {} + : { schemaVersion: bounded.schemaVersion }), + errorCode: bounded.code, + }); + throw bounded; } - return parseAgentMapWorkspaceState(decoded, projectId); } - /** - * The only E1 initializer. Concurrent calls share one per-project promise; - * no inventory, scanner, graph builder, or model dependency is reachable. - */ + async readSnapshot( + projectId: StudioProjectId, + ): Promise { + const aggregate = await this.readAggregate(projectId); + return { workspace: aggregate.workspace, proposal: aggregate.proposal }; + } + readOrCreate(projectId: StudioProjectId): Promise { - if (!isStudioProjectId(projectId)) { - return Promise.reject(new AgentMapWorkspaceStoreError("malformed_state")); - } - const active = this.reads.get(projectId); - if (active) return active; - const operation = this.read(projectId) - .catch((error: unknown) => { - const bounded = - error instanceof AgentMapWorkspaceStoreError ? error : storageError(); - this.emit({ - name: "agent_map.workspace_read_failed", - projectId, - ...(bounded.schemaVersion !== undefined - ? { schemaVersion: bounded.schemaVersion } - : {}), - errorCode: bounded.code, - }); - throw bounded; - }) - .finally(() => { - if (this.reads.get(projectId) === operation) { - this.reads.delete(projectId); - } - }); - this.reads.set(projectId, operation); - return operation; + return this.readSnapshot(projectId).then(({ workspace }) => workspace); + } + + transact( + projectId: StudioProjectId, + operation: ( + aggregate: AgentMapProjectAggregate, + ) => Promise<{ value: T; next?: AgentMapProjectAggregate }>, + ): Promise { + return this.locked(projectId, operation); } } diff --git a/packages/harness/src/core/durable-file-lock.test.ts b/packages/harness/src/core/durable-file-lock.test.ts new file mode 100644 index 00000000..b2a076f6 --- /dev/null +++ b/packages/harness/src/core/durable-file-lock.test.ts @@ -0,0 +1,39 @@ +import * as fs from "node:fs/promises"; +import * as os from "node:os"; +import * as path from "node:path"; +import { describe, expect, it } from "vitest"; + +import { DurableFileLock } from "./durable-file-lock.js"; + +describe("DurableFileLock", () => { + it("serializes live owners and reclaims a proven-dead owner", async () => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "durable-lock-")); + const target = path.join(root, "state.json"); + const first = new DurableFileLock(target); + const release = await first.acquire(); + let secondAcquired = false; + const second = new DurableFileLock(target).acquire().then((unlock) => { + secondAcquired = true; + return unlock; + }); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(secondAcquired).toBe(false); + await release(); + await ( + await second + )(); + await fs.writeFile( + `${target}.lock`, + `${JSON.stringify({ ownerId: "dead", pid: 999_999_999 })}\n`, + ); + await ( + await new DurableFileLock(target, { + hooks: { isPidAlive: () => false }, + }).acquire() + )(); + await expect(fs.access(`${target}.lock`)).rejects.toMatchObject({ + code: "ENOENT", + }); + await fs.rm(root, { recursive: true, force: true }); + }); +}); diff --git a/packages/harness/src/core/durable-file-lock.ts b/packages/harness/src/core/durable-file-lock.ts new file mode 100644 index 00000000..4d2a34cf --- /dev/null +++ b/packages/harness/src/core/durable-file-lock.ts @@ -0,0 +1,199 @@ +import { randomUUID } from "node:crypto"; +import * as fs from "node:fs/promises"; +import * as path from "node:path"; +import { setTimeout as delay } from "node:timers/promises"; + +export interface DurableFileLockOwner { + ownerId: string; + pid: number; +} + +export interface DurableFileLockTestHooks { + afterDeadOwnerObserved?: ( + owner: DurableFileLockOwner, + ) => void | Promise; + afterObservedOwnerChanged?: () => void | Promise; + afterLiveOwnerObserved?: ( + owner: DurableFileLockOwner, + ) => void | Promise; + afterLockAcquired?: (ownerId: string) => void | Promise; + isPidAlive?: (pid: number) => boolean; +} + +export interface DurableFileLockOptions { + timeoutMs?: number; + retryMs?: number; + hooks?: DurableFileLockTestHooks; + storageError?: () => Error; +} + +const isRecord = (value: unknown): value is Record => + typeof value === "object" && value !== null && !Array.isArray(value); + +const sameOwner = ( + left: DurableFileLockOwner | null, + right: DurableFileLockOwner, +): left is DurableFileLockOwner => + left !== null && left.ownerId === right.ownerId && left.pid === right.pid; + +/** Cross-process owner-file lock with live-PID protection and dead-owner fencing. */ +export class DurableFileLock { + private readonly timeoutMs: number; + private readonly retryMs: number; + private readonly hooks: DurableFileLockTestHooks; + private readonly failure: () => Error; + + constructor( + private readonly targetPath: string, + options: DurableFileLockOptions = {}, + ) { + this.timeoutMs = options.timeoutMs ?? 5_000; + this.retryMs = options.retryMs ?? 10; + this.hooks = options.hooks ?? {}; + this.failure = + options.storageError ?? (() => new Error("Storage unavailable")); + } + + async acquire(): Promise<() => Promise> { + const lockPath = `${this.targetPath}.lock`; + const owner = { ownerId: randomUUID(), pid: process.pid }; + const deadline = Date.now() + this.timeoutMs; + try { + await fs.mkdir(path.dirname(this.targetPath), { recursive: true }); + } catch { + throw this.failure(); + } + await this.cleanupArtifacts(lockPath); + for (;;) { + if (await this.tryCreate(lockPath, owner)) + return this.acquired(lockPath, owner); + const observed = await this.readOwner(lockPath); + if (observed === null || this.isAlive(observed.pid)) { + if (observed) await this.hooks.afterLiveOwnerObserved?.(observed); + if (Date.now() >= deadline) throw this.failure(); + await delay(this.retryMs); + continue; + } + await this.hooks.afterDeadOwnerObserved?.(observed); + const claimPath = `${lockPath}.claim-${observed.ownerId}`; + if (!(await this.tryCreate(claimPath, owner))) { + if (Date.now() >= deadline) throw this.failure(); + await delay(this.retryMs); + continue; + } + try { + const current = await this.readOwner(lockPath); + if (!sameOwner(current, observed) || this.isAlive(current.pid)) { + await this.hooks.afterObservedOwnerChanged?.(); + continue; + } + const tombstone = `${lockPath}.reclaim-${owner.ownerId}`; + try { + await fs.rename(lockPath, tombstone); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") continue; + throw this.failure(); + } + if (!(await this.tryCreate(lockPath, owner))) { + await fs.rm(tombstone, { force: true }).catch(() => {}); + continue; + } + await fs.rm(tombstone, { force: true }); + return this.acquired(lockPath, owner); + } finally { + await this.release(claimPath, owner); + } + } + } + + private async acquired(lockPath: string, owner: DurableFileLockOwner) { + try { + await this.hooks.afterLockAcquired?.(owner.ownerId); + } catch (error) { + await this.release(lockPath, owner); + throw error; + } + return () => this.release(lockPath, owner); + } + + private async tryCreate(lockPath: string, owner: DurableFileLockOwner) { + try { + await fs.writeFile(lockPath, `${JSON.stringify(owner)}\n`, { + encoding: "utf8", + flag: "wx", + }); + return true; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "EEXIST") return false; + throw this.failure(); + } + } + + private async readOwner( + lockPath: string, + ): Promise { + try { + const decoded = JSON.parse( + await fs.readFile(lockPath, "utf8"), + ) as unknown; + if ( + !isRecord(decoded) || + Object.keys(decoded).sort().join(",") !== "ownerId,pid" || + typeof decoded.ownerId !== "string" || + decoded.ownerId.length === 0 || + !Number.isSafeInteger(decoded.pid) || + (decoded.pid as number) <= 0 + ) + return null; + return { ownerId: decoded.ownerId, pid: decoded.pid as number }; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return null; + if (error instanceof SyntaxError) return null; + throw this.failure(); + } + } + + private isAlive(pid: number): boolean { + if (this.hooks.isPidAlive) return this.hooks.isPidAlive(pid); + try { + process.kill(pid, 0); + return true; + } catch (error) { + return (error as NodeJS.ErrnoException).code !== "ESRCH"; + } + } + + private async cleanupArtifacts(lockPath: string): Promise { + try { + const directory = path.dirname(lockPath); + const base = path.basename(lockPath); + for (const entry of await fs.readdir(directory)) { + if ( + !entry.startsWith(`${base}.claim-`) && + !entry.startsWith(`${base}.reclaim-`) + ) + continue; + const artifact = path.join(directory, entry); + const owner = await this.readOwner(artifact); + if (owner && !this.isAlive(owner.pid)) + await fs.rm(artifact, { force: true }); + } + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") + throw this.failure(); + } + } + + private async release( + lockPath: string, + owner: DurableFileLockOwner, + ): Promise { + if (!sameOwner(await this.readOwner(lockPath), owner)) return; + try { + await fs.unlink(lockPath); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") + throw this.failure(); + } + } +} diff --git a/packages/harness/src/core/studio-project-catalog.ts b/packages/harness/src/core/studio-project-catalog.ts index 176738e1..6779e725 100644 --- a/packages/harness/src/core/studio-project-catalog.ts +++ b/packages/harness/src/core/studio-project-catalog.ts @@ -1,7 +1,6 @@ import { randomUUID } from "node:crypto"; import * as fs from "node:fs/promises"; import * as path from "node:path"; -import { setTimeout as delay } from "node:timers/promises"; import { STUDIO_PROJECT_CATALOG_SCHEMA_VERSION, @@ -12,6 +11,10 @@ import { } from "../shared/agent-map.js"; import type { WorkspaceScopeSummary } from "../shared/system-graph.js"; import { canonicalGraphPath } from "./canonical-graph-path.js"; +import { + DurableFileLock, + type DurableFileLockTestHooks, +} from "./durable-file-lock.js"; export interface ProjectRootBinding { id: string; @@ -293,33 +296,8 @@ function storageError(): StudioProjectCatalogError { return new StudioProjectCatalogError("storage_unavailable"); } -const CATALOG_LOCK_TIMEOUT_MS = 5_000; -const CATALOG_LOCK_RETRY_MS = 10; - -interface CatalogLockOwner { - ownerId: string; - pid: number; -} - -function sameLockOwner( - left: CatalogLockOwner | null, - right: CatalogLockOwner, -): left is CatalogLockOwner { - return ( - left !== null && - left.ownerId === right.ownerId && - left.pid === right.pid - ); -} - /** Internal deterministic seams used only by file-lock race regressions. */ -export interface StudioProjectCatalogLockTestHooks { - afterDeadOwnerObserved?: (owner: CatalogLockOwner) => void | Promise; - afterObservedOwnerChanged?: () => void | Promise; - afterLiveOwnerObserved?: (owner: CatalogLockOwner) => void | Promise; - afterLockAcquired?: (ownerId: string) => void | Promise; - isPidAlive?: (pid: number) => boolean; -} +export type StudioProjectCatalogLockTestHooks = DurableFileLockTestHooks; /** * Durable, serialized owner of Studio project identity. Catalog reads never @@ -359,193 +337,10 @@ export class StudioProjectCatalog { } private async acquireFileLock(): Promise<() => Promise> { - const lockPath = `${this.catalogPath}.lock`; - const owner: CatalogLockOwner = { - ownerId: randomUUID(), - pid: process.pid, - }; - const deadline = Date.now() + CATALOG_LOCK_TIMEOUT_MS; - try { - await fs.mkdir(path.dirname(this.catalogPath), { recursive: true }); - } catch { - throw storageError(); - } - await this.cleanupDeadLockArtifacts(lockPath); - for (;;) { - if (await this.tryCreateLockFile(lockPath, owner)) { - try { - await this.lockTestHooks.afterLockAcquired?.(owner.ownerId); - } catch (error) { - await this.releaseFileLock(lockPath, owner); - throw error; - } - return () => this.releaseFileLock(lockPath, owner); - } - - const observedOwner = await this.readLockOwner(lockPath); - if (observedOwner === null) { - // Another process may be between exclusive create and owner write. - // Never evict an owner whose PID cannot be proven dead. - if (Date.now() >= deadline) throw storageError(); - await delay(CATALOG_LOCK_RETRY_MS); - continue; - } - if (this.isPidAlive(observedOwner.pid)) { - await this.lockTestHooks.afterLiveOwnerObserved?.(observedOwner); - if (Date.now() >= deadline) throw storageError(); - await delay(CATALOG_LOCK_RETRY_MS); - continue; - } - - await this.lockTestHooks.afterDeadOwnerObserved?.(observedOwner); - const claimPath = `${lockPath}.claim-${observedOwner.ownerId}`; - if (!(await this.tryCreateLockFile(claimPath, owner))) { - if (Date.now() >= deadline) throw storageError(); - await delay(CATALOG_LOCK_RETRY_MS); - continue; - } - - try { - // Every waiter that observed this dead owner competes for the same - // claim. Re-read under that claim: a delayed waiter must not rename a - // newer live owner's fixed lock. - const currentOwner = await this.readLockOwner(lockPath); - if ( - !sameLockOwner(currentOwner, observedOwner) || - this.isPidAlive(currentOwner.pid) - ) { - await this.lockTestHooks.afterObservedOwnerChanged?.(); - continue; - } - - const tombstone = `${lockPath}.reclaim-${owner.ownerId}`; - try { - await fs.rename(lockPath, tombstone); - } catch (error) { - if ((error as NodeJS.ErrnoException).code === "ENOENT") continue; - throw storageError(); - } - - // The fixed name is the fence. A normal contender may win this gap; - // this reclaimer proceeds only if its own exclusive create wins. - if (!(await this.tryCreateLockFile(lockPath, owner))) { - await fs.rm(tombstone, { force: true }).catch(() => {}); - continue; - } - - await fs.rm(tombstone, { force: true }); - try { - await this.lockTestHooks.afterLockAcquired?.(owner.ownerId); - } catch (error) { - await this.releaseFileLock(lockPath, owner); - throw error; - } - return () => this.releaseFileLock(lockPath, owner); - } finally { - await this.releaseFileLock(claimPath, owner); - } - } - } - - private async tryCreateLockFile( - lockPath: string, - owner: CatalogLockOwner, - ): Promise { - try { - await fs.writeFile(lockPath, `${JSON.stringify(owner)}\n`, { - encoding: "utf8", - flag: "wx", - }); - return true; - } catch (error) { - if ((error as NodeJS.ErrnoException).code === "EEXIST") return false; - throw storageError(); - } - } - - private async readLockOwner(lockPath: string): Promise { - let raw: string; - try { - raw = await fs.readFile(lockPath, "utf8"); - } catch (error) { - if ((error as NodeJS.ErrnoException).code === "ENOENT") return null; - throw storageError(); - } - try { - const decoded = JSON.parse(raw) as unknown; - if ( - !isRecord(decoded) || - !hasExactKeys(decoded, ["ownerId", "pid"]) || - !isOpaqueId(decoded.ownerId) || - !Number.isSafeInteger(decoded.pid) || - (decoded.pid as number) <= 0 - ) { - return null; - } - return { - ownerId: decoded.ownerId, - pid: decoded.pid as number, - }; - } catch { - // An exclusive creator may still be writing its owner record. Unknown - // ownership waits and fails closed; it is never reclaimed by age. - return null; - } - } - - private isPidAlive(pid: number): boolean { - if (this.lockTestHooks.isPidAlive) { - return this.lockTestHooks.isPidAlive(pid); - } - try { - process.kill(pid, 0); - return true; - } catch (error) { - return (error as NodeJS.ErrnoException).code !== "ESRCH"; - } - } - - private async cleanupDeadLockArtifacts(lockPath: string): Promise { - const directory = path.dirname(lockPath); - const base = path.basename(lockPath); - let entries: string[]; - try { - entries = await fs.readdir(directory); - } catch (error) { - if ((error as NodeJS.ErrnoException).code === "ENOENT") return; - throw storageError(); - } - await Promise.all( - entries - .filter( - (entry) => - entry.startsWith(`${base}.reclaim-`) || - entry.startsWith(`${base}.claim-`), - ) - .map(async (entry) => { - const artifact = path.join(directory, entry); - const artifactOwner = await this.readLockOwner(artifact); - if (artifactOwner && !this.isPidAlive(artifactOwner.pid)) { - await fs.rm(artifact, { force: true }); - } - }), - ); - } - - /** A live PID cannot be reclaimed, so this read-then-unlink is owner-safe. */ - private async releaseFileLock( - lockPath: string, - owner: CatalogLockOwner, - ): Promise { - const currentOwner = await this.readLockOwner(lockPath); - if (!sameLockOwner(currentOwner, owner)) return; - try { - await fs.unlink(lockPath); - } catch (error) { - if ((error as NodeJS.ErrnoException).code !== "ENOENT") { - throw storageError(); - } - } + return new DurableFileLock(this.catalogPath, { + hooks: this.lockTestHooks, + storageError, + }).acquire(); } private async load(force = false): Promise { diff --git a/packages/harness/src/index.ts b/packages/harness/src/index.ts index 650256cd..8330ab4f 100644 --- a/packages/harness/src/index.ts +++ b/packages/harness/src/index.ts @@ -4,6 +4,29 @@ */ export * from "./shared/types.js"; +export { + AGENT_MAP_PROPOSAL_SCHEMA_VERSION, + EXECUTION_MODES, + PLAN_NODE_KINDS, + RELATIONSHIP_KINDS, +} from "./shared/agent-map.js"; +export type { + AcceptedProposalDelta, + ExecutionMode, + MapOperation, + MapProposalId, + PlanNode, + PlanNodeChanges, + PlanNodeId, + PlanNodeKind, + PlanRelationship, + PlanRelationshipId, + ProposalActor, + ProposalOperationId, + RelationshipChanges, + RelationshipKind, + StudioProjectId, +} from "./shared/agent-map.js"; export type { WorkspaceScopeSummary } from "./shared/system-graph.js"; export { AGENT_STUDIO_PRODUCT_NAME } from "./shared/branding.js"; export { @@ -59,4 +82,9 @@ export type { SpawnTarget } from "./core/spawn-target.js"; // desktop app's --smoke mode to create a REAL session against a stub agent, so // per-OS session coverage doesn't require Claude Code installed on a CI runner. export { createClaudeCodeAdapter } from "./core/adapters/claude-code.js"; -export { loadSettings, saveSettings, recordRecentDir, hasStoredSettings } from "./cli/settings.js"; +export { + loadSettings, + saveSettings, + recordRecentDir, + hasStoredSettings, +} from "./cli/settings.js"; diff --git a/packages/harness/src/server/agent-map-proposal-wiring.test.ts b/packages/harness/src/server/agent-map-proposal-wiring.test.ts new file mode 100644 index 00000000..8e62138b --- /dev/null +++ b/packages/harness/src/server/agent-map-proposal-wiring.test.ts @@ -0,0 +1,56 @@ +import * as fs from "node:fs/promises"; +import * as os from "node:os"; +import * as path from "node:path"; +import { expect, it } from "vitest"; + +import type { AcceptedProposalDelta, BusMessage } from "../index.js"; +import { EventBus } from "../core/event-bus.js"; +import { AgentMapProposalService } from "../core/agent-map-proposal-service.js"; +import { AgentMapWorkspaceStore } from "../core/agent-map-workspace-store.js"; + +it("publishes exactly one accepted proposal delta after durable commit", async () => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "agent-map-wiring-")); + const bus = new EventBus(); + const messages: BusMessage[] = []; + bus.subscribe((message) => messages.push(message)); + const publishAccepted = (delta: AcceptedProposalDelta) => + bus.publish({ type: "agent-map.proposal.changed", delta }); + const service = new AgentMapProposalService( + new AgentMapWorkspaceStore(root), + { + onAccepted: publishAccepted, + }, + ); + const identity = { + projectId: "project_00000000-0000-4000-8000-000000000001", + userId: "user-1", + sessionId: "session-1", + role: "map-planner" as const, + }; + const request = { + schemaVersion: 1 as const, + proposalId: null, + expectedVersion: 0, + requestId: "request-1", + operations: [ + { + kind: "add-node" as const, + draftRef: "research" as import("../shared/agent-map.js").DraftRef, + node: { + kind: "agent" as const, + name: "Research", + purpose: "Research", + ownerAgent: null, + contractRefs: [], + }, + }, + ], + }; + const result = await service.propose(identity, request); + await service.propose(identity, request); + expect(messages).toEqual([ + { type: "agent-map.proposal.changed", delta: result.delta }, + ]); + expect((await service.read(identity.projectId)).proposal?.version).toBe(1); + await fs.rm(root, { recursive: true, force: true }); +}); diff --git a/packages/harness/src/server/agent-map.ts b/packages/harness/src/server/agent-map.ts index 6753d474..9dc93206 100644 --- a/packages/harness/src/server/agent-map.ts +++ b/packages/harness/src/server/agent-map.ts @@ -9,10 +9,7 @@ import { type PlannerSessionRequest, type StudioWorkspaceSelection, } from "../shared/agent-map.js"; -import { - SPAWNABLE_HARNESS_KINDS, - type WorkflowInfo, -} from "../shared/types.js"; +import { SPAWNABLE_HARNESS_KINDS, type WorkflowInfo } from "../shared/types.js"; import type { WorkspaceScopeSummary } from "../shared/system-graph.js"; import { AgentMapWorkspaceStore, @@ -276,11 +273,18 @@ export function createAgentMapRouter(options: AgentMapRouterOptions): Router { // Project resolution intentionally happens before the lazy initializer: // an arbitrary/cross-instance ID can never create a state directory. - const workspace = await options.store.readOrCreate(project.projectId); + const { workspace, proposal } = await options.store.readSnapshot( + project.projectId, + ); res .status(200) .setHeader("Cache-Control", "no-store") - .json({ project, workspace } satisfies AgentMapWorkspaceResponse); + .json({ + schemaVersion: 1, + project, + workspace, + proposal, + } satisfies AgentMapWorkspaceResponse); } catch (error) { const bounded = error instanceof AgentMapWorkspaceStoreError || @@ -354,27 +358,30 @@ export function createAgentMapRouter(options: AgentMapRouterOptions): Router { } }); - router.post("/projects/:projectId/planner-sessions", async (req, res, next) => { - if (!options.planningSessions || !options.plannerGreeting) { - res.status(501).json({ error: "Planner sessions are unavailable" }); - return; - } - const parsed = plannerSessionSchema.safeParse(req.body); - if (!parsed.success) { - res.status(400).json({ error: "Invalid planner session request" }); - return; - } - try { - await options.catalog.reconcile(await options.listWorkspaceScopes()); - const result = await options.planningSessions.open( - req.params.projectId, - parsed.data, - ); - res.status(result.resolution === "created" ? 201 : 200).json(result); - } catch (error) { - if (!sendPlanningError(res, error)) next(error); - } - }); + router.post( + "/projects/:projectId/planner-sessions", + async (req, res, next) => { + if (!options.planningSessions || !options.plannerGreeting) { + res.status(501).json({ error: "Planner sessions are unavailable" }); + return; + } + const parsed = plannerSessionSchema.safeParse(req.body); + if (!parsed.success) { + res.status(400).json({ error: "Invalid planner session request" }); + return; + } + try { + await options.catalog.reconcile(await options.listWorkspaceScopes()); + const result = await options.planningSessions.open( + req.params.projectId, + parsed.data, + ); + res.status(result.resolution === "created" ? 201 : 200).json(result); + } catch (error) { + if (!sendPlanningError(res, error)) next(error); + } + }, + ); router.post( "/projects/:projectId/planner-sessions/:sessionId/messages", diff --git a/packages/harness/src/shared/agent-map-codec.test.ts b/packages/harness/src/shared/agent-map-codec.test.ts new file mode 100644 index 00000000..09f4c96d --- /dev/null +++ b/packages/harness/src/shared/agent-map-codec.test.ts @@ -0,0 +1,97 @@ +import { describe, expect, it } from "vitest"; + +import { + parseAgentMapProposalReceipt, + parseMapChangeProposal, + parseProposalActor, +} from "./agent-map-codec.js"; + +const nodeId = "node_00000000-0000-7000-8000-000000000001"; +const proposalId = "proposal_00000000-0000-7000-8000-000000000002"; +const operationId = "operation_00000000-0000-7000-8000-000000000003"; +const acceptedAt = "2026-09-02T12:00:00.000Z"; +const actor = { + userId: "user-1", + sessionId: "session-1", + role: "map-planner", + assignment: null, +}; +const operation = { + kind: "add-node", + node: { + id: nodeId, + kind: "agent", + name: "Research", + purpose: "Research", + ownerAgentId: null, + contractRefs: [], + }, +}; +const proposal = { + schemaVersion: 1, + id: proposalId, + projectId: "project_00000000-0000-4000-8000-000000000001", + baseRevisionId: null, + version: 1, + nodes: [operation.node], + relationships: [], + history: [ + { + id: operationId, + requestId: "request-1", + acceptedVersion: 1, + operation, + actor, + acceptedAt, + }, + ], + createdAt: acceptedAt, + updatedAt: acceptedAt, +}; +const receipt = { + sessionId: "session-1", + requestId: "request-1", + requestDigest: "a".repeat(64), + version: 1, + allocatedNodeIds: { research: nodeId }, + allocatedRelationshipIds: {}, +}; + +describe("Agent Map persisted/public codecs", () => { + it("accepts the complete exact nested proposal and receipt", () => { + expect(parseMapChangeProposal(proposal)).toEqual(proposal); + expect(parseAgentMapProposalReceipt(receipt)).toEqual(receipt); + }); + + it.each([ + [ + "unknown operation", + (value: any) => (value.history[0].operation.kind = "execute"), + ], + [ + "spoofed assignment", + (value: any) => + (value.history[0].actor.assignment = { kind: "unplanned" }), + ], + [ + "nested extra field", + (value: any) => (value.history[0].operation.node.privatePath = "/secret"), + ], + ])("rejects %s in public history", (_name, mutate) => { + const value = structuredClone(proposal); + mutate(value); + expect(() => parseMapChangeProposal(value)).toThrow(); + }); + + it("rejects corrupt private receipt allocations", () => { + const value = structuredClone(receipt) as any; + value.allocatedNodeIds.research = "foreign"; + expect(() => parseAgentMapProposalReceipt(value)).toThrow(); + }); + + it("uses the same DEL-safe actor boundary as proposal requests", () => { + expect(() => + parseProposalActor({ ...actor, sessionId: "session\u007f1" }), + ).toThrow(); + }); +}); diff --git a/packages/harness/src/shared/agent-map-codec.ts b/packages/harness/src/shared/agent-map-codec.ts new file mode 100644 index 00000000..9334e5f6 --- /dev/null +++ b/packages/harness/src/shared/agent-map-codec.ts @@ -0,0 +1,410 @@ +import { + AGENT_MAP_PROPOSAL_SCHEMA_VERSION, + EXECUTION_MODES, + PLAN_NODE_KINDS, + RELATIONSHIP_KINDS, + type DraftRef, + type MapChangeProposal, + type MapOperation, + type PlanNode, + type PlanNodeId, + type PlanRelationship, + type PlanRelationshipId, + type ProposalActor, + type ProposalBatchResult, +} from "./agent-map.js"; + +export const AGENT_MAP_UUID_V7_PATTERN = + "[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}"; + +export function hasAgentMapControlCharacter(value: string): boolean { + return [...value].some((character) => { + const codePoint = character.codePointAt(0) ?? 0; + return codePoint <= 0x1f || codePoint === 0x7f; + }); +} + +export function isAgentMapBoundedText( + value: unknown, + maximum: number, + allowEmpty = false, +): value is string { + return ( + typeof value === "string" && + value.length <= maximum && + (allowEmpty || value.length > 0) && + value.trim() === value && + !hasAgentMapControlCharacter(value) + ); +} + +const isRecord = (value: unknown): value is Record => + typeof value === "object" && value !== null && !Array.isArray(value); + +const hasExactKeys = ( + value: Record, + keys: readonly string[], +): boolean => { + const actual = Object.keys(value).sort(); + const expected = [...keys].sort(); + return ( + actual.length === expected.length && + actual.every((key, index) => key === expected[index]) + ); +}; + +const isPlanId = (value: unknown, prefix: string): value is string => + typeof value === "string" && + new RegExp(`^${prefix}_${AGENT_MAP_UUID_V7_PATTERN}$`, "u").test(value); + +const isTimestamp = (value: unknown): value is string => { + if (typeof value !== "string") return false; + try { + return new Date(value).toISOString() === value; + } catch { + return false; + } +}; + +const isContractRefs = (value: unknown): value is string[] => + Array.isArray(value) && + value.length <= 64 && + value.every((entry) => isAgentMapBoundedText(entry, 512)) && + new Set(value).size === value.length; + +function parseNode(value: unknown): PlanNode { + if ( + !isRecord(value) || + !hasExactKeys(value, [ + "id", + "kind", + "name", + "purpose", + "ownerAgentId", + "contractRefs", + ]) || + !isPlanId(value.id, "node") || + !PLAN_NODE_KINDS.includes(value.kind as (typeof PLAN_NODE_KINDS)[number]) || + !isAgentMapBoundedText(value.name, 160) || + !isAgentMapBoundedText(value.purpose, 2_000) || + (value.ownerAgentId !== null && !isPlanId(value.ownerAgentId, "node")) || + !isContractRefs(value.contractRefs) + ) + throw new Error("invalid Agent Map node"); + return structuredClone(value) as unknown as PlanNode; +} + +function parseRelationship(value: unknown): PlanRelationship { + if ( + !isRecord(value) || + !hasExactKeys(value, [ + "id", + "fromNodeId", + "toNodeId", + "kind", + "executionMode", + "contractRef", + "description", + ]) || + !isPlanId(value.id, "rel") || + !isPlanId(value.fromNodeId, "node") || + !isPlanId(value.toNodeId, "node") || + !RELATIONSHIP_KINDS.includes( + value.kind as (typeof RELATIONSHIP_KINDS)[number], + ) || + (value.executionMode !== null && + !EXECUTION_MODES.includes( + value.executionMode as (typeof EXECUTION_MODES)[number], + )) || + (value.contractRef !== null && + !isAgentMapBoundedText(value.contractRef, 512)) || + !isAgentMapBoundedText(value.description, 2_000, true) + ) + throw new Error("invalid Agent Map relationship"); + return structuredClone(value) as unknown as PlanRelationship; +} + +function parseNodeChanges(value: unknown) { + if ( + !isRecord(value) || + Object.keys(value).length === 0 || + !Object.keys(value).every((key) => + ["name", "purpose", "contractRefs"].includes(key), + ) || + ("name" in value && !isAgentMapBoundedText(value.name, 160)) || + ("purpose" in value && !isAgentMapBoundedText(value.purpose, 2_000)) || + ("contractRefs" in value && !isContractRefs(value.contractRefs)) + ) + throw new Error("invalid Agent Map node changes"); + return structuredClone(value); +} + +function parseRelationshipChanges(value: unknown) { + if ( + !isRecord(value) || + Object.keys(value).length === 0 || + !Object.keys(value).every((key) => + ["description", "executionMode", "contractRef"].includes(key), + ) || + ("description" in value && + !isAgentMapBoundedText(value.description, 2_000, true)) || + ("executionMode" in value && + value.executionMode !== null && + !EXECUTION_MODES.includes( + value.executionMode as (typeof EXECUTION_MODES)[number], + )) || + ("contractRef" in value && + value.contractRef !== null && + !isAgentMapBoundedText(value.contractRef, 512)) + ) + throw new Error("invalid Agent Map relationship changes"); + return structuredClone(value); +} + +function parseMapOperation(value: unknown): MapOperation { + if (!isRecord(value) || typeof value.kind !== "string") + throw new Error("invalid Agent Map operation"); + switch (value.kind) { + case "add-node": + if (!hasExactKeys(value, ["kind", "node"])) + throw new Error("invalid Agent Map operation"); + return { kind: value.kind, node: parseNode(value.node) }; + case "update-node": + if ( + !hasExactKeys(value, ["kind", "nodeId", "changes"]) || + !isPlanId(value.nodeId, "node") + ) + throw new Error("invalid Agent Map operation"); + return { + kind: value.kind, + nodeId: value.nodeId as PlanNodeId, + changes: parseNodeChanges(value.changes), + } as MapOperation; + case "remove-node": + if ( + !hasExactKeys(value, ["kind", "nodeId"]) || + !isPlanId(value.nodeId, "node") + ) + throw new Error("invalid Agent Map operation"); + return { kind: value.kind, nodeId: value.nodeId as PlanNodeId }; + case "add-relationship": + if (!hasExactKeys(value, ["kind", "relationship"])) + throw new Error("invalid Agent Map operation"); + return { + kind: value.kind, + relationship: parseRelationship(value.relationship), + }; + case "update-relationship": + if ( + !hasExactKeys(value, ["kind", "relationshipId", "changes"]) || + !isPlanId(value.relationshipId, "rel") + ) + throw new Error("invalid Agent Map operation"); + return { + kind: value.kind, + relationshipId: value.relationshipId as PlanRelationshipId, + changes: parseRelationshipChanges(value.changes), + } as MapOperation; + case "remove-relationship": + if ( + !hasExactKeys(value, ["kind", "relationshipId"]) || + !isPlanId(value.relationshipId, "rel") + ) + throw new Error("invalid Agent Map operation"); + return { + kind: value.kind, + relationshipId: value.relationshipId as PlanRelationshipId, + }; + default: + throw new Error("invalid Agent Map operation"); + } +} + +export function parseProposalActor(value: unknown): ProposalActor { + if ( + !isRecord(value) || + !hasExactKeys(value, ["userId", "sessionId", "role", "assignment"]) || + !isAgentMapBoundedText(value.userId, 256) || + !isAgentMapBoundedText(value.sessionId, 256) + ) + throw new Error("invalid Agent Map actor"); + if (value.role === "map-planner" && value.assignment === null) + return structuredClone(value) as unknown as ProposalActor; + if ( + value.role !== "agent-builder" || + !isRecord(value.assignment) || + (value.assignment.kind === "planned" + ? !hasExactKeys(value.assignment, ["kind", "agentId"]) || + !isAgentMapBoundedText(value.assignment.agentId, 256) + : value.assignment.kind !== "unplanned" || + !hasExactKeys(value.assignment, ["kind"])) + ) + throw new Error("invalid Agent Map actor"); + return structuredClone(value) as unknown as ProposalActor; +} + +export function parseMapChangeProposal( + value: unknown, + projectId?: string, + activeProposalId?: string, +): MapChangeProposal { + if ( + !isRecord(value) || + !hasExactKeys(value, [ + "schemaVersion", + "id", + "projectId", + "baseRevisionId", + "version", + "nodes", + "relationships", + "history", + "createdAt", + "updatedAt", + ]) || + value.schemaVersion !== AGENT_MAP_PROPOSAL_SCHEMA_VERSION || + !isPlanId(value.id, "proposal") || + !isAgentMapBoundedText(value.projectId, 128) || + (projectId !== undefined && value.projectId !== projectId) || + (activeProposalId !== undefined && value.id !== activeProposalId) || + (value.baseRevisionId !== null && + !isAgentMapBoundedText(value.baseRevisionId, 256)) || + !Number.isSafeInteger(value.version) || + (value.version as number) < 1 || + !Array.isArray(value.nodes) || + !Array.isArray(value.relationships) || + !Array.isArray(value.history) || + value.history.length === 0 || + !isTimestamp(value.createdAt) || + !isTimestamp(value.updatedAt) + ) + throw new Error("invalid Agent Map proposal"); + + const nodes = value.nodes.map(parseNode); + const relationships = value.relationships.map(parseRelationship); + const history = value.history.map((record) => { + if ( + !isRecord(record) || + !hasExactKeys(record, [ + "id", + "requestId", + "acceptedVersion", + "operation", + "actor", + "acceptedAt", + ]) || + !isPlanId(record.id, "operation") || + !isAgentMapBoundedText(record.requestId, 128) || + !Number.isSafeInteger(record.acceptedVersion) || + (record.acceptedVersion as number) < 1 || + !isTimestamp(record.acceptedAt) + ) + throw new Error("invalid Agent Map history"); + return { + id: record.id, + requestId: record.requestId, + acceptedVersion: record.acceptedVersion as number, + operation: parseMapOperation(record.operation), + actor: parseProposalActor(record.actor), + acceptedAt: record.acceptedAt, + }; + }); + const versions = history.map(({ acceptedVersion }) => acceptedVersion); + const uniqueVersions = [...new Set(versions)]; + const nodeIds = new Set(nodes.map(({ id }) => id)); + if ( + new Set(nodes.map(({ id }) => id)).size !== nodes.length || + new Set(relationships.map(({ id }) => id)).size !== relationships.length || + new Set(history.map(({ id }) => id)).size !== history.length || + uniqueVersions.some((version, index) => version !== index + 1) || + versions.some( + (version, index) => index > 0 && version < versions[index - 1]!, + ) || + versions.at(-1) !== value.version || + nodes.some( + ({ ownerAgentId }) => ownerAgentId !== null && !nodeIds.has(ownerAgentId), + ) || + relationships.some( + ({ fromNodeId, toNodeId }) => + !nodeIds.has(fromNodeId) || !nodeIds.has(toNodeId), + ) + ) + throw new Error("inconsistent Agent Map proposal"); + return { + schemaVersion: AGENT_MAP_PROPOSAL_SCHEMA_VERSION, + id: value.id, + projectId: value.projectId, + baseRevisionId: value.baseRevisionId, + version: value.version as number, + nodes, + relationships, + history, + createdAt: value.createdAt, + updatedAt: value.updatedAt, + } as MapChangeProposal; +} + +export interface PersistedAgentMapProposalReceipt { + sessionId: string; + requestId: string; + requestDigest: string; + version: number; + allocatedNodeIds: ProposalBatchResult["allocatedNodeIds"]; + allocatedRelationshipIds: ProposalBatchResult["allocatedRelationshipIds"]; +} + +const parseAllocationMap = ( + value: unknown, + prefix: "node" | "rel", +): Record => { + if ( + !isRecord(value) || + Object.keys(value).length > 256 || + !Object.entries(value).every( + ([draftRef, id]) => + isAgentMapBoundedText(draftRef, 128) && isPlanId(id, prefix), + ) + ) + throw new Error("invalid Agent Map allocation map"); + return structuredClone(value) as Record< + DraftRef, + PlanNodeId | PlanRelationshipId + >; +}; + +export function parseAgentMapProposalReceipt( + value: unknown, +): PersistedAgentMapProposalReceipt { + if ( + !isRecord(value) || + !hasExactKeys(value, [ + "sessionId", + "requestId", + "requestDigest", + "version", + "allocatedNodeIds", + "allocatedRelationshipIds", + ]) || + !isAgentMapBoundedText(value.sessionId, 256) || + !isAgentMapBoundedText(value.requestId, 128) || + typeof value.requestDigest !== "string" || + !/^[0-9a-f]{64}$/u.test(value.requestDigest) || + !Number.isSafeInteger(value.version) || + (value.version as number) < 1 + ) + throw new Error("invalid Agent Map receipt"); + return { + sessionId: value.sessionId, + requestId: value.requestId, + requestDigest: value.requestDigest, + version: value.version as number, + allocatedNodeIds: parseAllocationMap( + value.allocatedNodeIds, + "node", + ) as ProposalBatchResult["allocatedNodeIds"], + allocatedRelationshipIds: parseAllocationMap( + value.allocatedRelationshipIds, + "rel", + ) as ProposalBatchResult["allocatedRelationshipIds"], + }; +} diff --git a/packages/harness/src/shared/agent-map.ts b/packages/harness/src/shared/agent-map.ts index 5ee6f653..dba53628 100644 --- a/packages/harness/src/shared/agent-map.ts +++ b/packages/harness/src/shared/agent-map.ts @@ -161,14 +161,17 @@ export type ProposalValidationResult = | { ok: true; value: T } | { ok: false; issues: ProposalValidationIssue[] }; -export type ProposalConflictCode = "stale_version" | "request_id_reused"; +export type ProposalConflictCode = + | "stale_version" + | "request_id_reused" + | "request_id_expired"; export interface ProposalConflict { code: ProposalConflictCode; currentVersion: number; affectedNodeIds: PlanNodeId[]; affectedRelationshipIds: PlanRelationshipId[]; - recovery: "reread" | "retry"; + recovery: "reread" | "retry" | "new_request"; } export type ProjectRootBindingStatus = "active" | "missing"; @@ -205,10 +208,8 @@ export interface AgentMapWorkspaceState { updatedAt: string; } -export interface AgentMapWorkspaceResponse { - project: StudioProjectSummary; - workspace: AgentMapWorkspaceState; -} +/** Backwards-compatible route name for the canonical Agent Map read shape. */ +export type AgentMapWorkspaceResponse = AgentMapReadSnapshot; /** Stable, path-free identity for the workspace currently open in Studio. */ export type StudioWorkspaceSelection = @@ -274,7 +275,7 @@ export type PlanningSessionIdentity = export interface ProposalActor { userId: string; sessionId: string; - role: PlanningSessionIdentity["role"]; + role: "map-planner" | "agent-builder"; assignment: | { kind: "planned"; agentId: string } | { kind: "unplanned" } diff --git a/packages/harness/src/shared/types.ts b/packages/harness/src/shared/types.ts index 33f4be7b..ece2891f 100644 --- a/packages/harness/src/shared/types.ts +++ b/packages/harness/src/shared/types.ts @@ -554,6 +554,10 @@ export type BusMessage = revision: number; state: SystemGraphLifecycleState; } + | { + type: "agent-map.proposal.changed"; + delta: import("./agent-map.js").AcceptedProposalDelta; + } /** * Full snapshot of one background task, re-broadcast on every change * (spawn, each new status line, completion/failure). Tasks are rare and diff --git a/packages/harness/vitest.config.ts b/packages/harness/vitest.config.ts index 78684376..51547410 100644 --- a/packages/harness/vitest.config.ts +++ b/packages/harness/vitest.config.ts @@ -16,10 +16,21 @@ export default defineConfig({ // Resolve "@shared/types" to the package's canonical contract so web // unit tests and server tests always build against the same source of // truth. Mirrors the alias in web/vite.config.ts. - "@shared/types": fileURLToPath(new URL("src/shared/types.ts", import.meta.url)), - "@shared/system-graph": fileURLToPath(new URL("src/shared/system-graph.ts", import.meta.url)), - "@shared/agent-map": fileURLToPath(new URL("src/shared/agent-map.ts", import.meta.url)), - "@shared/agent-name": fileURLToPath(new URL("src/shared/agent-name.ts", import.meta.url)), + "@shared/types": fileURLToPath( + new URL("src/shared/types.ts", import.meta.url), + ), + "@shared/system-graph": fileURLToPath( + new URL("src/shared/system-graph.ts", import.meta.url), + ), + "@shared/agent-map": fileURLToPath( + new URL("src/shared/agent-map.ts", import.meta.url), + ), + "@shared/agent-map-codec": fileURLToPath( + new URL("src/shared/agent-map-codec.ts", import.meta.url), + ), + "@shared/agent-name": fileURLToPath( + new URL("src/shared/agent-name.ts", import.meta.url), + ), }, }, test: { diff --git a/packages/harness/web/src/lib/agent-map.test.ts b/packages/harness/web/src/lib/agent-map.test.ts index 6296d6cd..0eabc2e6 100644 --- a/packages/harness/web/src/lib/agent-map.test.ts +++ b/packages/harness/web/src/lib/agent-map.test.ts @@ -13,6 +13,7 @@ const timestamp = "2026-09-01T12:00:00.000Z"; function validResponse(): unknown { return { + schemaVersion: 1, project: { projectId, identityVersion: 1, @@ -36,6 +37,7 @@ function validResponse(): unknown { createdAt: timestamp, updatedAt: timestamp, }, + proposal: null, }; } @@ -46,6 +48,58 @@ describe("parseAgentMapWorkspaceResponse", () => { ); }); + it("accepts and strictly parses a populated proposal", () => { + const value = validResponse() as any; + const proposalId = "proposal_00000000-0000-7000-8000-000000000001"; + const nodeId = "node_00000000-0000-7000-8000-000000000002"; + const operationId = "operation_00000000-0000-7000-8000-000000000003"; + const operation = { + kind: "add-node", + node: { + id: nodeId, + kind: "agent", + name: "Research", + purpose: "Research", + ownerAgentId: null, + contractRefs: [], + }, + }; + value.workspace.activeProposalId = proposalId; + value.workspace.recordVersion = 2; + value.proposal = { + schemaVersion: 1, + id: proposalId, + projectId, + baseRevisionId: null, + version: 1, + nodes: [operation.node], + relationships: [], + history: [ + { + id: operationId, + requestId: "request-1", + acceptedVersion: 1, + operation, + actor: { + userId: "user-1", + sessionId: "session-1", + role: "map-planner", + assignment: null, + }, + acceptedAt: timestamp, + }, + ], + createdAt: timestamp, + updatedAt: timestamp, + }; + + expect(parseAgentMapWorkspaceResponse(value, projectId)).toEqual(value); + value.proposal.history[0].operation.node.privatePath = "/secret"; + expect(() => parseAgentMapWorkspaceResponse(value, projectId)).toThrow( + "Invalid Agent Map workspace response", + ); + }); + it("uses the same public shape in mock mode", async () => { const api = new MockApi(); const state = await api.getState(); @@ -132,8 +186,7 @@ describe("resolveStudioWorkspaceSelection", () => { describe("mostSpecificStudioScope", () => { it("chooses the nearest containing durable project, not the first parent", () => { - const nestedProjectId = - "project_00000000-0000-4000-8000-000000000002"; + const nestedProjectId = "project_00000000-0000-4000-8000-000000000002"; expect( mostSpecificStudioScope( "/work/services/agent", diff --git a/packages/harness/web/src/lib/agent-map.ts b/packages/harness/web/src/lib/agent-map.ts index c71b944c..b6505527 100644 --- a/packages/harness/web/src/lib/agent-map.ts +++ b/packages/harness/web/src/lib/agent-map.ts @@ -6,6 +6,7 @@ import type { StudioCurrentWorkspaceResponse, StudioWorkspaceSelection, } from "@shared/agent-map"; +import { parseMapChangeProposal } from "@shared/agent-map-codec"; import type { WorkspaceScopeSummary } from "@shared/system-graph"; import { isWithinDir, stripTrailingSep } from "./paths"; @@ -154,12 +155,35 @@ function parseWorkspace( return value as unknown as AgentMapWorkspaceState; } +function parseProposal( + value: unknown, + projectId: string, + activeProposalId: string | null, +): AgentMapWorkspaceResponse["proposal"] | undefined { + if (value === null) return activeProposalId === null ? null : undefined; + if (activeProposalId === null) return undefined; + try { + return parseMapChangeProposal(value, projectId, activeProposalId); + } catch { + return undefined; + } +} + /** Strictly validates the path-free Agent Map HTTP boundary. */ export function parseAgentMapWorkspaceResponse( value: unknown, expectedProjectId?: string, ): AgentMapWorkspaceResponse { - if (!isRecord(value) || !hasExactKeys(value, ["project", "workspace"])) { + if ( + !isRecord(value) || + !hasExactKeys(value, [ + "schemaVersion", + "project", + "workspace", + "proposal", + ]) || + value.schemaVersion !== 1 + ) { throw new Error("Invalid Agent Map workspace response"); } const project = parseProject(value.project); @@ -171,7 +195,14 @@ export function parseAgentMapWorkspaceResponse( } const workspace = parseWorkspace(value.workspace, project.projectId); if (!workspace) throw new Error("Invalid Agent Map workspace response"); - return { project, workspace }; + const proposal = parseProposal( + value.proposal, + project.projectId, + workspace.activeProposalId, + ); + if (proposal === undefined) + throw new Error("Invalid Agent Map workspace response"); + return { schemaVersion: 1, project, workspace, proposal }; } function parseSelection( @@ -266,13 +297,12 @@ export function mostSpecificStudioScope( const projectIds = new Set(projects.map((project) => project.projectId)); return ( scopes - .filter( - (scope): scope is WorkspaceScopeSummary & { projectId: string } => - Boolean( - scope.projectId && - projectIds.has(scope.projectId) && - isWithinDir(scope.cwd, targetPath), - ), + .filter((scope): scope is WorkspaceScopeSummary & { projectId: string } => + Boolean( + scope.projectId && + projectIds.has(scope.projectId) && + isWithinDir(scope.cwd, targetPath), + ), ) .map((scope) => ({ scope, diff --git a/packages/harness/web/src/lib/api.ts b/packages/harness/web/src/lib/api.ts index de4da1f9..7fad5643 100644 --- a/packages/harness/web/src/lib/api.ts +++ b/packages/harness/web/src/lib/api.ts @@ -2152,6 +2152,7 @@ export class MockApi implements HarnessApi { } return parseAgentMapWorkspaceResponse( { + schemaVersion: 1, project, workspace: { projectId, @@ -2163,6 +2164,7 @@ export class MockApi implements HarnessApi { createdAt: project.createdAt, updatedAt: project.updatedAt, }, + proposal: null, }, projectId, ); @@ -2444,9 +2446,7 @@ export class MockApi implements HarnessApi { const retryFailure = typeof window === "undefined" ? null - : new URLSearchParams(window.location.search).get( - "mockGreetingRetry", - ); + : new URLSearchParams(window.location.search).get("mockGreetingRetry"); if (retryFailure === "error") { throw new ApiError( 503, diff --git a/packages/harness/web/tsconfig.json b/packages/harness/web/tsconfig.json index bd20af36..61a47fea 100644 --- a/packages/harness/web/tsconfig.json +++ b/packages/harness/web/tsconfig.json @@ -16,6 +16,7 @@ "@shared/types": ["../src/shared/types.ts"], "@shared/system-graph": ["../src/shared/system-graph.ts"], "@shared/agent-map": ["../src/shared/agent-map.ts"], + "@shared/agent-map-codec": ["../src/shared/agent-map-codec.ts"], "@shared/agent-name": ["../src/shared/agent-name.ts"], "@shared/render-local-run": ["../src/core/render-local-run.ts"], "@shared/stub-feedback": ["../src/core/stub-feedback.ts"], diff --git a/packages/harness/web/vite.config.ts b/packages/harness/web/vite.config.ts index 166d590b..39ada7a3 100644 --- a/packages/harness/web/vite.config.ts +++ b/packages/harness/web/vite.config.ts @@ -28,7 +28,9 @@ const STUDIO_VERSION = ( ).version; const DS_PACKAGE = "@sapiom/design-system"; -const DS_NEUTRAL_DIR = fileURLToPath(new URL("./src/styles/ds-neutral", import.meta.url)); +const DS_NEUTRAL_DIR = fileURLToPath( + new URL("./src/styles/ds-neutral", import.meta.url), +); /** * Resolve the design-system seam: prefer the private `@sapiom/design-system` @@ -88,25 +90,40 @@ export default defineConfig({ // resolves to the package's own canonical shared contract // (packages/harness/src/shared/types.ts) so the web and server always // build against one source of truth — no vendored copy to drift. - "@shared/types": fileURLToPath(new URL("../src/shared/types.ts", import.meta.url)), - "@shared/system-graph": fileURLToPath(new URL("../src/shared/system-graph.ts", import.meta.url)), - "@shared/agent-map": fileURLToPath(new URL("../src/shared/agent-map.ts", import.meta.url)), + "@shared/types": fileURLToPath( + new URL("../src/shared/types.ts", import.meta.url), + ), + "@shared/system-graph": fileURLToPath( + new URL("../src/shared/system-graph.ts", import.meta.url), + ), + "@shared/agent-map": fileURLToPath( + new URL("../src/shared/agent-map.ts", import.meta.url), + ), + "@shared/agent-map-codec": fileURLToPath( + new URL("../src/shared/agent-map-codec.ts", import.meta.url), + ), // One agent-name rule for the dialog and the create route: a name the // field accepts and the server refuses reads as a broken app. - "@shared/agent-name": fileURLToPath(new URL("../src/shared/agent-name.ts", import.meta.url)), + "@shared/agent-name": fileURLToPath( + new URL("../src/shared/agent-name.ts", import.meta.url), + ), // The local-run mapper is a pure fn shared with the server (its canonical // home is src/core/render-local-run.ts, per the ticket). The SPA imports // the SAME implementation to map an offline stub run's NDJSON traces into // the RunView the inspector renders — one mapper, no client/server drift. // It pulls in only the `LocalStepTrace` *type* from agent-core (erased at // build), so no agent-core runtime code enters the browser bundle. - "@shared/render-local-run": fileURLToPath(new URL("../src/core/render-local-run.ts", import.meta.url)), + "@shared/render-local-run": fileURLToPath( + new URL("../src/core/render-local-run.ts", import.meta.url), + ), // The stub-feedback derivations (stubbed-chip + read-only stub notice) are // pure fns over RunView; the SPA imports the SAME canonical implementation // the unit tests target so the inspector can never disagree with the tests. // Types-only import of RunView (erased at build) — no server code enters // the browser bundle. - "@shared/stub-feedback": fileURLToPath(new URL("../src/core/stub-feedback.ts", import.meta.url)), + "@shared/stub-feedback": fileURLToPath( + new URL("../src/core/stub-feedback.ts", import.meta.url), + ), // The design system is a private package. Official builds (private // package installed) render branded; public clones fall back to a // committed neutral token set. See designSystemAlias() above. diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 752fcd16..f8307e8d 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -398,6 +398,9 @@ importers: typescript: specifier: ~5.9.3 version: 5.9.3 + uuid: + specifier: ^10.0.0 + version: 10.0.0 ws: specifier: ^8.18.0 version: 8.21.0 @@ -426,6 +429,9 @@ importers: '@types/react-dom': specifier: ^19.0.0 version: 19.2.3(@types/react@19.2.17) + '@types/uuid': + specifier: ^10.0.0 + version: 10.0.0 '@types/ws': specifier: ^8.5.10 version: 8.18.1