diff --git a/kora_cli/web_server.py b/kora_cli/web_server.py
index 3bfa0fa108a7..41428b880b05 100644
--- a/kora_cli/web_server.py
+++ b/kora_cli/web_server.py
@@ -5090,79 +5090,222 @@ async def list_recent_agent_activity():
 #      at the API edge, not in the operator's browser.
 
 
+# KR-SLACK-DM-PANEL-FLIP constants.
+_SLACK_DM_LOG_FILENAME = "slack_dm_log.jsonl"
+_SLACK_DM_DEFAULT_LIMIT = 50
+_SLACK_DM_MAX_LIMIT = 200
+# Channel-mask shape: first 4 + last 4 chars; backend exposes the
+# field, the small follow-on bucket KR-SLACK-DM-PANEL-CHANNEL-MASK
+# handles FE rendering (kept decoupled to keep this PR small).
+_CHANNEL_MASK_HEAD = 4
+_CHANNEL_MASK_TAIL = 4
+
+
+def _mask_channel_id(channel_id: str) -> str:
+    """First 4 + last 4 chars + literal ellipsis. Short channel IDs
+    (≤ head+tail+1) pass through unchanged — masking a shorter string
+    would expose more of it via implication than just returning it."""
+    if len(channel_id) <= _CHANNEL_MASK_HEAD + _CHANNEL_MASK_TAIL + 1:
+        return channel_id
+    return f"{channel_id[:_CHANNEL_MASK_HEAD]}…{channel_id[-_CHANNEL_MASK_TAIL:]}"
+
+
+def _project_slack_dm_entry(
+    entry: Dict[str, Any],
+    lineno: int,
+    expected_joshua: str,
+) -> Optional[Dict[str, Any]]:
+    """Project a single JSONL entry to the FE's SlackDMMessage shape.
+
+    Returns None when the entry can't be classified as inbound or
+    outbound (defensive — malformed entries skip rather than crash).
+
+    Direction discrimination follows the actual writer shape in
+    ``kora_cli/handlers/slack_dm_handler.py:302-310`` (inbound — has
+    ``received_at``/``handled_status``) vs ``slack_dm_handler.py:775-805``
+    (outbound — has ``sent_at``/``send_status``). The bucket spec's
+    K-DG block mentioned a ``direction: outbound`` key but the actual
+    writer does NOT include one; we discriminate on the keys that DO
+    exist.
+
+    SECURITY: ``user_id_label`` is a label only (joshua / kora_bot /
+    unknown_user) — never the raw U... Slack user ID. Resolution
+    mirrors the handler's own check at ``slack_dm_handler.py:241``
+    against ``KORA_SLACK_JOSHUA_USER_ID``.
+    """
+    channel_id_raw = entry.get("channel_id", "") or ""
+
+    if "received_at" in entry and "handled_status" in entry:
+        # Inbound
+        user_id = entry.get("user_id") or ""
+        if expected_joshua and user_id == expected_joshua:
+            label = "joshua"
+        else:
+            # Non-Joshua inbound: label as unknown_user even if the
+            # raw user_id is present in the JSONL; we never echo it.
+            label = "unknown_user"
+        return {
+            "id": f"line-{lineno}",
+            "direction": "inbound",
+            "timestamp": entry.get("received_at", ""),
+            "channel_id": channel_id_raw,
+            "channel_id_truncated": _mask_channel_id(channel_id_raw),
+            "thread_ts": entry.get("thread_ts"),
+            "user_id_label": label,
+            "text": entry.get("text", ""),
+            "handled_status": entry.get("handled_status", ""),
+        }
+
+    if "sent_at" in entry and "send_status" in entry:
+        # Outbound — always Kora's bot identity. send_status is
+        # "ok" / "failed"; FE expects "sent_ok" / "sent_failed".
+        send_status = entry.get("send_status", "")
+        return {
+            "id": f"line-{lineno}",
+            "direction": "outbound",
+            "timestamp": entry.get("sent_at", ""),
+            "channel_id": channel_id_raw,
+            "channel_id_truncated": _mask_channel_id(channel_id_raw),
+            "thread_ts": entry.get("thread_ts"),
+            "user_id_label": "kora_bot",
+            "text": entry.get("text", ""),
+            "handled_status": f"sent_{send_status}",
+        }
+
+    return None
+
+
 @app.get("/api/slack-dm/recent")
-async def list_recent_slack_dm():
+async def list_recent_slack_dm(limit: int = _SLACK_DM_DEFAULT_LIMIT):
     """Return recent Kora ↔ Joshua DM messages for the operator lens.
 
-    v1 stub — pinned shape so CC#3's KR-FEAT-SLACK-DM ST2 can swap
-    the body without touching the FE.
+    Reads ``${KORA_HOME}/slack_dm_log.jsonl`` (written by
+    ``kora_cli/handlers/slack_dm_handler.py`` per PR #119/#122) and
+    projects each entry to the FE's ``SlackDMMessage`` shape.
 
-    Per-message fields:
-      id              — opaque id
-      direction       — "inbound" | "outbound"
-      timestamp       — ISO-8601
-      channel_id      — STUB label (D_STUB1 etc) in v1; real IDs
-                        must be hashed/truncated when flipped
-      thread_ts       — Slack thread parent ts (or null)
-      user_id_label   — LABEL only (joshua / kora_bot / unknown_user);
-                        never the raw U... Slack user ID
-      text            — message body (FE renders as plain text)
-      handled_status  — received | sent_ok | sent_failed |
-                        filtered_non_joshua | filtered_bot |
-                        filtered_subtype | handler_error | dropped_paused
+    Query params:
+      ``limit`` — number of newest entries to return; default 50,
+                  capped at 200 to bound large-file reads.
+
+    Per-message fields (matches FE TS interface from PR #120):
+      id                    — derived line-{N} id, stable within file
+      direction             — "inbound" | "outbound"
+      timestamp             — ISO-8601 (from received_at or sent_at)
+      channel_id            — raw channel ID (D... for DM channels)
+      channel_id_truncated  — masked form (head…tail) for the
+                              upcoming FE channel-mask bucket; FE
+                              consumers ignore this field today
+      thread_ts             — Slack thread parent ts (or null)
+      user_id_label         — LABEL only (joshua / kora_bot /
+                              unknown_user); NEVER the raw U... ID
+      text                  — message body (FE renders plain text)
+      handled_status        — received / filtered_non_joshua /
+                              filtered_bot / filtered_subtype /
+                              handler_error / dropped_paused
+                              (inbound) OR sent_ok / sent_failed
+                              (outbound; derived from send_status)
+
+    SECURITY (4-layer contract preserved from PR #120):
+      1. user_id_label is a derived LABEL; raw user_id never reaches
+         the wire. Tests sweep payload for U[A-Z0-9]{8,} shape.
+      2. channel_id starts with "D" (Slack DM channel) — backend
+         test asserts. channel_id_truncated companion field lets
+         the FE follow-on render a masked view.
+      3. text rendered as PLAIN TEXT by FE (React default escape +
+         dangerouslySetInnerHTML ban pinned in
+         test_web_server_slack_dm.py).
+      4. Walk-payload guard for xoxb-/xoxp- Slack token shapes.
+
+    Behaviour:
+      * Missing JSONL file → empty list + stub:false (fresh daemon
+        with no DMs yet — empty-state UI; no error).
+      * Malformed JSONL line → logged + skipped; other entries
+        still parsed (defensive against partial-write corruption).
+      * Sort: newest first by timestamp descending.
+      * stub: false always — this endpoint no longer serves stub
+        data, even on empty file (FE's STUB banner stays hidden).
     """
