diff --git a/kora_cli/handlers/__init__.py b/kora_cli/handlers/__init__.py
new file mode 100644
index 000000000000..bcd165502acb
--- /dev/null
+++ b/kora_cli/handlers/__init__.py
@@ -0,0 +1,7 @@
+"""Per-source webhook handlers (KR-FEAT-SLACK-DM, KR-FEAT-EMAIL, ...).
+
+Distinct from ``kora_cli/listeners/`` (which owns the daemon-coordinator
+lifecycle + transport-layer routing). A handler module takes a
+verified-payload dict from a listener + drives Kora-specific logic:
+identity filtering, state-gating, persistence, downstream emits.
+"""
diff --git a/kora_cli/handlers/slack_dm_handler.py b/kora_cli/handlers/slack_dm_handler.py
new file mode 100644
index 000000000000..abb597b0a3aa
--- /dev/null
+++ b/kora_cli/handlers/slack_dm_handler.py
@@ -0,0 +1,317 @@
+"""Slack-DM handler — KR-FEAT-SLACK-DM ST1.
+
+Called from ``kora_cli/listeners/webhooks.py:_handle_slack`` after
+HMAC verification + URL-verification handshake. Owns the Kora-side
+DM-processing logic:
+
+  - Identity check: sender must match ``KORA_SLACK_JOSHUA_USER_ID``.
+    Non-Joshua messages are dropped silently (don't echo back to a
+    third party).
+  - Channel-type filter: only ``"im"`` events. Channel messages,
+    app_mentions, etc. are filtered.
+  - Bot-message filter: events with ``event.bot_id`` set are filtered
+    (defense against echo-loops if Kora's own bot is ever in the
+    conversation).
+  - Subtype filter: only regular messages (no ``event.subtype``);
+    drop message_changed / message_deleted / message_replied etc.
+  - OperationalStateHolder gating: PAUSED or STOPPED → drop. Don't
+    process Joshua's DM during a pause.
+  - JSONL append-only persistence at ``<KORA_HOME>/slack_dm_log.jsonl``.
+  - ``[kora.slack_dm.received]`` structured-log emit on Joshua DMs
+    (chain-event vocab literal flagged for substrate follow-on; same
+    pattern as KR-D-DAEMON ST3 webhook dead-letter + KR-MCP-RUNTIME-
+    SURFACE ST2 audit log).
+
+# Security posture
+
+The signing secret is consumed by the HMAC verifier in the listener;
+this handler NEVER sees it. JSONL entries are bounded to a fixed
+allow-list of fields — body content is recorded (Joshua's own
+message text is the operator-visible record by design), but no
+header values, no signing secret, no bot token, no auth metadata.
+
+A unit test asserts the JSONL does NOT contain the signing-secret
+env value after a sequence of events.
+
+# Exception posture
+
+Any uncaught exception inside ``handle_event`` is caught at the
+listener-layer wrap (see ``webhooks.py``) and logged to the
+dead-letter logger; we return 200 to Slack so it doesn't retry
+indefinitely. The handler's own internal failure modes (JSONL write
+failure, holder unavailable, etc.) WARN-log + continue — never
+crash the request, never block the 200 response.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Dict, Optional
+
+logger = logging.getLogger(__name__)
+
+
+# Env vars.
+JOSHUA_USER_ID_ENV = "KORA_SLACK_JOSHUA_USER_ID"
+LOG_PATH_ENV = "KORA_SLACK_DM_LOG_PATH"  # test override; defaults to KORA_HOME
+
+# Handled-status enum for the JSONL ``handled_status`` field.
+HANDLED_RECEIVED = "received"
+HANDLED_FILTERED_NON_JOSHUA = "filtered_non_joshua"
+HANDLED_FILTERED_NON_IM = "filtered_non_im"
+HANDLED_FILTERED_BOT = "filtered_bot"
+HANDLED_FILTERED_SUBTYPE = "filtered_subtype"
+HANDLED_DROPPED_PAUSED = "dropped_paused"
+HANDLED_DROPPED_STOPPED = "dropped_stopped"
+HANDLED_HANDLER_ERROR = "handler_error"
+
+
+def _resolve_log_path() -> Path:
+    """Return the JSONL log path: env override → ``KORA_HOME/slack_dm_log.jsonl``."""
+    override = os.environ.get(LOG_PATH_ENV, "").strip()
+    if override:
+        return Path(override)
+    # Lazy import — keeps test paths that monkeypatch LOG_PATH_ENV from
+    # needing the full kora_constants resolution chain.
+    from kora_constants import get_kora_home
+
+    return get_kora_home() / "slack_dm_log.jsonl"
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _safe_extract(event: Dict[str, Any], *keys: str) -> Optional[Any]:
+    """Walk ``event[k1][k2]...`` defensively; return None on any miss."""
+    cur: Any = event
+    for k in keys:
+        if not isinstance(cur, dict):
+            return None
+        cur = cur.get(k)
+        if cur is None:
+            return None
+    return cur
+
+
+# ---------------------------------------------------------------------------
+# Handler
+# ---------------------------------------------------------------------------
+
+
+class SlackDMHandler:
+    """Processes a single verified Slack Events payload.
+
+    Stateless across requests — each ``handle_event`` call processes
+    one event independently. Persistent state (the JSONL log) is
+    file-backed; in-memory state is request-scoped.
+    """
+
+    def __init__(self, log_path: Optional[Path] = None) -> None:
+        self._log_path = log_path or _resolve_log_path()
+
+    async def handle_event(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+        """Process a Slack Events payload.
+
+        Returns the Slack-API-compliant response dict (always
+        ``{"ok": True}`` — Slack uses 200 + ok as the acknowledgement;
+        the daemon's response code is set in the listener layer).
+
+        Filter order:
+
+          1. PAUSED / STOPPED state → drop, no further processing.
+          2. ``event.type`` must be ``message`` AND no subtype.
+          3. ``event.channel_type`` must be ``im``.
+          4. ``event.bot_id`` must be absent.
+          5. ``event.user`` must match ``KORA_SLACK_JOSHUA_USER_ID``.
+
+        Each filter writes a JSONL entry with the appropriate
+        ``handled_status`` then returns ``{"ok": True}``.
+
+        Exceptions during filter / log / emit are caught + WARN-logged
+        + still return ``{"ok": True}`` — never let internal failures
+        cause Slack to retry.
+        """
+        try:
+            return await self._handle_event_inner(payload)
+        except Exception as exc:
+            # Last-resort guard. Log + return ok so Slack doesn't retry.
+            logger.warning(
+                "[kora.slack_dm] handler raised %r — returning ok to Slack",
+                exc,
+            )
+            try:
+                self._append_log_entry(
+                    payload, HANDLED_HANDLER_ERROR, error=repr(exc)
+                )
+            except Exception as inner:
+                logger.warning(
+                    "[kora.slack_dm] failed to log handler error: %r", inner
+                )
+            return {"ok": True}
+
+    async def _handle_event_inner(
+        self, payload: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        # Filter 1: OperationalStateHolder gating.
+        gate_status = self._check_state_gate()
+        if gate_status is not None:
+            self._append_log_entry(payload, gate_status)
+            logger.info(
+                "[kora.slack_dm] %s — message dropped",
+                gate_status,
+            )
+            return {"ok": True}
+
+        event_type = _safe_extract(payload, "event", "type")
+        subtype = _safe_extract(payload, "event", "subtype")
+        channel_type = _safe_extract(payload, "event", "channel_type")
+        bot_id = _safe_extract(payload, "event", "bot_id")
+        user_id = _safe_extract(payload, "event", "user")
+
+        # Filter 2: only regular messages (no subtype).
+        if event_type != "message" or subtype:
+            self._append_log_entry(
+                payload,
+                HANDLED_FILTERED_SUBTYPE,
+                extra={"event_type": event_type, "subtype": subtype},
+            )
+            return {"ok": True}
+
+        # Filter 3: only IM channel-type.
+        if channel_type != "im":
+            self._append_log_entry(
+                payload,
+                HANDLED_FILTERED_NON_IM,
+                extra={"channel_type": channel_type},
+            )
+            return {"ok": True}
+
+        # Filter 4: bot messages.
+        if bot_id:
+            self._append_log_entry(
+                payload,
+                HANDLED_FILTERED_BOT,
+                extra={"bot_id": bot_id},
+            )
+            return {"ok": True}
+
+        # Filter 5: identity — must be Joshua.
+        expected_joshua = os.environ.get(JOSHUA_USER_ID_ENV, "").strip()
+        if not expected_joshua:
+            # Misconfigured — fail-CLOSED. Without the Joshua ID set,
+            # we can't verify the sender, so drop everything.
+            logger.warning(
+                "[kora.slack_dm] %s unset — all messages dropped (fail-CLOSED)",
+                JOSHUA_USER_ID_ENV,
+            )
+            self._append_log_entry(
+                payload,
+                HANDLED_FILTERED_NON_JOSHUA,
+                extra={"reason": "joshua_id_env_unset"},
+            )
+            return {"ok": True}
+        if user_id != expected_joshua:
+            self._append_log_entry(
+                payload,
+                HANDLED_FILTERED_NON_JOSHUA,
+                extra={"actual_user_id": user_id},
+            )
+            return {"ok": True}
+
+        # All filters passed — Joshua DM received.
+        self._append_log_entry(payload, HANDLED_RECEIVED)
+        self._emit_received_event(payload)
+        return {"ok": True}
+
+    # ------------------------------------------------------------------
+    # Filters / helpers
+    # ------------------------------------------------------------------
+
+    def _check_state_gate(self) -> Optional[str]:
+        """Return a handled_status if the operational state should
+        drop this message; otherwise None."""
+        try:
+            from agent.operational_state import PrimaryState
+            from agent.operational_state_holder import get_holder
+        except Exception:
+            # If the operational-state module can't even be imported,
+            # we're in an unusual test path. Don't gate; let the rest
+            # of the filters apply.
+            return None
+
+        holder = get_holder()
+        if holder is None:
+            # No holder initialized → no gating. The handler is
+            # running outside the daemon (or in a partial-init test);
+            # let processing proceed.
+            return None
+
+        # holder.current is a @property — caught in KR-MCP-RUNTIME-SURFACE
+        # ST1 K-DG corrections.
+        state = holder.current
+        ps = state.primary_state
+
+        if ps is PrimaryState.PAUSED:
+            return HANDLED_DROPPED_PAUSED
+        if ps is PrimaryState.STOPPED:
+            return HANDLED_DROPPED_STOPPED
+        return None
+
+    def _append_log_entry(
+        self,
+        payload: Dict[str, Any],
+        handled_status: str,
+        *,
+        extra: Optional[Dict[str, Any]] = None,
+        error: Optional[str] = None,
+    ) -> None:
+        """Append one JSONL entry. Best-effort: any write failure is
+        WARN-logged + swallowed."""
+        entry: Dict[str, Any] = {
+            "received_at": _now_iso(),
+            "channel_id": _safe_extract(payload, "event", "channel") or "",
+            "thread_ts": _safe_extract(payload, "event", "thread_ts"),
+            "user_id": _safe_extract(payload, "event", "user") or "",
+            "text": _safe_extract(payload, "event", "text") or "",
+            "event_ts": _safe_extract(payload, "event", "ts") or "",
+            "handled_status": handled_status,
+        }
+        if extra:
+            entry["extra"] = extra
+        if error:
+            entry["error"] = error
+
+        try:
+            # Ensure parent dir exists (KORA_HOME may need to be created
+            # in test envs). Best-effort; failure path logged.
+            self._log_path.parent.mkdir(parents=True, exist_ok=True)
+            with self._log_path.open("a", encoding="utf-8") as f:
+                f.write(json.dumps(entry, default=str) + "\n")
+        except OSError as exc:
+            logger.warning(
+                "[kora.slack_dm] log write failed (%s): %r",
+                self._log_path,
+                exc,
+            )
+
+    def _emit_received_event(self, payload: Dict[str, Any]) -> None:
+        """Stable structured-log emit for an identified Joshua DM.
+
+        ``kora.slack_dm.received`` is the intended chain-event vocab
+        literal; if/when substrate ships the CHECK-constraint
+        addition, this can extend to also call
+        ``IsoKronMCPClient.invoke("kora__append_event", ...)``.
+        For now structured log is the audit seam.
+        """
+        logger.info(
+            "[kora.slack_dm.received] channel=%s user=%s ts=%s text_len=%d",
+            _safe_extract(payload, "event", "channel") or "",
+            _safe_extract(payload, "event", "user") or "",
+            _safe_extract(payload, "event", "ts") or "",
+            len(_safe_extract(payload, "event", "text") or ""),
+        )
diff --git a/kora_cli/listeners/webhooks.py b/kora_cli/listeners/webhooks.py
index d51f0db75c63..5a8b6c24ddda 100644
--- a/kora_cli/listeners/webhooks.py
+++ b/kora_cli/listeners/webhooks.py
@@ -184,10 +184,38 @@ async def _handle_slack(request: Request) -> Response:
         challenge = payload.get("challenge", "")
         return PlainTextResponse(challenge)
 
-    # All other event types — Feature 5 will land the real handler.
-    # ST3 scaffolding logs + acknowledges.
+    # KR-FEAT-SLACK-DM ST1 — route verified `event_callback` payloads
+    # to the Slack-DM handler. The handler owns Kora-specific filtering
+    # (identity / channel-type / bot / subtype / state-gate) + JSONL
+    # persistence + chain-event emit. Belt-and-suspenders exception
+    # guard at the listener boundary: the handler wraps its own body
+    # too, but if SlackDMHandler construction itself fails, we still
+    # need to 200 Slack to prevent its aggressive retries.
+    if isinstance(payload, dict) and payload.get("type") == "event_callback":
+        try:
+            from kora_cli.handlers.slack_dm_handler import SlackDMHandler
+
+            handler = SlackDMHandler()
+            await handler.handle_event(payload)
+        except Exception as exc:
+            logger.warning(
+                "[kora.webhook.slack] handler raised %r — dead-lettering",
+                exc,
+            )
+            emit_webhook_dead_letter(
+                source="slack",
+                reason=f"handler_error: {type(exc).__name__}",
+                headers=dict(request.headers),
+                peer_ip=_peer_ip(request),
+                request_id=request.headers.get("x-request-id"),
+                body_bytes=len(raw_body),
+            )
+        return JSONResponse({"ok": True})
+
+    # Other Slack event-wrapper types we don't route (e.g.
+    # rate_limit, app_rate_limited). Log + 200 OK.
     logger.info(
-        "[kora.webhook.slack] event accepted: type=%s",
+        "[kora.webhook.slack] event accepted but not routed: type=%s",
         payload.get("type") if isinstance(payload, dict) else "(unknown)",
     )
     return JSONResponse({"ok": True})
diff --git a/tests/kora_cli/handlers/__init__.py b/tests/kora_cli/handlers/__init__.py
new file mode 100644
index 000000000000..e69de29bb2d1
diff --git a/tests/kora_cli/handlers/test_slack_dm_handler.py b/tests/kora_cli/handlers/test_slack_dm_handler.py
new file mode 100644
index 000000000000..46ffabd87a24
--- /dev/null
+++ b/tests/kora_cli/handlers/test_slack_dm_handler.py
@@ -0,0 +1,434 @@
+"""Tests for ``kora_cli.handlers.slack_dm_handler`` — KR-FEAT-SLACK-DM ST1.
+
+Covers:
+  - Joshua DM → received + logged + chain-event-log emit
+  - Non-Joshua → filtered_non_joshua + logged + no emit
+  - Bot message → filtered_bot
+  - Subtype event → filtered_subtype
+  - Non-IM channel → filtered_non_im
+  - PAUSED state → dropped_paused
+  - STOPPED state → dropped_stopped
+  - Handler internal exception → handler_error + still 200
+  - JSONL format: one valid JSON per line, required fields present
+  - SECURITY: signing-secret env value never appears in JSONL
+  - JOSHUA_USER_ID env unset → all messages filtered (fail-CLOSED)
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from pathlib import Path
+from typing import Any, Dict
+
+import pytest
+
+from kora_cli.handlers import slack_dm_handler as sdm
+from kora_cli.handlers.slack_dm_handler import (
+    HANDLED_DROPPED_PAUSED,
+    HANDLED_DROPPED_STOPPED,
+    HANDLED_FILTERED_BOT,
+    HANDLED_FILTERED_NON_IM,
+    HANDLED_FILTERED_NON_JOSHUA,
+    HANDLED_FILTERED_SUBTYPE,
+    HANDLED_HANDLER_ERROR,
+    HANDLED_RECEIVED,
+    JOSHUA_USER_ID_ENV,
+    SlackDMHandler,
+)
+
+
+JOSHUA_ID = "UJOSHUA01"
+
+
+def _make_payload(
+    *,
+    user: str = JOSHUA_ID,
+    channel: str = "D01CHAN01",
+    text: str = "hi kora",
+    ts: str = "1700000000.001",
+    channel_type: str = "im",
+    bot_id: str | None = None,
+    subtype: str | None = None,
+    thread_ts: str | None = None,
+    event_type: str = "message",
+) -> Dict[str, Any]:
+    """Build a Slack Events `event_callback` payload."""
+    event: Dict[str, Any] = {
+        "type": event_type,
+        "user": user,
+        "channel": channel,
+        "channel_type": channel_type,
+        "text": text,
+        "ts": ts,
+    }
+    if bot_id is not None:
+        event["bot_id"] = bot_id
+    if subtype is not None:
+        event["subtype"] = subtype
+    if thread_ts is not None:
+        event["thread_ts"] = thread_ts
+    return {"type": "event_callback", "event": event}
+
+
+@pytest.fixture
+def log_path(tmp_path):
+    path = tmp_path / "slack_dm_log.jsonl"
+    return path
+
+
+@pytest.fixture
+def handler(log_path):
+    return SlackDMHandler(log_path=log_path)
+
+
+@pytest.fixture(autouse=True)
+def _joshua_env(monkeypatch):
+    monkeypatch.setenv(JOSHUA_USER_ID_ENV, JOSHUA_ID)
+
+
+@pytest.fixture(autouse=True)
+def _reset_holder(monkeypatch):
+    """Default to no operational-state holder so tests don't see
+    accidental PAUSED-state drops. Per-test fixtures can override."""
+    from agent import operational_state_holder as h_mod
+
+    monkeypatch.setattr(h_mod, "_HOLDER", None)
+
+
+def _read_log_lines(log_path: Path) -> list[dict]:
+    if not log_path.exists():
+        return []
+    return [
+        json.loads(line)
+        for line in log_path.read_text(encoding="utf-8").splitlines()
+        if line.strip()
+    ]
+
+
+# ---------------------------------------------------------------------------
+# Happy path — Joshua DM
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_joshua_dm_received_and_logged(handler, log_path, caplog):
+    caplog.set_level(logging.INFO)
+    result = await handler.handle_event(_make_payload(text="hello"))
+    assert result == {"ok": True}
+
+    lines = _read_log_lines(log_path)
+    assert len(lines) == 1
+    entry = lines[0]
+    assert entry["handled_status"] == HANDLED_RECEIVED
+    assert entry["user_id"] == JOSHUA_ID
+    assert entry["text"] == "hello"
+    assert entry["channel_id"] == "D01CHAN01"
+    assert entry["event_ts"] == "1700000000.001"
+    # Chain-event-log emit fires for Joshua only.
+    assert any(
+        "kora.slack_dm.received" in r.getMessage() for r in caplog.records
+    )
+
+
+@pytest.mark.asyncio
+async def test_joshua_dm_with_thread_logs_thread_ts(handler, log_path):
+    payload = _make_payload(thread_ts="1700000000.000")
+    await handler.handle_event(payload)
+    [entry] = _read_log_lines(log_path)
+    assert entry["thread_ts"] == "1700000000.000"
+
+
+# ---------------------------------------------------------------------------
+# Identity filter
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_non_joshua_filtered_no_emit(handler, log_path, caplog):
+    caplog.set_level(logging.INFO)
+    result = await handler.handle_event(_make_payload(user="USOMEONEELSE"))
+    assert result == {"ok": True}
+
+    [entry] = _read_log_lines(log_path)
+    assert entry["handled_status"] == HANDLED_FILTERED_NON_JOSHUA
+    assert entry["extra"]["actual_user_id"] == "USOMEONEELSE"
+    # Chain-event-log emit must NOT fire for non-Joshua.
+    assert not any(
+        "kora.slack_dm.received" in r.getMessage() for r in caplog.records
+    )
+
+
+@pytest.mark.asyncio
+async def test_joshua_env_unset_drops_all(handler, log_path, monkeypatch, caplog):
+    """Fail-CLOSED — without JOSHUA_USER_ID env, we can't verify
+    sender, so drop everything."""
+    monkeypatch.delenv(JOSHUA_USER_ID_ENV, raising=False)
+    caplog.set_level(logging.WARNING)
+    result = await handler.handle_event(_make_payload(user=JOSHUA_ID))
+    assert result == {"ok": True}
+
+    [entry] = _read_log_lines(log_path)
+    assert entry["handled_status"] == HANDLED_FILTERED_NON_JOSHUA
+    assert entry["extra"]["reason"] == "joshua_id_env_unset"
+    assert any(
+        JOSHUA_USER_ID_ENV in r.getMessage() and "fail-CLOSED" in r.getMessage()
+        for r in caplog.records
+    )
+
+
+# ---------------------------------------------------------------------------
+# Filter precedence — state gate first
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_paused_state_drops_before_other_filters(
+    handler, log_path, monkeypatch
+):
+    """Even a valid Joshua DM is dropped when PAUSED."""
+    from agent.operational_state import OperationalState, PrimaryState
+    from agent.operational_state_holder import OperationalStateHolder
+    from agent import operational_state_holder as h_mod
+
+    monkeypatch.setattr(
+        h_mod,
+        "_HOLDER",
+        OperationalStateHolder(
+            OperationalState(primary_state=PrimaryState.PAUSED)
+        ),
+    )
+
+    await handler.handle_event(_make_payload())
+    [entry] = _read_log_lines(log_path)
+    assert entry["handled_status"] == HANDLED_DROPPED_PAUSED
+
+
+@pytest.mark.asyncio
+async def test_stopped_state_drops(handler, log_path, monkeypatch):
+    from agent.operational_state import OperationalState, PrimaryState
+    from agent.operational_state_holder import OperationalStateHolder
+    from agent import operational_state_holder as h_mod
+
+    monkeypatch.setattr(
+        h_mod,
+        "_HOLDER",
+        OperationalStateHolder(
+            OperationalState(primary_state=PrimaryState.STOPPED)
+        ),
+    )
+
+    await handler.handle_event(_make_payload())
+    [entry] = _read_log_lines(log_path)
+    assert entry["handled_status"] == HANDLED_DROPPED_STOPPED
+
+
+@pytest.mark.asyncio
+async def test_ready_state_does_not_drop(handler, log_path, monkeypatch):
+    from agent.operational_state import OperationalState, PrimaryState
+    from agent.operational_state_holder import OperationalStateHolder
+    from agent import operational_state_holder as h_mod
+
+    monkeypatch.setattr(
+        h_mod,
+        "_HOLDER",
+        OperationalStateHolder(
+            OperationalState(primary_state=PrimaryState.READY)
+        ),
+    )
+
+    await handler.handle_event(_make_payload())
+    [entry] = _read_log_lines(log_path)
+    assert entry["handled_status"] == HANDLED_RECEIVED
+
+
+@pytest.mark.asyncio
+async def test_active_state_does_not_drop(handler, log_path, monkeypatch):
+    from agent.operational_state import OperationalState, PrimaryState
+    from agent.operational_state_holder import OperationalStateHolder
+    from agent import operational_state_holder as h_mod
+
+    monkeypatch.setattr(
+        h_mod,
+        "_HOLDER",
+        OperationalStateHolder(
+            OperationalState(primary_state=PrimaryState.ACTIVE)
+        ),
+    )
+
+    await handler.handle_event(_make_payload())
+    [entry] = _read_log_lines(log_path)
+    assert entry["handled_status"] == HANDLED_RECEIVED
+
+
+# ---------------------------------------------------------------------------
+# Bot / subtype / channel-type filters
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_bot_message_filtered(handler, log_path):
+    """Defense against echo-loops: if Kora's own bot ever shows up
+    in the chat, ignore its messages."""
+    await handler.handle_event(_make_payload(bot_id="B0KORA01"))
+    [entry] = _read_log_lines(log_path)
+    assert entry["handled_status"] == HANDLED_FILTERED_BOT
+    assert entry["extra"]["bot_id"] == "B0KORA01"
+
+
+@pytest.mark.asyncio
+async def test_subtype_message_filtered(handler, log_path):
+    """message_changed / message_deleted etc. — drop."""
+    await handler.handle_event(_make_payload(subtype="message_changed"))
+    [entry] = _read_log_lines(log_path)
+    assert entry["handled_status"] == HANDLED_FILTERED_SUBTYPE
+    assert entry["extra"]["subtype"] == "message_changed"
+
+
+@pytest.mark.asyncio
+async def test_non_message_event_filtered_as_subtype(handler, log_path):
+    """app_mention / reaction_added / etc. — caught by the
+    event_type != 'message' branch."""
+    await handler.handle_event(_make_payload(event_type="app_mention"))
+    [entry] = _read_log_lines(log_path)
+    assert entry["handled_status"] == HANDLED_FILTERED_SUBTYPE
+    assert entry["extra"]["event_type"] == "app_mention"
+
+
+@pytest.mark.asyncio
+async def test_channel_message_filtered(handler, log_path):
+    """channel_type='channel' (not im) — drop."""
+    await handler.handle_event(_make_payload(channel_type="channel"))
+    [entry] = _read_log_lines(log_path)
+    assert entry["handled_status"] == HANDLED_FILTERED_NON_IM
+
+
+@pytest.mark.asyncio
+async def test_group_channel_filtered(handler, log_path):
+    """channel_type='group' — drop."""
+    await handler.handle_event(_make_payload(channel_type="group"))
+    [entry] = _read_log_lines(log_path)
+    assert entry["handled_status"] == HANDLED_FILTERED_NON_IM
+
+
+# ---------------------------------------------------------------------------
+# Handler exception path
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_handler_internal_exception_returns_ok(
+    handler, log_path, monkeypatch
+):
+    """If _handle_event_inner raises, the outer handle_event catches +
+    logs handler_error + still returns 200 to Slack."""
+
+    async def _boom(self, payload):
+        raise RuntimeError("simulated handler failure")
+
+    monkeypatch.setattr(SlackDMHandler, "_handle_event_inner", _boom)
+
+    result = await handler.handle_event(_make_payload())
+    assert result == {"ok": True}
+
+    [entry] = _read_log_lines(log_path)
+    assert entry["handled_status"] == HANDLED_HANDLER_ERROR
+    assert "simulated handler failure" in entry["error"]
+
+
+@pytest.mark.asyncio
+async def test_malformed_payload_does_not_crash(handler, log_path):
+    """An event missing the inner ``event`` dict shouldn't crash —
+    filter logic returns the appropriate status."""
+    result = await handler.handle_event({"type": "event_callback"})
+    assert result == {"ok": True}
+    # No 'event' dict → all extractors return None → subtype filter
+    # path (event_type is None, treated as != 'message').
+    [entry] = _read_log_lines(log_path)
+    assert entry["handled_status"] == HANDLED_FILTERED_SUBTYPE
+
+
+# ---------------------------------------------------------------------------
+# JSONL format
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_jsonl_one_json_per_line(handler, log_path):
+    """Multiple events → multiple lines, each parseable as JSON."""
+    for i, user in enumerate([JOSHUA_ID, "USOMEONE", JOSHUA_ID]):
+        await handler.handle_event(
+            _make_payload(user=user, ts=f"170000000{i}.001")
+        )
+    raw = log_path.read_text(encoding="utf-8")
+    lines = raw.splitlines()
+    assert len(lines) == 3
+    for line in lines:
+        entry = json.loads(line)  # parse-or-raise
+        for required in ("received_at", "user_id", "text", "handled_status"):
+            assert required in entry
+
+
+@pytest.mark.asyncio
+async def test_jsonl_received_at_is_iso(handler, log_path):
+    from datetime import datetime
+
+    await handler.handle_event(_make_payload())
+    [entry] = _read_log_lines(log_path)
+    # Round-trip parse — no exception means it's a valid ISO timestamp.
+    datetime.fromisoformat(entry["received_at"])
+
+
+# ---------------------------------------------------------------------------
+# SECURITY — signing secret never logged
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_signing_secret_never_in_jsonl(handler, log_path, monkeypatch):
+    """Even though Joshua's text could theoretically contain the
+    secret, the handler MUST NOT write secret material via any of
+    its internal paths. The signing secret is verified-and-dropped
+    upstream in the listener; we assert by simulating multiple
+    events + checking the log file."""
+    secret_marker = "topsecret_signing_secret_DO_NOT_LOG_ME"
+    monkeypatch.setenv("KORA_SLACK_SIGNING_SECRET", secret_marker)
+
+    # Simulate a diverse set of events that exercise every code path.
+    for payload in [
+        _make_payload(),  # received
+        _make_payload(user="UOTHER"),  # filtered_non_joshua
+        _make_payload(bot_id="B"),  # filtered_bot
+        _make_payload(subtype="message_changed"),  # filtered_subtype
+        _make_payload(channel_type="channel"),  # filtered_non_im
+    ]:
+        await handler.handle_event(payload)
+
+    contents = log_path.read_text(encoding="utf-8")
+    assert secret_marker not in contents, (
+        "signing secret env value appeared in JSONL — handler must "
+        "NEVER log secret material"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Log path resolution
+# ---------------------------------------------------------------------------
+
+
+def test_log_path_env_override(monkeypatch, tmp_path):
+    """KORA_SLACK_DM_LOG_PATH env wins over the default."""
+    override = tmp_path / "custom.jsonl"
+    monkeypatch.setenv("KORA_SLACK_DM_LOG_PATH", str(override))
+    handler_obj = SlackDMHandler()
+    assert handler_obj._log_path == override
+
+
+def test_log_path_default_uses_kora_home(monkeypatch, tmp_path):
+    """No env override → resolves via kora_constants.get_kora_home()."""
+    monkeypatch.delenv("KORA_SLACK_DM_LOG_PATH", raising=False)
+    import kora_constants
+
+    monkeypatch.setattr(kora_constants, "get_kora_home", lambda: tmp_path)
+    handler_obj = SlackDMHandler()
+    assert handler_obj._log_path == tmp_path / "slack_dm_log.jsonl"