You've already forked agentic-coding-workflow
132 lines
4.1 KiB
Go
132 lines
4.1 KiB
Go
// turnbus.go 实现 TurnBus：进程内的极简 observer / pub-sub，作为"回合完成 /
|
|
// 会话空闲"的内部事件原语。与 notify.Dispatcher（面向用户的外发通知）解耦——
|
|
// TurnBus 是给未来的编排层（auto-continue / gate / hand-off）订阅用的内部信号，
|
|
// 不落库、不外发。
|
|
//
|
|
// 关键不变量：Publish 必须永不阻塞 relay reader 热路径。订阅者以 fire-and-forget
|
|
// 方式在独立 goroutine 中调用，并用 recover() 隔离 panic——一个慢/有 bug 的
|
|
// 订阅者不能拖垮协议 I/O。语义对齐 notify.Dispatcher 的非阻塞 fan-out。
|
|
package acp
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"log/slog"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
)
|
|
|
|
// TurnEventKind 标识 TurnEvent 的类别。
|
|
type TurnEventKind string
|
|
|
|
const (
|
|
// TurnCompletedEvent 表示一次回合正常完成（收到 session/prompt 响应的 stopReason）。
|
|
TurnCompletedEvent TurnEventKind = "turn_completed"
|
|
// SessionIdleEvent 表示会话当前空闲（回合完成或被中止后），编排层可据此决策。
|
|
SessionIdleEvent TurnEventKind = "session_idle"
|
|
)
|
|
|
|
// TurnEvent 是 TurnBus 上发布的事件载荷。RawStopReason 是 agent 上报的原始字符串
|
|
// （可能为空，例如 abort 路径），StopReason 是其归一化枚举。
|
|
type TurnEvent struct {
|
|
SessionID uuid.UUID
|
|
UserID uuid.UUID
|
|
WorkspaceID uuid.UUID
|
|
TurnIndex int
|
|
Kind TurnEventKind
|
|
StopReason StopReason
|
|
RawStopReason string
|
|
CompletedAt time.Time
|
|
}
|
|
|
|
// TurnBus 是回合事件的进程内 pub-sub 抽象。
|
|
type TurnBus interface {
|
|
// Subscribe 注册一个具名订阅者。name 仅用于日志/排障。重复 name 会追加，
|
|
// 不去重（与 notify 一致，调用方自行保证唯一）。
|
|
Subscribe(name string, fn func(context.Context, TurnEvent))
|
|
// Publish 把事件 fire-and-forget 派发给所有订阅者，永不阻塞调用方。
|
|
Publish(ctx context.Context, ev TurnEvent)
|
|
}
|
|
|
|
type deliver struct {
|
|
ctx context.Context
|
|
ev TurnEvent
|
|
}
|
|
|
|
type subscriber struct {
|
|
name string
|
|
fn func(context.Context, TurnEvent)
|
|
ch chan deliver
|
|
}
|
|
|
|
// subBuffer 是每个订阅者的投递缓冲容量。
|
|
const subBuffer = 256
|
|
|
|
// defaultTurnBus 是 TurnBus 的默认实现：每个订阅者一个常驻 worker goroutine，从带缓冲
|
|
// channel 顺序消费事件——因此对同一订阅者保证 per-subscriber FIFO（turn_completed 必先于
|
|
// 同一次 Publish 序列里随后的 session_idle 到达）。panic 用 recover() 隔离。Publish 以非阻塞
|
|
// 方式投递（缓冲满则丢弃并告警），永不阻塞 relay reader 热路径。
|
|
type defaultTurnBus struct {
|
|
mu sync.RWMutex
|
|
subs []*subscriber
|
|
log *slog.Logger
|
|
}
|
|
|
|
// NewTurnBus 构造默认 TurnBus。
|
|
func NewTurnBus(log *slog.Logger) TurnBus {
|
|
if log == nil {
|
|
log = slog.Default()
|
|
}
|
|
return &defaultTurnBus{log: log}
|
|
}
|
|
|
|
func (b *defaultTurnBus) Subscribe(name string, fn func(context.Context, TurnEvent)) {
|
|
if fn == nil {
|
|
return
|
|
}
|
|
s := &subscriber{name: name, fn: fn, ch: make(chan deliver, subBuffer)}
|
|
b.mu.Lock()
|
|
b.subs = append(b.subs, s)
|
|
b.mu.Unlock()
|
|
// 每个订阅者一个常驻 worker，顺序消费保证 FIFO。worker 与 app 同生命周期。
|
|
go b.worker(s)
|
|
}
|
|
|
|
func (b *defaultTurnBus) worker(s *subscriber) {
|
|
for d := range s.ch {
|
|
b.dispatch(s, d)
|
|
}
|
|
}
|
|
|
|
func (b *defaultTurnBus) dispatch(s *subscriber, d deliver) {
|
|
defer func() {
|
|
if rec := recover(); rec != nil {
|
|
b.log.Error("acp.turnbus.subscriber_panic",
|
|
"subscriber", s.name,
|
|
"session_id", d.ev.SessionID,
|
|
"panic", fmt.Sprintf("%v", rec))
|
|
}
|
|
}()
|
|
s.fn(d.ctx, d.ev)
|
|
}
|
|
|
|
func (b *defaultTurnBus) Publish(ctx context.Context, ev TurnEvent) {
|
|
b.mu.RLock()
|
|
subs := b.subs
|
|
b.mu.RUnlock()
|
|
|
|
// 解绑调用方 ctx 的取消（relay teardown 会 cancel ctx，但订阅者应能完成它的短暂处理）。
|
|
d := deliver{ctx: context.WithoutCancel(ctx), ev: ev}
|
|
for _, s := range subs {
|
|
select {
|
|
case s.ch <- d:
|
|
default:
|
|
// 缓冲满：丢弃以保证 Publish 永不阻塞 relay reader（正常负载不应发生）。
|
|
b.log.Warn("acp.turnbus.drop_full",
|
|
"subscriber", s.name, "session_id", ev.SessionID, "kind", string(ev.Kind))
|
|
}
|
|
}
|
|
}
|