diff --git a/kora_cli/audit/reasoning_xref.py b/kora_cli/audit/reasoning_xref.py
new file mode 100644
index 000000000000..3c1b4d965b13
--- /dev/null
+++ b/kora_cli/audit/reasoning_xref.py
@@ -0,0 +1,413 @@
+"""Reasoning panel ↔ slack_dm cross-reference — KR-REASONING-PANEL-MODEL-XREF.
+
+The reasoning panel's /api/reasoning/recent endpoint reads
+``kora_audit_log.jsonl`` rows where ``seam=reasoning.tool_called``,
+groups by ``caller_session_id``, and projects to ReasoningCall
+shape. PR #141 left ``model_used`` / ``cost_rung_at_call`` /
+``input_tokens`` / ``output_tokens`` / ``response_text_truncated_200``
+as null because those fields live in
+``slack_dm_log.jsonl`` outbound entries, not in audit.
+
+This module cross-references the two log files to populate those
+fields. Graceful-degradation: when the xref fails (slack_dm entry
+missing OR stale), the ReasoningCall row still renders with null
+fields — same as the pre-xref behavior from #141.
+
+K-DG verified against actual writer code (per the
+``feedback_no_pm_memory_assertions_grep_yourself`` rule):
+
+  * Audit ``caller_session_id`` shape per
+    ``kora_cli/reasoning/anthropic_engine.py:844-876``
+    (``_derive_caller_session_id``):
+        - slack_dm → ``"{channel_id}:{event_ts}"``
+        - email   → ``"email:{message_id}"``
+        - mcp     → ``"mcp:{actor_kind}:{tool_name}"``
+        - other   → ``"unknown"``
+
+  * slack_dm outbound writer at
+    ``kora_cli/handlers/slack_dm_handler.py:753-833`` does NOT
+    include ``caller_session_id`` in the JSONL entry (spec said
+    "verify; CC#3 may have added in #131" — it did NOT). Outbound
+    entries have ``channel_id`` + ``thread_ts`` + ``sent_at`` +
+    reasoning meta (``model_used``, tokens, ``reasoning_duration_ms``,
+    ``reasoning_error``).
+
+  * Correlation algorithm (only workable path given the above):
+        1. Parse audit caller_session_id as ``"{channel_id}:{event_ts}"``
+           for slack_dm-sourced groups. Other sources (email/mcp/
+           unknown) currently have no outbound JSONL to xref —
+           rows render with null fields per graceful degradation.
+        2. Find outbound entries where ``channel_id`` matches AND
+           (``thread_ts == event_ts`` OR ``sent_at`` within ±60s
+           of the audit group's latest ``emitted_at``).
+        3. Pick the closest-time match.
+
+  * Email reasoning replies don't write to slack_dm_log — that's
+    the KR-REASONING-PANEL-EMAIL-XREF follow-on per spec §3.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Tuple
+
+from kora_cli.audit.jsonl_reader import read_audit_entries
+
+logger = logging.getLogger(__name__)
+
+
+# Slack-DM JSONL path mirrors slack_dm_handler.py:74-82 — env override
+# (HERMES_HOME / KORA_HOME via kora_constants) → ``<KORA_HOME>/slack_dm_log.jsonl``.
+_SLACK_DM_LOG_FILENAME = "slack_dm_log.jsonl"
+
+# Match window for the fallback timestamp join. ±60s comfortably
+# covers reasoning latency (typical 1-5s, p99 ~30s) plus clock drift.
+_TIMESTAMP_WINDOW = timedelta(seconds=60)
+
+# 200-char body cap for response_text_truncated_200 — matches the
+# field name's semantic contract. The slack_dm outbound stores the
+# full text (no truncation at write time); this projection enforces
+# the cap at the panel edge.
+_RESPONSE_TEXT_CAP = 200
+
+
+def _slack_dm_log_path() -> Path:
+    """Re-resolve on every call so monkeypatch in tests works."""
+    from kora_constants import get_kora_home
+
+    return get_kora_home() / _SLACK_DM_LOG_FILENAME
+
+
+def _parse_slack_dm_session_id(
+    session_id: Optional[str],
+) -> Optional[Tuple[str, str]]:
+    """Parse audit ``caller_session_id`` as ``"{channel_id}:{event_ts}"``.
+
+    Returns ``(channel_id, event_ts)`` for slack_dm-shaped session
+    ids; ``None`` for other shapes (email/mcp/unknown). The
+    discrimination is: 2 segments separated by ``:``, first segment
+    starts with ``D`` or ``C`` (Slack channel prefix), neither
+    segment starts with a known prefix like ``email:`` or ``mcp:``.
+    """
+    if not session_id:
+        return None
+    # Reject other-source session ids by their prefixes.
+    if session_id.startswith(("email:", "mcp:", "unknown")):
+        return None
+    if session_id == "unknown":
+        return None
+    # The slack_dm-fallback shape is ``slack_dm:{channel_id or unknown}``
+    # (when event_ts is missing). Treat as unparseable.
+    if session_id.startswith("slack_dm:"):
+        return None
+    # Slack-DM happy path: exactly one ":" separator
+    parts = session_id.split(":", 1)
+    if len(parts) != 2:
+        return None
+    channel_id, event_ts = parts[0], parts[1]
+    if not channel_id or not event_ts:
+        return None
+    return channel_id, event_ts
+
+
+def _load_outbound_entries(limit: int = 500) -> List[Dict[str, Any]]:
+    """Read recent outbound slack_dm JSONL entries (those with
+    ``sent_at`` + ``send_status``). Tolerates missing file +
+    malformed lines (log + skip) per the same discipline as
+    ``audit/jsonl_reader.py``.
+
+    Returns the LAST ``limit`` outbound entries (file-position-based,
+    NOT timestamp-sorted) since the writer appends; the matcher
+    sorts by ``sent_at`` later if needed.
+    """
+    log_path = _slack_dm_log_path()
+    if not log_path.is_file():
+        return []
+
+    outbound: List[Dict[str, Any]] = []
+    try:
+        with log_path.open("r", encoding="utf-8") as f:
+            for lineno, raw_line in enumerate(f, start=1):
+                raw_line = raw_line.strip()
+                if not raw_line:
+                    continue
+                try:
+                    entry = json.loads(raw_line)
+                except json.JSONDecodeError as exc:
+                    logger.warning(
+                        "[kora.reasoning_xref] slack_dm line %d "
+                        "malformed JSON, skipped: %r",
+                        lineno,
+                        exc,
+                    )
+                    continue
+                if not isinstance(entry, dict):
+                    continue
+                # Outbound entries have ``sent_at`` + ``send_status``;
+                # inbound entries have ``received_at`` + ``handled_status``.
+                if "sent_at" in entry and "send_status" in entry:
+                    outbound.append(entry)
+    except OSError as exc:
+        logger.warning(
+            "[kora.reasoning_xref] failed to read %s: %r", log_path, exc
+        )
+        return []
+
+    return outbound[-limit:] if limit > 0 else outbound
+
+
+def _parse_iso(ts: str) -> Optional[datetime]:
+    """Parse the writer's ``_now_iso()`` Z-suffixed shape."""
+    if not ts:
+        return None
+    try:
+        dt = datetime.fromisoformat(ts.replace("Z", "+00:00"))
+    except (ValueError, AttributeError):
+        return None
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    return dt
+
+
+def _find_xref_for_slack_dm_group(
+    channel_id: str,
+    event_ts: str,
+    group_latest_emitted_at: datetime,
+    outbound_entries: List[Dict[str, Any]],
+) -> Optional[Dict[str, Any]]:
+    """Pick the outbound entry that best matches a reasoning group.
+
+    Algorithm:
+      1. Filter outbound to entries with matching ``channel_id``.
+      2. Within that, prefer entries where ``thread_ts == event_ts``
+         (the natural threading match — Kora's reply threads under
+         the user's inbound message).
+      3. Fallback: pick the entry with ``sent_at`` closest to the
+         group's latest ``emitted_at``, within ``±_TIMESTAMP_WINDOW``.
+      4. Return None when no candidate matches — caller renders the
+         row with null fields (graceful degradation).
+    """
+    same_channel = [
+        e for e in outbound_entries if e.get("channel_id") == channel_id
+    ]
+    if not same_channel:
+        return None
+
+    # Preferred: thread_ts == event_ts match. Within those, pick the
+    # closest by sent_at (multiple replies can thread under the same
+    # inbound; pick the one closest to the reasoning finish time).
+    thread_matches = [
+        e for e in same_channel if e.get("thread_ts") == event_ts
+    ]
+    if thread_matches:
+        return _pick_closest_by_sent_at(thread_matches, group_latest_emitted_at)
+
+    # Fallback: closest sent_at within the time window.
+    candidate = _pick_closest_by_sent_at(
+        same_channel, group_latest_emitted_at
+    )
+    if candidate is None:
+        return None
+    candidate_ts = _parse_iso(candidate.get("sent_at", ""))
+    if candidate_ts is None:
+        return None
+    if abs(candidate_ts - group_latest_emitted_at) > _TIMESTAMP_WINDOW:
+        return None
+    return candidate
+
+
+def _pick_closest_by_sent_at(
+    entries: List[Dict[str, Any]],
+    target: datetime,
+) -> Optional[Dict[str, Any]]:
+    """Pick the entry whose ``sent_at`` is closest to ``target``.
+    Entries with unparseable ``sent_at`` are skipped."""
+    best: Optional[Dict[str, Any]] = None
+    best_delta: Optional[timedelta] = None
+    for e in entries:
+        ts = _parse_iso(e.get("sent_at", ""))
+        if ts is None:
+            continue
+        delta = abs(ts - target)
+        if best_delta is None or delta < best_delta:
+            best = e
+            best_delta = delta
+    return best
+
+
+def _derive_cost_rung(
+    model_used: Optional[str],
+    reasoning_error: Optional[str],
+) -> str:
+    """Derive lowercase CostRung.value from model name + error code.
+
+    Per the cost-ladder model selection in
+    ``kora_cli/reasoning/anthropic_engine.py`` (model → rung
+    mapping) AND the agent/cost_state_holder.py:114-117 lowercase
+    Enum.value wire format (preserves the PR #132 + #141 K-DG pin).
+
+    Mapping:
+      * reasoning_error == "cost_ladder_halted" → "hard_stop_100"
+      * model contains "opus"   → "normal"
+      * model contains "sonnet" → "warn_75"
+      * model contains "haiku"  → "downshift_90"
+      * unmapped / missing      → "unknown"
+
+    Substring-match (instead of exact-equals) so future minor model
+    revs (claude-opus-4-7 → claude-opus-4-8 etc) keep mapping
+    correctly without code changes.
+    """
+    if reasoning_error == "cost_ladder_halted":
+        return "hard_stop_100"
+    if not model_used:
+        return "unknown"
+    lower = model_used.lower()
+    if "opus" in lower:
+        return "normal"
+    if "sonnet" in lower:
+        return "warn_75"
+    if "haiku" in lower:
+        return "downshift_90"
+    return "unknown"
+
+
+def _truncate_response_text(text: Optional[str]) -> Optional[str]:
+    """200-char cap matching the field name's semantic contract.
+    None passes through (no response captured)."""
+    if text is None:
+        return None
+    s = str(text)
+    if len(s) <= _RESPONSE_TEXT_CAP:
+        return s
+    return s[:_RESPONSE_TEXT_CAP] + "…"
+
+
+def _aggregate_status(
+    statuses: List[str],
+) -> Tuple[str, Optional[str]]:
+    """Mirror the aggregation in /api/reasoning/recent's projection:
+    all-ok → ok; any not_allowed → halted+capability_denied; any
+    execution_error → failed+handler_error; other non-ok → failed."""
+    if all(s == "ok" for s in statuses):
+        return "ok", None
+    if any(s == "not_allowed" for s in statuses):
+        return "halted", "capability_denied"
+    if any(s == "execution_error" for s in statuses):
+        return "failed", "handler_error"
+    return "failed", next((s for s in statuses if s != "ok"), "unknown")
+
+
+def load_reasoning_calls_with_xref(
+    *,
+    limit: int = 200,
+) -> Tuple[List[Dict[str, Any]], int]:
+    """Load + group reasoning audit rows + cross-reference slack_dm.
+
+    Returns ``(projected_calls, total_recent_24h_raw_rows)``.
+
+    The ``projected_calls`` list is newest-first (by group started_at)
+    and capped at ``limit`` groups. ``total_recent_24h_raw_rows`` is
+    the count of INDIVIDUAL audit rows in the 24h window (NOT
+    groups) — matches the aggregate-counts-from-individual-rows
+    pattern from PR #141 so the dashboard headline reflects activity
+    volume, not pagination choice.
+
+    SECURITY (carry-forward from #141 + #132 + xref-specific):
+      * model_used + tokens are inherently safe metadata.
+      * cost_rung_at_call is lowercase CostRung.value (PR #132 pin).
+      * response_text_truncated_200 IS Joshua-content (intentional
+        carve-out from PII regex sweep — same shape as #141's
+        message_id carve-out + slack_dm panel's text carve-out).
+        Plain-text rendering already enforced FE-side via
+        dangerouslySetInnerHTML ban from PR #132.
+    """
+    from kora_cli.audit.jsonl_sink import AuditEntry  # noqa: F401 — for type-doc
+
+    audit_rows = read_audit_entries(seam="reasoning.tool_called")
+    outbound_entries = _load_outbound_entries(limit=500)
+
+    capped_limit = max(1, min(limit, 200))
+    now = datetime.now(timezone.utc)
+    cutoff_24h = now - timedelta(hours=24)
+
+    # Group by caller_session_id (same logic as PR #141).
+    groups: Dict[str, List[Any]] = {}
+    for e in audit_rows:
+        key = e.caller_session_id or f"orphan-{id(e)}"
+        groups.setdefault(key, []).append(e)
+
+    projected: List[Dict[str, Any]] = []
+    for sid, group_rows in groups.items():
+        rows_sorted = sorted(group_rows, key=lambda e: e.emitted_at)
+        first = rows_sorted[0]
+        last = rows_sorted[-1]
+        tool_names = [str(e.details.get("tool_name", "")) for e in rows_sorted]
+        total_duration_ms = sum(
+            int(e.details.get("tool_duration_ms") or 0) for e in rows_sorted
+        )
+        statuses = [
+            str(e.details.get("tool_status", "ok")) for e in rows_sorted
+        ]
+        agg_status, error_code = _aggregate_status(statuses)
+        triggered_by = (
+            first.details.get("triggered_by") or first.source or "slack_dm"
+        )
+
+        # XREF: try to find a matching slack_dm outbound entry.
+        parsed = _parse_slack_dm_session_id(first.caller_session_id)
+        xref: Optional[Dict[str, Any]] = None
+        if parsed is not None:
+            channel_id, event_ts = parsed
+            xref = _find_xref_for_slack_dm_group(
+                channel_id=channel_id,
+                event_ts=event_ts,
+                group_latest_emitted_at=last.emitted_at,
+                outbound_entries=outbound_entries,
+            )
+
+        if xref is not None:
+            model_used = xref.get("model_used")
+            input_tokens = int(xref.get("input_tokens") or 0)
+            output_tokens = int(xref.get("output_tokens") or 0)
+            reasoning_error_x = xref.get("reasoning_error")
+            response_text = _truncate_response_text(xref.get("text"))
+            # If the xref surfaced a reasoning_error that supersedes
+            # the audit-derived status (e.g. cost_ladder_halted with
+            # no tool calls at all), reflect that in error_code.
+            if reasoning_error_x and reasoning_error_x != error_code:
+                if reasoning_error_x == "cost_ladder_halted":
+                    agg_status = "halted"
+                    error_code = reasoning_error_x
+            cost_rung = _derive_cost_rung(model_used, reasoning_error_x)
+        else:
+            # Graceful degradation: no xref → null fields, same as
+            # the pre-xref behavior from PR #141.
+            model_used = None
+            input_tokens = 0
+            output_tokens = 0
+            response_text = None
+            cost_rung = _derive_cost_rung(None, None)  # "unknown"
+
+        projected.append({
+            "id": f"audit-session-{sid or first.emitted_at.isoformat()}",
+            "triggered_by": str(triggered_by),
+            "started_at": first.emitted_at.strftime("%Y-%m-%dT%H:%M:%SZ"),
+            "duration_ms": total_duration_ms,
+            "model_used": model_used,
+            "cost_rung_at_call": cost_rung,
+            "input_tokens": input_tokens,
+            "output_tokens": output_tokens,
+            "status": agg_status,
+            "error_code": error_code,
+            "response_text_truncated_200": response_text,
+            "tools_used": tool_names,
+        })
+
+    projected.sort(key=lambda r: r["started_at"], reverse=True)
+
+    raw_in_window = sum(1 for e in audit_rows if e.emitted_at >= cutoff_24h)
+    return projected[:capped_limit], raw_in_window
diff --git a/kora_cli/web_server.py b/kora_cli/web_server.py
index 229d4559b727..42745a9aac21 100644
--- a/kora_cli/web_server.py
+++ b/kora_cli/web_server.py
@@ -5539,121 +5539,61 @@ async def list_recent_email():
 #
 # Status derivation: ok if every grouped tool has tool_status==ok;
 # otherwise the dominant non-ok status (capability_denied →
