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"

	"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 实现 ============

// Client ACP 客户端（管理单个 Agent 进程）
type Client struct {
	cmd       *exec.Cmd
	stdin     io.WriteCloser
	scanner   *bufio.Scanner
	sessionId string
	requestId int64
	mu        sync.Mutex // 保护 stdin 写入
	readMu    sync.Mutex // 保护 stdout 读取

	// 权限请求等待机制
	permissionCh chan string  // 接收前端选择的 optionId
	eventCh      chan<- Event // 当前 prompt 的事件通道（用于推送 permission 事件）

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

	// 文本累积（用于合并日志输出）
	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)
}

// 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...)

	// 若配置了降权用户，则以该用户身份拉起子进程
	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

	return &Client{
		cmd:          cmd,
		stdin:        stdin,
		scanner:      scanner,
		permissionCh: make(chan string, 1),
	}, nil
}

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

// 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 紧跟发送的初始用量 metadata 通知
	c.drainInitialMeta()

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

// drainInitialMeta 读取 session/new 之后紧跟的一批通知，捕获其中的初始上下文用量。
// Kiro 在 session/new 响应后会发送 commands/available、subagent/list_update、metadata 等
// 一批通知，且 metadata（含 contextUsagePercentage）位于这批通知的末尾。这里读取直到
// 遇到 metadata 通知即停止，避免读空后阻塞。设置行数上限作为安全兜底。
func (c *Client) drainInitialMeta() {
	const maxLines = 30
	for i := 0; i < maxLines; i++ {
		if !c.scanner.Scan() {
			return
		}
		line := c.scanner.Text()
		if line == "" {
			continue
		}

		var msg struct {
			Method string          `json:"method,omitempty"`
			Params json.RawMessage `json:"params,omitempty"`
			ID     json.RawMessage `json:"id,omitempty"`
		}
		if err := json.Unmarshal([]byte(line), &msg); err != nil {
			continue
		}

		if msg.Method == "_kiro.dev/metadata" {
			var data struct {
				ContextUsagePercentage float64 `json:"contextUsagePercentage"`
				TurnDurationMs         int64   `json:"turnDurationMs,omitempty"`
			}
			if err := json.Unmarshal(msg.Params, &data); err == nil {
				zap.L().Debug("[ACP] 初始 metadata", zap.Float64("contextUsage%", data.ContextUsagePercentage))
				c.setMeta(data.ContextUsagePercentage, data.TurnDurationMs)
			}
			return
		}
		// 其他 pre-prompt 通知（commands/available、subagent 等）忽略
	}
	zap.L().Debug("[ACP] drainInitialMeta 未在上限内捕获到 metadata")
}

