From 977dc6479d447a8d5fa50d8d4f07552de5817a39 Mon Sep 17 00:00:00 2001 From: devv-eve Date: Mon, 13 Apr 2026 23:00:27 -0700 Subject: [PATCH] fix(daemon): prevent task stall when agent process hangs on stdout (#947) When an agent CLI process hangs (e.g. a tool call blocks on unreachable I/O), the daemon's scanner blocks indefinitely on stdout, preventing the Result from ever being sent. This causes tasks to stay in "running" state permanently with no further events. Three-layer fix: 1. Agent backends (claude, opencode, openclaw, gemini): add a watchdog goroutine that closes the stdout/stderr pipe when the context is cancelled, forcing the scanner to unblock. Also set cmd.WaitDelay so Go force-closes pipes after 10s if the process doesn't exit. 2. daemon executeAndDrain: add an independent drain timeout (backend timeout + 30s buffer) with context-aware select on both the message channel and the result channel, so the daemon never blocks forever. 3. daemon ping path: add context-aware select so pings don't deadlock if the agent backend stalls. Closes #925 Co-authored-by: Devv Co-authored-by: Claude Opus 4.6 (1M context) --- server/internal/daemon/daemon.go | 163 +++++++++++++++++++------------ server/pkg/agent/claude.go | 7 ++ server/pkg/agent/gemini.go | 7 ++ server/pkg/agent/openclaw.go | 7 ++ server/pkg/agent/opencode.go | 7 ++ 5 files changed, 128 insertions(+), 63 deletions(-) diff --git a/server/internal/daemon/daemon.go b/server/internal/daemon/daemon.go index 54f43137e5..cec733a0dd 100644 --- a/server/internal/daemon/daemon.go +++ b/server/internal/daemon/daemon.go @@ -525,7 +525,18 @@ func (d *Daemon) handlePing(ctx context.Context, rt Runtime, pingID string) { } }() - result := <-session.Result + var result agent.Result + select { + case result = <-session.Result: + case <-pingCtx.Done(): + d.logger.Warn("ping timed out waiting for result", "runtime_id", rt.ID, "ping_id", pingID) + d.client.ReportPingResult(ctx, rt.ID, pingID, map[string]any{ + "status": "failed", + "error": "ping context cancelled while waiting for result", + "duration_ms": time.Since(start).Milliseconds(), + }) + return + } durationMs := time.Since(start).Milliseconds() if result.Status == "completed" { @@ -1078,6 +1089,17 @@ func (d *Daemon) executeAndDrain(ctx context.Context, backend agent.Backend, pro return agent.Result{}, 0, err } + // Create an independent drain deadline so we don't block forever if the + // backend's internal timeout fails to produce a Result (e.g. scanner + // stuck on a hung stdout pipe). The extra 30 s gives the backend time + // to clean up after its own timeout fires. + drainTimeout := opts.Timeout + 30*time.Second + if opts.Timeout == 0 { + drainTimeout = 21 * time.Minute + } + drainCtx, drainCancel := context.WithTimeout(ctx, drainTimeout) + defer drainCancel() + var toolCount atomic.Int32 go func() { var seq atomic.Int32 @@ -1135,77 +1157,92 @@ func (d *Daemon) executeAndDrain(ctx context.Context, backend agent.Backend, pro } }() - for msg := range session.Messages { - switch msg.Type { - case agent.MessageToolUse: - n := toolCount.Add(1) - taskLog.Info(fmt.Sprintf("tool #%d: %s", n, msg.Tool)) - if msg.CallID != "" { + for { + select { + case msg, ok := <-session.Messages: + if !ok { + goto drainDone + } + switch msg.Type { + case agent.MessageToolUse: + n := toolCount.Add(1) + taskLog.Info(fmt.Sprintf("tool #%d: %s", n, msg.Tool)) + if msg.CallID != "" { + mu.Lock() + callIDToTool[msg.CallID] = msg.Tool + mu.Unlock() + } + s := seq.Add(1) mu.Lock() - callIDToTool[msg.CallID] = msg.Tool + batch = append(batch, TaskMessageData{ + Seq: int(s), + Type: "tool_use", + Tool: msg.Tool, + Input: msg.Input, + }) + mu.Unlock() + case agent.MessageToolResult: + s := seq.Add(1) + output := msg.Output + if len(output) > 8192 { + output = output[:8192] + } + toolName := msg.Tool + if toolName == "" && msg.CallID != "" { + mu.Lock() + toolName = callIDToTool[msg.CallID] + mu.Unlock() + } + mu.Lock() + batch = append(batch, TaskMessageData{ + Seq: int(s), + Type: "tool_result", + Tool: toolName, + Output: output, + }) + mu.Unlock() + case agent.MessageThinking: + if msg.Content != "" { + mu.Lock() + pendingThinking.WriteString(msg.Content) + mu.Unlock() + } + case agent.MessageText: + if msg.Content != "" { + taskLog.Debug("agent", "text", truncateLog(msg.Content, 200)) + mu.Lock() + pendingText.WriteString(msg.Content) + mu.Unlock() + } + case agent.MessageError: + taskLog.Error("agent error", "content", msg.Content) + s := seq.Add(1) + mu.Lock() + batch = append(batch, TaskMessageData{ + Seq: int(s), + Type: "error", + Content: msg.Content, + }) mu.Unlock() } - s := seq.Add(1) - mu.Lock() - batch = append(batch, TaskMessageData{ - Seq: int(s), - Type: "tool_use", - Tool: msg.Tool, - Input: msg.Input, - }) - mu.Unlock() - case agent.MessageToolResult: - s := seq.Add(1) - output := msg.Output - if len(output) > 8192 { - output = output[:8192] - } - toolName := msg.Tool - if toolName == "" && msg.CallID != "" { - mu.Lock() - toolName = callIDToTool[msg.CallID] - mu.Unlock() - } - mu.Lock() - batch = append(batch, TaskMessageData{ - Seq: int(s), - Type: "tool_result", - Tool: toolName, - Output: output, - }) - mu.Unlock() - case agent.MessageThinking: - if msg.Content != "" { - mu.Lock() - pendingThinking.WriteString(msg.Content) - mu.Unlock() - } - case agent.MessageText: - if msg.Content != "" { - taskLog.Debug("agent", "text", truncateLog(msg.Content, 200)) - mu.Lock() - pendingText.WriteString(msg.Content) - mu.Unlock() - } - case agent.MessageError: - taskLog.Error("agent error", "content", msg.Content) - s := seq.Add(1) - mu.Lock() - batch = append(batch, TaskMessageData{ - Seq: int(s), - Type: "error", - Content: msg.Content, - }) - mu.Unlock() + case <-drainCtx.Done(): + goto drainDone } } - + drainDone: close(done) flush() }() - result := <-session.Result - return result, toolCount.Load(), nil + select { + case result := <-session.Result: + return result, toolCount.Load(), nil + case <-drainCtx.Done(): + return agent.Result{ + Status: "timeout", + Error: "agent did not produce result within drain timeout", + }, toolCount.Load(), nil + } } func mergeUsage(a, b map[string]agent.TokenUsage) map[string]agent.TokenUsage { diff --git a/server/pkg/agent/claude.go b/server/pkg/agent/claude.go index c4e4c95533..dbf98cf94f 100644 --- a/server/pkg/agent/claude.go +++ b/server/pkg/agent/claude.go @@ -37,6 +37,7 @@ func (b *claudeBackend) Execute(ctx context.Context, prompt string, opts ExecOpt args := buildClaudeArgs(opts) cmd := exec.CommandContext(runCtx, execPath, args...) + cmd.WaitDelay = 10 * time.Second if opts.Cwd != "" { cmd.Dir = opts.Cwd } @@ -90,6 +91,12 @@ func (b *claudeBackend) Execute(ctx context.Context, prompt string, opts ExecOpt var finalError string usage := make(map[string]TokenUsage) + // Close stdout when the context is cancelled so scanner.Scan() unblocks. + go func() { + <-runCtx.Done() + _ = stdout.Close() + }() + scanner := bufio.NewScanner(stdout) scanner.Buffer(make([]byte, 0, 1024*1024), 10*1024*1024) diff --git a/server/pkg/agent/gemini.go b/server/pkg/agent/gemini.go index b42616faac..e34fffdba7 100644 --- a/server/pkg/agent/gemini.go +++ b/server/pkg/agent/gemini.go @@ -39,6 +39,7 @@ func (b *geminiBackend) Execute(ctx context.Context, prompt string, opts ExecOpt args := buildGeminiArgs(prompt, opts) cmd := exec.CommandContext(runCtx, execPath, args...) + cmd.WaitDelay = 10 * time.Second if opts.Cwd != "" { cmd.Dir = opts.Cwd } @@ -61,6 +62,12 @@ func (b *geminiBackend) Execute(ctx context.Context, prompt string, opts ExecOpt msgCh := make(chan Message, 16) resCh := make(chan Result, 1) + // Close stdout when the context is cancelled so io.ReadAll unblocks. + go func() { + <-runCtx.Done() + _ = stdout.Close() + }() + go func() { defer cancel() defer close(msgCh) diff --git a/server/pkg/agent/openclaw.go b/server/pkg/agent/openclaw.go index f2e056ad44..0ed2df53a3 100644 --- a/server/pkg/agent/openclaw.go +++ b/server/pkg/agent/openclaw.go @@ -44,6 +44,7 @@ func (b *openclawBackend) Execute(ctx context.Context, prompt string, opts ExecO args = append(args, "--message", prompt) cmd := exec.CommandContext(runCtx, execPath, args...) + cmd.WaitDelay = 10 * time.Second if opts.Cwd != "" { cmd.Dir = opts.Cwd } @@ -67,6 +68,12 @@ func (b *openclawBackend) Execute(ctx context.Context, prompt string, opts ExecO msgCh := make(chan Message, 256) resCh := make(chan Result, 1) + // Close stderr when the context is cancelled so the scanner unblocks. + go func() { + <-runCtx.Done() + _ = stderr.Close() + }() + go func() { defer cancel() defer close(msgCh) diff --git a/server/pkg/agent/opencode.go b/server/pkg/agent/opencode.go index 4c17a039dd..8901ea8ffb 100644 --- a/server/pkg/agent/opencode.go +++ b/server/pkg/agent/opencode.go @@ -48,6 +48,7 @@ func (b *opencodeBackend) Execute(ctx context.Context, prompt string, opts ExecO args = append(args, prompt) cmd := exec.CommandContext(runCtx, execPath, args...) + cmd.WaitDelay = 10 * time.Second if opts.Cwd != "" { cmd.Dir = opts.Cwd } @@ -74,6 +75,12 @@ func (b *opencodeBackend) Execute(ctx context.Context, prompt string, opts ExecO msgCh := make(chan Message, 256) resCh := make(chan Result, 1) + // Close stdout when the context is cancelled so the scanner unblocks. + go func() { + <-runCtx.Done() + _ = stdout.Close() + }() + go func() { defer cancel() defer close(msgCh)