Agent-System/internal/stream/stream.go

package stream

import (
	"context"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
)

type contextKey struct{}

var streamerCtxKey = contextKey{}

// Listener 事件监听回调
type Listener func(StreamEvent)

// Streamer 流式事件发射器
type Streamer struct {
	mu             sync.RWMutex
	listeners      []Listener
	includeDetails bool
	eventCount     int64
}

// NewStreamer 创建新的发射器
func NewStreamer(includeDetails bool, listeners ...Listener) *Streamer {
	return &Streamer{
		listeners:      listeners,
		includeDetails: includeDetails,
	}
}

// AddListener 动态注册监听器
func (s *Streamer) AddListener(l Listener) {
	if s == nil || l == nil {
		return
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	s.listeners = append(s.listeners, l)
}

// Emit 发送一个流式事件
func (s *Streamer) Emit(ev StreamEvent) {
	if s == nil {
		return
	}
	// 详情按需过滤
	if ev.Type == EventDetail && !s.includeDetails {
		return
	}

	count := atomic.AddInt64(&s.eventCount, 1)
	if ev.ID == "" {
		ev.ID = fmt.Sprintf("evt_%d_%d", time.Now().UnixNano(), count)
	}
	if ev.Timestamp.IsZero() {
		ev.Timestamp = time.Now()
	}

	s.mu.RLock()
	listeners := make([]Listener, len(s.listeners))
	copy(listeners, s.listeners)
	s.mu.RUnlock()

	for _, l := range listeners {
		if l != nil {
			l(ev)
		}
	}
}

// EmitPhase 发送阶段性任务状态跃迁事件
func (s *Streamer) EmitPhase(phaseID, message, role string, stepIndex, totalSteps, progress int) {
	if s == nil {
		return
	}
	s.Emit(StreamEvent{
		Type: EventPhase,
		Phase: &PhasePayload{
			PhaseID:    phaseID,
			Message:    message,
			Role:       role,
			StepIndex:  stepIndex,
			TotalSteps: totalSteps,
			Progress:   progress,
		},
	})
}

// EmitStatus 发送意图与状态机事件
func (s *Streamer) EmitStatus(intent, branch, currentStage, message string, patches map[string]string) {
	if s == nil {
		return
	}
	s.Emit(StreamEvent{
		Type: EventStatus,
		Status: &StatusPayload{
			Intent:       intent,
			Branch:       branch,
			CurrentStage: currentStage,
			Patches:      patches,
			Message:      message,
		},
	})
}

// EmitSummary 发送核心进展摘要事件 (低认知负荷)
func (s *Streamer) EmitSummary(title, content string, maturityScore int, maturityLevel string) {
	if s == nil {
		return
	}
	s.Emit(StreamEvent{
		Type: EventSummary,
		Summary: &SummaryPayload{
			Title:         title,
			Content:       content,
			MaturityScore: maturityScore,
			MaturityLevel: maturityLevel,
		},
	})
}

// EmitBriefing 发送 B 轨业务参谋决策锦囊
func (s *Streamer) EmitBriefing(brief any) {
	if s == nil || brief == nil {
		return
	}
	s.Emit(StreamEvent{
		Type:     EventBriefing,
		Briefing: brief,
	})
}

// EmitDetail 发送详细推演草稿或思考链 (按需折叠)
func (s *Streamer) EmitDetail(category, stepID, role, title, content string, collapsible bool) {
	if s == nil {
		return
	}
	s.Emit(StreamEvent{
		Type: EventDetail,
		Detail: &DetailPayload{
			Category:    category,
			StepID:      stepID,
			Role:        role,
			Title:       title,
			Content:     content,
			Collapsible: collapsible,
		},
	})
}

// EmitDelta 发送对话自然语言流式文本
func (s *Streamer) EmitDelta(delta string) {
	if s == nil || delta == "" {
		return
	}
	s.Emit(StreamEvent{
		Type:  EventDelta,
		Delta: delta,
	})
}

// EmitDeliverable 发送 A 轨合规文档交付事件
func (s *Streamer) EmitDeliverable(taskKey, title, filePath string, wordCount int, score float64) {
	if s == nil {
		return
	}
	s.Emit(StreamEvent{
		Type: EventDeliverable,
		Deliverable: &DeliverablePayload{
			TaskKey:      taskKey,
			Title:        title,
			FilePath:     filePath,
			WordCount:    wordCount,
			QualityScore: score,
		},
	})
}

// EmitGovernance 发送文档治理生命周期与认领变更拦截事件
func (s *Streamer) EmitGovernance(payload GovernancePayload) {
	if s == nil {
		return
	}
	s.Emit(StreamEvent{
		Type:       EventGovernance,
		Governance: &payload,
	})
}

// EmitErrorInfo 发送结构化错误警报
func (s *Streamer) EmitErrorInfo(info ErrorInfo) {
	if s == nil {
		return
	}
	cp := info
	s.Emit(StreamEvent{
		Type:      EventError,
		Error:     info.UserMessage,
		ErrorInfo: &cp,
	})
}

// EmitError 发送错误警报（自动分类为结构化 ErrorInfo）
func (s *Streamer) EmitError(err error) {
	if s == nil || err == nil {
		return
	}
	s.EmitErrorInfo(ClassifyError(err, "", "", "", ""))
}

// EmitDone 发送流结束事件
func (s *Streamer) EmitDone(sessionID string, duration time.Duration) {
	if s == nil {
		return
	}
	s.Emit(StreamEvent{
		Type: EventDone,
		Done: &DonePayload{
			SessionID:   sessionID,
			DurationMs:  duration.Milliseconds(),
			TotalEvents: int(atomic.LoadInt64(&s.eventCount)),
		},
	})
}

// EmitDoneWithWorkspace 发送流结束事件并携带任务工作区目录（P2：供前端「继续任务」续跑）
func (s *Streamer) EmitDoneWithWorkspace(sessionID, workspaceID string, duration time.Duration) {
	if s == nil {
		return
	}
	s.Emit(StreamEvent{
		Type: EventDone,
		Done: &DonePayload{
			SessionID:   sessionID,
			WorkspaceID: workspaceID,
			DurationMs:  duration.Milliseconds(),
			TotalEvents: int(atomic.LoadInt64(&s.eventCount)),
		},
	})
}

// EmitDoneWithExport 发送流结束事件并携带落库消息 ID 与权威导出能力
func (s *Streamer) EmitDoneWithExport(sessionID string, duration time.Duration, messageID int64, revision int, sha256 string, capability any) {
	if s == nil {
		return
	}
	s.Emit(StreamEvent{
		Type: EventDone,
		Done: &DonePayload{
			SessionID:        sessionID,
			DurationMs:       duration.Milliseconds(),
			TotalEvents:      int(atomic.LoadInt64(&s.eventCount)),
			MessageID:        messageID,
			ContentRevision:  revision,
			ContentSha256:    sha256,
			ExportCapability: capability,
		},
	})
}

// EventCount 获取已发射事件总数
func (s *Streamer) EventCount() int {
	if s == nil {
		return 0
	}
	return int(atomic.LoadInt64(&s.eventCount))
}

// WithStreamer 将 Streamer 注入 Context
func WithStreamer(ctx context.Context, s *Streamer) context.Context {
	return context.WithValue(ctx, streamerCtxKey, s)
}

// WithListener 便捷方法：基于 Listener 创建 Streamer 并注入 Context
func WithListener(ctx context.Context, includeDetails bool, l Listener) (context.Context, *Streamer) {
	s := NewStreamer(includeDetails, l)
	return WithStreamer(ctx, s), s
}

// FromContext 从 Context 提取 Streamer，若未设置则返回安全空对象，避免空指针崩溃
func FromContext(ctx context.Context) *Streamer {
	if ctx == nil {
		return nil
	}
	if s, ok := ctx.Value(streamerCtxKey).(*Streamer); ok && s != nil {
		return s
	}
	return nil
}