diff --git a/changelog/entries/2026-06-22-mcp-next-actions.json b/changelog/entries/2026-06-22-mcp-next-actions.json new file mode 100644 index 000000000..7f7f902f1 --- /dev/null +++ b/changelog/entries/2026-06-22-mcp-next-actions.json @@ -0,0 +1,10 @@ +{ + "id": "2026-06-22-mcp-next-actions", + "version": "0.9.4", + "date": "2026-06-22", + "category": "feat", + "title": "MCP tool errors now carry recovery steps", + "summary": "Every failed MCP tool call returns structured next_actions — concrete recovery steps (and the tool to call) so an agent can recover in one turn instead of guessing.", + "features": ["mcp", "agents"], + "mcpTools": [] +} diff --git a/packages/mcp/src/__tests__/dispatch-persist.test.ts b/packages/mcp/src/__tests__/dispatch-persist.test.ts index d3077fd82..7a4d113a8 100644 --- a/packages/mcp/src/__tests__/dispatch-persist.test.ts +++ b/packages/mcp/src/__tests__/dispatch-persist.test.ts @@ -25,11 +25,14 @@ interface Fake { rows: Map>; posts: number; gets: number; + /** Appends to the event spine (rpc/append_session_event) — counted apart + * from the document persist POSTs so each is asserted independently. */ + appends: number; } function installFake(): Fake { const rows = new Map>(); - const fake: Fake = { rows, posts: 0, gets: 0 }; + const fake: Fake = { rows, posts: 0, gets: 0, appends: 0 }; const eq = (sp: URLSearchParams, k: string): string | null => { const v = sp.get(k); return v && v.startsWith("eq.") ? v.slice(3) : null; @@ -37,6 +40,15 @@ function installFake(): Fake { setSessionFetch((async (input: unknown, init: RequestInit = {}) => { const url = new URL(String(input)); const method = (init.method ?? "GET").toUpperCase(); + // Event spine append — a kernel mutation logs one row. Separate endpoint, + // separate counter; returns the RPC's jsonb result. + if (method === "POST" && url.pathname.endsWith("/rpc/append_session_event")) { + fake.appends++; + return new Response(JSON.stringify({ ok: true, id: fake.appends, seq: fake.appends }), { + status: 200, + headers: { "Content-Type": "application/json" }, + }); + } if (method === "POST") { fake.posts++; const body = JSON.parse(String(init.body)) as Array< @@ -118,8 +130,11 @@ describe("dispatch persist wrapper (end-to-end through createServer)", () => { const documentId = JSON.parse(firstText(open)).document_id as string; expect(fake.posts).toBe(1); expect(fake.rows.has(`user-1|mcp:${documentId}`)).toBe(true); + // A creator is a kernel mutation → exactly one event on the spine. + expect(fake.appends).toBe(1); const postsAfterOpen = fake.posts; + const appendsAfterOpen = fake.appends; // Reader: get_document runs in a fresh per-request scope, so it can only // succeed by hydrating from the durable store — and it must NOT persist. @@ -130,6 +145,7 @@ describe("dispatch persist wrapper (end-to-end through createServer)", () => { expect(JSON.parse(firstText(got)).version).toBeDefined(); // IR round-tripped expect(fake.gets).toBeGreaterThanOrEqual(1); // hydrate read happened expect(fake.posts).toBe(postsAfterOpen); // reader did not write + expect(fake.appends).toBe(appendsAfterOpen); // reader emitted no event await client.close(); await server.close(); @@ -152,6 +168,8 @@ describe("dispatch persist wrapper (end-to-end through createServer)", () => { const documentId = JSON.parse(firstText(res)).document_id as string; expect(fake.posts).toBe(1); expect(fake.rows.has(`user-2|mcp:${documentId}`)).toBe(true); + // …and the same call logged exactly one kernel event to the spine. + expect(fake.appends).toBe(1); await client.close(); await server.close(); diff --git a/packages/mcp/src/__tests__/live.test.ts b/packages/mcp/src/__tests__/live.test.ts new file mode 100644 index 000000000..1f4e857d4 --- /dev/null +++ b/packages/mcp/src/__tests__/live.test.ts @@ -0,0 +1,147 @@ +import { describe, it, expect } from "vitest"; +import { appendOverlay, listEvents, OVERLAY_TYPES } from "../tools/live.js"; +import type { + SessionEvent, + SessionEventStore, + StoredSessionEvent, +} from "../session-store.js"; + +class FakeEventStore implements SessionEventStore { + public appended: Array<{ sessionId: string } & SessionEvent> = []; + public lastSince: number | undefined; + private rows: StoredSessionEvent[] = []; + seed(rows: StoredSessionEvent[]): void { + this.rows = rows; + } + async append(sessionId: string, evt: SessionEvent): Promise { + this.appended.push({ sessionId, ...evt }); + } + async list(_sessionId: string, sinceSeq?: number): Promise { + this.lastSince = sinceSeq; + return this.rows; + } +} + +const row = (seq: number, over: Partial = {}): StoredSessionEvent => ({ + id: seq, + seq, + session_id: "doc_x", + author: "agent", + kind: "kernel", + type: "create", + payload: {}, + created_at: new Date(0).toISOString(), + ...over, +}); + +describe("appendOverlay", () => { + it("appends a valid pin as a kind:'overlay' event with the anchor payload", async () => { + const es = new FakeEventStore(); + const res = await appendOverlay(es, "doc_x", { + type: "pin", + payload: { anchor: { node: 3, face: 1 }, text: "too thin" }, + author: "reviewer", + }); + expect(res.ok).toBe(true); + expect(es.appended).toHaveLength(1); + const evt = es.appended[0]; + expect(evt.kind).toBe("overlay"); + expect(evt.type).toBe("pin"); + // An untrusted (anonymous) author is namespaced so it can't impersonate. + expect(evt.author).toBe("viewer:reviewer"); + expect(evt.payload).toEqual({ anchor: { node: 3, face: 1 }, text: "too thin" }); + }); + + it("namespaces an anonymous author as viewer: and caps it", async () => { + const es = new FakeEventStore(); + await appendOverlay(es, "doc_x", { type: "flag", payload: {} }); + expect(es.appended[0].author).toBe("viewer:anon"); + + const es2 = new FakeEventStore(); + await appendOverlay(es2, "doc_x", { type: "note", payload: {}, author: "x".repeat(200) }); + expect(es2.appended[0].author).toBe("viewer:" + "x".repeat(48)); + }); + + it("never lets an anonymous author impersonate a reserved identity", async () => { + const es = new FakeEventStore(); + await appendOverlay(es, "doc_x", { type: "pin", payload: {}, author: "agent" }); + expect(es.appended[0].author).toBe("viewer:agent"); // not the reserved "agent" + }); + + it("uses the verified identity authoritatively, ignoring a forged body author", async () => { + const es = new FakeEventStore(); + await appendOverlay( + es, + "doc_x", + { type: "pin", payload: {}, author: "victim@x.z" }, + { trustedAuthor: "alice@x.z" }, + ); + expect(es.appended[0].author).toBe("alice@x.z"); + }); + + it("rejects an oversized payload (no spine bloat / broadcast amplification)", async () => { + const es = new FakeEventStore(); + const res = await appendOverlay(es, "doc_x", { + type: "pin", + payload: { blob: "x".repeat(5000) }, + }); + expect(res.ok).toBe(false); + if (!res.ok) expect(res.error).toContain("too large"); + expect(es.appended).toHaveLength(0); + }); + + it("rejects an unknown overlay type (never reaches the store)", async () => { + const es = new FakeEventStore(); + const res = await appendOverlay(es, "doc_x", { type: "mutation", payload: {} }); + expect(res.ok).toBe(false); + expect(es.appended).toHaveLength(0); + }); + + it("rejects a missing session id", async () => { + const es = new FakeEventStore(); + const res = await appendOverlay(es, "", { type: "pin", payload: {} }); + expect(res.ok).toBe(false); + }); + + it("tolerates a non-object payload", async () => { + const es = new FakeEventStore(); + await appendOverlay(es, "doc_x", { type: "pin", payload: "oops" as unknown }); + expect(es.appended[0].payload).toEqual({}); + }); + + it("accepts every declared overlay type", async () => { + for (const t of OVERLAY_TYPES) { + const es = new FakeEventStore(); + const res = await appendOverlay(es, "doc_x", { type: t, payload: {} }); + expect(res.ok).toBe(true); + } + }); +}); + +describe("listEvents", () => { + it("returns all events in seq order", async () => { + const es = new FakeEventStore(); + es.seed([row(1), row(2), row(3)]); + const out = await listEvents(es, "doc_x"); + expect(out.map((e) => e.seq)).toEqual([1, 2, 3]); + }); + + it("returns only events after sinceSeq (late-join catch-up)", async () => { + const es = new FakeEventStore(); + es.seed([row(1), row(2), row(3), row(4)]); + const out = await listEvents(es, "doc_x", 2); + expect(out.map((e) => e.seq)).toEqual([3, 4]); + }); + + it("pushes sinceSeq down to the store (server-side filter)", async () => { + const es = new FakeEventStore(); + es.seed([row(3), row(4)]); + await listEvents(es, "doc_x", 2); + expect(es.lastSince).toBe(2); + }); + + it("returns [] for a missing session id", async () => { + const es = new FakeEventStore(); + expect(await listEvents(es, "")).toEqual([]); + }); +}); diff --git a/packages/mcp/src/__tests__/next-actions.test.ts b/packages/mcp/src/__tests__/next-actions.test.ts new file mode 100644 index 000000000..8d6d0df40 --- /dev/null +++ b/packages/mcp/src/__tests__/next-actions.test.ts @@ -0,0 +1,193 @@ +import { describe, it, expect } from "vitest"; +import { + suggestNextActions, + buildErrorResult, + enrichErrorResult, +} from "../tools/next-actions.js"; + +describe("suggestNextActions", () => { + it("steers a kernel trap toward different inputs (no blind retry tool)", () => { + const actions = suggestNextActions("fillet", { document_id: "d" }, "unreachable", { + kernelTrap: true, + }); + expect(actions).toHaveLength(1); + expect(actions[0].action.toLowerCase()).toContain("kernel reset"); + expect(actions[0].tool).toBeUndefined(); + }); + + it("points an unknown document_id at open_document", () => { + const actions = suggestNextActions( + "update", + { document_id: "doc_x" }, + "Unknown document_id: doc_x", + ); + expect(actions[0].tool).toBe("open_document"); + }); + + it("points a missing/bad part_id at read, carrying the document_id", () => { + const a1 = suggestNextActions("delete", { document_id: "d1" }, "delete: missing `part_id`"); + expect(a1[0].tool).toBe("read"); + expect(a1[0].args).toEqual({ document_id: "d1" }); + + const a2 = suggestNextActions("read", { document_id: "d1" }, 'read: no part with id "7"'); + expect(a2[0].tool).toBe("read"); + }); + + it("returns the type's param hint for a malformed create", () => { + const actions = suggestNextActions( + "create", + { document_id: "d", type: "cylinder", params: {} }, + "missing field `radius`", + ); + // First action carries the cylinder shape; a follow-up names the field. + expect(actions[0].tool).toBe("create"); + expect(actions[0].action).toContain("radius"); + expect(actions.some((a) => a.action.includes('"radius"'))).toBe(true); + }); + + it("adds a children-first hint for a malformed boolean", () => { + const actions = suggestNextActions( + "create", + { document_id: "d", type: "difference", params: {} }, + "invalid node reference", + ); + expect(actions.some((a) => a.action.toLowerCase().includes("child nodes first"))).toBe(true); + }); + + it("falls back to the catalog hint for an unknown create type", () => { + const actions = suggestNextActions("create", { document_id: "d", type: "blob" }, "bad type"); + expect(actions[0].tool).toBe("create"); + expect(actions[0].action.toLowerCase()).toContain("type catalog"); + }); + + it("marks a planner-unavailable error as non-recoverable from the client", () => { + const actions = suggestNextActions("create", {}, "Rust planner unavailable"); + expect(actions[0].tool).toBeUndefined(); + expect(actions[0].action.toLowerCase()).toContain("misconfiguration"); + }); + + it("routes a missing schematic to create_schematic", () => { + const actions = suggestNextActions( + "place_components", + { document_id: "d" }, + "Error: Document has no schematic", + ); + expect(actions[0].tool).toBe("create_schematic"); + expect(actions[0].args).toEqual({ document_id: "d" }); + }); + + it("routes a missing board to place_components", () => { + const actions = suggestNextActions( + "route_nets", + { document_id: "d" }, + "Error: Document has no PCB — run place_components first", + ); + expect(actions[0].tool).toBe("place_components"); + }); + + it("routes an unknown catalog part to search_parts (not read)", () => { + const actions = suggestNextActions( + "place_part", + { document_id: "d" }, + 'place_part: unknown part "m3_screw"', + ); + expect(actions[0].tool).toBe("search_parts"); + }); + + it("lets a generic update error fall through to the inspect floor", () => { + // `update` has no `type`, so it must not get a create-flavored catalog hint. + const actions = suggestNextActions("update", { document_id: "d" }, "some update failure"); + expect(actions[0].tool).toBe("read"); + expect(actions[0].args).toEqual({ document_id: "d" }); + }); + + it("falls back to inspect-then-retry for an unrecognized error", () => { + const actions = suggestNextActions("export_cad", { document_id: "d" }, "weird failure"); + expect(actions[0].tool).toBe("read"); + expect(actions[0].args).toEqual({ document_id: "d" }); + }); + + it("omits args when no document_id is in context", () => { + const actions = suggestNextActions("export_cad", {}, "weird failure"); + expect(actions[0].tool).toBe("read"); + expect(actions[0].args).toBeUndefined(); + }); +}); + +describe("buildErrorResult", () => { + it("carries the message, a machine-readable tail, and structured actions", () => { + const res = buildErrorResult("delete", { document_id: "d1" }, "delete: missing `part_id`"); + expect(res.isError).toBe(true); + const text = res.content[0].text; + expect(text).toContain("Error: delete: missing `part_id`"); + expect(text).toContain("next_actions:"); + // The tail is valid JSON the agent can parse. + const tail = text.slice(text.indexOf("next_actions:") + "next_actions:".length).trim(); + expect(JSON.parse(tail)).toEqual(res.structuredContent.next_actions); + expect(res.structuredContent.next_actions[0].tool).toBe("read"); + expect(res.structuredContent.error).toBe("delete: missing `part_id`"); + }); + + it("uses the kernel-trap headline for a trap", () => { + const res = buildErrorResult("fillet", { document_id: "d" }, "unreachable", { + kernelTrap: true, + }); + expect(res.content[0].text).toContain("kernel trap during 'fillet'"); + expect(res.content[0].text).toContain("was reset"); + }); +}); + +describe("enrichErrorResult (for tools that return isError instead of throwing)", () => { + it("attaches next_actions to a plain-text ECAD error (appended tail)", () => { + const result = { + content: [{ type: "text", text: "Error: Document has no schematic" }], + isError: true, + }; + enrichErrorResult(result, "place_components", { document_id: "d" }); + expect(result.content[0].text).toContain("next_actions:"); + expect( + (result as { structuredContent?: { next_actions?: Array<{ tool?: string }> } }) + .structuredContent?.next_actions?.[0].tool, + ).toBe("create_schematic"); + }); + + it("injects next_actions INTO a JSON error body, keeping it parseable", () => { + const result = { + content: [ + { type: "text", text: JSON.stringify({ error: 'place_part: unknown part "x"' }) }, + ], + isError: true, + }; + enrichErrorResult(result, "place_part", { document_id: "d" }); + const parsed = JSON.parse(result.content[0].text) as { + error: string; + next_actions: Array<{ tool?: string }>; + }; + expect(parsed.error).toContain("unknown part"); + expect(parsed.next_actions[0].tool).toBe("search_parts"); + }); + + it("is idempotent — a result already carrying next_actions is untouched", () => { + const result = { + content: [{ type: "text", text: "Error: x\nnext_actions: [{}]" }], + structuredContent: { next_actions: [{ action: "already here" }] }, + isError: true as const, + }; + const before = result.content[0].text; + enrichErrorResult(result, "route_nets", {}); + expect(result.content[0].text).toBe(before); + }); + + it("skips carve-outs (unknown tool / disabled pack)", () => { + const result = { content: [{ type: "text", text: "Unknown tool: frobnicate" }], isError: true }; + enrichErrorResult(result, "frobnicate", {}); + expect(result.content[0].text).toBe("Unknown tool: frobnicate"); + expect((result as { structuredContent?: unknown }).structuredContent).toBeUndefined(); + }); + + it("is a no-op on a successful result", () => { + const result = { content: [{ type: "text", text: "ok" }], isError: false }; + enrichErrorResult(result, "create", {}); + expect(result.content[0].text).toBe("ok"); + }); +}); diff --git a/packages/mcp/src/__tests__/ordering.test.ts b/packages/mcp/src/__tests__/ordering.test.ts new file mode 100644 index 000000000..5f751944a --- /dev/null +++ b/packages/mcp/src/__tests__/ordering.test.ts @@ -0,0 +1,258 @@ +import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import { authorizeSpend, placeOrder } from "../tools/ordering.js"; +import { InMemoryFabricateStore } from "../fabricate/store.js"; +import type { Order, SpendAuthorization } from "../fabricate/types.js"; +import type { AuthUser } from "../oauth.js"; +import type { SessionEvent, SessionEventStore, StoredSessionEvent } from "../session-store.js"; + +/** Records control events so we can assert the spine got them. */ +class RecordingEventStore implements SessionEventStore { + public events: Array = []; + async append(sessionId: string, evt: SessionEvent): Promise { + this.events.push({ sessionId, ...evt }); + } + async list(): Promise { + return []; + } + types(): string[] { + return this.events.map((e) => e.type); + } +} + +function makeOrder(over: Partial = {}): Order { + const now = new Date().toISOString(); + return { + order_id: "ord_1", + document_id: "doc_x", + quote_id: "q_1", + state: "QUOTED", + fab: "digitalmetal", + fab_order_ref: null, + amount_total_minor: 5000, + currency: "USD", + ship_to: null, + events: [{ state: "QUOTED", at: now, note: "quote" }], + created_at: now, + updated_at: now, + ...over, + }; +} + +const text = (r: { content: Array<{ text: string }> }) => r.content[0].text; +const json = (r: { content: Array<{ text: string }> }) => JSON.parse(text(r)); + +describe("Fabricate ordering — disabled by default (flag gate)", () => { + beforeEach(() => delete process.env.VCAD_FABRICATE_ORDERING); + + it("authorize_spend refuses unless VCAD_FABRICATE_ORDERING=1", async () => { + const res = await authorizeSpend({ order_id: "ord_1" }, new InMemoryFabricateStore(), new RecordingEventStore(), null); + expect(res.isError).toBe(true); + expect(text(res)).toContain("disabled"); + }); + + it("place_order refuses unless VCAD_FABRICATE_ORDERING=1", async () => { + const res = await placeOrder( + { order_id: "ord_1", authorization_id: "a" }, + new InMemoryFabricateStore(), + new RecordingEventStore(), + null, + ); + expect(res.isError).toBe(true); + expect(text(res)).toContain("disabled"); + }); +}); + +describe("Fabricate ordering — enabled (test-mode)", () => { + let prev: string | undefined; + beforeEach(() => { + prev = process.env.VCAD_FABRICATE_ORDERING; + process.env.VCAD_FABRICATE_ORDERING = "1"; + }); + afterEach(() => { + if (prev === undefined) delete process.env.VCAD_FABRICATE_ORDERING; + else process.env.VCAD_FABRICATE_ORDERING = prev; + }); + + it("runs propose → (human approve) → place: debits once, moves to PAID, emits spine events", async () => { + const user: AuthUser = { sub: "u-happy", email: "x@y.z" }; + const store = new InMemoryFabricateStore(); + const es = new RecordingEventStore(); + await store.saveOrder(makeOrder(), 4000, user.sub); + + // 1. Agent proposes — pending_human, propose_order on the spine, no money moved. + const aRes = await authorizeSpend({ order_id: "ord_1" }, store, es, user); + const a = json(aRes); + expect(a.status).toBe("pending_human"); + expect(es.types()).toContain("propose_order"); + const authId = a.authorization_id as string; + + // 2. Placing BEFORE human approval is refused. + const early = await placeOrder({ order_id: "ord_1", authorization_id: authId }, store, es, user); + expect(early.isError).toBe(true); + expect(text(early)).toContain("pending human approval"); + + // 3. The human approves (web app, simulated) and the wallet has credit. + expect(store.approveAuthorizationForTest(authId, user.sub)).toBe(true); + store.creditWalletForTest(user.sub, 10000); + + // 4. Now place succeeds: PAID, order_placed emitted. + const pRes = await placeOrder({ order_id: "ord_1", authorization_id: authId }, store, es, user); + expect(pRes.isError).toBeFalsy(); + expect(json(pRes).state).toBe("PAID"); + expect(es.types()).toContain("order_placed"); + expect((await store.getOrder("ord_1", user.sub))?.state).toBe("PAID"); + }); + + it("refuses place_order with an unknown authorization", async () => { + const user: AuthUser = { sub: "u-noauth", email: "x@y.z" }; + const store = new InMemoryFabricateStore(); + await store.saveOrder(makeOrder(), 4000, user.sub); + const res = await placeOrder({ order_id: "ord_1", authorization_id: "nope" }, store, new RecordingEventStore(), user); + expect(res.isError).toBe(true); + expect(text(res)).toContain("Unknown authorization_id"); + }); + + it("fails payment on insufficient funds and leaves the order QUOTED", async () => { + const user: AuthUser = { sub: "u-broke", email: "x@y.z" }; + const store = new InMemoryFabricateStore(); + const es = new RecordingEventStore(); + await store.saveOrder(makeOrder(), 4000, user.sub); + const a = json(await authorizeSpend({ order_id: "ord_1" }, store, es, user)); + store.approveAuthorizationForTest(a.authorization_id, user.sub); + // No credit added → balance 0 < 5000. + + const res = await placeOrder({ order_id: "ord_1", authorization_id: a.authorization_id }, store, es, user); + expect(res.isError).toBe(true); + expect(text(res)).toContain("insufficient_funds"); + expect((await store.getOrder("ord_1", user.sub))?.state).toBe("QUOTED"); + expect(es.types()).toContain("order_payment_failed"); + }); + + it("debits at most once across a retry (idempotent)", async () => { + const user: AuthUser = { sub: "u-idem", email: "x@y.z" }; + const store = new InMemoryFabricateStore(); + const a = { authorization_id: "" } as { authorization_id: string }; + await store.saveOrder(makeOrder(), 4000, user.sub); + a.authorization_id = json(await authorizeSpend({ order_id: "ord_1" }, store, new RecordingEventStore(), user)) + .authorization_id; + store.approveAuthorizationForTest(a.authorization_id, user.sub); + store.creditWalletForTest(user.sub, 10000); + + const first = await store.debit({ + userId: user.sub, + amountMinor: 5000, + orderId: "ord_1", + authorizationId: a.authorization_id, + idempotencyKey: "ord_1:debit", + }); + const second = await store.debit({ + userId: user.sub, + amountMinor: 5000, + orderId: "ord_1", + authorizationId: a.authorization_id, + idempotencyKey: "ord_1:debit", + }); + expect(first.ok).toBe(true); + expect(first.balance_minor).toBe(5000); // 10000 - 5000 + expect(second.ok).toBe(true); + expect(second.idempotent).toBe(true); + expect(second.balance_minor).toBe(5000); // not double-debited + }); + + it("won't authorize a non-QUOTED order", async () => { + const user: AuthUser = { sub: "u-paid", email: "x@y.z" }; + const store = new InMemoryFabricateStore(); + await store.saveOrder(makeOrder({ state: "PAID" }), 4000, user.sub); + const res = await authorizeSpend({ order_id: "ord_1" }, store, new RecordingEventStore(), user); + expect(res.isError).toBe(true); + expect(text(res)).toContain("not QUOTED"); + }); + + it("rejects a spend ceiling below the order total", async () => { + const user: AuthUser = { sub: "u-low", email: "x@y.z" }; + const store = new InMemoryFabricateStore(); + await store.saveOrder(makeOrder(), 4000, user.sub); + const res = await authorizeSpend({ order_id: "ord_1", max_amount_minor: 100 }, store, new RecordingEventStore(), user); + expect(res.isError).toBe(true); + expect(text(res)).toContain("below the order total"); + }); + + it("refuses an expired (but authorized) authorization — leaves the order QUOTED", async () => { + const user: AuthUser = { sub: "u-exp", email: "x@y.z" }; + const store = new InMemoryFabricateStore(); + await store.saveOrder(makeOrder(), 4000, user.sub); + const past = new Date(Date.now() - 3_600_000).toISOString(); + const authz: SpendAuthorization = { + id: "authz-exp", + user_id: user.sub, + quote_id: "q_1", + kind: "one_time", + max_amount_minor: 5000, + daily_cap_minor: null, + process_allowlist: null, + fab_allowlist: null, + doc_hash: null, + status: "authorized", + expires_at: past, + created_at: past, + }; + await store.createAuthorization(authz, user.sub); + store.creditWalletForTest(user.sub, 10000); + + const res = await placeOrder({ order_id: "ord_1", authorization_id: "authz-exp" }, store, new RecordingEventStore(), user); + expect(res.isError).toBe(true); + expect(text(res)).toContain("expired"); + expect((await store.getOrder("ord_1", user.sub))?.state).toBe("QUOTED"); + }); + + it("finalizes a paid-but-unrecorded order on retry without double-debiting (the blocker)", async () => { + const user: AuthUser = { sub: "u-crash", email: "x@y.z" }; + const store = new InMemoryFabricateStore(); + const es = new RecordingEventStore(); + await store.saveOrder(makeOrder(), 4000, user.sub); + const a = json(await authorizeSpend({ order_id: "ord_1" }, store, es, user)); + store.approveAuthorizationForTest(a.authorization_id, user.sub); + store.creditWalletForTest(user.sub, 10000); + + // Simulate a crash AFTER debit but BEFORE setOrderState: the debit commits + // (authz consumed, balance down), but the order is left QUOTED. + const d = await store.debit({ + userId: user.sub, + amountMinor: 5000, + orderId: "ord_1", + authorizationId: a.authorization_id, + idempotencyKey: "ord_1:debit", + }); + expect(d.ok).toBe(true); + expect((await store.getAuthorization(a.authorization_id, user.sub))?.status).toBe("consumed"); + expect((await store.getOrder("ord_1", user.sub))?.state).toBe("QUOTED"); + + // Retry: must finalize to PAID via an idempotent replay — not a 2nd charge, + // not a permanent strand. + const res = await placeOrder({ order_id: "ord_1", authorization_id: a.authorization_id }, store, es, user); + expect(res.isError).toBeFalsy(); + expect(json(res).state).toBe("PAID"); + expect(json(res).idempotent).toBe(true); + expect((await store.getOrder("ord_1", user.sub))?.state).toBe("PAID"); + }); + + it("rejects a reused idempotency key whose terms differ (no silent replay)", async () => { + const user: AuthUser = { sub: "u-reuse", email: "x@y.z" }; + const store = new InMemoryFabricateStore(); + await store.saveOrder(makeOrder(), 4000, user.sub); + const a = json(await authorizeSpend({ order_id: "ord_1" }, store, new RecordingEventStore(), user)); + store.approveAuthorizationForTest(a.authorization_id, user.sub); + store.creditWalletForTest(user.sub, 10000); + const key = "shared-key"; + + const first = await store.debit({ + userId: user.sub, amountMinor: 5000, orderId: "ord_1", authorizationId: a.authorization_id, idempotencyKey: key, + }); + expect(first.ok).toBe(true); + const reused = await store.debit({ + userId: user.sub, amountMinor: 9999, orderId: "ord_1", authorizationId: a.authorization_id, idempotencyKey: key, + }); + expect(reused.ok).toBe(false); + expect(reused.reason).toBe("idempotency_key_reused"); + }); +}); diff --git a/packages/mcp/src/__tests__/session-events.test.ts b/packages/mcp/src/__tests__/session-events.test.ts new file mode 100644 index 000000000..c1de5b1a1 --- /dev/null +++ b/packages/mcp/src/__tests__/session-events.test.ts @@ -0,0 +1,225 @@ +import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import { + SupabaseSessionEventStore, + NoopSessionEventStore, + createSessionEventStore, + setSessionFetch, + type SessionEvent, +} from "../session-store.js"; + +/** + * In-memory stand-in for the `session_events` table + the + * `append_session_event` RPC. Reproduces the RPC's contract — per-session + * monotonic `seq`, per-(session, key) idempotent replay — so the store's real + * request shaping (RPC body, GET filters) is exercised without a network or a + * live Postgres. + */ +function makeEventsFake() { + const sessions = new Map>>(); + const seenBodies: Array> = []; + let nextId = 1; + + const fetchImpl = (async (input: unknown, init: RequestInit = {}) => { + const url = new URL(String(input)); + const method = (init.method ?? "GET").toUpperCase(); + + // append_session_event RPC + if (method === "POST" && url.pathname.endsWith("/rpc/append_session_event")) { + const b = JSON.parse(String(init.body)) as Record; + seenBodies.push(b); + const sid = String(b.p_session_id); + const key = String(b.p_idempotency_key); + const list = sessions.get(sid) ?? []; + const existing = list.find((r) => r.idempotency_key === key); + if (existing) { + return jsonResponse({ + ok: true, + idempotent: true, + id: existing.id, + seq: existing.seq, + }); + } + const seq = list.length + 1; + const id = nextId++; + const row = { + id, + seq, + session_id: sid, + user_id: b.p_user ?? null, + author: b.p_author, + kind: b.p_kind, + type: b.p_type, + payload: b.p_payload ?? {}, + idempotency_key: key, + created_at: new Date(0).toISOString(), + }; + list.push(row); + sessions.set(sid, list); + return jsonResponse({ ok: true, id, seq }); + } + + // GET /session_events?session_id=eq.X&order=seq.asc + if (method === "GET" && url.pathname.endsWith("/session_events")) { + const sidParam = url.searchParams.get("session_id"); + const sid = sidParam?.startsWith("eq.") ? sidParam.slice(3) : null; + const rows = (sid && sessions.get(sid)) || []; + return jsonResponse(rows); + } + + return new Response("unexpected", { status: 400 }); + }) as unknown as typeof fetch; + + return { sessions, seenBodies, fetchImpl }; +} + +function jsonResponse(body: unknown): Response { + return new Response(JSON.stringify(body), { + status: 200, + headers: { "Content-Type": "application/json" }, + }); +} + +const CFG = { + supabaseUrl: "https://supa.test", + serviceRoleKey: "service-role-key", + userId: "user-me", +}; + +const kernelEvt = (overrides: Partial = {}): SessionEvent => ({ + author: "agent", + kind: "kernel", + type: "create", + payload: { tool: "create", args: { kind: "cube" } }, + ...overrides, +}); + +afterEach(() => { + setSessionFetch(((...args: Parameters) => + fetch(...args)) as typeof fetch); +}); + +describe("SupabaseSessionEventStore", () => { + it("assigns per-session monotonic seq and lists in order", async () => { + const fake = makeEventsFake(); + setSessionFetch(fake.fetchImpl); + const store = new SupabaseSessionEventStore(CFG); + + await store.append("doc_a", kernelEvt({ type: "create" })); + await store.append("doc_a", kernelEvt({ type: "update" })); + await store.append("doc_a", kernelEvt({ type: "delete" })); + + const events = await store.list("doc_a"); + expect(events.map((e) => e.seq)).toEqual([1, 2, 3]); + expect(events.map((e) => e.type)).toEqual(["create", "update", "delete"]); + }); + + it("replays idempotently on a repeated (session, key) — no new row", async () => { + const fake = makeEventsFake(); + setSessionFetch(fake.fetchImpl); + const store = new SupabaseSessionEventStore(CFG); + + await store.append("doc_a", kernelEvt({ idempotencyKey: "k1" })); + await store.append("doc_a", kernelEvt({ idempotencyKey: "k1" })); + + const events = await store.list("doc_a"); + expect(events).toHaveLength(1); + expect(events[0].seq).toBe(1); + }); + + it("scopes seq per session — independent counters", async () => { + const fake = makeEventsFake(); + setSessionFetch(fake.fetchImpl); + const store = new SupabaseSessionEventStore(CFG); + + await store.append("doc_a", kernelEvt()); + await store.append("doc_b", kernelEvt()); + await store.append("doc_b", kernelEvt()); + + expect((await store.list("doc_a")).map((e) => e.seq)).toEqual([1]); + expect((await store.list("doc_b")).map((e) => e.seq)).toEqual([1, 2]); + }); + + it("round-trips the payload and forwards the caller's user_id", async () => { + const fake = makeEventsFake(); + setSessionFetch(fake.fetchImpl); + const store = new SupabaseSessionEventStore(CFG); + + const payload = { tool: "update", args: { node: 3 }, changed: { modified: [{ part_id: "0" }] } }; + await store.append("doc_a", kernelEvt({ payload })); + + const [evt] = await store.list("doc_a"); + expect(evt.payload).toEqual(payload); + // The RPC body carries the caller's user id, not tool input. + expect(fake.seenBodies[0].p_user).toBe("user-me"); + expect(fake.seenBodies[0].p_kind).toBe("kernel"); + }); + + it("generates an idempotency key when none is supplied", async () => { + const fake = makeEventsFake(); + setSessionFetch(fake.fetchImpl); + const store = new SupabaseSessionEventStore(CFG); + + await store.append("doc_a", kernelEvt()); + await store.append("doc_a", kernelEvt()); + + // Distinct generated keys → two distinct rows. + expect(await store.list("doc_a")).toHaveLength(2); + const keys = fake.seenBodies.map((b) => b.p_idempotency_key); + expect(keys[0]).not.toBe(keys[1]); + }); + + it("passes null user_id for an anonymous session", async () => { + const fake = makeEventsFake(); + setSessionFetch(fake.fetchImpl); + const store = new SupabaseSessionEventStore({ ...CFG, userId: null }); + await store.append("doc_anon", kernelEvt()); + expect(fake.seenBodies[0].p_user).toBeNull(); + }); + + it("never throws on a transport error (best-effort append)", async () => { + setSessionFetch((() => { + throw new Error("network down"); + }) as unknown as typeof fetch); + const store = new SupabaseSessionEventStore(CFG); + await expect(store.append("doc_a", kernelEvt())).resolves.toBeUndefined(); + }); +}); + +describe("NoopSessionEventStore", () => { + it("append is a no-op and list is empty", async () => { + const store = new NoopSessionEventStore(); + await expect(store.append("doc_a", kernelEvt())).resolves.toBeUndefined(); + expect(await store.list("doc_a")).toEqual([]); + }); +}); + +describe("createSessionEventStore factory", () => { + let prevUrl: string | undefined; + let prevKey: string | undefined; + + beforeEach(() => { + prevUrl = process.env.SUPABASE_URL; + prevKey = process.env.SUPABASE_SERVICE_ROLE_KEY; + }); + afterEach(() => { + if (prevUrl === undefined) delete process.env.SUPABASE_URL; + else process.env.SUPABASE_URL = prevUrl; + if (prevKey === undefined) delete process.env.SUPABASE_SERVICE_ROLE_KEY; + else process.env.SUPABASE_SERVICE_ROLE_KEY = prevKey; + }); + + it("returns the Supabase spine with url + key (signed-in or anon)", () => { + process.env.SUPABASE_URL = "https://supa.test/"; + process.env.SUPABASE_SERVICE_ROLE_KEY = "k"; + expect(createSessionEventStore(null)).toBeInstanceOf(SupabaseSessionEventStore); + expect( + createSessionEventStore({ sub: "user-me", email: "a@b.c" }), + ).toBeInstanceOf(SupabaseSessionEventStore); + }); + + it("returns the no-op spine without Supabase env (stdio/local)", () => { + delete process.env.SUPABASE_URL; + delete process.env.SUPABASE_SERVICE_ROLE_KEY; + expect(createSessionEventStore(null)).toBeInstanceOf(NoopSessionEventStore); + }); +}); diff --git a/packages/mcp/src/fabricate/store.ts b/packages/mcp/src/fabricate/store.ts index a6ea71b25..4f4f28360 100644 --- a/packages/mcp/src/fabricate/store.ts +++ b/packages/mcp/src/fabricate/store.ts @@ -9,24 +9,56 @@ */ import type { AuthUser } from "../oauth.js"; -import type { Order, OrderState, Quote, QuoteEconomics } from "./types.js"; +import type { + DebitResult, + Order, + OrderState, + Quote, + QuoteEconomics, + SpendAuthorization, +} from "./types.js"; export interface OrderFilter { status?: OrderState; limit?: number; } +/** Parameters for an atomic wallet debit (forwarded to the debit_wallet RPC). */ +export interface DebitParams { + userId: string; + amountMinor: number; + orderId: string; + authorizationId: string; + idempotencyKey: string; +} + export interface FabricateStore { saveQuote(quote: Quote, econ: QuoteEconomics, userId: string): Promise; saveOrder(order: Order, fabCostMinor: number, userId: string): Promise; getOrder(orderId: string, userId: string): Promise; listOrders(userId: string, filter: OrderFilter): Promise; + // ── money plane (Phase 4, flag-gated) ── + /** Persist a proposed spend authorization (status pending_human). */ + createAuthorization(authz: SpendAuthorization, userId: string): Promise; + /** Read an authorization, ownership-scoped. */ + getAuthorization(id: string, userId: string): Promise; + /** Atomic, balance-floored, idempotent debit. The ONLY way credits leave. */ + debit(p: DebitParams): Promise; + /** Transition an order's state and append a lifecycle event. */ + setOrderState(orderId: string, userId: string, state: OrderState, note: string): Promise; } // ── In-memory (module-global so it survives across calls in one process) ── const memQuotes = new Map(); const memOrders = new Map(); +const memAuthz = new Map(); +const memWallets = new Map(); // userId → balance (minor) +// userId::key → the original request + its result, for replay-match idempotency. +const memDebits = new Map< + string, + { result: DebitResult; orderId: string; authorizationId: string; amountMinor: number } +>(); const memKey = (userId: string, id: string): string => `${userId}::${id}`; @@ -51,6 +83,87 @@ export class InMemoryFabricateStore implements FabricateStore { out.sort((a, b) => (a.created_at < b.created_at ? 1 : -1)); return typeof filter.limit === "number" ? out.slice(0, filter.limit) : out; } + + async createAuthorization(authz: SpendAuthorization, userId: string): Promise { + memAuthz.set(memKey(userId, authz.id), { ...authz, user_id: userId }); + } + async getAuthorization(id: string, userId: string): Promise { + return memAuthz.get(memKey(userId, id)) ?? null; + } + + /** + * Local mirror of the debit_wallet RPC contract — enough to exercise the + * place_order logic in tests, enforcing the SAME guards the RPC does: + * per-(user,key) idempotency WITH replay-match, authorized-only, NOT-expired, + * amount ceiling, balance floor, one_time consume. The Supabase impl defers + * to the real SECURITY DEFINER RPC, which is the source of truth. + */ + async debit(p: DebitParams): Promise { + const idk = memKey(p.userId, p.idempotencyKey); + const prior = memDebits.get(idk); + if (prior) { + // A reused key MUST match the original request, or it's an error — mirrors + // the RPC's idempotency_key_reused guard (027 lines 185-190). + if ( + prior.orderId !== p.orderId || + prior.authorizationId !== p.authorizationId || + prior.amountMinor !== p.amountMinor + ) { + return { ok: false, reason: "idempotency_key_reused" }; + } + return { ...prior.result, idempotent: true }; + } + + const authz = memAuthz.get(memKey(p.userId, p.authorizationId)); + if (!authz) return { ok: false, reason: "authz_not_found" }; + if (authz.status === "revoked") return { ok: false, reason: "authz_revoked" }; + if (authz.status !== "authorized") return { ok: false, reason: "authz_not_authorized" }; + if (new Date(authz.expires_at).getTime() <= Date.now()) { + return { ok: false, reason: "authz_expired" }; + } + if (p.amountMinor <= 0) return { ok: false, reason: "invalid_amount" }; + if (p.amountMinor > authz.max_amount_minor) { + return { ok: false, reason: "amount_exceeds_authz" }; + } + const balance = memWallets.get(p.userId) ?? 0; + if (balance < p.amountMinor) { + return { ok: false, reason: "insufficient_funds", balance_minor: balance }; + } + const after = balance - p.amountMinor; + memWallets.set(p.userId, after); + if (authz.kind === "one_time") { + memAuthz.set(memKey(p.userId, authz.id), { ...authz, status: "consumed" }); + } + const result: DebitResult = { ok: true, balance_minor: after }; + memDebits.set(idk, { + result, + orderId: p.orderId, + authorizationId: p.authorizationId, + amountMinor: p.amountMinor, + }); + return result; + } + + async setOrderState(orderId: string, userId: string, state: OrderState, note: string): Promise { + const rec = memOrders.get(memKey(userId, orderId)); + if (!rec) return; + rec.order.state = state; + rec.order.events.push({ state, at: new Date().toISOString(), note }); + rec.order.updated_at = new Date().toISOString(); + } + + // ── test-only seams (NOT on the FabricateStore interface) ── + /** Seed wallet credit for tests (production top-ups go via credit_wallet). */ + creditWalletForTest(userId: string, minor: number): void { + memWallets.set(userId, (memWallets.get(userId) ?? 0) + minor); + } + /** Simulate the HUMAN web-app approval for tests (never an agent action). */ + approveAuthorizationForTest(id: string, userId: string): boolean { + const a = memAuthz.get(memKey(userId, id)); + if (!a || a.status !== "pending_human") return false; + memAuthz.set(memKey(userId, id), { ...a, status: "authorized" }); + return true; + } } // ── Supabase (raw PostgREST + service role) ── @@ -156,6 +269,92 @@ export class SupabaseFabricateStore implements FabricateStore { } } + async createAuthorization(authz: SpendAuthorization, userId: string): Promise { + await this.insert("spend_authorizations", { + id: authz.id, + user_id: userId, + quote_id: authz.quote_id, + kind: authz.kind, + max_amount_minor: authz.max_amount_minor, + daily_cap_minor: authz.daily_cap_minor, + process_allowlist: authz.process_allowlist, + fab_allowlist: authz.fab_allowlist, + doc_hash: authz.doc_hash, + status: authz.status, + expires_at: authz.expires_at, + }); + } + + async getAuthorization(id: string, userId: string): Promise { + try { + const res = await fabricateFetch( + this.url( + "spend_authorizations", + `?id=eq.${encodeURIComponent(id)}&user_id=eq.${encodeURIComponent(userId)}&limit=1`, + ), + { method: "GET", headers: this.headers({ Accept: "application/vnd.pgrst.object+json" }) }, + ); + if (!res.ok) return null; + return rowToAuthorization(await res.json()); + } catch (err) { + console.error("[fabricate-store] getAuthorization failed:", err); + return null; + } + } + + async debit(p: DebitParams): Promise { + try { + const res = await fabricateFetch(this.url("rpc/debit_wallet"), { + method: "POST", + headers: this.headers(), + body: JSON.stringify({ + p_user: p.userId, + p_amount_minor: p.amountMinor, + p_order_id: p.orderId, + p_authorization_id: p.authorizationId, + p_idempotency_key: p.idempotencyKey, + }), + }); + if (!res.ok) { + return { ok: false, reason: `rpc_http_${res.status}` }; + } + return (await res.json()) as DebitResult; + } catch (err) { + console.error("[fabricate-store] debit failed:", err); + return { ok: false, reason: "rpc_unreachable" }; + } + } + + async setOrderState(orderId: string, userId: string, state: OrderState, note: string): Promise { + const order = await this.getOrder(orderId, userId); + const events = [ + ...(order?.events ?? []), + { state, at: new Date().toISOString(), note }, + ]; + try { + const res = await fabricateFetch( + this.url( + "orders", + `?id=eq.${encodeURIComponent(orderId)}&user_id=eq.${encodeURIComponent(userId)}`, + ), + { + method: "PATCH", + headers: this.headers({ Prefer: "return=minimal" }), + body: JSON.stringify({ state, events }), + }, + ); + if (!res.ok) { + console.error( + "[fabricate-store] setOrderState failed:", + res.status, + await res.text().catch(() => ""), + ); + } + } catch (err) { + console.error("[fabricate-store] setOrderState failed:", err); + } + } + private async insert( table: string, row: Record, @@ -204,6 +403,25 @@ function rowToOrder(row: unknown): Order { }; } +/** Map a PostgREST spend_authorizations row to a SpendAuthorization. */ +function rowToAuthorization(row: unknown): SpendAuthorization { + const r = (row ?? {}) as Record; + return { + id: String(r.id ?? ""), + user_id: String(r.user_id ?? ""), + quote_id: (r.quote_id as string) ?? null, + kind: (r.kind as SpendAuthorization["kind"]) ?? "one_time", + max_amount_minor: Number(r.max_amount_minor ?? 0), + daily_cap_minor: r.daily_cap_minor == null ? null : Number(r.daily_cap_minor), + process_allowlist: (r.process_allowlist as string[] | null) ?? null, + fab_allowlist: (r.fab_allowlist as string[] | null) ?? null, + doc_hash: (r.doc_hash as string) ?? null, + status: (r.status as SpendAuthorization["status"]) ?? "pending_human", + expires_at: String(r.expires_at ?? ""), + created_at: String(r.created_at ?? ""), + }; +} + /** * Choose the store impl: cloud-backed when a signed-in user AND Supabase * service-role env are present, else in-memory (preserves local/stdio behavior diff --git a/packages/mcp/src/fabricate/types.ts b/packages/mcp/src/fabricate/types.ts index 08db22509..1643e090f 100644 --- a/packages/mcp/src/fabricate/types.ts +++ b/packages/mcp/src/fabricate/types.ts @@ -159,6 +159,43 @@ export interface QuoteEconomics { margin_minor: number; } +/** Lifecycle of a spend authorization (mirrors the migration-027 check). */ +export type AuthorizationStatus = + | "pending_human" + | "authorized" + | "consumed" + | "revoked" + | "expired"; + +/** + * A DB-backed, revocable spend authorization (NOT a stateless JWT). The agent + * PROPOSES one (status pending_human); a HUMAN approves it (→ authorized) out of + * band via the web app — never through an MCP tool. Only then can place_order + * consume it. Mirrors the spend_authorizations table. + */ +export interface SpendAuthorization { + id: string; + user_id: string; + quote_id: string | null; + kind: "one_time" | "standing"; + max_amount_minor: number; + daily_cap_minor: number | null; + process_allowlist: string[] | null; + fab_allowlist: string[] | null; + doc_hash: string | null; + status: AuthorizationStatus; + expires_at: string; + created_at: string; +} + +/** Result of an atomic wallet debit (mirrors the debit_wallet RPC jsonb). */ +export interface DebitResult { + ok: boolean; + reason?: string; + balance_minor?: number; + idempotent?: boolean; +} + /** The persisted order lifecycle row. */ export interface Order { order_id: string; diff --git a/packages/mcp/src/http.ts b/packages/mcp/src/http.ts index 8d2ce7f7b..036284429 100644 --- a/packages/mcp/src/http.ts +++ b/packages/mcp/src/http.ts @@ -24,6 +24,8 @@ import { createServer } from "./server.js"; import { verifyAccessToken } from "./oauth.js"; import { getViewerHtml, MCP_APP_MIME_TYPE } from "./viewer.js"; import { flushTelemetry } from "./telemetry.js"; +import { createSessionEventStore } from "./session-store.js"; +import { appendOverlay, listEvents } from "./tools/live.js"; const PORT = parseInt(process.env.PORT || "8080", 10); @@ -126,15 +128,112 @@ async function handleMcpRequest( } } -/** Read request body as string, enforcing a max size. */ -function readBody(req: import("node:http").IncomingMessage): Promise { +/** + * Handle a live-review-window request. Capability-keyed by the session id in the + * path (`/live//`) — possession of the unguessable id is the grant, + * same model as mcp_sessions. Reads/appends go through the service-role event + * store, so they work for both anon and signed-in sessions by session id alone. + * + * GET /live//events[?since=N] → the session's spine events (replay) + * POST /live//annotate → append a viewer overlay (pin/flag/…) + * + * The geometry stream (GLB / fold) and the browser viewer app are a separate, + * env-gated slice — this is the data backbone the broadcast trigger feeds. + */ +async function handleLiveRequest( + req: import("node:http").IncomingMessage, + res: import("node:http").ServerResponse, +): Promise { + const url = new URL(req.url ?? "/", `http://${req.headers.host}`); + const parts = url.pathname.split("/").filter(Boolean); // ["live", id, action] + const sessionId = parts[1] ? decodeURIComponent(parts[1]) : ""; + const action = parts[2] ?? ""; + if (!sessionId) { + res.writeHead(400, { "Content-Type": "text/plain" }); + res.end("missing session id"); + return; + } + + // Viewers are usually anonymous (the link is the capability); a Bearer token, + // if present, just attributes the actor. The event store reads/writes by + // session id via the service role regardless. + const user = verifyAccessToken(req); + const eventStore = createSessionEventStore(user); + + if (req.method === "GET" && action === "events") { + // Each call does a per-session fetch; throttle this public read like annotate. + if (rateLimitExceeded(clientIp(req))) { + res.writeHead(429, { "Content-Type": "text/plain" }); + res.end("Too Many Requests"); + return; + } + const sinceRaw = url.searchParams.get("since"); + const since = sinceRaw != null && sinceRaw !== "" ? Number(sinceRaw) : undefined; + const events = await listEvents( + eventStore, + sessionId, + Number.isFinite(since) ? since : undefined, + ); + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ session_id: sessionId, events })); + return; + } + + if (req.method === "POST" && action === "annotate") { + if (rateLimitExceeded(clientIp(req))) { + res.writeHead(429, { "Content-Type": "text/plain" }); + res.end("Too Many Requests"); + return; + } + let parsed: unknown; + try { + // An overlay is tiny — a far smaller cap than the 10 MiB /mcp default. + parsed = JSON.parse(await readBody(req, 16 * 1024)); + } catch { + res.writeHead(400, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ ok: false, error: "invalid or oversized json body" })); + return; + } + // Tolerate null / array / scalar bodies — appendOverlay rejects the empty + // overlay with a clean 400 instead of throwing a 500. + const body = + parsed && typeof parsed === "object" && !Array.isArray(parsed) + ? (parsed as Record) + : {}; + const result = await appendOverlay( + eventStore, + sessionId, + { + type: typeof body.type === "string" ? body.type : "", + payload: body.payload, + author: typeof body.author === "string" ? body.author : undefined, + }, + // The verified token identity is authoritative; a body author is only ever + // honored (namespaced) for genuinely anonymous viewers. + { trustedAuthor: user?.email || undefined }, + ); + res.writeHead(result.ok ? 200 : 400, { "Content-Type": "application/json" }); + res.end(JSON.stringify(result)); + return; + } + + res.writeHead(404, { "Content-Type": "text/plain" }); + res.end("Not Found"); +} + +/** Read request body as string, enforcing a max size (default MAX_BODY_BYTES; + * callers like the live annotate route pass a much smaller cap). */ +function readBody( + req: import("node:http").IncomingMessage, + maxBytes: number = MAX_BODY_BYTES, +): Promise { return new Promise((resolve, reject) => { const chunks: Buffer[] = []; let total = 0; req.on("data", (chunk: Buffer) => { total += chunk.length; - if (total > MAX_BODY_BYTES) { - reject(new Error("Request body exceeds MCP_MAX_BODY_BYTES")); + if (total > maxBytes) { + reject(new Error("Request body exceeds limit")); req.destroy(); return; } @@ -226,6 +325,18 @@ const httpServer = createHttpServer(async (req, res) => { return; } + // Live review window — capability-keyed by the (unguessable) session id. + // Flag-gated: a new public read/append surface, off until VCAD_LIVE_WINDOW=1. + if (path.startsWith("/live/")) { + if (process.env.VCAD_LIVE_WINDOW !== "1") { + res.writeHead(404, { "Content-Type": "text/plain" }); + res.end("Not Found"); + return; + } + await handleLiveRequest(req, res); + return; + } + // Health check if (path === "/health") { const eng = await getEngine(); diff --git a/packages/mcp/src/server.ts b/packages/mcp/src/server.ts index b62e73665..3543bd6da 100644 --- a/packages/mcp/src/server.ts +++ b/packages/mcp/src/server.ts @@ -44,7 +44,7 @@ import { dropSession, runInSessionScope, } from "./tools/session.js"; -import { createSessionStore } from "./session-store.js"; +import { createSessionStore, createSessionEventStore } from "./session-store.js"; import { createFabricateStore } from "./fabricate/store.js"; import { quoteManufacturing, @@ -54,6 +54,12 @@ import { listOrders, listOrdersSchema, } from "./tools/order.js"; +import { + authorizeSpend, + authorizeSpendSchema, + placeOrder, + placeOrderSchema, +} from "./tools/ordering.js"; import type { AuthUser } from "./oauth.js"; /** Per-connection context threaded from the transport entry point — the @@ -67,6 +73,7 @@ import { registryDispatchableNames, dispatchRegistryTool, } from "./tools/registry-dispatch.js"; +import { buildErrorResult, enrichErrorResult } from "./tools/next-actions.js"; import { createRobotEnv, createRobotEnvSchema, @@ -404,7 +411,13 @@ const WIDGET_CALLABLE_META = { * tool in a disabled pack return an error pointing at the env var. */ const TOOL_PACKS: Record = { - fabricate: ["quote_manufacturing", "get_order_status", "list_orders"], + fabricate: [ + "quote_manufacturing", + "get_order_status", + "list_orders", + "authorize_spend", + "place_order", + ], dfm: ["dfm_check", "dfm_explain", "dfm_suggest_fix", "dfm_apply_fix"], sheet_metal: [ "sheet_metal_create", @@ -542,6 +555,11 @@ export async function createServer( // instance can't clobber each other's binding. const sessionStore = createSessionStore(context.user); + // The event spine for THIS connection. Every kernel mutation appends one row + // (state = fold(log)); the sessionStore's content write is the derived + // materialization. No-op without Supabase env, so stdio/local is unchanged. + const eventStore = createSessionEventStore(context.user); + // vcad Fabricate store (quotes + orders). Cloud-backed for a signed-in user // with the Supabase service-role key, else in-memory (local stdio). Held in // the connection closure like sessionStore so concurrent connections can't @@ -721,6 +739,18 @@ export async function createServer( "List the caller's Fabricate orders, newest first. Optional status filter and limit. Read-only.", inputSchema: listOrdersSchema, }, + { + name: "authorize_spend", + description: + "Propose a spend authorization for a QUOTED order. Creates a DB-backed, revocable authorization (status pending_human) and records the proposal on the session's event log. A HUMAN must approve it in the vcad app before place_order can charge — the agent cannot approve its own spend. Flag-gated (test-mode); no money moves here.", + inputSchema: authorizeSpendSchema, + }, + { + name: "place_order", + description: + "Place a QUOTED order once its authorization has been human-approved: performs one atomic wallet debit and moves the order to PAID (fab submission follows in a later step). Refuses if the authorization is still pending approval. Flag-gated (test-mode).", + inputSchema: placeOrderSchema, + }, // ── Stdlib parts library (session-aware) ────────────────── { name: "search_parts", @@ -1493,6 +1523,14 @@ export async function createServer( result = await listOrders(args, fabricateStore, context.user); break; + case "authorize_spend": + result = await authorizeSpend(args, fabricateStore, eventStore, context.user); + break; + + case "place_order": + result = await placeOrder(args, fabricateStore, eventStore, context.user); + break; + case "render_view": // Image content blocks don't fit the text-only local result // type; the MCP SDK accepts them as-is. @@ -1795,6 +1833,11 @@ export async function createServer( try { const result = await runTool(); + // Tools that RETURN {isError:true} (the ECAD / sheet-metal / DFM surface) + // never reach the throw-catch below, so enrich them here — every failure + // carries next_actions, not just the ones that throw. + if (result.isError) enrichErrorResult(result, name, args); + // ── Persist ───────────────────────────────────────────────────────── // After a creator/mutator settles, write the (possibly newly-minted) // session through to the durable store. Best-effort: the tool already @@ -1810,6 +1853,19 @@ export async function createServer( } catch { // best-effort durable write } + // Append the kernel event to the spine (state = fold(log)). Same + // best-effort discipline as persist — a spine write must never turn a + // successful tool call into an error. + try { + await eventStore.append(writtenId, { + author: context.user?.sub ?? "agent", + kind: "kernel", + type: name, + payload: buildKernelEventPayload(name, args, result), + }); + } catch { + // best-effort event append + } } } // close_document → also forget the durable row (flush-then-forget). @@ -1828,21 +1884,14 @@ export async function createServer( // handled the trap itself. Recover the shared instance so this one bad // document can't DoS every other session — the hosted server can't be // restarted by a client. - if (err instanceof WebAssembly.RuntimeError) { + const kernelTrap = err instanceof WebAssembly.RuntimeError; + if (kernelTrap) { resetKernelWasm(`${name} trapped: ${message}`); } - const errorResult = { - content: [ - { - type: "text", - text: - err instanceof WebAssembly.RuntimeError - ? `Error: kernel trap during '${name}' (${message}). The kernel was reset; other documents are unaffected. This is a kernel bug — please report the document that triggered it.` - : `Error: ${message}`, - }, - ], - isError: true, - }; + // Every failure carries structured `next_actions` so the agent can + // recover in one turn instead of flailing — the verified loop's + // error side. (The success side rides on the `changed` diff.) + const errorResult = buildErrorResult(name, args, message, { kernelTrap }); fireToolAlert(name, args, errorResult); return errorResult; } @@ -2009,6 +2058,45 @@ function resolvePreviewDocumentId( * does NOT call resolvePreviewDocumentId, which registers a fresh session for * import_step / create_cad_loon and would double-mint here. */ +/** + * Build the payload for a `kernel` session_events row: the tool name, its args + * (minus document_id), and the compact `changed` parts diff the registry path + * already merged into the result. Capped so a fat call can't bloat the spine — + * mirrors the >8KB result-slimming discipline; tool + changed (the cheap, + * high-value parts) are always kept. + */ +function buildKernelEventPayload( + name: string, + args: Record, + result: { content: Array<{ type: string; text: string }> }, +): Record { + const { document_id: _docId, ...rest } = args; + void _docId; + let changed: unknown; + for (const block of result.content) { + if (block.type !== "text") continue; + try { + const parsed = JSON.parse(block.text) as { changed?: unknown }; + if (parsed && parsed.changed !== undefined) { + changed = parsed.changed; + break; + } + } catch { + // not JSON — skip + } + } + const payload: Record = { tool: name, args: rest }; + if (changed !== undefined) payload.changed = changed; + try { + if (JSON.stringify(payload).length > 8192) { + payload.args = { _omitted: true }; + } + } catch { + payload.args = { _omitted: true }; + } + return payload; +} + function effectiveDocId( result: { content: Array<{ type: string; text: string }>; diff --git a/packages/mcp/src/session-store.ts b/packages/mcp/src/session-store.ts index 9b3651185..3cf406e89 100644 --- a/packages/mcp/src/session-store.ts +++ b/packages/mcp/src/session-store.ts @@ -16,6 +16,7 @@ * anonymous hosted call during the pre-MCP_REQUIRE_AUTH transition) the * in-memory impl reproduces today's behavior exactly. */ +import { randomUUID } from "node:crypto"; import type { Document } from "@vcad/ir"; import type { AuthUser } from "./oauth.js"; @@ -300,3 +301,154 @@ export function createSessionStore(user: AuthUser | null): SessionStore { } return new InMemorySessionStore(); } + +// ─── Event spine — the per-session append-only log (migration 028) ─────────── +// +// State = fold(log); the content snapshot the SessionStore writes is a derived +// materialization. This store is the canonical record: a kernel mutation, an +// overlay annotation, or a control event each appends one row, which the DB +// fans out over Realtime. Best-effort throughout (mirrors SessionStore): a +// failed append must never turn a successful tool call into an error. + +/** An event to append. `idempotencyKey` is generated when omitted. */ +export interface SessionEvent { + /** Emitter: a user sub, the literal "agent", or "human". */ + author: string; + /** kernel = folds into geometry; overlay = annotation; control = lifecycle. */ + kind: "kernel" | "overlay" | "control"; + /** Fine type: the tool name for kernel, "pin"/"flag" for overlay, etc. */ + type: string; + /** kernel: {tool, args, changed?}; overlay: {anchor, text, …}. */ + payload: Record; + /** Per-session idempotency key; a random uuid is used when omitted. */ + idempotencyKey?: string; +} + +/** A row read back from `session_events`. */ +export interface StoredSessionEvent { + id: number; + seq: number; + session_id: string; + author: string; + kind: string; + type: string; + payload: Record; + created_at: string; +} + +export interface SessionEventStore { + /** Append one event. Best-effort: errors are logged, never thrown. */ + append(sessionId: string, evt: SessionEvent): Promise; + /** Read a session's events in seq order (replay / live window). With + * `sinceSeq`, only events after it — the backend filters server-side so a + * late-join catch-up doesn't pull the whole log. */ + list(sessionId: string, sinceSeq?: number): Promise; +} + +/** No-op store = stdio/local: nothing durable, list is empty. */ +export class NoopSessionEventStore implements SessionEventStore { + async append(): Promise { + /* no spine without Supabase env */ + } + async list(): Promise { + return []; + } +} + +/** + * Cloud-backed spine. Appends via the `append_session_event` RPC (the sole + * writer; the table grants service_role SELECT only), reads the table directly. + * The service role bypasses RLS, so `list` sees every row for a session + * regardless of ownership — the unguessable session_id is the capability. + */ +export class SupabaseSessionEventStore implements SessionEventStore { + constructor( + private cfg: { + supabaseUrl: string; + serviceRoleKey: string; + /** Caller's user id, or null for an anonymous capability session. */ + userId: string | null; + }, + ) {} + + private headers(extra: Record = {}): Record { + return { + apikey: this.cfg.serviceRoleKey, + Authorization: `Bearer ${this.cfg.serviceRoleKey}`, + "Content-Type": "application/json", + ...extra, + }; + } + + async append(sessionId: string, evt: SessionEvent): Promise { + try { + const res = await sessionFetch( + `${this.cfg.supabaseUrl}/rest/v1/rpc/append_session_event`, + { + method: "POST", + headers: this.headers(), + body: JSON.stringify({ + p_session_id: sessionId, + p_user: this.cfg.userId, + p_author: evt.author, + p_kind: evt.kind, + p_type: evt.type, + p_payload: evt.payload ?? {}, + p_idempotency_key: evt.idempotencyKey ?? randomUUID(), + }), + }, + ); + if (!res.ok) { + console.error( + "[session-events] append failed:", + res.status, + await res.text().catch(() => ""), + ); + } + } catch (err) { + console.error("[session-events] append failed:", err); + } + } + + async list(sessionId: string, sinceSeq?: number): Promise { + try { + const sinceFilter = + typeof sinceSeq === "number" && Number.isFinite(sinceSeq) + ? `&seq=gt.${Math.trunc(sinceSeq)}` + : ""; + const res = await sessionFetch( + `${this.cfg.supabaseUrl}/rest/v1/session_events` + + `?session_id=eq.${encodeURIComponent(sessionId)}` + + sinceFilter + + `&order=seq.asc` + + `&select=id,seq,session_id,author,kind,type,payload,created_at`, + { method: "GET", headers: this.headers() }, + ); + if (!res.ok) return []; + return (await res.json()) as StoredSessionEvent[]; + } catch (err) { + console.error("[session-events] list failed:", err); + return []; + } + } +} + +/** + * Choose the spine impl from env + user. With Supabase env present, both + * signed-in and anonymous sessions append via the same RPC (it takes a nullable + * user). Without it (stdio/local) the no-op store reproduces today's behavior. + */ +export function createSessionEventStore( + user: AuthUser | null, +): SessionEventStore { + const url = (process.env.SUPABASE_URL || "").replace(/\/+$/, ""); + const key = process.env.SUPABASE_SERVICE_ROLE_KEY || ""; + if (url && key) { + return new SupabaseSessionEventStore({ + supabaseUrl: url, + serviceRoleKey: key, + userId: user?.sub ?? null, + }); + } + return new NoopSessionEventStore(); +} diff --git a/packages/mcp/src/tools/live.ts b/packages/mcp/src/tools/live.ts new file mode 100644 index 000000000..37421e5fb --- /dev/null +++ b/packages/mcp/src/tools/live.ts @@ -0,0 +1,115 @@ +/** + * Live review window — server-side backbone. + * + * The spine (session_events, migration 028) already fans every appended row out + * over Supabase Realtime to topic `session:`. This module adds the two + * things a viewer needs that aren't kernel mutations: + * + * appendOverlay — a viewer drops an annotation (pin / flag / stroke / note). + * It's a `kind:'overlay'` event, so it rides the same spine + * and broadcast as geometry, lands in the Receipt, and never + * touches the kernel — the asymmetry is structural (overlay vs + * kernel event class), not a rule. + * listEvents — replay / late-join catch-up: a session's events in order, + * optionally only those after a seq the client already has. + * + * Anchoring note: pins/flags should carry an `anchor` into the IR id space + * (e.g. {node, face}) in their payload so they survive a re-fold; this module + * doesn't mandate a shape yet, it just carries the payload through. + * + * The browser viewer app (subscribe to the topic, fold events, render the GLB, + * draw overlays) is a separate, env-gated slice — it needs a live browser + + * Realtime to verify, which this backbone does not. + */ + +import type { SessionEventStore, StoredSessionEvent } from "../session-store.js"; + +/** Annotation kinds a viewer may add. Anything else is rejected. */ +export const OVERLAY_TYPES = ["pin", "flag", "stroke", "note"] as const; +export type OverlayType = (typeof OVERLAY_TYPES)[number]; + +/** An overlay is a small annotation (a pin, a short note) — never a payload + * dump. Cap it so an unauthenticated annotate can't bloat the durable log or + * amplify over the broadcast topic. */ +export const OVERLAY_PAYLOAD_MAX_BYTES = 4096; + +export interface OverlayInput { + type: string; + payload?: unknown; + /** Viewer-supplied display name (UNTRUSTED) — namespaced as `viewer:` + * so it can never collide with a real user identity or 'agent'/'human'. */ + author?: string; +} + +export interface OverlayOpts { + /** A verified identity (e.g. the token email). When present it is + * authoritative and the untrusted `input.author` is ignored. */ + trustedAuthor?: string; +} + +export type AppendOverlayResult = { ok: true } | { ok: false; error: string }; + +/** Resolve the spine `author` for an overlay. A verified identity wins; an + * anonymous viewer's self-asserted name is sanitized and `viewer:`-namespaced + * so it can never impersonate a real sub/email or the reserved authors. */ +function resolveAuthor(input: OverlayInput, opts: OverlayOpts): string { + if (opts.trustedAuthor) return String(opts.trustedAuthor).slice(0, 64); + const raw = input.author ? String(input.author) : "anon"; + const safe = raw.slice(0, 48).replace(/[^\w .@-]/g, "") || "anon"; + return `viewer:${safe}`; +} + +/** + * Validate and append a viewer annotation as a `kind:'overlay'` spine event. + * The DB trigger broadcasts it to topic `session:`. Best-effort underneath + * (eventStore.append never throws), so this resolves true once validation passes. + */ +export async function appendOverlay( + eventStore: SessionEventStore, + sessionId: string, + input: OverlayInput, + opts: OverlayOpts = {}, +): Promise { + if (!sessionId) return { ok: false, error: "missing session id" }; + const type = String(input?.type ?? ""); + if (!(OVERLAY_TYPES as readonly string[]).includes(type)) { + return { ok: false, error: `overlay type must be one of: ${OVERLAY_TYPES.join(", ")}` }; + } + const payload = + input.payload && typeof input.payload === "object" && !Array.isArray(input.payload) + ? (input.payload as Record) + : {}; + let bytes: number; + try { + bytes = JSON.stringify(payload).length; + } catch { + return { ok: false, error: "overlay payload is not serializable" }; + } + if (bytes > OVERLAY_PAYLOAD_MAX_BYTES) { + return { ok: false, error: `overlay payload too large (${bytes} > ${OVERLAY_PAYLOAD_MAX_BYTES} bytes)` }; + } + + await eventStore.append(sessionId, { + author: resolveAuthor(input, opts), + kind: "overlay", + type, + payload, + }); + return { ok: true }; +} + +/** + * Replay / catch-up: a session's events in seq order. With `sinceSeq`, only the + * events after it (what a live client missed while reconnecting). + */ +export async function listEvents( + eventStore: SessionEventStore, + sessionId: string, + sinceSeq?: number, +): Promise { + if (!sessionId) return []; + // Push the filter to the backend (server-side seq>sinceSeq), and keep the JS + // filter as a guard so correctness holds even for a store that ignores it. + const rows = await eventStore.list(sessionId, sinceSeq); + return typeof sinceSeq === "number" ? rows.filter((e) => e.seq > sinceSeq) : rows; +} diff --git a/packages/mcp/src/tools/next-actions.ts b/packages/mcp/src/tools/next-actions.ts new file mode 100644 index 000000000..fef3755cf --- /dev/null +++ b/packages/mcp/src/tools/next-actions.ts @@ -0,0 +1,283 @@ +/** + * The error side of the verified agent loop: turn a failed tool call into a + * recoverable one. Instead of returning a bare `Error: ` string that + * leaves the agent guessing, every failure carries structured `next_actions` — + * an ordered list of recovery steps, each optionally naming a tool (and + * ready-to-run args) the agent can call to get unstuck in a single turn. + * + * Two entry points cover both failure conventions in the codebase: + * - buildErrorResult — for THROWN errors, used by the server's central catch + * (registry CRUD, planner errors, kernel traps, "Unknown document_id"). + * - enrichErrorResult — for tools that RETURN {isError:true} instead of + * throwing (the whole ECAD / sheet-metal / DFM surface). The only + * intentional carve-outs are disabled-pack and unknown-tool results. + * + * Pure, deterministic, and dependency-light so it's unit-testable without + * booting the server or the kernel. + */ + +import { CREATE_PARAM_HINTS } from "./registry-dispatch.js"; + +/** One recovery step. */ +export interface NextAction { + /** Imperative instruction the agent can act on. */ + action: string; + /** A tool to call to recover, when one applies. */ + tool?: string; + /** Ready-to-run args for `tool`, when derivable from context. */ + args?: Record; +} + +/** An MCP error result with structured recovery attached. */ +export interface McpErrorResult { + content: Array<{ type: "text"; text: string }>; + structuredContent: { error: string; next_actions: NextAction[] }; + isError: true; +} + +const docIdOf = (args: Record): string | undefined => + typeof args.document_id === "string" ? args.document_id : undefined; + +/** + * Map (tool, args, error message) to a small ordered list of recovery actions. + * Ordered most-specific-first; the generic "inspect then retry" is the floor so + * the agent always has somewhere to go. + */ +export function suggestNextActions( + toolName: string, + args: Record, + message: string, + opts: { kernelTrap?: boolean } = {}, +): NextAction[] { + const docId = docIdOf(args); + const lower = message.toLowerCase(); + + // A kernel trap already reset the shared instance — retrying verbatim will + // likely re-trap. Steer toward different inputs. + if (opts.kernelTrap) { + return [ + { + action: + "The kernel reset after a trap. Retry with simpler parameters (avoid degenerate or zero-size geometry, or split a chained boolean into steps). If it persists it's a kernel bug — report the document.", + }, + ]; + } + + // The session isn't live (cold instance, wrong id, or never opened). + if (lower.includes("unknown document_id")) { + return [ + { + action: + "No live session for that id. Open a session (or re-open the saved document), then pass its id as document_id.", + tool: "open_document", + }, + ]; + } + + // Server-side misconfiguration — not recoverable from the client. Checked + // before the create/update branch since this surfaces on those tools too. + if (lower.includes("planner unavailable") || lower.includes("setwasm")) { + return [ + { + action: + "Server misconfiguration (kernel planner not initialized). Not recoverable from the client — report it.", + }, + ]; + } + + // PCB order-of-operations: the board has no schematic / no board yet. These + // are the most common multi-step ECAD mistakes, and the recovery is exact. + if (lower.includes("no schematic")) { + return [ + { + action: + "This session has no schematic yet. Declare connectivity first with create_schematic, then retry.", + tool: "create_schematic", + ...(docId ? { args: { document_id: docId } } : {}), + }, + ]; + } + if (lower.includes("no pcb") || lower.includes("pcbboard") || lower.includes("no board")) { + return [ + { + action: + "This session has no board yet. Create one with place_components (or board_from_solid for an existing solid), then retry.", + tool: "place_components", + ...(docId ? { args: { document_id: docId } } : {}), + }, + ]; + } + + // A guessed catalog path that doesn't exist — list the library, don't `read` + // the current document (which only knows already-placed parts). + if (lower.includes("unknown part")) { + return [ + { + action: "List available catalog parts to get a valid id/path, then retry.", + tool: "search_parts", + }, + ]; + } + + // A bad or missing part reference — the fix is always to list the real ids. + if ( + lower.includes("missing `part_id`") || + lower.includes("missing part_id") || + lower.includes("no part with id") + ) { + return [ + { + action: "List the document's parts to get a valid part_id, then retry.", + tool: "read", + ...(docId ? { args: { document_id: docId } } : {}), + }, + ]; + } + + // Malformed create/update params — surface the exact expected shape. `update` + // carries no `type` (its schema is {node_id, params}), so the Type-Catalog + // hint is create-only; an unmatched update falls through to the floor below, + // which carries document_id and is more apt for a node edit. + if (toolName === "create" || toolName === "update") { + const type = String((args as { type?: unknown }).type ?? "").toLowerCase(); + const hint = CREATE_PARAM_HINTS[type]; + const actions: NextAction[] = []; + if (hint) { + actions.push({ + action: `Correct the params for "${type}" and retry: ${hint}`, + tool: toolName, + }); + } else if (toolName === "create") { + actions.push({ + action: `Check the Type Catalog in this server's instructions for "${type || "this type"}" params, then retry.`, + tool: "create", + }); + } + const miss = /missing field `([^`]+)`/.exec(message); + if (miss) { + actions.push({ + action: `Provide the required field "${miss[1]}".`, + tool: toolName, + }); + } + // Booleans take numeric ids of pre-created children, not inline geometry. + if ( + (type === "union" || type === "difference" || type === "intersection") && + lower.includes("node") + ) { + actions.push({ + action: + "Create both child nodes first, then reference their numeric ids — inline child definitions aren't supported.", + }); + } + if (actions.length) return actions; + } + + // Floor: inspect current state, then retry with corrected arguments. + return [ + { + action: "Inspect the current state, then retry with corrected arguments.", + tool: "read", + ...(docId ? { args: { document_id: docId } } : {}), + }, + ]; +} + +/** + * Build the MCP error result for a failed tool call: the human-readable error + * line plus a machine-readable `next_actions:` JSON tail (so text-only clients + * see recovery too) and the same actions in `structuredContent` for richer + * hosts. + */ +export function buildErrorResult( + toolName: string, + args: Record, + message: string, + opts: { kernelTrap?: boolean } = {}, +): McpErrorResult { + const next = suggestNextActions(toolName, args, message, opts); + const head = opts.kernelTrap + ? `Error: kernel trap during '${toolName}' (${message}). The kernel was reset; other documents are unaffected.` + : `Error: ${message}`; + const tail = next.length ? `\nnext_actions: ${JSON.stringify(next)}` : ""; + return { + content: [{ type: "text", text: `${head}${tail}` }], + structuredContent: { error: message, next_actions: next }, + isError: true, + }; +} + +/** Shape of an already-formed tool result (mutated in place by enrich). */ +interface MutableResult { + content: Array<{ type: string; text: string }>; + structuredContent?: Record; + isError?: boolean; +} + +/** Errors that should NOT get a generic recovery floor — either not recoverable + * (disabled pack / unknown tool / ordering off) or self-explanatory with their + * own instruction (a spend awaiting human approval must NOT be retried, and a + * generic "inspect then read" would be actively misleading). */ +function isCarveOut(message: string): boolean { + return ( + /^unknown tool:/i.test(message) || + /belongs to a pack/i.test(message) || + /ordering is disabled/i.test(message) || + /pending human approval/i.test(message) + ); +} + +/** + * Attach `next_actions` to a tool result that RETURNED `{isError:true}` instead + * of throwing. The whole ECAD / sheet-metal / DFM surface reports failures this + * way (a normal success-path return), so without this they'd bypass the central + * catch and carry no recovery — leaving the most multi-step, order-of-operations + * -prone tools (e.g. "Document has no schematic" → create_schematic) unhelped. + * Idempotent: a result already carrying next_actions (via buildErrorResult) is + * left alone. Injects into the JSON body when the text is a JSON object, else + * appends a parseable tail. + */ +export function enrichErrorResult( + result: MutableResult, + toolName: string, + args: Record, +): void { + if (!result.isError) return; + if (result.structuredContent && "next_actions" in result.structuredContent) { + return; + } + const block = result.content.find((b) => b.type === "text"); + if (!block) return; + + // Recover the human message: ECAD returns "Error: " text; order.ts and + // friends return a {"error": ""} JSON body. + let message = block.text; + try { + const parsed = JSON.parse(block.text) as { error?: unknown }; + if (parsed && typeof parsed === "object" && typeof parsed.error === "string") { + message = parsed.error; + } + } catch { + // not JSON — use the raw text + } + if (isCarveOut(message)) return; + + const next = suggestNextActions(toolName, args, message); + if (!next.length) return; + + try { + const parsed = JSON.parse(block.text) as Record; + if (parsed && typeof parsed === "object" && !Array.isArray(parsed)) { + parsed.next_actions = next; + block.text = JSON.stringify(parsed); + } else { + block.text = `${block.text}\nnext_actions: ${JSON.stringify(next)}`; + } + } catch { + block.text = `${block.text}\nnext_actions: ${JSON.stringify(next)}`; + } + result.structuredContent = { + ...(result.structuredContent ?? {}), + next_actions: next, + }; +} diff --git a/packages/mcp/src/tools/ordering.ts b/packages/mcp/src/tools/ordering.ts new file mode 100644 index 000000000..85a5d3a53 --- /dev/null +++ b/packages/mcp/src/tools/ordering.ts @@ -0,0 +1,248 @@ +/** + * vcad Fabricate — Phase 4 money gate (flag-gated, test-mode). + * + * The asymmetric-capability seam from the event-log design, applied to money: + * authorize_spend — the AGENT proposes a spend authorization for a QUOTED + * order (status pending_human) and emits a `propose_order` + * control event on the session spine. No money moves. + * place_order — the agent places the order ONLY after a HUMAN approved + * the authorization (out of band, in the web app — never an + * MCP tool). On approval it performs one atomic debit via + * the debit_wallet RPC, moves the order to PAID, and emits + * an `order_placed` control event. + * + * Ordering is OFF unless VCAD_FABRICATE_ORDERING=1 — and even then the live + * debit path (Supabase RPC) must be staging-verified before use. Fab submission + * (the idempotent outbox worker) is a deliberate follow-up: a placed order rests + * at PAID for a worker to pick up. + */ + +import { randomUUID } from "node:crypto"; +import type { AuthUser } from "../oauth.js"; +import { ownerId, type FabricateStore } from "../fabricate/store.js"; +import type { SessionEventStore } from "../session-store.js"; +import type { SpendAuthorization } from "../fabricate/types.js"; + +type ToolResult = { content: Array<{ type: "text"; text: string }>; isError?: boolean }; + +function ok(payload: unknown): ToolResult { + return { content: [{ type: "text", text: JSON.stringify(payload, null, 2) }] }; +} +function err(message: string): ToolResult { + return { content: [{ type: "text", text: JSON.stringify({ error: message }) }], isError: true }; +} + +/** Ordering is disabled unless explicitly enabled — test-mode, flag-gated. */ +export function orderingEnabled(): boolean { + return process.env.VCAD_FABRICATE_ORDERING === "1"; +} + +const DISABLED_MSG = + "Fabricate ordering is disabled (Phase 4, test-mode). It is flag-gated behind VCAD_FABRICATE_ORDERING and the live payment path requires staging verification before enabling. Use quote_manufacturing for estimates."; + +const AUTHZ_TTL_MS = 24 * 60 * 60 * 1000; // 24h + +const toUsd = (minor: number): number => Math.round(minor) / 100; + +/** Best-effort spine control event; never fails the tool (mirrors persist). */ +async function emitControl( + eventStore: SessionEventStore, + sessionId: string, + user: AuthUser | null, + type: string, + payload: Record, +): Promise { + try { + await eventStore.append(sessionId, { + author: user?.sub ?? "agent", + kind: "control", + type, + payload, + }); + } catch { + // best-effort — the order action already succeeded + } +} + +// ── authorize_spend ────────────────────────────────────────────────────────── + +export const authorizeSpendSchema = { + type: "object" as const, + properties: { + order_id: { type: "string" as const, description: "Order id from quote_manufacturing (must be QUOTED)." }, + max_amount_minor: { + type: "integer" as const, + minimum: 1, + description: + "Optional spend ceiling in minor units (USD cents). Defaults to the order total; must be ≥ the order total.", + }, + }, + required: ["order_id"], +}; + +export async function authorizeSpend( + input: unknown, + store: FabricateStore, + eventStore: SessionEventStore, + user: AuthUser | null, +): Promise { + if (!orderingEnabled()) return err(DISABLED_MSG); + + const args = (input ?? {}) as Record; + const orderId = String(args.order_id ?? ""); + if (!orderId) return err("order_id is required."); + const owner = ownerId(user); + + const order = await store.getOrder(orderId, owner); + if (!order) return err(`Unknown order_id: ${orderId}`); + if (order.state !== "QUOTED") { + return err(`Order ${orderId} is ${order.state}, not QUOTED — nothing to authorize.`); + } + + const max = typeof args.max_amount_minor === "number" + ? Math.round(args.max_amount_minor) + : order.amount_total_minor; + if (max < order.amount_total_minor) { + return err( + `max_amount_minor (${max}) is below the order total (${order.amount_total_minor}).`, + ); + } + + const now = new Date(); + const authz: SpendAuthorization = { + id: randomUUID(), + user_id: owner, + quote_id: order.quote_id, + kind: "one_time", + max_amount_minor: max, + daily_cap_minor: null, + process_allowlist: null, + fab_allowlist: order.fab ? [order.fab] : null, + doc_hash: null, + status: "pending_human", + expires_at: new Date(now.getTime() + AUTHZ_TTL_MS).toISOString(), + created_at: now.toISOString(), + }; + await store.createAuthorization(authz, owner); + await emitControl(eventStore, order.document_id, user, "propose_order", { + order_id: orderId, + authorization_id: authz.id, + amount_minor: order.amount_total_minor, + fab: order.fab, + }); + + return ok({ + authorization_id: authz.id, + order_id: orderId, + status: authz.status, + max_amount_usd: toUsd(max), + expires_at: authz.expires_at, + note: + "Proposed. A HUMAN must approve this authorization in the vcad app before place_order can charge — the agent cannot approve its own spend. Once approved, call place_order with this authorization_id.", + }); +} + +// ── place_order ────────────────────────────────────────────────────────────── + +export const placeOrderSchema = { + type: "object" as const, + properties: { + order_id: { type: "string" as const, description: "Order id from quote_manufacturing." }, + authorization_id: { + type: "string" as const, + description: "Authorization id from authorize_spend (must be human-approved).", + }, + idempotency_key: { + type: "string" as const, + description: "Optional. Reuse to retry safely; defaults to a per-order key.", + }, + }, + required: ["order_id", "authorization_id"], +}; + +export async function placeOrder( + input: unknown, + store: FabricateStore, + eventStore: SessionEventStore, + user: AuthUser | null, +): Promise { + if (!orderingEnabled()) return err(DISABLED_MSG); + + const args = (input ?? {}) as Record; + const orderId = String(args.order_id ?? ""); + const authorizationId = String(args.authorization_id ?? ""); + if (!orderId || !authorizationId) { + return err("order_id and authorization_id are required."); + } + const owner = ownerId(user); + + const order = await store.getOrder(orderId, owner); + if (!order) return err(`Unknown order_id: ${orderId}`); + if (order.state === "PAID" || order.state === "SUBMITTED" || order.state === "IN_PRODUCTION") { + return ok({ order_id: orderId, state: order.state, note: "Order already placed." }); + } + if (order.state !== "QUOTED") { + return err(`Order ${orderId} is ${order.state}, not placeable.`); + } + + const authz = await store.getAuthorization(authorizationId, owner); + if (!authz) return err(`Unknown authorization_id: ${authorizationId}`); + if (authz.quote_id && order.quote_id && authz.quote_id !== order.quote_id) { + return err("Authorization is bound to a different quote than this order."); + } + if (authz.status === "pending_human") { + return err( + `Authorization ${authorizationId} is pending human approval — a human must approve it in the vcad app before this order can be placed. Do not retry until approved.`, + ); + } + if (authz.status === "revoked" || authz.status === "expired") { + return err(`Authorization ${authorizationId} is ${authz.status}. Propose a new one with authorize_spend.`); + } + // Expiry guard for a still-authorized authz (mirrors the debit_wallet RPC). + // Deliberately skipped for a 'consumed' authz: that path is an idempotent + // retry finalizing an order whose debit committed but whose state write did + // not (e.g. a crash between debit and setOrderState) — blocking it on expiry + // would permanently strand an order the user already paid for. + if (authz.status === "authorized" && new Date(authz.expires_at).getTime() <= Date.now()) { + return err(`Authorization ${authorizationId} has expired. Propose a new one with authorize_spend.`); + } + // Both 'authorized' (first placement) and 'consumed' (idempotent retry) fall + // through: the idempotent debit below is the single chokepoint against double + // spend. A 'consumed' authz with no matching prior debit is rejected there + // (authz_not_authorized), so a stale authz can't place a fresh charge. + + const debit = await store.debit({ + userId: owner, + amountMinor: order.amount_total_minor, + orderId, + authorizationId, + idempotencyKey: typeof args.idempotency_key === "string" ? args.idempotency_key : `${orderId}:debit`, + }); + if (!debit.ok) { + await emitControl(eventStore, order.document_id, user, "order_payment_failed", { + order_id: orderId, + authorization_id: authorizationId, + reason: debit.reason, + }); + return err(`Payment failed: ${debit.reason ?? "unknown"}. The order remains QUOTED; resolve and retry.`); + } + + await store.setOrderState(orderId, owner, "PAID", `place_order debit ok${debit.idempotent ? " (idempotent replay)" : ""}`); + await emitControl(eventStore, order.document_id, user, "order_placed", { + order_id: orderId, + authorization_id: authorizationId, + amount_minor: order.amount_total_minor, + fab: order.fab, + idempotent: debit.idempotent ?? false, + }); + + return ok({ + order_id: orderId, + state: "PAID", + amount_usd: toUsd(order.amount_total_minor), + fab: order.fab, + idempotent: debit.idempotent ?? false, + note: + "Paid via atomic wallet debit. Fab submission runs in a follow-up outbox slice; the order rests at PAID until then. Track with get_order_status.", + }); +} diff --git a/packages/mcp/src/tools/registry-dispatch.ts b/packages/mcp/src/tools/registry-dispatch.ts index d25893848..3c88677c8 100644 --- a/packages/mcp/src/tools/registry-dispatch.ts +++ b/packages/mcp/src/tools/registry-dispatch.ts @@ -86,7 +86,7 @@ const CREATE_PARAM_DEFAULTS: Record> = { * errors. serde errors like "missing field `left`" are accurate but say * nothing about the full shape — this does. */ -const CREATE_PARAM_HINTS: Record = { +export const CREATE_PARAM_HINTS: Record = { cube: "{size: {x, y, z}}", cylinder: "{radius, height, segments? (default 64)} — axis along Z", sphere: "{radius, segments? (default 32)}", diff --git a/supabase/migrations/028_session_events.sql b/supabase/migrations/028_session_events.sql new file mode 100644 index 000000000..8e2aab5cf --- /dev/null +++ b/supabase/migrations/028_session_events.sql @@ -0,0 +1,213 @@ +-- session_events — the per-session append-only event log ("the spine"). +-- +-- Every durable, attributable thing that happens in an MCP session is a row +-- here: a kernel mutation, an overlay annotation (pin/flag), a control event +-- (propose_order/approve). State = fold(log); the content snapshot in +-- `documents` / `mcp_sessions` is a derived materialization of that fold. +-- +-- Design mirrors the money plane (migration 027): +-- 1. Append-only, bigserial total order; per-session monotonic `seq` assigned +-- inside the RPC under an advisory lock so concurrent serverless instances +-- can't collide. +-- 2. Idempotency is namespaced PER SESSION (session_id, idempotency_key) — one +-- session's key can never collide with another's, and a retried append is a +-- safe no-op replay. +-- 3. Writes happen ONLY inside the SECURITY DEFINER `append_session_event` RPC; +-- service_role gets SELECT only on the table, so even a leaked service key +-- can't forge history outside the audited append path. +-- 4. append = persist = broadcast: an after-insert trigger fans the row out via +-- Supabase Realtime (`realtime.send`) to topic `session:`. The +-- INSERT is the one write; the broadcast is a DB-side consequence, not an +-- app-side dual write. Best-effort — a missing realtime schema never blocks +-- the durable append. +-- +-- High-frequency liveness (cursor/camera) is NOT stored here — it rides a plain +-- ephemeral Realtime broadcast. The test for what belongs in this table: "would +-- you want it in the Receipt?" + +create table if not exists session_events ( + id bigserial primary key, -- global monotonic order + -- The MCP session/document id (text — MCP ids aren't uuids). Possession of an + -- unguessable id is the capability, same model as mcp_sessions. + session_id text not null, + -- The owning user, or null for an anonymous capability session. + user_id uuid references auth.users(id) on delete cascade, + -- Per-session 1..N, assigned in append_session_event() under an advisory lock. + seq integer not null, + -- Who emitted it: a user sub, the literal 'agent', or 'human'. + author text not null, + -- Coarse class. kernel = folds into geometry (driver-only later); overlay = + -- annotations (open to viewers); control = lifecycle (propose/approve). + kind text not null check (kind in ('kernel', 'overlay', 'control')), + -- Fine type: the tool name for kernel, 'pin'/'flag'/'stroke' for overlay, + -- 'propose_order'/'approve'/… for control. + type text not null, + -- kernel: {tool, args(slim), changed?}; overlay: {anchor, text, …}. + payload jsonb not null default '{}'::jsonb, + idempotency_key text not null, + created_at timestamptz not null default now(), + unique (session_id, idempotency_key), + unique (session_id, seq) +); + +create index if not exists session_events_session_idx + on session_events (session_id, seq); + +-- ─── guard: the append RPC is server-only (mirrors assert_money_caller) ─────── +create or replace function assert_session_caller() +returns void +language plpgsql +security definer +set search_path = public +as $$ +declare + v_claims text := current_setting('request.jwt.claims', true); +begin + if v_claims is not null + and (v_claims::jsonb ->> 'role') in ('authenticated', 'anon') then + raise exception 'forbidden: session_events functions are service-role only'; + end if; +end; +$$; + +-- ─── append_session_event() — atomic, per-session-ordered, idempotent ──────── +-- The ONLY way a row enters session_events. Returns jsonb: +-- { ok:true, id, seq } on append +-- { ok:true, id, seq, idempotent:true} on replay of the same (session, key) +-- { ok:false, reason } on a guarded failure (no raise) +create or replace function append_session_event( + p_session_id text, + p_user uuid, + p_author text, + p_kind text, + p_type text, + p_payload jsonb, + p_idempotency_key text +) +returns jsonb +language plpgsql +security definer +set search_path = public +as $$ +declare + v_existing session_events%rowtype; + v_seq integer; + v_id bigint; +begin + perform assert_session_caller(); + + if p_session_id is null or length(p_session_id) = 0 then + return jsonb_build_object('ok', false, 'reason', 'missing_session_id'); + end if; + if p_kind not in ('kernel', 'overlay', 'control') then + return jsonb_build_object('ok', false, 'reason', 'invalid_kind'); + end if; + if p_type is null or length(p_type) = 0 then + return jsonb_build_object('ok', false, 'reason', 'missing_type'); + end if; + if p_idempotency_key is null or length(p_idempotency_key) = 0 then + return jsonb_build_object('ok', false, 'reason', 'missing_idempotency_key'); + end if; + + -- Serialize same-session appends so seq assignment + idempotency check-then- + -- insert is race-free (the loser sees the winner's committed row). + perform pg_advisory_xact_lock(hashtext(p_session_id)); + + -- Idempotent replay (per-session key). + select * into v_existing + from session_events + where session_id = p_session_id and idempotency_key = p_idempotency_key; + if found then + return jsonb_build_object('ok', true, 'idempotent', true, + 'id', v_existing.id, 'seq', v_existing.seq); + end if; + + select coalesce(max(seq), 0) + 1 into v_seq + from session_events where session_id = p_session_id; + + insert into session_events + (session_id, user_id, seq, author, kind, type, payload, idempotency_key) + values (p_session_id, p_user, v_seq, p_author, p_kind, p_type, + coalesce(p_payload, '{}'::jsonb), p_idempotency_key) + returning id into v_id; + + return jsonb_build_object('ok', true, 'id', v_id, 'seq', v_seq); +end; +$$; + +-- ─── broadcast trigger — append = persist = broadcast ──────────────────────── +-- Fans each appended row out to topic `session:`. Wrapped so a +-- missing realtime schema or a transient broadcast error can NEVER roll back +-- the durable insert. v1 uses a public topic (the unguessable session_id is the +-- capability); signed-in sessions can move to private channels + an RLS policy +-- on realtime.messages later without touching this table. +create or replace function broadcast_session_event() +returns trigger +language plpgsql +security definer +set search_path = public +as $$ +begin + begin + perform realtime.send( + jsonb_build_object( + 'id', new.id, + 'seq', new.seq, + 'session_id', new.session_id, + 'author', new.author, + 'kind', new.kind, + 'type', new.type, + 'payload', new.payload, + 'created_at', new.created_at + ), + 'session_event', -- event name + 'session:' || new.session_id, -- topic + false -- public topic; secrecy = capability (v1) + ); + exception when others then + null; -- best-effort fan-out; the durable append already succeeded + end; + return new; +end; +$$; + +drop trigger if exists session_events_broadcast on session_events; +create trigger session_events_broadcast + after insert on session_events + for each row execute function broadcast_session_event(); + +-- ─── stale-event sweep (cron wiring is a follow-up; callable manually) ─────── +create or replace function cleanup_stale_session_events(max_age interval default '30 days') +returns integer +language plpgsql +security definer +set search_path = public +as $$ +declare + swept integer; +begin + delete from session_events where created_at < now() - max_age; + get diagnostics swept = row_count; + return swept; +end; +$$; + +-- ─── RLS — users read their own events; the RPC is the SOLE writer ─────────── +alter table session_events enable row level security; + +drop policy if exists "Users read own session events" on session_events; +create policy "Users read own session events" + on session_events for select using (auth.uid() = user_id); +-- Anonymous capability sessions have user_id = null → no direct table read; +-- the server (service_role, bypasses RLS) and the broadcast topic serve them. + +-- service_role: SELECT only on the table — the SECURITY DEFINER RPC is the sole +-- writer, mirroring wallets/wallet_ledger. +grant select on session_events to service_role; +grant usage, select on all sequences in schema public to service_role; + +-- The append RPC is server-only. Never grant execute to authenticated/anon. +revoke all on function append_session_event(text, uuid, text, text, text, jsonb, text) from public; +revoke all on function cleanup_stale_session_events(interval) from public; +grant execute on function append_session_event(text, uuid, text, text, text, jsonb, text) to service_role; +grant execute on function cleanup_stale_session_events(interval) to service_role;