From c5ffaec85b2c9e5f2b2c15d1bdd37e54edd344a1 Mon Sep 17 00:00:00 2001 From: dtoro Date: Sat, 11 Jul 2026 20:05:19 +0200 Subject: [PATCH] fix(agent): panic recovery on every background goroutine (B1+B2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes B1 and B2 of plans/2026-07-11-nomos-agent-code-review.md together, since the right granularity for B1 in the auto-continuation worker turned out to require B2's restructuring anyway (see below). B1: grep -rn "recover()" cmd/nomos/ internal/mcp/ internal/httpapi/ returned nothing before this — every explicitly-spawned goroutine (continuation worker, resumed chat turns, async execution dispatch, the SSE listener, two duplicate sshExec implementations' output-collector goroutines) crashed the whole process on an unhandled panic, not just that one goroutine. More consequential post-concurrency: more simultaneous unattended background work means more surface area for one bad input to end every running task. New internal/safego package: Go(label, fn) launches fn in a goroutine with a recover-and-log wrapper. Applied at every bare `go` spawn site across the three packages. Two sites needed bespoke handling instead of the generic helper because their callers block on a channel and a silent recover would just make them hang until timeout: sshExec's output-collector goroutine (two near-identical copies, internal/mcp/server.go and internal/httpapi/phase3.go) and httpapi's ListenAndServe goroutine — both now recover AND send a synthetic error result so the waiting select unblocks immediately instead of waiting out the full timeout. httpapi's sseListener got extra treatment: its per-notification handling was extracted into handleNotification with its own recover, so a panic decoding ONE malformed pg_notify payload can't kill the listener goroutine for every connected SSE client — the outer goroutine spawn only needs to guard the connection setup/reconnect code around it. B2: cmd/nomos/continue.go's processContinuations used to run every pending continuation SEQUENTIALLY in a plain for loop, in the SAME goroutine as the ticker — meaning (a) task B's continuation waited for task A's full (up to 10-minute) resumed turn to finish first, undercutting this session's earlier concurrency work on exactly the path autonomous tasks depend on most, and (b) an unrecovered panic anywhere in that call chain didn't just crash the process (B1) — even WITH B1's recovery wrapped only at the top-level worker spawn, the panic would still unwind the ENTIRE ticker-loop goroutine, silently ending auto-continuation for every task until nomos restarted. Fixed by spawning each pending item via safego.Go individually: real parallelism, and a bad item can now only ever take down its own goroutine. Added internal/safego/safego_test.go: TestGo_RecoversPanic is the concrete proof — a deliberate panic inside Go() that would otherwise crash the whole test binary; reaching the assertion after it IS the evidence recovery works. Verified live against the rebuilt containers: full chat turn round-tripped correctly (hostname lookup, 2 iterations, normal completion) — no regression from threading safego.Go through the tool-dispatch/continuation paths. Co-Authored-By: Claude Opus 4.8 --- cmd/nomos/continue.go | 17 ++++++- cmd/nomos/main.go | 14 +++--- internal/httpapi/phase3.go | 12 ++++- internal/httpapi/server.go | 16 ++++++- internal/httpapi/sse.go | 81 ++++++++++++++++++++-------------- internal/mcp/server.go | 24 +++++++++- internal/safego/safego.go | 35 +++++++++++++++ internal/safego/safego_test.go | 39 ++++++++++++++++ 8 files changed, 195 insertions(+), 43 deletions(-) create mode 100644 internal/safego/safego.go create mode 100644 internal/safego/safego_test.go diff --git a/cmd/nomos/continue.go b/cmd/nomos/continue.go index 5113667..072cf74 100644 --- a/cmd/nomos/continue.go +++ b/cmd/nomos/continue.go @@ -9,6 +9,7 @@ import ( "strings" "time" + "github.com/dtoro/oikos/internal/safego" "github.com/google/uuid" ) @@ -55,6 +56,20 @@ func (a *agent) runContinuationWorker(ctx context.Context) { } } +// processContinuations dispatches each pending item as its OWN goroutine +// (safego.Go, so a panic deep in one task's resumed turn — JSON parsing of +// model output, an unexpected nil in a tool result — is recovered and logged +// instead of taking down this whole function, which used to run every +// item sequentially in the SAME goroutine as the ticker loop. Two problems +// that fixed: (1) throughput — task B's continuation no longer waits for +// task A's full (up to 10-minute) resumed turn to finish first, the exact +// per-task blocking this session's earlier concurrency work removed from the +// live-chat path but had left in place here; (2) survivability — since Go +// panics unwind the goroutine they occur in, an unrecovered one here used to +// mean this call (and every future tick, since the whole ticker loop runs in +// one goroutine) would simply stop — auto-continuation for every task would +// silently die until nomos restarted. Now a single bad item can only ever +// take down its own goroutine. func (a *agent) processContinuations(ctx context.Context) { pending := a.store.pendingContinuations(ctx, 5) for _, p := range pending { @@ -70,7 +85,7 @@ func (a *agent) processContinuations(ctx context.Context) { continue } a.store.markContinued(ctx, p.ExecID) // stamp first: a failure here must not cause a re-continue loop - a.continueSession(ctx, p) + safego.Go("nomos:continue-session:"+p.SessionID, func() { a.continueSession(ctx, p) }) } } diff --git a/cmd/nomos/main.go b/cmd/nomos/main.go index ac0c7b3..a12a78d 100644 --- a/cmd/nomos/main.go +++ b/cmd/nomos/main.go @@ -15,6 +15,8 @@ import ( "sync" "syscall" "time" + + "github.com/dtoro/oikos/internal/safego" ) func main() { @@ -75,9 +77,9 @@ func main() { // Event-driven auto-continuation: feed finished async executions back // into the agent so an approved plan runs to completion (and recovers // from failures) without the operator ticking it forward each step. - go nAgent.runContinuationWorker(ctx) + safego.Go("nomos:continuation-worker", func() { nAgent.runContinuationWorker(ctx) }) - go func() { + safego.Go("nomos:mcp-pool-sweeper", func() { ticker := time.NewTicker(5 * time.Minute) defer ticker.Stop() for { @@ -88,7 +90,7 @@ func main() { clientPool.sweep() } } - }() + }) mux := http.NewServeMux() mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { @@ -114,12 +116,12 @@ func main() { } srv := &http.Server{Addr: addr, Handler: mux} - go func() { + safego.Go("nomos:http-server", func() { slog.Info("nomos: gateway listening", "addr", addr, "mcp", mcpURL, "db", databaseURL != "") if err := srv.ListenAndServe(); err != http.ErrServerClosed { slog.Error("nomos: serve", "error", err) } - }() + }) <-ctx.Done() slog.Info("nomos: shutting down") @@ -344,7 +346,7 @@ func handleAnswerQuestion(w http.ResponseWriter, r *http.Request, st *store, a * if a != nil { note := fmt.Sprintf("[System: the operator answered your question %q with: %q. "+ "Continue the task from here — do not re-ask.]", prompt, req.Answer) - go a.resumeSession(context.Background(), sessionID, note) + safego.Go("nomos:resume-session", func() { a.resumeSession(context.Background(), sessionID, note) }) } w.WriteHeader(202) } diff --git a/internal/httpapi/phase3.go b/internal/httpapi/phase3.go index 2e0de96..46fa8bc 100644 --- a/internal/httpapi/phase3.go +++ b/internal/httpapi/phase3.go @@ -18,6 +18,7 @@ import ( "github.com/dtoro/oikos/internal/domain" "github.com/dtoro/oikos/internal/httpapi/gen" "github.com/dtoro/oikos/internal/observability" + "github.com/dtoro/oikos/internal/safego" "github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgtype" @@ -117,6 +118,13 @@ func sshExec(ctx context.Context, host, user, command string) (string, error) { } done := make(chan result, 1) go func() { + // See internal/mcp/server.go's sshExec for why this recovers rather + // than letting a rare SSH-library panic crash the whole api process. + defer func() { + if r := recover(); r != nil { + done <- result{nil, fmt.Errorf("panic in ssh exec: %v", r)} + } + }() out, err := session.CombinedOutput(command) done <- result{out, err} }() @@ -1453,7 +1461,9 @@ func (s *Server) DecideApproval(ctx context.Context, req gen.DecideApprovalReque // Resolve target entity slug from targetID. _ = tx.QueryRow(ctx, "SELECT slug FROM entities WHERE id = $1", targetID).Scan(&targetSlug) - go executeApprovedAction(context.Background(), s.pool, execID, targetSlug, actionStr) + safego.Go("httpapi:executeApprovedAction", func() { + executeApprovedAction(context.Background(), s.pool, execID, targetSlug, actionStr) + }) // Status only — risk_class was set correctly at request time // (e.g. by policy.ClassifyCommand for `run`); overwriting it to // a hardcoded 'config_mutation' here corrupted the audit ledger diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index fdd0968..05ffe27 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -26,6 +26,7 @@ import ( "github.com/dtoro/oikos/internal/db" "github.com/dtoro/oikos/internal/httpapi/gen" mcphandler "github.com/dtoro/oikos/internal/mcp" + "github.com/dtoro/oikos/internal/safego" "github.com/go-chi/chi/v5" "github.com/go-chi/chi/v5/middleware" "github.com/golang-jwt/jwt/v5" @@ -78,7 +79,10 @@ func NewHandler(ctx context.Context, pool *db.Pool, cfg config.Config, uiHandler } // Start background SSE listener, tied to ctx for clean shutdown. - go s.sseListener(ctx) + // handleNotification (called per-message inside sseListener's loop) has + // its own recover for the common case; this outer one covers the + // connection-setup/reconnect code around it. + safego.Go("httpapi:sse-listener", func() { s.sseListener(ctx) }) r := chi.NewRouter() r.Use(middleware.Recoverer) @@ -531,6 +535,16 @@ func ListenAndServe(ctx context.Context, pool *db.Pool, cfg config.Config, uiHan errCh := make(chan error, 1) go func() { + // Recovers a panic in ListenAndServe (stdlib, so extremely unlikely, + // but an unrecovered panic here would crash the whole process rather + // than surfacing as a normal startup error) and reports it through + // errCh instead — the select below would otherwise just hang waiting + // for a value that never arrives. + defer func() { + if r := recover(); r != nil { + errCh <- fmt.Errorf("panic in ListenAndServe: %v", r) + } + }() slog.Info("api listening", "addr", cfg.APIListen) errCh <- srv.ListenAndServe() }() diff --git a/internal/httpapi/sse.go b/internal/httpapi/sse.go index 9cd928f..1b74c72 100644 --- a/internal/httpapi/sse.go +++ b/internal/httpapi/sse.go @@ -146,41 +146,58 @@ func (s *Server) sseListener(ctx context.Context) { continue } - var p notifyPayload - if err := json.Unmarshal([]byte(nt.Payload), &p); err != nil { - slog.Error("sse listener unmarshal failed", "error", err) - continue - } - - // Fetch full event from DB - q := sqlcgen.New(s.pool) - events, err := q.ListEventsAfter(ctx, sqlcgen.ListEventsAfterParams{ - ID: p.ID - 1, - Limit: 1, - }) - if err != nil || len(events) == 0 { - slog.Warn("sse listener event fetch failed", "id", p.ID, "error", err) - continue - } - ev := events[0] - - // Push to broker - s.sseBroker.push(ev) - - // Fan out to subscribers (non-blocking send) - s.sseMu.Lock() - for sub := range s.sseSubs { - select { - case sub.ch <- ev: - default: - // Subscriber too slow — drop event for them - // (they'll reconnect via Last-Event-ID) - } - } - s.sseMu.Unlock() + s.handleNotification(ctx, nt.Payload) } } +// handleNotification processes one pg_notify payload: decode, fetch the full +// event, push to the broker, fan out to live subscribers. Split out of +// sseListener's loop specifically so it can be wrapped in its own recover — +// a panic while handling ONE notification (a malformed payload, an +// unexpected nil somewhere in the fan-out) must not kill the whole listener +// goroutine, which would silently stop the live event stream for every +// connected client until the api process is restarted. +func (s *Server) handleNotification(ctx context.Context, payload string) { + defer func() { + if r := recover(); r != nil { + slog.Error("sse listener: panic recovered handling notification", "panic", r) + } + }() + + var p notifyPayload + if err := json.Unmarshal([]byte(payload), &p); err != nil { + slog.Error("sse listener unmarshal failed", "error", err) + return + } + + // Fetch full event from DB + q := sqlcgen.New(s.pool) + events, err := q.ListEventsAfter(ctx, sqlcgen.ListEventsAfterParams{ + ID: p.ID - 1, + Limit: 1, + }) + if err != nil || len(events) == 0 { + slog.Warn("sse listener event fetch failed", "id", p.ID, "error", err) + return + } + ev := events[0] + + // Push to broker + s.sseBroker.push(ev) + + // Fan out to subscribers (non-blocking send) + s.sseMu.Lock() + for sub := range s.sseSubs { + select { + case sub.ch <- ev: + default: + // Subscriber too slow — drop event for them + // (they'll reconnect via Last-Event-ID) + } + } + s.sseMu.Unlock() +} + // sqlcEventToGen converts a DB event row to the canonical wire shape so the // SSE `data:` payload matches GET /events (snake_case keys, decoded data // object) rather than leaking Go field names and base64-encoded JSONB. diff --git a/internal/mcp/server.go b/internal/mcp/server.go index e3bd92f..1aa84b3 100644 --- a/internal/mcp/server.go +++ b/internal/mcp/server.go @@ -24,6 +24,7 @@ import ( "github.com/dtoro/oikos/internal/db/sqlcgen" "github.com/dtoro/oikos/internal/observability" "github.com/dtoro/oikos/internal/policy" + "github.com/dtoro/oikos/internal/safego" "github.com/google/jsonschema-go/jsonschema" "github.com/google/uuid" "github.com/modelcontextprotocol/go-sdk/mcp" @@ -474,7 +475,9 @@ func newServer(pool *db.Pool, agentID uuid.UUID) *mcp.Server { // tool call, cancelled the instant the chat turn's HTTP // response completes (every normal turn) — a goroutine // meant to outlive the request must not inherit its context. - go executeApprovedViaAPI(context.Background(), id, targetSlug, "apt_upgrade:"+params) + safego.Go("mcp:executeApprovedViaAPI:apt_upgrade", func() { + executeApprovedViaAPI(context.Background(), id, targetSlug, "apt_upgrade:"+params) + }) slog.Info("mcp: apt_upgrade auto-approved via assent window", "execution_id", id) return textResult(fmt.Sprintf("apt_upgrade on %s auto-approved via assent window — execution %s running.", targetSlug, id)), nil } @@ -492,7 +495,9 @@ func newServer(pool *db.Pool, agentID uuid.UUID) *mcp.Server { // See the apt_upgrade case above for why there's no // pre-flip-status "autoApprove" step here anymore, and why // this uses context.Background(). - go executeApprovedViaAPI(context.Background(), id, targetSlug, "pct_create:"+params) + safego.Go("mcp:executeApprovedViaAPI:pct_create", func() { + executeApprovedViaAPI(context.Background(), id, targetSlug, "pct_create:"+params) + }) slog.Info("mcp: pct_create auto-approved via assent window", "execution_id", id) return textResult(fmt.Sprintf("pct_create on %s auto-approved via assent window — execution %s running. The LXC is being provisioned now.", targetSlug, id)), nil } @@ -1132,6 +1137,21 @@ func sshExec(ctx context.Context, host, user, command string) (string, error) { } done := make(chan result, 1) go func() { + // Recovers a panic in CombinedOutput (SSH library internals, rare but + // not impossible) and reports it as a failed command instead of + // crashing the whole api process — every gated action runs through + // this function, so an unrecovered panic here would take down every + // concurrently-running task's execution, not just this one. Without + // this, a panic would ALSO silently degrade to "wait out the full + // timeout" (done never receives, the select below falls through to + // its time.After case) rather than crashing outright — recovering + // and sending an immediate result is strictly better: the caller + // finds out now, not after sshExecTimeout. + defer func() { + if r := recover(); r != nil { + done <- result{nil, fmt.Errorf("panic in ssh exec: %v", r)} + } + }() out, err := session.CombinedOutput(command) done <- result{out, err} }() diff --git a/internal/safego/safego.go b/internal/safego/safego.go new file mode 100644 index 0000000..0769ee1 --- /dev/null +++ b/internal/safego/safego.go @@ -0,0 +1,35 @@ +// Package safego provides a goroutine launcher that recovers panics instead +// of letting them crash the whole process. +// +// Go's default behavior for a panic in ANY goroutine — not just the one +// serving an HTTP request, which net/http recovers automatically per +// request — is to take down the entire process. This codebase runs several +// long-lived or unattended background goroutines (the nomos auto- +// continuation worker, resumed chat turns, async execution dispatch, the SSE +// event listener) that do real work — JSON parsing of model/tool output, +// map/slice indexing — with no operator watching. Before this package, a +// single edge case in any of them (a malformed tool result, an unexpected +// nil) would crash nomos or the api process outright, taking down every +// concurrently-running task or request, not just the one that hit it. +package safego + +import ( + "log/slog" + "runtime/debug" +) + +// Go runs fn in a new goroutine. A panic inside fn is recovered and logged +// (with a stack trace) instead of crashing the process. label identifies the +// goroutine in logs — use something a reader can trace back to the call +// site, e.g. "nomos:continuation-worker" or "mcp:executeApprovedViaAPI". +func Go(label string, fn func()) { + go func() { + defer func() { + if r := recover(); r != nil { + slog.Error("panic recovered in background goroutine", + "goroutine", label, "panic", r, "stack", string(debug.Stack())) + } + }() + fn() + }() +} diff --git a/internal/safego/safego_test.go b/internal/safego/safego_test.go new file mode 100644 index 0000000..abc652f --- /dev/null +++ b/internal/safego/safego_test.go @@ -0,0 +1,39 @@ +package safego + +import ( + "sync" + "testing" +) + +// TestGo_RecoversPanic is the concrete proof for the B1 fix in +// plans/2026-07-11-nomos-agent-code-review.md: a panic inside a goroutine +// launched via Go must not crash the process (or, here, the test binary — +// the same guarantee). Before this package existed, every background +// goroutine in cmd/nomos/internal/mcp/internal/httpapi used a bare `go` +// statement; an unhandled panic in any of them takes down the entire Go +// process, not just that goroutine. +func TestGo_RecoversPanic(t *testing.T) { + var wg sync.WaitGroup + wg.Add(1) + + Go("test:deliberate-panic", func() { + defer wg.Done() + panic("this must be recovered, not crash the test binary") + }) + + // If the panic weren't recovered, the whole test binary would crash + // before ever reaching this line (a Go panic in any goroutine terminates + // the process, full stop) — Wait() returning normally IS the proof. + wg.Wait() +} + +// TestGo_RunsFnNormally confirms the non-panic path still just runs fn. +func TestGo_RunsFnNormally(t *testing.T) { + done := make(chan bool, 1) + Go("test:normal", func() { + done <- true + }) + if !<-done { + t.Fatal("fn did not run") + } +}