Agent-System-Backend/internal/runmgr/events.go

package runmgr

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

	"agent-system-backend/internal/bus"
	"agent-system-backend/internal/store"
	core "agent-system-core"
)

// eventBuffer 是会话事件批处理器（写放大治理）：
//   - 高频增量（message.delta / thinking.delta / agent.progress）缓冲合并落盘；
//   - 其余事件（终态、HITL、工具、阶段）立即落盘，落盘前先冲掉同会话缓冲保证顺序；
//   - seq 在落盘事务内分配（会话内连续），发布顺序 = seq 顺序；
//   - 停机清空缓冲；未冲盘的增量在崩溃时丢失（≤ interval 窗口，重放由终态消息树对账）。
//
// 会话内事件天然单线程（一个会话至多一个在途 run，子代理在父执行栈内同步运行），
// 少数跨 goroutine 路径（HITL resolve）与冲盘互斥由 flushing 锁保证不交叉。
type eventBuffer struct {
	store               *store.Store
	bus                 *bus.Bus
	accountGuardEnabled bool
	allowImplicitLocal  bool
	interval            time.Duration // 定时冲盘窗口
	maxBuf              int           // 单会话缓冲条数上限（触发立即冲盘）
	// onPersisted 落盘+本地发布后的回调（PG 档 NOTIFY 跨实例扇出；nil = 无）。
	onPersisted func(conversationID string, evs []core.StreamEvent)

	mu       sync.Mutex
	buf      map[string][]core.StreamEvent
	flushing sync.Mutex // 串行化落盘（批量插入很快，全局锁足够）

	ticker *time.Ticker
	done   chan struct{}
	wg     sync.WaitGroup
}

func newEventBuffer(st *store.Store, b *bus.Bus) *eventBuffer {
	eb := &eventBuffer{
		store: st, bus: b,
		interval: 150 * time.Millisecond,
		maxBuf:   64,
		buf:      map[string][]core.StreamEvent{},
		done:     make(chan struct{}),
	}
	eb.wg.Add(1)
	go eb.loop()
	return eb
}

func (eb *eventBuffer) loop() {
	defer eb.wg.Done()
	eb.ticker = time.NewTicker(eb.interval)
	defer eb.ticker.Stop()
	for {
		select {
		case <-eb.done:
			return
		case <-eb.ticker.C:
			eb.flushAll()
		}
	}
}

// stop 停机：停止定时器并清空全部缓冲（优雅停机时无增量丢失）。
func (eb *eventBuffer) stop() {
	close(eb.done)
	eb.wg.Wait()
	eb.flushAll()
}

// isDelta 标记高频可缓冲事件。
func isDelta(typ core.StreamEventType) bool {
	switch typ {
	case core.EvMessageDelta, core.EvThinkingDelta, core.EvAgentProgress, core.EvToolCallDelta, core.EvToolOutput:
		return true
	}
	return false
}

// emit 入队一条事件；返回已落盘事件的 seq（缓冲事件返回 0）与错误。
func (eb *eventBuffer) emit(conversationID, runID string, typ core.StreamEventType, payload any) (int64, error) {
	e := core.StreamEvent{
		ConversationID: conversationID, RunID: runID,
		TS: time.Now().UTC().Format(time.RFC3339Nano), V: core.ProtocolVersion, Type: typ,
	}
	// 项目流（D50）：伪会话键复用同一持久化/seq/重放机制；
	// 信封在此盖章项目归属，前端按 ProjectID 识别项目流事件。
	if pid, ok := core.ProjectFromChannel(core.ID(conversationID)); ok {
		e.ProjectID = pid
	}
	if payload != nil {
		buf, err := json.Marshal(payload)
		if err != nil {
			return 0, err
		}
		e.Payload = buf
	}
	if !isDelta(typ) {
		// 立即事件：先冲掉同会话缓冲（保序），再单条落盘发布。
		eb.flushConversation(conversationID)
		batch := []core.StreamEvent{e}
		if err := eb.append(conversationID, batch); err != nil {
			return 0, err
		}
		return batch[0].Seq, nil
	}
	eb.mu.Lock()
	eb.buf[conversationID] = append(eb.buf[conversationID], e)
	n := len(eb.buf[conversationID])
	eb.mu.Unlock()
	if n >= eb.maxBuf {
		eb.flushConversation(conversationID)
	}
	return 0, nil
}

