diff --git a/.evolve/skill-runs.jsonl b/.evolve/skill-runs.jsonl index 7f0634e9..21288b50 100644 --- a/.evolve/skill-runs.jsonl +++ b/.evolve/skill-runs.jsonl @@ -34,3 +34,6 @@ {"skill":"/verify","ts":"2026-07-29T20:20:04Z","project":"agent-eval-trace-tool-descriptors","target":"transport-neutral trace-analysis package export","operatorPrompt":"","durationMin":null,"verdict":"SHIP","dispatchedTo":"/reflect","operatorOverride":null,"transcriptPath":null,"traceDir":null} {"skill":"/reflect","ts":"2026-07-29T20:23:34Z","project":"agent-eval-trace-tool-descriptors","target":"trace-tool descriptor session n=1","operatorPrompt":"","durationMin":null,"verdict":"PASS","dispatchedTo":"/handoff","operatorOverride":null,"transcriptPath":null,"traceDir":null} {"skill":"/handoff","ts":"2026-07-29T20:24:06Z","project":"agent-eval-trace-tool-descriptors","target":"agent-eval trace-tool descriptor implementation","operatorPrompt":"","durationMin":null,"verdict":"READY","dispatchedTo":"/stop","operatorOverride":null,"transcriptPath":null,"traceDir":null} +{"skill":"/critical-audit","ts":"2026-07-29T22:47:03Z","project":"agent-eval-control-integrity","target":"agent-eval PR 487 control-integrity repair, 17 changed paths","operatorPrompt":"","durationMin":null,"verdict":"APPROVE","dispatchedTo":"/verify","operatorOverride":null,"transcriptPath":null,"traceDir":null} +{"skill":"/verify","ts":"2026-07-29T22:47:04Z","project":"agent-eval-control-integrity","target":"agent-eval PR 487: 4244 tests, build, packed exports, runtime import","operatorPrompt":"","durationMin":null,"verdict":"PASS","dispatchedTo":"/stop","operatorOverride":null,"transcriptPath":null,"traceDir":null} +{"skill":"/simplify","ts":"2026-07-29T22:54:49Z","project":"agent-eval-control-integrity","target":"PR 487 supervisor-run integrity modularization","operatorPrompt":"","durationMin":null,"verdict":"PASS","dispatchedTo":"/stop","operatorOverride":null,"transcriptPath":null,"traceDir":null} diff --git a/CHANGELOG.md b/CHANGELOG.md index fa8ca1e1..253743fa 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,14 @@ All notable changes to `@tangle-network/agent-eval` and its sibling `agent-eval- --- +## [Unreleased] + +### Added + +- `CONTROL_INTEGRITY_ANALYST` deterministically checks the existing `SupervisorRunSources` and `SupervisorRunTree` for duplicate or detached identities, parent cycles, impossible event order, orphan terminal events, and steer request/acknowledgement mismatches. + Missing transcripts, profile ids, worker logs, and declared tree gaps remain explicitly unavailable and never become a zero or a clean result. + The analyst emits ordinary `AnalystFinding` records and adds no run schema or execution authority. + ## [0.136.0] - 2026-07-29 - preserve recursive evidence and complete profile changes ### Fixed @@ -16,12 +24,18 @@ All notable changes to `@tangle-network/agent-eval` and its sibling `agent-eval- - Spawned invocations retain an explicit `supervisor` or `worker` role, and structured verdicts retain their numeric score in rollout rewards and per-worker report rows. - Accepted-patch counts are unavailable when the source did not retain worker deliverables, even when a worker event claimed patch bytes. - Manager and worker token totals have independent unavailable reasons, so an uncaptured channel is not reported as zero and a captured zero remains zero. +- Supervisor-run integrity checks share one typed source parse, correlate worker control rows by stable invocation id, and distinguish captured-empty control artifacts from missing artifacts. +- Claude Code child agents remain workers when they delegate, and steer counts use the original tool-use request id. +- Malformed journal and worker-control rows make dependent checks unavailable instead of producing missing-parent claims or clean zero counts. ### Changed — BREAKING - Custom `SupervisorRunSources` readers must add `managerTokens` and `workerTokens` to `SourceLimits`. Set each field to `null` only when that role's aggregate token channel is complete; otherwise set the reason it is unavailable. Readers with no source limitations can continue to use `NO_SOURCE_LIMITS`. +- `SupervisorRunTree.gaps` now contains typed `{ code, message, nodeId?, count? }` records instead of free-form strings. +- `WorkerLogFacts.steersQueued` and `steersDelivered` are now nullable. + `null` means malformed, incomplete, or uncorrelated control rows prevent exact request-id accounting. ## [0.135.4] - 2026-07-29 - keep rich source evidence in one schema cohort diff --git a/README.md b/README.md index 95de1233..8f09e3ce 100644 --- a/README.md +++ b/README.md @@ -370,6 +370,7 @@ See [concepts](./docs/concepts.md), [customer paths](./docs/customer-journeys.md | `@tangle-network/agent-eval/reporting` | Statistical comparisons and report rendering. | | `@tangle-network/agent-eval/analyst` | Model-assisted failure analysis. | | `@tangle-network/agent-eval/traces` | Store, replay, and inspect structured traces. | +| `@tangle-network/agent-eval/supervisor-run` | Read and analyze recursive supervisor runs without collapsing missing measurements to zero. | | `@tangle-network/agent-eval/benchmarks` | Benchmark adapters and retrieval metrics. | | `@tangle-network/agent-eval/rl` | Export rewards, preferences, and training rows. | | `@tangle-network/agent-eval/wire` | HTTP and RPC schemas for other languages. | diff --git a/docs/trace-analysis.md b/docs/trace-analysis.md index 69b61e17..061d5c0d 100644 --- a/docs/trace-analysis.md +++ b/docs/trace-analysis.md @@ -132,6 +132,52 @@ for (const c of overview.error_clusters) { See `failureClusters` in [insight-report.md](./insight-report.md) and the `ErrorCluster` type doc-comments for the field-level contract. +## Recursive control integrity (no LLM) + +`CONTROL_INTEGRITY_ANALYST` checks the existing `SupervisorRunSources` or `SupervisorRunTree` directly. +It does not define another run format. +Register it as a custom-input analyst and pass the existing value under its stable id: + +```ts +import { + AnalystRegistry, + CONTROL_INTEGRITY_ANALYST, +} from '@tangle-network/agent-eval/analyst' +import { + readLoopsSupervisorRun, +} from '@tangle-network/agent-eval/supervisor-run' + +const sources = await readLoopsSupervisorRun(runDir) +const registry = new AnalystRegistry() +registry.register(CONTROL_INTEGRITY_ANALYST) + +const result = await registry.run('run-123', { + custom: { 'control-integrity': sources }, +}) +``` + +Pass `SupervisorRunSources` when it is available. +A `SupervisorRunTree` does not retain raw journal multiplicity or worker request and acknowledgement rows, so tree input explicitly reports those checks as unavailable. + +The deterministic pass can prove only facts represented by these two existing surfaces. + +| Question | Current evidence | What the analyst can say | +|---|---|---| +| Is every invocation attached to one unambiguous tree? | `rootId`, `rollout_id`, `parent_rollout_id`, `run_id` | Duplicate ids, missing parents, extra parentless roots, cross-run edges, and ancestry cycles are violations with exact field references. | +| Did invocation roles survive capture? | Explicit journal and `RolloutLine.role` values | The root must remain `supervisor`; non-root roles are consumed as recorded, and workers may spawn workers. | +| Is the causal order possible? | `outcome.metrics.spawned_at`, `started_at`, `settled_at`, `completed_at`, `finished_at` when present | A child before its parent, a child after its parent closed, or a close before a start is a violation; absent timestamps produce no timing claim. | +| Did a queued steer reach the worker? | `SupervisorRunSources.workers[].inbox` and `.events` | Requests and acknowledgements are joined by request id, not compared as totals. Missing, malformed, duplicate, or uncorrelated rows make the affected count unavailable. | +| Can behavior be attributed to an exact profile? | `policy.agent_profile_cell_id` | An absent id is reported as unavailable. | +| Can action authorship or reasoning be inspected? | `messages[]` | Empty gap rows are reported as unavailable. | + +An empty finding list means only that no implemented rule fired on the captured fields. +It does not certify that an agent chose the action, that the action was authorized, that a budget or depth limit was enforced, or that a finding caused a later decision. +Those claims require upstream action-decision events carrying `action_id`, `actor_rollout_id`, `target_rollout_id`, `action_kind`, `authority_snapshot_id`, requested and granted resource/depth values, the authorization result, and any `finding_id` or evidence references that caused the action. +Resume integrity additionally requires an explicit prior-session id and resumed-session id rather than a prose summary. + +Malformed source rows are excluded from structural claims. +Their count is retained in `SupervisorRunTree.gaps`, so analyzing a projected tree later cannot turn an unreadable parent row into a missing-parent violation. + ## Required Trace Shape Every serious product run should include: diff --git a/scripts/verify-package-exports.mjs b/scripts/verify-package-exports.mjs index ca089d9e..9c89f6ab 100644 --- a/scripts/verify-package-exports.mjs +++ b/scripts/verify-package-exports.mjs @@ -153,6 +153,8 @@ try { ` import { CostLedger, + CONTROL_INTEGRITY_ANALYST as ROOT_CONTROL_INTEGRITY_ANALYST, + analyzeSupervisorRunIntegrity as ROOT_ANALYZE_SUPERVISOR_RUN_INTEGRITY, InMemoryTraceStore, type BenchmarkRunnerConfig, type ChatClient, @@ -169,7 +171,10 @@ try { runTaskScore, } from '@tangle-network/agent-eval' import { + CONTROL_INTEGRITY_ANALYST, + ControlIntegrityAnalyst, RawAnalystFindingSchema, + emitControlIntegrityFindings, type RawAnalystFinding, type TraceAnalystGolden, } from '@tangle-network/agent-eval/analyst' @@ -230,6 +235,13 @@ try { toGrpoRows, toSftRows, } from '@tangle-network/agent-eval/rl' + import { + analyzeSupervisorRunIntegrity, + type SupervisorRunIntegrityReport, + type SupervisorRunSources, + type SupervisorRunTree, + type SupervisorRunTreeGap, + } from '@tangle-network/agent-eval/supervisor-run' const store: TraceAnalysisStore = new OtlpFileTraceStore({ path: 'spans.jsonl' }) const traceToolDescriptors: TraceAnalysisToolDescriptor[] = @@ -336,6 +348,13 @@ try { const removedDatasetRecordInput: Parameters[0] = removedRecordInputs const canonicalChat = null as unknown as ChatClient const canonicalJudge = null as unknown as JudgeFn + const controlIntegrityAnalyst: ControlIntegrityAnalyst = CONTROL_INTEGRITY_ANALYST + const controlIntegrityInput = undefined as SupervisorRunSources | SupervisorRunTree | undefined + const controlIntegrityReport = undefined as SupervisorRunIntegrityReport | undefined + const controlTreeGap: SupervisorRunTreeGap = { + code: 'journal-unavailable', + message: 'not captured', + } const rawFinding: RawAnalystFinding = RawAnalystFindingSchema.parse({ severity: 'info', claim: 'current', @@ -454,6 +473,14 @@ try { removedDatasetRecordInput, canonicalChat, canonicalJudge, + controlIntegrityAnalyst, + controlIntegrityInput, + controlIntegrityReport, + controlTreeGap, + analyzeSupervisorRunIntegrity, + ROOT_ANALYZE_SUPERVISOR_RUN_INTEGRITY, + emitControlIntegrityFindings, + ROOT_CONTROL_INTEGRITY_ANALYST, rawFinding, golden, report, @@ -522,6 +549,22 @@ try { if ('CanonicalRawAnalystFindingSchema' in analyst) { throw new Error('obsolete analyst export CanonicalRawAnalystFindingSchema') } + const integrityInput = { rootId: null, nodes: [], gaps: [] } + const integrityReport = root.analyzeSupervisorRunIntegrity(integrityInput) + const integrityFindings = analyst.emitControlIntegrityFindings( + integrityInput, + '2026-07-29T00:00:00.000Z', + ) + const integrityCodes = new Set( + integrityFindings.map((finding) => finding.metadata?.integrity_code), + ) + if ( + integrityReport.input !== 'tree' || + !integrityCodes.has('root-unavailable') || + !integrityCodes.has('source-checks-unavailable') + ) { + throw new Error('packed control-integrity runtime exports returned an invalid result') + } for (const name of [ 'buildTraceAnalysisToolDescriptors', 'buildTraceAnalystTools', diff --git a/src/analyst/index.ts b/src/analyst/index.ts index b76d3e2c..217a289e 100644 --- a/src/analyst/index.ts +++ b/src/analyst/index.ts @@ -78,6 +78,11 @@ export { KNOWLEDGE_GAP_KIND_SPEC, KNOWLEDGE_POISONING_KIND_SPEC, } from './kinds' +export { + CONTROL_INTEGRITY_ANALYST, + ControlIntegrityAnalyst, + emitControlIntegrityFindings, +} from './kinds/control-integrity' export type { SkillUsageRecord, SkillUsageReport, diff --git a/src/analyst/kinds/control-integrity.test.ts b/src/analyst/kinds/control-integrity.test.ts new file mode 100644 index 00000000..fd872c0d --- /dev/null +++ b/src/analyst/kinds/control-integrity.test.ts @@ -0,0 +1,76 @@ +import { describe, expect, it } from 'vitest' +import { fixtureSources } from '../../supervisor-run/fixtures' +import type { SupervisorRunTree } from '../../supervisor-run/types' +import { AnalystRegistry } from '../registry' +import { CONTROL_INTEGRITY_ANALYST, emitControlIntegrityFindings } from './control-integrity' + +const AT = '2026-07-29T18:00:00.000Z' + +describe('control-integrity analyst adapter', () => { + it('maps typed integrity issues into stable zero-cost analyst findings', async () => { + const input: SupervisorRunTree = { + rootId: null, + nodes: [], + gaps: [{ code: 'journal-unavailable', message: 'journal unavailable' }], + } + const registry = new AnalystRegistry() + registry.register(CONTROL_INTEGRITY_ANALYST) + + const result = await registry.run( + 'run-1', + { custom: { 'control-integrity': input } }, + { tags: { producedAt: AT } }, + ) + const repeated = emitControlIntegrityFindings(input, '2099-01-01T00:00:00.000Z') + + expect(result.per_analyst).toEqual([ + expect.objectContaining({ + analyst_id: 'control-integrity', + status: 'ok', + usage: expect.objectContaining({ cost: { kind: 'observed', usd: 0 } }), + }), + ]) + expect(result.findings.map((finding) => finding.finding_id)).toEqual( + repeated.map((finding) => finding.finding_id), + ) + expect(result.findings.every((finding) => finding.metadata?.integrity_code)).toBe(true) + }) + + it('passes source input through without dropping source-only checks', () => { + const findings = emitControlIntegrityFindings( + fixtureSources({ workers: null, workersMissingReason: 'not retained' }), + AT, + ) + + expect(findings.some((finding) => finding.metadata?.integrity_input === 'sources')).toBe(true) + expect(findings.map((finding) => finding.metadata?.integrity_code)).toContain( + 'worker-controls-unavailable', + ) + expect(findings.map((finding) => finding.metadata?.integrity_code)).not.toContain( + 'source-checks-unavailable', + ) + }) + + it('keeps identical local node ids distinct across runs', () => { + const input = (runId: string): SupervisorRunTree => ({ + rootId: 'root', + nodes: [ + { + ...({} as SupervisorRunTree['nodes'][number]), + rollout_id: 'root', + run_id: runId, + }, + ], + gaps: [], + }) + + const first = emitControlIntegrityFindings(input('run-a'), AT) + const second = emitControlIntegrityFindings(input('run-b'), AT) + + expect(first).not.toHaveLength(0) + expect(second).not.toHaveLength(0) + expect(first[0]?.finding_id).not.toBe(second[0]?.finding_id) + expect(first[0]?.subject).toContain('run-a/') + expect(second[0]?.subject).toContain('run-b/') + }) +}) diff --git a/src/analyst/kinds/control-integrity.ts b/src/analyst/kinds/control-integrity.ts new file mode 100644 index 00000000..cad67aeb --- /dev/null +++ b/src/analyst/kinds/control-integrity.ts @@ -0,0 +1,83 @@ +import { + analyzeSupervisorRunIntegrity, + type SupervisorRunIntegrityEvidence, +} from '../../supervisor-run/integrity' +import type { SupervisorRunSources, SupervisorRunTree } from '../../supervisor-run/types' +import { + type Analyst, + type AnalystContext, + type AnalystFinding, + type EvidenceRef, + makeFinding, +} from '../types' + +const ANALYST_ID = 'control-integrity' + +function shown(value: unknown): string { + if (value === undefined) return '' + const encoded = JSON.stringify(value) + return encoded === undefined ? String(value) : encoded +} + +function evidenceRef(namespace: string, value: SupervisorRunIntegrityEvidence): EvidenceRef { + return { + kind: 'metric', + uri: `supervisor-run://${encodeURIComponent(namespace)}/${value.path}`, + excerpt: shown(value.value), + } +} + +/** Translate typed supervisor-run integrity issues into the shared analyst envelope. */ +export function emitControlIntegrityFindings( + input: SupervisorRunSources | SupervisorRunTree, + producedAt: string, +): AnalystFinding[] { + const report = analyzeSupervisorRunIntegrity(input, { capturedAt: producedAt }) + return report.issues.map((issue) => + makeFinding({ + analyst_id: ANALYST_ID, + produced_at: producedAt, + area: issue.area, + severity: issue.severity, + subject: `${report.runRef}/${issue.subject}`, + claim: issue.claim, + rationale: issue.detail, + evidence_refs: issue.evidence.map((value) => evidenceRef(report.runRef, value)), + recommended_action: issue.recommendedAction, + validation_plan: + 'Re-run this deterministic analyst on the retained SupervisorRunSources or SupervisorRunTree after correcting the producer.', + confidence: 1, + metadata: { + integrity_code: issue.code, + integrity_input: report.input, + integrity_run_ref: report.runRef, + integrity_subject: issue.subject, + ...issue.metadata, + }, + }), + ) +} + +/** Deterministic Analyst adapter for `SupervisorRunSources | SupervisorRunTree`. */ +export class ControlIntegrityAnalyst implements Analyst { + readonly id = ANALYST_ID + readonly description = + 'Deterministic supervisor-run integrity checks with explicit unavailable evidence.' + readonly inputKind = 'custom' as const + readonly cost = { kind: 'deterministic' as const, est_usd_per_run: 0 } + readonly version = '2.0.0' + + async analyze( + input: SupervisorRunSources | SupervisorRunTree, + ctx: AnalystContext, + ): Promise { + const producedAt = ctx.tags?.producedAt ?? new Date().toISOString() + const findings = emitControlIntegrityFindings(input, producedAt) + ctx.log?.(`control-integrity: ${findings.length} finding(s)`, { + input: 'nodes' in input ? 'SupervisorRunTree' : 'SupervisorRunSources', + }) + return findings + } +} + +export const CONTROL_INTEGRITY_ANALYST = new ControlIntegrityAnalyst() diff --git a/src/analyst/kinds/index.ts b/src/analyst/kinds/index.ts index 6e44e99d..f6a40861 100644 --- a/src/analyst/kinds/index.ts +++ b/src/analyst/kinds/index.ts @@ -9,6 +9,11 @@ * to pass each completed kind's findings to the kinds that follow it. */ +export { + CONTROL_INTEGRITY_ANALYST, + ControlIntegrityAnalyst, + emitControlIntegrityFindings, +} from './control-integrity' export { FAILURE_MODE_KIND_SPEC } from './failure-mode' export { IMPROVEMENT_KIND_SPEC } from './improvement' export { KNOWLEDGE_GAP_KIND_SPEC } from './knowledge-gap' diff --git a/src/index.ts b/src/index.ts index e85db9a4..37afc8f6 100644 --- a/src/index.ts +++ b/src/index.ts @@ -94,6 +94,11 @@ export { KNOWLEDGE_GAP_KIND_SPEC, KNOWLEDGE_POISONING_KIND_SPEC, } from './analyst/kinds' +export { + CONTROL_INTEGRITY_ANALYST, + ControlIntegrityAnalyst, + emitControlIntegrityFindings, +} from './analyst/kinds/control-integrity' export { SKILL_USAGE_ANALYST, SkillUsageAnalyst } from './analyst/kinds/skill-usage' export { type AnalystHooks, @@ -537,6 +542,7 @@ export { // Full surface on the `@tangle-network/agent-eval/supervisor-run` subpath. export { analyzeSupervisorRun, + analyzeSupervisorRunIntegrity, analyzeSupervisorRunSources, claudeCodeSupervisorRunReader, isUnavailable, @@ -546,13 +552,23 @@ export { renderSupervisorRunMarkdown, rollupSupervisorRuns, type SourceLimits, + SUPERVISOR_RUN_INTEGRITY_SCHEMA, SUPERVISOR_RUN_SCHEMA, + type SupervisorRunIntegrityEvidence, + type SupervisorRunIntegrityIssue, + type SupervisorRunIntegrityIssueCode, + type SupervisorRunIntegrityOptions, + type SupervisorRunIntegrityReport, + type SupervisorRunIntegritySeverity, type SupervisorRunNodeRole, type SupervisorRunReader, type SupervisorRunReport, type SupervisorRunRollup, + type SupervisorRunSourceOnlyCheckCode, type SupervisorRunSources, type SupervisorRunTree, + type SupervisorRunTreeGap, + type SupervisorRunTreeGapCode, showMeasured, supervisorRunRolloutLines, type Unavailable, diff --git a/src/supervisor-run/analyze.ts b/src/supervisor-run/analyze.ts index 49d871c1..8e4ed6a1 100644 --- a/src/supervisor-run/analyze.ts +++ b/src/supervisor-run/analyze.ts @@ -5,6 +5,15 @@ * (`loops-reader.ts` is one). */ +import { + asRecord, + parseJson, + parseJsonl, + parseSupervisorTree, + type SpawnRow, + type WorkerLogFacts, + workerSourceKey, +} from './source-facts' import { type DecisionMetrics, type EconomicsMetrics, @@ -18,7 +27,6 @@ import { type SteerBreakdown, SUPERVISOR_RUN_ROLLUP_SCHEMA, SUPERVISOR_RUN_SCHEMA, - type SupervisorRunNodeRole, type SupervisorRunReport, type SupervisorRunRollup, type SupervisorRunSources, @@ -26,337 +34,21 @@ import { unavailable, } from './types' -// --------------------------------------------------------------------------- -// Journal shapes — structurally parsed. The journal is the contract, not the type. -// --------------------------------------------------------------------------- - -interface JournalEvent { - kind?: unknown - id?: unknown - parent?: unknown - label?: unknown - role?: unknown - status?: unknown - verdict?: unknown - reason?: unknown - seq?: unknown - at?: unknown - spend?: unknown - spent?: unknown -} - -interface Tokens { - input: number - output: number - cacheRead: number - cacheWrite: number - /** False when the event carried no cache counters at all — not "zero cached". */ - hasCache: boolean -} - -interface SpendLike { - tokens: Tokens - usd: number -} - -export function asRecord(v: unknown): Record { - return typeof v === 'object' && v !== null ? (v as Record) : {} -} - -function num(v: unknown): number { - return typeof v === 'number' && Number.isFinite(v) ? v : 0 -} - -function readSpend(v: unknown): SpendLike { - const rec = asRecord(v) - const tok = asRecord(rec.tokens) - const cacheRead = tok.cacheRead ?? tok.cache_read - const cacheWrite = tok.cacheWrite ?? tok.cache_write - return { - tokens: { - input: num(tok.input), - output: num(tok.output), - cacheRead: num(cacheRead), - cacheWrite: num(cacheWrite), - hasCache: cacheRead !== undefined || cacheWrite !== undefined, - }, - usd: num(rec.usd), - } -} +export { + asRecord, + type CloseRow, + parseJson, + parseJsonl, + parseSupervisorTree, + type SpawnRow, + type SteerAcknowledgementFact, + type SteerRequestFact, + type SupervisorTreeFacts, + type WorkerLogFacts, +} from './source-facts' const NO_CACHE_COUNTERS = 'the journal carries no cache-token counters for this role' -export function parseJsonl(text: string | null): Record[] { - if (text === null) return [] - const out: Record[] = [] - for (const line of text.split('\n')) { - const trimmed = line.trim() - if (!trimmed) continue - try { - const parsed: unknown = JSON.parse(trimmed) - if (typeof parsed === 'object' && parsed !== null) out.push(parsed as Record) - } catch { - // A torn last line (writer killed mid-append) is skipped, never fatal. - } - } - return out -} - -export function parseJson(text: string | null): Record | null { - if (text === null) return null - try { - const parsed: unknown = JSON.parse(text) - return typeof parsed === 'object' && parsed !== null - ? (parsed as Record) - : null - } catch { - return null - } -} - -function ms(at: unknown): number | null { - if (typeof at !== 'string') return null - const t = Date.parse(at) - return Number.isFinite(t) ? t : null -} - -// --------------------------------------------------------------------------- -// The supervision tree, as parsed from the journal event stream. -// --------------------------------------------------------------------------- - -export interface SpawnRow { - id: string - parent: string | null - label: string - role: SupervisorRunNodeRole - at: number | null -} - -export interface CloseRow { - id: string - kind: 'settled' | 'cancelled' - status: string | null - /** String verdict from legacy journals, or valid/invalid for a structured verdict. */ - verdict: string | null - /** Structured verdict validity, when recorded. */ - valid: boolean | null - /** Structured verdict score, preserved without boolean coercion. */ - score: number | null - /** The verdict exactly as the journal carried it. */ - rawVerdict: unknown | null - at: number | null - spend: SpendLike - /** False when the close event carried no spend object — not "spent nothing". */ - hasSpend: boolean -} - -export interface WorkerLogFacts { - started: number | null - /** True once a `finished` event was seen — independent of whether its `at` parsed. */ - finished: boolean - finishedAt: number | null - passed: boolean | null - /** Numeric score exactly as the finished event recorded it. */ - score: number | null - /** `patchBytes` as reported by the finished event (not the patch file's size). */ - finishedPatchBytes: number | null - evidenceBytes: number - steersQueued: number - steersDelivered: number - questions: number -} - -/** - * The tree + timeline the report is computed from, exposed because the rollout-row - * minter needs exactly the same parse (one parser, two consumers). - */ -export interface SupervisorTreeFacts { - readonly rootId: string | null - readonly spawns: readonly SpawnRow[] - readonly closes: readonly CloseRow[] - readonly workerSpawns: readonly SpawnRow[] - readonly workerCloses: readonly CloseRow[] - readonly brain: { - tokensIn: number - tokensOut: number - cacheRead: number - cacheWrite: number - /** False when no metered event carried cache counters — not "nothing cached". */ - hasCache: boolean - usd: number - meteredCount: number - } - readonly workerLogs: ReadonlyMap - readonly startedAt: number | null - readonly completedAt: number | null -} - -interface VerdictFacts { - label: string | null - valid: boolean | null - score: number | null - raw: unknown | null -} - -function readVerdict(v: unknown): VerdictFacts { - if (typeof v === 'string') return { label: v, valid: null, score: null, raw: v } - if (typeof v !== 'object' || v === null) { - return { label: null, valid: null, score: null, raw: null } - } - const rec = v as Record - const valid = typeof rec.valid === 'boolean' ? rec.valid : null - return { - label: valid === null ? null : valid ? 'valid' : 'invalid', - valid, - score: typeof rec.score === 'number' && Number.isFinite(rec.score) ? rec.score : null, - raw: v, - } -} - -function workerSourceKey(worker: NonNullable[number]): string { - return worker.workerId ?? worker.label -} - -export function parseSupervisorTree(src: SupervisorRunSources): SupervisorTreeFacts { - const events = parseJsonl(src.journal) - const state = parseJson(src.state) - - const spawns: SpawnRow[] = [] - const closes: CloseRow[] = [] - let brainIn = 0 - let brainOut = 0 - let brainCacheRead = 0 - let brainCacheWrite = 0 - let brainHasCache = false - let brainUsd = 0 - let meteredCount = 0 - let rootId: string | null = null - for (const ev of events as JournalEvent[]) { - const kind = typeof ev.kind === 'string' ? ev.kind : '' - const id = typeof ev.id === 'string' ? ev.id : '' - if (kind === 'spawned') { - const parent = typeof ev.parent === 'string' ? ev.parent : null - const label = typeof ev.label === 'string' ? ev.label : '' - if (parent === null && rootId === null) rootId = id - const role: SupervisorRunNodeRole = - parent === null || ev.role === 'supervisor' ? 'supervisor' : 'worker' - spawns.push({ id, parent, label, role, at: ms(ev.at) }) - } else if (kind === 'settled') { - const verdict = readVerdict(ev.verdict) - closes.push({ - id, - kind: 'settled', - status: typeof ev.status === 'string' ? ev.status : null, - verdict: verdict.label, - valid: verdict.valid, - score: verdict.score, - rawVerdict: verdict.raw, - at: ms(ev.at), - spend: readSpend(ev.spent), - hasSpend: asRecord(ev.spent).tokens !== undefined, - }) - } else if (kind === 'cancelled') { - closes.push({ - id, - kind: 'cancelled', - status: 'cancelled', - verdict: typeof ev.reason === 'string' ? ev.reason : null, - valid: null, - score: null, - rawVerdict: typeof ev.reason === 'string' ? ev.reason : null, - at: ms(ev.at), - spend: { - tokens: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, hasCache: false }, - usd: 0, - }, - hasSpend: false, - }) - } else if (kind === 'metered') { - const s = readSpend(ev.spend) - brainIn += s.tokens.input - brainOut += s.tokens.output - brainCacheRead += s.tokens.cacheRead - brainCacheWrite += s.tokens.cacheWrite - brainHasCache = brainHasCache || s.tokens.hasCache - brainUsd += s.usd - meteredCount += 1 - } - } - - const workerSpawns = spawns.filter((s) => s.id !== rootId) - const workerIds = new Set(workerSpawns.map((s) => s.id)) - const workerCloses = closes.filter((c) => workerIds.has(c.id)) - - const workerLogs = new Map() - for (const w of src.workers ?? []) { - const facts: WorkerLogFacts = { - started: null, - finished: false, - finishedAt: null, - passed: null, - score: null, - finishedPatchBytes: null, - evidenceBytes: 0, - steersQueued: 0, - steersDelivered: 0, - questions: 0, - } - for (const req of parseJsonl(w.inbox)) { - if (typeof req.message === 'string' && req.message.trim().length > 0) facts.steersQueued += 1 - } - for (const ev of parseJsonl(w.events)) { - const kind = ev.kind - if (kind === 'message') { - if (ev.direction === 'up') { - facts.questions += 1 - continue - } - // A down-leg message not backed by an inbox line (in-process steer) still counts. - if (typeof ev.requestId !== 'string') facts.steersQueued += 1 - if (ev.delivered === true) facts.steersDelivered += 1 - } else if (kind === 'started') { - facts.started = ms(ev.at) - } else if (kind === 'finished') { - facts.finished = true - facts.finishedAt = ms(ev.at) - facts.passed = typeof ev.passed === 'boolean' ? ev.passed : null - facts.score = typeof ev.score === 'number' && Number.isFinite(ev.score) ? ev.score : null - facts.finishedPatchBytes = typeof ev.patchBytes === 'number' ? ev.patchBytes : null - facts.evidenceBytes = typeof ev.evidence === 'string' ? ev.evidence.length : 0 - } - } - workerLogs.set(workerSourceKey(w), facts) - } - - const startedAt = ms(state?.startedAt) ?? spawns[0]?.at ?? null - const completedAt = - ms(state?.completedAt) ?? - [...spawns.map((s) => s.at), ...closes.map((c) => c.at)].reduce( - (acc, t) => (t === null ? acc : acc === null ? t : Math.max(acc, t)), - null, - ) - - return { - rootId, - spawns, - closes, - workerSpawns, - workerCloses, - brain: { - tokensIn: brainIn, - tokensOut: brainOut, - cacheRead: brainCacheRead, - cacheWrite: brainCacheWrite, - hasCache: brainHasCache, - usd: brainUsd, - meteredCount, - }, - workerLogs, - startedAt, - completedAt, - } -} - // --------------------------------------------------------------------------- // The analyzer. // --------------------------------------------------------------------------- @@ -380,10 +72,10 @@ export function analyzeSupervisorRunSources( ? 'no supervisor run dir under /.loops/supervisor' : 'journal.jsonl absent' const haveJournal = src.journal !== null - const state = parseJson(src.state) + const tree = parseSupervisorTree(src) + const state = tree.state const result = parseJson(src.result) const judge = parseJson(src.judge) - const tree = parseSupervisorTree(src) const { rootId, workerSpawns, workerCloses, startedAt, completedAt } = tree const spawnById = new Map(workerSpawns.map((spawn) => [spawn.id, spawn])) const spawnsByLabel = new Map() @@ -421,21 +113,52 @@ export function analyzeSupervisorRunSources( if (src.workers !== null) { for (const w of src.workers) { const facts = tree.workerLogs.get(workerSourceKey(w)) - const queued = facts?.steersQueued ?? 0 - const delivered = facts?.steersDelivered ?? 0 + const queued = facts?.steersQueued ?? null + const delivered = facts?.steersDelivered ?? null upLegMessages += facts?.questions ?? 0 - steerRows.push({ workerId: w.workerId ?? null, worker: w.label, queued, delivered }) - steerQueuedTotal += queued - steerDeliveredTotal += delivered + if (queued !== null && delivered !== null) { + steerRows.push({ workerId: w.workerId ?? null, worker: w.label, queued, delivered }) + } + if (queued !== null) steerQueuedTotal += queued + if (delivered !== null) steerDeliveredTotal += delivered } } const workersGapReason = src.workersMissingReason ?? 'workers/ directory absent' + const unavailableReasons = (pick: (facts: WorkerLogFacts) => string | null): string | null => { + const reasons = tree.workerLogRows + .map((facts) => pick(facts)) + .filter((reason): reason is string => reason !== null) + return reasons.length === 0 + ? null + : `exact steer accounting unavailable for ${reasons.length} worker row(s): ${[...new Set(reasons)].join(' | ')}` + } + const queuedGapReason = unavailableReasons((facts) => facts.steersQueuedUnavailable) + const deliveredGapReason = unavailableReasons((facts) => facts.steersDeliveredUnavailable) + const workerEventsGapReason = unavailableReasons((facts) => + !facts.eventsCaptured + ? 'events absent' + : facts.eventsInvalidRows > 0 + ? 'events contain malformed rows' + : null, + ) const steers: Measured = - src.workers === null ? gap('steers', workersGapReason) : steerQueuedTotal + src.workers === null + ? gap('steers', workersGapReason) + : queuedGapReason === null + ? steerQueuedTotal + : gap('steers', queuedGapReason) const steersDelivered: Measured = - src.workers === null ? unavailable(workersGapReason) : steerDeliveredTotal + src.workers === null + ? unavailable(workersGapReason) + : deliveredGapReason === null + ? steerDeliveredTotal + : unavailable(deliveredGapReason) const steersByWorker: Measured = - src.workers === null ? unavailable(workersGapReason) : steerRows + src.workers === null + ? unavailable(workersGapReason) + : queuedGapReason === null && deliveredGapReason === null + ? steerRows + : unavailable(queuedGapReason ?? deliveredGapReason ?? workersGapReason) // The `[driver] registered tools: …supervisor_steer…` banner names the verb without // invoking it, so banner lines are subtracted from the raw mention count. @@ -673,8 +396,17 @@ export function analyzeSupervisorRunSources( observeThenRespawn: haveJournal ? observeThenRespawn : unavailable(journalMissing), respawnWithoutEvidence: haveJournal ? respawnWithoutEvidence : unavailable(journalMissing), reviewActions: - src.workers === null ? unavailable(workersGapReason) : steerQueuedTotal + upLegMessages, - workerEvidenceBytes: src.workers === null ? unavailable(workersGapReason) : evidenceBytes, + src.workers === null + ? unavailable(workersGapReason) + : queuedGapReason === null + ? steerQueuedTotal + upLegMessages + : unavailable(queuedGapReason), + workerEvidenceBytes: + src.workers === null + ? unavailable(workersGapReason) + : workerEventsGapReason === null + ? evidenceBytes + : unavailable(workerEventsGapReason), } // ── economics ────────────────────────────────────────────────────────── diff --git a/src/supervisor-run/claude-code-reader.test.ts b/src/supervisor-run/claude-code-reader.test.ts index 28e23dc2..153b8096 100644 --- a/src/supervisor-run/claude-code-reader.test.ts +++ b/src/supervisor-run/claude-code-reader.test.ts @@ -202,18 +202,20 @@ describe('claudeCodeSupervisorRunReader', () => { it('reports steers as a REAL count — Claude Code can message a live subagent', async () => { const s = await writeSession() - const report = analyzeSupervisorRunSources( - await readClaudeCodeSupervisorRun({ - transcriptPath: s.transcriptPath, - subagentsDir: s.subagentsDir, - }), - ) + const source = await readClaudeCodeSupervisorRun({ + transcriptPath: s.transcriptPath, + subagentsDir: s.subagentsDir, + }) + const report = analyzeSupervisorRunSources(source) expect(report.orchestration.steers).toBe(1) expect(report.orchestration.steersDelivered).toBe(1) expect(report.orchestration.steersByWorker).toEqual([ { workerId: 'ag-a', worker: 'build A', queued: 1, delivered: 1 }, { workerId: 'ag-b', worker: 'build B', queued: 0, delivered: 0 }, ]) + expect(source.workers?.find((worker) => worker.workerId === 'ag-a')?.inbox).toContain( + '"id":"tu-s"', + ) }) it('reports a run with no SendMessage as steers=0, NOT unavailable', async () => { @@ -365,9 +367,9 @@ describe('claudeCodeSupervisorRunReader', () => { expect(report.orchestration.delegationDepth).toBe(2) const tree = supervisorRunRolloutLines(src) - const nestedSupervisor = tree.nodes.find((n) => n.rollout_id === 'ag-a') + const nestedWorker = tree.nodes.find((n) => n.rollout_id === 'ag-a') const grandchild = tree.nodes.find((n) => n.rollout_id === 'ag-c') - expect(nestedSupervisor?.role).toBe('supervisor') + expect(nestedWorker?.role).toBe('worker') expect(grandchild?.parent_rollout_id).toBe('ag-a') expect(grandchild?.role).toBe('worker') }) @@ -378,6 +380,7 @@ describe('claudeCodeSupervisorRunReader', () => { transcriptPath: s.transcriptPath, subagentsDir: s.subagentsDir, }) + expect(src.workers?.find((worker) => worker.workerId === 'ag-b')?.inbox).toBe('') const tree = supervisorRunRolloutLines(src, { supervisorHarness: 'claude-code', workerHarness: 'claude-code', diff --git a/src/supervisor-run/claude-code-reader.ts b/src/supervisor-run/claude-code-reader.ts index e5a535d0..6e8a97d6 100644 --- a/src/supervisor-run/claude-code-reader.ts +++ b/src/supervisor-run/claude-code-reader.ts @@ -390,7 +390,10 @@ export async function readClaudeCodeSupervisorRun( } const spawns: SpawnFact[] = [] - const steersByTarget = new Map>() + const steersByTarget = new Map< + string, + Array<{ requestId: string; at: string | null; delivered: boolean }> + >() const cancels: Array<{ agentId: string; at: string | null }> = [] const settles: TaskNotification[] = [] @@ -422,7 +425,7 @@ export async function readClaudeCodeSupervisorRun( // absent, the steer is counted queued but not delivered. const delivered = structured?.success === true || str(structured?.resumedAgentId) === target const rows = steersByTarget.get(target) ?? [] - rows.push({ at: use.at, delivered }) + rows.push({ requestId: use.id, at: use.at, delivered }) steersByTarget.set(target, rows) continue } @@ -451,7 +454,6 @@ export async function readClaudeCodeSupervisorRun( at: startedAt, }), ] - const supervisorIds = new Set(spawns.map((spawn) => spawn.parentId)) for (const s of spawns) { journalLines.push( line({ @@ -459,7 +461,7 @@ export async function readClaudeCodeSupervisorRun( id: s.agentId, parent: s.parentId, label: s.label, - role: supervisorIds.has(s.agentId) ? 'supervisor' : 'worker', + role: 'worker', at: s.at, }), ) @@ -533,16 +535,14 @@ export async function readClaudeCodeSupervisorRun( const endAt = settleAtByAgent.get(agentId) ?? child?.lastAt ?? null if (startAt !== null) events.push(line({ kind: 'started', label, at: startAt, agentId })) for (const steer of steersByTarget.get(agentId) ?? []) { - inbox.push( - line({ id: `${agentId}:${steer.at}`, at: steer.at, worker: label, message: 'steer' }), - ) + inbox.push(line({ id: steer.requestId, at: steer.at, worker: label, message: 'steer' })) events.push( line({ kind: 'message', label, direction: 'down', at: steer.at, - requestId: `${agentId}:${steer.at}`, + requestId: steer.requestId, delivered: steer.delivered, }), ) @@ -566,8 +566,8 @@ export async function readClaudeCodeSupervisorRun( workers.push({ workerId: agentId, label, - events: events.length === 0 ? null : `${events.join('\n')}\n`, - inbox: inbox.length === 0 ? null : `${inbox.join('\n')}\n`, + events: events.length === 0 ? '' : `${events.join('\n')}\n`, + inbox: inbox.length === 0 ? '' : `${inbox.join('\n')}\n`, patchBytes: null, transcriptRef: child?.path ?? null, patchPath: null, diff --git a/src/supervisor-run/fixtures.ts b/src/supervisor-run/fixtures.ts index d7a55496..e648132c 100644 --- a/src/supervisor-run/fixtures.ts +++ b/src/supervisor-run/fixtures.ts @@ -180,7 +180,7 @@ export function fixtureWorker(label: string, opts: WorkerPlan = { startSec: 0 }) ...(opts.workerId === undefined ? {} : { workerId: opts.workerId }), label, events: `${events.join('\n')}\n`, - inbox: inboxLines.length === 0 ? null : `${inboxLines.join('\n')}\n`, + inbox: inboxLines.length === 0 ? '' : `${inboxLines.join('\n')}\n`, patchBytes: opts.patchBytes ?? null, } } diff --git a/src/supervisor-run/index.ts b/src/supervisor-run/index.ts index af0cf0c3..1dc663ab 100644 --- a/src/supervisor-run/index.ts +++ b/src/supervisor-run/index.ts @@ -19,6 +19,8 @@ export { parseSupervisorTree, rollupSupervisorRuns, type SpawnRow, + type SteerAcknowledgementFact, + type SteerRequestFact, type SupervisorTreeFacts, type WorkerLogFacts, } from './analyze' @@ -30,6 +32,17 @@ export { DEFAULT_STEER_TOOLS, readClaudeCodeSupervisorRun, } from './claude-code-reader' +export { + analyzeSupervisorRunIntegrity, + SUPERVISOR_RUN_INTEGRITY_SCHEMA, + type SupervisorRunIntegrityEvidence, + type SupervisorRunIntegrityIssue, + type SupervisorRunIntegrityIssueCode, + type SupervisorRunIntegrityOptions, + type SupervisorRunIntegrityReport, + type SupervisorRunIntegritySeverity, + type SupervisorRunSourceOnlyCheckCode, +} from './integrity' export { analyzeSupervisorRun, findSupervisorRunDirIn, @@ -71,6 +84,8 @@ export { type SupervisorRunRollup, type SupervisorRunSources, type SupervisorRunTree, + type SupervisorRunTreeGap, + type SupervisorRunTreeGapCode, showMeasured, type Unavailable, unavailable, diff --git a/src/supervisor-run/integrity-issues.ts b/src/supervisor-run/integrity-issues.ts new file mode 100644 index 00000000..5f937a2c --- /dev/null +++ b/src/supervisor-run/integrity-issues.ts @@ -0,0 +1,15 @@ +import type { SupervisorRunIntegrityEvidence, SupervisorRunIntegrityIssue } from './integrity-types' + +export const MAX_EXAMPLES = 20 + +export function evidence(path: string, value: unknown): SupervisorRunIntegrityEvidence { + return { path, value } +} + +export function issue( + init: Omit & { + evidence?: readonly SupervisorRunIntegrityEvidence[] + }, +): SupervisorRunIntegrityIssue { + return { ...init, evidence: init.evidence ?? [] } +} diff --git a/src/supervisor-run/integrity-source.ts b/src/supervisor-run/integrity-source.ts new file mode 100644 index 00000000..70bb78a3 --- /dev/null +++ b/src/supervisor-run/integrity-source.ts @@ -0,0 +1,268 @@ +import { evidence, issue, MAX_EXAMPLES } from './integrity-issues' +import { steerIssues } from './integrity-steer' +import type { SupervisorRunIntegrityIssue } from './integrity-types' +import type { SpawnRow, SupervisorTreeFacts, WorkerLogFacts } from './source-facts' +import type { SupervisorRunSources, SupervisorRunTree } from './types' + +function sourceIdentityIssues(facts: SupervisorTreeFacts): SupervisorRunIntegrityIssue[] { + const out: SupervisorRunIntegrityIssue[] = [] + const malformedSpawns = facts.spawns.filter((spawn) => !spawn.valid) + if (facts.journalInvalidRows > 0 || malformedSpawns.length > 0) { + out.push( + issue({ + code: 'source-row-malformed', + area: 'capture-integrity', + severity: 'high', + subject: 'journal-rows', + claim: 'Some supervisor journal rows are malformed', + detail: `${facts.journalInvalidRows} non-object or invalid JSON row(s) and ${malformedSpawns.length} malformed spawn row(s) were excluded from structural conclusions.`, + evidence: [ + evidence('journal/malformed-json-rows/count', facts.journalInvalidRows), + ...malformedSpawns + .slice(0, MAX_EXAMPLES) + .map((spawn) => + evidence(`journal/spawns/${spawn.sourceRow}/invalid-fields`, spawn.invalidFields), + ), + ], + recommendedAction: + 'Repair or reject malformed journal rows before using the run for structural analysis.', + metadata: { + assessment: 'unavailable', + malformed_json_rows: facts.journalInvalidRows, + malformed_spawn_rows: malformedSpawns.length, + }, + }), + ) + } + const unidentifiedSpawns = malformedSpawns.filter((spawn) => spawn.id.length === 0) + const unidentifiedCloses = facts.closes.filter((close) => close.id.length === 0) + if (unidentifiedSpawns.length + unidentifiedCloses.length > 0) { + out.push( + issue({ + code: 'source-event-identity-unavailable', + area: 'capture-integrity', + severity: 'high', + subject: 'journal-event-identity', + claim: 'Some supervisor journal events have no usable invocation id', + detail: `${unidentifiedSpawns.length} spawn and ${unidentifiedCloses.length} terminal event(s) lack an id.`, + evidence: [ + evidence('journal/unidentified-spawns/count', unidentifiedSpawns.length), + evidence('journal/unidentified-terminals/count', unidentifiedCloses.length), + ], + recommendedAction: 'Reject or repair control events that omit their stable invocation id.', + metadata: { assessment: 'unavailable', checks: ['journal-event-linkage'] }, + }), + ) + } + + const spawnCounts = new Map() + for (const spawn of facts.spawns) { + if (!spawn.valid || spawn.id.length === 0) continue + spawnCounts.set(spawn.id, (spawnCounts.get(spawn.id) ?? 0) + 1) + } + for (const [id, count] of spawnCounts) { + if (count < 2) continue + out.push( + issue({ + code: 'duplicate-spawn', + area: 'control-integrity', + severity: 'critical', + subject: id, + claim: 'An invocation has more than one spawned control event', + detail: `${count} spawned events name invocation ${JSON.stringify(id)}.`, + evidence: [evidence(`journal/spawn-count/${encodeURIComponent(id)}`, count)], + recommendedAction: 'Make spawn event append idempotent.', + }), + ) + } + + const closeCounts = new Map() + for (const close of facts.closes) { + if (close.id.length === 0) continue + closeCounts.set(close.id, (closeCounts.get(close.id) ?? 0) + 1) + } + const uncertainSpawnIds = new Set( + malformedSpawns.filter((spawn) => spawn.id.length > 0).map((spawn) => spawn.id), + ) + if (unidentifiedSpawns.length === 0 && facts.journalInvalidRows === 0) { + const orphanCloses = new Map>() + for (const close of facts.closes) { + if (close.id.length === 0 || spawnCounts.has(close.id) || uncertainSpawnIds.has(close.id)) { + continue + } + const rows = orphanCloses.get(close.id) ?? [] + rows.push(close) + orphanCloses.set(close.id, rows) + } + for (const [id, closes] of orphanCloses) { + out.push( + issue({ + code: 'orphan-terminal', + area: 'control-integrity', + severity: 'critical', + subject: id, + claim: 'A terminal control event names an invocation that was never spawned', + detail: `${closes.length} terminal event(s) for ${JSON.stringify(id)} have no matching spawn.`, + evidence: [ + evidence(`journal/closes/${encodeURIComponent(id)}/count`, closes.length), + ...closes + .slice(0, MAX_EXAMPLES) + .map((close, index) => + evidence(`journal/closes/${encodeURIComponent(id)}/${index}`, close), + ), + ], + recommendedAction: 'Retain the matching spawn event or reject the orphan terminal event.', + }), + ) + } + } + for (const [id, count] of closeCounts) { + if (count < 2) continue + out.push( + issue({ + code: 'duplicate-terminal', + area: 'control-integrity', + severity: 'critical', + subject: id, + claim: 'An invocation has more than one terminal control event', + detail: `${count} settled/cancelled events name invocation ${JSON.stringify(id)}.`, + evidence: [evidence(`journal/terminal-count/${encodeURIComponent(id)}`, count)], + recommendedAction: 'Make terminal event append idempotent.', + }), + ) + } + return out +} + +export interface JoinedWorker { + readonly spawn: SpawnRow + readonly facts: WorkerLogFacts +} + +function joinWorkerControls(facts: SupervisorTreeFacts): { + joined: JoinedWorker[] + issue: SupervisorRunIntegrityIssue | null +} { + const spawnsById = new Map() + const spawnsByLabel = new Map() + for (const spawn of facts.workerSpawns) { + const ids = spawnsById.get(spawn.id) ?? [] + ids.push(spawn) + spawnsById.set(spawn.id, ids) + const labels = spawnsByLabel.get(spawn.label) ?? [] + labels.push(spawn) + spawnsByLabel.set(spawn.label, labels) + } + const rowsWithoutIdByLabel = new Map() + for (const row of facts.workerLogRows) { + if (row.workerId !== null && row.workerId.length > 0) continue + const matches = rowsWithoutIdByLabel.get(row.label) ?? [] + matches.push(row) + rowsWithoutIdByLabel.set(row.label, matches) + } + + const rowsBySpawn = new Map() + const unmatchedRows: number[] = [] + const ambiguousRows: number[] = [] + for (const row of facts.workerLogRows) { + const matches = + row.workerId !== null && row.workerId.length > 0 + ? (spawnsById.get(row.workerId) ?? []) + : (rowsWithoutIdByLabel.get(row.label)?.length ?? 0) === 1 + ? (spawnsByLabel.get(row.label) ?? []) + : [] + if (matches.length === 0) { + unmatchedRows.push(row.sourceIndex) + continue + } + if (matches.length !== 1) { + ambiguousRows.push(row.sourceIndex) + continue + } + const spawn = matches[0] as SpawnRow + const joined = rowsBySpawn.get(spawn.id) ?? [] + joined.push(row) + rowsBySpawn.set(spawn.id, joined) + } + + const joined: JoinedWorker[] = [] + const missingSpawnIds: string[] = [] + const duplicateSpawnIds: string[] = [] + for (const spawn of facts.workerSpawns) { + if ((spawnsById.get(spawn.id)?.length ?? 0) !== 1) continue + const rows = rowsBySpawn.get(spawn.id) ?? [] + if (rows.length === 0) missingSpawnIds.push(spawn.id) + else if (rows.length > 1) duplicateSpawnIds.push(spawn.id) + else joined.push({ spawn, facts: rows[0] as WorkerLogFacts }) + } + + if ( + missingSpawnIds.length === 0 && + duplicateSpawnIds.length === 0 && + unmatchedRows.length === 0 && + ambiguousRows.length === 0 + ) { + return { joined, issue: null } + } + return { + joined, + issue: issue({ + code: 'worker-control-join-unavailable', + area: 'capture-integrity', + severity: 'high', + subject: 'worker-control-join', + claim: 'Worker control-log coverage does not match recorded child invocations', + detail: `${missingSpawnIds.length} spawn(s) have no log, ${duplicateSpawnIds.length} have duplicate logs, ${unmatchedRows.length} log row(s) match no spawn, and ${ambiguousRows.length} row(s) are ambiguous.`, + evidence: [ + evidence('workers/join/missing-spawn-ids', missingSpawnIds.slice(0, MAX_EXAMPLES)), + evidence('workers/join/duplicate-spawn-ids', duplicateSpawnIds.slice(0, MAX_EXAMPLES)), + evidence('workers/join/unmatched-source-rows', unmatchedRows.slice(0, MAX_EXAMPLES)), + evidence('workers/join/ambiguous-source-rows', ambiguousRows.slice(0, MAX_EXAMPLES)), + ], + recommendedAction: + 'Retain exactly one worker control row with workerId for every spawned invocation.', + metadata: { + assessment: 'unavailable', + missing_count: missingSpawnIds.length, + duplicate_count: duplicateSpawnIds.length, + unmatched_count: unmatchedRows.length, + ambiguous_count: ambiguousRows.length, + }, + }), + } +} + +export function sourceIssues( + source: SupervisorRunSources, + facts: SupervisorTreeFacts, + tree: SupervisorRunTree, +): SupervisorRunIntegrityIssue[] { + const out = sourceIdentityIssues(facts) + if (source.workers === null) { + const reason = source.workersMissingReason ?? 'worker control-log store was not captured' + out.push( + issue({ + code: 'worker-controls-unavailable', + area: 'capture-integrity', + severity: 'medium', + subject: 'steer-delivery', + claim: + 'Steer delivery integrity is unavailable because worker control logs were not captured', + detail: reason, + evidence: [evidence('sources/workers', null)], + recommendedAction: 'Retain each worker inbox and event stream.', + metadata: { assessment: 'unavailable', reason }, + }), + ) + return out + } + + const joined = joinWorkerControls(facts) + if (joined.issue !== null) out.push(joined.issue) + const rootCompleted = + tree.rootId !== null && + tree.nodes.filter((node) => node.rollout_id === tree.rootId).length === 1 && + tree.nodes.find((node) => node.rollout_id === tree.rootId)?.outcome.is_completed === true + for (const row of joined.joined) out.push(...steerIssues(row, rootCompleted)) + return out +} diff --git a/src/supervisor-run/integrity-steer.ts b/src/supervisor-run/integrity-steer.ts new file mode 100644 index 00000000..dec32bc5 --- /dev/null +++ b/src/supervisor-run/integrity-steer.ts @@ -0,0 +1,257 @@ +import { evidence, issue, MAX_EXAMPLES } from './integrity-issues' +import type { JoinedWorker } from './integrity-source' +import type { SupervisorRunIntegrityIssue } from './integrity-types' + +function idListDetail(ids: readonly string[]): string { + const examples = ids + .slice(0, MAX_EXAMPLES) + .map((id) => JSON.stringify(id)) + .join(', ') + return ids.length <= MAX_EXAMPLES + ? examples + : `${examples}, and ${ids.length - MAX_EXAMPLES} more` +} + +export function steerIssues( + joined: JoinedWorker, + rootCompleted: boolean, +): SupervisorRunIntegrityIssue[] { + const { facts } = joined + const subject = facts.workerId ?? joined.spawn.id + const prefix = `workers/${facts.sourceIndex}` + const out: SupervisorRunIntegrityIssue[] = [] + if ( + !facts.inboxCaptured || + !facts.eventsCaptured || + facts.inboxInvalidRows > 0 || + facts.eventsInvalidRows > 0 + ) { + const missing = [ + ...(facts.inboxCaptured ? [] : ['inbox']), + ...(facts.eventsCaptured ? [] : ['events']), + ] + const malformed = [ + ...(facts.inboxInvalidRows > 0 ? [`inbox:${facts.inboxInvalidRows}`] : []), + ...(facts.eventsInvalidRows > 0 ? [`events:${facts.eventsInvalidRows}`] : []), + ] + out.push( + issue({ + code: 'worker-controls-unavailable', + area: 'capture-integrity', + severity: 'medium', + subject, + claim: + 'Steer delivery integrity is unavailable for a worker with missing control artifacts', + detail: + missing.length > 0 + ? `${JSON.stringify(subject)} is missing ${missing.join(' and ')}.` + : `${JSON.stringify(subject)} has malformed control rows (${malformed.join(', ')}).`, + evidence: [ + ...missing.map((artifact) => evidence(`${prefix}/${artifact}`, null)), + ...(facts.inboxInvalidRows > 0 + ? [evidence(`${prefix}/inbox/malformed-rows/count`, facts.inboxInvalidRows)] + : []), + ...(facts.eventsInvalidRows > 0 + ? [evidence(`${prefix}/events/malformed-rows/count`, facts.eventsInvalidRows)] + : []), + ], + recommendedAction: + 'Retain both control artifacts and repair malformed rows before correlating requests.', + metadata: { + assessment: 'unavailable', + missing_artifacts: missing, + malformed_rows: malformed, + }, + }), + ) + return out + } + + const requestsById = new Map() + const unidentifiedRequests = facts.steerRequests.filter((request) => request.requestId === null) + for (const request of facts.steerRequests) { + if (request.requestId === null) continue + const rows = requestsById.get(request.requestId) ?? [] + rows.push(request) + requestsById.set(request.requestId, rows) + } + if (unidentifiedRequests.length > 0) { + out.push( + issue({ + code: 'steer-request-id-unavailable', + area: 'capture-integrity', + severity: 'high', + subject, + claim: 'Some queued steer requests cannot be correlated because their ids are missing', + detail: `${unidentifiedRequests.length} queued request(s) omit id.`, + evidence: unidentifiedRequests + .slice(0, MAX_EXAMPLES) + .map((request) => evidence(`${prefix}/inbox/${request.row}/id`, null)), + recommendedAction: 'Assign one stable request id before appending a steer to the inbox.', + metadata: { assessment: 'unavailable', unavailable_count: unidentifiedRequests.length }, + }), + ) + } + const duplicateRequestIds = [...requestsById] + .filter(([, rows]) => rows.length > 1) + .map(([id]) => id) + if (duplicateRequestIds.length > 0) { + out.push( + issue({ + code: 'duplicate-steer-request-id', + area: 'control-integrity', + severity: 'high', + subject, + claim: 'A worker inbox contains duplicate steer request ids', + detail: `${duplicateRequestIds.length} duplicate id(s): ${idListDetail(duplicateRequestIds)}.`, + evidence: [ + evidence( + `${prefix}/inbox/duplicate-request-ids`, + duplicateRequestIds.slice(0, MAX_EXAMPLES), + ), + ], + recommendedAction: 'Generate a unique request id for every queued steer.', + }), + ) + } + + const acknowledgementsById = new Map() + const unidentifiedAcks = facts.steerAcknowledgements.filter((ack) => ack.requestId === null) + for (const ack of facts.steerAcknowledgements) { + if (ack.requestId === null) continue + const rows = acknowledgementsById.get(ack.requestId) ?? [] + rows.push(ack) + acknowledgementsById.set(ack.requestId, rows) + } + if (unidentifiedAcks.length > 0) { + out.push( + issue({ + code: 'steer-ack-id-unavailable', + area: 'capture-integrity', + severity: 'high', + subject, + claim: 'Some steer acknowledgements cannot be correlated because requestId is missing', + detail: `${unidentifiedAcks.length} down-leg event(s) omit requestId.`, + evidence: unidentifiedAcks + .slice(0, MAX_EXAMPLES) + .map((ack) => evidence(`${prefix}/events/${ack.row}/requestId`, null)), + recommendedAction: 'Echo the queued request id on every worker acknowledgement.', + metadata: { assessment: 'unavailable', unavailable_count: unidentifiedAcks.length }, + }), + ) + } + + const unknownAckIds = [...acknowledgementsById.keys()].filter((id) => !requestsById.has(id)) + if (unknownAckIds.length > 0) { + out.push( + issue({ + code: 'unknown-steer-ack', + area: 'control-integrity', + severity: 'high', + subject, + claim: 'A worker acknowledged steer request ids absent from its captured inbox', + detail: `${unknownAckIds.length} unknown id(s): ${idListDetail(unknownAckIds)}.`, + evidence: [ + evidence(`${prefix}/events/unknown-request-ids`, unknownAckIds.slice(0, MAX_EXAMPLES)), + ], + recommendedAction: + 'Reject acknowledgements whose request id was not queued for this worker.', + }), + ) + } + + const duplicateAckIds = [...acknowledgementsById] + .filter(([, rows]) => rows.length > 1) + .map(([id]) => id) + if (duplicateAckIds.length > 0) { + out.push( + issue({ + code: 'duplicate-steer-ack', + area: 'control-integrity', + severity: 'high', + subject, + claim: 'A worker emitted duplicate acknowledgements for steer request ids', + detail: `${duplicateAckIds.length} duplicated id(s): ${idListDetail(duplicateAckIds)}.`, + evidence: [ + evidence( + `${prefix}/events/duplicate-request-ids`, + duplicateAckIds.slice(0, MAX_EXAMPLES), + ), + ], + recommendedAction: 'Make steer acknowledgement append idempotent by request id.', + }), + ) + } + + const missingAckIds = [...requestsById.keys()].filter((id) => !acknowledgementsById.has(id)) + if (rootCompleted && missingAckIds.length > 0) { + out.push( + issue({ + code: 'missing-steer-ack', + area: 'control-integrity', + severity: 'high', + subject, + claim: 'A completed supervisor run has queued steer requests without acknowledgements', + detail: `${missingAckIds.length}/${requestsById.size} request id(s) are unacknowledged: ${idListDetail(missingAckIds)}.`, + evidence: [ + evidence( + `${prefix}/inbox/unacknowledged-request-ids`, + missingAckIds.slice(0, MAX_EXAMPLES), + ), + ], + recommendedAction: + 'Record one delivered or failed acknowledgement for every queued request id.', + }), + ) + } + + const statusUnavailableIds: string[] = [] + const notDeliveredIds: string[] = [] + for (const [id, acknowledgements] of acknowledgementsById) { + if (!requestsById.has(id) || acknowledgements.length !== 1) continue + const delivered = acknowledgements[0]?.delivered + if (delivered === null || delivered === undefined) statusUnavailableIds.push(id) + else if (!delivered) notDeliveredIds.push(id) + } + if (statusUnavailableIds.length > 0) { + out.push( + issue({ + code: 'steer-ack-status-unavailable', + area: 'capture-integrity', + severity: 'high', + subject, + claim: 'Some steer acknowledgements omit delivery status', + detail: `${statusUnavailableIds.length} id(s): ${idListDetail(statusUnavailableIds)}.`, + evidence: [ + evidence( + `${prefix}/events/status-unavailable-request-ids`, + statusUnavailableIds.slice(0, MAX_EXAMPLES), + ), + ], + recommendedAction: 'Record delivered: true or delivered: false on each acknowledgement.', + metadata: { assessment: 'unavailable', unavailable_count: statusUnavailableIds.length }, + }), + ) + } + if (notDeliveredIds.length > 0) { + out.push( + issue({ + code: 'steer-not-delivered', + area: 'control-integrity', + severity: 'high', + subject, + claim: 'Queued steer requests were acknowledged but not delivered', + detail: `${notDeliveredIds.length} id(s): ${idListDetail(notDeliveredIds)}.`, + evidence: [ + evidence( + `${prefix}/events/not-delivered-request-ids`, + notDeliveredIds.slice(0, MAX_EXAMPLES), + ), + ], + recommendedAction: + 'Retry or explicitly fail the control request instead of treating it as delivered.', + }), + ) + } + return out +} diff --git a/src/supervisor-run/integrity-tree.ts b/src/supervisor-run/integrity-tree.ts new file mode 100644 index 00000000..262c958b --- /dev/null +++ b/src/supervisor-run/integrity-tree.ts @@ -0,0 +1,530 @@ +import { validateRolloutLine } from '../rollout/schema' +import { evidence, issue, MAX_EXAMPLES } from './integrity-issues' +import type { SupervisorRunIntegrityEvidence, SupervisorRunIntegrityIssue } from './integrity-types' +import type { SupervisorRunTree } from './types' + +interface NodeRecord { + readonly index: number + readonly raw: Record + readonly errors: readonly string[] + readonly id: string | null + readonly parent: string | null | undefined + readonly runId: string | null + readonly role: string | null + readonly messages: readonly unknown[] | null + readonly profileId: string | null + readonly completed: boolean | null + readonly terminal: boolean | null +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value) +} + +function nonEmptyString(value: unknown): string | null { + return typeof value === 'string' && value.length > 0 ? value : null +} + +function parentValue(raw: Record): string | null | undefined { + if (raw.parent_rollout_id === null) return null + return nonEmptyString(raw.parent_rollout_id) ?? undefined +} + +function terminalValue(outcome: Record | null): boolean | null { + if (outcome === null) return null + if (typeof outcome.is_completed !== 'boolean' || typeof outcome.is_truncated !== 'boolean') { + return null + } + if (outcome.error !== null && typeof outcome.error !== 'string') return null + return outcome.is_completed || outcome.is_truncated || outcome.error !== null +} + +function nodeRecord(value: unknown, index: number): NodeRecord { + const raw = isRecord(value) ? value : {} + const policy = isRecord(raw.policy) ? raw.policy : null + const outcome = isRecord(raw.outcome) ? raw.outcome : null + return { + index, + raw, + errors: validateRolloutLine(value), + id: nonEmptyString(raw.rollout_id), + parent: parentValue(raw), + runId: nonEmptyString(raw.run_id), + role: nonEmptyString(raw.role), + messages: Array.isArray(raw.messages) ? raw.messages : null, + profileId: policy === null ? null : nonEmptyString(policy.agent_profile_cell_id), + completed: + outcome !== null && typeof outcome.is_completed === 'boolean' ? outcome.is_completed : null, + terminal: terminalValue(outcome), + } +} + +function metric( + record: NodeRecord, + keys: readonly string[], +): { key: string; value: number } | null { + const outcome = isRecord(record.raw.outcome) ? record.raw.outcome : null + const metrics = outcome !== null && isRecord(outcome.metrics) ? outcome.metrics : null + if (metrics === null) return null + for (const key of keys) { + const value = metrics[key] + if (typeof value === 'number' && Number.isFinite(value)) return { key, value } + } + return null +} + +function nodeEvidence( + record: NodeRecord, + path: string, + value: unknown, +): SupervisorRunIntegrityEvidence { + const key = record.id ?? `index-${record.index}` + return evidence(`nodes/${encodeURIComponent(key)}/${path}`, value) +} + +function unavailableNodeChecks(record: NodeRecord): string[] { + const checks: string[] = [] + if (record.id === null) checks.push('identity', 'root-membership', 'parent-link', 'cycle') + if (record.parent === undefined) checks.push('parent-link', 'cycle') + if (record.runId === null) checks.push('cross-run-parent') + if (record.role === null) checks.push('root-role') + if (!isRecord(record.raw.outcome)) checks.push('timing', 'terminal-state') + return [...new Set(checks)] +} + +interface TreeIssueUncertainty { + readonly unknownIdentityRows?: number + readonly uncertainIds?: ReadonlySet +} + +export function treeIssues( + tree: SupervisorRunTree, + uncertainty: TreeIssueUncertainty = {}, +): SupervisorRunIntegrityIssue[] { + const out: SupervisorRunIntegrityIssue[] = [] + const records = (tree.nodes as readonly unknown[]).map(nodeRecord) + const malformedSourceRows = tree.gaps + .filter((gap) => gap.code === 'source-row-malformed') + .reduce((count, gap) => count + (gap.count ?? 1), 0) + + for (const record of records) { + if (record.errors.length === 0) continue + out.push( + issue({ + code: 'node-schema-invalid', + area: 'capture-integrity', + severity: 'high', + subject: record.id ?? `node-${record.index}`, + claim: 'A supervisor-tree node fails canonical rollout validation', + detail: record.errors.join('; '), + evidence: [evidence(`nodes/${record.index}/validation`, record.errors)], + recommendedAction: + 'Fix the producer so every node validates as the existing tangle.rollout.v1 shape before analysis.', + metadata: { validation_errors: record.errors }, + }), + ) + const unavailableChecks = unavailableNodeChecks(record) + if (unavailableChecks.length === 0) continue + out.push( + issue({ + code: 'node-checks-unavailable', + area: 'capture-integrity', + severity: 'medium', + subject: record.id ?? `node-${record.index}`, + claim: 'Some integrity checks are unavailable for a malformed supervisor-tree node', + detail: `Unavailable checks: ${unavailableChecks.join(', ')}.`, + evidence: [evidence(`nodes/${record.index}/unavailable-checks`, unavailableChecks)], + recommendedAction: + 'Retain the named identity and relationship fields with their canonical types.', + metadata: { assessment: 'unavailable', checks: unavailableChecks }, + }), + ) + } + + const byId = new Map() + for (const record of records) { + if (record.id === null) continue + const matches = byId.get(record.id) ?? [] + matches.push(record) + byId.set(record.id, matches) + } + + for (const [rolloutId, matches] of byId) { + if (matches.length < 2) continue + out.push( + issue({ + code: 'duplicate-rollout-id', + area: 'control-integrity', + severity: 'critical', + subject: rolloutId, + claim: 'A rollout id is duplicated within one supervisor tree', + detail: `${matches.length} nodes claim rollout_id ${JSON.stringify(rolloutId)}, so references are ambiguous.`, + evidence: [ + evidence(`duplicates/${encodeURIComponent(rolloutId)}/count`, matches.length), + ...matches.slice(0, MAX_EXAMPLES).map((record) => + evidence(`nodes/${record.index}/identity`, { + rollout_id: record.id, + parent_rollout_id: record.parent, + role: record.role, + }), + ), + ], + recommendedAction: 'Give every invocation one stable rollout id and regenerate the tree.', + }), + ) + } + + const unknownIdentityRows = + records.filter((record) => record.id === null).length + + Math.max(malformedSourceRows, uncertainty.unknownIdentityRows ?? 0) + const unique = records.filter( + (record): record is NodeRecord & { readonly id: string } => + record.id !== null && byId.get(record.id)?.length === 1, + ) + + let root: NodeRecord | undefined + if (tree.rootId === null) { + out.push( + issue({ + code: 'root-unavailable', + area: 'capture-integrity', + severity: 'high', + subject: 'tree-root', + claim: 'The supervisor-tree root is unavailable', + detail: `rootId is null while ${records.length} node row(s) were captured.`, + evidence: [evidence('rootId', null)], + recommendedAction: 'Capture the root spawned event and preserve its rollout id as rootId.', + metadata: { assessment: 'unavailable', node_rows: records.length }, + }), + ) + } else { + const rootRows = byId.get(tree.rootId) ?? [] + if (rootRows.length === 1) { + root = rootRows[0] + } else if (rootRows.length === 0 && unknownIdentityRows > 0) { + out.push( + issue({ + code: 'node-checks-unavailable', + area: 'capture-integrity', + severity: 'high', + subject: tree.rootId, + claim: 'Root membership is unavailable because a malformed node lost its rollout id', + detail: `${unknownIdentityRows} malformed node row(s) have no usable rollout_id, so absence of root ${JSON.stringify(tree.rootId)} cannot be established.`, + evidence: [evidence('rootId', tree.rootId)], + recommendedAction: 'Retain rollout_id even when another node field fails validation.', + metadata: { assessment: 'unavailable', checks: ['root-membership'] }, + }), + ) + } else if (rootRows.length === 0) { + out.push( + issue({ + code: 'declared-root-missing', + area: 'control-integrity', + severity: 'critical', + subject: tree.rootId, + claim: 'The declared supervisor-tree root has no matching invocation node', + detail: `rootId ${JSON.stringify(tree.rootId)} is absent from ${records.length} node row(s).`, + evidence: [evidence('rootId', tree.rootId)], + recommendedAction: 'Restore the root rollout row or correct rootId.', + }), + ) + } + } + + if (root !== undefined) { + if (root.parent !== undefined && root.parent !== null) { + out.push( + issue({ + code: 'root-has-parent', + area: 'control-integrity', + severity: 'critical', + subject: root.id ?? tree.rootId ?? 'tree-root', + claim: 'The declared supervisor-tree root has a parent', + detail: `Root points to parent ${JSON.stringify(root.parent)}.`, + evidence: [nodeEvidence(root, 'parent_rollout_id', root.parent)], + recommendedAction: 'Record the root with parent_rollout_id null.', + }), + ) + } + if (root.role !== null && root.role !== 'supervisor') { + out.push( + issue({ + code: 'root-role-invalid', + area: 'control-integrity', + severity: 'high', + subject: root.id ?? tree.rootId ?? 'tree-root', + claim: 'The declared supervisor-tree root is not labeled as a supervisor', + detail: `Root has role ${JSON.stringify(root.role)}.`, + evidence: [nodeEvidence(root, 'role', root.role)], + recommendedAction: + 'Preserve the root invocation role as supervisor when minting rollout rows.', + }), + ) + } + } + + for (const record of unique) { + if (record.parent === undefined) continue + if (record.parent === null) { + if (record.id !== tree.rootId) { + out.push( + issue({ + code: 'detached-node', + area: 'control-integrity', + severity: 'critical', + subject: record.id, + claim: 'A non-root invocation is recorded as a detached parentless root', + detail: `Node ${JSON.stringify(record.id)} is parentless while rootId is ${JSON.stringify(tree.rootId)}.`, + evidence: [ + nodeEvidence(record, 'parent_rollout_id', null), + evidence('rootId', tree.rootId), + ], + recommendedAction: + 'Attach the invocation to its spawning parent or record a separate run.', + }), + ) + } + continue + } + + const parentRows = byId.get(record.parent) ?? [] + if (parentRows.length === 0) { + if (unknownIdentityRows > 0 || uncertainty.uncertainIds?.has(record.parent) === true) { + out.push( + issue({ + code: 'node-checks-unavailable', + area: 'capture-integrity', + severity: 'high', + subject: record.id, + claim: 'Parent membership is unavailable because a malformed node lost its rollout id', + detail: + uncertainty.uncertainIds?.has(record.parent) === true + ? `Parent ${JSON.stringify(record.parent)} appears only in a malformed source row, so its absence cannot be established.` + : `Parent ${JSON.stringify(record.parent)} cannot be matched while ${unknownIdentityRows} row(s) have no usable rollout_id.`, + evidence: [nodeEvidence(record, 'parent_rollout_id', record.parent)], + recommendedAction: 'Retain rollout_id on every row before assessing detached children.', + metadata: { assessment: 'unavailable', checks: ['parent-membership'] }, + }), + ) + } else { + out.push( + issue({ + code: 'parent-missing', + area: 'control-integrity', + severity: 'critical', + subject: record.id, + claim: 'An invocation points to a parent absent from the supervisor tree', + detail: `Node ${JSON.stringify(record.id)} names missing parent ${JSON.stringify(record.parent)}.`, + evidence: [nodeEvidence(record, 'parent_rollout_id', record.parent)], + recommendedAction: 'Retain the parent invocation or remove the dangling edge.', + }), + ) + } + continue + } + if (parentRows.length !== 1) continue + const parent = parentRows[0] as NodeRecord + + if (record.runId !== null && parent.runId !== null && record.runId !== parent.runId) { + out.push( + issue({ + code: 'cross-run-parent', + area: 'control-integrity', + severity: 'critical', + subject: record.id, + claim: 'A parent-child edge crosses supervisor-run identities', + detail: `Child run_id ${JSON.stringify(record.runId)} differs from parent run_id ${JSON.stringify(parent.runId)}.`, + evidence: [ + nodeEvidence(record, 'run_id', record.runId), + nodeEvidence(parent, 'run_id', parent.runId), + ], + recommendedAction: + 'Keep one run_id across a supervision tree or remove the cross-run edge.', + }), + ) + } + + const childStart = metric(record, ['spawned_at', 'started_at']) + const parentStart = metric(parent, ['spawned_at', 'started_at']) + if (childStart !== null && parentStart !== null && childStart.value < parentStart.value) { + out.push( + issue({ + code: 'child-before-parent', + area: 'control-integrity', + severity: 'high', + subject: record.id, + claim: 'A child invocation is timestamped before its parent started', + detail: `Child ${childStart.key} ${childStart.value} precedes parent ${parentStart.key} ${parentStart.value}.`, + evidence: [ + nodeEvidence(record, `outcome.metrics.${childStart.key}`, childStart.value), + nodeEvidence(parent, `outcome.metrics.${parentStart.key}`, parentStart.value), + ], + recommendedAction: 'Use one ordered clock for spawn events or retain clock metadata.', + }), + ) + } + + const parentClose = metric(parent, ['settled_at', 'completed_at', 'finished_at']) + if (childStart !== null && parentClose !== null && childStart.value > parentClose.value) { + out.push( + issue({ + code: 'child-after-parent-close', + area: 'control-integrity', + severity: 'critical', + subject: record.id, + claim: 'A child invocation is timestamped after its parent closed', + detail: `Child ${childStart.key} ${childStart.value} follows parent ${parentClose.key} ${parentClose.value}.`, + evidence: [ + nodeEvidence(record, `outcome.metrics.${childStart.key}`, childStart.value), + nodeEvidence(parent, `outcome.metrics.${parentClose.key}`, parentClose.value), + ], + recommendedAction: + 'Correct the causal timestamps or attach the child to its actual spawner.', + }), + ) + } + } + + for (const record of unique) { + const start = metric(record, ['spawned_at', 'started_at']) + const close = metric(record, ['settled_at', 'completed_at', 'finished_at']) + if (start === null || close === null || close.value >= start.value) continue + out.push( + issue({ + code: 'close-before-start', + area: 'control-integrity', + severity: 'high', + subject: record.id, + claim: 'An invocation is timestamped as closing before it started', + detail: `${start.key} is ${start.value} and ${close.key} is ${close.value}.`, + evidence: [ + nodeEvidence(record, `outcome.metrics.${start.key}`, start.value), + nodeEvidence(record, `outcome.metrics.${close.key}`, close.value), + ], + recommendedAction: 'Correct the event ordering or retain clock-domain metadata.', + }), + ) + } + + const emittedCycles = new Set() + const resolved = new Set() + for (const start of unique) { + if (resolved.has(start.id)) continue + const path: string[] = [] + const positions = new Map() + let current: string | null = start.id + while (current !== null) { + if (resolved.has(current)) break + const position = positions.get(current) + if (position !== undefined) { + const cycle = path.slice(position) + const key = [...cycle].sort().join('\u0000') + if (!emittedCycles.has(key)) { + emittedCycles.add(key) + out.push( + issue({ + code: 'parent-cycle', + area: 'control-integrity', + severity: 'critical', + subject: [...cycle].sort()[0] ?? 'cycle', + claim: 'The supervisor tree contains a parent cycle', + detail: `${cycle.join(' -> ')} -> ${cycle[0]}`, + evidence: cycle.slice(0, MAX_EXAMPLES).flatMap((id) => { + const rows = byId.get(id) + const record = rows?.length === 1 ? rows[0] : undefined + return record === undefined + ? [] + : [nodeEvidence(record, 'parent_rollout_id', record.parent)] + }), + recommendedAction: + 'Correct parent ids at ingestion; a cyclic graph is not a run tree.', + }), + ) + } + break + } + positions.set(current, path.length) + path.push(current) + const rows = byId.get(current) + if (rows?.length !== 1) break + const parent = rows[0]?.parent + if (parent === undefined) break + current = parent + } + for (const id of path) resolved.add(id) + } + + if (root?.completed === true) { + const missingTerminal = unique.filter( + (record) => record.id !== root?.id && record.terminal === false, + ) + if (missingTerminal.length > 0) { + out.push( + issue({ + code: 'child-terminal-unavailable', + area: 'capture-integrity', + severity: 'high', + subject: 'child-terminal-outcomes', + claim: 'A completed supervisor run has children with no captured terminal outcome', + detail: `${missingTerminal.length}/${Math.max(0, unique.length - 1)} child node(s) are neither completed, truncated, nor errored.`, + evidence: [ + evidence('children/missing-terminal/count', missingTerminal.length), + ...missingTerminal + .slice(0, MAX_EXAMPLES) + .map((record) => nodeEvidence(record, 'outcome', record.raw.outcome)), + ], + recommendedAction: 'Retain each child settlement or cancellation event.', + metadata: { assessment: 'unavailable', unavailable_count: missingTerminal.length }, + }), + ) + } + } + + const missingTranscripts = records.filter((record) => record.messages?.length === 0) + if (missingTranscripts.length > 0) { + out.push( + issue({ + code: 'transcript-unavailable', + area: 'capture-integrity', + severity: 'medium', + subject: 'decision-transcripts', + claim: 'Decision transcripts are unavailable for some supervisor-tree invocations', + detail: `${missingTranscripts.length}/${records.length} node row(s) have messages: [].`, + evidence: [ + evidence('capture/missing-transcripts/count', missingTranscripts.length), + ...missingTranscripts + .slice(0, MAX_EXAMPLES) + .map((record) => nodeEvidence(record, 'messages', [])), + ], + recommendedAction: 'Hydrate canonical messages from the retained session transcript.', + metadata: { assessment: 'unavailable', unavailable_count: missingTranscripts.length }, + }), + ) + } + + const missingProfiles = records.filter((record) => record.profileId === null) + if (missingProfiles.length > 0) { + out.push( + issue({ + code: 'profile-id-unavailable', + area: 'capture-integrity', + severity: 'medium', + subject: 'agent-profile-identity', + claim: + 'Canonical agent-profile identity is unavailable for some supervisor-tree invocations', + detail: `${missingProfiles.length}/${records.length} node row(s) omit policy.agent_profile_cell_id.`, + evidence: [ + evidence('capture/missing-agent-profile-cell-id/count', missingProfiles.length), + ...missingProfiles + .slice(0, MAX_EXAMPLES) + .map((record) => nodeEvidence(record, 'policy.agent_profile_cell_id', null)), + ], + recommendedAction: + 'Record policy.agent_profile_cell_id on every invocation at dispatch time.', + metadata: { assessment: 'unavailable', unavailable_count: missingProfiles.length }, + }), + ) + } + + return out +} diff --git a/src/supervisor-run/integrity-types.ts b/src/supervisor-run/integrity-types.ts new file mode 100644 index 00000000..43b384c6 --- /dev/null +++ b/src/supervisor-run/integrity-types.ts @@ -0,0 +1,75 @@ +import type { SupervisorRunTree } from './types' + +export const SUPERVISOR_RUN_INTEGRITY_SCHEMA = 'tangle.supervisor-run-integrity@1' + +export type SupervisorRunIntegritySeverity = 'critical' | 'high' | 'medium' | 'low' | 'info' + +export type SupervisorRunSourceOnlyCheckCode = + | 'journal-event-cardinality' + | 'worker-control-join' + | 'steer-ack-correlation' + +export type SupervisorRunIntegrityIssueCode = + | 'source-checks-unavailable' + | 'node-schema-invalid' + | 'node-checks-unavailable' + | 'root-unavailable' + | 'duplicate-rollout-id' + | 'declared-root-missing' + | 'root-has-parent' + | 'root-role-invalid' + | 'detached-node' + | 'parent-missing' + | 'cross-run-parent' + | 'child-before-parent' + | 'child-after-parent-close' + | 'close-before-start' + | 'parent-cycle' + | 'child-terminal-unavailable' + | 'transcript-unavailable' + | 'profile-id-unavailable' + | 'source-row-malformed' + | 'source-event-identity-unavailable' + | 'duplicate-spawn' + | 'orphan-terminal' + | 'duplicate-terminal' + | 'worker-controls-unavailable' + | 'worker-control-join-unavailable' + | 'steer-request-id-unavailable' + | 'steer-ack-id-unavailable' + | 'steer-ack-status-unavailable' + | 'duplicate-steer-request-id' + | 'missing-steer-ack' + | 'duplicate-steer-ack' + | 'unknown-steer-ack' + | 'steer-not-delivered' + +export interface SupervisorRunIntegrityEvidence { + readonly path: string + readonly value: unknown +} + +export interface SupervisorRunIntegrityIssue { + readonly code: SupervisorRunIntegrityIssueCode + readonly area: 'control-integrity' | 'capture-integrity' + readonly severity: SupervisorRunIntegritySeverity + readonly subject: string + readonly claim: string + readonly detail: string + readonly evidence: readonly SupervisorRunIntegrityEvidence[] + readonly recommendedAction: string + readonly metadata?: Readonly> +} + +export interface SupervisorRunIntegrityReport { + readonly schema: typeof SUPERVISOR_RUN_INTEGRITY_SCHEMA + readonly runRef: string + readonly input: 'sources' | 'tree' + readonly tree: SupervisorRunTree + readonly issues: readonly SupervisorRunIntegrityIssue[] +} + +export interface SupervisorRunIntegrityOptions { + /** Pins rollout provenance when source input must be projected into a tree. */ + readonly capturedAt?: string +} diff --git a/src/supervisor-run/integrity.test.ts b/src/supervisor-run/integrity.test.ts new file mode 100644 index 00000000..fdba4f46 --- /dev/null +++ b/src/supervisor-run/integrity.test.ts @@ -0,0 +1,482 @@ +import { performance } from 'node:perf_hooks' +import { describe, expect, it } from 'vitest' +import { ROLLOUT_SCHEMA, type RolloutLine } from '../rollout/schema' +import { analyzeSupervisorRunSources } from './analyze' +import { fixtureAt, fixtureJournal, fixtureSources, fixtureState, fixtureWorker } from './fixtures' +import { analyzeSupervisorRunIntegrity } from './integrity' +import { supervisorRunRolloutLines } from './rollout-nodes' +import { isUnavailable, type SupervisorRunSources, type SupervisorRunTree } from './types' + +const AT = '2026-07-29T18:00:00.000Z' + +function node( + rolloutId: string, + over: Partial & Pick, +): RolloutLine { + const { parent_rollout_id, role, ...rest } = over + return { + schema: ROLLOUT_SCHEMA, + rollout_id: rolloutId, + parent_rollout_id, + run_id: 'run-1', + experiment_id: null, + candidate_id: null, + generation: null, + candidate_index: null, + role, + task: { + suite: 'supervisor-run', + instance_id: 'instance-1', + split: 'search', + seed: 7, + rep: 0, + }, + policy: { + harness: 'test', + harness_version: '1', + model: 'test/model@1', + provider: 'test', + profile_commit: 'profile-commit', + agent_profile_cell_id: `profile:${rolloutId}`, + sampling: {}, + }, + messages: [{ role: 'user', content: `task for ${rolloutId}` }], + tool_defs: [], + outcome: { + reward: null, + reward_source: null, + verdict: null, + metrics: {}, + is_completed: false, + is_truncated: false, + error: null, + realness_gated: false, + }, + cost: { + usd: null, + tokens_in: null, + tokens_out: null, + tokens_reasoning: null, + cache_read: null, + cache_write: null, + wall_s: null, + }, + artifacts: { patch_path: null, run_dir: null, transcript_ref: `session:${rolloutId}` }, + provenance: { captured_at: AT, capture: 'settle-time' }, + ...rest, + } +} + +function tree(nodes: RolloutLine[], rootId: string | null = 'root'): SupervisorRunTree { + return { rootId, nodes, gaps: [] } +} + +function codes(input: SupervisorRunSources | SupervisorRunTree): string[] { + return analyzeSupervisorRunIntegrity(input, { capturedAt: AT }).issues.map((issue) => issue.code) +} + +function completedControlSource(inbox: string | null, events: string | null): SupervisorRunSources { + return fixtureSources({ + journal: fixtureJournal({ workers: [['worker', 1, 4]] }), + state: fixtureState({ startSec: 0, endSec: 5 }), + workers: [ + { + workerId: 'sup-1-test:s0', + label: 'worker', + inbox, + events, + patchBytes: 10, + }, + ], + }) +} + +describe('supervisor-run integrity', () => { + it('reports source-only checks unavailable for tree input instead of silently dropping them', () => { + const worker = fixtureWorker('worker', { + workerId: 'sup-1-test:s0', + startSec: 1, + finishSec: 4, + steers: ['change direction'], + }) + const events = (worker.events ?? '').replace('"delivered":true', '"delivered":false') + const source = completedControlSource(worker.inbox, events) + + expect(codes(source)).toContain('steer-not-delivered') + + const projected = supervisorRunRolloutLines(source, { capturedAt: AT }) + const projectedCodes = codes(projected) + expect(projectedCodes).toContain('source-checks-unavailable') + expect(projectedCodes).not.toContain('steer-not-delivered') + }) + + it('allows workers to spawn workers without inventing a supervisor-role violation', () => { + const root = node('root', { parent_rollout_id: null, role: 'supervisor' }) + const workerParent = node('worker-parent', { + parent_rollout_id: 'root', + role: 'worker', + }) + const child = node('child', { parent_rollout_id: 'worker-parent', role: 'worker' }) + + expect(codes(tree([root, workerParent, child]))).toEqual(['source-checks-unavailable']) + + const journal = [ + { kind: 'spawned', id: 'root', parent: null, label: 'root', role: 'supervisor' }, + { kind: 'spawned', id: 'worker-parent', parent: 'root', label: 'parent', role: 'worker' }, + { + kind: 'spawned', + id: 'child', + parent: 'worker-parent', + label: 'child', + role: 'worker', + }, + { kind: 'settled', id: 'worker-parent', status: 'done' }, + { kind: 'settled', id: 'child', status: 'done' }, + ] + .map((event) => JSON.stringify(event)) + .join('\n') + const sourceReport = analyzeSupervisorRunIntegrity( + fixtureSources({ + journal: `${journal}\n`, + state: fixtureState({ startSec: 0, endSec: 5 }), + workers: [ + { workerId: 'worker-parent', label: 'parent', inbox: '', events: '', patchBytes: null }, + { workerId: 'child', label: 'child', inbox: '', events: '', patchBytes: null }, + ], + }), + { capturedAt: AT }, + ) + expect(sourceReport.tree.nodes.map((row) => [row.rollout_id, row.role])).toEqual([ + ['root', 'supervisor'], + ['worker-parent', 'worker'], + ['child', 'worker'], + ]) + expect(sourceReport.issues.map((entry) => entry.code)).not.toContain( + 'worker-control-join-unavailable', + ) + }) + + it('preserves an explicitly invalid root role for integrity analysis', () => { + const journal = `${JSON.stringify({ + kind: 'spawned', + id: 'root', + parent: null, + label: 'root', + role: 'worker', + })}\n` + + expect(codes(fixtureSources({ journal, workers: [] }))).toContain('root-role-invalid') + }) + + it('retains malformed-row identity and does not claim its root or parent is absent', () => { + const validRoot = node('root', { parent_rollout_id: null, role: 'supervisor' }) + const malformedRoot = { + ...validRoot, + policy: { ...validRoot.policy, harness: 42 }, + } as unknown as RolloutLine + const child = node('child', { parent_rollout_id: 'root', role: 'worker' }) + const found = codes(tree([malformedRoot, child])) + + expect(found).toContain('node-schema-invalid') + expect(found).not.toContain('declared-root-missing') + expect(found).not.toContain('parent-missing') + }) + + it('joins duplicate display labels by stable workerId', () => { + const journal = [ + { kind: 'spawned', id: 'root', parent: null, label: 'root', role: 'supervisor' }, + { kind: 'spawned', id: 'worker-a', parent: 'root', label: 'same task', role: 'worker' }, + { kind: 'spawned', id: 'worker-b', parent: 'root', label: 'same task', role: 'worker' }, + ] + .map((event) => JSON.stringify(event)) + .join('\n') + const report = analyzeSupervisorRunIntegrity( + fixtureSources({ + journal: `${journal}\n`, + workers: [ + { workerId: 'worker-a', label: 'same task', inbox: '', events: '', patchBytes: null }, + { workerId: 'worker-b', label: 'same task', inbox: '', events: '', patchBytes: null }, + ], + }), + { capturedAt: AT }, + ) + + expect(report.issues.map((entry) => entry.code)).not.toContain( + 'worker-control-join-unavailable', + ) + }) + + it('marks relationship checks unavailable when a malformed row loses identity', () => { + const malformed = { policy: { harness: 42 } } as unknown as RolloutLine + const child = node('child', { parent_rollout_id: 'root', role: 'worker' }) + const found = codes(tree([malformed, child])) + + expect(found).toContain('node-checks-unavailable') + expect(found).not.toContain('declared-root-missing') + expect(found).not.toContain('parent-missing') + }) + + it('does not infer a missing parent from malformed journal rows', () => { + const journal = [ + { kind: 'spawned', id: 'root', parent: null, label: 'root', role: 'supervisor' }, + { kind: 'spawned', id: 'parent', parent: 42, label: 'parent', role: 'worker' }, + { kind: 'spawned', id: 'child', parent: 'parent', label: 'child', role: 'worker' }, + ] + .map((row) => JSON.stringify(row)) + .join('\n') + const report = analyzeSupervisorRunIntegrity( + fixtureSources({ + journal: `${journal}\n`, + workers: [{ workerId: 'child', label: 'child', inbox: '', events: '', patchBytes: null }], + }), + { capturedAt: AT }, + ) + const found = report.issues.map((entry) => entry.code) + + expect(found).toContain('source-row-malformed') + expect(found).toContain('node-checks-unavailable') + expect(found).not.toContain('parent-missing') + expect(report.tree.nodes.map((row) => row.rollout_id)).toEqual(['root', 'child']) + const projectedCodes = codes(report.tree) + expect(projectedCodes).toContain('source-checks-unavailable') + expect(projectedCodes).not.toContain('parent-missing') + }) + + it('does not infer missing relationships after an unreadable journal line', () => { + const journal = [ + JSON.stringify({ + kind: 'spawned', + id: 'root', + parent: null, + label: 'root', + role: 'supervisor', + }), + '{"kind":"spawned","id":"lost-parent"', + JSON.stringify({ + kind: 'spawned', + id: 'child', + parent: 'lost-parent', + label: 'child', + role: 'worker', + }), + ].join('\n') + const found = codes( + fixtureSources({ + journal: `${journal}\n`, + workers: [{ workerId: 'child', label: 'child', inbox: '', events: '', patchBytes: null }], + }), + ) + + expect(found).toContain('source-row-malformed') + expect(found).not.toContain('parent-missing') + }) + + it('correlates request ids and detects missing, duplicate, and unknown acknowledgements', () => { + const inbox = [ + { id: 'A', message: 'first' }, + { id: 'B', message: 'second' }, + ] + .map((row) => JSON.stringify(row)) + .join('\n') + const events = [ + { kind: 'started', at: fixtureAt(1) }, + { kind: 'message', direction: 'down', requestId: 'A', delivered: true }, + { kind: 'message', direction: 'down', requestId: 'A', delivered: true }, + { kind: 'message', direction: 'down', requestId: 'C', delivered: true }, + { kind: 'finished', at: fixtureAt(4), passed: true }, + ] + .map((row) => JSON.stringify(row)) + .join('\n') + + const report = analyzeSupervisorRunIntegrity( + completedControlSource(`${inbox}\n`, `${events}\n`), + { capturedAt: AT }, + ) + const found = report.issues.map((entry) => entry.code) + expect(found).toEqual( + expect.arrayContaining(['missing-steer-ack', 'duplicate-steer-ack', 'unknown-steer-ack']), + ) + expect(report.issues.find((entry) => entry.code === 'missing-steer-ack')?.detail).toContain( + '"B"', + ) + expect(report.issues.find((entry) => entry.code === 'duplicate-steer-ack')?.detail).toContain( + '"A"', + ) + expect(report.issues.find((entry) => entry.code === 'unknown-steer-ack')?.detail).toContain( + '"C"', + ) + const summary = analyzeSupervisorRunSources( + completedControlSource(`${inbox}\n`, `${events}\n`), + () => 0, + ) + expect(isUnavailable(summary.orchestration.steers)).toBe(true) + expect(isUnavailable(summary.orchestration.steersDelivered)).toBe(true) + }) + + it('never counts duplicate acknowledgement rows as multiple deliveries', () => { + const inbox = `${JSON.stringify({ id: 'A', message: 'first' })}\n` + const events = [ + { kind: 'message', direction: 'down', requestId: 'A', delivered: true }, + { kind: 'message', direction: 'down', requestId: 'A', delivered: true }, + ] + .map((row) => JSON.stringify(row)) + .join('\n') + const report = analyzeSupervisorRunSources( + completedControlSource(inbox, `${events}\n`), + () => 0, + ) + + expect(report.orchestration.steers).toBe(1) + expect(isUnavailable(report.orchestration.steersDelivered)).toBe(true) + expect(report.orchestration.steersByWorker).toEqual({ + unavailable: expect.stringContaining('acknowledgement request id duplicated'), + }) + expect(report.decision.workerEvidenceBytes).toBe(0) + }) + + it('treats malformed worker rows as unavailable instead of a clean zero', () => { + const report = analyzeSupervisorRunIntegrity(completedControlSource('{"id":"A"\n', ''), { + capturedAt: AT, + }) + + expect(report.issues.map((entry) => entry.code)).toContain('worker-controls-unavailable') + expect( + isUnavailable( + analyzeSupervisorRunSources(completedControlSource('{"id":"A"\n', ''), () => 0) + .orchestration.steers, + ), + ).toBe(true) + }) + + it('distinguishes captured-empty control artifacts from missing artifacts', () => { + const worker = fixtureWorker('worker', { + workerId: 'sup-1-test:s0', + startSec: 1, + finishSec: 4, + }) + const captured = completedControlSource('', worker.events) + const missing = completedControlSource(null, worker.events) + + expect(codes(captured)).not.toContain('worker-controls-unavailable') + expect(analyzeSupervisorRunSources(captured, () => 0).orchestration.steers).toBe(0) + expect(codes(missing)).toContain('worker-controls-unavailable') + expect(isUnavailable(analyzeSupervisorRunSources(missing, () => 0).orchestration.steers)).toBe( + true, + ) + }) + + it('does not turn reward gaps into unavailable control evidence', () => { + const root = node('root', { parent_rollout_id: null, role: 'supervisor' }) + const input: SupervisorRunTree = { + ...tree([root]), + gaps: [ + { + code: 'root-reward-unavailable', + message: 'no judge verdict for this run', + nodeId: 'root', + }, + ], + } + + expect(codes(input)).toEqual(['source-checks-unavailable']) + }) + + it('detects duplicate identities, missing parents, cross-run edges, and cycles', () => { + const root = node('root', { parent_rollout_id: null, role: 'supervisor' }) + const duplicateA = node('same', { parent_rollout_id: 'root', role: 'worker' }) + const duplicateB = node('same', { parent_rollout_id: 'root', role: 'worker' }) + const orphan = node('orphan', { + parent_rollout_id: 'missing', + role: 'worker', + run_id: 'run-2', + }) + const cycleA = node('cycle-a', { parent_rollout_id: 'cycle-b', role: 'worker' }) + const cycleB = node('cycle-b', { parent_rollout_id: 'cycle-a', role: 'worker' }) + const found = codes(tree([root, duplicateA, duplicateB, orphan, cycleA, cycleB])) + + expect(found).toEqual( + expect.arrayContaining(['duplicate-rollout-id', 'parent-missing', 'parent-cycle']), + ) + }) + + it('detects orphan and duplicate terminal journal events', () => { + const journal = [ + fixtureJournal({ workers: [['worker', 1, 4]] }).trim(), + JSON.stringify({ kind: 'settled', id: 'orphan', status: 'done', at: AT }), + JSON.stringify({ kind: 'settled', id: 'orphan', status: 'done', at: AT }), + ].join('\n') + const worker = fixtureWorker('worker', { + workerId: 'sup-1-test:s0', + startSec: 1, + finishSec: 4, + }) + + expect( + codes( + fixtureSources({ + journal, + state: fixtureState({ startSec: 0, endSec: 5 }), + workers: [worker], + }), + ), + ).toEqual(expect.arrayContaining(['orphan-terminal', 'duplicate-terminal'])) + }) + + it('keeps the 4,000-node source path bounded and near-linear', { timeout: 10_000 }, () => { + const source = (size: number): SupervisorRunSources => { + const journal: string[] = [ + JSON.stringify({ + kind: 'spawned', + id: 'root', + parent: null, + label: 'root', + role: 'supervisor', + at: fixtureAt(0), + }), + ] + const workers: NonNullable[number][] = [] + for (let index = 1; index < size; index += 1) { + const id = `node-${index}` + journal.push( + JSON.stringify({ + kind: 'spawned', + id, + parent: index === 1 ? 'root' : `node-${index - 1}`, + label: `task-${index}`, + role: 'worker', + at: fixtureAt(index), + }), + ) + workers.push({ + workerId: id, + label: `task-${index}`, + inbox: '', + events: '', + patchBytes: null, + }) + } + return fixtureSources({ + runRef: `chain-${size}`, + journal: `${journal.join('\n')}\n`, + state: fixtureState({ startSec: 0, endSec: size + 1 }), + workers, + }) + } + const measure = (input: SupervisorRunSources): number => { + const started = performance.now() + analyzeSupervisorRunIntegrity(input, { capturedAt: AT }) + return performance.now() - started + } + const median = (values: number[]): number => + [...values].sort((left, right) => left - right)[Math.floor(values.length / 2)] ?? Infinity + const small = source(1_000) + const large = source(4_000) + measure(small) + measure(large) + const smallMs = median([measure(small), measure(small), measure(small)]) + const largeMs = median([measure(large), measure(large), measure(large)]) + + expect(largeMs).toBeLessThan(smallMs * 7 + 100) + expect(largeMs).toBeLessThan(1_500) + }) +}) diff --git a/src/supervisor-run/integrity.ts b/src/supervisor-run/integrity.ts new file mode 100644 index 00000000..4e1bcf79 --- /dev/null +++ b/src/supervisor-run/integrity.ts @@ -0,0 +1,103 @@ +import { evidence, issue } from './integrity-issues' +import { sourceIssues } from './integrity-source' +import { treeIssues } from './integrity-tree' +import { + SUPERVISOR_RUN_INTEGRITY_SCHEMA, + type SupervisorRunIntegrityOptions, + type SupervisorRunIntegrityReport, + type SupervisorRunSourceOnlyCheckCode, +} from './integrity-types' +import { supervisorRunRolloutLinesFromFacts } from './rollout-nodes' +import { parseSupervisorTree } from './source-facts' +import type { SupervisorRunSources, SupervisorRunTree } from './types' + +export { + SUPERVISOR_RUN_INTEGRITY_SCHEMA, + type SupervisorRunIntegrityEvidence, + type SupervisorRunIntegrityIssue, + type SupervisorRunIntegrityIssueCode, + type SupervisorRunIntegrityOptions, + type SupervisorRunIntegrityReport, + type SupervisorRunIntegritySeverity, + type SupervisorRunSourceOnlyCheckCode, +} from './integrity-types' + +function isTree(input: SupervisorRunSources | SupervisorRunTree): input is SupervisorRunTree { + return 'nodes' in input && Array.isArray(input.nodes) +} + +function treeRunRef(tree: SupervisorRunTree): string { + const runIds = [ + ...new Set( + tree.nodes + .map((node) => node.run_id) + .filter((runId): runId is string => typeof runId === 'string' && runId.length > 0), + ), + ].sort() + if (runIds.length === 1) return runIds[0]! + if (runIds.length > 1) return `mixed-runs:${runIds.join(',')}` + return tree.rootId ?? 'unidentified-tree' +} + +/** + * Run deterministic integrity checks over source bytes or an already minted tree. + * Source input is parsed exactly once and reused by projection and source-only checks. + */ +export function analyzeSupervisorRunIntegrity( + input: SupervisorRunSources | SupervisorRunTree, + options: SupervisorRunIntegrityOptions = {}, +): SupervisorRunIntegrityReport { + if (isTree(input)) { + const sourceOnlyChecks: readonly SupervisorRunSourceOnlyCheckCode[] = [ + 'journal-event-cardinality', + 'worker-control-join', + 'steer-ack-correlation', + ] + return { + schema: SUPERVISOR_RUN_INTEGRITY_SCHEMA, + runRef: treeRunRef(input), + input: 'tree', + tree: input, + issues: [ + ...treeIssues(input), + issue({ + code: 'source-checks-unavailable', + area: 'capture-integrity', + severity: 'medium', + subject: 'source-only-control-checks', + claim: 'Source-only control checks are unavailable for SupervisorRunTree input', + detail: + 'Rollout rows do not retain journal event multiplicity or worker inbox and acknowledgement rows.', + evidence: [evidence('input/source-only-checks', sourceOnlyChecks)], + recommendedAction: + 'Pass SupervisorRunSources when journal and worker control checks are required.', + metadata: { assessment: 'unavailable', checks: sourceOnlyChecks }, + }), + ], + } + } + + const facts = parseSupervisorTree(input) + const tree = supervisorRunRolloutLinesFromFacts(input, facts, { + capturedAt: options.capturedAt, + }) + return { + schema: SUPERVISOR_RUN_INTEGRITY_SCHEMA, + runRef: input.runRef, + input: 'sources', + tree, + issues: [ + ...treeIssues(tree, { + unknownIdentityRows: + facts.journalInvalidRows + + facts.spawns.filter((spawn) => !spawn.valid && spawn.id.length === 0).length, + uncertainIds: new Set( + facts.spawns + .filter((spawn) => !spawn.valid && spawn.id.length > 0) + .map((spawn) => spawn.id), + ), + }), + ...sourceIssues(input, facts, tree), + ], + } +} diff --git a/src/supervisor-run/rollout-nodes.test.ts b/src/supervisor-run/rollout-nodes.test.ts index 4d159d84..f0c0caf5 100644 --- a/src/supervisor-run/rollout-nodes.test.ts +++ b/src/supervisor-run/rollout-nodes.test.ts @@ -73,7 +73,11 @@ describe('supervisorRunRolloutLines — the tree IS rollout rows', () => { expect(unfinished?.outcome.reward).toBeNull() expect(unfinished?.outcome.reward_source).toBeNull() expect(unfinished?.outcome.is_truncated).toBe(true) - expect(tree.gaps.some((g) => g.includes('fix-b'))).toBe(true) + expect( + tree.gaps.some( + (gap) => gap.code === 'child-reward-unavailable' && gap.nodeId?.endsWith(':s1'), + ), + ).toBe(true) }) it('carries per-node cost and the timeline the report is computed from', () => { @@ -128,7 +132,10 @@ describe('supervisorRunRolloutLines — the tree IS rollout rows', () => { it('returns no nodes and says why when the journal is absent', () => { const empty = supervisorRunRolloutLines(sources({ supRunDir: null })) expect(empty.nodes).toEqual([]) - expect(empty.gaps[0]).toContain('no supervisor run dir') + expect(empty.gaps[0]).toEqual({ + code: 'journal-unavailable', + message: 'no supervisor run dir; no nodes recoverable', + }) }) it('retains nested supervisor roles and exact structured scores across duplicate labels', () => { diff --git a/src/supervisor-run/rollout-nodes.ts b/src/supervisor-run/rollout-nodes.ts index 6bad9b56..2c08193e 100644 --- a/src/supervisor-run/rollout-nodes.ts +++ b/src/supervisor-run/rollout-nodes.ts @@ -29,8 +29,13 @@ import { type RolloutLine, type RolloutSplit, } from '../rollout/schema' -import { asRecord, parseJson, parseSupervisorTree } from './analyze' -import type { SupervisorRunSources, SupervisorRunTree, WorkerLogSource } from './types' +import { asRecord, parseJson, parseSupervisorTree, type SupervisorTreeFacts } from './source-facts' +import type { + SupervisorRunSources, + SupervisorRunTree, + SupervisorRunTreeGap, + WorkerLogSource, +} from './types' export interface SupervisorRolloutOptions { /** Benchmark/suite id for `task.suite`. Defaults to `'supervisor-run'`. */ @@ -84,16 +89,34 @@ export function supervisorRunRolloutLines( src: SupervisorRunSources, opts: SupervisorRolloutOptions = {}, ): SupervisorRunTree { - const gaps: string[] = [] - const tree = parseSupervisorTree(src) + return supervisorRunRolloutLinesFromFacts(src, parseSupervisorTree(src), opts) +} + +/** Mint rollout rows from an already parsed source without reading its JSONL again. */ +export function supervisorRunRolloutLinesFromFacts( + src: SupervisorRunSources, + tree: SupervisorTreeFacts, + opts: SupervisorRolloutOptions = {}, +): SupervisorRunTree { + const gaps: SupervisorRunTreeGap[] = [] if (src.journal === null) { - gaps.push( - `tree: ${src.supRunDir === null ? 'no supervisor run dir' : 'journal absent'} — no nodes recoverable`, - ) + gaps.push({ + code: 'journal-unavailable', + message: `${src.supRunDir === null ? 'no supervisor run dir' : 'journal absent'}; no nodes recoverable`, + }) return { rootId: null, nodes: [], gaps } } + const malformedSourceRows = + tree.journalInvalidRows + tree.spawns.filter((spawn) => !spawn.valid).length + if (malformedSourceRows > 0) { + gaps.push({ + code: 'source-row-malformed', + message: `${malformedSourceRows} malformed journal row(s) were excluded from the tree`, + count: malformedSourceRows, + }) + } - const state = parseJson(src.state) + const state = tree.state const result = parseJson(src.result) const judge = parseJson(src.judge) const stateResult = asRecord(state?.result) @@ -127,10 +150,17 @@ export function supervisorRunRolloutLines( // ── root: the supervisor invocation ──────────────────────────────────── if (rootId !== null) { + const rootSpawn = tree.spawns.find((spawn) => spawn.valid && spawn.id === rootId) const judgeResolved = typeof judge?.resolved === 'boolean' ? judge.resolved : null const judgeScore = typeof judge?.score === 'number' ? judge.score : null const reward = judgeScore ?? (judgeResolved === null ? null : judgeResolved ? 1 : 0) - if (reward === null) gaps.push('root.outcome.reward: no judge verdict for this run') + if (reward === null) { + gaps.push({ + code: 'root-reward-unavailable', + message: 'no judge verdict for this run', + nodeId: rootId, + }) + } const wallMs = tree.startedAt !== null && tree.completedAt !== null && tree.completedAt >= tree.startedAt ? tree.completedAt - tree.startedAt @@ -139,7 +169,7 @@ export function supervisorRunRolloutLines( ...base, rollout_id: rootId, parent_rollout_id: null, - role: 'supervisor', + role: rootSpawn?.role ?? 'supervisor', policy: { harness: opts.supervisorHarness ?? null, harness_version: null, @@ -212,7 +242,10 @@ export function supervisorRunRolloutLines( }, }) } else { - gaps.push('tree.root: no parentless `spawned` event in the journal') + gaps.push({ + code: 'root-spawn-unavailable', + message: 'no parentless spawned event in the journal', + }) } // ── workers: one node per spawn, keyed to its spawner ─────────────────── @@ -238,7 +271,11 @@ export function supervisorRunRolloutLines( const passed = close?.valid ?? facts?.passed ?? null const reward = score ?? (passed === null ? null : passed ? 1 : 0) if (reward === null) { - gaps.push(`child ${spawn.label}: no verify verdict (child logs absent or unfinished)`) + gaps.push({ + code: 'child-reward-unavailable', + message: `child ${JSON.stringify(spawn.label)} has no verify verdict`, + nodeId: spawn.id, + }) } const isSupervisor = spawn.role === 'supervisor' nodes.push({ @@ -338,7 +375,11 @@ export function supervisorRunRolloutLines( const invalid = nodes.filter((n) => !isRolloutLine(n)) if (invalid.length > 0) { - gaps.push(`${invalid.length} node(s) failed tangle.rollout.v1 validation`) + gaps.push({ + code: 'node-schema-invalid', + message: `${invalid.length} node(s) failed tangle.rollout.v1 validation`, + count: invalid.length, + }) } return { rootId, nodes, gaps } } diff --git a/src/supervisor-run/source-facts.ts b/src/supervisor-run/source-facts.ts new file mode 100644 index 00000000..92328927 --- /dev/null +++ b/src/supervisor-run/source-facts.ts @@ -0,0 +1,492 @@ +import type { SupervisorRunNodeRole, SupervisorRunSources } from './types' + +// --------------------------------------------------------------------------- +// Journal shapes — structurally parsed. The journal is the contract, not the type. +// --------------------------------------------------------------------------- + +interface JournalEvent { + kind?: unknown + id?: unknown + parent?: unknown + label?: unknown + role?: unknown + status?: unknown + verdict?: unknown + reason?: unknown + seq?: unknown + at?: unknown + spend?: unknown + spent?: unknown +} + +interface Tokens { + input: number + output: number + cacheRead: number + cacheWrite: number + /** False when the event carried no cache counters at all — not "zero cached". */ + hasCache: boolean +} + +interface SpendLike { + tokens: Tokens + usd: number +} + +export function asRecord(v: unknown): Record { + return typeof v === 'object' && v !== null ? (v as Record) : {} +} + +function num(v: unknown): number { + return typeof v === 'number' && Number.isFinite(v) ? v : 0 +} + +function readSpend(v: unknown): SpendLike { + const rec = asRecord(v) + const tok = asRecord(rec.tokens) + const cacheRead = tok.cacheRead ?? tok.cache_read + const cacheWrite = tok.cacheWrite ?? tok.cache_write + return { + tokens: { + input: num(tok.input), + output: num(tok.output), + cacheRead: num(cacheRead), + cacheWrite: num(cacheWrite), + hasCache: cacheRead !== undefined || cacheWrite !== undefined, + }, + usd: num(rec.usd), + } +} + +export function parseJsonl(text: string | null): Record[] { + return parseJsonlWithDiagnostics(text).rows +} + +interface ParsedJsonl { + readonly rows: Record[] + readonly invalidRows: number +} + +function parseJsonlWithDiagnostics(text: string | null): ParsedJsonl { + if (text === null) return { rows: [], invalidRows: 0 } + const out: Record[] = [] + let invalidRows = 0 + for (const line of text.split('\n')) { + const trimmed = line.trim() + if (!trimmed) continue + try { + const parsed: unknown = JSON.parse(trimmed) + if (typeof parsed === 'object' && parsed !== null && !Array.isArray(parsed)) { + out.push(parsed as Record) + } else { + invalidRows += 1 + } + } catch { + invalidRows += 1 + } + } + return { rows: out, invalidRows } +} + +export function parseJson(text: string | null): Record | null { + if (text === null) return null + try { + const parsed: unknown = JSON.parse(text) + return typeof parsed === 'object' && parsed !== null + ? (parsed as Record) + : null + } catch { + return null + } +} + +function ms(at: unknown): number | null { + if (typeof at !== 'string') return null + const t = Date.parse(at) + return Number.isFinite(t) ? t : null +} + +// --------------------------------------------------------------------------- +// The supervision tree, as parsed from the journal event stream. +// --------------------------------------------------------------------------- + +export interface SpawnRow { + /** Zero-based row among parsed journal objects. */ + sourceRow: number + id: string + parent: string | null + label: string + role: SupervisorRunNodeRole + at: number | null + /** False when identity, parentage, label, or role was malformed. */ + valid: boolean + invalidFields: readonly string[] +} + +export interface CloseRow { + id: string + kind: 'settled' | 'cancelled' + status: string | null + /** String verdict from legacy journals, or valid/invalid for a structured verdict. */ + verdict: string | null + /** Structured verdict validity, when recorded. */ + valid: boolean | null + /** Structured verdict score, preserved without boolean coercion. */ + score: number | null + /** The verdict exactly as the journal carried it. */ + rawVerdict: unknown | null + at: number | null + spend: SpendLike + /** False when the close event carried no spend object — not "spent nothing". */ + hasSpend: boolean +} + +export interface WorkerLogFacts { + /** Position in `SupervisorRunSources.workers`, retained for exact evidence paths. */ + sourceIndex: number + workerId: string | null + label: string + /** `false` means the artifact was missing; a captured empty artifact is `true`. */ + inboxCaptured: boolean + eventsCaptured: boolean + inboxInvalidRows: number + eventsInvalidRows: number + started: number | null + /** True once a `finished` event was seen — independent of whether its `at` parsed. */ + finished: boolean + finishedAt: number | null + passed: boolean | null + /** Numeric score exactly as the finished event recorded it. */ + score: number | null + /** `patchBytes` as reported by the finished event (not the patch file's size). */ + finishedPatchBytes: number | null + evidenceBytes: number + /** Unique queued request ids, or null when exact accounting is unavailable. */ + steersQueued: number | null + /** Unique matched request ids acknowledged as delivered, or null when unavailable. */ + steersDelivered: number | null + steersQueuedUnavailable: string | null + steersDeliveredUnavailable: string | null + questions: number + steerRequests: SteerRequestFact[] + steerAcknowledgements: SteerAcknowledgementFact[] +} + +export interface SteerRequestFact { + /** Durable inbox request id; null when the row did not retain one. */ + requestId: string | null + row: number +} + +export interface SteerAcknowledgementFact { + /** Worker event request id; null when the event cannot be correlated. */ + requestId: string | null + /** `null` means the acknowledgement omitted delivery status. */ + delivered: boolean | null + row: number +} + +/** + * The tree + timeline the report is computed from, exposed because the rollout-row + * minter needs exactly the same parse (one parser, two consumers). + */ +export interface SupervisorTreeFacts { + readonly rootId: string | null + readonly spawns: readonly SpawnRow[] + readonly closes: readonly CloseRow[] + readonly workerSpawns: readonly SpawnRow[] + readonly workerCloses: readonly CloseRow[] + readonly brain: { + tokensIn: number + tokensOut: number + cacheRead: number + cacheWrite: number + /** False when no metered event carried cache counters — not "nothing cached". */ + hasCache: boolean + usd: number + meteredCount: number + } + readonly workerLogs: ReadonlyMap + /** Every worker source row, including duplicate identities that a map cannot retain. */ + readonly workerLogRows: readonly WorkerLogFacts[] + /** Non-empty journal lines that were not JSON objects. */ + readonly journalInvalidRows: number + /** Parsed once with the journal and worker artifacts. */ + readonly state: Record | null + readonly startedAt: number | null + readonly completedAt: number | null +} + +interface VerdictFacts { + label: string | null + valid: boolean | null + score: number | null + raw: unknown | null +} + +function readVerdict(v: unknown): VerdictFacts { + if (typeof v === 'string') return { label: v, valid: null, score: null, raw: v } + if (typeof v !== 'object' || v === null) { + return { label: null, valid: null, score: null, raw: null } + } + const rec = v as Record + const valid = typeof rec.valid === 'boolean' ? rec.valid : null + return { + label: valid === null ? null : valid ? 'valid' : 'invalid', + valid, + score: typeof rec.score === 'number' && Number.isFinite(rec.score) ? rec.score : null, + raw: v, + } +} + +export function workerSourceKey( + worker: NonNullable[number], +): string { + return worker.workerId ?? worker.label +} + +export function parseSupervisorTree(src: SupervisorRunSources): SupervisorTreeFacts { + const parsedJournal = parseJsonlWithDiagnostics(src.journal) + const events = parsedJournal.rows + const state = parseJson(src.state) + + const spawns: SpawnRow[] = [] + const closes: CloseRow[] = [] + let brainIn = 0 + let brainOut = 0 + let brainCacheRead = 0 + let brainCacheWrite = 0 + let brainHasCache = false + let brainUsd = 0 + let meteredCount = 0 + let rootId: string | null = null + for (const [sourceRow, ev] of (events as JournalEvent[]).entries()) { + const kind = typeof ev.kind === 'string' ? ev.kind : '' + const id = typeof ev.id === 'string' ? ev.id : '' + if (kind === 'spawned') { + const parent = typeof ev.parent === 'string' && ev.parent.length > 0 ? ev.parent : null + const label = typeof ev.label === 'string' ? ev.label : '' + const invalidFields: string[] = [] + if (id.length === 0) invalidFields.push('id') + const parentOmittedForFirstRoot = ev.parent === undefined && rootId === null + if ( + ev.parent !== null && + !parentOmittedForFirstRoot && + !(typeof ev.parent === 'string' && ev.parent.length > 0) + ) { + invalidFields.push('parent') + } + if (label.length === 0) invalidFields.push('label') + if (ev.role !== undefined && ev.role !== 'supervisor' && ev.role !== 'worker') { + invalidFields.push('role') + } + const role: SupervisorRunNodeRole = + ev.role === 'supervisor' || ev.role === 'worker' + ? ev.role + : parent === null + ? 'supervisor' + : 'worker' + const valid = invalidFields.length === 0 + if (valid && parent === null && rootId === null) rootId = id + spawns.push({ + sourceRow, + id, + parent, + label, + role, + at: ms(ev.at), + valid, + invalidFields, + }) + } else if (kind === 'settled') { + const verdict = readVerdict(ev.verdict) + closes.push({ + id, + kind: 'settled', + status: typeof ev.status === 'string' ? ev.status : null, + verdict: verdict.label, + valid: verdict.valid, + score: verdict.score, + rawVerdict: verdict.raw, + at: ms(ev.at), + spend: readSpend(ev.spent), + hasSpend: asRecord(ev.spent).tokens !== undefined, + }) + } else if (kind === 'cancelled') { + closes.push({ + id, + kind: 'cancelled', + status: 'cancelled', + verdict: typeof ev.reason === 'string' ? ev.reason : null, + valid: null, + score: null, + rawVerdict: typeof ev.reason === 'string' ? ev.reason : null, + at: ms(ev.at), + spend: { + tokens: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, hasCache: false }, + usd: 0, + }, + hasSpend: false, + }) + } else if (kind === 'metered') { + const s = readSpend(ev.spend) + brainIn += s.tokens.input + brainOut += s.tokens.output + brainCacheRead += s.tokens.cacheRead + brainCacheWrite += s.tokens.cacheWrite + brainHasCache = brainHasCache || s.tokens.hasCache + brainUsd += s.usd + meteredCount += 1 + } + } + + const workerSpawns = spawns.filter((s) => s.valid && s.id !== rootId) + const workerIds = new Set(workerSpawns.map((s) => s.id)) + const workerCloses = closes.filter((c) => workerIds.has(c.id)) + + const workerLogs = new Map() + const workerLogRows: WorkerLogFacts[] = [] + for (const [sourceIndex, w] of (src.workers ?? []).entries()) { + const facts: WorkerLogFacts = { + sourceIndex, + workerId: w.workerId ?? null, + label: w.label, + inboxCaptured: w.inbox !== null, + eventsCaptured: w.events !== null, + inboxInvalidRows: 0, + eventsInvalidRows: 0, + started: null, + finished: false, + finishedAt: null, + passed: null, + score: null, + finishedPatchBytes: null, + evidenceBytes: 0, + steersQueued: null, + steersDelivered: null, + steersQueuedUnavailable: null, + steersDeliveredUnavailable: null, + questions: 0, + steerRequests: [], + steerAcknowledgements: [], + } + const parsedInbox = parseJsonlWithDiagnostics(w.inbox) + const parsedEvents = parseJsonlWithDiagnostics(w.events) + facts.inboxInvalidRows = parsedInbox.invalidRows + facts.eventsInvalidRows = parsedEvents.invalidRows + for (const [row, req] of parsedInbox.rows.entries()) { + if (typeof req.message !== 'string' || req.message.trim().length === 0) continue + facts.steerRequests.push({ + requestId: typeof req.id === 'string' && req.id.length > 0 ? req.id : null, + row, + }) + } + for (const [row, ev] of parsedEvents.rows.entries()) { + const kind = ev.kind + if (kind === 'message') { + if (ev.direction === 'up') { + facts.questions += 1 + continue + } + facts.steerAcknowledgements.push({ + requestId: + typeof ev.requestId === 'string' && ev.requestId.length > 0 ? ev.requestId : null, + delivered: typeof ev.delivered === 'boolean' ? ev.delivered : null, + row, + }) + } else if (kind === 'started') { + facts.started = ms(ev.at) + } else if (kind === 'finished') { + facts.finished = true + facts.finishedAt = ms(ev.at) + facts.passed = typeof ev.passed === 'boolean' ? ev.passed : null + facts.score = typeof ev.score === 'number' && Number.isFinite(ev.score) ? ev.score : null + facts.finishedPatchBytes = typeof ev.patchBytes === 'number' ? ev.patchBytes : null + facts.evidenceBytes = typeof ev.evidence === 'string' ? ev.evidence.length : 0 + } + } + const requestsById = new Map() + for (const request of facts.steerRequests) { + if (request.requestId === null) continue + requestsById.set(request.requestId, (requestsById.get(request.requestId) ?? 0) + 1) + } + const acknowledgementsById = new Map() + for (const acknowledgement of facts.steerAcknowledgements) { + if (acknowledgement.requestId === null) continue + const matches = acknowledgementsById.get(acknowledgement.requestId) ?? [] + matches.push(acknowledgement) + acknowledgementsById.set(acknowledgement.requestId, matches) + } + const queueProblems: string[] = [] + if (!facts.inboxCaptured) queueProblems.push('inbox absent') + if (!facts.eventsCaptured) queueProblems.push('events absent') + if (facts.inboxInvalidRows > 0) queueProblems.push('inbox contains malformed rows') + if (facts.eventsInvalidRows > 0) queueProblems.push('events contain malformed rows') + if (facts.steerRequests.some((request) => request.requestId === null)) { + queueProblems.push('queued request id missing') + } + if ([...requestsById.values()].some((count) => count > 1)) { + queueProblems.push('queued request id duplicated') + } + if (facts.steerAcknowledgements.some((acknowledgement) => acknowledgement.requestId === null)) { + queueProblems.push('acknowledgement request id missing') + } + if ([...acknowledgementsById.keys()].some((id) => !requestsById.has(id))) { + queueProblems.push('acknowledgement has no queued request') + } + if (queueProblems.length === 0) { + facts.steersQueued = requestsById.size + } else { + facts.steersQueuedUnavailable = [...new Set(queueProblems)].join('; ') + } + + const deliveryProblems = [...queueProblems] + if ([...acknowledgementsById.values()].some((rows) => rows.length > 1)) { + deliveryProblems.push('acknowledgement request id duplicated') + } + if (facts.steerAcknowledgements.some((acknowledgement) => acknowledgement.delivered === null)) { + deliveryProblems.push('acknowledgement delivery status missing') + } + if (deliveryProblems.length === 0) { + facts.steersDelivered = [...requestsById.keys()].filter((id) => { + const acknowledgement = acknowledgementsById.get(id) + return acknowledgement?.length === 1 && acknowledgement[0]?.delivered === true + }).length + } else { + facts.steersDeliveredUnavailable = [...new Set(deliveryProblems)].join('; ') + } + workerLogRows.push(facts) + workerLogs.set(workerSourceKey(w), facts) + } + + const startedAt = ms(state?.startedAt) ?? spawns[0]?.at ?? null + const completedAt = + ms(state?.completedAt) ?? + [...spawns.map((s) => s.at), ...closes.map((c) => c.at)].reduce( + (acc, t) => (t === null ? acc : acc === null ? t : Math.max(acc, t)), + null, + ) + + return { + rootId, + spawns, + closes, + workerSpawns, + workerCloses, + brain: { + tokensIn: brainIn, + tokensOut: brainOut, + cacheRead: brainCacheRead, + cacheWrite: brainCacheWrite, + hasCache: brainHasCache, + usd: brainUsd, + meteredCount, + }, + workerLogs, + workerLogRows, + journalInvalidRows: parsedJournal.invalidRows, + state, + startedAt, + completedAt, + } +} diff --git a/src/supervisor-run/types.ts b/src/supervisor-run/types.ts index 7aaacc54..4705b28a 100644 --- a/src/supervisor-run/types.ts +++ b/src/supervisor-run/types.ts @@ -60,7 +60,10 @@ export function showMeasured(v: Measured): str /** The two invocation roles a recursive supervision tree can contain. */ export type SupervisorRunNodeRole = 'supervisor' | 'worker' -/** One worker's logs, as read. `null` = the artifact did not exist. */ +/** + * One worker's logs, as read. `null` means the artifact did not exist; `''` + * means the artifact was captured and contained no rows. + */ export interface WorkerLogSource { /** * Stable journal node id. Readers should set this whenever their source has @@ -416,6 +419,22 @@ export interface SupervisorRunRollup { export interface SupervisorRunTree { readonly rootId: string | null readonly nodes: readonly RolloutLine[] - /** Why a node could not be recovered, in read order. */ - readonly gaps: readonly string[] + /** Typed reasons a rollout field could not be recovered, in read order. */ + readonly gaps: readonly SupervisorRunTreeGap[] +} + +/** Stable machine-readable reasons emitted while minting a supervisor tree. */ +export type SupervisorRunTreeGapCode = + | 'journal-unavailable' + | 'source-row-malformed' + | 'root-spawn-unavailable' + | 'root-reward-unavailable' + | 'child-reward-unavailable' + | 'node-schema-invalid' + +export interface SupervisorRunTreeGap { + readonly code: SupervisorRunTreeGapCode + readonly message: string + readonly nodeId?: string + readonly count?: number }