Something went wrong. Try again.
A local-first event pipeline for independent agents, built on Jazz.
Something went wrong. Try again.
35 kB · 829 lines
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830import fs from "node:fs/promises";import path from "node:path";import { afterEach, describe, expect, test } from "vitest";import { buildSubscribedTelegramConversationContextPacket } from "../src/agents/context.js";import { loadAgentDeclarations } from "../src/agents/declarations.js";import { OBSERVATION_OUTPUT_CONTRACT, outputContractIdentityJson,} from "../src/agents/output-contracts.js";import { CORRECTION_TARGET_LATEST, proposalCapabilitiesSchema } from "../src/agents/proposals.js";import { ThoughtAgentRuntime } from "../src/agents/runtime.js";import type { AgentRunner, AgentRunInput, ThoughtAgentDeclaration } from "../src/agents/types.js";import { CORRECTION_PROPOSAL_EVENT_TYPE, MEMORY_MATERIALIZATION_FAILED_EVENT_TYPE, MEMORY_MATERIALIZED_EVENT_TYPE, MEMORY_PROPOSAL_EVENT_TYPE, PROPOSAL_DECISION_EVENT_TYPE,} from "../src/agent-proposals/contracts.js";import { materializeMemoryDecision } from "../src/agent-proposals/memory-materializer.js";import { recordProposalDecision } from "../src/agent-proposals/review.js";import { FilesystemConnector } from "../src/connectors/filesystem.js";import { stableKey } from "../src/core/ids.js";import type { JsonObject } from "../src/core/json.js";import type { EventCandidate, ThoughtEvent } from "../src/events/types.js";import type { JazzThoughtStore } from "../src/jazz/store.js";import type { AgentRun, ConsumerProgress } from "../src/store/types.js";import { projectTrainingExamples, projectPrivateTrainingExamples, recordJudgment } from "../src/training/judgments.js";import { exportPrivateTrainingDataset } from "../src/training/private-training.js";import { temporaryProject, testDeclarationEnvironment, testStore } from "./helpers.js";import { FOCUS_DECLARATION_PROPOSED_EVENT_TYPE, focusDeclarationProposalPayloadSchema,} from "../src/focuses/types.js";
const stores: JazzThoughtStore[] = [];const roots: string[] = [];const telegramSource = "telegram:thoughtstream-bot-webhook";const memorySource = "filesystem:telegram-agent-context";const chatId = "fixture-chat";const senderId = "fixture-user";const baseMemory = "---\nid: stream-memory\n---\n# Stream memory\n\nExisting fact.\n";
class ProposalFixtureRunner implements AgentRunner { readonly mode = "pi" as const; constructor( private readonly invalidEvidence = false, private readonly memoryOperation: "append" | "replace-document" = "append", private readonly useLatestCorrectionTarget = false, ) {}
async run(input: AgentRunInput) { const capabilities = proposalCapabilitiesSchema.parse(input.context.manifest.proposalCapabilities); const target = capabilities.correctionTargets.at(-1); if (!capabilities.memoryTarget || !target) throw new Error("Fixture requires both proposal capabilities"); const evidence = this.invalidEvidence ? ["evt-outside-snapshot"] : [input.event.id]; return { summary: "I understand.", tags: ["conversation"], importance: "normal" as const, confidence: 0.5, model: { provider: "tinker", id: input.declaration.model! }, proposals: [ { toolCallId: "call-memory", kind: "memory-change" as const, arguments: { operation: this.memoryOperation, proposed_text: "## Response preference\n\nKeep technical replies compact.", reason: "Cameron stated a durable response preference.", evidence_event_ids: evidence, }, }, { toolCallId: "call-correction", kind: "self-correction" as const, arguments: { target_output: this.useLatestCorrectionTarget ? CORRECTION_TARGET_LATEST : target.outputEventId, replacement: "The corrected prior reply.", reason: "The prior delivered reply misstated the fact.", evidence_event_ids: evidence, }, }, ], }; }}
class FocusProposalFixtureRunner implements AgentRunner { readonly mode = "pi" as const;
constructor(private readonly enforceCapabilities = true) {}
async run(input: AgentRunInput) { const capabilities = proposalCapabilitiesSchema.parse(input.context.manifest.proposalCapabilities); if (this.enforceCapabilities && !capabilities.enabled.includes("focus-declaration")) { throw new Error("Fixture requires focus proposal capability"); } return { summary: "I saved that as a focus proposal.", tags: ["conversation"], importance: "normal" as const, confidence: 0.5, model: { provider: "tinker", id: input.declaration.model! }, proposals: [{ toolCallId: "call-focus", kind: "focus-declaration" as const, arguments: { declaration: focusDeclarationFixture() }, }], }; }}
interface ProposalFixture { root: string; contextRoot: string; store: JazzThoughtStore; declaration: ThoughtAgentDeclaration; trigger: ThoughtEvent; priorRunId: string; priorOutputEventId: string; memoryProposal: ThoughtEvent; correctionProposal: ThoughtEvent;}
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("agent-originated proposals", () => { test("settles one inert focus proposal atomically and reconstructs its native tool history", async () => { const setup = await setupConversation("/focus AI news that changes how I understand agents and training"); const runtime = new ThoughtAgentRuntime(setup.store, [new FocusProposalFixtureRunner()]); const results = await runtime.consumeBacklog([setup.declaration]); expect(results).toHaveLength(1); expect(results[0]?.error).toBeUndefined(); const run = (await setup.store.listRuns()).find((candidate) => candidate.triggerEventId === setup.trigger.id)!; const output = await setup.store.getEvent(run.outputEventIds[0]!); const focusEvent = (await setup.store.listEvents({ types: [FOCUS_DECLARATION_PROPOSED_EVENT_TYPE] }))[0]!; const focus = focusDeclarationProposalPayloadSchema.parse(focusEvent.payload); const completed = (await setup.store.listEvents({ types: ["stream.thought.agent.run.completed"] })) .find((event) => event.payload.runId === run.id)!;
expect(focusEvent).toMatchObject({ privacy: "sensitive", rootEventId: setup.trigger.rootEventId, parentEventId: output?.id, payload: { authority: { activatesFocus: false, expandsDataAccess: false, grantsExternalActions: false, startsTraining: false, promotesAdapter: false, }, }, }); expect(focus.declaration).toEqual(focusDeclarationFixture()); expect(focus.proposer).toMatchObject({ kind: "agent", runId: run.id, outputEventId: output?.id, triggerEventId: setup.trigger.id, agentId: setup.declaration.id, agentVersion: setup.declaration.version, }); expect(completed.payload.proposalEventIds).toEqual([focusEvent.id]);
await setup.store.appendEvent({ type: "stream.thought.action.telegram.send.delivered", schemaVersion: 1, source: `telegram-dispatcher:${telegramSource}:${chatId}`, sourceKind: "system", externalId: "delivery-focus", idempotencyKey: "delivery-focus", occurredAt: new Date().toISOString(), actor: `telegram-dispatcher:${telegramSource}:${chatId}`, rootEventId: setup.trigger.rootEventId, parentEventId: output!.id, correlationId: setup.trigger.correlationId, privacy: "sensitive", payload: { chatId, messageId: "out-focus", runIds: [run.id] }, }); const next = (await setup.store.appendEvent(telegramMessage( "after-focus", "/focus AI news that changes how I understand agents and training", ))).event; const packet = await buildSubscribedTelegramConversationContextPacket(setup.declaration, next, setup.store); expect(packet.messages).toEqual(expect.arrayContaining([ expect.objectContaining({ role: "assistant", toolCalls: [expect.objectContaining({ name: "propose_focus", arguments: { declaration: focusDeclarationFixture() }, })], }), expect.objectContaining({ role: "toolResult", toolName: "propose_focus", isError: false, }), ]));
const repeated = await runtime.consumeBacklog([setup.declaration]); expect(repeated).toHaveLength(1); expect(repeated[0]?.error).toBeUndefined(); const focusEvents = await setup.store.listEvents({ types: [FOCUS_DECLARATION_PROPOSED_EVENT_TYPE] }); expect(focusEvents).toHaveLength(2); expect(new Set(focusEvents.map((event) => event.id)).size).toBe(2); expect(focusEvents.map((event) => event.payload.declarationFingerprint)) .toEqual([focus.declarationFingerprint, focus.declarationFingerprint]); });
test("rejects an alternate runner focus proposal outside an exact /focus turn", async () => { const setup = await setupConversation(); const runtime = new ThoughtAgentRuntime(setup.store, [new FocusProposalFixtureRunner(false)]); const results = await runtime.consumeBacklog([setup.declaration]); const run = (await setup.store.listRuns()).find((candidate) => candidate.triggerEventId === setup.trigger.id)!;
expect(results).toHaveLength(1); expect(results[0]?.error).toBe("Agent runner failed"); expect(run.status).toBe("failed"); expect(run.outputEventIds).toEqual([]); expect(await setup.store.listEvents({ types: [FOCUS_DECLARATION_PROPOSED_EVENT_TYPE] })).toEqual([]); expect((await setup.store.listEvents({ types: ["stream.thought.agent.run.completed"] })) .filter((event) => event.payload.runId === run.id)).toEqual([]); const progress = (await setup.store.listConsumerProgress()) .find((candidate) => candidate.consumerVersion === setup.declaration.version && candidate.source === telegramSource)!; expect(progress.lastEventId).toBe(setup.trigger.id); });
test("freezes exact proposal capability evidence and settles output, proposals, lifecycle, run, and progress atomically", async () => { const fixture = await proposalFixture(); const { store, trigger, memoryProposal, correctionProposal } = fixture; const run = (await store.listRuns()).find((candidate) => candidate.triggerEventId === trigger.id)!; const output = await store.getEvent(run.outputEventIds[0]!); const completed = (await store.listEvents({ types: ["stream.thought.agent.run.completed"] })) .find((event) => event.payload.runId === run.id)!; const progress = (await store.listConsumerProgress()).find((item) => item.source === telegramSource)!;
expect(run.status).toBe("completed"); expect(run.result).not.toHaveProperty("proposals"); expect(output?.payload.structuredOutput).not.toHaveProperty("proposals"); expect([memoryProposal, correctionProposal].map((event) => event.sourceSequence)).toEqual([ output!.sourceSequence + 1, output!.sourceSequence + 2, ]); expect(completed.sourceSequence).toBe(output!.sourceSequence + 3); expect(completed.payload.proposalEventIds).toEqual([memoryProposal.id, correctionProposal.id]); expect(progress.lastEventId).toBe(trigger.id); expect(memoryProposal).toMatchObject({ type: MEMORY_PROPOSAL_EVENT_TYPE, privacy: "sensitive", parentEventId: output?.id, payload: { proposalState: "agent-proposed", target: expect.objectContaining({ source: memorySource, path: "memory.md" }), publicationEligible: false, }, }); expect(correctionProposal).toMatchObject({ type: CORRECTION_PROPOSAL_EVENT_TYPE, parentEventId: fixture.priorOutputEventId, payload: { proposalState: "agent-proposed", target: expect.objectContaining({ runId: fixture.priorRunId, outputEventId: fixture.priorOutputEventId }), qualityEligible: false, externalExportEligible: false, publicationEligible: false, }, });
const snapshotId = String(run.contextManifest.contextSnapshot && (run.contextManifest.contextSnapshot as JsonObject).id); const snapshotVersion = await store.getDocumentVersion(snapshotId); expect(snapshotVersion).toBeDefined(); const snapshot = JSON.parse(snapshotVersion!.content) as { manifest: JsonObject }; expect(snapshot.manifest.proposalCapabilities).toEqual(run.contextManifest.proposalCapabilities);
const before = (await store.listEvents()).length; const replay = await new ThoughtAgentRuntime(store, [new ProposalFixtureRunner()]) .consumeBacklog([fixture.declaration]); expect(replay).toHaveLength(0); expect(await store.listEvents()).toHaveLength(before); });
test("resolves the latest correction shorthand to exact snapshot-bound target evidence", async () => { const fixture = await proposalFixture("append", true);
expect(fixture.correctionProposal.payload.target).toMatchObject({ runId: fixture.priorRunId, outputEventId: fixture.priorOutputEventId, }); expect(JSON.stringify(fixture.correctionProposal.payload)).not.toContain(CORRECTION_TARGET_LATEST); });
test("allows exactly one concurrent human decision and rejects fabricated proposal lineage", async () => { const fixture = await proposalFixture(); const decisions = await Promise.allSettled([ recordProposalDecision(fixture.store, { proposalEventId: fixture.memoryProposal.id, disposition: "accept", submissionId: "concurrent-accept", }), recordProposalDecision(fixture.store, { proposalEventId: fixture.memoryProposal.id, disposition: "reject", submissionId: "concurrent-reject", }), ]); expect(decisions.filter((result) => result.status === "fulfilled")).toHaveLength(1); expect(decisions.filter((result) => result.status === "rejected")).toHaveLength(1); expect((await fixture.store.listEvents({ types: [PROPOSAL_DECISION_EVENT_TYPE] })) .filter((event) => event.payload.proposalEventId === fixture.memoryProposal.id)).toHaveLength(1);
const forged = (await fixture.store.appendEvent({ type: MEMORY_PROPOSAL_EVENT_TYPE, schemaVersion: 1, source: fixture.memoryProposal.source, sourceKind: "agent", externalId: "forged-proposal", idempotencyKey: "forged-proposal", occurredAt: new Date().toISOString(), actor: fixture.memoryProposal.actor, rootEventId: fixture.memoryProposal.rootEventId, parentEventId: fixture.memoryProposal.parentEventId!, correlationId: fixture.memoryProposal.correlationId, privacy: "sensitive", payload: fixture.memoryProposal.payload, })).event; await expect(recordProposalDecision(fixture.store, { proposalEventId: forged.id, disposition: "accept", submissionId: "forged-decision", })).rejects.toThrow("atomic completed-run receipt"); });
test("keeps invalid proposal settlement from leaking a partial output or proposal", async () => { const setup = await setupConversation(); const declaration = setup.declaration; const runtime = new ThoughtAgentRuntime(setup.store, [new ProposalFixtureRunner(true)]); const results = await runtime.consumeBacklog([declaration]); const run = (await setup.store.listRuns()).find((candidate) => candidate.triggerEventId === setup.trigger.id)!; const outputs = (await setup.store.listEvents()).filter((event) => event.payload.runId === run.id && ( event.type === declaration.outputEventType || event.type === MEMORY_PROPOSAL_EVENT_TYPE || event.type === CORRECTION_PROPOSAL_EVENT_TYPE ));
expect(results[0]).toMatchObject({ error: "Agent runner failed" }); expect(run.status).toBe("failed"); expect(run.outputEventIds).toEqual([]); expect(outputs).toEqual([]); expect((await setup.store.listEvents({ types: ["stream.thought.agent.run.failed"] }))) .toEqual([expect.objectContaining({ payload: expect.objectContaining({ runId: run.id }) })]); });
test("records append-only decisions, projects accepted corrections, materializes memory with scan receipts, and exports private quality data only through the private path", async () => { const fixture = await proposalFixture(); const correction = await recordProposalDecision(fixture.store, { proposalEventId: fixture.correctionProposal.id, disposition: "accept", submissionId: "submission-correction", actor: "operator:test", }); const correctionAgain = await recordProposalDecision(fixture.store, { proposalEventId: fixture.correctionProposal.id, disposition: "accept", submissionId: "submission-correction", actor: "operator:test", }); expect(correctionAgain.decision.id).toBe(correction.decision.id); expect(correction.projectedJudgment).toMatchObject({ type: "stream.thought.judgment.training-example", schemaVersion: 2, payload: { runId: fixture.priorRunId, outputEventId: fixture.priorOutputEventId, kind: "correct", criterion: "agent-self-correction", criterionVersion: 1, qualityEligible: true, externalExportEligible: false, }, }); await expect(recordProposalDecision(fixture.store, { proposalEventId: fixture.correctionProposal.id, disposition: "reject", submissionId: "different-decision", })).rejects.toThrow("already has a human decision");
expect(await projectTrainingExamples(fixture.store, { includeSensitivePrivate: true })).toEqual([]); await recordJudgment(fixture.store, { runId: fixture.priorRunId, kind: "accept", criterion: "manual-private-without-feedback-provenance", criterionVersion: 1, qualityEligible: true, externalExportEligible: false, actor: "operator:test", }); const privateExamples = await projectPrivateTrainingExamples(fixture.store); expect(privateExamples).toEqual([ expect.objectContaining({ format: "thoughtstream.private-training-example.v1", privateProvenance: expect.objectContaining({ judgmentEventId: correction.projectedJudgment!.id, runId: fixture.priorRunId, outputEventId: fixture.priorOutputEventId, feedbackSourceEventId: correction.decision.id, }), }), ]); const privateDestination = path.join(fixture.root, "private-training", "stream.jsonl"); const manifest = await exportPrivateTrainingDataset(fixture.store, privateDestination, { acknowledgeSensitivePrivate: true, }); expect(manifest).toMatchObject({ format: "thoughtstream.private-training-dataset-manifest.v1", examples: 1, exactPrivateProvenance: true, externalExportAuthorityRequired: false, }); expect((await fs.stat(privateDestination)).mode & 0o777).toBe(0o600); expect((await fs.stat(`${privateDestination}.manifest.json`)).mode & 0o777).toBe(0o600);
const memory = await recordProposalDecision(fixture.store, { proposalEventId: fixture.memoryProposal.id, disposition: "accept", submissionId: "submission-memory", actor: "operator:test", }); const materialized = await materializeMemoryDecision(fixture.store, memory.decision.id, { contextRoot: fixture.contextRoot, }); expect(materialized).toMatchObject({ status: "materialized", event: { type: MEMORY_MATERIALIZED_EVENT_TYPE, payload: { proposalEventId: fixture.memoryProposal.id, decisionEventId: memory.decision.id, result: expect.objectContaining({ filesystemEventId: expect.any(String) }), }, }, }); expect(await fs.readFile(path.join(fixture.contextRoot, "memory.md"), "utf8")).toContain("Keep technical replies compact."); expect((await fs.stat(path.join(fixture.contextRoot, "memory.md"))).mode & 0o777).toBe(0o600); const resultVersionId = String(materialized.event.payload.result && (materialized.event.payload.result as JsonObject).versionId); expect(await fixture.store.getDocumentVersion(resultVersionId)).toMatchObject({ source: memorySource, path: "memory.md", }); });
test("reclaims a verified stale materializer lock while retaining live/invalid-lock fail-closed behavior", async () => { const staleOwner = await proposalFixture(); const decision = await recordProposalDecision(staleOwner.store, { proposalEventId: staleOwner.memoryProposal.id, disposition: "accept", submissionId: "stale-lock", }); const lockPath = path.join(staleOwner.contextRoot, ".thoughtstream-memory-materializer.lock"); await fs.writeFile(lockPath, `${JSON.stringify({ bootId: "definitely-another-boot", pid: 999_999, processStart: "1", token: "stale-lock-token-123456789", createdAt: "2026-01-01T00:00:00.000Z", })}\n`, { mode: 0o600 }); const materialized = await materializeMemoryDecision(staleOwner.store, decision.decision.id, { contextRoot: staleOwner.contextRoot, }); expect(materialized.status).toBe("materialized"); await expect(fs.lstat(lockPath)).rejects.toMatchObject({ code: "ENOENT" });
const invalidOwner = await proposalFixture(); const invalidDecision = await recordProposalDecision(invalidOwner.store, { proposalEventId: invalidOwner.memoryProposal.id, disposition: "accept", submissionId: "invalid-lock", }); const invalidLockPath = path.join(invalidOwner.contextRoot, ".thoughtstream-memory-materializer.lock"); await fs.writeFile(invalidLockPath, "not valid lock evidence\n", { mode: 0o600 }); const failed = await materializeMemoryDecision(invalidOwner.store, invalidDecision.decision.id, { contextRoot: invalidOwner.contextRoot, }); expect(failed).toMatchObject({ status: "failed", event: { payload: { reasonCode: "write-failed" } } }); expect(await fs.readFile(invalidLockPath, "utf8")).toBe("not valid lock evidence\n"); });
test("fails closed on stale bases, symlink swaps, and frontmatter-removing edits without overwriting the file", async () => { const stale = await proposalFixture(); const stalePath = path.join(stale.contextRoot, "memory.md"); const concurrent = `${baseMemory.trimEnd()}\n\nConcurrent operator edit.\n`; await fs.writeFile(stalePath, concurrent); await new FilesystemConnector({ id: memorySource, root: stale.contextRoot }).scan(stale.store); const staleDecision = await recordProposalDecision(stale.store, { proposalEventId: stale.memoryProposal.id, disposition: "accept", submissionId: "stale", }); const staleResult = await materializeMemoryDecision(stale.store, staleDecision.decision.id, { contextRoot: stale.contextRoot }); expect(staleResult).toMatchObject({ status: "failed", event: { payload: { reasonCode: "stale-base", contentRedacted: true } } }); expect(await fs.readFile(stalePath, "utf8")).toBe(concurrent);
const symlink = await proposalFixture(); const symlinkPath = path.join(symlink.contextRoot, "memory.md"); const outside = path.join(symlink.root, "outside-memory.md"); await fs.writeFile(outside, "OUTSIDE SENTINEL\n"); const symlinkDecision = await recordProposalDecision(symlink.store, { proposalEventId: symlink.memoryProposal.id, disposition: "accept", submissionId: "symlink", }); const symlinkResult = await materializeMemoryDecision(symlink.store, symlinkDecision.decision.id, { contextRoot: symlink.contextRoot, beforeRename: async () => { await fs.rm(symlinkPath); await fs.symlink(outside, symlinkPath); }, }); expect(symlinkResult).toMatchObject({ status: "failed", event: { payload: { reasonCode: "symlink-refused" } } }); expect(await fs.readFile(outside, "utf8")).toBe("OUTSIDE SENTINEL\n");
const frontmatter = await proposalFixture("replace-document"); const frontmatterPath = path.join(frontmatter.contextRoot, "memory.md"); const before = await fs.readFile(frontmatterPath, "utf8"); const frontmatterDecision = await recordProposalDecision(frontmatter.store, { proposalEventId: frontmatter.memoryProposal.id, disposition: "edit", replacementText: "# No stable identity\n", submissionId: "frontmatter", }); const frontmatterResult = await materializeMemoryDecision(frontmatter.store, frontmatterDecision.decision.id, { contextRoot: frontmatter.contextRoot }); expect(frontmatterResult).toMatchObject({ status: "failed", event: { payload: { reasonCode: "frontmatter-invalid" } } }); expect(await fs.readFile(frontmatterPath, "utf8")).toBe(before); expect((await frontmatter.store.listEvents({ types: [MEMORY_MATERIALIZATION_FAILED_EVENT_TYPE] }))).toHaveLength(1); });
test("rolls back a terminal settlement when a side-effect proposal candidate is invalid", async () => { const root = await temporaryProject("thoughtstream-proposal-rollback-"); roots.push(root); const store = await testStore(root); stores.push(store); const input = (await store.appendEvent(telegramMessage("rollback", "Rollback fixture"))).event; const run = runningRun(input); await store.upsertRun(run); const output: EventCandidate = { type: "stream.thought.derived.message.observation", schemaVersion: 1, source: "agent:telegram-conversation", sourceKind: "agent", externalId: run.id, idempotencyKey: `${run.id}:output`, occurredAt: new Date().toISOString(), actor: "telegram-conversation", rootEventId: input.rootEventId, parentEventId: input.id, correlationId: input.correlationId, privacy: "sensitive", payload: { runId: run.id, summary: "Would have settled" }, }; const invalidProposal: EventCandidate = { ...output, type: MEMORY_PROPOSAL_EVENT_TYPE, externalId: `${run.id}:proposal`, idempotencyKey: `${run.id}:proposal`, payload: { invalid: true }, }; const completed: EventCandidate = { ...output, type: "stream.thought.agent.run.completed", externalId: `${run.id}:completed`, idempotencyKey: `${run.id}:completed`, payload: { runId: run.id, agentId: run.agentId, agentVersion: run.agentVersion, inputEventIds: [input.id], attempt: 1, status: "completed", }, }; const progress: ConsumerProgress = { id: stableKey("consumer-progress", run.agentId, String(run.agentVersion), input.source), consumerId: run.agentId, consumerVersion: run.agentVersion, source: input.source, lastSequence: input.sourceSequence, lastEventId: input.id, updatedAt: new Date().toISOString(), };
await expect(store.settleConsumerSuccess({ run: { ...run, status: "completed", outputEventIds: ["would-be-output"] }, inputEvent: input, output, sideEffects: [invalidProposal], completed, progress, })).rejects.toThrow(); expect((await store.getRun(run.id))?.status).toBe("running"); expect((await store.listEvents()).filter((event) => event.source === output.source)).toEqual([]); expect(await store.getConsumerProgress(progress.id)).toBeUndefined(); });});
async function proposalFixture( memoryOperation: "append" | "replace-document" = "append", useLatestCorrectionTarget = false,): Promise<ProposalFixture> { const setup = await setupConversation(); const runtime = new ThoughtAgentRuntime(setup.store, [ new ProposalFixtureRunner(false, memoryOperation, useLatestCorrectionTarget), ]); const results = await runtime.consumeBacklog([setup.declaration]); expect(results).toHaveLength(1); expect(results[0]?.error).toBeUndefined(); const proposals = (await setup.store.listEvents({ types: [MEMORY_PROPOSAL_EVENT_TYPE, CORRECTION_PROPOSAL_EVENT_TYPE], })).filter((event) => event.payload.proposer && (event.payload.proposer as JsonObject).triggerEventId === setup.trigger.id); const memoryProposal = proposals.find((event) => event.type === MEMORY_PROPOSAL_EVENT_TYPE); const correctionProposal = proposals.find((event) => event.type === CORRECTION_PROPOSAL_EVENT_TYPE); if (!memoryProposal || !correctionProposal) throw new Error("Proposal fixture did not settle both proposals"); return { ...setup, memoryProposal, correctionProposal, };}
async function setupConversation(currentText = "Please remember that and correct yourself.") { const root = await temporaryProject("thoughtstream-agent-proposals-"); roots.push(root); const store = await testStore(root); stores.push(store); const contextRoot = path.join(root, "stream-context"); await fs.mkdir(contextRoot); await fs.writeFile(path.join(contextRoot, "identity.md"), "---\nid: stream-identity\n---\n# Stream identity\n"); await fs.writeFile(path.join(contextRoot, "memory.md"), baseMemory); await new FilesystemConnector({ id: memorySource, root: contextRoot }).scan(store);
const loadedDeclaration = (await loadAgentDeclarations(path.join(process.cwd(), "agents"), testDeclarationEnvironment)) .find((candidate) => candidate.id === "telegram-conversation")!; const declaration = structuredClone(loadedDeclaration); declaration.enabled = true; delete declaration.conversationCompaction; const prior = (await store.appendEvent(telegramMessage("prior", "The prior user fact."))).event; const priorContext = await buildSubscribedTelegramConversationContextPacket(declaration, prior, store); const priorOutput = (await store.appendEvent({ type: "stream.thought.derived.message.observation", schemaVersion: 1, source: "agent:telegram-conversation", sourceKind: "agent", externalId: "run-prior", idempotencyKey: "run-prior:output", occurredAt: new Date().toISOString(), actor: "telegram-conversation", rootEventId: prior.rootEventId, parentEventId: prior.id, correlationId: prior.correlationId, privacy: "sensitive", payload: { runId: "run-prior", executionKey: "execution-prior", inputEventId: prior.id, inputSourceSequence: prior.sourceSequence, summary: "The incorrect prior reply.", tags: ["conversation"], importance: "normal", confidence: 0.5, outputContract: outputContractIdentityJson(OBSERVATION_OUTPUT_CONTRACT.identity), structuredOutput: { summary: "The incorrect prior reply.", tags: ["conversation"], importance: "normal", confidence: 0.5, }, }, traceId: "run-prior", })).event; const priorRun: AgentRun = { id: "run-prior", executionKey: "execution-prior", triggerEventId: prior.id, agentId: "telegram-conversation", agentVersion: declaration.version, status: "completed", inputEventIds: [prior.id], outputEventIds: [priorOutput.id], attempt: 1, provider: "tinker", model: "thinkingmachines/Inkling-Small", privacy: "sensitive", promptHash: "prompt-prior", contextManifest: priorContext.manifest, result: { summary: "The incorrect prior reply.", tags: ["conversation"], importance: "normal", confidence: 0.5, }, createdAt: prior.observedAt, startedAt: prior.observedAt, completedAt: prior.observedAt, updatedAt: prior.observedAt, }; await store.upsertRun(priorRun); await store.appendEvent({ type: "stream.thought.action.telegram.send.delivered", schemaVersion: 1, source: `telegram-dispatcher:${telegramSource}:${chatId}`, sourceKind: "system", externalId: "delivery-prior", idempotencyKey: "delivery-prior", occurredAt: new Date().toISOString(), actor: `telegram-dispatcher:${telegramSource}:${chatId}`, rootEventId: prior.rootEventId, parentEventId: priorOutput.id, correlationId: prior.correlationId, privacy: "sensitive", payload: { chatId, messageId: "out-prior", runIds: [priorRun.id] }, }); await store.initializeConsumerProgress({ id: stableKey("consumer-progress", declaration.id, String(declaration.version), telegramSource), consumerId: declaration.id, consumerVersion: declaration.version, source: telegramSource, lastSequence: prior.sourceSequence, lastEventId: prior.id, updatedAt: new Date().toISOString(), }); const trigger = (await store.appendEvent(telegramMessage("current", currentText))).event; return { root, contextRoot, store, declaration, trigger, priorRunId: priorRun.id, priorOutputEventId: priorOutput.id, };}
function telegramMessage(id: string, text: string): EventCandidate { return { type: "stream.thought.source.telegram.message", schemaVersion: 1, source: telegramSource, sourceKind: "telegram", externalId: id, idempotencyKey: id, occurredAt: new Date().toISOString(), actor: `telegram-user:${senderId}`, correlationId: `conversation-${chatId}`, privacy: "sensitive", payload: { accountId: "thoughtstream-bot", updateId: id, chatId, messageId: id, senderId, chatType: "direct", text, occurredAt: new Date().toISOString(), transport: "telegram-bot-api-webhook", }, };}
function runningRun(input: ThoughtEvent): AgentRun { const at = new Date().toISOString(); return { id: "run-rollback", executionKey: "execution-rollback", triggerEventId: input.id, agentId: "telegram-conversation", agentVersion: 15, status: "running", inputEventIds: [input.id], outputEventIds: [], attempt: 1, provider: "tinker", model: "thinkingmachines/Inkling-Small", privacy: "sensitive", promptHash: "prompt-rollback", contextManifest: { agentRole: "standard", outputContract: outputContractIdentityJson(OBSERVATION_OUTPUT_CONTRACT.identity), }, createdAt: at, startedAt: at, updatedAt: at, };}
function focusDeclarationFixture() { return { id: "ai-news", version: 1, name: "AI news", description: "Notice public AI news that changes Cameron's model of agents and training.", parentFocusIds: ["news"], scope: { summary: "Public AI research, product, and infrastructure developments.", includeTags: ["ai-news"], excludeTags: [], }, objective: { statement: "Produce concise evidence-grounded observations about meaningful AI developments." }, subscriptions: { eventTypes: ["stream.thought.source.rss.item", "stream.thought.source.atproto.commit"], sourcePatterns: ["rss:*", "jetstream:*"], privacy: ["public-source" as const], replay: "now" as const, }, budgets: { inference: { maxCallsPerDay: 24, maxInputTokensPerDay: 480_000, maxOutputTokensPerDay: 12_288, maxCostMicrousdPerDay: 2_400_000, }, attention: { maxDeliveriesPerDay: 6 }, }, permissions: { requestedReadTools: [], requestedExternalActions: [] }, retirementRule: "Retire or revise after 100 eligible events without accepted value.", };}