// 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 请求
	id := c.sendRequest("session/load", params)

	// 收集回放消息
	var replayMessages []ReplayMessage
	currentMsg := &ReplayMessage{}

	// 读取回放事件直到收到 session/load 的 response
	for c.scanner.Scan() {
		line := c.scanner.Text()
		if line == "" {
			continue
		}

		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 {
			continue
		}

		// 检查是否是 session/load 的响应
		hasId := len(msg.ID) > 0 && string(msg.ID) != "null"
		if hasId && msg.Method == "" {
			var numId int64
			json.Unmarshal(msg.ID, &numId)
			if numId == id {
				// session/load 完成
				if msg.Error != nil {
					return "", "", "", "", nil, fmt.Errorf("session/load 错误: %s", msg.Error.FullMessage())
				}
				c.sessionId = sessionId
				// 追加最后一条未关闭的消息
				if currentMsg.Content != "" {
					replayMessages = append(replayMessages, *currentMsg)
				}

				// 解析响应中的 welcomeMessage、实际生效的 agentId 和 modelId
				var welcomeMsg string
				var agentId string
				var modelId string
				if len(msg.Result) > 0 {
					var res NewSessionResult
					if err := json.Unmarshal(msg.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 通知
				c.drainInitialMeta()

				return sessionId, welcomeMsg, agentId, modelId, replayMessages, nil
			}
		}

		// 处理回放的 session/update 通知
		if msg.Method == "session/update" || msg.Method == "_kiro.dev/session/update" {
			var notif SessionUpdate
			if err := json.Unmarshal(msg.Params, &notif); err != nil {
				continue
			}
			var base UpdateBase
			if err := json.Unmarshal(notif.Update, &base); err != nil {
				continue
			}

			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":
				// AI 消息
				var chunk AgentMessageChunk
				json.Unmarshal(notif.Update, &chunk)
				if currentMsg.Role != "assistant" || (chunk.MessageId != "" && chunk.MessageId != currentMsg.MessageId) {
					// 新的 AI 消息 → 保存上一条
					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":
				// 思考内容忽略（不显示给用户）

			case "tool_call", "tool_call_update", "tool_call_chunk":
				// 工具调用回放忽略

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

		// 忽略其他通知（kiro 扩展等）
	}

	return "", "", "", "", nil, fmt.Errorf("session/load 未收到响应（进程可能已退出）")
}

// Prompt 发送消息并流式读取事件
// eventCh 会在 turn 结束或出错时关闭
func (c *Client) Prompt(sessionId string, message string, eventCh chan<- Event) {
	defer close(eventCh)

	// 设置当前事件通道（供 handleAgentRequest 推送 permission 事件）
	c.eventCh = eventCh

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

	// 发送 prompt 请求
	promptId := c.sendRequest("session/prompt", params)
	zap.L().Debug("[ACP] prompt 已发送", zap.Int64("promptId", promptId), zap.String("sessionId", sessionId))

	// 持续读取直到收到 prompt 的 response
	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
		}

		// 解析 ID（兼容 int64 和 string）
		hasId := len(msg.ID) > 0 && string(msg.ID) != "null"
		var numericId int64
		isPromptResp := false
		if hasId {
			json.Unmarshal(msg.ID, &numericId)
			isPromptResp = numericId == promptId && msg.Method == ""
		}

		// 收到 prompt 的响应 → turn 结束
		if isPromptResp {
			zap.L().Debug("[ACP] 收到 prompt 响应（turn结束）", zap.Int64("id", numericId))
			// 打印本轮完整回复
			if c.textBuffer != "" {
				zap.L().Info("[ACP] AI 回复完整内容", zap.String("text", c.textBuffer))
				c.textBuffer = ""
			}
			var result PromptResult
			if msg.Error != nil {
				eventCh <- Event{Type: "error", Content: msg.Error.FullMessage()}
			} else if err := json.Unmarshal(msg.Result, &result); err == nil {
				eventCh <- Event{Type: "done", SessionId: sessionId, StopReason: result.StopReason}
			}
			return
		}

		// 处理 notification (session/update 或 _kiro.dev/session/update)
		if !hasId && (msg.Method == "session/update" || msg.Method == "_kiro.dev/session/update") {
			c.handleSessionUpdate(msg.Params, eventCh)
			continue
		}

		// 处理 Agent→Client 反向请求（有 ID + Method = 请求）
		if hasId && msg.Method != "" {
			zap.L().Debug("[ACP] 收到 Agent 反向请求", zap.String("method", msg.Method), zap.String("id", string(msg.ID)))
			c.handleAgentRequest(msg.ID, msg.Method, msg.Params)
			continue
		}

		// 未匹配的消息（kiro扩展通知等）
		if msg.Method != "" {
			c.handleExtensionNotification(msg.Method, msg.Params, eventCh)
		} else {
			zap.L().Debug("[ACP] 未处理的消息", zap.String("line", line[:min(len(line), 200)]))
		}
	}

	// Scanner 结束说明进程退出
	eventCh <- 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()

	// 终止进程
	if c.cmd.Process != nil {
		c.cmd.Process.Kill()
		// Wait 回收子进程资源，防止僵尸进程
		go c.cmd.Wait()
	}

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

// IsAlive 检查 Agent 进程是否存活
func (c *Client) IsAlive() bool {
	return c.cmd.ProcessState == nil
}

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

// sendRequest 发送请求（不等待响应），返回 request ID
func (c *Client) sendRequest(method string, params any) int64 {
	c.mu.Lock()
	defer c.mu.Unlock()

	id := atomic.AddInt64(&c.requestId, 1)
	req := Request{
		JSONRPC: "2.0",
		ID:      id,
		Method:  method,
		Params:  params,
	}

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

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

	data = append(data, '\n')
	if _, err := c.stdin.Write(data); err != nil {
		zap.L().Error("[ACP] 写入 stdin 失败", zap.String("method", method), zap.Error(err))
	}

	return id
}

// sendNotification 发送 JSON-RPC Notification（不带 id，不期望响应）
func (c *Client) sendNotification(method string, params any) {
	c.mu.Lock()
	defer c.mu.Unlock()

	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)))

	data = append(data, '\n')
	if _, err := c.stdin.Write(data); err != nil {
		zap.L().Error("[ACP] 写入 stdin 失败（通知）", zap.String("method", method), zap.Error(err))
	}
}

