import json from typing import Any, Dict, List, Optional def _short(x: Any, n: int = 160) -> str: try: s = json.dumps(x, ensure_ascii=False) except Exception: s = str(x) if len(s) > n: return s[:n] + "…" return s def _fmt_reward(x: Optional[float]) -> str: if x is None: return "n/a" # stable, readable return f"{x:.6g}" def render_report_markdown(diff: Dict[str, Any]) -> str: s = diff.get("summary", {}) counts = diff.get("class_counts", {}) first = s.get("first_divergence_index") # punchy one-liner line_bits: List[str] = [] if first is None: line_bits.append("Timelines identical") else: line_bits.append(f"Identical until index {first}") line_bits.append(f"{s.get('diff_event_count', 0)} events differ") if s.get("missing_event_count", 0): line_bits.append(f"{s.get('missing_event_count')} missing") if s.get("final_reward_delta") is not None: line_bits.append(f"Final reward delta: {s.get('final_reward_delta'):.6g}") one_liner = " · ".join(line_bits) lines: List[str] = [] lines.append("# DRP Differential Report") lines.append("") lines.append(f"**{one_liner}**") lines.append("") lines.append("## Run identity") lines.append(f"- Run A: `{s.get('run_a')}`") lines.append(f"- Run B: `{s.get('run_b')}`") lines.append(f"- Framework A/B: `{s.get('framework_a')}` / `{s.get('framework_b')}`") lines.append(f"- Model A/B: `{s.get('model_a')}` / `{s.get('model_b')}`") lines.append(f"- Events A/B: `{s.get('events_a')}` / `{s.get('events_b')}`") lines.append("") if s.get("run_link_a") or s.get("run_link_b"): lines.append("## Viewer links (if provided)") if s.get("run_link_a"): lines.append(f"- Run A viewer: {s.get('run_link_a')}") if s.get("run_link_b"): lines.append(f"- Run B viewer: {s.get('run_link_b')}") lines.append("") lines.append("## Summary stats") lines.append(f"- First divergence index: `{s.get('first_divergence_index')}`") lines.append(f"- Diff events: `{s.get('diff_event_count')}`") lines.append(f"- Missing events: `{s.get('missing_event_count')}`") lines.append(f"- Final reward A/B/Δ: `{_fmt_reward(s.get('final_reward_a'))}` / `{_fmt_reward(s.get('final_reward_b'))}` / `{_fmt_reward(s.get('final_reward_delta'))}`") lines.append("") lines.append("## Divergence classes") if counts: for k in sorted(counts.keys()): lines.append(f"- **{k}**: {counts[k]}") else: lines.append("- (no diffs)") lines.append("") # First divergence detail lines.append("## First divergence detail") if first is None: lines.append("- No divergence found.") return "\n".join(lines) first_item = None for item in diff.get("differences", []): if item.get("i") == first: first_item = item break if not first_item: # likely a length mismatch only lines.append("- Divergence is a length mismatch or occurs where one side is missing.") return "\n".join(lines) lines.append(f"- Index: `{first_item.get('i')}`") lines.append(f"- Class: `{first_item.get('class')}`") lines.append(f"- Kind A/B: `{first_item.get('kind_a')}` / `{first_item.get('kind_b')}`") lines.append(f"- Step A/B: `{first_item.get('step_a')}` / `{first_item.get('step_b')}`") if first_item.get("link_a") or first_item.get("link_b"): lines.append("") lines.append("### Jump to replay (if provided)") if first_item.get("link_a"): lines.append(f"- Open A @ i={first}: {first_item.get('link_a')}") if first_item.get("link_b"): lines.append(f"- Open B @ i={first}: {first_item.get('link_b')}") lines.append("") lines.append("### Key diffs (first 25)") for d in (first_item.get("diffs") or [])[:25]: lines.append(f"- `{d.get('path')}` ({d.get('kind')}): A={_short(d.get('a'))} | B={_short(d.get('b'))}") if "text_unified_diff" in first_item: lines.append("") lines.append("### Text unified diff (truncated)") lines.append("```diff") lines.append(first_item["text_unified_diff"][:8000]) lines.append("```") return "\n".join(lines)