FranklinWH-Toolbox/service/utils/acp/acp.go

// Package acp 实现 ACP (Agent Client Protocol) 客户端
// 基于 ACP v1 规范，通过 stdin/stdout ND-JSON 与 Agent 子进程通讯
package acp

import (
	"bufio"
	"encoding/json"
	"fmt"
	"io"
	"os"
	"os/exec"
	"os/user"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"syscall"
	"time"

	"fwh-toolbox-server/utils/filestore"

	"go.uber.org/zap"
)

// ============ JSON-RPC 2.0 基础类型 ============

// Request JSON-RPC 请求
type Request struct {
	JSONRPC string `json:"jsonrpc"`
	ID      int64  `json:"id"`
	Method  string `json:"method"`
	Params  any    `json:"params,omitempty"`
}

// Response JSON-RPC 响应
type Response struct {
	JSONRPC string          `json:"jsonrpc"`
	ID      *int64          `json:"id,omitempty"`
	Method  string          `json:"method,omitempty"`
	Result  json.RawMessage `json:"result,omitempty"`
	Error   *RPCError       `json:"error,omitempty"`
	Params  json.RawMessage `json:"params,omitempty"`
}

// RPCError JSON-RPC 错误
type RPCError struct {
	Code    int    `json:"code"`
	Message string `json:"message"`
	Data    string `json:"data,omitempty"`
}

// FullMessage 返回包含 data 的完整错误信息
func (e *RPCError) FullMessage() string {
	if e.Data != "" {
		return fmt.Sprintf("%s: %s", e.Message, e.Data)
	}
	return e.Message
}

// ============ ACP 协议类型 ============

// InitializeParams 初始化请求参数
type InitializeParams struct {
	ProtocolVersion    int                `json:"protocolVersion"`
	ClientInfo         ClientInfo         `json:"clientInfo"`
	ClientCapabilities ClientCapabilities `json:"clientCapabilities"`
}

// ClientInfo 客户端信息
type ClientInfo struct {
	Name    string `json:"name"`
	Version string `json:"version"`
}

// ClientCapabilities 客户端能力声明
type ClientCapabilities struct {
	FS       *FSCapabilities `json:"fs,omitempty"`
	Terminal bool            `json:"terminal,omitempty"`
}

// FSCapabilities 文件系统能力
type FSCapabilities struct {
	ReadTextFile  bool `json:"readTextFile"`
	WriteTextFile bool `json:"writeTextFile"`
}

// NewSessionParams 创建会话参数
type NewSessionParams struct {
	CWD        string `json:"cwd"`
	MCPServers []any  `json:"mcpServers"`
}

// NewSessionResult 创建会话响应
type NewSessionResult struct {
	SessionId string         `json:"sessionId"`
	Modes     *SessionModes  `json:"modes,omitempty"`
	Models    *SessionModels `json:"models,omitempty"`
}

// SessionModes 会话模式
type SessionModes struct {
	CurrentModeId  string        `json:"currentModeId"`
	AvailableModes []SessionMode `json:"availableModes"`
}

// SessionMode 单个模式
type SessionMode struct {
	Id          string       `json:"id"`
	Name        string       `json:"name"`
	Description string       `json:"description"`
	Meta        *SessionMeta `json:"_meta,omitempty"`
}

// SessionMeta 模式元数据
type SessionMeta struct {
	WelcomeMessage string `json:"welcomeMessage"`
}

// SessionModels 会话可用模型
type SessionModels struct {
	CurrentModelId  string         `json:"currentModelId"`
	AvailableModels []SessionModel `json:"availableModels"`
}

// SessionModel 单个模型
type SessionModel struct {
	ModelId     string `json:"modelId"`
	Name        string `json:"name"`
	Description string `json:"description"`
}

// PromptParams 发送消息参数
type PromptParams struct {
	SessionId string         `json:"sessionId"`
	Prompt    []ContentBlock `json:"prompt"`
}

// ContentBlock 内容块
type ContentBlock struct {
	Type string `json:"type"`
	Text string `json:"text"`
}

// PromptResult prompt 响应
type PromptResult struct {
	StopReason string `json:"stopReason"`
}

// CancelParams 取消参数
type CancelParams struct {
	SessionId string `json:"sessionId"`
}

// ============ Session Update 事件类型 ============

// SessionUpdate 会话更新通知
type SessionUpdate struct {
	SessionId string          `json:"sessionId"`
	Update    json.RawMessage `json:"update"`
}

// UpdateBase 更新基础类型
type UpdateBase struct {
	SessionUpdate string `json:"sessionUpdate"`
}

// AgentMessageChunk Agent 文本流式输出
type AgentMessageChunk struct {
	SessionUpdate string       `json:"sessionUpdate"` // "agent_message_chunk"
	MessageId     string       `json:"messageId,omitempty"`
	Content       ContentBlock `json:"content"`
}

// ToolCall 工具调用通知
type ToolCall struct {
	SessionUpdate string `json:"sessionUpdate"` // "tool_call"
	ToolCallId    string `json:"toolCallId"`
	Title         string `json:"title"`
	Status        string `json:"status"` // pending, running, completed, failed
}

// ToolCallUpdate 工具调用状态更新
type ToolCallUpdate struct {
	SessionUpdate string `json:"sessionUpdate"` // "tool_call_update"
	ToolCallId    string `json:"toolCallId"`
	Title         string `json:"title,omitempty"`
	Status        string `json:"status"`
}

// ============ 客户端事件（对外输出） ============

// Event 输出给上层的事件
type Event struct {
	Type       string // "text", "tool_call", "tool_update", "done", "error", "permission"
	Content    string
	ToolCallId string // 工具调用唯一ID（用于前端按 id 关联状态更新）
	ToolName   string
	ToolStatus string
	SessionId  string
	StopReason string
	// 权限请求相关
	Permission *PermissionEvent `json:"permission,omitempty"`
}

// PermissionEvent 权限请求事件（推给前端）
type PermissionEvent struct {
	RequestId string             `json:"requestId"` // 用于前端回传
	Title     string             `json:"title"`     // 权限标题
	Options   []PermissionOption `json:"options"`   // 可选选项
}

// PermissionOption 权限选项
type PermissionOption struct {
	OptionId string `json:"optionId"`
	Name     string `json:"name"`
	Kind     string `json:"kind"` // allow_once, allow_always, deny
}

// ============ ACP Client 实现 ============

// PermissionWaitTimeout 前端授权回传的最长等待时长。
// 超时后默认按 deny 处理，避免读循环因前端断开而永久阻塞。
const PermissionWaitTimeout = 5 * time.Minute

// pendingResponse 等待中的请求：读循环收到匹配 id 的响应后通过 ch 投递
type pendingResponse struct {
	ch chan Response
}

// notifySink 通知回调：读循环把 notification 派发给当前阶段的接收者。
// 返回值无意义，仅用于让不同阶段（Prompt / LoadSession / 默认）自定义处理。
type notifySink func(method string, params json.RawMessage)

