weichenandCursor
7ddcb2180d
HTTP failures now carry catalog codes and a trace id, and the parent replace directives are the only copies used at build time. Co-authored-by: Cursor <cursoragent@cursor.com>
118 lines
3.2 KiB
Go
118 lines
3.2 KiB
Go
package store
|
||
|
||
import (
|
||
"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]++
|
||
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'))
|
||
}
|
||
|
||
// 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
|
||
}
|