Langchain Debug Bundle

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,
    }

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

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},
    },
)

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 closed
    return rt.get_url()  # https://smith.langchain.com/o/.../r/<run_id>

# Write to langsmith.url in the bundle:
url = capture_langsmith_url()
(staging / "langsmith.url").write_text(url or "(no trace URL available)")

import re

PATTERNS = [
    ("openai_key",    r"sk-proj-[A-Za-z0-9_-]{16,}|sk-[A-Za-z0-9_-]{32,}"),
    ("anthropic_key", r"sk-ant-[A-Za-z0-9_-]{16,}"),
    ("google_key",    r"AIza[A-Za-z0-9_-]{35}"),
    ("langsmith_key", r"lsv2_(?:pt|sk)_[A-Za-z0-9]{32,}"),
    ("bearer",        r"(?i)bearer\s+[A-Za-z0-9._~+/=-]{20,}"),
    ("db_uri",        r"[a-z]+://[^:/\s]+:[^@\s]+@[^/\s]+"),
    ("private_key",   r"-----BEGIN [A-Z ]*PRIVATE KEY-----[\s\S]*?-----END [A-Z ]*PRIVATE KEY-----"),
]

def sanitize_file(path, patterns=PATTERNS) -> dict[str, int]:
    text, counts = path.read_text(), {}
    for name, pat in patterns:
        new, n = re.subn(pat, f"[REDACTED:{name}]", text)
        if n: counts[name] = n; text = new
    path.write_text(text)
    return counts

import tarfile, yaml
from pathlib import Path

def write_bundle(staging: Path, manifest: dict, sanitize_report: dict,
                 out: Path, events_count: int, callback_count: int) -> Path:
    index = {
        "bundle_spec_version": "1.0",
        "incident_id": manifest["incident_id"],
        "generated_at": manifest["generated_at"],
        "files": [
            {"name": "manifest.yaml",  "purpose": "env + version snapshot"},
            {"name": "events.jsonl",   "purpose": f"filtered astream_events(v2), {events_count} events"},
            {"name": "callbacks.txt",  "purpose": f"DebugCallbackHandler records, {callback_count} entries"},
            {"name": "langsmith.url",  "purpose": "trace URL (shared) or (none)"},
            {"name": "notes.txt",      "purpose": "free-form engineer notes, sanitized"},
        ],
        "sanitization": sanitize_report,
    }
    (staging / "MANIFEST.yaml").write_text(yaml.safe_dump(index, sort_keys=False))
    with tarfile.open(out, "w:gz") as tar:
        tar.add(staging, arcname=out.stem)
    return out

bundle = write_bundle(
    staging=Path("/tmp/bundle-INC-2026-0421-A"),
    manifest=m, sanitize_report=report,
    out=Path("/tmp/bundle-INC-2026-0421-A.tar.gz"),
    events_count=n_events, callback_count=len(debug.records),
)