+    from datetime import datetime, timedelta, timezone
+    import json as _json
+    import os as _os
+
+    capped_limit = max(1, min(limit, _SLACK_DM_MAX_LIMIT))
+    log_path = get_kora_home() / _SLACK_DM_LOG_FILENAME
+    expected_joshua = _os.environ.get(
+        "KORA_SLACK_JOSHUA_USER_ID", ""
+    ).strip()
+    now = datetime.now(timezone.utc)
+    generated_at = now.strftime("%Y-%m-%dT%H:%M:%SZ")
+    cutoff_24h = now - timedelta(hours=24)
+
+    projected: List[Dict[str, Any]] = []
+
+    if log_path.is_file():
+        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:
+                        _log.warning(
+                            "[kora.slack_dm_panel] line %d malformed JSON, "
+                            "skipped: %r",
+                            lineno,
+                            exc,
+                        )
+                        continue
+                    if not isinstance(entry, dict):
+                        _log.warning(
+                            "[kora.slack_dm_panel] line %d not a JSON object, "
+                            "skipped",
+                            lineno,
+                        )
+                        continue
+                    msg = _project_slack_dm_entry(entry, lineno, expected_joshua)
+                    if msg is not None:
+                        projected.append(msg)
+        except OSError as exc:
+            _log.warning(
+                "[kora.slack_dm_panel] failed to read %s: %r",
+                log_path,
+                exc,
+            )
+
+    # Newest-first sort. ISO-8601 lex order == chronological order
+    # for UTC Z-suffixed timestamps (same shape the writer uses).
+    projected.sort(key=lambda m: m.get("timestamp", ""), reverse=True)
+
+    # Aggregate counts use the FULL projected set (not the limited
+    # slice) so the headline numbers reflect the whole 24h window.
+    def _within_24h(ts_str: str) -> bool:
+        if not ts_str:
+            return False
+        try:
+            dt = datetime.fromisoformat(ts_str.replace("Z", "+00:00"))
+        except ValueError:
+            return False
+        if dt.tzinfo is None:
+            dt = dt.replace(tzinfo=timezone.utc)
+        return dt >= cutoff_24h
+
+    in_window = [m for m in projected if _within_24h(m["timestamp"])]
+    by_direction: Dict[str, int] = {"inbound": 0, "outbound": 0}
+    by_status: Dict[str, int] = {}
+    for m in in_window:
+        by_direction[m["direction"]] = by_direction.get(m["direction"], 0) + 1
+        status = m["handled_status"]
+        by_status[status] = by_status.get(status, 0) + 1
+
     return {
-        "messages": [
-            {
-                "id": "stub-1",
-                "direction": "inbound",
-                "timestamp": "2026-05-22T17:55:13Z",
-                "channel_id": "D_STUB1",
-                "thread_ts": None,
-                "user_id_label": "joshua",
-                "text": "Kora, what's the daemon status?",
-                "handled_status": "received",
-            },
-            {
-                "id": "stub-2",
-                "direction": "outbound",
-                "timestamp": "2026-05-22T17:55:14Z",
-                "channel_id": "D_STUB1",
-                "thread_ts": "1779380123.456",
-                "user_id_label": "kora_bot",
-                "text": "Kora received: Kora, what's the daemon status?",
-                "handled_status": "sent_ok",
-            },
-            {
-                "id": "stub-3",
-                "direction": "inbound",
-                "timestamp": "2026-05-22T17:52:01Z",
-                "channel_id": "D_STUB1",
-                "thread_ts": None,
-                "user_id_label": "joshua",
-                "text": "hey",
-                "handled_status": "received",
-            },
-            {
-                "id": "stub-4",
-                "direction": "inbound",
-                "timestamp": "2026-05-22T17:48:22Z",
-                "channel_id": "D_STUB2",
-                "thread_ts": None,
-                "user_id_label": "unknown_user",
-                "text": "[filtered: non-Joshua sender]",
-                "handled_status": "filtered_non_joshua",
-            },
-        ],
-        "stub": True,
-        "generated_at": "2026-05-22T18:00:00Z",
-        "total_recent_24h": 12,
-        "by_direction_24h": {"inbound": 7, "outbound": 5},
-        "by_status_24h": {
-            "received": 6,
-            "sent_ok": 5,
-            "filtered_non_joshua": 1,
-        },
+        "messages": projected[:capped_limit],
+        "stub": False,
+        "generated_at": generated_at,
+        "total_recent_24h": len(in_window),
+        "by_direction_24h": by_direction,
+        "by_status_24h": by_status,
     }
 
 
