"""Claude SDK turn runner for Möbius.

This module isolates the Claude Agent SDK integration behind one
function that executes exactly one Möbius chat turn and publishes the
same event shapes the rest of the backend already understands.

Design choices:

- `permission_mode="bypassPermissions"` preserves Möbius's existing
  trust posture. The agent is expected to have the normal Claude Code
  tool surface. We intercept only `AskUserQuestion`, because that tool
  must become an explicit partner choice in the Möbius UI.
- `ClaudeSDKClient` is used instead of one-shot `query()` because
  Möbius needs the bidirectional control surface: explicit `connect()`,
  `query()`, streaming `receive_response()`, and external
  `interrupt()` support for Stop.
- `AskUserQuestion` is intercepted via the SDK's `can_use_tool`
  callback (NOT PreToolUse/PostToolUse hooks). The callback parks a
  future in the shared `pending_questions` registry, broadcasts the
  question event, and awaits the user's answer. When the answer
  arrives (POST /messages with body.answers, resolved by
  routes/chats_stream.py), the callback returns
  `PermissionResultAllow(updated_input={"questions": ..., "answers": ...})`
  — the SDK then runs `AskUserQuestion` with the answers as input and
  the tool's headless implementation echoes them back as the result
  the model sees. PreToolUse/PostToolUse was explored and rejected:
  the SDK does NOT fire PostToolUse for AskUserQuestion in headless
  mode, so the two-hook flow never worked.
- Stop support is wired through `active_clients[chat_id]`. The caller
  interrupts the live SDK client while this runner keeps draining
  `receive_response()` until the terminal result arrives.
- `system_prompt` is passed on EVERY turn, not just the first. The
  installed SDK transport
  (`claude_agent_sdk/_internal/transport/subprocess_cli.py:227-228`)
  serializes `system_prompt is None → --system-prompt ""`, which on
  resume silently wipes the original session's system prompt. Since
  ClaudeAgentOptions defaults `system_prompt` to `None`, omitting the
  kwarg has the same effect. Always passing `skill_text` keeps the
  skill load-bearing across resumes and matches our "skill is always-
  on" contract — passing the same text on resume is a no-op; passing
  updated text after a deploy correctly updates the resumed session.
- We deliberately pass `skill_text` as a custom string (not
  `SystemPromptPreset{append=skill_text, exclude_dynamic_sections=True}`).
  The preset+append form would layer Claude Code's default
  engineer-facing preset on top of our Möbius skill — adding
  generic tool-use / communication guidance that our skill already
  defines in Möbius-specific terms (and sometimes contradicts).
  `exclude_dynamic_sections` only applies with the default preset
  (the CLI ignores it with `--system-prompt`, per the CLI's own
  `--help`), so for our custom-string path it would be a no-op
  even if we set it. Möbius owns its system prompt end-to-end;
  the skill is the contract, not a layer on top of someone else's.
"""

from __future__ import annotations

import asyncio
import inspect
import json
import logging
import os
from typing import Any
from uuid import uuid4

log = logging.getLogger(__name__)

from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient, HookMatcher
from claude_agent_sdk.types import (
  AssistantMessage,
  PermissionResultAllow,
  PermissionResultDeny,
  RateLimitEvent,
  ResultMessage,
  StreamEvent,
  SystemMessage,
  TaskNotificationMessage,
  TaskProgressMessage,
  TaskStartedMessage,
  TextBlock,
  ThinkingBlock,
  ToolResultBlock,
  ToolUseBlock,
  UserMessage,
)

from app import activity
from app.pending_questions import PendingQuestion
from app.runner_registry import RunnerKind, registry
from app.runtime_types import RunnerResult
from app.sdk_emit import emit_unknown_enabled, unknown_event
from app.tool_summaries import summarize_tool_input


async def _persist_session_id(db, chat_id: str, session_id: str | None) -> None:
  """Best-effort early persistence for provider resume continuity."""
  if db is None or not chat_id or not session_id:
    return
  try:
    from app.chat_writer import PersistSessionId, await_ack, get_writer
    ack = get_writer().submit(
      PersistSessionId(chat_id=chat_id, session_id=session_id)
    )
    await await_ack(ack)
  except Exception:
    log.warning(
      "Claude session id persistence failed chat_id=%s session_id=%s",
      chat_id,
      session_id,
      exc_info=True,
    )


