langchain-debug-bundle

What this skill does

# LangChain Debug Bundle (Python)

## Overview

An on-call engineer pages you at 2am: the production agent loops, `ToolMessage`
outputs are empty strings, the user sees "I could not find the answer." Someone
asks the right question — *what state was the graph in when it gave up?* — and
there is no answer, because the terminal that caught the failure is already
gone, the Kubernetes pod has restarted, and the LangSmith URL was never
recorded.

This skill produces one artifact: a single `bundle-<incident_id>.tar.gz` (typically
1-10 MB) containing everything a second engineer needs to reproduce the failure
without a live terminal — environment and version manifest, filtered
`astream_events(version="v2")` JSONL, a propagating callback stack, the
LangSmith trace URL, and a post-write sanitization pass.

Four pitfalls make naive bundles useless:

- **P01** — `ChatAnthropic.stream()` reports `token_usage` only on stream close; token math read from `on_llm_end` lags by stream duration, so cost context in the bundle is wrong.
- **P28** — `BaseCallbackHandler.with_config(callbacks=[...])` does NOT propagate into subgraphs or inner `create_react_agent` loops. A debug callback bound that way silently captures zero events from the place the incident actually happened.
- **P47** — `astream_events(version="v2")` emits 2,000+ events per invocation. A raw dump is 50 MB and unreadable; an SSE viewer crashes on it.
- **P67** — `astream_log()` is soft-deprecated in 1.0. Diagnostic tooling built on it breaks on the next minor version.

The skill's answer: assemble the manifest, capture v2 events with a whitelist
(drop lifecycle noise, keep `on_chat_model_stream` / `on_tool_*` / any `*_error`
event), attach `DebugCallbackHandler` via `config["callbacks"]` at invoke time,
pull the LangSmith URL from the active `RunTree`, run the sanitization pass,
tar it up. Pinned: `langchain-core 1.0.x`, `langgraph 1.0.x`, `langsmith 0.1.x`.
Pain-catalog anchors: P01, P28, P47, P67.

## Prerequisites

- Python 3.10+
- `langchain-core >= 1.0, < 2.0`, `langgraph >= 1.0, < 2.0`
- `langsmith >= 0.1.40` for `RunTree` access
- Active LangSmith project (`LANGSMITH_TRACING=true`, `LANGSMITH_API_KEY=...`,
  `LANGSMITH_PROJECT=...`) — canonical 1.0 env-var names, not the legacy
  `LANGCHAIN_TRACING_V2` (see P26).
- Write access to a staging directory outside the repo tree.

## Instructions

### Step 1 — Assemble the environment manifest

Record the runtime snapshot that lets a colleague reproduce on a different
host. See [env-manifest-template.md](references/env-manifest-template.md) for
the exact YAML shape.

```python
import platform, sys, os, subprocess, datetime

RELEVANT = [
    "langchain-core", "langchain", "langgraph",
    "langchain-anthropic", "langchain-openai",
    "langsmith", "anthropic", "openai", "pydantic",
]

def pip_show(name: str) -> str | None:
    try:
        out = subprocess.check_output(
            [sys.executable, "-m", "pip", "show", name],
            stderr=subprocess.DEVNULL, text=True,
        )
        for line in out.splitlines():
            if line.startswith("Version:"):
                return line.split(":", 1)[1].strip()
    except subprocess.CalledProcessError:
        return None

def build_manifest(incident_id: str, invoke_meta: dict) -> dict:
    return {
        "bundle_spec_version": "1.0",
        "generated_at": datetime.datetime.utcnow().isoformat() + "Z",
        "incident_id": incident_id,
        "runtime": {
            "python": sys.version.split()[0],
            "platform": platform.platform(),
            "cpu_count": os.cpu_count(),
        },
        "packages": [
            {"name": n, "version": pip_show(n)}
            for n in RELEVANT if pip_show(n) is not None
        ],
        # NAMES only — never values. Sanitized by design (P27 posture).
        "env_var_names_present": sorted(
            k for k in os.environ
            if k.startswith(("LANGSMITH_", "LANGCHAIN_", "ANTHROPIC_", "OPENAI_", "GOOGLE_"))
        ),
        "invocation": invoke_meta,
    }
```

