"""
lambdagent.providers.claude_code_provider — Claude Code CLI as LLM provider.

No API Key required — uses Claude Code Max Plan authentication.

Optimizations:
  - First call: creates session with system prompt, captures session_id
  - Subsequent calls: --resume <session_id> (Claude retains full memory)
  - Falls back to messages-in-prompt if --resume unavailable
  - Auto-detects a working claude binary across nvm versions
"""
from __future__ import annotations

import fcntl
import json
import logging
import os
import select
import shutil
import subprocess
import time
from typing import Dict, List, Optional

from .base import LLMProvider, ProviderConfig, ProviderError

# Lazy import to avoid hard dependency cycle if someone uses providers
# stand-alone. cancel module is small + side-effect-free.
try:
    from lambdagent.agentruntime import cancel as _cancel
except Exception:   # pragma: no cover
    _cancel = None


# Default idle-timeout: if claude produces no new bytes for this many seconds
# we conclude it has silently stalled (the common claude-code API failure
# mode — exit code 0 never comes, no error written, just dead air for the
# full hard timeout). Killing fast turns 10-min disasters into 1-min ones.
# Override per-provider via config.extra["claude_idle_timeout"].
_DEFAULT_IDLE_TIMEOUT_S = 60.0
# Initial idle window is more generous since cold start of claude-code's
# session can take 30-60s before the first byte (model load + auth).
_DEFAULT_FIRST_BYTE_TIMEOUT_S = 120.0
# Resume calls reuse a WARM session — healthy first-byte is 5-8s (observed).
# 120s here means a wedged large-context resume burns 120s×2=240s before
# failing; warm resume never legitimately needs that long. Use a tighter
# window so a stalled resume is detected fast and the engine can recover.
# 但实测 Anthropic 首字节延迟偶尔飙到 ~43s(仍活着、慢)——45s 会把这些
# "慢但活着"的响应误判成卡死、白烧一次 45s 再重试。抬到 60s 让它们在
# attempt 1 直接通过;真卡死多等 15s 也有连续卡死熔断兜底。
# Override via config.extra["claude_resume_first_byte_timeout"].
_DEFAULT_RESUME_FIRST_BYTE_TIMEOUT_S = 60.0
# 会话轮换:claude-code `--resume` 一个变大的 session 会首字节卡死(实测 resume
# 一个 204 行/395KB 的 session 超过 60s 没首字节，而全新会话 8s 秒回)。每 N 次
# resume 后主动开新会话，让 session 永远停在"小而快"的区间。0=不轮换。
# 可配 config.extra["claude_session_rotate_after"]。
_DEFAULT_SESSION_ROTATE_AFTER = 8
# 连续多少次 _call_resume 整体首字节卡死后,判定该 session 本身太大/废了,
# 强制清掉换新会话。保留会话只兜瞬时抖动;连卡说明是 session 大小问题,必须换。
_DEFAULT_RESUME_STALL_GIVEUP = 2

# claude-code 订阅 OAuth 令牌过期/无效时返回 401。把原始 JSON 翻译成人话,
# 让用户一眼知道去重新登录,而不是来排查平台。auth 错误 NON-retryable(重试
# 一个过期令牌只是白等)。
_AUTH_ERROR_MARKERS = (
    "401",
    "invalid authentication",
    "failed to authenticate",
    "authentication_error",
    "oauth token has expired",
    "invalid bearer token",
)
_AUTH_HINT = (
    "claude-code 登录已过期或无效(API 401)。请在终端运行 `claude` 并 `/login` "
    "重新登录(或重新登录 Claude 桌面应用,二者共用同一份凭据),令牌刷新后即可恢复。"
)


def _is_auth_error(text: str) -> bool:
    """判断一段错误文本是否为 claude-code 认证(401)失败。"""
    if not text:
        return False
    low = text.lower()
    if "401" in low and ("authenticat" in low or "credential" in low or "bearer" in low):
        return True
    return any(m in low for m in _AUTH_ERROR_MARKERS if m != "401")