def _resumable(
  session_id: str | None, cwd: str, config_dir: str | None = None
) -> bool:
  """True iff a transcript .jsonl for session_id exists for this cwd.

  `claude --resume <id>` reads the transcript the CLI stored under
  `<CLAUDE_CONFIG_DIR>/projects/<encoded-cwd>/<id>.jsonl`, where the
  project dir encodes the cwd by stripping the leading slash and
  replacing every `/` with `-` (cwd `/data` -> `-data`, cwd
  `/data/apps/news-2` -> `-data-apps-news-2`). A stored id can fail to
  resolve two ways, both of which make `--resume` die "No conversation
  found" (exit 1): a pre-fix PHANTOM id (the codex plugin's SessionStart
  hook minted an id that got a `session-env/<id>` dir but never a
  transcript), or a real id whose transcript the CLI's ~30-day cleanup
  has since deleted. Callers use this to fall back to a DB-transcript
  reseed instead of letting the turn hard-fail.

  The `-data` derivation is verified against prod: every stored
  session id that resolves on disk lives under `projects/-data/`, and
  `fork-chat.sh` resumes chat sessions from the same `/data` cwd.
  """
  if not session_id:
    return False
  base = config_dir or os.environ.get("CLAUDE_CONFIG_DIR", "")
  if not base:
    return False
  proj = "-" + cwd.strip("/").replace("/", "-")
  return os.path.isfile(
    os.path.join(base, "projects", proj, f"{session_id}.jsonl")
  )


class ActiveClaudeClient:
  """Stop handle stored in `chat._active_clients` for SDK-backed turns.

  Ordering contract: `interrupt()` signals the SDK and then awaits
  `_finished`, which the runner resolves ONLY after `client.disconnect()`
  returns. Callers (stop_chat / stop_chat_for) therefore block until
  the SDK subprocess is fully torn down, so a late `bc.publish(done)`
  from the runner cannot land after `bc.mark_completed()` has already
  closed the broadcast for live SSE subscribers.
  """

  def __init__(self, client: ClaudeSDKClient, chat_id: str):
    self.chat_id = chat_id
    self.kind = RunnerKind.CLAUDE_SDK
    self._client = client
    # FIFO of mid-turn steer texts: two rapid sends must both reach Claude
    # (both are already persisted to the transcript), so a single slot would
    # silently drop the first. The runner drains the whole list on interrupt.
    self.pending_steer: list[str] = []
    self._finished: asyncio.Future[None] = (
      asyncio.get_running_loop().create_future()
    )

  async def steer(self, text: str) -> bool:
    """Interrupts the live response so the runner can re-prompt.

    Claude's SDK cannot append to an in-flight tool loop. The closest
    faithful behavior is to record the redirect text, interrupt the
    current response, drain the interrupt result, then query the same
    connected client again so session context is preserved.
    """
    if self._finished.done():
      return False
    self.pending_steer.append(text)
    await self._client.interrupt()
    return True

  async def interrupt(self) -> None:
    """Interrupts the live run and waits for runner-side drain.

    Bounds the `_finished` wait at 5s as a defense-in-depth so a
    wedged runner (one that never reaches its `finally` block) can't
    hang Stop indefinitely. `chat.py:stop_chat_for` adds its own 2s
    bound at the call site; this inner timeout protects any other
    direct caller.
    """
    self.pending_steer = []
    await self._client.interrupt()
    try:
      await asyncio.wait_for(self._finished, timeout=5.0)
    except asyncio.TimeoutError:
      import logging
      logging.getLogger("moebius.chat").warning(
        "ActiveClaudeClient._finished never resolved within 5s; "
        "runner is wedged",
      )

  async def stop(self, timeout: float = 2.0) -> bool:
    """Interrupts the SDK run and waits up to `timeout` seconds."""
    try:
      await asyncio.wait_for(self.interrupt(), timeout=timeout)
      return True
    except asyncio.CancelledError:
      raise
    except asyncio.TimeoutError:
      log.warning(
        "Claude SDK stop timed out chat_id=%s", self.chat_id,
      )
      return False
    except Exception:
      log.exception(
        "Claude SDK stop failed chat_id=%s", self.chat_id,
      )
      return False

  def mark_finished(self) -> None:
    """Resolves the stop waiter once the runner is fully drained."""
    if not self._finished.done():
      self._finished.set_result(None)


