cmdc

defmodule CMDC do
  @moduledoc """
  CMDC — Elixir Agent Kernel 公共 API 入口。

  提供从零创建 Agent 会话到收集回复的完整链路。
  内部通过 `CMDC.EventBus` 实现异步解耦。

  ## 公开 API 矩阵

  所有公开 API 同时接受 **session pid** 和 **session_id 字符串**。
  字符串入参通过 `CMDC.SessionRegistry`（`:via` 注册表）反查 pid。

  | 能力 | API | 入参类型 | 备注 |
  |---|---|---|---|
  | 创建会话 | `create_agent/1` | keyword \\| `Options.t()` | 返回 `{:ok, pid}` |
  | 发送 prompt | `prompt/2` | pid \\| string | 立刻或入队 |
  | 收集回复 | `collect_reply/2` | pid \\| string | 阻塞等待 |
  | 中段干预 | `steer/2` | pid \\| string | 软中断 |
  | 中止 | `abort/1` | pid \\| string | Agent → idle |
  | 停止 | `stop/2` | pid \\| string | 关闭 Supervisor |
  | 订阅事件 | `subscribe/1` | pid \\| string | 当前进程接收事件 |
  | 取消订阅 | `unsubscribe/1` | pid \\| string | — |
  | 取 session_id | `session_id/1` | pid \\| string | 字符串入参直接返回 |
  | 取 Agent pid | `agent_pid/1` | pid \\| string | — |
  | 状态快照 | `status/1` | pid \\| string | `%{state, turns, ...}` |
  | HITL 批准 | `approve/3` | pid \\| string | 默认 auto_resume |
  | HITL 拒绝 | `reject/3` | pid \\| string | 默认不续 turn |
  | 用户回应 | `user_respond/3` | pid \\| string | 应答 AskUser 工具 |

  字符串入参对应的 session 进程必须存活；不存在时 raise `ArgumentError`。

  ## 快速开始

      # 1. 创建 Agent 会话
      {:ok, session} = CMDC.create_agent(
        session_id: "my-session",
        model: "anthropic:claude-sonnet-4-5",
        tools: [CMDC.Tool.ReadFile, CMDC.Tool.Shell],
        system_prompt: "你是一个专业的代码助手。",
        working_dir: "/my/project"
      )

      # 2. 发送 prompt（pid 或 session_id 字符串都行）
      CMDC.prompt(session, "帮我列出当前目录的文件")
      CMDC.prompt("my-session", "帮我列出当前目录的文件")

      # 3. 等待回复（阻塞直到 Agent 回到 idle）
      {:ok, reply} = CMDC.collect_reply(session, timeout: 30_000)
      IO.puts(reply)

      # 4. 停止会话
      CMDC.stop(session)

  ## 流式接收

      CMDC.subscribe(session)

      CMDC.prompt(session, "写一首诗")

      receive do
        {:cmdc_event, _sid, {:message_delta, %{delta: chunk}}} -> IO.write(chunk)
        {:cmdc_event, _sid, {:agent_end, _msgs, _usage}} -> :done
      end

  ## HITL 审批

      CMDC.subscribe(session)
      CMDC.prompt(session, "删除所有日志文件")

      receive do
        {:cmdc_event, _sid, {:approval_required, %{id: ref, tool: tool, args: args}}} ->
          IO.inspect({tool, args}, label: "批准工具调用？")
          CMDC.approve(session, ref)   # 或 CMDC.reject(session, ref)
      end

  ## AskUser 工具回应

      receive do
        {:cmdc_event, sid, {:ask_user, ^sid, question, options, ref}} ->
          # 通过 string session_id 也可以
          CMDC.user_respond(sid, ref, "我的答案")
      end
  """

  alias CMDC.{Agent, Checkpoint, EventBus, SessionServer}
  alias CMDC.Checkpoint.Snapshot

  # ==========================================================================
  # 类型
  # ==========================================================================

  @typedoc """
  会话引用 — 既支持 SessionServer pid，也支持 session_id 字符串。

  字符串入参会通过 `CMDC.SessionRegistry` 反查 pid;
  当 session 不存在或已死亡时，所有公开 API 会 raise `ArgumentError`。
  """
  @type session :: pid() | String.t()

  @type create_opts :: keyword() | CMDC.Options.t()

  # ==========================================================================
  # create_agent/1
  # ==========================================================================

  @doc """
  创建并启动一个 Agent 会话。

  支持三种调用方式：

  **1. keyword list（推荐）**

      {:ok, session} = CMDC.create_agent(
        model: "anthropic:claude-sonnet-4-5",
        tools: [CMDC.Tool.Shell],
        working_dir: "/project"
      )

  **2. `CMDC.Options.t()` struct**

      opts = CMDC.Options.new!(model: "anthropic:claude-sonnet-4-5")
      {:ok, session} = CMDC.create_agent(opts)

  **3. Blueprint 模块**

      {:ok, session} = CMDC.create_agent(blueprint: MyApp.CodingAgent)

  ## 必填选项

  - `:model` — 模型标识符，如 `"anthropic:claude-sonnet-4-5"`（Blueprint 启动时可省略）

  ## 可选选项

  - `:session_id` — 会话唯一标识符，默认自动生成
  - `:blueprint` — Blueprint 模块或 `CMDC.Blueprint.t()` struct
  - `:working_dir` — 工作目录，默认当前目录
  - `:system_prompt` — 系统提示词
  - `:tools` — Tool 模块列表
  - `:plugins` — Plugin 列表
  - `:config` — `CMDC.Config.t()` struct
  - `:provider_opts` — 传给 req_llm 的选项，如 `[api_key: "sk-..."]`
  - `:max_turns` — 最大轮数，默认 100
  - `:user_data` — 业务上下文 map，原样写入 `ctx.user_data`，SubAgent 自动继承
  - `:messages` — 历史消息列表（`[CMDC.Message.t()]`），按时间顺序传入；
    Agent 启动时直接写入 `state.messages`，跳过让 LLM 重读历史的轮次,
    适用于 idle 休眠唤醒 / 跨进程恢复

  ## 历史消息恢复

      history = [
        CMDC.Message.user("帮我审核这段代码"),
        CMDC.Message.assistant("好的，请贴上来", [], [])
      ]

      {:ok, session} = CMDC.create_agent(
        model: "anthropic:claude-sonnet-4-5",
        messages: history
      )

      CMDC.prompt(session, "def hello, do: :world")
      # → 续接历史，不会让 LLM 重新读 history

  返回 `{:ok, session_pid}` 或 `{:error, reason}`。
  """
  @spec create_agent(create_opts()) :: {:ok, session()} | {:error, term()}
  def create_agent(%CMDC.Options{} = opts) do
    opts
    |> options_to_session_opts()
    |> start_session_with_registry_check()
  end

  def create_agent(opts) when is_list(opts) do
    opts
    |> ensure_session_id()
    |> start_session_with_registry_check()
  end

  # v0.5.1：model 是 "registry:profile:model_id" 协议时，在 SessionServer.start_link
  # 前**预检** Registry。命中则放行，未命中直接返 {:error, {:registry_profile_missing, name}}，
  # 避免 Supervisor restart 3 次后才暴露错误。Agent.init/1 内部仍保留同样的解析作为兜底。
  defp start_session_with_registry_check(session_opts) do
    case preflight_registry_model(session_opts) do
      :ok -> SessionServer.start_link(session_opts)
      {:error, _} = err -> err
    end
  end

  defp preflight_registry_model(session_opts) do
    case Keyword.get(session_opts, :model) do
      "registry:" <> rest when is_binary(rest) -> resolve_registry_prefix(rest)
      _ -> :ok
    end
  end

  defp resolve_registry_prefix(rest) do
    case String.split(rest, ":", parts: 2) do
      [name, _model_id] when name != "" -> lookup_registry_profile(name)
      _ -> {:error, {:invalid_registry_model, "registry:" <> rest}}
    end
  end

  defp lookup_registry_profile(name) do
    case CMDC.Provider.Registry.lookup(name) do
      {:ok, _profile} -> :ok
      {:error, :not_found} -> {:error, {:registry_profile_missing, name}}
      # v0.6+ resolver_fn 失败：把错误暴露到 create_agent 调用方，避免 Supervisor restart 3 次
      {:error, {:resolver_failed, _} = err} -> {:error, {:registry_resolver_failed, name, err}}
      {:error, :resolver_invalid_return} -> {:error, {:registry_resolver_invalid, name}}
    end
  end

  # ==========================================================================
  # prompt/2
  # ==========================================================================

  @doc """
  向 Agent 发送用户 prompt。

  - 若 Agent 处于 idle，立即开始处理，返回 `%{queued: false}`
  - 若 Agent 正忙，消息入队，处理完当前任务后自动处理，返回 `%{queued: true}`

  ## 示例

      %{queued: false} = CMDC.prompt(session, "帮我写一个 Elixir GenServer")
  """
  @spec prompt(session(), String.t()) ::
          %{queued: boolean()} | {:error, :invalid_session | :not_alive}
  def prompt(session, text) when is_binary(text) do
    with_agent(session, &Agent.prompt(&1, text))
  end

  # ==========================================================================
  # collect_reply/2
  # ==========================================================================

  @doc """
  等待并收集 Agent 的最终文本回复。

  订阅当前会话的 EventBus，等待 `{:agent_end, messages, _usage}` 事件，
  从最后一条 assistant 消息中提取文本内容返回。

  ## 选项

  - `:timeout` — 等待超时时间（毫秒），默认 60_000（60 秒）

  ## 返回

  - `{:ok, text}` — 成功收到回复文本
  - `{:error, :timeout}` — 超时
  - `{:error, reason}` — Agent 出错或被中止

  ## 示例

      CMDC.prompt(session, "你好")
      {:ok, reply} = CMDC.collect_reply(session, timeout: 30_000)
      IO.puts(reply)
  """
  @spec collect_reply(session(), keyword()) ::
          {:ok, String.t()} | {:error, :timeout | :invalid_session | term()}
  def collect_reply(session, opts \\ []) do
    timeout = Keyword.get(opts, :timeout, 60_000)

    with_sid(session, fn sid ->
      EventBus.subscribe(sid)
      result = await_reply(sid, timeout)
      EventBus.unsubscribe(sid)
      result
    end)
  end

  # ==========================================================================
  # stop/2
  # ==========================================================================

  @doc """
  停止 Agent 会话，清理所有资源。

  优雅停止 `SessionServer` 监督树（级联停止 SubAgent.Supervisor 和主 Agent）。

  ## 选项

  - `:reason` — 停止原因，默认 `:normal`
  - `:timeout` — 等待停止超时（毫秒），默认 5_000

  ## 示例

      :ok = CMDC.stop(session)
  """
  @spec stop(session(), keyword()) :: :ok | {:error, :invalid_session | :not_alive}
  def stop(session, opts \\ []) do
    reason = Keyword.get(opts, :reason, :normal)
    timeout = Keyword.get(opts, :timeout, 5_000)

    case normalize_to_pid(session) do
      {:ok, pid} -> Supervisor.stop(pid, reason, timeout)
      {:error, _} = err -> err
    end
  end

  # ==========================================================================
  # subscribe / unsubscribe
  # ==========================================================================

  @doc """
  订阅当前进程接收指定 Agent 会话的所有事件。

  订阅后，当前进程将收到 `{:cmdc_event, session_id, event}` 格式的消息。

  常见事件：
  - `{:stream_chunk, session_id, chunk}` — 流式 token
  - `{:tool_start, session_id, tool_name, args}` — 工具开始
  - `{:tool_end, session_id, tool_name, result}` — 工具完成
  - `{:agent_end, messages, %CMDC.TokenUsage{}}` — 回复完成
  - `{:approval_required, session_id, tool_name, args, ref}` — 等待审批

  ## 选项

  - `:since :: non_neg_integer()` — 从该 index 之后的事件开始 replay。
    要求 Agent 启动时已开启 ring buffer（`Options.event_buffer_size > 0`）。
    用于 WebSocket / Channel 重连补帧，避免断网那一秒丢的 stream_chunk / tool_end 永远丢失。
    可通过 `CMDC.EventBus.last_index/1` 获取当前最新 index 作为下次重连的 `:since`。

  ## 示例

      CMDC.subscribe(session)
      CMDC.prompt(session, "hello")

      receive do
        {:cmdc_event, _sid, {:agent_end, _msgs, _usage}} -> IO.puts("done")
      end

      # 重连补帧（需 Options.event_buffer_size > 0）
      CMDC.subscribe(session, since: last_seen_index)
  """
  @spec subscribe(session(), keyword()) ::
          {:ok, pid()} | {:error, :invalid_session | term()}
  def subscribe(session, opts \\ []) do
    with_sid(session, &EventBus.subscribe(&1, opts))
  end

  @doc "取消当前进程对指定 Agent 会话的事件订阅。"
  @spec unsubscribe(session()) :: :ok | {:error, :invalid_session}
  def unsubscribe(session) do
    with_sid(session, &EventBus.unsubscribe/1)
  end

  # ==========================================================================
  # monitor / demonitor
  # ==========================================================================

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

  与 `Process.monitor/1` 的区别：CMDC.monitor 返回的 `reference()` 是 CMDC 自维护的,
  Agent 进程退出时观察者收到**结构化 reason**：

      {:cmdc_down, ref, session_id, structured_reason}

  ## Structured Reason 表

  - `:normal` — 正常退出（supervisor shutdown / `CMDC.stop/2` 明确调用）
  - `{:exception, exception}` — 未捕获异常（含 RuntimeError / ArgumentError 等）
  - `:max_turns_exceeded` — 超出 `Options.max_turns`（预留）
  - `:provider_timeout` — Provider.stream 超时（预留）
  - `{:plugin_aborted, plugin_name, why}` — Plugin 返回 `:abort` 导致 Agent 终止
    （预留；当前 plugin abort 仅将 Agent 回到 idle 不退出进程）
  - 其他原子/元组 — 原样透传（未知 reason）

  ## 示例

      ref = CMDC.monitor(session)

      receive do
        {:cmdc_down, ^ref, sid, :normal} ->
          IO.puts("Agent \#{sid} exited normally")

        {:cmdc_down, ^ref, sid, {:exception, e}} ->
          IO.puts("Agent \#{sid} crashed: \#{Exception.message(e)}")
      end

  ## 与 `Process.monitor/1` 区别

  - `Process.monitor` 只给 `{:DOWN, ref, :process, pid, reason}`，reason 是 raw 的
    `:shutdown` / `{:shutdown, _}` / 异常元组，需要调用方自行解析
  - `CMDC.monitor` 返回 CMDC 自己的 ref，reason 已结构化，可直接做 pattern match
  """
  @spec monitor(session()) :: reference() | {:error, :invalid_session | :not_alive}
  def monitor(session) do
    with_agent(session, &CMDC.Agent.monitor/1)
  end

  @doc "取消通过 `monitor/1` 登记的 CMDC 崩溃监控。"
  @spec demonitor(session(), reference()) :: :ok | {:error, :invalid_session | :not_alive}
  def demonitor(session, ref) when is_reference(ref) do
    with_agent(session, &CMDC.Agent.demonitor(&1, ref))
  end

  # ==========================================================================
  # abort / approve / reject
  # ==========================================================================

  @doc """
  中止当前 Agent 运行中的任务，Agent 回到 `:idle` 状态。

  ## 选项

  - `:reason :: term()` — 中止原因。如设置则事件为 `{:agent_abort, reason}`,
    否则为裸 `:agent_abort`（向后兼容）。
  - `:clear_queue :: boolean()` — 是否清空 prompt pending queue，默认 `true`;
    清空时为每条被丢弃的 user prompt emit `{:prompt_dropped, text}`。
  - `:kill_tools :: :all | :killable | :none` — 工具任务清理策略，默认 `:killable`:
    - `:all` — 杀全部 in-flight 工具
    - `:killable` — 只杀非 immune 工具（与 Steering 一致）
    - `:none` — 不杀，让工具自然完成
    每杀一个工具 emit `{:tool_killed, %{name, call_id, reason}}`。

  ## 状态行为

  | 调用时状态 | `:agent_abort` 是否发出 | 工具行为（默认 `:killable`） |
  |---|---|---|
  | `:idle` | ✅ no-op，但仍发出便于订阅方对账 | — |
  | `:running` | ✅ | cancel stream task |
  | `:streaming` | ✅ | cancel stream task |
  | `:executing_tools` | ✅ | 杀非 immune 工具，immune 工具继续 |

  事件保证：所有状态下 `:agent_abort` 在 100ms 内到达订阅方
  （BEAM 调度延迟，本地 EventBus 通常 <10ms）。多次 abort 幂等。
  """
  @spec abort(session(), keyword()) :: :ok | {:error, :invalid_session | :not_alive}
  def abort(session, opts \\ []) do
    with_agent(session, &Agent.abort(&1, opts))
  end

  # ==========================================================================
  # switch_model/2 — 运行期切换 LLM 模型
  # ==========================================================================

  @doc """
  运行期切换 LLM 模型。

  下一次 LLM 请求立即用新模型，不重启 Agent，**messages / tools / plugin 状态全部保留**。

  ## 行为

  - **idle 状态**：立即更新 `state.model`，emit `{:model_switched, %{from, to}}`,
    后续 `prompt/2` 用新模型
  - **running / streaming**：当前请求继续用旧模型完成；`state.model` 已更新；
    `{:model_switched, ...}` 立即发出；下一个 turn 用新模型
  - **executing_tools**：同上，工具继续执行，下一轮 LLM 调用用新模型
  - **同模型切换**（new_model == 当前 model）：no-op，**不发事件**

  ## 触发方式

  ### 方式一：调用方主动切换

      :ok = CMDC.switch_model(session, "openai:gpt-4o-mini")

  ### 方式二：Plugin Action

  Plugin 在 `:on_tool_error` / `:after_response` / `:after_tool_batch` 等钩子中返回:

      def handle_event({:on_tool_error, _name, _call_id, _err, _attempt}, state, _ctx) do
        {:switch_model, "openai:gpt-4o-mini", state}
      end

  Pipeline 收集后由 Agent 自动执行（无需调用方介入）。

  ## 兼容性警告

  - **system_prompt 兼容性**：不同 Provider 对 system prompt 的支持差异由 Provider 层处理
  - **上下文窗口差异**：从 200k 模型切到 8k 模型时，调用方需自行评估是否要先压缩上下文
  - **tool_calling schema 差异**：由 Provider 适配（OpenAI / Anthropic / Gemini 各自不同）

  ## 事件

      {:model_switched, %{from: "anthropic:claude-sonnet-4-5", to: "openai:gpt-4o-mini"}}
  """
  @spec switch_model(session(), CMDC.Provider.model()) ::
          :ok | {:error, :invalid_session | :not_alive}
  def switch_model(session, new_model) when is_binary(new_model) do
    with_agent(session, &Agent.switch_model(&1, new_model))
  end

  @doc """
  运行期切换 LLM 模型并同步替换 provider 参数（base_url / api_key / timeout 等）。

  ## 选项

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

  事件 `{:model_switched, %{from, to, provider_opts_changed?}}` 含 `provider_opts_changed?`
  字段，便于订阅方判断是否需要刷新连接池。

  ## 示例

      :ok = CMDC.switch_model(session, "openai:gpt-4o-mini",
              provider_opts: [base_url: "https://api.custom.com/v1", api_key: "sk-..."])
  """
  @spec switch_model(session(), CMDC.Provider.model(), keyword()) ::
          :ok | {:error, :invalid_session | :not_alive}
  def switch_model(session, new_model, opts)
      when is_binary(new_model) and is_list(opts) do
    with_agent(session, &Agent.switch_model(&1, new_model, opts))
  end

  # ==========================================================================
  # attach_tool/2 + detach_tool/2 — 运行期挂载/卸载工具
  # ==========================================================================

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

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

  ## 返回

  - `:ok` — 挂载成功，emit `{:tool_attached, name}`
  - `{:error, :already_attached}` — 已存在同名 tool（按 `tool.name()` 比较）
  - `{:error, :invalid_tool}` — 模块未实现 `CMDC.Tool` behaviour 必需回调

  ## 示例

      :ok = CMDC.attach_tool(session, MyApp.Tools.GitHubMCP)
      # → 下一次 prompt 时 LLM 即可看到该 tool 的 schema 并调用

  ## 适用场景

  - **MCP 热加载**：用户对话中途说"装个 GitHub MCP"，无需重启 Agent
  - **Skill 进化**：升级一个 Skill 不必重启会话
  - **基于上下文动态扩展**：根据 LLM 当前任务挂载相关工具
  """
  @spec attach_tool(session(), module()) ::
          :ok | {:error, :already_attached | :invalid_tool | :invalid_session | :not_alive}
  def attach_tool(session, tool_module) when is_atom(tool_module) do
    with_agent(session, &Agent.attach_tool(&1, tool_module))
  end

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

  立即从 `state.tools` 移除（按 `tool.name()` 字符串匹配），
  **下一次** LLM 请求生效。已 in-flight 的 tool 调用**不受影响**（继续执行完）。

  ## 返回

  - `:ok` — 卸载成功，emit `{:tool_detached, name}`
  - `{:error, :not_found}` — 不存在同名 tool

  ## 示例

      :ok = CMDC.detach_tool(session, "shell")

  ## 注意

  若 LLM 在 detach 后仍调用该 tool（已在 streaming 中或下一轮），
  Agent 会发 `{:tool_call_unknown, name, call_id}` 事件，
  并在工具结果中注入 `{:error, "tool not found"}` 让 LLM 自我纠正。
  """
  @spec detach_tool(session(), String.t()) ::
          :ok | {:error, :not_found | :invalid_session | :not_alive}
  def detach_tool(session, tool_name) when is_binary(tool_name) do
    with_agent(session, &Agent.detach_tool(&1, tool_name))
  end

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

  dry-run 全部 validate 通过才一次性挂载；任何一个失败 → 全回滚。
  典型场景：用户启用 MCP Server 时一次性挂上 N 个工具。

  返回 `{:ok, [name, ...]}` 或 `{:error, {:validation_failed, [{module, reason}, ...]}}`,
  reason: `:invalid_tool | :already_attached`。

  emit 单个 `{:tool_attached, name}` × N + 汇总 `{:tools_updated, %{attached, detached: []}}`。
  """
  @spec attach_tools(session(), [module()]) ::
          {:ok, [String.t()]}
          | {:error, {:validation_failed, [{module(), atom()}]}}
          | {:error, :invalid_session | :not_alive}
  def attach_tools(session, tool_modules) when is_list(tool_modules) do
    with_agent(session, &Agent.attach_tools(&1, tool_modules))
  end

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

  返回 `{:ok, [name, ...]}` 或 `{:error, {:validation_failed, [{name, :not_found}, ...]}}`。
  emit 单个 `{:tool_detached, name}` × N + 汇总 `{:tools_updated, %{attached: [], detached}}`。
  """
  @spec detach_tools(session(), [String.t()]) ::
          {:ok, [String.t()]}
          | {:error, {:validation_failed, [{String.t(), atom()}]}}
          | {:error, :invalid_session | :not_alive}
  def detach_tools(session, tool_names) when is_list(tool_names) do
    with_agent(session, &Agent.detach_tools(&1, tool_names))
  end

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

  自动 diff：保留已存在的（按 `tool.name()`）、attach 新的、detach 缺失的。
  全部 validate 失败 → 全回滚。典型场景：MCP Server 重启 / 配置变更。

  返回 `{:ok, %{attached: [name, ...], detached: [name, ...]}}` 或
  `{:error, {:validation_failed, [{module, :invalid_tool | :duplicate_in_list}, ...]}}`。

  emit 单个 `{:tool_attached/detached, name}` × N + 汇总 `{:tools_updated, %{attached, detached}}`。
  """
  @spec replace_tools(session(), [module()]) ::
          {:ok, %{attached: [String.t()], detached: [String.t()]}}
          | {:error, {:validation_failed, [{module(), atom()}]}}
          | {:error, :invalid_session | :not_alive}
  def replace_tools(session, tool_modules) when is_list(tool_modules) do
    with_agent(session, &Agent.replace_tools(&1, tool_modules))
  end

  # ==========================================================================
  # steer/2 — 中段软中断
  # ==========================================================================

  @doc """
  中段软中断（Steering）：在 Agent 执行中段追加新指引，**不需要先 abort 再 prompt**。

  详见 `guides/cookbook.md` 中的「中段干预（Steering）」配方。

  ## 行为

  - `idle` 状态：等同 `prompt/2`，立刻进入新 turn
  - `running` / `streaming` / `executing_tools`：入 `:steering_queue`，下个 turn 间隙合并注入
  - `executing_tools` 同时杀掉 killable in-flight tool（白名单豁免见 `Options.interrupt_immune_tools`）

  ## 返回

  - `{:ok, ref}` — 已入队（不等同已生效；生效信号通过 `:steering_applied` 事件广播）
  - `{:error, :queue_full}` — Steering queue 已满（默认 3，可由 `Options.max_steering_queue` 配置）
  - `{:error, :rejected}` — Plugin 拦截 `:before_steering` 返回 `:abort`
  - `{:error, :invalid_text}` — 入参非合法字符串

  ## 示例

      CMDC.subscribe(session)
      CMDC.prompt(session, "搜索 Elixir gen_statem 教程，分析每个并整理成文档")

      # 一段时间后用户改主意
      {:ok, _ref} = CMDC.steer(session, "改成只看官方文档，不要第三方教程")

      receive do
        {:cmdc_event, _sid, {:steering_applied, %{count: n}}} ->
          IO.puts("已注入 \#{n} 条 steering")
      end
  """
  @spec steer(session(), String.t()) ::
          {:ok, reference()}
          | {:error, :queue_full | :rejected | :invalid_text | :invalid_session | :not_alive}
  def steer(session, text) when is_binary(text) and byte_size(text) > 0 do
    ref = make_ref()

    case with_agent(session, &Agent.steer(&1, ref, text)) do
      :ok -> {:ok, ref}
      {:error, _} = error -> error
    end
  end

  def steer(_session, _text), do: {:error, :invalid_text}

  # ==========================================================================
  # update_plugin_opts/3 — 运行期 Plugin opts 热更（v0.6+）
  # ==========================================================================

  @doc """
  运行期热更 Plugin opts（v0.6+）。

  长会话中无需停 Agent 重启即可调整已注册 Plugin 的运行时参数。新 opts 立即入队
  `state.pending_plugin_opts`，Agent 进入 `idle` 状态时统一 flush；`running` /
  `streaming` / `executing_tools` 期间走 `:postpone` 推迟，到 idle 后由
  gen_statem 自动重投递。

  ## 工作流程

  1. cast 立即返回 `:ok`，不阻塞
  2. 同一 plugin 多次更新 LWW 覆盖（后写覆盖前写）
  3. 进入 idle 后按入队顺序 flush
  4. 每条 flush 先走 `:before_plugin_opts_update` Plugin hook
     （其他 Plugin 可 `:abort` / `:skip` 拦截，典型用途：SecurityGuard 校验合法性）
  5. 通过 hook 后调用 plugin 的 `on_config_update/2`（若实现）或默认替换语义
  6. 单条 flush 完成 emit `{:plugin_opts_updated, %{plugin, success?, error, mode}}`

  ## 默认替换语义（未实现 `on_config_update/2`）

  - `new_opts` 是 keyword list 且老 plugin state 是 map → 自动 `Map.merge`
  - 其他情况 → `new_opts` 直接替换为新 state

  详见 `CMDC.Plugin` moduledoc 的「运行期热更 Plugin opts」段。

  ## 示例

      :ok = CMDC.update_plugin_opts(session, MyApp.Plugins.CostGuard,
              max_cost_usd: 10.0, soft_warning_pct: 0.8)

      # 订阅结果
      receive do
        {:cmdc_event, _sid,
         {:plugin_opts_updated, %{plugin: MyApp.Plugins.CostGuard, success?: true}}} ->
          :ok
      end

  ## 返回

  - `:ok` — cast 已入队（不等于已生效；生效信号见 `:plugin_opts_updated` 事件）
  - `{:error, :invalid_session | :not_alive}` — 会话不存在或 Agent 已退出
  """
  @spec update_plugin_opts(session(), module(), keyword() | term()) ::
          :ok | {:error, :invalid_session | :not_alive}
  def update_plugin_opts(session, plugin_module, new_opts) when is_atom(plugin_module) do
    with_agent(session, &Agent.update_plugin_opts(&1, plugin_module, new_opts))
  end

  @doc """
  批准一个待审批的工具调用（HITL 审批流）。

  `approval_id` 来自 `{:approval_required, approval_map}` 事件中的 `approval_map.id`。

  ## 选项

  - `:auto_resume` — boolean，默认 `true`。Agent 处于 idle 时（被 `block_tool` 拦下后回到 idle）,
    审批通过后自动开新 turn 让 LLM 重试，**无需调用方再 `prompt/2`**。
    设为 `false` 保持旧版"只放行不续 turn"行为。

  自动续接成功时会广播 `{:agent_resumed, %{trigger: :tool_approved, approval_id: id}}`,
  上层可订阅事件确认 Agent 已重新进入 running 状态。

  ## 示例

      receive do
        {:cmdc_event, _sid, {:approval_required, %{id: aid}}} ->
          :ok = CMDC.approve(session, aid)
      end

      receive do
        {:cmdc_event, _sid, {:agent_resumed, %{approval_id: ^aid}}} ->
          IO.puts("Agent 已自动续 turn")
      end

  关闭自动续（保留旧行为，需要自行 prompt）：

      CMDC.approve(session, aid, auto_resume: false)
      CMDC.prompt(session, "继续刚才被拦截的操作")
  """
  @spec approve(session(), String.t(), keyword()) ::
          :ok | {:error, :invalid_session | :not_alive}
  def approve(session, approval_id, opts \\ []) when is_binary(approval_id) and is_list(opts) do
    with_agent(session, &Agent.approve(&1, approval_id, opts))
  end

  @doc """
  拒绝一个待审批的工具调用（HITL 审批流）。

  ## 选项

  - `:auto_resume` — boolean，默认 `false`。reject 通常希望让 Agent 保持 idle 等待新 prompt,
    被拒命令的 ToolMessage 已记录为 is_error，下次用户 prompt 时 LLM 自然能感知拒绝结果。

    若希望 reject 后 Agent 立刻重新规划（让 LLM 走"被拒后换个方案"分支），传 `auto_resume: true`,
    会广播 `{:agent_resumed, %{trigger: :tool_rejected, approval_id: id}}` 并自动续 turn。

  ## 示例

      :ok = CMDC.reject(session, approval_id)
      :ok = CMDC.reject(session, approval_id, auto_resume: true)
  """
  @spec reject(session(), String.t(), keyword()) ::
          :ok | {:error, :invalid_session | :not_alive}
  def reject(session, approval_id, opts \\ []) when is_binary(approval_id) and is_list(opts) do
    with_agent(session, &Agent.reject(&1, approval_id, opts))
  end

  # ==========================================================================
  # user_respond/3
  # ==========================================================================

  @doc """
  应答 `CMDC.Tool.AskUser` 工具的等待请求（HITL 输入流）。

  AskUser 工具会广播 `{:ask_user, session_id, question, options, ref}` 并阻塞等待
  `{:user_responded, session_id, ref, response}` 事件。本函数提供与 `approve/3` /
  `reject/3` 对称的公开 API，避免上层直接调用 `EventBus.broadcast/2`。

  ## 参数

  - `session` — pid 或 session_id 字符串
  - `ref` — `:ask_user` 事件中的 `ref`（字符串）
  - `response` — 任意可序列化的应答（字符串、字符串列表、map）

  ## 示例

      receive do
        {:cmdc_event, sid, {:ask_user, ^sid, _question, _options, ref}} ->
          CMDC.user_respond(sid, ref, "我的回答")
      end
  """
  @spec user_respond(session(), String.t(), term()) ::
          :ok | {:error, :invalid_session | :not_alive}
  def user_respond(session, ref, response) when is_binary(ref) do
    with_sid(session, fn sid ->
      EventBus.broadcast(sid, {:user_responded, sid, ref, response})
      # 让前端在统一的 :agent_resumed 监听点感知用户响应
      EventBus.broadcast(sid, {:agent_resumed, %{trigger: :user_respond, ref: ref}})
      :ok
    end)
  end

  # ==========================================================================
  # status / agent_pid / session_id
  # ==========================================================================

  @doc """
  获取 Agent 当前状态快照。

  返回：

      %{
        state:        :idle | :running | :streaming | :executing_tools,
        session_id:   String.t(),
        turns:        non_neg_integer(),
        cost_usd:     float(),
        uptime_ms:    non_neg_integer(),
        timestamp_ms: integer()
      }
  """
  @spec status(session()) :: map() | {:error, :invalid_session | :not_alive}
  def status(session) do
    with_agent(session, &Agent.status/1)
  end

  @doc """
  获取会话的完整消息列表（按时间顺序，含 system / user / assistant / tool_result）。

  适用场景：
  - 切换模型前查看历史
  - 持久化对话用于断点恢复
  - 审计日志
  """
  @spec messages(session()) :: [CMDC.Message.t()] | {:error, :invalid_session | :not_alive}
  def messages(session) do
    with_agent(session, &Agent.messages/1)
  end

  @doc "返回会话对应的主 Agent pid。"
  @spec agent_pid(session()) :: pid() | nil | {:error, :invalid_session | :not_alive}
  def agent_pid(session) do
    with_agent(session, & &1)
  end

  @doc """
  返回会话的 session_id 字符串。

  字符串入参直接返回（不查注册表，便于纯 string 流水线）。
  pid 入参时会查询 Agent 状态获取 session_id。
  """
  @spec session_id(session()) :: String.t() | {:error, :invalid_session | :not_alive}
  def session_id(session) when is_binary(session), do: session

  def session_id(session) when is_pid(session) do
    with_agent(session, &(&1 |> Agent.status() |> Map.fetch!(:session_id)))
  end

  def session_id(_other), do: {:error, :invalid_session}

  # ==========================================================================
  # checkpoint! / resume_session! — 会话快照 + 跨进程恢复（v0.5）
  # ==========================================================================

  @doc """
  抓取当前会话的状态快照并持久化到 backend。

  适用场景：BEAM 节点滚动重启、跨设备恢复、长会话定期存档、HITL 审批前留底。
  返回的 `Snapshot.t()` 含 `checkpoint_id` / `saved_at`，可后续传给 `resume_session!/2`。

  ## 选项

  - `:backend` — 覆盖默认 backend（`Application.get_env(:cmdc, :checkpoint_backend, ETS)`）;
    生产环境推荐用 `cmdc_memory_pg` 子库提供的 `CMDCMemoryPg.CheckpointBackend`
  - `:label` — 可选标签（如 `"before_compact"`），便于回溯
  - `:metadata` — 用户附加 map（如 `%{trigger: :manual, actor: "alice"}`）

  ## 序列化策略

  快照只保留 **可重建的核心状态**：

  - **保留**：`session_id` / `model` / `working_dir` / `messages`（正序化）/
    `tools` / `plugins`（模块名）/ `disabled_tools` / `subagents` / `user_data` /
    `prompt_mode` / `blueprint_system_prompt` / `todos` / `memory_contents` /
    `dynamic_context_sections` / `steering_queue` / `token_usage` / `turn_count` / `cost_usd`
  - **丢弃**：`plugin_states`（resume 后由 plugin `:session_start` hook 重建）/
    `pending_tool_tasks` / `stream_task_pid` / `streaming_resp` / `monitors` /
    `turn_start_marker` / 流式累积字段（`current_text` / `current_tool_calls` / 等）

  这一策略避免序列化死引用 + 强制 plugin 走 `:session_start` 重建,
  与 `CMDC.Checkpoint.Snapshot` 模块文档一致。

  ## 示例

      {:ok, snap} = CMDC.checkpoint!(session, label: "after_plan_approved")
      :ok = save_to_db(snap.checkpoint_id, snap)

      # 配合 redact helper 写前去敏
      {:ok, snap} = CMDC.checkpoint!(session)
      sanitized = CMDC.Checkpoint.Snapshot.redact(snap, &MyApp.Vault.encrypt/1)

  ## 错误返回

  - `{:error, :invalid_session}` — session_id 字符串查不到对应进程
  - `{:error, :not_alive}` — pid 已退出
  - `{:error, term()}` — backend `save/3` 返回的错误（如 PG 连接断）
  """
  @spec checkpoint!(session(), keyword()) ::
          {:ok, Snapshot.t()} | {:error, :invalid_session | :not_alive | term()}
  def checkpoint!(session, opts \\ []) when is_list(opts) do
    with_agent(session, fn agent_pid ->
      state = Agent.get_state(agent_pid)
      serializable = state_to_serializable(state)
      Checkpoint.save(state.session_id, serializable, opts)
    end)
  end

  @doc """
  从 Snapshot 重建一个 Agent 会话，messages / tools / plugins / user_data 等全部还原。

  ## 行为

  - 启动新的 `SessionServer` 监督树（session_id 默认沿用 snapshot,可通过
    `:session_id` 选项覆盖以避免 Registry 冲突）
  - **messages** 正序灌入（Agent.init 内部 reverse 成头新顺序）
  - **plugins** 从模块名列表重建,初始 plugin_state 走每个 plugin 的 `init/1` 默认值;
    plugin 通过订阅自动触发的 `:session_start` hook 恢复需要的状态
  - **运行期字段**（streaming / pending tools / monitors / turn_start_marker 等）
    全部由 Agent.init 重新初始化

  ## 选项

  允许覆盖 snapshot 中的字段（典型用途：换 working_dir、换 provider_opts 接生产 LLM 等）:

  - `:session_id` — 默认沿用 snapshot.session_id;若同名 session 已运行需指定新 ID
  - `:provider_opts` — 重连 LLM Provider 时传新 api_key / base_url
  - `:hibernate_after_ms` — 新增/覆盖 hibernate 配置
  - 其他任意 `CMDC.Options` 字段

  返回 `{:ok, pid()}`（与 `create_agent/1` 一致），调用方可直接传给 `prompt/2` 续接对话。

  ## 示例

      # 跨 BEAM 节点恢复
      {:ok, snap} = CMDC.Checkpoint.load("sess-prod-001")
      {:ok, session} = CMDC.resume_session!(snap, provider_opts: [api_key: System.get_env("API_KEY")])
      CMDC.prompt(session, "继续刚才的话题")

      # 同进程内 fork 一个新会话（不同 session_id）
      {:ok, snap} = CMDC.checkpoint!(running_session)
      {:ok, fork} = CMDC.resume_session!(snap, session_id: "sess-fork-1")

  ## 错误返回

  - `{:error, {:already_started, pid}}` — session_id 已运行
  - `{:error, :missing_model}` — snapshot.state 没有 `:model` 字段（损坏的快照）
  - 其他 `create_agent/1` 可能返回的错误
  """
  @spec resume_session!(Snapshot.t(), keyword()) ::
          {:ok, session()} | {:error, term()}
  def resume_session!(%Snapshot{} = snap, extra_opts \\ []) when is_list(extra_opts) do
    case snapshot_to_start_opts(snap) do
      {:ok, base_opts} ->
        base_opts
        |> Keyword.merge(extra_opts)
        |> create_agent()

      {:error, _} = err ->
        err
    end
  end

  # ==========================================================================
  # 私有辅助 — Snapshot ↔ Agent.State 转换
  # ==========================================================================

  # 运行期字段(plugin / stream / monitor / tool task 等)在 resume 时由 Agent.init
  # 重建，不进 Snapshot；与 CMDC.Checkpoint.Snapshot moduledoc 描述的"不可序列化字段策略"对齐。
  @resume_dropped_state_fields [
    :pending_tool_tasks,
    :stream_task_pid,
    :streaming_resp,
    :monitors,
    :turn_start_marker,
    :tag_buffers,
    :tool_call_arg_buffers,
    :current_text,
    :current_tool_calls,
    :current_thinking,
    :message_started,
    :last_chunk_at,
    :stream_errored,
    :stall_count,
    :plugin_states,
    :exit_reason
  ]

  defp state_to_serializable(%CMDC.Agent.State{} = state) do
    # State.messages 内部头新（逆序），Snapshot / resume 用正序，
    # 这样和 Options.messages / Agent.init 的输入语义一致。
    ordered_messages = Enum.reverse(state.messages)

    # plugins 字段在 State 中是 [{module, plugin_state}]，
    # resume 时只需要 [module] / [{module, opts}] 形式重新 init。
    plugin_specs =
      Enum.map(state.plugins, fn
        {mod, _ps} when is_atom(mod) -> mod
        mod when is_atom(mod) -> mod
      end)

    state
    |> Map.from_struct()
    |> Map.put(:messages, ordered_messages)
    |> Map.put(:plugins, plugin_specs)
    |> Map.drop(@resume_dropped_state_fields)
  end

  defp snapshot_to_start_opts(%Snapshot{state: state, session_id: sid}) when is_map(state) do
    case Map.get(state, :model) do
      nil ->
        {:error, :missing_model}

      model when is_binary(model) ->
        {:ok,
         [
           session_id: sid,
           model: model,
           working_dir: Map.get(state, :working_dir, "."),
           tools: Map.get(state, :tools, []),
           plugins: Map.get(state, :plugins, []),
           disabled_tools: Map.get(state, :disabled_tools, []),
           subagents: Map.get(state, :subagents, []),
           user_data: Map.get(state, :user_data, %{}),
           messages: Map.get(state, :messages, []),
           prompt_mode: Map.get(state, :prompt_mode, :full),
           system_prompt: Map.get(state, :blueprint_system_prompt, "")
         ]
         |> Enum.reject(fn {_k, v} -> v in [nil, "", []] end)
         |> Keyword.put(:session_id, sid)
         |> Keyword.put(:model, model)
         |> Keyword.put(:messages, Map.get(state, :messages, []))}
    end
  end

  defp snapshot_to_start_opts(%Snapshot{}), do: {:error, :invalid_snapshot}

  # ==========================================================================
  # 私有辅助 — session 入参归一化（pid 与 session_id 字符串都接受）
  # ==========================================================================

  # 所有 normalize/with_* 内部 helper 返回 `{:ok, _} | {:error, _}`
  # 风格，不再 raise / exit 调用方。Facade API 通过 `with_agent/2` / `with_sid/2`
  # 把内部错误透传为标准 `{:error, :invalid_session | :not_alive}`。
  #
  # 成功路径的返回值保持 :ok / map / Message struct 等不变;
  # 失败路径才返回 `{:error, _}`——老 `case ... do :ok -> ... ; {:error, _} -> ... end`
  # 模式继续工作；旧 `:ok = CMDC.xxx(invalid_sid)` 模式会 MatchError，但 invalid
  # session 本来就是程序错误。

  @typedoc "normalize_to_pid 返回类型。"
  @type pid_lookup :: {:ok, pid()} | {:error, :invalid_session | :not_alive}

  @spec normalize_to_pid(session()) :: pid_lookup()
  defp normalize_to_pid(session) when is_pid(session) do
    if Process.alive?(session), do: {:ok, session}, else: {:error, :not_alive}
  end

  defp normalize_to_pid(session_id) when is_binary(session_id) do
    case SessionServer.whereis(session_id) do
      nil -> {:error, :invalid_session}
      pid when is_pid(pid) -> {:ok, pid}
    end
  end

  defp normalize_to_pid(_other), do: {:error, :invalid_session}

  @spec normalize_to_session_id(session()) :: {:ok, String.t()} | {:error, atom()}
  defp normalize_to_session_id(session_id) when is_binary(session_id), do: {:ok, session_id}

  defp normalize_to_session_id(session) when is_pid(session) do
    with {:ok, pid} <- normalize_to_pid(session) do
      try do
        sid = pid |> SessionServer.agent() |> Agent.status() |> Map.fetch!(:session_id)
        {:ok, sid}
      catch
        :exit, _ -> {:error, :not_alive}
      end
    end
  end

  defp normalize_to_session_id(_other), do: {:error, :invalid_session}

  # 包装 Agent.* 调用：自动 normalize + 透传 invalid_session/not_alive。
  # 成功 → 调 fun，返回 fun 结果（不强行包成 ok-tuple）;
  # 失败 → 立刻返回 `{:error, _}`。
  defp with_agent(session, fun) when is_function(fun, 1) do
    case normalize_to_pid(session) do
      {:ok, pid} ->
        try do
          fun.(SessionServer.agent(pid))
        catch
          :exit, {:noproc, _} -> {:error, :not_alive}
          :exit, {:normal, _} -> {:error, :not_alive}
        end

      {:error, _} = err ->
        err
    end
  end

  defp with_sid(session, fun) when is_function(fun, 1) do
    case normalize_to_session_id(session) do
      {:ok, sid} -> fun.(sid)
      {:error, _} = err -> err
    end
  end

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

  defp ensure_session_id(opts) do
    if Keyword.has_key?(opts, :session_id) do
      opts
    else
      Keyword.put(opts, :session_id, generate_session_id())
    end
  end

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

  defp options_to_session_opts(%CMDC.Options{} = opts) do
    opts
    |> Map.from_struct()
    |> Enum.reject(fn {_k, v} -> is_nil(v) end)
    |> Keyword.new()
    |> ensure_session_id()
  end

  defp await_reply(session_id, timeout) do
    receive do
      {:cmdc_event, ^session_id, {:agent_end, messages, _token_usage}} ->
        reply_text = extract_last_assistant(messages)
        {:ok, reply_text}

      {:cmdc_event, ^session_id, {:agent_abort, reason}} ->
        {:error, {:aborted, reason}}

      {:cmdc_event, ^session_id, {:error, _sid, reason}} ->
        {:error, reason}
    after
      timeout -> {:error, :timeout}
    end
  end

  defp extract_last_assistant(messages) do
    messages
    |> Enum.filter(&(&1.role == :assistant))
    |> List.last()
    |> case do
      nil -> ""
      msg -> msg.content || ""
    end
  end
end