Agent-Squad/store/eventsink.go

package store

import (
	"bufio"
	"encoding/json"
	"os"
	"path/filepath"
	"regexp"
	"strings"
	"sync"
	"time"

	"agent-squad/squad"
)

// FileEventSink 事件持久化装饰器：把引擎事件按行追加到
// runs/<run_id>/events.jsonl（每行带自增 seq，天然即回放顺序），
// 落盘前做秘密形状脱敏，然后转发给下游（FanoutSink → SSE）。
// 机制参考 Multica 的 (task_id, seq) 消息表 + 写前 redact，文件版实现。
type FileEventSink struct {
	DataDir string
	Next    squad.ProgressSink

	mu    sync.Mutex
	files map[string]*os.File
	seqs  map[string]int64
}

// NewFileEventSink 创建事件落盘 sink。
func NewFileEventSink(dataDir string, next squad.ProgressSink) *FileEventSink {
	return &FileEventSink{DataDir: dataDir, Next: next, files: map[string]*os.File{}, seqs: map[string]int64{}}
}

// Emit 落盘（脱敏）并转发。
func (s *FileEventSink) Emit(ev squad.Event) {
	defer func() {
		if s.Next != nil {
			s.Next.Emit(ev)
		}
	}()
	if ev.RunID == "" {
		return
	}
	if ev.At.IsZero() {
		ev.At = time.Now()
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	f, ok := s.files[ev.RunID]
	if !ok {
		dir := filepath.Join(s.DataDir, "runs", ev.RunID)
		if err := os.MkdirAll(dir, 0o755); err != nil {
			return
		}
		fp, err := os.OpenFile(filepath.Join(dir, "events.jsonl"),
			os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
		if err != nil {
			return
		}
		f = fp
		s.files[ev.RunID] = f
		s.seqs[ev.RunID] = lastEventSeq(filepath.Join(dir, "events.jsonl"))
	}
	s.seqs[ev.RunID]++
	line := persistedEvent{Seq: s.seqs[ev.RunID], Type: ev.Type, StepID: ev.StepID,
		Role: ev.Role, Text: Redact(ev.Text), Code: ev.Code, At: ev.At}
	if len(ev.Data) > 0 {
		line.Data = json.RawMessage(Redact(string(ev.Data)))
	}
	data, err := json.Marshal(line)
	if err != nil {
		return
	}
	_, _ = f.Write(append(data, '\n'))
}

func lastEventSeq(path string) int64 {
	f, err := os.Open(path)
	if err != nil {
		return 0
	}
	defer f.Close()
	var last int64
	sc := bufio.NewScanner(f)
	for sc.Scan() {
		var line persistedEvent
		if err := json.Unmarshal(sc.Bytes(), &line); err == nil && line.Seq > last {
			last = line.Seq
		}
	}
	return last
}

// Close 关闭全部文件句柄。
func (s *FileEventSink) Close() {
	s.mu.Lock()
	defer s.mu.Unlock()
	for _, f := range s.files {
		_ = f.Close()
	}
	s.files = map[string]*os.File{}
}

type persistedEvent struct {
	Seq    int64           `json:"seq"`
	Type   string          `json:"type"`
	StepID string          `json:"step_id,omitempty"`
	Role   string          `json:"role,omitempty"`
	Text   string          `json:"text,omitempty"`
	Code   string          `json:"code,omitempty"`
	Data   json.RawMessage `json:"data,omitempty"`
	At     time.Time       `json:"at"`
}

// secret 形状：API key（sk-/api key 风格长串）与 Bearer 凭证。
var (
	reSKKey  = regexp.MustCompile(`(?:sk|rk|pk)-[A-Za-z0-9_\-]{12,}`)
	reBearer = regexp.MustCompile(`(?i)bearer\s+[A-Za-z0-9._\-]{12,}`)
	reAPIKey = regexp.MustCompile(`(?i)(api[_-]?key["':=\s]+)[A-Za-z0-9_\-\.]{12,}`)
	reJWT    = regexp.MustCompile(`eyJ[A-Za-z0-9_\-]{8,}\.[A-Za-z0-9_\-]{8,}\.[A-Za-z0-9_\-]{8,}`)
)

// Redact 脱敏：把秘密形状子串替换为 ***（宁可多脱，不漏一个）。
func Redact(s string) string {
	if s == "" {
		return s
	}
	s = reJWT.ReplaceAllString(s, "***")
	s = reSKKey.ReplaceAllString(s, "***")
	s = reBearer.ReplaceAllString(s, "Bearer ***")
	s = reAPIKey.ReplaceAllString(s, "${1}***")
	if strings.Contains(strings.ToLower(s), "password") {
		s = regexp.MustCompile(`(?i)(password["':=\s]+)[^\s"',}]+`).ReplaceAllString(s, "${1}***")
	}
	return s
}