Agent-Squad/squad/groupchat.go

package squad

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"time"
)

// groupchat.go — group_chat 拓扑驱动（Q2 WP3）。
// 形态（计划 §三 WP3 与边界 §五）：共享转录（BB 字段 transcript +
// group_chat/transcript.md 全量落盘）；发言人策略只做**轮转**与**主持人**
// 两种（脚本可判定，LLM 选人策略不进库）；预算停机（max_turns，缺省 8）。
// 蓝图步骤 = 参与者名册（按序，不直接执行，driver 标记 skipped 并派生轮次
// 步骤）；任一发言含 [close] 标记即收口；主持人模式下主持人输出 JSON 决策
// （{"next":"role"} / {"action":"close"}），未知角色/非法 JSON fail-closed。
// 驱动幂等可恢复：轮次由已落盘的 gc 步骤数重建，挂起（权限/HITL）后
// Respond 重入续转。

func init() {
	_ = RegisterTopologyDriver(TopologyGroupChat)
}

// DefaultMaxGroupChatTurns 群聊轮次预算缺省值（预算停机）。
const DefaultMaxGroupChatTurns = 8

// gcRosterSkipReason 名册步骤的 skipped 原因（非执行步骤，不算降级）。
const gcRosterSkipReason = "group_chat: 名册步骤（参与者由驱动接管）"

// gcCloseMarker 任一发言包含该标记（大小写不敏感）即收口。
const gcCloseMarker = "[close]"

// gcTranscriptCapBB 是 BB 转录字段的容量上限（TASK.md 快照只带近端）。
const gcTranscriptCapBB = 4000

// runGroupChatDriver group_chat 拓扑入口。
func (e *SquadEngine) runGroupChatDriver(ctx context.Context, st *RunState) (*RunResult, error) {
	drv := &flowDriver{opts: e.deps}

	// 名册：首次进入从蓝图步骤取参与者序列并持久化到元数据（恢复时直读）。
	roster, err := e.gcRoster(st)
	if err != nil {
		return e.driverFail(st, err.Error()), nil
	}
	policy := orDefault(st.Meta.Metadata["group_chat_policy"], "rotate")
	moderator := st.Meta.Metadata["group_chat_moderator"]
	if policy != "rotate" && policy != "moderator" {
		return e.driverFail(st, fmt.Sprintf("group_chat: 未知发言人策略 %q（只支持 rotate/moderator）", policy)), nil
	}
	if policy == "moderator" {
		if moderator == "" || !containsString(roster, moderator) {
			return e.driverFail(st, fmt.Sprintf("group_chat: 主持人 %q 不在名册内（fail-closed）", moderator)), nil
		}
	}
	maxTurns := DefaultMaxGroupChatTurns
	if v := st.Meta.Metadata["group_chat_max_turns"]; v != "" {
		if n := parseIntDefault(v, DefaultMaxGroupChatTurns); n > 0 {
			maxTurns = n
		}
	}

	// 名册步骤标记 skipped（首次进入时；恢复重入无 pending 名册步骤）。
	st.mu.Lock()
	for i := range st.Plan.Workflow.Steps {
		s := &st.Plan.Workflow.Steps[i]
		if !strings.HasPrefix(s.ID, "gc") && (s.Status == "pending" || s.Status == "running") {
			s.Status = "skipped"
			s.Error = gcRosterSkipReason
		}
	}
	_ = st.save()
	st.mu.Unlock()

	for {
		// 恢复重入：上次挂起留下的未完 gc 轮次步骤优先续跑（不重新选人）。
		if idx := e.gcHeldStepIdx(st); idx >= 0 {
			out, gate, err := e.gcRunStep(ctx, drv, st, idx)
			if err != nil {
				return nil, err
			}
			if gate != nil {
				if out == nil {
					out = heldResult(st, gate)
				}
				return out, nil
			}
			if out != nil {
				return out, nil
			}
			if e.gcMessageClosed(st, idx) {
				drv.emit(st, Event{Type: EvPhase, Text: fmt.Sprintf("group_chat：%s 发出收口标记，停机", st.Plan.Workflow.Steps[idx].Role)})
				break
			}
			continue
		}
		turns := e.gcTurnCount(st)
		if turns >= maxTurns {
			drv.emit(st, Event{Type: EvPhase, Text: fmt.Sprintf("group_chat：轮次预算 %d 用尽，停机收口", maxTurns)})
			break
		}
		speaker, closed, derr := e.gcNextSpeaker(st, roster, policy, moderator, turns)
		if derr != nil {
			return e.driverFail(st, derr.Error()), nil
		}
		if closed {
			drv.emit(st, Event{Type: EvPhase, Text: "group_chat：主持人收口"})
			break
		}
		out, gate, err := e.gcRunTurn(ctx, drv, st, turns, speaker)
		if err != nil {
			return nil, err
		}
		if gate != nil {
			if out == nil {
				out = heldResult(st, gate)
			}
			return out, nil
		}
		if out != nil { // 轮次步骤失败（关键）→ fail-closed 传播
			return out, nil
		}
		if e.gcMessageClosed(st, len(st.Plan.Workflow.Steps)-1) {
			drv.emit(st, Event{Type: EvPhase, Text: fmt.Sprintf("group_chat：%s 发出收口标记，停机", speaker)})
			break
		}
	}
	return e.gcFinish(st, drv), nil
}