// sendAndWait 发送请求并等待对应 ID 的响应
func (c *Client) sendAndWait(method string, params any) (json.RawMessage, error) {
	id := c.sendRequest(method, params)

	// 读取响应（跳过 notification 和其他消息）
	for c.scanner.Scan() {
		line := c.scanner.Text()
		if line == "" {
			continue
		}

		var resp Response
		if err := json.Unmarshal([]byte(line), &resp); err != nil {
			continue
		}

		// 匹配 ID 的响应
		if resp.ID != nil && *resp.ID == id {
			if resp.Error != nil {
				return nil, fmt.Errorf("ACP 错误 [%d]: %s", resp.Error.Code, resp.Error.FullMessage())
			}
			return resp.Result, nil
		}

		// 跳过 notification（session/new 之前可能收到 kiro 扩展通知）
	}

	return nil, fmt.Errorf("未收到 %s 的响应（进程可能已退出）", method)
}

// handleSessionUpdate 解析并分发 session/update 通知
func (c *Client) handleSessionUpdate(params json.RawMessage, eventCh chan<- Event) {
	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.textBuffer += chunk.Content.Text
			eventCh <- Event{Type: "text", Content: chunk.Content.Text}
		}

	case "agent_thought_chunk":
		var chunk AgentMessageChunk
		if err := json.Unmarshal(notif.Update, &chunk); err == nil {
			c.textBuffer += chunk.Content.Text
			eventCh <- 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))
			eventCh <- 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))
			eventCh <- 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, eventCh chan<- Event) {
	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)
				}
				eventCh <- 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
			}
			eventCh <- 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)
}

// handleAgentRequest 处理 Agent→Client 的反向请求
func (c *Client) handleAgentRequest(rawId json.RawMessage, method string, rawParams json.RawMessage) {
	c.mu.Lock()
	defer c.mu.Unlock()

	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 {
				result, _ := json.Marshal(map[string]string{"content": content})
				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
		}

		// 判断是否涉及写操作（需要手动授权）
		isWriteOp := true
		// isWriteOp := false
		// if title != "" {
		// 	for _, keyword := range []string{"write", "Write", "create", "Create", "delete", "Delete", "修改", "写入", "创建", "删除"} {
		// 		if len(title) >= len(keyword) && containsStr(title, keyword) {
		// 			isWriteOp = true
		// 			break
		// 		}
		// 	}
		// }

		if isWriteOp && c.eventCh != nil {
			// 写操作：推送权限请求给前端，等待前端回复
			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))
			c.eventCh <- Event{
				Type: "permission",
				Permission: &PermissionEvent{
					RequestId: requestId,
					Title:     title,
					Options:   options,
				},
			}

			// 释放 mu 锁后等待前端回复（否则会死锁）
			c.mu.Unlock()
			selectedOptionId := <-c.permissionCh
			c.mu.Lock()

			respData, _ = json.Marshal(map[string]any{
				"jsonrpc": "2.0",
				"id":      rawId,
				"result": map[string]any{
					"outcome": map[string]any{
						"outcome":  "selected",
						"optionId": selectedOptionId,
					},
				},
			})
		} else {
			// 非写操作：自动允许
			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)])))
		respData = append(respData, '\n')
		c.stdin.Write(respData)
	} else {
		zap.L().Warn("[ACP] 反向请求无响应数据", zap.String("method", method))
	}
}

// containsStr 简单字符串包含检测
func containsStr(s, substr string) bool {
	return len(s) >= len(substr) && (s == substr || len(s) > 0 && stringContains(s, substr))
}

func stringContains(s, substr string) bool {
	for i := 0; i <= len(s)-len(substr); i++ {
		if s[i:i+len(substr)] == substr {
			return true
		}
	}
	return false
}

func min(a, b int) int {
	if a < b {
		return a
	}
	return b
}