// Client ACP 客户端（管理单个 Agent 进程）
//
// 并发模型：构造时启动唯一的 readLoop goroutine，负责读取 stdout 的每一行并派发：
//   - 响应（有 id、无 method）→ 按 id 投递到 pending map 中等待的 channel；
//   - Agent→Client 反向请求（有 id、有 method）→ 交给 handleAgentRequest（在独立
//     goroutine 中处理，避免权限等待阻塞读循环）；
//   - 通知（无 id、有 method）→ 交给当前 notifySink。
//
// 所有对 stdin 的写入通过 writeMu 串行化；Cancel 与权限响应因此不会互相死锁。
type Client struct {
	cmd     *exec.Cmd
	stdin   io.WriteCloser
	scanner *bufio.Scanner

	sessionId string
	requestId int64

	writeMu sync.Mutex // 保护 stdin 写入

	// 请求分发：id -> 等待响应的 channel
	pendingMu sync.Mutex
	pending   map[int64]*pendingResponse

	// 通知分发：读循环把 notification 交给当前 sink
	sinkMu  sync.Mutex
	sink    notifySink
	eventCh chan<- Event  // 当前 prompt 的事件通道（用于推送 permission 事件），受 sinkMu 保护
	abortCh chan struct{} // 当前 prompt 的中止信号（上层停止消费时用于解除发送阻塞），受 sinkMu 保护
	emitWg  sync.WaitGroup // 统计在途 emit，endTurn 借此在 close(eventCh) 前等待发送退出

	// 权限请求等待机制
	permissionCh chan string // 接收前端选择的 optionId

	// 进程存活标记：由 readLoop / waitProcess 单点置位，IsAlive 读取
	dead   atomic.Bool
	doneCh chan struct{} // 进程退出/读循环结束时关闭

	// 文件存储：会话身份（用于定位 MinIO 中的文件）
	UserId string

	// 文本累积（仅用于在 turn 结束时打一条完整回复日志）。
	// 由 sink（readLoop goroutine）写入、Prompt goroutine 读取，故用 textMu 保护。
	textMu     sync.Mutex
	textBuffer string

	// 最近一次上下文用量与耗时（供首屏展示，读取需加 metaMu）
	metaMu             sync.Mutex
	lastContextUsage   float64
	lastTurnDurationMs int64
}

// ClientConfig 客户端配置
type ClientConfig struct {
	Command    string   // Agent 命令（如 "kiro-cli"）
	Args       []string // 命令参数（如 ["acp"]）
	WorkingDir string   // 工作目录
	Env        []string // 额外环境变量（如 "HOME=/home/user"）
	RunAsUser  string   // 降权运行的目标系统用户名，为空则不降权
}

// ApplyRunAsUser 让 cmd 以指定系统用户身份运行（降权）。
// runAsUser 为空时不做任何处理（沿用当前进程用户）。
// 该函数会：
//  1. 解析目标用户的 uid/gid，并设置 cmd.SysProcAttr.Credential；
//  2. 附带解析该用户的补充组（supplementary groups）；
//  3. 修正 HOME/USER/LOGNAME 环境变量为目标用户的值，避免 kiro-cli
//     仍读取 root 的家目录导致配置/凭证路径错误。
//
// 注意：设置 Credential 需要当前进程具备相应权限（通常为 root）。
func ApplyRunAsUser(cmd *exec.Cmd, runAsUser string) error {
	runAsUser = strings.TrimSpace(runAsUser)
	if runAsUser == "" {
		return nil
	}

	u, err := user.Lookup(runAsUser)
	if err != nil {
		return fmt.Errorf("查找降权用户 %q 失败: %w", runAsUser, err)
	}

	uid, err := strconv.ParseUint(u.Uid, 10, 32)
	if err != nil {
		return fmt.Errorf("解析用户 %q 的 uid(%s) 失败: %w", runAsUser, u.Uid, err)
	}
	gid, err := strconv.ParseUint(u.Gid, 10, 32)
	if err != nil {
		return fmt.Errorf("解析用户 %q 的 gid(%s) 失败: %w", runAsUser, u.Gid, err)
	}

	// 解析补充组（失败不致命，仅记录告警）
	var groups []uint32
	if gidStrs, gErr := u.GroupIds(); gErr == nil {
		for _, gs := range gidStrs {
			if g, pErr := strconv.ParseUint(gs, 10, 32); pErr == nil {
				groups = append(groups, uint32(g))
			}
		}
	} else {
		zap.L().Warn("[ACP] 读取降权用户补充组失败", zap.String("user", runAsUser), zap.Error(gErr))
	}

	if cmd.SysProcAttr == nil {
		cmd.SysProcAttr = &syscall.SysProcAttr{}
	}
	cmd.SysProcAttr.Credential = &syscall.Credential{
		Uid:    uint32(uid),
		Gid:    uint32(gid),
		Groups: groups,
	}

	// 修正环境变量，确保子进程使用目标用户的身份。
	// USER/LOGNAME 强制设为目标用户；HOME 尊重上层（配置）已传入的值，
	// 仅在未提供时才回退到用户家目录。
	if cmd.Env == nil {
		cmd.Env = os.Environ()
	}
	cmd.Env = overrideEnv(cmd.Env, map[string]string{
		"USER":    u.Username,
		"LOGNAME": u.Username,
	})
	cmd.Env = fillEnvIfAbsent(cmd.Env, "HOME", u.HomeDir)

	effectiveHome := u.HomeDir
	for _, e := range cmd.Env {
		if strings.HasPrefix(e, "HOME=") {
			effectiveHome = strings.TrimPrefix(e, "HOME=")
			break
		}
	}

	zap.L().Info("[ACP] 子进程降权运行",
		zap.String("user", u.Username),
		zap.String("uid", u.Uid),
		zap.String("gid", u.Gid),
		zap.String("home", effectiveHome),
	)
	return nil
}

// fillEnvIfAbsent 仅当 env 中不存在 key 时才追加 key=value，否则原样返回。
func fillEnvIfAbsent(env []string, key, value string) []string {
	prefix := key + "="
	for _, e := range env {
		if strings.HasPrefix(e, prefix) {
			return env
		}
	}
	return append(env, prefix+value)
}

// envHasKey 判断 env（"KEY=VALUE" 形式的切片）中是否已存在指定 key。
func envHasKey(env []string, key string) bool {
	prefix := key + "="
	for _, e := range env {
		if strings.HasPrefix(e, prefix) {
			return true
		}
	}
	return false
}

