From ef9ebfdb45680ee18df412ea4d5c32121bd3cefe Mon Sep 17 00:00:00 2001 From: Will Washburn Date: Sun, 20 Sep 2026 13:56:48 -0700 Subject: [PATCH 1/2] feat(ingest): collect GitHub Copilot CLI usage via OTEL file exporter Copilot CLI emits one OTEL JSONL span per API call when the user sets COPILOT_OTEL_FILE_EXPORTER_PATH (tokscale-style: chat spans preferred, invoke_agent aggregates only as a per-trace fallback to avoid double counting). Burn now scans that env-var file plus $COPILOT_HOME/otel/*.jsonl on every ingest sweep, recording each usage-bearing span as a per-message copilot-cli TurnRecord with usage-only fidelity. - reader::copilot: incremental JSONL span parser with byte-offset resume, partial-tail protection, trace-id context fill for model/session, dotted + underscored cache attribute spellings, and input tokens normalized to cache-exclusive. - ingest: CopilotCursor (inode/offset plus per-session turn counters and chat-trace set) so exporter rotation re-reads without double counting and turn_index survives across passes; env-gated sources no-op silently when unset. - New `burn init copilot` prints the exact exporter setup (nothing is recoverable retroactively, so the opt-in has to come first). Closes #14 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 2 + README.md | 32 +- crates/relayburn-cli/src/cli.rs | 20 + crates/relayburn-cli/src/commands/init.rs | 40 ++ crates/relayburn-cli/src/commands/mod.rs | 1 + crates/relayburn-cli/src/main.rs | 2 + crates/relayburn-cli/tests/smoke.rs | 16 + crates/relayburn-sdk/src/analyze/provider.rs | 1 + crates/relayburn-sdk/src/ingest.rs | 3 +- crates/relayburn-sdk/src/ingest/cursors.rs | 22 +- crates/relayburn-sdk/src/ingest/gap.rs | 2 + .../src/ingest/gap_warning_tests.rs | 1 + crates/relayburn-sdk/src/ingest/ingest.rs | 221 ++++++- .../src/ingest/orchestration_tests.rs | 94 ++- crates/relayburn-sdk/src/ingest_verb.rs | 1 + crates/relayburn-sdk/src/lib.rs | 11 +- crates/relayburn-sdk/src/reader.rs | 5 + crates/relayburn-sdk/src/reader/copilot.rs | 562 ++++++++++++++++++ .../relayburn-sdk/src/reader/copilot/tests.rs | 195 ++++++ crates/relayburn-sdk/src/reader/types.rs | 2 + crates/relayburn-sdk/tests/integration.rs | 2 + tests/fixtures/copilot-cli/chat-spans.jsonl | 10 + 22 files changed, 1218 insertions(+), 27 deletions(-) create mode 100644 crates/relayburn-cli/src/commands/init.rs create mode 100644 crates/relayburn-sdk/src/reader/copilot.rs create mode 100644 crates/relayburn-sdk/src/reader/copilot/tests.rs create mode 100644 tests/fixtures/copilot-cli/chat-spans.jsonl diff --git a/CHANGELOG.md b/CHANGELOG.md index fde707b5..5d66801c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,8 @@ Cross-package release notes for relayburn. Package changelogs contain package-le ## [Unreleased] +- `burn ingest` collects GitHub Copilot CLI usage from the OpenTelemetry file exporter while `COPILOT_OTEL_FILE_EXPORTER_PATH` is set (the exporter file plus `~/.copilot/otel/*.jsonl`), recording per-API-call token usage as `copilot-cli` turns with usage-only fidelity; the opt-in setup is printed by the new `burn init copilot` helper. + ## [4.1.0] - 2026-09-20 - `burn measure` and `@relayburn/sdk.measureSession()` turn one explicit Claude Code, Codex, or OpenCode session source into a versioned per-model token/cost document without discovery or a ledger; incomplete and zero-turn inputs fail closed. diff --git a/README.md b/README.md index 2ed402db..86551885 100644 --- a/README.md +++ b/README.md @@ -1,8 +1,9 @@ ![relayburn](./burn-readme-banner.png) Understand how you're spending tokens in agent CLIs. Burn ingests Claude Code, -Codex, and OpenCode session logs into a local ledger, then shows cost by model, -provider, tool, file, workflow, agent, session, and overhead file. +Codex, and OpenCode session logs — plus GitHub Copilot CLI's opt-in OTEL span +export — into a local ledger, then shows cost by model, provider, tool, file, +workflow, agent, session, and overhead file. ## Quick Start @@ -29,6 +30,7 @@ Burn stores data under `~/.agentworkforce/burn/` by default. Set | [`burn flow`](#burn-flow) | Render a session's inference and subagent flow as Mermaid, SVG, or JSON. | | [`burn stamps`](#burn-stamps) | Export enrichment stamps as JSONL. | | [`burn ingest`](#burn-ingest) | Import existing or live session logs without wrapping the harness. | +| [`burn init`](#collector-setup) | Print setup instructions for opt-in collectors (Copilot CLI). | | [`burn mcp-server`](#burn-mcp-server) | Expose read-only cost queries to an agent through stdio MCP. | | [`burn update`](#burn-update) | Check for releases, install an update, or configure automatic checks. | @@ -205,7 +207,8 @@ Run `burn summary --by-provider` to discover model IDs present in your ledger. ## `burn ingest` Use `burn ingest` when sessions already exist, or when another process owns the -harness spawn. Default mode scans Claude Code, Codex, and OpenCode stores once. +harness spawn. Default mode scans Claude Code, Codex, and OpenCode stores once, +plus the GitHub Copilot CLI OTEL export when it is enabled (see below). | Option | What it does | |---|---| @@ -213,6 +216,23 @@ harness spawn. Default mode scans Claude Code, Codex, and OpenCode stores once. | `--interval ` | Poll interval in milliseconds. Default: `1000`. | | `--quiet` | Suppress stderr progress spinner / breadcrumbs. One-shot mode still writes the final summary on stdout. | | `--hook claude` | Read one Claude Code hook payload from stdin and ingest its single transcript via the SDK fast-path. | + +### Collector setup + +**GitHub Copilot CLI** does not write a session log; it emits usage through its +OpenTelemetry file exporter, and only when you opt in: + +```bash +export COPILOT_OTEL_FILE_EXPORTER_PATH="$HOME/.copilot/otel/copilot.jsonl" +``` + +Add that to your shell rc file, restart the shell, and subsequent `copilot` +sessions emit one JSONL span per API call, which `burn ingest` picks up +automatically (files under `~/.copilot/otel/` are scanned even without the env +var on the ingest side). Nothing is recoverable retroactively — the exporter +must be on before sessions are captured. Run `burn init copilot` to print these +instructions. Coverage is usage-only: spans carry per-call token counts (input, +output, cache read/write, reasoning) and the model, but no tool-call content. | `--no-fsevents` | In watch mode, use polling instead of filesystem events. | | Example | Result | @@ -497,7 +517,8 @@ burn ingest burn ingest --watch --interval 1000 ``` -`burn ingest` scans Claude, Codex, and OpenCode stores once and uses the same +`burn ingest` scans Claude, Codex, and OpenCode stores (plus the Copilot CLI +OTEL export when enabled) once and uses the same cursor and dedup path as the reporting commands. `burn ingest --watch` keeps that scan loop running in the foreground. @@ -521,7 +542,8 @@ the same session. ### What does `burn ingest` do? Each harness (Claude Code, Codex, OpenCode) writes its own session transcripts -to disk in its own format. `burn ingest` reads those transcripts, normalizes +to disk in its own format; GitHub Copilot CLI instead appends OTEL spans to the +file named by `COPILOT_OTEL_FILE_EXPORTER_PATH` when that export is enabled. `burn ingest` reads those transcripts, normalizes them, and writes them into burn's local SQLite ledger so the query commands (`summary`, `hotspots`, `overhead`, `compare`) have something to read against. Three modes: diff --git a/crates/relayburn-cli/src/cli.rs b/crates/relayburn-cli/src/cli.rs index b960c328..e397ff38 100644 --- a/crates/relayburn-cli/src/cli.rs +++ b/crates/relayburn-cli/src/cli.rs @@ -126,6 +126,26 @@ pub enum Command { /// Check for, install, or configure `burn` self-updates. Update(UpdateArgs), + + /// Print setup instructions for collectors that need opt-in + /// configuration (e.g. GitHub Copilot CLI's OTEL exporter). + Init(InitArgs), +} + +/// Per-command flags for `burn init`. +#[derive(Debug, Clone, ClapArgs)] +pub struct InitArgs { + #[command(subcommand)] + pub action: InitAction, +} + +/// Nested subcommand for `burn init`. Required — `burn init` on its own +/// would have nothing to print. +#[derive(Debug, Clone, Subcommand)] +pub enum InitAction { + /// Print the COPILOT_OTEL_FILE_EXPORTER_PATH setup for the GitHub + /// Copilot CLI collector. + Copilot, } #[derive(Debug, Clone, ClapArgs)] diff --git a/crates/relayburn-cli/src/commands/init.rs b/crates/relayburn-cli/src/commands/init.rs new file mode 100644 index 00000000..6f18fe95 --- /dev/null +++ b/crates/relayburn-cli/src/commands/init.rs @@ -0,0 +1,40 @@ +//! `burn init` — setup helpers for collectors that need opt-in +//! configuration before burn can see their data. +//! +//! `burn init copilot` prints the exact `COPILOT_OTEL_FILE_EXPORTER_PATH` +//! setup for the GitHub Copilot CLI collector (#14). It never touches the +//! user's shell rc files — Copilot's exporter is opt-in and which rc file +//! to edit is the user's call, so this just prints copy-pasteable lines. + +use crate::cli::InitArgs; + +pub fn run(args: InitArgs) -> i32 { + match args.action { + crate::cli::InitAction::Copilot => { + print!( + "GitHub Copilot CLI collector setup\n\ + \n\ + Copilot CLI only emits usage data when its OpenTelemetry file\n\ + exporter is enabled, and nothing is recoverable retroactively —\n\ + export must be on before sessions are captured.\n\ + \n\ + 1. Point the exporter at a file burn scans:\n\ + \n\ + \x20 export COPILOT_OTEL_FILE_EXPORTER_PATH=\"$HOME/.copilot/otel/copilot.jsonl\"\n\ + \n\ + \x20 Add that line to your shell rc file (~/.zshrc, ~/.bashrc, …) so\n\ + \x20 new shells pick it up. Any path works; under ~/.copilot/otel/ it is\n\ + \x20 picked up even on shells where the export isn't set.\n\ + \n\ + 2. Restart your shell (or `source` the rc file), then run Copilot CLI.\n\ + \n\ + 3. Ingest as usual — `burn ingest` picks the spans up automatically.\n\ + \n\ + Coverage note: OTEL spans carry per-API-call token usage (input,\n\ + output, cache read/write, reasoning) and the model, but no tool-call\n\ + content, so Copilot turns report as usage-only fidelity.\n" + ); + 0 + } + } +} diff --git a/crates/relayburn-cli/src/commands/mod.rs b/crates/relayburn-cli/src/commands/mod.rs index 0ce2eb3e..d26b3301 100644 --- a/crates/relayburn-cli/src/commands/mod.rs +++ b/crates/relayburn-cli/src/commands/mod.rs @@ -23,6 +23,7 @@ pub mod flow; mod freshness; pub mod hotspots; pub mod ingest; +pub mod init; pub mod mcp_server; pub mod measure; pub mod overhead; diff --git a/crates/relayburn-cli/src/main.rs b/crates/relayburn-cli/src/main.rs index c078e042..4889f0bd 100644 --- a/crates/relayburn-cli/src/main.rs +++ b/crates/relayburn-cli/src/main.rs @@ -60,6 +60,7 @@ fn dispatch(args: Args) -> i32 { Command::Ingest(args) => commands::ingest::run(&globals, args), Command::McpServer(args) => commands::mcp_server::run(&globals, args), Command::Update(args) => commands::update::run(&globals, args), + Command::Init(args) => commands::init::run(args), } } @@ -87,5 +88,6 @@ fn command_name(command: &Command) -> &'static str { Command::Ingest(_) => "ingest", Command::McpServer(_) => "mcp-server", Command::Update(_) => "update", + Command::Init(_) => "init", } } diff --git a/crates/relayburn-cli/tests/smoke.rs b/crates/relayburn-cli/tests/smoke.rs index 0b1dba5a..ef5c330c 100644 --- a/crates/relayburn-cli/tests/smoke.rs +++ b/crates/relayburn-cli/tests/smoke.rs @@ -36,6 +36,7 @@ const SUBCOMMANDS: &[&str] = &[ "ingest", "mcp-server", "update", + "init", ]; #[test] @@ -437,6 +438,21 @@ fn json_mode_emits_error_envelope_on_argument_failure() { ); } +#[test] +fn init_copilot_prints_otel_exporter_setup() { + let output = burn() + .args(["init", "copilot"]) + .assert() + .success() + .get_output() + .clone(); + let stdout = String::from_utf8(output.stdout).expect("stdout should be valid UTF-8"); + assert!( + stdout.contains("COPILOT_OTEL_FILE_EXPORTER_PATH"), + "expected `init copilot` to print the exporter env var; got:\n{stdout}", + ); +} + #[test] fn version_flag_exits_zero() { burn() diff --git a/crates/relayburn-sdk/src/analyze/provider.rs b/crates/relayburn-sdk/src/analyze/provider.rs index ac8a8488..d648b2ae 100644 --- a/crates/relayburn-sdk/src/analyze/provider.rs +++ b/crates/relayburn-sdk/src/analyze/provider.rs @@ -339,6 +339,7 @@ fn provider_from_source(source: SourceKind) -> String { SourceKind::Codex | SourceKind::OpenaiApi => "openai".into(), SourceKind::GeminiApi => "google".into(), SourceKind::Opencode => "opencode".into(), + SourceKind::CopilotCli => "github-copilot".into(), } } diff --git a/crates/relayburn-sdk/src/ingest.rs b/crates/relayburn-sdk/src/ingest.rs index 5bd72f08..10389683 100644 --- a/crates/relayburn-sdk/src/ingest.rs +++ b/crates/relayburn-sdk/src/ingest.rs @@ -64,7 +64,8 @@ pub(crate) static TEST_GAP_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(() pub use gap::{restore_ingest_gap_writer, set_ingest_gap_writer}; pub use ingest::{ default_session_roots, ingest_all, ingest_claude_session, ingest_claude_transcript_path, - ingest_codex_sessions, ingest_opencode_sessions, IngestOptions, IngestReport, IngestRoots, + ingest_codex_sessions, ingest_copilot_sessions, ingest_opencode_sessions, IngestOptions, + IngestReport, IngestRoots, }; pub use pending_stamps::{ cleanup_stale_pending_stamps, write_pending_stamp, PendingStamp, PendingStampHarness, diff --git a/crates/relayburn-sdk/src/ingest/cursors.rs b/crates/relayburn-sdk/src/ingest/cursors.rs index 3711d167..bf414524 100644 --- a/crates/relayburn-sdk/src/ingest/cursors.rs +++ b/crates/relayburn-sdk/src/ingest/cursors.rs @@ -7,7 +7,8 @@ //! ## Wire layout //! //! `{"files": {"": }}`. The cursor variant is tagged -//! by `kind`: `"claude" | "codex" | "opencode" | "opencode-stream"`. Field +//! by `kind`: `"claude" | "codex" | "opencode" | "copilot" | +//! "opencode-stream"`. Field //! names are camelCase to match the TS schema so a Rust ingest can pick up //! cursors a TS ingest wrote, and vice versa, during the migration. @@ -68,6 +69,23 @@ pub struct OpencodeCursor { pub seen_message_ids: Vec, } +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CopilotCursor { + pub inode: u64, + pub offset_bytes: u64, + pub mtime_ms: i64, + /// Next `turn_index` per session id, so a continued session keeps + /// numbering across incremental passes and exporter rotations. + #[serde(default)] + pub session_turn_counts: BTreeMap, + /// Trace ids known to contain a `chat` span; suppresses the + /// double-counting `invoke_agent` aggregate for those traces. Bounded + /// by the parser at a few thousand entries. + #[serde(default)] + pub chat_trace_ids: Vec, +} + /// Tagged-union cursor variants. The Codex variant is heap-boxed because it /// carries a per-turn map and dwarfs the others — keeping the enum payload /// size sane keeps `Cursors` cheap to clone in the orchestration hot path. @@ -77,6 +95,8 @@ pub enum FileCursor { Claude(ClaudeCursor), Codex(Box), Opencode(OpencodeCursor), + #[serde(rename = "copilot")] + Copilot(CopilotCursor), #[serde(rename = "opencode-stream")] OpencodeStream(Value), } diff --git a/crates/relayburn-sdk/src/ingest/gap.rs b/crates/relayburn-sdk/src/ingest/gap.rs index 90799ce5..ceb2ab77 100644 --- a/crates/relayburn-sdk/src/ingest/gap.rs +++ b/crates/relayburn-sdk/src/ingest/gap.rs @@ -48,6 +48,7 @@ pub enum AdapterName { Claude, Codex, Opencode, + Copilot, } impl AdapterName { @@ -56,6 +57,7 @@ impl AdapterName { AdapterName::Claude => "claude", AdapterName::Codex => "codex", AdapterName::Opencode => "opencode", + AdapterName::Copilot => "copilot", } } } diff --git a/crates/relayburn-sdk/src/ingest/gap_warning_tests.rs b/crates/relayburn-sdk/src/ingest/gap_warning_tests.rs index 664ad5d3..e16be21e 100644 --- a/crates/relayburn-sdk/src/ingest/gap_warning_tests.rs +++ b/crates/relayburn-sdk/src/ingest/gap_warning_tests.rs @@ -59,6 +59,7 @@ fn pinned_roots(tmp: &TempDir) -> IngestRoots { claude_projects_dir: Some(tmp.path().join("claude").join("projects")), codex_sessions_dir: Some(tmp.path().join("codex").join("sessions")), opencode_storage_dir: Some(tmp.path().join("opencode").join("storage")), + copilot_otel_files: Some(vec![]), } } diff --git a/crates/relayburn-sdk/src/ingest/ingest.rs b/crates/relayburn-sdk/src/ingest/ingest.rs index 60642593..f5548561 100644 --- a/crates/relayburn-sdk/src/ingest/ingest.rs +++ b/crates/relayburn-sdk/src/ingest/ingest.rs @@ -27,18 +27,20 @@ use serde_json::Value; use crate::ledger::{load_config, Ledger}; use crate::reader::{ parse_claude_session, parse_claude_session_incremental, parse_codex_session_incremental, - parse_opencode_session_incremental, reconcile_claude_session_relationships, - ClaudeParseIncrementalOptions, ClaudeParseIncrementalResult, ClaudeParseOptions, - ClaudeParseResult, CodexLastCompletedTurn, CodexResumeState, CodexTurnContext, CompactionEvent, - ContentRecord, ContentStoreMode, CumulativeUsage as ReaderCumulativeUsage, - ParseCodexIncrementalOptions, ParseCodexIncrementalResult, ParseOpencodeIncrementalOptions, - ParseOpencodeIncrementalResult, PersistedUserTurnSlot, ReconcileClaudeRelationshipsInput, - SessionRelationshipRecord, ToolResultEventRecord, TurnRecord, UserTurnRecord, + parse_copilot_otel_incremental, parse_opencode_session_incremental, + reconcile_claude_session_relationships, ClaudeParseIncrementalOptions, + ClaudeParseIncrementalResult, ClaudeParseOptions, ClaudeParseResult, CodexLastCompletedTurn, + CodexResumeState, CodexTurnContext, CompactionEvent, ContentRecord, ContentStoreMode, + CopilotResumeState, CumulativeUsage as ReaderCumulativeUsage, ParseCodexIncrementalOptions, + ParseCodexIncrementalResult, ParseCopilotIncrementalOptions, ParseCopilotIncrementalResult, + ParseOpencodeIncrementalOptions, ParseOpencodeIncrementalResult, PersistedUserTurnSlot, + ReconcileClaudeRelationshipsInput, SessionRelationshipRecord, ToolResultEventRecord, + TurnRecord, UserTurnRecord, }; use crate::ingest::cursors::{ - load_cursors, save_cursors_if_changed, ClaudeCursor, CodexCumulative, CodexCursor, Cursors, - FileCursor, OpencodeCursor, + load_cursors, save_cursors_if_changed, ClaudeCursor, CodexCumulative, CodexCursor, + CopilotCursor, Cursors, FileCursor, OpencodeCursor, }; use crate::ingest::gap::{ count_new_tool_calls, count_new_tool_results, emit_gap_warning, record_session_gap, AdapterName, @@ -116,6 +118,11 @@ pub struct IngestRoots { pub claude_projects_dir: Option, pub codex_sessions_dir: Option, pub opencode_storage_dir: Option, + /// Copilot CLI OTEL export files. `None` resolves the defaults: the + /// `COPILOT_OTEL_FILE_EXPORTER_PATH` env var (when set to an existing + /// file) plus `$COPILOT_HOME/otel/*.jsonl` (default `~/.copilot/otel`). + /// Tests inject explicit paths so they don't depend on process env. + pub copilot_otel_files: Option>, } pub(crate) fn claude_projects_dir(roots: &IngestRoots) -> PathBuf { @@ -146,6 +153,42 @@ pub(crate) fn opencode_session_root(roots: &IngestRoots) -> PathBuf { opencode_storage_dir(roots).join("session") } +/// Default OTEL export directory Copilot CLI writes when the exporter is +/// pointed at a directory: `$COPILOT_HOME/otel`, falling back to +/// `~/.copilot/otel`. +pub(crate) fn copilot_otel_dir() -> PathBuf { + std::env::var("COPILOT_HOME") + .map(PathBuf::from) + .unwrap_or_else(|_| home_dir().join(".copilot")) + .join("otel") +} + +/// Resolve the Copilot CLI OTEL JSONL files ingest scans. Unlike the other +/// harnesses this source is env-gated: nothing exists on disk until the +/// user sets `COPILOT_OTEL_FILE_EXPORTER_PATH` (see #14), so an empty list +/// is the normal steady state and never an error. +pub(crate) fn copilot_otel_files(roots: &IngestRoots) -> Vec { + if let Some(files) = &roots.copilot_otel_files { + return files.clone(); + } + let mut files: Vec = Vec::new(); + if let Ok(var) = std::env::var("COPILOT_OTEL_FILE_EXPORTER_PATH") { + let trimmed = var.trim(); + if !trimmed.is_empty() { + let p = PathBuf::from(trimmed); + if p.is_file() { + files.push(p); + } + } + } + for file in list_jsonl_files(&copilot_otel_dir()) { + if !files.contains(&file) { + files.push(file); + } + } + files +} + /// Resolve the default session-store roots ingest scans, in the same /// order `ingest_all` walks them. Used by the watch loop to drive its /// `notify`-backed FS-event driver against the harness home dirs the @@ -153,15 +196,30 @@ pub(crate) fn opencode_session_root(roots: &IngestRoots) -> PathBuf { /// [`IngestRoots`] to override individual paths; defaults still come /// from `$HOME` for fields left `None`. /// -/// Returns the Claude / Codex / OpenCode roots in that order — the -/// caller doesn't have to filter for existence; the FS-event driver -/// silently skips any path that doesn't yet exist. +/// Returns the Claude / Codex / OpenCode roots, then the Copilot OTEL +/// export directory, in that order — the caller doesn't have to filter +/// for existence; the FS-event driver silently skips any path that +/// doesn't yet exist. pub fn default_session_roots(roots: &IngestRoots) -> Vec { - vec![ + let mut dirs = vec![ claude_projects_dir(roots), codex_sessions_dir(roots), opencode_storage_dir(roots), - ] + ]; + // Copilot: watch the injected files' parents in tests / overrides, the + // default otel dir otherwise (the env-var file's appends are also caught + // by the polling fingerprint even without an FS event). + match &roots.copilot_otel_files { + Some(files) => { + for parent in files.iter().filter_map(|f| f.parent()) { + if !dirs.iter().any(|d| d == parent) { + dirs.push(parent.to_path_buf()); + } + } + } + None => dirs.push(copilot_otel_dir()), + } + dirs } pub(crate) fn opencode_message_root(roots: &IngestRoots) -> PathBuf { @@ -264,6 +322,17 @@ fn source_fingerprint(roots: &IngestRoots) -> String { } } + // Copilot CLI: the OTEL export files (env-var path plus otel dir glob). + // Env-gated, so normally empty — but when set, appends must move the + // fingerprint like any other source. + for file in copilot_otel_files(roots) { + if let Ok(meta) = fs::metadata(&file) { + count = count.wrapping_add(1); + total_bytes = total_bytes.wrapping_add(meta.len()); + hash_sum = hash_sum.wrapping_add(fingerprint_entry_hash("copilot", &file, &meta)); + } + } + format!("{count}:{total_bytes}:{hash_sum:016x}") } @@ -381,6 +450,19 @@ pub fn ingest_all(ledger: &mut Ledger, opts: &IngestOptions) -> anyhow::Result anyhow::Result { + run_single_harness(ledger, opts, AdapterName::Copilot, ingest_copilot_into) +} + /// Shared boilerplate for the per-harness verbs: clean stale stamps, snapshot /// cursors, resolve content mode, run the harness body, emit any pending gap /// warning for that adapter, then persist cursor mutations. The per-harness @@ -978,6 +1067,107 @@ fn ingest_opencode_into( // --- Codex cursor <-> reader resume-state conversions ------------------- +/// Iterate the Copilot CLI OTEL export files (`COPILOT_OTEL_FILE_EXPORTER_PATH` +/// plus `$COPILOT_HOME/otel/*.jsonl`), driving +/// [`parse_copilot_otel_incremental`] with the carried per-session turn +/// counters and chat-trace set. Env-gated: when neither source exists the +/// file list is empty and this is a silent no-op (#14). +/// +/// Rotation handling matches the Claude/Codex adapters: an inode change or +/// a shrunken file restarts the byte offset at 0, but the per-session +/// `turn_index` counters and `chat_trace_ids` survive — the exporter's new +/// file continues the same Copilot sessions, and the ledger's +/// `(source, session_id, message_id)` key dedups any span re-read at the +/// rotation boundary. +fn ingest_copilot_into( + ledger: &mut Ledger, + cursors: &mut Cursors, + roots: &IngestRoots, + _content_mode: ContentStoreMode, + _ledger_home: Option<&Path>, + had_skips: &mut bool, +) -> anyhow::Result { + let mut report = IngestReport::empty(); + for file in copilot_otel_files(roots) { + report.scanned_sessions += 1; + let key = file.to_string_lossy().into_owned(); + let meta = match fs::metadata(&file) { + Ok(m) => m, + Err(err) if err.kind() == std::io::ErrorKind::NotFound => { + // The env var can point at a file the exporter has not + // created yet; that's the pre-first-session steady state, + // not a failure. + continue; + } + Err(err) => { + eprintln!("[burn] skipping {}: {}", file.display(), err); + *had_skips = true; + continue; + } + }; + let prior = match cursors.get_typed(&key) { + Some(FileCursor::Copilot(c)) => Some(c), + _ => None, + }; + let inode = file_inode(&meta); + let mtime = mtime_ms(&meta); + let size = meta.len(); + let rotated = match &prior { + None => true, + Some(c) => c.inode != inode || mtime < c.mtime_ms || size < c.offset_bytes, + }; + let start_offset = if rotated { + 0 + } else { + prior.as_ref().map(|c| c.offset_bytes).unwrap_or(0) + }; + + if !rotated && start_offset >= size { + if let Some(mut c) = prior.clone() { + c.mtime_ms = mtime; + cursors.insert(key, FileCursor::Copilot(c)); + } + continue; + } + + let resume = prior.map(|c| CopilotResumeState { + session_turn_counts: c.session_turn_counts, + chat_trace_ids: c.chat_trace_ids, + }); + let parse_opts = ParseCopilotIncrementalOptions { + session_path: Some(file.to_string_lossy().into_owned()), + start_offset: Some(start_offset), + resume, + fallback_ts_ms: Some(mtime), + }; + let parsed: ParseCopilotIncrementalResult = + match parse_copilot_otel_incremental(&file, &parse_opts) { + Ok(r) => r, + Err(err) => { + eprintln!("[burn] skipping {}: {}", file.display(), err); + *had_skips = true; + continue; + } + }; + + if !parsed.turns.is_empty() { + report.appended_turns += parsed.turns.len(); + report.ingested_sessions += 1; + ledger.append_turns(&parsed.turns)?; + } + + let next = CopilotCursor { + inode, + offset_bytes: parsed.end_offset, + mtime_ms: mtime, + session_turn_counts: parsed.resume.session_turn_counts, + chat_trace_ids: parsed.resume.chat_trace_ids, + }; + cursors.insert(key, FileCursor::Copilot(next)); + } + Ok(report) +} + fn codex_cursor_to_resume_state(c: &CodexCursor) -> CodexResumeState { let mut turn_contexts: HashMap = HashMap::new(); for (k, v) in &c.turn_contexts { @@ -1386,6 +1576,7 @@ mod tests { claude_projects_dir: Some(PathBuf::from("/x/claude")), codex_sessions_dir: Some(PathBuf::from("/x/codex")), opencode_storage_dir: Some(PathBuf::from("/x/oc")), + copilot_otel_files: Some(vec![]), }; assert_eq!(claude_projects_dir(&roots), PathBuf::from("/x/claude")); assert_eq!(codex_sessions_dir(&roots), PathBuf::from("/x/codex")); @@ -1399,6 +1590,7 @@ mod tests { claude_projects_dir: Some(tmp.path().join("claude")), codex_sessions_dir: Some(tmp.path().join("codex")), opencode_storage_dir: Some(tmp.path().join("opencode")), + copilot_otel_files: Some(vec![]), }; // Empty roots: well-formed, stable, and identical across calls. let empty = source_fingerprint(&roots); @@ -1442,6 +1634,7 @@ mod tests { claude_projects_dir: Some(tmp.path().join("claude")), codex_sessions_dir: Some(tmp.path().join("codex")), opencode_storage_dir: Some(storage.clone()), + copilot_otel_files: Some(vec![]), }; let session_dir = storage.join("session"); diff --git a/crates/relayburn-sdk/src/ingest/orchestration_tests.rs b/crates/relayburn-sdk/src/ingest/orchestration_tests.rs index 1add6c2d..e8dc7f00 100644 --- a/crates/relayburn-sdk/src/ingest/orchestration_tests.rs +++ b/crates/relayburn-sdk/src/ingest/orchestration_tests.rs @@ -32,7 +32,8 @@ use std::path::{Path, PathBuf}; use crate::ingest::cursors::{load_cursors, ClaudeCursor, FileCursor}; use crate::ingest::ingest::{ ingest_all, ingest_claude_projects, ingest_claude_session, ingest_claude_transcript_path, - ingest_codex_sessions, ingest_opencode_sessions, IngestOptions, IngestRoots, + ingest_codex_sessions, ingest_copilot_sessions, ingest_opencode_sessions, IngestOptions, + IngestRoots, }; use crate::ingest::pending_stamps::{write_pending_stamp, PendingStampHarness, WriteOptions}; use crate::ledger::{Enrichment, Ledger, LedgerLayout, Query}; @@ -123,6 +124,7 @@ fn pinned_roots(tmp: &TempDir) -> IngestRoots { claude_projects_dir: Some(tmp.path().join("claude").join("projects")), codex_sessions_dir: Some(tmp.path().join("codex").join("sessions")), opencode_storage_dir: Some(tmp.path().join("opencode").join("storage")), + copilot_otel_files: Some(vec![]), } } @@ -282,6 +284,96 @@ fn ingest_opencode_sessions_round_trips_a_fixture_session() { } } +#[test] +fn ingest_copilot_sessions_round_trips_otel_spans() { + let tmp = TempDir::new().unwrap(); + let _env = isolated_relayburn_home(&tmp); + + let otel_dir = tmp.path().join("copilot").join("otel"); + fs::create_dir_all(&otel_dir).unwrap(); + let export_file = otel_dir.join("copilot.jsonl"); + let span = |trace: &str, span: &str, input: u64, output: u64| { + format!( + "{{\"type\":\"span\",\"traceId\":\"{trace}\",\"spanId\":\"{span}\",\"name\":\"chat claude-sonnet-4.6\",\"startTime\":[1775934260,0],\"attributes\":{{\"gen_ai.operation.name\":\"chat\",\"gen_ai.response.model\":\"claude-sonnet-4.6\",\"gen_ai.conversation.id\":\"conv-1\",\"gen_ai.usage.input_tokens\":{input},\"gen_ai.usage.output_tokens\":{output}}}}}" + ) + }; + fs::write( + &export_file, + format!( + "{}\n{}\n", + span("t-1", "s-1", 100, 10), + span("t-1", "s-2", 200, 20) + ), + ) + .unwrap(); + + let roots = IngestRoots { + copilot_otel_files: Some(vec![export_file.clone()]), + ..pinned_roots(&tmp) + }; + let mut ledger = open_ledger_in(&tmp); + let opts = IngestOptions { + roots, + ..Default::default() + }; + + let report = ingest_copilot_sessions(&mut ledger, &opts).unwrap(); + assert_eq!(report.appended_turns, 2); + assert_eq!(report.ingested_sessions, 1); + + let turns = ledger.query_turns(&Query::for_session("conv-1")).unwrap(); + assert_eq!(turns.len(), 2); + assert_eq!(turns[0].turn.source, crate::reader::SourceKind::CopilotCli); + assert_eq!(turns[0].turn.turn_index, 0); + assert_eq!(turns[1].turn.turn_index, 1); + assert_eq!(turns[0].turn.usage.input, 100); + assert_eq!(turns[1].turn.usage.output, 20); + + let cursors = load_cursors(&ledger).unwrap(); + let key = export_file.to_string_lossy().into_owned(); + match cursors.get_typed(&key) { + Some(FileCursor::Copilot(_)) => {} + other => panic!("expected CopilotCursor for {key}, got {other:?}"), + } + + // No-op sweep: unchanged file appends nothing. + let report = ingest_copilot_sessions(&mut ledger, &opts).unwrap(); + assert_eq!(report.appended_turns, 0); + + // A new span picked up incrementally keeps the per-session numbering. + let mut f = fs::OpenOptions::new() + .append(true) + .open(&export_file) + .unwrap(); + writeln!(f, "{}", span("t-2", "s-3", 300, 30)).unwrap(); + drop(f); + let report = ingest_copilot_sessions(&mut ledger, &opts).unwrap(); + assert_eq!(report.appended_turns, 1); + let turns = ledger.query_turns(&Query::for_session("conv-1")).unwrap(); + assert_eq!(turns.len(), 3); + assert_eq!(turns[2].turn.turn_index, 2); +} + +#[test] +fn ingest_copilot_sessions_is_a_noop_without_exporter_files() { + let tmp = TempDir::new().unwrap(); + let _env = isolated_relayburn_home(&tmp); + // Explicitly empty: mirrors the env-gated steady state where the user + // has never set COPILOT_OTEL_FILE_EXPORTER_PATH. + let roots = IngestRoots { + copilot_otel_files: Some(vec![]), + ..pinned_roots(&tmp) + }; + let mut ledger = open_ledger_in(&tmp); + let opts = IngestOptions { + roots, + ..Default::default() + }; + let report = ingest_copilot_sessions(&mut ledger, &opts).unwrap(); + assert_eq!(report.scanned_sessions, 0); + assert_eq!(report.appended_turns, 0); +} + #[test] fn ingest_all_walks_each_harness_root_once() { let tmp = TempDir::new().unwrap(); diff --git a/crates/relayburn-sdk/src/ingest_verb.rs b/crates/relayburn-sdk/src/ingest_verb.rs index 68116190..27159a58 100644 --- a/crates/relayburn-sdk/src/ingest_verb.rs +++ b/crates/relayburn-sdk/src/ingest_verb.rs @@ -101,6 +101,7 @@ mod tests { claude_projects_dir: Some(claude.path().to_path_buf()), codex_sessions_dir: Some(codex.path().to_path_buf()), opencode_storage_dir: Some(opencode.path().to_path_buf()), + copilot_otel_files: Some(vec![]), }, on_progress: None, on_warn: None, diff --git a/crates/relayburn-sdk/src/lib.rs b/crates/relayburn-sdk/src/lib.rs index 9ffe0a9a..4a1b3040 100644 --- a/crates/relayburn-sdk/src/lib.rs +++ b/crates/relayburn-sdk/src/lib.rs @@ -133,11 +133,12 @@ pub use crate::analyze::{ pub use crate::ingest::{ cleanup_stale_pending_stamps, default_session_roots, ingest_all, ingest_claude_session, - ingest_claude_transcript_path, ingest_codex_sessions, ingest_opencode_sessions, - start_watch_loop, write_pending_stamp, ErrorSink, IngestFn, IngestOptions as RawIngestOptions, - IngestReport, IngestRoots, PendingStamp, PendingStampHarness, PendingStampWriteResult, - ReportSink, StartWatchLoopOptions, WatchController, WriteOptions as PendingStampWriteOptions, - DEFAULT_FS_DEBOUNCE, DEFAULT_SLOW_FALLBACK, + ingest_claude_transcript_path, ingest_codex_sessions, ingest_copilot_sessions, + ingest_opencode_sessions, start_watch_loop, write_pending_stamp, ErrorSink, IngestFn, + IngestOptions as RawIngestOptions, IngestReport, IngestRoots, PendingStamp, + PendingStampHarness, PendingStampWriteResult, ReportSink, StartWatchLoopOptions, + WatchController, WriteOptions as PendingStampWriteOptions, DEFAULT_FS_DEBOUNCE, + DEFAULT_SLOW_FALLBACK, }; // --- LedgerOpenOptions ----------------------------------------------------- diff --git a/crates/relayburn-sdk/src/reader.rs b/crates/relayburn-sdk/src/reader.rs index 8f4af424..8301bb34 100644 --- a/crates/relayburn-sdk/src/reader.rs +++ b/crates/relayburn-sdk/src/reader.rs @@ -16,6 +16,7 @@ pub mod user_turn; pub mod claude; pub mod codex; +pub mod copilot; pub mod opencode; pub use codex::span_tree::{build_codex_span_tree, CodexSpanTreeInputs}; @@ -24,6 +25,10 @@ pub use codex::{ CodexResumeState, CodexTurnContext, CumulativeUsage, ParseCodexIncrementalOptions, ParseCodexIncrementalResult, PersistedUserTurnSlot, }; +pub use copilot::{ + parse_copilot_otel_incremental, CopilotResumeState, ParseCopilotIncrementalOptions, + ParseCopilotIncrementalResult, +}; pub use opencode::{ parse_opencode_session_incremental, ParseOpencodeIncrementalOptions, ParseOpencodeIncrementalResult, diff --git a/crates/relayburn-sdk/src/reader/copilot.rs b/crates/relayburn-sdk/src/reader/copilot.rs new file mode 100644 index 00000000..c62e87ad --- /dev/null +++ b/crates/relayburn-sdk/src/reader/copilot.rs @@ -0,0 +1,562 @@ +//! GitHub Copilot CLI OTEL span parser (AgentWorkforce/burn#14). +//! +//! Copilot CLI does not write a session log; it emits usage through its +//! OpenTelemetry file exporter when the user sets +//! `COPILOT_OTEL_FILE_EXPORTER_PATH` (auto-enables +//! `COPILOT_OTEL_EXPORTER_TYPE=file`). The exporter appends one JSON record +//! per line — a `chat` span per API call, an `invoke_agent` summary span per +//! turn, plus metrics and log records we ignore. Burn additionally globs +//! `$COPILOT_HOME/otel/*.jsonl` (default `~/.copilot/otel`) so a directory +//! pointed at by the exporter is covered too. +//! +//! Format reference: tokscale's `sessions/copilot.rs` and the GitHub Docs +//! Copilot CLI command reference. Notable shapes handled here: +//! +//! - `{"type":"span","traceId","spanId","name":"chat ", +//! "startTime":[secs,nanos],"endTime":[secs,nanos],"attributes":{…}}` +//! - usage attrs `gen_ai.usage.{input,output}_tokens` (input is *inclusive* +//! of cache reads), cache buckets under both the dotted semconv spelling +//! (`gen_ai.usage.cache_read.input_tokens`) and the underscored variant +//! the CLI actually emits (`gen_ai.usage.cache_read_input_tokens`), and +//! reasoning under `gen_ai.usage.reasoning{,.output}_tokens`. +//! - session identity by priority: `gen_ai.conversation.id`, +//! `copilot_chat.session_id`, `copilot_chat.chat_session_id`, +//! `session.id`, `github.copilot.interaction_id`, `gen_ai.response.id`, +//! then the OTEL `traceId`, then `"unknown"`. +//! +//! An `invoke_agent` span reports *totals across all turns* of its trace, so +//! it is only used as a fallback when no `chat` span was seen for the same +//! `traceId` (tracked across incremental passes in +//! [`CopilotResumeState::chat_trace_ids`]). Spans with zero total tokens are +//! dropped. +//! +//! Tool calls are not available: OTEL spans carry structured metrics, not +//! content, so the emitted [`TurnRecord`]s are usage-only +//! ([`UsageGranularity::PerMessage`] — one record per API call). + +use std::collections::BTreeMap; +use std::fs::File; +use std::io::{BufReader, Read, Seek, SeekFrom}; +use std::path::Path; + +use serde_json::{Map, Value}; + +use crate::reader::types::{ + Coverage, Fidelity, SourceKind, StopReason, TurnRecord, Usage, UsageGranularity, +}; +use crate::util::time::format_iso_ms; + +/// Cap on `chat_trace_ids` carried in the cursor so a long-lived export file +/// can't grow ingest state without bound. Fallback double-count protection +/// only needs traces whose `invoke_agent` summary hasn't flushed yet, which +/// is a working set of in-flight turns — far below this cap. +const CHAT_TRACE_ID_CAP: usize = 4096; + +#[derive(Debug, Clone, Default)] +pub struct ParseCopilotIncrementalOptions { + pub session_path: Option, + pub start_offset: Option, + pub resume: Option, + /// Timestamp fallback (file mtime, ms) for spans that carry no usable + /// `startTime`/`endTime`. tokscale uses the same fallback. + pub fallback_ts_ms: Option, +} + +/// State carried across incremental passes over the same export file. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct CopilotResumeState { + /// Next `turn_index` per session id. + pub session_turn_counts: BTreeMap, + /// Trace ids known to contain at least one `chat` span; `invoke_agent` + /// summaries for these traces are suppressed to avoid double counting. + pub chat_trace_ids: Vec, +} + +#[derive(Debug, Clone, Default)] +pub struct ParseCopilotIncrementalResult { + pub turns: Vec, + pub end_offset: u64, + pub resume: CopilotResumeState, +} + +#[cfg(test)] +#[derive(Debug, Clone, Default)] +pub struct ParseCopilotResult { + pub turns: Vec, +} + +#[cfg(test)] +pub fn parse_copilot_file(path: &Path) -> std::io::Result { + let parsed = parse_copilot_otel_incremental( + path, + &ParseCopilotIncrementalOptions { + session_path: Some(path.to_string_lossy().into_owned()), + ..Default::default() + }, + )?; + Ok(ParseCopilotResult { + turns: parsed.turns, + }) +} + +/// Incrementally parse a Copilot OTEL JSONL export starting at +/// `start_offset`, emitting one [`TurnRecord`] per usage-bearing span. +/// +/// Reading stops at the last complete line: a trailing partial line (a +/// span still being flushed by the exporter) is left for the next pass and +/// `end_offset` points at its first byte. +pub fn parse_copilot_otel_incremental( + path: &Path, + opts: &ParseCopilotIncrementalOptions, +) -> std::io::Result { + let start_offset = opts.start_offset.unwrap_or(0); + let mut resume = opts.resume.clone().unwrap_or_default(); + let session_path = opts.session_path.clone(); + + let mut reader = BufReader::new(File::open(path)?); + reader.seek(SeekFrom::Start(start_offset))?; + let mut buf = Vec::new(); + reader.read_to_end(&mut buf)?; + + // Only consume up to the final newline; bytes after it are a partial + // line still being written. + let consumed = match buf.iter().rposition(|b| *b == b'\n') { + Some(pos) => pos + 1, + None => 0, + }; + let end_offset = start_offset + consumed as u64; + + // A usage span does not always carry its own model/session — those can + // arrive on a sibling span sharing the trace id — so collect per-trace + // context for this chunk first, then resolve candidates against it. + // (tokscale reads the whole file for this; our incremental cursor makes + // the chunk boundary the practical horizon, which matches how spans for + // one trace land within milliseconds of each other.) + let mut trace_contexts: BTreeMap = BTreeMap::new(); + let mut candidates: Vec = Vec::new(); + + for (index, line) in buf[..consumed].split(|b| *b == b'\n').enumerate() { + let trimmed = trim_ascii(line); + if trimmed.is_empty() { + continue; + } + let record: Value = match serde_json::from_slice(trimmed) { + Ok(v) => v, + Err(_) => continue, // lossy per line: one bad record never truncates the file + }; + accumulate_trace_context(&mut trace_contexts, &record); + if let Some(candidate) = candidate_from_record(&record, index, opts.fallback_ts_ms) { + candidates.push(candidate); + } + } + + let mut turns = Vec::new(); + for candidate in candidates { + let Some(turn) = candidate.resolve(&trace_contexts, &mut resume, session_path.as_deref()) + else { + continue; + }; + turns.push(turn); + } + + Ok(ParseCopilotIncrementalResult { + turns, + end_offset, + resume, + }) +} + +// --------------------------------------------------------------------------- +// Record classification +// --------------------------------------------------------------------------- + +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +enum SpanKind { + /// `gen_ai.operation.name == "chat"` — one span per API call. + Chat, + /// `gen_ai.operation.name == "invoke_agent"` — per-turn aggregate; + /// fallback only, suppressed when the trace has chat spans. + AgentSummary, +} + +fn is_span_record(value: &Value) -> bool { + match value.get("type").and_then(Value::as_str) { + Some(t) => t == "span", + // VS Code Copilot Chat exports omit `type`; infer span-ness from a + // top-level `name` plus span identity. Harmless for the CLI lane. + None => value.get("name").and_then(Value::as_str).is_some() && span_id(value).is_some(), + } +} + +fn classify_span(value: &Value, attributes: &Map) -> Option { + if !is_span_record(value) { + return None; + } + let op = attr_str(attributes, "gen_ai.operation.name"); + let name = value.get("name").and_then(Value::as_str).unwrap_or(""); + if op == Some("chat") || name.starts_with("chat ") { + return Some(SpanKind::Chat); + } + if op == Some("invoke_agent") || name.starts_with("invoke_agent ") || name == "invoke_agent" { + return Some(SpanKind::AgentSummary); + } + None +} + +// --------------------------------------------------------------------------- +// Attribute extraction +// --------------------------------------------------------------------------- + +fn attr_str<'a>(attributes: &'a Map, key: &str) -> Option<&'a str> { + attributes.get(key).and_then(Value::as_str) +} + +fn first_non_empty_attr<'a>(attributes: &'a Map, keys: &[&str]) -> Option<&'a str> { + keys.iter() + .filter_map(|key| attributes.get(*key).and_then(Value::as_str)) + .map(str::trim) + .find(|value| !value.is_empty()) +} + +fn value_as_i64(value: &Value) -> Option { + value + .as_i64() + .or_else(|| value.as_u64().and_then(|v| i64::try_from(v).ok())) + .or_else(|| value.as_f64().map(|v| v as i64)) + .or_else(|| value.as_str().and_then(|v| v.parse::().ok())) +} + +fn attr_i64(attributes: &Map, key: &str) -> i64 { + attributes + .get(key) + .and_then(value_as_i64) + .unwrap_or(0) + .max(0) +} + +fn attr_i64_first(attributes: &Map, keys: &[&str]) -> i64 { + keys.iter() + .map(|key| attr_i64(attributes, key)) + .find(|value| *value > 0) + .unwrap_or(0) +} + +const MODEL_ATTRS: &[&str] = &["gen_ai.response.model", "gen_ai.request.model"]; + +/// Session id attribute priority, mirroring tokscale's `SESSION_ATTRS`. +const SESSION_ATTRS: &[&str] = &[ + "gen_ai.conversation.id", + "copilot_chat.session_id", + "copilot_chat.chat_session_id", + "session.id", + "github.copilot.interaction_id", + "gen_ai.response.id", +]; + +fn best_session_attr(attributes: &Map) -> Option<&str> { + SESSION_ATTRS + .iter() + .find_map(|key| attr_str(attributes, key)) + .map(str::trim) + .filter(|value| !value.is_empty()) +} + +fn trace_id(value: &Value) -> Option<&str> { + non_empty_id(value.get("traceId").and_then(Value::as_str)).or_else(|| { + non_empty_id( + value + .get("spanContext") + .and_then(|sc| sc.get("traceId")) + .and_then(Value::as_str), + ) + }) +} + +fn span_id(value: &Value) -> Option<&str> { + non_empty_id(value.get("spanId").and_then(Value::as_str)).or_else(|| { + non_empty_id( + value + .get("spanContext") + .and_then(|sc| sc.get("spanId")) + .and_then(Value::as_str), + ) + }) +} + +/// Treat empty and all-zero W3C sentinel ids as absent. +fn non_empty_id(id: Option<&str>) -> Option<&str> { + id.map(str::trim) + .filter(|s| !s.is_empty() && s.chars().any(|c| c != '0' && c != '-')) +} + +/// `[seconds, nanos]` OTEL time pair, or a scalar whose unit is inferred +/// from magnitude (ns / µs / ms / s), or an ISO string. +fn timestamp_ms_from_value(value: &Value) -> Option { + match value { + Value::Array(parts) if !parts.is_empty() => { + let secs = value_as_i64(&parts[0])?; + let nanos = parts.get(1).and_then(value_as_i64).unwrap_or(0); + Some(secs.saturating_mul(1_000).saturating_add(nanos / 1_000_000)) + } + Value::String(s) => crate::util::time::parse_iso_ms(s), + _ => { + let raw = value_as_i64(value)?; + Some(match raw { + // ~1e18 → nanoseconds, ~1e15 → microseconds, ~1e12 → ms. + n if n >= 1_000_000_000_000_000_000 => n / 1_000_000, + n if n >= 1_000_000_000_000_000 => n / 1_000, + n if n >= 1_000_000_000_000 => n, + n => n.saturating_mul(1_000), + }) + } + } +} + +fn timestamp_ms_from_record(value: &Value) -> Option { + value + .get("startTime") + .and_then(timestamp_ms_from_value) + .or_else(|| value.get("hrTime").and_then(timestamp_ms_from_value)) + .or_else(|| value.get("_hrTime").and_then(timestamp_ms_from_value)) + .or_else(|| value.get("timeUnixNano").and_then(timestamp_ms_from_value)) + .or_else(|| { + // End-only span: back-calculate the start from the duration. + let end = value.get("endTime").and_then(timestamp_ms_from_value)?; + let start = value.get("startTime").and_then(timestamp_ms_from_value); + Some(start.unwrap_or(end)) + }) +} + +fn trim_ascii(bytes: &[u8]) -> &[u8] { + let start = bytes + .iter() + .position(|b| !b.is_ascii_whitespace()) + .unwrap_or(bytes.len()); + let end = bytes + .iter() + .rposition(|b| !b.is_ascii_whitespace()) + .map(|p| p + 1) + .unwrap_or(start); + &bytes[start..end] +} + +// --------------------------------------------------------------------------- +// Trace context + candidate resolution +// --------------------------------------------------------------------------- + +#[derive(Default)] +struct TraceContext { + model: Option, + session_id: Option, +} + +fn accumulate_trace_context(contexts: &mut BTreeMap, record: &Value) { + let Some(trace) = trace_id(record).map(str::to_string) else { + return; + }; + let attributes = record + .get("attributes") + .and_then(Value::as_object) + .cloned() + .unwrap_or_default(); + let ctx = contexts.entry(trace).or_default(); + if ctx.model.is_none() { + ctx.model = first_non_empty_attr(&attributes, MODEL_ATTRS).map(str::to_string); + } + if ctx.session_id.is_none() { + ctx.session_id = best_session_attr(&attributes).map(str::to_string); + } +} + +struct PendingCandidate { + kind: SpanKind, + trace_id: Option, + span_id: Option, + response_id: Option, + model: Option, + session_id: Option, + stop_reason: Option, + ts_ms: i64, + usage: Usage, + /// Position among successfully parsed records in this chunk; fallback + /// identity for spans with no `spanId`. + index: usize, +} + +fn candidate_from_record( + record: &Value, + index: usize, + fallback_ts_ms: Option, +) -> Option { + let attributes = record.get("attributes").and_then(Value::as_object)?; + let kind = classify_span(record, attributes)?; + + let input = attr_i64(attributes, "gen_ai.usage.input_tokens"); + let output = attr_i64(attributes, "gen_ai.usage.output_tokens"); + let cache_read = attr_i64_first( + attributes, + &[ + "gen_ai.usage.cache_read.input_tokens", + "gen_ai.usage.cache_read_input_tokens", + ], + ); + let cache_write = attr_i64_first( + attributes, + &[ + "gen_ai.usage.cache_write.input_tokens", + "gen_ai.usage.cache_creation.input_tokens", + "gen_ai.usage.cache_write_input_tokens", + "gen_ai.usage.cache_creation_input_tokens", + ], + ); + let reasoning = attr_i64_first( + attributes, + &[ + "gen_ai.usage.reasoning.output_tokens", + "gen_ai.usage.reasoning_tokens", + ], + ); + if input + output + cache_read + cache_write + reasoning == 0 { + return None; + } + + // OTEL reports input_tokens inclusive of cache reads; burn's + // `Usage.input` is exclusive (same convention as the Claude/Codex + // readers), so subtract the cached-read portion while keeping the + // cache buckets intact for pricing. + let cache_read_in_input = cache_read.min(input); + let usage = Usage { + input: (input - cache_read_in_input) as u64, + output: output as u64, + reasoning: reasoning as u64, + cache_read: cache_read as u64, + // Copilot doesn't split cache creation by TTL; fold into the 5m bucket. + cache_create_5m: cache_write as u64, + cache_create_1h: 0, + }; + + let ts_ms = timestamp_ms_from_record(record) + .or(fallback_ts_ms) + .unwrap_or(0); + + let stop_reason = record + .get("attributes") + .and_then(|a| a.get("gen_ai.response.finish_reasons")) + .and_then(|v| match v { + Value::Array(items) => items.first().and_then(Value::as_str), + Value::String(s) => Some(s.as_str()), + _ => None, + }) + .and_then(StopReason::from_wire); + + Some(PendingCandidate { + kind, + trace_id: trace_id(record).map(str::to_string), + span_id: span_id(record).map(str::to_string), + response_id: attr_str(attributes, "gen_ai.response.id") + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(str::to_string), + model: first_non_empty_attr(attributes, MODEL_ATTRS).map(str::to_string), + session_id: best_session_attr(attributes).map(str::to_string), + stop_reason, + ts_ms, + usage, + index, + }) +} + +impl PendingCandidate { + fn resolve( + self, + trace_contexts: &BTreeMap, + resume: &mut CopilotResumeState, + session_path: Option<&str>, + ) -> Option { + let ctx = self.trace_id.as_deref().and_then(|t| trace_contexts.get(t)); + + // Aggregate `invoke_agent` spans double count when the same trace + // already produced `chat` spans — whether in this chunk or an + // earlier incremental pass. + match self.kind { + SpanKind::Chat => { + if let Some(trace) = &self.trace_id { + if !resume.chat_trace_ids.contains(trace) { + resume.chat_trace_ids.push(trace.clone()); + if resume.chat_trace_ids.len() > CHAT_TRACE_ID_CAP { + let overflow = resume.chat_trace_ids.len() - CHAT_TRACE_ID_CAP; + resume.chat_trace_ids.drain(..overflow); + } + } + } + } + SpanKind::AgentSummary => { + if self + .trace_id + .as_ref() + .is_some_and(|t| resume.chat_trace_ids.contains(t)) + { + return None; + } + } + } + + let session_id = self + .session_id + .or_else(|| ctx.and_then(|c| c.session_id.clone())) + .or_else(|| self.trace_id.clone()) + .unwrap_or_else(|| "unknown".to_string()); + let model = self + .model + .or_else(|| ctx.and_then(|c| c.model.clone())) + .unwrap_or_else(|| "unknown".to_string()); + // spanId is the stable per-span identity; the ledger's + // (source, session_id, message_id) PRIMARY KEY with INSERT OR + // IGNORE makes re-parse idempotent. + let message_id = self + .span_id + .or(self.response_id) + .unwrap_or_else(|| format!("line-{}", self.index)); + + let turn_index = resume + .session_turn_counts + .entry(session_id.clone()) + .or_insert(0); + let this_turn = *turn_index; + *turn_index += 1; + + let coverage = Coverage { + has_input_tokens: true, + has_output_tokens: true, + has_reasoning_tokens: true, + has_cache_read_tokens: true, + has_cache_create_tokens: true, + ..Coverage::EMPTY + }; + + Some(TurnRecord { + v: 1, + source: SourceKind::CopilotCli, + session_id, + session_path: session_path.map(str::to_string), + message_id, + turn_index: this_turn, + ts: format_iso_ms(self.ts_ms), + model, + project: None, + project_key: None, + usage: self.usage, + tool_calls: Vec::new(), + files_touched: None, + subagent: None, + stop_reason: self.stop_reason, + activity: None, + retries: None, + has_edits: None, + fidelity: Some(Fidelity::new(UsageGranularity::PerMessage, coverage)), + }) + } +} + +#[cfg(test)] +mod tests; diff --git a/crates/relayburn-sdk/src/reader/copilot/tests.rs b/crates/relayburn-sdk/src/reader/copilot/tests.rs new file mode 100644 index 00000000..81ef7e2e --- /dev/null +++ b/crates/relayburn-sdk/src/reader/copilot/tests.rs @@ -0,0 +1,195 @@ +//! Copilot CLI OTEL parser tests. Fixtures live at the repo root +//! (`tests/fixtures/copilot-cli/*.jsonl`) and model the span shapes Copilot +//! CLI's OTEL file exporter emits (see tokscale's `sessions/copilot.rs` test +//! module and the GitHub Docs Copilot CLI command reference). + +use std::io::Write; +use std::path::PathBuf; + +use tempfile::tempdir; + +use super::*; +use crate::reader::types::{FidelityClass, UsageGranularity}; +use crate::util::time::format_iso_ms; + +fn fixture(name: &str) -> PathBuf { + let mut p = PathBuf::from(env!("CARGO_MANIFEST_DIR")); + // crates/relayburn-sdk/ -> repo root + p.pop(); + p.pop(); + p.push("tests/fixtures/copilot-cli"); + p.push(name); + p +} + +#[test] +fn chat_spans_fixture() { + let result = parse_copilot_file(&fixture("chat-spans.jsonl")).unwrap(); + let turns = &result.turns; + + // 9 usage-relevant lines in, 5 turns out: the invoke_agent summary for + // trace-1 is suppressed (chat spans present), the zero-token chat span + // and the execute_tool span carry no usage, the metric line and the + // malformed line are skipped. + assert_eq!(turns.len(), 5); + assert!(turns + .iter() + .all(|t| matches!(t.source, SourceKind::CopilotCli))); + + let first = &turns[0]; + assert_eq!(first.session_id, "conv-1"); + assert_eq!(first.message_id, "span-1"); + assert_eq!(first.turn_index, 0); + assert_eq!(first.model, "claude-sonnet-4.6"); + assert_eq!(first.ts, format_iso_ms(1_775_934_260_133)); + // input is exclusive of cache reads: 19452 - 123. + assert_eq!(first.usage.input, 19_329); + assert_eq!(first.usage.output, 281); + assert_eq!(first.usage.cache_read, 123); + assert_eq!(first.usage.cache_create_5m, 21_881); + assert_eq!(first.usage.reasoning, 128); + assert_eq!(first.stop_reason, Some(StopReason::EndTurn)); + + let second = &turns[1]; + assert_eq!(second.session_id, "conv-1"); + assert_eq!(second.turn_index, 1); + assert_eq!(second.usage.input, 300); // 20100 - 19800 + assert_eq!(second.usage.cache_read, 19_800); + assert_eq!(second.stop_reason, None); + + // invoke_agent for trace-2: no chat spans in that trace, so the + // aggregate is the only usage signal and must be kept. Model falls back + // to gen_ai.request.model; cache attrs use the CLI's underscored + // spelling. + let summary = &turns[2]; + assert_eq!(summary.message_id, "invoke-2"); + assert_eq!(summary.session_id, "conv-2"); + assert_eq!(summary.turn_index, 0); + assert_eq!(summary.model, "gpt-5.4-mini"); + assert_eq!(summary.usage.input, 200); // 4200 - 4000 + assert_eq!(summary.usage.output, 310); + assert_eq!(summary.usage.cache_read, 4_000); + assert_eq!(summary.usage.cache_create_5m, 150); + + // String-coerced token counts; no session attr → trace id is the + // session fallback. + let coerced = &turns[3]; + assert_eq!(coerced.session_id, "trace-3"); + assert_eq!(coerced.usage.input, 7); + assert_eq!(coerced.usage.output, 9); + + // VS-Code-style record: no `type`, span identity nested under + // `spanContext`, timestamp under `hrTime`. + let vsc = &turns[4]; + assert_eq!(vsc.message_id, "spanctx-1"); + assert_eq!(vsc.session_id, "trace-6"); + assert_eq!(vsc.model, "gpt-5.4"); + assert_eq!(vsc.ts, format_iso_ms(1_775_934_700_250)); + + // Every turn declares usage-only fidelity at per-message granularity. + for t in turns { + let fidelity = t.fidelity.as_ref().expect("fidelity set"); + assert_eq!(fidelity.granularity, UsageGranularity::PerMessage); + assert_eq!(fidelity.class, FidelityClass::UsageOnly); + assert!(t.tool_calls.is_empty()); + } +} + +#[test] +fn incremental_resume_and_partial_tail() { + let tmp = tempdir().unwrap(); + let path = tmp.path().join("copilot.jsonl"); + + let span_a = br#"{"type":"span","traceId":"t-a","spanId":"s-a","name":"chat m","startTime":[1775934260,0],"attributes":{"gen_ai.operation.name":"chat","gen_ai.conversation.id":"conv-a","gen_ai.usage.input_tokens":10,"gen_ai.usage.output_tokens":2}}"#; + let span_b = br#"{"type":"span","traceId":"t-b","spanId":"s-b","name":"chat m","startTime":[1775934270,0],"attributes":{"gen_ai.operation.name":"chat","gen_ai.conversation.id":"conv-a","gen_ai.usage.input_tokens":20,"gen_ai.usage.output_tokens":4}}"#; + + // First pass: one complete span plus a partial second line still being + // flushed. The partial tail must not be consumed. + let mut f = std::fs::File::create(&path).unwrap(); + f.write_all(span_a).unwrap(); + f.write_all(b"\n").unwrap(); + f.write_all(&span_b[..40]).unwrap(); // partial, no newline + drop(f); + + let first = + parse_copilot_otel_incremental(&path, &ParseCopilotIncrementalOptions::default()).unwrap(); + assert_eq!(first.turns.len(), 1); + assert_eq!(first.turns[0].message_id, "s-a"); + assert_eq!(first.turns[0].turn_index, 0); + assert_eq!(first.end_offset, (span_a.len() + 1) as u64); + + // Complete the second span and re-ingest from the cursor. + let mut f = std::fs::OpenOptions::new() + .append(true) + .open(&path) + .unwrap(); + f.write_all(&span_b[40..]).unwrap(); + f.write_all(b"\n").unwrap(); + drop(f); + + let second = parse_copilot_otel_incremental( + &path, + &ParseCopilotIncrementalOptions { + start_offset: Some(first.end_offset), + resume: Some(first.resume.clone()), + ..Default::default() + }, + ) + .unwrap(); + assert_eq!(second.turns.len(), 1); + assert_eq!(second.turns[0].message_id, "s-b"); + // turn_index continues per session across incremental passes. + assert_eq!(second.turns[0].turn_index, 1); +} + +#[test] +fn invoke_agent_suppressed_across_incremental_passes() { + let tmp = tempdir().unwrap(); + let path = tmp.path().join("copilot.jsonl"); + + let chat = br#"{"type":"span","traceId":"t-x","spanId":"s-x","name":"chat m","startTime":[1775934260,0],"attributes":{"gen_ai.operation.name":"chat","gen_ai.conversation.id":"conv-x","gen_ai.usage.input_tokens":10,"gen_ai.usage.output_tokens":2}}"#; + let summary = br#"{"type":"span","traceId":"t-x","spanId":"i-x","name":"invoke_agent","startTime":[1775934259,0],"attributes":{"gen_ai.operation.name":"invoke_agent","gen_ai.conversation.id":"conv-x","gen_ai.usage.input_tokens":10,"gen_ai.usage.output_tokens":2}}"#; + + std::fs::write(&path, [chat.as_slice(), b"\n"].concat()).unwrap(); + let first = + parse_copilot_otel_incremental(&path, &ParseCopilotIncrementalOptions::default()).unwrap(); + assert_eq!(first.turns.len(), 1); + + // The summary span flushes in a later pass; the carried chat_trace_ids + // must still suppress it. + let mut f = std::fs::OpenOptions::new() + .append(true) + .open(&path) + .unwrap(); + f.write_all(summary).unwrap(); + f.write_all(b"\n").unwrap(); + drop(f); + + let second = parse_copilot_otel_incremental( + &path, + &ParseCopilotIncrementalOptions { + start_offset: Some(first.end_offset), + resume: Some(first.resume), + ..Default::default() + }, + ) + .unwrap(); + assert!(second.turns.is_empty()); +} + +#[test] +fn empty_and_missing_files() { + let tmp = tempdir().unwrap(); + let path = tmp.path().join("empty.jsonl"); + std::fs::write(&path, b"").unwrap(); + let parsed = + parse_copilot_otel_incremental(&path, &ParseCopilotIncrementalOptions::default()).unwrap(); + assert!(parsed.turns.is_empty()); + assert_eq!(parsed.end_offset, 0); + + let missing = tmp.path().join("nope.jsonl"); + assert!( + parse_copilot_otel_incremental(&missing, &ParseCopilotIncrementalOptions::default()) + .is_err() + ); +} diff --git a/crates/relayburn-sdk/src/reader/types.rs b/crates/relayburn-sdk/src/reader/types.rs index b055a518..50340b01 100644 --- a/crates/relayburn-sdk/src/reader/types.rs +++ b/crates/relayburn-sdk/src/reader/types.rs @@ -35,6 +35,7 @@ pub enum SourceKind { ClaudeCode, Codex, Opencode, + CopilotCli, AnthropicApi, OpenaiApi, GeminiApi, @@ -47,6 +48,7 @@ impl SourceKind { Self::ClaudeCode => "claude-code", Self::Codex => "codex", Self::Opencode => "opencode", + Self::CopilotCli => "copilot-cli", Self::AnthropicApi => "anthropic-api", Self::OpenaiApi => "openai-api", Self::GeminiApi => "gemini-api", diff --git a/crates/relayburn-sdk/tests/integration.rs b/crates/relayburn-sdk/tests/integration.rs index 40b1719c..7e494ff2 100644 --- a/crates/relayburn-sdk/tests/integration.rs +++ b/crates/relayburn-sdk/tests/integration.rs @@ -335,6 +335,7 @@ fn ingest_with_empty_roots_returns_zero_report_via_handle_and_free_fn() { claude_projects_dir: Some(claude.path().to_path_buf()), codex_sessions_dir: Some(codex.path().to_path_buf()), opencode_storage_dir: Some(opencode.path().to_path_buf()), + copilot_otel_files: Some(vec![]), }, ..Default::default() }) @@ -348,6 +349,7 @@ fn ingest_with_empty_roots_returns_zero_report_via_handle_and_free_fn() { claude_projects_dir: Some(claude.path().to_path_buf()), codex_sessions_dir: Some(codex.path().to_path_buf()), opencode_storage_dir: Some(opencode.path().to_path_buf()), + copilot_otel_files: Some(vec![]), }, ..Default::default() }) diff --git a/tests/fixtures/copilot-cli/chat-spans.jsonl b/tests/fixtures/copilot-cli/chat-spans.jsonl new file mode 100644 index 00000000..90cb9125 --- /dev/null +++ b/tests/fixtures/copilot-cli/chat-spans.jsonl @@ -0,0 +1,10 @@ +{"type":"metric","name":"gen_ai.client.token.usage","description":"Number of tokens used"} +{"type":"span","traceId":"trace-1","spanId":"span-1","name":"chat claude-sonnet-4.6","startTime":[1775934260,133000000],"endTime":[1775934264,967317833],"resource":{"attributes":{"service.name":"github-copilot","service.version":"1.0.62"}},"attributes":{"gen_ai.operation.name":"chat","gen_ai.provider.name":"github","gen_ai.request.model":"claude-sonnet-4.6","gen_ai.response.model":"claude-sonnet-4.6","gen_ai.conversation.id":"conv-1","gen_ai.response.id":"resp-1","gen_ai.usage.input_tokens":19452,"gen_ai.usage.output_tokens":281,"gen_ai.usage.cache_read.input_tokens":123,"gen_ai.usage.cache_creation.input_tokens":21881,"gen_ai.usage.reasoning.output_tokens":128,"gen_ai.response.finish_reasons":["stop"],"github.copilot.interaction_id":"interaction-1"}} +{"type":"span","traceId":"trace-1","spanId":"span-2","name":"chat claude-sonnet-4.6","startTime":[1775934265,0],"endTime":[1775934268,500000000],"attributes":{"gen_ai.operation.name":"chat","gen_ai.response.model":"claude-sonnet-4.6","gen_ai.conversation.id":"conv-1","gen_ai.response.id":"resp-2","gen_ai.usage.input_tokens":20100,"gen_ai.usage.output_tokens":95,"gen_ai.usage.cache_read.input_tokens":19800}} +{"type":"span","traceId":"trace-1","spanId":"invoke-1","name":"invoke_agent burn-agent","startTime":[1775934258,0],"endTime":[1775934270,0],"attributes":{"gen_ai.operation.name":"invoke_agent","gen_ai.agent.id":"burn-agent","gen_ai.conversation.id":"conv-1","gen_ai.usage.input_tokens":39552,"gen_ai.usage.output_tokens":376}} +{"type":"span","traceId":"trace-2","spanId":"invoke-2","name":"invoke_agent","startTime":[1775934300,0],"endTime":[1775934310,0],"attributes":{"gen_ai.operation.name":"invoke_agent","gen_ai.conversation.id":"conv-2","gen_ai.request.model":"gpt-5.4-mini","gen_ai.usage.input_tokens":4200,"gen_ai.usage.output_tokens":310,"gen_ai.usage.cache_read_input_tokens":4000,"gen_ai.usage.cache_creation_input_tokens":150}} +{"type":"span","traceId":"trace-3","spanId":"span-3","name":"chat gpt-5.4-mini","startTime":[1775934400,0],"attributes":{"gen_ai.operation.name":"chat","gen_ai.request.model":"gpt-5.4-mini","gen_ai.usage.input_tokens":"7","gen_ai.usage.output_tokens":"9"}} +{"type":"span","traceId":"trace-4","spanId":"span-4","name":"chat claude-sonnet-4.6","startTime":[1775934500,0],"attributes":{"gen_ai.operation.name":"chat","gen_ai.conversation.id":"conv-1","gen_ai.usage.input_tokens":0,"gen_ai.usage.output_tokens":0}} +{"type":"span","traceId":"trace-5","spanId":"tool-1","name":"execute_tool rg","startTime":[1775934600,0],"attributes":{"gen_ai.operation.name":"execute_tool","gen_ai.tool.name":"rg"}} +{"this is not json +{"type":"span","traceId":"trace-6","name":"chat gpt-5.4","hrTime":[1775934700,250000000],"spanContext":{"traceId":"trace-6","spanId":"spanctx-1"},"attributes":{"gen_ai.operation.name":"chat","gen_ai.response.model":"gpt-5.4","gen_ai.usage.input_tokens":100,"gen_ai.usage.output_tokens":12}} From 184dd39a5bf56856a550d31625ba6641a5e5ed63 Mon Sep 17 00:00:00 2001 From: Will Washburn Date: Sun, 20 Sep 2026 16:02:04 -0700 Subject: [PATCH 2/2] fix(ingest): address copilot-cli review findings - Gate Copilot OTEL discovery on COPILOT_OTEL_FILE_EXPORTER_PATH; without it ingest is a silent no-op even when ~/.copilot/otel holds files. - Suppress invoke_agent summaries independent of record order by pre-collecting this chunk's chat trace ids. - Fall back to the 'chat ' span-name suffix when a span carries no model attributes, instead of recording 'unknown'. - Derive the id-less span message_id fallback from the absolute file byte offset so incremental passes can't collide in the ledger key. - Materialize per-API-call inferences for Copilot turns via apply_parsed_extras, like every other harness path. - Keep the ingest option table contiguous in README; align init output, README, and CHANGELOG with the gated collector behavior. --- README.md | 7 +- crates/relayburn-cli/src/commands/init.rs | 6 +- crates/relayburn-sdk/src/ingest/ingest.rs | 61 ++++++++++--- .../src/ingest/orchestration_tests.rs | 74 ++++++++++++++++ crates/relayburn-sdk/src/reader/copilot.rs | 87 +++++++++++++++---- .../relayburn-sdk/src/reader/copilot/tests.rs | 83 ++++++++++++++++++ 6 files changed, 281 insertions(+), 37 deletions(-) diff --git a/README.md b/README.md index 86551885..86cf286d 100644 --- a/README.md +++ b/README.md @@ -216,6 +216,7 @@ plus the GitHub Copilot CLI OTEL export when it is enabled (see below). | `--interval ` | Poll interval in milliseconds. Default: `1000`. | | `--quiet` | Suppress stderr progress spinner / breadcrumbs. One-shot mode still writes the final summary on stdout. | | `--hook claude` | Read one Claude Code hook payload from stdin and ingest its single transcript via the SDK fast-path. | +| `--no-fsevents` | In watch mode, use polling instead of filesystem events. | ### Collector setup @@ -228,12 +229,12 @@ export COPILOT_OTEL_FILE_EXPORTER_PATH="$HOME/.copilot/otel/copilot.jsonl" Add that to your shell rc file, restart the shell, and subsequent `copilot` sessions emit one JSONL span per API call, which `burn ingest` picks up -automatically (files under `~/.copilot/otel/` are scanned even without the env -var on the ingest side). Nothing is recoverable retroactively — the exporter +automatically while the variable is set (the exporter file plus +`~/.copilot/otel/*.jsonl`; without it Copilot ingest is a silent no-op). +Nothing is recoverable retroactively — the exporter must be on before sessions are captured. Run `burn init copilot` to print these instructions. Coverage is usage-only: spans carry per-call token counts (input, output, cache read/write, reasoning) and the model, but no tool-call content. -| `--no-fsevents` | In watch mode, use polling instead of filesystem events. | | Example | Result | |---|---| diff --git a/crates/relayburn-cli/src/commands/init.rs b/crates/relayburn-cli/src/commands/init.rs index 6f18fe95..68a1b7c4 100644 --- a/crates/relayburn-cli/src/commands/init.rs +++ b/crates/relayburn-cli/src/commands/init.rs @@ -8,6 +8,8 @@ use crate::cli::InitArgs; +/// Print the opt-in setup for one collector (`burn init copilot`). +/// Read-only: never edits shell rc files, only prints copy-pasteable lines. pub fn run(args: InitArgs) -> i32 { match args.action { crate::cli::InitAction::Copilot => { @@ -23,8 +25,8 @@ pub fn run(args: InitArgs) -> i32 { \x20 export COPILOT_OTEL_FILE_EXPORTER_PATH=\"$HOME/.copilot/otel/copilot.jsonl\"\n\ \n\ \x20 Add that line to your shell rc file (~/.zshrc, ~/.bashrc, …) so\n\ - \x20 new shells pick it up. Any path works; under ~/.copilot/otel/ it is\n\ - \x20 picked up even on shells where the export isn't set.\n\ + \x20 new shells pick it up — burn ingest only scans Copilot\n\ + \x20 files while this variable is set. Any path works.\n\ \n\ 2. Restart your shell (or `source` the rc file), then run Copilot CLI.\n\ \n\ diff --git a/crates/relayburn-sdk/src/ingest/ingest.rs b/crates/relayburn-sdk/src/ingest/ingest.rs index f5548561..03cb537f 100644 --- a/crates/relayburn-sdk/src/ingest/ingest.rs +++ b/crates/relayburn-sdk/src/ingest/ingest.rs @@ -164,22 +164,25 @@ pub(crate) fn copilot_otel_dir() -> PathBuf { } /// Resolve the Copilot CLI OTEL JSONL files ingest scans. Unlike the other -/// harnesses this source is env-gated: nothing exists on disk until the -/// user sets `COPILOT_OTEL_FILE_EXPORTER_PATH` (see #14), so an empty list -/// is the normal steady state and never an error. +/// harnesses this source is env-gated: nothing is scanned until the user +/// sets `COPILOT_OTEL_FILE_EXPORTER_PATH` (see #14), so an empty list is +/// the normal steady state and never an error. An explicit +/// [`IngestRoots::copilot_otel_files`] override bypasses the gate so tests +/// can inject files directly. pub(crate) fn copilot_otel_files(roots: &IngestRoots) -> Vec { if let Some(files) = &roots.copilot_otel_files { return files.clone(); } + let configured = std::env::var("COPILOT_OTEL_FILE_EXPORTER_PATH") + .map(|var| var.trim().to_owned()) + .unwrap_or_default(); + if configured.is_empty() { + return Vec::new(); + } let mut files: Vec = Vec::new(); - if let Ok(var) = std::env::var("COPILOT_OTEL_FILE_EXPORTER_PATH") { - let trimmed = var.trim(); - if !trimmed.is_empty() { - let p = PathBuf::from(trimmed); - if p.is_file() { - files.push(p); - } - } + let p = PathBuf::from(&configured); + if p.is_file() { + files.push(p); } for file in list_jsonl_files(&copilot_otel_dir()) { if !files.contains(&file) { @@ -1070,8 +1073,9 @@ fn ingest_opencode_into( /// Iterate the Copilot CLI OTEL export files (`COPILOT_OTEL_FILE_EXPORTER_PATH` /// plus `$COPILOT_HOME/otel/*.jsonl`), driving /// [`parse_copilot_otel_incremental`] with the carried per-session turn -/// counters and chat-trace set. Env-gated: when neither source exists the -/// file list is empty and this is a silent no-op (#14). +/// counters and chat-trace set. Env-gated: without +/// `COPILOT_OTEL_FILE_EXPORTER_PATH` (and without an explicit roots +/// override) the file list is empty and this is a silent no-op (#14). /// /// Rotation handling matches the Claude/Codex adapters: an inode change or /// a shrunken file restarts the byte offset at 0, but the per-session @@ -1155,6 +1159,10 @@ fn ingest_copilot_into( report.ingested_sessions += 1; ledger.append_turns(&parsed.turns)?; } + // Keep the inference table in lockstep with the persisted turns, + // like every other harness path (issue #434) — otherwise + // `burn flow` and span-tree reads see no Copilot API calls. + apply_parsed_extras(ledger, &parsed)?; let next = CopilotCursor { inode, @@ -1405,6 +1413,33 @@ impl_derived_records_common!(ClaudeParseIncrementalResult); impl_derived_records_common!(ParseCodexIncrementalResult); impl_derived_records_common!(ParseOpencodeIncrementalResult); +/// The Copilot OTEL parser emits usage-only turns with no trailing content, +/// compaction, relationship, tool-result, or user-turn buckets, so every +/// bucket but `turns` is empty. The `turns` slice feeds the +/// `apply_parsed_extras` inference materializer; the request-id lookup +/// stays the trait default (empty) since Copilot spans carry no `requestId` +/// equivalent, and the inference builder falls back to `message_id`. +impl DerivedRecords for ParseCopilotIncrementalResult { + fn content(&self) -> &[ContentRecord] { + &[] + } + fn events(&self) -> &[CompactionEvent] { + &[] + } + fn relationships(&self) -> &[SessionRelationshipRecord] { + &[] + } + fn tool_result_events(&self) -> &[ToolResultEventRecord] { + &[] + } + fn user_turns(&self) -> &[UserTurnRecord] { + &[] + } + fn turns(&self) -> &[TurnRecord] { + &self.turns + } +} + /// Per-type adapter for the `request_id_lookup` override (issue #434). /// Specialized for Claude's two result types so they borrow the parser's /// real lookup; the generic fallback returns an empty owned lookup so diff --git a/crates/relayburn-sdk/src/ingest/orchestration_tests.rs b/crates/relayburn-sdk/src/ingest/orchestration_tests.rs index e8dc7f00..7ae78ba1 100644 --- a/crates/relayburn-sdk/src/ingest/orchestration_tests.rs +++ b/crates/relayburn-sdk/src/ingest/orchestration_tests.rs @@ -329,6 +329,15 @@ fn ingest_copilot_sessions_round_trips_otel_spans() { assert_eq!(turns[0].turn.usage.input, 100); assert_eq!(turns[1].turn.usage.output, 20); + // Inferences stay in lockstep with the persisted turns (issue #434): + // one per API call, carrying its usage. + let inferences = ledger + .query_inferences(&Query::for_session("conv-1")) + .unwrap(); + assert_eq!(inferences.len(), 2); + assert_eq!(inferences[0].usage.input, 100); + assert_eq!(inferences[1].usage.output, 20); + let cursors = load_cursors(&ledger).unwrap(); let key = export_file.to_string_lossy().into_owned(); match cursors.get_typed(&key) { @@ -354,6 +363,71 @@ fn ingest_copilot_sessions_round_trips_otel_spans() { assert_eq!(turns[2].turn.turn_index, 2); } +/// Restores one env var to its prior value on drop so env-mutating tests +/// can't leak into other tests sharing the process. +struct RestoreEnv { + key: &'static str, + prior: Option, +} + +impl Drop for RestoreEnv { + fn drop(&mut self) { + match &self.prior { + Some(value) => std::env::set_var(self.key, value), + None => std::env::remove_var(self.key), + } + } +} + +#[test] +fn ingest_copilot_sessions_requires_exporter_env_var() { + let tmp = TempDir::new().unwrap(); + let _env = isolated_relayburn_home(&tmp); + let _restore_exporter = RestoreEnv { + key: "COPILOT_OTEL_FILE_EXPORTER_PATH", + prior: std::env::var("COPILOT_OTEL_FILE_EXPORTER_PATH").ok(), + }; + let _restore_home = RestoreEnv { + key: "COPILOT_HOME", + prior: std::env::var("COPILOT_HOME").ok(), + }; + + // A valid export file sits in the default OTEL dir, but the exporter + // variable is unset: the collector must stay a silent no-op (#14 + // opt-in gate). No explicit roots override, so discovery reads env. + let otel_dir = tmp.path().join("copilot-home").join("otel"); + fs::create_dir_all(&otel_dir).unwrap(); + let export_file = otel_dir.join("copilot.jsonl"); + fs::write( + &export_file, + "{\"type\":\"span\",\"traceId\":\"t-g\",\"spanId\":\"s-g\",\"name\":\"chat m\",\"startTime\":[1775934260,0],\"attributes\":{\"gen_ai.operation.name\":\"chat\",\"gen_ai.conversation.id\":\"conv-g\",\"gen_ai.usage.input_tokens\":10,\"gen_ai.usage.output_tokens\":2}}\n", + ) + .unwrap(); + std::env::remove_var("COPILOT_OTEL_FILE_EXPORTER_PATH"); + std::env::set_var("COPILOT_HOME", tmp.path().join("copilot-home")); + + let roots = IngestRoots { + copilot_otel_files: None, + ..pinned_roots(&tmp) + }; + let mut ledger = open_ledger_in(&tmp); + let opts = IngestOptions { + roots, + ..Default::default() + }; + + let report = ingest_copilot_sessions(&mut ledger, &opts).unwrap(); + assert_eq!(report.scanned_sessions, 0); + assert_eq!(report.appended_turns, 0); + + // Opting in picks the same file up. + std::env::set_var("COPILOT_OTEL_FILE_EXPORTER_PATH", &export_file); + let report = ingest_copilot_sessions(&mut ledger, &opts).unwrap(); + assert_eq!(report.appended_turns, 1); + let turns = ledger.query_turns(&Query::for_session("conv-g")).unwrap(); + assert_eq!(turns.len(), 1); +} + #[test] fn ingest_copilot_sessions_is_a_noop_without_exporter_files() { let tmp = TempDir::new().unwrap(); diff --git a/crates/relayburn-sdk/src/reader/copilot.rs b/crates/relayburn-sdk/src/reader/copilot.rs index c62e87ad..46e68897 100644 --- a/crates/relayburn-sdk/src/reader/copilot.rs +++ b/crates/relayburn-sdk/src/reader/copilot.rs @@ -7,7 +7,9 @@ //! per line — a `chat` span per API call, an `invoke_agent` summary span per //! turn, plus metrics and log records we ignore. Burn additionally globs //! `$COPILOT_HOME/otel/*.jsonl` (default `~/.copilot/otel`) so a directory -//! pointed at by the exporter is covered too. +//! pointed at by the exporter is covered too. Both are scanned only while +//! `COPILOT_OTEL_FILE_EXPORTER_PATH` is set — without it ingest is a +//! silent no-op (test roots can still inject files directly). //! //! Format reference: tokscale's `sessions/copilot.rs` and the GitHub Docs //! Copilot CLI command reference. Notable shapes handled here: @@ -34,7 +36,7 @@ //! content, so the emitted [`TurnRecord`]s are usage-only //! ([`UsageGranularity::PerMessage`] — one record per API call). -use std::collections::BTreeMap; +use std::collections::{BTreeMap, BTreeSet}; use std::fs::File; use std::io::{BufReader, Read, Seek, SeekFrom}; use std::path::Path; @@ -72,6 +74,10 @@ pub struct CopilotResumeState { pub chat_trace_ids: Vec, } +/// Turns parsed from one incremental pass over a Copilot OTEL export, +/// plus the cursor state (`end_offset` + `resume`) the next pass continues +/// from. `end_offset` stops at the last complete line so a partial tail +/// still being flushed is re-read next pass. #[derive(Debug, Clone, Default)] pub struct ParseCopilotIncrementalResult { pub turns: Vec, @@ -135,7 +141,13 @@ pub fn parse_copilot_otel_incremental( let mut trace_contexts: BTreeMap = BTreeMap::new(); let mut candidates: Vec = Vec::new(); - for (index, line) in buf[..consumed].split(|b| *b == b'\n').enumerate() { + // `consumed` ends on a newline, so every piece (including blanks) + // occupies `len + 1` bytes of the file; track the absolute offset so + // the `message_id` fallback below stays stable across passes. + let mut line_offset = start_offset; + for line in buf[..consumed].split(|b| *b == b'\n') { + let this_offset = line_offset; + line_offset += line.len() as u64 + 1; let trimmed = trim_ascii(line); if trimmed.is_empty() { continue; @@ -145,15 +157,29 @@ pub fn parse_copilot_otel_incremental( Err(_) => continue, // lossy per line: one bad record never truncates the file }; accumulate_trace_context(&mut trace_contexts, &record); - if let Some(candidate) = candidate_from_record(&record, index, opts.fallback_ts_ms) { + if let Some(candidate) = candidate_from_record(&record, this_offset, opts.fallback_ts_ms) { candidates.push(candidate); } } + // Suppression must not depend on record order: an `invoke_agent` + // summary can precede its trace's `chat` spans in the same chunk + // (exporter flush order isn't contractual), so collect this chunk's + // chat traces before resolving anything. + let chunk_chat_traces: BTreeSet = candidates + .iter() + .filter(|c| c.kind == SpanKind::Chat) + .filter_map(|c| c.trace_id.clone()) + .collect(); + let mut turns = Vec::new(); for candidate in candidates { - let Some(turn) = candidate.resolve(&trace_contexts, &mut resume, session_path.as_deref()) - else { + let Some(turn) = candidate.resolve( + &trace_contexts, + &chunk_chat_traces, + &mut resume, + session_path.as_deref(), + ) else { continue; }; turns.push(turn); @@ -243,6 +269,15 @@ fn attr_i64_first(attributes: &Map, keys: &[&str]) -> i64 { const MODEL_ATTRS: &[&str] = &["gen_ai.response.model", "gen_ai.request.model"]; +/// Model suffix of a `chat ` span name. The exporter always names +/// chat spans this way, so a usage-bearing span without model attributes +/// still identifies its model instead of falling back to `"unknown"`. +fn model_from_span_name(name: &str) -> Option<&str> { + name.strip_prefix("chat ") + .map(str::trim) + .filter(|model| !model.is_empty()) +} + /// Session id attribute priority, mirroring tokscale's `SESSION_ATTRS`. const SESSION_ATTRS: &[&str] = &[ "gen_ai.conversation.id", @@ -361,7 +396,15 @@ fn accumulate_trace_context(contexts: &mut BTreeMap, recor .unwrap_or_default(); let ctx = contexts.entry(trace).or_default(); if ctx.model.is_none() { - ctx.model = first_non_empty_attr(&attributes, MODEL_ATTRS).map(str::to_string); + ctx.model = first_non_empty_attr(&attributes, MODEL_ATTRS) + .map(str::to_string) + .or_else(|| { + record + .get("name") + .and_then(Value::as_str) + .and_then(model_from_span_name) + .map(str::to_string) + }); } if ctx.session_id.is_none() { ctx.session_id = best_session_attr(&attributes).map(str::to_string); @@ -378,18 +421,23 @@ struct PendingCandidate { stop_reason: Option, ts_ms: i64, usage: Usage, - /// Position among successfully parsed records in this chunk; fallback - /// identity for spans with no `spanId`. - index: usize, + /// Absolute byte offset of this span's line in the export file; + /// fallback identity for spans with no `spanId`. Chunk-local indexes + /// would restart at zero on every incremental pass and collide in the + /// ledger's `(source, session_id, message_id)` key; the absolute + /// offset is stable for a given file generation, so a re-read span + /// dedups instead of dropping a later, distinct turn. + line_offset: u64, } fn candidate_from_record( record: &Value, - index: usize, + line_offset: u64, fallback_ts_ms: Option, ) -> Option { let attributes = record.get("attributes").and_then(Value::as_object)?; let kind = classify_span(record, attributes)?; + let span_name = record.get("name").and_then(Value::as_str).unwrap_or(""); let input = attr_i64(attributes, "gen_ai.usage.input_tokens"); let output = attr_i64(attributes, "gen_ai.usage.output_tokens"); @@ -457,12 +505,14 @@ fn candidate_from_record( .map(str::trim) .filter(|s| !s.is_empty()) .map(str::to_string), - model: first_non_empty_attr(attributes, MODEL_ATTRS).map(str::to_string), + model: first_non_empty_attr(attributes, MODEL_ATTRS) + .map(str::to_string) + .or_else(|| model_from_span_name(span_name).map(str::to_string)), session_id: best_session_attr(attributes).map(str::to_string), stop_reason, ts_ms, usage, - index, + line_offset, }) } @@ -470,6 +520,7 @@ impl PendingCandidate { fn resolve( self, trace_contexts: &BTreeMap, + chunk_chat_traces: &BTreeSet, resume: &mut CopilotResumeState, session_path: Option<&str>, ) -> Option { @@ -491,11 +542,9 @@ impl PendingCandidate { } } SpanKind::AgentSummary => { - if self - .trace_id - .as_ref() - .is_some_and(|t| resume.chat_trace_ids.contains(t)) - { + if self.trace_id.as_ref().is_some_and(|t| { + chunk_chat_traces.contains(t) || resume.chat_trace_ids.contains(t) + }) { return None; } } @@ -516,7 +565,7 @@ impl PendingCandidate { let message_id = self .span_id .or(self.response_id) - .unwrap_or_else(|| format!("line-{}", self.index)); + .unwrap_or_else(|| format!("line-{}", self.line_offset)); let turn_index = resume .session_turn_counts diff --git a/crates/relayburn-sdk/src/reader/copilot/tests.rs b/crates/relayburn-sdk/src/reader/copilot/tests.rs index 81ef7e2e..7c372522 100644 --- a/crates/relayburn-sdk/src/reader/copilot/tests.rs +++ b/crates/relayburn-sdk/src/reader/copilot/tests.rs @@ -177,6 +177,89 @@ fn invoke_agent_suppressed_across_incremental_passes() { assert!(second.turns.is_empty()); } +#[test] +fn invoke_agent_before_chat_in_same_chunk_is_suppressed() { + let tmp = tempdir().unwrap(); + let path = tmp.path().join("copilot.jsonl"); + + // Exporter flush order isn't contractual: the aggregate summary can + // precede its trace's chat spans in the same chunk. Suppression must + // not depend on record order, or the trace is double counted. + let summary = br#"{"type":"span","traceId":"t-ord","spanId":"i-ord","name":"invoke_agent","startTime":[1775934259,0],"attributes":{"gen_ai.operation.name":"invoke_agent","gen_ai.conversation.id":"conv-ord","gen_ai.usage.input_tokens":10,"gen_ai.usage.output_tokens":2}}"#; + let chat = br#"{"type":"span","traceId":"t-ord","spanId":"s-ord","name":"chat m","startTime":[1775934260,0],"attributes":{"gen_ai.operation.name":"chat","gen_ai.conversation.id":"conv-ord","gen_ai.usage.input_tokens":10,"gen_ai.usage.output_tokens":2}}"#; + + std::fs::write( + &path, + [summary.as_slice(), b"\n", chat.as_slice(), b"\n"].concat(), + ) + .unwrap(); + let parsed = + parse_copilot_otel_incremental(&path, &ParseCopilotIncrementalOptions::default()).unwrap(); + assert_eq!(parsed.turns.len(), 1); + assert_eq!(parsed.turns[0].message_id, "s-ord"); +} + +#[test] +fn chat_span_name_provides_model_fallback() { + let tmp = tempdir().unwrap(); + let path = tmp.path().join("copilot.jsonl"); + + // No gen_ai.{request,response}.model attributes: the `chat ` + // span name is the only model signal and must beat "unknown". + let span = br#"{"type":"span","traceId":"t-nm","spanId":"s-nm","name":"chat claude-sonnet-4.6","startTime":[1775934260,0],"attributes":{"gen_ai.operation.name":"chat","gen_ai.conversation.id":"conv-nm","gen_ai.usage.input_tokens":10,"gen_ai.usage.output_tokens":2}}"#; + std::fs::write(&path, [span.as_slice(), b"\n"].concat()).unwrap(); + let parsed = + parse_copilot_otel_incremental(&path, &ParseCopilotIncrementalOptions::default()).unwrap(); + assert_eq!(parsed.turns.len(), 1); + assert_eq!(parsed.turns[0].model, "claude-sonnet-4.6"); +} + +#[test] +fn id_less_spans_get_offset_stable_message_ids() { + let tmp = tempdir().unwrap(); + let path = tmp.path().join("copilot.jsonl"); + + // Spans with neither spanId nor gen_ai.response.id fall back to a + // file-offset identity. A chunk-local index would restart at zero on + // the second pass and collide in the ledger key; the absolute offset + // keeps the two turns distinct. + let line = |trace: &str| { + format!( + "{{\"type\":\"span\",\"traceId\":\"{trace}\",\"name\":\"chat m\",\"startTime\":[1775934260,0],\"attributes\":{{\"gen_ai.operation.name\":\"chat\",\"gen_ai.conversation.id\":\"conv-off\",\"gen_ai.usage.input_tokens\":10,\"gen_ai.usage.output_tokens\":2}}}}" + ) + }; + let line_a = line("t-off-a"); + let line_b = line("t-off-b"); + std::fs::write(&path, format!("{line_a}\n")).unwrap(); + let first = + parse_copilot_otel_incremental(&path, &ParseCopilotIncrementalOptions::default()).unwrap(); + assert_eq!(first.turns.len(), 1); + assert_eq!(first.turns[0].message_id, "line-0"); + + let mut f = std::fs::OpenOptions::new() + .append(true) + .open(&path) + .unwrap(); + writeln!(f, "{line_b}").unwrap(); + drop(f); + + let second = parse_copilot_otel_incremental( + &path, + &ParseCopilotIncrementalOptions { + start_offset: Some(first.end_offset), + resume: Some(first.resume), + ..Default::default() + }, + ) + .unwrap(); + assert_eq!(second.turns.len(), 1); + assert_eq!( + second.turns[0].message_id, + format!("line-{}", first.end_offset) + ); + assert_ne!(first.turns[0].message_id, second.turns[0].message_id); +} + #[test] fn empty_and_missing_files() { let tmp = tempdir().unwrap();