"""
ClaudeCodeAdapter — drives `claude` via a single persistent subprocess using
--input-format stream-json / --output-format stream-json.

One process lives for the full session lifetime. Messages are fed via stdin;
blocks stream from stdout continuously. The system:init event on startup
provides the live tool list and slash_commands — overrides the startup fallback list.

Stdin message format (one JSON line per turn):
  {"type":"user","message":{"role":"user","content":[{"type":"text","text":"…"}]}}

Permission responses (when Claude Code awaits approval):
  {"type":"permission_response","permission_id":"…","approved":true}

Working directory:
  Pass cwd= to __init__ to scope the agent to a specific project folder.
  known_working_dirs() returns all previously used Claude Code project dirs.
"""
from __future__ import annotations
import asyncio
import base64
import json
import logging
import os
import shutil
import uuid
from pathlib import Path
from typing import AsyncIterator

log = logging.getLogger(__name__)

from ..attachments import Attachment, build_text_with_inline
from ..blocks import (
    AgentInfoBlock, AudioHintBlock, BgTaskBlock, Block, CodeBlock, ErrorBlock, InputRequestBlock,
    PermissionRequestBlock, SessionEndBlock, SessionStatusHintBlock, SystemStatusBlock, TextBlock,
    ThinkingBlock, ToolCallBlock, tool_result_or_large,
)
from ..sentinel_tools import (
    SENTINEL_NAMES, SentinelServer, make_mcp_config,
    handle_sentinel_tool_use, has_pending_input, resolve_input,
)
from .base import (BaseAdapter, DetectedAgent, ExternalSession, split_markdown, BRIDGE_SLASH_COMMANDS,
                   agent_env, write_session_claude_md, build_append_system_prompt, tool_label_subject,
                   probe_binary)
from ..adapter_logging import setup_adapter_logger


# Well-known Claude Code slash commands shown in the iOS menu immediately,
# before system/init arrives. system/init overrides with the live list.
# Coalesce runs of partial text/thinking deltas into small batches before delivery. Claude
# Code (with --include-partial-messages) streams one content_block_delta per token; emitting
# a Block per token meant one JSON-serialize + AES-GCM encrypt + frame + WS-send PER TOKEN —
# the dominant active-turn CPU cost (bridge asyncio loop pegged >200% on a fast stream). We
# flush the buffer on either threshold; the concatenated chunk is byte-identical to the sum of
# the individual deltas, so iOS append-streaming is unchanged and done=True still carries the
# full text. Tunable via env for A/B without a code change.
_STREAM_COALESCE_MAX_CHARS = int(os.environ.get("BEACON_STREAM_COALESCE_CHARS", "256"))
_STREAM_COALESCE_MAX_SECS = float(os.environ.get("BEACON_STREAM_COALESCE_SECS", "0.05"))

_CC_NATIVE_COMMANDS = ["/compact", "/review", "/cost", "/memory", "/model", "/think"]
# Reasoning-effort levels accepted by `claude --effort` (empty applied value = CLI default).
_CC_EFFORT_LEVELS = ["low", "medium", "high", "xhigh", "max"]


def find_claude_binary() -> str | None:
    """Locate the `claude` CLI, PATH-independent.

    `shutil.which` alone fails under a non-login launch (open/launchd/systemd) whose
    PATH is the minimal GUI set without ~/.local/bin — the exact reason claude silently
    dropped out of detection after a menu-bar/`open` relaunch. `detector.ensure_cli_path`
    normally re-adds those dirs before detection; this is the belt-and-suspenders that
    also probes claude's known install locations directly, so detection can't depend on
    PATH being fixed first. Returns an absolute path or None."""
    hit = shutil.which("claude")
    if hit:
        return hit
    home = Path.home()
    for cand in (
        home / ".local" / "bin" / "claude",       # curl|sh installer default
        home / ".claude" / "local" / "claude",    # claude's own local install
        home / "bin" / "claude",
        home / ".bun" / "bin" / "claude",          # bun install -g
        home / ".npm-global" / "bin" / "claude",
        Path("/opt/homebrew/bin/claude"),
        Path("/usr/local/bin/claude"),
    ):
        if cand.is_file() and os.access(cand, os.X_OK):
            return str(cand)
    return None


def _read_claude_mcp_tools() -> list[dict]:
    config_path = Path.home() / ".claude" / "claude_desktop_config.json"
    if not config_path.exists():
        return []
    try:
        data = json.loads(config_path.read_text())
        servers = data.get("mcpServers", {})
        return [{"name": name, "type": "mcp_server", **cfg} for name, cfg in servers.items()]
    except Exception:
        return []


# known_working_dirs() only feeds the iOS new-session folder-picker SUGGESTIONS
# (NewSessionSheet.knownDirs) — not load-bearing (the picker also has client-side recent
# cwds + a live folder browser). So it's capped to the most-recently-active projects and
# cached briefly: it's called on every spawn / system-init / status-refresh, and there's no
# value in rescanning ~/.claude/projects (or surfacing every repo ever opened) each time.
_KWD_LIMIT = 24            # most-recently-modified project dirs to offer
_KWD_TTL = 30.0           # seconds to reuse the cached result
_kwd_cache: "tuple[float, list[str]] | None" = None


def _safe_mtime(p: Path) -> float:
    try:
        return p.stat().st_mtime
    except OSError:
        return 0.0


def known_working_dirs() -> list[str]:
    """Real filesystem paths for the most-recently-active Claude Code project dirs
    (most-recent first, capped at `_KWD_LIMIT`). Cached for `_KWD_TTL` s."""
    global _kwd_cache
    import time as _t
    now = _t.monotonic()
    if _kwd_cache is not None and now - _kwd_cache[0] < _KWD_TTL:
        return _kwd_cache[1]
    projects = Path.home() / ".claude" / "projects"
    result: list[str] = []
    if projects.exists():
        try:
            entries = [e for e in projects.iterdir() if e.is_dir()]
        except OSError:
            entries = []
        # Most-recently-modified first, then cap — the picker wants your ACTIVE projects,
        # and this bounds the decode work no matter how many stale dirs accumulate.
        entries.sort(key=_safe_mtime, reverse=True)
        home = str(Path.home())
        for entry in entries[:_KWD_LIMIT]:
            path = _decode_project_name(entry.name)
            if path and path != home:  # skip bare home dir
                result.append(path)
    _kwd_cache = (now, result)
    return result


def _decode_project_name(name: str) -> str | None:
    """Recover the real path from a ~/.claude/projects/<name> entry.

    Claude Code encodes project paths by stripping the leading slash and
    replacing every path separator with a hyphen. This is ambiguous when
    directory names contain hyphens, so we use backtracking with filesystem
    existence checks to find the correct split.
    """
    if not name.startswith("-"):
        return None
    tokens = name[1:].split("-")

    def search(idx: int, parts: list[str]) -> str | None:
        # Ancestor prune (CRITICAL): every element of `parts` except the last is a
        # *finalized* path component (no further tokens can be merged into it), so if
        # their joined prefix isn't a real directory, NO completion below can exist —
        # skip the entire subtree. Without this the hyphen-split backtracking is
        # O(2^hyphens): a name with many hyphens (e.g. a deleted pytest temp dir
        # `-private-var-folders-…-pytest-77-…`, 18 hyphens, decoding to a non-existent
        # path) explores the full 2^18 tree of splits, each doing an isdir() stat.
        # Measured: this blocked known_working_dirs() for ~6 s idle / 44–53 s under
        # load — on the event loop, inside adapter.start() on EVERY session spawn —
        # which stalled the whole bridge (slow spawns, "session expired" split-brain,
        # "no response"). See memory project_adapter_pathdecode_cpu_spin. The prune
        # turns it into a near-linear walk over only directories that actually exist.
        if len(parts) > 1:
            parent = "/" + "/".join(parts[:-1])
            if not os.path.isdir(parent):
                return None
        if idx == len(tokens):
            path = "/" + "/".join(parts)
            return path if os.path.isdir(path) else None
        token = tokens[idx]
        # Try token as a new path component
        r = search(idx + 1, parts + [token])
        if r:
            return r
        # Try merging token into the last component with a hyphen
        if parts:
            r = search(idx + 1, parts[:-1] + [parts[-1] + "-" + token])
            if r:
                return r
        return None

    return search(1, [tokens[0]])


