Something went wrong. Try again.
A local-first event pipeline for independent agents, built on Jazz.
Something went wrong. Try again.
10 kB · 210 lines
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211import { z } from "zod";import { contextPacketFromSnapshot, snapshotManifestMatchesRunContext } from "../agents/context.js";import { outputContractIdentityJson, parseOutputContractIdentity } from "../agents/output-contracts.js";import { proposalCapabilitiesSchema } from "../agents/proposals.js";import { canonicalJson, sha256, type JsonObject } from "../core/json.js";import type { ThoughtEvent } from "../events/types.js";import type { JazzThoughtStore } from "../jazz/store.js";import type { AgentRun } from "../store/types.js";import { CORRECTION_PROPOSAL_EVENT_TYPE, correctionProposalPayloadSchema, MEMORY_PROPOSAL_EVENT_TYPE, memoryProposalPayloadSchema,} from "./contracts.js";
export type ValidatedMemoryProposalPayload = z.infer<typeof memoryProposalPayloadSchema>;export type ValidatedCorrectionProposalPayload = z.infer<typeof correctionProposalPayloadSchema>;export type ValidatedProposalPayload = ValidatedMemoryProposalPayload | ValidatedCorrectionProposalPayload;
export interface ValidatedProposalLineage { proposal: ThoughtEvent; payload: ValidatedProposalPayload; proposerRun: AgentRun; proposerOutput: ThoughtEvent; proposerTrigger: ThoughtEvent; completion: ThoughtEvent;}
export async function validateProposalLineage( store: JazzThoughtStore, proposal: ThoughtEvent,): Promise<ValidatedProposalLineage> { if (proposal.type !== MEMORY_PROPOSAL_EVENT_TYPE && proposal.type !== CORRECTION_PROPOSAL_EVENT_TYPE) { throw new Error(`Event is not an agent proposal: ${proposal.id}`); } if (proposal.sourceKind !== "agent" || proposal.privacy !== "sensitive") { throw new Error("Agent proposal source authority or privacy is invalid"); } const payload = proposal.type === MEMORY_PROPOSAL_EVENT_TYPE ? memoryProposalPayloadSchema.parse(proposal.payload) : correctionProposalPayloadSchema.parse(proposal.payload); const proposer = payload.proposer; const [proposerRun, proposerOutput, proposerTrigger] = await Promise.all([ store.getRun(proposer.runId), store.getEvent(proposer.outputEventId), store.getEvent(proposer.triggerEventId), ]); if ( !proposerRun || proposerRun.status !== "completed" || !proposerRun.result || proposerRun.outputEventIds.length !== 1 || proposerRun.outputEventIds[0] !== proposer.outputEventId || proposerRun.triggerEventId !== proposer.triggerEventId || proposerRun.agentId !== proposer.agentId || proposerRun.agentVersion !== proposer.agentVersion || proposerRun.provider !== proposer.provider || proposerRun.model !== proposer.model || !proposerOutput || proposerOutput.type !== "stream.thought.derived.message.observation" || proposerOutput.source !== `agent:${proposer.agentId}` || proposerOutput.sourceKind !== "agent" || proposerOutput.actor !== proposer.agentId || proposerOutput.parentEventId !== proposer.triggerEventId || proposerOutput.payload.runId !== proposer.runId || proposerOutput.payload.inputEventId !== proposer.triggerEventId || !proposerTrigger || proposerOutput.rootEventId !== proposerTrigger.rootEventId || proposal.source !== proposerOutput.source || proposal.actor !== proposer.agentId ) { throw new Error("Agent proposal proposer run/output lineage is incomplete or inconsistent"); } if (proposerRun.contextManifest.declarationFingerprint !== proposer.declarationFingerprint) { throw new Error("Agent proposal declaration fingerprint does not match its completed run"); } const snapshotIdentity = objectField(proposerRun.contextManifest.contextSnapshot); if (snapshotIdentity?.id !== proposer.contextSnapshotId) { throw new Error("Agent proposal context snapshot identity does not match its completed run"); } const snapshotVersion = await store.getDocumentVersion(proposer.contextSnapshotId); if (!snapshotVersion || snapshotVersion.id !== proposer.contextSnapshotId || snapshotVersion.documentId !== proposer.contextSnapshotId || snapshotVersion.source !== `context:${proposer.agentId}` || sha256(snapshotVersion.content) !== snapshotVersion.sha256 || Buffer.byteLength(snapshotVersion.content) !== snapshotVersion.sizeBytes) { throw new Error("Agent proposal context snapshot storage evidence is missing or inconsistent"); } const snapshotPacket = contextPacketFromSnapshot(snapshotVersion.content, proposer.contextSnapshotId); if (!snapshotManifestMatchesRunContext(snapshotPacket.manifest, proposerRun.contextManifest)) { throw new Error("Agent proposal context snapshot does not match its completed run manifest"); } const capabilities = proposalCapabilitiesSchema.parse(snapshotPacket.manifest.proposalCapabilities); const admittedEvidence = new Set(capabilities.evidenceEventIds); for (const evidenceEventId of payload.evidenceEventIds) { if (!admittedEvidence.has(evidenceEventId) || !(await store.getEvent(evidenceEventId))) { throw new Error("Agent proposal cites evidence outside or missing from its frozen context snapshot"); } } const completions = (await store.listEvents({ source: `agent:${proposer.agentId}`, types: ["stream.thought.agent.run.completed"], })).filter((event) => ( event.payload.runId === proposer.runId && event.payload.outputEventId === proposer.outputEventId && Array.isArray(event.payload.proposalEventIds) && event.payload.proposalEventIds.includes(proposal.id) )); if (completions.length !== 1) { throw new Error("Agent proposal is not named by exactly one atomic completed-run receipt"); } const completion = completions[0]!; if ( completion.sourceKind !== "agent" || completion.actor !== proposer.agentId || completion.parentEventId !== proposer.triggerEventId || completion.rootEventId !== proposerTrigger.rootEventId || completion.payload.declarationFingerprint !== proposer.declarationFingerprint ) { throw new Error("Agent proposal completed-run receipt lineage is inconsistent"); }
if (proposal.type === MEMORY_PROPOSAL_EVENT_TYPE) { const memoryPayload = memoryProposalPayloadSchema.parse(proposal.payload); if (!capabilities.memoryTarget || canonicalJson(capabilities.memoryTarget as unknown as JsonObject) !== canonicalJson(memoryPayload.target as unknown as JsonObject) || proposal.parentEventId !== proposer.outputEventId || proposal.rootEventId !== proposerTrigger.rootEventId || proposal.correlationId !== proposer.runId) { throw new Error("Memory proposal target is not bound to its frozen proposer snapshot"); } } else { const correctionPayload = correctionProposalPayloadSchema.parse(proposal.payload); const target = capabilities.correctionTargets.find((candidate) => candidate.outputEventId === correctionPayload.target.outputEventId); if (!target || canonicalJson(target as unknown as JsonObject) !== canonicalJson(correctionPayload.target as unknown as JsonObject)) { throw new Error("Correction proposal target is not bound to its frozen proposer snapshot"); } await validateCorrectionTarget(store, proposerTrigger, correctionPayload, proposal); } return { proposal, payload, proposerRun, proposerOutput, proposerTrigger, completion };}
async function validateCorrectionTarget( store: JazzThoughtStore, proposerTrigger: ThoughtEvent, payload: ValidatedCorrectionProposalPayload, proposal: ThoughtEvent,): Promise<void> { const target = payload.target; const [run, output, delivery] = await Promise.all([ store.getRun(target.runId), store.getEvent(target.outputEventId), store.getEvent(target.deliveryReceiptEventId), ]); const targetTrigger = run ? await store.getEvent(run.triggerEventId) : undefined; const chatId = stringField(proposerTrigger.payload.chatId); const senderId = stringField(proposerTrigger.payload.senderId); const contract = run ? outputContractIdentityJson(parseOutputContractIdentity(run.contextManifest.outputContract)) : undefined; const outputContract = output ? objectField(output.payload.outputContract) : undefined; if ( !run || run.status !== "completed" || !run.result || run.outputEventIds.length !== 1 || run.outputEventIds[0] !== target.outputEventId || !targetTrigger || targetTrigger.source !== proposerTrigger.source || targetTrigger.payload.chatId !== chatId || targetTrigger.payload.senderId !== senderId || !output || output.type !== "stream.thought.derived.message.observation" || output.source !== `agent:${run.agentId}` || output.sourceKind !== "agent" || output.actor !== run.agentId || output.parentEventId !== targetTrigger.id || output.rootEventId !== target.sourceRootEventId || output.payload.runId !== run.id || output.payload.inputEventId !== targetTrigger.id || !delivery || delivery.type !== "stream.thought.action.telegram.send.delivered" || delivery.source !== `telegram-dispatcher:${proposerTrigger.source}:${chatId}` || delivery.sourceKind !== "system" || delivery.actor !== delivery.source || delivery.parentEventId !== output.id || delivery.rootEventId !== target.sourceRootEventId || delivery.payload.chatId !== chatId || !Array.isArray(delivery.payload.runIds) || delivery.payload.runIds.length !== 1 || delivery.payload.runIds[0] !== run.id || proposal.parentEventId !== output.id || proposal.rootEventId !== target.sourceRootEventId || proposal.correlationId !== run.id || !contract || !outputContract || canonicalJson(contract) !== canonicalJson(target.outputContract as unknown as JsonObject) || canonicalJson(outputContract) !== canonicalJson(target.outputContract as unknown as JsonObject) ) { throw new Error("Correction proposal target delivery/run/output lineage is incomplete or inconsistent"); }}
function objectField(value: unknown): JsonObject | undefined { return value && typeof value === "object" && !Array.isArray(value) ? value as JsonObject : undefined;}
function stringField(value: unknown): string | undefined { return typeof value === "string" && value.length > 0 ? value : undefined;}