mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-15 12:14:17 +00:00
fix(agent): add panic recovery to prevent zombie state after agent loop crash (#39)
Root cause: runLoop() had no recover() — panics killed the goroutine while leaving the session queue's activeRuns entry orphaned. Subsequent messages to the same session would hang indefinitely (zombie state). Fix: Two-layer panic recovery: 1. loop.go: runLoop() now returns error on panic instead of crashing goroutine, allowing normal cleanup flow (session queue, traces, events) 2. queue.go: executeRun() defense-in-depth recovery ensures activeRuns cleanup and scheduleNext() even if agent-level recovery fails
This commit is contained in:
1 parent
4ac611530a
commit
d10236241e
2 files changed
+31
-1
No files matched your search
+13
-1
@@ -5,6 +5,7 @@ import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"runtime"
|
||||
"sort"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -31,7 +32,18 @@ type indexedResult struct {
|
||||
spanStart time.Time
|
||||
}
|
||||
|
||||
func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) {
|
||||
func (l *Loop) runLoop(ctx context.Context, req RunRequest) (result *RunResult, err error) {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
buf := make([]byte, 8192)
|
||||
n := runtime.Stack(buf, false)
|
||||
slog.Error("agent loop panicked", "agent", l.id, "session", req.SessionKey,
|
||||
"panic", fmt.Sprint(r), "stack", string(buf[:n]))
|
||||
result = nil
|
||||
err = fmt.Errorf("agent loop panic: %v", r)
|
||||
}
|
||||
}()
|
||||
|
||||
// Per-run emit wrapper: enriches every AgentEvent with delegation + routing context.
|
||||
emitRun := func(event AgentEvent) {
|
||||
event.RunKind = req.RunKind
|
||||
|
||||
@@ -2,6 +2,7 @@ package scheduler
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -294,6 +295,23 @@ func (sq *SessionQueue) startOne(ctx context.Context) {
|
||||
|
||||
// executeRun runs the agent and then starts the next queued message(s) if capacity allows.
|
||||
func (sq *SessionQueue) executeRun(ctx context.Context, runID string, runGeneration uint64, pending *PendingRequest) {
|
||||
// Defense-in-depth: if runFn panics despite agent-level recovery,
|
||||
// ensure cleanup still runs so the session queue doesn't orphan this run.
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
slog.Error("scheduler: executeRun panicked", "run_id", runID, "panic", fmt.Sprint(r))
|
||||
pending.ResultCh <- RunOutcome{Err: fmt.Errorf("run panic: %v", r)}
|
||||
close(pending.ResultCh)
|
||||
sq.mu.Lock()
|
||||
delete(sq.activeRuns, runID)
|
||||
sq.removeFromOrder(runID)
|
||||
if sq.hasCapacity() && len(sq.queue) > 0 {
|
||||
sq.scheduleNext(sq.parentCtx)
|
||||
}
|
||||
sq.mu.Unlock()
|
||||
}
|
||||
}()
|
||||
|
||||
result, err := sq.runFn(ctx, pending.Req)
|
||||
pending.ResultCh <- RunOutcome{Result: result, Err: err}
|
||||
close(pending.ResultCh)
|
||||
|
||||
Reference in new issue
Block a user