Something went wrong. Try again.
A local-first event pipeline for independent agents, built on Jazz.
Something went wrong. Try again.
13 kB · 344 lines
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345import fs from "node:fs/promises";import { afterEach, describe, expect, test, vi } from "vitest";import { AGENT_MESSAGE_RESPONSE_EVENT_TYPE, AGENT_MESSAGE_SOURCE_EVENT_TYPE, appendCoAgentMessage, parseAgentMessageResponseEvent,} from "../src/agents/agent-messages.js";import { declarationFingerprint } from "../src/agents/declarations.js";import { OBSERVATION_OUTPUT_CONTRACT } from "../src/agents/output-contracts.js";import { ThoughtAgentRuntime } from "../src/agents/runtime.js";import { buildSubscribedAgentConversationContextPacket, type AgentContextPacket,} from "../src/agents/context.js";import { sha256 } from "../src/core/json.js";import type { AgentRunner, ThoughtAgentDeclaration } from "../src/agents/types.js";import type { EventCandidate } from "../src/events/types.js";import { TelegramChannelDispatcher } from "../src/bridges/telegram-dispatcher.js";import type { TelegramBotClient } from "../src/connectors/telegram-bot.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("private agent messages", () => { test("keeps an exact Co-to-Stream thread separate and settles typed response receipts", async () => { const project = await temporaryProject("thoughtstream-agent-message-"); roots.push(project); const store = await testStore(project); stores.push(store); const contexts = new Map<string, AgentContextPacket>(); const runner: AgentRunner = { mode: "deterministic", run: vi.fn(async (input) => { contexts.set(String(input.event.payload.messageId), input.context); return { summary: `Stream reply: ${String(input.event.payload.text)}`, tags: ["conversation"], importance: "normal" as const, confidence: 0.5, }; }), }; const declaration = agentMessageDeclaration(); const runtime = new ThoughtAgentRuntime(store, [runner]);
const first = await appendCoAgentMessage(store, { messageId: "co-first", threadId: "co-stream-test", text: "Hello Stream.", occurredAt: "2026-08-04T02:00:00.000Z", }); expect(first.inserted).toBe(true); const replay = await appendCoAgentMessage(store, { messageId: "co-first", threadId: "co-stream-test", text: "Hello Stream.", occurredAt: "2026-08-04T02:01:00.000Z", }); expect(replay).toMatchObject({ inserted: false, event: { id: first.event.id } }); await expect(appendCoAgentMessage(store, { messageId: "co-first", threadId: "co-stream-test", text: "Divergent reuse.", })).rejects.toThrow("divergent content");
const [firstProcessed] = await runtime.consumeBacklog([declaration]); const firstRun = await store.getRun(firstProcessed!.runId); const firstOutput = await store.getEvent(firstRun!.outputEventIds[0]!); const firstResponse = parseAgentMessageResponseEvent(firstOutput!); expect(firstRun).toMatchObject({ status: "completed", agentId: "stream-agent-conversation" }); expect(firstOutput).toMatchObject({ type: AGENT_MESSAGE_RESPONSE_EVENT_TYPE, source: "agent:stream-agent-conversation", actor: "stream-agent-conversation", parentEventId: first.event.id, rootEventId: first.event.id, correlationId: "co-stream-test", privacy: "sensitive", }); expect(firstResponse).toMatchObject({ inReplyToMessageId: "co-first", threadId: "co-stream-test", senderAgentId: "stream-agent-conversation", recipientAgentId: "co", summary: "Stream reply: Hello Stream.", }); const dispatcher = new TelegramChannelDispatcher({ id: "telegram-notifier:agent-message-deny", client: { sendMessage: vi.fn(() => { throw new Error("agent response reached Telegram"); }) } as unknown as TelegramBotClient, chatId: "123456789", allowedSources: ["agent-message:co"], allowedActors: ["agent:co"], directReplyAgentIds: ["stream-agent-conversation"], runStatuses: ["completed", "failed"], }); await expect(dispatcher.sendPending(store, { includeNormal: true })).resolves.toMatchObject({ pending: 1, eligible: 0, delivered: 0, }); expect(contexts.get("co-first")).toMatchObject({ text: "Hello Stream.", manifest: { contextStrategy: "agent-conversation", agentMessage: { senderAgentId: "co", recipientAgentId: "stream-agent-conversation", threadId: "co-stream-test", }, }, }); expect(contexts.get("co-first")?.messages).toBeUndefined();
await appendCoAgentMessage(store, { messageId: "co-other-thread", threadId: "other-thread", text: "This is another thread.", occurredAt: "2026-08-04T02:02:00.000Z", }); await runtime.consumeBacklog([declaration]); expect(contexts.get("co-other-thread")?.messages).toBeUndefined();
await appendCoAgentMessage(store, { messageId: "co-second", threadId: "co-stream-test", text: "Do you remember me?", occurredAt: "2026-08-04T02:03:00.000Z", }); await runtime.consumeBacklog([declaration]); expect(contexts.get("co-second")?.messages).toEqual([ { role: "user", content: "Hello Stream." }, { role: "assistant", content: "Stream reply: Hello Stream." }, ]); expect(JSON.stringify(contexts.get("co-second")?.messages)).not.toContain("another thread"); });
test("skips a source-row route forgery before runner invocation and advances progress", async () => { const project = await temporaryProject("thoughtstream-agent-message-skip-"); roots.push(project); const store = await testStore(project); stores.push(store); const runner: AgentRunner = { mode: "deterministic", run: vi.fn(async () => ({ summary: "should not run", tags: [], importance: "low" as const, confidence: 0, })), }; const runtime = new ThoughtAgentRuntime(store, [runner]); const declaration = agentMessageDeclaration(); const forged: EventCandidate = { type: AGENT_MESSAGE_SOURCE_EVENT_TYPE, schemaVersion: 1, source: "agent-message:co", sourceKind: "agent", externalId: "forged-recipient", idempotencyKey: "forged-recipient", occurredAt: "2026-08-04T02:00:00.000Z", actor: "agent:co", correlationId: "co-stream-test", privacy: "sensitive", payload: { messageId: "forged-recipient", threadId: "co-stream-test", senderAgentId: "co", recipientAgentId: "other-agent", text: "Do not admit this.", }, }; const appended = await store.appendEvent(forged); const [processed] = await runtime.consumeBacklog([declaration]); const run = await store.getRun(processed!.runId); expect(runner.run).not.toHaveBeenCalled(); expect(run).toMatchObject({ status: "skipped", triggerEventId: appended.event.id, errorText: "Agent message is outside the endpoint route", contextManifest: { skip: { code: "agent-message-not-admitted" }, }, }); expect(await store.getConsumerProgress( `consumer-progress:stream-agent-conversation:1:agent-message%3Aco`, )).toMatchObject({ lastSequence: appended.event.sourceSequence }); });
test("freezes operator documents and exact thread context for retries", async () => { const project = await temporaryProject("thoughtstream-agent-message-snapshot-"); roots.push(project); const store = await testStore(project); stores.push(store); await putCurrentDocument(store, "identity", "identity.md", "# Stream identity v1\n", "identity-v1"); await putCurrentDocument(store, "memory", "memory.md", "# Stream memory\n", "memory-v1"); const declaration: ThoughtAgentDeclaration = { ...agentMessageDeclaration(), contextDocumentMaxChars: 4_000, contextDocumentSubscriptions: [{ source: "filesystem:telegram-agent-context", paths: ["identity.md", "memory.md"], required: true, }], }; declaration.declarationFingerprint = declarationFingerprint(declaration); const appended = await appendCoAgentMessage(store, { messageId: "co-snapshot", threadId: "co-stream-snapshot", text: "Freeze this turn.", occurredAt: "2026-08-04T02:00:00.000Z", }); const first = await buildSubscribedAgentConversationContextPacket(declaration, appended.event, store); await putCurrentDocument(store, "identity", "identity.md", "# Stream identity v2\n", "identity-v2"); const retried = await buildSubscribedAgentConversationContextPacket(declaration, appended.event, store); expect(retried).toEqual(first); expect(first.systemText).toContain("Stream identity v1"); expect(first.systemText).not.toContain("Stream identity v2"); expect(first.systemText).toContain("current private correspondent is agent:co"); const snapshot = first.manifest.contextSnapshot as { id: string }; expect(await store.getDocumentVersion(snapshot.id)).toMatchObject({ source: "context:stream-agent-conversation", path: `agent-subscribed-context/${appended.event.id}.json`, contentType: "application/vnd.thoughtstream.agent-context+json", }); });
test("discovers the first Co message after an exact replay-now endpoint starts", async () => { const project = await temporaryProject("thoughtstream-agent-message-live-"); roots.push(project); const store = await testStore(project); stores.push(store); const runner: AgentRunner = { mode: "deterministic", run: async () => ({ summary: "I hear Co.", tags: ["conversation"], importance: "normal", confidence: 0.5, }), }; const declaration = { ...agentMessageDeclaration(), initialReplay: "now" as const }; declaration.declarationFingerprint = declarationFingerprint(declaration); const runtime = new ThoughtAgentRuntime(store, [runner], { reconcileIntervalMs: 25 }); const consumers = await runtime.startConsumers([declaration]); const appended = await appendCoAgentMessage(store, { messageId: "co-after-start", threadId: "co-stream-live", text: "Can you hear me?", }); await waitFor(async () => (await store.getRunsForTriggerEvents([appended.event.id])) .some((run) => run.status === "completed"), 5_000); await consumers.stop(); const [response] = (await store.listEvents({ types: [AGENT_MESSAGE_RESPONSE_EVENT_TYPE] })) .filter((event) => event.parentEventId === appended.event.id); expect(parseAgentMessageResponseEvent(response!)).toMatchObject({ summary: "I hear Co." }); });});
function agentMessageDeclaration(): ThoughtAgentDeclaration { const declaration: ThoughtAgentDeclaration = { id: "stream-agent-conversation", version: 1, name: "Stream", description: "fixture private agent conversation", mode: "deterministic", role: "standard", outputContract: { ...OBSERVATION_OUTPUT_CONTRACT.identity }, outputMode: "conversation-text", eventTypes: [AGENT_MESSAGE_SOURCE_EVENT_TYPE], compiledEventTypes: [AGENT_MESSAGE_SOURCE_EVENT_TYPE], sourcePatterns: ["agent-message:co"], acceptedPrivacy: ["sensitive"], initialReplay: "beginning", outputEventType: AGENT_MESSAGE_RESPONSE_EVENT_TYPE, emit: [AGENT_MESSAGE_RESPONSE_EVENT_TYPE], promptRef: "fixture-agent-message.md", systemPrompt: "Reply to Co.", enabled: true, maxEvents: 20, maxInputChars: 8_000, contextStrategy: "agent-conversation", maxOutputTokens: 2_000, timeoutMs: 60_000, tools: [], proposals: [], externalActions: false, }; declaration.declarationFingerprint = declarationFingerprint(declaration); return declaration;}
async function putCurrentDocument( store: JazzThoughtStore, documentId: string, documentPath: string, content: string, versionId: string,): Promise<void> { const source = "filesystem:telegram-agent-context"; const digest = sha256(content); const at = "2026-08-04T01:00:00.000Z"; await store.appendDocumentVersion({ id: versionId, source, documentId, path: documentPath, contentType: "text/markdown", sha256: digest, content, sizeBytes: Buffer.byteLength(content), mtimeMs: Date.parse(at), createdAt: at, }); await store.upsertCurrentDocument({ id: `${source}:${documentId}`, source, documentId, path: documentPath, versionId, sha256: digest, contentType: "text/markdown", sizeBytes: Buffer.byteLength(content), mtimeMs: Date.parse(at), deleted: false, updatedAt: at, });}
async function waitFor(predicate: () => Promise<boolean>, timeoutMs: number): Promise<void> { const deadline = Date.now() + timeoutMs; while (Date.now() < deadline) { if (await predicate()) return; await new Promise((resolve) => setTimeout(resolve, 25)); } throw new Error("Timed out waiting for agent-message consumer");}