Something went wrong. Try again.
A local-first event pipeline for independent agents, built on Jazz.
Something went wrong. Try again.
36 kB · 880 lines
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881import path from "node:path";import { createHash } from "node:crypto";import { contextPacketFromSnapshot, snapshotManifestMatchesRunContext } from "../agents/context.js";import { createOutputContractRegistry, REVIEW_RESPONSE_OUTPUT_CONTRACT, outputContractIdentityJson, parseOutputContractIdentity, canonicalStructuredOutput, type OutputContractIdentity,} from "../agents/output-contracts.js";import { CORRECTION_PROPOSAL_EVENT_TYPE } from "../agents/repairs.js";import { canonicalJson, type JsonObject } from "../core/json.js";import { stableKey } from "../core/ids.js";import type { ThoughtEvent } from "../events/types.js";import type { JazzThoughtStore } from "../jazz/store.js";import { activeJudgments, rebuildEffectiveOutputForRun } from "../projections/effective-output.js";import type { AgentRun, TraceChunk } from "../store/types.js";import { assertPrivateDestination, openPrivateDirectory } from "../security/private-files.js";import { joinPrivacy, runPrivacy } from "../security/privacy.js";import { activeReviewTrainingRecords } from "../review/review.js";
export type JudgmentKind = "accept" | "reject" | "correct" | "prefer";
export interface RecordJudgmentInput { runId: string; kind: JudgmentKind; criterion: string; criterionVersion: number; qualityEligible: boolean; externalExportEligible: boolean; sensitiveExternalExportAuthorized?: boolean | undefined; comparedRunId?: string | undefined; replacementOutput?: JsonObject | undefined; notes?: string | undefined; actor?: string | undefined; source?: string | undefined; feedbackSourceEventId?: string | undefined; deliveryReceiptEventId?: string | undefined; supersedesJudgmentEventId?: string | undefined;}
export interface RetractJudgmentInput { runId: string; criterion: string; criterionVersion: number; retractedJudgmentEventId: string; feedbackSourceEventId: string; deliveryReceiptEventId: string; actor?: string | undefined; source?: string | undefined;}
export interface TrainingTraceReference { sequence: number; type: string;}
export interface TrainingRunProvenance { agentVersion: number; provider: string; model: string; checkpointRevision?: string | undefined; executionAdapterRevision?: string | undefined; modelAdapter?: JsonObject | undefined; adapterCatalogDigest?: string | undefined; adapterCatalogGeneration?: number | undefined; promptHash: string; outputContract: JsonObject; repair?: true | undefined;}
export interface TrainingExample { format: "thoughtstream.training-example.v3" | "thoughtstream.training-example.v4" | "thoughtstream.private-training-example.v1"; kind: JudgmentKind; judgment: { criterion: string; criterionVersion: number; preferenceStrength?: "slight" | "strong" | undefined; confidence?: "low" | "medium" | "high" | undefined; reasonCodes?: string[] | undefined; responseTags?: string[] | undefined; }; input: { event: { type: string; schemaVersion: number; sourceKind: ThoughtEvent["sourceKind"]; privacy: ThoughtEvent["privacy"]; }; contextManifest: JsonObject; prompt?: string | undefined; evidence?: string | undefined; }; trajectory: TrainingTraceReference[]; comparedTrajectory?: TrainingTraceReference[] | undefined; chosen?: JsonObject | undefined; rejected?: JsonObject | undefined; additionalRejected?: JsonObject[] | undefined; provenance: TrainingRunProvenance; originalProvenance?: TrainingRunProvenance | undefined; comparedProvenance?: TrainingRunProvenance | undefined; review?: { campaignId: string; campaignVersion: number; disposition: "prefer" | "correct"; candidateCount: 2; } | undefined; privateProvenance?: { judgmentEventId: string; runId: string; outputEventId: string; triggerEventId: string; feedbackSourceEventId: string; deliveryReceiptEventId: string; contextSnapshotId: string; } | undefined;}
export interface TrainingProjectionOptions { includeSensitivePrivate?: boolean | undefined; includeRestrictedModelAdapters?: boolean | undefined;}
export interface TrainingWriteOptions { authorizeSensitivePrivateExport?: boolean | undefined; publicContentRoots?: string[] | undefined; beforeFinalize?: (() => void | Promise<void>) | undefined;}
export interface TrainingDatasetManifest { format: "thoughtstream.training-dataset-manifest.v4"; datasetId: string; generatedAt: string; dataFile: string; sha256: string; examples: number; kinds: Partial<Record<JudgmentKind, number>>; models: string[]; exampleFormats: Record<string, number>; reviewCampaigns: string[];}
export async function recordJudgment(store: JazzThoughtStore, input: RecordJudgmentInput): Promise<ThoughtEvent> { validateInput(input); const run = await requireCompletedRun(store, input.runId); const outputEventId = requireOutputEventId(run); const outputEvent = await requireEvent(store, outputEventId); const sourceEvent = await requireEvent(store, run.triggerEventId); const replacementOutput = input.kind === "correct" ? validateReplacementOutput(run, outputEvent, input.replacementOutput!) : undefined; let comparedRun: AgentRun | undefined; let comparedOutputEventId: string | undefined; let comparedOutputEvent: ThoughtEvent | undefined; if (input.kind === "prefer") { comparedRun = await requireCompletedRun(store, input.comparedRunId!); if (comparedRun.triggerEventId !== run.triggerEventId) { throw new Error("Preference judgments require runs over the same trigger event"); } comparedOutputEventId = requireOutputEventId(comparedRun); comparedOutputEvent = await requireEvent(store, comparedOutputEventId); } if ( input.externalExportEligible && joinPrivacy( runPrivacy(run), sourceEvent.privacy, outputEvent.privacy, comparedRun ? runPrivacy(comparedRun) : undefined, comparedOutputEvent?.privacy, ) !== "public-source" && input.sensitiveExternalExportAuthorized !== true ) { throw new Error("Sensitive/private external export eligibility requires explicit authorization"); } const feedbackSourceEvent = input.feedbackSourceEventId ? await requireEvent(store, input.feedbackSourceEventId) : undefined; if (feedbackSourceEvent && feedbackSourceEvent.rootEventId !== sourceEvent.rootEventId) { throw new Error("Judgment feedback source must share the run source root"); } const deliveryReceipt = input.deliveryReceiptEventId ? await requireDeliveryReceipt(store, input.deliveryReceiptEventId, run, sourceEvent, outputEventId) : undefined; const supersededJudgment = input.supersedesJudgmentEventId ? await requireSupersededJudgment(store, input.supersedesJudgmentEventId, run.id, input.criterion, input.criterionVersion) : undefined; const payload: JsonObject = { runId: run.id, outputEventId, kind: input.kind, criterion: input.criterion, criterionVersion: input.criterionVersion, qualityEligible: input.qualityEligible, externalExportEligible: input.externalExportEligible, ...(comparedRun ? { comparedRunId: comparedRun.id } : {}), ...(comparedOutputEventId ? { comparedOutputEventId } : {}), ...(replacementOutput ? { replacementOutput } : {}), ...(input.notes ? { notes: input.notes } : {}), ...(feedbackSourceEvent ? { feedbackSourceEventId: feedbackSourceEvent.id } : {}), ...(input.deliveryReceiptEventId ? { deliveryReceiptEventId: input.deliveryReceiptEventId } : {}), ...(input.supersedesJudgmentEventId ? { supersedesJudgmentEventId: input.supersedesJudgmentEventId } : {}), }; const idempotencyKey = feedbackSourceEvent ? stableKey("judgment-feedback", feedbackSourceEvent.id, run.id, input.criterion, String(input.criterionVersion)) : stableKey( "judgment", run.id, input.kind, input.criterion, String(input.criterionVersion), comparedRun?.id ?? "none", ); const result = await store.appendEvent({ type: "stream.thought.judgment.training-example", schemaVersion: 2, source: input.source ?? "judgment:local", sourceKind: "system", externalId: idempotencyKey, idempotencyKey, occurredAt: feedbackSourceEvent?.occurredAt ?? new Date().toISOString(), actor: input.actor ?? "operator:local", rootEventId: sourceEvent.rootEventId, parentEventId: feedbackSourceEvent?.id ?? outputEventId, correlationId: run.id, privacy: joinPrivacy( runPrivacy(run), sourceEvent.privacy, outputEvent.privacy, comparedRun ? runPrivacy(comparedRun) : undefined, comparedOutputEvent?.privacy, feedbackSourceEvent?.privacy, deliveryReceipt?.privacy, supersededJudgment?.privacy, ), payload, createdByRuntime: "thoughtstream-judgment-v2", }); await rebuildAfterAuthorityChange(store, run); return result.event;}
export async function retractJudgment(store: JazzThoughtStore, input: RetractJudgmentInput): Promise<ThoughtEvent> { if (!input.runId || !input.criterion || !input.retractedJudgmentEventId || !input.feedbackSourceEventId || !input.deliveryReceiptEventId) { throw new Error("Judgment retraction requires run, criterion, target judgment, feedback source, and delivery receipt ids"); } if (!Number.isSafeInteger(input.criterionVersion) || input.criterionVersion <= 0) { throw new Error("Judgment criterion version must be a positive integer"); } const run = await requireCompletedRun(store, input.runId); const outputEventId = requireOutputEventId(run); const outputEvent = await requireEvent(store, outputEventId); const sourceEvent = await requireEvent(store, run.triggerEventId); const feedbackSourceEvent = await requireEvent(store, input.feedbackSourceEventId); if (feedbackSourceEvent.rootEventId !== sourceEvent.rootEventId) { throw new Error("Judgment feedback source must share the run source root"); } const deliveryReceipt = await requireDeliveryReceipt(store, input.deliveryReceiptEventId, run, sourceEvent, outputEventId); const retractedJudgment = await requireSupersededJudgment( store, input.retractedJudgmentEventId, run.id, input.criterion, input.criterionVersion, ); const idempotencyKey = stableKey("judgment-retraction", feedbackSourceEvent.id, input.retractedJudgmentEventId); const result = await store.appendEvent({ type: "stream.thought.judgment.training-example.retracted", schemaVersion: 1, source: input.source ?? "judgment:local", sourceKind: "system", externalId: idempotencyKey, idempotencyKey, occurredAt: feedbackSourceEvent.occurredAt, actor: input.actor ?? "operator:local", rootEventId: sourceEvent.rootEventId, parentEventId: feedbackSourceEvent.id, correlationId: run.id, privacy: joinPrivacy( runPrivacy(run), sourceEvent.privacy, outputEvent.privacy, feedbackSourceEvent.privacy, deliveryReceipt.privacy, retractedJudgment.privacy, ), payload: { runId: run.id, outputEventId, criterion: input.criterion, criterionVersion: input.criterionVersion, retractedJudgmentEventId: input.retractedJudgmentEventId, feedbackSourceEventId: feedbackSourceEvent.id, deliveryReceiptEventId: input.deliveryReceiptEventId, }, createdByRuntime: "thoughtstream-judgment-v1", }); await rebuildAfterAuthorityChange(store, run); return result.event;}
export async function projectTrainingExamples( store: JazzThoughtStore, options: TrainingProjectionOptions = {},): Promise<TrainingExample[]> { return projectTrainingExamplesByAuthority(store, options, "external");}
export async function projectPrivateTrainingExamples( store: JazzThoughtStore, options: Omit<TrainingProjectionOptions, "includeSensitivePrivate"> = {},): Promise<TrainingExample[]> { return projectTrainingExamplesByAuthority(store, { ...options, includeSensitivePrivate: true }, "private-quality");}
async function projectTrainingExamplesByAuthority( store: JazzThoughtStore, options: TrainingProjectionOptions, authority: "external" | "private-quality",): Promise<TrainingExample[]> { const judgmentSet = await activeJudgments(store); const examples: TrainingExample[] = []; for (const judgment of judgmentSet.active) { if (authority === "external" ? !hasExternalExportAuthority(judgment) : !hasPrivateQualityAuthority(judgment)) continue; const run = await requireCompletedRun(store, stringField(judgment.payload.runId, "runId")); if (!modelAdapterExportAllowed(run, options)) continue; const outputEvent = await requireEvent(store, requireOutputEventId(run)); let contract: OutputContractIdentity; try { contract = outputContractForRun(run, outputEvent); } catch { continue; } const isRepair = run.contextManifest.agentRole === "repair" || outputEvent.type === CORRECTION_PROPOSAL_EVENT_TYPE; const kind = judgment.payload.kind as JudgmentKind; if (isRepair && kind !== "accept" && kind !== "correct") continue; const runOutput = validatedEventOutput(outputEvent, contract); const replacement = objectField(judgment.payload.replacementOutput); let chosen: JsonObject | undefined; let rejected: JsonObject | undefined; let comparedRun: AgentRun | undefined; let comparedEvent: ThoughtEvent | undefined; let originalRun: AgentRun | undefined; if (kind === "accept") chosen = runOutput; if (kind === "reject") rejected = runOutput; if (kind === "correct") { chosen = replacement ? canonicalStructuredOutput(createOutputContractRegistry(), contract, replacement) : undefined; rejected = runOutput; } if (kind === "prefer") { comparedRun = await requireCompletedRun(store, stringField(judgment.payload.comparedRunId, "comparedRunId")); if (!modelAdapterExportAllowed(comparedRun, options)) continue; comparedEvent = await requireEvent(store, requireOutputEventId(comparedRun)); chosen = runOutput; rejected = validatedEventOutput(comparedEvent, outputContractForRun(comparedRun, comparedEvent)); } if (!chosen && !rejected) continue;
let inputEvent = await requireEvent(store, run.triggerEventId); if (isRepair) { const originalRunId = stringField(outputEvent.payload.originalRunId, "originalRunId"); originalRun = await requireRun(store, originalRunId); if (!modelAdapterExportAllowed(originalRun, options)) continue; const originalTriggerEventId = stringField(outputEvent.payload.originalTriggerEventId, "originalTriggerEventId"); if (originalRun.triggerEventId !== originalTriggerEventId) continue; inputEvent = await requireEvent(store, originalTriggerEventId); } const combinedPrivacy = joinPrivacy( judgment.privacy, inputEvent.privacy, outputEvent.privacy, runPrivacy(run), originalRun ? runPrivacy(originalRun) : undefined, comparedEvent?.privacy, comparedRun ? runPrivacy(comparedRun) : undefined, ); if (!options.includeSensitivePrivate && combinedPrivacy !== "public-source") continue; const privateProvenance = authority === "private-quality" ? await exactPrivateProvenance(store, judgment, run, outputEvent, inputEvent) : undefined; if (authority === "private-quality" && !privateProvenance) continue;
const trace = await store.listTrace(run.id); examples.push({ format: authority === "external" ? "thoughtstream.training-example.v3" : "thoughtstream.private-training-example.v1", kind, judgment: { criterion: stringField(judgment.payload.criterion, "criterion"), criterionVersion: numberField(judgment.payload.criterionVersion, "criterionVersion"), }, input: { event: trainingEventReference(inputEvent, combinedPrivacy), contextManifest: sanitizeContextManifest(run.contextManifest), }, trajectory: (isRepair ? trace.filter(isAllowlistedRepairTrace) : trace).map(redactTrainingTrace), ...(comparedRun ? { comparedTrajectory: (await store.listTrace(comparedRun.id)).map(redactTrainingTrace) } : {}), ...(chosen ? { chosen } : {}), ...(rejected ? { rejected } : {}), provenance: trainingRunProvenance(run, contract, isRepair), ...(originalRun ? { originalProvenance: trainingRunProvenance( originalRun, parseOutputContractIdentity(originalRun.contextManifest.outputContract), false, ), } : {}), ...(comparedRun && comparedEvent ? { comparedProvenance: trainingRunProvenance( comparedRun, outputContractForRun(comparedRun, comparedEvent), false, ), } : {}), ...(privateProvenance ? { privateProvenance } : {}), }); } if (authority === "external") for (const record of await activeReviewTrainingRecords(store)) { const promptPayload = record.prompt.payload; const campaign = objectField(promptPayload.campaign); const criterion = objectField(promptPayload.criterion); if (!campaign || !criterion) continue; const disposition = record.decision.payload.disposition; if (disposition !== "prefer" && disposition !== "correct") continue; if (record.runs.some((run) => !modelAdapterExportAllowed(run, options))) continue; const candidateOutputs = record.outputs.map((event) => canonicalStructuredOutput( createOutputContractRegistry(), REVIEW_RESPONSE_OUTPUT_CONTRACT.identity, event.payload.structuredOutput, )) as [JsonObject, JsonObject]; let primaryIndex = 0; let chosen: JsonObject; let rejected: JsonObject; let additionalRejected: JsonObject[] | undefined; if (disposition === "prefer") { const preferredRunId = stringField(record.decision.payload.preferredRunId, "preferredRunId"); primaryIndex = record.runs.findIndex((run) => run.id === preferredRunId); if (primaryIndex < 0) continue; chosen = candidateOutputs[primaryIndex]!; rejected = candidateOutputs[primaryIndex === 0 ? 1 : 0]!; } else { chosen = canonicalStructuredOutput( createOutputContractRegistry(), REVIEW_RESPONSE_OUTPUT_CONTRACT.identity, { response: stringField(record.decision.payload.replacementResponse, "replacementResponse") }, ); rejected = candidateOutputs[0]; additionalRejected = [candidateOutputs[1]]; } const comparedIndex = primaryIndex === 0 ? 1 : 0; const primaryRun = record.runs[primaryIndex]!; const comparedRun = record.runs[comparedIndex]!; const primaryEvent = record.outputs[primaryIndex]!; const comparedEvent = record.outputs[comparedIndex]!; const combinedPrivacy = joinPrivacy( record.prompt.privacy, record.item.privacy, record.decision.privacy, runPrivacy(primaryRun), runPrivacy(comparedRun), primaryEvent.privacy, comparedEvent.privacy, ); if (!options.includeSensitivePrivate && combinedPrivacy !== "public-source") continue; examples.push({ format: "thoughtstream.training-example.v4", kind: disposition, judgment: { criterion: stringField(criterion.id, "criterion.id"), criterionVersion: numberField(criterion.version, "criterion.version"), ...(record.decision.payload.preferenceStrength ? { preferenceStrength: record.decision.payload.preferenceStrength as "slight" | "strong", } : {}), ...(record.decision.payload.confidence ? { confidence: record.decision.payload.confidence as "low" | "medium" | "high", } : {}), reasonCodes: stringArray(record.decision.payload.reasonCodes), responseTags: stringArray(record.decision.payload.responseTags), }, input: { event: trainingEventReference(record.prompt, combinedPrivacy), contextManifest: sanitizeContextManifest(primaryRun.contextManifest), prompt: stringField(promptPayload.prompt, "prompt"), ...(typeof promptPayload.evidence === "string" ? { evidence: promptPayload.evidence } : {}), }, // Review labels bind prompt/response pairs, not mutable post-run trace projections. trajectory: [], comparedTrajectory: [], chosen, rejected, ...(additionalRejected ? { additionalRejected } : {}), provenance: trainingRunProvenance(primaryRun, outputContractForRun(primaryRun, primaryEvent), false), comparedProvenance: trainingRunProvenance(comparedRun, outputContractForRun(comparedRun, comparedEvent), false), review: { campaignId: stringField(campaign.id, "campaign.id"), campaignVersion: numberField(campaign.version, "campaign.version"), disposition, candidateCount: 2, }, }); } return examples;}
async function exactPrivateProvenance( store: JazzThoughtStore, judgment: ThoughtEvent, run: AgentRun, outputEvent: ThoughtEvent, triggerEvent: ThoughtEvent,): Promise<NonNullable<TrainingExample["privateProvenance"]> | undefined> { const feedbackSourceEventId = typeof judgment.payload.feedbackSourceEventId === "string" ? judgment.payload.feedbackSourceEventId : undefined; const deliveryReceiptEventId = typeof judgment.payload.deliveryReceiptEventId === "string" ? judgment.payload.deliveryReceiptEventId : undefined; const contextSnapshotId = typeof objectField(run.contextManifest.contextSnapshot)?.id === "string" ? String(objectField(run.contextManifest.contextSnapshot)!.id) : undefined; if (!feedbackSourceEventId || !deliveryReceiptEventId || !contextSnapshotId) return undefined; const [feedback, delivery, snapshot] = await Promise.all([ store.getEvent(feedbackSourceEventId), store.getEvent(deliveryReceiptEventId), store.getDocumentVersion(contextSnapshotId), ]); if ( !feedback || feedback.rootEventId !== triggerEvent.rootEventId || !delivery || delivery.type !== "stream.thought.action.telegram.send.delivered" || delivery.sourceKind !== "system" || delivery.actor !== delivery.source || delivery.parentEventId !== outputEvent.id || delivery.rootEventId !== triggerEvent.rootEventId || !Array.isArray(delivery.payload.runIds) || delivery.payload.runIds.length !== 1 || delivery.payload.runIds[0] !== run.id || judgment.rootEventId !== triggerEvent.rootEventId || judgment.payload.runId !== run.id || judgment.payload.outputEventId !== outputEvent.id || run.triggerEventId !== triggerEvent.id || run.outputEventIds.length !== 1 || run.outputEventIds[0] !== outputEvent.id || !snapshot || snapshot.id !== contextSnapshotId || snapshot.documentId !== contextSnapshotId || snapshot.source !== `context:${run.agentId}` || createHash("sha256").update(snapshot.content).digest("hex") !== snapshot.sha256 || Buffer.byteLength(snapshot.content) !== snapshot.sizeBytes ) return undefined; try { const packet = contextPacketFromSnapshot(snapshot.content, contextSnapshotId); if (!snapshotManifestMatchesRunContext(packet.manifest, run.contextManifest)) return undefined; } catch { return undefined; } return { judgmentEventId: judgment.id, runId: run.id, outputEventId: outputEvent.id, triggerEventId: triggerEvent.id, feedbackSourceEventId, deliveryReceiptEventId, contextSnapshotId, };}
export async function writeTrainingJsonl( destination: string, examples: TrainingExample[], options: TrainingWriteOptions = {},): Promise<TrainingDatasetManifest> { const absolute = path.resolve(destination); const manifestPath = `${absolute}.manifest.json`; const containsSensitivePrivate = examples.some((example) => example.input.event.privacy !== "public-source"); if (containsSensitivePrivate) { if (options.authorizeSensitivePrivateExport !== true) { throw new Error("Sensitive/private training export requires explicit authorization"); } await assertPrivateDestination(absolute, { publicContentRoots: options.publicContentRoots }); await assertPrivateDestination(manifestPath, { publicContentRoots: options.publicContentRoots }); }
const content = examples.map((example) => canonicalJson(example as unknown as JsonObject)).join("\n"); const serialized = content ? `${content}\n` : ""; const sha256 = createHash("sha256").update(serialized).digest("hex"); const manifest: TrainingDatasetManifest = { format: "thoughtstream.training-dataset-manifest.v4", datasetId: `sha256:${sha256}`, generatedAt: new Date().toISOString(), dataFile: path.basename(absolute), sha256, examples: examples.length, kinds: examples.reduce<Partial<Record<JudgmentKind, number>>>((counts, example) => { counts[example.kind] = (counts[example.kind] ?? 0) + 1; return counts; }, {}), models: [...new Set(examples.flatMap((example) => [ trainingModelLabel(example.provenance), ...(example.originalProvenance ? [trainingModelLabel(example.originalProvenance)] : []), ...(example.comparedProvenance ? [trainingModelLabel(example.comparedProvenance)] : []), ]))].sort(), exampleFormats: examples.reduce<Record<string, number>>((counts, example) => { counts[example.format] = (counts[example.format] ?? 0) + 1; return counts; }, {}), reviewCampaigns: [...new Set(examples.flatMap((example) => example.review ? [`${example.review.campaignId}@${example.review.campaignVersion}`] : []))].sort(), }; const privateDirectory = await openPrivateDirectory(path.dirname(absolute), { publicContentRoots: options.publicContentRoots, beforeFinalize: options.beforeFinalize, }); try { await privateDirectory.write(path.basename(absolute), serialized); await privateDirectory.write(path.basename(manifestPath), `${canonicalJson(manifest as unknown as JsonObject)}\n`); } finally { await privateDirectory.close(); } return manifest;}
function modelAdapterExportAllowed(run: AgentRun, options: TrainingProjectionOptions): boolean { if (run.modelAdapter?.exportClass === "forbidden") return false; if (run.modelAdapter?.exportClass === "restricted" && options.includeRestrictedModelAdapters !== true) return false; return true;}
function trainingRunProvenance( run: AgentRun, contract: OutputContractIdentity, repair: boolean,): TrainingRunProvenance { return { agentVersion: run.agentVersion, provider: run.provider, model: run.model, ...(run.checkpointRevision ? { checkpointRevision: run.checkpointRevision } : {}), ...(run.executionAdapterRevision ? { executionAdapterRevision: run.executionAdapterRevision } : {}), ...(run.modelAdapter ? { modelAdapter: JSON.parse(JSON.stringify(run.modelAdapter)) as JsonObject } : {}), ...(run.adapterCatalogDigest ? { adapterCatalogDigest: run.adapterCatalogDigest } : {}), ...(run.adapterCatalogGeneration ? { adapterCatalogGeneration: run.adapterCatalogGeneration } : {}), promptHash: run.promptHash, outputContract: outputContractIdentityJson(contract), ...(repair ? { repair: true as const } : {}), };}
function trainingModelLabel(provenance: TrainingRunProvenance): string { const catalog = provenance.adapterCatalogDigest ? `@catalog:${provenance.adapterCatalogDigest}${provenance.adapterCatalogGeneration ? `:g${provenance.adapterCatalogGeneration}` : ""}` : ""; return `${provenance.provider}:${provenance.model}${provenance.checkpointRevision ? `#${provenance.checkpointRevision}` : ""}${provenance.executionAdapterRevision ? `@execution:${provenance.executionAdapterRevision}` : ""}${adapterTrainingLabel(provenance.modelAdapter)}${catalog}`;}
function adapterTrainingLabel(value: JsonObject | undefined): string { if (!value) return ""; const id = typeof value.id === "string" ? value.id : "unknown"; const version = typeof value.version === "number" ? value.version : 0; const manifestSha256 = typeof value.manifestSha256 === "string" ? value.manifestSha256 : "unknown"; const binding = value.binding; const checkpointReferenceSha256 = binding && typeof binding === "object" && !Array.isArray(binding) && typeof binding.checkpointReferenceSha256 === "string" ? binding.checkpointReferenceSha256 : "unknown"; return `@model-adapter:${id}@${version}#${manifestSha256}#${checkpointReferenceSha256}`;}
function trainingEventReference( event: ThoughtEvent, privacy: ThoughtEvent["privacy"] = event.privacy,): TrainingExample["input"]["event"] { return { type: event.type, schemaVersion: event.schemaVersion, sourceKind: event.sourceKind, privacy, };}
function redactTrainingTrace(trace: TraceChunk): TrainingTraceReference { return { sequence: trace.sequence, type: trace.type };}
function sanitizeContextManifest(manifest: JsonObject): JsonObject { const sanitized: JsonObject = {}; for (const key of [ "contextStrategy", "maxEvents", "maxChars", "truncated", "truncationReason", "agentVersion", "agentRole", "outputContract", "executionAdapterRevision", "modelAdapter", "adapterCatalogDigest", "adapterCatalogGeneration", "tools", "externalActions", ]) { if (manifest[key] !== undefined) sanitized[key] = manifest[key]!; } return sanitized;}
function hasPrivateQualityAuthority(judgment: ThoughtEvent): boolean { return judgment.schemaVersion >= 2 && judgment.payload.qualityEligible === true;}
function hasExternalExportAuthority(judgment: ThoughtEvent): boolean { if (judgment.schemaVersion >= 2) { return judgment.payload.qualityEligible === true && judgment.payload.externalExportEligible === true; } return judgment.privacy === "public-source" && judgment.payload.exportEligible === true;}
const allowlistedRepairTraceTypes = new Set([ "prompt", "sandbox.provider.request", "sandbox.provider.response", "sandbox.provider.failed", "agent_start", "agent_end", "turn_start", "turn_end", "message_start", "message_end",]);
function isAllowlistedRepairTrace(trace: TraceChunk): boolean { return allowlistedRepairTraceTypes.has(trace.type);}
function validatedEventOutput(event: ThoughtEvent, contract: OutputContractIdentity): JsonObject { const candidate = objectField(event.payload.structuredOutput); if (!candidate) throw new Error(`Output event has no canonical structured output: ${event.id}`); return canonicalStructuredOutput(createOutputContractRegistry(), contract, candidate);}
async function rebuildAfterAuthorityChange(store: JazzThoughtStore, run: AgentRun): Promise<void> { if (run.contextManifest.agentRole === "repair") { await rebuildEffectiveOutputForRun(store, run.id); return; } try { await rebuildEffectiveOutputForRun(store, run.id); } catch { // Historical pre-contract runs remain judgment-compatible but do not gain an effective-output projection. }}
function validateReplacementOutput(run: AgentRun, outputEvent: ThoughtEvent, replacement: JsonObject): JsonObject { const contract = outputContractForRun(run, outputEvent); return canonicalStructuredOutput(createOutputContractRegistry(), contract, replacement);}
function outputContractForRun(run: AgentRun, outputEvent: ThoughtEvent): OutputContractIdentity { const fromOutput = outputEvent.payload.outputContract; const identity = parseOutputContractIdentity(fromOutput ?? run.contextManifest.outputContract); const canonical = outputContractIdentityJson(identity); if (fromOutput && canonicalJson(canonical) !== canonicalJson(outputContractIdentityJson(parseOutputContractIdentity(run.contextManifest.outputContract)))) { throw new Error(`Run ${run.id} output contract does not match its context manifest`); } return identity;}
function validateInput(input: RecordJudgmentInput): void { if (!input.runId || !input.criterion) throw new Error("Judgments require a run id and criterion"); if (typeof input.qualityEligible !== "boolean" || typeof input.externalExportEligible !== "boolean") { throw new Error("Judgments require separate quality and external-export eligibility booleans"); } if (input.externalExportEligible && !input.qualityEligible) { throw new Error("External export eligibility requires quality eligibility"); } if (!Number.isSafeInteger(input.criterionVersion) || input.criterionVersion <= 0) { throw new Error("Judgment criterion version must be a positive integer"); } if (input.kind === "correct" && !input.replacementOutput) throw new Error("Correction judgments require replacement output"); if (input.kind === "prefer" && !input.comparedRunId) throw new Error("Preference judgments require --compared-run");}
async function requireCompletedRun(store: JazzThoughtStore, id: string): Promise<AgentRun> { const run = await store.getRun(id); if (!run) throw new Error(`Run not found: ${id}`); if (run.status !== "completed" || !run.result) throw new Error(`Run is not a completed training candidate: ${id}`); return run;}
async function requireRun(store: JazzThoughtStore, id: string): Promise<AgentRun> { const run = await store.getRun(id); if (!run) throw new Error(`Run not found: ${id}`); return run;}
async function requireEvent(store: JazzThoughtStore, id: string): Promise<ThoughtEvent> { const event = await store.getEvent(id); if (!event) throw new Error(`Event not found: ${id}`); return event;}
async function requireDeliveryReceipt( store: JazzThoughtStore, id: string, run: AgentRun, sourceEvent: ThoughtEvent, outputEventId: string,): Promise<ThoughtEvent> { const receipt = await requireEvent(store, id); const chatId = typeof sourceEvent.payload.chatId === "string" ? sourceEvent.payload.chatId : undefined; const runIds = Array.isArray(receipt.payload.runIds) ? receipt.payload.runIds : []; const chatBindingValid = chatId === undefined || ( receipt.source === `telegram-dispatcher:${sourceEvent.source}:${chatId}` && receipt.payload.chatId === chatId ); if ( receipt.type !== "stream.thought.action.telegram.send.delivered" || !chatBindingValid || receipt.sourceKind !== "system" || receipt.actor !== receipt.source || receipt.parentEventId !== outputEventId || receipt.rootEventId !== sourceEvent.rootEventId || runIds.length !== 1 || runIds[0] !== run.id ) { throw new Error(`Telegram delivery receipt does not exactly bind the judged run output: ${id}`); } return receipt;}
async function requireSupersededJudgment( store: JazzThoughtStore, id: string, runId: string, criterion: string, criterionVersion: number,): Promise<ThoughtEvent> { const judgment = await requireEvent(store, id); if (judgment.type !== "stream.thought.judgment.training-example") { throw new Error(`Event is not a training judgment: ${id}`); } if ( judgment.payload.runId !== runId || judgment.payload.criterion !== criterion || judgment.payload.criterionVersion !== criterionVersion ) { throw new Error("Superseded judgment must address the same run and criterion"); } return judgment;}
function requireOutputEventId(run: AgentRun): string { const output = run.outputEventIds[0]; if (!output) throw new Error(`Completed run has no output event: ${run.id}`); return output;}
function stringField(value: unknown, name: string): string { if (typeof value !== "string" || value.length === 0) throw new Error(`Judgment has invalid ${name}`); return value;}
function numberField(value: unknown, name: string): number { if (typeof value !== "number" || !Number.isSafeInteger(value)) throw new Error(`Judgment has invalid ${name}`); return value;}
function objectField(value: unknown): JsonObject | undefined { return value && typeof value === "object" && !Array.isArray(value) ? value as JsonObject : undefined;}
function stringArray(value: unknown): string[] { return Array.isArray(value) && value.every((part) => typeof part === "string") ? value as string[] : [];}