diff --git a/tests/kora_cli/test_web_server_slack_dm.py b/tests/kora_cli/test_web_server_slack_dm.py
index 02afafb37fc4..43042f3181ae 100644
--- a/tests/kora_cli/test_web_server_slack_dm.py
+++ b/tests/kora_cli/test_web_server_slack_dm.py
@@ -1,26 +1,39 @@
-"""Tests for the KR-SLACK-DM-PANEL stub endpoint.
-
-Bucket §3 scenarios:
-  1. GET /api/slack-dm/recent returns 200
-  2. Top-level shape (messages + stub:true + generated_at +
-     total_recent_24h + by_direction_24h + by_status_24h)
-  3. 4 representative stub messages present
-  4. Stub spans inbound + outbound + filtered_non_joshua so the
-     operator's first look surfaces filtering posture
-  5. Per-entry shape + valid direction + valid handled_status enum
-  6. SECURITY: no raw Slack user IDs (U[A-Z0-9]{8,}) anywhere in
-     payload — walk-whole-payload guard
-  7. SECURITY: no Slack token shapes (xoxb-/xoxp-/xapp-/xoxa-) or
-     signing-secret-shaped hex runs anywhere in payload — walk-whole
-  8. channel_id v1 stub shape (D_<label>)
-  9. SECURITY: companion FE pin — SlackDMPanel.tsx never uses
-     dangerouslySetInnerHTML for message text (comment-stripped grep)
- 10. by_direction_24h sum reconciles to total_recent_24h
- 11. Cron-regression sanity
+"""Tests for the KR-SLACK-DM-PANEL-FLIP endpoint.
+
+After PR #134's stub shipped, the endpoint reads from the live
+``${KORA_HOME}/slack_dm_log.jsonl`` written by the handler in
+``kora_cli/handlers/slack_dm_handler.py`` (PR #119 inbound, #122
+outbound, #131 reasoning meta, #130 caller_actor_kind). The
+4-layer security contract from the original panel is preserved.
+
+Scenarios:
+  1. Empty / missing JSONL → empty list + stub:false
+  2. Single inbound entry → projects with handled_status pass-through
+  3. Single outbound entry → projects with handled_status=sent_<send_status>
+  4. user_id_label resolution: joshua / kora_bot / unknown_user
+  5. Newest-first ordering by timestamp
+  6. ?limit query param respected; capped at 200
+  7. Malformed JSONL line → logged + skipped; other lines parsed
+  8. JSON-object-but-not-dict (e.g., array on a line) → skipped
+  9. Entry missing both received_at + sent_at → skipped (defensive)
+ 10. channel_id_truncated companion field present + correctly masked
+ 11. SECURITY: user_id_label is a label, NOT raw U... Slack user ID
+ 12. SECURITY: walk-payload sweep finds no raw U... anywhere
+ 13. SECURITY: walk-payload sweep finds no xoxb-/xoxp- Slack token shapes
+ 14. SECURITY: channel_id starts with "D" (DM channel shape)
+ 15. SECURITY: FE companion pin — no dangerouslySetInnerHTML
+ 16. SECURITY: FE companion pin — text rendered as JSX child node
+ 17. by_direction_24h sum reconciles to total_recent_24h
+ 18. by_status_24h keys are drawn from the valid enum
+ 19. Aggregate counts use 24h window (not the limited slice)
+ 20. Cron-regression sanity
 """
 
+import json
 import re
+from datetime import datetime, timedelta, timezone
 from pathlib import Path
+from typing import Any, Dict, List
 
 import pytest
 
@@ -37,75 +50,156 @@
     "dropped_paused",
 }
 
-# Raw Slack user IDs are "U" + 8+ uppercase alphanumerics
-# (e.g. U01ABCDEF). The payload's user_id_label is a LABEL only
-# (joshua / kora_bot / unknown_user); a Slack ID showing up would
-# signal the backend leaked the raw value through.
 _RAW_SLACK_USER_ID = re.compile(r"\bU[A-Z0-9]{8,}\b")
-
-# Slack token shapes — bot tokens (xoxb-), user tokens (xoxp-),
-# app tokens (xapp-), legacy/test tokens (xoxa-), workflow tokens
-# (xoxr-). Match the prefix + at least the first segment so we
-# catch real tokens but not the bare "xoxb-" word in code.
 _SLACK_TOKEN_SHAPE = re.compile(r"\bxox[abprs]-[0-9A-Za-z-]{8,}\b")
-
-# Slack signing secrets are 32-char hex strings. The DM endpoint
-# should never carry one — they live in env-var settings, not the
-# wire format. Catch any 32-char hex run as a precaution.
 _SIGNING_SECRET_SHAPE = re.compile(r"\b[0-9a-f]{32}\b")
 
-# Bucket §2(a): v1 channel_id is a STUB label of shape D_<label>.
-# Real channel IDs (C…, G…, D… + 8+ alphanumerics) must be
-# hashed/truncated by CC#3 when real data flips; this pin ensures
-# the v1 stub doesn't accidentally use a real-shaped value.
-_CHANNEL_ID_STUB = re.compile(r"^D_[A-Z0-9]+$")
-_REAL_SLACK_CHANNEL_SHAPE = re.compile(r"\b[CGD][A-Z0-9]{8,}\b")
-
-
 _REPO_ROOT = Path(__file__).resolve().parents[2]
 _PANEL_PATH = _REPO_ROOT / "web" / "src" / "pages" / "SlackDMPanel.tsx"
 
 
 def _strip_ts_comments(src: str) -> str:
-    """Strip /* … */ block comments, // line comments, and {/* … */}
-    JSX block comments so source-pin tests check live code only, not
-    explanatory prose that may legitimately mention the banned pattern.
-    """
     src = re.sub(r"\{/\*.*?\*/\}", "", src, flags=re.DOTALL)
     src = re.sub(r"/\*.*?\*/", "", src, flags=re.DOTALL)
     src = re.sub(r"(^|[^:])//[^\n]*", r"\1", src)
     return src
 
 
-@pytest.fixture(autouse=True)
-def _isolate_config(tmp_path, monkeypatch):
+# ---- Test env isolation + JSONL fixture helpers --------------------
+
+
+# Match the env-var name the handler reads (slack_dm_handler.py:59).
+_JOSHUA_USER_ID = "U01JOSHUA"
+_OTHER_USER_ID = "U02OTHER"
+_DM_CHANNEL = "D0123456789ABCDEF"
+
+
+@pytest.fixture
+def env(tmp_path, monkeypatch):
+    """Isolated KORA_HOME + Joshua env var set to a known U... ID.
+
+    Monkeypatches the ``get_kora_home`` symbol DIRECTLY in
+    ``kora_cli.web_server`` (where the endpoint imports it via
+    ``from kora_cli.config import get_kora_home`` — a copy in the
+    module namespace, not a transparent re-export). Also patches the
+    upstream ``kora_constants`` reference for code paths that import
+    it from there. Doing both belt-and-braces because the
+    set_kora_home_override ContextVar approach leaks across parallel
+    pytest-xdist tests within the same worker process — each test
+    rebinds the override but the ContextVar chain isn't restored
+    cleanly across multiple fixtures, leaving stale state.
+    """
     monkeypatch.setenv("HERMES_HOME", str(tmp_path))
-    monkeypatch.setattr("kora_cli.config.get_kora_home", lambda: tmp_path)
+    monkeypatch.setenv("KORA_HOME", str(tmp_path))
+    monkeypatch.setenv("KORA_SLACK_JOSHUA_USER_ID", _JOSHUA_USER_ID)
     monkeypatch.setattr(
-        "kora_cli.config.get_config_path", lambda: tmp_path / "config.yaml"
+        "kora_cli.config.get_config_path",
+        lambda: tmp_path / "config.yaml",
     )
     monkeypatch.setattr(
         "kora_cli.config.get_env_path", lambda: tmp_path / ".env"
     )
+    monkeypatch.setattr("kora_constants.get_kora_home", lambda: tmp_path)
+    monkeypatch.setattr("kora_cli.config.get_kora_home", lambda: tmp_path)
+    # Critical: the endpoint resolves get_kora_home from its own
+    # module namespace (line 5233), not via a fresh import. Monkeypatch
+    # there too.
+    monkeypatch.setattr("kora_cli.web_server.get_kora_home", lambda: tmp_path)
     return tmp_path
 
 
-# ---- 1. 200 -----------------------------------------------------------
+def write_log(env_path: Path, lines: List[Dict[str, Any]]) -> Path:
+    """Write a JSONL fixture at ``${KORA_HOME}/slack_dm_log.jsonl``."""
+    log_path = env_path / "slack_dm_log.jsonl"
+    with log_path.open("w", encoding="utf-8") as f:
+        for entry in lines:
+            f.write(json.dumps(entry) + "\n")
+    return log_path
+
+
+def _iso(minutes_ago: int = 0) -> str:
+    """ISO-8601 UTC Z-suffixed, N minutes in the past. Matches the
+    writer's ``_now_iso()`` output shape."""
+    ts = datetime.now(timezone.utc) - timedelta(minutes=minutes_ago)
+    return ts.strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def _inbound(
+    *,
+    user_id: str = _JOSHUA_USER_ID,
+    text: str = "hello",
+    handled_status: str = "received",
+    minutes_ago: int = 5,
+    thread_ts: Any = None,
+    channel_id: str = _DM_CHANNEL,
+) -> Dict[str, Any]:
+    """Inbound JSONL entry shape per slack_dm_handler.py:302-310."""
+    return {
+        "received_at": _iso(minutes_ago),
+        "channel_id": channel_id,
+        "thread_ts": thread_ts,
+        "user_id": user_id,
+        "text": text,
+        "event_ts": f"{int(datetime.now().timestamp())}.000000",
+        "handled_status": handled_status,
+    }
+
+
+def _outbound(
+    *,
+    text: str = "ack",
+    send_status: str = "ok",
+    minutes_ago: int = 4,
+    thread_ts: str = "1779380123.456",
+    channel_id: str = _DM_CHANNEL,
+    **extra: Any,
+) -> Dict[str, Any]:
+    """Outbound JSONL entry shape per slack_dm_handler.py:775-805."""
+    entry: Dict[str, Any] = {
+        "sent_at": _iso(minutes_ago),
+        "channel_id": channel_id,
+        "thread_ts": thread_ts,
+        "text": text,
+        "slack_message_ts": f"{int(datetime.now().timestamp())}.000001",
+        "send_status": send_status,
+    }
+    entry.update(extra)
+    return entry
+
+
+# ---- 1. Endpoint always returns stub:false (no more hardcoded stub)
 
 
 @pytest.mark.asyncio
-async def test_endpoint_returns_200(_isolate_config):
+async def test_endpoint_returns_200_with_stub_false(env):
     from kora_cli import web_server
 
     result = await web_server.list_recent_slack_dm()
     assert isinstance(result, dict)
+    assert result["stub"] is False, (
+        "Post-flip endpoint must always return stub:false (no more "
+        "hardcoded sample data)"
+    )
+
+
+@pytest.mark.asyncio
+async def test_empty_log_returns_empty_messages_list(env):
+    """Fresh daemon with no DMs yet → log file missing → empty list
+    + stub:false. NOT an error; FE renders existing empty-state UI."""
+    from kora_cli import web_server
+
+    result = await web_server.list_recent_slack_dm()
+    assert result["messages"] == []
+    assert result["total_recent_24h"] == 0
+    assert result["stub"] is False
 
 
 # ---- 2. Top-level shape ---------------------------------------------
 
 
 @pytest.mark.asyncio
-async def test_response_shape_has_required_keys(_isolate_config):
+async def test_response_shape_has_required_keys(env):
+    write_log(env, [_inbound()])
     from kora_cli import web_server
 
     result = await web_server.list_recent_slack_dm()
@@ -117,285 +211,469 @@ async def test_response_shape_has_required_keys(_isolate_config):
         "by_direction_24h",
         "by_status_24h",
     }