// gcRoster 参与者名册（首次从蓝图步骤派生并写入元数据；恢复直读）。
func (e *SquadEngine) gcRoster(st *RunState) ([]string, error) {
	if v := st.Meta.Metadata["group_chat_roster"]; v != "" {
		return strings.Split(v, ","), nil
	}
	var roster []string
	for _, s := range st.Plan.Workflow.Steps {
		if s.Role != "" && !containsString(roster, s.Role) {
			roster = append(roster, s.Role)
		}
	}
	if len(roster) == 0 {
		return nil, fmt.Errorf("group_chat: 名册为空（蓝图步骤即参与者）")
	}
	st.mu.Lock()
	if st.Meta.Metadata == nil {
		st.Meta.Metadata = map[string]string{}
	}
	st.Meta.Metadata["group_chat_roster"] = strings.Join(roster, ",")
	_ = st.save()
	st.mu.Unlock()
	return roster, nil
}

// gcTurnCount 已完成的轮次数（预算与轮转序号的基准；恢复重入自然续算）。
func (e *SquadEngine) gcTurnCount(st *RunState) int {
	n := 0
	for _, s := range st.Plan.Workflow.Steps {
		if strings.HasPrefix(s.ID, "gc") && s.Status == "completed" {
			n++
		}
	}
	return n
}

// gcHeldStepIdx 返回上次挂起留下的未完 gc 轮次步骤下标（无则 -1）。
func (e *SquadEngine) gcHeldStepIdx(st *RunState) int {
	for i := range st.Plan.Workflow.Steps {
		s := &st.Plan.Workflow.Steps[i]
		if strings.HasPrefix(s.ID, "gc") && (s.Status == "pending" || s.Status == "running") {
			return i
		}
	}
	return -1
}

// gcNextSpeaker 决定下一发言人。rotate = 轮转；moderator = 主持人裁决
// （上一步是参与者发言 → 主持人接手；主持人决策 close → 收口，next → 指名，
// 非法 JSON / 未知角色 fail-closed）。
func (e *SquadEngine) gcNextSpeaker(st *RunState, roster []string, policy, moderator string, turns int) (string, bool, error) {
	if policy == "rotate" {
		return roster[turns%len(roster)], false, nil
	}
	last := e.gcLastStep(st)
	if last == nil || last.Role == moderator {
		// 首轮或上一步就是主持人（不该连续，防御性让主持人接手）。
		if last != nil && last.Role == moderator {
			// 解析主持人决策。
			dec, err := e.gcParseDecision(st, last)
			if err != nil {
				return "", false, err
			}
			if dec.Action == "close" {
				return "", true, nil
			}
			if !containsString(roster, dec.Next) {
				return "", false, fmt.Errorf("group_chat: 主持人指名 %q 不在名册内（fail-closed）", dec.Next)
			}
			return dec.Next, false, nil
		}
		return moderator, false, nil
	}
	// 上一步是参与者发言 → 主持人接手裁决。
	return moderator, false, nil
}

