Remove unused raw protocol recorders from agent sessions
This commit is contained in:
@@ -41,7 +41,6 @@ type Options struct {
|
||||
Hooks map[string][]HookMatcher
|
||||
MCPServers []*mcp.Server
|
||||
StdioMCP []mcp.Stdio
|
||||
Transcript io.Writer
|
||||
OnBadLine func(line []byte, reason string)
|
||||
}
|
||||
|
||||
@@ -197,9 +196,7 @@ type Session struct {
|
||||
events chan Event
|
||||
done chan struct{}
|
||||
|
||||
writeMu sync.Mutex
|
||||
transcript io.Writer
|
||||
scrub []byte
|
||||
writeMu sync.Mutex
|
||||
|
||||
pendingMu sync.Mutex
|
||||
pending map[string]chan controlResult
|
||||
@@ -235,15 +232,14 @@ type envelope struct {
|
||||
|
||||
func Start(ctx context.Context, opts Options) (*Session, error) {
|
||||
s := &Session{
|
||||
events: make(chan Event, 256),
|
||||
done: make(chan struct{}),
|
||||
transcript: opts.Transcript,
|
||||
pending: map[string]chan controlResult{},
|
||||
inflight: map[string]context.CancelFunc{},
|
||||
hooks: map[string]HookFunc{},
|
||||
servers: map[string]*mcp.Server{},
|
||||
id: initialID(opts),
|
||||
onBadLine: opts.OnBadLine,
|
||||
events: make(chan Event, 256),
|
||||
done: make(chan struct{}),
|
||||
pending: map[string]chan controlResult{},
|
||||
inflight: map[string]context.CancelFunc{},
|
||||
hooks: map[string]HookFunc{},
|
||||
servers: map[string]*mcp.Server{},
|
||||
id: initialID(opts),
|
||||
onBadLine: opts.OnBadLine,
|
||||
}
|
||||
s.lastLine.Store(time.Now().UnixNano())
|
||||
s.ctx, s.stop = context.WithCancel(context.Background())
|
||||
@@ -524,30 +520,10 @@ func (s *Session) write(v any) error {
|
||||
s.writeMu.Lock()
|
||||
defer s.writeMu.Unlock()
|
||||
|
||||
s.record("send", b)
|
||||
|
||||
_, err = s.stdin.Write(append(b, '\n'))
|
||||
return err
|
||||
}
|
||||
|
||||
func (s *Session) record(dir string, line []byte) {
|
||||
if s.transcript == nil {
|
||||
return
|
||||
}
|
||||
|
||||
if len(s.scrub) > 0 {
|
||||
line = bytes.ReplaceAll(line, s.scrub, []byte("[redacted]"))
|
||||
}
|
||||
|
||||
b, err := json.Marshal(map[string]any{"dir": dir, "msg": json.RawMessage(line)})
|
||||
if err != nil {
|
||||
slog.Error("record transcript", "err", err)
|
||||
return
|
||||
}
|
||||
|
||||
s.transcript.Write(append(b, '\n'))
|
||||
}
|
||||
|
||||
func (s *Session) readLoop(stdout io.Reader) {
|
||||
defer close(s.done)
|
||||
defer close(s.events)
|
||||
@@ -587,15 +563,6 @@ func (s *Session) handleLine(line []byte) {
|
||||
s.badLine(line, fmt.Sprintf("unrecognized type %q", env.Type))
|
||||
}
|
||||
|
||||
s.writeMu.Lock()
|
||||
if env.Type == "control_response" {
|
||||
s.learnScrub(env.Response)
|
||||
}
|
||||
if env.Type != "stream_event" {
|
||||
s.record("recv", line)
|
||||
}
|
||||
s.writeMu.Unlock()
|
||||
|
||||
switch env.Type {
|
||||
case "control_response":
|
||||
s.handleControlResponse(env.Response)
|
||||
@@ -638,21 +605,6 @@ func (s *Session) badLine(line []byte, reason string) {
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Session) learnScrub(raw json.RawMessage) {
|
||||
resp := struct {
|
||||
Response struct {
|
||||
Account struct {
|
||||
Email string `json:"email"`
|
||||
} `json:"account"`
|
||||
} `json:"response"`
|
||||
}{}
|
||||
err := json.Unmarshal(raw, &resp)
|
||||
if err != nil || resp.Response.Account.Email == "" {
|
||||
return
|
||||
}
|
||||
s.scrub = []byte(resp.Response.Account.Email)
|
||||
}
|
||||
|
||||
func (s *Session) handleControlResponse(raw json.RawMessage) {
|
||||
resp := struct {
|
||||
Subtype string `json:"subtype"`
|
||||
|
||||
@@ -11,7 +11,6 @@ import (
|
||||
"io"
|
||||
"log/slog"
|
||||
"os"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
@@ -33,7 +32,6 @@ type Options struct {
|
||||
Reminder string
|
||||
MCPServers []*mcp.Server
|
||||
StdioMCP []mcp.Stdio
|
||||
Transcript io.Writer
|
||||
OnBadLine func(line []byte, reason string)
|
||||
}
|
||||
|
||||
@@ -88,9 +86,7 @@ type Session struct {
|
||||
events chan Event
|
||||
done chan struct{}
|
||||
|
||||
writeLock chan struct{}
|
||||
recordMu sync.Mutex
|
||||
transcript io.Writer
|
||||
writeLock chan struct{}
|
||||
|
||||
nextID atomic.Int64
|
||||
pendingMu sync.Mutex
|
||||
@@ -181,15 +177,14 @@ func spawn(opts Options) (*Session, error) {
|
||||
|
||||
func newSession(opts Options, stdin *os.File, stdout io.Reader) *Session {
|
||||
s := &Session{
|
||||
stdin: stdin,
|
||||
events: make(chan Event, 256),
|
||||
writeLock: make(chan struct{}, 1),
|
||||
done: make(chan struct{}),
|
||||
transcript: opts.Transcript,
|
||||
pending: map[int64]chan rpcResult{},
|
||||
servers: map[string]*mcp.Server{},
|
||||
reminder: opts.Reminder,
|
||||
onBadLine: opts.OnBadLine,
|
||||
stdin: stdin,
|
||||
events: make(chan Event, 256),
|
||||
writeLock: make(chan struct{}, 1),
|
||||
done: make(chan struct{}),
|
||||
pending: map[int64]chan rpcResult{},
|
||||
servers: map[string]*mcp.Server{},
|
||||
reminder: opts.Reminder,
|
||||
onBadLine: opts.OnBadLine,
|
||||
}
|
||||
for _, srv := range opts.MCPServers {
|
||||
s.servers[srv.Name] = srv
|
||||
@@ -552,8 +547,6 @@ func (s *Session) write(m message, deadline time.Time) error {
|
||||
return os.ErrDeadlineExceeded
|
||||
}
|
||||
|
||||
s.record("send", b)
|
||||
|
||||
err = s.stdin.SetWriteDeadline(deadline)
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -566,26 +559,6 @@ func (s *Session) write(m message, deadline time.Time) error {
|
||||
return err
|
||||
}
|
||||
|
||||
var emailField = regexp.MustCompile(`"email":"[^"]*"`)
|
||||
|
||||
func (s *Session) record(dir string, line []byte) {
|
||||
if s.transcript == nil {
|
||||
return
|
||||
}
|
||||
|
||||
line = emailField.ReplaceAll(line, []byte(`"email":"[redacted]"`))
|
||||
|
||||
b, err := json.Marshal(map[string]any{"dir": dir, "msg": json.RawMessage(line)})
|
||||
if err != nil {
|
||||
slog.Error("record transcript", "err", err)
|
||||
return
|
||||
}
|
||||
|
||||
s.recordMu.Lock()
|
||||
defer s.recordMu.Unlock()
|
||||
s.transcript.Write(append(b, '\n'))
|
||||
}
|
||||
|
||||
func (s *Session) readLoop(stdout io.Reader) {
|
||||
defer close(s.done)
|
||||
defer close(s.events)
|
||||
@@ -620,8 +593,6 @@ func (s *Session) handleLine(line []byte) {
|
||||
return
|
||||
}
|
||||
|
||||
s.record("recv", line)
|
||||
|
||||
if m.Method == "" {
|
||||
s.handleResponse(line, m)
|
||||
return
|
||||
|
||||
Reference in New Issue
Block a user