-    assert isinstance(result["messages"], list)
-    assert isinstance(result["generated_at"], str)
-    assert isinstance(result["total_recent_24h"], int)
-    assert isinstance(result["by_direction_24h"], dict)
-    assert isinstance(result["by_status_24h"], dict)
-    assert result["stub"] is True
 
 
-# ---- 3. Expected stub messages --------------------------------------
+# ---- 3. Inbound entry projection -----------------------------------
 
 
 @pytest.mark.asyncio
-async def test_stub_returns_four_representative_messages(_isolate_config):
-    """Pin the bucket §2(a) canonical 4-message stub list. CC#3's
-    KR-FEAT-SLACK-DM ST2 will swap the body to read from
-    ${HERMES_HOME}/slack_dm_log.jsonl, but the shape stays stable
-    so the FE keeps rendering during cut-over."""
+async def test_inbound_entry_projects_to_inbound_direction(env):
+    write_log(env, [_inbound(handled_status="received")])
     from kora_cli import web_server
 
     result = await web_server.list_recent_slack_dm()
-    assert len(result["messages"]) == 4
-    ids = {m["id"] for m in result["messages"]}
-    assert ids == {"stub-1", "stub-2", "stub-3", "stub-4"}
+    assert len(result["messages"]) == 1
+    msg = result["messages"][0]
+    assert msg["direction"] == "inbound"
+    assert msg["handled_status"] == "received"
+    assert msg["user_id_label"] == "joshua"
+    assert msg["channel_id"] == _DM_CHANNEL
 
 
 @pytest.mark.asyncio