def _steer_redirect_message(text: str) -> str:
  """Frames a Claude steer as a redirect on the still-connected client."""
  return (
    "The user added this while you were working. Incorporate it and "
    "continue the same task:\n\n"
    f"{text}"
  )


async def steer_into_active_turn(chat_id: str, text: str) -> bool:
  """Interrupts a live Claude SDK turn so it can resume with `text`."""
  handle = registry.get_handle(chat_id, RunnerKind.CLAUDE_SDK)
  if not isinstance(handle, ActiveClaudeClient):
    return False
  return await handle.steer(text)


def _skill_file_read_name(
  tool_name: str, input_data: Any, cwd: str,
) -> str:
  """Returns the skill name when a Read targets a Möbius skill file.

  The in-product agent loads its skills by Reading
  `<data_dir>/shared/skills/<name>.md` — on the default posture
  (skills_enabled off) the SDK Skill tool is never offered, so the
  Read input is the only place skill loads are actually observable.
  The match is purely lexical (normpath, no filesystem access) and
  returns "" for anything that isn't a direct skill-file read. A
  relative path is resolved against the turn's cwd: the agent runs
  with cwd=/data, so `shared/skills/mind.md` is the same load.
  """
  if tool_name != "Read" or not isinstance(input_data, dict):
    return ""
  raw = input_data.get("file_path")
  if not isinstance(raw, str) or not raw.strip():
    return ""
  path = raw.strip()
  if not os.path.isabs(path):
    path = os.path.join(cwd or "/", path)
  path = os.path.normpath(path)
  from app.config import get_settings
  skills_dir = os.path.normpath(
    os.path.join(get_settings().data_dir, "shared", "skills")
  )
  parent, filename = os.path.split(path)
  if parent != skills_dir or not filename.endswith(".md"):
    return ""
  return filename[: -len(".md")]


def observe_skill_file_read(
  tool_name: str,
  input_data: Any,
  *,
  bc,
  chat_id: str,
  cwd: str,
) -> None:
  """Fire-and-forget skill observability for skill-file Reads.

  Publishes the same `skill_loaded` event + activity record the Skill
  tool path emits (see the dispatch below), so the activity log's
  most-used-skills cross-check sees Read-based loads too — before
  this, the cross-check endpoint returned empty every night because
  the agent never goes through the Skill tool. Never raises: a broken
  broadcast or a full disk must not block or fail the tool call being
  intercepted.
  """
  try:
    skill = _skill_file_read_name(tool_name, input_data, cwd)
    if not skill:
      return
    bc.publish({"type": "skill_loaded", "skill": skill})
    activity.log_skill_load(chat_id, skill)
  except Exception:
    log.debug("skill_loaded read observability failed", exc_info=True)


def _memory_node_read_id(
  tool_name: str, input_data: Any, cwd: str,
) -> str:
  """Returns the memory-graph node id when a Read targets a note or MOC.

  The agent descends the graph by Reading
  `<data_dir>/shared/memory/{notes,mocs}/<slug>.md` — exactly the
  explicit-read signal the per-chat read-trace wants (the injected
  block is recorded separately at the injection site). index.md,
  inbox.md, and recent-chats.md don't count: they arrive injected, so
  a Read of them says nothing about what the agent dug for. Purely
  lexical, like `_skill_file_read_name` above; returns "" for anything
  that isn't a direct note/MOC read."""
  if tool_name != "Read" or not isinstance(input_data, dict):
    return ""
  raw = input_data.get("file_path")
  if not isinstance(raw, str) or not raw.strip():
    return ""
  path = raw.strip()
  if not os.path.isabs(path):
    path = os.path.join(cwd or "/", path)
  path = os.path.normpath(path)
  from app.config import get_settings
  memory_root = os.path.normpath(
    os.path.join(get_settings().data_dir, "shared", "memory")
  )
  parent, filename = os.path.split(path)
  if os.path.dirname(parent) != memory_root:
    return ""
  if os.path.basename(parent) not in ("notes", "mocs"):
    return ""
  if not filename.endswith(".md"):
    return ""
  return filename[: -len(".md")]


