pch/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix: add panic recovery to core-path goroutines

Browse source 
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
This commit is contained in:
SiYue-ZO 2026-06-15 23:06:11 +08:00
parent c1ff5aa6f4
commit b292defd95
4 changed files with 100 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

30
pkg/channels/manager.go
View file

@ -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()
}

17
pkg/events/subscription.go
View file

@ -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()

41
pkg/tools/shell.go
View file

@ -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

12
pkg/tools/toolloop.go
View file

@ -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)