-async def test_stub_spans_inbound_outbound_and_filtered(_isolate_config):
-    """The 4 stub messages deliberately span inbound + outbound +
-    filtered_non_joshua so the operator's first look shows what the
-    filtering posture looks like. Pin so a future stub edit can't
-    accidentally homogenize."""
+async def test_inbound_filtered_non_joshua_passes_status_through(env):
+    """handled_status enum values flow through unchanged for inbound."""
+    write_log(env, [
+        _inbound(user_id=_OTHER_USER_ID, handled_status="filtered_non_joshua")
+    ])
     from kora_cli import web_server
 
     result = await web_server.list_recent_slack_dm()
-    directions = {m["direction"] for m in result["messages"]}
-    statuses = {m["handled_status"] for m in result["messages"]}
-    assert directions == {"inbound", "outbound"}
-    assert "filtered_non_joshua" in statuses
-    assert "received" in statuses
-    assert "sent_ok" in statuses
+    msg = result["messages"][0]
+    assert msg["direction"] == "inbound"
+    assert msg["handled_status"] == "filtered_non_joshua"
+    assert msg["user_id_label"] == "unknown_user", (
+        "Non-Joshua inbound must label as unknown_user — never echo "
+        "the raw user_id"
+    )
 
 
-# ---- 4. Per-entry shape + enums ------------------------------------
+# ---- 4. Outbound entry projection ----------------------------------
 
 
 @pytest.mark.asyncio
-async def test_each_message_has_required_keys_and_valid_enums(_isolate_config):
+async def test_outbound_entry_projects_to_outbound_direction(env):
+    write_log(env, [_outbound(send_status="ok")])
     from kora_cli import web_server
 
     result = await web_server.list_recent_slack_dm()