Record env-var *names*, not values. Values go through the sanitization pass in
Step 5, but the safest design is never to capture them.

### Step 2 — Capture `astream_events(version="v2")` with a filter

Raw v2 events flood 2,000+ per invocation (P47). A server-side filter drops
lifecycle noise (`on_chain_start`/`on_chain_end`) and keeps model, tool, and
error events — yielding 50-200 events per invocation and a ~500 KB JSONL.

```python
import json, itertools
from pathlib import Path

KEEP = {
    "on_chat_model_start", "on_chat_model_end",
    "on_tool_start", "on_tool_end", "on_tool_error",
    "on_retriever_start", "on_retriever_end",
    "on_custom_event",
}
# Additionally: any event whose name ends in "_error"
# Additionally: 1-in-10 sampled on_chat_model_stream (for response reconstruction)

async def capture_events(graph, inputs, config, out_path: Path) -> int:
    sample = itertools.count()
    written = 0
    with out_path.open("w") as f:
        async for evt in graph.astream_events(inputs, config=config, version="v2"):
            name = evt["event"]
            if name == "on_chat_model_stream" and next(sample) % 10 != 0:
                continue
            if name not in KEEP and not name.endswith("_error"):
                continue
            f.write(json.dumps({
                "event": name,
                "name": evt.get("name"),
                "run_id": str(evt.get("run_id")),
                "tags": evt.get("tags"),
                "metadata": evt.get("metadata"),
                "data": _json_safe(evt.get("data", {})),
            }, default=str) + "\n")
            written += 1
    return written
```

Never use `astream_log()` (P67). The full event taxonomy and `_json_safe`
helper live in [astream-events-capture.md](references/astream-events-capture.md).

### Step 3 — Attach callbacks that propagate into subgraphs (P28)

Callbacks bound via `Runnable.with_config(callbacks=[...])` fire on the outer
chain only. They go silent the moment the graph crosses into a subgraph or an
inner `create_react_agent` loop — exactly where incidents happen. Pass them via
`config["callbacks"]` at invoke time instead.

```python
from langchain_core.callbacks import BaseCallbackHandler

class DebugCallbackHandler(BaseCallbackHandler):
    def __init__(self): self.records: list[dict] = []
    def on_tool_start(self, serialized, input_str, *, run_id, parent_run_id=None, **kw):
        self.records.append({
            "kind": "tool_start", "run_id": str(run_id),
            "parent_run_id": str(parent_run_id) if parent_run_id else None,
            "tool": serialized.get("name"), "input": input_str[:500],
        })
    def on_tool_error(self, error, *, run_id, **kw):
        self.records.append({
            "kind": "tool_error", "run_id": str(run_id),
            "error_type": type(error).__name__, "error_message": str(error)[:1000],
        })

debug = DebugCallbackHandler()

result = await agent.ainvoke(
    {"messages": [("user", reproducer_prompt)]},
    config={
        "configurable": {"thread_id": thread_id},
        "callbacks": [debug],                      # propagates into subgraphs
        "tags": ["debug-bundle", incident_id],
        "metadata": {"incident_id": incident_id},
    },
)
```

The full handler (LLM + retriever + tool lifecycle) and a propagation smoke
test live in [callback-propagation.md](references/callback-propagation.md).

### Step 4 — Record the LangSmith trace URL

A trace URL is cheaper than any local artifact — one click and the colleague
sees the full run with latency, token counts, and input/output per node. Pull
it from the active `RunTree` if you have a live handle; otherwise construct it
from the invoke's `run_id`:

```python
from langsmith.run_helpers import get_current_run_tree

def capture_langsmith_url() -> str | None:
    rt = get_current_run_tree()
    if rt is None:
        return None  # tracing not enabled or run already cl