# Only read details for this many most-recently-modified transcripts per scan.
# Older sessions are dropped by the per-adapter cap in headless anyway, and
# reading every transcript in full was a hundreds-of-MB spike every refresh.
_SCAN_LIMIT = 40
# Bytes read from the tail of a transcript when hunting for the last-prompt record.
_TAIL_BYTES = 64 * 1024
# A transcript written within this window of the scan is treated as a turn in flight
# (the agent appends to the .jsonl continuously while generating). Sized at ~1/3 of the
# client-connected scan cadence (60 s): a session actively working — including during
# tool waits, when the transcript goes briefly quiet — reliably reads as "turn" rather
# than flicking to "open". Too tight (the old 2.5 s) and a busy session almost never
# lands inside the window at scan time, so it shows as merely "open" (not running).
_TURN_RECENT_SECS = 20.0


def _read_session_meta(jsonl_file: Path, size: int) -> tuple[str, str]:
    """Return (started_at, last_prompt) by reading only the head + tail of the
    transcript — never the whole file. Claude transcripts are often multi-MB, so
    a full read() per file every refresh dominated bridge memory."""
    started_at = ""
    last_prompt = ""
    # started_at: the first record carrying a timestamp lives at the very top.
    try:
        with jsonl_file.open("r", errors="replace") as f:
            for _ in range(50):  # first handful of lines is more than enough
                line = f.readline()
                if not line:
                    break
                line = line.strip()
                if not line:
                    continue
                try:
                    obj = json.loads(line)
                except Exception:
                    continue
                if "timestamp" in obj:
                    started_at = obj["timestamp"]
                    break
    except Exception:
        pass
    # last_prompt: the "last-prompt" record sits near the end — scan only the tail.
    try:
        with jsonl_file.open("rb") as f:
            if size > _TAIL_BYTES:
                f.seek(-_TAIL_BYTES, os.SEEK_END)
                f.readline()  # drop the partial first line after the seek
            chunk = f.read().decode("utf-8", errors="replace")
        for line in reversed(chunk.splitlines()):
            line = line.strip()
            if not line:
                continue
            try:
                obj = json.loads(line)
            except Exception:
                continue
            if obj.get("type") == "last-prompt":
                last_prompt = obj.get("lastPrompt", "")
                break
    except Exception:
        pass
    return started_at, last_prompt


# Cmdline markers of a claude *daemon worker* (vs a top-level interactive TUI).
# The post-2.1 daemon architecture runs a thin `claude` TUI per session plus one or
# more helper processes — also basename `claude`, also lacking `--resume` — that carry
# these flags. They must NOT be counted as separate native sessions (they'd inflate the
# anonymous "?" count and make idle transcripts spuriously appear active).
_CLAUDE_WORKER_MARKERS = (
    "--session-id", "--agent", "--bg-pty-host", "--bg-spare", "--bg-", "--mcp-host",
    " daemon ", " daemon\t", "--output-format",
)


def _scan_claude_procs() -> tuple[dict[str, set[str]], dict[str, int]]:
    """Single ps/lsof sweep → (sids_by_cwd, anon_counts).

    sids_by_cwd: realpath(cwd) → set of session IDs (--resume arg, or "?" for a fresh
        process with no --resume). This is the ownership view — multiple "?" collapse.
    anon_counts: realpath(cwd) → number of *top-level interactive* claude TUIs with no
        --resume. Unlike the set, this preserves the count so the session scanner can
        surface N concurrent fresh sessions in one directory (not just the newest one).
        Daemon-worker helpers (see _CLAUDE_WORKER_MARKERS) are excluded from the count.

    Excludes the bridge's own spawned subprocesses (own_subtree_pids + beacon cmdline
    markers). Used by is_cwd_externally_owned for session-aware ownership: an external
    claude only conflicts with a Beacon session if it resumes the SAME session id (or is
    a fresh unknown process). A different --resume in the same dir is a different
    conversation — no conflict."""
    import re
    import subprocess
    from .base import own_subtree_pids, is_beacon_spawned_cmdline
    sids_by_cwd: dict[str, set[str]] = {}
    anon_counts: dict[str, int] = {}
    own = own_subtree_pids()  # exclude Beacon's own spawned claude subprocesses
    try:
        # Full command line (not just comm) so Beacon-spawned claude — including
        # orphans from a prior bridge instance, not in our subtree — can be excluded
        # by signature. The executable basename must be `claude` to count.
        out = subprocess.check_output(
            ["ps", "-axo", "pid=,command="], text=True, stderr=subprocess.DEVNULL
        )
        for line in out.splitlines():
            line = line.strip()
            parts = line.split(None, 1)
            if len(parts) != 2:
                continue
            pid, cmd = parts
            if not pid.isdigit():
                continue
            exe = cmd.split()[0] if cmd else ""
            if os.path.basename(exe).lower() not in ("claude",):
                continue
            if int(pid) in own or is_beacon_spawned_cmdline(cmd):
                continue
            # Extract the --resume <sid> from this process's command line.
            # "?" = fresh session with no known ID (conservative: may collide).
            m = re.search(r'--resume\s+(\S+)', cmd)
            sid = m.group(1) if m else "?"
            # A top-level interactive TUI is a fresh `claude` (no --resume) that
            # carries none of the daemon-worker markers — only these count toward the
            # per-cwd anonymous-session tally.
            is_worker = any(mk in cmd for mk in _CLAUDE_WORKER_MARKERS)
            is_anon_tui = sid == "?" and not is_worker
            try:
                # -a ANDs -p and -d (without it lsof ORs them → every process's
                # cwd). Yields exactly this pid's cwd as an `n<path>` line.
                cwd_out = subprocess.check_output(
                    ["lsof", "-p", pid, "-a", "-d", "cwd", "-Fn"],
                    text=True, stderr=subprocess.DEVNULL,
                )
                for lline in cwd_out.splitlines():
                    if lline.startswith("n/"):
                        cwd_real = os.path.realpath(lline[1:])
                        sids_by_cwd.setdefault(cwd_real, set()).add(sid)
                        if is_anon_tui:
                            anon_counts[cwd_real] = anon_counts.get(cwd_real, 0) + 1
            except Exception:
                pass
    except Exception:
        pass
    return sids_by_cwd, anon_counts


def _active_claude_sessions() -> dict[str, set[str]]:
    """External claude processes mapped by realpath(cwd) → set of session IDs.
    Thin wrapper over _scan_claude_procs (ownership view only). See that function."""
    return _scan_claude_procs()[0]


def _active_claude_cwds() -> set[str]:
    """Set of cwds with external claude processes. Shim over _active_claude_sessions
    for the session scanner (is_active/activity marking)."""
    return set(_active_claude_sessions().keys())