def observe_memory_node_read(
  tool_name: str,
  input_data: Any,
  *,
  chat_id: str,
  cwd: str,
) -> None:
  """Fire-and-forget read-trace entry for memory-node Reads.

  Mirrors `observe_skill_file_read`: called from `can_use_tool` on
  every non-question tool, filters down to note/MOC reads, and merges
  the slug into this chat's read-trace file so the nightly Dreaming
  pass can see what the agent went looking for. Never raises — a full
  disk or broken trace file must not block or fail the Read being
  intercepted."""
  try:
    node_id = _memory_node_read_id(tool_name, input_data, cwd)
    if not node_id:
      return
    from app import memory_trace
    from app.config import get_settings
    memory_trace.record_note_read(
      get_settings().data_dir, chat_id, node_id
    )
  except Exception:
    log.debug("memory read-trace observability failed", exc_info=True)


def _skill_name_from_input(input_data: Any) -> str:
  """Extracts the loaded skill's name from a Skill tool_use input.

  The Skill tool's input is `{"skill": "<name>", "args": "..."}` — the
  skill name lives under the `skill` key. Older / plugin-namespaced
  forms can carry it as `command` (the slash-command name), so fall
  back to that. Returns an empty string when neither is present so the
  caller can decide not to emit an empty chip.
  """
  if not isinstance(input_data, dict):
    return ""
  name = input_data.get("skill") or input_data.get("command") or ""
  return name.strip() if isinstance(name, str) else ""


def _result_error_message(result: ResultMessage) -> str:
  """Builds a user-facing error string from an SDK result."""
  if isinstance(result.result, str) and result.result.strip():
    return result.result.strip()
  if result.errors:
    return "\n".join(err for err in result.errors if err).strip()
  if result.subtype == "error_during_execution":
    return "Execution interrupted."
  return "Claude SDK turn failed."


def _should_retry_without_model(error_text: str | None) -> bool:
  """True when Claude rejected the explicit model selection."""
  if not error_text:
    return False
  text = error_text.lower()
  return (
    "selected model" in text
    and "may not exist or you may not have access" in text
  )


def _format_tool_output(content: Any) -> str:
  """Formats SDK tool-result content for Möbius tool_output events."""
  if content is None:
    return ""
  if isinstance(content, str):
    return content
  if isinstance(content, list):
    parts: list[str] = []
    for item in content:
      if isinstance(item, str):
        parts.append(item)
        continue
      if isinstance(item, dict) and item.get("type") == "text":
        text = item.get("text")
        if isinstance(text, str):
          parts.append(text)
          continue
      parts.append(json.dumps(item, ensure_ascii=True))
    return "\n".join(part for part in parts if part).strip()
  return json.dumps(content, ensure_ascii=True)


async def _maybe_await(value: Any) -> Any:
  """Awaits a value only when the callback returned an awaitable."""
  if inspect.isawaitable(value):
    return await value
  return value


def _emit_unknown(bc, kind: str, raw: Any) -> None:
  """Logs an unknown SDK event and emits it on the wire when enabled.

  The DEBUG log fires unconditionally so noisy sessions stay
  inspectable in `chat.log` even when wire emission is turned off
  via ``MOBIUS_EMIT_UNKNOWN=0``.
  """
  event = unknown_event(kind, raw)
  if emit_unknown_enabled():
    bc.publish(event)


def _usage_event(usage: dict[str, Any]) -> dict:
  """Builds the wire-shape `usage` event from an SDK usage dict.

  The SDK's usage shape evolves — we extract the fields we know
  about today and pass the full dict through under ``raw`` so a
  later UI can pick up newly-added counters without a runner change.
  """
  return {
    "type": "usage",
    "input_tokens": usage.get("input_tokens"),
    "output_tokens": usage.get("output_tokens"),
    "cache_creation_input_tokens": usage.get("cache_creation_input_tokens"),
    "cache_read_input_tokens": usage.get("cache_read_input_tokens"),
    "raw": dict(usage),
  }