// ensureDirWritableByRunAsUser 确保日志目录可被降权用户写入。
// 主服务常以 root 运行，创建的目录属主为 root；子进程降权到 runAsUser 后会写不进去，
// 导致 kiro-cli 日志被静默丢弃。这里将目录（含其直接父目录）chown 到 runAsUser 的 uid/gid。
// runAsUser 为空表示不降权（与主进程同用户），无需调整。所有失败仅记录告警，不致命。
func ensureDirWritableByRunAsUser(dir string, runAsUser string) {
	runAsUser = strings.TrimSpace(runAsUser)
	if runAsUser == "" {
		return
	}
	u, err := user.Lookup(runAsUser)
	if err != nil {
		zap.L().Warn("[ACP] 查找降权用户失败，日志目录属主未调整", zap.String("user", runAsUser), zap.Error(err))
		return
	}
	uid, err1 := strconv.Atoi(u.Uid)
	gid, err2 := strconv.Atoi(u.Gid)
	if err1 != nil || err2 != nil {
		zap.L().Warn("[ACP] 解析降权用户 uid/gid 失败，日志目录属主未调整",
			zap.String("user", runAsUser), zap.String("uid", u.Uid), zap.String("gid", u.Gid))
		return
	}
	// 同时调整父目录（如 /tmp/kiro-log）与目标目录（如 /tmp/kiro-log/logs）。
	for _, d := range []string{filepath.Dir(dir), dir} {
		if err := os.Chown(d, uid, gid); err != nil {
			zap.L().Warn("[ACP] 调整日志目录属主失败（降权用户可能无法写入日志）",
				zap.String("dir", d), zap.String("user", runAsUser), zap.Error(err))
		}
	}
}

// overrideEnv 用 kv 覆盖 env 中已存在的同名变量，不存在则追加。
func overrideEnv(env []string, kv map[string]string) []string {
	out := make([]string, 0, len(env)+len(kv))
	seen := make(map[string]bool, len(kv))
	for _, e := range env {
		key := e
		if idx := strings.IndexByte(e, '='); idx >= 0 {
			key = e[:idx]
		}
		if v, ok := kv[key]; ok {
			out = append(out, key+"="+v)
			seen[key] = true
		} else {
			out = append(out, e)
		}
	}
	for k, v := range kv {
		if !seen[k] {
			out = append(out, k+"="+v)
		}
	}
	return out
}

// NewClient 创建并启动 ACP Client
func NewClient(cfg ClientConfig) (*Client, error) {
	cmd := exec.Command(cfg.Command, cfg.Args...)

	// 确保工作目录是绝对路径
	absDir, err := filepath.Abs(cfg.WorkingDir)
	if err != nil {
		absDir = cfg.WorkingDir
	}
	cmd.Dir = absDir

	// 设置环境变量
	cmd.Env = append(os.Environ(), cfg.Env...)

	// A 方案：为每次启动的子进程分配带启动时间戳的独立日志文件，避免覆盖历史日志。
	// 仅在开启了 kiro-cli debug 日志（env 含 KIRO_LOG_LEVEL）、且调用方未显式指定
	// KIRO_CHAT_LOG_FILE 时生效。KIRO_CHAT_LOG_FILE 是 kiro-cli acp 官方支持的日志路径变量。
	if envHasKey(cmd.Env, "KIRO_LOG_LEVEL") && !envHasKey(cmd.Env, "KIRO_CHAT_LOG_FILE") {
		logDir := "/tmp/kiro-log/logs"
		if err := os.MkdirAll(logDir, 0o777); err != nil {
			zap.L().Warn("[ACP] 创建日志目录失败，沿用 kiro-cli 默认日志路径",
				zap.String("dir", logDir), zap.Error(err))
		} else {
			// 主服务常以 root 运行，创建出的目录默认属主为 root；而子进程会被
			// ApplyRunAsUser 降权到 cfg.RunAsUser，若目录不可写日志将被静默丢弃。
			// 因此这里把目录属主/权限调整到降权用户可写。
			ensureDirWritableByRunAsUser(logDir, cfg.RunAsUser)

			logPath := filepath.Join(logDir, fmt.Sprintf("kiro-chat-%s.log", time.Now().Format("20060102T150405.000")))
			cmd.Env = append(cmd.Env, "KIRO_CHAT_LOG_FILE="+logPath)
			// 顺带关闭 ANSI 颜色码，便于后续直接阅读/grep 日志。
			if !envHasKey(cmd.Env, "KIRO_LOG_NO_COLOR") {
				cmd.Env = append(cmd.Env, "KIRO_LOG_NO_COLOR=1")
			}
			zap.L().Info("[ACP] 本次子进程日志文件", zap.String("logFile", logPath))
		}
	}

	// 若配置了降权用户，则以该用户身份拉起子进程
	if err := ApplyRunAsUser(cmd, cfg.RunAsUser); err != nil {
		return nil, err
	}

	stdin, err := cmd.StdinPipe()
	if err != nil {
		return nil, fmt.Errorf("创建 stdin pipe 失败: %w", err)
	}

	stdout, err := cmd.StdoutPipe()
	if err != nil {
		return nil, fmt.Errorf("创建 stdout pipe 失败: %w", err)
	}

	// 捕获 stderr
	stderrPipe, _ := cmd.StderrPipe()
	go func() {
		if stderrPipe != nil {
			s := bufio.NewScanner(stderrPipe)
			for s.Scan() {
				zap.L().Warn("[ACP] agent stderr", zap.String("line", s.Text()))
			}
		}
	}()

	if err := cmd.Start(); err != nil {
		return nil, fmt.Errorf("启动 Agent 进程失败: %w", err)
	}

	scanner := bufio.NewScanner(stdout)
	scanner.Buffer(make([]byte, 0, 2*1024*1024), 2*1024*1024) // 2MB buffer

	c := &Client{
		cmd:          cmd,
		stdin:        stdin,
		scanner:      scanner,
		pending:      make(map[int64]*pendingResponse),
		permissionCh: make(chan string, 1),
		doneCh:       make(chan struct{}),
	}

	// 监听子进程退出，单点置位 dead 标记（供 IsAlive 可靠判活）
	go c.waitProcess()
	// 启动唯一的读循环，统一分发响应/请求/通知
	go c.readLoop()

	return c, nil
}

// waitProcess 阻塞等待子进程退出，退出后标记 dead。
// 这是回收子进程、置位存活标记的唯一入口，因此 IsAlive 不再依赖
// cmd.ProcessState（只有调用 Wait 后才会被填充）的偶然状态。
func (c *Client) waitProcess() {
	_ = c.cmd.Wait()
	c.markDead()
}

// markDead 标记进程已退出，并关闭 doneCh（幂等）。
func (c *Client) markDead() {
	if c.dead.CompareAndSwap(false, true) {
		close(c.doneCh)
	}
}

// ============ 读循环与分发 ============

