Something went wrong. Try again.
A local-first event pipeline for independent agents, built on Jazz.
Something went wrong. Try again.
12 kB · 319 lines
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320import fs from "node:fs/promises";import { afterEach, describe, expect, test } from "vitest";import { CONCEPTUALIZATION_OUTPUT_CONTRACT, createOutputContractRegistry, OutputContractValidationError,} from "../src/agents/output-contracts.js";import { ThoughtAgentRuntime } from "../src/agents/runtime.js";import type { AgentRunner, ThoughtAgentDeclaration } from "../src/agents/types.js";import { createDefaultRegistry } from "../src/events/registry.js";import type { JazzThoughtStore } from "../src/jazz/store.js";import { temporaryProject, testStore } from "./helpers.js";
const stores: JazzThoughtStore[] = [];const roots: string[] = [];
afterEach(async () => { 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("conceptualizer output", () => { test("rejects manually constructed contract and event mismatches at runtime registration", async () => { const project = await temporaryProject(); roots.push(project); const store = await testStore(project); stores.push(store); const runtime = new ThoughtAgentRuntime(store, []); const mismatched = conceptualizerDeclaration("conceptualizer-mismatch"); mismatched.outputEventType = "stream.thought.derived.topics"; mismatched.emit = ["stream.thought.derived.topics"];
await expect(runtime.registerDeclarations([mismatched])) .rejects.toThrow("must bind conceptualization output and concept graph events together"); });
test("validates one strict bounded graph and rejects invalid indices", () => { const registry = createOutputContractRegistry(); const valid = { summary: "Two related ideas", concepts: [ { text: "agent memory", relationship: "DESCRIBES" }, { text: "durable receipts", relationship: "SUPPORTS" }, ], links: [{ fromIndex: 1, toIndex: 0, relationship: "SUPPORTS" }], confidence: 0.8, };
expect(registry.validate(CONCEPTUALIZATION_OUTPUT_CONTRACT.identity, valid)).toEqual(valid); expect(() => registry.validate(CONCEPTUALIZATION_OUTPUT_CONTRACT.identity, { ...valid, links: [{ fromIndex: 2, toIndex: 0, relationship: "SUPPORTS" }], })).toThrow(OutputContractValidationError); expect(() => registry.validate(CONCEPTUALIZATION_OUTPUT_CONTRACT.identity, { ...valid, concepts: [{ text: "Agent Memory", relationship: "DESCRIBES" }], })).toThrow(OutputContractValidationError); expect(() => registry.validate(CONCEPTUALIZATION_OUTPUT_CONTRACT.identity, { ...valid, publicationUri: "at://synthetic/not-a-receipt", })).toThrow(OutputContractValidationError); });
test("validates correction proposals against the contract named in their payload", () => { const registry = createDefaultRegistry(); const payload = { originalRunId: "run-original", repairRequestEventId: "evt-request", repairRunId: "run-repair", originalTriggerEventId: "evt-source", sourceRootEventId: "evt-source", outputContract: { ...CONCEPTUALIZATION_OUTPUT_CONTRACT.identity }, originalModel: { provider: "tinker", id: "public-base-model" }, repairModel: { provider: "tinker", id: "repair-model" }, structuredOutput: { summary: "One concept", concepts: [{ text: "agent memory", relationship: "DESCRIBES" }], confidence: 0.8, }, };
expect(registry.validate( "stream.thought.derived.output.correction.proposed", 1, payload, )).toMatchObject(payload); expect(() => registry.validate( "stream.thought.derived.output.correction.proposed", 1, { ...payload, structuredOutput: { ...payload.structuredOutput, links: [{ fromIndex: 0, toIndex: 4, relationship: "RELATES_TO" }], }, }, )).toThrow("Correction proposal must satisfy its registered output contract"); });
test("rejects graph envelopes whose common summary disagrees with structured output", () => { const registry = createDefaultRegistry(); expect(() => registry.validate("stream.thought.derived.concept.graph", 1, { runId: "run-graph", executionKey: "execution-graph", inputEventId: "evt-source", inputSourceSequence: 1, summary: "Contradictory envelope", confidence: 0.8, outputContract: { ...CONCEPTUALIZATION_OUTPUT_CONTRACT.identity }, structuredOutput: { summary: "Canonical graph", concepts: [{ text: "agent memory", relationship: "DESCRIBES" }], confidence: 0.8, }, })).toThrow("Summary must equal canonical structured output"); });
test("rejects graph envelopes that claim a different contract identity", () => { const registry = createDefaultRegistry(); expect(() => registry.validate("stream.thought.derived.concept.graph", 1, { runId: "run-graph", executionKey: "execution-graph", inputEventId: "evt-source", inputSourceSequence: 1, summary: "Canonical graph", confidence: 0.8, outputContract: { ...CONCEPTUALIZATION_OUTPUT_CONTRACT.identity, sha256: "0".repeat(64), }, structuredOutput: { summary: "Canonical graph", concepts: [{ text: "agent memory", relationship: "DESCRIBES" }], confidence: 0.8, }, })).toThrow("Concept graph must name the canonical conceptualization contract identity"); });
test("rejects direct public-source graph appends at the durable store boundary", async () => { const project = await temporaryProject(); roots.push(project); const store = await testStore(project); stores.push(store); const payload = { runId: "run-direct-graph", executionKey: "execution-direct-graph", inputEventId: "evt-source", inputSourceSequence: 1, summary: "Canonical graph", confidence: 0.8, outputContract: { ...CONCEPTUALIZATION_OUTPUT_CONTRACT.identity }, structuredOutput: { summary: "Canonical graph", concepts: [{ text: "agent memory", relationship: "DESCRIBES" }], confidence: 0.8, }, };
await expect(store.appendEvent({ type: "stream.thought.derived.concept.graph", schemaVersion: 1, source: "agent:direct-fixture", sourceKind: "agent", externalId: "direct-public-graph", idempotencyKey: "direct-public-graph", occurredAt: "2026-07-25T20:00:00.000Z", actor: "direct-fixture", correlationId: "direct-public-graph", privacy: "public-source", payload, })).rejects.toThrow("requires private or stricter privacy");
const inserted = await store.appendEvent({ type: "stream.thought.derived.concept.graph", schemaVersion: 1, source: "agent:direct-fixture", sourceKind: "agent", externalId: "direct-private-graph", idempotencyKey: "direct-private-graph", occurredAt: "2026-07-25T20:00:00.000Z", actor: "direct-fixture", correlationId: "direct-private-graph", privacy: "private", payload, }); expect(inserted.event.privacy).toBe("private"); });
test("settles one private graph event atomically with exact source and run lineage", async () => { const project = await temporaryProject(); roots.push(project); const store = await testStore(project); stores.push(store); const source = (await store.appendEvent({ type: "stream.thought.source.rss.item", schemaVersion: 1, source: "rss:concept-fixture", sourceKind: "rss", externalId: "concept-fixture", idempotencyKey: "concept-fixture", occurredAt: "2026-07-25T20:00:00.000Z", actor: "rss:concept-fixture", correlationId: "concept-fixture", privacy: "public-source", payload: { title: "Durable memory requires receipts" }, })).event; const runner: AgentRunner = { mode: "deterministic", run: async () => ({ summary: "Receipts make memory changes inspectable", concepts: [ { text: "durable memory", relationship: "DESCRIBES" }, { text: "execution receipts", relationship: "SUPPORTS" }, ], links: [{ fromIndex: 1, toIndex: 0, relationship: "SUPPORTS" }], confidence: 0.9, }), }; const runtime = new ThoughtAgentRuntime(store, [runner]);
const [result] = await runtime.consumeBacklog([conceptualizerDeclaration("conceptualizer-valid")]); const graph = result?.derivedEvent;
expect(result?.error).toBeUndefined(); expect(graph).toMatchObject({ type: "stream.thought.derived.concept.graph", parentEventId: source.id, rootEventId: source.id, privacy: "private", payload: { inputEventId: source.id, inputSourceSequence: source.sourceSequence, structuredOutput: { concepts: [{ text: "durable memory" }, { text: "execution receipts" }], links: [{ fromIndex: 1, toIndex: 0, relationship: "SUPPORTS" }], }, }, }); expect(graph?.payload).not.toHaveProperty("uri"); expect(graph?.payload).not.toHaveProperty("cid"); expect((await store.listRuns())[0]?.outputEventIds).toEqual([graph?.id]); });
test("emits no graph when any link falls outside the concept array", async () => { const project = await temporaryProject(); roots.push(project); const store = await testStore(project); stores.push(store); await store.appendEvent({ type: "stream.thought.source.rss.item", schemaVersion: 1, source: "rss:invalid-concept-fixture", sourceKind: "rss", externalId: "invalid-concept-fixture", idempotencyKey: "invalid-concept-fixture", occurredAt: "2026-07-25T20:00:00.000Z", actor: "rss:invalid-concept-fixture", correlationId: "invalid-concept-fixture", privacy: "private", payload: { title: "Invalid graph fixture" }, }); const runner: AgentRunner = { mode: "deterministic", run: async () => ({ summary: "Invalid graph", concepts: [{ text: "one concept", relationship: "DESCRIBES" }], links: [{ fromIndex: 0, toIndex: 4, relationship: "RELATES_TO" }], confidence: 0.5, }), }; const runtime = new ThoughtAgentRuntime(store, [runner]);
const [result] = await runtime.consumeBacklog([ conceptualizerDeclaration("conceptualizer-invalid", "rss:invalid-concept-fixture"), ]);
expect(result).toMatchObject({ error: "Agent final output rejected by canonical contract" }); expect((await store.listEvents({ types: ["stream.thought.derived.concept.graph"] }))).toHaveLength(0); expect((await store.listEvents({ types: ["stream.thought.agent.repair.requested"] }))).toHaveLength(0); expect((await store.listRuns())[0]?.result).toMatchObject({ failureDiagnostic: { reason: "output-contract-invalid" }, }); });});
function conceptualizerDeclaration( id: string, source = "rss:concept-fixture",): ThoughtAgentDeclaration { return { id, version: 1, name: id, description: "Conceptualizer fixture", mode: "deterministic", role: "standard", outputContract: { ...CONCEPTUALIZATION_OUTPUT_CONTRACT.identity }, declarationFingerprint: `${id}-fingerprint`, provider: "openai-compatible", model: "fixture-model", eventTypes: ["stream.thought.source.rss.item"], compiledEventTypes: ["stream.thought.source.rss.item"], sourcePatterns: [source], acceptedPrivacy: ["public-source", "private"], initialReplay: "beginning", outputEventType: "stream.thought.derived.concept.graph", emit: ["stream.thought.derived.concept.graph"], promptRef: "prompts/conceptualizer.md", systemPrompt: "Extract concepts.", enabled: true, maxEvents: 1, maxInputChars: 64_000, contextStrategy: "single-event", maxOutputTokens: 2_000, timeoutMs: 60_000, tools: [], externalActions: false, };}