def dispatch_sdk_message(
  sdk_msg: Any,
  bc,
  current_session_id: str | None,
) -> tuple[str | None, dict | None]:
  """Translates one SDK message into broadcast events.

  Returns ``(new_session_id, terminal_result_or_None)``. When the
  message is a ResultMessage, the caller receives the final result
  dict and stops draining the SDK stream. For every other message
  type the caller updates ``current_session_id`` from the first
  return value and keeps reading.

  Extracted from the runner loop so unit tests can exercise the
  full dispatch matrix (named events, unknown fallthrough, usage
  + stop_reason side channels) without spinning up a live SDK
  subprocess.
  """
  if isinstance(sdk_msg, SystemMessage):
    if isinstance(sdk_msg, TaskStartedMessage):
      bc.publish({
        "type": "task_start",
        "task_id": sdk_msg.task_id,
        "description": sdk_msg.description,
        "task_type": sdk_msg.task_type,
      })
      return current_session_id, None
    if isinstance(sdk_msg, TaskProgressMessage):
      bc.publish({
        "type": "task_progress",
        "task_id": sdk_msg.task_id,
        "usage": dict(sdk_msg.usage) if sdk_msg.usage else None,
        "last_tool_name": sdk_msg.last_tool_name,
      })
      return current_session_id, None
    if isinstance(sdk_msg, TaskNotificationMessage):
      bc.publish({
        "type": "task_done",
        "task_id": sdk_msg.task_id,
        "status": sdk_msg.status,
        "summary": sdk_msg.summary,
      })
      return current_session_id, None
    if sdk_msg.subtype == "init":
      # Setup metadata only — no Möbius-side render.
      return current_session_id, None
    _emit_unknown(bc, f"system:{sdk_msg.subtype}", sdk_msg)
    return current_session_id, None

  if isinstance(sdk_msg, StreamEvent):
    if sdk_msg.session_id:
      current_session_id = sdk_msg.session_id
    event = sdk_msg.event
    event_type = event.get("type")
    if event_type == "content_block_delta":
      delta = event.get("delta", {})
      delta_type = delta.get("type")
      if delta_type == "text_delta":
        text = delta.get("text")
        if text:
          bc.publish({"type": "text", "content": text})
        return current_session_id, None
      if delta_type == "thinking_delta":
        thinking = delta.get("thinking") or delta.get("text") or ""
        if thinking:
          bc.publish({"type": "thinking", "content": thinking})
        return current_session_id, None
      _emit_unknown(bc, f"stream:content_block_delta:{delta_type}", delta)
      return current_session_id, None
    _emit_unknown(bc, f"stream:{event_type}", event)
    return current_session_id, None

  if isinstance(sdk_msg, AssistantMessage):
    if sdk_msg.session_id:
      current_session_id = sdk_msg.session_id
    for block in sdk_msg.content:
      if isinstance(block, ToolUseBlock):
        bc.publish({
          "type": "tool_start",
          "tool": block.name,
          "input": "",
        })
        summary = summarize_tool_input(block.name, block.input)
        if summary:
          bc.publish({
            "type": "tool_input",
            "tool": block.name,
            "input": summary,
          })
        # Skill observability: when the agent loads a skill, surface it
        # as its own `skill_loaded` event (the frontend stamps a chip
        # onto the Skill tool block) and append a record to the activity
        # log so "most-used skills" can be aggregated. This fires
        # whenever the Skill tool runs at all; whether skills are even
        # OFFERED to the agent is the separate, gated `skills_enabled`
        # decision below — observability is correct either way.
        if block.name == "Skill":
          skill = _skill_name_from_input(block.input)
          if skill:
            bc.publish({"type": "skill_loaded", "skill": skill})
            activity.log_skill_load(getattr(bc, "chat_id", None), skill)
        continue
      if isinstance(block, ThinkingBlock):
        bc.publish({"type": "thinking", "content": block.thinking})
        continue
      if isinstance(block, TextBlock):
        # Streamed via text_delta already — snapshot duplicate.
        continue
      _emit_unknown(
        bc, f"assistant_block:{type(block).__name__}", block,
      )
    if sdk_msg.usage:
      bc.publish(_usage_event(sdk_msg.usage))
    if sdk_msg.stop_reason:
      bc.publish({
        "type": "stop_reason",
        "reason": sdk_msg.stop_reason,
      })
    return current_session_id, None

  if isinstance(sdk_msg, UserMessage):
    content = sdk_msg.content if isinstance(sdk_msg.content, list) else []
    for block in content:
      if isinstance(block, ToolResultBlock):
        bc.publish({
          "type": "tool_output",
          "content": _format_tool_output(block.content),
        })
        bc.publish({"type": "tool_end"})
        continue
      _emit_unknown(bc, f"user_block:{type(block).__name__}", block)
    return current_session_id, None

  if isinstance(sdk_msg, RateLimitEvent):
    info = sdk_msg.rate_limit_info
    bc.publish({
      "type": "rate_limit",
      "status": info.status,
      "resets_at": info.resets_at,
      "rate_limit_type": info.rate_limit_type,
      "utilization": info.utilization,
    })
    return current_session_id, None

  if isinstance(sdk_msg, ResultMessage):
    if sdk_msg.session_id:
      current_session_id = sdk_msg.session_id
    if sdk_msg.usage:
      bc.publish(_usage_event(sdk_msg.usage))
    if sdk_msg.stop_reason:
      bc.publish({
        "type": "stop_reason",
        "reason": sdk_msg.stop_reason,
      })
    return current_session_id, {
      "session_id": current_session_id,
      "cost_usd": sdk_msg.total_cost_usd,
      "usage": dict(sdk_msg.usage) if sdk_msg.usage else None,
      "model_usage": (
        dict(sdk_msg.model_usage) if sdk_msg.model_usage else None
      ),
      "permission_denials": sdk_msg.permission_denials or None,
      "api_error_status": sdk_msg.api_error_status,
      "error": (
        _result_error_message(sdk_msg)
        if sdk_msg.is_error else None
      ),
    }

  # Any SDK message class we didn't enumerate — never silently dropped.
  _emit_unknown(bc, f"sdk_message:{type(sdk_msg).__name__}", sdk_msg)
  return current_session_id, None