// readLoop 是唯一读取 stdout 的 goroutine。它把每行消息按类型派发：
//   - 响应（有 id、无 method）→ 投递到 pending map 中等待的 channel；
//   - 反向请求（有 id、有 method）→ 在独立 goroutine 中处理（权限等待不阻塞读循环）；
//   - 通知（无 id、有 method）→ 交给当前 notifySink。
//
// scanner 结束（EOF / 进程退出）时标记 dead，并唤醒所有等待中的请求。
func (c *Client) readLoop() {
	defer func() {
		c.markDead()
		c.failAllPending()
	}()

	for c.scanner.Scan() {
		line := c.scanner.Text()
		if line == "" {
			continue
		}

		zap.L().Debug("[ACP] <<< 收到", zap.String("raw", line[:min(len(line), 500)]))

		var msg struct {
			JSONRPC string          `json:"jsonrpc"`
			ID      json.RawMessage `json:"id,omitempty"`
			Method  string          `json:"method,omitempty"`
			Result  json.RawMessage `json:"result,omitempty"`
			Error   *RPCError       `json:"error,omitempty"`
			Params  json.RawMessage `json:"params,omitempty"`
		}
		if err := json.Unmarshal([]byte(line), &msg); err != nil {
			zap.L().Warn("[ACP] JSON解析失败，跳过", zap.String("line", line[:min(len(line), 100)]))
			continue
		}

		hasId := len(msg.ID) > 0 && string(msg.ID) != "null"

		switch {
		case hasId && msg.Method == "":
			// 响应：按 id 派发给等待者
			var id int64
			if err := json.Unmarshal(msg.ID, &id); err != nil {
				zap.L().Warn("[ACP] 响应 id 解析失败", zap.String("id", string(msg.ID)))
				continue
			}
			c.deliverResponse(id, Response{
				JSONRPC: msg.JSONRPC,
				ID:      &id,
				Result:  msg.Result,
				Error:   msg.Error,
			})

		case hasId && msg.Method != "":
			// Agent→Client 反向请求：独立 goroutine 处理，避免权限等待阻塞读循环
			id := append(json.RawMessage(nil), msg.ID...)
			params := append(json.RawMessage(nil), msg.Params...)
			method := msg.Method
			go c.handleAgentRequest(id, method, params)

		case !hasId && msg.Method != "":
			// 通知：交给当前 sink
			c.dispatchNotification(msg.Method, msg.Params)

		default:
			zap.L().Debug("[ACP] 未处理的消息", zap.String("line", line[:min(len(line), 200)]))
		}
	}
}

// registerPending 注册一个等待响应的请求，返回接收 channel。
func (c *Client) registerPending(id int64) chan Response {
	ch := make(chan Response, 1)
	c.pendingMu.Lock()
	c.pending[id] = &pendingResponse{ch: ch}
	c.pendingMu.Unlock()
	return ch
}

// unregisterPending 移除等待项（请求完成或超时后调用）。
func (c *Client) unregisterPending(id int64) {
	c.pendingMu.Lock()
	delete(c.pending, id)
	c.pendingMu.Unlock()
}

// deliverResponse 把响应投递给等待者；无人等待则忽略。
func (c *Client) deliverResponse(id int64, resp Response) {
	c.pendingMu.Lock()
	p, ok := c.pending[id]
	if ok {
		delete(c.pending, id)
	}
	c.pendingMu.Unlock()
	if ok {
		p.ch <- resp
	} else {
		zap.L().Debug("[ACP] 收到无等待者的响应", zap.Int64("id", id))
	}
}

// failAllPending 进程退出时唤醒所有等待中的请求，返回错误响应。
func (c *Client) failAllPending() {
	c.pendingMu.Lock()
	pend := c.pending
	c.pending = make(map[int64]*pendingResponse)
	c.pendingMu.Unlock()
	for id, p := range pend {
		p.ch <- Response{ID: &id, Error: &RPCError{Code: -1, Message: "Agent 进程已断开"}}
	}
}

// setSink 设置当前通知接收者，返回旧 sink（便于恢复）。
func (c *Client) setSink(s notifySink) notifySink {
	c.sinkMu.Lock()
	old := c.sink
	c.sink = s
	c.sinkMu.Unlock()
	return old
}

// beginTurn 在一次 prompt 开始时设置事件通道与中止信号（受 sinkMu 保护），
// 返回本轮的 abort channel 供上层（如断连时）主动中止。
func (c *Client) beginTurn(ch chan<- Event) chan struct{} {
	abort := make(chan struct{})
	c.sinkMu.Lock()
	c.eventCh = ch
	c.abortCh = abort
	c.sinkMu.Unlock()
	return abort
}

// endTurn 在一次 prompt 结束时清理，并保证调用方随后可安全 close(eventCh)：
//  1. 关闭 abort，唤醒可能阻塞在发送上的 emit；
//  2. 等待所有在途 emit（emitWg）退出；
//  3. 在锁内清空 eventCh/abortCh。
//
// 注意：emit 不持 sinkMu 阻塞发送，故这里关闭 abort 不会与 emit 形成锁环。
func (c *Client) endTurn() {
	c.sinkMu.Lock()
	abort := c.abortCh
	c.sinkMu.Unlock()

	if abort != nil {
		safeClose(abort)
	}
	c.emitWg.Wait()

	c.sinkMu.Lock()
	c.eventCh = nil
	c.abortCh = nil
	c.sinkMu.Unlock()
}

// abortTurn 主动中止当前 turn 的事件发送（上层连接断开时调用）。
// 仅关闭 abort 信号，唤醒阻塞的 emit；turn 的最终结束仍由 Prompt 收到响应/取消驱动。
func (c *Client) abortTurn() {
	c.sinkMu.Lock()
	abort := c.abortCh
	c.sinkMu.Unlock()
	if abort != nil {
		safeClose(abort)
	}
}

// AbortEvents 供上层在连接断开时调用：唤醒所有阻塞在事件发送上的 goroutine，
// 使当前 turn 的事件流尽快收敛（配合 Cancel 通知 Agent 停止本轮）。
func (c *Client) AbortEvents() {
	c.abortTurn()
}

// safeClose 幂等关闭一个 struct{} channel。
func safeClose(ch chan struct{}) {
	defer func() { _ = recover() }()
	close(ch)
}

// emit 向当前 turn 的事件通道发送一个事件，返回是否成功送达。
// 发送不持有 sinkMu（避免与 endTurn 关闭 abort 形成锁环），改用 emitWg 标记在途状态，
// 使 endTurn 能在 close(eventCh) 前等待所有发送退出，杜绝向已关闭通道发送。
// 发送同时监听 abort（上层停止消费）与 doneCh（进程退出），任一触发即放弃发送。
func (c *Client) emit(ev Event) bool {
	c.emitWg.Add(1)
	defer c.emitWg.Done()

	c.sinkMu.Lock()
	ch := c.eventCh
	abort := c.abortCh
	c.sinkMu.Unlock()

	if ch == nil {
		return false
	}
	select {
	case ch <- ev:
		return true
	case <-abort:
		return false
	case <-c.doneCh:
		return false
	}
}

// pushPermissionEvent 把权限事件推给当前 eventCh；返回是否成功（无活跃通道时返回 false）。
func (c *Client) pushPermissionEvent(ev Event) bool {
	return c.emit(ev)
}

// dispatchNotification 把通知交给当前 sink；无 sink 时走默认处理（仅缓存 metadata）。
func (c *Client) dispatchNotification(method string, params json.RawMessage) {
	c.sinkMu.Lock()
	s := c.sink
	c.sinkMu.Unlock()
	if s != nil {
		s(method, params)
		return
	}
	c.defaultNotify(method, params)
}