class _StallError(Exception):
    """Raised when claude subprocess produced no bytes within idle_timeout
    (or exceeded the hard timeout). Caller converts to ProviderError with
    context (first-turn vs resume, session_id, etc.)."""

    def __init__(self, kind: str, elapsed: float,
                 stdout: str, stderr: str, returncode: Optional[int] = None):
        self.kind = kind            # "idle" | "hard" | "killed"
        self.elapsed = elapsed
        self.stdout = stdout
        self.stderr = stderr
        self.returncode = returncode
        super().__init__(f"claude {kind}-timeout after {elapsed:.1f}s")


def _run_with_idle_timeout(
    cmd: List[str],
    *,
    idle_timeout: float,
    first_byte_timeout: float,
    hard_timeout: float,
) -> subprocess.CompletedProcess:
    """subprocess.run replacement that kills the child the moment it stops
    producing output.

    Three independent watchdogs:
      - first_byte_timeout: before any bytes arrive (cold start)
      - idle_timeout:       between subsequent bytes
      - hard_timeout:       total wall-clock ceiling (existing behavior)

    Returns a CompletedProcess (text-decoded) on clean exit. Raises
    _StallError on any timeout, with whatever partial output was captured.

    Implementation: select() on non-blocking stdout/stderr pipes, loop
    until poll() returns or a watchdog fires. macOS/Linux only — Windows
    has no select-on-pipe, but agentpaas doesn't target Windows anyway.
    """
    proc = subprocess.Popen(
        cmd,
        stdin=subprocess.DEVNULL,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        # Put claude into its own process group so we can SIGKILL the
        # whole tree (claude CLI -> node helpers -> MCP servers -> etc.)
        # with a single os.killpg, instead of just signalling the
        # immediate parent and leaving grandchildren as init-orphans.
        # The accumulated orphans were the direct cause of the macOS
        # Jetsam SIGKILL on agentpaas earlier (131 claude descendants
        # eating ~25 GB of RAM).
        start_new_session=True,
    )

    # Bind to per-run cancel registry — request_cancel(run_id) can now reach
    # in and kill this specific subprocess from another thread without
    # process-wide pkill. The bind is no-op if no run is registered on the
    # current thread (e.g. someone using the provider stand-alone).
    if _cancel is not None:
        _cancel.bind_proc(proc)

    # Helper used by every kill site below — kill the whole process group,
    # then wait for proc.poll() to confirm so we never leak.
    def _kill_tree(reason: str = "") -> None:
        if proc.poll() is not None:
            return
        try:
            os.killpg(os.getpgid(proc.pid), 9)  # SIGKILL whole group
        except (ProcessLookupError, OSError):
            # Fall back to direct kill if pgid lookup failed.
            try:
                proc.kill()
            except Exception:
                pass
        try:
            proc.wait(timeout=2)
        except Exception:
            pass

    # Make the read pipes non-blocking so select() + read() can't accidentally
    # park waiting for more bytes once the kernel buffer is drained.
    for fd in (proc.stdout.fileno(), proc.stderr.fileno()):
        flags = fcntl.fcntl(fd, fcntl.F_GETFL)
        fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)

    out_chunks: List[bytes] = []
    err_chunks: List[bytes] = []
    start = time.time()
    last_byte_at: Optional[float] = None   # None = haven't seen the first byte

    try:
        while proc.poll() is None:
            now = time.time()
            elapsed = now - start

            # Cancellation check — fires at most ~2 Hz (the select polling
            # cadence) so user "Stop" reaches the killing path within ~0.5s
            # even when claude is mid-thought.
            if _cancel is not None and _cancel.is_cancelled():
                raise _cancel.CancelledRun()

            # Watchdogs
            if elapsed > hard_timeout:
                raise _StallError(
                    "hard", elapsed,
                    b"".join(out_chunks).decode(errors="replace"),
                    b"".join(err_chunks).decode(errors="replace"),
                )
            if last_byte_at is None:
                if elapsed > first_byte_timeout:
                    raise _StallError(
                        "first-byte", elapsed,
                        b"".join(out_chunks).decode(errors="replace"),
                        b"".join(err_chunks).decode(errors="replace"),
                    )
            else:
                idle = now - last_byte_at
                if idle > idle_timeout:
                    raise _StallError(
                        "idle", elapsed,
                        b"".join(out_chunks).decode(errors="replace"),
                        b"".join(err_chunks).decode(errors="replace"),
                    )

            # Poll. Short select timeout so the watchdogs above run regularly.
            ready, _, _ = select.select(
                [proc.stdout, proc.stderr], [], [], 0.5
            )
            for f in ready:
                try:
                    chunk = f.read()
                except (BlockingIOError, OSError):
                    chunk = b""
                if chunk:
                    (out_chunks if f is proc.stdout else err_chunks).append(chunk)
                    last_byte_at = now

        # Process exited — drain anything still buffered.
        try:
            rem_out, rem_err = proc.communicate(timeout=5)
            if rem_out:
                out_chunks.append(rem_out)
            if rem_err:
                err_chunks.append(rem_err)
        except subprocess.TimeoutExpired:
            pass

    except _StallError:
        # Kill the stalled child before re-raising; otherwise it sits
        # around eating tokens (and money).
        _kill_tree("stall")
        raise
    except BaseException:
        # Cancellation (or any other exception) — kill before re-raising.
        # CancelledRun is BaseException-safe since it inherits from Exception,
        # but using BaseException here also covers KeyboardInterrupt cleanly.
        _kill_tree("cancel-or-exception")
        raise
    finally:
        # Final safety net: if any path above missed cleanup, this kills
        # the whole process group. No-op when the proc already exited.
        _kill_tree("finally-cleanup")
        # Always release the proc binding so a subsequent run doesn't see
        # a dead Popen registered for it.
        if _cancel is not None:
            _cancel.unbind_proc()

    return subprocess.CompletedProcess(
        args=cmd,
        returncode=proc.returncode,
        stdout=b"".join(out_chunks).decode("utf-8", errors="replace"),
        stderr=b"".join(err_chunks).decode("utf-8", errors="replace"),
    )

