// turn.go 实现 TurnTracker:单 relay 维度的回合状态机,外加 stopReason 解析辅助。 // // 回合(turn)定义:一次 session/prompt 请求到其响应之间的窗口。relay reader 在解析到 // - session/update 通知 → OnUpdate(累加 update_count,内存计数,完成时一次性写库) // - 该 prompt 的字符串-id 响应 → OnPromptResponse(标完成 + 写 last_stop_reason + 发事件) // // MVP 假设:同一 session 同时只有一个开放回合。若在已有开放回合时收到新的 StartTurn, // 自动中止前一个(auto-abort),避免悬挂。所有状态由 mu 保护;Abort 只触碰 repo+bus, // 不持有 supervisor 锁,故可安全地从 onExit 调用(不违反 onExit-must-not-relock 不变量)。 package acp import ( "context" "encoding/json" "log/slog" "sync" "time" "github.com/google/uuid" "github.com/yan1h/agent-coding-workflow/internal/acp/handlers" ) // TurnRepo 是 TurnTracker 需要的最小 repo 子集,由 acp.Repository 满足。 type TurnRepo interface { InsertTurn(ctx context.Context, t *Turn) (*Turn, error) MarkTurnCompleted(ctx context.Context, sessionID uuid.UUID, turnIndex int, stop string, updateCount int) error MarkTurnAborted(ctx context.Context, sessionID uuid.UUID, turnIndex int) error GetLatestTurnBySession(ctx context.Context, sessionID uuid.UUID) (*Turn, error) UpdateSessionLastStopReason(ctx context.Context, sessionID uuid.UUID, reason string) error } // TurnTracker 是单 relay 的回合状态机。 type TurnTracker struct { repo TurnRepo bus TurnBus sess handlers.SessionContext log *slog.Logger mu sync.Mutex open bool // 是否有开放回合 turnIndex int // 当前/最近回合的 index promptID string // 当前开放回合的 prompt 请求 id(用于响应相关联) updateCount int // 当前回合的 session/update 计数(内存累计,完成时落库) nextIndex int // 下一个回合的 index(构造时由 GetLatestTurnBySession 播种) seeded bool // nextIndex 是否已播种 } // NewTurnTracker 构造一个 TurnTracker。bus / repo 为 nil 时方法均为安全 no-op // (部分测试可不注入)。 func NewTurnTracker(repo TurnRepo, bus TurnBus, sess handlers.SessionContext, log *slog.Logger) *TurnTracker { if log == nil { log = slog.Default() } return &TurnTracker{repo: repo, bus: bus, sess: sess, log: log} } // seedLocked 懒播种 nextIndex:从 DB 取最近回合 index+1。失败时从 0 开始(best-effort)。 // 调用方必须已持有 t.mu。 func (t *TurnTracker) seedLocked(ctx context.Context) { if t.seeded { return } t.seeded = true if t.repo == nil { return } latest, err := t.repo.GetLatestTurnBySession(ctx, t.sess.SessionID) if err != nil { t.log.Warn("acp.turn.seed_index", "session_id", t.sess.SessionID, "err", err.Error()) return } if latest != nil { t.nextIndex = latest.TurnIndex + 1 } } // StartTurn 在发送 session/prompt 之前调用,登记一个 in_progress 回合并把 promptRequestID // 与之关联。若已有开放回合,先 auto-abort 之(MVP 单回合假设)。返回新回合的 index。 func (t *TurnTracker) StartTurn(ctx context.Context, promptRequestID string) (int, error) { if t == nil { return 0, nil } t.mu.Lock() defer t.mu.Unlock() t.seedLocked(ctx) // 已有开放回合 → auto-abort,避免悬挂。 if t.open { t.abortLocked(ctx, string(StopCancelled)) } idx := t.nextIndex if t.repo != nil { var pid *string if promptRequestID != "" { p := promptRequestID pid = &p } if _, err := t.repo.InsertTurn(ctx, &Turn{ SessionID: t.sess.SessionID, TurnIndex: idx, PromptRequestID: pid, Status: TurnInProgress, }); err != nil { return 0, err } } t.open = true t.turnIndex = idx t.promptID = promptRequestID t.updateCount = 0 t.nextIndex = idx + 1 return idx, nil } // OnUpdate 在收到一条 session/update 通知时调用,累加当前开放回合的 update_count // (内存计数,避免每 chunk 一次 UPDATE;完成时一次性落库)。无开放回合时 no-op。 func (t *TurnTracker) OnUpdate(_ context.Context) { if t == nil { return } t.mu.Lock() defer t.mu.Unlock() if t.open { t.updateCount++ } } // IsTrackedPrompt 报告 id 是否为当前开放回合关联的 prompt 请求 id。relay reader 用它 // 判断一条字符串-id 响应是否应触发 OnPromptResponse(而非沿用既有 fanout 路径)。 func (t *TurnTracker) IsTrackedPrompt(id string) bool { if t == nil || id == "" { return false } t.mu.Lock() defer t.mu.Unlock() return t.open && t.promptID == id } // OnPromptResponse 在收到匹配当前回合的 session/prompt 响应时调用:标完成、写 // session.last_stop_reason、发布 turn_completed 与 session_idle 事件。id 不匹配时 no-op。 func (t *TurnTracker) OnPromptResponse(ctx context.Context, promptRequestID string, rawStopReason string) { if t == nil { return } t.mu.Lock() if !t.open || t.promptID != promptRequestID { t.mu.Unlock() return } idx := t.turnIndex updateCount := t.updateCount t.open = false t.promptID = "" t.mu.Unlock() norm := NormalizeStopReason(rawStopReason) completedAt := time.Now() if t.repo != nil { if err := t.repo.MarkTurnCompleted(ctx, t.sess.SessionID, idx, rawStopReason, updateCount); err != nil { t.log.Error("acp.turn.mark_completed", "session_id", t.sess.SessionID, "turn_index", idx, "err", err.Error()) } if err := t.repo.UpdateSessionLastStopReason(ctx, t.sess.SessionID, rawStopReason); err != nil { t.log.Error("acp.turn.update_last_stop_reason", "session_id", t.sess.SessionID, "err", err.Error()) } } if t.bus != nil { base := TurnEvent{ SessionID: t.sess.SessionID, UserID: t.sess.UserID, WorkspaceID: t.sess.WorkspaceID, TurnIndex: idx, StopReason: norm, RawStopReason: rawStopReason, CompletedAt: completedAt, } completed := base completed.Kind = TurnCompletedEvent t.bus.Publish(ctx, completed) idle := base idle.Kind = SessionIdleEvent t.bus.Publish(ctx, idle) } } // Abort 中止当前开放回合(如果有)并发布 session_idle 事件。从 supervisor.onExit // 在子进程退出时调用,确保崩溃/被杀时也能闭合悬挂回合。无开放回合时 no-op。 func (t *TurnTracker) Abort(ctx context.Context, reason string) { if t == nil { return } t.mu.Lock() defer t.mu.Unlock() if !t.open { return } t.abortLocked(ctx, reason) } // abortLocked 闭合当前开放回合并发布 session_idle。调用方必须已持有 t.mu。 func (t *TurnTracker) abortLocked(ctx context.Context, reason string) { idx := t.turnIndex t.open = false t.promptID = "" if t.repo != nil { if err := t.repo.MarkTurnAborted(ctx, t.sess.SessionID, idx); err != nil { t.log.Error("acp.turn.mark_aborted", "session_id", t.sess.SessionID, "turn_index", idx, "err", err.Error()) } } if t.bus != nil { norm := NormalizeStopReason(reason) t.bus.Publish(ctx, TurnEvent{ SessionID: t.sess.SessionID, UserID: t.sess.UserID, WorkspaceID: t.sess.WorkspaceID, TurnIndex: idx, Kind: SessionIdleEvent, StopReason: norm, RawStopReason: reason, CompletedAt: time.Now(), }) } } // ParseStopReason 从 session/prompt 响应的 Result JSON 中提取 stopReason 字段。 // 字段缺失或解析失败时返回空串(调用方归一化为 StopOther)。 func ParseStopReason(result json.RawMessage) string { if len(result) == 0 { return "" } var r struct { StopReason string `json:"stopReason"` } if err := json.Unmarshal(result, &r); err != nil { return "" } return r.StopReason }