type gcDecision struct {
	Action string `json:"action"`
	Next   string `json:"next"`
}

// gcParseDecision 宽松解析主持人 message.md 里的 JSON 决策（首 `{` 末 `}`）。
func (e *SquadEngine) gcParseDecision(st *RunState, step *StepSpec) (*gcDecision, error) {
	msg := e.gcMessageOf(st, step)
	start := strings.Index(msg, "{")
	end := strings.LastIndex(msg, "}")
	if start < 0 || end <= start {
		return nil, fmt.Errorf("group_chat: 主持人输出无 JSON 决策（fail-closed）：%q", truncateStr(msg, 80))
	}
	var dec gcDecision
	if err := json.Unmarshal([]byte(msg[start:end+1]), &dec); err != nil {
		return nil, fmt.Errorf("group_chat: 主持人决策解析失败（fail-closed）: %v", err)
	}
	if dec.Action == "close" {
		return &dec, nil
	}
	if strings.TrimSpace(dec.Next) == "" {
		return nil, fmt.Errorf("group_chat: 主持人决策缺 next/action（fail-closed）")
	}
	return &dec, nil
}

// gcRunTurn 派生一轮发言步骤并执行。
func (e *SquadEngine) gcRunTurn(ctx context.Context, drv *flowDriver, st *RunState, turns int, speaker string) (*RunResult, *HitlGate, error) {
	step := StepSpec{
		ID:           fmt.Sprintf("gc%02d", turns+1),
		Role:         speaker,
		Title:        fmt.Sprintf("群聊第 %d 轮：%s 发言（把发言写入 message.md；同意收口就写 [close]）", turns+1, speaker),
		Targets:      []string{"message.md"},
		SubWorkspace: fmt.Sprintf("group_chat/turn_%02d_%s", turns+1, speaker),
		Status:       "pending",
	}
	st.mu.Lock()
	st.Plan.Workflow.Steps = append(st.Plan.Workflow.Steps, step)
	idx := len(st.Plan.Workflow.Steps) - 1
	_ = st.save()
	st.mu.Unlock()
	return e.gcRunStep(ctx, drv, st, idx)
}

// gcRunStep 执行一个轮次步骤：过门 → runOneStep → 转录并入共享状态。
// 挂起（gate）时步骤保持未完，恢复重入经 gcHeldStepIdx 续跑同一步骤。
func (e *SquadEngine) gcRunStep(ctx context.Context, drv *flowDriver, st *RunState, idx int) (*RunResult, *HitlGate, error) {
	out, gate, err := drv.gateBeforeRun(ctx, st, idx)
	if err != nil || gate != nil || out != nil {
		return out, gate, err
	}
	step := st.Plan.Workflow.Steps[idx]
	drv.emit(st, Event{Type: EvPhase, StepID: step.ID, Role: step.Role,
		Text: fmt.Sprintf("group_chat：%s 发言（第 %d 轮）", step.Role, e.gcTurnCount(st)+1)})
	out, gate, err = drv.runOneStep(ctx, st, idx)
	if err != nil || gate != nil || out != nil {
		return out, gate, err
	}
	// 转录并入共享状态：全量落盘 transcript.md，近端进 BB 字段（下一位
	// 发言人的 TASK.md 黑板快照可见）。
	msg := e.gcMessageOf(st, &st.Plan.Workflow.Steps[idx])
	e.gcAppendTranscript(st, e.gcTurnCount(st), step.Role, msg)
	return nil, nil, nil
}

