Show agent liveness, stuck/zombie detection and parse errors in admin
This commit is contained in:
@@ -0,0 +1,55 @@
|
||||
package claude
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestBadLines(t *testing.T) {
|
||||
bad := []string{}
|
||||
s := &Session{
|
||||
events: make(chan Event, 16),
|
||||
done: make(chan struct{}),
|
||||
onBadLine: func(line []byte, reason string) {
|
||||
bad = append(bad, string(line)+" | "+reason)
|
||||
},
|
||||
}
|
||||
|
||||
input := strings.Join([]string{
|
||||
`{"type":"assistant","message":{"content":[]}}`,
|
||||
``,
|
||||
`warning: something on stdout`,
|
||||
`{"type":"assistant",`,
|
||||
`{"type":"mystery","x":1}`,
|
||||
`{"type":"keep_alive"}`,
|
||||
`{"type":"tool_progress","heartbeat":true}`,
|
||||
}, "\n")
|
||||
|
||||
before := time.Now()
|
||||
s.readLoop(strings.NewReader(input))
|
||||
|
||||
want := []string{
|
||||
`warning: something on stdout | not a JSON object`,
|
||||
`{"type":"assistant", | unexpected end of JSON input`,
|
||||
`{"type":"mystery","x":1} | unrecognised type "mystery"`,
|
||||
}
|
||||
if strings.Join(bad, "\n") != strings.Join(want, "\n") {
|
||||
t.Errorf("bad lines:\n%s\nwant:\n%s", strings.Join(bad, "\n"), strings.Join(want, "\n"))
|
||||
}
|
||||
|
||||
types := []string{}
|
||||
for ev := range s.events {
|
||||
types = append(types, ev.Type)
|
||||
}
|
||||
if strings.Join(types, ",") != "assistant,mystery,tool_progress" {
|
||||
t.Errorf("events = %v", types)
|
||||
}
|
||||
|
||||
if s.LastLine().Before(before) {
|
||||
t.Errorf("last line time %v not updated", s.LastLine())
|
||||
}
|
||||
if !s.Exited() {
|
||||
t.Errorf("session not exited after stdout closed")
|
||||
}
|
||||
}
|
||||
@@ -15,6 +15,7 @@ import (
|
||||
"os"
|
||||
"os/exec"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"g.fc.run/misc/byre/internal/mcp"
|
||||
@@ -39,6 +40,21 @@ type Options struct {
|
||||
Hooks map[string][]HookMatcher
|
||||
MCPServers []*mcp.Server
|
||||
Transcript io.Writer
|
||||
OnBadLine func(line []byte, reason string)
|
||||
}
|
||||
|
||||
var knownTypes = map[string]bool{
|
||||
"assistant": true,
|
||||
"user": true,
|
||||
"result": true,
|
||||
"system": true,
|
||||
"stream_event": true,
|
||||
"tool_progress": true,
|
||||
"rate_limit_event": true,
|
||||
"control_request": true,
|
||||
"control_response": true,
|
||||
"control_cancel_request": true,
|
||||
"keep_alive": true,
|
||||
}
|
||||
|
||||
type Event struct {
|
||||
@@ -175,6 +191,9 @@ type Session struct {
|
||||
|
||||
idMu sync.Mutex
|
||||
id string
|
||||
|
||||
onBadLine func(line []byte, reason string)
|
||||
lastLine atomic.Int64
|
||||
}
|
||||
|
||||
type controlResult struct {
|
||||
@@ -200,7 +219,9 @@ func Start(ctx context.Context, opts Options) (*Session, error) {
|
||||
hooks: map[string]HookFunc{},
|
||||
servers: map[string]*mcp.Server{},
|
||||
id: initialID(opts),
|
||||
onBadLine: opts.OnBadLine,
|
||||
}
|
||||
s.lastLine.Store(time.Now().UnixNano())
|
||||
|
||||
args, err := s.buildArgs(opts)
|
||||
if err != nil {
|
||||
@@ -355,6 +376,23 @@ func (s *Session) Done() <-chan struct{} {
|
||||
return s.done
|
||||
}
|
||||
|
||||
func (s *Session) Pid() int {
|
||||
return s.cmd.Process.Pid
|
||||
}
|
||||
|
||||
func (s *Session) LastLine() time.Time {
|
||||
return time.Unix(0, s.lastLine.Load())
|
||||
}
|
||||
|
||||
func (s *Session) Exited() bool {
|
||||
select {
|
||||
case <-s.done:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Session) ID() string {
|
||||
s.idMu.Lock()
|
||||
defer s.idMu.Unlock()
|
||||
@@ -493,8 +531,14 @@ func (s *Session) readLoop(stdout io.Reader) {
|
||||
r := bufio.NewReader(stdout)
|
||||
for {
|
||||
line, err := r.ReadBytes('\n')
|
||||
if len(line) > 1 && line[0] == '{' {
|
||||
s.handleLine(line[:len(line)-1])
|
||||
line = bytes.TrimRight(line, "\r\n")
|
||||
if len(line) > 0 {
|
||||
s.lastLine.Store(time.Now().UnixNano())
|
||||
}
|
||||
if len(line) > 0 && line[0] == '{' {
|
||||
s.handleLine(line)
|
||||
} else if len(line) > 0 {
|
||||
s.badLine(line, "not a JSON object")
|
||||
}
|
||||
if err != nil {
|
||||
if err != io.EOF {
|
||||
@@ -510,10 +554,14 @@ func (s *Session) handleLine(line []byte) {
|
||||
env := envelope{}
|
||||
err := json.Unmarshal(line, &env)
|
||||
if err != nil {
|
||||
slog.Error("parse claude message", "err", err, "line", string(line))
|
||||
s.badLine(line, err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
if !knownTypes[env.Type] {
|
||||
s.badLine(line, fmt.Sprintf("unrecognised type %q", env.Type))
|
||||
}
|
||||
|
||||
s.writeMu.Lock()
|
||||
if env.Type == "control_response" {
|
||||
s.learnScrub(env.Response)
|
||||
@@ -545,6 +593,13 @@ func (s *Session) handleLine(line []byte) {
|
||||
s.events <- Event{Type: env.Type, Subtype: env.Subtype, Raw: append(json.RawMessage{}, line...), Read: read}
|
||||
}
|
||||
|
||||
func (s *Session) badLine(line []byte, reason string) {
|
||||
slog.Warn("bad claude line", "reason", reason, "line", string(line))
|
||||
if s.onBadLine != nil {
|
||||
s.onBadLine(line, reason)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Session) learnScrub(raw json.RawMessage) {
|
||||
resp := struct {
|
||||
Response struct {
|
||||
|
||||
Reference in New Issue
Block a user