async def run_claude_sdk_turn(
  user_message: str,
  session_id: str | None,
  base_env: dict[str, str],
  cwd: str,
  chat_id: str,
  skill_text: str,
  bc,
  pending_questions: dict,
  db,
  agent_settings: dict | None = None,
  skills_enabled: bool = False,
) -> RunnerResult:
  """Runs one Claude SDK turn and translates SDK messages to Möbius events.

  Args:
    user_message: Fully prepared user prompt for this turn.
    session_id: Existing Claude session to resume, or None on first turn.
    base_env: Environment passed through to the Claude subprocess.
    cwd: Working directory for the SDK run.
    chat_id: Möbius chat identifier used for registries.
    skill_text: Möbius skill/system prompt text for first turn only.
    bc: Chat broadcast object with a publish(event) method.
    pending_questions: Shared AskUserQuestion registry owned by chat.py.
    db: SQLAlchemy session used by runner-side persistence.
    skills_enabled: When True, offer SDK skills to the agent
      (`setting_sources` including user+project + `skills="all"`). This
      is behavior-shifting and defaults OFF so the skill-observability
      path can ship without changing what the agent does — skill loads
      are still observed (chip + activity log) whenever a skill does
      load, regardless of this flag.

  Returns:
    A dict containing the resulting session ID, final cost, and error.
  """
  current_session_id = session_id
  cost_usd: float | None = None

  # Canonical AskUserQuestion handling via can_use_tool, per
  # https://code.claude.com/docs/en/agent-sdk/user-input
  # The SDK does NOT fire PostToolUse for AskUserQuestion (empirically
  # confirmed). The correct injection point is `can_use_tool`: return
  # PermissionResultAllow with updated_input containing the original
  # questions array plus an `answers` dict {question_text: label}.
  # The SDK then runs the tool with that input and the model sees the
  # answers as the tool's result.
  #
  # `bypassPermissions` would skip the can_use_tool callback entirely —
  # use the documented dummy PreToolUse keepalive + default permission
  # mode so the callback fires only on AskUserQuestion (other tools are
  # auto-approved by the keepalive hook returning continue_=True).
  async def can_use_tool(
    tool_name: str,
    input_data: dict[str, Any],
    context,
  ) -> PermissionResultAllow | PermissionResultDeny:
    del context
    # Auto-approve every tool except AskUserQuestion — this preserves
    # the "trust the agent" posture (no tool gating) while still
    # intercepting AskUserQuestion for the partner UX. The callback is
    # also the canonical observation point for skill-file Reads (the
    # agent loads /data/shared/skills/*.md via Read, not the Skill
    # tool); the observe call is fire-and-forget and never blocks or
    # fails the tool.
    if tool_name != "AskUserQuestion":
      observe_skill_file_read(
        tool_name, input_data, bc=bc, chat_id=chat_id, cwd=cwd,
      )
      observe_memory_node_read(
        tool_name, input_data, chat_id=chat_id, cwd=cwd,
      )
      return PermissionResultAllow(updated_input=input_data)

    questions = input_data.get("questions", [])
    if not isinstance(questions, list):
      questions = []

    existing = pending_questions.get(chat_id)
    if existing is not None and not existing.future.done():
      return PermissionResultDeny(
        message=f"AskUserQuestion already pending for chat {chat_id}"
      )

    future = asyncio.get_running_loop().create_future()
    pending = PendingQuestion(
      question_id=str(uuid4()),
      questions=questions,
      future=future,
      # The turn's persistence run token (the sink carries it). The
      # answer route submits AnswerQuestion keyed on this so the writer
      # actor fences the right (chat_id, run_token) snapshot before
      # merging the answer; None for a sink/broadcast without one
      # (legacy/test) → the answer route broad-fences by chat.
      run_token=getattr(bc, "run_token", None),
    )
    pending_questions[chat_id] = pending

    # Save-before-broadcast (Candidate B): the card's question_id MUST be
    # durably persisted before the SSE event shows it, or a fast Submit
    # races the DB write and the answer is lost. `publish_question`
    # submits a QuestionCommit, awaits its ack, then broadcasts — and
    # RAISES if the commit didn't land. We register the pending entry
    # first (so the finally below always pops it) but on a failed commit
    # DENY the tool with a persistence-unavailable message rather than
    # broadcasting an unpersisted card or writing the blob directly.
    #
    # Push notification on AskUserQuestion is AGENT-DRIVEN: the skill/seed
    # tells the agent to `curl POST /api/notifications/send` itself, so it
    # has direct visibility into push success/failure and decides whether
    # a given question is worth a phone buzz.
    try:
      await bc.publish_question({
        "type": "question",
        "question_id": pending.question_id,
        "questions": questions,
      })
    except Exception as exc:
      log.error(
        "AskUserQuestion save-before-broadcast failed chat_id=%s: %s",
        chat_id, exc,
      )
      if pending_questions.get(chat_id) is pending:
        pending_questions.pop(chat_id, None)
      return PermissionResultDeny(
        message=(
          "Could not save the question (persistence unavailable); not "
          "asking. Please try again."
        )
      )

    try:
      answers = await future
    except asyncio.CancelledError:
      if pending_questions.get(chat_id) is pending:
        pending_questions.pop(chat_id, None)
      return PermissionResultDeny(
        message="AskUserQuestion cancelled."
      )

    if pending_questions.get(chat_id) is pending:
      pending_questions.pop(chat_id, None)

    # Per docs: return updated_input with BOTH the original questions
    # array AND an answers dict {question_text: selected_label}.
    # The SDK passes this through as the tool input; AskUserQuestion's
    # implementation in headless mode echoes the answers back as the
    # tool result the model sees.
    return PermissionResultAllow(
      updated_input={
        "questions": questions,
        "answers": answers,
      }
    )

  # Required workaround per the SDK docs: a dummy PreToolUse hook
  # returning continue_=True keeps the stream open so can_use_tool can
  # be invoked. Without this, the stream closes before the callback
  # fires. See https://code.claude.com/docs/en/agent-sdk/user-input
  async def keepalive_hook(
    hook_input: dict[str, Any],
    tool_use_id: str | None,
    context: dict[str, Any],
  ) -> dict[str, Any]:
    del hook_input, tool_use_id, context
    return {"continue_": True}

  # Per-chat model/effort overrides flow in via `agent_settings`
  # (merged in chat.py from global defaults + Chat.agent_settings_json).
  # Both are session-wide on the SDK but Möbius spawns one `query()`
  # per turn, so passing them here applies to *this* turn — which is
  # exactly the "apply on next turn" semantics the slash picker promises.
  _model = (agent_settings or {}).get("model") or None
  _effort = (agent_settings or {}).get("effort") or None
  # Cross-provider mismatch defense (mirrors codex_sdk_runner).
  # Chats persisted before the snapshot logic learned to
  # provider-validate (see chat.py snapshot-on-first-send and
  # effective_agent_settings) can carry a Codex model on a Claude
  # chat. Sending that through here would surface as an obscure SDK
  # error. Quietly normalize so existing chats keep working.
  from app.providers import _model_belongs_to_other_provider, DEFAULT_MODELS
  if _model and _model_belongs_to_other_provider(_model, "claude"):
    log.warning(
      "claude turn started with non-claude model %r — normalizing to %r",
      _model, DEFAULT_MODELS["claude"],
    )
    _model = DEFAULT_MODELS["claude"]
  async def _run_once(model_override: str | None) -> RunnerResult:
    nonlocal current_session_id, cost_usd
    # Skills are gated behind the per-owner `skills_enabled` flag. OFF
    # (the default) keeps the historical posture: `setting_sources=None`
    # means the SDK loads NO user/project settings, so the Skill tool is
    # never offered and no skill can load. ON enables user+project
    # setting sources and `skills="all"` so the agent may load any
    # installed skill — a behavior-shifting change the owner opts into.
    # Observability (the skill_loaded event + activity log) lives in the
    # tool-use dispatch and works whenever a skill loads, independent of
    # this flag.
    options_kwargs = {
      "system_prompt": skill_text,
      "resume": session_id if session_id is not None else None,
      "cwd": cwd,
      "env": base_env,
      "setting_sources": (
        ["user", "project"] if skills_enabled else None
      ),
      "include_partial_messages": True,
      "can_use_tool": can_use_tool,
      "cli_path": "/usr/local/bin/claude",
      "hooks": {
        "PreToolUse": [
          HookMatcher(matcher=None, hooks=[keepalive_hook]),
        ],
      },
    }
    if skills_enabled:
      options_kwargs["skills"] = "all"
    if model_override:
      options_kwargs["model"] = model_override
    if _effort:
      options_kwargs["effort"] = _effort
    options = ClaudeAgentOptions(**options_kwargs)

    client = ClaudeSDKClient(options)
    active_client = ActiveClaudeClient(client, chat_id=chat_id)
    registry.register(active_client)

    try:
      try:
        await asyncio.wait_for(client.connect(), timeout=30.0)
      except asyncio.TimeoutError:
        bc.publish({
          "type": "error",
          "message": "Claude SDK failed to start (connect timeout)",
        })
        return {
          "session_id": current_session_id,
          "cost_usd": None,
          "error": "connect timeout",
        }
      await client.query(user_message)

      while True:
        async for sdk_msg in client.receive_response():
          # Persist the session id ONLY from real conversation messages.
          # SystemMessage and its subclasses — notably HookEventMessage,
          # which the codex plugin's SessionStart hook emits on every
          # resumed turn — carry a PHANTOM session id that gets a
          # `session-env/<id>` dir but never a transcript `.jsonl`.
          # Persisting that phantom overwrites Chat.session_id with an id
          # the CLI cannot resume, so the next turn dies "No conversation
          # found". Only StreamEvent/Assistant/User/Result carry the
          # resumable id (the same types dispatch advances the session from).
          if isinstance(
            sdk_msg,
            (StreamEvent, AssistantMessage, UserMessage, ResultMessage),
          ):
            incoming_session_id = getattr(sdk_msg, "session_id", None)
            if incoming_session_id and incoming_session_id != current_session_id:
              await _persist_session_id(db, chat_id, incoming_session_id)
          current_session_id, terminal = dispatch_sdk_message(
            sdk_msg, bc, current_session_id,
          )
          if terminal is None:
            continue
          steer_texts = active_client.pending_steer
          if steer_texts:
            active_client.pending_steer = []
            await client.query(
              _steer_redirect_message("\n\n".join(steer_texts))
            )
            break
          cost_usd = terminal.get("cost_usd")
          return terminal
        else:
          steer_texts = active_client.pending_steer
          if steer_texts:
            active_client.pending_steer = []
            await client.query(
              _steer_redirect_message("\n\n".join(steer_texts))
            )
            continue
          break

      return {
        "session_id": current_session_id,
        "cost_usd": cost_usd,
        "usage": None,
        "error": None,
      }
    except Exception as exc:
      return {
        "session_id": current_session_id,
        "cost_usd": None,
        "usage": None,
        "error": str(exc),
      }
    finally:
      current_handle = registry.get_handle(chat_id, RunnerKind.CLAUDE_SDK)
      if current_handle is active_client:
        registry.unregister(chat_id, RunnerKind.CLAUDE_SDK)
      pending = pending_questions.get(chat_id)
      if pending is not None and not pending.future.done():
        pending.future.cancel()
      if pending is not None:
        pending_questions.pop(chat_id, None)
      try:
        await client.disconnect()
      finally:
        active_client.mark_finished()

  result = await _run_once(_model)
  if _model and _should_retry_without_model(result.get("error")):
    log.warning(
      "Claude model %r unavailable for chat %s; retrying without explicit model",
      _model,
      chat_id,
    )
    return await _run_once(None)
  return result
