Something went wrong. Try again.
A local-first event pipeline for independent agents, built on Jazz.
Something went wrong. Try again.
19 kB · 472 lines
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473import fs from "node:fs/promises";import path from "node:path";import { afterEach, describe, expect, test } from "vitest";import { REVIEW_RESPONSE_OUTPUT_CONTRACT, canonicalStructuredOutput, createOutputContractRegistry, outputContractIdentityJson, reviewResponseSummary,} from "../src/agents/output-contracts.js";import type { JsonObject } from "../src/core/json.js";import type { PrivacyClass, ThoughtEvent } from "../src/events/types.js";import type { JazzThoughtStore } from "../src/jazz/store.js";import { activeReviewDecisions, appendReviewPrompt, createReviewItem, projectReviewQueue, recordBrowserReviewDecision,} from "../src/review/review.js";import { REVIEW_RESPONSE_EVENT_TYPE } from "../src/review/types.js";import { REVIEW_NONCE_HEADER, REVIEW_SIGNATURE_HEADER, REVIEW_TIMESTAMP_HEADER, signReviewRequest,} from "../src/review/web-capability.js";import type { AgentRun } from "../src/store/types.js";import { projectTrainingExamples, writeTrainingJsonl } from "../src/training/judgments.js";import { startInspectorServer } from "../src/web/inspector.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 })));});
describe("thought stream Review", () => { test("freezes blinded same-trigger pairs and exports only active judgeable public decisions", async () => { const { store, item } = await fixture(); const unresolved = await projectReviewQueue(store); expect(unresolved.counts).toMatchObject({ total: 1, unresolved: 1, decided: 0 }); expect(unresolved.items[0]?.candidates.map((candidate) => Object.keys(candidate))).toEqual([ ["label", "response"], ["label", "response"], ]); expect(JSON.stringify(unresolved)).not.toContain("fixture/model-a"); expect(JSON.stringify(unresolved)).not.toContain("run_a");
const underdetermined = await recordBrowserReviewDecision(store, item.id, { disposition: "underdetermined", confidence: "medium", reasonCodes: ["missing-mechanism"], responseTags: [], notes: "The prompt lacks the critique that would make the mechanism change judgeable.", trainingEligible: false, submissionId: "submission-underdetermined-0001", }); expect((await projectTrainingExamples(store))).toEqual([]);
const decidedQueue = await projectReviewQueue(store); expect(decidedQueue.items[0]?.decision).toMatchObject({ eventId: underdetermined.id, disposition: "underdetermined", trainingEligible: false, externalExportEligible: false, }); expect(decidedQueue.items[0]?.candidates[0]?.provenance).toBeDefined();
const preferredLabel = decidedQueue.items[0]!.candidates[0].label; const preferredResponse = decidedQueue.items[0]!.candidates[0].response; const rejectedResponse = decidedQueue.items[0]!.candidates[1].response; const preferred = await recordBrowserReviewDecision(store, item.id, { disposition: "prefer", preferredCandidate: preferredLabel, preferenceStrength: "strong", confidence: "high", reasonCodes: ["mechanism-update"], responseTags: ["specific"], notes: "private note excluded from the dataset", trainingEligible: true, submissionId: "submission-preference-00000002", supersedesDecisionEventId: underdetermined.id, }); const repeated = await recordBrowserReviewDecision(store, item.id, { disposition: "prefer", preferredCandidate: preferredLabel, preferenceStrength: "strong", confidence: "high", reasonCodes: ["mechanism-update"], responseTags: ["specific"], notes: "private note excluded from the dataset", trainingEligible: true, submissionId: "submission-preference-00000002", supersedesDecisionEventId: underdetermined.id, }); expect(repeated.id).toBe(preferred.id); const active = await activeReviewDecisions(store); expect(active.active.map((event) => event.id)).toEqual([preferred.id]); expect(active.inactiveIds.has(underdetermined.id)).toBe(true);
const examples = await projectTrainingExamples(store); expect(examples).toHaveLength(1); expect(examples[0]).toMatchObject({ format: "thoughtstream.training-example.v4", kind: "prefer", judgment: { criterion: "response-quality", criterionVersion: 1, preferenceStrength: "strong", confidence: "high", reasonCodes: ["mechanism-update"], responseTags: ["specific"], }, input: { prompt: "Given the critique and evidence, explain which mechanism should change.", evidence: "The user critique says the retrieval gate was bypassed in the failed run.", }, chosen: { response: preferredResponse, summary: preferredResponse, confidence: 1 }, rejected: { response: rejectedResponse, summary: rejectedResponse, confidence: 1 }, review: { campaignId: "review-pipeline-canary", campaignVersion: 1, disposition: "prefer", candidateCount: 2, }, }); const serialized = JSON.stringify(examples[0]); expect(serialized).not.toContain("private note excluded"); expect(serialized).not.toContain("submission-preference"); expect(serialized).not.toContain(item.id); expect(serialized).not.toContain(preferred.id);
const output = path.join(roots[0]!, "review-dataset.jsonl"); const manifest = await writeTrainingJsonl(output, examples); expect(manifest).toMatchObject({ format: "thoughtstream.training-dataset-manifest.v4", examples: 1, exampleFormats: { "thoughtstream.training-example.v4": 1 }, reviewCampaigns: ["review-pipeline-canary@1"], }); });
test("records contract-valid corrections with both rejected candidates", async () => { const { store, item } = await fixture(); await expect(recordBrowserReviewDecision(store, item.id, { disposition: "correct", replacementResponse: "", reasonCodes: [], responseTags: [], trainingEligible: true, submissionId: "submission-invalid-correction", })).rejects.toThrow(); await recordBrowserReviewDecision(store, item.id, { disposition: "correct", replacementResponse: "The critique shows retrieval was bypassed, so the mechanism change is to enforce retrieval before generation and test that invariant.", confidence: "high", reasonCodes: ["mechanism-update"], responseTags: ["specific"], trainingEligible: true, submissionId: "submission-valid-correction-01", }); const [example] = await projectTrainingExamples(store); expect(example).toMatchObject({ format: "thoughtstream.training-example.v4", kind: "correct", chosen: { response: "The critique shows retrieval was bypassed, so the mechanism change is to enforce retrieval before generation and test that invariant.", }, additionalRejected: [expect.objectContaining({ response: "Candidate B proposes a vague tone adjustment." })], }); });
test("keeps private review useful but browser-ineligible for external training export", async () => { const { store, item } = await fixture({ privacy: "private", externalExportEligible: false }); const queue = await projectReviewQueue(store); await recordBrowserReviewDecision(store, item.id, { disposition: "prefer", preferredCandidate: queue.items[0]!.candidates[0].label, preferenceStrength: "slight", reasonCodes: [], responseTags: [], trainingEligible: true, submissionId: "submission-private-preference", }); const reviewed = await projectReviewQueue(store); expect(reviewed.items[0]?.decision).toMatchObject({ trainingEligible: true, externalExportEligible: false, }); expect(await projectTrainingExamples(store, { includeSensitivePrivate: true })).toEqual([]); await expect(appendReviewPrompt(store, { externalId: "invalid-private-declassification", privacy: "private", payload: promptPayload(true), })).rejects.toThrow("Only public-source"); });
test("rejects cross-trigger pairs, undeclared agents, and forced labels on unjudgeable items", async () => { const { store, prompt, runs } = await fixture({ materialize: false }); const otherPrompt = await appendReviewPrompt(store, { externalId: "other-prompt", privacy: "public-source", payload: promptPayload(true), }); const other = await candidate(store, otherPrompt, "agent-b", "run_other", "Other response."); await expect(createReviewItem(store, { promptEventId: prompt.id, candidateRunIds: [runs[0].id, other.id], })).rejects.toThrow("exact review prompt"); const undeclared = await candidate(store, prompt, "agent-z", "run_undeclared", "Undeclared response."); await expect(createReviewItem(store, { promptEventId: prompt.id, candidateRunIds: [runs[0].id, undeclared.id], })).rejects.toThrow("not declared"); const item = await createReviewItem(store, { promptEventId: prompt.id, candidateRunIds: [runs[0].id, runs[1].id], }); await expect(recordBrowserReviewDecision(store, item.id, { disposition: "underdetermined", reasonCodes: [], responseTags: [], trainingEligible: true, submissionId: "submission-forced-label-invalid", })).rejects.toThrow(); await expect(recordBrowserReviewDecision(store, item.id, { disposition: "prefer", preferredCandidate: "A", preferenceStrength: "strong", reasonCodes: ["invented-reason"], responseTags: [], trainingEligible: false, submissionId: "submission-unknown-reason", })).rejects.toThrow("Unknown review reason code"); });
test("accepts only a fresh body-bound loopback capability on the exact Review route", async () => { const { store, item } = await fixture(); const key = Buffer.alloc(32, 29); const server = await startInspectorServer(store, { port: 0, reviewCapability: key }); servers.push(server); const address = server.address(); if (!address || typeof address === "string") throw new Error("Missing inspector address"); const base = `http://127.0.0.1:${address.port}`; const pathName = `/api/reviews/${encodeURIComponent(item.id)}/decisions`; const body = Buffer.from(JSON.stringify({ disposition: "skip", reasonCodes: [], responseTags: [], trainingEligible: false, submissionId: "submission-loopback-canary-01", }), "utf8"); const unsigned = await fetch(`${base}${pathName}`, { method: "POST", headers: { "content-type": "application/json" }, body, }); expect(unsigned.status).toBe(403); expect((await activeReviewDecisions(store)).active).toEqual([]);
const signed = signReviewRequest(key, { method: "POST", path: pathName, body }); const headers = { "content-type": "application/json", [REVIEW_TIMESTAMP_HEADER]: signed.timestamp, [REVIEW_NONCE_HEADER]: signed.nonce, [REVIEW_SIGNATURE_HEADER]: signed.signature, }; const accepted = await fetch(`${base}${pathName}`, { method: "POST", headers, body }); expect(accepted.status).toBe(201); expect((await accepted.json()) as JsonObject).toMatchObject({ active: true }); expect((await activeReviewDecisions(store)).active).toHaveLength(1); const replay = await fetch(`${base}${pathName}`, { method: "POST", headers, body }); expect(replay.status).toBe(403); expect((await activeReviewDecisions(store)).active).toHaveLength(1); });
test("fails queue, decision, and export closed when a frozen run receipt or output pointer changes", async () => { const { store, item, runs } = await fixture(); const queue = await projectReviewQueue(store); await recordBrowserReviewDecision(store, item.id, { disposition: "prefer", preferredCandidate: queue.items[0]!.candidates[0].label, preferenceStrength: "strong", reasonCodes: [], responseTags: [], trainingEligible: true, submissionId: "submission-before-run-tamper", }); await store.upsertRun({ ...runs[0], model: "fixture/tampered-model", updatedAt: "2026-07-26T23:00:00.000Z" }); await expect(projectReviewQueue(store)).rejects.toThrow("receipt changed after materialization"); await expect(projectTrainingExamples(store)).rejects.toThrow("receipt changed after materialization"); await expect(recordBrowserReviewDecision(store, item.id, { disposition: "skip", reasonCodes: [], responseTags: [], trainingEligible: false, submissionId: "submission-after-run-tamper", })).rejects.toThrow("receipt changed after materialization");
const second = await fixture(); await second.store.upsertRun({ ...second.runs[0], outputEventIds: [second.runs[1].outputEventIds[0]!], updatedAt: "2026-07-26T23:00:00.000Z", }); await expect(projectReviewQueue(second.store)).rejects.toThrow("output pointer changed after materialization"); });
test("rejects truncated, projected, enriched, or multi-event candidate context before materialization", async () => { const { store, prompt, runs } = await fixture({ materialize: false }); const create = () => createReviewItem(store, { promptEventId: prompt.id, candidateRunIds: [runs[0].id, runs[1].id], }); await store.upsertRun({ ...runs[0], contextManifest: { ...runs[0].contextManifest, truncated: true, sourceIncludedChars: 999 }, updatedAt: "2026-07-26T23:00:00.000Z", }); await expect(create()).rejects.toThrow("truncated, projected, enriched, or action-capable");
await store.upsertRun({ ...runs[0], inputEventIds: [prompt.id, "event_extra"], contextManifest: { ...runs[0].contextManifest, inputEventIds: [prompt.id, "event_extra"], includedEventIds: [prompt.id, "event_extra"], }, updatedAt: "2026-07-26T23:00:01.000Z", }); await expect(create()).rejects.toThrow("exactly one prompt input");
await store.upsertRun({ ...runs[0], contextManifest: { ...runs[0].contextManifest, payloadFields: ["prompt"], tools: ["web.download-image"] }, updatedAt: "2026-07-26T23:00:02.000Z", }); await expect(create()).rejects.toThrow("truncated, projected, enriched, or action-capable"); });});
async function fixture(options: { privacy?: PrivacyClass; externalExportEligible?: boolean; materialize?: boolean;} = {}): Promise<{ store: JazzThoughtStore; prompt: ThoughtEvent; runs: [AgentRun, AgentRun]; item: ThoughtEvent;}> { const project = await temporaryProject("thoughtstream-review-"); roots.push(project); const store = await testStore(project); stores.push(store); const privacy = options.privacy ?? "public-source"; const prompt = await appendReviewPrompt(store, { externalId: "prompt-1", occurredAt: "2026-07-26T22:00:00.000Z", privacy, payload: promptPayload(options.externalExportEligible ?? true), }); const runs: [AgentRun, AgentRun] = [ await candidate(store, prompt, "agent-a", "run_a", "Candidate A identifies the retrieval mechanism and proposes a concrete gate."), await candidate(store, prompt, "agent-b", "run_b", "Candidate B proposes a vague tone adjustment."), ]; const item = options.materialize === false ? ({ id: "not-materialized" } as ThoughtEvent) : await createReviewItem(store, { promptEventId: prompt.id, candidateRunIds: [runs[0].id, runs[1].id] }); return { store, prompt, runs, item };}
function promptPayload(externalExportEligible: boolean): JsonObject { return { campaign: { id: "review-pipeline-canary", version: 1, label: "Review pipeline canary" }, prompt: "Given the critique and evidence, explain which mechanism should change.", evidence: "The user critique says the retrieval gate was bypassed in the failed run.", criterion: { id: "response-quality", version: 1, label: "Mechanism-changing response quality", instructions: "Prefer the response that identifies the changed mechanism and preserves evidence boundaries.", reasonCodes: ["missing-mechanism", "mechanism-update"], responseTags: ["specific", "vague"], }, candidateAgentIds: ["agent-a", "agent-b"], externalExportEligible, };}
async function candidate( store: JazzThoughtStore, prompt: ThoughtEvent, agentId: string, runId: string, response: string,): Promise<AgentRun> { const structuredOutput = canonicalStructuredOutput( createOutputContractRegistry(), REVIEW_RESPONSE_OUTPUT_CONTRACT.identity, { response }, ); const output = (await store.appendEvent({ type: REVIEW_RESPONSE_EVENT_TYPE, schemaVersion: 1, source: `agent:${agentId}`, sourceKind: "agent", externalId: runId, idempotencyKey: `${runId}:output`, occurredAt: "2026-07-26T22:00:01.000Z", actor: agentId, rootEventId: prompt.id, parentEventId: prompt.id, correlationId: prompt.correlationId, privacy: prompt.privacy, payload: { runId, executionKey: `execution:${runId}`, inputEventId: prompt.id, inputSourceSequence: prompt.sourceSequence, summary: reviewResponseSummary(response), outputContract: outputContractIdentityJson(REVIEW_RESPONSE_OUTPUT_CONTRACT.identity), structuredOutput, model: { provider: "fixture", id: `fixture/${agentId}` }, }, })).event; const run: AgentRun = { id: runId, executionKey: `execution:${runId}`, triggerEventId: prompt.id, agentId, agentVersion: 1, status: "completed", inputEventIds: [prompt.id], outputEventIds: [output.id], attempt: 1, provider: "fixture", model: `fixture/${agentId}`, privacy: prompt.privacy, promptHash: `prompt-hash-${agentId}`, contextManifest: { contextStrategy: "single-event", inputEventIds: [prompt.id], includedEventIds: [prompt.id], omittedEventIds: [], maxEvents: 1, maxChars: 220_000, sourceOriginalChars: 1_000, sourceIncludedChars: 1_000, truncated: false, outputContract: outputContractIdentityJson(REVIEW_RESPONSE_OUTPUT_CONTRACT.identity), tools: [], externalActions: false, }, result: structuredOutput, createdAt: "2026-07-26T22:00:00.000Z", completedAt: "2026-07-26T22:00:02.000Z", updatedAt: "2026-07-26T22:00:02.000Z", }; await store.upsertRun(run); return run;}