From 571a02f06b649fa4001d6e12234ab5bf19e3f913 Mon Sep 17 00:00:00 2001 From: Jack Champagne Date: Wed, 29 Jul 2026 05:39:28 -0400 Subject: [PATCH] Add corpus-side trace distiller + transcript exporter otlp_export.py can only export your OWN sessions: it reads opencode's SQLite at ~/.local/share/opencode, which lives on the user's machine. So looking at a teammate's run meant asking them to send you their database. It does not have to. The raw wire capture in the run corpus already carries every semantic attribute the AI SDK emits -- ai.prompt, ai.response.text/reasoning, ai.toolCall.args/result, ai.usage.*, ai.model.id, and the real opencode session id. It is just diluted ~300:1 by opencode's own internal instrumentation (sql.execute 3.6k spans, Session.getPart, SessionProcessor.*, http.server). The clean view is a filter over data we already own, not a different source. - corpus_distill.py: corpus *.otlp.gz -> AGENT/LLM/TOOL OpenInference trace, POSTed as protobuf. Measured 18918 -> 67 spans (282:1) on a real session; output is structurally identical to otlp_export.py's. - corpus_export.py: corpus *.otlp.gz -> readable .md transcript plus .spans.json, same format as the SQLite path. Identity is the token-verified `submitter` from S3 object metadata, never the client-reported ai.telemetry.metadata.userId or the machineId hash -- both of those are client-supplied and spoofable. All three are recorded, and the trace is tagged identity:verified or identity:client-reported so provenance is never ambiguous. Langfuse's mapping was verified against a live instance: langfuse.user.id -> userId (a plain user.id is silently ignored), langfuse.session.id -> sessionId, langfuse.tags -> tags. Neither script relies on ai.prompt parsing. The 16 KB attribute cap clips 27 of 28 prompts into invalid JSON, so both scrape it with a regex fallback and rebuild the transcript from the untruncated pieces instead. ai.prompt is the whole conversation re-sent every call, so it is quadratically redundant -- the content is already held as reasoning/text/ tool spans. Root titles come from opencode's own title-generator call, which is small enough to escape the cap; the boilerplate prompt is skipped so roots are not all named "Generate a title for this conversation:". Genuinely lost content is limited to tool results over 16 KB, which are marked [TRUNCATED at 16 KB cap] rather than dropped. gitignore extended for the new outputs: they carry raw prompts and chain-of-thought, same as the existing exports. --- tools/trace-export/.gitignore | 2 + tools/trace-export/README.md | 89 ++++++- tools/trace-export/corpus_distill.py | 331 +++++++++++++++++++++++++++ tools/trace-export/corpus_export.py | 212 +++++++++++++++++ 4 files changed, 632 insertions(+), 2 deletions(-) create mode 100644 tools/trace-export/corpus_distill.py create mode 100644 tools/trace-export/corpus_export.py diff --git a/tools/trace-export/.gitignore b/tools/trace-export/.gitignore index aac98c97..8e7aa7cf 100644 --- a/tools/trace-export/.gitignore +++ b/tools/trace-export/.gitignore @@ -1,4 +1,6 @@ # session exports contain raw prompts + chain-of-thought (amicode.reasoning) -- never commit *.otlp.json ses_*.md +*.spans.json +_submitter __pycache__/ diff --git a/tools/trace-export/README.md b/tools/trace-export/README.md index 084ee46f..dc7ae043 100644 --- a/tools/trace-export/README.md +++ b/tools/trace-export/README.md @@ -1,12 +1,97 @@ # amicode → Phoenix / Langfuse trace export -Converts an amicode/opencode session (SQLite) into an OTLP / OpenInference trace. +Two sources, one output shape. + +**Local (your own sessions):** `otlp_export.py` reads opencode's SQLite +(`~/.local/share/opencode`) and emits a clean OTLP / OpenInference trace. + +**Cloud (anyone's sessions):** `corpus_distill.py` / `corpus_export.py` read the +run corpus instead. This matters because SQLite lives on the *user's machine* — +so the local path can only ever export your own runs. To look at a teammate's +run you would otherwise have to ask them to send you their database. + +You don't have to. The raw wire capture already carries every semantic +attribute the AI SDK emits; it is just diluted ~300:1 by opencode's own internal +instrumentation (`sql.execute`, `Session.getPart`, `SessionProcessor.*`, +`http.server`). The clean view is a **filter over the corpus**, not a different +data source. ## Files -- `otlp_export.py` — pure-stdlib converter (no pip deps to *produce* the trace) +- `otlp_export.py` — pure-stdlib converter, SQLite → OTLP JSON + transcript +- `corpus_distill.py` — corpus `*.otlp.gz` → distilled trace, POSTed to Langfuse/Phoenix +- `corpus_export.py` — corpus `*.otlp.gz` → `.md` transcript + `.spans.json` +- `push_otlp.py` — OTLP JSON → protobuf → collector (viewers reject OTLP/JSON; see below) - `.otlp.json` — OTLP `ExportTraceServiceRequest` JSON, OpenInference attributes - `.md` — human-readable transcript +## Corpus workflow + +```bash +# 1. pull a session's raw objects (prefix is the per-activation session id) +aws s3 sync s3://harmoniqs-run-corpus-production-/v1/production/// ./corpus/ + +# 2. record the token-verified submitter -- S3 object metadata is NOT preserved +# by `aws s3 sync`, so carry it out-of-band (or pass --submitter) +aws s3api head-object --bucket harmoniqs-run-corpus-production- \ + --key --query 'Metadata.submitter' --output text > ./corpus/_submitter + +# 3. readable transcript +python3 corpus_export.py ./corpus ~/traces-out + +# 4. into Langfuse +OTLP_AUTH_BASIC=pk-lf-…:sk-lf-… python3 corpus_distill.py \ + --submitter raghav ./corpus http://localhost:3000/api/public/otel/v1/traces raghav +``` + +### Identity: use the submitter, not the client-reported id + +Three user identities appear in this data and only one is trustworthy: + +| source | trustworthy? | what it is | +| --- | --- | --- | +| `submitter` (S3 object metadata) | **yes** | derived by the ingest lambda from sha256 of the caller's bearer token; never client-supplied | +| `ai.telemetry.metadata.userId` | no | opencode config value, e.g. `raghavchari` — spoofable | +| `amicode.user` | no | VS Code `machineId` hash; pseudonymous per-install handle | + +`corpus_distill.py` prefers `--submitter`, falls back to the client-reported id, +and **tags the trace `identity:verified` or `identity:client-reported`** so the +provenance is never ambiguous. All three are also written as +`amicode.submitter` / `amicode.opencode_user` / `amicode.machine_id`. + +Langfuse's OTLP attribute mapping was verified empirically against a local +instance: **`langfuse.user.id` → `trace.userId`** (it wins over a plain +`user.id`, which is silently ignored), `langfuse.session.id` → `trace.sessionId`, +`langfuse.tags` (JSON array string) → tags. + +### Known fidelity limit: the 16 KB attribute cap + +amicode injects `OTEL_ATTRIBUTE_VALUE_LENGTH_LIMIT=16384`, which clips long +attribute values. Measured on a real session: + +| attribute | intact | note | +| --- | --- | --- | +| `ai.toolCall.args` | 38/38 | tool inputs complete | +| `ai.toolCall.result` | 35/37 | tool outputs ~95% complete | +| `ai.response.text` | 56/56 | complete | +| `ai.response.reasoning` | 52/52 | complete | +| `ai.prompt` | 1/28 | **clipped — often invalid JSON** | + +`ai.prompt` is the whole conversation re-sent on every call, so it is +quadratically redundant: turn N restates turns 1..N-1, which we already hold as +untruncated reasoning/text/tool spans. Both corpus scripts therefore rebuild the +transcript from the pieces and never rely on `ai.prompt` parsing — they scrape it +with a regex fallback and mark spans `amicode.prompt.truncated`. + +The user's opening request still survives, via a quirk worth knowing: opencode's +first LLM call in a session is its own *title generator* (`[system, "Generate a +title for this conversation:", ]`) and is small enough to escape +the cap. That is where root titles come from — and why naming a root from "first +user message" without skipping the boilerplate labels every session +`Generate a title for this conversation:`. + +What is genuinely lost: any single tool result over 16 KB. Those are marked +`[TRUNCATED at 16 KB cap]` in the transcript rather than silently dropped. + ## Span model ``` AGENT (session) diff --git a/tools/trace-export/corpus_distill.py b/tools/trace-export/corpus_distill.py new file mode 100644 index 00000000..095b4242 --- /dev/null +++ b/tools/trace-export/corpus_distill.py @@ -0,0 +1,331 @@ +#!/usr/bin/env python3 +"""Distil raw run-corpus OTLP into the clean OpenInference shape, SERVER-SIDE. + +WHY THIS EXISTS (and why otlp_export.py cannot be used here): +otlp_export.py builds its clean AGENT->LLM->TOOL spans by reading opencode's +LOCAL SQLite session store (~/.local/share/opencode). That file lives on the +USER's machine. For anyone but yourself it is unreachable, so the good pipeline +does not apply to corpus data at all. + +It does not need to. The raw wire capture already carries every semantic +attribute the AI SDK emits -- ai.prompt, ai.response.text/reasoning, +ai.toolCall.args/result, ai.usage.*, ai.model.id, and the real opencode session +id -- it is just diluted ~140:1 by opencode's own internal instrumentation +(sql.execute, Session.getPart, SessionProcessor.*, http.server). So the clean +view is a FILTER over the corpus, not a different data source. + +Mapping (mirrors otlp_export.py's output so both render identically): + ai.streamText -> openinference.span.kind=LLM (+ llm.model_name, token counts) + ai.toolCall -> openinference.span.kind=TOOL (+ tool.name, args/result) + synthetic root -> openinference.span.kind=AGENT (one per opencode session) + +Spans are re-parented onto the synthetic session root, because the real parents +(LLM.run, SessionTools.resolve) are exactly the noise being removed -- keeping +them would drag the whole internal tree back in. + +Usage: + OTLP_AUTH_BASIC=pk:sk python3 corpus_distill.py [endpoint] [label] +""" +import base64 +import glob +import gzip +import hashlib +import json +import os +import re +import sys +import urllib.error +import urllib.request +from collections import Counter, defaultdict + +from opentelemetry.proto.collector.trace.v1 import trace_service_pb2 as ts +from opentelemetry.proto.common.v1 import common_pb2 as cp +from opentelemetry.proto.resource.v1 import resource_pb2 as rp + + +def attrs(s): + out = {} + for a in s.get("attributes", []): + v = a.get("value", {}) + out[a["key"]] = next(iter(v.values())) if v else None + return out + + +def load(src): + spans, resource = {}, {} + for f in sorted(glob.glob(os.path.join(src, "**", "*.otlp.gz"), recursive=True)): + raw = gzip.decompress(open(f, "rb").read()) + if not raw.lstrip().startswith(b"{"): + continue + doc = json.loads(raw) + for rs in doc.get("resourceSpans", []): + for a in rs.get("resource", {}).get("attributes", []): + v = a.get("value", {}) + resource.setdefault(a["key"], next(iter(v.values())) if v else None) + for ss in rs.get("scopeSpans", []): + for s in ss.get("spans", []): + spans[s["spanId"]] = s + return spans, resource + + +def sv(x): + av = cp.AnyValue() + av.string_value = "" if x is None else (x if isinstance(x, str) else json.dumps(x)) + return av + + +def iv(x): + av = cp.AnyValue() + av.int_value = int(x) + return av + + +def kv(k, val): + o = cp.KeyValue(key=k) + o.value.CopyFrom(val) + return o + + +def synth_id(*parts, n=8): + return hashlib.sha256("|".join(str(p) for p in parts).encode()).digest()[:n] + + +# opencode's FIRST llm call in a session is its own title generator (system +# prompt: "You are a title generator"), and its message list is +# [system, user("Generate a title for this conversation:"), user()]. +# Naming the root from "first user message" therefore labels every session with +# opencode's internal housekeeping prompt instead of what the user asked. +TITLE_BOILERPLATE = "generate a title for this conversation" + +# ai.prompt is frequently INVALID JSON: OTEL_ATTRIBUTE_VALUE_LENGTH_LIMIT=16384 +# (set by amicode's env injection) truncates it mid-string. So never rely on +# json.loads succeeding -- fall back to scraping "text" values out of the +# fragment. This is the fidelity cost of the attribute cap, visible in practice. +TEXT_RE = re.compile(r'"text"\s*:\s*"((?:[^"\\]|\\.){4,})"') + + +def user_texts(prompt_json): + """Every user-authored text in an ai.prompt, tolerating truncation.""" + out = [] + try: + for m in json.loads(prompt_json or "{}").get("messages", []): + if m.get("role") != "user": + continue + c = m.get("content") + if isinstance(c, str): + out.append(c) + elif isinstance(c, list): + out.extend(p.get("text", "") for p in c if isinstance(p, dict)) + except Exception: + out.extend(m.group(1) for m in TEXT_RE.finditer(prompt_json or "")) + return [t.strip() for t in out if t and t.strip()] + + +def prompt_parses(p): + try: + json.loads(p or "{}") + return True + except Exception: + return False + + +def extract_title(llm_spans): + """The user's actual opening request, skipping opencode's title-gen prompt.""" + for s, a in sorted(llm_spans, key=lambda x: int(x[0]["startTimeUnixNano"])): + for t in user_texts(a.get("ai.prompt")): + if TITLE_BOILERPLATE in t.lower(): + continue + t = t.replace("\\n", " ").strip() + if t: + return t.splitlines()[0][:100] + return None + + +def resolve_submitter(src, explicit): + """The token-verified submitter. Explicit flag wins; otherwise read the + sidecar that fetch_corpus writes (S3 object metadata is NOT preserved by + `aws s3 sync`, so it has to be carried out-of-band or passed in).""" + if explicit: + return explicit + side = os.path.join(src, "_submitter") + if os.path.exists(side): + v = open(side).read().strip() + if v: + return v + return None + + +def main(): + argv = sys.argv[1:] + submitter = None + if "--submitter" in argv: + i = argv.index("--submitter") + submitter = argv[i + 1] + del argv[i:i + 2] + src = argv[0] + endpoint = argv[1] if len(argv) > 1 else "http://localhost:3000/api/public/otel/v1/traces" + label = argv[2] if len(argv) > 2 else "distilled" + submitter = resolve_submitter(src, submitter) + + spans, resource = load(src) + # Group the semantic spans by the REAL opencode session id (a per-span + # attribute), not by the corpus prefix -- the prefix is a per-activation stub + # and BatchSpanProcessor coalesces multiple sessions into one export. + by_session = defaultdict(lambda: {"llm": [], "tool": []}) + for s in spans.values(): + a = attrs(s) + sid = a.get("ai.telemetry.metadata.sessionId") or a.get("session.id") + if not sid: + continue + if s["name"] == "ai.streamText": + by_session[sid]["llm"].append((s, a)) + elif s["name"] == "ai.toolCall": + by_session[sid]["tool"].append((s, a)) + + req = ts.ExportTraceServiceRequest() + pb_rs = req.resource_spans.add() + res_attrs = [kv(k, sv(v)) for k, v in resource.items()] + res_attrs.append(kv("amicode.view", sv("distilled"))) + pb_rs.resource.CopyFrom(rp.Resource(attributes=res_attrs)) + pb_ss = pb_rs.scope_spans.add() + pb_ss.scope.CopyFrom(cp.InstrumentationScope(name="amicode.corpus.distill", version="1")) + + stats = Counter() + for sid, groups in sorted(by_session.items()): + members = groups["llm"] + groups["tool"] + if not members: + continue + trace_id = synth_id("trace", label, sid, n=16) + root_id = synth_id("root", label, sid) + start = min(int(s["startTimeUnixNano"]) for s, _ in members) + end = max(int(s["endTimeUnixNano"]) for s, _ in members) + + title = extract_title(groups["llm"]) or sid + + root = pb_ss.spans.add() + root.trace_id, root.span_id = trace_id, root_id + root.name = title + root.kind = 1 + root.start_time_unix_nano, root.end_time_unix_nano = start, end + # --- Identity ------------------------------------------------------ + # Langfuse's OTLP mapping was verified empirically against this instance: + # `langfuse.user.id` -> trace.userId (it WINS over a plain `user.id`), + # `langfuse.session.id` -> trace.sessionId, `langfuse.tags` -> tags. + # + # WHICH identity to use matters. Three candidates exist and only one is + # trustworthy: + # submitter <- AUTHORITATIVE. Derived by the + # ingest lambda from the sha256 of the caller's bearer token and + # stamped into S3 object metadata. Never client-supplied, so it + # cannot be spoofed -- that is the entire point of the bearer-auth + # design. It is NOT in the OTLP payload, so it must be passed in. + # ai.telemetry.metadata.userId <- client-reported (opencode config, + # e.g. "raghavchari"). Convenient, spoofable. + # amicode.user <- VS Code machineId hash. Pseudonymous + # per-install handle, also client-supplied. + # So: prefer the submitter; fall back to the client value but TAG the + # trace `identity:client-reported` so nobody mistakes it for verified. + client_user = next((a.get("ai.telemetry.metadata.userId") + for _, a in groups["llm"] + if a.get("ai.telemetry.metadata.userId")), None) + user_id = submitter or client_user or resource.get("amicode.user") + verified = bool(submitter) + tags = ["run-corpus", "distilled", + f"identity:{'verified' if verified else 'client-reported'}"] + if resource.get("amicode.repo"): + tags.append(f"repo:{resource['amicode.repo']}") + + root.attributes.extend([ + kv("openinference.span.kind", sv("AGENT")), + # Langfuse-native attribution (verified mapping above). + kv("langfuse.user.id", sv(user_id)), + kv("langfuse.session.id", sv(sid)), + kv("langfuse.tags", sv(json.dumps(tags))), + # Keep every candidate visible and labelled, so the provenance of + # the userId above is auditable rather than implicit. + kv("amicode.submitter", sv(submitter or "")), + kv("amicode.identity.verified", sv(str(verified).lower())), + kv("amicode.opencode_user", sv(client_user or "")), + kv("amicode.machine_id", sv(resource.get("amicode.user") or "")), + kv("amicode.session.id", sv(sid)), + kv("amicode.repo", sv(resource.get("amicode.repo"))), + kv("amicode.git_ref", sv(resource.get("amicode.git_ref"))), + kv("input.value", sv(title)), + kv("input.mime_type", sv("text/plain")), + ]) + stats["AGENT"] += 1 + + for s, a in sorted(members, key=lambda x: int(x[0]["startTimeUnixNano"])): + sp = pb_ss.spans.add() + sp.trace_id = trace_id + sp.span_id = synth_id("span", label, s["spanId"]) + sp.parent_span_id = root_id # flatten onto the agent root + sp.start_time_unix_nano = int(s["startTimeUnixNano"]) + sp.end_time_unix_nano = int(s["endTimeUnixNano"]) + sp.kind = 1 + if s["name"] == "ai.streamText": + sp.name = f"LLM {a.get('ai.model.id','?')}" + out = a.get("ai.response.text") or a.get("ai.response.reasoning") or "" + sp.attributes.extend([ + kv("openinference.span.kind", sv("LLM")), + kv("llm.model_name", sv(a.get("ai.model.id"))), + kv("llm.provider", sv(a.get("ai.model.provider"))), + kv("input.value", sv(a.get("ai.prompt"))), + kv("amicode.prompt.truncated", sv(str(not prompt_parses(a.get("ai.prompt"))).lower())), + kv("input.mime_type", sv("application/json")), + kv("output.value", sv(out)), + kv("output.mime_type", sv("text/plain")), + kv("llm.finish_reason", sv(a.get("ai.response.finishReason"))), + ]) + for src_k, dst_k in ( + ("ai.usage.inputTokens", "llm.token_count.prompt"), + ("ai.usage.outputTokens", "llm.token_count.completion"), + ("ai.usage.totalTokens", "llm.token_count.total"), + ("ai.usage.reasoningTokens", "llm.token_count.completion_details.reasoning"), + ("ai.usage.cachedInputTokens", "llm.token_count.prompt_details.cache_read"), + ): + if a.get(src_k) is not None: + sp.attributes.append(kv(dst_k, iv(a[src_k]))) + if a.get("ai.response.toolCalls"): + sp.attributes.append(kv("llm.output_messages.0.message.tool_calls", sv(a["ai.response.toolCalls"]))) + stats["LLM"] += 1 + else: + name = a.get("ai.toolCall.name", "tool") + sp.name = name + sp.attributes.extend([ + kv("openinference.span.kind", sv("TOOL")), + kv("tool.name", sv(name)), + kv("tool.parameters", sv(a.get("ai.toolCall.args"))), + kv("input.value", sv(a.get("ai.toolCall.args"))), + kv("input.mime_type", sv("application/json")), + kv("output.value", sv(a.get("ai.toolCall.result"))), + kv("output.mime_type", sv("application/json")), + kv("amicode.tool.call_id", sv(a.get("ai.toolCall.id"))), + ]) + stats["TOOL"] += 1 + + body = req.SerializeToString() + headers = {"Content-Type": "application/x-protobuf"} + auth = os.environ.get("OTLP_AUTH_BASIC") + if auth: + headers["Authorization"] = "Basic " + base64.b64encode(auth.encode()).decode() + try: + with urllib.request.urlopen(urllib.request.Request(endpoint, data=body, headers=headers), timeout=120) as r: + status, resp = r.status, r.read()[:200] + except urllib.error.HTTPError as e: + status, resp = e.code, e.read()[:300] + + total = sum(len(s.get("spans", [])) for s in [{"spans": pb_ss.spans}]) + print(f"raw spans in : {len(spans)}") + print(f"sessions found : {len(by_session)}") + print(f"distilled spans : {len(pb_ss.spans)} {dict(stats)}") + print(f"reduction : {len(spans)} -> {len(pb_ss.spans)} ({len(spans)/max(len(pb_ss.spans),1):.0f}:1)") + print(f"repo : {resource.get('amicode.repo')}@{resource.get('amicode.git_ref')}") + print(f"user (langfuse) : {submitter or '(none passed)'}" + f"{' [VERIFIED submitter]' if submitter else ' [falling back to CLIENT-REPORTED id -- pass --submitter]'}") + print(f"POST {endpoint} -> HTTP {status} {resp[:120]!r}") + return 0 if status in (200, 202, 204) else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tools/trace-export/corpus_export.py b/tools/trace-export/corpus_export.py new file mode 100644 index 00000000..d845330b --- /dev/null +++ b/tools/trace-export/corpus_export.py @@ -0,0 +1,212 @@ +#!/usr/bin/env python3 +"""Reconstruct a readable session transcript + distilled OTLP from the RUN CORPUS only. + +This is otlp_export.py's output, produced WITHOUT the user's machine. +otlp_export.py reads opencode's local SQLite (~/.local/share/opencode), so it can +only ever export your own sessions. This reads the cloud corpus instead, so it +works for any submitter. + +That is possible because the wire capture keeps the pieces that matter, and the +16 KB attribute cap lands almost entirely on the ONE attribute that is redundant: + + ai.toolCall.args 38/38 intact (max 437 B) <- tool INPUTS complete + ai.toolCall.result 35/37 intact <- tool OUTPUTS ~95% complete + ai.response.text 56/56 intact (max 721 B) + ai.response.reasoning 52/52 intact (max 10.8 KB) + ai.prompt 1/28 intact <- CLIPPED, but see below + +ai.prompt is the whole conversation history re-sent on every single call, so it +is ~quadratically redundant: turn N's prompt restates turns 1..N-1, which we +already hold as untruncated reasoning/text/tool spans. The only content that is +irrecoverable is a single user or tool message larger than 16 KB. So the +transcript below is rebuilt from the untruncated pieces, not from ai.prompt. + +Usage: + python3 corpus_export.py +""" +import glob +import gzip +import json +import os +import re +import sys +from collections import defaultdict + +CAP = 16384 +TITLE_BOILERPLATE = "generate a title for this conversation" +TEXT_RE = re.compile(r'"text"\s*:\s*"((?:[^"\\]|\\.)+)"') + + +def attrs(s): + out = {} + for a in s.get("attributes", []): + v = a.get("value", {}) + out[a["key"]] = next(iter(v.values())) if v else None + return out + + +def load(src): + spans, resource = {}, {} + for f in sorted(glob.glob(os.path.join(src, "**", "*.otlp.gz"), recursive=True)): + raw = gzip.decompress(open(f, "rb").read()) + if not raw.lstrip().startswith(b"{"): + continue + for rs in json.loads(raw).get("resourceSpans", []): + for a in rs.get("resource", {}).get("attributes", []): + v = a.get("value", {}) + resource.setdefault(a["key"], next(iter(v.values())) if v else None) + for ss in rs.get("scopeSpans", []): + for s in ss.get("spans", []): + spans[s["spanId"]] = s + return spans, resource + + +def clipped(v): + return isinstance(v, str) and len(v) >= CAP - 2 + + +def first_user_request(llm): + """The user's opening ask. Recoverable because opencode's title-generator call + is SMALL -- it carries [system, 'Generate a title...', ] and + so escapes the cap even when every later prompt is clipped.""" + for s, a in sorted(llm, key=lambda x: int(x[0]["startTimeUnixNano"])): + p = a.get("ai.prompt") or "" + cands = [] + try: + for m in json.loads(p).get("messages", []): + if m.get("role") != "user": + continue + c = m.get("content") + if isinstance(c, str): + cands.append(c) + elif isinstance(c, list): + cands += [x.get("text", "") for x in c if isinstance(x, dict)] + except Exception: + cands += [m.group(1) for m in TEXT_RE.finditer(p)] + for t in cands: + t = (t or "").replace("\\n", "\n").strip() + if t and TITLE_BOILERPLATE not in t.lower(): + return t + return None + + +def jfmt(v): + if v is None: + return "" + try: + return json.dumps(json.loads(v), indent=2)[:4000] + except Exception: + return str(v)[:4000] + + +def main(): + src, outdir = sys.argv[1], sys.argv[2] + os.makedirs(outdir, exist_ok=True) + spans, resource = load(src) + + sessions = defaultdict(lambda: {"llm": [], "tool": [], "exec": []}) + for s in spans.values(): + a = attrs(s) + sid = a.get("ai.telemetry.metadata.sessionId") or a.get("session.id") + if not sid: + continue + if s["name"] == "ai.streamText": + sessions[sid]["llm"].append((s, a)) + elif s["name"] == "ai.toolCall": + sessions[sid]["tool"].append((s, a)) + elif s["name"] == "Tool.execute": + sessions[sid]["exec"].append((s, a)) + + written = [] + for sid, g in sessions.items(): + if not g["llm"] and not g["tool"]: + continue + # Drop opencode's title-generator call from the step list -- it is + # housekeeping, not a turn the user took. + steps = [(s, a) for s, a in g["llm"] + if TITLE_BOILERPLATE not in (a.get("ai.prompt") or "").lower()] + if not steps: + steps = g["llm"] + steps.sort(key=lambda x: int(x[0]["startTimeUnixNano"])) + tools = sorted(g["tool"], key=lambda x: int(x[0]["startTimeUnixNano"])) + status = {a.get("tool.call_id"): "completed" for _, a in g["exec"]} + + title = first_user_request(g["llm"]) or sid + model = next((a.get("ai.model.id") for _, a in g["llm"] if a.get("ai.model.id")), "?") + tin = sum(int(a.get("ai.usage.inputTokens") or 0) for _, a in g["llm"]) + tout = sum(int(a.get("ai.usage.outputTokens") or 0) for _, a in g["llm"]) + treas = sum(int(a.get("ai.usage.reasoningTokens") or 0) for _, a in g["llm"]) + tcache = sum(int(a.get("ai.usage.cachedInputTokens") or 0) for _, a in g["llm"]) + n_clip = sum(1 for _, a in tools if clipped(a.get("ai.toolCall.result"))) + + L = [f"# {title.splitlines()[0][:110]}", ""] + L += [f"- **session**: `{sid}`", + f"- **submitter**: `{resource.get('amicode.user','?')[:12]}…` (identity verified server-side)", + f"- **repo**: `{resource.get('amicode.repo')}` @ `{resource.get('amicode.git_ref')}`", + f"- **model**: {model} **client**: {resource.get('amicode.client')} / opencode {resource.get('service.version')}", + f"- **tokens** in/out/reason/cacheR: {tin}/{tout}/{treas}/{tcache}", + f"- **steps**: {len(steps)} llm · {len(tools)} tool calls", + f"- **source**: run corpus (cloud) — reconstructed WITHOUT local sqlite", + ""] + if n_clip: + L += [f"> ⚠️ {n_clip} of {len(tools)} tool results were truncated at the 16 KB " + f"attribute cap and are shown clipped.", ""] + L += ["---", "", "## 👤 user", "", title, ""] + + # Pair each llm step with the tool calls it requested (ai.response.toolCalls + # lists the ids), falling back to time order for anything unmatched. + claimed = set() + for i, (s, a) in enumerate(steps, 1): + fin = a.get("ai.response.finishReason", "?") + L += [f"## 🤖 assistant · step {i} · {fin}", ""] + if a.get("ai.response.reasoning"): + L += ["
reasoning", ""] + L += ["> " + ln for ln in a["ai.response.reasoning"].splitlines()] + L += ["", "
", ""] + if a.get("ai.response.text", "").strip(): + L += [a["ai.response.text"], ""] + want = [] + try: + want = [tc.get("toolCallId") for tc in json.loads(a.get("ai.response.toolCalls") or "[]")] + except Exception: + pass + for _, ta in tools: + cid = ta.get("ai.toolCall.id") + if cid in want and cid not in claimed: + claimed.add(cid) + L += [f"**🔧 {ta.get('ai.toolCall.name')}** `{status.get(cid,'?')}`", + "```json", jfmt(ta.get("ai.toolCall.args")), "```", + "output:", "```", jfmt(ta.get("ai.toolCall.result")), + ("… [TRUNCATED at 16 KB cap]" if clipped(ta.get("ai.toolCall.result")) else ""), + "```", ""] + u = [f"{k.split('.')[-1]}={a[k]}" for k in + ("ai.usage.inputTokens", "ai.usage.outputTokens", "ai.usage.reasoningTokens") if a.get(k)] + if u: + L += [f"_tokens: {' '.join(u)}_", ""] + + orphan = [t for t in tools if t[1].get("ai.toolCall.id") not in claimed] + if orphan: + L += [f"## 🔧 unpaired tool calls ({len(orphan)})", "", + "_Requested by an llm span whose ai.response.toolCalls was clipped or not captured._", ""] + for _, ta in orphan: + L += [f"**🔧 {ta.get('ai.toolCall.name')}**", "```json", + jfmt(ta.get("ai.toolCall.args")), "```", "output:", "```", + jfmt(ta.get("ai.toolCall.result")), "```", ""] + + md = os.path.join(outdir, f"{sid}.md") + open(md, "w").write("\n".join(L)) + js = os.path.join(outdir, f"{sid}.spans.json") + json.dump({"session": sid, "resource": resource, + "steps": [{"name": s["name"], **a} for s, a in steps], + "tools": [{"name": s["name"], **a} for s, a in tools]}, + open(js, "w"), indent=2) + written.append((sid, title, len(steps), len(tools), n_clip, md)) + + print(f"raw spans: {len(spans)} sessions reconstructed: {len(written)}\n") + for sid, title, ns, nt, nc, md in written: + print(f" {sid}\n {title.splitlines()[0][:80]!r}\n {ns} steps, {nt} tool calls, {nc} clipped -> {md}") + return 0 + + +if __name__ == "__main__": + sys.exit(main())