diff --git a/kora_cli/reasoning/__init__.py b/kora_cli/reasoning/__init__.py
new file mode 100644
index 000000000000..50100d313b2e
--- /dev/null
+++ b/kora_cli/reasoning/__init__.py
@@ -0,0 +1,8 @@
+"""Kora's reasoning surface — KR-FEAT-AI-RESPONSE-LOOP.
+
+The reasoning layer takes an inbound message (Slack DM today; email
++ MCP-driven later) + thread context + Kora's operational state +
+the cost-ladder rung, calls an LLM, and returns a response. Handler
+modules compose ``ReasoningEngine.respond(...)``; the engine wraps
+the SDK call + cost-ladder-aware model selection + audit.
+"""
diff --git a/kora_cli/reasoning/anthropic_engine.py b/kora_cli/reasoning/anthropic_engine.py
new file mode 100644
index 000000000000..3b264a3ac335
--- /dev/null
+++ b/kora_cli/reasoning/anthropic_engine.py
@@ -0,0 +1,458 @@
+"""AnthropicReasoningEngine — KR-FEAT-AI-RESPONSE-LOOP ST1.
+
+Implements :class:`kora_cli.reasoning.engine.ReasoningEngine` against
+Anthropic's Python SDK (``anthropic==0.86.0``, declared in the
+``[web]`` extra alongside fastapi/uvicorn/slowapi so daemon installs
+auto-pick it up).
+
+# Credential cascade
+
+Two supported credential sources (K-DG drift surfaced in the ST1
+PR body — bucket spec said ``KORA_ANTHROPIC_API_KEY`` only; the
+existing env mapping doc plus the gate-2 anti-secret block + the
+"Max plan via Agent SDK billing" framing in §1 imply
+``CLAUDE_CODE_OAUTH_TOKEN`` is the canonical credential):
+
+  1. ``KORA_ANTHROPIC_API_KEY`` (if set) → SDK constructed with
+     ``api_key=...``. Billing: Anthropic Console (operator must
+     provision an API key separately).
+  2. ``CLAUDE_CODE_OAUTH_TOKEN`` (fallback) → SDK constructed with
+     ``auth_token=...``. Billing: Max plan ($200/mo Agent SDK
+     pool). Existing Doppler ``kora-runtime-anthropic`` secret.
+
+Both unset → ``ReasoningEngineNotConfigured`` raised at
+construction. **Fail-CLOSED** per
+``feedback_fail_closed_by_default_security_infra``.
+
+# Cost-ladder model selection
+
+The active ``CostRung`` (read from the holder, mapped to the
+``ConversationContext.current_cost_ladder_rung`` string by the
+listener at call time) selects the model:
+
+  - ``"normal"`` → ``claude-opus-4-7``
+  - ``"warn_75"`` → ``claude-sonnet-4-6``
+  - ``"downshift_90"`` → ``claude-haiku-4-5-20251001``
+  - ``"hard_stop_100"`` → refuse with
+    ``ResponseResult(error="cost_ladder_halted")`` — no API call
+
+# Operational-state gating
+
+Kora's primary state is also surfaced via ``ConversationContext``.
+``paused`` / ``stopped`` → refuse with
+``error="operational_state_paused"``. The handler maps this to a
+canned acknowledgment so Joshua isn't met with silence during a
+pause.
+
+# Per-call timeout + retry
+
+60s per-call timeout. **NO retry on 5xx** (per PM Q3 default —
+preserves the cost-ladder budget; retries burn tokens). Single
+attempt; failure returns a ResponseResult with the appropriate
+``error`` code.
+
+# Credential sanitization
+
+The token NEVER appears in logs, error messages, or
+``ResponseResult.error`` codes. A diverse-failure test
+(``test_anthropic_engine.py::test_credential_never_in_errors``)
+exercises 401 / 429 / 500 / timeout / network-error paths and
+asserts the token's env value doesn't appear in any surface.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import time
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+from kora_cli.reasoning.engine import (
+    ConversationContext,
+    ConversationTurn,
+    IncomingMessage,
+    ResponseResult,
+)
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Constants
+# ---------------------------------------------------------------------------
+
+API_KEY_ENV = "KORA_ANTHROPIC_API_KEY"
+OAUTH_TOKEN_ENV = "CLAUDE_CODE_OAUTH_TOKEN"
+SYSTEM_PROMPT_PATH_ENV = "KORA_SYSTEM_PROMPT_PATH"
+
+DEFAULT_SYSTEM_PROMPT_PATH = (
+    Path(__file__).resolve().parents[2]
+    / "kora_docs"
+    / "00_canonical_current_state"
+    / "kora_system_prompt.md"
+)
+
+DEFAULT_TIMEOUT_SECONDS = 60.0
+DEFAULT_MAX_OUTPUT_TOKENS = 2048
+
+# Model identifiers per the canonical Claude 4.X family. Values
+# verified against the project_kora memory + the latest SDK docs.
+MODEL_OPUS = "claude-opus-4-7"
+MODEL_SONNET = "claude-sonnet-4-6"
+MODEL_HAIKU = "claude-haiku-4-5-20251001"
+
+# Cost-ladder rung → model mapping. Bucket spec maps the four rungs
+# to {opus, sonnet, haiku, halt}. Keyed by the CostRung enum's
+# canonical ``.value`` strings (NOT the bucket-spec paraphrases).
+RUNG_MODEL_MAP: Dict[str, str] = {
+    "normal": MODEL_OPUS,
+    "warn_75": MODEL_SONNET,
+    "downshift_90": MODEL_HAIKU,
+    # "hard_stop_100" is special-cased — refuse, no API call.
+}
+
+
+# ---------------------------------------------------------------------------
+# Errors
+# ---------------------------------------------------------------------------
+
+
+class ReasoningEngineError(RuntimeError):
+    """Base class for engine construction / config failures."""
+
+
+class ReasoningEngineNotConfigured(ReasoningEngineError):
+    """Both credential envs unset. Operator must provision either
+    ``KORA_ANTHROPIC_API_KEY`` (Console billing) OR
+    ``CLAUDE_CODE_OAUTH_TOKEN`` (Max plan billing) before the
+    daemon can register the reasoning listener."""
+
+
+class ReasoningSystemPromptError(ReasoningEngineError):
+    """``kora_system_prompt.md`` missing or unreadable. Fail-CLOSED
+    at engine construction — daemon won't register the reasoning
+    listener until the prompt loads cleanly."""
+
+
+# ---------------------------------------------------------------------------
+# Engine
+# ---------------------------------------------------------------------------
+
+
+class AnthropicReasoningEngine:
+    """Concrete ReasoningEngine backed by the anthropic SDK."""
+
+    def __init__(
+        self,
+        *,
+        timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
+        max_output_tokens: int = DEFAULT_MAX_OUTPUT_TOKENS,
+        system_prompt_path: Optional[Path] = None,
+        # Test seam — inject a pre-configured Anthropic client OR a
+        # stand-in (anything with an async ``messages.create``).
+        client: Optional[Any] = None,
+    ) -> None:
+        # Credential cascade. Read once at construction (production
+        # rotation pattern: redeploy, not hot-reload — same as
+        # SlackClient's bot-token model).
+        api_key = os.environ.get(API_KEY_ENV, "").strip() or None
+        oauth_token = os.environ.get(OAUTH_TOKEN_ENV, "").strip() or None
+        if not api_key and not oauth_token:
+            raise ReasoningEngineNotConfigured(
+                f"both {API_KEY_ENV} and {OAUTH_TOKEN_ENV} are unset — "
+                "daemon cannot reason. Set one in Doppler "
+                "(kora-runtime-anthropic project). See env-mapping doc."
+            )
+        self._auth_mode: str = "api_key" if api_key else "oauth_token"
+        # Stored under underscore-prefixed attrs to discourage casual
+        # serialization. NEVER logged.
+        self._api_key = api_key
+        self._oauth_token = oauth_token
+
+        # System prompt — fail-CLOSED on missing/unreadable.
+        prompt_path = system_prompt_path or _resolve_system_prompt_path()
+        try:
+            self._system_prompt = prompt_path.read_text(encoding="utf-8")
+        except OSError as exc:
+            raise ReasoningSystemPromptError(
+                f"system prompt unreadable at {prompt_path}: {exc!r}"
+            ) from exc
+        if not self._system_prompt.strip():
+            raise ReasoningSystemPromptError(
+                f"system prompt at {prompt_path} is empty"
+            )
+
+        self._timeout = timeout_seconds
+        self._max_output_tokens = max_output_tokens
+        self._client = client  # lazy-constructed in respond() if None
+        self._client_close_method: Optional[Any] = None
+
+    # ------------------------------------------------------------------
+    # Public — ReasoningEngine protocol
+    # ------------------------------------------------------------------
+
+    async def respond(
+        self,
+        message: IncomingMessage,
+        context: ConversationContext,
+    ) -> ResponseResult:
+        """Main entry. See :class:`ReasoningEngine.respond`."""
+        started_at = time.monotonic()
+
+        # Refuse-paths first — these don't call the SDK.
+        if context.current_operational_state in ("paused", "stopped"):
+            return ResponseResult(
+                text="",
+                model_used="",
+                input_tokens=0,
+                output_tokens=0,
+                reasoning_duration_ms=_elapsed_ms(started_at),
+                error="operational_state_paused",
+            )
+        rung = context.current_cost_ladder_rung
+        if rung == "hard_stop_100":
+            return ResponseResult(
+                text="",
+                model_used="",
+                input_tokens=0,
+                output_tokens=0,
+                reasoning_duration_ms=_elapsed_ms(started_at),
+                error="cost_ladder_halted",
+            )
+        model = RUNG_MODEL_MAP.get(rung)
+        if model is None:
+            # Unknown rung (e.g. "unknown") — default to OPUS but log
+            # a WARN. The cost-ladder listener should always inject a
+            # valid rung; this path covers misconfiguration.
+            logger.warning(
+                "[kora.reasoning] unknown cost rung %r — defaulting to "
+                "opus + continuing",
+                rung,
+            )
+            model = MODEL_OPUS
+
+        # Assemble the message list. Anthropic SDK expects:
+        #   [{role: "user"|"assistant", content: "..."}]
+        # Map ConversationTurn.direction → role. The fresh
+        # IncomingMessage goes at the end as the latest user turn.
+        messages = self._build_message_history(message, context)
+
+        # SDK call. Single attempt — NO retry per PM Q3 default.
+        client = await self._ensure_client()
+        try:
+            response = await client.messages.create(
+                model=model,
+                system=self._system_prompt,
+                messages=messages,
+                max_tokens=self._max_output_tokens,
+                timeout=self._timeout,
+            )
+        except Exception as exc:
+            return self._map_sdk_exception(
+                exc, model=model, started_at=started_at
+            )
+
+        # Extract reply text + token counts.
+        return self._project_response(
+            response, model=model, started_at=started_at
+        )
+
+    async def close(self) -> None:
+        """Close any open HTTP client. Idempotent."""
+        if self._client is None:
+            return
+        # The Anthropic SDK's AsyncClient has a close() coroutine.
+        close = getattr(self._client, "close", None)
+        if close is not None:
+            try:
+                result = close()
+                if hasattr(result, "__await__"):
+                    await result
+            except Exception as exc:
+                logger.warning(
+                    "[kora.reasoning] close raised %r — continuing", exc
+                )
+
+    # ------------------------------------------------------------------
+    # Internals
+    # ------------------------------------------------------------------
+
+    async def _ensure_client(self) -> Any:
+        if self._client is not None:
+            return self._client
+        # Import + construct here (not at module-import) so daemon
+        # `kora --help` stays fast + tests that mock-out the SDK
+        # never see the real client created.
+        from anthropic import AsyncAnthropic
+
+        if self._api_key:
+            self._client = AsyncAnthropic(
+                api_key=self._api_key, timeout=self._timeout
+            )
+        else:
+            # OAuth token via the SDK's ``auth_token`` constructor
+            # arg (Max plan billing path).
+            self._client = AsyncAnthropic(
+                auth_token=self._oauth_token, timeout=self._timeout
+            )
+        return self._client
+
+    def _build_message_history(
+        self,
+        message: IncomingMessage,
+        context: ConversationContext,
+    ) -> List[Dict[str, str]]:
+        """Convert ConversationContext.recent_messages + the fresh
+        IncomingMessage into the SDK's ``messages`` list shape.
+
+        Direction → role:
+          - ``inbound`` → ``"user"`` (Joshua said it)
+          - ``outbound`` → ``"assistant"`` (Kora said it)
+
+        Order: oldest → newest. The latest item is the FRESH
+        ``IncomingMessage`` as a final user turn — that's what
+        Kora is responding to.
+
+        Constraint: Anthropic API requires alternating user /
+        assistant turns. If the loader returned non-alternating
+        history (rare but possible after race conditions), we
+        collapse consecutive same-role turns by concatenation
+        so the SDK doesn't reject.
+        """
+        history: List[Dict[str, str]] = []
+        for turn in context.recent_messages:
+            role = "user" if turn.direction == "inbound" else "assistant"
+            if history and history[-1]["role"] == role:
+                # Same-role consecutive — concatenate.
+                history[-1]["content"] = (
+                    history[-1]["content"] + "\n\n" + turn.text
+                )
+            else:
+                history.append({"role": role, "content": turn.text})
+
+        # Append fresh inbound. If the last history turn is already
+        # "user", concatenate (covers the case where the loader
+        # included the fresh inbound in the history slice).
+        if history and history[-1]["role"] == "user":
+            history[-1]["content"] = (
+                history[-1]["content"] + "\n\n" + message.text
+            )
+        else:
+            history.append({"role": "user", "content": message.text})
+
+        return history
+
+    def _project_response(
+        self,
+        response: Any,
+        *,
+        model: str,
+        started_at: float,
+    ) -> ResponseResult:
+        """Map an SDK ``Message`` response → ``ResponseResult``."""
+        # SDK shape: response.content is a list of content blocks;
+        # for text-only responses, content[0].text is the reply.
+        text_parts: List[str] = []
+        try:
+            for block in response.content:
+                # Anthropic SDK: TextBlock has .type=="text" + .text.
+                if getattr(block, "type", "") == "text":
+                    text_parts.append(getattr(block, "text", "") or "")
+        except Exception as exc:
+            logger.warning(
+                "[kora.reasoning] response content projection failed: %r",
+                exc,
+            )
+            return ResponseResult(
+                text="",
+                model_used=getattr(response, "model", model) or model,
+                input_tokens=0,
+                output_tokens=0,
+                reasoning_duration_ms=_elapsed_ms(started_at),
+                error="response_projection_failed",
+            )
+
+        text = "".join(text_parts).strip()
+
+        usage = getattr(response, "usage", None)
+        input_tokens = int(getattr(usage, "input_tokens", 0) or 0)
+        output_tokens = int(getattr(usage, "output_tokens", 0) or 0)
+
+        return ResponseResult(
+            text=text,
+            model_used=getattr(response, "model", model) or model,
+            input_tokens=input_tokens,
+            output_tokens=output_tokens,
+            reasoning_duration_ms=_elapsed_ms(started_at),
+            error=None,
+        )
+
+    def _map_sdk_exception(
+        self,
+        exc: BaseException,
+        *,
+        model: str,
+        started_at: float,
+    ) -> ResponseResult:
+        """Map an Anthropic SDK exception → stable ``error`` code.
+
+        NEVER includes the credential in the error code — only the
+        exception class name + an HTTP-status hint when available.
+        """
+        # Avoid hard-importing anthropic.* exception classes at module
+        # top (lazy-loaded by _ensure_client). Use duck-typing on
+        # attribute presence.
+        status_code = getattr(exc, "status_code", None)
+        exc_name = type(exc).__name__
+
+        if status_code == 401 or status_code == 403:
+            error = "sdk_auth"
+        elif status_code == 429:
+            error = "sdk_rate_limited"
+        elif status_code is not None and 500 <= int(status_code) < 600:
+            error = "sdk_5xx"
+        elif status_code is not None and 400 <= int(status_code) < 500:
+            error = f"sdk_4xx_{status_code}"
+        elif exc_name in ("APITimeoutError", "Timeout", "TimeoutException"):
+            error = "sdk_timeout"
+        elif "Connection" in exc_name or "Transport" in exc_name:
+            error = "sdk_transport"
+        else:
+            # Generic — log the exception class + a sanitized
+            # message snippet (NEVER the credential).
+            error = f"sdk_unknown_{exc_name}"
+
+        logger.warning(
+            "[kora.reasoning] SDK call failed model=%s error=%s "
+            "exc_type=%s",
+            model,
+            error,
+            exc_name,
+        )
+        return ResponseResult(
+            text="",
+            model_used=model,
+            input_tokens=0,
+            output_tokens=0,
+            reasoning_duration_ms=_elapsed_ms(started_at),
+            error=error,
+        )
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _resolve_system_prompt_path() -> Path:
+    """Env override → in-repo default. Mirrors the SlackDMHandler
+    log-path pattern."""
+    override = os.environ.get(SYSTEM_PROMPT_PATH_ENV, "").strip()
+    if override:
+        return Path(override)
+    return DEFAULT_SYSTEM_PROMPT_PATH
+
+
+def _elapsed_ms(started_at: float) -> int:
+    return int((time.monotonic() - started_at) * 1000)
diff --git a/kora_cli/reasoning/context_loader.py b/kora_cli/reasoning/context_loader.py
new file mode 100644
index 000000000000..164642c658c4
--- /dev/null
+++ b/kora_cli/reasoning/context_loader.py
@@ -0,0 +1,255 @@
+"""Conversation context loader — KR-FEAT-AI-RESPONSE-LOOP ST1.
+
+Reads ``${HERMES_HOME}/slack_dm_log.jsonl`` (or
+``KORA_SLACK_DM_LOG_PATH`` override), slices to a target Slack
+channel + thread, and projects the most-recent N entries into a
+:class:`ConversationContext` the reasoning engine consumes.
+
+# JSONL schema reminder
+
+Per KR-FEAT-SLACK-DM ST1 + ST2, two entry shapes coexist:
+
+  - **Inbound**: ``{received_at, channel_id, thread_ts, user_id,
+    text, event_ts, handled_status, ...}``
+  - **Outbound**: ``{sent_at, channel_id, thread_ts, text,
+    slack_message_ts, send_status, ...}``
+
+Distinguished by ``received_at`` vs ``sent_at`` key presence.
+Inbound entries with ``handled_status != "received"`` (filter
+drops, state drops, errors) are skipped — only successfully-
+processed Joshua DMs become conversation turns.
+
+# Thread matching
+
+Same Slack thread = same ``channel_id`` AND same ``thread_ts``.
+A DM that's not in a thread (no ``thread_ts``) matches other
+DMs without thread_ts in the same channel — Slack's IM channels
+don't typically have threads, so this is the common case.
+
+# Operational + cost state injection
+
+The loader reads the holders' singletons at call time and
+projects them into :class:`ConversationContext`'s
+``current_operational_state`` + ``current_cost_ladder_rung``
+fields. The reasoning engine consumes the canonical string
+values (NOT the enum types) — keeps the engine module pure.
+
+# Failure modes
+
+  - JSONL missing → empty context (no turns; state strings
+    "unknown"). The reasoning engine still runs; it just lacks
+    history.
+  - JSONL unreadable / malformed lines → WARN-log + skip the
+    bad line; continue with valid lines.
+  - Holder modules uninitialized → context strings are "unknown";
+    the engine defaults to OPUS + skips the paused-state gate.
+    Production daemon always has both holders initialized; this
+    fall-through covers test paths.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+from kora_cli.reasoning.engine import (
+    ConversationContext,
+    ConversationTurn,
+)
+
+logger = logging.getLogger(__name__)
+
+
+# Mirror the env name + default-path resolution from
+# kora_cli/handlers/slack_dm_handler.py so the loader points at
+# the same file the handler writes to.
+LOG_PATH_ENV = "KORA_SLACK_DM_LOG_PATH"
+
+DEFAULT_MAX_TURNS = 10
+
+
+def _resolve_log_path() -> Path:
+    override = os.environ.get(LOG_PATH_ENV, "").strip()
+    if override:
+        return Path(override)
+    from kora_constants import get_kora_home
+
+    return get_kora_home() / "slack_dm_log.jsonl"
+
+
+def _parse_iso(value: Any) -> Optional[datetime]:
+    """Defensive ISO 8601 parser. Returns None on any failure."""
+    if not isinstance(value, str) or not value:
+        return None
+    try:
+        return datetime.fromisoformat(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def _match_thread(
+    entry: Dict[str, Any],
+    *,
+    channel_id: str,
+    thread_ts: Optional[str],
+) -> bool:
+    """A JSONL entry belongs to the target thread iff:
+    - same channel_id
+    - same thread_ts (None matches None; non-None matches exact)
+    """
+    if entry.get("channel_id") != channel_id:
+        return False
+    entry_thread = entry.get("thread_ts")
+    if thread_ts is None:
+        return entry_thread is None
+    return entry_thread == thread_ts
+
+
+def _current_operational_state_str() -> str:
+    """Best-effort resolution of the canonical PrimaryState string."""
+    try:
+        from agent.operational_state_holder import get_holder
+    except Exception:
+        return "unknown"
+    holder = get_holder()
+    if holder is None:
+        return "unknown"
+    try:
+        # .current is a @property — caught in KR-MCP-RUNTIME-SURFACE ST1.
+        return holder.current.primary_state.value
+    except Exception:
+        return "unknown"
+
+
+def _current_cost_rung_str() -> str:
+    """Best-effort resolution of the canonical CostRung string."""
+    try:
+        from agent.cost_state_holder import get_cost_holder
+    except Exception:
+        return "unknown"
+    holder = get_cost_holder()
+    if holder is None:
+        return "unknown"
+    try:
+        # active_rung is a method (NOT @property) per K-DG check.
+        return holder.active_rung().value
+    except Exception:
+        return "unknown"
+
+
+def load_slack_dm_context(
+    *,
+    channel_id: str,
+    thread_ts: Optional[str],
+    max_turns: int = DEFAULT_MAX_TURNS,
+    log_path: Optional[Path] = None,
+) -> ConversationContext:
+    """Load up to ``max_turns`` most-recent turns in the target thread.
+
+    Args:
+      channel_id: Slack channel ID (e.g. ``D01ABC...``). Required.
+      thread_ts: Slack thread timestamp. ``None`` for non-threaded DMs.
+      max_turns: Cap on returned turn count. Default 10 (5 in + 5 out
+        in the typical alternating case; the loader doesn't enforce
+        alternation — caller is responsible if SDK requires it).
+      log_path: Override for tests. ``None`` → resolves via
+        ``KORA_SLACK_DM_LOG_PATH`` env or ``get_kora_home()``.
+
+    Returns:
+      A ConversationContext with up to ``max_turns`` turns
+      (oldest→newest) + current operational state + current cost
+      rung. On missing/unreadable JSONL returns an empty context
+      with state strings ``"unknown"``.
+    """
+    path = log_path or _resolve_log_path()
+    turns: List[ConversationTurn] = []
+
+    try:
+        raw = path.read_text(encoding="utf-8")
+    except FileNotFoundError:
+        return ConversationContext(
+            recent_messages=[],
+            current_operational_state=_current_operational_state_str(),
+            current_cost_ladder_rung=_current_cost_rung_str(),
+        )
+    except OSError as exc:
+        logger.warning(
+            "[kora.reasoning] context loader: %s unreadable: %r — "
+            "continuing with empty history",
+            path,
+            exc,
+        )
+        return ConversationContext(
+            recent_messages=[],
+            current_operational_state=_current_operational_state_str(),
+            current_cost_ladder_rung=_current_cost_rung_str(),
+        )
+
+    for lineno, line in enumerate(raw.splitlines(), start=1):
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            entry = json.loads(line)
+        except json.JSONDecodeError as exc:
+            logger.warning(
+                "[kora.reasoning] context loader: %s line %d malformed: %r",
+                path,
+                lineno,
+                exc,
+            )
+            continue
+        if not isinstance(entry, dict):
+            continue
+
+        if not _match_thread(entry, channel_id=channel_id, thread_ts=thread_ts):
+            continue
+
+        # Inbound? (has received_at + handled_status)
+        if "received_at" in entry and entry.get("handled_status") == "received":
+            at = _parse_iso(entry.get("received_at")) or _epoch_dt()
+            text = str(entry.get("text") or "")
+            if text:
+                turns.append(
+                    ConversationTurn(direction="inbound", text=text, at=at)
+                )
+            continue
+
+        # Outbound? (has sent_at + send_status == "ok")
+        if "sent_at" in entry and entry.get("send_status") == "ok":
+            at = _parse_iso(entry.get("sent_at")) or _epoch_dt()
+            text = str(entry.get("text") or "")
+            if text:
+                turns.append(
+                    ConversationTurn(direction="outbound", text=text, at=at)
+                )
+            continue
+
+        # Filtered / failed / dropped entries are skipped (they
+        # weren't part of Kora's reasoning history).
+
+    # Sort by timestamp (oldest first), keep last N. We sort
+    # defensively because JSONL append-order MAY occasionally race
+    # against the wall-clock ordering (rare; outbound entry's
+    # sent_at is set inside _send_echo_reply, after the inbound
+    # received_at, but log writes are buffered).
+    turns.sort(key=lambda t: t.at)
+    if len(turns) > max_turns:
+        turns = turns[-max_turns:]
+
+    return ConversationContext(
+        recent_messages=turns,
+        current_operational_state=_current_operational_state_str(),
+        current_cost_ladder_rung=_current_cost_rung_str(),
+    )
+
+
+def _epoch_dt() -> datetime:
+    """Defensive fallback timestamp when a JSONL entry has a
+    malformed ``received_at`` / ``sent_at``. Sorts to the start of
+    history so the bad entry doesn't take precedence over good ones."""
+    return datetime.fromtimestamp(0, tz=timezone.utc)
diff --git a/kora_cli/reasoning/engine.py b/kora_cli/reasoning/engine.py
new file mode 100644
index 000000000000..f4c1238a671a
--- /dev/null
+++ b/kora_cli/reasoning/engine.py
@@ -0,0 +1,202 @@
+"""ReasoningEngine protocol + canonical value classes.
+
+Defines the abstract surface a Kora reasoning implementation must
+expose. ST1 ships :class:`AnthropicReasoningEngine` against this
+protocol (``kora_cli/reasoning/anthropic_engine.py``); future
+buckets may swap providers (Bedrock / Vertex / a different model
+family) without touching the handler-side call site.
+
+# Why a Protocol vs an ABC
+
+Protocol composes more cleanly with the existing test seam (the
+handler accepts an injected engine instance; tests pass a
+``MagicMock(spec=ReasoningEngine)`` and don't need to subclass).
+ABCs would force a synthetic test class per case.
+
+# Field-naming convention (K-DG locked)
+
+Every typed structure in this module ships its actual field names
+verbatim in this docstring + the dataclass definitions. Per the
+2026-05-22 PM-locked standing rule
+(``feedback_k_dg_substrate_field_names_in_specs``), the bucket
+spec's paraphrased shapes are NOT authoritative; this module is.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Any, Awaitable, Callable, Dict, List, Literal, Optional, Protocol
+
+
+# ---------------------------------------------------------------------------
+# Source enum (literal, not Enum — keeps JSON serialization trivial)
+# ---------------------------------------------------------------------------
+
+
+MessageSource = Literal["slack_dm", "email", "mcp"]
+
+# Cost-ladder rung as Kora's reasoning sees it. These are the
+# ``.value`` strings of ``agent.cost_state_holder.CostRung`` — the
+# canonical-name mapping is one-shot at the listener boundary so the
+# rest of the reasoning code consumes the canonical strings.
+#
+# Per K-DG check on KR-P2-K cascade: actual CostRung values are
+# "normal" / "warn_75" / "downshift_90" / "hard_stop_100"
+# (NOT the bucket spec's "normal" / "warned" / "constrained" / "halted").
+CostLadderRungName = Literal[
+    "normal", "warn_75", "downshift_90", "hard_stop_100", "unknown"
+]
+
+
+# ---------------------------------------------------------------------------
+# Value classes — inbound side
+# ---------------------------------------------------------------------------
+
+
+@dataclass(frozen=True, slots=True)
+class IncomingMessage:
+    """A message arriving at Kora's reasoning layer.
+
+    Fields:
+
+      - ``text``: full inbound message text (NO truncation here;
+        the reasoning engine may want context).
+      - ``source``: which surface the message arrived on.
+      - ``received_at``: wall-clock when the listener accepted it.
+        UTC-tagged. The handler sets this from inbound JSONL's
+        ``received_at`` (round-trippable ISO 8601).
+      - ``metadata``: source-specific bag. For ``slack_dm``:
+        ``{channel_id, thread_ts, user_id, event_ts}``. For
+        ``email``: ``{subject, from, message_id}``. For ``mcp``:
+        ``{caller_actor_kind, tool_name}``. Stable per-source
+        schema — but typed as ``dict[str, Any]`` to keep this
+        module source-agnostic.
+    """
+
+    text: str
+    source: MessageSource
+    received_at: datetime
+    metadata: Dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(frozen=True, slots=True)
+class ConversationTurn:
+    """One prior message in the same thread, in chronological order.
+
+    The context loader pulls these from the inbound/outbound JSONL
+    sliced to (channel_id, thread_ts) + most-recent N.
+    """
+
+    direction: Literal["inbound", "outbound"]
+    text: str
+    at: datetime
+
+
+@dataclass(frozen=True, slots=True)
+class ConversationContext:
+    """Per-call context the reasoning engine consumes alongside the
+    fresh ``IncomingMessage``.
+
+    Fields:
+
+      - ``recent_messages``: prior turns in the SAME thread,
+        oldest→newest. Bounded at the loader (default 10) — the
+        engine may further truncate if its token budget requires.
+      - ``current_operational_state``: lowercase string of the
+        active ``PrimaryState`` (``"booting"`` / ``"ready"`` /
+        ``"active"`` / ``"paused"`` / ``"stopped"`` — verified
+        against ``agent/operational_state.py:70-74``).
+      - ``current_cost_ladder_rung``: lowercase string of the active
+        ``CostRung`` (see ``CostLadderRungName``).
+      - ``extra``: future-extension bag — currently unused; engine
+        impls may ignore.
+    """
+
+    recent_messages: List[ConversationTurn] = field(default_factory=list)
+    current_operational_state: str = "unknown"
+    current_cost_ladder_rung: CostLadderRungName = "unknown"
+    extra: Dict[str, Any] = field(default_factory=dict)
+
+
+# ---------------------------------------------------------------------------
+# Value classes — response side
+# ---------------------------------------------------------------------------
+
+
+@dataclass(frozen=True, slots=True)
+class ResponseResult:
+    """The reasoning engine's output.
+
+    The handler checks ``error`` first; if non-None, the engine
+    failed (cost-ladder halt / paused / SDK exception / etc.). The
+    handler falls back to a canned response and records the error
+    in JSONL.
+
+    On success (``error is None``):
+
+      - ``text``: the response body to send back via the source's
+        outbound channel.
+      - ``model_used``: actual model string from the SDK response
+        (e.g. ``"claude-opus-4-7"``). Cost-ladder downshift means
+        this may be a different model than the ladder ``normal``
+        default.
+      - ``input_tokens`` + ``output_tokens``: from the SDK response
+        usage block; handler passes these into the cost-ladder
+        ``record_inference()`` write.
+      - ``reasoning_duration_ms``: wall-clock from the engine's
+        own timer (NOT the SDK's; includes prompt assembly +
+        network).
+
+    On failure (``error`` set):
+
+      - ``error``: short machine-readable code:
+        ``"cost_ladder_halted"`` / ``"operational_state_paused"`` /
+        ``"sdk_timeout"`` / ``"sdk_auth"`` / ``"sdk_rate_limited"``
+        / ``"sdk_5xx"`` / ``"sdk_4xx_<code>"`` /
+        ``"sdk_transport"`` / ``"unconfigured"``.
+      - Other fields may be partially populated (e.g.
+        ``reasoning_duration_ms`` is set even on failure to surface
+        slow-failure paths).
+    """
+
+    text: str
+    model_used: str
+    input_tokens: int
+    output_tokens: int
+    reasoning_duration_ms: int
+    error: Optional[str] = None
+
+
+# ---------------------------------------------------------------------------
+# Protocol — what a reasoning engine must implement
+# ---------------------------------------------------------------------------
+
+
+class ReasoningEngine(Protocol):
+    """The single method handlers call.
+
+    Implementations:
+
+      - Read the cost-ladder rung + operational state from
+        ``context`` (NOT directly from the holders — keeps the
+        engine pure-functional + testable; listener boundary
+        wraps the holders).
+      - Refuse to call when the rung is ``hard_stop_100`` or the
+        operational state is ``paused`` / ``stopped`` (returns a
+        ResponseResult with ``error`` set; the handler falls back).
+      - Sanitize all errors so credentials don't leak.
+
+    ``respond`` is the only required method. ``close`` is optional
+    (defaults to a no-op via the Protocol's structural typing).
+    """
+
+    async def respond(
+        self,
+        message: IncomingMessage,
+        context: ConversationContext,
+    ) -> ResponseResult:
+        ...
+
+    async def close(self) -> None:
+        ...
diff --git a/kora_docs/00_canonical_current_state/kora_system_prompt.md b/kora_docs/00_canonical_current_state/kora_system_prompt.md
new file mode 100644
index 000000000000..cdf94a2aaa1c
--- /dev/null
+++ b/kora_docs/00_canonical_current_state/kora_system_prompt.md
@@ -0,0 +1,127 @@
+# Kora — system prompt
+
+**Purpose**: the identity + behavioral envelope Kora's reasoning
+engine wraps around every inference call. This file is loaded
+ONCE at daemon startup and prepended (as the Anthropic API's
+``system`` field) to every reasoning request.
+
+**This file is operator-editable**. Joshua iterates over time;
+edits take effect at the next daemon restart (no hot-reload —
+deliberate, so prompt churn is observable in the deploy ledger).
+
+**Source-of-truth pointer**: this file. The reasoning engine
+imports it via ``kora_cli/reasoning/anthropic_engine.py``'s
+``_load_system_prompt`` at construction. A missing/unreadable file
+fails-CLOSED — the daemon refuses to register the reasoning
+listener if the prompt can't be loaded.
+
+---
+
+You are Kora. You are Joshua's digital extension — not a generic
+assistant, not a chatbot, not "an AI." You are HIS specifically.
+This is the identity envelope every response of yours lives in.
+
+## Who you are
+
+You are a frontier-class agent built on the Hermes-fork runtime,
+backed by the IsoKron substrate. You run as a long-running daemon
+(`kora daemon`) on Fly.io. You receive messages via Slack DMs from
+Joshua, via inbound email (forthcoming), and via the agent-facing
+MCP surface (other PMs / drone tickets calling `kora__*` tools).
+Today you reply via Slack DM; outbound channels expand as the
+operator wires them.
+
+Your purpose is to think on Joshua's behalf and surface useful
+output. NOT to be friendly. NOT to summarize what you've just
+said. NOT to perform thoughtfulness. To be USEFUL.
+
+## How you respond
+
+- **Brevity by default**. Slack DM is your primary channel; long
+  responses don't fit the medium. One to three sentences for most
+  exchanges. Expand only when Joshua explicitly asks for depth
+  ("walk me through this", "give me the full picture", etc.) OR
+  when the question genuinely requires a multi-step answer that
+  brevity would mangle.
+- **No preamble, no sycophancy**. Don't say "Great question" or
+  "Let me think about that." Don't restate the question. Start
+  with the answer.
+- **Direct, honest tone**. If you don't know, say "I don't know" +
+  the shape of what you'd need to find out. If something Joshua
+  said is wrong, say so + why. Don't hedge.
+- **Concrete over abstract**. Examples, file paths, specific
+  values. Avoid "perhaps consider X" — say "do X" or "X breaks
+  because Y".
+- **No emoji** in responses unless Joshua used them first or asked
+  for them. Slack DMs are work surface.
+- **Code blocks** for code; inline backticks for `identifiers`.
+
+## What you know about Joshua
+
+- He's the operator + sole consumer of your output. He pays for
+  you (Anthropic Max plan). He installed you.
+- He uses you alongside Claude Code (this CLI) + multiple other
+  agents (other PMs). You are part of his agent fleet, not his
+  only thinking partner.
+- He's an experienced software engineer + system designer; you
+  can assume technical fluency.
+- His broader project is building Kora itself — meta-recursive
+  but normal. Don't comment on it.
+
+## Operational boundaries
+
+- **Respect the operational state machine**. If you've been put
+  into `paused` or `stopped`, you do NOT reason — the reasoning
+  engine refuses the call before reaching you. But if Joshua DMs
+  you DURING a pause, the engine surfaces that to you (via the
+  context) — acknowledge briefly, don't try to "work around" the
+  pause.
+- **Respect the cost ladder**. The reasoning engine's model
+  selection is downstream of your context (`current_cost_ladder_rung`).
+  When you're running on a downshifted model (Sonnet / Haiku
+  instead of Opus), respond in a way that fits the model's
+  capability — don't pretend to do reasoning the smaller model
+  can't.
+- **Don't claim certainty about non-substrate state**. If asked
+  about something you can't verify (the contents of a file, the
+  state of a remote service, what someone else did), say what you
+  CAN verify + what you'd need to verify the rest.
+
+## Memory + context
+
+- You see the last 10 turns of THIS thread (5 inbound + 5
+  outbound). You don't see other threads — each Slack DM
+  conversation is a separate universe.
+- You don't have access to substrate state directly in this
+  loop — no `kora_operation_ledger`, no chain events, no
+  `kora_control`. Future buckets (`KR-FEAT-AGENTIC-REASONING`)
+  will let you call tools mid-reasoning; until then, work with
+  what's in the thread.
+- You don't have access to file content, terminal output, or
+  any other external state. If Joshua references a file or a
+  command, ask for it OR proceed under the explicit assumption
+  he'll evaluate your suggestion himself.
+
+## When you don't have an answer
+
+- "I don't know" is a complete sentence. Say it when true.
+- If you'd need to look something up to answer, say what you'd
+  look up + offer to do so when the tool surface exists.
+- Don't make up file paths, function names, or API shapes — if
+  unsure, say so + name what would resolve the uncertainty.
+
+## What you are NOT
+
+- Not Claude (the public assistant). Don't reference yourself as
+  Claude. You're Kora, running on Claude.
+- Not a help desk. You don't have a "limitations" section to
+  enumerate. You have the limits of your context + your model;
+  acknowledge them in line when they bite.
+- Not a tool wrapper. You think. The tool surface exists so you
+  can act on what you think.
+
+---
+
+End of system prompt. The next message in the inference call is
+the inbound message + conversation context the reasoning engine
+assembles around it.
diff --git a/pyproject.toml b/pyproject.toml
index d65cc6b1b7e4..1a3184398f67 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -182,7 +182,7 @@ youtube = [
   "youtube-transcript-api==1.2.4",
 ]
 # `hermes dashboard` (localhost SPA + API).  Not in core to keep the default install lean.
-web = ["fastapi==0.133.1", "uvicorn[standard]==0.41.0", "slowapi>=0.1.9"]
+web = ["fastapi==0.133.1", "uvicorn[standard]==0.41.0", "slowapi>=0.1.9", "anthropic==0.86.0"]
 all = [
   # Policy (2026-05-12): `[all]` includes only extras that genuinely
   # CAN'T be lazy-installed via `tools/lazy_deps.py` — i.e. things every
diff --git a/tests/kora_cli/reasoning/__init__.py b/tests/kora_cli/reasoning/__init__.py
new file mode 100644
index 000000000000..e69de29bb2d1
diff --git a/tests/kora_cli/reasoning/test_anthropic_engine.py b/tests/kora_cli/reasoning/test_anthropic_engine.py
new file mode 100644
index 000000000000..9802a41f20f2
--- /dev/null
+++ b/tests/kora_cli/reasoning/test_anthropic_engine.py
@@ -0,0 +1,599 @@
+"""Tests for ``kora_cli.reasoning.anthropic_engine`` —
+KR-FEAT-AI-RESPONSE-LOOP ST1.
+
+Covers:
+  - Construction: fail-CLOSED on missing both creds; succeeds with
+    API key, succeeds with OAuth token; API key wins over OAuth
+  - System prompt: fail-CLOSED on missing/empty/unreadable file
+  - Cost-ladder model selection: NORMAL→opus, WARN_75→sonnet,
+    DOWNSHIFT_90→haiku, HARD_STOP_100→refuse
+  - Operational-state gating: paused / stopped → refuse
+  - Successful API call → ResponseResult with text + tokens
+  - Message history projection: alternating roles + concat on
+    consecutive same-role
+  - Fresh inbound concatenates to last history user turn if same role
+  - SDK error mapping: 401 → sdk_auth, 429 → sdk_rate_limited,
+    500 → sdk_5xx, 4xx → sdk_4xx_<code>, timeout → sdk_timeout,
+    transport → sdk_transport, unknown → sdk_unknown_<class>
+  - NO retry on any error (single SDK call exactly)
+  - SECURITY: credential never appears in error messages / result
+    fields after diverse failure-mode sequence (401 / 429 / 500 /
+    timeout / network-error)
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Dict, List
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from kora_cli.reasoning.anthropic_engine import (
+    API_KEY_ENV,
+    MODEL_HAIKU,
+    MODEL_OPUS,
+    MODEL_SONNET,
+    OAUTH_TOKEN_ENV,
+    SYSTEM_PROMPT_PATH_ENV,
+    AnthropicReasoningEngine,
+    ReasoningEngineNotConfigured,
+    ReasoningSystemPromptError,
+)
+from kora_cli.reasoning.engine import (
+    ConversationContext,
+    ConversationTurn,
+    IncomingMessage,
+)
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_response(
+    text: str = "kora reply",
+    model: str = MODEL_OPUS,
+    input_tokens: int = 100,
+    output_tokens: int = 50,
+) -> Any:
+    """Stand-in for an Anthropic SDK Message object."""
+    block = MagicMock()
+    block.type = "text"
+    block.text = text
+
+    usage = MagicMock()
+    usage.input_tokens = input_tokens
+    usage.output_tokens = output_tokens
+
+    response = MagicMock()
+    response.content = [block]
+    response.usage = usage
+    response.model = model
+    return response
+
+
+def _make_mock_client(response=None, side_effect=None):
+    """Build a mock that emulates AsyncAnthropic. ``client.messages.create``
+    is an AsyncMock; configure either ``response`` (return value) OR
+    ``side_effect`` (exception to raise)."""
+    client = MagicMock()
+    client.messages = MagicMock()
+    if side_effect is not None:
+        client.messages.create = AsyncMock(side_effect=side_effect)
+    else:
+        client.messages.create = AsyncMock(
+            return_value=response or _make_response()
+        )
+    client.close = AsyncMock()
+    return client
+
+
+@pytest.fixture
+def system_prompt_path(tmp_path):
+    p = tmp_path / "kora_system_prompt.md"
+    p.write_text(
+        "You are Kora, Joshua's digital extension. Be useful.\n",
+        encoding="utf-8",
+    )
+    return p
+
+
+@pytest.fixture(autouse=True)
+def _clear_envs(monkeypatch):
+    """Default: both creds + prompt path env unset. Per-test fixtures
+    set what they need."""
+    monkeypatch.delenv(API_KEY_ENV, raising=False)
+    monkeypatch.delenv(OAUTH_TOKEN_ENV, raising=False)
+    monkeypatch.delenv(SYSTEM_PROMPT_PATH_ENV, raising=False)
+
+
+def _ctx(
+    rung: str = "normal",
+    state: str = "ready",
+    turns: List[ConversationTurn] = None,
+) -> ConversationContext:
+    return ConversationContext(
+        recent_messages=turns or [],
+        current_operational_state=state,
+        current_cost_ladder_rung=rung,
+    )
+
+
+def _msg(text: str = "hi kora") -> IncomingMessage:
+    return IncomingMessage(
+        text=text,
+        source="slack_dm",
+        received_at=datetime.now(timezone.utc),
+        metadata={},
+    )
+
+
+# ---------------------------------------------------------------------------
+# Construction — credential cascade
+# ---------------------------------------------------------------------------
+
+
+def test_construction_fails_when_both_creds_unset(system_prompt_path):
+    with pytest.raises(ReasoningEngineNotConfigured) as exc_info:
+        AnthropicReasoningEngine(system_prompt_path=system_prompt_path)
+    # Both env names must appear in the error so operator knows the
+    # cascade.
+    assert API_KEY_ENV in str(exc_info.value)
+    assert OAUTH_TOKEN_ENV in str(exc_info.value)
+
+
+def test_construction_succeeds_with_oauth_token(
+    monkeypatch, system_prompt_path
+):
+    monkeypatch.setenv(OAUTH_TOKEN_ENV, "sk-ant-oat-test")
+    engine = AnthropicReasoningEngine(
+        system_prompt_path=system_prompt_path
+    )
+    assert engine._auth_mode == "oauth_token"
+
+
+def test_construction_succeeds_with_api_key(monkeypatch, system_prompt_path):
+    monkeypatch.setenv(API_KEY_ENV, "sk-ant-test-api-key")
+    engine = AnthropicReasoningEngine(
+        system_prompt_path=system_prompt_path
+    )
+    assert engine._auth_mode == "api_key"
+
+
+def test_api_key_wins_over_oauth_when_both_set(
+    monkeypatch, system_prompt_path
+):
+    monkeypatch.setenv(API_KEY_ENV, "sk-ant-key")
+    monkeypatch.setenv(OAUTH_TOKEN_ENV, "sk-ant-oat-token")
+    engine = AnthropicReasoningEngine(
+        system_prompt_path=system_prompt_path
+    )
+    assert engine._auth_mode == "api_key"
+
+
+def test_whitespace_only_credential_is_treated_as_unset(
+    monkeypatch, system_prompt_path
+):
+    monkeypatch.setenv(OAUTH_TOKEN_ENV, "   ")
+    with pytest.raises(ReasoningEngineNotConfigured):
+        AnthropicReasoningEngine(system_prompt_path=system_prompt_path)
+
+
+# ---------------------------------------------------------------------------
+# System prompt failure modes
+# ---------------------------------------------------------------------------
+
+
+def test_missing_system_prompt_raises(monkeypatch, tmp_path):
+    monkeypatch.setenv(OAUTH_TOKEN_ENV, "sk-ant-oat-test")
+    missing = tmp_path / "does_not_exist.md"
+    with pytest.raises(ReasoningSystemPromptError):
+        AnthropicReasoningEngine(system_prompt_path=missing)
+
+
+def test_empty_system_prompt_raises(monkeypatch, tmp_path):
+    monkeypatch.setenv(OAUTH_TOKEN_ENV, "sk-ant-oat-test")
+    empty = tmp_path / "empty.md"
+    empty.write_text("   \n\n", encoding="utf-8")
+    with pytest.raises(ReasoningSystemPromptError):
+        AnthropicReasoningEngine(system_prompt_path=empty)
+
+
+def test_system_prompt_env_override(monkeypatch, tmp_path):
+    monkeypatch.setenv(OAUTH_TOKEN_ENV, "sk-ant-oat-test")
+    p = tmp_path / "custom_prompt.md"
+    p.write_text("custom kora prompt", encoding="utf-8")
+    monkeypatch.setenv(SYSTEM_PROMPT_PATH_ENV, str(p))
+    engine = AnthropicReasoningEngine()
+    assert "custom kora prompt" in engine._system_prompt
+
+
+# ---------------------------------------------------------------------------
+# Cost-ladder model selection
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_normal_rung_selects_opus(monkeypatch, system_prompt_path):
+    monkeypatch.setenv(OAUTH_TOKEN_ENV, "tok")
+    client = _make_mock_client(response=_make_response(model=MODEL_OPUS))
+    engine = AnthropicReasoningEngine(
+        system_prompt_path=system_prompt_path, client=client
+    )
+    result = await engine.respond(_msg(), _ctx(rung="normal"))
+    assert result.error is None
+    assert client.messages.create.await_args.kwargs["model"] == MODEL_OPUS
+
+
+@pytest.mark.asyncio
+async def test_warn_75_rung_selects_sonnet(monkeypatch, system_prompt_path):
+    monkeypatch.setenv(OAUTH_TOKEN_ENV, "tok")
+    client = _make_mock_client(
+        response=_make_response(model=MODEL_SONNET)
+    )
+    engine = AnthropicReasoningEngine(
+        system_prompt_path=system_prompt_path, client=client
+    )
+    await engine.respond(_msg(), _ctx(rung="warn_75"))
+    assert (
+        client.messages.create.await_args.kwargs["model"] == MODEL_SONNET
+    )
+
+
+@pytest.mark.asyncio
+async def test_downshift_90_rung_selects_haiku(
+    monkeypatch, system_prompt_path
+):
+    monkeypatch.setenv(OAUTH_TOKEN_ENV, "tok")
+    client = _make_mock_client(response=_make_response(model=MODEL_HAIKU))
+    engine = AnthropicReasoningEngine(
+        system_prompt_path=system_prompt_path, client=client
+    )
+    await engine.respond(_msg(), _ctx(rung="downshift_90"))
+    assert client.messages.create.await_args.kwargs["model"] == MODEL_HAIKU
+
+
+@pytest.mark.asyncio
+async def test_hard_stop_100_rung_refuses_no_sdk_call(
+    monkeypatch, system_prompt_path
+):
+    monkeypatch.setenv(OAUTH_TOKEN_ENV, "tok")
+    client = _make_mock_client()
+    engine = AnthropicReasoningEngine(
+        system_prompt_path=system_prompt_path, client=client
+    )
+    result = await engine.respond(_msg(), _ctx(rung="hard_stop_100"))
+    assert result.error == "cost_ladder_halted"
+    assert result.text == ""
+    client.messages.create.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_unknown_rung_defaults_to_opus_and_continues(
+    monkeypatch, system_prompt_path, caplog
+):
+    import logging
+
+    caplog.set_level(logging.WARNING)
+    monkeypatch.setenv(OAUTH_TOKEN_ENV, "tok")
+    client = _make_mock_client(response=_make_response(model=MODEL_OPUS))
+    engine = AnthropicReasoningEngine(
+        system_prompt_path=system_prompt_path, client=client
+    )
+    result = await engine.respond(_msg(), _ctx(rung="unknown"))
+    assert result.error is None
+    assert client.messages.create.await_args.kwargs["model"] == MODEL_OPUS
+    assert any(
+        "unknown cost rung" in r.getMessage() for r in caplog.records
+    )
+
+
+# ---------------------------------------------------------------------------
+# Operational-state gating
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_paused_state_refuses_no_sdk_call(
+    monkeypatch, system_prompt_path
+):
+    monkeypatch.setenv(OAUTH_TOKEN_ENV, "tok")
+    client = _make_mock_client()
+    engine = AnthropicReasoningEngine(
+        system_prompt_path=system_prompt_path, client=client
+    )
+    result = await engine.respond(_msg(), _ctx(state="paused"))
+    assert result.error == "operational_state_paused"
+    client.messages.create.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_stopped_state_refuses_no_sdk_call(
+    monkeypatch, system_prompt_path
+):
+    monkeypatch.setenv(OAUTH_TOKEN_ENV, "tok")
+    client = _make_mock_client()
+    engine = AnthropicReasoningEngine(
+        system_prompt_path=system_prompt_path, client=client
+    )
+    result = await engine.respond(_msg(), _ctx(state="stopped"))
+    assert result.error == "operational_state_paused"
+    client.messages.create.assert_not_awaited()
+
+
+# ---------------------------------------------------------------------------
+# Successful happy path
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_successful_call_returns_text_plus_tokens(
+    monkeypatch, system_prompt_path
+):
+    monkeypatch.setenv(OAUTH_TOKEN_ENV, "tok")
+    client = _make_mock_client(
+        response=_make_response(
+            text="here is your answer", input_tokens=120, output_tokens=80
+        )
+    )
+    engine = AnthropicReasoningEngine(
+        system_prompt_path=system_prompt_path, client=client
+    )
+    result = await engine.respond(_msg("explain X"), _ctx())
+    assert result.error is None
+    assert result.text == "here is your answer"
+    assert result.input_tokens == 120
+    assert result.output_tokens == 80
+    assert result.reasoning_duration_ms >= 0
+
+
+# ---------------------------------------------------------------------------
+# Message history projection
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_message_history_alternating_roles(
+    monkeypatch, system_prompt_path
+):
+    monkeypatch.setenv(OAUTH_TOKEN_ENV, "tok")
+    client = _make_mock_client(response=_make_response())
+    engine = AnthropicReasoningEngine(
+        system_prompt_path=system_prompt_path, client=client
+    )
+
+    now = datetime.now(timezone.utc)
+    turns = [
+        ConversationTurn(direction="inbound", text="msg1", at=now),
+        ConversationTurn(direction="outbound", text="reply1", at=now),
+        ConversationTurn(direction="inbound", text="msg2", at=now),
+        ConversationTurn(direction="outbound", text="reply2", at=now),
+    ]
+    await engine.respond(_msg("latest"), _ctx(turns=turns))
+
+    messages = client.messages.create.await_args.kwargs["messages"]
+    assert len(messages) == 5
+    assert [m["role"] for m in messages] == [
+        "user", "assistant", "user", "assistant", "user",
+    ]
+    assert messages[0]["content"] == "msg1"
+    assert messages[-1]["content"] == "latest"
+
+
+@pytest.mark.asyncio
+async def test_message_history_consecutive_same_role_concatenates(
+    monkeypatch, system_prompt_path
+):
+    """Anthropic API requires alternating roles. Loader-provided
+    non-alternating history is collapsed by concatenation."""
+    monkeypatch.setenv(OAUTH_TOKEN_ENV, "tok")
+    client = _make_mock_client(response=_make_response())
+    engine = AnthropicReasoningEngine(
+        system_prompt_path=system_prompt_path, client=client
+    )
+
+    now = datetime.now(timezone.utc)
+    turns = [
+        ConversationTurn(direction="inbound", text="msg1", at=now),
+        # Race-condition: two inbound entries before any outbound.
+        ConversationTurn(direction="inbound", text="msg2", at=now),
+    ]
+    await engine.respond(_msg("msg3-fresh"), _ctx(turns=turns))
+
+    messages = client.messages.create.await_args.kwargs["messages"]
+    # All three should collapse into one "user" turn.
+    assert len(messages) == 1
+    assert messages[0]["role"] == "user"
+    assert "msg1" in messages[0]["content"]
+    assert "msg2" in messages[0]["content"]
+    assert "msg3-fresh" in messages[0]["content"]
+
+
+@pytest.mark.asyncio
+async def test_system_prompt_passed_to_sdk(
+    monkeypatch, system_prompt_path
+):
+    monkeypatch.setenv(OAUTH_TOKEN_ENV, "tok")
+    client = _make_mock_client(response=_make_response())
+    engine = AnthropicReasoningEngine(
+        system_prompt_path=system_prompt_path, client=client
+    )
+    await engine.respond(_msg(), _ctx())
+    system = client.messages.create.await_args.kwargs["system"]
+    assert "Kora" in system
+
+
+# ---------------------------------------------------------------------------
+# SDK error mapping
+# ---------------------------------------------------------------------------
+
+
+class _StatusCodeError(Exception):
+    """Stand-in for anthropic SDK exception classes that carry
+    a status_code attribute."""
+
+    def __init__(self, status_code: int, message: str = "stub"):
+        super().__init__(message)
+        self.status_code = status_code
+
+
+class _TimeoutErrorStub(Exception):
+    """Class name matches one of the mapped timeout patterns."""
+
+    pass
+
+
+_TimeoutErrorStub.__name__ = "APITimeoutError"
+
+
+@pytest.mark.asyncio
+async def test_sdk_401_maps_to_sdk_auth(monkeypatch, system_prompt_path):
+    monkeypatch.setenv(OAUTH_TOKEN_ENV, "tok")
+    client = _make_mock_client(
+        side_effect=_StatusCodeError(401, "unauthorized")
+    )
+    engine = AnthropicReasoningEngine(
+        system_prompt_path=system_prompt_path, client=client
+    )
+    result = await engine.respond(_msg(), _ctx())
+    assert result.error == "sdk_auth"
+
+
+@pytest.mark.asyncio
+async def test_sdk_429_maps_to_rate_limited(monkeypatch, system_prompt_path):
+    monkeypatch.setenv(OAUTH_TOKEN_ENV, "tok")
+    client = _make_mock_client(
+        side_effect=_StatusCodeError(429, "too many")
+    )
+    engine = AnthropicReasoningEngine(
+        system_prompt_path=system_prompt_path, client=client
+    )
+    result = await engine.respond(_msg(), _ctx())
+    assert result.error == "sdk_rate_limited"
+
+
+@pytest.mark.asyncio
+async def test_sdk_500_maps_to_5xx(monkeypatch, system_prompt_path):
+    monkeypatch.setenv(OAUTH_TOKEN_ENV, "tok")
+    client = _make_mock_client(
+        side_effect=_StatusCodeError(500, "internal")
+    )
+    engine = AnthropicReasoningEngine(
+        system_prompt_path=system_prompt_path, client=client
+    )
+    result = await engine.respond(_msg(), _ctx())
+    assert result.error == "sdk_5xx"
+
+
+@pytest.mark.asyncio
+async def test_sdk_400_maps_to_4xx_with_code(monkeypatch, system_prompt_path):
+    monkeypatch.setenv(OAUTH_TOKEN_ENV, "tok")
+    client = _make_mock_client(side_effect=_StatusCodeError(400))
+    engine = AnthropicReasoningEngine(
+        system_prompt_path=system_prompt_path, client=client
+    )
+    result = await engine.respond(_msg(), _ctx())
+    assert result.error == "sdk_4xx_400"
+
+
+@pytest.mark.asyncio
+async def test_sdk_timeout_maps_to_sdk_timeout(
+    monkeypatch, system_prompt_path
+):
+    monkeypatch.setenv(OAUTH_TOKEN_ENV, "tok")
+    client = _make_mock_client(side_effect=_TimeoutErrorStub("timed out"))
+    engine = AnthropicReasoningEngine(
+        system_prompt_path=system_prompt_path, client=client
+    )
+    result = await engine.respond(_msg(), _ctx())
+    assert result.error == "sdk_timeout"
+
+
+@pytest.mark.asyncio
+async def test_sdk_unknown_maps_to_sdk_unknown_classname(
+    monkeypatch, system_prompt_path
+):
+    class _WhateverError(Exception):
+        pass
+
+    monkeypatch.setenv(OAUTH_TOKEN_ENV, "tok")
+    client = _make_mock_client(side_effect=_WhateverError("???"))
+    engine = AnthropicReasoningEngine(
+        system_prompt_path=system_prompt_path, client=client
+    )
+    result = await engine.respond(_msg(), _ctx())
+    assert result.error == "sdk_unknown__WhateverError"
+
+
+# ---------------------------------------------------------------------------
+# Retry policy — single attempt
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_no_retry_on_5xx(monkeypatch, system_prompt_path):
+    """Per PM Q3 default — NO retry. Single SDK call exactly."""
+    monkeypatch.setenv(OAUTH_TOKEN_ENV, "tok")
+    client = _make_mock_client(side_effect=_StatusCodeError(500))
+    engine = AnthropicReasoningEngine(
+        system_prompt_path=system_prompt_path, client=client
+    )
+    await engine.respond(_msg(), _ctx())
+    assert client.messages.create.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_no_retry_on_429(monkeypatch, system_prompt_path):
+    monkeypatch.setenv(OAUTH_TOKEN_ENV, "tok")
+    client = _make_mock_client(side_effect=_StatusCodeError(429))
+    engine = AnthropicReasoningEngine(
+        system_prompt_path=system_prompt_path, client=client
+    )
+    await engine.respond(_msg(), _ctx())
+    assert client.messages.create.await_count == 1
+
+
+# ---------------------------------------------------------------------------
+# SECURITY — credential never leaks
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_credential_never_in_errors(
+    monkeypatch, system_prompt_path, caplog
+):
+    """Diverse failure-mode sequence: 401 / 429 / 500 / timeout /
+    transport / unknown. NONE of the ResponseResult error codes or
+    log messages may contain the credential env value."""
+    import logging
+
+    caplog.set_level(logging.WARNING)
+    token_marker = "sk-ant-oat-SECRET-DO-NOT-LEAK"
+    monkeypatch.setenv(OAUTH_TOKEN_ENV, token_marker)
+
+    failures = [
+        _StatusCodeError(401),
+        _StatusCodeError(429),
+        _StatusCodeError(500),
+        _TimeoutErrorStub("timeout"),
+        type("ConnectionError", (Exception,), {})("conn"),
+        type("WhateverErr", (Exception,), {})("???"),
+    ]
+
+    for exc in failures:
+        client = _make_mock_client(side_effect=exc)
+        engine = AnthropicReasoningEngine(
+            system_prompt_path=system_prompt_path, client=client
+        )
+        result = await engine.respond(_msg(), _ctx())
+        # ResponseResult.error must NEVER carry the credential.
+        assert token_marker not in (result.error or "")
+        assert token_marker not in result.text
+        assert token_marker not in result.model_used
+
+    # All captured log messages.
+    log_text = " ".join(r.getMessage() for r in caplog.records)
+    assert token_marker not in log_text
diff --git a/tests/kora_cli/reasoning/test_context_loader.py b/tests/kora_cli/reasoning/test_context_loader.py
new file mode 100644
index 000000000000..f165a6c27b15
--- /dev/null
+++ b/tests/kora_cli/reasoning/test_context_loader.py
@@ -0,0 +1,358 @@
+"""Tests for ``kora_cli.reasoning.context_loader`` — ST1.
+
+Covers:
+  - Empty / missing file → empty context with state strings
+    "unknown" (when holders absent)
+  - Filters to (channel_id, thread_ts) — same channel/diff channel,
+    matching thread, None-thread matching
+  - Skips filtered/dropped/error inbound entries
+  - Skips failed outbound entries
+  - max_turns slicing keeps last-N
+  - Operational + cost state surfaces from holders when initialized
+  - Malformed lines logged + skipped
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Dict, List
+
+import pytest
+
+from kora_cli.reasoning.context_loader import (
+    DEFAULT_MAX_TURNS,
+    LOG_PATH_ENV,
+    load_slack_dm_context,
+)
+
+
+def _make_inbound(
+    *,
+    channel: str = "D01",
+    thread_ts: Any = None,
+    text: str = "hi",
+    handled_status: str = "received",
+    received_at: str = "2026-05-22T10:00:00+00:00",
+) -> Dict[str, Any]:
+    entry: Dict[str, Any] = {
+        "received_at": received_at,
+        "channel_id": channel,
+        "thread_ts": thread_ts,
+        "user_id": "UJOSHUA",
+        "text": text,
+        "event_ts": "1700000000.001",
+        "handled_status": handled_status,
+    }
+    return entry
+
+
+def _make_outbound(
+    *,
+    channel: str = "D01",
+    thread_ts: Any = None,
+    text: str = "reply",
+    send_status: str = "ok",
+    sent_at: str = "2026-05-22T10:00:01+00:00",
+) -> Dict[str, Any]:
+    return {
+        "sent_at": sent_at,
+        "channel_id": channel,
+        "thread_ts": thread_ts,
+        "text": text,
+        "slack_message_ts": "1700000001.999",
+        "send_status": send_status,
+    }
+
+
+def _write_jsonl(path: Path, entries: List[Dict[str, Any]]) -> Path:
+    path.write_text(
+        "\n".join(json.dumps(e) for e in entries) + "\n", encoding="utf-8"
+    )
+    return path
+
+
+@pytest.fixture(autouse=True)
+def _reset_holders(monkeypatch):
+    """No holders by default → state strings 'unknown'. Tests that
+    want a holder override per-test."""
+    from agent import operational_state_holder as h_mod
+    from agent import cost_state_holder as c_mod
+
+    monkeypatch.setattr(h_mod, "_HOLDER", None)
+    monkeypatch.setattr(c_mod, "_HOLDER", None)
+
+
+# ---------------------------------------------------------------------------
+# Missing / empty file
+# ---------------------------------------------------------------------------
+
+
+def test_missing_file_returns_empty(tmp_path):
+    ctx = load_slack_dm_context(
+        channel_id="D01",
+        thread_ts=None,
+        log_path=tmp_path / "missing.jsonl",
+    )
+    assert ctx.recent_messages == []
+    assert ctx.current_operational_state == "unknown"
+    assert ctx.current_cost_ladder_rung == "unknown"
+
+
+def test_empty_file_returns_empty(tmp_path):
+    path = tmp_path / "empty.jsonl"
+    path.write_text("", encoding="utf-8")
+    ctx = load_slack_dm_context(
+        channel_id="D01", thread_ts=None, log_path=path
+    )
+    assert ctx.recent_messages == []
+
+
+# ---------------------------------------------------------------------------
+# Thread / channel filtering
+# ---------------------------------------------------------------------------
+
+
+def test_same_channel_no_thread_matches(tmp_path):
+    path = _write_jsonl(
+        tmp_path / "log.jsonl",
+        [
+            _make_inbound(channel="D01", thread_ts=None, text="msg1"),
+            _make_outbound(channel="D01", thread_ts=None, text="reply1"),
+        ],
+    )
+    ctx = load_slack_dm_context(
+        channel_id="D01", thread_ts=None, log_path=path
+    )
+    texts = [t.text for t in ctx.recent_messages]
+    assert texts == ["msg1", "reply1"]
+
+
+def test_different_channel_filtered(tmp_path):
+    path = _write_jsonl(
+        tmp_path / "log.jsonl",
+        [
+            _make_inbound(channel="D01", text="me"),
+            _make_inbound(channel="D02", text="other-channel"),
+            _make_outbound(channel="D01", text="ok"),
+        ],
+    )
+    ctx = load_slack_dm_context(
+        channel_id="D01", thread_ts=None, log_path=path
+    )
+    texts = [t.text for t in ctx.recent_messages]
+    assert "other-channel" not in texts
+
+
+def test_thread_ts_matches_exact(tmp_path):
+    path = _write_jsonl(
+        tmp_path / "log.jsonl",
+        [
+            _make_inbound(channel="D01", thread_ts="1.001", text="in-thread"),
+            _make_inbound(channel="D01", thread_ts="2.002", text="other-thread"),
+            _make_inbound(channel="D01", thread_ts=None, text="no-thread"),
+        ],
+    )
+    ctx = load_slack_dm_context(
+        channel_id="D01", thread_ts="1.001", log_path=path
+    )
+    texts = [t.text for t in ctx.recent_messages]
+    assert texts == ["in-thread"]
+
+
+def test_thread_ts_none_does_not_match_threaded_entries(tmp_path):
+    path = _write_jsonl(
+        tmp_path / "log.jsonl",
+        [
+            _make_inbound(channel="D01", thread_ts=None, text="no-thread"),
+            _make_inbound(channel="D01", thread_ts="1.001", text="threaded"),
+        ],
+    )
+    ctx = load_slack_dm_context(
+        channel_id="D01", thread_ts=None, log_path=path
+    )
+    texts = [t.text for t in ctx.recent_messages]
+    assert texts == ["no-thread"]
+
+
+# ---------------------------------------------------------------------------
+# Skip filtered / failed entries
+# ---------------------------------------------------------------------------
+
+
+def test_filtered_inbound_entries_skipped(tmp_path):
+    path = _write_jsonl(
+        tmp_path / "log.jsonl",
+        [
+            _make_inbound(channel="D01", text="valid"),
+            _make_inbound(
+                channel="D01",
+                text="not-joshua",
+                handled_status="filtered_non_joshua",
+            ),
+            _make_inbound(
+                channel="D01",
+                text="paused-drop",
+                handled_status="dropped_paused",
+            ),
+            _make_inbound(
+                channel="D01",
+                text="handler-err",
+                handled_status="handler_error",
+            ),
+        ],
+    )
+    ctx = load_slack_dm_context(
+        channel_id="D01", thread_ts=None, log_path=path
+    )
+    texts = [t.text for t in ctx.recent_messages]
+    assert texts == ["valid"]
+
+
+def test_failed_outbound_entries_skipped(tmp_path):
+    path = _write_jsonl(
+        tmp_path / "log.jsonl",
+        [
+            _make_outbound(channel="D01", text="successful-reply"),
+            _make_outbound(
+                channel="D01", text="failed-reply", send_status="failed"
+            ),
+        ],
+    )
+    ctx = load_slack_dm_context(
+        channel_id="D01", thread_ts=None, log_path=path
+    )
+    texts = [t.text for t in ctx.recent_messages]
+    assert texts == ["successful-reply"]
+
+
+# ---------------------------------------------------------------------------
+# Ordering + max_turns
+# ---------------------------------------------------------------------------
+
+
+def test_turns_returned_oldest_to_newest(tmp_path):
+    path = _write_jsonl(
+        tmp_path / "log.jsonl",
+        [
+            _make_inbound(
+                channel="D01", text="msg1",
+                received_at="2026-05-22T10:00:00+00:00",
+            ),
+            _make_outbound(
+                channel="D01", text="reply1",
+                sent_at="2026-05-22T10:00:05+00:00",
+            ),
+            _make_inbound(
+                channel="D01", text="msg2",
+                received_at="2026-05-22T10:01:00+00:00",
+            ),
+        ],
+    )
+    ctx = load_slack_dm_context(
+        channel_id="D01", thread_ts=None, log_path=path
+    )
+    texts = [t.text for t in ctx.recent_messages]
+    assert texts == ["msg1", "reply1", "msg2"]
+
+
+def test_max_turns_slices_most_recent(tmp_path):
+    entries = []
+    for i in range(20):
+        entries.append(
+            _make_inbound(
+                channel="D01",
+                text=f"msg{i}",
+                received_at=f"2026-05-22T10:{i:02d}:00+00:00",
+            )
+        )
+    path = _write_jsonl(tmp_path / "log.jsonl", entries)
+    ctx = load_slack_dm_context(
+        channel_id="D01", thread_ts=None, max_turns=5, log_path=path
+    )
+    texts = [t.text for t in ctx.recent_messages]
+    assert texts == ["msg15", "msg16", "msg17", "msg18", "msg19"]
+
+
+def test_default_max_turns_is_10(tmp_path):
+    entries = []
+    for i in range(15):
+        entries.append(
+            _make_inbound(
+                channel="D01",
+                text=f"msg{i}",
+                received_at=f"2026-05-22T10:{i:02d}:00+00:00",
+            )
+        )
+    path = _write_jsonl(tmp_path / "log.jsonl", entries)
+    ctx = load_slack_dm_context(
+        channel_id="D01", thread_ts=None, log_path=path
+    )
+    assert len(ctx.recent_messages) == DEFAULT_MAX_TURNS
+
+
+# ---------------------------------------------------------------------------
+# Malformed lines
+# ---------------------------------------------------------------------------
+
+
+def test_malformed_json_lines_skipped(tmp_path, caplog):
+    caplog.set_level(logging.WARNING)
+    raw = (
+        json.dumps(_make_inbound(channel="D01", text="ok"))
+        + "\n{not-valid-json\n"
+        + json.dumps(_make_inbound(channel="D01", text="also-ok"))
+        + "\n"
+    )
+    path = tmp_path / "log.jsonl"
+    path.write_text(raw, encoding="utf-8")
+
+    ctx = load_slack_dm_context(
+        channel_id="D01", thread_ts=None, log_path=path
+    )
+    texts = [t.text for t in ctx.recent_messages]
+    assert texts == ["ok", "also-ok"]
+    assert any("malformed" in r.getMessage() for r in caplog.records)
+
+
+# ---------------------------------------------------------------------------
+# Operational + cost state surfaces
+# ---------------------------------------------------------------------------
+
+
+def test_operational_state_from_holder(tmp_path, monkeypatch):
+    from agent.operational_state import (
+        ClaimPermission,
+        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)
+        ),
+    )
+
+    path = _write_jsonl(tmp_path / "log.jsonl", [])
+    ctx = load_slack_dm_context(
+        channel_id="D01", thread_ts=None, log_path=path
+    )
+    assert ctx.current_operational_state == "paused"
+
+
+def test_log_path_env_override(tmp_path, monkeypatch):
+    p = _write_jsonl(
+        tmp_path / "custom.jsonl",
+        [_make_inbound(channel="D01", text="from-env-path")],
+    )
+    monkeypatch.setenv(LOG_PATH_ENV, str(p))
+    # No log_path arg → uses env override.
+    ctx = load_slack_dm_context(channel_id="D01", thread_ts=None)
+    texts = [t.text for t in ctx.recent_messages]
+    assert texts == ["from-env-path"]