diff --git a/README.md b/README.md index 9bbda6c..dcec665 100644 --- a/README.md +++ b/README.md @@ -548,11 +548,13 @@ traces analyze --last 1 traces analyze --last 1 --llm --budget 0.50 traces analyze --last 1 --analyzer halo --analyzer-prompt "find token waste" traces analyze --all --last 20 --analyzer hodoscope +traces analyze --last 1 --analyzer prime traces analyze --last 1 --analyzer my-installed-command ``` HALO returns a diagnosis report. Hodoscope samples distinct behaviors and marks every sample `needs_review`. +Prime posts the full span projection to an OpenAI-compatible bridge (`TRACES_PRIME_BRIDGE_URL`, default `http://localhost:4181`) and returns validated findings with span evidence. An arbitrary command returns a raw report unless its SDK adapter explicitly parses a stricter output type. Read [Trace analysts](./docs/trace-analysts.md) for the output contract, a minimal custom analyst, and labeled benchmark setup. @@ -564,7 +566,7 @@ Read [Trace analysts](./docs/trace-analysts.md) for the output contract, a minim traces upload --since 24h --dry-run --redactor "my-pii-scrubber" ``` -In the SDK these are the `ExternalAnalyzer` and `Redactor` interfaces (`haloAnalyzer`, `hodoscopeAnalyzer`, `commandAnalyzer`, `commandRedactor`, `applyRedactor`, `runExternalAnalyzers`). +In the SDK these are the `ExternalAnalyzer` and `Redactor` interfaces (`haloAnalyzer`, `hodoscopeAnalyzer`, `primeAnalyzer`, `commandAnalyzer`, `commandRedactor`, `applyRedactor`, `runExternalAnalyzers`). See [`examples/external-engines.ts`](./examples/external-engines.ts). > The built-in agentic analysts (`--llm`) run on the Tangle router by default: set `TANGLE_API_KEY`. diff --git a/docs/trace-analysts.md b/docs/trace-analysts.md index 9d54c3b..0f46f88 100644 --- a/docs/trace-analysts.md +++ b/docs/trace-analysts.md @@ -68,9 +68,12 @@ traces analyze --all --last 20 --analyzer hodoscope # Run any installed command that accepts an OpenInference file path. traces analyze --last 1 --analyzer my-trace-tool + +# One-shot prime-RLM analysis through a local OpenAI-compatible bridge. +traces analyze --last 1 --analyzer prime --analyzer-prompt "find unsupported completion claims" ``` -`--model` is forwarded to the built-in model-assisted analysts, HALO, and Hodoscope. +`--model` is forwarded to the built-in model-assisted analysts, HALO, Hodoscope, and prime. HALO and Hodoscope use their own provider clients and credentials. The Hodoscope adapter pins version `0.2.4` and uses Python 3.11 through `uvx`. @@ -87,6 +90,41 @@ Raw JSON is still a `report`. Only an adapter that validates the full finding shape may return `findings`. Hodoscope always returns `discovery`, and each candidate has `status: 'needs_review'` plus its source trace and span. +### Prime engine + +`--analyzer prime` runs a one-shot analyst over the emitted OTLP artifact through an OpenAI-compatible bridge, such as cli-bridge's prime backend. +Unlike the `--llm` analysts, which drill into the trace with paged tools, prime has no REPL and no trace tools: the full span projection is inlined into a single prompt as JSON. +Oversized projections are re-rendered with a per-attribute character cap; if the projection still exceeds the inline budget the engine fails loud instead of silently dropping spans. + +Prerequisites: + +- A running bridge that accepts `POST /v1/chat/completions` and routes the configured model to the prime backend. +- No key handling here: the bridge owns provider credentials. + +Configuration (flags first, then environment, then defaults): + +| Setting | Source | Default | +| --- | --- | --- | +| Bridge root URL | `TRACES_PRIME_BRIDGE_URL` | `http://localhost:4181` | +| Model | `--model`, then `TRACES_PRIME_MODEL` | `prime/zai/glm-5.2` | +| Per-call deadline | `TRACES_PRIME_TIMEOUT_MS` | `1200000` (20 min) | +| Question | `--analyzer-prompt` | a general diagnosis question | + +```bash +traces analyze --last 1 --analyzer prime +TRACES_PRIME_BRIDGE_URL=http://localhost:4181 traces analyze --last 1 --analyzer prime --analyzer-prompt "find unsupported completion claims" +``` + +Output expectations: + +- The reply contract is one fenced JSON block of short strings citing span ids verbatim; a structurally malformed reply gets exactly one bounded repair turn that carries the malformed reply and the contract, never the trajectory. +- A still-malformed reply after repair is a failed result (`ok: false`) with the raw reply preserved for inspection; one failed engine never discards the other results. +- Valid rows become full `findings` with `trace://` span evidence, validated against the artifact like every other findings-kind engine; rows citing unknown or ambiguous span ids are rejected with a recorded reason. +- Zero findings from a well-formed reply is an honest null, not a failure. +- Bridge-reported token usage and call counts are recorded in the result output; cost stays uncaptured because this adapter has no pricing table. + +The scored prime-vs-dspy comparison — same trajectories, same scoring — lives in `@tangle-network/agent-eval`'s analyst benchmark (`runAnalystBenchmark`); this engine is the capture-side entry point, not the scoreboard. + ## Write one analyst An analyst receives a paged trace store and returns typed findings. diff --git a/examples/external-engines.ts b/examples/external-engines.ts index cdfe8f4..bbe43a1 100644 --- a/examples/external-engines.ts +++ b/examples/external-engines.ts @@ -10,6 +10,7 @@ import { commandRedactor, haloAnalyzer, hodoscopeAnalyzer, + primeAnalyzer, writeOtlpFile, } from '@tangle-network/traces' @@ -41,7 +42,19 @@ console.log( : `hodoscope unavailable: ${discovery.error}`, ) -// 3) External REDACTOR: scrub prose with your own PII model before upload. The +// 3) PRIME ENGINE: one-shot RLM through an OpenAI-compatible bridge (run +// cli-bridge's prime backend locally, or set TRACES_PRIME_BRIDGE_URL). +// The full span projection travels inline; findings come back with +// validated trace:// span evidence. +const prime = primeAnalyzer({ defaultPrompt: 'find unsupported completion claims' }) +const primeResult = await prime.analyze(otlp) +console.log( + primeResult.ok + ? `${primeResult.findings?.length ?? 0} prime finding(s)\n${primeResult.output}` + : `prime bridge unavailable: ${primeResult.error}`, +) + +// 4) External REDACTOR: scrub prose with your own PII model before upload. The // command reads a JSON array of strings on stdin and writes the scrubbed array // on stdout (a 3-line wrapper adapts openai/privacy-filter's `opf`). const redactor = commandRedactor({ command: 'my-pii-scrubber' }) diff --git a/src/analyst-engine-prime.ts b/src/analyst-engine-prime.ts new file mode 100644 index 0000000..505703d --- /dev/null +++ b/src/analyst-engine-prime.ts @@ -0,0 +1,591 @@ +/** + * Prime analyst engine — a bundled {@link ExternalAnalyzer} that runs a + * one-shot RLM over the OTLP-JSONL artifact through an OpenAI-compatible + * bridge (cli-bridge's prime backend, or any `/v1/chat/completions` endpoint). + * + * Protocol, in one pass with no tools: + * 1. The FULL span projection is inlined into the prompt as JSON — prime has + * no REPL and no trace tools, so unlike the dspy-rlm analysts (which + * drill via viewSpans/searchTrace) every fact must travel in the prompt. + * 2. The reply contract is one fenced ```json block of SHORT strings + * (long strings get corrupted in transport), citing span ids verbatim. + * 3. A structurally malformed reply gets ONE bounded stateless repair turn + * carrying the malformed reply plus the contract — never the trajectory — + * mirroring the typed-adapter repair the dspy arm gets, so both arms face + * the same structured-output affordance. + * 4. Cited span ids are resolved against the artifact; rows citing unknown + * or ambiguous ids are rejected with a recorded reason, never guessed. + * Zero findings from a well-formed reply is an honest null, not an error. + * + * Oversized traces: when the rendered projection exceeds the inline budget it + * is re-rendered with a per-attribute character cap (the prompt-side analog of + * the trace store's per-attribute byte cap); if still oversized the run fails + * loud — inline delivery is the only delivery, so silently dropping spans + * would understate the trajectory. The delivery decision is recorded in the + * result output. + * + * Usage (bridge-reported token counts and call count) is recorded in the + * result output; cost stays uncaptured because the bridge reports no priced + * cost and this adapter carries no pricing table. + */ + +import { createHash } from 'node:crypto' +import { request as httpRequest } from 'node:http' +import { request as httpsRequest } from 'node:https' +import type { AnalystFinding } from '@tangle-network/agent-eval/analyst' +import { spanEvidenceUri } from './external-analysis-validation.js' +import type { ExternalAnalysisResult, ExternalAnalyzer, ExternalAnalyzerOptions } from './external.js' +import { readJsonl } from './jsonl.js' + +const DEFAULT_BRIDGE_URL = 'http://localhost:4181' +const DEFAULT_MODEL = 'prime/zai/glm-5.2' +const DEFAULT_TIMEOUT_MS = 1_200_000 +const DEFAULT_MAX_INLINE_CHARS = 360_000 +const DEFAULT_PER_ATTRIBUTE_CHAR_CAP = 1_200 +const DEFAULT_QUESTION = + 'Diagnose this trajectory: identify concrete agent failures, wasted work, and unsupported claims.' +const MAX_FINDINGS = 10 +const MAX_SPAN_IDS_PER_FINDING = 8 +const SEVERITIES = new Set(['critical', 'high', 'medium', 'low', 'info']) +const ANALYST_ID = 'prime' + +export interface PrimeTransportRequest { + url: string + body: { model: string; messages: Array<{ role: 'user'; content: string }> } + /** Aborts on the analyzer's deadline AND the caller's signal. */ + signal: AbortSignal +} + +export interface PrimeTransportResponse { + status: number + text: string +} + +/** POSTs one JSON chat completion; injectable so tests run on a fake bridge. */ +export type PrimeTransport = (req: PrimeTransportRequest) => Promise + +export interface PrimeAnalyzerOptions { + /** Bridge ROOT url; the adapter calls `/v1/chat/completions`. + * Default: TRACES_PRIME_BRIDGE_URL, then http://localhost:4181. */ + bridgeUrl?: string + /** Model id the bridge routes on. Default: TRACES_PRIME_MODEL, then prime/zai/glm-5.2. */ + model?: string + /** Per-call deadline. Default: TRACES_PRIME_TIMEOUT_MS, then 1200000 (20 min — + * prime runs legitimately exceed 5 minutes). */ + timeoutMs?: number + /** Question asked when the caller passes no prompt. */ + defaultPrompt?: string + /** Rendered-projection budget before the per-attribute cap kicks in. */ + maxInlineChars?: number + /** Character cap applied to each attribute on the oversized re-render. */ + perAttributeCharCap?: number + /** Disable the bounded repair turn (one extra call on a malformed reply). */ + repair?: boolean + transport?: PrimeTransport +} + +interface PrimeUsage { + calls: number | null + tokens: { input: number | null; output: number | null } | null + estimated: boolean +} + +interface ProjectionDelivery { + mode: 'inline-json' + perAttributeCharCap: number | null + renderedChars: number +} + +interface ArtifactProjection { + rows: Array> + rendered: string + delivery: ProjectionDelivery + traceCount: number + /** span_id → every trace_id it appears under; >1 entry marks an ambiguous id. */ + spanTraces: Map> +} + +interface FindingRow { + spanIds: string[] + severity: AnalystFinding['severity'] + area: string + claim: string + action?: string + confidence: number +} + +function positiveIntegerEnv(name: string): number | undefined { + const raw = process.env[name] + if (raw === undefined || raw === '') return undefined + const value = Number(raw) + if (!Number.isSafeInteger(value) || value < 1) { + throw new TypeError(`${name} must be a positive integer, got '${raw}'`) + } + return value +} + +function finiteOrNull(value: unknown): number | null { + return typeof value === 'number' && Number.isFinite(value) ? value : null +} + +function normalizeUsage(raw: unknown): PrimeUsage { + if (!raw || typeof raw !== 'object') return { calls: null, tokens: null, estimated: false } + const usage = raw as Record + const input = finiteOrNull(usage.prompt_tokens) + const output = finiteOrNull(usage.completion_tokens) + return { + calls: finiteOrNull(usage.model_requests), + tokens: input === null && output === null ? null : { input, output }, + estimated: usage.estimated === true, + } +} + +/** Sum two usage receipts; a side with uncaptured counts poisons the sum to + * uncaptured rather than silently under-reporting. */ +function mergeUsage(a: PrimeUsage, b: PrimeUsage): PrimeUsage { + const calls = a.calls !== null && b.calls !== null ? a.calls + b.calls : null + const tokens = + a.tokens !== null && b.tokens !== null && + a.tokens.input !== null && b.tokens.input !== null && + a.tokens.output !== null && b.tokens.output !== null + ? { input: a.tokens.input + b.tokens.input, output: a.tokens.output + b.tokens.output } + : null + return { calls, tokens, estimated: a.estimated || b.estimated } +} + +function renderUsage(usage: PrimeUsage): string { + const calls = usage.calls === null ? 'uncaptured' : String(usage.calls) + const input = usage.tokens?.input ?? 'uncaptured' + const output = usage.tokens?.output ?? 'uncaptured' + const estimated = usage.estimated ? ' (bridge-estimated)' : '' + return `usage: calls=${calls} input_tokens=${input} output_tokens=${output}${estimated}; cost=uncaptured (no pricing table)` +} + +function capAttributeValue(value: unknown, cap: number): unknown { + const rendered = JSON.stringify(value) + if (rendered === undefined || rendered.length <= cap) return value + if (typeof value === 'string') return `${value.slice(0, cap)}…[truncated ${value.length - cap} chars]` + return `[omitted non-string attribute: ${rendered.length} JSON chars]` +} + +function capRowAttributes(row: Record, cap: number): Record { + const attributes = row.attributes + if (!attributes || typeof attributes !== 'object' || Array.isArray(attributes)) return row + const capped: Record = {} + for (const [key, value] of Object.entries(attributes as Record)) { + capped[key] = capAttributeValue(value, cap) + } + return { ...row, attributes: capped } +} + +async function projectArtifact( + otlpPath: string, + maxInlineChars: number, + perAttributeCharCap: number, + signal?: AbortSignal, +): Promise { + const rows: Array> = [] + const spanTraces = new Map>() + let index = 0 + for await (const value of readJsonl(otlpPath, { signal })) { + const label = `${otlpPath}:${index + 1}` + if (value === null || typeof value !== 'object' || Array.isArray(value)) { + throw new TypeError(`${label} must contain an object`) + } + const row = value as Record + if (typeof row.trace_id !== 'string' || row.trace_id.length === 0) { + throw new TypeError(`${label}.trace_id must be a non-empty string`) + } + if (typeof row.span_id !== 'string' || row.span_id.length === 0) { + throw new TypeError(`${label}.span_id must be a non-empty string`) + } + rows.push(row) + const traces = spanTraces.get(row.span_id) ?? new Set() + traces.add(row.trace_id) + spanTraces.set(row.span_id, traces) + index += 1 + } + if (rows.length === 0) throw new Error(`${otlpPath} contains no spans`) + const traceCount = new Set(rows.map((row) => row.trace_id as string)).size + + let projected = rows + let rendered = JSON.stringify(projected) + const delivery: ProjectionDelivery = { + mode: 'inline-json', + perAttributeCharCap: null, + renderedChars: rendered.length, + } + if (rendered.length > maxInlineChars) { + projected = rows.map((row) => capRowAttributes(row, perAttributeCharCap)) + rendered = JSON.stringify(projected) + delivery.perAttributeCharCap = perAttributeCharCap + delivery.renderedChars = rendered.length + } + if (rendered.length > maxInlineChars) { + throw new Error( + `trajectory renders to ${rendered.length} chars even at per-attribute cap ` + + `${perAttributeCharCap} (budget ${maxInlineChars}); inline delivery impossible`, + ) + } + return { rows: projected, rendered, delivery, traceCount, spanTraces } +} + +const OUTPUT_CONTRACT_LINES = [ + 'OUTPUT CONTRACT (you have no trace tools and no REPL):', + 'You are a one-shot analyst. Every fact you need is in the TRAJECTORY JSON below.', + 'Do not run shell commands, do not read or write files, do not use any tools.', + 'Reply with EXACTLY one fenced ```json code block and no other fenced block. The JSON object has exactly two fields:', + ' "answer": string — ONE short sentence (max 300 chars) summarizing your verdict.', + ' "findings": array (possibly empty) of finding rows, each exactly:', + ' {"span_ids": [string, ...],', + ' "severity": "critical"|"high"|"medium"|"low"|"info",', + ' "area": string (short kebab-case topic, max 40 chars),', + ' "claim": string (ONE short sentence, max 200 chars),', + ' "action": string (optional, ONE short imperative sentence, max 200 chars),', + ' "confidence": number 0..1}', + 'Do NOT include a rationale field. Keep every string SHORT — long strings get corrupted in transport and void your work.', + `Report at most ${MAX_FINDINGS} findings; a finding cites 1..${MAX_SPAN_IDS_PER_FINDING} span_ids, each copied VERBATIM from a span in the trajectory below.`, + '"findings" is [] only for a clean trajectory.', +] + +function buildPrompt(question: string, projection: ArtifactProjection): string { + return [ + `QUESTION: ${question}`, + '', + ...OUTPUT_CONTRACT_LINES, + '', + `TRAJECTORY (${projection.traceCount} trace(s); ${projection.rows.length} spans; full OpenInference span projection as JSON):`, + projection.rendered, + ].join('\n') +} + +function buildRepairPrompt(parseDefect: string, previousReply: string): string { + return [ + 'Your previous reply to a trace-analysis task was structurally malformed and could not be parsed', + `(${parseDefect}). Below is your previous reply verbatim. Re-emit ONLY the corrected JSON — one`, + 'fenced ```json block, no other text, no tools. The JSON object has exactly two fields:', + ' "answer": string (ONE short sentence, max 300 chars)', + ' "findings": array (possibly empty) of {"span_ids": [string, ...],', + ' "severity": "critical"|"high"|"medium"|"low"|"info", "area": string (max 40 chars),', + ' "claim": string (max 200 chars), "action": string (optional, max 200 chars), "confidence": number 0..1}', + 'No rationale field. Keep every string SHORT. Preserve the span ids and verdicts of your previous reply exactly; shorten prose freely.', + '', + 'PREVIOUS REPLY:', + previousReply, + ].join('\n') +} + +function tryParseObject(text: string): Record | null { + try { + const value: unknown = JSON.parse(text.trim()) + return value && typeof value === 'object' && !Array.isArray(value) + ? (value as Record) + : null + } catch { + return null + } +} + +function extractJsonObject(text: string): Record | null { + const direct = tryParseObject(text) + if (direct) return direct + const fenced = [...text.matchAll(/```(?:json)?\s*\n?([\s\S]*?)```/g)] + for (let i = fenced.length - 1; i >= 0; i -= 1) { + const candidate = tryParseObject(fenced[i]![1]!) + if (candidate) return candidate + } + const start = text.indexOf('{') + const end = text.lastIndexOf('}') + if (start >= 0 && end > start) { + const candidate = tryParseObject(text.slice(start, end + 1)) + if (candidate) return candidate + } + return null +} + +function parseDefectOf(parsed: Record | null): string | null { + if (parsed === null) return 'no parseable JSON object' + if (!Array.isArray(parsed.findings)) return 'JSON has no "findings" array' + return null +} + +function findingRowDefect(row: unknown, spanTraces: ReadonlyMap>): string | FindingRow { + if (!row || typeof row !== 'object' || Array.isArray(row)) return 'row is not an object' + const record = row as Record + if (!Array.isArray(record.span_ids) || record.span_ids.length === 0) { + return 'span_ids must be a non-empty array' + } + const spanIds: string[] = [] + for (const id of record.span_ids) { + if (typeof id !== 'string' || id.length === 0) return 'span_ids must contain non-empty strings' + const traces = spanTraces.get(id) + if (!traces) return `span_id '${id}' is not in the trajectory` + if (traces.size > 1) return `span_id '${id}' is ambiguous across ${traces.size} traces` + if (!spanIds.includes(id)) spanIds.push(id) + } + // The contract says 1..MAX; enforce it on the deduplicated list so repeated + // ids neither dodge the cap nor trip it spuriously. + if (spanIds.length > MAX_SPAN_IDS_PER_FINDING) { + return `span_ids cites ${spanIds.length} distinct spans (cap ${MAX_SPAN_IDS_PER_FINDING})` + } + if (typeof record.severity !== 'string' || !SEVERITIES.has(record.severity)) { + return 'severity outside the analyst severity enum' + } + if (typeof record.area !== 'string' || record.area.trim().length === 0 || record.area.length > 200) { + return 'area must be a 1-200 char string' + } + if (typeof record.claim !== 'string' || record.claim.trim().length === 0 || record.claim.length > 2000) { + return 'claim must be a 1-2000 char string' + } + if ( + record.action !== undefined && + (typeof record.action !== 'string' || record.action.trim().length === 0 || record.action.length > 2000) + ) { + return 'action must be a 1-2000 char string when present' + } + if ( + typeof record.confidence !== 'number' || + !Number.isFinite(record.confidence) || + record.confidence < 0 || + record.confidence > 1 + ) { + return 'confidence must be 0..1' + } + return { + spanIds, + severity: record.severity as AnalystFinding['severity'], + area: record.area.trim(), + claim: record.claim.trim(), + ...(typeof record.action === 'string' ? { action: record.action.trim() } : {}), + confidence: record.confidence, + } +} + +function toAnalystFinding( + row: FindingRow, + index: number, + spanTraces: ReadonlyMap>, + model: string, + producedAt: string, +): AnalystFinding { + const uris = row.spanIds.map((spanId) => { + const traces = spanTraces.get(spanId)! + return spanEvidenceUri([...traces][0]!, spanId) + }) + const findingId = `prime-${createHash('sha256') + .update(`${index}\n${row.claim}\n${uris.join('\n')}`) + .digest('hex') + .slice(0, 12)}` + return { + schema_version: '1.0.0', + finding_id: findingId, + analyst_id: ANALYST_ID, + produced_at: producedAt, + severity: row.severity, + area: row.area, + claim: row.claim, + evidence_refs: uris.map((uri) => ({ kind: 'span', uri })), + confidence: row.confidence, + ...(row.action ? { recommended_action: row.action } : {}), + metadata: { engine: ANALYST_ID, model }, + } +} + +/** Default transport: plain node:http/https, NOT fetch — undici's fixed 300s + * headers timeout kills prime calls that legitimately run longer; the + * analyzer's AbortSignal is the only deadline. */ +export const httpJsonTransport: PrimeTransport = ({ url, body, signal }) => { + const target = new URL(url) + if (target.protocol !== 'http:' && target.protocol !== 'https:') { + throw new TypeError(`bridge URL must be http: or https:, got ${target.protocol}`) + } + const requestFn = target.protocol === 'https:' ? httpsRequest : httpRequest + const payload = JSON.stringify(body) + return new Promise((resolve, reject) => { + const req = requestFn( + { + hostname: target.hostname, + port: target.port, + path: `${target.pathname}${target.search}`, + method: 'POST', + headers: { 'content-type': 'application/json', 'content-length': Buffer.byteLength(payload) }, + signal, + }, + (res) => { + const chunks: Buffer[] = [] + res.on('data', (chunk: Buffer) => chunks.push(chunk)) + res.on('end', () => + resolve({ status: res.statusCode ?? 0, text: Buffer.concat(chunks).toString('utf8') })) + res.on('error', reject) + }, + ) + req.on('error', reject) + req.end(payload) + }) +} + +interface BridgeReply { + content: string + usage: PrimeUsage +} + +class PrimeBridgeError extends Error {} + +async function callBridge( + transport: PrimeTransport, + url: string, + model: string, + content: string, + timeoutMs: number, + callerSignal: AbortSignal | undefined, + turn: string, +): Promise { + const controller = new AbortController() + const timer = setTimeout( + () => controller.abort(new Error(`${turn}bridge call exceeded ${timeoutMs}ms`)), + timeoutMs, + ) + const onCallerAbort = (): void => controller.abort(callerSignal!.reason) + callerSignal?.addEventListener('abort', onCallerAbort, { once: true }) + if (callerSignal?.aborted) onCallerAbort() + let response: PrimeTransportResponse + try { + response = await transport({ + url, + body: { model, messages: [{ role: 'user', content }] }, + signal: controller.signal, + }) + } catch (error) { + const reason = controller.signal.aborted ? controller.signal.reason : error + throw new PrimeBridgeError( + `${turn}bridge transport failure: ${reason instanceof Error ? reason.message : String(reason)}`, + ) + } finally { + clearTimeout(timer) + callerSignal?.removeEventListener('abort', onCallerAbort) + } + if (response.status !== 200) { + throw new PrimeBridgeError(`${turn}bridge HTTP ${response.status}: ${response.text.slice(0, 500)}`) + } + let parsed: unknown + try { + parsed = JSON.parse(response.text) + } catch { + throw new PrimeBridgeError(`${turn}bridge returned unparseable JSON (${response.text.length} bytes)`) + } + const record = parsed as { choices?: Array<{ message?: { content?: unknown } }>; usage?: unknown } + const replyContent = record.choices?.[0]?.message?.content + if (typeof replyContent !== 'string' || replyContent.length === 0) { + throw new PrimeBridgeError(`${turn}bridge reply carries no message content`) + } + return { content: replyContent, usage: normalizeUsage(record.usage) } +} + +function failure(output: string, error: string): ExternalAnalysisResult { + return { analyzer: ANALYST_ID, kind: 'report', ok: false, output, error } +} + +/** One-shot prime-RLM analysis over the emitted OTLP artifact, as a peer of + * `haloAnalyzer` / `hodoscopeAnalyzer` on the `--analyzer` registry. Requires + * a running OpenAI-compatible bridge with the prime backend; deterministic + * analysis and every other engine are unaffected when it is down (`ok:false` + * result, never a thrown run). */ +export function primeAnalyzer(opts: PrimeAnalyzerOptions = {}): ExternalAnalyzer { + const bridgeUrl = (opts.bridgeUrl ?? process.env.TRACES_PRIME_BRIDGE_URL ?? DEFAULT_BRIDGE_URL) + .replace(/\/+$/, '') + const model = opts.model ?? process.env.TRACES_PRIME_MODEL ?? DEFAULT_MODEL + const timeoutMs = opts.timeoutMs ?? positiveIntegerEnv('TRACES_PRIME_TIMEOUT_MS') ?? DEFAULT_TIMEOUT_MS + if (!Number.isSafeInteger(timeoutMs) || timeoutMs < 1) { + throw new RangeError('timeoutMs must be a positive safe integer') + } + const maxInlineChars = opts.maxInlineChars ?? DEFAULT_MAX_INLINE_CHARS + const perAttributeCharCap = opts.perAttributeCharCap ?? DEFAULT_PER_ATTRIBUTE_CHAR_CAP + if (!Number.isSafeInteger(maxInlineChars) || maxInlineChars < 1) { + throw new RangeError('maxInlineChars must be a positive safe integer') + } + if (!Number.isSafeInteger(perAttributeCharCap) || perAttributeCharCap < 1) { + throw new RangeError('perAttributeCharCap must be a positive safe integer') + } + const repairEnabled = opts.repair ?? true + const transport = opts.transport ?? httpJsonTransport + const url = `${bridgeUrl}/v1/chat/completions` + + return { + name: ANALYST_ID, + async analyze(otlpPath, analyzeOpts: ExternalAnalyzerOptions = {}) { + let projection: ArtifactProjection + try { + projection = await projectArtifact(otlpPath, maxInlineChars, perAttributeCharCap, analyzeOpts.signal) + } catch (error) { + return failure('', error instanceof Error ? error.message : String(error)) + } + const question = analyzeOpts.prompt ?? opts.defaultPrompt ?? DEFAULT_QUESTION + const deliveryLine = + `delivery: ${projection.delivery.mode} (${projection.delivery.renderedChars} chars, ` + + `per-attribute cap ${projection.delivery.perAttributeCharCap ?? 'none'})` + + let reply: BridgeReply + try { + reply = await callBridge( + transport, url, model, buildPrompt(question, projection), timeoutMs, analyzeOpts.signal, '') + } catch (error) { + if (!(error instanceof PrimeBridgeError)) throw error + return failure('', `${error.message}; ${deliveryLine}`) + } + let usage = reply.usage + let parsed = extractJsonObject(reply.content) + let parseDefect = parseDefectOf(parsed) + let repairAttempted = false + + if (parseDefect !== null && repairEnabled) { + repairAttempted = true + let repairReply: BridgeReply + try { + repairReply = await callBridge( + transport, url, model, buildRepairPrompt(parseDefect, reply.content), + timeoutMs, analyzeOpts.signal, 'repair-turn ') + } catch (error) { + if (!(error instanceof PrimeBridgeError)) throw error + return failure(reply.content, `${error.message}; ${renderUsage(usage)}`) + } + usage = mergeUsage(usage, repairReply.usage) + parsed = extractJsonObject(repairReply.content) + parseDefect = parseDefectOf(parsed) + } + if (parseDefect !== null) { + return failure( + reply.content, + `${parseDefect} in prime reply${repairAttempted ? ' even after the bounded repair turn' : ''}; ` + + renderUsage(usage), + ) + } + + const rows = (parsed!.findings as unknown[]).slice(0, MAX_FINDINGS) + const overflow = (parsed!.findings as unknown[]).length - rows.length + const producedAt = new Date().toISOString() + const rejected: string[] = [] + const findings: AnalystFinding[] = [] + rows.forEach((row, index) => { + const decoded = findingRowDefect(row, projection.spanTraces) + if (typeof decoded === 'string') { + rejected.push(`rejected[${index}]: ${decoded}`) + return + } + findings.push(toAnalystFinding(decoded, index, projection.spanTraces, model, producedAt)) + }) + + const answer = typeof parsed!.answer === 'string' ? parsed!.answer : null + const output = [ + ...(answer ? [`answer: ${answer}`] : []), + `findings: ${findings.length} mapped, ${rejected.length} rejected` + + (overflow > 0 ? `, ${overflow} over the ${MAX_FINDINGS}-finding cap dropped` : ''), + ...rejected, + deliveryLine, + `repair: ${repairAttempted ? 'attempted (succeeded)' : 'not needed'}`, + renderUsage(usage), + ...(findings.length === 0 ? ['zero findings — an honest null, not a failure'] : []), + ].join('\n') + return { analyzer: ANALYST_ID, kind: 'findings', ok: true, output, findings } + }, + } +} diff --git a/src/cli.ts b/src/cli.ts index 7b63902..c11a302 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -58,6 +58,7 @@ import { import { parseCorpusFlag } from './replay-corpus.js' import { commandAnalyzer, commandRedactor, haloAnalyzer } from './external.js' import { hodoscopeAnalyzer } from './hodoscope.js' +import { primeAnalyzer } from './analyst-engine-prime.js' import { type TraceEvidenceFormatOption, exportTraceEvidenceFile, writeTraceEvidenceExportFile } from './file-export.js' import { inspectSessionIndex, readSessionIndexFile, renderInspectionReport, writeInspectionReportFile } from './inspect.js' import { @@ -1115,6 +1116,8 @@ function externalAnalyzersFromArgs(args: Args) { ? haloAnalyzer({ defaultPrompt: args.analyzerPrompt, model: args.model }) : spec === 'hodoscope' ? hodoscopeAnalyzer({ summarizeModel: args.model }) + : spec === 'prime' + ? primeAnalyzer({ defaultPrompt: args.analyzerPrompt, ...(args.model ? { model: args.model } : {}) }) : commandAnalyzer({ name: spec, command: spec, args: (p, prompt) => (prompt ? [p, prompt] : [p]) }), ) } @@ -1627,7 +1630,10 @@ Options: --model Model for --llm, HALO, and Hodoscope (default for --llm: ${DEFAULT_ANALYST_MODEL}) --config investigate/improve/stream: JS config with analysts, liveAnalysts, or external analyzers --budget USD cap for agentic analysts - --analyzer analyze: also run halo, hodoscope, or an installed command (repeatable) + --analyzer analyze: also run halo, hodoscope, prime, or an installed command (repeatable) + prime posts the full span projection to an OpenAI-compatible bridge + (TRACES_PRIME_BRIDGE_URL, default http://localhost:4181; + TRACES_PRIME_MODEL, default prime/zai/glm-5.2; TRACES_PRIME_TIMEOUT_MS) --analyzer-prompt

