From b73a301bf9bb4b9dc645bd69b968fd449d42cf91 Mon Sep 17 00:00:00 2001 From: Bohan Jiang <52446949+Bohan-J@users.noreply.github.com> Date: Sat, 9 May 2026 17:34:25 +0800 Subject: [PATCH] fix(agent): drain stderr before deciding ACP failure promotion (#2333) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `hermes`, `kimi`, and `kiro` all wired stderr through `cmd.Stderr = io.MultiWriter(logWriter, providerErrSniffer)`. The OS-pipe → MultiWriter copy goroutine that exec spawns for that form is only joined by `cmd.Wait()`, which the lifecycle goroutine fires in deferred cleanup — *after* `promoteACPResultOnProviderError` already consulted the sniffer. When stopReason=end_turn (success) raced ahead of the stderr drain, the sniffer's `lines` slice was empty, the helper fell through to the synthetic agent-text fallback ("hermes provider error: API call failed after 3 retries"), and the actionable upstream signal (HTTP 429 / usage limit) was lost. This was visible as a flaky `TestHermesBackendPromotesProviderErrorWithNonEmptyOutput` in CI under high parallelism — a real prod bug, not a test issue: live runs hit the same race when an upstream LLM returns 429 and hermes' synthetic agent turn beats the stderr drain to the parent. Replace the MultiWriter wiring with `cmd.StderrPipe()` + an explicit copier goroutine that signals on `stderrDone`. The lifecycle goroutine already awaits `<-readerDone` for stdout; add `<-stderrDone` next to it before `promoteACPResultOnProviderError` runs. The deferred `cmd.Wait()` ordering is unchanged — it just becomes a cheap reap by the time it fires. Verified: `go test ./pkg/agent/ -run "TestHermes|TestKimi|TestKiro" -count=10 -race`, then full package `-count=3 -race`, all green. Co-authored-by: multica-agent --- server/pkg/agent/hermes.go | 30 +++++++++++++++++++++++++++++- server/pkg/agent/kimi.go | 21 ++++++++++++++++++++- server/pkg/agent/kiro.go | 20 +++++++++++++++++++- 3 files changed, 68 insertions(+), 3 deletions(-) diff --git a/server/pkg/agent/hermes.go b/server/pkg/agent/hermes.go index af8db9c488..c14359825c 100644 --- a/server/pkg/agent/hermes.go +++ b/server/pkg/agent/hermes.go @@ -76,14 +76,35 @@ func (b *hermesBackend) Execute(ctx context.Context, prompt string, opts ExecOpt // without this we'd report a misleading "empty output" and hide // the real cause (wrong model for the current provider, bad // credentials, rate limit, …) in the daemon log. + // + // We use StderrPipe + an explicit copier goroutine instead of + // `cmd.Stderr = io.MultiWriter(...)` so we have a join point + // (`stderrDone`) before the failure-promotion decision. With the + // MultiWriter form, exec's internal copy goroutine is only + // joined by `cmd.Wait()`, which runs in the deferred cleanup — + // after `promoteACPResultOnProviderError` already consulted the + // sniffer. That race lost the 429 / usage-limit message under + // CI load and surfaced as a flaky test + // (TestHermesBackendPromotesProviderErrorWithNonEmptyOutput). providerErr := newACPProviderErrorSniffer("hermes") - cmd.Stderr = io.MultiWriter(newLogWriter(b.cfg.Logger, "[hermes:stderr] "), providerErr) + stderr, err := cmd.StderrPipe() + if err != nil { + cancel() + return nil, fmt.Errorf("hermes stderr pipe: %w", err) + } if err := cmd.Start(); err != nil { cancel() return nil, fmt.Errorf("start hermes: %w", err) } + stderrSink := io.MultiWriter(newLogWriter(b.cfg.Logger, "[hermes:stderr] "), providerErr) + stderrDone := make(chan struct{}) + go func() { + defer close(stderrDone) + _, _ = io.Copy(stderrSink, stderr) + }() + b.cfg.Logger.Info("hermes acp started", "pid", cmd.Process.Pid, "cwd", opts.Cwd) msgCh := make(chan Message, 256) @@ -307,6 +328,13 @@ func (b *hermesBackend) Execute(ctx context.Context, prompt string, opts ExecOpt // Wait for the reader goroutine to finish so all output is accumulated. <-readerDone + // Wait for the stderr copier as well so the provider-error sniffer + // has every byte the child wrote before we consult it for failure + // promotion. Skipping this leaves a small race where stopReason= + // end_turn arrives over stdout while the stderr 429 / usage-limit + // lines are still in transit, causing the promoted error message + // to fall through to the synthetic agent-text fallback. + <-stderrDone outputMu.Lock() finalOutput := output.String() diff --git a/server/pkg/agent/kimi.go b/server/pkg/agent/kimi.go index cf8cd6afe0..a4ed0ce8ba 100644 --- a/server/pkg/agent/kimi.go +++ b/server/pkg/agent/kimi.go @@ -75,14 +75,30 @@ func (b *kimiBackend) Execute(ctx context.Context, prompt string, opts ExecOptio // without this the daemon reports a misleading "empty output" // and the actionable error (expired token, rate limit, upstream // 5xx, …) stays buried in the daemon log. + // + // StderrPipe + an explicit copier give us a join point + // (`stderrDone`) that fires before the failure-promotion + // decision; see the matching comment in hermes.go for why the + // io.MultiWriter form races with stopReason=end_turn under load. providerErr := newACPProviderErrorSniffer("kimi") - cmd.Stderr = io.MultiWriter(newLogWriter(b.cfg.Logger, "[kimi:stderr] "), providerErr) + stderr, err := cmd.StderrPipe() + if err != nil { + cancel() + return nil, fmt.Errorf("kimi stderr pipe: %w", err) + } if err := cmd.Start(); err != nil { cancel() return nil, fmt.Errorf("start kimi: %w", err) } + stderrSink := io.MultiWriter(newLogWriter(b.cfg.Logger, "[kimi:stderr] "), providerErr) + stderrDone := make(chan struct{}) + go func() { + defer close(stderrDone) + _, _ = io.Copy(stderrSink, stderr) + }() + b.cfg.Logger.Info("kimi acp started", "pid", cmd.Process.Pid, "cwd", opts.Cwd) msgCh := make(chan Message, 256) @@ -297,6 +313,9 @@ func (b *kimiBackend) Execute(ctx context.Context, prompt string, opts ExecOptio cancel() <-readerDone + // Ensure the stderr copier has drained before consulting the + // provider-error sniffer; see hermes.go for the failure mode. + <-stderrDone outputMu.Lock() finalOutput := output.String() diff --git a/server/pkg/agent/kiro.go b/server/pkg/agent/kiro.go index 4a9442b97e..d9cdc0cf12 100644 --- a/server/pkg/agent/kiro.go +++ b/server/pkg/agent/kiro.go @@ -69,14 +69,29 @@ func (b *kiroBackend) Execute(ctx context.Context, prompt string, opts ExecOptio cancel() return nil, fmt.Errorf("kiro stdin pipe: %w", err) } + // StderrPipe + an explicit copier give us a join point + // (`stderrDone`) that fires before the failure-promotion + // decision; see the matching comment in hermes.go for why the + // io.MultiWriter form races with stopReason=end_turn under load. providerErr := newACPProviderErrorSniffer("kiro") - cmd.Stderr = io.MultiWriter(newLogWriter(b.cfg.Logger, "[kiro:stderr] "), providerErr) + stderr, err := cmd.StderrPipe() + if err != nil { + cancel() + return nil, fmt.Errorf("kiro stderr pipe: %w", err) + } if err := cmd.Start(); err != nil { cancel() return nil, fmt.Errorf("start kiro: %w", err) } + stderrSink := io.MultiWriter(newLogWriter(b.cfg.Logger, "[kiro:stderr] "), providerErr) + stderrDone := make(chan struct{}) + go func() { + defer close(stderrDone) + _, _ = io.Copy(stderrSink, stderr) + }() + b.cfg.Logger.Info("kiro acp started", "pid", cmd.Process.Pid, "cwd", opts.Cwd) msgCh := make(chan Message, 256) @@ -292,6 +307,9 @@ func (b *kiroBackend) Execute(ctx context.Context, prompt string, opts ExecOptio cancel() <-readerDone + // Ensure the stderr copier has drained before consulting the + // provider-error sniffer; see hermes.go for the failure mode. + <-stderrDone outputMu.Lock() finalOutput := output.String()