From b292defd95f5ae6bc344509deb683acd8d5227d6 Mon Sep 17 00:00:00 2001 From: SiYue-ZO <2835601846@qq.com> Date: Mon, 15 Jun 2026 23:06:11 +0800 Subject: [PATCH] fix: add panic recovery to core-path goroutines Add defer-recover to 11 goroutines across 4 files to prevent ungoroutine panics from crashing the entire process: - pkg/tools/toolloop.go: parallel tool execution - pkg/channels/manager.go: HTTP server (x2), channel registration - pkg/events/subscription.go: concurrent dispatch, timeout handler, watchContext - pkg/tools/shell.go: cmd.Wait, PTY cmd.Wait, PTY read, pipe read Key design decisions: - Recover handlers send fallback values to channels (shell done, subscription done) to prevent deadlocks when the producer panics - PTY cmd.Wait sets session.Status='error' on panic for consistency - toolloop sets ErrorResult on panic so the LLM gets a meaningful response instead of a nil result - subscription.go uses log.Printf to match existing invokeHandler style - Other files use project logger (ErrorCF) with stack traces Refs: FIX-PLAN-0.3.0 #2 --- pkg/channels/manager.go | 30 ++++++++++++++++++++++++++++ pkg/events/subscription.go | 17 ++++++++++++++++ pkg/tools/shell.go | 41 ++++++++++++++++++++++++++++++++++++++ pkg/tools/toolloop.go | 12 +++++++++++ 4 files changed, 100 insertions(+) diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index c50b1999..d7c6553a 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -13,6 +13,7 @@ import ( "math" "net" "net/http" + "runtime/debug" "sort" "strings" "sync" @@ -1279,6 +1280,16 @@ func (m *Manager) StartAll(ctx context.Context) error { for _, listener := range m.httpListeners { ln := listener go func() { + defer func() { + if r := recover(); r != nil { + logger.ErrorCF("channels", "HTTP server goroutine panic recovered", + map[string]any{ + "addr": ln.Addr().String(), + "panic": fmt.Sprintf("%v", r), + "stack": string(debug.Stack()), + }) + } + }() logger.InfoCF("channels", "Shared HTTP server listening", map[string]any{ "addr": ln.Addr().String(), }) @@ -1292,6 +1303,16 @@ func (m *Manager) StartAll(ctx context.Context) error { } } else { go func() { + defer func() { + if r := recover(); r != nil { + logger.ErrorCF("channels", "HTTP server goroutine panic recovered", + map[string]any{ + "addr": m.httpServer.Addr, + "panic": fmt.Sprintf("%v", r), + "stack": string(debug.Stack()), + }) + } + }() logger.InfoCF("channels", "Shared HTTP server listening", map[string]any{ "addr": m.httpServer.Addr, }) @@ -1943,6 +1964,15 @@ func (m *Manager) Reload(ctx context.Context, cfg *config.Config) error { // Commit hashes only on full success. m.channelHashes = list go func() { + defer func() { + if r := recover(); r != nil { + logger.ErrorCF("channels", "channel registration goroutine panic recovered", + map[string]any{ + "panic": fmt.Sprintf("%v", r), + "stack": string(debug.Stack()), + }) + } + }() for _, f := range deferFuncs { f() } diff --git a/pkg/events/subscription.go b/pkg/events/subscription.go index 6619707a..d8486ee5 100644 --- a/pkg/events/subscription.go +++ b/pkg/events/subscription.go @@ -4,6 +4,7 @@ import ( "context" "errors" "log" + "runtime/debug" "sync" "sync/atomic" "time" @@ -227,6 +228,11 @@ func (s *eventSubscription) dispatch(ctx context.Context, evt Event) { s.wg.Add(1) go func() { defer s.wg.Done() + defer func() { + if r := recover(); r != nil { + log.Printf("events: subscriber %q goroutine panic recovered: %v\n%s", s.name, r, debug.Stack()) + } + }() s.handle(ctx, evt) }() case Keyed: @@ -253,6 +259,12 @@ func (s *eventSubscription) handle(ctx context.Context, evt Event) { done := make(chan handlerResult, 1) go func() { + defer func() { + if r := recover(); r != nil { + log.Printf("events: subscriber %q timeout-handler goroutine panic recovered: %v\n%s", s.name, r, debug.Stack()) + done <- handlerResult{panicked: true} + } + }() done <- s.invokeHandler(ctx, evt) }() @@ -302,6 +314,11 @@ func (s *eventSubscription) watchContext(ctx context.Context) { } go func() { + defer func() { + if r := recover(); r != nil { + log.Printf("events: subscriber %q watchContext goroutine panic recovered: %v\n%s", s.name, r, debug.Stack()) + } + }() select { case <-ctx.Done(): _ = s.Close() diff --git a/pkg/tools/shell.go b/pkg/tools/shell.go index b105b539..85bdcb8c 100644 --- a/pkg/tools/shell.go +++ b/pkg/tools/shell.go @@ -12,6 +12,7 @@ import ( "path/filepath" "regexp" "runtime" + "runtime/debug" "strings" "sync" "time" @@ -417,6 +418,16 @@ func (t *ExecTool) runSync(ctx context.Context, command, cwd string) *ToolResult done := make(chan error, 1) go func() { + defer func() { + if r := recover(); r != nil { + logger.ErrorCF("shell", "cmd.Wait goroutine panic recovered", + map[string]any{ + "panic": fmt.Sprintf("%v", r), + "stack": string(debug.Stack()), + }) + done <- fmt.Errorf("panic in cmd.Wait: %v", r) + } + }() done <- cmd.Wait() }() @@ -573,6 +584,18 @@ func (t *ExecTool) runBackground(ctx context.Context, command, cwd string, ptyEn // so we need cmd.Wait() in a separate goroutine to detect process exit. if session.PTY && session.ptyMaster != nil { go func() { + defer func() { + if r := recover(); r != nil { + logger.ErrorCF("shell", "PTY cmd.Wait goroutine panic recovered", + map[string]any{ + "panic": fmt.Sprintf("%v", r), + "stack": string(debug.Stack()), + }) + session.mu.Lock() + session.Status = "error" + session.mu.Unlock() + } + }() cmd.Wait() // Wait for process to exit session.mu.Lock() if cmd.ProcessState != nil { @@ -583,6 +606,15 @@ func (t *ExecTool) runBackground(ctx context.Context, command, cwd string, ptyEn }() go func() { + defer func() { + if r := recover(); r != nil { + logger.ErrorCF("shell", "PTY read goroutine panic recovered", + map[string]any{ + "panic": fmt.Sprintf("%v", r), + "stack": string(debug.Stack()), + }) + } + }() buf := make([]byte, 4096) for { n, err := session.ptyMaster.Read(buf) @@ -613,6 +645,15 @@ func (t *ExecTool) runBackground(ctx context.Context, command, cwd string, ptyEn // When Read() returns EOF (pipe closed), we break. // When process exits, OS closes pipe write end → Read() returns EOF → we exit. go func() { + defer func() { + if r := recover(); r != nil { + logger.ErrorCF("shell", "pipe read goroutine panic recovered", + map[string]any{ + "panic": fmt.Sprintf("%v", r), + "stack": string(debug.Stack()), + }) + } + }() buf := make([]byte, 4096) // Read stdout diff --git a/pkg/tools/toolloop.go b/pkg/tools/toolloop.go index 5cbf5662..c925c774 100644 --- a/pkg/tools/toolloop.go +++ b/pkg/tools/toolloop.go @@ -10,6 +10,7 @@ import ( "context" "encoding/json" "fmt" + "runtime/debug" "sync" "github.com/sipeed/picoclaw/pkg/logger" @@ -165,6 +166,17 @@ func RunToolLoop( wg.Add(1) go func(idx int, tc providers.ToolCall) { defer wg.Done() + defer func() { + if r := recover(); r != nil { + logger.ErrorCF("toolloop", "tool execution goroutine panic recovered", + map[string]any{ + "tool": tc.Name, + "panic": fmt.Sprintf("%v", r), + "stack": string(debug.Stack()), + }) + results[idx].result = ErrorResult(fmt.Sprintf("internal panic in tool %s", tc.Name)) + } + }() argsJSON, _ := json.Marshal(tc.Arguments) argsPreview := utils.Truncate(string(argsJSON), 200)