// defaultNotify 无活跃 sink 时的兜底：仅捕获 metadata 用量，其余忽略。
func (c *Client) defaultNotify(method string, params json.RawMessage) {
	if method == "_kiro.dev/metadata" {
		var data struct {
			ContextUsagePercentage float64 `json:"contextUsagePercentage"`
			TurnDurationMs         int64   `json:"turnDurationMs,omitempty"`
		}
		if err := json.Unmarshal(params, &data); err == nil {
			c.setMeta(data.ContextUsagePercentage, data.TurnDurationMs)
		}
	}
}

// ResolvePermission 前端回传权限选择结果
func (c *Client) ResolvePermission(optionId string) {
	select {
	case c.permissionCh <- optionId:
	default:
		// 没有等待中的权限请求，忽略
	}
}

// appendText 累积本轮 AI 文本（sink/readLoop goroutine 调用）。
func (c *Client) appendText(s string) {
	c.textMu.Lock()
	c.textBuffer += s
	c.textMu.Unlock()
}

// takeText 取出并清空累积文本（Prompt goroutine 在 turn 结束时调用）。
func (c *Client) takeText() string {
	c.textMu.Lock()
	s := c.textBuffer
	c.textBuffer = ""
	c.textMu.Unlock()
	return s
}

// resetText 清空累积文本（turn 开始前调用，避免跨 turn 串味）。
func (c *Client) resetText() {
	c.textMu.Lock()
	c.textBuffer = ""
	c.textMu.Unlock()
}

// setMeta 更新最近的上下文用量与耗时（值 >0 时才覆盖）
func (c *Client) setMeta(usage float64, turnMs int64) {
	c.metaMu.Lock()
	defer c.metaMu.Unlock()
	if usage > 0 {
		c.lastContextUsage = usage
	}
	if turnMs > 0 {
		c.lastTurnDurationMs = turnMs
	}
}

// GetMeta 返回最近的上下文用量（百分比）与上一轮耗时（毫秒）
func (c *Client) GetMeta() (float64, int64) {
	c.metaMu.Lock()
	defer c.metaMu.Unlock()
	return c.lastContextUsage, c.lastTurnDurationMs
}

// Initialize 执行初始化握手
func (c *Client) Initialize() error {
	params := InitializeParams{
		ProtocolVersion: 1,
		ClientInfo: ClientInfo{
			Name:    "fwh-toolbox",
			Version: "1.0.0",
		},
		ClientCapabilities: ClientCapabilities{
			FS: &FSCapabilities{
				ReadTextFile:  true,
				WriteTextFile: true,
			},
			Terminal: false, // 不声明终端能力，强制 Agent 使用 writeTextFile
		},
	}

	_, err := c.sendAndWait("initialize", params)
	return err
}

// NewSession 创建新会话，返回 sessionId、welcomeMessage、实际生效的 agentId 和 modelId
func (c *Client) NewSession(cwd string) (sessionId string, welcomeMsg string, agentId string, modelId string, err error) {
	params := NewSessionParams{
		CWD:        cwd,
		MCPServers: []any{},
	}

	result, err := c.sendAndWait("session/new", params)
	if err != nil {
		return "", "", "", "", err
	}

	var res NewSessionResult
	if err := json.Unmarshal(result, &res); err != nil {
		return "", "", "", "", fmt.Errorf("解析 session/new 响应失败: %w", err)
	}

	zap.L().Info("session/new", zap.String("result", string(result)))

	c.sessionId = res.SessionId

	// 提取当前 mode 的 welcomeMessage 和实际生效的 agentId
	if res.Modes != nil {
		agentId = res.Modes.CurrentModeId
		for _, mode := range res.Modes.AvailableModes {
			if mode.Id == res.Modes.CurrentModeId && mode.Meta != nil {
				welcomeMsg = mode.Meta.WelcomeMessage
				break
			}
		}
	}

	// 提取实际生效的 modelId
	if res.Models != nil {
		modelId = res.Models.CurrentModelId
	}

	// session/new 之后 Kiro 紧跟发送的 commands/available、metadata 等通知，
	// 由读循环交给默认 sink 处理：其中 metadata 的初始上下文用量会被 setMeta 缓存。
	// 不再需要单独的 drainInitialMeta 阻塞读取。

	return res.SessionId, welcomeMsg, agentId, modelId, nil
}

// ReplayMessage 回放的单条消息
type ReplayMessage struct {
	Role      string `json:"role"` // "user" 或 "assistant"
	Content   string `json:"content"`
	MessageId string `json:"messageId,omitempty"`
}

// LoadSession 加载已有会话（恢复历史对话上下文），返回回放的消息列表
func (c *Client) LoadSession(sessionId string, cwd string) (string, string, string, string, []ReplayMessage, error) {
	params := map[string]any{
		"sessionId":  sessionId,
		"cwd":        cwd,
		"mcpServers": []any{},
	}

	// 回放消息在 session/load 响应之前以通知形式到达。用一个临时 sink 收集它们，
	// 由读循环在等待响应期间调用（sendAndWaitWithSink 会在返回前恢复原 sink）。
	var replayMessages []ReplayMessage
	currentMsg := &ReplayMessage{}
	sink := func(method string, p json.RawMessage) {
		if method != "session/update" && method != "_kiro.dev/session/update" {
			// 非回放通知（metadata 等）走默认处理，避免丢失用量
			c.defaultNotify(method, p)
			return
		}
		var notif SessionUpdate
		if err := json.Unmarshal(p, &notif); err != nil {
			return
		}
		var base UpdateBase
		if err := json.Unmarshal(notif.Update, &base); err != nil {
			return
		}

		switch base.SessionUpdate {
		case "user_message_chunk":
			if currentMsg.Content != "" {
				replayMessages = append(replayMessages, *currentMsg)
			}
			var chunk AgentMessageChunk
			json.Unmarshal(notif.Update, &chunk)
			currentMsg = &ReplayMessage{Role: "user", Content: chunk.Content.Text, MessageId: chunk.MessageId}

		case "agent_message_chunk":
			var chunk AgentMessageChunk
			json.Unmarshal(notif.Update, &chunk)
			if currentMsg.Role != "assistant" || (chunk.MessageId != "" && chunk.MessageId != currentMsg.MessageId) {
				if currentMsg.Content != "" {
					replayMessages = append(replayMessages, *currentMsg)
				}
				currentMsg = &ReplayMessage{Role: "assistant", Content: chunk.Content.Text, MessageId: chunk.MessageId}
			} else {
				currentMsg.Content += chunk.Content.Text
			}

		case "agent_thought_chunk", "tool_call", "tool_call_update", "tool_call_chunk":
			// 思考内容与工具调用回放忽略

		default:
			// 其他类型忽略
		}
	}

	result, err := c.sendAndWaitWithSink("session/load", params, sink)
	if err != nil {
		return "", "", "", "", nil, fmt.Errorf("session/load 错误: %w", err)
	}

	c.sessionId = sessionId
	// 追加最后一条未关闭的消息
	if currentMsg.Content != "" {
		replayMessages = append(replayMessages, *currentMsg)
	}

	// 解析响应中的 welcomeMessage、实际生效的 agentId 和 modelId
	var welcomeMsg string
	var agentId string
	var modelId string
	if len(result) > 0 {
		var res NewSessionResult
		if err := json.Unmarshal(result, &res); err == nil {
			if res.Modes != nil {
				agentId = res.Modes.CurrentModeId
				for _, mode := range res.Modes.AvailableModes {
					if mode.Id == res.Modes.CurrentModeId && mode.Meta != nil {
						welcomeMsg = mode.Meta.WelcomeMessage
						break
					}
				}
			}
			if res.Models != nil {
				modelId = res.Models.CurrentModelId
			}
		}
	}

	// session/load 之后的初始用量 metadata 通知由默认 sink 捕获，无需单独处理。
	return sessionId, welcomeMsg, agentId, modelId, replayMessages, nil
}