# Attach under "agentpaas" so the host's logging handler picks us up.
# Using __name__ ("lambdagent.providers.claude_code_provider") gives a
# silent logger when running inside agentpaas — its setup_logging() only
# configures handlers on the "agentpaas" tree.
logger = logging.getLogger("agentpaas.claude-code")


def _find_working_claude(preferred: str = "claude") -> Optional[str]:
    """
    Find a claude binary that actually supports modern flags (--tools, --mcp-config).

    Strategy:
      1. If preferred is an absolute path, use it directly.
      2. Test whatever shutil.which finds first.
      3. Scan ~/.nvm/versions/node in reverse order (newest first).

    A binary is "working" if:
      - It's executable and exits 0 for --version
      - Its version output contains "Claude Code"
      - Its --help output mentions "--tools" (rules out old v1.x CLI)
    """
    def _works(path: str) -> bool:
        try:
            r = subprocess.run([path, "--version"], capture_output=True, text=True, timeout=10)
            if r.returncode != 0 or "Claude Code" not in r.stdout:
                return False
            # v1.x (old) doesn't have --tools; v2.x does. Use --help as a proxy check.
            h = subprocess.run([path, "--help"], capture_output=True, text=True, timeout=10)
            help_text = h.stdout + h.stderr
            return "--tools" in help_text
        except Exception:
            return False

    # 1. Absolute path given — trust it.
    if os.path.isabs(preferred):
        return preferred if os.path.isfile(preferred) else None

    # 2. Whatever is first in PATH.
    found = shutil.which(preferred)
    if found and _works(found):
        return found

    # 3. Scan nvm versions, newest first.
    nvm_base = os.path.expanduser("~/.nvm/versions/node")
    if os.path.isdir(nvm_base):
        try:
            versions = sorted(os.listdir(nvm_base), reverse=True)
        except OSError:
            versions = []
        for ver in versions:
            candidate = os.path.join(nvm_base, ver, "bin", "claude")
            if os.path.isfile(candidate) and _works(candidate):
                return candidate

    # Fall back to whatever shutil found, even if it might not work.
    return found or preferred


