weichenandCursor
bfec3c8930
Pin Agent-Native as a submodule so the module builds without the multi-repo sibling path. Co-authored-by: Cursor <cursoragent@cursor.com>
127 lines
3.1 KiB
Go
127 lines
3.1 KiB
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"`
|
|
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()
|
|
go func() {
|
|
for _, ev := range replay {
|
|
ch <- ev
|
|
}
|
|
}()
|
|
return ch, next, func() {
|
|
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) {}
|