// Prompt 发送消息并流式读取事件
// eventCh 会在 turn 结束或出错时关闭
func (c *Client) Prompt(sessionId string, message string, eventCh chan<- Event) {
	// 开始一个 turn：登记事件通道与 abort 信号。退出时 endTurn 会关闭 abort、
	// 等待所有在途 emit 结束，再清空通道引用，之后才 close(eventCh)——
	// 保证权限/流式 goroutine 不会向已关闭的通道发送。
	c.beginTurn(eventCh)
	defer func() {
		c.endTurn()
		close(eventCh)
	}()

	params := PromptParams{
		SessionId: sessionId,
		Prompt: []ContentBlock{
			{Type: "text", Text: message},
		},
	}

	// 注册等待 prompt 响应
	promptId := atomic.AddInt64(&c.requestId, 1)
	respCh := c.registerPending(promptId)

	// 清空上一轮可能残留的累积文本，避免跨 turn 串味
	c.resetText()

	// 安装 sink：turn 期间的 session/update 与扩展通知都转成 Event 推给 eventCh。
	// 反向请求（requestPermission）由读循环单独派发，不经过 sink。
	old := c.setSink(func(method string, p json.RawMessage) {
		switch method {
		case "session/update", "_kiro.dev/session/update":
			c.handleSessionUpdate(p)
		default:
			c.handleExtensionNotification(method, p)
		}
	})
	defer c.setSink(old)

	if err := c.writeRequest(promptId, "session/prompt", params); err != nil {
		c.unregisterPending(promptId)
		c.emit(Event{Type: "error", Content: fmt.Sprintf("发送 prompt 失败: %v", err)})
		return
	}
	zap.L().Debug("[ACP] prompt 已发送", zap.Int64("promptId", promptId), zap.String("sessionId", sessionId))

	// 等待 prompt 响应（turn 结束）或进程退出
	select {
	case resp := <-respCh:
		if full := c.takeText(); full != "" {
			zap.L().Info("[ACP] AI 回复完整内容", zap.String("text", full))
		}
		if resp.Error != nil {
			c.emit(Event{Type: "error", Content: resp.Error.FullMessage()})
			return
		}
		var result PromptResult
		if err := json.Unmarshal(resp.Result, &result); err == nil {
			c.emit(Event{Type: "done", SessionId: sessionId, StopReason: result.StopReason})
		}
	case <-c.doneCh:
		c.unregisterPending(promptId)
		c.emit(Event{Type: "error", Content: "Agent 进程已断开"})
	}
}

// Cancel 取消当前 prompt turn（ACP 协议：Notification, method="session/cancel", params={sessionId}）
func (c *Client) Cancel(sessionId string) {
	c.sendNotification("session/cancel", CancelParams{SessionId: sessionId})
}

// GetSessionId 获取当前会话 ID
func (c *Client) GetSessionId() string {
	return c.sessionId
}

// Close 关闭 ACP Client，终止 Agent 进程
func (c *Client) Close() {
	// 关闭 stdin 通知 Agent 退出
	c.stdin.Close()

	// 终止进程（回收由 waitProcess 中的 cmd.Wait 负责，此处不再重复 Wait）
	if c.cmd.Process != nil {
		c.cmd.Process.Kill()
	}

	// 解除可能阻塞在 permissionCh 上的 goroutine
	select {
	case c.permissionCh <- "deny":
	default:
	}
}

// IsAlive 检查 Agent 进程是否存活。
// 依赖 waitProcess/readLoop 单点设置的 dead 标记，而非 cmd.ProcessState
// （后者只有在 Wait 之后才被填充，子进程自行崩溃时会误判为存活）。
func (c *Client) IsAlive() bool {
	return !c.dead.Load()
}

// ============ 内部方法 ============

// sendRequest 发送请求（不等待响应），返回 request ID。
// 注意：调用方若需要等待响应，应使用 sendAndWait；直接使用本方法的场景
// （如 Prompt）需自行通过 registerPending/读循环获取响应。
func (c *Client) sendRequest(method string, params any) int64 {
	id := atomic.AddInt64(&c.requestId, 1)
	if err := c.writeRequest(id, method, params); err != nil {
		zap.L().Error("[ACP] 发送请求失败", zap.String("method", method), zap.Error(err))
	}
	return id
}

// writeRequest 序列化并写入一个带 id 的请求，写入通过 writeMu 串行化。
func (c *Client) writeRequest(id int64, method string, params any) error {
	req := Request{
		JSONRPC: "2.0",
		ID:      id,
		Method:  method,
		Params:  params,
	}

	data, err := json.Marshal(req)
	if err != nil {
		return fmt.Errorf("序列化请求失败: %w", err)
	}

	zap.L().Debug("[ACP] >>> 发送", zap.String("raw", string(data)))
	return c.writeLine(data)
}

// writeLine 追加换行并串行写入 stdin。
func (c *Client) writeLine(data []byte) error {
	c.writeMu.Lock()
	defer c.writeMu.Unlock()
	data = append(data, '\n')
	_, err := c.stdin.Write(data)
	return err
}

// sendNotification 发送 JSON-RPC Notification（不带 id，不期望响应）
func (c *Client) sendNotification(method string, params any) {
	notif := struct {
		JSONRPC string `json:"jsonrpc"`
		Method  string `json:"method"`
		Params  any    `json:"params,omitempty"`
	}{
		JSONRPC: "2.0",
		Method:  method,
		Params:  params,
	}

	data, err := json.Marshal(notif)
	if err != nil {
		zap.L().Error("[ACP] 序列化通知失败", zap.Error(err))
		return
	}

	zap.L().Debug("[ACP] >>> 发送(notification)", zap.String("raw", string(data)))
	if err := c.writeLine(data); err != nil {
		zap.L().Error("[ACP] 写入 stdin 失败（通知）", zap.String("method", method), zap.Error(err))
	}
}

// sendAndWait 发送请求并等待对应 ID 的响应（通过读循环分发）。
// 若传入非 nil sink，在等待期间临时接管通知（用于捕获 init/load 阶段的通知），
// 返回前恢复原 sink。
func (c *Client) sendAndWait(method string, params any) (json.RawMessage, error) {
	return c.sendAndWaitWithSink(method, params, nil)
}