def _scan_claude_sessions(beacon_sids: frozenset[str] = frozenset()) -> list[ExternalSession]:
    """Scan ~/.claude/projects/ JSONL files for sessions started outside Beacon."""
    projects = Path.home() / ".claude" / "projects"
    if not projects.exists():
        return []

    # Find cwds of running `claude` processes and the specific external session IDs
    # per cwd (so we don't let beacon-managed transcripts in the same directory
    # pollute the tier-2 active markers — their files are freshly written and would
    # otherwise always win, making genuine external sessions appear as "idle").
    ext_sessions_by_cwd, anon_counts = _scan_claude_procs()  # cwd → set[sid|"?"], cwd → int
    active_cwds = set(ext_sessions_by_cwd.keys())

    # Tier-2 active transcript stems per cwd. A cwd running N anonymous (no --resume)
    # claude TUIs gets its N newest non-beacon transcripts marked active — so several
    # concurrent fresh sessions in one directory all surface, not just the newest.
    tier2_active_stems: dict[str, set[str]] = {}

    # Pass 1 — cheap stat-only discovery of every transcript (mtime + size).
    # We sort by recency and only read details for the newest _SCAN_LIMIT, so a
    # huge history doesn't translate into a huge per-refresh read.
    candidates: list[tuple[float, int, Path, str]] = []  # (mtime, size, file, cwd)
    for proj_dir in projects.iterdir():
        if not proj_dir.is_dir():
            continue
        cwd = _decode_project_name(proj_dir.name)
        if not cwd:
            continue
        for jsonl_file in proj_dir.glob("*.jsonl"):
            try:
                st = jsonl_file.stat()
            except OSError:
                continue
            candidates.append((st.st_mtime, st.st_size, jsonl_file, cwd))

    candidates.sort(key=lambda x: x[0], reverse=True)

    # Compute active transcripts per cwd using two tiers:
    #   1. Confirmed: session_id matches a known --resume <sid> from ps — highest
    #      priority, marked individually below (every confirmed sid is active).
    #   2. Fallback: cwd has unidentified "?" process(es) and no confirmed sids — pick
    #      the N newest transcripts that aren't beacon-managed sessions (beacon_sids),
    #      where N is the count of anonymous top-level TUIs in that cwd. candidates are
    #      sorted newest-first, so per cwd we accumulate the first N non-beacon stems.
    for _mtime, _size, jsonl_file, cwd in candidates:
        if cwd not in active_cwds:
            continue
        ext_sids = ext_sessions_by_cwd.get(cwd, set())
        known_ext = ext_sids - {"?"}
        if known_ext:
            continue  # tier-1 handled per-session via sid match
        if "?" not in ext_sids:
            continue
        # N = number of concurrent anonymous TUIs (≥1 if any "?" process exists, even
        # if only daemon workers were seen — preserves the old single-active fallback).
        n = anon_counts.get(cwd, 0) or 1
        active = tier2_active_stems.setdefault(cwd, set())
        if len(active) >= n:
            continue
        stem = jsonl_file.stem
        if stem in beacon_sids:
            continue
        active.add(stem)

    sessions_raw: list[tuple[float, ExternalSession]] = []  # (mtime, session)
    for mtime, size, jsonl_file, cwd in candidates[:_SCAN_LIMIT]:
        session_id = jsonl_file.stem
        started_at, last_prompt = _read_session_meta(jsonl_file, size)
        sessions_raw.append((mtime, ExternalSession(
            session_id=session_id,
            agent_id="claude-code",
            cwd=cwd,
            last_prompt=last_prompt,
            started_at=started_at,
            is_active=False,  # set below
        )))

    # Mark sessions as active and derive finer-grained liveness from transcript
    # write-recency: a turn in flight keeps appending, so fresh mtime ⇒ "turn";
    # alive process with a quiet transcript ⇒ "open" (idle at the prompt).
    #
    # is_active rules (in priority order):
    #   1. session_id is a *confirmed* external SID (appeared as --resume <sid>
    #      in ps output) → always active, regardless of mtime rank in the cwd.
    #   2. cwd has only "?" process(es) → the N newest non-beacon transcripts chosen
    #      by the tier-2 fallback (tier2_active_stems) are active, N = anon TUI count.
    import time as _time
    now = _time.time()
    result: list[ExternalSession] = []
    for mtime, s in sorted(sessions_raw, key=lambda x: x[0], reverse=True):
        ext_sids = ext_sessions_by_cwd.get(s.cwd, set())
        confirmed = ext_sids - {"?"}
        is_active = s.cwd in active_cwds and (
            s.session_id in confirmed                           # tier-1: explicit SID
            or (not confirmed and s.session_id in tier2_active_stems.get(s.cwd, set()))  # tier-2: N newest "?"
        )
        if not is_active:
            activity = "idle"
        elif (now - mtime) <= _TURN_RECENT_SECS:
            activity = "turn"
        else:
            activity = "open"
        result.append(ExternalSession(
            session_id=s.session_id,
            agent_id=s.agent_id,
            cwd=s.cwd,
            last_prompt=s.last_prompt,
            started_at=s.started_at,
            is_active=is_active,
            activity=activity,
        ))

    return result


