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())