func (c *Client) sendAndWaitWithSink(method string, params any, sink notifySink) (json.RawMessage, error) {
	id := atomic.AddInt64(&c.requestId, 1)
	ch := c.registerPending(id)

	if sink != nil {
		old := c.setSink(sink)
		defer c.setSink(old)
	}

	if err := c.writeRequest(id, method, params); err != nil {
		c.unregisterPending(id)
		return nil, err
	}

	select {
	case resp := <-ch:
		if resp.Error != nil {
			return nil, fmt.Errorf("ACP 错误 [%d]: %s", resp.Error.Code, resp.Error.FullMessage())
		}
		return resp.Result, nil
	case <-c.doneCh:
		c.unregisterPending(id)
		return nil, fmt.Errorf("未收到 %s 的响应（进程已退出）", method)
	}
}

// handleSessionUpdate 解析并分发 session/update 通知
func (c *Client) handleSessionUpdate(params json.RawMessage) {
	var notif SessionUpdate
	if err := json.Unmarshal(params, &notif); err != nil {
		return
	}

	// 解析 update 的类型
	var base UpdateBase
	if err := json.Unmarshal(notif.Update, &base); err != nil {
		return
	}

	// zap.L().Debug("[ACP] session/update", zap.String("type", base.SessionUpdate))

	switch base.SessionUpdate {
	case "agent_message_chunk":
		var chunk AgentMessageChunk
		if err := json.Unmarshal(notif.Update, &chunk); err == nil {
			c.appendText(chunk.Content.Text)
			c.emit(Event{Type: "text", Content: chunk.Content.Text})
		}

	case "agent_thought_chunk":
		var chunk AgentMessageChunk
		if err := json.Unmarshal(notif.Update, &chunk); err == nil {
			c.appendText(chunk.Content.Text)
			c.emit(Event{Type: "text", Content: chunk.Content.Text})
		}

	case "tool_call":
		var tc ToolCall
		if err := json.Unmarshal(notif.Update, &tc); err == nil {
			zap.L().Debug("[ACP] → tool_call", zap.String("title", tc.Title), zap.String("status", tc.Status), zap.String("id", tc.ToolCallId))
			c.emit(Event{Type: "tool_call", ToolCallId: tc.ToolCallId, ToolName: tc.Title, ToolStatus: tc.Status})
		}

	case "tool_call_update":
		var tcu ToolCallUpdate
		if err := json.Unmarshal(notif.Update, &tcu); err == nil {
			zap.L().Debug("[ACP] → tool_call_update", zap.String("title", tcu.Title), zap.String("status", tcu.Status), zap.String("id", tcu.ToolCallId))
			c.emit(Event{Type: "tool_update", ToolCallId: tcu.ToolCallId, ToolName: tcu.Title, ToolStatus: tcu.Status})
		}

	case "tool_call_chunk":
		// 工具调用的流式参数输出（Kiro 扩展）
		zap.L().Debug("[ACP] → tool_call_chunk (ignored)")

	case "plan":
		zap.L().Debug("[ACP] → plan update")

	default:
		zap.L().Debug("[ACP] → 未知 session/update 类型", zap.String("type", base.SessionUpdate))
	}
}

// handleExtensionNotification 处理 Kiro 扩展通知
func (c *Client) handleExtensionNotification(method string, params json.RawMessage) {
	switch method {
	case "_kiro.dev/commands/available":
		// 可用命令和工具列表
		var data struct {
			SessionId string `json:"sessionId"`
			Commands  []struct {
				Name        string `json:"name"`
				Description string `json:"description"`
			} `json:"commands"`
			Tools []struct {
				Name        string `json:"name"`
				Description string `json:"description"`
				Source      string `json:"source"`
			} `json:"tools"`
		}
		if err := json.Unmarshal(params, &data); err == nil {
			zap.L().Debug("[ACP] 可用命令/工具更新",
				zap.Int("commands", len(data.Commands)),
				zap.Int("tools", len(data.Tools)),
			)
			// 推送工具列表给前端（可选）
			if len(data.Tools) > 0 {
				toolNames := make([]string, 0, len(data.Tools))
				for _, t := range data.Tools {
					toolNames = append(toolNames, t.Name)
				}
				c.emit(Event{Type: "tools_available", Content: toJSONStr(toolNames)})
			}
		}

	case "_kiro.dev/metadata":
		// 上下文使用量和耗时
		var data struct {
			SessionId              string  `json:"sessionId"`
			ContextUsagePercentage float64 `json:"contextUsagePercentage"`
			TurnDurationMs         int64   `json:"turnDurationMs,omitempty"`
		}
		if err := json.Unmarshal(params, &data); err == nil {
			zap.L().Debug("[ACP] metadata",
				zap.Float64("contextUsage%", data.ContextUsagePercentage),
				zap.Int64("turnMs", data.TurnDurationMs),
			)
			// 缓存最近用量/耗时，供首屏读取
			c.setMeta(data.ContextUsagePercentage, data.TurnDurationMs)
			// 上下文用量每条通知都会带（turnDurationMs 仅 turn 结束那条才有），
			// 因此总是转发用量，耗时有值时才附带
			content := map[string]any{
				"contextUsagePercent": data.ContextUsagePercentage,
			}
			if data.TurnDurationMs > 0 {
				content["turnDurationMs"] = data.TurnDurationMs
			}
			c.emit(Event{
				Type:    "metadata",
				Content: toJSONStr(content),
			})
		}

	case "_kiro.dev/subagent/list_update":
		// 子代理列表更新
		zap.L().Debug("[ACP] 子代理列表更新")

	case "_kiro.dev/mcp/server_initialized":
		// MCP Server 初始化完成
		var data struct {
			ServerName string `json:"serverName"`
		}
		json.Unmarshal(params, &data)
		zap.L().Info("[ACP] MCP Server 初始化完成", zap.String("server", data.ServerName))

	case "_kiro.dev/compaction/status":
		// 上下文压缩状态
		zap.L().Debug("[ACP] 上下文压缩中")

	case "_kiro.dev/clear/status":
		// 会话清除状态
		zap.L().Debug("[ACP] 会话清除中")

	default:
		zap.L().Debug("[ACP] 未处理的扩展通知", zap.String("method", method))
	}
}

// toJSONStr 序列化为 JSON 字符串
func toJSONStr(v any) string {
	data, _ := json.Marshal(v)
	return string(data)
}

// applyLineWindow 按 ACP readTextFile 语义对文本做行窗口截取。
//   - line 为 1 基起始行号（nil 或 <=0 视为从第 1 行开始）；
//   - limit 为返回的最大行数（nil 或 <=0 视为不限制，返回到文件末尾）。
//
// 保留行间换行；起始行越界时返回空串。用于避免把整份大文件回传给 Agent。
func applyLineWindow(content string, line, limit *int) string {
	start := 1
	if line != nil && *line > 0 {
		start = *line
	}
	// 无窗口需求：整文件返回
	if start <= 1 && (limit == nil || *limit <= 0) {
		return content
	}

	lines := strings.Split(content, "\n")
	if start > len(lines) {
		return ""
	}
	from := start - 1 // 转 0 基
	to := len(lines)
	if limit != nil && *limit > 0 {
		if end := from + *limit; end < to {
			to = end
		}
	}
	return strings.Join(lines[from:to], "\n")
}

