Something went wrong. Try again.
A local-first event pipeline for independent agents, built on Jazz.
Something went wrong. Try again.
13 kB · 201 lines
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202import fs from "node:fs/promises";import { afterEach, describe, expect, test } from "vitest";import type { JazzThoughtStore } from "../src/jazz/store.js";import type { ThoughtEvent } from "../src/events/types.js";import { REVIEW_NONCE_HEADER, REVIEW_SIGNATURE_HEADER, REVIEW_TIMESTAMP_HEADER, signReviewRequest,} from "../src/review/web-capability.js";import { startInspectorServer } from "../src/web/inspector.js";import { PROPOSAL_PROPOSED_EVENT_TYPE, WORKBENCH_DOCUMENT_SOURCE } from "../src/workbench/contracts.js";import { FIXTURE_RUNNER_ID } from "../src/workbench/runners.js";import { temporaryProject, testStore } from "./helpers.js";
const stores: JazzThoughtStore[] = [];const roots: string[] = [];const servers: import("node:http").Server[] = [];
afterEach(async () => { await Promise.all(servers.splice(0).map((server) => new Promise<void>((resolve) => { server.closeAllConnections(); server.close(() => resolve()); }))); await Promise.all(stores.splice(0).map((store) => store.close())); await Promise.all(roots.splice(0).map((root) => fs.rm(root, { recursive: true, force: true })));});
async function fixture(options: { reviewCapability?: Buffer } = {}): Promise<{ store: JazzThoughtStore; base: string; origin: ThoughtEvent }> { const root = await temporaryProject("thoughtstream-workbench-inspector-"); roots.push(root); const store = await testStore(root); stores.push(store); const origin = (await store.appendEvent({ type: "stream.thought.source.rss.item", schemaVersion: 1, source: "rss:fixture", sourceKind: "rss", externalId: "item-1", idempotencyKey: "rss:item-1", occurredAt: "2026-07-14T00:00:00.000Z", actor: "rss:fixture", correlationId: "poll-1", privacy: "public-source", payload: { title: "Fixture item", summary: "Synthetic <script>alert(1)</script> summary" }, })).event; const server = await startInspectorServer(store, { port: 0, proposalActor: "operator:test", ...(options.reviewCapability ? { reviewCapability: options.reviewCapability } : {}), }); servers.push(server); const address = server.address(); if (!address || typeof address === "string") throw new Error("Missing inspector address"); return { store, base: `http://127.0.0.1:${address.port}`, origin };}
function signedHeaders(key: Buffer, pathName: string, body: Buffer): Record<string, string> { const signed = signReviewRequest(key, { method: "POST", path: pathName, body }); return { "content-type": "application/json", [REVIEW_TIMESTAMP_HEADER]: signed.timestamp, [REVIEW_NONCE_HEADER]: signed.nonce, [REVIEW_SIGNATURE_HEADER]: signed.signature, };}
async function signedPost(base: string, key: Buffer, pathName: string, payload: unknown): Promise<{ status: number; body: Record<string, unknown> }> { const body = Buffer.from(JSON.stringify(payload), "utf8"); const response = await fetch(`${base}${pathName}`, { method: "POST", headers: signedHeaders(key, pathName, body), body }); return { status: response.status, body: (await response.json()) as Record<string, unknown> };}
describe("workbench inspector routes", () => { test("returns 405 for every workbench write when no Review verifier is configured", async () => { const { base, origin } = await fixture(); for (const pathName of [ "/api/workbench/documents", "/api/workbench/documents/doc/versions", "/api/workbench/documents/doc/selections", "/api/workbench/documents/doc/proposals", "/api/workbench/proposals/evt/decisions", ]) { const response = await fetch(`${base}${pathName}`, { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ originEventId: origin.id, title: "x", body: "y", requestId: "req-00000001" }), }); expect(response.status, pathName).toBe(405); } const unknown = await fetch(`${base}/api/workbench/unknown`, { method: "POST", body: "{}" }); expect(unknown.status).toBe(404); const list = await fetch(`${base}/api/workbench/documents`); expect(list.status).toBe(200); expect(await list.json()).toMatchObject({ items: [], runners: [{ id: FIXTURE_RUNNER_ID, inference: "none" }] }); const missing = await fetch(`${base}/api/workbench/documents/missing`); expect(missing.status).toBe(404); const candidates = await fetch(`${base}/api/workbench/candidates?documentId=missing`); expect(candidates.status).toBe(404); const badCandidates = await fetch(`${base}/api/workbench/candidates`); expect(badCandidates.status).toBe(400); });
test("rejects unsigned writes and applies signed writes through the trusted workflow", async () => { const key = Buffer.alloc(32, 31); const { store, base, origin } = await fixture({ reviewCapability: key }); const createBody = Buffer.from(JSON.stringify({ originEventId: origin.id, title: "Notes", body: "# Notes\n", requestId: "req-create-0001" }), "utf8"); const unsigned = await fetch(`${base}/api/workbench/documents`, { method: "POST", headers: { "content-type": "application/json" }, body: createBody }); expect(unsigned.status).toBe(403); expect(await store.listCurrentDocuments(WORKBENCH_DOCUMENT_SOURCE)).toEqual([]); const wrongPath = await fetch(`${base}/api/workbench/documents`, { method: "POST", headers: signedHeaders(key, "/api/workbench/documents/other/versions", createBody), body: createBody, }); expect(wrongPath.status).toBe(403); const malformed = await signedPost(base, key, "/api/workbench/documents", { originEventId: origin.id, title: "", body: "x", requestId: "short" }); expect(malformed.status).toBe(400); const missingOrigin = await signedPost(base, key, "/api/workbench/documents", { originEventId: "evt_missing", title: "x", body: "y", requestId: "req-create-0009" }); expect(missingOrigin.status).toBe(404);
const created = await signedPost(base, key, "/api/workbench/documents", { originEventId: origin.id, title: "Notes", body: "# Notes\n", requestId: "req-create-0001" }); expect(created.status).toBe(201); const documentId = String(created.body.documentId); const headVersionId = String(created.body.versionId); const documentPath = `/api/workbench/documents/${encodeURIComponent(documentId)}`;
const list = (await (await fetch(`${base}/api/workbench/documents`)).json()) as { items: Array<Record<string, unknown>> }; expect(list.items).toHaveLength(1); expect(list.items[0]).toMatchObject({ documentId, title: "Notes", originEventId: origin.id, headVersionId });
const candidates = (await (await fetch(`${base}/api/workbench/candidates?documentId=${encodeURIComponent(documentId)}`)).json()) as Record<string, unknown>; expect((candidates.events as Array<Record<string, unknown>>).some((event) => event.eventId === origin.id && event.origin === true)).toBe(true); expect(candidates.versions).toEqual([expect.objectContaining({ versionId: headVersionId, head: true })]);
const selection = await signedPost(base, key, `${documentPath}/selections`, { eventIds: [origin.id], versionIds: [headVersionId], requestId: "req-select-0001" }); expect(selection.status).toBe(201); const selectionId = String(selection.body.selectionId);
const proposal = await signedPost(base, key, `${documentPath}/proposals`, { selectionId, runnerId: FIXTURE_RUNNER_ID, requestId: "req-prop-00001" }); expect(proposal.status).toBe(201); const proposalEventId = String(proposal.body.proposalEventId); const unknownRunner = await signedPost(base, key, `${documentPath}/proposals`, { selectionId, runnerId: "missing", requestId: "req-prop-00002" }); expect(unknownRunner.status).toBe(404);
const detail = (await (await fetch(`${base}${documentPath}`)).json()) as Record<string, any>; expect(detail.head).toMatchObject({ documentId, title: "Notes", body: "# Notes\n", versionId: headVersionId }); expect(detail.origin).toMatchObject({ eventId: origin.id }); expect(detail.selections).toHaveLength(1); expect(detail.selections[0].selectedEvents[0]).toMatchObject({ eventId: origin.id, payloadHash: origin.payloadHash }); expect(detail.selections[0].selectedEvents[0].excerpt).toContain("<script>alert(1)</script>"); expect(detail.proposals).toHaveLength(1); expect(detail.proposals[0]).toMatchObject({ eventId: proposalEventId, status: "pending", stale: false, inference: "none" }); expect(detail.proposals[0].diff).toContain("+## Sources");
const edit = await signedPost(base, key, `${documentPath}/versions`, { baseVersionId: headVersionId, title: "Notes", body: "# Notes\n\nOperator edit.\n", requestId: "req-edit-00001" }); expect(edit.status).toBe(201); const editedVersionId = String(edit.body.versionId); const staleEdit = await signedPost(base, key, `${documentPath}/versions`, { baseVersionId: headVersionId, title: "Notes", body: "# Other\n", requestId: "req-edit-00002" }); expect(staleEdit.status).toBe(409); expect(staleEdit.body).toMatchObject({ stale: true, headVersionId: editedVersionId });
const staleAccept = await signedPost(base, key, `/api/workbench/proposals/${encodeURIComponent(proposalEventId)}/decisions`, { disposition: "accept", submissionId: "sub-accept-0001" }); expect(staleAccept.status).toBe(409); expect(staleAccept.body).toMatchObject({ stale: true, headVersionId: editedVersionId }); const afterStale = (await (await fetch(`${base}${documentPath}`)).json()) as Record<string, any>; expect(afterStale.head.versionId).toBe(editedVersionId); expect(afterStale.head.body).toBe("# Notes\n\nOperator edit.\n"); expect(afterStale.proposals[0]).toMatchObject({ status: "stale", stale: true });
const rejected = await signedPost(base, key, `/api/workbench/proposals/${encodeURIComponent(proposalEventId)}/decisions`, { disposition: "reject", submissionId: "sub-reject-0001" }); expect(rejected.status).toBe(201); expect(rejected.body).toMatchObject({ disposition: "reject", replayed: false }); const replay = await signedPost(base, key, `/api/workbench/proposals/${encodeURIComponent(proposalEventId)}/decisions`, { disposition: "reject", submissionId: "sub-reject-0001" }); expect(replay.status).toBe(201); expect(replay.body).toMatchObject({ decisionEventId: rejected.body.decisionEventId, replayed: true }); const conflict = await signedPost(base, key, `/api/workbench/proposals/${encodeURIComponent(proposalEventId)}/decisions`, { disposition: "accept", submissionId: "sub-accept-0002" }); expect(conflict.status).toBe(409);
const selection2 = await signedPost(base, key, `${documentPath}/selections`, { eventIds: [origin.id], versionIds: [], requestId: "req-select-0002" }); const proposal2 = await signedPost(base, key, `${documentPath}/proposals`, { selectionId: String(selection2.body.selectionId), runnerId: FIXTURE_RUNNER_ID, requestId: "req-prop-00003" }); expect(proposal2.status).toBe(201); const accepted = await signedPost(base, key, `/api/workbench/proposals/${encodeURIComponent(String(proposal2.body.proposalEventId))}/decisions`, { disposition: "accept", submissionId: "sub-accept-0003" }); expect(accepted.status).toBe(201); expect(accepted.body).toMatchObject({ disposition: "accept", replayed: false }); expect(accepted.body.judgmentEventId).toBeDefined(); const final = (await (await fetch(`${base}${documentPath}`)).json()) as Record<string, any>; expect(final.head.versionId).toBe(accepted.body.resultVersionId); expect(final.head.body).toContain("## Sources"); expect(final.versions.map((version: { reason: string }) => version.reason)).toEqual(["created", "operator-edit", "proposal-accepted"]); expect((await store.listEvents({ types: [PROPOSAL_PROPOSED_EVENT_TYPE] }))).toHaveLength(2);
// Signature replay is refused: the nonce was consumed. const body = Buffer.from(JSON.stringify({ disposition: "accept", submissionId: "sub-accept-0003" }), "utf8"); const headers = signedHeaders(key, `/api/workbench/proposals/${encodeURIComponent(String(proposal2.body.proposalEventId))}/decisions`, body); const first = await fetch(`${base}/api/workbench/proposals/${encodeURIComponent(String(proposal2.body.proposalEventId))}/decisions`, { method: "POST", headers, body }); expect(first.status).toBe(201); const second = await fetch(`${base}/api/workbench/proposals/${encodeURIComponent(String(proposal2.body.proposalEventId))}/decisions`, { method: "POST", headers, body }); expect(second.status).toBe(403); });});