weichenandCursor
44aa764b56
ci / vet-and-test (pull_request) Successful in 43s
Why: A hard kill left runs in pending, running, or waiting_hitl, a busy send stored a message before the 409, finishRun could mark a run completed without an assistant message, and a full SSE buffer dropped events without closing the stream. Evidence: GOWORK=off go vet ./... and GOWORK=off go test -count=1 ./... passed, including store, runmgr, bus, and server recovery tests. Impact: Migration v14 adds runs.instance_id and messages.request_id. Startup reconciliation and shutdown share one seal. Turn and workflow sends claim a slot before writing. Duplicate request ids return the existing message. SettleRun updates the run and appends the assistant message in one transaction. A full subscriber buffer closes the channel so SSE can replay from the cursor. Refs: Dev feature/pi-durable-borrow Co-authored-by: Cursor <cursoragent@cursor.com>
76 lines
1.7 KiB
Go
76 lines
1.7 KiB
Go
// Package bus 是 streamhub：按会话分发已持久化的事件（SSE 通道）。
|
|
// 事件权威在 store（seq 游标）；bus 只做在线转发。
|
|
// 订阅缓冲满时关闭该订阅，客户端用 Last-Event-ID 从游标重放，不再静默丢事件。
|
|
package bus
|
|
|
|
import (
|
|
"sync"
|
|
|
|
core "agent-system-core"
|
|
)
|
|
|
|
type subscriber struct {
|
|
ch chan core.StreamEvent
|
|
filter string // conversationID；空 = 全部
|
|
closed bool
|
|
}
|
|
|
|
type Bus struct {
|
|
mu sync.Mutex
|
|
subs map[int64]*subscriber
|
|
next int64
|
|
buf int
|
|
}
|
|
|
|
func New() *Bus { return NewWithBuffer(512) }
|
|
|
|
// NewWithBuffer 指定订阅缓冲长度。测试用小缓冲观察溢出关闭。
|
|
func NewWithBuffer(n int) *Bus {
|
|
if n < 1 {
|
|
n = 1
|
|
}
|
|
return &Bus{subs: map[int64]*subscriber{}, buf: n}
|
|
}
|
|
|
|
// Subscribe 返回新订阅；close(unsubscribe) 注销。注销是幂等的。
|
|
func (b *Bus) Subscribe(conversationID string) (<-chan core.StreamEvent, func()) {
|
|
b.mu.Lock()
|
|
id := b.next
|
|
b.next++
|
|
ch := make(chan core.StreamEvent, b.buf)
|
|
b.subs[id] = &subscriber{ch: ch, filter: conversationID}
|
|
b.mu.Unlock()
|
|
return ch, func() {
|
|
b.mu.Lock()
|
|
b.dropLocked(id)
|
|
b.mu.Unlock()
|
|
}
|
|
}
|
|
|
|
// Publish 非阻塞广播。缓冲满的订阅被移除并关闭，避免静默丢事件。
|
|
func (b *Bus) Publish(e core.StreamEvent) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
for id, s := range b.subs {
|
|
if s.closed || (s.filter != "" && s.filter != e.ConversationID) {
|
|
continue
|
|
}
|
|
select {
|
|
case s.ch <- e:
|
|
default:
|
|
b.dropLocked(id)
|
|
}
|
|
}
|
|
}
|
|
|
|
// dropLocked 在 b.mu 下关闭并移除订阅。重复调用无效果。
|
|
func (b *Bus) dropLocked(id int64) {
|
|
s, ok := b.subs[id]
|
|
if !ok || s.closed {
|
|
return
|
|
}
|
|
s.closed = true
|
|
delete(b.subs, id)
|
|
close(s.ch)
|
|
}
|