def _strip_tool_docs(system_prompt: str) -> tuple:
    """Strip the auto-generated tool-schema section and extract tool names.

    _generate_tool_schema_docs() appends a '## 工具参数参考' block that
    lists Bash, WriteFile, etc. with JSON call signatures.  When this reaches
    claude-code via --system-prompt, claude decides it *has* those native tools
    and tries to use them directly.  We strip the full schema block (to stop
    that), but extract the tool names so we can re-inject a lightweight
    "these tools are available via JSON format" notice — without that notice,
    claude only sees its MCP session tools (Gmail, Drive, …) and gives up.

    Returns:
        (stripped_prompt, tool_names_list)
    """
    import re as _re
    tool_names: list = []
    for marker in (
        "\n\n## 工具参数参考",   # Chinese header added by _generate_tool_schema_docs
        "\n\n## Available Tools",
        "\n## 工具参数参考",
        "\n## Available Tools",
    ):
        idx = system_prompt.find(marker)
        if idx != -1:
            # Extract tool names from the block before discarding it
            block = system_prompt[idx:]
            # Matches "- **ToolName**:" or "**terminate**:"
            tool_names = _re.findall(r'\*\*(\w+)\*\*', block)
            return system_prompt[:idx], tool_names
    return system_prompt, tool_names


