Agent-Squad/squad/progress.go

package squad

import (
	"encoding/json"
	"sync"
	"time"
)

// Event 类型常量（SSE 事件名与此一致）。
const (
	EvRunStarted         = "run_started"
	EvPhase              = "phase"
	EvStepStarted        = "step_started"
	EvStepCompleted      = "step_completed"
	EvStepFailed         = "step_failed"
	EvStepSkipped        = "step_skipped"
	EvDetail             = "detail" // 成员执行过程帧（agent-native ProcessEvent 投影）
	EvDeliverable        = "deliverable"
	EvHumanInputRequired = "human_input_required"
	EvHumanResponded     = "human_responded"
	EvRunCompleted       = "run_completed"
	EvRunFailed          = "run_failed"
	EvRunCancelled       = "run_cancelled"
)

// Event 引擎对外事件（可 JSON 序列化，SSE 直接下发）。
type Event struct {
	Type   string          `json:"type"`
	RunID  string          `json:"run_id"`
	StepID string          `json:"step_id,omitempty"`
	Role   string          `json:"role,omitempty"`
	Text   string          `json:"text,omitempty"`
	Code   string          `json:"code,omitempty"`
	Data   json.RawMessage `json:"data,omitempty"`
	At     time.Time       `json:"at"`
}

// ProgressSink 事件出口；实现方必须非阻塞或自行缓冲。
type ProgressSink interface {
	Emit(ev Event)
}

// ProgressFunc 函数式 ProgressSink。
type ProgressFunc func(ev Event)

func (f ProgressFunc) Emit(ev Event) { f(ev) }

// FanoutSink 多路分发 + 环形缓冲（server 层 SSE 重放用）。
type FanoutSink struct {
	mu      sync.Mutex
	buffer  []Event
	subs    map[int]chan Event
	nextSub int
	limit   int
}

// NewFanoutSink 创建分发器；limit 为事件缓冲上限（超过丢弃最旧）。
func NewFanoutSink(limit int) *FanoutSink {
	if limit <= 0 {
		limit = 4096
	}
	return &FanoutSink{subs: make(map[int]chan Event), limit: limit}
}

// Emit 分发事件并写入缓冲。
func (s *FanoutSink) Emit(ev Event) {
	if ev.At.IsZero() {
		ev.At = time.Now()
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	s.buffer = append(s.buffer, ev)
	if len(s.buffer) > s.limit {
		s.buffer = s.buffer[len(s.buffer)-s.limit:]
	}
	for _, ch := range s.subs {
		select {
		case ch <- ev:
		default: // 慢消费者丢帧，不阻塞引擎
		}
	}
}

// Subscribe 订阅后续事件；cursor 为已消费的缓冲长度（0 = 从头重放）。
// 返回事件通道与取消函数。重放部分一次性送入通道后进入实时流。
func (s *FanoutSink) Subscribe(cursor int) (<-chan Event, int, func()) {
	ch := make(chan Event, 256)
	s.mu.Lock()
	id := s.nextSub
	s.nextSub++
	s.subs[id] = ch
	if cursor < 0 {
		cursor = 0
	}
	if cursor > len(s.buffer) {
		cursor = len(s.buffer)
	}
	replay := make([]Event, len(s.buffer)-cursor)
	copy(replay, s.buffer[cursor:])
	next := len(s.buffer)
	s.mu.Unlock()
	done := make(chan struct{})
	var once sync.Once
	var replayDone sync.WaitGroup
	replayDone.Add(1)
	go func() {
		defer replayDone.Done()
		for _, ev := range replay {
			select {
			case <-done:
				return
			case ch <- ev:
			}
		}
	}()
	return ch, next, func() {
		once.Do(func() { close(done) })
		replayDone.Wait()
		s.mu.Lock()
		delete(s.subs, id)
		s.mu.Unlock()
		close(ch)
	}
}

// Snapshot 返回缓冲快照与当前游标（重放起点）。
func (s *FanoutSink) Snapshot() ([]Event, int) {
	s.mu.Lock()
	defer s.mu.Unlock()
	out := make([]Event, len(s.buffer))
	copy(out, s.buffer)
	return out, len(s.buffer)
}

// NopSink 丢弃所有事件。
type NopSink struct{}

func (NopSink) Emit(Event) {}

// processEventData 把过程帧投影成 detail 事件的 Data（JSON 对象）。
//
// 工具调用身份与真实结局（call_id / status / error_reason）非空时随帧透传，
// 使 detail 事件流能按 CallID 配对交错的并行工具调用，而不是按到达顺序猜。
// 三者全空（legacy 上游未填充）时输出与既有投影逐字节一致，不引入新键。
func processEventData(ev ProcessEvent) json.RawMessage {
	data := map[string]string{"process_type": ev.Type, "tool": ev.Tool, "detail": ev.Detail}
	if ev.CallID != "" {
		data["call_id"] = ev.CallID
	}
	if ev.Status != "" {
		data["status"] = ev.Status
	}
	if ev.ErrorReason != "" {
		data["error_reason"] = ev.ErrorReason
	}
	return marshalJSON(data)
}