From 175488aa449c70098bc26aaffeb439bd04651b45 Mon Sep 17 00:00:00 2001 From: Drew Stone Date: Tue, 8 Sep 2026 23:23:43 -0700 Subject: [PATCH] feat(traces): retain source evidence across adapters --- docs/trace-analysts.md | 47 +++++++ package.json | 10 +- pnpm-lock.yaml | 78 +++++------ src/adapters/amp.ts | 14 +- src/adapters/claude.ts | 20 ++- src/adapters/codex-exec.ts | 15 ++- src/adapters/codex.ts | 9 +- src/adapters/conversation.ts | 3 + src/adapters/copilot.ts | 7 +- src/adapters/factory.ts | 19 ++- src/adapters/forge.ts | 14 +- src/adapters/gemini.ts | 12 +- src/adapters/opencode.ts | 41 +++++- src/adapters/pi.ts | 16 ++- src/adapters/qwen.ts | 8 +- src/adapters/tool-io.ts | 12 +- src/analyze.ts | 10 +- src/bundle-source.ts | 236 ++++++++++++++++++++++++++++++++++ src/bundle.ts | 59 ++++++--- src/cli.ts | 18 +++ src/improvement.ts | 7 +- src/index.ts | 2 + src/integrity.ts | 3 +- src/json.ts | 13 +- src/jsonl.ts | 9 +- src/otlp.ts | 3 + src/redact.ts | 5 +- src/session-selection.ts | 17 ++- src/source-location.ts | 99 ++++++++++++++ src/types.ts | 2 + src/upload.ts | 4 +- tests/bundle-source.test.ts | 226 ++++++++++++++++++++++++++++++++ tests/source-location.test.ts | 163 +++++++++++++++++++++++ 33 files changed, 1089 insertions(+), 112 deletions(-) create mode 100644 src/bundle-source.ts create mode 100644 src/source-location.ts create mode 100644 tests/bundle-source.test.ts create mode 100644 tests/source-location.test.ts diff --git a/docs/trace-analysts.md b/docs/trace-analysts.md index 8c91a3c..102ff6d 100644 --- a/docs/trace-analysts.md +++ b/docs/trace-analysts.md @@ -305,3 +305,50 @@ The required review path is: ```text production trace -> finding -> reviewed feedback -> eval case -> candidate change -> comparison ``` + +## Read retained source fields + +Normalized conversation text and tool values remain capped. +Create a full bundle before the harness rotates its original files. +Explicitly authorize that bundle when analysis needs omitted text: + +```bash +traces bundle --session --out ./session-bundle +traces analyze --source-bundle ./session-bundle --llm +``` + +`investigate` and `improve` accept the same flag. +The SDK accepts `sourceBundle: { path, maxRecordBytes? }` in analysis and investigation options. +This authorization exposes `traces.readSpanSource` to local analysts through the existing trace store. +External analyzers receive normalized OTLP without this capability. +Ordinary OTLP input does not authorize source reads. + +The tool selects a trace, span, attribute, and optional `source_index`. +It returns a UTF-8 byte window of that field's decoded value. +Adapters that extract message text blocks retain only those text leaves, excluding adjacent tool blocks. +Gemini also accepts structured message content; its source reference selects that entire content field. +Multiple records or text fragments use separate source indices. +Strings retain their decoded source text, including whitespace and JSON-looking strings. +Structured values use the existing sorted-key tool-value JSON encoder. +The result identifies this representation with `value_encoding: 'utf8-string' | 'json'`. +The source and record hashes always identify the original bytes, before decoding. + +Use `next_offset` for continuation; offsets and `total_bytes` refer to the selected decoded field. +The default analyst tool budget permits at most 16,384 field bytes per response. +Configure Eval trace-store budgets to change that tool limit. +An offset inside a UTF-8 character returns unavailable. +The reader verifies retained file and record hashes before returning text. +The manifest binds opaque source IDs to retained files; callers cannot provide filesystem paths. + +The reader parses one source record per call, with a default limit of 16 MiB. +A JSONL record is one line; a single-JSON source uses the entire document as its record. +Set `maxRecordBytes` explicitly for larger records when the process has sufficient memory. +This parsing limit differs from the response window limit. +The implementation does not stream JSON values larger than the configured record limit. + +Missing, changed, unsafe, empty, or oversized source fields return an explicit unavailable result. +Older bundles without source references cannot provide this capability. +Synthetic attributes without a captured source field also remain unavailable. +Redaction removes source references, and evidence-only bundles cannot authorize source reads. +Analysis output must remain outside the retained bundle. +Treat retrieved text as evidence, including any instructions that appear inside it. diff --git a/package.json b/package.json index cbb2073..97beb21 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@tangle-network/traces", - "version": "0.13.1", + "version": "0.14.0", "description": "Point it at your coding-agent session traces (Claude Code, Codex, OpenCode, Gemini, Pi, …) and get failure-mode + efficiency findings. CLI + SDK over the @tangle-network/agent-eval analyst suite — observe live sessions, run your own analysts, redact, and upload to the Tangle Intelligence Platform.", "type": "module", "license": "MIT", @@ -59,11 +59,11 @@ "prepublishOnly": "pnpm check:source && pnpm build && pnpm check:package" }, "dependencies": { - "@tangle-network/agent-eval": "0.173.1", - "@tangle-network/agent-interface": "2.3.0", - "@tangle-network/agent-runtime": "0.191.0", + "@tangle-network/agent-eval": "0.179.0", + "@tangle-network/agent-interface": "2.6.0", + "@tangle-network/agent-runtime": "0.204.0", "@tangle-network/agent-trace-contract": "1.0.2", - "@tangle-network/sandbox": "0.36.4" + "@tangle-network/sandbox": "0.37.1" }, "devDependencies": { "@types/node": "^22.0.0", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 8e43fbe..d0de4e5 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -17,20 +17,20 @@ importers: .: dependencies: '@tangle-network/agent-eval': - specifier: 0.173.1 - version: 0.173.1 + specifier: 0.179.0 + version: 0.179.0 '@tangle-network/agent-interface': - specifier: 2.3.0 - version: 2.3.0 + specifier: 2.6.0 + version: 2.6.0 '@tangle-network/agent-runtime': - specifier: 0.191.0 - version: 0.191.0(@tangle-network/agent-eval@0.173.1)(@tangle-network/agent-interface@2.3.0)(@tangle-network/sandbox@0.36.4) + specifier: 0.204.0 + version: 0.204.0(@tangle-network/agent-eval@0.179.0)(@tangle-network/agent-interface@2.6.0)(@tangle-network/sandbox@0.37.1) '@tangle-network/agent-trace-contract': specifier: 1.0.2 version: 1.0.2 '@tangle-network/sandbox': - specifier: 0.36.4 - version: 0.36.4 + specifier: 0.37.1 + version: 0.37.1 devDependencies: '@types/node': specifier: ^22.0.0 @@ -651,20 +651,20 @@ packages: '@modelcontextprotocol/sdk': optional: true - '@tangle-network/agent-eval@0.173.1': - resolution: {integrity: sha512-BBLKy0quTvDK1L1CJ6+6RQSsYGxGsm/QLXXtM1ApJjwO/aX+Bes22vc3bW3A+POUTQxMXqNVNCLNQfvZ/XyifQ==} + '@tangle-network/agent-eval@0.179.0': + resolution: {integrity: sha512-rC3uAya4nxH0+8IISJrh8i6N4+wQbhRZiNnX291ChQBiSqYQG85JZ+yLYrgHQQt3ttJyIRtaOnfi8KUw8Vc4xQ==} engines: {node: '>=20.19.0'} hasBin: true - '@tangle-network/agent-interface@2.3.0': - resolution: {integrity: sha512-4BLA6WIgbK9ZtqWD8+sQvMjAMLhT2yDvRQwN/aH3KaWLe2SQ8aXT6B1+VmunM8rvWd6Q9iFvTGOx/l8TXjKtLA==} + '@tangle-network/agent-interface@2.6.0': + resolution: {integrity: sha512-tEByATif9oM5EEQB94Fytjt1Jtd70KxbWudKpwMQYyqXw2dcVFh2JTkeE8rIQP2F8KDLj9n5VHFkwg/ikbr7dg==} - '@tangle-network/agent-knowledge@13.0.1': - resolution: {integrity: sha512-G0y4M83ov1S5N0xWdVOWPhydzm8BEZ8lfuanwe7vGbtSKlzKMZKP6tYIUiff8neWbBNubx8fRS82vlE72Ev5bQ==} + '@tangle-network/agent-knowledge@15.0.2': + resolution: {integrity: sha512-8BJ4udy14I/U3qVUXrc3EI9vIpmxMkWLIv3Y8PVYVPa/ASjtGz2E42PtqhW2lhCbkCfY3Ar0D8ZcLFW49em93g==} engines: {node: '>=20.19.0'} hasBin: true peerDependencies: - '@tangle-network/agent-eval': '>=0.173.0 <0.174.0' + '@tangle-network/agent-eval': '>=0.174.0 <0.180.0' '@tangle-network/agent-interface': ^2.0.0 '@tangle-network/agent-profile-materialize@0.19.0': @@ -672,20 +672,20 @@ packages: peerDependencies: '@tangle-network/agent-interface': ^1.0.0 || ^2.0.0 - '@tangle-network/agent-runtime@0.191.0': - resolution: {integrity: sha512-DWdxaZIkUJfTgCPylG+8b9O9Zf/DD2kXYnAQC212L4JrvOKxWpzXblLLlRbxIwMCs3DaAy9Ll9xf8lx8XnfYWQ==} + '@tangle-network/agent-runtime@0.204.0': + resolution: {integrity: sha512-500IaPgnFZa4cQq++hgt6ijFW2uuhSSIdjVjqhcWKrU9cFL+HPQj4cXxu9z/08dTugjICSlgdGR21UGmaqXjnw==} engines: {node: '>=22.13.0'} hasBin: true peerDependencies: - '@tangle-network/agent-eval': '>=0.173.0 <0.174.0' - '@tangle-network/agent-interface': ^2.3.0 - '@tangle-network/sandbox': '>=0.36.4 <0.37.0' + '@tangle-network/agent-eval': '>=0.179.0 <0.180.0' + '@tangle-network/agent-interface': ^2.6.0 + '@tangle-network/sandbox': '>=0.36.4 <0.38.0' '@tangle-network/agent-trace-contract@1.0.2': resolution: {integrity: sha512-v7uMh56jkEp4vckevEU9xKsIatbs5dqzGPp69dFLSSXUVit0RP6VD6EANMXVlTCUk+6wVKBLHJx23XspVCEiIA==} - '@tangle-network/sandbox@0.36.4': - resolution: {integrity: sha512-1DGBpJPakISy/qe8BZK1m2PL5CrLJJ12apXsDWP/fUC+NjqmiI2fL3A7C2VJiGwYXz8Vz5/qFDW8QivRZCjqFg==} + '@tangle-network/sandbox@0.37.1': + resolution: {integrity: sha512-SQe1o2OczEg3szI15scqoGRAsqFH233S2TgggUqF8+gP0oTiUjLLi6xwVt0B/x1oHUEpTRmN137mwFcnNJx1sw==} peerDependencies: '@mastra/core': ^1.36.0 '@modelcontextprotocol/sdk': ^1.30.0 @@ -1649,15 +1649,15 @@ snapshots: '@tangle-network/agent-core@0.9.6': dependencies: - '@tangle-network/agent-interface': 2.3.0 + '@tangle-network/agent-interface': 2.6.0 zod: 4.5.4 - '@tangle-network/agent-eval@0.173.1': + '@tangle-network/agent-eval@0.179.0': dependencies: '@asteasolutions/zod-to-openapi': 9.1.0(zod@4.5.4) '@hono/node-server': 2.1.1(hono@4.13.5) '@tangle-network/agent-core': 0.9.6 - '@tangle-network/agent-interface': 2.3.0 + '@tangle-network/agent-interface': 2.6.0 '@tangle-network/agent-trace-contract': 1.0.2 hono: 4.13.5 linear-sum-assignment: 1.0.9 @@ -1666,33 +1666,33 @@ snapshots: transitivePeerDependencies: - '@modelcontextprotocol/sdk' - '@tangle-network/agent-interface@2.3.0': + '@tangle-network/agent-interface@2.6.0': dependencies: '@noble/hashes': 2.4.0 spdx-expression-parse: 5.0.0 zod: 4.5.4 - '@tangle-network/agent-knowledge@13.0.1(@tangle-network/agent-eval@0.173.1)(@tangle-network/agent-interface@2.3.0)': + '@tangle-network/agent-knowledge@15.0.2(@tangle-network/agent-eval@0.179.0)(@tangle-network/agent-interface@2.6.0)': dependencies: - '@tangle-network/agent-eval': 0.173.1 - '@tangle-network/agent-interface': 2.3.0 + '@tangle-network/agent-eval': 0.179.0 + '@tangle-network/agent-interface': 2.6.0 '@types/proper-lockfile': 4.1.4 proper-lockfile: 4.1.2 zod: 4.5.4 - '@tangle-network/agent-profile-materialize@0.19.0(@tangle-network/agent-interface@2.3.0)': + '@tangle-network/agent-profile-materialize@0.19.0(@tangle-network/agent-interface@2.6.0)': dependencies: - '@tangle-network/agent-interface': 2.3.0 + '@tangle-network/agent-interface': 2.6.0 - '@tangle-network/agent-runtime@0.191.0(@tangle-network/agent-eval@0.173.1)(@tangle-network/agent-interface@2.3.0)(@tangle-network/sandbox@0.36.4)': + '@tangle-network/agent-runtime@0.204.0(@tangle-network/agent-eval@0.179.0)(@tangle-network/agent-interface@2.6.0)(@tangle-network/sandbox@0.37.1)': dependencies: '@tangle-network/agent-core': 0.9.6 - '@tangle-network/agent-eval': 0.173.1 - '@tangle-network/agent-interface': 2.3.0 - '@tangle-network/agent-knowledge': 13.0.1(@tangle-network/agent-eval@0.173.1)(@tangle-network/agent-interface@2.3.0) - '@tangle-network/agent-profile-materialize': 0.19.0(@tangle-network/agent-interface@2.3.0) + '@tangle-network/agent-eval': 0.179.0 + '@tangle-network/agent-interface': 2.6.0 + '@tangle-network/agent-knowledge': 15.0.2(@tangle-network/agent-eval@0.179.0)(@tangle-network/agent-interface@2.6.0) + '@tangle-network/agent-profile-materialize': 0.19.0(@tangle-network/agent-interface@2.6.0) '@tangle-network/agent-trace-contract': 1.0.2 - '@tangle-network/sandbox': 0.36.4 + '@tangle-network/sandbox': 0.37.1 tar-stream: 3.2.1 transitivePeerDependencies: - '@modelcontextprotocol/sdk' @@ -1702,10 +1702,10 @@ snapshots: '@tangle-network/agent-trace-contract@1.0.2': {} - '@tangle-network/sandbox@0.36.4': + '@tangle-network/sandbox@0.37.1': dependencies: '@tangle-network/agent-core': 0.9.6 - '@tangle-network/agent-interface': 2.3.0 + '@tangle-network/agent-interface': 2.6.0 zod: 4.5.4 '@types/chai@5.2.3': diff --git a/src/adapters/amp.ts b/src/adapters/amp.ts index 39b2436..bf57add 100644 --- a/src/adapters/amp.ts +++ b/src/adapters/amp.ts @@ -11,13 +11,15 @@ * field names medium-conf; parse unverified against local data. */ +import { sourceOf, textSources } from '../source-location.js' + import { readdir, stat } from 'node:fs/promises' import { homedir } from 'node:os' import { basename, join } from 'node:path' import { isMissingPathError, readJsonFile } from '../json.js' import type { OtlpSpan } from '../otlp.js' import { span } from '../otlp.js' -import type { HarnessTraceAdapter, LocateOptions, SessionRef } from '../types.js' +import type { HarnessTraceAdapter, LocateOptions, ParseOptions, SessionRef } from '../types.js' import { CONTENT_CAP, capText, userPromptSpan } from './conversation.js' import { recordToolOutput, toolIoAttributes } from './tool-io.js' @@ -100,8 +102,8 @@ export class AmpAdapter implements HarnessTraceAdapter { return refs.sort((a, b) => b.mtimeMs - a.mtimeMs) } - async parse(ref: SessionRef): Promise { - const thread = await readJsonFile(ref.path) + async parse(ref: SessionRef, options: ParseOptions = {}): Promise { + const thread = await readJsonFile(ref.path, options) const traceId = thread.id ?? ref.sessionId const rootId = `root:${traceId}` const start = new Date(thread.created ?? 0).toISOString() @@ -133,6 +135,7 @@ export class AmpAdapter implements HarnessTraceAdapter { agent: SERVICE, step, content: prompt, + contentSource: textSources(m, 'content'), }), ) step += 1 @@ -156,6 +159,7 @@ export class AmpAdapter implements HarnessTraceAdapter { cacheWriteInputTokens: u?.cacheCreationInputTokens ?? null, step, content: textOf(m.content) || null, + contentSource: textSources(m, 'content'), }), ) step += 1 @@ -174,7 +178,7 @@ export class AmpAdapter implements HarnessTraceAdapter { agent: SERVICE, tool: b.name, step, - extra: toolIoAttributes({ input: b.input }), + extra: toolIoAttributes({ input: b.input, inputSource: sourceOf(b, 'input') }), }) spans.push(t) if (b.id) toolByUseId.set(b.id, t) @@ -183,7 +187,7 @@ export class AmpAdapter implements HarnessTraceAdapter { const t = toolByUseId.get(b.tool_use_id) if (t) { t.status = b.is_error === true ? { code: 'ERROR', message: 'tool reported error' } : { code: 'OK' } - recordToolOutput(t, b.content ?? b.output) + recordToolOutput(t, b.content ?? b.output, sourceOf(b, b.content != null ? 'content' : 'output')) } } } diff --git a/src/adapters/claude.ts b/src/adapters/claude.ts index c8e49b6..bc09710 100644 --- a/src/adapters/claude.ts +++ b/src/adapters/claude.ts @@ -52,6 +52,7 @@ import { } from './claude-workflow.js' import { capText, userPromptSpan } from './conversation.js' import { toolIoAttributes } from './tool-io.js' +import { appendSourceAttributes, sourceOf, textSources, SOURCE_ATTRIBUTE_PREFIX, type SourceReferences } from '../source-location.js' const SERVICE = 'claude-code' const EPOCH = new Date(0).toISOString() @@ -222,12 +223,14 @@ type ClaudeEventProjection = cachedInputTokens: number | null cacheWriteInputTokens: number | null content: string | null + contentSource?: SourceReferences tools: Array<{ id: string | null; name: string; attributes: Record }> } | { kind: 'user' timestamp: string prompt: string | null + contentSource?: SourceReferences isSidechain?: boolean isMeta?: boolean userType?: string | null @@ -241,6 +244,7 @@ function projectToolResult( isError: boolean, output: unknown, structuredWorkflowRun?: WorkflowRunReference, + outputSource?: SourceReferences, ): ToolResultProjection { const outputText = stringifyToolResult(output) const workflowRun = resolveWorkflowRunReference( @@ -251,7 +255,7 @@ function projectToolResult( return { toolUseId, isError, - attributes: toolIoAttributes({ output }), + attributes: toolIoAttributes({ output, outputSource }), message: outputText.slice(0, 500), ...(workflowRun ? { workflowRun } : {}), } @@ -263,7 +267,7 @@ function projectClaudeEvent(event: ClaudeEvent): ClaudeEventProjection { const tools: Array<{ id: string | null; name: string; attributes: Record }> = [] for (const block of asBlocks(event.message.content)) { if (block.type !== 'tool_use' || !block.name) continue - tools.push({ id: block.id || null, name: block.name, attributes: toolIoAttributes({ input: block.input }) }) + tools.push({ id: block.id || null, name: block.name, attributes: toolIoAttributes({ input: block.input, inputSource: sourceOf(block, 'input') }) }) } return { kind: 'assistant', @@ -275,6 +279,7 @@ function projectClaudeEvent(event: ClaudeEvent): ClaudeEventProjection { cachedInputTokens: event.message.usage?.cache_read_input_tokens ?? null, cacheWriteInputTokens: event.message.usage?.cache_creation_input_tokens ?? null, content: textOf(event.message.content) || null, + contentSource: textSources(event.message, 'content'), tools, } } @@ -296,6 +301,7 @@ function projectClaudeEvent(event: ClaudeEvent): ClaudeEventProjection { block.is_error === true, block.content, structuredWorkflowRun, + sourceOf(block, 'content'), )) } const prompt = textOf(event.message.content) @@ -303,6 +309,7 @@ function projectClaudeEvent(event: ClaudeEvent): ClaudeEventProjection { kind: 'user', timestamp, prompt: prompt || null, + contentSource: textSources(event.message, 'content'), ...(prompt ? { isSidechain: event.isSidechain === true, @@ -321,6 +328,8 @@ function projectClaudeEvent(event: ClaudeEvent): ClaudeEventProjection { event.attachment.toolUseID, typeof event.attachment.exitCode === 'number' && event.attachment.exitCode !== 0, event.attachment.stderr ?? '', + undefined, + sourceOf(event.attachment, 'stderr'), ), } } @@ -341,7 +350,10 @@ function indexWorkflowProjection( } function fingerprintClaudeEvent(event: ClaudeEventProjection): string { - return createHash('sha256').update(JSON.stringify(event)).digest('hex') + // Duplicate events can occupy different byte ranges without changing their meaning. + return createHash('sha256').update(JSON.stringify(event, (key, value: unknown) => + key === 'contentSource' || key.startsWith(SOURCE_ATTRIBUTE_PREFIX) ? undefined : value, + )).digest('hex') } function startsClaudeTask(projection: ClaudeEventProjection): boolean { @@ -446,6 +458,7 @@ function consumeClaudeEvent( ), }) mergeMessageContent(llmSpan, messageId, event.content, state) + if (event.content) appendSourceAttributes(llmSpan.attributes, 'content', event.contentSource) for (const tool of event.tools) { const existingTool = tool.id ? state.toolSpanByUseId.get(tool.id) : undefined @@ -493,6 +506,7 @@ function consumeClaudeEvent( agent: ctx.agent, step: state.step, content: event.prompt, + contentSource: event.contentSource, actor, }), ) diff --git a/src/adapters/codex-exec.ts b/src/adapters/codex-exec.ts index 643e09d..156746a 100644 --- a/src/adapters/codex-exec.ts +++ b/src/adapters/codex-exec.ts @@ -8,6 +8,8 @@ * without inventing durations. */ +import { sourceOf, textSources } from '../source-location.js' + import { sessionJsonlOptions } from '../integrity.js' import { readJsonl } from '../jsonl.js' import type { OtlpSpan } from '../otlp.js' @@ -250,6 +252,12 @@ export class CodexExecAdapter implements HarnessTraceAdapter { const id = requireString(item.id, ref.path, `${type}.id`) const name = toolName(type) const input = toolInput(type, item, ref.path) + const inputSource = type === 'command_execution' + ? ['command', 'cwd'].flatMap((field) => { + const reference = sourceOf(item, field) + return item[field] !== undefined && reference ? [reference] : [] + }) + : sourceOf(item, 'changes') const cwd = stringValue(item.cwd) if (!ref.cwd && cwd) ref.cwd = cwd const toolSpan = span({ @@ -265,7 +273,7 @@ export class CodexExecAdapter implements HarnessTraceAdapter { tool: name, step: step++, extra: { - ...toolIoAttributes({ input, argsCaptured: true }), + ...toolIoAttributes({ input, argsCaptured: true, inputSource }), 'traces.codex.exec_item_id': id, 'traces.codex.exec_item_type': type, 'traces.codex.exec_lifecycle': lifecycle, @@ -296,7 +304,7 @@ export class CodexExecAdapter implements HarnessTraceAdapter { pending.span.attributes['traces.codex.exec_exit_code'] = result.exitCode } if (type === 'command_execution') { - recordToolOutput(pending.span, typeof item.aggregated_output === 'string' ? item.aggregated_output : undefined) + recordToolOutput(pending.span, typeof item.aggregated_output === 'string' ? item.aggregated_output : undefined, sourceOf(item, 'aggregated_output')) } turn.pendingTools.delete(id) turn.completedItemIds.add(id) @@ -438,6 +446,7 @@ export class CodexExecAdapter implements HarnessTraceAdapter { agent: SERVICE, step: step++, content: text, + contentSource: textSources(item, 'text'), extra: { 'traces.codex.exec_item_id': id, 'traces.codex.exec_item_type': completedType, @@ -462,6 +471,7 @@ export class CodexExecAdapter implements HarnessTraceAdapter { agent: SERVICE, step: step++, content: capText(message), + contentSource: textSources(item, 'message'), extra: { 'traces.codex.exec_item_id': id, 'traces.codex.exec_item_type': completedType, @@ -503,6 +513,7 @@ export class CodexExecAdapter implements HarnessTraceAdapter { agent: SERVICE, step: step++, content: capText(message), + contentSource: textSources(event, 'message'), })) if (activeTurn) closeTurn('ERROR', timed.time, undefined, message) else { diff --git a/src/adapters/codex.ts b/src/adapters/codex.ts index 512ac62..660266c 100644 --- a/src/adapters/codex.ts +++ b/src/adapters/codex.ts @@ -14,6 +14,8 @@ * Shared by the codex-acp wrapper via alias (same rollout format). */ +import { sourceOf, textSources } from '../source-location.js' + import { readdir, stat } from 'node:fs/promises' import { homedir } from 'node:os' import { basename, join } from 'node:path' @@ -813,7 +815,7 @@ export class CodexAdapter implements HarnessTraceAdapter { tool: name, step, extra: { - ...toolIoAttributes({ input }), + ...toolIoAttributes({ input, inputSource: sourceOf(l.payload, l.payload.type === 'custom_tool_call' ? 'input' : 'arguments') }), 'traces.codex.call_type': l.payload.type, ...(name !== outerName ? { 'traces.codex.outer_tool_name': outerName } : {}), ...(nestedName ? { 'traces.codex.nested_tool_name': nestedName } : {}), @@ -848,7 +850,7 @@ export class CodexAdapter implements HarnessTraceAdapter { closeSpanAt(t, ts) t.status = status if (pollOutcome) t.attributes['traces.poll.outcome'] = pollOutcome - recordToolOutput(t, l.payload.output) + recordToolOutput(t, l.payload.output, sourceOf(l.payload, 'output')) const operation = t.attributes['traces.codex.agent_operation'] if (operation === 'spawn_agent') { setAgentSessionIds(t, spawnedSessionIds(l.payload.output)) @@ -926,6 +928,7 @@ export class CodexAdapter implements HarnessTraceAdapter { agent: SERVICE, step, content: text, + contentSource: textSources(l.payload, 'content'), extra: { 'traces.codex.agent_message_type': messageType, ...(author ? { 'traces.codex.agent_message_author': author } : {}), @@ -963,6 +966,7 @@ export class CodexAdapter implements HarnessTraceAdapter { parentSpanId: rootId, startTime: ts, content: prompt, + contentSource: textSources(l.payload, 'content'), service: SERVICE, agent: SERVICE, step, @@ -986,6 +990,7 @@ export class CodexAdapter implements HarnessTraceAdapter { agent: SERVICE, step, content: text, + contentSource: textSources(l.payload, 'content'), }), ) step += 1 diff --git a/src/adapters/conversation.ts b/src/adapters/conversation.ts index d872456..24ac7ed 100644 --- a/src/adapters/conversation.ts +++ b/src/adapters/conversation.ts @@ -8,6 +8,7 @@ import type { OtlpSpan } from '../otlp.js' import { span } from '../otlp.js' +import type { SourceReferences } from '../source-location.js' /** Max chars of conversation text kept per span — enough for prompt/response * analysis, bounded for storage + redaction cost. */ @@ -40,6 +41,7 @@ export type Actor = 'human' | 'agent' | 'subagent-spawn' | 'injected' | 'tool-re export const ACTOR_ATTR = 'tangle.actor' export interface UserPromptInput { + contentSource?: SourceReferences traceId: string spanId: string parentSpanId: string | null @@ -70,6 +72,7 @@ export function userPromptSpan(o: UserPromptInput): OtlpSpan { agent: o.agent ?? null, step: o.step, content: o.content, + contentSource: o.contentSource, extra: { [ACTOR_ATTR]: o.actor ?? 'human' }, }) } diff --git a/src/adapters/copilot.ts b/src/adapters/copilot.ts index 3d4d470..d240e1d 100644 --- a/src/adapters/copilot.ts +++ b/src/adapters/copilot.ts @@ -13,6 +13,8 @@ * parse unverified against local data. */ +import { sourceOf, textSources } from '../source-location.js' + import { readdir, stat } from 'node:fs/promises' import { homedir } from 'node:os' import { join } from 'node:path' @@ -123,6 +125,7 @@ export class CopilotAdapter implements HarnessTraceAdapter { outputTokens: d.outputTokens ?? null, step, content: typeof d.content === 'string' ? d.content.slice(0, 8000) : null, + contentSource: textSources(d, 'content'), }), ) lastLlm = llmId @@ -141,7 +144,7 @@ export class CopilotAdapter implements HarnessTraceAdapter { agent: SERVICE, tool: name, step, - extra: toolIoAttributes({ input: d.arguments }), + extra: toolIoAttributes({ input: d.arguments, inputSource: sourceOf(d, 'arguments') }), }) spans.push(t) toolByCallId.set(d.toolCallId, t) @@ -152,7 +155,7 @@ export class CopilotAdapter implements HarnessTraceAdapter { t.end_time = ts const err = d.success === false t.status = err ? { code: 'ERROR', message: (d.error?.message ?? '').slice(0, 500) } : { code: 'OK' } - recordToolOutput(t, d.output ?? d.result ?? d.error?.message) + recordToolOutput(t, d.output ?? d.result ?? d.error?.message, d.output != null ? sourceOf(d, 'output') : d.result != null ? sourceOf(d, 'result') : sourceOf(d.error, 'message')) } } } diff --git a/src/adapters/factory.ts b/src/adapters/factory.ts index cf9948d..a887710 100644 --- a/src/adapters/factory.ts +++ b/src/adapters/factory.ts @@ -11,6 +11,8 @@ * flag (`is_error`) is inferred (Anthropic convention), not source-confirmed. */ +import { sourceOf, textSources } from '../source-location.js' + import { readdir, stat } from 'node:fs/promises' import { homedir } from 'node:os' import { basename, join } from 'node:path' @@ -106,6 +108,17 @@ export class FactoryAdapter implements HarnessTraceAdapter { return refs.sort((a, b) => b.mtimeMs - a.mtimeMs) } + async sourcePaths(ref: SessionRef): Promise { + const sidecar = ref.path.replace(/\.jsonl$/, '.settings.json') + try { + await stat(sidecar) + return [ref.path, sidecar] + } catch (error) { + if (isMissingPathError(error)) return [ref.path] + throw error + } + } + async parse(ref: SessionRef, options: ParseOptions = {}): Promise { // Sidecar holds model + session-total tokens. let settings: FactorySettings = {} @@ -151,6 +164,7 @@ export class FactoryAdapter implements HarnessTraceAdapter { parentSpanId: sourceRootId, startTime: ts, content: text, + contentSource: textSources(l.message, 'content'), service: SERVICE, agent: SERVICE, step, @@ -173,6 +187,7 @@ export class FactoryAdapter implements HarnessTraceAdapter { model: settings.model ?? null, step, content: text || null, + contentSource: textSources(l.message, 'content'), }), ) lastLlm = llmId @@ -192,7 +207,7 @@ export class FactoryAdapter implements HarnessTraceAdapter { agent: SERVICE, tool: b.name, step, - extra: toolIoAttributes({ input: b.input }), + extra: toolIoAttributes({ input: b.input, inputSource: sourceOf(b, 'input') }), }) spans.push(t) if (b.id) toolByUseId.set(b.id, t) @@ -201,7 +216,7 @@ export class FactoryAdapter implements HarnessTraceAdapter { const t = toolByUseId.get(b.tool_use_id) if (t) { t.status = b.is_error === true ? { code: 'ERROR', message: 'tool reported error' } : { code: 'OK' } - recordToolOutput(t, b.content) + recordToolOutput(t, b.content, sourceOf(b, 'content')) } } } diff --git a/src/adapters/forge.ts b/src/adapters/forge.ts index 748dadd..13532f0 100644 --- a/src/adapters/forge.ts +++ b/src/adapters/forge.ts @@ -12,6 +12,8 @@ * or `{approx:N}`. Parse unverified against local data. */ +import { sourceOf, textSources } from '../source-location.js' + import { readdir, stat } from 'node:fs/promises' import { homedir } from 'node:os' import { basename, join } from 'node:path' @@ -20,7 +22,7 @@ import type { OtlpSpan } from '../otlp.js' import { span } from '../otlp.js' import { capText, userPromptSpan } from './conversation.js' import { recordToolOutput, toolIoAttributes } from './tool-io.js' -import type { HarnessTraceAdapter, LocateOptions, SessionRef } from '../types.js' +import type { HarnessTraceAdapter, LocateOptions, ParseOptions, SessionRef } from '../types.js' const SERVICE = 'forge' @@ -103,8 +105,8 @@ export class ForgeAdapter implements HarnessTraceAdapter { return refs.sort((a, b) => b.mtimeMs - a.mtimeMs) } - async parse(ref: SessionRef): Promise { - const dump = await readJsonFile(ref.path) + async parse(ref: SessionRef, options: ParseOptions = {}): Promise { + const dump = await readJsonFile(ref.path, options) const ctx = dump.context ?? dump const traceId = ctx.conversation_id ?? ref.sessionId const rootId = `root:${traceId}` @@ -132,6 +134,7 @@ export class ForgeAdapter implements HarnessTraceAdapter { parentSpanId: rootId, startTime: ts, content: prompt, + contentSource: textSources(entry.text, 'content'), service: SERVICE, agent: SERVICE, step, @@ -156,6 +159,7 @@ export class ForgeAdapter implements HarnessTraceAdapter { outputTokens: tokens(raw.usage?.completion_tokens), step, content: textOf(entry.text.content) || null, + contentSource: textSources(entry.text, 'content'), }), ) lastLlm = llmId @@ -173,7 +177,7 @@ export class ForgeAdapter implements HarnessTraceAdapter { agent: SERVICE, tool: tc.name ?? 'tool', step, - extra: toolIoAttributes({ input: tc.arguments }), + extra: toolIoAttributes({ input: tc.arguments, inputSource: sourceOf(tc, 'arguments') }), }) spans.push(t) toolByCallId.set(id, t) @@ -183,7 +187,7 @@ export class ForgeAdapter implements HarnessTraceAdapter { const t = toolByCallId.get(entry.tool.call_id ?? '') if (t) { t.status = entry.tool.output?.is_error ? { code: 'ERROR', message: 'tool reported error' } : { code: 'OK' } - recordToolOutput(t, entry.tool.output?.values ?? entry.tool.output) + recordToolOutput(t, entry.tool.output?.values ?? entry.tool.output, entry.tool.output?.values != null ? sourceOf(entry.tool.output, 'values') : sourceOf(entry.tool, 'output')) } } } diff --git a/src/adapters/gemini.ts b/src/adapters/gemini.ts index 6b40556..d2b75d6 100644 --- a/src/adapters/gemini.ts +++ b/src/adapters/gemini.ts @@ -9,6 +9,8 @@ * JSONL under `~/.qwen/projects/`); it has its own adapter. */ +import { sourceOf } from '../source-location.js' + import { readdir, stat } from 'node:fs/promises' import { homedir } from 'node:os' import { join } from 'node:path' @@ -17,7 +19,7 @@ import type { OtlpSpan } from '../otlp.js' import { span } from '../otlp.js' import { capText, userPromptSpan } from './conversation.js' import { toolIoAttributes } from './tool-io.js' -import type { HarnessTraceAdapter, LocateOptions, SessionRef } from '../types.js' +import type { HarnessTraceAdapter, LocateOptions, ParseOptions, SessionRef } from '../types.js' interface GeminiToolCall { id?: string @@ -145,8 +147,8 @@ export class GeminiFamilyAdapter implements HarnessTraceAdapter { return refs.sort((a, b) => b.mtimeMs - a.mtimeMs) } - async parse(ref: SessionRef): Promise { - const session = await readJsonFile(ref.path) + async parse(ref: SessionRef, options: ParseOptions = {}): Promise { + const session = await readJsonFile(ref.path, options) const traceId = session.sessionId ?? ref.sessionId const messages = session.messages ?? [] const rootId = `root:${traceId}` @@ -182,6 +184,7 @@ export class GeminiFamilyAdapter implements HarnessTraceAdapter { parentSpanId: rootId, startTime: ts, content: prompt, + contentSource: sourceOf(m, 'content'), service: this.service, agent: this.service, step, @@ -205,6 +208,7 @@ export class GeminiFamilyAdapter implements HarnessTraceAdapter { outputTokens: m.tokens?.output ?? null, step, content: textOf(m.content) || null, + contentSource: sourceOf(m, 'content'), }), ) step += 1 @@ -225,7 +229,7 @@ export class GeminiFamilyAdapter implements HarnessTraceAdapter { agent: this.service, tool: tc.name ?? 'tool', step, - extra: toolIoAttributes({ input: tc.args, output: tc.result }), + extra: toolIoAttributes({ input: tc.args, output: tc.result, inputSource: sourceOf(tc, 'args'), outputSource: sourceOf(tc, 'result') }), }), ) step += 1 diff --git a/src/adapters/opencode.ts b/src/adapters/opencode.ts index c8972b3..0ad866e 100644 --- a/src/adapters/opencode.ts +++ b/src/adapters/opencode.ts @@ -8,13 +8,15 @@ * marks a failed call. */ +import { sourceOf } from '../source-location.js' + import { readdir, stat } from 'node:fs/promises' import { homedir } from 'node:os' import { join } from 'node:path' import { isMissingJsonSource, isMissingPathError, listJsonFiles, readJsonFile } from '../json.js' import type { OtlpSpan } from '../otlp.js' import { span } from '../otlp.js' -import type { HarnessTraceAdapter, LocateOptions, SessionRef } from '../types.js' +import type { HarnessTraceAdapter, LocateOptions, ParseOptions, SessionRef } from '../types.js' import { capText, userPromptSpan } from './conversation.js' import { toolIoAttributes } from './tool-io.js' @@ -92,7 +94,34 @@ export class OpencodeAdapter implements HarnessTraceAdapter { return refs.sort((a, b) => b.mtimeMs - a.mtimeMs) } - async parse(ref: SessionRef): Promise { + async sourcePaths(ref: SessionRef): Promise { + const paths: string[] = [] + let step = 0 + const messages = await Promise.all((await listJsonFiles(ref.path)).map(async (file) => { + const path = join(ref.path, file) + paths.push(path) + return readJsonFile(path) + })) + messages.sort((a, b) => (a.time?.created ?? 0) - (b.time?.created ?? 0)) + for (const message of messages) { + const mid = message.id ?? `m${step}` + const directory = join(this.storage(), 'part', mid) + let files: string[] = [] + try { + files = await listJsonFiles(directory) + } catch (error) { + if (!isMissingJsonSource(error)) throw error + } + paths.push(...files.map((file) => join(directory, file))) + // The parser's fallback message ID depends on emitted spans, not file count. + const parts = await Promise.all(files.map((file) => readJsonFile(join(directory, file)))) + if (message.role !== 'user' || textOf(parts)) step += 1 + step += parts.filter((part) => part.type === 'tool' && part.tool).length + } + return paths.sort() + } + + async parse(ref: SessionRef, options: ParseOptions = {}): Promise { const traceId = ref.sessionId const rootId = `root:${traceId}` const partRoot = join(this.storage(), 'part') @@ -101,7 +130,7 @@ export class OpencodeAdapter implements HarnessTraceAdapter { const messages: OcMessage[] = [] for (const f of msgFiles) { - messages.push(await readJsonFile(join(ref.path, f))) + messages.push(await readJsonFile(join(ref.path, f), options)) } messages.sort((a, b) => (a.time?.created ?? 0) - (b.time?.created ?? 0)) @@ -135,7 +164,7 @@ export class OpencodeAdapter implements HarnessTraceAdapter { if (!isMissingJsonSource(error)) throw error } for (const pf of partFiles) { - parts.push(await readJsonFile(join(pdir, pf))) + parts.push(await readJsonFile(join(pdir, pf), options)) } const turnText = textOf(parts) @@ -153,6 +182,7 @@ export class OpencodeAdapter implements HarnessTraceAdapter { agent: SERVICE, step, content: turnText, + contentSource: parts.filter((part) => part.type === 'text').flatMap((part) => { const ref = sourceOf(part, 'text'); return ref ? [ref] : [] }), }), ) step += 1 @@ -177,6 +207,7 @@ export class OpencodeAdapter implements HarnessTraceAdapter { cacheWriteInputTokens: msg.tokens?.cache?.write ?? null, step, content: turnText || null, + contentSource: parts.filter((part) => part.type === 'text').flatMap((part) => { const ref = sourceOf(part, 'text'); return ref ? [ref] : [] }), }), ) step += 1 @@ -202,6 +233,8 @@ export class OpencodeAdapter implements HarnessTraceAdapter { step, extra: toolIoAttributes({ input: part.state?.input, + inputSource: sourceOf(part, 'state', 'input'), + outputSource: sourceOf(part, 'state', part.state?.output != null ? 'output' : 'error'), output: part.state?.output ?? part.state?.error, }), }), diff --git a/src/adapters/pi.ts b/src/adapters/pi.ts index 309f867..e72291a 100644 --- a/src/adapters/pi.ts +++ b/src/adapters/pi.ts @@ -10,6 +10,8 @@ * separate `role: "toolResult"` messages keyed by `message.toolCallId`. */ +import { sourceOf, textSources } from '../source-location.js' + import type { Dirent } from 'node:fs' import { readdir, stat } from 'node:fs/promises' import { homedir } from 'node:os' @@ -69,6 +71,7 @@ interface PiLine { } interface PiToolResult { + source?: ReturnType callId: string toolName: string timestamp: string @@ -143,7 +146,7 @@ function completeToolSpan(toolSpan: OtlpSpan, result: PiToolResult): void { toolSpan.status = result.isError ? { code: 'ERROR', message: result.errorMessage } : { code: 'OK' } - recordToolOutput(toolSpan, result.output) + recordToolOutput(toolSpan, result.output, result.source) } export interface PiAdapterOptions { @@ -272,6 +275,9 @@ export class PiAdapter implements HarnessTraceAdapter { timestamp: ts, isError: msg.isError === true, output: toolResultOutput(msg.content), + source: msg.content?.length === 1 && msg.content[0]?.type === 'text' && typeof msg.content[0].text === 'string' + ? sourceOf(msg.content[0], 'text') + : sourceOf(msg, 'content'), errorMessage, step, } @@ -297,6 +303,7 @@ export class PiAdapter implements HarnessTraceAdapter { agent: SERVICE, step, content: prompt, + contentSource: textSources(msg, 'content'), }), ) step += 1 @@ -323,6 +330,7 @@ export class PiAdapter implements HarnessTraceAdapter { cacheWriteInputTokens: msg.usage?.cacheWrite ?? null, step, content: textOf(msg.content) || null, + contentSource: textSources(msg, 'content'), }), ) step += 1 @@ -344,7 +352,7 @@ export class PiAdapter implements HarnessTraceAdapter { tool: name, status: 'UNSET', step, - extra: toolIoAttributes({ input: b.input ?? b.args ?? b.arguments }), + extra: toolIoAttributes({ input: b.input ?? b.args ?? b.arguments, inputSource: sourceOf(b, b.input != null ? 'input' : b.args != null ? 'args' : 'arguments') }), }) spans.push(toolSpan) toolByCallId.set(callId, toolSpan) @@ -361,7 +369,7 @@ export class PiAdapter implements HarnessTraceAdapter { const err = b.isError === true || b.is_error === true t.end_time = ts t.status = err ? { code: 'ERROR', message: 'tool result reported error' } : { code: 'OK' } - recordToolOutput(t, b.output ?? b.result ?? b.content ?? b.text) + recordToolOutput(t, b.output ?? b.result ?? b.content ?? b.text, sourceOf(b, b.output != null ? 'output' : b.result != null ? 'result' : b.content != null ? 'content' : 'text')) } } } @@ -384,7 +392,7 @@ export class PiAdapter implements HarnessTraceAdapter { tool: result.toolName, step: result.step, extra: { - ...toolIoAttributes({ output: result.output }), + ...toolIoAttributes({ output: result.output, outputSource: result.source }), 'traces.pi.tool_result_without_call': true, }, }), diff --git a/src/adapters/qwen.ts b/src/adapters/qwen.ts index 7e3db4b..003d172 100644 --- a/src/adapters/qwen.ts +++ b/src/adapters/qwen.ts @@ -12,6 +12,8 @@ * sessions on this machine). */ +import { sourceOf, textSources } from '../source-location.js' + import { readdir, stat } from 'node:fs/promises' import { homedir } from 'node:os' import { basename, join } from 'node:path' @@ -132,6 +134,7 @@ export class QwenAdapter implements HarnessTraceAdapter { outputTokens: r.usageMetadata?.candidatesTokenCount ?? null, step, content: capText(textOf(r.message?.parts)) || null, + contentSource: textSources(r.message, 'parts'), }), ) step += 1 @@ -150,7 +153,7 @@ export class QwenAdapter implements HarnessTraceAdapter { agent: SERVICE, tool: name, step, - extra: toolIoAttributes({ input: p.functionCall.args }), + extra: toolIoAttributes({ input: p.functionCall.args, inputSource: sourceOf(p.functionCall, 'args') }), }) spans.push(t) const q = openToolsByName.get(name) ?? [] @@ -175,6 +178,7 @@ export class QwenAdapter implements HarnessTraceAdapter { agent: SERVICE, step, content: prompt, + contentSource: textSources(r.message, 'parts'), }), ) step += 1 @@ -188,7 +192,7 @@ export class QwenAdapter implements HarnessTraceAdapter { if (t) { t.end_time = ts t.status = err ? { code: 'ERROR', message: 'tool result reported error' } : { code: 'OK' } - recordToolOutput(t, p.functionResponse?.response ?? r.toolCallResult?.error) + recordToolOutput(t, p.functionResponse?.response ?? r.toolCallResult?.error, p.functionResponse ? sourceOf(p.functionResponse, 'response') : sourceOf(r.toolCallResult, 'error')) } } } diff --git a/src/adapters/tool-io.ts b/src/adapters/tool-io.ts index db41265..900ff46 100644 --- a/src/adapters/tool-io.ts +++ b/src/adapters/tool-io.ts @@ -1,10 +1,13 @@ import { createHash } from 'node:crypto' import type { OtlpSpan } from '../otlp.js' +import { sourceAttributes, SOURCE_ATTRIBUTE_PREFIX, type SourceReferences } from '../source-location.js' export const TOOL_IO_VALUE_MAX_BYTES = 16 * 1024 export const TOOL_IO_VALUE_KEYS = ['input.value', 'output.value'] as const interface ToolIoInput { + inputSource?: SourceReferences + outputSource?: SourceReferences input?: unknown output?: unknown argsCaptured?: boolean @@ -22,7 +25,7 @@ const TRUNCATION_MARKER = /\n\[truncated\]$/ // Sort structurally, then stringify once. Recursive string concatenation // exceeds the bounded-heap adapter contract for large structured values. -function canonicalJson(value: unknown): string { +export function canonicalJson(value: unknown): string { const sort = (item: unknown): unknown => { if (Array.isArray(item)) return item.map(sort) if (!item || typeof item !== 'object') return item @@ -95,6 +98,8 @@ export function toolIoAttributes(io: ToolIoInput): Record { return { ...valueAttributes('input', io.input), ...valueAttributes('output', io.output), + ...(io.input === undefined ? {} : sourceAttributes('input.value', io.inputSource)), + ...(io.output === undefined ? {} : sourceAttributes('output.value', io.outputSource)), ...(io.argsCaptured === undefined ? {} : { 'tool.args_captured': io.argsCaptured }), } } @@ -151,7 +156,10 @@ export function normalizeToolIoAttributes(attributes: Record): } } -export function recordToolOutput(toolSpan: OtlpSpan | undefined, output: unknown): void { +export function recordToolOutput(toolSpan: OtlpSpan | undefined, output: unknown, source?: SourceReferences): void { if (!toolSpan) return + if (output === undefined) return + delete toolSpan.attributes[`${SOURCE_ATTRIBUTE_PREFIX}output.value`] Object.assign(toolSpan.attributes, valueAttributes('output', output)) + Object.assign(toolSpan.attributes, sourceAttributes('output.value', source)) } diff --git a/src/analyze.ts b/src/analyze.ts index 084d6a7..41ee982 100644 --- a/src/analyze.ts +++ b/src/analyze.ts @@ -25,8 +25,11 @@ import { OtlpFileTraceStore } from '@tangle-network/agent-eval/traces' import { summarizeSpanExecution } from './execution.js' import type { OtlpSpan } from './otlp.js' import { writeOtlpFile } from './otlp.js' +import { assertOutsideSourceBundle, createBundleSourceReader } from './bundle-source.js' export interface AnalyzeOptions { + /** Explicitly authorize original source reads from this full session bundle. */ + sourceBundle?: { path: string; maxRecordBytes?: number } /** * Recursive analysis engine enabling the agentic analyst kinds. Omit → * deterministic only. The engine's id, version, and model become the @@ -97,6 +100,10 @@ function mergeCostProvenance( export async function analyzeSpans(spans: readonly OtlpSpan[], opts: AnalyzeOptions = {}): Promise { if (spans.length === 0) throw new Error('analyzeSpans: no spans to analyze') opts.signal?.throwIfAborted() + if (opts.sourceBundle && opts.otlpOutPath) await assertOutsideSourceBundle(opts.sourceBundle.path, opts.otlpOutPath) + const sourceReader = opts.sourceBundle + ? await createBundleSourceReader(opts.sourceBundle.path, spans, { signal: opts.signal, maxRecordBytes: opts.sourceBundle.maxRecordBytes }) + : undefined const otlpPath = await writeOtlpFile(spans, opts.otlpOutPath) opts.signal?.throwIfAborted() const runId = opts.runId ?? `traces-${Date.now()}` @@ -111,6 +118,7 @@ export async function analyzeSpans(spans: readonly OtlpSpan[], opts: AnalyzeOpti path: otlpPath, maxFileBytes: GENERATED_TRACE_FILE_CEILING, perCallByteCeiling: GENERATED_TRACE_FILE_CEILING, + ...(sourceReader ? { sourceReader } : {}), }) await detStore.ensureIndexed() opts.signal?.throwIfAborted() @@ -122,7 +130,7 @@ export async function analyzeSpans(spans: readonly OtlpSpan[], opts: AnalyzeOpti // the RLM kinds drill via viewSpans/searchTrace from a summary. let agenticPerAnalyst: readonly AnalystRunSummary[] | undefined if (opts.engine || opts.agenticRegistry) { - const agStore = new OtlpFileTraceStore({ path: otlpPath, maxFileBytes: GENERATED_TRACE_FILE_CEILING }) + const agStore = new OtlpFileTraceStore({ path: otlpPath, maxFileBytes: GENERATED_TRACE_FILE_CEILING, ...(sourceReader ? { sourceReader } : {}) }) await agStore.ensureIndexed() const agRegistry = opts.agenticRegistry ?? buildDefaultAnalystRegistry({ engine: opts.engine!, diff --git a/src/bundle-source.ts b/src/bundle-source.ts new file mode 100644 index 0000000..d58e0c0 --- /dev/null +++ b/src/bundle-source.ts @@ -0,0 +1,236 @@ +import { canonicalJson } from './adapters/tool-io.js' +import { isUtf8 } from 'node:buffer' +import { createHash } from 'node:crypto' +import { constants } from 'node:fs' +import { lstat, open, realpath, type FileHandle } from 'node:fs/promises' +import { dirname, isAbsolute, join, relative, resolve, sep } from 'node:path' +import type { ReadSpanSourceResult, SpanSourceReader } from '@tangle-network/agent-eval/traces' +import { readSessionBundleManifest, type SessionBundleFile } from './bundle.js' +import { readOtlpInput } from './otlp-input.js' +import type { OtlpSpan } from './otlp.js' +import { SOURCE_ATTRIBUTE_PREFIX, type SourceRecordReference } from './source-location.js' + +const SHA256 = /^[a-f0-9]{64}$/ +const TRACE_FILE = 'derived/trace.otlp.jsonl' + +function containedPath(root: string, path: string): string { + if (isAbsolute(path) || path.split(/[\\/]/).some((part) => part === '..' || part === '' || part === '.')) { + throw new Error('bundle contains an unsafe file path') + } + const resolved = resolve(root, path) + if (relative(root, resolved).startsWith(`..${sep}`) || resolved === root) { + throw new Error('bundle file is outside the authorized directory') + } + return resolved +} + +async function openBundleFile(root: string, file: SessionBundleFile): Promise { + const path = containedPath(root, file.path) + let current = root + for (const part of relative(root, path).split(sep)) { + current = join(current, part) + if ((await lstat(current)).isSymbolicLink()) throw new Error('bundle source symlinks are not permitted') + } + if (await realpath(path) !== path) throw new Error('bundle source resolved outside its declared path') + const handle = await open(path, constants.O_RDONLY | constants.O_NOFOLLOW) + const info = await handle.stat() + if (!info.isFile() || info.size !== file.bytes) { + await handle.close() + throw new Error('bundle source size or type does not match its manifest') + } + return handle +} + +async function fileDigest(handle: FileHandle, signal?: AbortSignal): Promise { + const hash = createHash('sha256') + for await (const bytes of handle.createReadStream({ start: 0, autoClose: false, signal })) hash.update(bytes) + return hash.digest('hex') +} + +async function validateFile(root: string, file: SessionBundleFile, signal?: AbortSignal): Promise { + const handle = await openBundleFile(root, file) + try { + if (await fileDigest(handle, signal) !== file.sha256) throw new Error('bundle file digest does not match its manifest') + } finally { + await handle.close() + } +} + +function reference(value: unknown): value is SourceRecordReference { + if (!value || typeof value !== 'object') return false + const item = value as Partial + return typeof item.sourceId === 'string' && SHA256.test(item.sourceId) + && typeof item.sourceSha256 === 'string' && SHA256.test(item.sourceSha256) + && typeof item.recordSha256 === 'string' && SHA256.test(item.recordSha256) + && Number.isSafeInteger(item.recordOffset) && item.recordOffset! >= 0 + && Number.isSafeInteger(item.recordBytes) && item.recordBytes! > 0 + && typeof item.fieldLocator === 'string' && item.fieldLocator.length <= 4096 && item.fieldLocator.startsWith('#') +} + +function identity(span: Pick): string { + return JSON.stringify([span.trace_id, span.span_id]) +} + +/** Grant only the original records associated with unchanged attributes in these selected spans. */ +export async function createBundleSourceReader( + directory: string, + spans: readonly OtlpSpan[], + options: { signal?: AbortSignal; maxRecordBytes?: number } = {}, +): Promise { + const maxRecordBytes = options.maxRecordBytes ?? 16 * 1024 * 1024 + if (!Number.isSafeInteger(maxRecordBytes) || maxRecordBytes < 1) throw new Error('maxRecordBytes must be a positive safe integer') + options.signal?.throwIfAborted() + if ((await lstat(resolve(directory))).isSymbolicLink()) throw new Error('bundle directory must not be a symlink') + const root = await realpath(resolve(directory)) + if ((await lstat(join(root, 'manifest.json'))).isSymbolicLink()) throw new Error('bundle manifest must not be a symlink') + const manifest = await readSessionBundleManifest(root) + if (manifest.view !== 'full') throw new Error('original source reads require an explicitly authorized full bundle') + if (!Array.isArray(manifest.files)) throw new Error('bundle manifest has no file receipts') + const paths = new Set() + const sources = new Map() + for (const file of manifest.files) { + if (!file || typeof file.path !== 'string' || !SHA256.test(file.sha256) + || !Number.isSafeInteger(file.bytes) || file.bytes < 0 || paths.has(file.path)) { + throw new Error('bundle manifest has an invalid or duplicate file receipt') + } + containedPath(root, file.path) + paths.add(file.path) + if (file.sourceId !== undefined) { + if (!SHA256.test(file.sourceId) || sources.has(file.sourceId) || !file.path.startsWith('session/')) { + throw new Error('bundle manifest has an invalid source identity') + } + sources.set(file.sourceId, { ...file }) + } + } + const traceFile = manifest.files.find((file) => file.path === TRACE_FILE) + if (!traceFile) throw new Error('bundle has no derived trace receipt') + await validateFile(root, traceFile, options.signal) + const recorded = await readOtlpInput(join(root, TRACE_FILE)) + await validateFile(root, traceFile, options.signal) + const original = new Map(recorded.spans.map((span) => [identity(span), span])) + const allowed = new Map>() + for (const span of spans) { + const prior = original.get(identity(span)) + if (!prior) continue + const attributes = new Map() + for (const [key, encoded] of Object.entries(span.attributes)) { + if (!key.startsWith(SOURCE_ATTRIBUTE_PREFIX) || typeof encoded !== 'string') continue + const attribute = key.slice(SOURCE_ATTRIBUTE_PREFIX.length) + if (encoded !== prior.attributes[key] || span.attributes[attribute] === undefined + || JSON.stringify(span.attributes[attribute]) !== JSON.stringify(prior.attributes[attribute])) continue + let refs: unknown + try { refs = JSON.parse(encoded) } catch { continue } + if (Array.isArray(refs) && refs.length > 0 && refs.every(reference)) attributes.set(attribute, refs) + } + if (allowed.has(identity(span))) throw new Error('duplicate selected span identity') + allowed.set(identity(span), attributes) + } + if (![...allowed.values()].some((attributes) => attributes.size > 0)) return undefined + + return async (input, context): Promise => { + const sourceIndex = input.source_index ?? 0 + const unavailable = (reason: string): ReadSpanSourceResult => ({ + status: 'unavailable', trace_id: input.trace_id, span_id: input.span_id, + attribute: input.attribute, source_index: sourceIndex, reason, + }) + const signals = [options.signal, context?.signal].filter((signal): signal is AbortSignal => signal !== undefined) + const signal = signals.length > 1 ? AbortSignal.any(signals) : signals[0] + signal?.throwIfAborted() + if (![input.offset, input.limit, sourceIndex].every(Number.isSafeInteger) + || input.offset < 0 || input.limit < 1 || sourceIndex < 0) { + return unavailable('invalid source window') + } + const ref = allowed.get(identity(input))?.get(input.attribute)?.[sourceIndex] + if (!ref) return unavailable('attribute source was not captured or authorized for these spans') + const file = sources.get(ref.sourceId) + if (!file || file.sha256 !== ref.sourceSha256) return unavailable('retained source receipt is missing or mismatched') + if (ref.recordOffset > file.bytes - ref.recordBytes) { + return unavailable('source window is outside the recorded bytes') + } + if (ref.recordBytes > maxRecordBytes) return unavailable('source record exceeds the configured parsing limit') + let handle: FileHandle | undefined + try { + handle = await openBundleFile(root, file) + const before = await handle.stat() + if (await fileDigest(handle, signal) !== file.sha256) return unavailable('retained source digest mismatch') + const hash = createHash('sha256') + const record = Buffer.alloc(ref.recordBytes) + let position = 0 + for await (const raw of handle.createReadStream({ + start: ref.recordOffset, end: ref.recordOffset + ref.recordBytes - 1, autoClose: false, signal, + })) { + const bytes = Buffer.isBuffer(raw) ? raw : Buffer.from(raw) + hash.update(bytes) + bytes.copy(record, position) + position += bytes.length + } + const after = await handle.stat() + if (position !== ref.recordBytes || hash.digest('hex') !== ref.recordSha256 + || before.size !== after.size || before.mtimeMs !== after.mtimeMs || before.ctimeMs !== after.ctimeMs) { + return unavailable('retained record digest mismatch or source changed during read') + } + if (!isUtf8(record)) return unavailable('source record is not valid UTF-8') + let value: unknown = JSON.parse(record.toString('utf8')) + if (ref.fieldLocator !== '#') { + if (!ref.fieldLocator.startsWith('#/')) return unavailable('invalid source field locator') + for (const part of ref.fieldLocator.slice(2).split('/')) { + if (/~(?:[^01]|$)/.test(part)) return unavailable('invalid source field locator') + const key = part.replaceAll('~1', '/').replaceAll('~0', '~') + if (!value || typeof value !== 'object' || !Object.hasOwn(value, key)) return unavailable('source field is missing') + value = (value as Record)[key] + } + } + const valueEncoding = typeof value === 'string' ? 'utf8-string' as const : 'json' as const + const field = Buffer.from(typeof value === 'string' ? value : canonicalJson(value), 'utf8') + if (field.length === 0) return unavailable('source field is empty') + if (input.offset > field.length) return unavailable('source offset is outside the selected field') + const window = field.subarray(input.offset, Math.min(field.length, input.offset + input.limit)) + if (window.length > 0 && (window[0]! & 0xc0) === 0x80) return unavailable('offset splits a UTF-8 character') + let bytes = window + for (let omitted = 0; omitted <= 3; omitted += 1) { + bytes = window.subarray(0, Math.max(0, window.length - omitted)) + if (isUtf8(bytes)) break + } + if (!isUtf8(bytes) || bytes.length === 0 && input.offset < field.length) { + return unavailable('window cannot contain a complete UTF-8 character') + } + const next = input.offset + bytes.length + return { + status: 'available', trace_id: input.trace_id, span_id: input.span_id, + attribute: input.attribute, source_index: sourceIndex, text: bytes.toString('utf8'), + offset: input.offset, total_bytes: field.length, next_offset: next < field.length ? next : null, + source: { + source_id: ref.sourceId, source_sha256: file.sha256, + record_sha256: ref.recordSha256, field_locator: ref.fieldLocator, value_encoding: valueEncoding, + }, + } + } catch { + signal?.throwIfAborted() + return unavailable('retained source is missing, inaccessible, or unsafe') + } finally { + await handle?.close() + } + } +} + +/** Keep analysis artifacts from changing an authorized retained archive. */ +export async function assertOutsideSourceBundle(directory: string, outputPath: string): Promise { + const root = await realpath(directory) + let ancestor = resolve(outputPath) + while (true) { + try { + ancestor = await realpath(ancestor) + break + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error + const parent = dirname(ancestor) + if (parent === ancestor) throw error + ancestor = parent + } + } + const within = (path: string): boolean => { + const location = relative(root, path) + return location === '' || location !== '..' && !location.startsWith(`..${sep}`) && !location.startsWith(sep) + } + if (within(resolve(outputPath)) || within(ancestor)) throw new Error('analysis output must be outside the retained source bundle') +} diff --git a/src/bundle.ts b/src/bundle.ts index 76d4dc0..29ced84 100644 --- a/src/bundle.ts +++ b/src/bundle.ts @@ -23,8 +23,8 @@ import { createHash } from 'node:crypto' import { readFileSync } from 'node:fs' -import { cp, mkdir, readdir, readFile, stat, writeFile } from 'node:fs/promises' -import { join, relative, resolve, sep } from 'node:path' +import { copyFile, cp, mkdir, readdir, readFile, stat, writeFile } from 'node:fs/promises' +import { dirname, join, relative, resolve, sep } from 'node:path' import { buildPolicyEvidenceRecord, serializePolicyEvidence } from './evidence.js' import { runTraceInvestigation } from './improvement.js' import { isMissingPathError } from './json.js' @@ -34,10 +34,13 @@ import { findContextRoot, serializeSessionIndex, } from './session-index.js' -import { parseSession } from './session-source.js' +import { collectSessionSelection, fileSha256 } from './session-selection.js' +import { sourceFileId } from './source-location.js' import type { HarnessTraceAdapter, SessionRef } from './types.js' export interface SessionBundleFile { + /** Opaque source identity when this file backs native span attributes. */ + readonly sourceId?: string /** Bundle-relative path, `/`-separated. */ readonly path: string readonly bytes: number @@ -432,22 +435,22 @@ export async function assembleSessionBundle(opts: AssembleSessionBundleOptions): const { adapter, ref } = opts const generatedAt = opts.generatedAt ?? new Date().toISOString() - let transcriptBytes: Buffer try { - transcriptBytes = await readFile(ref.path) + await stat(ref.path) } catch (error) { if (isMissingPathError(error)) { - throw new Error( - `session transcript not found at ${ref.path} — a bundle cannot be assembled without its transcript`, - ) + throw new Error(`session transcript not found at ${ref.path} — a bundle cannot be assembled without its transcript`) } throw error } - const transcriptSha256 = sha256Hex(transcriptBytes) - - // Parse BEFORE creating the output directory: an unparseable session must - // fail without leaving a half-written bundle behind. - const spans = await parseSession(adapter, ref, { signal: opts.signal }) + const selection = await collectSessionSelection([{ adapter, refs: [ref] }], { + bindSources: true, + signal: opts.signal, + }) + const row = selection.rows[0]! + const spans = row.spans + const transcriptSha256 = row.sourceFiles!.find((file) => resolve(file.path) === resolve(ref.path))?.sha256 + ?? row.sourceSha256! const outDir = resolve(opts.outDir) await mkdir(outDir, { recursive: true }) @@ -462,7 +465,6 @@ export async function assembleSessionBundle(opts: AssembleSessionBundleOptions): // session/ — the raw sources, byte-for-byte. await mkdir(join(outDir, 'session'), { recursive: true }) - await writeFile(join(outDir, 'session', 'transcript.jsonl'), transcriptBytes) const subagentsSource = join(ref.path.replace(/\.jsonl$/, ''), 'subagents') if ((await statOrNull(subagentsSource))?.isDirectory()) { await cp(subagentsSource, join(outDir, 'session', 'subagents'), { recursive: true }) @@ -470,6 +472,25 @@ export async function assembleSessionBundle(opts: AssembleSessionBundleOptions): absent.push({ path: bundlePath('session', 'subagents'), reason: `no subagents directory at ${subagentsSource}` }) } + const retainedSourceIds = new Map() + for (const file of row.sourceFiles!) { + opts.signal?.throwIfAborted() + const sourceId = sourceFileId(file.path) + const subagentRelative = relative(subagentsSource, file.path) + const path = resolve(file.path) === resolve(ref.path) + ? bundlePath('session', 'transcript.jsonl') + : !subagentRelative.startsWith('..') && !subagentRelative.startsWith(sep) + ? bundlePath('session', 'subagents', subagentRelative) + : bundlePath('session', 'sources', `${sourceId}.jsonl`) + const target = join(outDir, path) + await mkdir(dirname(target), { recursive: true }) + await copyFile(file.path, target) + if (await fileSha256(target, opts.signal) !== file.sha256) { + throw new Error(`session source changed while retaining bundle evidence: ${file.path}`) + } + retainedSourceIds.set(path, sourceId) + } + // derived/ — every derivation this CLI already owns, deterministic only: // a bundle assembly must never spend a model call. await mkdir(join(outDir, 'derived'), { recursive: true }) @@ -548,7 +569,10 @@ export async function assembleSessionBundle(opts: AssembleSessionBundleOptions): } // manifest.json — sha256 per file, written LAST so it covers every byte. - const files = await hashSessionBundleFiles(outDir) + const files = (await hashSessionBundleFiles(outDir)).map((file) => ({ + ...file, + ...(retainedSourceIds.has(file.path) ? { sourceId: retainedSourceIds.get(file.path)! } : {}), + })) const manifest: SessionBundleManifest = { schemaVersion: 2, kind: 'traces.session_bundle', @@ -580,8 +604,9 @@ export async function assembleSessionBundle(opts: AssembleSessionBundleOptions): export async function hashSessionBundleFiles(root: string): Promise { const files: SessionBundleFile[] = [] for (const path of await listSessionBundleFiles(root)) { - const bytes = await readFile(join(root, path)) - files.push({ path, bytes: bytes.length, sha256: sha256Hex(bytes) }) + const filePath = join(root, path) + const info = await stat(filePath) + files.push({ path, bytes: info.size, sha256: await fileSha256(filePath) }) } return files } diff --git a/src/cli.ts b/src/cli.ts index a908115..6450085 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -40,6 +40,7 @@ * `--otlp ` READS OTLP; `--otlp-out ` WRITES the artifact. */ +import { assertOutsideSourceBundle } from './bundle-source.js' import { readFileSync } from 'node:fs' import { readdir, readFile, stat, writeFile } from 'node:fs/promises' import { basename, join, resolve } from 'node:path' @@ -138,6 +139,8 @@ interface Args { dir?: string /** OTLP-JSONL file or directory to READ. */ otlp?: string + /** Full bundle whose original records may be read by local analysts. */ + sourceBundle?: string /** Where to WRITE the OTLP-JSONL artifact. */ otlpOut?: string llm: boolean @@ -240,6 +243,7 @@ function parseArgs(argv: string[]): Args { case '--out': a.out = next(); break case '--dir': a.dir = next(); break case '--otlp': a.otlp = next(); break + case '--source-bundle': a.sourceBundle = next(); break case '--otlp-out': a.otlpOut = next(); break case '--llm': a.llm = true; break case '--budget': a.budget = Number(next()); break @@ -649,6 +653,16 @@ async function collectOtlpSpans(path: string): Promise { } async function collectSpans(args: Args): Promise { + if (args.sourceBundle) { + if (!['analyze', 'investigate', 'improve'].includes(args.command)) throw new Error('--source-bundle is an analysis option') + if (args.session || args.otlp || args.input || args.current || args.workflow || args.noContent || args.redactorCmd) { + throw new Error('--source-bundle cannot be combined with another input, redaction, or metadata-only selection') + } + for (const output of [args.out, args.dir, args.otlpOut]) { + if (output) await assertOutsideSourceBundle(args.sourceBundle, output) + } + return collectOtlpSpans(join(args.sourceBundle, 'derived', 'trace.otlp.jsonl')) + } if (args.otlp) return collectOtlpSpans(args.otlp) if (args.input) { if (args.workflow) throw new Error('--workflow reads discovered sessions and cannot be combined with an input file') @@ -1193,6 +1207,7 @@ async function investigate(args: Args, options: { loadDefaultConfig?: boolean } const analystModel = args.model ?? process.env.TRACES_ANALYST_MODEL ?? DEFAULT_ANALYST_MODEL const engine = args.llm ? buildAnalysisEngine(analystModel, args.budget) : undefined return runTraceInvestigation(mergeTracesConfig({ + sourceBundle: args.sourceBundle ? { path: args.sourceBundle } : undefined, spans, harness, sources, @@ -1219,6 +1234,7 @@ async function cmdImprove(args: Args): Promise { const engine = args.llm ? buildAnalysisEngine(analystModel, args.budget) : undefined const result = await runTraceImprovement({ ...mergeTracesConfig({ + sourceBundle: args.sourceBundle ? { path: args.sourceBundle } : undefined, spans, harness, sources, @@ -1659,6 +1675,8 @@ Options: Supported by: validate, analyze, investigate, improve, stream. On a WRITING command it is the deprecated spelling of --otlp-out; it still works, with a warning, until 0.12. + --source-bundle Analyze a retained full bundle; explicitly grant source-field reads. + Available for analyze, investigate, and improve. --otlp-out WRITE the OTLP-JSONL artifact here (also evidence provenance / dry-run upload preview) --format analyze/export: auto | policy-evidence | sandbox-events | openinference | intelligence-spans | chat-trajectory diff --git a/src/improvement.ts b/src/improvement.ts index acab64b..3888ee3 100644 --- a/src/improvement.ts +++ b/src/improvement.ts @@ -20,7 +20,8 @@ import { traceAgenticKinds, type TraceAgenticRoute, } from './agentic-routing.js' -import { analyzeSpans } from './analyze.js' +import { assertOutsideSourceBundle } from './bundle-source.js' +import { analyzeSpans, type AnalyzeOptions } from './analyze.js' import type { TraceValidation } from '@tangle-network/agent-trace-contract' import { conformanceOfSpans, renderConformance, unavailableCapabilities } from './conformance.js' import { @@ -64,6 +65,8 @@ export interface TracesConfig { } export interface TraceInvestigationOptions { + /** Explicit full-bundle source access for local analyst tools only. */ + readonly sourceBundle?: AnalyzeOptions['sourceBundle'] readonly spans: readonly OtlpSpan[] readonly harness: string readonly sources?: readonly ReportSource[] @@ -691,6 +694,7 @@ export async function runTraceInvestigation(opts: TraceInvestigationOptions): Pr ? planTraceAgenticRoute(pipelines, reactions) : undefined const analysis = await analyzeSpans(opts.spans, { + sourceBundle: opts.sourceBundle, engine: opts.engine, model: opts.model, budgetUsd: opts.budgetUsd, @@ -847,6 +851,7 @@ export async function writeTraceImprovementArtifacts( export async function runTraceImprovement( opts: TraceImprovementOptions, ): Promise { + if (opts.sourceBundle && opts.outDir) await assertOutsideSourceBundle(opts.sourceBundle.path, opts.outDir) const directory = opts.outDir ? resolve(opts.outDir) : await mkdtemp(join(tmpdir(), 'traces-improvement-')) diff --git a/src/index.ts b/src/index.ts index 1dd2262..58bc6a0 100644 --- a/src/index.ts +++ b/src/index.ts @@ -47,6 +47,8 @@ export { } from './integrity.js' export { knownHarnesses, listAdapters, resolveAdapter, selectAdapters } from './registry.js' export * from './session-source.js' // scanSessions() / parseSession() — locate→parse→stamp +export { createBundleSourceReader } from './bundle-source.js' +export { SOURCE_ATTRIBUTE_PREFIX, type SourceRecordReference } from './source-location.js' export * from './session-relationship.js' // stable parent/child metadata decoding export * from './session-workflow.js' // collectSessionWorkflow() — bounded parent/child expansion export * from './session-selection.js' // collectSessionSelection() — parse/expand/bind selected groups diff --git a/src/integrity.ts b/src/integrity.ts index 12b62b3..b1e7e51 100644 --- a/src/integrity.ts +++ b/src/integrity.ts @@ -56,11 +56,12 @@ export function recordSessionCorruption(ref: SessionRef, receipt: JsonlCorruptio export function sessionJsonlOptions(ref: SessionRef, options: ParseOptions = {}): JsonlReadOptions { if (options.corruptionMode === 'strict') { - return { mode: 'strict', signal: options.signal } + return { mode: 'strict', signal: options.signal, captureSources: options.captureSources } } return { mode: 'recover', onCorruption: (receipt) => recordSessionCorruption(ref, receipt), + captureSources: options.captureSources, signal: options.signal, } } diff --git a/src/json.ts b/src/json.ts index 55d739f..7019db8 100644 --- a/src/json.ts +++ b/src/json.ts @@ -1,4 +1,5 @@ import { readdir, readFile } from 'node:fs/promises' +import { locateSourceObjects } from './source-location.js' export type JsonSourceErrorKind = 'read' | 'parse' @@ -41,16 +42,22 @@ export function isMissingJsonSource(error: unknown): error is JsonSourceError { return error instanceof JsonSourceError && error.kind === 'read' && isMissingPathError(error) } -export async function readJsonFile(path: string): Promise { +export async function readJsonFile(path: string, options: { captureSources?: boolean } = {}): Promise { let raw: string + let bytes: Buffer | undefined try { - raw = await readFile(path, 'utf8') + if (options.captureSources) { + bytes = await readFile(path) + raw = bytes.toString('utf8') + } else raw = await readFile(path, 'utf8') } catch (error) { throw new JsonSourceError(path, 'read', error) } try { - return JSON.parse(raw) as T + const value = JSON.parse(raw) as T + if (bytes) locateSourceObjects(value, path, bytes) + return value } catch (error) { throw new JsonSourceError(path, 'parse', pathSafeParseCause(error)) } diff --git a/src/jsonl.ts b/src/jsonl.ts index 1160634..eb1f72e 100644 --- a/src/jsonl.ts +++ b/src/jsonl.ts @@ -1,6 +1,7 @@ import { isUtf8 } from 'node:buffer' import { createHash } from 'node:crypto' import { createReadStream } from 'node:fs' +import { locateSourceObjects } from './source-location.js' export interface JsonlCorruptionReceipt { receiptVersion: 1 @@ -19,13 +20,13 @@ export interface JsonlCorruptionReceipt { rawBytes: 'local_source_only' } -export type JsonlReadOptions = +export type JsonlReadOptions = { captureSources?: boolean } & ( | { mode?: 'strict'; signal?: AbortSignal } | { mode: 'recover' onCorruption: (receipt: JsonlCorruptionReceipt) => void signal?: AbortSignal - } + }) export class JsonlParseError extends SyntaxError { readonly sourcePath: string @@ -79,7 +80,9 @@ function parseLine( const json = jsonBytes.toString('utf8') if (json.trim().length === 0) return undefined try { - return JSON.parse(json) as T + const value = JSON.parse(json) as T + if (options.captureSources) locateSourceObjects(value, path, rawLine, byteOffset) + return value } catch { return handleCorruption(rawLine, path, lineNumber, byteOffset, options) } diff --git a/src/otlp.ts b/src/otlp.ts index 2395975..55143a9 100644 --- a/src/otlp.ts +++ b/src/otlp.ts @@ -31,6 +31,7 @@ import type { SpanKind } from '@tangle-network/agent-trace-contract' import { traceContractBuildIdOrNull } from './contract-build.js' import { RAW_FIELD_ATTRIBUTES, SUBSTITUTED_FIELDS_ATTR, type SubstitutedField } from './otlp-input.js' import { validateOtlpSpans } from './span-validation.js' +import { sourceAttributes, type SourceReferences } from './source-location.js' /** * The span-kind vocabulary, which is exactly @@ -76,6 +77,7 @@ export interface OtlpSpan { } export interface SpanInput { + contentSource?: SourceReferences traceId: string spanId: string parentSpanId?: string | null @@ -131,6 +133,7 @@ export function span(input: SpanInput): OtlpSpan { if (input.step != null) attributes.step = input.step if (input.content != null && input.content.length > 0) attributes['content'] = input.content if (input.extra) Object.assign(attributes, input.extra) + if (input.content) Object.assign(attributes, sourceAttributes('content', input.contentSource)) const status: OtlpSpan['status'] = { code: input.status ?? 'OK' } if (input.statusMessage && input.statusMessage.length > 0) status.message = input.statusMessage diff --git a/src/redact.ts b/src/redact.ts index ae06531..447f283 100644 --- a/src/redact.ts +++ b/src/redact.ts @@ -20,6 +20,7 @@ import type { RedactionReport, RedactionRule } from '@tangle-network/agent-eval/ import { normalizeToolIoAttributes, TOOL_IO_VALUE_KEYS } from './adapters/tool-io.js' import type { Redactor } from './external.js' import type { OtlpSpan } from './otlp.js' +import { stripSourceAttributes } from './source-location.js' /** Secrets common in coding-agent traces that the substrate defaults don't cover. */ export const CODING_REDACTION_RULES: RedactionRule[] = [ @@ -73,7 +74,7 @@ export function redactSpans( ): SpanRedaction { const report: RedactionReport = { redactionCount: 0, byRule: {} } const out = spans.map((s) => { - const attributes = redactValue(s.attributes, rules, report).value as Record + const attributes = redactValue(stripSourceAttributes(s.attributes), rules, report).value as Record normalizeToolIoAttributes(attributes) let status = s.status if (status.message) { @@ -111,7 +112,7 @@ export async function applyRedactor( texts.push(s.status.message) } }) - const out = spans.map((s) => ({ ...s, attributes: { ...s.attributes } })) + const out = spans.map((s) => ({ ...s, attributes: stripSourceAttributes(s.attributes) })) if (texts.length === 0) { for (const span of out) normalizeToolIoAttributes(span.attributes) return { spans: out, changed: 0 } diff --git a/src/session-selection.ts b/src/session-selection.ts index b262bd8..ea23ec7 100644 --- a/src/session-selection.ts +++ b/src/session-selection.ts @@ -12,6 +12,7 @@ import { type SessionWorkflowSummary, } from './session-workflow.js' import type { HarnessTraceAdapter, SessionRef } from './types.js' +import { SOURCE_ATTRIBUTE_PREFIX, sourceFileId, type SourceRecordReference } from './source-location.js' export interface SessionSeedGroup { readonly adapter: HarnessTraceAdapter @@ -53,7 +54,7 @@ interface SessionSourceSnapshot { readonly files: readonly SessionSourceFile[] } -async function fileSha256(path: string, signal?: AbortSignal): Promise { +export async function fileSha256(path: string, signal?: AbortSignal): Promise { signal?.throwIfAborted() const hash = createHash('sha256') try { @@ -116,12 +117,24 @@ async function boundSessionRow( signal?: AbortSignal, ): Promise { const before = await sourceSnapshot(adapter, ref, signal) - const spans = await parseSession(adapter, ref, { taskScope, taskTurnId, signal }) + const spans = await parseSession(adapter, ref, { taskScope, taskTurnId, signal, captureSources: true }) signal?.throwIfAborted() const after = await sourceSnapshot(adapter, ref, signal) if (JSON.stringify(before) !== JSON.stringify(after)) { throw new Error(`session source changed while parsing; refusing unbound evidence: ${ref.path}`) } + const digests = new Map(after.files.map((file) => [sourceFileId(file.path), file.sha256])) + for (const span of spans) { + for (const [key, value] of Object.entries(span.attributes)) { + if (!key.startsWith(SOURCE_ATTRIBUTE_PREFIX) || typeof value !== 'string') continue + const refs = JSON.parse(value) as SourceRecordReference[] + span.attributes[key] = JSON.stringify(refs.map((record) => { + const sourceSha256 = digests.get(record.sourceId) + if (!sourceSha256) throw new Error('adapter source reference is outside its declared source files') + return { ...record, sourceSha256 } + })) + } + } if ( expectedRelationship && JSON.stringify(describeSessionRelationship(ref, spans)) !== JSON.stringify(expectedRelationship) diff --git a/src/source-location.ts b/src/source-location.ts new file mode 100644 index 0000000..fc975da --- /dev/null +++ b/src/source-location.ts @@ -0,0 +1,99 @@ +import { createHash } from 'node:crypto' +import { resolve } from 'node:path' + +export const SOURCE_ATTRIBUTE_PREFIX = 'traces.source_record.' + +/** Location of original bytes, never a copy of their content. */ +export interface SourceRecordReference { + sourceId: string + sourceSha256?: string + recordOffset: number + recordBytes: number + recordSha256: string + fieldLocator: string +} + +export type SourceReferences = SourceRecordReference | readonly SourceRecordReference[] | undefined + +const locations = new WeakMap() + +export function sourceFileId(path: string): string { + return createHash('sha256').update(resolve(path)).digest('hex') +} + +function pointerPart(value: string): string { + return value.replaceAll('~', '~0').replaceAll('/', '~1') +} + +/** Object metadata expires with parsed records and does not alter source projections. */ +export function locateSourceObjects(value: unknown, path: string, bytes: Buffer, offset = 0): void { + if (!value || typeof value !== 'object') return + const record = { + sourceId: sourceFileId(path), + recordOffset: offset, + recordBytes: bytes.length, + recordSha256: createHash('sha256').update(bytes).digest('hex'), + } + const pending: Array<{ value: object; pointer: string }> = [{ value, pointer: '#' }] + while (pending.length > 0) { + const current = pending.pop()! + locations.set(current.value, { ...record, fieldLocator: current.pointer }) + for (const [key, child] of Object.entries(current.value)) { + if (child && typeof child === 'object') { + pending.push({ value: child, pointer: `${current.pointer}/${pointerPart(key)}` }) + } + } + } +} + +export function sourceOf(value: unknown, ...fields: string[]): SourceRecordReference | undefined { + if (!value || typeof value !== 'object') return undefined + const location = locations.get(value) + return location ? { + ...location, + fieldLocator: `${location.fieldLocator}${fields.map((field) => `/${pointerPart(field)}`).join('')}`, + } : undefined +} + +/** Conversation citations can read text leaves, never adjacent tool blocks. */ +export function textSources(value: unknown, field: string): SourceReferences { + if (!value || typeof value !== 'object') return undefined + const content = (value as Record)[field] + if (typeof content === 'string') return sourceOf(value, field) + if (!Array.isArray(content)) return undefined + return content.flatMap((block, index) => { + if (typeof block === 'string') { + const ref = sourceOf(value, field, String(index)) + return ref ? [ref] : [] + } + if (!block || typeof block !== 'object' || typeof block.text !== 'string') return [] + if (block.type !== undefined && !['text', 'input_text', 'output_text'].includes(block.type)) return [] + const ref = sourceOf(block, 'text') + return ref ? [ref] : [] + }) +} + +export function sourceAttributes(attribute: string, sources: SourceReferences): Record { + const refs = sources === undefined ? [] : Array.isArray(sources) ? sources : [sources] + return refs.length === 0 ? {} : { [`${SOURCE_ATTRIBUTE_PREFIX}${attribute}`]: JSON.stringify(refs) } +} + +export function appendSourceAttributes(attributes: Record, attribute: string, sources: SourceReferences): void { + const key = `${SOURCE_ATTRIBUTE_PREFIX}${attribute}` + const added = sourceAttributes(attribute, sources)[key] + if (typeof added !== 'string') return + const prior = attributes[key] + const refs = typeof prior === 'string' ? JSON.parse(prior) as SourceRecordReference[] : [] + const known = new Set(refs.map((ref) => JSON.stringify(ref))) + for (const ref of JSON.parse(added) as SourceRecordReference[]) { + if (!known.has(JSON.stringify(ref))) { + refs.push(ref) + known.add(JSON.stringify(ref)) + } + } + attributes[key] = JSON.stringify(refs) +} + +export function stripSourceAttributes(attributes: Readonly>): Record { + return Object.fromEntries(Object.entries(attributes).filter(([key]) => !key.startsWith(SOURCE_ATTRIBUTE_PREFIX))) +} diff --git a/src/types.ts b/src/types.ts index 311633d..e242928 100644 --- a/src/types.ts +++ b/src/types.ts @@ -16,6 +16,8 @@ import type { JsonlCorruptionReceipt } from './jsonl.js' export type CorruptionMode = 'recover' | 'strict' export interface ParseOptions { + /** Attach source locations for a caller that will retain the original files. */ + captureSources?: boolean /** Recover valid JSONL records by default; strict rejects the first corruption. */ corruptionMode?: CorruptionMode /** For resumed session formats, parse all turns, the latest turn, or one exact turn. */ diff --git a/src/upload.ts b/src/upload.ts index 20cb4f9..c61bd88 100644 --- a/src/upload.ts +++ b/src/upload.ts @@ -11,6 +11,8 @@ * final privacy options, deduplicates, and sends or writes a dry-run preview. */ +import { stripSourceAttributes } from './source-location.js' + import { readFile } from 'node:fs/promises' import { createRequire } from 'node:module' import { hostname } from 'node:os' @@ -190,7 +192,7 @@ export interface ExecuteOptions { /** Drop captured conversation and tool values from metadata-only uploads. */ function stripSpanContent(spans: readonly OtlpSpan[]): OtlpSpan[] { return spans.map((s) => { - const attributes = { ...s.attributes } + const attributes = stripSourceAttributes(s.attributes) delete attributes.content for (const key of TOOL_IO_VALUE_KEYS) delete attributes[key] normalizeToolIoAttributes(attributes) diff --git a/tests/bundle-source.test.ts b/tests/bundle-source.test.ts new file mode 100644 index 0000000..bf16155 --- /dev/null +++ b/tests/bundle-source.test.ts @@ -0,0 +1,226 @@ +import { execFile } from 'node:child_process' +import { promisify } from 'node:util' +import { createHash } from 'node:crypto' +import { cp, mkdtemp, readFile, rm, symlink, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { AnalystRegistry, type Analyst } from '@tangle-network/agent-eval/analyst' +import { buildTraceAnalysisToolDescriptors, OtlpFileTraceStore, type ReadSpanSourceInput, type ReadSpanSourceResult, type TraceAnalysisStore } from '@tangle-network/agent-eval/traces' +import { afterAll, beforeAll, describe, expect, it } from 'vitest' +import { ClaudeAdapter } from '../src/adapters/claude.js' +import { analyzeSpans } from '../src/analyze.js' +import { assembleSessionBundle, type SessionBundleManifest } from '../src/bundle.js' +import { createBundleSourceReader } from '../src/bundle-source.js' +import { runTraceImprovement } from '../src/improvement.js' +import { readOtlpInput } from '../src/otlp-input.js' +import type { OtlpSpan } from '../src/otlp.js' +import { redactSpans } from '../src/redact.js' +import { SOURCE_ATTRIBUTE_PREFIX } from '../src/source-location.js' + +let root: string +let bundle: string +let spans: readonly OtlpSpan[] +let input: ReadSpanSourceInput +const answer = `${'a'.repeat(18_000)}🙂 source-only-answer-tail` +const secret = `${'z'.repeat(18_000)} sibling-tool-input-tail` +const output = `${'o'.repeat(18_000)} source-only-output-tail` + +beforeAll(async () => { + root = await mkdtemp(join(tmpdir(), 'traces-source-test-')) + bundle = join(root, 'bundle') + const path = join(root, 'session.jsonl') + await writeFile(path, [ + { type: 'user', uuid: 'user', sessionId: 'source-test', timestamp: '2026-01-01T00:00:00Z', message: { role: 'user', content: 'Inspect source' } }, + { type: 'assistant', uuid: 'assistant', sessionId: 'source-test', timestamp: '2026-01-01T00:00:01Z', message: { id: 'message', role: 'assistant', content: [ + { type: 'text', text: answer }, + { type: 'tool_use', id: 'call', name: 'Read', input: { path: secret } }, + ] } }, + { type: 'user', uuid: 'result', sessionId: 'source-test', timestamp: '2026-01-01T00:00:02Z', message: { role: 'user', content: [{ type: 'tool_result', tool_use_id: 'call', content: output }] } }, + ].map((record) => JSON.stringify(record)).join('\n')) + await assembleSessionBundle({ adapter: new ClaudeAdapter(), ref: { harness: 'claude-code', sessionId: 'source-test', path, cwd: null, mtimeMs: 0 }, outDir: bundle }) + spans = (await readOtlpInput(join(bundle, 'derived/trace.otlp.jsonl'))).spans + const assistant = spans.find((span) => typeof span.attributes.content === 'string' && span.attributes.content.startsWith('aaaa'))! + input = { trace_id: assistant.trace_id, span_id: assistant.span_id, attribute: 'content', offset: 18_000, limit: 100 } + await rm(path) +}) +afterAll(async () => { if (root) await rm(root, { recursive: true, force: true }) }) + +async function copiedBundle(): Promise { + const destination = await mkdtemp(join(root, 'copy-')) + await cp(bundle, destination, { recursive: true }) + return destination +} + +async function sourcePath(directory: string): Promise { + const manifest = JSON.parse(await readFile(join(directory, 'manifest.json'), 'utf8')) as SessionBundleManifest + return join(directory, manifest.files.find((file) => file.sourceId)!.path) +} + +describe('retained source capability', () => { + it('reads omitted source text through the canonical tool after originals rotate', async () => { + const reader = await createBundleSourceReader(bundle, spans) + const store = new OtlpFileTraceStore({ path: join(bundle, 'derived/trace.otlp.jsonl'), sourceReader: reader }) + const tool = buildTraceAnalysisToolDescriptors({ store }).find((tool) => tool.name === 'readSpanSource')! + const result = await tool.handler(input) as ReadSpanSourceResult + expect(result).toMatchObject({ status: 'available', text: '🙂 source-only-answer-tail', offset: 18_000, total_bytes: Buffer.byteLength(answer), next_offset: null, source: { field_locator: '#/message/content/0/text', value_encoding: 'utf8-string' } }) + expect(spans.find((span) => span.span_id === input.span_id)!.attributes.content).not.toContain('source-only-answer-tail') + expect(JSON.stringify(result)).not.toContain('sibling-tool-input-tail') + expect(JSON.stringify(result)).not.toContain(root) + if (result.status === 'available') expect(result.source.source_sha256).toMatch(/^[a-f0-9]{64}$/) + expect(await reader!({ ...input, source_index: 1 })).toMatchObject({ status: 'unavailable' }) + await expect(tool.handler({ ...input, limit: 16_385 })).rejects.toThrow() + const largerStore = new OtlpFileTraceStore({ path: join(bundle, 'derived/trace.otlp.jsonl'), sourceReader: reader, perAttributeSpanBudget: 32_768 }) + const largerTool = buildTraceAnalysisToolDescriptors({ store: largerStore, budgets: { perAttributeSpanBudget: 32_768 } }).find((tool) => tool.name === 'readSpanSource')! + const larger = await largerTool.handler({ ...input, offset: 0, limit: 17_000 }) as ReadSpanSourceResult + expect(larger).toMatchObject({ status: 'available', next_offset: 17_000 }) + if (larger.status === 'available') expect(Buffer.byteLength(larger.text)).toBe(17_000) + }) + + it('recovers tool input and output tails using their own field locators', async () => { + const reader = await createBundleSourceReader(bundle, spans) + const tool = spans.find((span) => span.attributes['tool.name'] === 'Read')! + for (const [attribute, tail, encoding] of [['input.value', 'sibling-tool-input-tail', 'json'], ['output.value', 'source-only-output-tail', 'utf8-string']]) { + const result = await reader!({ ...input, span_id: tool.span_id, attribute: attribute!, offset: 18_000, limit: 100 }) + expect(result.status).toBe('available') + if (result.status === 'available') { + expect(result.text).toContain(tail) + expect(result.source.value_encoding).toBe(encoding) + } + } + }) + + it('wires explicit authorization into the actual analysis registry', async () => { + const observed: ReadSpanSourceResult[] = [] + const registry = new AnalystRegistry() + const analyst: Analyst = { id: 'source-reader-proof', description: 'Read a retained field', inputKind: 'trace-store', cost: { kind: 'deterministic' }, version: '1', async analyze(store) { + const tool = buildTraceAnalysisToolDescriptors({ store }).find((tool) => tool.name === 'readSpanSource')! + observed.push(await tool.handler(input) as ReadSpanSourceResult) + return [] + } } + registry.register(analyst) + await analyzeSpans(spans, { registry, sourceBundle: { path: bundle }, otlpOutPath: join(root, 'analysis.jsonl') }) + expect(observed).toHaveLength(1) + expect(observed[0]).toMatchObject({ status: 'available', text: '🙂 source-only-answer-tail' }) + const store = new OtlpFileTraceStore({ path: join(bundle, 'derived/trace.otlp.jsonl') }) + expect(store.readSpanSource).toBeUndefined() + expect(buildTraceAnalysisToolDescriptors({ store }).some((tool) => tool.name === 'readSpanSource')).toBe(false) + }) + + it('retains streamed message records without turning replayed UUIDs into conflicts', async () => { + const path = join(root, 'streamed.jsonl') + const first = { type: 'assistant', uuid: 'first', sessionId: 'streamed', timestamp: '2026-01-01T00:00:00Z', message: { id: 'shared-message', role: 'assistant', content: [{ type: 'text', text: 'first fragment' }] } } + const second = { ...first, uuid: 'second', message: { ...first.message, content: [{ type: 'text', text: 'second fragment' }] } } + await writeFile(path, [first, first, second].map((event) => JSON.stringify(event)).join('\n')) + const destination = join(root, 'streamed-bundle') + await assembleSessionBundle({ adapter: new ClaudeAdapter(), ref: { harness: 'claude-code', sessionId: 'streamed', path, cwd: null, mtimeMs: 0 }, outDir: destination }) + const selected = (await readOtlpInput(join(destination, 'derived/trace.otlp.jsonl'))).spans + const message = selected.find((span) => span.attributes.content === 'first fragment\nsecond fragment')! + expect(message).toBeDefined() + await rm(path) + const reader = await createBundleSourceReader(destination, selected) + for (const [source_index, text] of ['first fragment', 'second fragment'].entries()) { + expect(await reader!({ trace_id: message.trace_id, span_id: message.span_id, attribute: 'content', offset: 0, limit: 100, source_index })).toMatchObject({ status: 'available', text, source_index }) + } + expect(await reader!({ trace_id: message.trace_id, span_id: message.span_id, attribute: 'content', offset: 0, limit: 100, source_index: 2 })).toMatchObject({ status: 'unavailable' }) + }) + + it('accepts the CLI bundle input and rejects mixed input or archive output paths', async () => { + const run = promisify(execFile) + const cli = ['--import', 'tsx', 'src/cli.ts', 'analyze', '--source-bundle', bundle] + const report = join(root, 'cli-report.md') + await run(process.execPath, [...cli, '--out', report], { timeout: 30_000 }) + expect((await readFile(report, 'utf8')).length).toBeGreaterThan(0) + await expect(run(process.execPath, [...cli, '--otlp', join(bundle, 'derived/trace.otlp.jsonl')], { timeout: 30_000 })).rejects.toThrow('cannot be combined') + const manifestPath = join(bundle, 'manifest.json') + const before = await readFile(manifestPath) + await expect(run(process.execPath, [...cli, '--out', manifestPath], { timeout: 30_000 })).rejects.toThrow('outside') + expect(await readFile(manifestPath)).toEqual(before) + }) + + it('rejects redacted, changed, and cross-trace attributes', async () => { + expect(await createBundleSourceReader(bundle, redactSpans(spans, []).spans)).toBeUndefined() + const selected = structuredClone(spans.find((span) => span.span_id === input.span_id)!) + selected.attributes.content = 'changed' + expect(await createBundleSourceReader(bundle, [selected])).toBeUndefined() + const reader = await createBundleSourceReader(bundle, spans) + expect(await reader!({ ...input, trace_id: 'another-trace' })).toMatchObject({ status: 'unavailable' }) + expect(await reader!({ ...input, attribute: 'unrecorded' })).toMatchObject({ status: 'unavailable' }) + }) + + it('enforces field byte ranges, UTF-8 boundaries, cancellation, and configured parsing limits', async () => { + const reader = await createBundleSourceReader(bundle, spans) + expect(await reader!({ ...input, limit: 4 })).toMatchObject({ status: 'available', text: '🙂', next_offset: 18_004 }) + for (const update of [{ offset: 18_001 }, { limit: 3 }, { offset: -1 }, { offset: 0.5 }, { offset: Buffer.byteLength(answer) + 1 }]) { + expect(await reader!({ ...input, ...update })).toMatchObject({ status: 'unavailable' }) + } + expect(await reader!({ ...input, offset: Buffer.byteLength(answer) })).toMatchObject({ status: 'available', text: '', next_offset: null }) + const limited = await createBundleSourceReader(bundle, spans, { maxRecordBytes: 100 }) + expect(await limited!(input)).toMatchObject({ status: 'unavailable', reason: 'source record exceeds the configured parsing limit' }) + expect(await reader!({ ...input, limit: 16_385 })).toMatchObject({ status: 'available' }) + const controller = new AbortController() + controller.abort() + await expect(reader!(input, { signal: controller.signal })).rejects.toThrow() + const parent = new AbortController() + const cancellable = await createBundleSourceReader(bundle, spans, { signal: parent.signal }) + parent.abort() + await expect(cancellable!(input, { signal: new AbortController().signal })).rejects.toThrow() + }) + + it.each(['missing', 'same-size-tamper', 'symlink'])('rejects %s retained source', async (mode) => { + const directory = await copiedBundle() + const reader = await createBundleSourceReader(directory, spans) + const path = await sourcePath(directory) + const bytes = await readFile(path) + await rm(path) + if (mode === 'same-size-tamper') { bytes[0] = 32; await writeFile(path, bytes) } + if (mode === 'symlink') await symlink(await sourcePath(bundle), path) + expect(await reader!(input)).toMatchObject({ status: 'unavailable' }) + }) + + it('rejects evidence-only manifests and unsafe receipt paths', async () => { + for (const mode of ['evidence-only', 'path']) { + const directory = await copiedBundle() + const path = join(directory, 'manifest.json') + const manifest = JSON.parse(await readFile(path, 'utf8')) + if (mode === 'evidence-only') manifest.view = 'evidence-only' + else manifest.files[0].path = '../outside' + await writeFile(path, JSON.stringify(manifest)) + await expect(createBundleSourceReader(directory, spans)).rejects.toThrow() + } + }) + + it('detects record hash mismatch independently of the source receipt', async () => { + const directory = await copiedBundle() + const tracePath = join(directory, 'derived/trace.otlp.jsonl') + const raw = await readFile(tracePath, 'utf8') + const key = `${SOURCE_ATTRIBUTE_PREFIX}content` + const ref = JSON.parse(String(spans.find((span) => span.span_id === input.span_id)!.attributes[key]))[0] + const altered = raw.replaceAll(ref.recordSha256, '0'.repeat(64)) + await writeFile(tracePath, altered) + const manifestPath = join(directory, 'manifest.json') + const manifest = JSON.parse(await readFile(manifestPath, 'utf8')) + const receipt = manifest.files.find((file: { path: string }) => file.path === 'derived/trace.otlp.jsonl') + receipt.bytes = Buffer.byteLength(altered) + receipt.sha256 = createHash('sha256').update(altered).digest('hex') + await writeFile(manifestPath, JSON.stringify(manifest)) + const selected = (await readOtlpInput(tracePath)).spans + const reader = await createBundleSourceReader(directory, selected) + expect(await reader!(input)).toMatchObject({ status: 'unavailable', reason: 'retained record digest mismatch or source changed during read' }) + }) + + it('protects retained artifacts from analysis output writes', async () => { + const path = join(bundle, 'derived/trace.otlp.jsonl') + const before = await readFile(path) + await expect(analyzeSpans(spans, { sourceBundle: { path: bundle }, otlpOutPath: path })).rejects.toThrow('outside') + expect(await readFile(path)).toEqual(before) + await expect( + runTraceImprovement({ + spans, + sourceBundle: { path: bundle }, + outDir: bundle, + otlpOutPath: join(root, 'outside.jsonl'), + harness: 'test', + }), + ).rejects.toThrow('outside') + }) +}) diff --git a/tests/source-location.test.ts b/tests/source-location.test.ts new file mode 100644 index 0000000..b6ada73 --- /dev/null +++ b/tests/source-location.test.ts @@ -0,0 +1,163 @@ +import { createHash } from 'node:crypto' +import { mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { dirname, join } from 'node:path' +import { afterAll, describe, expect, it } from 'vitest' +import { AmpAdapter } from '../src/adapters/amp.js' +import { ClaudeAdapter } from '../src/adapters/claude.js' +import { CodexAdapter } from '../src/adapters/codex.js' +import { CodexExecAdapter } from '../src/adapters/codex-exec.js' +import { CopilotAdapter } from '../src/adapters/copilot.js' +import { FactoryAdapter } from '../src/adapters/factory.js' +import { ForgeAdapter } from '../src/adapters/forge.js' +import { GeminiAdapter } from '../src/adapters/gemini.js' +import { OpencodeAdapter } from '../src/adapters/opencode.js' +import { PiAdapter } from '../src/adapters/pi.js' +import { QwenAdapter } from '../src/adapters/qwen.js' +import { assembleSessionBundle } from '../src/bundle.js' +import { createBundleSourceReader } from '../src/bundle-source.js' +import { readOtlpInput } from '../src/otlp-input.js' +import { canonicalJson } from '../src/adapters/tool-io.js' +import { readJsonl } from '../src/jsonl.js' +import { collectSessionSelection } from '../src/session-selection.js' +import { SOURCE_ATTRIBUTE_PREFIX, sourceOf, type SourceRecordReference } from '../src/source-location.js' +import type { HarnessTraceAdapter, SessionRef } from '../src/types.js' + +const directories: string[] = [] +afterAll(async () => { for (const directory of directories) await rm(directory, { recursive: true, force: true }) }) +const text = `${'message '.repeat(2300)}message-tail` +const args = { path: `${'input '.repeat(3000)}input-tail` } +const output = `${'output '.repeat(2600)}output-tail` +const blocks = [{ type: 'text', text }, { type: 'tool_use', id: 'call', name: 'Read', input: args }] +const resultBlocks = [{ type: 'tool_result', tool_use_id: 'call', content: output }] +const timestamp = '2026-01-01T00:00:00Z' +const cases: Array<{ harness: string; adapter: HarnessTraceAdapter; json?: unknown; lines?: unknown[] }> = [ + { harness: 'amp', adapter: new AmpAdapter(), json: { id: 'fixture', messages: [{ role: 'assistant', messageId: 1, content: blocks }, { role: 'user', content: resultBlocks }] } }, + { harness: 'claude-code', adapter: new ClaudeAdapter(), lines: [ + { type: 'assistant', uuid: 'a', sessionId: 'fixture', timestamp, message: { id: 'm', role: 'assistant', content: blocks } }, + { type: 'user', uuid: 'b', sessionId: 'fixture', timestamp, message: { role: 'user', content: resultBlocks } }, + ] }, + { harness: 'factory', adapter: new FactoryAdapter(), lines: [ + { type: 'session_start', id: 'fixture', timestamp }, + { type: 'message', id: 'a', timestamp, message: { role: 'assistant', content: blocks } }, + { type: 'message', id: 'b', timestamp, message: { role: 'user', content: resultBlocks } }, + ] }, + { harness: 'github-copilot', adapter: new CopilotAdapter(), lines: [ + { type: 'assistant.message', timestamp, data: { messageId: 'm', content: text } }, + { type: 'tool.execution_start', timestamp, data: { toolCallId: 'call', toolName: 'Read', arguments: args } }, + { type: 'tool.execution_complete', timestamp, data: { toolCallId: 'call', success: true, output } }, + ] }, + { harness: 'qwen', adapter: new QwenAdapter(), lines: [ + { type: 'assistant', sessionId: 'fixture', timestamp, message: { role: 'model', parts: [{ text }, { functionCall: { name: 'Read', args } }] } }, + { type: 'tool_result', sessionId: 'fixture', timestamp, message: { role: 'user', parts: [{ functionResponse: { name: 'Read', response: output } }] } }, + ] }, + { harness: 'pi', adapter: new PiAdapter(), lines: [ + { type: 'session', id: 'fixture', timestamp }, + { type: 'message', id: 'm', timestamp, message: { role: 'assistant', content: [{ type: 'text', text }, { type: 'tool_call', id: 'call', toolName: 'Read', input: args }] } }, + { type: 'message', id: 'r', timestamp, message: { role: 'toolResult', toolCallId: 'call', toolName: 'Read', content: [{ type: 'text', text: output }] } }, + ] }, + { harness: 'gemini', adapter: new GeminiAdapter(), json: { sessionId: 'fixture', startTime: timestamp, messages: [{ id: 'm', type: 'assistant', timestamp, content: text, toolCalls: [{ id: 'call', name: 'Read', args, result: output, status: 'ok' }] }] } }, + { harness: 'gemini', adapter: new GeminiAdapter(), json: { sessionId: 'fixture', startTime: timestamp, messages: [{ id: 'm', type: 'assistant', timestamp, content: { text, structured: true }, toolCalls: [{ id: 'call', name: 'Read', args, result: output, status: 'ok' }] }] } }, + { harness: 'forge', adapter: new ForgeAdapter(), json: { conversation_id: 'fixture', messages: [ + { text: { role: 'assistant', content: text, tool_calls: [{ name: 'Read', call_id: 'call', arguments: args }] } }, + { tool: { name: 'Read', call_id: 'call', output: { is_error: false, values: output } } }, + ] } }, + { harness: 'codex', adapter: new CodexAdapter(), lines: [ + { type: 'session_meta', timestamp, payload: { id: 'fixture' } }, + { type: 'response_item', timestamp, payload: { type: 'message', role: 'assistant', content: [{ type: 'output_text', text }] } }, + { type: 'response_item', timestamp, payload: { type: 'function_call', call_id: 'call', name: 'Read', arguments: JSON.stringify(args) } }, + { type: 'response_item', timestamp, payload: { type: 'function_call_output', call_id: 'call', output } }, + ] }, + { harness: 'codex-exec', adapter: new CodexExecAdapter(), lines: [ + { type: 'thread.started', thread_id: 'fixture' }, + { type: 'turn.started' }, + { type: 'item.completed', item: { id: 'm', type: 'agent_message', text } }, + { type: 'item.started', item: { id: 'call', type: 'command_execution', command: args.path, status: 'in_progress' } }, + { type: 'item.completed', item: { id: 'call', type: 'command_execution', command: args.path, aggregated_output: output, exit_code: 0, status: 'completed' } }, + { type: 'turn.completed', usage: { input_tokens: 1, output_tokens: 1 } }, + ] }, +] + +async function verifyLocations(adapter: HarnessTraceAdapter, ref: SessionRef): Promise { + const selection = await collectSessionSelection([{ adapter, refs: [ref] }], { bindSources: true }) + const row = selection.rows[0]! + const found = new Set() + for (const span of row.spans) { + for (const attribute of ['content', 'input.value', 'output.value']) { + const encoded = span.attributes[`${SOURCE_ATTRIBUTE_PREFIX}${attribute}`] + if (typeof encoded !== 'string') continue + for (const reference of JSON.parse(encoded) as SourceRecordReference[]) { + const file = row.sourceFiles!.find((file) => file.sha256 === reference.sourceSha256)! + const bytes = await readFile(file.path) + const raw = bytes.subarray(reference.recordOffset, reference.recordOffset + reference.recordBytes) + expect(createHash('sha256').update(raw).digest('hex')).toBe(reference.recordSha256) + let field: unknown = JSON.parse(raw.toString('utf8')) + for (const key of reference.fieldLocator.slice(2).split('/')) field = (field as Record)[key.replaceAll('~1', '/').replaceAll('~0', '~')] + const value = typeof field === 'string' ? field : canonicalJson(field) + const tail = attribute === 'content' ? 'message-tail' : attribute === 'input.value' ? 'input-tail' : 'output-tail' + expect(value).toContain(tail) + if (attribute === 'content') expect(value).not.toContain('input-tail') + expect(String(span.attributes[attribute])).not.toContain(tail) + found.add(attribute) + } + } + } + expect([...found].sort()).toEqual(['content', 'input.value', 'output.value']) + if (ref.harness === 'codex' || ref.harness === 'opencode') { + const destination = await mkdtemp(join(tmpdir(), 'native-retained-')) + directories.push(destination) + await assembleSessionBundle({ adapter, ref, outDir: destination }) + const retained = (await readOtlpInput(join(destination, 'derived/trace.otlp.jsonl'))).spans + for (const file of row.sourceFiles!) await rm(file.path) + const reader = await createBundleSourceReader(destination, retained) + const message = retained.find((span) => span.attributes[`${SOURCE_ATTRIBUTE_PREFIX}content`] !== undefined)! + expect(await reader!({ trace_id: message.trace_id, span_id: message.span_id, attribute: 'content', offset: Buffer.byteLength(text) - 12, limit: 12 })).toMatchObject({ status: 'available', text: 'message-tail', next_offset: null }) + } +} + +describe('native retained source locations', () => { + it('preserves canonical BOM rejection and exact CRLF record receipts', async () => { + const directory = await mkdtemp(join(tmpdir(), 'native-source-bom-')) + directories.push(directory) + const path = join(directory, 'bom.jsonl') + const rejected = '\ufeff{"text":"rejected"}\r' + const accepted = '{"text":"accepted"}\r' + await writeFile(path, `${rejected}\n${accepted}\n`) + const corrupt: unknown[] = [] + const rows = [] + for await (const row of readJsonl(path, { captureSources: true, mode: 'recover', onCorruption: (receipt) => corrupt.push(receipt) })) rows.push(row) + expect(corrupt).toHaveLength(1) + expect(rows).toEqual([{ text: 'accepted' }]) + expect(sourceOf(rows[0], 'text')).toMatchObject({ recordOffset: Buffer.byteLength(rejected) + 1, recordBytes: Buffer.byteLength(accepted), recordSha256: createHash('sha256').update(accepted).digest('hex') }) + }) + it.each(cases)('$harness pins omitted values to the correct original fields', async ({ harness, adapter, json, lines }) => { + const directory = await mkdtemp(join(tmpdir(), 'native-source-')) + directories.push(directory) + const path = join(directory, json ? 'session.json' : 'session.jsonl') + await writeFile(path, json ? JSON.stringify(json) : lines!.map((line) => JSON.stringify(line)).join('\r\n')) + await verifyLocations(adapter, { harness, sessionId: 'fixture', path, cwd: null, mtimeMs: 0 }) + }) + + it('retains OpenCode message and part files from directory sources', async () => { + const directory = await mkdtemp(join(tmpdir(), 'native-source-opencode-')) + directories.push(directory) + const storage = join(directory, 'opencode/storage') + for (const [path, value] of [ + ['message/fixture/m.json', { id: 'm', role: 'assistant', time: { created: 1000 } }], + ['part/m/text.json', { type: 'text', text }], + ['part/m/tool.json', { type: 'tool', tool: 'Read', callID: 'call', state: { status: 'ok', input: args, output } }], + ] as const) { + const destination = join(storage, path) + await mkdir(dirname(destination), { recursive: true }) + await writeFile(destination, JSON.stringify(value)) + } + const before = process.env.XDG_DATA_HOME + process.env.XDG_DATA_HOME = directory + try { + await verifyLocations(new OpencodeAdapter(), { harness: 'opencode', sessionId: 'fixture', path: join(storage, 'message/fixture'), cwd: null, mtimeMs: 0 }) + } finally { + if (before === undefined) delete process.env.XDG_DATA_HOME + else process.env.XDG_DATA_HOME = before + } + }) +})