From 81a60a0847d3237656a332e12286830e3df9abb3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Manso?= Date: Sat, 8 Aug 2026 14:42:26 +0100 Subject: [PATCH] fix(agentctl): explain why a one-shot ACP probe failed (#2417) --- .../agentctl/server/utility/acp_executor.go | 100 +++++++++++++++- .../server/utility/acp_executor_test.go | 113 ++++++++++++++++++ 2 files changed, 209 insertions(+), 4 deletions(-) diff --git a/apps/backend/internal/agentctl/server/utility/acp_executor.go b/apps/backend/internal/agentctl/server/utility/acp_executor.go index 460e68a4f2..6732e45dd3 100644 --- a/apps/backend/internal/agentctl/server/utility/acp_executor.go +++ b/apps/backend/internal/agentctl/server/utility/acp_executor.go @@ -1,6 +1,7 @@ package utility import ( + "bytes" "context" "errors" "fmt" @@ -83,6 +84,11 @@ func (e *ACPInferenceExecutor) Execute(ctx context.Context, req *PromptRequest) cmd.Env = sanitizeEnvForAgent(req.InferenceConfig) configureACPCommand(cmd, e.logger) + // Same reasoning as the probe: without this the child's own account of why + // it died is discarded. + var stderr stderrBuffer + cmd.Stderr = &stderr + stdin, err := cmd.StdinPipe() if err != nil { return &PromptResponse{Success: false, Error: fmt.Sprintf("stdin pipe: %v", err)}, nil @@ -112,6 +118,10 @@ func (e *ACPInferenceExecutor) Execute(ctx context.Context, req *PromptRequest) } response, err := e.executeACPSession(ctx, stdin, stdout, workDir, req.AgentID, req.Prompt, model, modelConfigOptions, req.Mode, mcpServers) if err != nil { + e.logger.Error("ACP inference failed", + zap.String("agent_id", req.AgentID), + zap.Error(err), + zap.String("stderr", stderr.tail())) return &PromptResponse{ Success: false, Error: err.Error(), @@ -188,7 +198,7 @@ func (e *ACPInferenceExecutor) executeACPSession( }, }) if err != nil { - return "", fmt.Errorf("ACP initialize failed: %w", err) + return "", describeACPFailure(ctx, "initialize", err) } // Create new session. ACP requires McpServers to be a non-nil slice; @@ -201,7 +211,7 @@ func (e *ACPInferenceExecutor) executeACPSession( McpServers: mcpServers, }) if err != nil { - return "", fmt.Errorf("ACP session/new failed: %w", err) + return "", describeACPFailure(ctx, "session/new", err) } sessionID := sessionResp.SessionId @@ -371,6 +381,14 @@ func (e *ACPInferenceExecutor) Probe(ctx context.Context, req *ProbeRequest) (*P cmd.Env = sanitizeEnvForAgent(req.InferenceConfig) configureACPCommand(cmd, e.logger) + // Keep the child's stderr. A probe that dies before answering otherwise + // reports only "peer disconnected before response", and the reason the + // process gave — an npx resolution failure, a missing runtime, a panic — + // goes to the void, leaving the UI's "check agent logs" pointing at logs + // that never carried it. + var stderr stderrBuffer + cmd.Stderr = &stderr + stdin, err := cmd.StdinPipe() if err != nil { return &ProbeResponse{Success: false, Error: fmt.Sprintf("stdin pipe: %v", err)}, nil @@ -391,6 +409,14 @@ func (e *ACPInferenceExecutor) Probe(ctx context.Context, req *ProbeRequest) (*P resp, err := e.probeACPSession(ctx, stdin, stdout, workDir, req.AgentID) if err != nil { + // The tail goes to the log rather than into the response: handlers + // deliberately keep subprocess diagnostics and tmp paths out of client + // payloads. An empty tail is itself a result — the child produced no + // diagnostics, rather than producing some we failed to record. + e.logger.Error("ACP probe failed", + zap.String("agent_id", req.AgentID), + zap.Error(err), + zap.String("stderr", stderr.tail())) return &ProbeResponse{ Success: false, Error: err.Error(), @@ -567,7 +593,7 @@ func (e *ACPInferenceExecutor) probeACPSession( }, }) if err != nil { - return nil, fmt.Errorf("ACP initialize failed: %w", err) + return nil, describeACPFailure(ctx, "initialize", err) } sessionResp, err := conn.NewSession(ctx, acp.NewSessionRequest{ @@ -575,7 +601,7 @@ func (e *ACPInferenceExecutor) probeACPSession( McpServers: []acp.McpServer{}, }) if err != nil { - return nil, fmt.Errorf("ACP session/new failed: %w", err) + return nil, describeACPFailure(ctx, "session/new", err) } // Wait up to 1s for the available_commands_update notification. Agents @@ -854,6 +880,72 @@ func resolveProbeCommand(name string) string { return allowedProbeCommands[filepath.Base(name)] } +// stderrTailLimit bounds how much of a spawned agent's stderr reaches the log: +// enough for an npm failure or the head of a panic, small enough that a chatty +// agent cannot flood it. +const stderrTailLimit = 4 << 10 + +// stderrBuffer collects a spawned agent's stderr. +// +// os/exec copies stderr on a goroutine of its own whenever the writer is not an +// *os.File, and the tail is read while the child is still being torn down — +// cleanup calls Wait from a defer, after the response has been built. Writer and +// reader therefore overlap, and the mutex is what makes that legal: an +// unsynchronised bytes.Buffer here trips the race detector in +// TestProbeCleansUpDescendantProcessOnTimeout. +type stderrBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +// Write keeps only the trailing stderrTailLimit bytes, so an agent that chatters +// for the length of a probe cannot grow this without bound. Trimming on read +// instead would still have retained every byte the child ever wrote. The full +// length of p is reported as written: a short count means failure to io.Writer, +// and discarding an old prefix is not a failure to record the new bytes. +func (b *stderrBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + written := len(p) + if len(p) > stderrTailLimit { + p = p[len(p)-stderrTailLimit:] + } + b.buf.Write(p) + if excess := b.buf.Len() - stderrTailLimit; excess > 0 { + b.buf.Next(excess) + } + return written, nil +} + +// tail returns what was retained, trimmed — the end of the output rather than +// the beginning, because what a dying process printed last is what explains it. +// It is a best-effort snapshot: the child may still be writing while its process +// group is torn down, so the tail can be short or empty. +func (b *stderrBuffer) tail() string { + b.mu.Lock() + defer b.mu.Unlock() + return strings.TrimSpace(b.buf.String()) +} + +// describeACPFailure names an expired or cancelled context as the cause instead +// of reporting whatever the SDK saw on the wire. +// +// When the deadline fires, cleanup kills the child's process group; the ACP +// client then hits EOF on stdout and returns "peer disconnected before +// response". That makes a plain timeout indistinguishable from an agent that +// crashed — two failures calling for opposite responses, one of which is to +// wait longer and the other to go read the agent's own output. +func describeACPFailure(ctx context.Context, phase string, err error) error { + switch { + case errors.Is(ctx.Err(), context.DeadlineExceeded): + return fmt.Errorf("ACP %s timed out: %w", phase, ctx.Err()) + case errors.Is(ctx.Err(), context.Canceled): + return fmt.Errorf("ACP %s cancelled: %w", phase, ctx.Err()) + default: + return fmt.Errorf("ACP %s failed: %w", phase, err) + } +} + // sanitizeEnvForAgent returns a child-process environment with agent-declared // variables (InferenceConfigDTO.StripEnv) removed. Applied to one-shot // probe/inference subprocesses; the persistent session path strips in diff --git a/apps/backend/internal/agentctl/server/utility/acp_executor_test.go b/apps/backend/internal/agentctl/server/utility/acp_executor_test.go index 6d65c49f29..715f847646 100644 --- a/apps/backend/internal/agentctl/server/utility/acp_executor_test.go +++ b/apps/backend/internal/agentctl/server/utility/acp_executor_test.go @@ -1,7 +1,9 @@ package utility import ( + "context" "encoding/json" + "errors" "maps" "path/filepath" "slices" @@ -13,6 +15,117 @@ import ( func ptr[T any](v T) *T { return &v } +// A probe that outlives its deadline is killed by cleanup, so the ACP client +// reports the dead pipe rather than the deadline. Reporting that verbatim sends +// readers after a crashed agent that never crashed. +func TestDescribeACPFailureNamesTheContextCause(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + err := describeACPFailure(ctx, "initialize", errors.New("peer disconnected before response")) + if !errors.Is(err, context.Canceled) { + t.Fatalf("err = %v, want it to wrap context.Canceled", err) + } + if strings.Contains(err.Error(), "peer disconnected") { + t.Fatalf("err = %q, want the context cause instead of the wire symptom", err) + } +} + +// The deadline is the case the change exists for: the probe timeout kills the +// child, the SDK reports the dead pipe, and the timeout has to win over it. +func TestDescribeACPFailureNamesTheDeadline(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithTimeout(context.Background(), 0) + defer cancel() + <-ctx.Done() + + err := describeACPFailure(ctx, "initialize", errors.New("peer disconnected before response")) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("err = %v, want it to wrap context.DeadlineExceeded", err) + } + if !strings.Contains(err.Error(), "timed out") { + t.Fatalf("err = %q, want it to say the phase timed out", err) + } +} + +func TestDescribeACPFailureKeepsGenuineErrors(t *testing.T) { + t.Parallel() + + inner := errors.New("peer disconnected before response") + err := describeACPFailure(context.Background(), "initialize", inner) + if !errors.Is(err, inner) { + t.Fatalf("err = %v, want the original failure preserved", err) + } + if strings.Contains(err.Error(), "timed out") { + t.Fatalf("err = %q, must not claim a timeout while the context is live", err) + } +} + +func TestStderrBufferTailKeepsTheEnd(t *testing.T) { + t.Parallel() + + var buf stderrBuffer + if _, err := buf.Write([]byte(strings.Repeat("a", stderrTailLimit))); err != nil { + t.Fatalf("write: %v", err) + } + if _, err := buf.Write([]byte("\nnpm error code ECONNREFUSED")); err != nil { + t.Fatalf("write: %v", err) + } + + tail := buf.tail() + if len(tail) > stderrTailLimit { + t.Fatalf("tail is %d bytes, want at most %d", len(tail), stderrTailLimit) + } + if !strings.HasSuffix(tail, "npm error code ECONNREFUSED") { + t.Fatalf("tail = %q, want the final line kept", tail) + } +} + +// Retention is bounded on write, not on read, so a chatty child cannot hold +// everything it ever printed in memory until something asks for the tail. +func TestStderrBufferBoundsRetentionOnWrite(t *testing.T) { + t.Parallel() + + var buf stderrBuffer + chunk := []byte(strings.Repeat("b", 1024)) + for range 64 { + if n, err := buf.Write(chunk); err != nil || n != len(chunk) { + t.Fatalf("Write = (%d, %v), want (%d, nil)", n, err, len(chunk)) + } + } + + if got := buf.buf.Len(); got > stderrTailLimit { + t.Fatalf("retained %d bytes after 64 KiB of writes, want at most %d", got, stderrTailLimit) + } +} + +// os/exec writes into this buffer from its stderr-copying goroutine while the +// tail is read during teardown. Exercising both at once means the race detector +// asserts the synchronisation rather than it being assumed. +func TestStderrBufferSurvivesConcurrentUse(t *testing.T) { + t.Parallel() + + var buf stderrBuffer + written := make(chan struct{}) + go func() { + defer close(written) + for range 200 { + _, _ = buf.Write([]byte("npm error code ECONNREFUSED\n")) + } + }() + for range 200 { + _ = buf.tail() + } + <-written + + if buf.tail() == "" { + t.Fatal("tail is empty after concurrent writes") + } +} + func TestProbeConfigOptions_PreservesDescriptions(t *testing.T) { t.Parallel()