diff --git a/internal/mcp/adapter.go b/internal/mcp/adapter.go index 5d729ab3f..38dae07bb 100644 --- a/internal/mcp/adapter.go +++ b/internal/mcp/adapter.go @@ -26,6 +26,10 @@ type Adapter struct { // registeredTools are the names THIS adapter actually owns in the // registry (collision-skips excluded, #1594-A). registeredTools []string + // breaker is the per-server circuit breaker (sa-21) shared by every + // mcpTool this adapter registers. One dead server must fast-fail all + // of its tools, so the state lives here, not per tool. + breaker *serverBreaker } // NewAdapter creates an MCP adapter from server config and tool definitions. @@ -34,6 +38,7 @@ func NewAdapter(serverName string, caller toolCaller, tools []ToolDefinition) *A serverName: serverName, caller: caller, tools: tools, + breaker: newServerBreaker(serverName), } } @@ -44,6 +49,7 @@ func NewReadOnlyAdapter(serverName string, caller toolCaller, tools []ToolDefini caller: caller, tools: tools, readOnly: true, + breaker: newServerBreaker(serverName), } } @@ -76,6 +82,7 @@ func (a *Adapter) RegisterTools(registry *tool.Registry) error { readOnly: a.readOnly, blocked: blocked, srvName: a.serverName, + breaker: a.breaker, // shared per-server state (sa-21) } if err := registry.Register(t); err != nil { // Log but continue — name collision is non-fatal. Whoever already @@ -126,6 +133,13 @@ type mcpTool struct { blocked bool srvName string + // breaker is the per-server circuit breaker (sa-21). Shared by ALL + // tools of the same server - including across Registry.Clone() copies + // (Clone copies the pointer, so swarm teammates share outage state; + // that is correct: a dead server is dead for every agent). Nil in + // hand-built test fixtures → breaker disabled, zero behavior change. + breaker *serverBreaker + // ContextFill mirrors the agent guard's fill ratio (current tokens / // compaction threshold, 0.0-1.0+). When ≥0.50 the result cap shrinks to // stay under the guard's corresponding limit, avoiding a second @@ -172,6 +186,7 @@ func (t *mcpTool) Clone() tool.Tool { readOnly: t.readOnly, blocked: t.blocked, srvName: t.srvName, + breaker: t.breaker, } } @@ -193,6 +208,14 @@ func (t *mcpTool) Parameters() json.RawMessage { } func (t *mcpTool) Execute(ctx context.Context, input json.RawMessage) (tool.Result, error) { + // sa-21: per-server circuit breaker gate. When OPEN, fail fast WITHOUT + // a transport attempt - a dead server cannot succeed, and every retry + // burns a full LLM round-trip (often 120s+ of stdio timeout). + if t.breaker != nil { + if blocked, msg := t.breaker.gate(); blocked { + return tool.Result{Content: msg, IsError: true}, nil + } + } if t.blocked { return tool.Result{ Content: fmt.Sprintf("MCP server '%s' is in read-only mode, tool '%s' is not allowed", t.srvName, t.toolName), @@ -206,6 +229,12 @@ func (t *mcpTool) Execute(ctx context.Context, input json.RawMessage) (tool.Resu } } if t.caller == nil { + // Never-connected counts as infrastructure: no tool on this server + // can work until it is (re)started, so it feeds the breaker. + err := error(errNotConnected{server: t.srvName}) + if t.breaker != nil { + t.breaker.recordFailure(err) + } return tool.Result{ Content: fmt.Sprintf("mcp[%s]: tool '%s' is not connected (server may have crashed or not started)", t.srvName, t.toolName), IsError: true, @@ -213,11 +242,24 @@ func (t *mcpTool) Execute(ctx context.Context, input json.RawMessage) (tool.Resu } result, err := t.caller.CallTool(ctx, t.toolName, args) if err != nil { + // Classify before surfacing: only transport-level failures feed the + // breaker. Semantic errors (server answered) never trip it. + if t.breaker != nil { + if isInfraError(err) { + t.breaker.recordFailure(err) + } else { + t.breaker.recordSuccess() + } + } return tool.Result{ Content: fmt.Sprintf("mcp[%s]: %s → %v", t.srvName, t.toolName, err), IsError: true, }, nil } + // The server answered (even isError=true results mean it is reachable). + if t.breaker != nil { + t.breaker.recordSuccess() + } // Extract text from content blocks var parts []string diff --git a/internal/mcp/breaker.go b/internal/mcp/breaker.go new file mode 100644 index 000000000..82fed66c9 --- /dev/null +++ b/internal/mcp/breaker.go @@ -0,0 +1,277 @@ +package mcp + +// Per-server MCP circuit breaker (sa-21). +// +// Research basis (2026 agentic-reliability engineering): +// - "Building Resilient AI Agents: Error Handling, Retries and Circuit +// Breakers" (TechMango, 2026) and "Agentic AI systems fail in storms: +// one dead dependency, N wasted retries" (2026 agent-ops surveys) - +// the canonical pattern is classify → trip → fast-fail → graceful +// degrade (tell the agent WHAT failed and WHY, let it re-plan). +// +// THE GAP IN GGCODE: an MCP server that crashed or became unreachable +// fails EVERY call to EVERY tool it exposes. Before this breaker, each +// such call burned the full pipeline - often a 120s stdio request +// timeout or the agent's adaptive tool timeout - and the model would +// typically retry the same tool (or a sibling tool of the same dead +// server) several times, costing minutes of wall-clock time and tokens +// per outage before it gave up. recurring_error.go / error_compound.go +// only inject guidance text at the trajectory level; nothing at the +// execution layer short-circuits the wasted retries. +// +// DESIGN: +// - One breaker per SERVER (not per tool): a dead server fails all of +// its tools, and the breaker state must be shared across every +// mcpTool instance, every agent clone (Registry.Clone shares the +// pointer via Clone), and sibling tools. +// - Only INFRASTRUCTURE failures trip it: connection refused/reset/ +// closed, EOF, broken pipe, DNS failure, i/o/deadline timeouts, TLS +// failures, HTTP 5xx. Semantic failures (the server answered: JSON- +// RPC "tool not found", "invalid params", or a tool result with +// isError=true) are the server WORKING - they never trip the +// breaker. Plain user-initiated context cancellation is also +// excluded (Esc is not an outage). +// - CLOSED → (threshold consecutive infra failures) → OPEN → (cooldown) +// → HALF-OPEN probe: one call allowed; success closes, infra failure +// reopens with a fresh cooldown. +// - Fail-fast message follows the permissionDeniedMessage convention: +// state what happened, why further retries waste a full LLM +// round-trip, and give the agent an actionable recovery path. +// - Zero LLM cost: deterministic counters, O(1) state. + +import ( + "fmt" + "strings" + "sync" + "time" + + "github.com/topcheer/ggcode/internal/debug" +) + +const ( + // breakerThreshold is the number of consecutive infrastructure + // failures before the circuit opens. 3 balances against transient + // hiccups (a single dropped WS frame) and against burning retries. + breakerThreshold = 3 + + // breakerCooldown is how long an OPEN circuit waits before allowing + // one half-open probe call. + breakerCooldown = 60 * time.Second +) + +type breakerState int + +const ( + breakerClosed breakerState = iota + breakerOpen + breakerHalfOpen +) + +// serverBreaker is the per-server circuit breaker shared by all mcpTool +// instances of one MCP server. All methods are goroutine-safe. +type serverBreaker struct { + mu sync.Mutex + server string + state breakerState + failures int // consecutive infra failures (closed) or current trip count + lastErr string + openedAt time.Time + probing bool // a half-open probe call is in flight + threshold int + cooldown time.Duration +} + +func newServerBreaker(server string) *serverBreaker { + return &serverBreaker{ + server: server, + threshold: breakerThreshold, + cooldown: breakerCooldown, + } +} + +// gate decides whether a tool call may proceed. When blocked is true, msg +// is a ready-to-return fast-fail error content. A blocked call is +// instantaneous (no transport I/O), which is the entire point: it saves +// the full timeout + LLM round-trip the retry would have burned. +func (b *serverBreaker) gate() (blocked bool, msg string) { + b.mu.Lock() + defer b.mu.Unlock() + switch b.state { + case breakerClosed: + return false, "" + case breakerOpen: + if time.Since(b.openedAt) >= b.cooldown { + // Cooldown elapsed: this call becomes the half-open probe. + b.state = breakerHalfOpen + b.probing = true + return false, "" + } + remain := b.cooldown - time.Since(b.openedAt) + if remain < 0 { + remain = 0 + } + return true, b.fastFailMessage(fmt.Sprintf( + "cooldown ends in ~%ds; the next call after that will be a single probe", + int(remain.Seconds()))) + case breakerHalfOpen: + if b.probing { + return true, b.fastFailMessage( + "a probe call is already in flight; await its result before retrying") + } + // Should not happen (half-open implies probing), but allow the call. + b.probing = true + return false, "" + } + return false, "" +} + +// recordFailure registers one infrastructure failure. When the consecutive +// count reaches the threshold, the circuit opens (closed→open) or reopens +// (half-open probe failed → fresh cooldown). +func (b *serverBreaker) recordFailure(err error) { + b.mu.Lock() + defer b.mu.Unlock() + b.failures++ + b.lastErr = errString(err) + switch b.state { + case breakerClosed: + if b.failures >= b.threshold { + b.state = breakerOpen + b.openedAt = time.Now() + b.probing = false + debugLogBreaker(b, "OPEN", "%d consecutive infra failures (last: %s)", b.failures, b.lastErr) + } else { + debugLogBreaker(b, "failure", "%d/%d (last: %s)", b.failures, b.threshold, b.lastErr) + } + case breakerHalfOpen: + // Probe failed: reopen with a fresh cooldown. + b.state = breakerOpen + b.openedAt = time.Now() + b.probing = false + debugLogBreaker(b, "REOPEN", "probe failed: %s", b.lastErr) + case breakerOpen: + // Redundant failure raced the trip; keep the earliest openedAt. + } +} + +// recordSuccess registers one successful (transport-level) call. A server +// semantic error (result.IsError) is still a SUCCESS for breaker purposes: +// the server answered, so it is reachable. +func (b *serverBreaker) recordSuccess() { + b.mu.Lock() + defer b.mu.Unlock() + if b.state != breakerClosed || b.failures > 0 { + debugLogBreaker(b, "CLOSED", "recovered after %d failures", b.failures) + } + b.state = breakerClosed + b.failures = 0 + b.lastErr = "" + b.probing = false +} + +// fastFailMessage renders the actionable guidance given to the model when +// the circuit is open. Mirrors permissionDeniedMessage conventions: +// what happened, why retrying wastes a full LLM round-trip, and what to +// do instead. +func (b *serverBreaker) fastFailMessage(next string) string { + return fmt.Sprintf( + "MCP server %q circuit breaker is OPEN after %d consecutive infrastructure failures "+ + "(last error: %q). All tool calls to this server are being fast-failed WITHOUT "+ + "transport attempts because each retry costs a full agent round-trip that "+ + "cannot succeed while the server is down. %s. Do NOT retry this tool now; "+ + "accomplish the task with other tools, or report the MCP server outage to "+ + "the user and ask them to restart/reconnect the server.", + b.server, b.failures, b.lastErr, next) +} + +// statusSnapshot is a read-only diagnostic view (tests / future TUI status). +type statusSnapshot struct { + State string + Failure int + LastErr string +} + +func (b *serverBreaker) snapshot() statusSnapshot { + b.mu.Lock() + defer b.mu.Unlock() + var s string + switch b.state { + case breakerClosed: + s = "closed" + case breakerOpen: + s = "open" + case breakerHalfOpen: + s = "half-open" + } + return statusSnapshot{State: s, Failure: b.failures, LastErr: b.lastErr} +} + +// debugLogBreaker emits a state-transition log; helper keeps call sites tidy. +func debugLogBreaker(b *serverBreaker, event, format string, args ...interface{}) { + debug.Log("mcp", "breaker[%s]: %s: %s", b.server, event, fmt.Sprintf(format, args...)) +} + +func errString(err error) string { + if err == nil { + return "" + } + return err.Error() +} + +// infraErrorPatterns are transport-level failure signatures. Matching is +// lowercase substring against the wrapped error chain rendered by Error(). +// This is intentionally conservative: anything NOT matched is treated as a +// semantic failure and never trips the breaker. +var infraErrorPatterns = []string{ + "connection refused", + "connection reset", + "connection closed", + "broken pipe", + "unexpected eof", + " eof", + "no such host", + "host unreachable", + "network unreachable", + "network is unreachable", + "i/o timeout", + "deadline exceeded", // includes context deadline exceeded (server hang) + "tls:", + "x509", + "http 5", + "status 5", + "server error", + "transport is closing", + "read goroutine did not return after abort", + "pipe is being closed", + "not connected", + "connection timed out", +} + +// isInfraError classifies a CallTool transport error. Semantic errors - +// the server responded with a JSON-RPC error or executed the tool and +// returned isError=true - must NOT trip the breaker. Plain user-initiated +// context cancellation ("context cancelled/canceled", ggcode's Esc path) +// matches NO infra pattern below, so no explicit exclusion is needed; +// transport aborts that EMBED a cancel ("...after abort: context canceled") +// still match infra patterns and are correctly counted as outages. +func isInfraError(err error) bool { + if err == nil { + return false + } + msg := strings.ToLower(err.Error()) + for _, p := range infraErrorPatterns { + if strings.Contains(msg, p) { + return true + } + } + return false +} + +// errNotConnected is recorded when the adapter's caller is nil (the server +// never connected). Treated as infrastructure: no tool on that server can +// work. +type errNotConnected struct{ server string } + +func (e errNotConnected) Error() string { + return fmt.Sprintf("mcp[%s]: caller not connected (server may have crashed or not started)", e.server) +} diff --git a/internal/mcp/breaker_test.go b/internal/mcp/breaker_test.go new file mode 100644 index 000000000..2c9c68784 --- /dev/null +++ b/internal/mcp/breaker_test.go @@ -0,0 +1,261 @@ +package mcp + +// sa-21: per-server MCP circuit breaker tests. + +import ( + "context" + "encoding/json" + "errors" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +func TestIsInfraErrorClassification(t *testing.T) { + infra := []string{ + "mcp[srv]: tools/call x: dial tcp 127.0.0.1:9999: connect: connection refused", + "mcp[srv]: connection closed", + "mcp[srv]: tools/call x: EOF", + "mcp[srv]: tools/call x: read tcp ...: i/o timeout", + "mcp[srv]: tools/call x: context deadline exceeded", + "mcp[srv]: read goroutine did not return after abort: context canceled", + "mcp[srv]: caller not connected (server may have crashed or not started)", + } + for _, m := range infra { + if !isInfraError(errors.New(m)) { + t.Errorf("expected INFRA: %q", m) + } + } + semantic := []string{ + "mcp[srv]: tools/call query: Tool not found: query", + "mcp[srv]: tools/call x: invalid arguments: missing param p", + "mcp[srv]: context cancelled: context canceled", // user pressed Esc + "mcp[srv]: tools/call x: permission denied for path /etc", + } + for _, m := range semantic { + if isInfraError(errors.New(m)) { + t.Errorf("expected SEMANTIC: %q", m) + } + } + if isInfraError(nil) { + t.Error("nil error must not be infra") + } +} + +func TestBreakerOpensAfterThresholdAndFastFails(t *testing.T) { + b := newServerBreaker("srv") + for i := 0; i < b.threshold; i++ { + blocked, _ := b.gate() + if blocked { + t.Fatalf("iteration %d: circuit must be closed", i) + } + b.recordFailure(errors.New("connection refused")) + } + blocked, msg := b.gate() + if !blocked { + t.Fatal("circuit must be OPEN after threshold failures") + } + if !strings.Contains(msg, `"srv"`) || !strings.Contains(msg, "Do NOT retry") { + t.Errorf("fast-fail message missing actionable guidance: %q", msg) + } +} + +func TestBreakerHalfOpenProbeThenClose(t *testing.T) { + b := newServerBreaker("srv") + b.cooldown = 10 * time.Millisecond + for i := 0; i < b.threshold; i++ { + b.recordFailure(errors.New("i/o timeout")) + } + if blocked, _ := b.gate(); !blocked { + t.Fatal("expected OPEN") + } + time.Sleep(15 * time.Millisecond) // cooldown elapses + if blocked, _ := b.gate(); blocked { + t.Fatal("after cooldown the call must become the half-open probe") + } + // Concurrent callers while the probe is in flight are fast-failed. + if blocked, msg := b.gate(); !blocked || !strings.Contains(msg, "probe") { + t.Errorf("expected probe-in-flight fast-fail, got blocked=%v msg=%q", blocked, msg) + } + b.recordSuccess() + if b.snapshot().State != "closed" { + t.Fatalf("success after probe must close, got %s", b.snapshot().State) + } + if b.snapshot().Failure != 0 { + t.Errorf("failure count must reset, got %d", b.snapshot().Failure) + } +} + +func TestBreakerProbeFailureReopens(t *testing.T) { + b := newServerBreaker("srv") + b.cooldown = 5 * time.Millisecond + for i := 0; i < b.threshold; i++ { + b.recordFailure(errors.New("connection reset")) + } + time.Sleep(10 * time.Millisecond) + if blocked, _ := b.gate(); blocked { + t.Fatal("probe must be allowed after cooldown") + } + b.recordFailure(errors.New("connection reset")) + blocked, _ := b.gate() + if !blocked { + t.Fatal("failed probe must reopen the circuit") + } + // Fresh cooldown: the reopened circuit only unlocks after a FRESH + // cooldown, not the original one (10ms elapsed here < 50ms fresh). + b.cooldown = 50 * time.Millisecond + time.Sleep(10 * time.Millisecond) + blocked, msg := b.gate() + if !blocked || !strings.Contains(msg, "OPEN") { + t.Fatalf("reopened circuit must block until a fresh cooldown; got blocked=%v msg=%q", blocked, msg) + } +} + +func TestBreakerSuccessResetsCounter(t *testing.T) { + b := newServerBreaker("srv") + b.recordFailure(errors.New("connection refused")) + b.recordFailure(errors.New("connection refused")) + b.recordSuccess() + b.recordFailure(errors.New("connection refused")) + b.recordFailure(errors.New("connection refused")) + if blocked, _ := b.gate(); blocked { + t.Fatal("success must reset the consecutive-failure counter") + } +} + +func TestMCPTripWireAcrossSiblingTools(t *testing.T) { + // Two tools of the SAME server share one breaker: trips on tool A must + // fast-fail tool B without any transport attempt. + var calls atomic.Int32 + failing := &countingCaller{calls: &calls, err: errors.New("connection refused")} + ok := &countingCaller{calls: &calls} + br := newServerBreaker("srv") + + ta := &mcpTool{name: "mcp__srv__a", caller: failing, toolName: "a", srvName: "srv", breaker: br} + tb := &mcpTool{name: "mcp__srv__b", caller: ok, toolName: "b", srvName: "srv", breaker: br} + + for i := 0; i < br.threshold; i++ { + res, _ := ta.Execute(context.Background(), json.RawMessage(`{}`)) + if !res.IsError { + t.Fatalf("call %d must fail", i) + } + } + if got := calls.Load(); got != int32(br.threshold) { + t.Fatalf("expected exactly %d transport calls, got %d", br.threshold, got) + } + res, _ := tb.Execute(context.Background(), json.RawMessage(`{}`)) + if !res.IsError { + t.Fatal("sibling tool must be fast-failed by the open breaker") + } + if calls.Load() != int32(br.threshold) { + t.Fatalf("sibling fast-fail must NOT attempt transport; calls=%d", calls.Load()) + } + if !strings.Contains(res.Content, "circuit breaker is OPEN") { + t.Errorf("sibling fast-fail must carry breaker message, got: %q", res.Content) + } +} + +func TestMCPSemanticErrorsNeverTripBreaker(t *testing.T) { + var calls atomic.Int32 + failing := &countingCaller{calls: &calls, err: errors.New("mcp[srv]: tools/call x: Tool not found: x")} + br := newServerBreaker("srv") + mt := &mcpTool{name: "mcp__srv__x", caller: failing, toolName: "x", srvName: "srv", breaker: br} + for i := 0; i < br.threshold*2; i++ { + if res, _ := mt.Execute(context.Background(), json.RawMessage(`{}`)); !res.IsError { + t.Fatal("expected error result") + } + } + if got := calls.Load(); got != int32(br.threshold*2) { + t.Fatalf("semantic errors must always reach transport; calls=%d", got) + } + if br.snapshot().State != "closed" { + t.Fatalf("semantic failures must not open the breaker, got %s", br.snapshot().State) + } +} + +func TestAdapterSharesBreakerAcrossRegisteredTools(t *testing.T) { + adapter := NewAdapter("srv", nil, []ToolDefinition{ + {Name: "a"}, {Name: "b"}, + }) + if adapter.breaker == nil { + t.Fatal("NewAdapter must create the per-server breaker") + } + adapter.breaker.threshold = 1 + adapter.breaker.recordFailure(errNotConnected{server: "srv"}) + blocked, msg := adapter.breaker.gate() + if !blocked || !strings.Contains(msg, "OPEN") { + t.Fatalf("expected open circuit, blocked=%v msg=%q", blocked, msg) + } +} + +func TestBreakerConcurrentGateDuringOpen(t *testing.T) { + b := newServerBreaker("srv") + for i := 0; i < b.threshold; i++ { + b.recordFailure(errors.New("connection refused")) + } + var wg sync.WaitGroup + var fastFailed atomic.Int32 + for i := 0; i < 50; i++ { + wg.Add(1) + go func() { + defer wg.Done() + if blocked, _ := b.gate(); blocked { + fastFailed.Add(1) + } + }() + } + wg.Wait() + if fastFailed.Load() != 50 { + t.Fatalf("all 50 concurrent gates must fast-fail while OPEN, got %d", fastFailed.Load()) + } + if b.snapshot().State != "open" { + t.Fatalf("state must remain open, got %s", b.snapshot().State) + } +} + +func TestBreakerCloneSharesBreaker(t *testing.T) { + br := newServerBreaker("srv") + mt := &mcpTool{name: "mcp__srv__a", caller: nil, toolName: "a", srvName: "srv", breaker: br} + c, ok := mt.Clone().(*mcpTool) + if !ok { + t.Fatal("Clone must return *mcpTool") + } + if c.breaker != br { + t.Fatal("Clone must share the per-server breaker pointer") + } + // nil-caller is infra and feeds the shared breaker. + for i := 0; i < br.threshold; i++ { + c.Execute(context.Background(), json.RawMessage(`{}`)) + } + if br.snapshot().State != "open" { + t.Fatalf("nil-caller failures must trip the breaker, got %s", br.snapshot().State) + } + res, _ := mt.Execute(context.Background(), json.RawMessage(`{}`)) + if !strings.Contains(res.Content, "circuit breaker is OPEN") { + t.Errorf("original tool must also fast-fail, got: %q", res.Content) + } +} + +// countingCaller is a toolCaller that counts transport attempts. +type countingCaller struct { + calls *atomic.Int32 + err error + result *CallToolResult + onCall func() +} + +func (c *countingCaller) CallTool(ctx context.Context, name string, args map[string]interface{}) (*CallToolResult, error) { + c.calls.Add(1) + if c.onCall != nil { + c.onCall() + } + if c.err != nil { + return nil, c.err + } + if c.result != nil { + return c.result, nil + } + return &CallToolResult{}, nil +}