class ClaudeCodeAdapter(BaseAdapter):
    cap_images = True
    cap_documents = False
    cap_sentinel = True
    cap_thinking = True
    cap_permissions = True
    cap_subagents = True
    cap_interrupt = True
    eager_respawn_on_exit = True   # claude self-exits rc=0 between turns; keep it warm
    cap_input_during_turn = True
    provides_models = True
    provides_usage = True
    agent_id = "claude-code"
    display_name = "Claude Code"
    beta = False              # live-verified via the adapter matrix

    @classmethod
    def detect(cls) -> "DetectedAgent | None":
        binary = find_claude_binary()
        if not binary:
            return None
        return DetectedAgent(
            id=cls.agent_id, name=cls.display_name, binary=binary,
            mcp_tools=_read_claude_mcp_tools(),
            available=probe_binary(binary),
            working_dirs=known_working_dirs(),
        )

    def __init__(
        self,
        binary: str | None = None,
        mcp_tools: list[dict] | None = None,
        cwd: str | None = None,
        sentinel_server: SentinelServer | None = None,
        skip_permissions: bool = False,
        permission_mode: str = "default",
        effort: str | None = None,
        resume_session_id: str | None = None,
        on_claude_session_id: "Callable[[str], None] | None" = None,
        model: str | None = None,
        beacon_id: str | None = None,
    ) -> None:
        self._binary = binary or find_claude_binary() or "claude"
        self._cwd = cwd
        self._beacon_id = beacon_id
        self._skip_permissions = skip_permissions
        self._permission_mode = permission_mode
        self._effort = effort
        self._startup_model = model
        self._resume_session_id = resume_session_id
        self._on_claude_session_id = on_claude_session_id
        self._proc: asyncio.subprocess.Process | None = None
        self._fetch_store: dict[str, str] = {}
        self._pending_tools: dict[str, ToolCallBlock] = {}
        self._pending_ask_question: tuple[str, list[str]] | None = None  # (tool_use_id, option_labels)
        # Live plan via the granular task tools (TaskCreate/TaskUpdate/TaskList,
        # which replaced the whole-list TodoWrite). They arrive one change at a
        # time, so we accumulate them and re-emit the full TodoBlock on each edit.
        self._tasks: dict[str, dict] = {}            # CLI task id -> {content, status}
        self._pending_task_create: dict[str, str] = {}  # tool_use_id -> subject (id learned from result)
        self._sentinel_server = sentinel_server
        self._session_id: str = ""
        self._mcp_config_path: Path | None = None
        # Streaming state — tracks active content_block_delta sequences
        self._stream_block_id: str | None = None      # id of the in-progress TextBlock
        self._stream_block_text: str = ""             # accumulated text so far
        self._thinking_active: bool = False           # True while inside a thinking content block
        self._thinking_text: str = ""                 # accumulated thinking text
        # Usage accumulators — updated from each "result" event
        self._total_input_tokens: int = 0
        self._total_output_tokens: int = 0
        self._total_cache_read_tokens: int = 0
        self._total_cache_creation_tokens: int = 0
        self._total_cost_usd: float = 0.0
        self._total_turns: int = 0
        self._last_input_tokens: int = 0   # last turn only — proxy for current context size
        self._total_web_searches: int = 0
        self._total_web_fetches: int = 0
        # Background tasks (background Bash + Agent/Task sub-agents) surface on stdout as
        # structured `system/task_*` events (task_started / task_progress / task_notification
        # / task_updated), each keyed by a stable `task_id`. These dedupe sets coalesce the
        # start (and the duplicate completion from task_notification + task_updated) so we
        # emit exactly one BgTaskBlock(started) and one BgTaskBlock(completed/failed) per id.
        self._seen_bg_starts: set[str] = set()
        self._done_bg_tasks: set[str] = set()

        # Adapter-specific per-session log
        self._adapter_logger: logging.Logger | None = None
        self._log = self._log_debug = self._log_warn = self._log_err = lambda _: None
        self._permission_denials: int = 0
        self._model: str = ""
        self._context_window: int = 0
        self._agent_info_key: str = ""
        self._current_claude_session_id: str = ""
        self._pending_model_switch: str | None = None
        self._pending_restart: bool = False   # generic restart (e.g. effort change)

    @classmethod
    def scan_external_sessions(cls, *, beacon_sids: frozenset[str] = frozenset()) -> list[ExternalSession]:  # type: ignore[override]
        return _scan_claude_sessions(beacon_sids=beacon_sids)

    @classmethod
    def transcript_path(cls, ext_sid: str) -> "Path | None":
        """Return the JSONL transcript file for `ext_sid`, or None if not found."""
        projects = Path.home() / ".claude" / "projects"
        if not projects.exists():
            return None
        for proj_dir in projects.iterdir():
            p = proj_dir / f"{ext_sid}.jsonl"
            if p.exists():
                return p
        return None

    @classmethod
    def has_changes_since(cls, ts: float) -> bool:
        """True if any JSONL in ~/.claude/projects/ has been written since `ts`."""
        projects = Path.home() / ".claude" / "projects"
        if not projects.exists():
            return False
        for proj_dir in projects.iterdir():
            try:
                if proj_dir.stat().st_mtime > ts:
                    return True
                for f in proj_dir.glob("*.jsonl"):
                    if f.stat().st_mtime > ts:
                        return True
            except OSError:
                pass
        return False

    @classmethod
    def is_cwd_externally_owned(cls, cwd: str, sid: str = "") -> bool:
        if not cwd:
            return False
        target = os.path.realpath(os.path.expanduser(cwd))
        external_sids = _active_claude_sessions().get(target)
        if not external_sids:
            return False
        if not sid:
            # No specific session to match — any external claude blocks.
            return True
        # Block only if the external claude is resuming OUR specific session.
        # A "?" (fresh, no --resume) or a different --resume are separate conversations
        # that share no session file with this sid — no write conflict.
        return sid in external_sids

    @classmethod
    def make_transcript_mirror(cls, cwd: str, sid: str, *, from_end: bool = False):
        from ..transcript_mirror import make_claude_mirror
        return make_claude_mirror(cwd, sid, from_end=from_end)

    @classmethod
    def terminal_resume_command(cls, sid: str) -> str | None:
        import shlex
        return f"claude --resume {shlex.quote(sid)}" if sid else "claude"

    @classmethod
    def latest_sid_for_cwd(cls, cwd: str) -> str | None:
        """Newest claude session id whose transcript lives under `cwd`.

        `--resume` can mint a *new* session id for the resumed branch, so the
        stored `claude_session_id` drifts behind reality after a native-CLI turn.
        Re-resolving from disk (newest transcript mtime in the cwd's project dir)
        gives the true lineage head to resume — used by the session-handoff path so
        Beacon picks up whatever the terminal last wrote. See docs/session-handoff.md."""
        projects = Path.home() / ".claude" / "projects"
        if not projects.exists():
            return None
        target = os.path.realpath(os.path.expanduser(cwd))
        newest: tuple[float, str] | None = None
        for proj_dir in projects.iterdir():
            if not proj_dir.is_dir():
                continue
            decoded = _decode_project_name(proj_dir.name)
            if not decoded or os.path.realpath(decoded) != target:
                continue
            for jsonl_file in proj_dir.glob("*.jsonl"):
                try:
                    mtime = jsonl_file.stat().st_mtime
                except OSError:
                    continue
                if newest is None or mtime > newest[0]:
                    newest = (mtime, jsonl_file.stem)
        return newest[1] if newest else None

    @classmethod
    async def list_models(cls, extra: dict | None = None) -> list[dict]:
        from .base import fetch_anthropic_models
        return await fetch_anthropic_models(fallback=[
            {"id": "claude-sonnet-4-6",         "label": "Sonnet 4.6", "detail": "Balanced speed & capability"},
            {"id": "claude-opus-4-8",           "label": "Opus 4.8",   "detail": "Most powerful, slower"},
            {"id": "claude-fable-5",            "label": "Fable 5",    "detail": "Newest"},
            {"id": "claude-haiku-4-5-20251001", "label": "Haiku 4.5",  "detail": "Fastest, cheapest"},
        ])

    @classmethod
    def readiness(cls) -> tuple[bool, str]:
        home = Path.home()
        if os.environ.get("ANTHROPIC_API_KEY") or (home / ".claude.json").exists() \
                or (home / ".claude").exists():
            return True, ""
        return False, "Sign in with `claude` (or set ANTHROPIC_API_KEY)."

    # ── Lifecycle ────────────────────────────────────────────────────────────

    async def start(self, session_id: str) -> AgentInfoBlock:
        self._session_id = session_id
        
        # Setup session-specific adapter log
        self._adapter_logger = setup_adapter_logger(
            "claude_code",
            session_id,
            extra={"model": self._startup_model or "default", "cwd": self._cwd}
        )
        self._log = self._adapter_logger.info
        self._log_debug = self._adapter_logger.debug
        self._log_warn = self._adapter_logger.warning
        self._log_err = self._adapter_logger.error
        self._log(f"Starting claude process: model={self._startup_model}")
        
        cmd = [
            self._binary,
            "--output-format", "stream-json",
            "--input-format", "stream-json",
            "--include-partial-messages",
            "--verbose", "-p",
            # Disable the native AskUserQuestion: in headless -p mode the CLI auto-stubs
            # it with "Answer questions?" and ends the turn (it can't present its own UI),
            # while the bridge separately shows a card whose answer then lands as a stray
            # tool_result on an already-finished turn. The sentinel request_user_input MCP
            # tool does the same job and works correctly, so steer the model to that.
            "--disallowedTools", "AskUserQuestion",
        ]
        if self._skip_permissions:
            cmd.append("--dangerously-skip-permissions")
        else:
            cmd += ["--permission-mode", self._permission_mode]
        if self._effort:
            cmd += ["--effort", self._effort]
        if self._startup_model:
            cmd += ["--model", self._startup_model]
        if self._resume_session_id:
            cmd += ["--resume", self._resume_session_id]
        if self._sentinel_server and self._sentinel_server.port:
            import tempfile
            cfg = make_mcp_config(self._sentinel_server.port, session_id)
            tmp = tempfile.NamedTemporaryFile(
                mode="w", suffix=".json", delete=False, prefix="beacon_mcp_"
            )
            import json as _json
            tmp.write(_json.dumps(cfg))
            tmp.flush()
            tmp.close()
            self._mcp_config_path = Path(tmp.name)
            cmd += ["--mcp-config", str(self._mcp_config_path)]
            # Pre-allow beacon sentinel tools so they run without per-call permission prompts,
            # while leaving all other tool permissions at their normal level.
            if not self._skip_permissions:
                cmd += ["--allowedTools", "mcp__terminull__*"]
        write_session_claude_md(self._cwd)
        append_prompt = build_append_system_prompt(beacon_id=self._beacon_id)
        if append_prompt:
            cmd += ["--append-system-prompt", append_prompt]
        log.debug("invoke: %s", " ".join(cmd))
        self._proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.DEVNULL,
            cwd=self._cwd,
            # The sentinel request_user_input / request_permission tools are MCP tools
            # that block until the *human* answers. Claude Code's default MCP tool-call
            # timeout (~60s) would abort them while the user is still deciding, leaving
            # the model to give up on the dialog (the "empty message" / broken-dialog
            # symptom). Give human-in-the-loop tools effectively unlimited time (24h).
            env=agent_env({"MCP_TOOL_TIMEOUT": "86400000", "MCP_TIMEOUT": "86400000"}),
            limit=16 * 1024 * 1024,  # 16 MB — large tool results can exceed the 64 KB default
        )
        # Return a basic AgentInfoBlock immediately — the real tool list and version
        # arrive in the first "system/init" event from the CLI, which only comes
        # after the first stdin message.  stream() will emit a fresh AgentInfoBlock
        # when it sees the init event so the UI gets updated automatically.
        return self.make_agent_info(
            name="Claude Code",
            slash_commands=_CC_NATIVE_COMMANDS,
            working_dirs=known_working_dirs(),
            effort_levels=_CC_EFFORT_LEVELS,
            effort=self._effort or "",
        )

    def clear_history(self) -> None:
        # Claude Code keeps history inside its own subprocess — there's no in-memory
        # history list to wipe. To actually clear context we must kill the process;
        # the next send() will start a fresh one without --resume.
        self._resume_session_id = None
        # Also forget the live claude session id — otherwise the run loop's restart
        # branch restores _resume_session_id from it and resurrects the old context.
        self._current_claude_session_id = ""
        # Context is now empty — reset the "% full" proxy so a fresh status hint
        # reports 0 rather than the pre-clear occupancy (the pill stays stale until
        # the next real turn otherwise).
        self._last_input_tokens = 0
        if self._on_claude_session_id:
            self._on_claude_session_id("")
        # Restart the held process fresh (no --resume) rather than letting the run loop
        # break on the dead process — so the next turn has a live, empty session ready.
        self._pending_restart = True
        if self._proc and self._proc.returncode is None:
            try:
                self._proc.stdin.close()
            except Exception:
                pass
            self._proc.terminate()

    async def stop(self) -> None:
        if self._proc and self._proc.returncode is None:
            try:
                self._proc.stdin.close()
            except Exception:
                pass
            # SIGINT (Ctrl+C) lets Claude Code flush its session DB before exiting,
            # which keeps --resume working after /stop. Fall back to SIGTERM if it
            # doesn't exit within 5 s.
            import signal as _signal
            try:
                self._proc.send_signal(_signal.SIGINT)
            except ProcessLookupError:
                pass
            try:
                await asyncio.wait_for(self._proc.wait(), timeout=5.0)
            except asyncio.TimeoutError:
                self._proc.terminate()
                await self._proc.wait()
        if self._mcp_config_path and self._mcp_config_path.exists():
            try:
                self._mcp_config_path.unlink()
            except Exception:
                pass
            self._mcp_config_path = None

    # ── I/O ──────────────────────────────────────────────────────────────────

    async def send(self, text: str, attachments: list[Attachment] | None = None) -> None:
        if self._pending_ask_question is not None:
            question_id, opts = self._pending_ask_question
            self._pending_ask_question = None
            answer = text.strip()
            if opts:
                try:
                    idx = int(answer) - 1
                    if 0 <= idx < len(opts):
                        answer = opts[idx]
                except ValueError:
                    pass
            await self._write({
                "type": "user",
                "message": {"role": "user", "content": [
                    {"type": "tool_result", "tool_use_id": question_id,
                     "content": json.dumps({"answers": [{"selectedOptions": [answer]}]})},
                ]},
            })
            return

        atts = attachments or []
        content: list[dict] = []

        doc_atts = [a for a in atts if not a.media_type.startswith("image/")]
        full_text = await asyncio.to_thread(build_text_with_inline, text, doc_atts)
        content.append({"type": "text", "text": full_text})

        for att in atts:
            if att.media_type.startswith("image/"):
                ocr = att.ocr_text()
                if ocr:
                    content.append({"type": "text", "text": f"[Image OCR text: {ocr}]"})
                data = att.local_path.read_bytes()
                content.append({
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": att.media_type,
                        "data": base64.b64encode(data).decode(),
                    },
                })

        await self._write({"type": "user", "message": {"role": "user", "content": content}})

    def set_model(self, model: str) -> None:
        self._startup_model = model
        self._model = model
        if self._proc and self._proc.returncode is None:
            # Mark a restart-with-new-model is pending, then close stdin so the
            # subprocess finishes its current turn and exits.  _run() in Session
            # will detect _pending_model_switch and restart with --resume + --model.
            self._pending_model_switch = model
            try:
                self._proc.stdin.close()
            except Exception:
                pass
            log.info("ClaudeCodeAdapter: model switch to %r requested — subprocess will restart after current turn", model)
        else:
            log.info("ClaudeCodeAdapter: model set to %r (no running subprocess)", model)

    def set_effort(self, level: str) -> None:
        """Change the reasoning-effort level (`--effort`). Like set_model, a running
        subprocess finishes its turn then restarts (via _pending_restart) so the new
        `--effort` applies; otherwise it takes effect on the next start()."""
        self._effort = (level or "").strip() or None
        if self._proc and self._proc.returncode is None:
            self._pending_restart = True
            try:
                self._proc.stdin.close()
            except Exception:
                pass
            log.info("ClaudeCodeAdapter: effort switch to %r requested — subprocess will restart after current turn", level)
        else:
            log.info("ClaudeCodeAdapter: effort set to %r (no running subprocess)", level)

    async def respond_permission(self, permission_id: str, approved: bool) -> None:
        if not approved:
            self._permission_denials += 1
        await self._write({"type": "permission_response", "permission_id": permission_id, "approved": approved})

    async def stream(self) -> AsyncIterator[Block]:
        assert self._proc is not None
        # Delta coalescer state (see _STREAM_COALESCE_* above). We buffer consecutive
        # same-id partial TextBlocks / partial ThinkingBlocks and flush a single merged
        # block on size/time threshold, when a different block must be yielded (order
        # preserved), or when the stream ends. done=True blocks pass straight through.
        import time as _t
        pend_text: list[str] = []
        pend_text_id: str | None = None
        pend_think: list[str] = []
        pend_since = 0.0

        def _flush_text():
            nonlocal pend_text, pend_text_id, pend_since
            if pend_text:
                out = TextBlock(text="".join(pend_text), done=False, id=pend_text_id or "")
                pend_text = []
                pend_since = 0.0
                return out
            return None

        def _flush_think():
            nonlocal pend_think, pend_since
            if pend_think:
                out = ThinkingBlock(thinking="".join(pend_think), done=False)
                pend_think = []
                pend_since = 0.0
                return out
            return None

        def _ripe(n: int) -> bool:
            return bool(n >= _STREAM_COALESCE_MAX_CHARS
                        or (pend_since > 0 and _t.monotonic() - pend_since >= _STREAM_COALESCE_MAX_SECS))

        async for raw in self._proc.stdout:
            line = raw.decode().strip()
            if not line:
                continue
            log.debug("← %s", line[:500])
            try:
                event = json.loads(line)
            except json.JSONDecodeError:
                continue
            async for block in self._parse(event):
                # Partial text delta → buffer (flushing any pending thinking first).
                if isinstance(block, TextBlock) and not block.done and block.id:
                    if (b := _flush_think()) is not None:
                        yield b
                    if pend_text_id is not None and block.id != pend_text_id:
                        if (b := _flush_text()) is not None:
                            yield b
                    pend_text_id = block.id
                    if not pend_text:
                        pend_since = _t.monotonic()
                    pend_text.append(block.text)
                    if _ripe(sum(len(c) for c in pend_text)) and (b := _flush_text()) is not None:
                        yield b
                    continue
                # Partial thinking delta → buffer (flushing any pending text first).
                if isinstance(block, ThinkingBlock) and not block.done:
                    if (b := _flush_text()) is not None:
                        yield b
                    if not pend_think:
                        pend_since = _t.monotonic()
                    pend_think.append(block.thinking)
                    if _ripe(sum(len(c) for c in pend_think)) and (b := _flush_think()) is not None:
                        yield b
                    continue
                # Any other block (incl. done=True text/thinking): flush buffers, preserve order.
                if (b := _flush_text()) is not None:
                    yield b
                if (b := _flush_think()) is not None:
                    yield b
                yield block
        # Drain any trailing partials on stream close.
        if (b := _flush_text()) is not None:
            yield b
        if (b := _flush_think()) is not None:
            yield b

    # ── Parsing ───────────────────────────────────────────────────────────────

    async def _parse(self, event: dict) -> AsyncIterator[Block]:
        etype = event.get("type")
        subtype = event.get("subtype", "")

        if etype == "system" and subtype in ("hook_started", "hook_response", "hook_stop_occurred"):
            return
        # Background-task lifecycle (background Bash + Agent/Task sub-agents). These
        # structured events are the authoritative signal — they supersede the old
        # text-regex on the tool_result and the file-stability watcher. `task_id` is
        # stable across the lifecycle and equals the launch agentId / bg id.
        if etype == "system" and subtype == "task_started":
            tid = event.get("task_id", "")
            if tid and tid not in self._seen_bg_starts:
                self._seen_bg_starts.add(tid)
                yield BgTaskBlock(task_id=tid, description=event.get("description", ""),
                                  status="started")
            return
        if etype == "system" and subtype in ("task_notification", "task_updated"):
            tid = event.get("task_id", "")
            status = event.get("status") or (event.get("patch") or {}).get("status") or ""
            if tid and status in ("completed", "failed", "error", "cancelled") \
                    and tid not in self._done_bg_tasks:
                self._done_bg_tasks.add(tid)
                yield BgTaskBlock(
                    task_id=tid, description=event.get("summary", ""),
                    status="completed" if status == "completed" else "failed",
                )
            return
        if etype == "system" and subtype == "task_progress":
            # Rich per-step progress (usage / last_tool_name / subagent_type). Not surfaced
            # as a block yet; the deferred-end progress line already keeps the indicator alive.
            return
        if etype == "system" and subtype == "init":
            claude_session_id = event.get("session_id", "")
            if claude_session_id:
                self._current_claude_session_id = claude_session_id
                if self._on_claude_session_id:
                    self._on_claude_session_id(claude_session_id)
            tools = [{"name": n} for n in event.get("tools", [])]
            version = event.get("claude_code_version", "")
            info_key = f"{version}:{','.join(t['name'] for t in tools)}"
            if info_key == self._agent_info_key:
                return
            self._agent_info_key = info_key
            # Normalise: Claude Code emits commands without a leading "/" (e.g. "compact"),
            # but BRIDGE_SLASH_COMMANDS all have one. Ensure every entry starts with "/".
            bridge_set = {c.lstrip("/") for c in BRIDGE_SLASH_COMMANDS}
            native = [
                (c if c.startswith("/") else f"/{c}")
                for c in event.get("slash_commands", [])
                if c.lstrip("/") not in bridge_set
            ]
            yield self.make_agent_info(
                name="Claude Code",
                version=version,
                tools=tools,
                slash_commands=(native or _CC_NATIVE_COMMANDS),
                working_dirs=known_working_dirs(),
                effort_levels=_CC_EFFORT_LEVELS,
                effort=self._effort or "",
            )
            return
        if etype == "stream_event":
            inner = event.get("event", {})
            itype = inner.get("type", "")
            if itype == "message_start":
                msg = inner.get("message", {})
                m = msg.get("model", "")
                # Only the MAIN agent's model drives the session pill. Task/Explore
                # sub-agents (parent_tool_use_id set) run on cheaper models (haiku/
                # sonnet) and must NOT clobber self._model — otherwise the pill flips
                # to the sub-agent's model mid-turn (the "session switched to haiku"
                # UI bug). get_usage() reports self._model straight to the iOS pill.
                if m and event.get("parent_tool_use_id") is None:
                    self._model = m
                # Current context occupancy = the prompt size of THIS single API request
                # (input + both cache buckets, all known at message_start). Each request in
                # the agentic loop overwrites it, so it ends at the most-recent request's
                # size. This is the correct "% full" source — unlike the result event's
                # `usage`, which is the CUMULATIVE sum across every request in the turn
                # (num_turns can be dozens) and would push the ratio far past 100%.
                u = msg.get("usage") or {}
                req_input = (u.get("input_tokens", 0)
                             + u.get("cache_read_input_tokens", 0)
                             + u.get("cache_creation_input_tokens", 0))
                if req_input > 0:
                    self._last_input_tokens = req_input
            if itype == "content_block_start":
                cb = inner.get("content_block", {})
                if cb.get("type") == "text":
                    import uuid as _uuid
                    self._stream_block_id = str(_uuid.uuid4())
                    self._stream_block_text = cb.get("text", "")
                    yield AudioHintBlock(state="agent_speaking")
                elif cb.get("type") == "thinking":
                    self._thinking_active = True
                    self._thinking_text = ""
                    yield AudioHintBlock(state="agent_silent")
                    # Don't emit a ThinkingBlock yet. Current Claude Code streams
                    # thinking_delta with empty text (only `estimated_tokens` progress) —
                    # the raw reasoning is withheld upstream. Emitting on start produced an
                    # empty "thinking" card that rendered as just a header. We now emit only
                    # once real thinking text actually arrives (below); the live "Thinking…"
                    # indicator covers the turn meanwhile.
            elif itype == "content_block_delta":
                delta = inner.get("delta", {})
                if delta.get("type") == "text_delta" and self._stream_block_id:
                    chunk = delta.get("text", "")
                    self._stream_block_text += chunk
                    yield TextBlock(text=chunk, done=False, id=self._stream_block_id)
                elif delta.get("type") == "thinking_delta" and self._thinking_active:
                    chunk = delta.get("thinking", "")
                    if chunk:                        # skip empty progress-only deltas
                        self._thinking_text += chunk
                        yield ThinkingBlock(thinking=chunk, done=False)
            elif itype == "content_block_stop":
                if self._stream_block_id:
                    yield TextBlock(text=self._stream_block_text, done=True, id=self._stream_block_id)
                    self._stream_block_id = None
                    self._stream_block_text = ""
                elif self._thinking_active:
                    if self._thinking_text:          # no empty card when text was withheld
                        yield ThinkingBlock(thinking=self._thinking_text, done=True)
                    self._thinking_active = False
                    self._thinking_text = ""
            return

        if etype == "debug":
            return

        if etype == "rate_limit_event":
            import time
            info     = event.get("rate_limit_info") or {}
            resets_at = info.get("resetsAt") or 0
            util     = info.get("utilization") or 0.0
            warning  = info.get("status") == "allowed_warning"
            if resets_at:
                remaining_m = max(0, int((resets_at - time.time()) / 60))
                reset_str = f"{remaining_m}m" if remaining_m else "soon"
                msg = f"{'Rate limit warning' if warning else 'Rate limited'} — resets in {reset_str} ({int(util * 100)}% used)"
            else:
                retry_s = (event.get("retry_after_ms") or 0) // 1000
                msg = f"Rate limited — retry in {retry_s}s" if retry_s else "Rate limited"
            yield SystemStatusBlock(event="rate_limited", message=msg)
            return

        if etype == "compacting" or (etype == "system" and subtype == "compacting") \
                or (etype == "system" and subtype == "status" and event.get("status") == "compacting"):
            yield SystemStatusBlock(event="compacting")
            return

        if etype == "system" and subtype == "compact_boundary":
            meta = event.get("compact_metadata") or {}
            pre  = meta.get("pre_tokens", 0)
            post = meta.get("post_tokens", 0)
            self._last_input_tokens = post
            log.info("compact_boundary: %d → %d tokens (%.0fs)",
                     pre, post, (meta.get("duration_ms") or 0) / 1000)
            yield SystemStatusBlock(event="compaction_done")
            # Push the shrunk occupancy NOW. The pill is otherwise refreshed only on
            # the `result` event (which may be far off, or not arrive at all for a bare
            # /compact), so it would stay pinned at the pre-compact value — i.e. ~100%,
            # since you compact precisely when the context is full.
            if not self._context_window:
                from .base import context_window_for_model
                self._context_window = (context_window_for_model(self._startup_model or "")
                                        or context_window_for_model(self._model or ""))
            if self._context_window and self._last_input_tokens > self._context_window:
                self._context_window = 1_000_000
            if self._context_window:
                log.debug("ctx-debug(compact): last_input=%d window=%d pct=%d",
                         self._last_input_tokens, self._context_window,
                         int(self._last_input_tokens / self._context_window * 100))
                yield SessionStatusHintBlock(
                    model=self._model,
                    turns=self._total_turns,
                    cost_usd=self._total_cost_usd,
                    output_tokens=self._total_output_tokens,
                    cache_creation_tokens=self._total_cache_creation_tokens,
                    last_input_tokens=self._last_input_tokens,
                    context_window=self._context_window,
                )
            # Mid-turn boundary: the turn resumes after compaction. partial=True so the
            # client doesn't finalize the turn (and doesn't show a real end pill).
            yield SessionEndBlock(reason="completed", partial=True)
            return

        if etype == "system" and subtype == "status" and event.get("compact_result"):
            return  # compaction success/fail status — compact_boundary carries the real info

        if etype == "compaction_delta":
            delta = event.get("delta", "")
            if delta:
                yield TextBlock(text=delta)
            return

        if etype == "assistant":
            yield AudioHintBlock(state="agent_speaking")
            stop_reason = event.get("message", {}).get("stop_reason")
            for content in event.get("message", {}).get("content", []):
                ctype = content.get("type")
                if ctype == "text":
                    text = content.get("text", "")
                    # Skip if already streamed via stream_event deltas (stop_reason is None
                    # on partial assistant events; only the final one has stop_reason set)
                    if text and stop_reason is not None:
                        for block in split_markdown(text):
                            yield block
                elif ctype == "tool_use":
                    name = content["name"]
                    # Claude Code prefixes MCP tools as "mcp__<server>__<tool>"; strip for sentinel lookup
                    bare_name = name.removeprefix("mcp__terminull__")
                    if bare_name in SENTINEL_NAMES:
                        # Sentinel tool — MCP subprocess handles execution and IPC.
                        # We just record the id so we can suppress the result event.
                        self._pending_tools[content["id"]] = None  # type: ignore[assignment]
                    elif name.lower() == "todowrite":
                        # Phase/checklist updates. Claude Code names this tool
                        # "TodoWrite"; OpenCode (which reuses this adapter) names it
                        # "todowrite" — match case-insensitively so both get the live
                        # card. The agent re-sends the whole list each time; emit a
                        # generic TodoBlock with a stable per-session id so iOS upserts
                        # one live card in place. Suppress the noisy "Todos have updated
                        # successfully" result like sentinel tools.
                        todos = content.get("input", {}).get("todos", [])
                        self._pending_tools[content["id"]] = None  # type: ignore[assignment]
                        yield self.todo_block(self._session_id, todos)
                        yield AudioHintBlock(state="agent_silent")
                    elif name in ("TaskCreate", "TaskUpdate", "TaskList"):
                        # Granular task tools (replaced the whole-list TodoWrite).
                        # Accumulate into per-session state and re-emit the full live
                        # TodoBlock so iOS upserts one card in place. Raw tool cards +
                        # noisy results are suppressed like TodoWrite/sentinel tools.
                        inp = content.get("input", {})
                        self._pending_tools[content["id"]] = None  # type: ignore[assignment]
                        if name == "TaskCreate":
                            # The CLI assigns the numeric id; we learn it from the
                            # result text, so defer emitting the card until then.
                            self._pending_task_create[content["id"]] = inp.get("subject", "")
                        elif name == "TaskUpdate":
                            tid = str(inp.get("taskId", ""))
                            task = self._tasks.get(tid)
                            if task is not None:
                                status = inp.get("status")
                                if status == "deleted":
                                    self._tasks.pop(tid, None)
                                elif status:
                                    task["status"] = status
                                if inp.get("subject"):
                                    task["content"] = inp["subject"]
                            if self._tasks:
                                yield self.todo_block(self._session_id, list(self._tasks.values()))
                        # TaskList is read-only — nothing to update.
                        yield AudioHintBlock(state="agent_silent")
                    elif name == "AskUserQuestion":
                        questions = content.get("input", {}).get("questions", [])
                        q = questions[0] if questions else {}
                        opts = [o["label"] for o in q.get("options", []) if "label" in o]
                        self._pending_ask_question = (content["id"], opts)
                        self._pending_tools[content["id"]] = None  # type: ignore[assignment]
                        yield InputRequestBlock(
                            id=content["id"],
                            prompt=q.get("question", ""),
                            input_type="choice" if opts else "text",
                            options=opts,
                        )
                        yield AudioHintBlock(state="agent_silent")
                    else:
                        args = content.get("input", {})
                        lbl, subj = tool_label_subject(name, args)
                        tc = ToolCallBlock(
                            id=content["id"],
                            name=name,
                            args=args,
                            status="running",
                            label=lbl,
                            subject=subj,
                        )
                        self._pending_tools[tc.id] = tc
                        yield AudioHintBlock(state="agent_silent")
                        yield tc
                elif ctype == "server_tool_use":
                    # Anthropic's native web-search/web-fetch tool: executed server-side
                    # (billed via usage.server_tool_use, see _total_web_searches below),
                    # unlike every other tool_use above which the CLI dispatches itself.
                    # Previously unhandled — silently rendered nothing, which let two
                    # otherwise-unrelated TextBlocks (the answer before and after the
                    # search) land back-to-back with no other block between them.
                    name = content.get("name") or "web_search"
                    args = content.get("input", {})
                    lbl, subj = tool_label_subject(name, args)
                    tc = ToolCallBlock(
                        id=content["id"],
                        name=name,
                        args=args,
                        status="running",
                        label=lbl,
                        subject=subj,
                    )
                    self._pending_tools[tc.id] = tc
                    yield AudioHintBlock(state="agent_silent")
                    yield tc
                elif ctype in ("web_search_tool_result", "web_fetch_tool_result"):
                    tool_use_id = content.get("tool_use_id", "")
                    pending = self._pending_tools.pop(tool_use_id, None)
                    result = content.get("content")
                    is_error = isinstance(result, dict) and "error_code" in result
                    if pending is not None:
                        if is_error:
                            preview = f"error: {result.get('error_code', 'search failed')}"
                            status = "error"
                        else:
                            n = len(result) if isinstance(result, list) else 0
                            preview = f"{n} result{'s' if n != 1 else ''}"
                            status = "done"
                        yield ToolCallBlock(
                            id=pending.id,
                            name=pending.name,
                            args=pending.args,
                            status=status,
                            label=pending.label,
                            subject=pending.subject,
                            preview=preview,
                        )
                    yield AudioHintBlock(state="agent_speaking")
                elif ctype == "thinking":
                    pass  # already emitted as ThinkingBlock(done=True) via stream_event content_block_stop

        elif etype == "tool":
            # Legacy format (older claude CLI versions)
            tool_use_id = event.get("tool_use_id", "")
            async for block in self._handle_tool_result(tool_use_id, str(event.get("content", "")), False):
                yield block

        elif etype == "user":
            # New format: tool results embedded in user messages.
            # After compaction Claude Code reinjects the summary as a plain string — skip those.
            content = event.get("message", {}).get("content", [])
            if isinstance(content, str):
                return
            for item in content:
                if item.get("type") == "tool_result":
                    tool_use_id = item.get("tool_use_id", "")
                    content = item.get("content", "")
                    is_error = item.get("is_error", False)
                    if isinstance(content, list):
                        content = " ".join(c.get("text", "") for c in content if c.get("type") == "text")
                    async for block in self._handle_tool_result(tool_use_id, str(content), is_error):
                        yield block

        elif etype == "system" and subtype == "permission_request":
            yield PermissionRequestBlock(
                id=event.get("id", ""),
                action=event.get("action", ""),
                risk=event.get("risk", "medium"),
                details=event.get("details", ""),
            )

        elif etype == "result":
            yield AudioHintBlock(state="agent_silent")
            if self._thinking_active:
                if self._thinking_text:
                    yield ThinkingBlock(thinking=self._thinking_text, done=True)
                self._thinking_active = False
                self._thinking_text = ""
            if self._stream_block_id:
                yield TextBlock(text=self._stream_block_text, done=True, id=self._stream_block_id)
                self._stream_block_id = None
                self._stream_block_text = ""
            usage = event.get("usage") or {}
            turn_input    = usage.get("input_tokens", 0)
            turn_output   = usage.get("output_tokens", 0)
            turn_cache    = usage.get("cache_read_input_tokens", 0)
            turn_creation = usage.get("cache_creation_input_tokens", 0)
            turn_cost     = event.get("cost_usd") or 0.0
            server_tools  = usage.get("server_tool_use") or {}
            self._total_input_tokens          += turn_input
            self._total_output_tokens         += turn_output
            self._total_cache_read_tokens     += turn_cache
            self._total_cache_creation_tokens += turn_creation
            self._total_web_searches          += server_tools.get("web_search_requests", 0)
            self._total_web_fetches           += server_tools.get("web_fetch_requests", 0)
            self._total_cost_usd = event.get("total_cost_usd") or (
                self._total_cost_usd + turn_cost
            )
            self._total_turns += 1
            # NOTE: do NOT set _last_input_tokens from this event's `usage` — it is the
            # CUMULATIVE total across every API request in the turn (sum of all internal
            # loop steps), so dividing by the context window yields >100% ("1275% full").
            # The live per-request value is captured at message_start above; we only adjust
            # it here for compaction (which genuinely shrinks the resident context).
            compaction = event.get("compactionResult") or event.get("compaction_result")
            if compaction:
                # After compaction the context is smaller — use post-compact output tokens
                post_tokens = (
                    (compaction.get("compactionOutputTokens") or 0)
                    or (compaction.get("output_tokens") or 0)
                )
                if post_tokens:
                    self._last_input_tokens = post_tokens
                yield SystemStatusBlock(event="compaction_done")
            # modelUsage lists EVERY model the turn touched — incl. haiku/sonnet
            # Task/Explore sub-agents — so keys()[0] could be a sub-agent. Prefer the
            # main session model (set from the main agent's message_start, or the launch
            # model) so the pill never flips to a sub-agent's model at turn end.
            usage_models = event.get("modelUsage") or {}
            if usage_models:
                if self._model in usage_models:
                    main_model = self._model
                elif self._startup_model in usage_models:
                    main_model = self._startup_model
                elif self._model or self._startup_model:
                    # Neither known session model ran this turn — e.g. a trivial or
                    # sub-agent-only turn that routed entirely to haiku/sonnet. KEEP the
                    # established session model; do NOT repaint the pill to a sub-agent
                    # model via next(iter(...)) (the "session flipped to haiku" bug, seen
                    # even with startup='claude-opus-4-8' on an Opus-idle turn).
                    main_model = self._model or self._startup_model
                else:
                    # No session model established yet (model never picked, no main-agent
                    # message_start captured) — adopt whatever the turn actually used.
                    main_model = next(iter(usage_models))
                self._model = main_model or self._model
                cw = (usage_models.get(self._model) or {}).get("contextWindow", 0)
                if cw: self._context_window = cw
            # Fallback: the stream omits contextWindow AND strips the "[1m]" marker from
            # the reported model — so derive from the LAUNCH model (self._startup_model
            # keeps "[1m]") first, then the stream model.
            if not self._context_window:
                from .base import context_window_for_model
                self._context_window = (context_window_for_model(self._startup_model or "")
                                        or context_window_for_model(self._model or ""))
            # Self-correct: if the actual prompt already exceeds the resolved window, the
            # session is running an extended context (e.g. Opus [1m]) the table couldn't
            # detect — bump to 1M so "% full" stays meaningful instead of pinned >100%.
            if self._context_window and self._last_input_tokens > self._context_window:
                self._context_window = 1_000_000
            log.debug("ctx-debug: model=%r startup=%r last_input=%d window=%d pct=%s",
                     self._model, self._startup_model, self._last_input_tokens,
                     self._context_window,
                     (int(self._last_input_tokens / self._context_window * 100)
                      if self._context_window else "n/a"))
            reason = "error" if event.get("is_error") else "completed"
            yield SessionEndBlock(
                reason=reason,
                cost_usd=turn_cost,
                input_tokens=turn_input,
                output_tokens=turn_output,
                cache_read_tokens=turn_cache,
            )
            yield SessionStatusHintBlock(
                model=self._model,
                turns=self._total_turns,
                cost_usd=self._total_cost_usd,
                output_tokens=self._total_output_tokens,
                cache_creation_tokens=self._total_cache_creation_tokens,
                last_input_tokens=self._last_input_tokens,
                context_window=self._context_window,
            )

        elif etype == "system" and subtype == "error":
            yield ErrorBlock(message=event.get("message", "unknown error"), recoverable=True)

    # ── Helpers ───────────────────────────────────────────────────────────────

    async def _handle_tool_result(self, tool_use_id: str, content: str, is_error: bool) -> AsyncIterator[Block]:
        pending = self._pending_tools.get(tool_use_id)
        if pending is None and tool_use_id in self._pending_tools:
            # Sentinel / TodoWrite / task-tool result — suppress, clean up.
            self._pending_tools.pop(tool_use_id, None)
            if tool_use_id in self._pending_task_create:
                # TaskCreate result carries the CLI-assigned id ("Task #3 created…").
                subject = self._pending_task_create.pop(tool_use_id)
                import re as _re_task
                m = _re_task.search(r"#(\d+)", content)
                tid = m.group(1) if m else str(len(self._tasks) + 1)
                self._tasks[tid] = {"content": subject, "status": "pending"}
                yield self.todo_block(self._session_id, list(self._tasks.values()))
            yield AudioHintBlock(state="agent_speaking")
        else:
            if pending is not None:
                preview = "" if is_error else content.split("\n")[0][:120]
                yield ToolCallBlock(
                    id=pending.id,
                    name=pending.name,
                    args=pending.args,
                    status="error" if is_error else "done",
                    label=pending.label,
                    subject=pending.subject,
                    preview=preview,
                )
            if is_error:
                import re as _re
                m = _re.match(r"<tool_use_error>(.*?)</tool_use_error>", content, _re.DOTALL)
                if m:
                    content = m.group(1).strip()
            yield tool_result_or_large(
                tool_call_id=tool_use_id,
                content=content,
                fetch_store=self._fetch_store,
                is_error=is_error,
            )
            yield AudioHintBlock(state="agent_speaking")
            # NOTE: background-task detection no longer scrapes this tool_result text.
            # Background Bash AND Agent/Task sub-agents both surface authoritative
            # `system/task_started` + `system/task_notification` events on stdout (handled
            # in _parse), keyed by a stable task_id — see _seen_bg_starts / _done_bg_tasks.

    async def get_usage(self) -> dict:
        return {
            "model":                   self._model,
            "input_tokens":            self._total_input_tokens,
            "output_tokens":           self._total_output_tokens,
            "cache_read_tokens":       self._total_cache_read_tokens,
            "cache_creation_tokens":   self._total_cache_creation_tokens,
            "cost_usd":                self._total_cost_usd,
            "turns":                   self._total_turns,
            "last_input_tokens":       self._last_input_tokens,
            "web_searches":            self._total_web_searches,
            "web_fetches":             self._total_web_fetches,
            "permission_denials":      self._permission_denials,
        }

    async def get_live_limits(self) -> list[dict]:
        """Subscription usage windows (5h / weekly) from the Claude OAuth token.
        Empty for API-key-only setups (the endpoint is OAuth-Bearer only)."""
        from .base import fetch_anthropic_oauth_usage, anthropic_usage_rows
        return anthropic_usage_rows(await fetch_anthropic_oauth_usage())

    async def get_status_hint(self) -> SessionStatusHintBlock | None:
        if not self._model:
            return None
        return SessionStatusHintBlock(
            model=self._model,
            turns=self._total_turns,
            cost_usd=self._total_cost_usd,
            output_tokens=self._total_output_tokens,
            cache_creation_tokens=self._total_cache_creation_tokens,
            last_input_tokens=self._last_input_tokens,
            context_window=self._context_window,
        )

    async def _write(self, obj: dict) -> None:
        assert self._proc is not None
        if self._proc.returncode is not None:
            raise BrokenPipeError(
                f"Claude Code process has exited (returncode={self._proc.returncode})"
            )
        otype = obj.get("type", "?")
        if otype == "user":
            parts = obj.get("message", {}).get("content", [])
            texts = [p.get("text", "")[:300] for p in parts if p.get("type") == "text"]
            imgs  = sum(1 for p in parts if p.get("type") == "image")
            summary = " ".join(texts).replace("\n", "↵")
            if imgs:
                summary += f" (+{imgs} image)"
            log.debug("→ [user] %s", summary)
        elif otype == "permission_response":
            log.debug("→ [permission_response] id=%s approved=%s",
                     obj.get("permission_id"), obj.get("approved"))
        else:
            log.debug("→ [%s]", otype)
        try:
            self._proc.stdin.write((json.dumps(obj) + "\n").encode())
            await self._proc.stdin.drain()
        except (BrokenPipeError, ConnectionResetError, OSError) as e:
            raise BrokenPipeError(f"stdin write failed ({otype}): {e}") from e