class ClaudeCodeProvider(LLMProvider):
    """
    LLM Provider backed by Claude Code CLI.

    Uses `claude -p` subprocess. Session persistence via `--resume`.
    """

    def __init__(self, config: ProviderConfig):
        super().__init__(config)
        preferred = config.extra.get("claude_bin", "claude") if config.extra else "claude"
        self.claude_bin = _find_working_claude(preferred) or preferred
        self._session_id = None
        self._resume_count = 0  # 会话轮换计数(每 N 次 resume 换新会话，防大会话 resume 卡死)
        self._resume_stall_streak = 0  # 连续首字节卡死计数(熔断:连卡 K 次判 session 废了→清掉换新)

        if not shutil.which(self.claude_bin) and not os.path.isfile(self.claude_bin):
            raise ProviderError(
                f"claude command not found at '{self.claude_bin}'. "
                f"Install: npm install -g @anthropic-ai/claude-code",
                "claude-code",
            )

    def chat(self, messages: List[Dict[str, str]]) -> str:
        if self._session_id:
            return self._call_resume(messages)
        else:
            return self._call_new(messages)

    # Injected into user message on every call to prevent native tool execution.
    _REACT_CONSTRAINT = (
        "\n\n[REACT_MODE] 你是决策模块，不是执行模块。"
        "只输出下一步的JSON工具调用，格式：{\"tool\":\"工具名\",\"input\":\"...\"}，"
        "或输出 terminate 表示任务完成。"
        "严禁直接执行任何搜索、Bash命令或文件读写操作。"
    )

    # Appended to system prompt to override any "do everything" framing in agent YAML.
    _SYSTEM_PROMPT_SUFFIX = (
        "\n\n===核心约束（优先级最高）==="
        "\n你是工具路由决策器，不是执行器。"
        "\n每次只输出一个JSON工具调用命令，不做任何实际操作。"
        "\n格式：{\"tool\": \"工具名\", \"input\": \"...\"} 或 terminate"
    )

    @staticmethod
    def _build_tool_notice(tool_names: list) -> str:
        """Build a lightweight notice listing available JSON-format tools.

        This is injected AFTER the schema block is stripped so Claude knows
        which tools to route through the JSON call format — without this,
        claude sees only its MCP session tools (Gmail, Drive, …) and gives up
        instead of emitting JSON tool calls.
        """
        if not tool_names:
            return ""
        names = "、".join(tool_names)
        return (
            f"\n\n[可用工具（JSON格式调用，与当前MCP工具无关）]\n"
            f"以下工具通过 {{\"tool\":\"工具名\",\"input\":...}} JSON格式调用，"
            f"不是当前session的MCP原生工具：\n{names}\n"
            f"当你需要读文件、列目录、写文件时，请用上述JSON格式调用对应工具，"
            f"不要说【工具不可用】，直接用JSON格式调用即可。"
        )

    def _call_new(self, messages: List[Dict[str, str]]) -> str:
        """First call: create session with system prompt, capture session_id."""
        system_prompt = ""
        user_content = ""
        for m in messages:
            if m["role"] == "system":
                system_prompt = m["content"]
            elif m["role"] == "user":
                user_content = m["content"]

        # Strip tool-schema docs (## 工具参数参考 section) — they expose Bash/WriteFile
        # which makes claude-code try to use them natively. Extract tool names first so
        # we can re-inject a lightweight "these tools exist via JSON" notice, which stops
        # Claude from looking at its MCP session tools and giving up.
        stripped_prompt, tool_names = _strip_tool_docs(system_prompt)
        tool_notice = self._build_tool_notice(tool_names)
        effective_system = stripped_prompt + tool_notice + self._SYSTEM_PROMPT_SUFFIX
        prompt_arg = user_content + self._REACT_CONSTRAINT

        cmd = [
            self.claude_bin, "-p", prompt_arg,
            "--output-format", "json",
            "--model", self.config.model,
            "--system-prompt", effective_system,
            # Disable claude-code's native tools entirely. The orchestrator
            # is supposed to OUTPUT a JSON tool call for the engine to route
            # through agentpaas's tool registry — it should never execute
            # Bash/WriteFile etc. itself. Without this, --dangerously-skip-
            # permissions lets claude run native tools whenever it pleases,
            # bypassing mcp.localTools entirely (observed in run_4e07736d7f60:
            # orchestrator fabricated 150 KB of fake `sim_results.json` via
            # native WriteFile while every `call_physsim` invocation STALLed).
            "--tools", "",
            # 同时隔离用户的 MCP 服务。--tools "" 只关原生工具,但用户 ~/.claude
            # 配置里的 MCP(如 Gmail/Google Calendar)仍会被加载——claude-code 一
            # 自省就只看到这些 MCP 工具、看不到平台 react 工具(ReadFile/WriteFile
            # 只是 prompt 里的"输出 JSON"协议、不算它持有的真工具),于是判定"工具
            # 不匹配、写不了文件"并拒绝(run_f2a1eca9c6ca 实锤)。--strict-mcp-config
            # 不配 --mcp-config = 不加载任何 MCP → 工具集清空 → 纯走 react JSON 协议。
            "--strict-mcp-config",
            "--dangerously-skip-permissions",
        ]

        # Diagnostics: log argv sizes before spawning. argv on macOS is
        # capped at ARG_MAX (~1MB); large system prompts can push us close.
        total_argv = sum(len(s) for s in cmd)
        extra = self.config.extra or {}
        idle_timeout = float(extra.get("claude_idle_timeout", _DEFAULT_IDLE_TIMEOUT_S))
        first_byte_timeout = float(extra.get("claude_first_byte_timeout", _DEFAULT_FIRST_BYTE_TIMEOUT_S))
        logger.info(
            "claude-code first-turn spawn: prompt=%d sys=%d total_argv=%d model=%s "
            "hard=%ds idle=%.0fs first-byte=%.0fs",
            len(prompt_arg), len(effective_system), total_argv,
            self.config.model, self.config.timeout, idle_timeout, first_byte_timeout,
        )

        # First-turn STALLs are usually transient Anthropic-side hiccups
        # (silent throttle during burst, transient 5xx). audit on
        # 2026-06-06 showed 38% (19/50) first-turn STALL rate when the
        # API got flaky — most of those would have succeeded on a second
        # try. We retry ONCE with the same params before giving up;
        # second attempt usually completes within normal latency.
        #
        # NOTE: this retry burns up to first_byte_timeout (120s default)
        # in the worst case. If both attempts stall, the engine still
        # gets a retryable ProviderError so its own logic can decide
        # whether to give up the whole step.
        max_attempts = 2
        last_stall: Optional[_StallError] = None
        result = None
        t0 = time.time()
        for attempt in range(1, max_attempts + 1):
            try:
                result = _run_with_idle_timeout(
                    cmd,
                    idle_timeout=idle_timeout,
                    first_byte_timeout=first_byte_timeout,
                    hard_timeout=float(self.config.timeout),
                )
                if attempt > 1:
                    logger.info(
                        "claude-code first-turn ok on retry %d after %.1fs total",
                        attempt, time.time() - t0,
                    )
                last_stall = None
                break
            except _StallError as exc:
                last_stall = exc
                logger.warning(
                    "claude-code STALL (%s, first-turn attempt %d/%d) after %.1fs "
                    "partial_stdout=%r partial_stderr=%r",
                    exc.kind, attempt, max_attempts, exc.elapsed,
                    exc.stdout[:300], exc.stderr[:300],
                )
                if attempt < max_attempts:
                    # Brief pause so any burst-window throttle on Anthropic's
                    # side has a chance to reset. Tested: 2s is enough; 0s
                    # often re-STALLs back-to-back.
                    time.sleep(2.0)

        if last_stall is not None:
            raise ProviderError(
                f"Claude Code stalled ({last_stall.kind}-timeout after "
                f"{last_stall.elapsed:.0f}s, no output produced) on first turn "
                f"(retried {max_attempts}x)",
                "claude-code", retryable=True,
            )
        assert result is not None  # for type checker; loop guarantees this

        elapsed = time.time() - t0
        if result.returncode != 0:
            stderr = (result.stderr or "").strip()
            stdout_hint = (result.stdout or "").strip()[:200]
            detail = stderr or stdout_hint or "(no output)"
            logger.warning(
                "claude-code FAILED exit=%d after %.1fs stderr=%r stdout=%r",
                result.returncode, elapsed, stderr[:500], stdout_hint,
            )
            if _is_auth_error(detail):
                raise ProviderError(_AUTH_HINT, "claude-code", retryable=False)
            raise ProviderError(
                f"Claude Code error (exit {result.returncode}): {detail}",
                "claude-code",
            )
        logger.info("claude-code first-turn ok: elapsed=%.1fs out_len=%d", elapsed, len(result.stdout or ""))

        raw = result.stdout.strip()
        if not raw:
            raise ProviderError(
                f"Claude Code returned empty output (exit {result.returncode}). "
                f"stderr: {(result.stderr or '').strip()[:200] or '(empty)'}",
                "claude-code", retryable=True,
            )

        try:
            data = json.loads(raw)
            # returncode 0 但 is_error:true(如 api_error_status 401)——claude 把错误
            # 塞进 result 字段还 exit 0。若不拦,会把 "Failed to authenticate" 当成
            # agent 的正常回答返回 → 静默失败。显式识别并抛出。
            if data.get("is_error"):
                err = str(data.get("result") or data.get("error") or "").strip()
                status = data.get("api_error_status")
                if (status == 401) or _is_auth_error(err):
                    raise ProviderError(_AUTH_HINT, "claude-code", retryable=False)
                raise ProviderError(
                    f"Claude Code error (is_error, status={status}): {err[:300]}",
                    "claude-code", retryable=bool(status and int(status) >= 500),
                )
            self._session_id = data.get("session_id")
            return data.get("result", raw)
        except json.JSONDecodeError:
            return raw

    def _call_resume(self, messages: List[Dict[str, str]]) -> str:
        """Subsequent calls: resume session (Claude has full memory)."""
        last_user = ""
        for m in reversed(messages):
            if m["role"] == "user":
                last_user = m["content"]
                break

        # Same ReAct constraint to keep each resume turn as a single JSON decision.
        prompt_arg = last_user + self._REACT_CONSTRAINT

        cmd = [
            self.claude_bin, "-p", prompt_arg,
            "--output-format", "text",
            "--model", self.config.model,
            "--resume", self._session_id,
            # See _call_new comment — keep parity on resume so claude can
            # never run native tools, regardless of session phase.
            "--tools", "",
            "--strict-mcp-config",  # 同 _call_new:隔离用户 MCP,防"我只有 Gmail/Calendar"误判
            "--dangerously-skip-permissions",
        ]

        extra_top = self.config.extra or {}
        _it = float(extra_top.get("claude_idle_timeout", _DEFAULT_IDLE_TIMEOUT_S))
        _fbt = float(extra_top.get(
            "claude_resume_first_byte_timeout", _DEFAULT_RESUME_FIRST_BYTE_TIMEOUT_S))
        logger.info(
            "claude-code resume spawn: prompt=%d session=%s hard=%ds idle=%.0fs first-byte=%.0fs",
            len(prompt_arg), self._session_id[:8] if self._session_id else "?",
            self.config.timeout, _it, _fbt,
        )

        # Transport-error patterns worth retrying once. Long-running sessions
        # sporadically die from Anthropic-side socket drops or 5xx — losing the
        # whole 22-turn paper-writing session over one flake is unacceptable.
        _TRANSPORT_RETRY_MARKERS = (
            "socket connection",       # "API Error: The socket connection ..."
            "socket hang up",
            "ECONNRESET",
            "fetch failed",
            "ETIMEDOUT",
            "503",
            "502",
            "overloaded",
        )

        def _is_transport_error(text: str) -> bool:
            low = text.lower()
            return any(m.lower() in low for m in _TRANSPORT_RETRY_MARKERS)

        extra = self.config.extra or {}
        idle_timeout = float(extra.get("claude_idle_timeout", _DEFAULT_IDLE_TIMEOUT_S))
        # Resume is warm → tighter first-byte window than cold first-turn.
        first_byte_timeout = float(extra.get(
            "claude_resume_first_byte_timeout", _DEFAULT_RESUME_FIRST_BYTE_TIMEOUT_S))

        def _run_once():
            t0 = time.time()
            try:
                result = _run_with_idle_timeout(
                    cmd,
                    idle_timeout=idle_timeout,
                    first_byte_timeout=first_byte_timeout,
                    hard_timeout=float(self.config.timeout),
                )
                return result, time.time() - t0, None
            except _StallError as exc:
                return None, time.time() - t0, exc

        for attempt in (1, 2):
            result, elapsed, stall_exc = _run_once()

            if stall_exc is not None:
                logger.warning(
                    "claude-code RESUME STALL (%s, attempt %d) after %.1fs "
                    "partial_stdout=%r partial_stderr=%r",
                    stall_exc.kind, attempt, elapsed,
                    stall_exc.stdout[:300], stall_exc.stderr[:300],
                )
                # first-byte stalls on attempt 1 are often just slow API
                # responses for large-context sessions (e.g. long PDF reviews)
                # — retry once before giving up. idle/hard stalls (session
                # wedged / hit hard ceiling) and second first-byte stalls
                # are unrecoverable: clear session and fail.
                if stall_exc.kind == "first-byte" and attempt == 1:
                    logger.info(
                        "claude-code resume first-byte stall on attempt 1 — "
                        "retrying (large context may need extra time)"
                    )
                    time.sleep(2.0)
                    continue
                # 关键:区分卡死类型,决定是否保留会话。
                # · first-byte 卡死 = 一个字节都没来 = API 一时慢,会话仍在
                #   Anthropic 侧 → **保留 session_id**,让上层(ConversationLam)
                #   重试 resume 同一会话、不丢记忆。曾经一律清 session → 每次首字节
                #   卡死都换 fresh 会话 → agent 丢上下文反复重读重分析(实测一条
                #   run 因此连换 8 个 session、76 步在打转)。
                # · idle/hard 卡死 = 已开始输出又卡住 / 超总时限 = 会话可能真卡死
                #   → 清掉,下次退化成 fresh first-turn 重启。
                if stall_exc.kind != "first-byte":
                    self._session_id = None
                    self._resume_stall_streak = 0
                else:
                    # 首字节卡死本应保留会话(兜瞬时抖动)。但若**连续**多次
                    # _call_resume 都整体卡死,说明不是抖动而是这个 session 已
                    # 大到 resume 必卡 → 熔断:清掉会话,下次退化成 fresh
                    # first-turn(快)。否则会像那条 run 一样在同一大 session 上
                    # 连卡 8 次打转到死。
                    self._resume_stall_streak += 1
                    _giveup = int((self.config.extra or {}).get(
                        "claude_resume_stall_giveup", _DEFAULT_RESUME_STALL_GIVEUP))
                    if _giveup > 0 and self._resume_stall_streak >= _giveup:
                        logger.warning(
                            "claude-code resume 连续首字节卡死 %d 次 — 判定 session "
                            "太大/已废,清会话换新(下次走 fresh first-turn)",
                            self._resume_stall_streak,
                        )
                        self._session_id = None
                        self._resume_count = 0
                        self._resume_stall_streak = 0
                raise ProviderError(
                    f"Claude Code stalled ({stall_exc.kind}-timeout after "
                    f"{stall_exc.elapsed:.0f}s, no output) on resume",
                    "claude-code", retryable=True,
                )

            if result.returncode == 0:
                logger.info(
                    "claude-code resume ok (attempt %d): elapsed=%.1fs out_len=%d",
                    attempt, elapsed, len(result.stdout or ""),
                )
                self._resume_stall_streak = 0  # 成功一次即清零连卡计数
                break

            # Non-zero exit. Inspect for transport-error marker.
            stderr = (result.stderr or "").strip()
            stdout_hint = (result.stdout or "").strip()[:500]
            detail = stderr or stdout_hint or "(no output)"

            if attempt == 1 and _is_transport_error(detail):
                logger.warning(
                    "claude-code RESUME transport error (attempt 1) exit=%d after %.1fs — retrying in 2s. detail=%r",
                    result.returncode, elapsed, detail[:300],
                )
                time.sleep(2)
                continue  # Keep session_id; retry the same resume

            logger.warning(
                "claude-code RESUME FAILED (attempt %d) exit=%d after %.1fs stderr=%r stdout=%r",
                attempt, result.returncode, elapsed, stderr[:500], stdout_hint,
            )
            self._session_id = None
            if _is_auth_error(detail):
                raise ProviderError(_AUTH_HINT, "claude-code", retryable=False)
            raise ProviderError(
                f"Claude Code resume error (exit {result.returncode}): {detail}",
                "claude-code",
            )

        output = result.stdout.strip()

        # text 模式下 401 有时 exit 0、把 "Failed to authenticate" 直接打到 stdout。
        # 不拦就会当成 agent 正常回答返回 → 静默失败。短输出且命中 auth 标记才拦,
        # 避免误伤正常含 "401" 的内容。
        if output and len(output) < 200 and _is_auth_error(output):
            self._session_id = None
            raise ProviderError(_AUTH_HINT, "claude-code", retryable=False)

        # 会话轮换:resume 成功后计数，到阈值就清掉 session_id → 下一次 chat()
        # 走全新 first-turn(快)。引擎(compiler)检测到 session 丢失会把原始指令
        # + 最新结果重注入到新会话，上下文延续(工作区 agent 还能从磁盘文件看进度)。
        # 这样 session 永远不会长到 resume 卡死的体量。
        self._resume_count += 1
        _rotate_after = int((self.config.extra or {}).get(
            "claude_session_rotate_after", _DEFAULT_SESSION_ROTATE_AFTER))
        if _rotate_after > 0 and self._resume_count >= _rotate_after:
            logger.info(
                "claude-code 轮换会话(已 resume %d 次)——开新会话避免大会话 resume 卡死",
                self._resume_count,
            )
            self._session_id = None
            self._resume_count = 0

        return output if output else "[no output]"

    def reset_session(self):
        """Reset session — next call creates a new conversation."""
        self._session_id = None

    def get_usage(self) -> dict:
        """Read cumulative token usage from Claude Code's JSONL session transcript.

        Claude Code CLI writes a JSONL file at
          ~/.claude/projects/{project_hash}/{session_id}.jsonl
        Every assistant turn contains a `message.usage` dict with:
          input_tokens, output_tokens,
          cache_creation_input_tokens, cache_read_input_tokens

        Returns a dict with those four keys (all ints, 0 if unavailable).
        """
        import glob as _glob

        empty = {
            "input_tokens": 0,
            "output_tokens": 0,
            "cache_creation_input_tokens": 0,
            "cache_read_input_tokens": 0,
        }

        sid = self._session_id
        if not sid:
            return empty

        pattern = os.path.expanduser(f"~/.claude/projects/*/{sid}.jsonl")
        files = _glob.glob(pattern)
        if not files:
            logger.debug("get_usage: no JSONL found for session %s", sid)
            return empty

        path = files[0]
        totals = dict(empty)
        try:
            with open(path, encoding="utf-8") as f:
                for line in f:
                    line = line.strip()
                    if not line:
                        continue
                    try:
                        obj = json.loads(line)
                        if obj.get("type") == "assistant":
                            usage = obj.get("message", {}).get("usage", {})
                            for k in totals:
                                totals[k] += usage.get(k, 0)
                    except (json.JSONDecodeError, TypeError):
                        continue
        except OSError as e:
            logger.warning("get_usage: error reading %s: %s", path, e)
            return empty

        logger.debug(
            "get_usage: session=%s input=%d output=%d cache_read=%d cache_create=%d",
            sid, totals["input_tokens"], totals["output_tokens"],
            totals["cache_read_input_tokens"], totals["cache_creation_input_tokens"],
        )
        return totals