-# halted; execution_error → handler_error). cost_rung_at_call is
-# "unknown" (lowercase CostRung.value literal that satisfies the
-# FE's ReasoningCostRung union without surfacing a null we can't
-# actually validate from audit).
-
-
-def _project_reasoning_group(
-    session_id: str,
-    rows: list,
-) -> Dict[str, Any]:
-    """Collapse N audit rows sharing caller_session_id → 1 ReasoningCall.
-
-    Per spec §2 Flip 2: extends the FE payload with ``tools_used``
-    (list of tool names). Existing TS ReasoningCall doesn't have
-    this field — extras pass through unused; a follow-on FE bucket
-    can render it. Doesn't break the existing FE since JS object
-    structural-typing tolerates extras.
-    """
-    # rows are AuditEntry; newest first within the group.
-    rows = sorted(rows, key=lambda e: e.emitted_at)
-    first = rows[0]
-    last = rows[-1]
-    tool_names = [str(e.details.get("tool_name", "")) for e in rows]
-    total_duration_ms = sum(
-        int(e.details.get("tool_duration_ms") or 0) for e in rows
-    )
-    statuses = [e.details.get("tool_status", "ok") for e in rows]
-
-    if all(s == "ok" for s in statuses):
-        agg_status = "ok"
-        error_code = None
-    elif any(s == "not_allowed" for s in statuses):
-        agg_status = "halted"
-        error_code = "capability_denied"
-    elif any(s == "execution_error" for s in statuses):
-        agg_status = "failed"
-        error_code = "handler_error"
-    else:
-        # Unknown non-ok status — surface as failed without a specific
-        # code so the FE renders the destructive tone.
-        agg_status = "failed"
-        # Find first non-ok status as the dominant error indicator.
-        error_code = next((s for s in statuses if s != "ok"), "unknown")
-
-    triggered_by = first.details.get("triggered_by") or first.source or "slack_dm"
-
-    return {
-        "id": f"audit-session-{session_id or first.emitted_at.isoformat()}",
-        "triggered_by": str(triggered_by),
-        "started_at": first.emitted_at.strftime("%Y-%m-%dT%H:%M:%SZ"),
-        "duration_ms": total_duration_ms,
-        # Not in audit; cross-ref to slack_dm_log.jsonl happens in
-        # KR-REASONING-PANEL-MODEL-XREF follow-on bucket.
-        "model_used": None,
-        # CostRung.value lowercase "unknown" satisfies the FE's
-        # ReasoningCostRung union (engine.py:47-49) without
-        # claiming a rung we can't actually read from audit.
-        "cost_rung_at_call": "unknown",
-        "input_tokens": 0,
-        "output_tokens": 0,
-        "status": agg_status,
-        "error_code": error_code,
-        "response_text_truncated_200": None,
-        # FE extension — extra field; existing TS ignores it.
-        "tools_used": tool_names,
-    }
+# halted; execution_error → handler_error).
+#
+# KR-REASONING-PANEL-MODEL-XREF (this file's update): model_used /
+# input_tokens / output_tokens / cost_rung_at_call /
+# response_text_truncated_200 are populated by cross-referencing
+# kora_audit_log.jsonl reasoning rows with slack_dm_log.jsonl
+# outbound entries via the channel_id + thread_ts/timestamp join
+# in ``kora_cli/audit/reasoning_xref.py``. Graceful degradation:
+# when no slack_dm match is found, those fields remain null (same
+# behavior as the pre-xref projection from PR #141).
 
 
 @app.get("/api/reasoning/recent")
 async def list_recent_reasoning(limit: int = 50):
     """Return recent Kora ReasoningEngine calls for the operator lens.
 
-    Reads ``${KORA_HOME}/kora_audit_log.jsonl``, filters to
-    ``seam=reasoning.tool_called``, and groups consecutive tool
-    calls by ``caller_session_id`` so a multi-tool reasoning
-    iteration collapses into a single ReasoningCall row with
-    ``tools_used: [...]``. Newest-first by the session's first
-    emitted_at.
-
-    Limitations until the KR-REASONING-PANEL-MODEL-XREF follow-on:
-      * model_used / tokens / response_text_truncated_200 are null
-        (those fields live in slack_dm_log.jsonl outbound entries,
-        not in the audit log).
-      * cost_rung_at_call is "unknown" (same reason — audit doesn't
-        capture the rung at call time).
+    Reads ``${KORA_HOME}/kora_audit_log.jsonl`` filtered to
+    ``seam=reasoning.tool_called``, groups consecutive tool calls
+    by ``caller_session_id``, and cross-references each group with
+    ``slack_dm_log.jsonl`` outbound entries (via
+    ``kora_cli/audit/reasoning_xref.py``) to populate
+    ``model_used`` / ``input_tokens`` / ``output_tokens`` /
+    ``cost_rung_at_call`` / ``response_text_truncated_200``.
+
+    Graceful degradation: when the xref lookup fails for a group
+    (slack_dm entry missing or outside the ±60s correlation
+    window), those fields render as null. Same shape as the
+    pre-xref behavior from PR #141, so the FE handles both.
+
+    Aggregates (total_recent_24h, by_status_24h, by_model_24h,
+    tokens_total_24h) operate on INDIVIDUAL audit rows + xref'd
+    outbound entries — NOT groups — so headline counts reflect
+    activity volume.
     """
     from datetime import datetime, timedelta, timezone
     from kora_cli.audit.jsonl_reader import read_audit_entries