// emitAccountChecked flushes any earlier deltas and the new structural event
// in one final account-checked Store transaction. A disabled owner loses its
// buffered deltas instead of persisting them after disable.
func (eb *eventBuffer) emitAccountChecked(actor, conversationID, runID string, typ core.StreamEventType, payload any, allowImplicitDefault bool) error {
	e := core.StreamEvent{ConversationID: conversationID, RunID: runID,
		TS: time.Now().UTC().Format(time.RFC3339Nano), V: core.ProtocolVersion, Type: typ}
	if projectID, ok := core.ProjectFromChannel(core.ID(conversationID)); ok {
		e.ProjectID = projectID
	}
	if payload != nil {
		raw, err := json.Marshal(payload)
		if err != nil {
			return err
		}
		e.Payload = raw
	}
	eb.flushing.Lock()
	defer eb.flushing.Unlock()
	eb.mu.Lock()
	batch := append([]core.StreamEvent(nil), eb.buf[conversationID]...)
	delete(eb.buf, conversationID)
	eb.mu.Unlock()
	batch = append(batch, e)
	if _, err := eb.store.AppendEventsAccountChecked(actor, conversationID, batch, allowImplicitDefault); err != nil {
		return err
	}
	for _, event := range batch {
		eb.bus.Publish(event)
	}
	if eb.onPersisted != nil {
		eb.onPersisted(conversationID, batch)
	}
	return nil
}

// emitProjectShared persists an explicit internal publication and its proof
// before fan-out. A client-visible payload cannot opt itself into this path.
func (eb *eventBuffer) emitProjectShared(owner, producer, channel, runID string, typ core.StreamEventType, payload any) error {
	project, ok := core.ProjectFromChannel(channel)
	if !ok {
		return store.ErrNotFound
	}
	encoded, err := json.Marshal(payload)
	if err != nil {
		return err
	}
	event := core.StreamEvent{ConversationID: channel, ProjectID: project, RunID: runID, Type: typ, Payload: encoded, TS: time.Now().UTC().Format(time.RFC3339Nano), V: core.ProtocolVersion}
	eb.flushConversation(channel)
	eb.flushing.Lock()
	defer eb.flushing.Unlock()
	events := []core.StreamEvent{event}
	if eb.accountGuardEnabled {
		_, err = eb.store.AppendProjectSharedEventsAccountChecked(owner, producer, channel, events, eb.allowImplicitLocal)
	} else {
		_, err = eb.store.AppendProjectSharedEvents(owner, producer, channel, events)
	}
	if err != nil {
		return err
	}
	for _, event := range events {
		eb.bus.Publish(event)
	}
	if eb.onPersisted != nil {
		eb.onPersisted(channel, events)
	}
	return nil
}

// append 落盘并按 seq 顺序发布。
func (eb *eventBuffer) append(conversationID string, evs []core.StreamEvent) error {
	eb.flushing.Lock()
	defer eb.flushing.Unlock()
	var err error
	if eb.accountGuardEnabled {
		_, err = eb.store.AppendEventsForEffectiveChannel(conversationID, evs, eb.allowImplicitLocal)
	} else {
		_, err = eb.store.AppendEvents(conversationID, evs)
	}
	if err != nil {
		return err
	}
	for i := range evs {
		eb.bus.Publish(evs[i])
	}
	if eb.onPersisted != nil {
		eb.onPersisted(conversationID, evs)
	}
	return nil
}

// flushConversation 冲掉指定会话的缓冲（失败记日志丢弃：增量可由重放补偿）。
func (eb *eventBuffer) flushConversation(conversationID string) {
	eb.mu.Lock()
	evs := eb.buf[conversationID]
	delete(eb.buf, conversationID)
	eb.mu.Unlock()
	if len(evs) == 0 {
		return
	}
	if err := eb.append(conversationID, evs); err != nil {
		log.Printf("[events] flush %s: %v（%d 条增量待重放补偿）", conversationID, err, len(evs))
	}
}

func (eb *eventBuffer) flushAll() {
	eb.mu.Lock()
	ids := make([]string, 0, len(eb.buf))
	for id := range eb.buf {
		ids = append(ids, id)
	}
	eb.mu.Unlock()
	for _, id := range ids {
		eb.flushConversation(id)
	}
}