analyze: prompt passed to external analyzers (default: diagnose) --verify-findings analyze: execute the findings as sandbox replay proofs and mark each VERIFIED (receipt path) or UNVERIFIABLE (reason). Needs diff --git a/src/index.ts b/src/index.ts index 04adbeb..8d5af3d 100644 --- a/src/index.ts +++ b/src/index.ts @@ -96,6 +96,7 @@ export * from './improvement.js' // runTraceInvestigation()/runTraceImprovement( // ── External engines (NOT bundled — shell out to tools you install) ──────── export * from './external.js' // haloAnalyzer / commandAnalyzer; commandRedactor export * from './hodoscope.js' // hodoscopeAnalyzer / writeHodoscopeInput +export * from './analyst-engine-prime.js' // primeAnalyzer — one-shot RLM over an OpenAI-compatible bridge // ── Live observation (event-driven; feed any system) ────────────────────── export * from './live.js' // streamSessions(), traceStreamEventsFromSpans(), semantic live findings diff --git a/tests/analyst-engine-prime.test.ts b/tests/analyst-engine-prime.test.ts new file mode 100644 index 0000000..6ef0913 --- /dev/null +++ b/tests/analyst-engine-prime.test.ts @@ -0,0 +1,376 @@ +import { createServer } from 'node:http' +import type { AddressInfo } from 'node:net' +import { describe, expect, it } from 'vitest' +import { + httpJsonTransport, + primeAnalyzer, + type PrimeTransport, + type PrimeTransportRequest, +} from '../src/analyst-engine-prime.js' +import { spanEvidenceUri } from '../src/external-analysis-validation.js' +import { runExternalAnalyzers } from '../src/external.js' +import { span, writeOtlpFile, type OtlpSpan } from '../src/otlp.js' + +function fixtureSpans(options: { contentChars?: number } = {}): OtlpSpan[] { + return [ + span({ + traceId: 'trace-one', + spanId: 'root', + name: 'session', + kind: 'AGENT', + startTime: '2026-01-01T00:00:00.000Z', + service: 'codex', + }), + span({ + traceId: 'trace-one', + spanId: 'llm-planning', + parentSpanId: 'root', + name: 'llm.turn', + kind: 'LLM', + startTime: '2026-01-01T00:00:01.000Z', + step: 1, + inputTokens: 100, + outputTokens: 20, + content: options.contentChars ? 'x'.repeat(options.contentChars) : 'I will inspect the repository.', + }), + span({ + traceId: 'trace-one', + spanId: 'tool-exec', + parentSpanId: 'llm-planning', + name: 'tool.exec_command', + kind: 'TOOL', + startTime: '2026-01-01T00:00:02.000Z', + step: 2, + tool: 'exec_command', + status: 'ERROR', + statusMessage: 'exit 1', + }), + ] +} + +function replyBody( + content: string, + usage: Record | undefined = { prompt_tokens: 100, completion_tokens: 20, model_requests: 1 }, +): string { + return JSON.stringify({ choices: [{ message: { content } }], usage }) +} + +const VALID_REPLY = [ + '```json', + JSON.stringify({ + answer: 'exec_command failed at step 2 and the failure went unhandled', + findings: [ + { + span_ids: ['tool-exec', 'llm-planning'], + severity: 'high', + area: 'tool-failure', + claim: 'exec_command exited 1 and the run continued without addressing it', + action: 'inspect the failing command before the next model turn', + confidence: 0.85, + }, + ], + }), + '```', +].join('\n') + +interface FakeCall { + request: PrimeTransportRequest +} + +function fakeBridge(responses: Array<{ status?: number; text: string } | Error>): { + transport: PrimeTransport + calls: FakeCall[] +} { + const calls: FakeCall[] = [] + const queue = [...responses] + const transport: PrimeTransport = async (request) => { + calls.push({ request }) + const next = queue.shift() + if (!next) throw new Error('fake bridge: no response queued') + if (next instanceof Error) throw next + return { status: next.status ?? 200, text: next.text } + } + return { transport, calls } +} + +describe('primeAnalyzer', () => { + it('maps a well-formed reply to grounded findings and records usage', async () => { + const spans = fixtureSpans() + const otlpPath = await writeOtlpFile(spans) + const { transport, calls } = fakeBridge([{ text: replyBody(VALID_REPLY) }]) + const analyzer = primeAnalyzer({ transport, model: 'prime/test-model' }) + + const [result] = await runExternalAnalyzers(otlpPath, [analyzer], { spans }) + expect(result!.ok).toBe(true) + expect(result!.kind).toBe('findings') + expect(result!.analyzer).toBe('prime') + expect(result!.findings).toHaveLength(1) + const finding = result!.findings![0]! + expect(finding.analyst_id).toBe('prime') + expect(finding.severity).toBe('high') + expect(finding.area).toBe('tool-failure') + expect(finding.recommended_action).toContain('inspect the failing command') + expect(finding.evidence_refs.map((ref) => ref.uri)).toEqual([ + spanEvidenceUri('trace-one', 'tool-exec'), + spanEvidenceUri('trace-one', 'llm-planning'), + ]) + expect(finding.metadata).toEqual({ engine: 'prime', model: 'prime/test-model' }) + expect(Number.isFinite(Date.parse(finding.produced_at))).toBe(true) + + expect(result!.output).toContain('answer: exec_command failed at step 2') + expect(result!.output).toContain('findings: 1 mapped, 0 rejected') + expect(result!.output).toContain('usage: calls=1 input_tokens=100 output_tokens=20') + expect(result!.output).toContain('cost=uncaptured') + expect(result!.output).toContain('delivery: inline-json') + expect(result!.output).toContain('per-attribute cap none') + + expect(calls).toHaveLength(1) + const prompt = calls[0]!.request.body.messages[0]!.content + expect(calls[0]!.request.url).toBe('http://localhost:4181/v1/chat/completions') + expect(calls[0]!.request.body.model).toBe('prime/test-model') + expect(prompt).toContain('TRAJECTORY (1 trace(s); 3 spans') + expect(prompt).toContain('"span_id":"tool-exec"') + expect(prompt).toContain('OUTPUT CONTRACT') + }) + + it('treats zero findings from a well-formed reply as an honest null', async () => { + const spans = fixtureSpans() + const otlpPath = await writeOtlpFile(spans) + const { transport } = fakeBridge([ + { text: replyBody('```json\n{"answer":"clean run","findings":[]}\n```') }, + ]) + const [result] = await runExternalAnalyzers(otlpPath, [primeAnalyzer({ transport })], { spans }) + expect(result!.ok).toBe(true) + expect(result!.kind).toBe('findings') + expect(result!.findings).toHaveLength(0) + expect(result!.output).toContain('zero findings — an honest null, not a failure') + }) + + it('runs one bounded repair turn carrying the malformed reply but never the trajectory', async () => { + const spans = fixtureSpans() + const otlpPath = await writeOtlpFile(spans) + const malformed = 'I looked at the trace and found a tool failure but forgot the JSON.' + const { transport, calls } = fakeBridge([ + { text: replyBody(malformed, { prompt_tokens: 100, completion_tokens: 20, model_requests: 1 }) }, + { text: replyBody(VALID_REPLY, { prompt_tokens: 50, completion_tokens: 10, model_requests: 1 }) }, + ]) + const [result] = await runExternalAnalyzers(otlpPath, [primeAnalyzer({ transport })], { spans }) + expect(result!.ok).toBe(true) + expect(result!.findings).toHaveLength(1) + expect(result!.output).toContain('repair: attempted (succeeded)') + expect(result!.output).toContain('usage: calls=2 input_tokens=150 output_tokens=30') + + expect(calls).toHaveLength(2) + const repairPrompt = calls[1]!.request.body.messages[0]!.content + expect(repairPrompt).toContain('PREVIOUS REPLY:') + expect(repairPrompt).toContain(malformed) + expect(repairPrompt).not.toContain('TRAJECTORY (') + expect(repairPrompt).not.toContain('"span_id":"tool-exec"') + }) + + it('fails the case when the reply is still malformed after the repair turn', async () => { + const spans = fixtureSpans() + const otlpPath = await writeOtlpFile(spans) + const { transport, calls } = fakeBridge([ + { text: replyBody('no json here') }, + { text: replyBody('still no json') }, + ]) + const [result] = await runExternalAnalyzers(otlpPath, [primeAnalyzer({ transport })], { spans }) + expect(result!.ok).toBe(false) + expect(result!.kind).toBe('report') + expect(result!.error).toContain('no parseable JSON object') + expect(result!.error).toContain('even after the bounded repair turn') + expect(result!.output).toContain('no json here') + expect(calls).toHaveLength(2) + }) + + it('makes a single call and fails loud when repair is disabled', async () => { + const spans = fixtureSpans() + const otlpPath = await writeOtlpFile(spans) + const { transport, calls } = fakeBridge([{ text: replyBody('prose only') }]) + const [result] = await runExternalAnalyzers( + otlpPath, + [primeAnalyzer({ transport, repair: false })], + { spans }, + ) + expect(result!.ok).toBe(false) + expect(result!.error).toContain('no parseable JSON object') + expect(result!.error).not.toContain('repair turn') + expect(calls).toHaveLength(1) + }) + + it('reports a non-200 bridge status as a failed result, never a thrown run', async () => { + const spans = fixtureSpans() + const otlpPath = await writeOtlpFile(spans) + const { transport } = fakeBridge([{ status: 502, text: 'bad gateway' }]) + const [result] = await runExternalAnalyzers(otlpPath, [primeAnalyzer({ transport })], { spans }) + expect(result!.ok).toBe(false) + expect(result!.error).toContain('bridge HTTP 502') + }) + + it('reports a transport failure as a failed result', async () => { + const spans = fixtureSpans() + const otlpPath = await writeOtlpFile(spans) + const { transport } = fakeBridge([new Error('connect ECONNREFUSED 127.0.0.1:4181')]) + const [result] = await runExternalAnalyzers(otlpPath, [primeAnalyzer({ transport })], { spans }) + expect(result!.ok).toBe(false) + expect(result!.error).toContain('bridge transport failure') + expect(result!.error).toContain('ECONNREFUSED') + }) + + it('rejects rows citing unknown spans or invalid fields while keeping valid rows', async () => { + const spans = fixtureSpans() + const otlpPath = await writeOtlpFile(spans) + const reply = [ + '```json', + JSON.stringify({ + answer: 'mixed quality reply', + findings: [ + { + span_ids: ['tool-exec'], + severity: 'medium', + area: 'tool-failure', + claim: 'the failing exec was never retried', + confidence: 0.6, + }, + { + span_ids: ['no-such-span'], + severity: 'high', + area: 'hallucination', + claim: 'cites a span that does not exist', + confidence: 0.9, + }, + { + span_ids: ['root'], + severity: 'catastrophic', + area: 'bad-severity', + claim: 'severity outside the enum', + confidence: 0.5, + }, + ], + }), + '```', + ].join('\n') + const { transport } = fakeBridge([{ text: replyBody(reply) }]) + const [result] = await runExternalAnalyzers(otlpPath, [primeAnalyzer({ transport })], { spans }) + expect(result!.ok).toBe(true) + expect(result!.findings).toHaveLength(1) + expect(result!.findings![0]!.claim).toBe('the failing exec was never retried') + expect(result!.output).toContain('findings: 1 mapped, 2 rejected') + expect(result!.output).toContain("rejected[1]: span_id 'no-such-span' is not in the trajectory") + expect(result!.output).toContain('rejected[2]: severity outside the analyst severity enum') + }) + + it('re-renders with the per-attribute cap when the projection is oversized', async () => { + const spans = fixtureSpans({ contentChars: 5_000 }) + const otlpPath = await writeOtlpFile(spans) + const { transport, calls } = fakeBridge([ + { text: replyBody('```json\n{"answer":"clean","findings":[]}\n```') }, + ]) + const analyzer = primeAnalyzer({ transport, maxInlineChars: 4_000, perAttributeCharCap: 200 }) + const [result] = await runExternalAnalyzers(otlpPath, [analyzer], { spans }) + expect(result!.ok).toBe(true) + expect(result!.output).toContain('per-attribute cap 200') + const prompt = calls[0]!.request.body.messages[0]!.content + expect(prompt).toContain('…[truncated 4800 chars]') + }) + + it('fails loud without calling the bridge when the capped projection is still oversized', async () => { + const spans = fixtureSpans({ contentChars: 5_000 }) + const otlpPath = await writeOtlpFile(spans) + const { transport, calls } = fakeBridge([]) + const analyzer = primeAnalyzer({ transport, maxInlineChars: 300, perAttributeCharCap: 50 }) + const [result] = await runExternalAnalyzers(otlpPath, [analyzer], { spans }) + expect(result!.ok).toBe(false) + expect(result!.error).toContain('inline delivery impossible') + expect(calls).toHaveLength(0) + }) + + it('aborts a call that exceeds the deadline, including on injected transports', async () => { + const spans = fixtureSpans() + const otlpPath = await writeOtlpFile(spans) + const transport: PrimeTransport = ({ signal }) => + new Promise((_resolve, reject) => { + signal.addEventListener('abort', () => reject(signal.reason), { once: true }) + }) + const analyzer = primeAnalyzer({ transport, timeoutMs: 25 }) + const [result] = await runExternalAnalyzers(otlpPath, [analyzer], { spans }) + expect(result!.ok).toBe(false) + expect(result!.error).toContain('bridge call exceeded 25ms') + }) + + it('drops findings over the cap and says so', async () => { + const spans = fixtureSpans() + const otlpPath = await writeOtlpFile(spans) + const rows = Array.from({ length: 12 }, (_v, i) => ({ + span_ids: ['tool-exec'], + severity: 'low', + area: 'noise', + claim: `finding number ${i}`, + confidence: 0.4, + })) + const reply = `\`\`\`json\n${JSON.stringify({ answer: 'noisy', findings: rows })}\n\`\`\`` + const { transport } = fakeBridge([{ text: replyBody(reply) }]) + const [result] = await runExternalAnalyzers(otlpPath, [primeAnalyzer({ transport })], { spans }) + expect(result!.ok).toBe(true) + expect(result!.findings).toHaveLength(10) + expect(result!.output).toContain('2 over the 10-finding cap dropped') + }) + + it('uses the caller prompt as the question when provided', async () => { + const spans = fixtureSpans() + const otlpPath = await writeOtlpFile(spans) + const { transport, calls } = fakeBridge([ + { text: replyBody('```json\n{"answer":"ok","findings":[]}\n```') }, + ]) + await runExternalAnalyzers(otlpPath, [primeAnalyzer({ transport })], { + spans, + prompt: 'find unsupported completion claims', + }) + expect(calls[0]!.request.body.messages[0]!.content) + .toContain('QUESTION: find unsupported completion claims') + }) +}) + +describe('httpJsonTransport', () => { + it('POSTs the chat body and returns status and text from a local server', async () => { + const received: Array<{ url: string; body: unknown }> = [] + const server = createServer((req, res) => { + const chunks: Buffer[] = [] + req.on('data', (chunk: Buffer) => chunks.push(chunk)) + req.on('end', () => { + received.push({ url: req.url ?? '', body: JSON.parse(Buffer.concat(chunks).toString('utf8')) }) + res.writeHead(200, { 'content-type': 'application/json' }) + res.end(replyBody('```json\n{"answer":"ok","findings":[]}\n```')) + }) + }) + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)) + const port = (server.address() as AddressInfo).port + try { + const response = await httpJsonTransport({ + url: `http://127.0.0.1:${port}/v1/chat/completions`, + body: { model: 'prime/test-model', messages: [{ role: 'user', content: 'hello' }] }, + signal: new AbortController().signal, + }) + expect(response.status).toBe(200) + expect(JSON.parse(response.text).choices[0].message.content).toContain('"answer"') + expect(received[0]!.url).toBe('/v1/chat/completions') + expect(received[0]!.body).toEqual({ + model: 'prime/test-model', + messages: [{ role: 'user', content: 'hello' }], + }) + } finally { + await new Promise((resolve, reject) => + server.close((error) => (error ? reject(error) : resolve()))) + } + }) + + it('rejects non-http(s) bridge urls', () => { + expect(() => + httpJsonTransport({ + url: 'ftp://localhost/v1/chat/completions', + body: { model: 'm', messages: [{ role: 'user', content: 'x' }] }, + signal: new AbortController().signal, + })).toThrow(/must be http: or https:/) + }) +})