// gcAppendTranscript 追加转录（全量文件 + BB 近端快照）。
func (e *SquadEngine) gcAppendTranscript(st *RunState, turn int, speaker, msg string) {
	dir := filepath.Join(st.Meta.WorkDir, "group_chat")
	_ = os.MkdirAll(dir, 0o755)
	line := fmt.Sprintf("%d. %s：%s\n", turn, speaker, strings.TrimSpace(msg))
	f, _ := os.OpenFile(filepath.Join(dir, "transcript.md"), os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
	if f != nil {
		_, _ = f.WriteString(line)
		_ = f.Close()
	}
	st.mu.Lock()
	if st.BB.Fields == nil {
		st.BB.Fields = map[string]string{}
	}
	cur := st.BB.Fields["transcript"] + line
	if len(cur) > gcTranscriptCapBB {
		cur = cur[len(cur)-gcTranscriptCapBB:]
	}
	st.BB.Fields["transcript"] = cur
	_ = st.save()
	st.mu.Unlock()
}

// gcLastStep 最后一个已完成的派生轮次步骤（主持人决策解析的锚点）。
func (e *SquadEngine) gcLastStep(st *RunState) *StepSpec {
	for i := len(st.Plan.Workflow.Steps) - 1; i >= 0; i-- {
		s := &st.Plan.Workflow.Steps[i]
		if strings.HasPrefix(s.ID, "gc") && s.Status == "completed" {
			return s
		}
	}
	return nil
}

// gcMessageOf 读取某轮的 message.md（缺文件 = 空串）。
func (e *SquadEngine) gcMessageOf(st *RunState, step *StepSpec) string {
	data, err := os.ReadFile(filepath.Join(st.Meta.WorkDir, filepath.FromSlash(step.SubWorkspace), "message.md"))
	if err != nil {
		return ""
	}
	return string(data)
}

// gcMessageClosed 第 idx 步的发言是否带收口标记。
func (e *SquadEngine) gcMessageClosed(st *RunState, idx int) bool {
	if idx < 0 || idx >= len(st.Plan.Workflow.Steps) {
		return false
	}
	msg := e.gcMessageOf(st, &st.Plan.Workflow.Steps[idx])
	return strings.Contains(strings.ToLower(msg), gcCloseMarker)
}

// gcFinish 收口：交付物 = 全量转录；写 output.md；状态落盘；构造结果。
func (e *SquadEngine) gcFinish(st *RunState, drv *flowDriver) *RunResult {
	transcript := ""
	if data, err := os.ReadFile(filepath.Join(st.Meta.WorkDir, "group_chat", "transcript.md")); err == nil {
		transcript = string(data)
	}
	report := "# 群聊转录\n\n" + transcript
	_ = os.WriteFile(filepath.Join(st.Meta.WorkDir, "output.md"), []byte(report), 0o644)
	turns := e.gcTurnCount(st)
	st.mu.Lock()
	wf := st.Workflow()
	wf.Status = string(RunCompleted)
	st.setWorkflow(wf)
	st.Meta.Status = string(RunCompleted)
	now := time.Now()
	st.Meta.CompletedAt = &now
	_ = st.save()
	st.mu.Unlock()
	summary := fmt.Sprintf("group_chat：%d 轮发言，转录已收口", turns)
	drv.emit(st, Event{Type: EvPhase, Text: summary})
	return &RunResult{RunID: st.Meta.ID, Status: RunCompleted, Plan: planPtr(st),
		Deliverable: "output.md", Summary: summary, WorkDir: st.Meta.WorkDir}
}

func truncateStr(s string, n int) string {
	if len(s) <= n {
		return s
	}
	return s[:n] + "…"
}