// handleAgentRequest 处理 Agent→Client 的反向请求。
// 该方法由读循环在独立 goroutine 中调用，因此权限等待不会阻塞读循环；
// stdin 写入通过 writeLine（writeMu）串行化，不与 Cancel 争锁。
func (c *Client) handleAgentRequest(rawId json.RawMessage, method string, rawParams json.RawMessage) {
	zap.L().Debug("[ACP] 处理反向请求", zap.String("method", method), zap.String("id", string(rawId)), zap.String("params", string(rawParams[:min(len(rawParams), 200)])))

	var respData []byte

	switch method {
	case "readTextFile":
		// 读取文件：先从 FileStore 查找（MinIO），再从本地文件系统
		var params struct {
			Path  string `json:"path"`
			Line  *int   `json:"line,omitempty"`
			Limit *int   `json:"limit,omitempty"`
		}
		if err := json.Unmarshal(rawParams, &params); err == nil && params.Path != "" {
			var content string
			var readErr error

			// 优先从 FileStore（MinIO）读取：先按路径解析出 object key
			key, keyErr := filestore.FindKeyByPath("upload", c.UserId, c.sessionId, params.Path)
			if keyErr == nil && key != "" {
				content, readErr = filestore.ReadFile(key)
				zap.L().Debug("[ACP] readTextFile 从 MinIO 读取", zap.String("path", params.Path), zap.String("key", key))
			} else {
				// 回退到本地文件系统
				data, err := os.ReadFile(params.Path)
				if err == nil {
					content = string(data)
					zap.L().Debug("[ACP] readTextFile 从本地读取", zap.String("path", params.Path))
				} else {
					readErr = err
				}
			}

			if readErr == nil {
				windowed := applyLineWindow(content, params.Line, params.Limit)
				result, _ := json.Marshal(map[string]string{"content": windowed})
				respData, _ = json.Marshal(map[string]any{
					"jsonrpc": "2.0",
					"id":      rawId,
					"result":  json.RawMessage(result),
				})
			}
		}
		if respData == nil {
			respData, _ = json.Marshal(map[string]any{
				"jsonrpc": "2.0",
				"id":      rawId,
				"error":   map[string]any{"code": -1, "message": "file not found"},
			})
		}

	case "writeTextFile":
		// 写入文件到 MinIO（不写本地）
		var params struct {
			Path    string `json:"path"`
			Content string `json:"content"`
		}
		if err := json.Unmarshal(rawParams, &params); err == nil && params.Path != "" {
			record, err := filestore.UploadFile(c.UserId, c.sessionId, filepath.Base(params.Path), params.Content)
			if err == nil {
				respData, _ = json.Marshal(map[string]any{
					"jsonrpc": "2.0",
					"id":      rawId,
					"result":  map[string]any{},
				})
				_ = record // 文件下载通过前端标记检测实现，不再推送事件
			} else {
				zap.L().Error("[ACP] writeTextFile MinIO上传失败", zap.Error(err))
			}
		}
		if respData == nil {
			respData, _ = json.Marshal(map[string]any{
				"jsonrpc": "2.0",
				"id":      rawId,
				"error":   map[string]any{"code": -1, "message": "write failed"},
			})
		}

	case "requestPermission", "session/request_permission":
		// 解析权限请求参数
		var params struct {
			ToolCall *struct {
				Title string `json:"title"`
			} `json:"toolCall"`
			Options []struct {
				OptionId string `json:"optionId"`
				Name     string `json:"name"`
				Kind     string `json:"kind"`
			} `json:"options"`
		}
		json.Unmarshal(rawParams, &params)

		title := ""
		if params.ToolCall != nil {
			title = params.ToolCall.Title
		}

		// 当前策略：所有权限请求都推给前端授权。
		options := make([]PermissionOption, 0, len(params.Options))
		for _, opt := range params.Options {
			options = append(options, PermissionOption{
				OptionId: opt.OptionId,
				Name:     opt.Name,
				Kind:     opt.Kind,
			})
		}

		requestId := fmt.Sprintf("perm_%s", string(rawId))
		pushed := c.pushPermissionEvent(Event{
			Type: "permission",
			Permission: &PermissionEvent{
				RequestId: requestId,
				Title:     title,
				Options:   options,
			},
		})

		if pushed {
			// 丢弃可能残留的旧授权结果，避免消费到上一次请求/Close 写入的值
			select {
			case <-c.permissionCh:
			default:
			}

			// 有界等待前端回复：超时 / 进程退出 时按 deny 处理，避免永久阻塞。
			// 本 goroutine 阻塞不影响读循环（读循环在另一 goroutine）。
			selectedOptionId := c.selectDenyOption(params.Options)
			timer := time.NewTimer(PermissionWaitTimeout)
			select {
			case selectedOptionId = <-c.permissionCh:
			case <-timer.C:
				zap.L().Warn("[ACP] 权限等待超时，默认拒绝", zap.String("title", title))
			case <-c.doneCh:
				zap.L().Warn("[ACP] 权限等待期间进程已退出", zap.String("title", title))
			}
			timer.Stop()

			respData, _ = json.Marshal(map[string]any{
				"jsonrpc": "2.0",
				"id":      rawId,
				"result": map[string]any{
					"outcome": map[string]any{
						"outcome":  "selected",
						"optionId": selectedOptionId,
					},
				},
			})
		} else {
			// 无事件通道（非 prompt 阶段）：自动允许
			respData, _ = json.Marshal(map[string]any{
				"jsonrpc": "2.0",
				"id":      rawId,
				"result": map[string]any{
					"outcome": map[string]any{
						"outcome":  "selected",
						"optionId": "allow_once",
					},
				},
			})
		}

	default:
		// 未知请求返回错误
		respData, _ = json.Marshal(map[string]any{
			"jsonrpc": "2.0",
			"id":      rawId,
			"error":   map[string]any{"code": -32601, "message": "method not supported"},
		})
	}

	if respData != nil {
		zap.L().Debug("[ACP] 发送反向响应", zap.String("method", method), zap.String("resp", string(respData[:min(len(respData), 200)])))
		if err := c.writeLine(respData); err != nil {
			zap.L().Error("[ACP] 写入反向响应失败", zap.String("method", method), zap.Error(err))
		}
	} else {
		zap.L().Warn("[ACP] 反向请求无响应数据", zap.String("method", method))
	}
}

// selectDenyOption 从可选项中挑选一个 deny 语义的 optionId 作为超时/退出时的默认值。
// 找不到显式 deny 选项时回退到字面量 "deny"。
func (c *Client) selectDenyOption(options []struct {
	OptionId string `json:"optionId"`
	Name     string `json:"name"`
	Kind     string `json:"kind"`
}) string {
	for _, opt := range options {
		if strings.Contains(strings.ToLower(opt.Kind), "deny") || strings.Contains(strings.ToLower(opt.OptionId), "deny") {
			return opt.OptionId
		}
	}
	return "deny"
}