+    from kora_cli.audit.reasoning_xref import (
+        load_reasoning_calls_with_xref,
+    )
 
-    capped_limit = max(1, min(limit, 200))
     now = datetime.now(timezone.utc)
     cutoff_24h = now - timedelta(hours=24)
 
-    all_rows = read_audit_entries(seam="reasoning.tool_called")
-
-    # Group by caller_session_id; rows without a session_id each get
-    # their own group (defensive — shouldn't happen since the writer
-    # always passes one).
-    groups: Dict[str, list] = {}
-    for e in all_rows:
-        key = e.caller_session_id or f"orphan-{id(e)}"
-        groups.setdefault(key, []).append(e)
-
-    # Project + sort newest-first by the LATEST event in each group.
-    projected = [
-        _project_reasoning_group(sid, group_rows)
-        for sid, group_rows in groups.items()
-    ]
-    projected.sort(key=lambda r: r["started_at"], reverse=True)
+    projected, raw_in_window_count = load_reasoning_calls_with_xref(
+        limit=limit,
+    )
 
-    # 24h-window aggregates over individual audit rows (not groups)
-    # so the headline counts reflect raw activity volume.
-    in_window = [e for e in all_rows if e.emitted_at >= cutoff_24h]
+    # Per-status aggregate (individual rows). Re-read audit since
+    # the xref helper returns groups; we count INDIVIDUAL rows
+    # within the 24h window for the headline (per PR #141 rationale).
+    audit_rows = read_audit_entries(seam="reasoning.tool_called")
     by_status: Dict[str, int] = {"ok": 0, "failed": 0, "halted": 0}
