cmdc

defmodule CMDC.Agent do
  @moduledoc """
  Agent 状态机 — 使用 `:gen_statem` 编排四状态循环。

  ## 状态图

  ```
      ┌──────┐  prompt   ┌─────────┐  stream   ┌───────────┐
      │ idle │──────────▶│ running │─────────▶│ streaming │
      └──┬───┘           └────▲────┘           └─────┬─────┘
         │                    │                       │
         │                    │          finalize     │
         │                    │                       │
         │     ┌──────────────┴────────┐              │
         │     │  tool_calls?          │◀─────────────┘
         │     │  yes → executing_tools│
         │     │  no  → finish/retry   │
         │     └──────────┬────────────┘
         │                │
         │        ┌───────▼──────────┐
         │        │ executing_tools  │ ← Task 完成后
         │        │ collect results  │   回到 running
         │        └───────┬──────────┘
         │                │ all done
         │                ▼
         │           run_turn (loop)
         │                │ no more tool_calls
         └────────────────┘ finish → idle
  ```

  ## 流式消息协议

  Agent 通过 `CMDC.Provider.stream/4` 发起 LLM 请求。
  `CMDC.Provider.StreamBridge` 将 req_llm 的 `StreamResponse` 转为消息：

  | 消息 | 处理 |
  |------|------|
  | `{:cmdc_stream_chunk, %StreamChunk{}}` | 由内部 Stream 处理器更新状态 |
  | `:cmdc_stream_done` | 触发 `finalize_response/1` |
  | `{:cmdc_stream_error, reason}` | 设置 `stream_errored`，触发错误恢复 |

  ## 关键改造（相对旧项目）

  - 移除 `Agent.Registry` 进程注册表（随 Orchestrator 走）
  - 移除 shadow 功能
  - Emitter 适配新 `CMDC.EventBus`（替代旧 `CMDC.Events`）
  - Plugin.Registry 从旧 `from_specs/1` 移植，支持 `Module` 和 `{Module, opts}` 两种格式
  - `config` 字段保持为 map（待 1.9 任务改为 `Config.t()`）

  ## 使用示例

      {:ok, pid} = CMDC.Agent.start_link(
        session_id: "abc",
        model: "anthropic:claude-sonnet-4-5",
        working_dir: "/project",
        tools: [CMDC.Tool.ReadFile, CMDC.Tool.Shell],
        provider_opts: [api_key: "sk-..."]
      )

      %{queued: false} = CMDC.Agent.prompt(pid, "Hello")
  """

  @behaviour :gen_statem

  @dialyzer [
    {:nowarn_function, route_approval_cast: 4},
    {:nowarn_function, maybe_auto_resume: 4},
    {:nowarn_function, enqueue: 3},
    {:nowarn_function, run_turn: 1},
    {:nowarn_function, start_streaming: 2},
    {:nowarn_function, safe_provider_stream: 2},
    {:nowarn_function, finalize_response: 1},
    {:nowarn_function, dispatch: 2},
    {:nowarn_function, split_tool_calls: 2},
    {:nowarn_function, try_finish: 1},
    {:nowarn_function, finish: 1},
    {:nowarn_function, next: 1},
    {:nowarn_function, handle_provider_error: 2},
    {:nowarn_function, recover_stream_error: 1},
    {:nowarn_function, schedule_retry: 2},
    {:nowarn_function, do_abort: 2},
    {:nowarn_function, cancel_stream: 1},
    {:nowarn_function, drain_pending: 1},
    {:nowarn_function, has_pending_user_message?: 1},
    {:nowarn_function, broadcast: 2}
  ]

  require Logger

  alias CMDC.Agent.{
    Compactor,
    Compactor.ArgTruncator,
    Emitter,
    Overflow,
    Repair,
    Retry,
    State,
    Stream,
    SystemPrompt,
    ToolRunner
  }

  alias CMDC.Plugin.{Pipeline, Registry}

  # ==========================================================================
  # Public API
  # ==========================================================================

  @type start_opts :: [
          {:session_id, String.t()}
          | {:model, CMDC.Provider.model()}
          | {:working_dir, String.t()}
          | {:blueprint_system_prompt, String.t()}
          | {:tools, [module()]}
          | {:disabled_tools, [String.t()]}
          | {:plugins, [Registry.plugin_spec()]}
          | {:config, map()}
          | {:provider_opts, keyword()}
          | {:messages, [CMDC.Message.t()]}
          | {:sandbox, module() | nil}
          | {:subagents, [map()]}
          | {:user_data, map()}
          | {:prompt_mode, :full | :task | :minimal | :none}
          | {:event_buffer_size, non_neg_integer()}
          | {:hibernate_after_ms, pos_integer() | nil}
        ]

  @doc """
  启动 Agent 状态机。

  支持 `:hibernate_after_ms` 选项，透传到 `:gen_statem.start_link/3` 的
  `{:hibernate_after, ms}` OTP 原生选项；`nil` 时不主动 hibernate。
  """
  @spec start_link(start_opts()) :: GenServer.on_start()
  def start_link(opts) do
    gen_opts = build_gen_opts(opts)
    :gen_statem.start_link(__MODULE__, opts, gen_opts)
  end

  @doc false
  @spec build_gen_opts(start_opts()) :: keyword()
  def build_gen_opts(opts) do
    case Keyword.get(opts, :hibernate_after_ms) do
      nil -> []
      ms when is_integer(ms) and ms > 0 -> [hibernate_after: ms]
      _ -> []
    end
  end

  @spec child_spec(start_opts()) :: Supervisor.child_spec()
  def child_spec(opts) do
    %{id: __MODULE__, start: {__MODULE__, :start_link, [opts]}, type: :worker}
  end

  @doc """
  发送用户 prompt。

  - idle 状态下立即处理，返回 `%{queued: false}`
  - 忙碌时入队，返回 `%{queued: true}`
  """
  @spec prompt(GenServer.server(), String.t()) :: %{queued: boolean()}
  def prompt(agent, text) when is_binary(text) do
    :gen_statem.call(agent, {:prompt, text})
  end

  @doc """
  中止当前运行。

  ## 选项

  - `:reason :: term()` — 中止原因，写入 `{:agent_abort, reason}` payload。
    默认 `nil`（透传，发裸 `:agent_abort` 事件）。

    入参可以是 atom 或 string。下列 6 个标准 string 会自动归一为同名 atom，
    防止前端通过 JSON 反序列化注入任意 atom 进 BEAM atom table:

        "user_cancelled" / "timeout" / "shutdown" /
        "budget_exceeded" / "permission_denied" / "provider_error"

    其他 string 一律归并为 `:unknown` 并 `Logger.warning`。Atom 入参保持原样透传。
    string 入参 **Since v0.3**。
  - `:clear_queue :: boolean()` — 是否清空 `pending_messages`（排队中的 prompt）;
    清空时为每条被丢弃的 prompt emit `{:prompt_dropped, text}` 事件。
    默认 `true`。
  - `:kill_tools :: :all | :killable | :none` — 工具任务清理策略：
    - `:all` — brutal_kill 所有 in-flight 工具（含 interrupt_immune_tools）
    - `:killable` — 只杀非 immune 工具（与 Steering 一致），默认值
    - `:none` — 不杀任何工具，让它们自然完成
    每杀一个工具 emit `{:tool_killed, %{name, call_id, reason}}` 事件。

  ## 4 状态行为

  | 状态 | 默认行为（killable）| `:all` | `:none` |
  |---|---|---|---|
  | `:idle` | 仅 emit `:agent_abort`（no-op） | 同 | 同 |
  | `:running` | cancel stream task | 同 | 仅 emit |
  | `:streaming` | cancel stream task | 同 | 仅 emit |
  | `:executing_tools` | 杀非 immune 工具 + cancel stream | 杀全部工具 | 留全部工具 |

  无论何种状态，`:agent_abort` 事件保证发出（订阅方 100ms 内收到，BEAM 调度延迟）。
  """
  @spec abort(GenServer.server(), keyword()) :: :ok
  def abort(agent, opts \\ []) when is_list(opts) do
    :gen_statem.cast(agent, {:abort, opts})
  end

  @doc """
  中段软中断（Steering）。

  - `idle` 状态：等同 `prompt/2`，立刻进入新 turn
  - 其他状态：text 入 `:steering_queue`，下个 turn 间隙合并注入

  详见 `guides/cookbook.md` 中的「中段干预（Steering）」配方。
  """
  @spec steer(GenServer.server(), reference(), String.t()) ::
          :ok | {:error, :queue_full | :rejected}
  def steer(agent, ref, text) when is_reference(ref) and is_binary(text) do
    :gen_statem.call(agent, {:steer, ref, text})
  end

  @doc """
  批准指定的工具审批请求。

  ## 选项

  - `:auto_resume` — boolean，默认 `true`。当 Agent 已经处于 idle（被 block_tool 拦下后回到 idle），
    审批通过后自动开启新 turn 让 LLM 重试被拦截的工具。设为 `false` 保持旧版行为
    （需调用方再 `prompt/2` 才能续）。

  自动续接成功时会 emit `{:agent_resumed, %{trigger: :tool_approved, approval_id: id}}`,
  上层可订阅事件确认 Agent 已重新进入 running 状态。
  """
  @spec approve(GenServer.server(), String.t(), keyword()) :: :ok
  def approve(agent, approval_id, opts \\ []) when is_binary(approval_id) and is_list(opts) do
    :gen_statem.cast(agent, {:tool_approved, approval_id, opts})
  end

  @doc """
  拒绝指定的工具审批请求。

  ## 选项

  - `:auto_resume` — boolean，默认 `false`。reject 默认不自动续 turn——HumanApproval Plugin
    只是把 awaiting 清掉并 emit `:approval_resolved` (status=:rejected)，调用方通常希望让
    Agent 保持 idle 等待新 prompt（被拒命令的 ToolMessage 已记录为 is_error，下次 prompt 时 LLM
    自然能感知到拒绝结果）。

    若希望 reject 后 Agent 立刻重新规划（例如让 LLM 走"被拒后换个方案"分支），传 `auto_resume: true`,
    会和 approve 一样开新 turn + emit `{:agent_resumed, %{trigger: :tool_rejected, ...}}`。
  """
  @spec reject(GenServer.server(), String.t(), keyword()) :: :ok
  def reject(agent, approval_id, opts \\ []) when is_binary(approval_id) and is_list(opts) do
    :gen_statem.cast(agent, {:tool_rejected, approval_id, opts})
  end

  @doc """
  运行期切换 model。

  下一次 LLM 调用立即生效。当前 streaming / executing_tools 不会被打断
  （语义：本轮跑完，下一轮再换；如果想立刻打断换模型，请先 `abort/2` 再 `switch_model/2` + `prompt/2`）。

  ## 选项

  - `:provider_opts :: keyword()` — 与 model 一并替换 provider 参数（base_url / api_key / timeout）;
    传 `nil` 或不传则保留现有 provider_opts;
    典型场景：从 Anthropic 切到 OpenAI 自建网关，需同步 `base_url`、`api_key`

  ## 行为

  - **不切换模型**（new_model 与 state.model 相同且无 provider_opts 变化）→ no-op，不发事件
  - **idle 状态**：立即更新 state.model / state.config.provider_opts，
    emit `{:model_switched, %{from, to, provider_opts_changed?}}`
  - **running / streaming / executing_tools**：更新 state.model，本轮继续用旧模型，
    下一轮自动用新模型；event 立即发出
  - **messages / tools / plugin_states 都保留**（同一会话，换模型继续）

  ## 兼容性警告

  - 不同模型对 system_prompt 的支持差异（如 OpenAI 的 system 消息 vs Anthropic 的 system 字段）
    由 Provider 层处理，不在 switch_model 范围内
  - 上下文窗口差异由调用方自行评估（如从 200k token 模型切到 8k 模型，需先 compact）
  - 不同模型的 tool_calling schema 兼容性由 Provider 适配
  """
  @spec switch_model(GenServer.server(), CMDC.Provider.model()) :: :ok
  def switch_model(agent, new_model) when is_binary(new_model) do
    :gen_statem.cast(agent, {:switch_model, new_model})
  end

  @spec switch_model(GenServer.server(), CMDC.Provider.model(), keyword()) :: :ok
  def switch_model(agent, new_model, opts) when is_binary(new_model) and is_list(opts) do
    provider_opts = Keyword.get(opts, :provider_opts)
    :gen_statem.cast(agent, {:switch_model, new_model, provider_opts})
  end

  @doc """
  运行期挂载新工具。

  立即写入 `state.tools`，下一次 LLM 请求生效（重生成 tools schema）。
  In-flight 请求不受影响。

  - 已存在同名 tool → `{:error, :already_attached}`
  - 模块未实现 `CMDC.Tool` behaviour → `{:error, :invalid_tool}`
  - 成功 → `:ok` + emit `{:tool_attached, name}`
  """
  @spec attach_tool(GenServer.server(), module()) ::
          :ok | {:error, :already_attached | :invalid_tool}
  def attach_tool(agent, tool_module) when is_atom(tool_module) do
    :gen_statem.call(agent, {:attach_tool, tool_module})
  end

  @doc """
  批量挂载多个工具（原子操作）。

  先对所有 tool 做 dry-run 校验（避免 invalid_tool / already_attached / 列表内重名）;
  任何一个失败 → 全部回滚，state.tools 完全不变。

  - 全部成功 → `{:ok, [name, ...]}` + 每个 emit `{:tool_attached, name}` + 汇总 emit
    `{:tools_updated, %{attached: [...], detached: []}}`
  - 失败 → `{:error, {:validation_failed, failures}}`，其中 failures 是 `[{module, reason}, ...]`

  典型场景：用户启用 MCP Server 时一次性挂载该 Server 提供的 N 个工具。
  """
  @spec attach_tools(GenServer.server(), [module()]) ::
          {:ok, [String.t()]} | {:error, {:validation_failed, [{module(), atom()}]}}
  def attach_tools(agent, tool_modules) when is_list(tool_modules) do
    :gen_statem.call(agent, {:attach_tools, tool_modules})
  end

  @doc """
  运行期卸载工具。

  立即从 `state.tools` 移除（按 `tool.name()` 字符串匹配），
  下一次 LLM 请求生效。In-flight tool 调用不受影响（仍跑完）。

  - 不存在同名 tool → `{:error, :not_found}`
  - 成功 → `:ok` + emit `{:tool_detached, name}`

  > **注意**：若 LLM 在 detach 后仍调用该 tool（已在 streaming 中或下一轮），
  > Agent 会发 `{:tool_call_unknown, name, call_id}` 并自动注入 error
  > tool_result，让 LLM 自我纠正。
  """
  @spec detach_tool(GenServer.server(), String.t()) :: :ok | {:error, :not_found}
  def detach_tool(agent, tool_name) when is_binary(tool_name) do
    :gen_statem.call(agent, {:detach_tool, tool_name})
  end

  @doc """
  批量卸载工具（原子操作）。

  先 dry-run 全部找到对应模块，任何一个 :not_found → 全回滚，state.tools 完全不变。

  - 全部成功 → `{:ok, [name, ...]}` + 每个 emit `{:tool_detached, name}` + 汇总 emit
    `{:tools_updated, %{attached: [], detached: [...]}}`
  - 失败 → `{:error, {:validation_failed, [{name, :not_found}, ...]}}`
  """
  @spec detach_tools(GenServer.server(), [String.t()]) ::
          {:ok, [String.t()]} | {:error, {:validation_failed, [{String.t(), atom()}]}}
  def detach_tools(agent, tool_names) when is_list(tool_names) do
    :gen_statem.call(agent, {:detach_tools, tool_names})
  end

  @doc """
  替换整张工具表（原子操作）。

  自动计算 diff：
  - 老 tools 在新列表里 → 保留
  - 老 tools 不在新列表里 → detach（emit `{:tool_detached, name}`）
  - 新 tools 不在老列表里 → attach（emit `{:tool_attached, name}`）

  全部 attach validate 失败 → 全回滚，emit `{:error, _}`。
  典型场景：MCP Server 重启 / 配置变更，新 tools 列表完全覆盖旧列表。

  返回 `{:ok, %{attached: [name, ...], detached: [name, ...]}}` + 汇总 emit
  `{:tools_updated, %{attached, detached}}`。
  """
  @spec replace_tools(GenServer.server(), [module()]) ::
          {:ok, %{attached: [String.t()], detached: [String.t()]}}
          | {:error, {:validation_failed, [{module(), atom()}]}}
  def replace_tools(agent, tool_modules) when is_list(tool_modules) do
    :gen_statem.call(agent, {:replace_tools, tool_modules})
  end

  @doc false
  @spec get_state(GenServer.server()) :: State.t()
  def get_state(agent), do: :gen_statem.call(agent, :get_state)

  @doc "获取完整的消息列表（含系统提示词，按时间顺序）。"
  @spec get_messages(GenServer.server()) :: [CMDC.Message.t()]
  def get_messages(agent), do: :gen_statem.call(agent, :get_messages)

  @doc """
  获取 Agent 的状态快照（含运行期可观测字段）。

  返回结构：

  - `:state` — 当前 gen_statem 状态（`:idle | :running | :streaming | :executing_tools`）
  - `:session_id` — 会话 ID
  - `:model` — 当前 LLM 模型字符串（可被 `switch_model/2` 改变）
  - `:turns` — 已完成轮次数（与 `:turns_count` 等同，保留兼容别名）
  - `:turns_count` — 已完成轮次数
  - `:tool_calls` — 已执行工具调用总次数
  - `:messages_count` — 当前 messages 列表长度（含 system + user + assistant + tool_result）
  - `:active_since_ms` — Agent 进程启动时间戳（毫秒，绝对时间排序；
    与 `:uptime_ms` 互补，后者为运行时长）
  - `:total_tokens` — 累计 token 用量（整数，向后兼容字段）
  - `:cost_usd` — 累计美元成本
  - `:token_usage` — `%CMDC.TokenUsage{}` struct，含
    `prompt_tokens / completion_tokens / total_tokens / cost_usd / cached_tokens`。
  - `:uptime_ms` — 运行时长（毫秒）
  - `:timestamp_ms` — 快照时间戳
  - `:pending_tools` — 当前正在执行的工具列表，每项包含
    `name / call_id / args / started_at_ms`（`started_at_ms` 单位
    `System.system_time(:millisecond)`，用于无须自己埋点即可统计 tool 耗时）。
  - `:pending_approvals` — 当前等待人类审批的工具调用，
    从启用了审批的 Plugin（如 `CMDC.Plugin.Builtin.HumanApproval`）的状态汇总;
    每项包含 `id / tool / args / session_id / requested_at` 等字段。
  - `:queues` — 各种内部队列长度:

      ```
      %{
        prompt_queue:   non_neg_integer(),  # 排队等执行的 prompt 数
        steering_queue: non_neg_integer()   # 中段软中断 queue 长度
      }
      ```
  """
  @spec status(GenServer.server()) :: %{
          state: :idle | :running | :streaming | :executing_tools,
          session_id: String.t(),
          model: String.t(),
          turns: non_neg_integer(),
          turns_count: non_neg_integer(),
          tool_calls: non_neg_integer(),
          messages_count: non_neg_integer(),
          total_tokens: non_neg_integer(),
          cost_usd: float(),
          token_usage: CMDC.TokenUsage.t(),
          uptime_ms: non_neg_integer(),
          active_since_ms: integer(),
          timestamp_ms: integer(),
          pending_tools: [
            %{
              name: String.t(),
              call_id: String.t(),
              args: map(),
              started_at_ms: integer()
            }
          ],
          pending_approvals: [map()],
          queues: %{prompt_queue: non_neg_integer(), steering_queue: non_neg_integer()}
        }
  def status(agent), do: :gen_statem.call(agent, :status)

  @doc "获取完整的消息列表（含系统提示词，按时间顺序）。"
  @spec messages(GenServer.server()) :: [CMDC.Message.t()]
  def messages(agent), do: :gen_statem.call(agent, :get_messages)

  @doc """
  登记当前进程对 Agent 的崩溃监控。

  Agent 退出（任何原因）时，观察者进程会收到：

      {:cmdc_down, ref, session_id, structured_reason}

  `structured_reason` 已结构化，常见值：
  `:normal | :shutdown | {:exception, term}`，未来扩展：
  `:max_turns_exceeded | :provider_timeout | {:plugin_aborted, name, why}`。

  返回 `reference()`，用 `demonitor/2` 取消监听。
  """
  @spec monitor(GenServer.server()) :: reference()
  def monitor(agent), do: :gen_statem.call(agent, {:monitor, self()})

  @doc "取消通过 `monitor/1` 登记的崩溃监控。"
  @spec demonitor(GenServer.server(), reference()) :: :ok
  def demonitor(agent, ref) when is_reference(ref),
    do: :gen_statem.call(agent, {:demonitor, ref})

  # ==========================================================================
  # gen_statem Callbacks
  # ==========================================================================

  @impl :gen_statem
  def callback_mode, do: :state_functions

  @impl :gen_statem
  def init(opts) do
    Process.flag(:trap_exit, true)
    session_id = Keyword.fetch!(opts, :session_id)
    raw_model = Keyword.fetch!(opts, :model)
    working_dir = Keyword.fetch!(opts, :working_dir)
    user_provider_opts = Keyword.get(opts, :provider_opts, [])
    :proc_lib.set_label("cmdc_agent:#{session_id}")

    case parse_model_spec(raw_model, user_provider_opts) do
      {:error, {:registry_profile_missing, name}} ->
        Logger.warning(
          "[Agent] init failed session=#{session_id}: registry profile #{inspect(name)} missing"
        )

        {:stop, {:registry_profile_missing, name}}

      {:error, reason} ->
        Logger.warning("[Agent] init failed session=#{session_id}: #{inspect(reason)}")
        {:stop, reason}

      {:ok, resolved_model, merged_opts, original_model_spec} ->
        do_init(
          opts,
          session_id,
          raw_model,
          resolved_model,
          merged_opts,
          original_model_spec,
          working_dir
        )
    end
  end

  defp do_init(
         opts,
         session_id,
         raw_model,
         resolved_model,
         merged_opts,
         original_model_spec,
         working_dir
       ) do
    plugin_specs = Keyword.get(opts, :plugins, [])
    registry = Registry.from_specs(plugin_specs)
    plugins = Registry.entries(registry)
    plugin_states = Map.new(plugins, fn {mod, ps} -> {mod, ps} end)

    initial_messages =
      opts
      |> Keyword.get(:messages, [])
      |> Enum.reverse()

    base_config =
      Keyword.get(opts, :config, %{})
      |> Map.merge(%{provider_opts: merged_opts})

    config =
      if original_model_spec do
        Map.put(base_config, :resolved_provider_model, resolved_model)
      else
        base_config
      end

    state =
      State.new(
        session_id: session_id,
        model: raw_model,
        working_dir: working_dir,
        blueprint_system_prompt: Keyword.get(opts, :blueprint_system_prompt, ""),
        messages: initial_messages,
        tools: Keyword.get(opts, :tools, []),
        disabled_tools: Keyword.get(opts, :disabled_tools, []),
        plugins: plugins,
        plugin_states: plugin_states,
        sandbox: Keyword.get(opts, :sandbox),
        subagents: Keyword.get(opts, :subagents, []),
        user_data: normalize_user_data(Keyword.get(opts, :user_data, %{})),
        prompt_mode: normalize_prompt_mode(Keyword.get(opts, :prompt_mode, :full)),
        config: config
      )

    case Keyword.get(opts, :event_buffer_size, 0) do
      n when is_integer(n) and n > 0 -> CMDC.EventBus.enable_buffer(session_id, n)
      _ -> :ok
    end

    hibernate_ms = Keyword.get(opts, :hibernate_after_ms)

    CMDC.Telemetry.execute(
      [:cmdc, :agent, :hibernate, :configured],
      %{ms: hibernate_ms || 0},
      %{session_id: session_id, enabled?: is_integer(hibernate_ms) and hibernate_ms > 0}
    )

    Logger.debug(
      "[Agent] ready session=#{session_id} plugins=#{length(plugins)} tools=#{length(state.tools)}"
    )

    {:ok, :idle, state}
  end

  @impl :gen_statem
  def terminate(reason, _state_name, %State{session_id: sid} = state) when is_binary(sid) do
    CMDC.EventBus.disable_buffer(sid)
    notify_monitors(state, reason)
    :ok
  end

  def terminate(_reason, _state_name, _state), do: :ok

  defp notify_monitors(%State{monitors: monitors, session_id: sid, exit_reason: override}, reason)
       when is_map(monitors) and map_size(monitors) > 0 do
    structured = structured_reason(override, reason)

    Enum.each(monitors, fn {ref, observer_pid} ->
      if is_pid(observer_pid) and Process.alive?(observer_pid) do
        send(observer_pid, {:cmdc_down, ref, sid, structured})
      end
    end)

    :ok
  end

  defp notify_monitors(_state, _reason), do: :ok

  defp structured_reason(nil, reason), do: normalize_reason(reason)
  defp structured_reason(override, _reason), do: override

  defp normalize_reason(:normal), do: :normal
  defp normalize_reason(:shutdown), do: :normal
  defp normalize_reason({:shutdown, _}), do: :normal

  defp normalize_reason({exception, _stacktrace}) when is_exception(exception),
    do: {:exception, exception}

  defp normalize_reason(exception) when is_exception(exception), do: {:exception, exception}
  defp normalize_reason(other), do: other

  # ╔═══════════════════════════════════════════════════════════════════╗
  # ║  :idle — 等待 prompt                                              ║
  # ╚═══════════════════════════════════════════════════════════════════╝

  @doc false
  def idle({:call, from}, {:prompt, text}, state) do
    Logger.debug("[Agent] prompt received session=#{state.session_id}")

    case run_pipeline(state, {:before_prompt, text}) do
      {:abort, state, reason} ->
        Logger.warning("[Agent] prompt rejected by hook: #{reason}")
        broadcast(state, {:prompt_rejected, reason})
        {:next_state, :idle, %{state | status: :idle}, [{:reply, from, %{queued: false}}]}

      {:ok, state, result} ->
        state =
          state
          |> State.mark_turn_start()
          |> State.append_message(CMDC.Message.user(text))
          |> maybe_append_before_prompt_intervention(result)
          |> Map.put(:status, :running)

        broadcast(state, {:prompt_received, text})
        maybe_broadcast_before_prompt_intervention(state, result)

        case run_pipeline(state, :session_start) do
          {:ok, state, _result} ->
            broadcast(state, :agent_start)

            {:next_state, :running, state,
             [{:reply, from, %{queued: false}}, {:next_event, :internal, :run_turn}]}

          {:abort, state, reason} ->
            state = emit_after_turn_for_abort(state, reason)
            broadcast(state, {:agent_abort, reason})
            {:next_state, :idle, %{state | status: :idle}, [{:reply, from, %{queued: false}}]}
        end
    end
  end

  def idle({:call, from}, {:steer, ref, text}, state),
    do: handle_steer(from, ref, text, state, :passthrough_to_prompt)

  def idle({:call, from}, msg, state), do: shared_call(from, msg, state)
  def idle(:cast, :abort, state), do: do_abort(state, [])
  def idle(:cast, {:abort, opts}, state), do: do_abort(state, opts)

  def idle(:cast, {:switch_model, new_model}, state),
    do: do_switch_model(state, new_model, nil)

  def idle(:cast, {:switch_model, new_model, provider_opts}, state),
    do: do_switch_model(state, new_model, provider_opts)

  def idle(:cast, {:tool_approved, id, opts}, state),
    do: route_approval_cast(:tool_approved, id, opts, state)

  def idle(:cast, {:tool_approved, id}, state),
    do: route_approval_cast(:tool_approved, id, [], state)

  def idle(:cast, {:tool_rejected, id, opts}, state),
    do: route_approval_cast(:tool_rejected, id, opts, state)

  def idle(:cast, {:tool_rejected, id}, state),
    do: route_approval_cast(:tool_rejected, id, [], state)

  def idle(:info, {:cmdc_approval_timeout, id}, state),
    do: route_approval_cast(:tool_approval_timeout, id, [], state)

  def idle(_event_type, _event, state), do: {:keep_state, state}

  # ╔═══════════════════════════════════════════════════════════════════╗
  # ║  :running — 准备调用 Provider 或等待重试                            ║
  # ╚═══════════════════════════════════════════════════════════════════╝

  @doc false
  def running(:internal, :run_turn, state), do: run_turn(state)
  def running(:info, :retry_turn, state), do: run_turn(state)
  def running({:call, from}, {:prompt, text}, state), do: enqueue(from, text, state)

  def running({:call, from}, {:steer, ref, text}, state),
    do: handle_steer(from, ref, text, state, :enqueue)

  def running({:call, from}, msg, state), do: shared_call(from, msg, state)
  def running(:cast, :abort, state), do: do_abort(state, [])
  def running(:cast, {:abort, opts}, state), do: do_abort(state, opts)

  def running(:cast, {:switch_model, new_model}, state),
    do: do_switch_model(state, new_model, nil)

  def running(:cast, {:switch_model, new_model, provider_opts}, state),
    do: do_switch_model(state, new_model, provider_opts)

  def running(:cast, {:tool_approved, id, opts}, state),
    do: route_approval_cast(:tool_approved, id, opts, state)

  def running(:cast, {:tool_approved, id}, state),
    do: route_approval_cast(:tool_approved, id, [], state)

  def running(:cast, {:tool_rejected, id, opts}, state),
    do: route_approval_cast(:tool_rejected, id, opts, state)

  def running(:cast, {:tool_rejected, id}, state),
    do: route_approval_cast(:tool_rejected, id, [], state)

  def running(:info, {:cmdc_approval_timeout, id}, state),
    do: route_approval_cast(:tool_approval_timeout, id, [], state)

  def running(_event_type, _event, state), do: {:keep_state, state}

  # ╔═══════════════════════════════════════════════════════════════════╗
  # ║  :streaming — 消费 LLM 响应 chunk                                 ║
  # ╚═══════════════════════════════════════════════════════════════════╝

  @stall_ms 10_000
  @max_stall_count 3

  @doc false
  def streaming(:info, {:cmdc_stream_chunk, chunk}, state) do
    state = handle_stream_chunk(chunk, state)

    if state.stream_errored != false do
      recover_stream_error(state)
    else
      {:keep_state, %{state | stall_count: 0}, [stall_timeout()]}
    end
  end

  def streaming(:info, :cmdc_stream_done, state) do
    finalize_response(state)
  end

  def streaming(:info, {:cmdc_stream_error, reason}, state) do
    Logger.error("[Agent] stream error: #{inspect(reason)}")
    broadcast(state, {:stream_error, reason})
    handle_provider_error(%{state | stream_errored: false}, reason)
  end

  def streaming(:state_timeout, :stall_check, state) do
    new_count = state.stall_count + 1
    elapsed_s = div(@stall_ms + new_count * 5_000, 1_000)

    broadcast(state, {:stream_stalled, elapsed_s})

    if new_count >= @max_stall_count do
      Logger.warning(
        "[Agent] stream stalled #{elapsed_s}s (count=#{new_count}), killing bridge and retrying"
      )

      if is_pid(state.stream_task_pid) and Process.alive?(state.stream_task_pid) do
        Process.exit(state.stream_task_pid, :kill)
      end

      state = %{state | stall_count: 0, stream_task_pid: nil}
      handle_provider_error(state, "stream_timeout: no chunk for #{elapsed_s}s")
    else
      {:keep_state, %{state | stall_count: new_count}, [{:state_timeout, 5_000, :stall_check}]}
    end
  end

  def streaming({:call, from}, {:prompt, text}, state), do: enqueue(from, text, state)

  def streaming({:call, from}, {:steer, ref, text}, state),
    do: handle_steer(from, ref, text, state, :enqueue)

  def streaming({:call, from}, msg, state), do: shared_call(from, msg, state)
  def streaming(:cast, :abort, state), do: do_abort(state, [])
  def streaming(:cast, {:abort, opts}, state), do: do_abort(state, opts)

  def streaming(:cast, {:switch_model, new_model}, state),
    do: do_switch_model(state, new_model, nil)

  def streaming(:cast, {:switch_model, new_model, provider_opts}, state),
    do: do_switch_model(state, new_model, provider_opts)

  def streaming(:cast, {:tool_approved, id, opts}, state),
    do: route_approval_cast(:tool_approved, id, opts, state)

  def streaming(:cast, {:tool_approved, id}, state),
    do: route_approval_cast(:tool_approved, id, [], state)

  def streaming(:cast, {:tool_rejected, id, opts}, state),
    do: route_approval_cast(:tool_rejected, id, opts, state)

  def streaming(:cast, {:tool_rejected, id}, state),
    do: route_approval_cast(:tool_rejected, id, [], state)

  def streaming(:info, {:cmdc_approval_timeout, id}, state),
    do: route_approval_cast(:tool_approval_timeout, id, [], state)

  def streaming(_event_type, _event, state), do: {:keep_state, state}

  # ╔═══════════════════════════════════════════════════════════════════╗
  # ║  :executing_tools — 并发执行工具调用                                ║
  # ╚═══════════════════════════════════════════════════════════════════╝

  @doc false
  def executing_tools(
        :info,
        {ref, result},
        %State{pending_tool_tasks: tasks} = state
      )
      when is_reference(ref) and is_map_key(tasks, ref) do
    {task, tc, _started_at_ms} = Map.fetch!(tasks, ref)
    Process.demonitor(task.ref, [:flush])
    ToolRunner.collect_result(ref, tc, result, state) |> next()
  end

  def executing_tools(
        :info,
        {:DOWN, ref, :process, _pid, reason},
        %State{pending_tool_tasks: tasks} = state
      )
      when is_map_key(tasks, ref) do
    {_task, tc, _started_at_ms} = Map.fetch!(tasks, ref)
    Logger.error("[Agent] tool crashed: #{inspect(reason)}")

    ToolRunner.collect_result(ref, tc, {:error, "Tool crashed: #{inspect(reason)}"}, state)
    |> next()
  end

  def executing_tools({:call, from}, {:prompt, text}, state), do: enqueue(from, text, state)

  def executing_tools({:call, from}, {:steer, ref, text}, state),
    do: handle_steer(from, ref, text, state, :enqueue)

  def executing_tools({:call, from}, msg, state), do: shared_call(from, msg, state)
  def executing_tools(:cast, :abort, state), do: do_abort(state, [])
  def executing_tools(:cast, {:abort, opts}, state), do: do_abort(state, opts)

  def executing_tools(:cast, {:switch_model, new_model}, state),
    do: do_switch_model(state, new_model, nil)

  def executing_tools(:cast, {:switch_model, new_model, provider_opts}, state),
    do: do_switch_model(state, new_model, provider_opts)

  def executing_tools(:cast, {:tool_approved, id, opts}, state),
    do: route_approval_cast(:tool_approved, id, opts, state)

  def executing_tools(:cast, {:tool_approved, id}, state),
    do: route_approval_cast(:tool_approved, id, [], state)

  def executing_tools(:cast, {:tool_rejected, id, opts}, state),
    do: route_approval_cast(:tool_rejected, id, opts, state)

  def executing_tools(:cast, {:tool_rejected, id}, state),
    do: route_approval_cast(:tool_rejected, id, [], state)

  def executing_tools(:info, {:cmdc_approval_timeout, id}, state),
    do: route_approval_cast(:tool_approval_timeout, id, [], state)

  def executing_tools(_event_type, _event, state), do: {:keep_state, state}

  # ==========================================================================
  # Approval Cast Routing
  # ==========================================================================

  defp route_approval_cast(event_name, approval_id, opts, state) do
    case run_pipeline(state, {event_name, approval_id}) do
      {:ok, new_state, _} ->
        maybe_auto_resume(event_name, approval_id, opts, new_state)

      {:abort, new_state, _} ->
        {:keep_state, new_state}
    end
  end

  # 默认 auto_resume：approve / approval_timeout 自动续；reject 默认不续。
  # 仅在 status == :idle 时实际触发；其他状态 Agent 已在工作，opts 等同 no-op。
  defp maybe_auto_resume(event_name, approval_id, opts, state) do
    default = event_name in [:tool_approved, :tool_approval_timeout]
    auto_resume? = Keyword.get(opts, :auto_resume, default)

    if auto_resume? and state.status == :idle do
      Logger.debug(
        "[Agent] auto-resume after #{event_name} approval_id=#{approval_id} session=#{state.session_id}"
      )

      broadcast(state, {:agent_resumed, %{trigger: event_name, approval_id: approval_id}})
      state = state |> State.mark_turn_start() |> Map.put(:status, :running)
      {:next_state, :running, state, [{:next_event, :internal, :run_turn}]}
    else
      {:keep_state, state}
    end
  end

  # ==========================================================================
  # Shared Call Dispatch
  # ==========================================================================

  defp shared_call(from, msg, state) do
    {reply, state} = handle_call(msg, state)
    {:keep_state, state, [{:reply, from, reply}]}
  end

  defp handle_call(:get_state, state), do: {state, state}
  defp handle_call(:get_messages, state), do: {build_messages(state), state}

  defp handle_call({:attach_tool, tool_module}, state) do
    do_attach_tool(state, tool_module)
  end

  defp handle_call({:detach_tool, tool_name}, state) do
    do_detach_tool(state, tool_name)
  end

  defp handle_call({:attach_tools, tool_modules}, state) do
    do_attach_tools(state, tool_modules)
  end

  defp handle_call({:detach_tools, tool_names}, state) do
    do_detach_tools(state, tool_names)
  end

  defp handle_call({:replace_tools, tool_modules}, state) do
    do_replace_tools(state, tool_modules)
  end

  defp handle_call({:monitor, observer_pid}, state) when is_pid(observer_pid) do
    ref = make_ref()
    monitors = Map.put(state.monitors || %{}, ref, observer_pid)
    {ref, %{state | monitors: monitors}}
  end

  defp handle_call({:demonitor, ref}, state) when is_reference(ref) do
    monitors = Map.delete(state.monitors || %{}, ref)
    {:ok, %{state | monitors: monitors}}
  end

  defp handle_call(:status, state) do
    now = System.system_time(:millisecond)
    token_usage = build_token_usage_struct(state)

    info = %{
      state: state.status,
      session_id: state.session_id,
      model: state.model,
      turns: state.turn_count,
      turns_count: state.turn_count,
      tool_calls: state.tool_call_count,
      messages_count: length(state.messages),
      total_tokens: state.token_usage.total_tokens,
      cost_usd: state.cost_usd,
      token_usage: token_usage,
      uptime_ms: now - state.started_at,
      active_since_ms: state.started_at,
      timestamp_ms: now,
      pending_tools: build_pending_tools(state),
      pending_approvals: build_pending_approvals(state),
      queues: build_queue_info(state)
    }

    {info, state}
  end

  defp build_pending_tools(%State{pending_tool_tasks: tasks}) do
    tasks
    |> Enum.map(fn {_ref, {_task, tc, started_at_ms}} ->
      %{
        name: Map.get(tc, :name) || "",
        call_id: Map.get(tc, :call_id) || "",
        args: Map.get(tc, :arguments) || %{},
        started_at_ms: started_at_ms
      }
    end)
  end

  # 从 plugin_states 汇总所有 Plugin 暴露的 awaiting_approvals。
  # 兼容两种格式：
  #   1. HumanApproval 当前格式 {approval_map, command_key, timer_ref}
  #      → 直接取 approval_map（自带 :id 字段）
  #   2. 通用格式 value 直接是 approval map → 自动补 :id 键
  defp build_pending_approvals(%State{plugin_states: ps}) do
    Enum.flat_map(ps, fn
      {_module, %{awaiting_approvals: aa}} when is_map(aa) ->
        Enum.flat_map(aa, &normalize_awaiting_entry/1)

      _ ->
        []
    end)
  end

  defp normalize_awaiting_entry({_id, {%{} = approval, _key, _timer}}), do: [approval]

  defp normalize_awaiting_entry({id, %{} = approval}), do: [Map.put(approval, :id, id)]

  defp normalize_awaiting_entry(_), do: []

  defp build_queue_info(%State{pending_messages: pq, steering_queue: sq}) do
    %{
      prompt_queue: length(pq),
      steering_queue: length(sq)
    }
  end

  # ==========================================================================
  # Prompt Queuing
  # ==========================================================================

  defp enqueue(from, text, state) do
    Logger.debug("[Agent] prompt queued session=#{state.session_id}")
    broadcast(state, {:prompt_queued, text})

    {:keep_state, %{state | pending_messages: state.pending_messages ++ [text]},
     [{:reply, from, %{queued: true}}]}
  end

  # ==========================================================================
  # Steering — 中段软中断
  # ==========================================================================

  # mode = :passthrough_to_prompt （idle 状态）/ :enqueue（其他 3 个状态）
  defp handle_steer(from, ref, text, state, mode) do
    case run_pipeline(state, {:before_steering, text}) do
      {:abort, state, reason} ->
        emit_steering_received(state, ref, text, :rejected_by_plugin)
        Logger.warning("[Agent] steering rejected by plugin: #{inspect(reason)}")
        {:keep_state, state, [{:reply, from, {:error, :rejected}}]}

      {:ok, state, %{action: :intervene} = result} ->
        effective = Pipeline.merged_interventions(result) || text
        do_steer(from, ref, effective, state, mode)

      {:ok, state, _result} ->
        do_steer(from, ref, text, state, mode)
    end
  end

  # idle: 透传到 prompt 路径，但回 :ok 给 steer 调用方
  # 注意：idle 不入 queue，所以不 emit :steering_received，
  # 直接在最终注入成功时 emit :steering_applied（表示"立刻生效"）。
  defp do_steer(from, ref, text, state, :passthrough_to_prompt) do
    case run_pipeline(state, {:before_prompt, text}) do
      {:abort, state, reason} ->
        Logger.warning("[Agent] steer (idle) rejected by before_prompt: #{inspect(reason)}")
        broadcast(state, {:prompt_rejected, reason})
        {:keep_state, state, [{:reply, from, {:error, :rejected}}]}

      {:ok, state, result} ->
        state =
          state
          |> State.mark_turn_start()
          |> State.append_message(CMDC.Message.user(text))
          |> maybe_append_before_prompt_intervention(result)
          |> Map.put(:status, :running)

        broadcast(state, {:prompt_received, text})
        maybe_broadcast_before_prompt_intervention(state, result)

        case run_pipeline(state, :session_start) do
          {:ok, state, _result} ->
            broadcast(state, :agent_start)
            broadcast(state, {:steering_applied, %{refs: [ref], count: 1}})

            {:next_state, :running, state,
             [{:reply, from, :ok}, {:next_event, :internal, :run_turn}]}

          {:abort, state, reason} ->
            state = emit_after_turn_for_abort(state, reason)
            broadcast(state, {:agent_abort, reason})

            {:next_state, :idle, %{state | status: :idle}, [{:reply, from, {:error, :rejected}}]}
        end
    end
  end

  # 其他 3 状态：入 queue + emit
  defp do_steer(from, ref, text, state, :enqueue) do
    if State.steering_queue_full?(state) do
      emit_steering_received(state, ref, text, :rejected_full)
      Logger.warning("[Agent] steering queue full session=#{state.session_id}")
      {:keep_state, state, [{:reply, from, {:error, :queue_full}}]}
    else
      state = State.enqueue_steering(state, ref, text)
      Logger.debug("[Agent] steering enqueued session=#{state.session_id} ref=#{inspect(ref)}")
      [entry | _] = Enum.reverse(state.steering_queue)
      payload = %{ref: ref, text: text, queued_at: entry.queued_at, status: :queued}
      broadcast(state, {:steering_received, payload})
      {:keep_state, state, [{:reply, from, :ok}]}
    end
  end

  defp emit_steering_received(state, ref, text, status) do
    payload = %{
      ref: ref,
      text: text,
      queued_at: System.system_time(:millisecond),
      status: status
    }

    broadcast(state, {:steering_received, payload})
  end

  # 把 :steering_queue 中的全部条目合并为 1 条 `[Steering]` 前缀 user message，
  # append 到 messages，emit :steering_applied，返回 `{:next_turn, state}` 供 next/1 处理。
  # ToolRunner.check_steering 在 in-flight tool 全部清理完后也会调这个 helper。
  @doc false
  @spec inject_steering(State.t()) :: {:next_turn, State.t()}
  def inject_steering(%State{} = state) do
    {entries, state} = State.drain_steering(state)
    text = format_steering_message(entries)
    msg = CMDC.Message.user(text)
    state = State.append_message(state, msg)

    refs = Enum.map(entries, & &1.ref)
    count = length(entries)

    broadcast(state, {:steering_applied, %{refs: refs, count: count}})
    # 让前端在统一的 :agent_resumed 监听点感知 steering 接管
    broadcast(state, {:agent_resumed, %{trigger: :steering, refs: refs, count: count}})

    {:next_turn, %{state | status: :running}}
  end

  defp format_steering_message(entries) do
    numbered =
      entries
      |> Enum.with_index(1)
      |> Enum.map_join("\n", fn {e, i} -> "#{i}. #{e.text}" end)

    "[Steering] 用户在执行中追加了以下指引（按时间顺序）：\n\n" <>
      numbered <>
      "\n\n请基于这些新信息重新规划下一步。"
  end

  # ==========================================================================
  # Turn Lifecycle
  # ==========================================================================

  defp run_turn(state) do
    state = repair_orphaned_calls(state)

    {state, _truncated} = ArgTruncator.truncate(state)

    state = maybe_compact_with_plugins(state)

    messages = build_messages(state)

    case run_pipeline(state, {:before_request, messages}) do
      {:ok, state, _result} ->
        broadcast(state, {:request_start, %{model: state.model, messages: length(messages)}})
        start_streaming(state, messages)

      {:abort, state, reason} ->
        Logger.warning("[Agent] before_request aborted: #{reason}")
        state = emit_after_turn_for_abort(state, reason)
        broadcast(state, {:agent_abort, reason})
        {:next_state, :idle, %{state | status: :idle}}
    end
  end

  defp start_streaming(state, messages) do
    case safe_provider_stream(state, messages) do
      {:ok, %{bridge_pid: bridge_pid}} ->
        state =
          state
          |> State.reset_stream_fields()
          |> Map.put(:status, :streaming)
          |> Map.put(:stream_task_pid, bridge_pid)
          |> State.increment_turn()

        {:next_state, :streaming, state, [stall_timeout()]}

      {:error, reason} ->
        Logger.error("[Agent] provider stream failed: #{inspect(reason)}")
        handle_provider_error(state, reason)
    end
  end

  defp safe_provider_stream(state, messages) do
    provider_opts = Map.get(state.config, :provider_opts) || []
    opts = [{:agent_pid, self()} | provider_opts]

    case Map.get(state.config, :provider_fn) do
      nil ->
        CMDC.Provider.stream(provider_model_for_request(state), messages, state.tools, opts)

      provider_fn when is_function(provider_fn, 4) ->
        provider_fn.(provider_model_for_request(state), messages, state.tools, opts)
    end
  rescue
    e -> {:error, {:provider_crash, e}}
  end

  # 取真实给 Provider 调用的 model 字符串。
  # 若 state.model 是 "registry:..." 协议，Agent.init/do_switch_model_state
  # 已解析后存到 state.config[:resolved_provider_model]，这里优先取这个；
  # 否则 state.model 本身就是 raw "provider:model_id"。
  defp provider_model_for_request(state) do
    Map.get(state.config, :resolved_provider_model) || state.model
  end

  defp handle_stream_chunk(chunk, state) do
    state = %{state | last_chunk_at: System.system_time(:millisecond)}
    Stream.handle_chunk(chunk, state)
  end

  # ==========================================================================
  # Response Finalization
  # ==========================================================================

  defp finalize_response(state) do
    tool_calls =
      state.current_tool_calls
      |> merge_arg_buffers(state.tool_call_arg_buffers)
      |> finalize_tool_calls()

    thinking =
      if state.current_thinking in [nil, ""],
        do: [],
        else: [thinking: state.current_thinking]

    assistant = CMDC.Message.assistant(state.current_text, tool_calls, thinking)

    state =
      state
      |> State.append_message(assistant)
      |> Map.put(:streaming_resp, nil)
      |> Map.put(:status, :running)
      |> Map.put(:retry_count, 0)

    broadcast(state, {:response_complete, assistant})

    case run_pipeline(state, {:after_response, assistant}) do
      {:ok, state, _result} ->
        dispatch(state, tool_calls)

      {:abort, state, reason} ->
        Logger.warning("[Agent] after_response aborted: #{reason}")
        state = emit_after_turn_for_abort(state, reason)
        broadcast(state, {:agent_abort, reason})
        {:next_state, :idle, %{state | status: :idle}}
    end
  end

  # ==========================================================================
  # Dispatch — 决定下一步动作
  # ==========================================================================

  defp dispatch(state, tool_calls) when tool_calls != [] do
    {known_calls, unknown_calls} = split_tool_calls(tool_calls, state.tools)
    state = handle_unknown_tool_calls(state, unknown_calls)

    case known_calls do
      [] ->
        dispatch(state, [])

      calls ->
        broadcast(state, {:tool_calls, length(calls)})
        ToolRunner.execute_batch(calls, state) |> next()
    end
  end

  defp dispatch(%State{steering_queue: [_ | _]} = state, []) do
    state |> inject_steering() |> next()
  end

  defp dispatch(state, []) do
    state = drain_pending(state)

    if has_pending_user_message?(state) do
      run_turn(state)
    else
      try_finish(state)
    end
  end

  defp split_tool_calls(tool_calls, tools) do
    tool_names = MapSet.new(tools, & &1.name())

    Enum.split_with(tool_calls, fn tc ->
      MapSet.member?(tool_names, tc.name)
    end)
  end

  # 当 LLM 调用不在 tools 列表里的工具（通常是 detach 后已 streaming 的引用）：
  # 1. emit {:tool_call_unknown, name, call_id} 事件，便于订阅方诊断
  # 2. 注入 synthetic tool_result {:error, "tool not found: ..."} 让 LLM 自我纠正
  defp handle_unknown_tool_calls(state, []), do: state

  defp handle_unknown_tool_calls(state, unknown_calls) do
    Logger.debug(
      "[Agent] #{length(unknown_calls)} unknown tool call(s): " <>
        Enum.map_join(unknown_calls, ", ", & &1.name)
    )

    Enum.reduce(unknown_calls, state, fn tc, acc ->
      broadcast(acc, {:tool_call_unknown, tc.name, tc.call_id})

      msg =
        CMDC.Message.tool_result(
          tc.call_id,
          "tool not found: #{tc.name} (it may have been detached at runtime)",
          true
        )

      %{acc | messages: [msg | acc.messages]}
    end)
  end

  # ==========================================================================
  # Tool Execution — 委托 ToolRunner（任务 1.11）
  # ==========================================================================

  # ==========================================================================
  # before_finish — 结束前检查
  # ==========================================================================

  defp try_finish(state) do
    case run_pipeline(state, :before_finish) do
      {:ok, state, %{action: :abort, halt_reason: reason}} ->
        Logger.warning("[Agent] before_finish aborted: #{reason}")
        state = emit_after_turn_for_abort(state, reason)
        broadcast(state, {:agent_abort, reason})
        {:next_state, :idle, %{state | status: :idle}}

      {:ok, state, %{action: :intervene} = result} ->
        prompt = Pipeline.merged_interventions(result)
        Logger.info("[Agent] before_finish intervened, injecting prompt")
        broadcast(state, {:intervention, prompt})

        state =
          state
          |> State.append_message(CMDC.Message.user(prompt))
          |> Map.put(:status, :running)

        {:next_state, :running, state, [{:next_event, :internal, :run_turn}]}

      {:ok, state, _result} ->
        finish(state)

      {:abort, state, reason} ->
        Logger.warning("[Agent] before_finish pipeline error: #{inspect(reason)}")
        state = emit_after_turn_for_abort(state, reason)
        broadcast(state, {:agent_abort, reason})
        {:next_state, :idle, %{state | status: :idle}}
    end
  end

  defp finish(state) do
    state = emit_after_turn(state, :finished, nil)
    run_pipeline(state, :session_end)
    broadcast(state, {:agent_end, Enum.reverse(state.messages), build_token_usage_struct(state)})
    {:next_state, :idle, %{state | status: :idle}}
  end

  # ==========================================================================
  # :after_turn Plugin hook
  # ==========================================================================
  #
  # 在 Agent 即将回 idle 前触发，payload 含本次 prompt cycle 的 messages_diff
  # 与 token_usage_diff，方便 Plugin 写入审计 / 长期记忆 / 计费等场景。
  #
  # 与 `:session_end` 的区别：
  #   - `:session_end` 保持现状，无 payload，主要供老 Plugin（如 EventLogger）使用
  #   - `:after_turn` 带结构化 payload，新 Plugin 推荐使用此 hook
  #
  # marker 为 nil 时（例如 prompt 在 `:before_prompt` 被拒、未进入 prompt cycle）跳过。

  defp emit_after_turn(state, outcome, abort_reason) do
    case State.consume_turn_marker(state, outcome, abort_reason) do
      {nil, state} ->
        state

      {payload, state} ->
        case run_pipeline(state, {:after_turn, payload}) do
          {:ok, new_state, _result} -> new_state
          {:abort, new_state, _reason} -> new_state
        end
    end
  end

  defp emit_after_turn_for_abort(state, reason) do
    emit_after_turn(state, :aborted, reason)
  end

  @spec build_token_usage_struct(State.t()) :: CMDC.TokenUsage.t()
  defp build_token_usage_struct(%State{} = state) do
    CMDC.TokenUsage.from_state_token_map(state.token_usage, state.cost_usd, state.cached_tokens)
  end

  # ==========================================================================
  # Continuation — 执行结果 → gen_statem 动作
  # ==========================================================================

  defp next({:next_turn, state}) do
    {:next_state, :running, state, [{:next_event, :internal, :run_turn}]}
  end

  defp next({:intervene, prompt, state}) do
    Logger.info("[Agent] loop detection intervened, injecting prompt")
    broadcast(state, {:intervention, prompt})

    state =
      state
      |> State.append_message(CMDC.Message.user(prompt))
      |> Map.put(:status, :running)

    {:next_state, :running, state, [{:next_event, :internal, :run_turn}]}
  end

  defp next({:abort, state}) do
    state = emit_after_turn_for_abort(state, :pipeline_abort)
    {:next_state, :idle, %{state | status: :idle}}
  end

  defp next({:next_state, state_name, state}) do
    {:next_state, state_name, state}
  end

  defp next(%State{} = state) do
    {:next_state, State.state_name(state), state}
  end

  # ==========================================================================
  # Error Handling
  # ==========================================================================

  defp handle_provider_error(state, reason) do
    cond do
      Overflow.context_overflow?(reason) ->
        Logger.warning("[Agent] context overflow: #{inspect(reason)}")
        broadcast(state, {:context_overflow, reason})
        state = emit_before_compact(state)

        case Compactor.force_compact(state) do
          {:compacted, state} ->
            Logger.info("[Agent] compacted after overflow, retrying turn")
            state = %{state | status: :running, overflow_detected: true}
            {:next_state, :running, state, [{:next_event, :internal, :run_turn}]}

          {:skip, state} ->
            state = emit_after_turn_for_abort(state, {:context_overflow_unrecoverable, reason})
            broadcast(state, {:error, state.session_id, reason})

            {:next_state, :idle,
             %{state | status: :idle, retry_count: 0, overflow_detected: true}}
        end

      Retry.retryable?(reason) and state.retry_count < state.max_retries ->
        schedule_retry(state, reason)

      true ->
        state = emit_after_turn_for_abort(state, {:provider_error, reason})
        broadcast(state, {:error, state.session_id, reason})
        {:next_state, :idle, %{state | status: :idle, retry_count: 0}}
    end
  end

  defp recover_stream_error(state) do
    reason = state.stream_errored
    broadcast(state, {:stream_error, reason})
    state = %{state | stream_errored: false}
    handle_provider_error(state, reason)
  end

  defp schedule_retry(state, reason) do
    attempt = state.retry_count + 1

    delay =
      Retry.delay(attempt, base_ms: state.retry_base_delay_ms, max_ms: state.retry_max_delay_ms)

    Logger.info("[Agent] retrying in #{delay}ms (attempt #{attempt}/#{state.max_retries})")
    broadcast(state, {:retry, attempt, delay, reason})
    Process.send_after(self(), :retry_turn, delay)
    {:next_state, :running, %{state | status: :running, retry_count: attempt}}
  end

  # ==========================================================================
  # Abort
  # ==========================================================================

  defp do_abort(state, opts) do
    reason = opts |> Keyword.get(:reason) |> normalize_abort_reason()
    clear_queue? = Keyword.get(opts, :clear_queue, true)
    kill_mode = Keyword.get(opts, :kill_tools, :killable)

    case run_pipeline(state, :before_stop) do
      {:ok, new_state, %{action: :intervene} = result} ->
        prompt = Pipeline.merged_interventions(result)
        Logger.info("[Agent] stop blocked by hook, injecting prompt")
        broadcast(new_state, {:stop_blocked, prompt})

        new_state =
          new_state
          |> State.append_message(CMDC.Message.user(prompt))
          |> Map.put(:status, :running)

        {:next_state, :running, new_state, [{:next_event, :internal, :run_turn}]}

      {:ok, new_state, _result} ->
        finalize_abort(new_state, reason, kill_mode, clear_queue?)

      {:abort, new_state, _abort_reason} ->
        finalize_abort(new_state, reason, kill_mode, clear_queue?)
    end
  end

  defp finalize_abort(state, reason, kill_mode, clear_queue?) do
    state =
      state
      |> cancel_stream()
      |> kill_tools_for_abort(kill_mode, reason)
      |> drop_pending_messages(clear_queue?)
      |> emit_after_turn(:aborted, reason)

    run_pipeline(state, :session_end)
    broadcast(state, build_agent_abort_event(reason))
    {:next_state, :idle, %{state | status: :idle}}
  end

  defp kill_tools_for_abort(state, mode, reason) do
    abort_reason = reason || :aborted
    {state, _killed} = ToolRunner.cancel_for_abort(state, mode, abort_reason)
    state
  end

  defp drop_pending_messages(%State{pending_messages: []} = state, _), do: state
  defp drop_pending_messages(state, false), do: state

  defp drop_pending_messages(%State{pending_messages: pending} = state, true) do
    Enum.each(pending, fn
      text when is_binary(text) ->
        broadcast(state, {:prompt_dropped, text})

      %CMDC.Message{role: :user, content: text} when is_binary(text) ->
        broadcast(state, {:prompt_dropped, text})

      _ ->
        :ok
    end)

    %{state | pending_messages: []}
  end

  defp build_agent_abort_event(nil), do: :agent_abort
  defp build_agent_abort_event(reason), do: {:agent_abort, reason}

  # abort/2 reason 白名单 + 字符串归一化。
  #
  # 标准 reason 列表（接受 atom 或对应 string）:
  #   :user_cancelled / :timeout / :shutdown / :budget_exceeded /
  #   :permission_denied / :provider_error
  # nil 透传（保持现状的无 reason 行为）
  # 其他 atom 原样保留（向后兼容老调用方）
  # 未识别的 string → :unknown + Logger.warning（防止前端注入任意 atom 进 BEAM atom table）。

  defp normalize_abort_reason(nil), do: nil
  defp normalize_abort_reason(reason) when is_atom(reason), do: reason

  defp normalize_abort_reason("user_cancelled"), do: :user_cancelled
  defp normalize_abort_reason("timeout"), do: :timeout
  defp normalize_abort_reason("shutdown"), do: :shutdown
  defp normalize_abort_reason("budget_exceeded"), do: :budget_exceeded
  defp normalize_abort_reason("permission_denied"), do: :permission_denied
  defp normalize_abort_reason("provider_error"), do: :provider_error

  defp normalize_abort_reason(reason) when is_binary(reason) do
    Logger.warning("[Agent] abort reason #{inspect(reason)} not in whitelist; using :unknown")
    :unknown
  end

  defp normalize_abort_reason(other) do
    Logger.warning(
      "[Agent] abort reason #{inspect(other)} unsupported type; coercing to :unknown"
    )

    :unknown
  end

  defp cancel_stream(state) do
    if is_pid(state.stream_task_pid) and Process.alive?(state.stream_task_pid) do
      Process.exit(state.stream_task_pid, :kill)
    end

    State.reset_stream_fields(state)
  end

  # ==========================================================================
  # Messages
  # ==========================================================================

  defp build_messages(state) do
    chronological = Enum.reverse(state.messages)
    chronological = Repair.ensure_tool_results(chronological)
    sys_prompt = build_system_prompt(state)

    if sys_prompt != "" do
      [CMDC.Message.system(sys_prompt) | chronological]
    else
      chronological
    end
  end

  defp build_system_prompt(state) do
    SystemPrompt.build(state)
  end

  # ==========================================================================
  # Pending Messages
  # ==========================================================================

  defp drain_pending(%{pending_messages: []} = state), do: state

  defp drain_pending(%{pending_messages: [msg | rest]} = state) do
    state
    |> State.append_message(CMDC.Message.user(msg))
    |> Map.put(:pending_messages, rest)
  end

  defp has_pending_user_message?(state) do
    case state.messages do
      [%{role: :user} | _] -> true
      _ -> false
    end
  end

  # ==========================================================================
  # Events
  # ==========================================================================

  defp broadcast(%State{} = state, event) do
    Emitter.broadcast(state, event)
  end

  # 容错把 keyword 转成 map；用于直接用 Agent.start_link/1 时的 user_data 入参。
  defp normalize_user_data(value) when is_map(value), do: value
  defp normalize_user_data(value) when is_list(value), do: Map.new(value)
  defp normalize_user_data(_), do: %{}

  defp normalize_prompt_mode(mode) when mode in [:full, :task, :minimal, :none], do: mode
  defp normalize_prompt_mode(_), do: :full

  # ==========================================================================
  # Repair — 孤立 tool_call 修复
  # ==========================================================================

  defp repair_orphaned_calls(state) do
    orphaned = Repair.find_orphaned_calls(state.messages)

    Enum.reduce(orphaned, state, fn call_id, st ->
      Logger.warning("[Agent] orphaned tool_call: #{call_id}, injecting abort result")
      msg = CMDC.Message.tool_result(call_id, "[Aborted by user]", true)
      State.append_message(st, msg)
    end)
  end

  # ==========================================================================
  # Tool Call Finalization
  # ==========================================================================

  defp merge_arg_buffers(tool_calls, buffers) when map_size(buffers) == 0, do: tool_calls

  defp merge_arg_buffers(tool_calls, buffers) do
    Enum.map(tool_calls, &merge_tc_args(&1, buffers))
  end

  defp merge_tc_args(tc, buffers) do
    block_idx = tc[:block_index]

    case block_idx && Map.get(buffers, block_idx) do
      nil ->
        tc

      iodata ->
        json_str = IO.iodata_to_binary(iodata)

        case Jason.decode(json_str) do
          {:ok, parsed} when is_map(parsed) ->
            Map.update(tc, :arguments, parsed, &Map.merge(&1, parsed))

          _ ->
            Logger.warning(
              "[Agent] failed to parse tool_call args JSON for block_index=#{block_idx}"
            )

            tc
        end
    end
  end

  defp finalize_tool_calls(raw) do
    raw
    |> Enum.map(fn tc ->
      args = tc[:arguments] || %{}

      %{
        call_id: tc[:call_id] || generate_call_id(),
        name: tc[:name] || "",
        arguments: args
      }
    end)
    |> Enum.reject(&(&1.name == ""))
  end

  defp generate_call_id do
    :crypto.strong_rand_bytes(8) |> Base.encode16(case: :lower)
  end

  # ==========================================================================
  # 私有辅助
  # ==========================================================================

  defp stall_timeout, do: {:state_timeout, @stall_ms, :stall_check}

  # ==========================================================================
  # Compactor + before_compact Plugin 事件
  # ==========================================================================

  # 检查是否需要压缩，需要时先 emit {:before_compact, messages} 让 Plugins（如
  # MemoryFlush）有机会提取关键事实到持久记忆，然后执行压缩。
  defp maybe_compact_with_plugins(state) do
    case Compactor.should_compact?(state) do
      true ->
        state = emit_before_compact(state)

        case Compactor.maybe_compact(state) do
          {:compacted, state} -> state
          {:skip, state} -> state
        end

      false ->
        state
    end
  end

  defp emit_before_compact(state) do
    messages = Enum.reverse(state.messages)

    case run_pipeline(state, {:before_compact, messages}) do
      {:ok, state, _result} -> state
      {:abort, state, _reason} -> state
    end
  end

  # ==========================================================================
  # Pipeline 辅助 — 执行并同步状态
  # ==========================================================================

  defp run_pipeline(state, event) do
    ctx = State.to_context(state)

    case Pipeline.run(state.plugins, event, ctx) do
      {:ok, result} ->
        state = State.sync_plugin_states(state, result)
        state = apply_emitted_state_updates(state, result.emitted_events)
        state = apply_pipeline_model_switch(state, result)
        broadcast_emitted_events(state, result.emitted_events)

        case result.action do
          :abort -> {:abort, state, result.halt_reason}
          _ -> {:ok, state, result}
        end

      {:error, reason} ->
        Logger.error("[Agent] pipeline error: #{inspect(reason)}")

        {:ok, state,
         %{
           action: :continue,
           interventions: [],
           emitted_events: [],
           replaced_args: nil,
           model_switch: nil,
           halted_by: nil,
           halt_reason: nil,
           plugin_states: state.plugin_states
         }}
    end
  end

  defp broadcast_emitted_events(_state, []), do: :ok

  defp broadcast_emitted_events(state, events) do
    user_data = state.user_data || %{}

    Enum.each(events, fn
      {name, payload} when is_atom(name) ->
        broadcast(state, {:plugin_event, name, inject_user_data(payload, user_data)})

      {name, a, b} when is_atom(name) ->
        broadcast(state, {:plugin_event, name, {a, b}})

      _other ->
        :ok
    end)
  end

  # `:plugin_event` 自动注入 user_data。
  #
  # 仅对 map payload 生效；Plugin 可在 payload 里塞 `:_no_user_data` 键 opt-out
  # （broadcast 前会 pop 掉，订阅方看不到这个内部 marker）。
  # 已含 `:user_data` 字段时不覆盖（Plugin 可显式提供自定义 user_data 子集）。
  defp inject_user_data(payload, user_data)
       when is_map(payload) and map_size(user_data) > 0 do
    case Map.pop(payload, :_no_user_data, false) do
      {true, rest} ->
        rest

      {_, rest} ->
        if is_map_key(rest, :user_data) do
          rest
        else
          Map.put(rest, :user_data, user_data)
        end
    end
  end

  defp inject_user_data(payload, _user_data) when is_map(payload) do
    # user_data 空时也要剥掉 _no_user_data 防止泄露内部 marker
    case Map.pop(payload, :_no_user_data, false) do
      {_, rest} -> rest
    end
  end

  defp inject_user_data(payload, _user_data), do: payload

  # ==========================================================================
  # Pipeline 辅助 — :before_prompt intervene 注入
  # ==========================================================================

  defp maybe_append_before_prompt_intervention(state, %{action: :intervene} = result) do
    case Pipeline.merged_interventions(result) do
      nil -> state
      prompt -> State.append_message(state, CMDC.Message.user(prompt))
    end
  end

  defp maybe_append_before_prompt_intervention(state, _result), do: state

  defp maybe_broadcast_before_prompt_intervention(state, %{action: :intervene} = result) do
    case Pipeline.merged_interventions(result) do
      nil -> :ok
      prompt -> broadcast(state, {:intervention, prompt})
    end
  end

  defp maybe_broadcast_before_prompt_intervention(_state, _result), do: :ok

  # ==========================================================================
  # Pipeline 辅助 — Plugin 触发的 model 切换
  # ==========================================================================

  defp apply_pipeline_model_switch(state, %{model_switch: nil}), do: state

  defp apply_pipeline_model_switch(state, %{model_switch: new_model})
       when is_binary(new_model) do
    do_switch_model_state(state, new_model, nil)
  end

  defp apply_pipeline_model_switch(state, %{model_switch: {new_model, provider_opts}})
       when is_binary(new_model) and is_list(provider_opts) do
    do_switch_model_state(state, new_model, provider_opts)
  end

  # ==========================================================================
  # Model 切换核心
  # ==========================================================================

  defp do_switch_model(state, new_model, provider_opts)
       when is_binary(new_model) do
    {:keep_state, do_switch_model_state(state, new_model, provider_opts)}
  end

  # 同 model + 无 provider_opts + 非 registry: 协议 → no-op
  # registry: 协议下即使 model 字符串相同也要走 parse 路径，因为外部 register
  # 可能已经更新了对应 profile 的 opts（运行时改写 hook）。
  defp do_switch_model_state(state, new_model, nil)
       when is_binary(new_model) and state.model == new_model do
    if String.starts_with?(new_model, "registry:") do
      apply_parsed_model_switch(state, new_model, nil)
    else
      state
    end
  end

  defp do_switch_model_state(state, new_model, provider_opts)
       when is_binary(new_model) do
    apply_parsed_model_switch(state, new_model, provider_opts)
  end

  defp apply_parsed_model_switch(state, new_model, provider_opts) do
    case parse_model_spec(new_model, provider_opts) do
      {:ok, resolved_model, merged_opts, original_model_spec} ->
        commit_model_switch(state, new_model, resolved_model, merged_opts, original_model_spec)

      {:error, {:registry_profile_missing, name} = reason} ->
        Logger.warning(
          "[Agent] switch_model session=#{state.session_id} requested=#{new_model} " <>
            "rejected: registry profile #{inspect(name)} missing,保持原 model=#{state.model}"
        )

        broadcast(state, {:model_switch_failed, %{requested: new_model, reason: reason}})
        state

      {:error, reason} ->
        Logger.warning(
          "[Agent] switch_model session=#{state.session_id} requested=#{new_model} " <>
            "rejected: #{inspect(reason)},保持原 model=#{state.model}"
        )

        broadcast(state, {:model_switch_failed, %{requested: new_model, reason: reason}})
        state
    end
  end

  defp commit_model_switch(state, new_model, resolved_model, merged_opts, original_model_spec) do
    old_model = state.model
    {base_config, provider_opts_changed?} = maybe_update_provider_opts(state.config, merged_opts)

    new_config =
      if original_model_spec do
        Map.put(base_config, :resolved_provider_model, resolved_model)
      else
        Map.delete(base_config, :resolved_provider_model)
      end

    state = %{state | model: new_model, config: new_config}

    Logger.debug(
      "[Agent] switch_model session=#{state.session_id} from=#{old_model} to=#{new_model} " <>
        "resolved=#{resolved_model} provider_opts_changed?=#{provider_opts_changed?}"
    )

    broadcast(
      state,
      {:model_switched,
       %{from: old_model, to: new_model, provider_opts_changed?: provider_opts_changed?}}
    )

    state
  end

  defp maybe_update_provider_opts(config, nil), do: {config, false}

  defp maybe_update_provider_opts(config, provider_opts) when is_list(provider_opts) do
    current = Map.get(config, :provider_opts, []) || []
    changed? = Keyword.equal?(current, provider_opts) == false
    {Map.put(config, :provider_opts, provider_opts), changed?}
  end

  # ==========================================================================
  # Model 字符串解析 — "registry:profile:model_id" 协议（v0.5.1）
  # ==========================================================================

  # 解析 model 字符串：
  # - "registry:profile_name:provider:model_id" → 查 CMDC.Provider.Registry，
  #   返回 {:ok, "provider:model_id", merged_opts, original_spec}
  # - 其他 → 直接透传，返回 {:ok, model, opts_or_nil, nil}
  #
  # merged_opts 策略：profile.opts 是基础，user_provider_opts(非 nil) 覆盖。
  # 调用方 nil 表示"不更新 opts"，registry 协议下也尊重该语义但仍提供 profile.opts。
  defp parse_model_spec("registry:" <> rest, user_provider_opts) when is_binary(rest) do
    case String.split(rest, ":", parts: 2) do
      [name, resolved_model] when name != "" and resolved_model != "" ->
        case CMDC.Provider.Registry.lookup(name) do
          {:ok, %{opts: profile_opts}} ->
            merged_opts = merge_registry_opts(profile_opts, user_provider_opts)
            {:ok, resolved_model, merged_opts, "registry:" <> rest}

          {:error, :not_found} ->
            {:error, {:registry_profile_missing, name}}
        end

      _ ->
        {:error, {:invalid_registry_model, "registry:" <> rest}}
    end
  end

  defp parse_model_spec(raw_model, user_provider_opts) when is_binary(raw_model) do
    {:ok, raw_model, user_provider_opts, nil}
  end

  defp parse_model_spec(other, _opts), do: {:error, {:invalid_model, other}}

  defp merge_registry_opts(profile_opts, nil), do: profile_opts

  defp merge_registry_opts(profile_opts, user_opts) when is_list(user_opts) do
    Keyword.merge(profile_opts || [], user_opts)
  end

  # ==========================================================================
  # Tool attach / detach 核心
  # ==========================================================================

  defp do_attach_tool(state, tool_module) when is_atom(tool_module) do
    cond do
      not valid_tool?(tool_module) ->
        {{:error, :invalid_tool}, state}

      tool_already_attached?(state.tools, tool_module) ->
        {{:error, :already_attached}, state}

      true ->
        new_state = %{state | tools: state.tools ++ [tool_module]}
        name = tool_module.name()

        Logger.debug(
          "[Agent] attach_tool session=#{state.session_id} tool=#{tool_module} name=#{name}"
        )

        broadcast(new_state, {:tool_attached, name})
        {:ok, new_state}
    end
  end

  defp do_detach_tool(state, tool_name) when is_binary(tool_name) do
    case Enum.split_with(state.tools, &(&1.name() == tool_name)) do
      {[], _} ->
        {{:error, :not_found}, state}

      {[_ | _], remaining} ->
        new_state = %{state | tools: remaining}

        Logger.debug("[Agent] detach_tool session=#{state.session_id} name=#{tool_name}")

        broadcast(new_state, {:tool_detached, tool_name})
        {:ok, new_state}
    end
  end

  defp valid_tool?(module) do
    Code.ensure_loaded?(module) and function_exported?(module, :name, 0) and
      function_exported?(module, :description, 0) and
      function_exported?(module, :parameters, 0) and
      function_exported?(module, :execute, 2)
  end

  # ==========================================================================
  # 批量 attach/detach/replace
  # ==========================================================================

  defp do_attach_tools(state, tool_modules) do
    failures = validate_attach_batch(state.tools, tool_modules)

    if failures == [] do
      names = Enum.map(tool_modules, & &1.name())
      new_state = %{state | tools: state.tools ++ tool_modules}

      Enum.each(names, &broadcast(new_state, {:tool_attached, &1}))
      broadcast(new_state, {:tools_updated, %{attached: names, detached: []}})

      {{:ok, names}, new_state}
    else
      {{:error, {:validation_failed, failures}}, state}
    end
  end

  defp validate_attach_batch(existing_tools, new_modules) do
    existing_names = MapSet.new(existing_tools, & &1.name())

    {failures, _seen_names} =
      Enum.reduce(new_modules, {[], existing_names}, fn mod, {acc, seen} ->
        case validate_attach_one(mod, seen) do
          :ok -> {acc, MapSet.put(seen, safe_tool_name(mod))}
          {:error, reason} -> {[{mod, reason} | acc], seen}
        end
      end)

    Enum.reverse(failures)
  end

  defp validate_attach_one(mod, seen_names) do
    cond do
      not valid_tool?(mod) -> {:error, :invalid_tool}
      MapSet.member?(seen_names, safe_tool_name(mod)) -> {:error, :already_attached}
      true -> :ok
    end
  end

  defp do_detach_tools(state, tool_names) do
    existing_names = MapSet.new(state.tools, & &1.name())

    failures =
      tool_names
      |> Enum.reject(&MapSet.member?(existing_names, &1))
      |> Enum.map(&{&1, :not_found})

    if failures == [] do
      names_set = MapSet.new(tool_names)
      remaining = Enum.reject(state.tools, &MapSet.member?(names_set, &1.name()))
      new_state = %{state | tools: remaining}

      Enum.each(tool_names, &broadcast(new_state, {:tool_detached, &1}))
      broadcast(new_state, {:tools_updated, %{attached: [], detached: tool_names}})

      {{:ok, tool_names}, new_state}
    else
      {{:error, {:validation_failed, failures}}, state}
    end
  end

  defp do_replace_tools(state, new_modules) do
    case validate_replace_batch(new_modules) do
      [] ->
        old_names = MapSet.new(state.tools, & &1.name())
        new_names_set = MapSet.new(new_modules, & &1.name())

        to_attach =
          Enum.reject(new_modules, fn mod ->
            MapSet.member?(old_names, mod.name())
          end)

        to_detach =
          Enum.reject(state.tools, fn mod ->
            MapSet.member?(new_names_set, mod.name())
          end)

        attached_names = Enum.map(to_attach, & &1.name())
        detached_names = Enum.map(to_detach, & &1.name())
        new_state = %{state | tools: new_modules}

        Enum.each(attached_names, &broadcast(new_state, {:tool_attached, &1}))
        Enum.each(detached_names, &broadcast(new_state, {:tool_detached, &1}))

        broadcast(
          new_state,
          {:tools_updated, %{attached: attached_names, detached: detached_names}}
        )

        {{:ok, %{attached: attached_names, detached: detached_names}}, new_state}

      failures ->
        {{:error, {:validation_failed, failures}}, state}
    end
  end

  # replace 时校验：每个 module 必须 valid + 列表内不能同名重复
  defp validate_replace_batch(new_modules) do
    {failures, _seen} =
      Enum.reduce(new_modules, {[], MapSet.new()}, fn mod, {acc, seen} ->
        cond do
          not valid_tool?(mod) ->
            {[{mod, :invalid_tool} | acc], seen}

          MapSet.member?(seen, safe_tool_name(mod)) ->
            {[{mod, :duplicate_in_list} | acc], seen}

          true ->
            {acc, MapSet.put(seen, safe_tool_name(mod))}
        end
      end)

    Enum.reverse(failures)
  end

  defp tool_already_attached?(tools, new_tool) do
    new_name = safe_tool_name(new_tool)
    new_name != nil and Enum.any?(tools, &(&1.name() == new_name))
  end

  defp safe_tool_name(module) do
    module.name()
  rescue
    _ -> nil
  end

  defp apply_emitted_state_updates(state, []), do: state

  defp apply_emitted_state_updates(state, events) do
    Enum.reduce(events, state, fn
      {:memory_contents, contents}, s when is_map(contents) ->
        %{s | memory_contents: contents}

      {:synthetic_tool_results, synthetics}, s when is_list(synthetics) ->
        apply_synthetic_tool_results(s, synthetics)

      {:update_message_metadata, updates}, s when is_list(updates) ->
        apply_message_metadata_updates(s, updates)

      {:update_system_context, key, nil}, s when is_atom(key) ->
        new_sections = Map.delete(s.dynamic_context_sections || %{}, key)
        %{s | dynamic_context_sections: new_sections}

      {:update_system_context, key, text}, s when is_atom(key) and is_binary(text) ->
        new_sections = Map.put(s.dynamic_context_sections || %{}, key, text)
        %{s | dynamic_context_sections: new_sections}

      _other, s ->
        s
    end)
  end

  defp apply_message_metadata_updates(state, updates) do
    update_map = Map.new(updates, fn {id, meta} -> {id, meta} end)

    updated_messages =
      Enum.map(state.messages, fn msg ->
        case Map.get(update_map, msg.id) do
          nil -> msg
          new_meta -> %{msg | metadata: Map.merge(msg.metadata || %{}, new_meta)}
        end
      end)

    %{state | messages: updated_messages}
  end

  defp apply_synthetic_tool_results(state, synthetics) do
    existing_result_ids =
      state.messages
      |> Enum.filter(&(&1.role == :tool_result))
      |> Enum.map(& &1.call_id)
      |> Enum.filter(&is_binary/1)
      |> MapSet.new()

    new_messages =
      synthetics
      |> Enum.reject(fn {call_id, _, _} -> MapSet.member?(existing_result_ids, call_id) end)
      |> Enum.map(fn {call_id, content, is_error} ->
        CMDC.Message.tool_result(call_id, content, is_error)
      end)

    State.append_messages(state, new_messages)
  end
end