-    required = {
-        "id",
-        "direction",
-        "timestamp",
-        "channel_id",
-        "thread_ts",
-        "user_id_label",
-        "text",
-        "handled_status",
-    }
-    for msg in result["messages"]:
-        assert set(msg.keys()) == required, (
-            f"{msg.get('id', '?')}: keys mismatch {set(msg.keys())}"
-        )
-        assert msg["direction"] in _VALID_DIRECTION, (
-            f"{msg['id']}: direction={msg['direction']!r}"
-        )
-        assert msg["handled_status"] in _VALID_STATUS, (
-            f"{msg['id']}: handled_status={msg['handled_status']!r} not in "
-            f"{_VALID_STATUS}"
+    msg = result["messages"][0]
+    assert msg["direction"] == "outbound"
+    assert msg["user_id_label"] == "kora_bot", (
+        "Outbound always Kora's bot identity"
+    )
+    assert msg["handled_status"] == "sent_ok", (
+        "Outbound send_status:ok → handled_status:sent_ok per FE enum"
+    )
+
+
+@pytest.mark.asyncio
+async def test_outbound_failed_status_maps_to_sent_failed(env):
+    write_log(env, [_outbound(send_status="failed", failure_reason="rate_limited")])
+    from kora_cli import web_server
+
+    result = await web_server.list_recent_slack_dm()
+    msg = result["messages"][0]
+    assert msg["handled_status"] == "sent_failed"
+
+
+@pytest.mark.asyncio
+async def test_outbound_with_reasoning_meta_still_projects(env):
+    """PR #131 outbound entries may include model_used / tokens /
+    reasoning_error. The projection function must ignore these
+    gracefully — extra fields don't break parsing."""
+    write_log(env, [
+        _outbound(
+            model_used="claude-opus-4-7",
+            input_tokens=120,
+            output_tokens=34,
+            reasoning_duration_ms=890,
         )
-        assert isinstance(msg["timestamp"], str) and msg["timestamp"].endswith("Z")
-        assert isinstance(msg["channel_id"], str)
-        # thread_ts is either null or a Slack timestamp string
-        assert msg["thread_ts"] is None or isinstance(msg["thread_ts"], str)
-        assert isinstance(msg["user_id_label"], str) and msg["user_id_label"]
-        assert isinstance(msg["text"], str)
+    ])
+    from kora_cli import web_server
+
+    result = await web_server.list_recent_slack_dm()
+    assert len(result["messages"]) == 1
+
+
+@pytest.mark.asyncio
+async def test_outbound_with_caller_actor_kind_still_projects(env):
+    """PR #130 may add caller_actor_kind to outbound entries."""
+    write_log(env, [_outbound(caller_actor_kind="kora_drone_7")])
+    from kora_cli import web_server
+
+    result = await web_server.list_recent_slack_dm()
+    assert len(result["messages"]) == 1
+
+
+# ---- 5. user_id_label resolution ----------------------------------
+
+
+@pytest.mark.asyncio
+async def test_user_id_label_resolves_joshua_via_env_match(env):
+    write_log(env, [_inbound(user_id=_JOSHUA_USER_ID)])
+    from kora_cli import web_server
+
+    result = await web_server.list_recent_slack_dm()
+    assert result["messages"][0]["user_id_label"] == "joshua"
+
+
+@pytest.mark.asyncio
+async def test_user_id_label_falls_back_to_unknown_user(env):
+    """Non-Joshua inbound (or Joshua env unset) → unknown_user. NEVER
+    the raw user_id."""
+    write_log(env, [_inbound(user_id=_OTHER_USER_ID)])
+    from kora_cli import web_server
+
+    result = await web_server.list_recent_slack_dm()
+    msg = result["messages"][0]
+    assert msg["user_id_label"] == "unknown_user"
+    assert _OTHER_USER_ID not in msg["user_id_label"]
+
+
+# ---- 6. Newest-first ordering -------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_newest_first_ordering(env):
+    """File-on-disk order doesn't matter — endpoint sorts by timestamp
+    descending so the newest event is messages[0]."""
+    write_log(env, [
+        _inbound(text="oldest", minutes_ago=30),
+        _outbound(text="middle", minutes_ago=20),
+        _inbound(text="newest", minutes_ago=5),
+    ])
+    from kora_cli import web_server
+
+    result = await web_server.list_recent_slack_dm()
+    texts = [m["text"] for m in result["messages"]]
+    assert texts == ["newest", "middle", "oldest"]
+
+
+# ---- 7. ?limit query param + 200 cap ------------------------------
+
+
+@pytest.mark.asyncio
+async def test_limit_query_param_caps_returned_messages(env):
+    write_log(env, [
+        _inbound(text=f"msg-{i}", minutes_ago=i) for i in range(20)
+    ])
+    from kora_cli import web_server
+
+    result = await web_server.list_recent_slack_dm(limit=5)
+    assert len(result["messages"]) == 5
+
+
+@pytest.mark.asyncio
+async def test_limit_query_param_capped_at_200(env):
+    """Performance guard: callers can't request unbounded message
+    counts — limit=99999 caps at 200."""
+    write_log(env, [
+        _inbound(text=f"msg-{i}", minutes_ago=i) for i in range(5)
+    ])
+    from kora_cli import web_server
+
+    result = await web_server.list_recent_slack_dm(limit=99999)
+    assert len(result["messages"]) <= 200
+
+
+@pytest.mark.asyncio
+async def test_limit_zero_or_negative_clamps_to_at_least_one(env):
+    """Defensive: limit=0 / limit=-5 shouldn't return everything OR
+    crash. Clamp to a minimum of 1."""
+    write_log(env, [_inbound()])
+    from kora_cli import web_server
+
+    result = await web_server.list_recent_slack_dm(limit=0)
+    assert isinstance(result["messages"], list)
 
 
-# ---- 5. SECURITY: no raw Slack user IDs in payload -----------------
+# ---- 8. Malformed JSONL handling ----------------------------------
 
 
 @pytest.mark.asyncio
-async def test_user_id_label_is_label_not_raw_slack_id(_isolate_config):
-    """Bucket §2(a) HARD CONSTRAINT: user_id_label is a LABEL
-    (joshua / kora_bot / unknown_user) — never a raw Slack user ID
-    (U + 8+ uppercase alphanumerics). If the real handler ever
-    defaults to the raw Slack ID when no label resolves, this
-    guard catches the leak before it ships to the operator's
-    browser."""
+async def test_malformed_json_line_skipped_other_lines_parsed(env, caplog):
+    """A corrupt line (partial write / disk-full truncation) must
+    not bring down the endpoint. Log + skip; other entries return."""
+    log_path = env / "slack_dm_log.jsonl"
+    with log_path.open("w", encoding="utf-8") as f:
+        f.write(json.dumps(_inbound(text="before")) + "\n")
+        f.write("{NOT VALID JSON{{{\n")
+        f.write(json.dumps(_inbound(text="after")) + "\n")
+
+    from kora_cli import web_server
+
+    import logging
+    with caplog.at_level(logging.WARNING):
+        result = await web_server.list_recent_slack_dm()
+
+    texts = sorted(m["text"] for m in result["messages"])
+    assert texts == ["after", "before"]
+
+
+@pytest.mark.asyncio
+async def test_json_array_line_skipped(env):
+    """A line that parses as JSON but is an array (not a dict) gets
+    skipped — projection expects dict shape."""
+    log_path = env / "slack_dm_log.jsonl"
+    with log_path.open("w", encoding="utf-8") as f:
+        f.write(json.dumps(_inbound(text="ok-1")) + "\n")
+        f.write(json.dumps([1, 2, 3]) + "\n")
+        f.write(json.dumps(_inbound(text="ok-2")) + "\n")
+
+    from kora_cli import web_server
+
+    result = await web_server.list_recent_slack_dm()
+    texts = sorted(m["text"] for m in result["messages"])
+    assert texts == ["ok-1", "ok-2"]
+
+
+@pytest.mark.asyncio
+async def test_entry_missing_both_received_at_and_sent_at_skipped(env):
+    """Defensive against partial-shape entries (handler bug or
+    forward-compat experiment). Skip without crashing."""
+    log_path = env / "slack_dm_log.jsonl"
+    with log_path.open("w", encoding="utf-8") as f:
+        f.write(json.dumps(_inbound()) + "\n")
+        f.write(json.dumps({"random_field": "value"}) + "\n")
+
+    from kora_cli import web_server
+
+    result = await web_server.list_recent_slack_dm()
+    assert len(result["messages"]) == 1
+
+
+@pytest.mark.asyncio
+async def test_empty_lines_skipped(env):
+    """Blank lines in JSONL files are common (manual editing). Must
+    be tolerated."""
+    log_path = env / "slack_dm_log.jsonl"
+    with log_path.open("w", encoding="utf-8") as f:
+        f.write(json.dumps(_inbound(text="one")) + "\n")
+        f.write("\n")
+        f.write("   \n")
+        f.write(json.dumps(_inbound(text="two")) + "\n")
+
+    from kora_cli import web_server
+
+    result = await web_server.list_recent_slack_dm()
+    assert len(result["messages"]) == 2
+
+
+# ---- 9. channel_id_truncated companion field ----------------------
+
+
+@pytest.mark.asyncio
+async def test_channel_id_truncated_field_present_and_masked(env):
+    """Per spec §2(b): backend exposes the masked form. FE follow-on
+    bucket (KR-SLACK-DM-PANEL-CHANNEL-MASK) renders it. Pinned so
+    the field stays available for the follow-on PR to consume."""
+    write_log(env, [_inbound(channel_id="D0123456789ABCDEF")])
+    from kora_cli import web_server
+
+    result = await web_server.list_recent_slack_dm()
+    msg = result["messages"][0]
+    assert "channel_id_truncated" in msg
+    assert msg["channel_id_truncated"] == "D012…CDEF", (
+        f"channel_id_truncated should mask middle: got "
+        f"{msg['channel_id_truncated']!r}"
+    )
+    assert msg["channel_id"] == "D0123456789ABCDEF"
+
+
+@pytest.mark.asyncio
+async def test_short_channel_id_passes_through_unmasked(env):
+    """Pathological short channel IDs (test fixtures, weird daemons)
+    pass through unchanged — masking a short string would expose
+    more by implication."""
+    write_log(env, [_inbound(channel_id="DABC")])
+    from kora_cli import web_server
+
+    result = await web_server.list_recent_slack_dm()
+    msg = result["messages"][0]
+    assert msg["channel_id_truncated"] == "DABC"
+
+
+# ---- 10. SECURITY: user_id_label layer 1 ---------------------------
+
+
+@pytest.mark.asyncio
+async def test_user_id_label_never_contains_raw_slack_id(env):
+    """Per-field SECURITY pin: the resolved user_id_label must
+    NEVER be a raw U... shape, regardless of writer behavior."""
+    write_log(env, [
+        _inbound(user_id=_JOSHUA_USER_ID),
+        _inbound(user_id=_OTHER_USER_ID, handled_status="filtered_non_joshua"),
+        _inbound(user_id="UABCDEFG12", handled_status="filtered_non_joshua"),
+    ])
     from kora_cli import web_server
 
     result = await web_server.list_recent_slack_dm()
     for msg in result["messages"]:
         label = msg["user_id_label"]
         assert _RAW_SLACK_USER_ID.search(label) is None, (
-            f"{msg['id']}: user_id_label={label!r} matches raw Slack "
-            f"user ID shape — contract requires a human label only"
+            f"user_id_label={label!r} matches raw Slack user ID shape"
         )
 
 
 @pytest.mark.asyncio
-async def test_no_raw_slack_user_ids_anywhere_in_payload(_isolate_config):
-    """Walk-the-whole-payload guard (standardizing the pattern from
-    KR-WEBHOOK-EVENTS / KR-AGENT-ACTIVITY). Asserts no field
-    anywhere in the response — top-level, per-message, or any
-    nested dict — contains a raw Slack user ID shape. Catches a
-    future drift like adding "raw_user_id" diagnostic or stuffing
-    a user ID into the message text."""
+async def test_no_raw_slack_user_ids_anywhere_in_payload(env):
+    """Walk-payload SECURITY sweep: regardless of writer behavior,
+    NO raw U... pattern appears anywhere in the response. Catches
+    a future drift where the projector accidentally surfaces
+    entry.user_id into a diagnostic field."""
+    write_log(env, [
+        _inbound(user_id=_JOSHUA_USER_ID),
+        _inbound(user_id=_OTHER_USER_ID, handled_status="filtered_non_joshua"),
+    ])
     from kora_cli import web_server
-    import json as _json
 
     result = await web_server.list_recent_slack_dm()
-    blob = _json.dumps(result)
+    blob = json.dumps(result)
     leaks = _RAW_SLACK_USER_ID.findall(blob)
     assert leaks == [], (
-        f"payload contains raw Slack user ID(s): {leaks} — Slack "
-        f"user IDs are PII and must be resolved to labels at the API "
-        f"edge (bucket §2(a) SECURITY contract)"
+        f"payload contains raw Slack user ID(s): {leaks}"
     )
 
 
-# ---- 6. SECURITY: no Slack token / signing-secret shapes -----------
+# ---- 11. SECURITY: Slack token shapes -----------------------------
 
 
 @pytest.mark.asyncio
-async def test_no_slack_token_shapes_anywhere_in_payload(_isolate_config):
-    """Bucket §2(a) SECURITY: walk-payload regex catching xoxb-/xoxp-/
-    xapp-/xoxa-/xoxr- Slack token shapes anywhere. A backend bug or
-    future log entry that leaks creds gets caught at the API edge,
-    not in the operator's browser (where it could end up in
-    diag-bundles, screenshots, etc.)."""
+async def test_no_slack_token_shapes_anywhere_in_payload(env):
+    """Walk-payload guard for xoxb-/xoxp- token shapes — backend
+    bug or future field that leaks credential material gets caught."""
+    write_log(env, [_inbound(text="legitimate user message")])
     from kora_cli import web_server
-    import json as _json
 
     result = await web_server.list_recent_slack_dm()
-    blob = _json.dumps(result)
+    blob = json.dumps(result)
     leaks = _SLACK_TOKEN_SHAPE.findall(blob)
     assert leaks == [], (
-        f"payload contains Slack token shape(s): {leaks} — Slack "
-        f"tokens must never appear in API responses (bucket §2(a) "
-        f"SECURITY contract layer 4)"
+        f"payload contains Slack token shape(s): {leaks}"
     )
 
 
 @pytest.mark.asyncio
-async def test_no_signing_secret_shapes_anywhere_in_payload(_isolate_config):
-    """Belt+braces companion to the Slack-token guard: 32-char hex
-    runs are the shape of a Slack signing secret. The DM endpoint
-    has no business carrying one — they live in env-var settings,
-    not in the wire format."""
+async def test_no_signing_secret_shapes_anywhere_in_payload(env):
+    """32-char hex run = Slack signing secret shape."""
+    write_log(env, [_inbound()])
     from kora_cli import web_server
-    import json as _json
 
     result = await web_server.list_recent_slack_dm()
-    blob = _json.dumps(result)
+    blob = json.dumps(result)
     leaks = _SIGNING_SECRET_SHAPE.findall(blob)
     assert leaks == [], (
-        f"payload contains 32-char hex string(s): {leaks} — these "
-        f"shape-match a Slack signing secret; the DM endpoint must "
-        f"never carry one"
+        f"payload contains 32-char hex string(s): {leaks}"
     )
 
 
-# ---- 7. channel_id stub shape pinning ------------------------------
+# ---- 12. SECURITY: channel_id starts with D ----------------------
 
 
 @pytest.mark.asyncio
-async def test_channel_id_uses_stub_label_format(_isolate_config):
-    """Bucket §2(a): v1 channel_id is a STUB label of shape D_<label>
-    (e.g. D_STUB1, D_STUB2). Real Slack channel IDs (C…/G…/D… +
-    8+ alphanumerics) must be hashed/truncated by CC#3 before the
-    flip — this pin ensures the v1 stub doesn't accidentally use a
-    real-shaped value that would mask the missing-redaction step."""
+async def test_channel_id_starts_with_d_for_dm_channels(env):
+    """Per spec §2(b): real DM channel IDs start with D. Pin so a
+    future writer change (e.g., starting to log group/channel chat
+    accidentally) gets caught — DM-panel should only ever show DMs."""
+    write_log(env, [
+        _inbound(channel_id="D0123456789ABCDEF"),
+        _outbound(channel_id="D0123456789ABCDEF"),
+    ])
     from kora_cli import web_server
 
     result = await web_server.list_recent_slack_dm()
     for msg in result["messages"]:
-        cid = msg["channel_id"]
-        assert _CHANNEL_ID_STUB.match(cid), (
-            f"{msg['id']}: channel_id={cid!r} doesn't match the v1 stub "
-            f"shape D_<label>"
-        )
-        # And explicitly: must NOT match the real Slack channel shape
-        assert not _REAL_SLACK_CHANNEL_SHAPE.match(cid), (
-            f"{msg['id']}: channel_id={cid!r} looks like a real Slack "
-            f"channel ID — v1 stub must use the D_<label> placeholder"
+        assert msg["channel_id"].startswith("D"), (
+            f"channel_id={msg['channel_id']!r} should be a Slack DM "
+            f"channel (D-prefix). The DM panel should only ever show "
+            f"DM channels — group/public channels would leak via this "
+            f"endpoint."
         )
 
 
-# ---- 8. SECURITY: companion FE pin for plain-text rendering --------
+# ---- 13. SECURITY: FE companion pins ------------------------------
 
 
 def test_panel_uses_no_dangerously_set_inner_html_for_message_text():
-    """Bucket §2(a) layer 3: message text rendered as PLAIN TEXT.
-    React's default child escaping handles this — but this guard
-    catches a future edit that flips to dangerouslySetInnerHTML
-    for "richer message formatting". Real DM text may contain
-    arbitrary user-typed content (Markdown, HTML, scripts); an
-    HTML-injection vector here would execute attacker-controlled
-    content in the operator's browser."""
     code = _strip_ts_comments(_PANEL_PATH.read_text())
-    assert "dangerouslySetInnerHTML" not in code, (
-        "SlackDMPanel.tsx must not use dangerouslySetInnerHTML — "
-        "message text is arbitrary user-typed content and must "
-        "render as plain text only"
-    )
+    assert "dangerouslySetInnerHTML" not in code
 
 
 def test_panel_renders_message_text_as_child_text_node():
-    """Belt+braces complement: confirm message.text is rendered as a
-    JSX child (escaped) not assigned to an innerHTML attribute or
-    piped through a markdown lib. The collapsed view uses
-    truncateText(message.text); both paths must be plain child
-    expressions."""
     src = _PANEL_PATH.read_text()
-    # message.text must be referenced as a JSX child expression
-    # somewhere in the file. The exact wrapping varies (ternary
-    # between expanded/collapsed paths), so pin the substring rather
-    # than the full JSX shape.
-    assert "message.text" in src, (
-        "SlackDMPanel.tsx should reference message.text in the render"
-    )
-    # Collapsed path: via truncateText, also as a child expression —
-    # this pins that the truncation goes through the pure helper, not
-    # an inline regex or a markdown-aware renderer.
-    assert "truncateText(message.text)" in src, (
-        "SlackDMPanel.tsx should render the truncated message text "
-        "via the pure truncateText() helper (no formatting)"
-    )
-    # Confirm message.text appears inside a JSX expression container
-    # at least once (i.e. between { and } in a JSX position), as a
-    # weak structural guarantee that it's used as a child not an
-    # innerHTML-assigned attribute string.
-    assert re.search(r"\{[^{}]*message\.text[^{}]*\}", src), (
-        "message.text should appear inside a JSX expression "
-        "container, indicating it's rendered as a child / value"
-    )
+    assert "message.text" in src
+    assert "truncateText(message.text)" in src
+    assert re.search(r"\{[^{}]*message\.text[^{}]*\}", src)
 
 
-# ---- 9. by_direction_24h reconciliation ---------------------------
+# ---- 14. Aggregate reconciliation --------------------------------
 
 
 @pytest.mark.asyncio
-async def test_by_direction_24h_sum_reconciles_to_total(_isolate_config):
-    """The 24h direction breakdown must sum to total_recent_24h —
-    otherwise the dashboard card's "X msgs / Y drops" headline
-    won't reconcile to the panel's per-direction breakdown."""
+async def test_by_direction_24h_sum_reconciles_to_total(env):
+    write_log(env, [
+        _inbound(text="i1"),
+        _inbound(text="i2"),
+        _outbound(text="o1"),
+    ])
     from kora_cli import web_server
 
     result = await web_server.list_recent_slack_dm()
     direction_sum = sum(result["by_direction_24h"].values())
-    assert direction_sum == result["total_recent_24h"], (
-        f"by_direction_24h sums to {direction_sum} but "
-        f"total_recent_24h is {result['total_recent_24h']}"
-    )
-    assert set(result["by_direction_24h"].keys()) == _VALID_DIRECTION
+    assert direction_sum == result["total_recent_24h"]
 
 
 @pytest.mark.asyncio
-async def test_by_status_24h_only_contains_valid_status_values(_isolate_config):
-    """Every key in the status breakdown must be a documented
-    handled_status enum value — keeps the dashboard's status
-    pivot reconcilable with the per-message badges."""
+async def test_by_status_24h_only_contains_valid_status_values(env):
+    write_log(env, [
+        _inbound(handled_status="received"),
+        _outbound(send_status="ok"),
+    ])
     from kora_cli import web_server
 
     result = await web_server.list_recent_slack_dm()
     invalid = set(result["by_status_24h"].keys()) - _VALID_STATUS
     assert not invalid, (
-        f"by_status_24h has unknown status key(s): {invalid} — must "
-        f"be drawn from {_VALID_STATUS}"
+        f"by_status_24h has unknown status key(s): {invalid}"
     )
 
 
-# ---- 10. Cron-regression sanity ------------------------------------
+@pytest.mark.asyncio
+async def test_aggregate_counts_use_full_24h_window_not_limited_slice(env):
+    """If limit=5 but 20 entries fell within 24h, total_recent_24h
+    must still be 20. Otherwise the headline number on the dashboard
+    misrepresents activity."""
+    write_log(env, [
+        _inbound(text=f"msg-{i}", minutes_ago=i) for i in range(20)
+    ])
+    from kora_cli import web_server
+
+    result = await web_server.list_recent_slack_dm(limit=5)
+    assert len(result["messages"]) == 5
+    assert result["total_recent_24h"] == 20
+
+
+@pytest.mark.asyncio
+async def test_entries_older_than_24h_excluded_from_aggregate(env):
+    """An entry from 25h ago appears in the messages list (no
+    timestamp filter on the slice) but doesn't count toward
+    total_recent_24h."""
+    write_log(env, [
+        _inbound(text="recent", minutes_ago=10),
+        _inbound(text="ancient", minutes_ago=25 * 60),
+    ])
+    from kora_cli import web_server
+
+    result = await web_server.list_recent_slack_dm()
+    assert result["total_recent_24h"] == 1
+
+
+# ---- 15. Cron-regression sanity ----------------------------------
 
 
 @pytest.mark.asyncio
-async def test_cron_endpoint_still_works_with_slack_dm_registered(_isolate_config):
+async def test_cron_endpoint_still_works_with_slack_dm_registered(env):
     from kora_cli import web_server
 
     jobs = await web_server.list_cron_jobs(profile="all")