-    for e in in_window:
+    for e in audit_rows:
+        if e.emitted_at < cutoff_24h:
+            continue
         st = e.details.get("tool_status", "ok")
         if st == "ok":
             by_status["ok"] += 1
@@ -5661,17 +5601,37 @@ async def list_recent_reasoning(limit: int = 50):
             by_status["halted"] += 1
         else:
             by_status["failed"] += 1
-    # Model + token aggregates are derived from the slack_dm log in
-    # the follow-on. Until then, surface the structure with zeros so
-    # the FE dashboard tile renders without conditional NaN handling.
+
+    # Model + token aggregates from the xref'd groups within window.
+    # Falls back to empty / zero if no xref enrichment happened.
     by_model: Dict[str, int] = {}
     tokens_total = {"input": 0, "output": 0}
+    for call in projected:
+        # Per-group emitted_at corresponds to started_at; only count
+        # those within the 24h window.
+        ts_str = call.get("started_at", "")
+        try:
+            ts = datetime.fromisoformat(ts_str.replace("Z", "+00:00"))
+            if ts.tzinfo is None:
+                ts = ts.replace(tzinfo=timezone.utc)
+        except (ValueError, AttributeError):
+            continue
+        if ts < cutoff_24h:
+            continue
+        model = call.get("model_used")
+        if model:
+            by_model[model] = by_model.get(model, 0) + 1
+        elif call.get("status") == "halted":
+            by_model["halted_no_model"] = by_model.get("halted_no_model", 0) + 1
+        tokens_total["input"] += int(call.get("input_tokens") or 0)
+        tokens_total["output"] += int(call.get("output_tokens") or 0)
 
     return {
-        "calls": projected[:capped_limit],
+        # projected already capped by the helper per its limit arg.
+        "calls": projected,
         "stub": False,
         "generated_at": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
-        "total_recent_24h": len(in_window),
+        "total_recent_24h": raw_in_window_count,
         "by_model_24h": by_model,
         "by_status_24h": by_status,
         "tokens_total_24h": tokens_total,
diff --git a/tests/kora_cli/audit/test_reasoning_xref.py b/tests/kora_cli/audit/test_reasoning_xref.py
new file mode 100644
index 000000000000..1189d2265928
--- /dev/null
+++ b/tests/kora_cli/audit/test_reasoning_xref.py
@@ -0,0 +1,532 @@
+"""Tests for kora_cli.audit.reasoning_xref.
+
+Cross-references kora_audit_log.jsonl reasoning rows with
+slack_dm_log.jsonl outbound entries to populate model_used /
+tokens / cost_rung_at_call / response_text_truncated_200 on
+reasoning panel rows.
+
+Scenarios:
+  1. Audit-only path (no slack_dm log) → ReasoningCall with null
+     model fields (graceful degradation; same shape as PR #141
+     pre-xref behavior)
+  2. Successful xref via channel_id + thread_ts match → model
+     fields populated
+  3. Successful xref via timestamp-window fallback → model fields
+     populated even when thread_ts doesn't match
+  4. Outbound entry outside ±60s window → no xref, graceful
+     degradation
+  5. Cost-rung derivation per model (opus / sonnet / haiku /
+     unknown / hard-stop)
+  6. caller_session_id shape for other sources (email/mcp/unknown)
+     → no xref attempt; graceful degradation
+  7. response_text_truncated_200 capped at 200 chars
+  8. Multiple groups within window — each picks its own match
+  9. SECURITY: walk-payload sweep (with response_text carve-out)
+ 10. Malformed slack_dm log line tolerated
+ 11. Empty slack_dm log → all rows graceful-degrade
+ 12. cost_ladder_halted xref status supersedes audit status
+"""
+
+import json
+import re
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from typing import Any, Dict, List
+
+import pytest
+
+from kora_cli.audit.jsonl_sink import AUDIT_LOG_FILENAME
+from kora_cli.audit.reasoning_xref import (
+    _derive_cost_rung,
+    _parse_slack_dm_session_id,
+    _truncate_response_text,
+    load_reasoning_calls_with_xref,
+)
+
+
+_ANTHROPIC_KEY_SHAPE = re.compile(r"\bsk-ant-[A-Za-z0-9_-]{16,}\b")
+_HEX_SECRET_SHAPE = re.compile(r"\b[0-9a-fA-F]{32,}\b")
+
+_DM_CHANNEL = "D0123456789ABCDEF"
+_EVENT_TS = "1779380123.456"
+
+
+@pytest.fixture
+def env(tmp_path, monkeypatch):
+    """Per #137/#141 fixture lesson — monkeypatch get_kora_home in
+    all 3 module namespaces. The reasoning_xref helper uses a local
+    kora_constants import in _slack_dm_log_path() so the patch in
+    kora_constants is sufficient for the xref reader; the audit
+    reader has the same local-import pattern."""
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    monkeypatch.setenv("KORA_HOME", str(tmp_path))
+    monkeypatch.setattr("kora_constants.get_kora_home", lambda: tmp_path)
+    monkeypatch.setattr("kora_cli.config.get_kora_home", lambda: tmp_path)
+    monkeypatch.setattr("kora_cli.web_server.get_kora_home", lambda: tmp_path)
+    monkeypatch.setattr(
+        "kora_cli.config.get_config_path",
+        lambda: tmp_path / "config.yaml",
+    )
+    monkeypatch.setattr(
+        "kora_cli.config.get_env_path", lambda: tmp_path / ".env"
+    )
+    return tmp_path
+
+
+def _iso(minutes_ago: int = 5) -> str:
+    ts = datetime.now(timezone.utc) - timedelta(minutes=minutes_ago)
+    return ts.strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def _audit_entry(
+    *,
+    seam: str = "reasoning.tool_called",
+    minutes_ago: int = 5,
+    tool_name: str = "get_state",
+    tool_status: str = "ok",
+    tool_duration_ms: int = 100,
+    caller_session_id: str = f"{_DM_CHANNEL}:{_EVENT_TS}",
+    triggered_by: str = "slack_dm",
+    source: str = "reasoning",
+) -> Dict[str, Any]:
+    ts = datetime.now(timezone.utc) - timedelta(minutes=minutes_ago)
+    return {
+        "emitted_at": ts.isoformat(),
+        "seam": seam,
+        "details": {
+            "tool_name": tool_name,
+            "triggered_by": triggered_by,
+            "tool_duration_ms": tool_duration_ms,
+            "tool_status": tool_status,
+        },
+        "caller_session_id": caller_session_id,
+        "source": source,
+    }
+
+
+def _outbound_entry(
+    *,
+    minutes_ago: int = 5,
+    channel_id: str = _DM_CHANNEL,
+    thread_ts: str = _EVENT_TS,
+    text: str = "ok",
+    model_used: str | None = "claude-opus-4-7",
+    input_tokens: int | None = 842,
+    output_tokens: int | None = 127,
+    reasoning_duration_ms: int | None = 1247,
+    reasoning_error: str | None = None,
+    send_status: str = "ok",
+) -> Dict[str, Any]:
+    """Outbound JSONL shape per slack_dm_handler.py:811-833."""
+    entry: Dict[str, Any] = {
+        "sent_at": _iso(minutes_ago),
+        "channel_id": channel_id,
+        "thread_ts": thread_ts,
+        "text": text,
+        "slack_message_ts": f"{minutes_ago}.0",
+        "send_status": send_status,
+    }
+    if model_used is not None:
+        entry["model_used"] = model_used
+    if input_tokens is not None:
+        entry["input_tokens"] = input_tokens
+    if output_tokens is not None:
+        entry["output_tokens"] = output_tokens
+    if reasoning_duration_ms is not None:
+        entry["reasoning_duration_ms"] = reasoning_duration_ms
+    if reasoning_error is not None:
+        entry["reasoning_error"] = reasoning_error
+    return entry
+
+
+def write_audit(env_path: Path, entries: List[Dict[str, Any]]) -> Path:
+    log_path = env_path / AUDIT_LOG_FILENAME
+    with log_path.open("w", encoding="utf-8") as f:
+        for entry in entries:
+            f.write(json.dumps(entry) + "\n")
+    return log_path
+
+
+def write_slack_dm(env_path: Path, entries: List[Dict[str, Any]]) -> Path:
+    log_path = env_path / "slack_dm_log.jsonl"
+    with log_path.open("w", encoding="utf-8") as f:
+        for entry in entries:
+            f.write(json.dumps(entry) + "\n")
+    return log_path
+
+
+# ---- _parse_slack_dm_session_id ----------------------------------
+
+
+def test_parse_slack_dm_session_id_happy_path():
+    result = _parse_slack_dm_session_id(f"{_DM_CHANNEL}:{_EVENT_TS}")
+    assert result == (_DM_CHANNEL, _EVENT_TS)
+
+
+def test_parse_session_id_other_sources_return_none():
+    """email / mcp / unknown / fallback shapes don't parse — they
+    have no slack_dm correlation target."""
+    assert _parse_slack_dm_session_id("email:msg-id-123") is None
+    assert _parse_slack_dm_session_id("mcp:claude_pm:get_state") is None
+    assert _parse_slack_dm_session_id("unknown") is None
+    assert _parse_slack_dm_session_id("slack_dm:unknown") is None
+    assert _parse_slack_dm_session_id("") is None
+    assert _parse_slack_dm_session_id(None) is None
+
+
+# ---- _derive_cost_rung -------------------------------------------
+
+
+def test_derive_cost_rung_opus_normal():
+    assert _derive_cost_rung("claude-opus-4-7", None) == "normal"
+    assert _derive_cost_rung("claude-opus-5-0", None) == "normal", (
+        "Substring match so future model revs keep working"
+    )
+
+
+def test_derive_cost_rung_sonnet_warn_75():
+    assert _derive_cost_rung("claude-sonnet-4-6", None) == "warn_75"
+
+
+def test_derive_cost_rung_haiku_downshift_90():
+    assert _derive_cost_rung("claude-haiku-4-5-20251001", None) == "downshift_90"
+
+
+def test_derive_cost_rung_cost_ladder_halted_overrides_model():
+    """When reasoning_error is cost_ladder_halted, the rung is
+    hard_stop_100 regardless of model_used (engine refused before
+    making the SDK call, but the writer may still have a stale
+    model_used from a previous turn)."""
+    assert _derive_cost_rung("claude-opus-4-7", "cost_ladder_halted") == "hard_stop_100"
+    assert _derive_cost_rung(None, "cost_ladder_halted") == "hard_stop_100"
+
+
+def test_derive_cost_rung_unmapped_model_unknown():
+    assert _derive_cost_rung("some-future-model-99", None) == "unknown"
+    assert _derive_cost_rung(None, None) == "unknown"
+    assert _derive_cost_rung("", None) == "unknown"
+
+
+# ---- _truncate_response_text -------------------------------------
+
+
+def test_truncate_response_text_under_cap_passes_through():
+    short = "hello world"
+    assert _truncate_response_text(short) == short
+
+
+def test_truncate_response_text_at_cap():
+    text = "x" * 200
+    assert _truncate_response_text(text) == text
+
+
+def test_truncate_response_text_over_cap_truncated_with_ellipsis():
+    text = "x" * 250
+    result = _truncate_response_text(text)
+    assert len(result) == 201  # 200 chars + ellipsis
+    assert result.endswith("…")
+
+
+def test_truncate_response_text_none_passes_through():
+    assert _truncate_response_text(None) is None
+
+
+# ---- 1. Audit-only path: graceful degradation -----------------
+
+
+def test_audit_only_no_slack_dm_log_returns_null_model_fields(env):
+    write_audit(env, [_audit_entry()])
+    calls, raw_count = load_reasoning_calls_with_xref()
+    assert len(calls) == 1
+    call = calls[0]
+    assert call["model_used"] is None
+    assert call["input_tokens"] == 0
+    assert call["output_tokens"] == 0
+    assert call["response_text_truncated_200"] is None
+    assert call["cost_rung_at_call"] == "unknown"
+    # Group structure still present
+    assert call["tools_used"] == ["get_state"]
+    assert call["status"] == "ok"
+
+
+# ---- 2. Successful xref via thread_ts match -------------------
+
+
+def test_successful_xref_populates_model_fields(env):
+    write_audit(env, [_audit_entry()])
+    write_slack_dm(env, [_outbound_entry()])
+    calls, _ = load_reasoning_calls_with_xref()
+    call = calls[0]
+    assert call["model_used"] == "claude-opus-4-7"
+    assert call["input_tokens"] == 842
+    assert call["output_tokens"] == 127
+    assert call["response_text_truncated_200"] == "ok"
+    assert call["cost_rung_at_call"] == "normal"
+
+
+def test_xref_uses_thread_ts_match_when_available(env):
+    """thread_ts == event_ts is the natural threading match; even
+    when multiple outbound entries are in the channel, the matcher
+    picks the one whose thread_ts ties back to this inbound."""
+    write_audit(env, [_audit_entry()])
+    write_slack_dm(env, [
+        # Different thread, would be picked by time but ignored
+        # because thread_ts mismatches.
+        _outbound_entry(
+            minutes_ago=5,
+            thread_ts="other-thread.001",
+            model_used="claude-sonnet-4-6",
+            text="other",
+        ),
+        # Correct thread match
+        _outbound_entry(
+            minutes_ago=5,
+            thread_ts=_EVENT_TS,
+            model_used="claude-opus-4-7",
+            text="right",
+        ),
+    ])
+    calls, _ = load_reasoning_calls_with_xref()
+    assert calls[0]["model_used"] == "claude-opus-4-7"
+    assert calls[0]["response_text_truncated_200"] == "right"
+
+
+# ---- 3. Timestamp-window fallback -----------------------------
+
+
+def test_timestamp_window_fallback_when_thread_ts_mismatches(env):
+    """If no outbound thread_ts matches event_ts but an outbound
+    in the same channel is within ±60s of the audit emitted_at,
+    use it (best-effort correlation)."""
+    write_audit(env, [_audit_entry(minutes_ago=5)])
+    write_slack_dm(env, [
+        # No thread_ts match, but in-window by time
+        _outbound_entry(
+            minutes_ago=5,
+            thread_ts="completely-unrelated",
+            model_used="claude-sonnet-4-6",
+        ),
+    ])
+    calls, _ = load_reasoning_calls_with_xref()
+    assert calls[0]["model_used"] == "claude-sonnet-4-6"
+
+
+# ---- 4. Outside-window degrades gracefully ---------------------
+
+
+def test_outbound_outside_60s_window_no_xref(env):
+    write_audit(env, [_audit_entry(minutes_ago=5)])
+    write_slack_dm(env, [
+        # 30 min ago — way outside ±60s window
+        _outbound_entry(
+            minutes_ago=30,
+            thread_ts="unrelated",
+            model_used="claude-opus-4-7",
+        ),
+    ])
+    calls, _ = load_reasoning_calls_with_xref()
+    assert calls[0]["model_used"] is None, (
+        "Outbound outside ±60s window must NOT match — graceful "
+        "degradation to null fields"
+    )
+
+
+# ---- 5. Different-channel outbound ignored --------------------
+
+
+def test_outbound_different_channel_ignored(env):
+    write_audit(env, [_audit_entry()])
+    write_slack_dm(env, [
+        _outbound_entry(channel_id="DOTHERCHANNEL", model_used="claude-opus-4-7"),
+    ])
+    calls, _ = load_reasoning_calls_with_xref()
+    assert calls[0]["model_used"] is None, (
+        "Outbound from different channel must NOT cross-correlate"
+    )
+
+
+# ---- 6. Non-slack_dm session IDs degrade -----------------------
+
+
+def test_email_session_id_skips_xref(env):
+    """email-shaped session ids have no slack_dm correlation
+    target — KR-REASONING-PANEL-EMAIL-XREF follow-on handles."""
+    write_audit(env, [_audit_entry(caller_session_id="email:msg-123")])
+    write_slack_dm(env, [_outbound_entry()])
+    calls, _ = load_reasoning_calls_with_xref()
+    # Group still rendered, fields stay null
+    assert len(calls) == 1
+    assert calls[0]["model_used"] is None
+
+
+def test_mcp_session_id_skips_xref(env):
+    write_audit(env, [_audit_entry(caller_session_id="mcp:claude_pm:get_state")])
+    write_slack_dm(env, [_outbound_entry()])
+    calls, _ = load_reasoning_calls_with_xref()
+    assert calls[0]["model_used"] is None
+
+
+# ---- 7. Multiple groups + multiple outbound ---------------------
+
+
+def test_multiple_groups_each_pick_own_match(env):
+    """Two distinct sessions; each xrefs to its own outbound."""
+    write_audit(env, [
+        _audit_entry(
+            caller_session_id=f"{_DM_CHANNEL}:111.111",
+            tool_name="t-a",
+            minutes_ago=10,
+        ),
+        _audit_entry(
+            caller_session_id=f"{_DM_CHANNEL}:222.222",
+            tool_name="t-b",
+            minutes_ago=5,
+        ),
+    ])
+    write_slack_dm(env, [
+        _outbound_entry(
+            minutes_ago=10,
+            thread_ts="111.111",
+            model_used="claude-opus-4-7",
+            text="reply-a",
+        ),
+        _outbound_entry(
+            minutes_ago=5,
+            thread_ts="222.222",
+            model_used="claude-sonnet-4-6",
+            text="reply-b",
+        ),
+    ])
+    calls, _ = load_reasoning_calls_with_xref()
+    # Newest first by group started_at
+    by_id = {c["id"]: c for c in calls}
+    a = next(c for c in calls if c["response_text_truncated_200"] == "reply-a")
+    b = next(c for c in calls if c["response_text_truncated_200"] == "reply-b")
+    assert a["model_used"] == "claude-opus-4-7"
+    assert a["cost_rung_at_call"] == "normal"
+    assert b["model_used"] == "claude-sonnet-4-6"
+    assert b["cost_rung_at_call"] == "warn_75"
+
+
+# ---- 8. response_text truncation -----------------------------
+
+
+def test_long_response_text_truncated_to_200_chars(env):
+    long_text = "x" * 500
+    write_audit(env, [_audit_entry()])
+    write_slack_dm(env, [_outbound_entry(text=long_text)])
+    calls, _ = load_reasoning_calls_with_xref()
+    text = calls[0]["response_text_truncated_200"]
+    assert len(text) == 201  # 200 + ellipsis
+    assert text.endswith("…")
+
+
+# ---- 9. cost_ladder_halted xref supersedes status ------------
+
+
+def test_cost_ladder_halted_xref_supersedes_audit_status(env):
+    """When the xref'd outbound has reasoning_error=cost_ladder_halted,
+    the call's status surfaces as halted even if the audit rows
+    showed ok (e.g., the engine refused before logging any tool
+    calls; alternatively a stale ok row + a halt on a later turn)."""
+    write_audit(env, [_audit_entry(tool_status="ok")])
+    write_slack_dm(env, [
+        _outbound_entry(
+            model_used=None,
+            reasoning_error="cost_ladder_halted",
+            text=None,
+        ),
+    ])
+    calls, _ = load_reasoning_calls_with_xref()
+    assert calls[0]["status"] == "halted"
+    assert calls[0]["error_code"] == "cost_ladder_halted"
+    assert calls[0]["cost_rung_at_call"] == "hard_stop_100"
+
+
+# ---- 10. Malformed slack_dm log tolerated ---------------------
+
+
+def test_malformed_slack_dm_line_skipped(env, caplog):
+    """Malformed line in slack_dm log → log + skip, other entries
+    still parsed (same discipline as audit reader)."""
+    write_audit(env, [_audit_entry()])
+    log_path = env / "slack_dm_log.jsonl"
+    with log_path.open("w", encoding="utf-8") as f:
+        f.write("{NOT VALID JSON{{{\n")
+        f.write(json.dumps(_outbound_entry()) + "\n")
+
+    import logging
+    with caplog.at_level(logging.WARNING):
+        calls, _ = load_reasoning_calls_with_xref()
+    assert calls[0]["model_used"] == "claude-opus-4-7"
+
+
+# ---- 11. raw_in_window_count semantics ------------------------
+
+
+def test_raw_in_window_count_uses_individual_rows_not_groups(env):
+    """Per PR #141 rationale: aggregate counts must reflect
+    INDIVIDUAL audit rows, not groups, so the headline number
+    represents activity volume."""
+    write_audit(env, [
+        _audit_entry(caller_session_id="s1", tool_name="a"),
+        _audit_entry(caller_session_id="s1", tool_name="b"),
+        _audit_entry(caller_session_id="s1", tool_name="c"),
+        _audit_entry(caller_session_id="s2", tool_name="d"),
+    ])
+    calls, raw_count = load_reasoning_calls_with_xref()
+    assert len(calls) == 2  # 2 groups
+    assert raw_count == 4  # 4 individual rows
+
+
+# ---- 12. SECURITY walk-payload --------------------------------
+
+
+def test_no_token_shapes_anywhere_in_xref_output(env):
+    write_audit(env, [_audit_entry()])
+    write_slack_dm(env, [_outbound_entry(text="ok response text")])
+    calls, _ = load_reasoning_calls_with_xref()
+    blob = json.dumps(calls)
+    assert _ANTHROPIC_KEY_SHAPE.findall(blob) == []
+    assert _HEX_SECRET_SHAPE.findall(blob) == []
+
+
+# ---- 13. Endpoint integration --------------------------------
+
+
+@pytest.mark.asyncio
+async def test_endpoint_xref_populates_fields_when_slack_dm_present(env):
+    """End-to-end: hit the endpoint with both audit + slack_dm
+    fixture files. Verifies the endpoint actually wires through
+    the xref helper, not just the helper itself."""
+    write_audit(env, [_audit_entry()])
+    write_slack_dm(env, [_outbound_entry()])
+
+    from kora_cli import web_server
+
+    result = await web_server.list_recent_reasoning()
+    assert len(result["calls"]) == 1
+    call = result["calls"][0]
+    assert call["model_used"] == "claude-opus-4-7"
+    assert call["input_tokens"] == 842
+    assert call["cost_rung_at_call"] == "normal"
+    # by_model_24h aggregate also reflects xref
+    assert result["by_model_24h"].get("claude-opus-4-7") == 1
+    assert result["tokens_total_24h"]["input"] == 842
+
+
+@pytest.mark.asyncio
+async def test_endpoint_graceful_degradation_when_slack_dm_missing(env):
+    """End-to-end graceful degradation: audit present, slack_dm
+    absent → endpoint returns rows with null model fields, no
+    crash."""
+    write_audit(env, [_audit_entry()])
+    # No slack_dm file written
+
+    from kora_cli import web_server
+
+    result = await web_server.list_recent_reasoning()
+    assert len(result["calls"]) == 1
+    assert result["calls"][0]["model_used"] is None
+    assert result["by_model_24h"] == {}
+    assert result["tokens_total_24h"] == {"input": 0, "output": 0}