diff --git a/apps/backend/internal/agent/runtime/agentctl/agent.go b/apps/backend/internal/agent/runtime/agentctl/agent.go index 49e767acc3..692c6b71cd 100644 --- a/apps/backend/internal/agent/runtime/agentctl/agent.go +++ b/apps/backend/internal/agent/runtime/agentctl/agent.go @@ -3,6 +3,7 @@ package client import ( "context" "encoding/json" + "errors" "fmt" "sync" "time" @@ -477,15 +478,13 @@ func (c *Client) readUpdatesStream( // This is an MCP request - dispatch it if wsMsg.Type == ws.MessageTypeRequest { - if mcpHandler != nil { - sessionID, pendingID := extractMCPRequestCorrelation(wsMsg.Payload) - c.logger.Debug("received MCP request from agent stream", - zap.String("request_id", wsMsg.ID), - zap.String("action", wsMsg.Action), - zap.String("session_id", sessionID), - zap.String("pending_id", pendingID)) - go c.dispatchMCPRequest(ctx, wsMsg, mcpHandler, writeMessage) - } + sessionID, pendingID := extractMCPRequestCorrelation(wsMsg.Payload) + c.logger.Info("received MCP request from agent stream", + zap.String("request_id", wsMsg.ID), + zap.String("action", wsMsg.Action), + zap.String("session_id", sessionID), + zap.String("pending_id", pendingID)) + go c.dispatchMCPRequest(ctx, wsMsg, mcpHandler, writeMessage) continue } } @@ -603,7 +602,13 @@ func (c *Client) dispatchMCPRequest(ctx context.Context, msg ws.Message, mcpHand zap.String("session_id", sessionID), zap.String("pending_id", pendingID)) - resp, err := mcpHandler.Dispatch(ctx, &msg) + var resp *ws.Message + var err error + if mcpHandler == nil { + err = errors.New("MCP handler is not configured") + } else { + resp, err = mcpHandler.Dispatch(ctx, &msg) + } if err != nil { c.logger.Error("MCP dispatch error", zap.String("request_id", msg.ID), @@ -614,6 +619,16 @@ func (c *Client) dispatchMCPRequest(ctx context.Context, msg ws.Message, mcpHand zap.Error(err)) tracing.TraceMCPResponse(span, err) resp, _ = ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) + } else if resp == nil { + err = errors.New("MCP handler returned no response") + c.logger.Error("MCP dispatch returned no response", + zap.String("request_id", msg.ID), + zap.String("action", msg.Action), + zap.String("session_id", sessionID), + zap.String("pending_id", pendingID), + zap.Duration("duration", time.Since(start))) + tracing.TraceMCPResponse(span, err) + resp, _ = ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) } else { c.logger.Debug("MCP request dispatched", zap.String("request_id", msg.ID), diff --git a/apps/backend/internal/agent/runtime/agentctl/agent_mcp_stream_test.go b/apps/backend/internal/agent/runtime/agentctl/agent_mcp_stream_test.go new file mode 100644 index 0000000000..453d98e7eb --- /dev/null +++ b/apps/backend/internal/agent/runtime/agentctl/agent_mcp_stream_test.go @@ -0,0 +1,70 @@ +package client + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/gorilla/websocket" + ws "github.com/kandev/kandev/pkg/websocket" +) + +// @covers AC-AGENTS-MCP-BRIDGE-RELIABILITY-001.3 +func TestStreamUpdatesReturnsCorrelatedErrorWhenMCPHandlerIsUnavailable(t *testing.T) { + type serverResult struct { + message ws.Message + err error + } + resultCh := make(chan serverResult, 1) + upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + resultCh <- serverResult{err: err} + return + } + defer func() { _ = conn.Close() }() + request, _ := ws.NewRequest("req-unavailable", "mcp.tools.call", nil) + data, _ := json.Marshal(request) + if err := conn.WriteMessage(websocket.TextMessage, data); err != nil { + resultCh <- serverResult{err: err} + return + } + _ = conn.SetReadDeadline(time.Now().Add(500 * time.Millisecond)) + _, data, err = conn.ReadMessage() + if err != nil { + resultCh <- serverResult{err: err} + return + } + var response ws.Message + err = json.Unmarshal(data, &response) + resultCh <- serverResult{message: response, err: err} + })) + t.Cleanup(server.Close) + + client := &Client{ + baseURL: server.URL, + httpClient: &http.Client{Timeout: time.Second}, + logger: newTestLogger(), + pendingRequests: make(map[string]chan *ws.Message), + } + t.Cleanup(client.Close) + if err := client.StreamUpdates(context.Background(), func(AgentEvent) {}, nil, nil); err != nil { + t.Fatalf("StreamUpdates: %v", err) + } + + select { + case result := <-resultCh: + if result.err != nil { + t.Fatalf("read response: %v", result.err) + } + if result.message.Type != ws.MessageTypeError || result.message.ID != "req-unavailable" { + t.Fatalf("response = %+v, want correlated error", result.message) + } + case <-time.After(time.Second): + t.Fatal("timed out waiting for unavailable-handler response") + } +} diff --git a/apps/backend/internal/agent/runtime/agentctl/agent_session_test.go b/apps/backend/internal/agent/runtime/agentctl/agent_session_test.go index 1e0afacd4f..28cf8d1941 100644 --- a/apps/backend/internal/agent/runtime/agentctl/agent_session_test.go +++ b/apps/backend/internal/agent/runtime/agentctl/agent_session_test.go @@ -411,9 +411,8 @@ type errUnauthorizedStub struct{} func (errUnauthorizedStub) Error() string { return "not authorized" } -// A nil response with no error means the handler intentionally answered -// nothing (e.g. a notification); nothing should go on the wire. -func TestDispatchMCPRequest_WritesNothingForANilResponse(t *testing.T) { +// @covers AC-AGENTS-MCP-BRIDGE-RELIABILITY-001.4 +func TestDispatchMCPRequest_WritesAnErrorFrameForANilResponse(t *testing.T) { handler := &stubMCPHandler{} var written [][]byte @@ -423,8 +422,15 @@ func TestDispatchMCPRequest_WritesNothingForANilResponse(t *testing.T) { handler, func(data []byte) error { written = append(written, data); return nil }) - if len(written) != 0 { - t.Errorf("wrote %d frames, want none for a nil handler response", len(written)) + if len(written) != 1 { + t.Fatalf("wrote %d frames, want 1 error frame", len(written)) + } + var sent ws.Message + if err := json.Unmarshal(written[0], &sent); err != nil { + t.Fatalf("decode written frame: %v", err) + } + if sent.Type != ws.MessageTypeError || sent.ID != "req-3" { + t.Errorf("written frame = %+v, want a correlated error", sent) } } diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager.go b/apps/backend/internal/agent/runtime/lifecycle/manager.go index 4b85efbe23..df909ce4d0 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager.go @@ -79,10 +79,6 @@ type Manager struct { // preparerRegistry maps executor types to environment preparers. preparerRegistry *PreparerRegistry - // mcpHandler is the MCP request dispatcher for handling MCP requests - // from agentctl instances through the agent stream. - mcpHandler agentctl.MCPHandler - // sessionAccessCheck enforces per-user workspace scoping on session-scoped // surfaces (opt-in auth). Nil = no scoping. See SetSessionAccessChecker. sessionAccessCheck func(ctx context.Context, sessionID string) error @@ -421,12 +417,10 @@ func (m *Manager) WorktreeManager() *worktree.Manager { // where they are dispatched to this handler. This enables agents to use MCP tools // like listing workspaces, boards, tasks, and asking user questions. // -// This must be called before agents start making MCP calls. Typically set during -// initialization after the MCP handlers are created. +// Streams resolve the current handler for each request, so recovered streams +// can connect before startup wiring installs the dispatcher. func (m *Manager) SetMCPHandler(handler agentctl.MCPHandler) { - m.mcpHandler = handler - // Update the stream manager with the handler - m.streamManager.mcpHandler = handler + m.streamManager.setMCPHandler(handler) } // SetMCPIdentityScoper installs the per-user scoping hook for in-session MCP @@ -442,14 +436,14 @@ func (m *Manager) SetMCPHandler(handler agentctl.MCPHandler) { // Set once during startup wiring, before agents start making MCP calls. Leave // unset to keep dispatch unscoped (single-user instances). func (m *Manager) SetMCPIdentityScoper(scoper MCPIdentityScoper) { - m.streamManager.mcpIdentityScoper = scoper + m.streamManager.setMCPIdentityScoper(scoper) } // SetMCPPrincipalScoper installs the trusted in-session MCP principal resolver. // The resolver derives automation identity and workspace boundaries from the // execution's own task and session, never from the agent request payload. func (m *Manager) SetMCPPrincipalScoper(scoper MCPPrincipalScoper) { - m.streamManager.mcpPrincipalScoper = scoper + m.streamManager.setMCPPrincipalScoper(scoper) } // SetSessionAccessChecker installs the per-user session visibility check used diff --git a/apps/backend/internal/agent/runtime/lifecycle/mcp_handler_binding_test.go b/apps/backend/internal/agent/runtime/lifecycle/mcp_handler_binding_test.go new file mode 100644 index 0000000000..a11bebbf55 --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/mcp_handler_binding_test.go @@ -0,0 +1,31 @@ +package lifecycle + +import ( + "context" + "testing" +) + +// @covers AC-AGENTS-MCP-BRIDGE-RELIABILITY-001.1 +func TestMCPHandlerForUsesDispatcherInstalledAfterStreamCreation(t *testing.T) { + log := newTestLogger() + sm := NewStreamManager(log, StreamCallbacks{}, nil, nil) + handler := sm.mcpHandlerFor(&AgentExecution{ + ID: "exec-recovered", + TaskID: "task-recovered", + SessionID: "session-recovered", + }) + if handler == nil { + t.Fatal("stream captured no MCP handler before dispatcher setup") + } + + inner := &recordingMCPHandler{} + manager := &Manager{streamManager: sm} + manager.SetMCPHandler(inner) + + if _, err := handler.Dispatch(context.Background(), mcpRequest(t, nil)); err != nil { + t.Fatalf("Dispatch: %v", err) + } + if inner.calls != 1 { + t.Fatalf("dispatcher calls = %d, want 1", inner.calls) + } +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/mcp_identity.go b/apps/backend/internal/agent/runtime/lifecycle/mcp_identity.go index 4c55cdeb97..e51eefea80 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/mcp_identity.go +++ b/apps/backend/internal/agent/runtime/lifecycle/mcp_identity.go @@ -2,6 +2,7 @@ package lifecycle import ( "context" + "errors" "go.uber.org/zap" @@ -11,6 +12,8 @@ import ( ws "github.com/kandev/kandev/pkg/websocket" ) +var errMCPHandlerUnavailable = errors.New("MCP handler is not configured") + // MCPIdentityScoper attaches the identity of the user who owns taskID to ctx, // so the in-session MCP tools an agent gets automatically are authorized as // that user instead of running unscoped. Returning an error denies the @@ -41,6 +44,32 @@ type taskScopedMCPHandler struct { logger *logger.Logger } +type currentMCPHandler struct { + streamManager *StreamManager + executionID string + taskID string + sessionID string +} + +func (h *currentMCPHandler) Dispatch(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + inner, scope, principal := h.streamManager.mcpDispatchState() + if inner == nil { + return nil, errMCPHandlerUnavailable + } + if h.taskID == "" { + return inner.Dispatch(ctx, msg) + } + return (&taskScopedMCPHandler{ + inner: inner, + scope: scope, + principal: principal, + executionID: h.executionID, + taskID: h.taskID, + sessionID: h.sessionID, + logger: h.streamManager.logger, + }).Dispatch(ctx, msg) +} + func (h *taskScopedMCPHandler) Dispatch(ctx context.Context, msg *ws.Message) (*ws.Message, error) { ctx = streams.WithMCPExecutionContext(ctx, streams.MCPExecutionContext{ ExecutionID: h.executionID, @@ -80,16 +109,34 @@ func (h *taskScopedMCPHandler) Dispatch(ctx context.Context, msg *ws.Message) (* // backend-owned execution identity is always attached; user identity is also // attached when per-user scoping has been wired. func (sm *StreamManager) mcpHandlerFor(execution *AgentExecution) agentctl.MCPHandler { - if sm.mcpHandler == nil || execution.TaskID == "" { - return sm.mcpHandler - } - return &taskScopedMCPHandler{ - inner: sm.mcpHandler, - scope: sm.mcpIdentityScoper, - principal: sm.mcpPrincipalScoper, - executionID: execution.ID, - taskID: execution.TaskID, - sessionID: execution.SessionID, - logger: sm.logger, + return ¤tMCPHandler{ + streamManager: sm, + executionID: execution.ID, + taskID: execution.TaskID, + sessionID: execution.SessionID, } } + +func (sm *StreamManager) mcpDispatchState() (agentctl.MCPHandler, MCPIdentityScoper, MCPPrincipalScoper) { + sm.mcpMu.RLock() + defer sm.mcpMu.RUnlock() + return sm.mcpHandler, sm.mcpIdentityScoper, sm.mcpPrincipalScoper +} + +func (sm *StreamManager) setMCPHandler(handler agentctl.MCPHandler) { + sm.mcpMu.Lock() + defer sm.mcpMu.Unlock() + sm.mcpHandler = handler +} + +func (sm *StreamManager) setMCPIdentityScoper(scoper MCPIdentityScoper) { + sm.mcpMu.Lock() + defer sm.mcpMu.Unlock() + sm.mcpIdentityScoper = scoper +} + +func (sm *StreamManager) setMCPPrincipalScoper(scoper MCPPrincipalScoper) { + sm.mcpMu.Lock() + defer sm.mcpMu.Unlock() + sm.mcpPrincipalScoper = scoper +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/mcp_identity_test.go b/apps/backend/internal/agent/runtime/lifecycle/mcp_identity_test.go index 360861f3b0..992208f613 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/mcp_identity_test.go +++ b/apps/backend/internal/agent/runtime/lifecycle/mcp_identity_test.go @@ -32,7 +32,7 @@ func newMCPStreamManager(t *testing.T, inner *recordingMCPHandler, scoper MCPIde t.Fatalf("logger: %v", err) } sm := NewStreamManager(log, StreamCallbacks{}, inner, nil) - sm.mcpIdentityScoper = scoper + sm.setMCPIdentityScoper(scoper) return sm } @@ -171,8 +171,12 @@ func TestMCPHandlerForPassesThroughWithoutTaskID(t *testing.T) { return ctx, nil }) - if got := sm.mcpHandlerFor(&AgentExecution{ID: "exec-1"}); got != inner { - t.Errorf("handler = %T, want the unwrapped inner handler", got) + handler := sm.mcpHandlerFor(&AgentExecution{ID: "exec-1"}) + if _, err := handler.Dispatch(context.Background(), mcpRequest(t, nil)); err != nil { + t.Fatalf("Dispatch: %v", err) + } + if inner.calls != 1 { + t.Errorf("dispatcher calls = %d, want 1", inner.calls) } } @@ -185,7 +189,8 @@ func TestSetMCPIdentityScoperReachesStreamManager(t *testing.T) { m.SetMCPIdentityScoper(func(ctx context.Context, _ string) (context.Context, error) { return ctx, nil }) - if m.streamManager.mcpIdentityScoper == nil { + _, scoper, _ := m.streamManager.mcpDispatchState() + if scoper == nil { t.Error("SetMCPIdentityScoper did not reach the stream manager") } } diff --git a/apps/backend/internal/agent/runtime/lifecycle/streams.go b/apps/backend/internal/agent/runtime/lifecycle/streams.go index 0a470a5776..0ec5cdc922 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/streams.go +++ b/apps/backend/internal/agent/runtime/lifecycle/streams.go @@ -34,6 +34,7 @@ type StreamCallbacks struct { type StreamManager struct { logger *logger.Logger callbacks StreamCallbacks + mcpMu sync.RWMutex mcpHandler agentctl.MCPHandler // mcpIdentityScoper scopes in-session MCP dispatches to the owner of the // stream's task. Nil leaves dispatch unscoped (single-user instances and diff --git a/apps/backend/internal/agentctl/server/api/agent.go b/apps/backend/internal/agentctl/server/api/agent.go index 4c13d24f79..e5912ee554 100644 --- a/apps/backend/internal/agentctl/server/api/agent.go +++ b/apps/backend/internal/agentctl/server/api/agent.go @@ -137,6 +137,7 @@ func (s *Server) handleAgentStreamWS(c *gin.Context) { } s.logger.Info("agent stream WebSocket connected") + streamID := uuid.NewString() ctx, cancel := context.WithCancel(c.Request.Context()) defer cancel() @@ -162,8 +163,11 @@ func (s *Server) handleAgentStreamWS(c *gin.Context) { wg.Add(1) go s.runAgentStreamReader(ctx, conn, writeMessage, cancel, &wg) wg.Add(1) - go s.runAgentStreamWriter(ctx, conn, updatesCh, mcpRequestCh, writeMessage, &wg) + go s.runAgentStreamWriter(ctx, conn, streamID, updatesCh, mcpRequestCh, writeMessage, &wg) wg.Wait() + if s.mcpBackendClient != nil { + s.mcpBackendClient.FailStreamRequests(streamID, errors.New("agent stream disconnected")) + } } // runAgentStreamReader reads MCP responses and agent operation requests from the backend connection. @@ -209,7 +213,7 @@ func (s *Server) runAgentStreamReader(ctx context.Context, conn *websocket.Conn, } // runAgentStreamWriter sends agent events and MCP requests to the backend connection. -func (s *Server) runAgentStreamWriter(ctx context.Context, conn *websocket.Conn, updatesCh <-chan adapter.AgentEvent, mcpRequestCh <-chan *ws.Message, writeMessage func([]byte) error, wg *sync.WaitGroup) { +func (s *Server) runAgentStreamWriter(ctx context.Context, conn *websocket.Conn, streamID string, updatesCh <-chan adapter.AgentEvent, mcpRequestCh <-chan *ws.Message, writeMessage func([]byte) error, wg *sync.WaitGroup) { defer wg.Done() defer func() { if err := conn.Close(); err != nil { @@ -244,9 +248,18 @@ func (s *Server) runAgentStreamWriter(ctx context.Context, conn *websocket.Conn, continue } if err := writeMessage(data); err != nil { - s.logger.Debug("failed to write MCP request", zap.Error(err)) + s.logger.Warn("failed to write MCP request", + zap.String("request_id", mcpReq.ID), + zap.String("action", mcpReq.Action), + zap.Error(err)) + if s.mcpBackendClient != nil { + s.mcpBackendClient.FailRequest(mcpReq.ID, fmt.Errorf("failed to write MCP request to agent stream: %w", err)) + } return } + if s.mcpBackendClient != nil { + s.mcpBackendClient.BindRequestToStream(mcpReq.ID, streamID) + } } } } diff --git a/apps/backend/internal/agentctl/server/api/agent_mcp_stream_test.go b/apps/backend/internal/agentctl/server/api/agent_mcp_stream_test.go new file mode 100644 index 0000000000..b6b6ac879d --- /dev/null +++ b/apps/backend/internal/agentctl/server/api/agent_mcp_stream_test.go @@ -0,0 +1,113 @@ +package api + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/gorilla/websocket" + "github.com/kandev/kandev/internal/agentctl/server/config" + "github.com/kandev/kandev/internal/agentctl/server/process" + mcpserver "github.com/kandev/kandev/internal/mcp/server" + ws "github.com/kandev/kandev/pkg/websocket" +) + +// @covers AC-AGENTS-MCP-BRIDGE-RELIABILITY-001.5 +func TestAgentStreamWriterReleasesMCPRequestAfterWriteFailure(t *testing.T) { + connectionCh := make(chan *websocket.Conn, 1) + releaseServer := make(chan struct{}) + upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} + httpServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + return + } + connectionCh <- conn + <-releaseServer + })) + t.Cleanup(func() { + close(releaseServer) + httpServer.Close() + }) + + wsURL := "ws" + strings.TrimPrefix(httpServer.URL, "http") + clientConn, _, err := websocket.DefaultDialer.Dial(wsURL, nil) + if err != nil { + t.Fatalf("dial WebSocket: %v", err) + } + t.Cleanup(func() { _ = clientConn.Close() }) + serverConn := <-connectionCh + + log := newTestLogger() + backend := mcpserver.NewChannelBackendClient(log) + t.Cleanup(backend.Close) + server := &Server{mcpBackendClient: backend, logger: log} + var wg sync.WaitGroup + wg.Add(1) + go server.runAgentStreamWriter( + context.Background(), serverConn, "stream-write-failure", nil, + backend.GetRequestChannel(), func([]byte) error { return errors.New("socket write failed") }, &wg, + ) + + errCh := make(chan error, 1) + go func() { + errCh <- backend.RequestPayload(context.Background(), "mcp.tools.call", nil, nil) + }() + + select { + case err := <-errCh: + if err == nil || !strings.Contains(err.Error(), "failed to write MCP request") { + t.Fatalf("request error = %v, want write failure", err) + } + case <-time.After(500 * time.Millisecond): + t.Fatal("MCP request remained blocked after stream write failure") + } + wg.Wait() +} + +// @covers AC-AGENTS-MCP-BRIDGE-RELIABILITY-001.5 +func TestAgentStreamWSDisconnectReleasesDeliveredMCPRequest(t *testing.T) { + log := newTestLogger() + cfg := &config.InstanceConfig{Port: 0, WorkDir: t.TempDir()} + backend := mcpserver.NewChannelBackendClient(log) + t.Cleanup(backend.Close) + server := httptest.NewServer(NewServer(cfg, process.NewManager(cfg, log), nil, backend, log).router) + t.Cleanup(server.Close) + conn := dialTestWS(t, server) + + errCh := make(chan error, 1) + go func() { + errCh <- backend.RequestPayload(context.Background(), "mcp.tools.call", nil, nil) + }() + + _ = conn.SetReadDeadline(time.Now().Add(time.Second)) + _, data, err := conn.ReadMessage() + if err != nil { + t.Fatalf("read MCP request: %v", err) + } + var request ws.Message + if err := json.Unmarshal(data, &request); err != nil { + t.Fatalf("decode MCP request: %v", err) + } + if request.Type != ws.MessageTypeRequest { + t.Fatalf("message type = %q, want request", request.Type) + } + if err := conn.Close(); err != nil { + t.Fatalf("close stream: %v", err) + } + + select { + case err := <-errCh: + if err == nil || !strings.Contains(err.Error(), "agent stream disconnected") { + t.Fatalf("request error = %v, want agent stream disconnected", err) + } + case <-time.After(500 * time.Millisecond): + t.Fatal("delivered MCP request remained blocked after stream disconnect") + } +} diff --git a/apps/backend/internal/mcp/server/backend_client.go b/apps/backend/internal/mcp/server/backend_client.go index 3555cc0b81..d3ce808b3a 100644 --- a/apps/backend/internal/mcp/server/backend_client.go +++ b/apps/backend/internal/mcp/server/backend_client.go @@ -27,13 +27,26 @@ type MCPResponse struct { Payload json.RawMessage `json:"payload"` } +type backendResponse struct { + message *ws.Message + err error + sessionID string +} + +type pendingRequest struct { + result chan backendResponse + streamID string + sessionID string +} + // ChannelBackendClient implements BackendClient using channels. // It sends MCP requests through a channel that will be read by the agent stream handler, // and receives responses through a callback mechanism. type ChannelBackendClient struct { requestCh chan *ws.Message - pending map[string]chan *ws.Message + pending map[string]*pendingRequest pendingMu sync.Mutex + sessionID string done chan struct{} closeOnce sync.Once closeMu sync.Mutex @@ -50,8 +63,8 @@ func NewChannelBackendClient(log *logger.Logger) *ChannelBackendClient { } clientLogger = clientLogger.WithFields(zap.String("component", "mcp-backend-client")) return &ChannelBackendClient{ - requestCh: make(chan *ws.Message, 100), - pending: make(map[string]chan *ws.Message), + requestCh: make(chan *ws.Message), + pending: make(map[string]*pendingRequest), done: make(chan struct{}), logger: clientLogger, } @@ -63,16 +76,17 @@ func (c *ChannelBackendClient) GetRequestChannel() <-chan *ws.Message { return c.requestCh } +// SetSessionID sets the server-owned session correlation used in bridge logs. +func (c *ChannelBackendClient) SetSessionID(sessionID string) { + c.pendingMu.Lock() + defer c.pendingMu.Unlock() + c.sessionID = sessionID +} + // HandleResponse handles an incoming MCP response from the backend. // This should be called by the agent stream handler when it receives a response. func (c *ChannelBackendClient) HandleResponse(msg *ws.Message) { - c.pendingMu.Lock() - ch, ok := c.pending[msg.ID] - delete(c.pending, msg.ID) - c.pendingMu.Unlock() - - if ok { - ch <- msg + if c.completeRequest(msg.ID, backendResponse{message: msg}) { return } c.logger.Debug("dropping MCP response with no pending request", @@ -81,6 +95,53 @@ func (c *ChannelBackendClient) HandleResponse(msg *ws.Message) { zap.String("action", msg.Action)) } +// BindRequestToStream records which backend stream delivered a request. +func (c *ChannelBackendClient) BindRequestToStream(requestID, streamID string) { + c.pendingMu.Lock() + defer c.pendingMu.Unlock() + if pending, ok := c.pending[requestID]; ok { + pending.streamID = streamID + } +} + +// FailRequest releases one pending request after a transport failure. +func (c *ChannelBackendClient) FailRequest(requestID string, err error) { + c.completeRequest(requestID, backendResponse{err: err}) +} + +// FailStreamRequests releases requests delivered by one disconnected stream. +func (c *ChannelBackendClient) FailStreamRequests(streamID string, err error) { + c.pendingMu.Lock() + failed := make([]*pendingRequest, 0) + for id, pending := range c.pending { + if pending.streamID == streamID { + delete(c.pending, id) + failed = append(failed, pending) + } + } + c.pendingMu.Unlock() + + for _, pending := range failed { + pending.result <- backendResponse{err: err, sessionID: pending.sessionID} + } +} + +func (c *ChannelBackendClient) completeRequest(requestID string, response backendResponse) bool { + c.pendingMu.Lock() + pending, ok := c.pending[requestID] + if ok { + delete(c.pending, requestID) + } + c.pendingMu.Unlock() + if ok { + if response.sessionID == "" { + response.sessionID = pending.sessionID + } + pending.result <- response + } + return ok +} + // RequestPayload sends a request to the backend and unmarshals the response. // The request will be cancelled if the context is cancelled or if Reset() is called. func (c *ChannelBackendClient) RequestPayload(ctx context.Context, action string, payload, result interface{}) error { @@ -103,14 +164,16 @@ func (c *ChannelBackendClient) RequestPayload(ctx context.Context, action string } // Create response channel - respChan := make(chan *ws.Message, 1) + respChan := make(chan backendResponse, 1) c.pendingMu.Lock() - c.pending[id] = respChan + sessionID := c.sessionID + c.pending[id] = &pendingRequest{result: respChan, sessionID: sessionID} c.pendingMu.Unlock() c.logger.Debug("sending MCP request through agent stream", zap.String("request_id", id), zap.String("action", action), + zap.String("session_id", sessionID), zap.Any("payload", backendPayloadForLog(action, payload))) // Ensure cleanup on exit @@ -130,6 +193,7 @@ func (c *ChannelBackendClient) RequestPayload(ctx context.Context, action string c.logger.Debug("MCP request cancelled before send", zap.String("request_id", id), zap.String("action", action), + zap.String("session_id", sessionID), zap.Duration("duration", time.Since(start)), zap.Error(ctx.Err())) return ctx.Err() @@ -137,6 +201,7 @@ func (c *ChannelBackendClient) RequestPayload(ctx context.Context, action string c.logger.Warn("timed out sending MCP request to agent stream", zap.String("request_id", id), zap.String("action", action), + zap.String("session_id", sessionID), zap.Duration("duration", time.Since(start))) return fmt.Errorf("timeout sending request to agent stream") } @@ -145,15 +210,17 @@ func (c *ChannelBackendClient) RequestPayload(ctx context.Context, action string // Wait for response select { - case resp, ok := <-respChan: - if !ok { - // Channel was closed by Reset() - session was cancelled/reset - c.logger.Warn("MCP request cancelled by session reset", + case response := <-respChan: + if response.err != nil { + c.logger.Warn("MCP request failed after publication", zap.String("request_id", id), zap.String("action", action), - zap.Duration("duration", time.Since(start))) - return fmt.Errorf("MCP request cancelled: session reset") + zap.String("session_id", response.sessionID), + zap.Duration("duration", time.Since(start)), + zap.Error(response.err)) + return response.err } + resp := response.message c.logger.Debug("received MCP response from backend", zap.String("request_id", id), zap.String("action", action), @@ -219,12 +286,18 @@ func backendPayloadForLog(action string, payload interface{}) interface{} { // stale requests from a previous session from interfering. func (c *ChannelBackendClient) Reset() { c.pendingMu.Lock() - defer c.pendingMu.Unlock() - - // Close all pending response channels to unblock waiting goroutines - for id, ch := range c.pending { - close(ch) + pending := make([]*pendingRequest, 0, len(c.pending)) + for id, request := range c.pending { delete(c.pending, id) + pending = append(pending, request) + } + c.pendingMu.Unlock() + + for _, request := range pending { + request.result <- backendResponse{ + err: fmt.Errorf("MCP request cancelled: session reset"), + sessionID: request.sessionID, + } } } diff --git a/apps/backend/internal/mcp/server/backend_client_test.go b/apps/backend/internal/mcp/server/backend_client_test.go index f2c27a1f33..627462cc47 100644 --- a/apps/backend/internal/mcp/server/backend_client_test.go +++ b/apps/backend/internal/mcp/server/backend_client_test.go @@ -2,6 +2,7 @@ package mcp import ( "context" + "errors" "testing" "time" @@ -52,6 +53,79 @@ func TestChannelBackendClientCloseReleasesPublishedRequest(t *testing.T) { } } +// @covers AC-AGENTS-MCP-BRIDGE-RELIABILITY-001.2 +func TestChannelBackendClientDoesNotQueueRequestWithoutStreamConsumer(t *testing.T) { + client := NewChannelBackendClient(nil) + t.Cleanup(client.Close) + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + t.Cleanup(cancel) + + err := client.RequestPayload(ctx, "test.action", nil, nil) + require.ErrorIs(t, err, context.DeadlineExceeded) + + select { + case msg := <-client.GetRequestChannel(): + t.Fatalf("request %q was queued without a stream consumer", msg.ID) + default: + } +} + +// @covers AC-AGENTS-MCP-BRIDGE-RELIABILITY-001.5 +func TestChannelBackendClientDisconnectOnlyFailsRequestsOwnedByThatStream(t *testing.T) { + client := NewChannelBackendClient(nil) + t.Cleanup(client.Close) + + oldErrCh := make(chan error, 1) + go func() { + oldErrCh <- client.RequestPayload(context.Background(), "old.action", nil, nil) + }() + oldRequest := <-client.GetRequestChannel() + client.BindRequestToStream(oldRequest.ID, "old-stream") + + newErrCh := make(chan error, 1) + go func() { + newErrCh <- client.RequestPayload(context.Background(), "new.action", nil, nil) + }() + newRequest := <-client.GetRequestChannel() + client.BindRequestToStream(newRequest.ID, "new-stream") + + client.FailStreamRequests("old-stream", errors.New("agent stream disconnected")) + require.EqualError(t, <-oldErrCh, "agent stream disconnected") + select { + case err := <-newErrCh: + t.Fatalf("replacement stream request completed early: %v", err) + default: + } + + response, err := ws.NewResponse(newRequest.ID, newRequest.Action, nil) + require.NoError(t, err) + client.HandleResponse(response) + require.NoError(t, <-newErrCh) +} + +// @covers AC-AGENTS-MCP-BRIDGE-RELIABILITY-001.7 +func TestChannelBackendClientTerminalFailureLogIncludesConfiguredSession(t *testing.T) { + core, observed := observer.New(zap.WarnLevel) + log, err := logger.NewFromZap(zap.New(core)) + require.NoError(t, err) + client := NewChannelBackendClient(log) + t.Cleanup(client.Close) + _ = New(client, "session-log", "task-log", 0, log, "", false, ModeTask) + + errCh := make(chan error, 1) + go func() { + errCh <- client.RequestPayload(context.Background(), "test.action", nil, nil) + }() + request := <-client.GetRequestChannel() + client.BindRequestToStream(request.ID, "failed-stream") + client.FailStreamRequests("failed-stream", errors.New("agent stream disconnected")) + require.EqualError(t, <-errCh, "agent stream disconnected") + + entries := observed.FilterMessage("MCP request failed after publication").All() + require.Len(t, entries, 1) + require.Equal(t, "session-log", entries[0].ContextMap()["session_id"]) +} + func TestChannelBackendClientRedactsPluginInvocationPayload(t *testing.T) { core, observed := observer.New(zap.DebugLevel) log, err := logger.NewFromZap(zap.New(core)) diff --git a/apps/backend/internal/mcp/server/server.go b/apps/backend/internal/mcp/server/server.go index 49949a7de8..777a5b2678 100644 --- a/apps/backend/internal/mcp/server/server.go +++ b/apps/backend/internal/mcp/server/server.go @@ -37,6 +37,10 @@ type BackendClient interface { RequestPayload(ctx context.Context, action string, payload, result interface{}) error } +type backendSessionSetter interface { + SetSessionID(sessionID string) +} + // MCP mode constants control which tools are registered. const ( // ModeTask registers kanban, plan, and interaction tools (default for task-solving agents). @@ -222,6 +226,9 @@ func newServer(backend BackendClient, sessionID, taskID string, log *logger.Logg func newServerWithProfile(backend BackendClient, sessionID, taskID string, log *logger.Logger, mcpLogFile string, profileContext mcpprofile.Context, options ...ServerOption) *Server { profileContext = mcpprofile.New(profileContext.Surface, profileContext.Capabilities, profileContext.Providers) + if setter, ok := backend.(backendSessionSetter); ok { + setter.SetSessionID(sessionID) + } s := &Server{ backend: backend, sessionID: sessionID, diff --git a/docs/plans/mcp-bridge-reliability/plan.md b/docs/plans/mcp-bridge-reliability/plan.md new file mode 100644 index 0000000000..9366578ea8 --- /dev/null +++ b/docs/plans/mcp-bridge-reliability/plan.md @@ -0,0 +1,96 @@ +--- +created: 2026-09-04 +status: done +requirements: + - REQ-AGENTS-MCP-BRIDGE-RELIABILITY-001 +system_design: + - ../../specs/agents/system-design/mcp-bridge-reliability.md +legacy_specs: [] +--- + +# Implementation Plan: MCP Bridge Reliability + +## Overview + +Make backend-backed MCP requests use the current dispatcher and explicit +transport ownership. Add regression tests before the implementation changes. + +## Scope + +### In scope + +- Remove the recovered-stream dispatcher capture race. +- Return correlated errors for missing handlers and empty request responses. +- Make the existing send timeout detect an absent stream consumer. +- Release pending requests after write errors and owning-stream disconnects. +- Add default-level diagnostic events without tool arguments. + +### Out of scope + +- Business-operation deadlines for tools and agent launch. +- User-interface stall indicators. +- Third-party MCP traffic. +- Changes to MCP tool schemas or results. + +## Technical approach + +### Lifecycle dispatcher binding + +Update `StreamManager` to return a stable execution-bound handler proxy. The +proxy reads the current dispatcher and scope functions for each request under +a read-write mutex. + +Update manager setters to use the synchronized `StreamManager` boundary. Keep +task, session, execution, user, and automation scope derived from the current +server-owned execution. + +### Backend request handling + +Update `readUpdatesStream` to write an error when a direct caller supplies no +handler. Update `dispatchMCPRequest` to reject an empty response for a request. + +Record request receipt at `Info`. Record terminal bridge errors at `Warn` or +`Error` without payload values. + +### Agentctl request ownership + +Make `ChannelBackendClient.requestCh` unbuffered. Replace bare response +channels with pending result records that can carry a response or transport +error. + +Assign one identifier to each backend stream. Bind a request after its writer +sends it. Complete the request after a write error or the owning stream closes. +Do not cancel requests owned by an overlapping replacement stream. + +## Tests + +- `AC-AGENTS-MCP-BRIDGE-RELIABILITY-001.1` maps to lifecycle handler-binding + tests in `internal/agent/runtime/lifecycle`. +- `AC-AGENTS-MCP-BRIDGE-RELIABILITY-001.2` maps to channel-send tests in + `internal/mcp/server`. +- `AC-AGENTS-MCP-BRIDGE-RELIABILITY-001.3` and `.4` map to stream-dispatch + tests in `internal/agent/runtime/agentctl`. +- `AC-AGENTS-MCP-BRIDGE-RELIABILITY-001.5` maps to stream ownership tests in + `internal/agentctl/server/api` and channel-client tests. +- `AC-AGENTS-MCP-BRIDGE-RELIABILITY-001.6` and `.7` map to existing behavior + tests plus new log assertions. + +## Work orders + +- [x] [Task 01: Stabilize the internal MCP bridge](task-01-stabilize-internal-mcp-bridge.md) + +## Verification results + +- `cd apps/backend && go test ./internal/mcp/server ./internal/agent/runtime/agentctl ./internal/agentctl/server/api ./internal/agent/runtime/lifecycle -count=1` + passed 3,560 tests in four packages. +- `cd apps/backend && go test -race ./internal/mcp/server ./internal/agent/runtime/agentctl ./internal/agentctl/server/api ./internal/agent/runtime/lifecycle -run 'MCP|ChannelBackendClient|StreamUpdates' -count=1` + passed 109 tests in four packages. + +## Risks + +- Stream replacement can overlap with old-stream cleanup. Request ownership + must use the stream identifier and must not use a global reset. +- Dispatcher and scope setters can run after recovery starts. All related + reads and writes must use the same synchronization boundary. +- An unbuffered request channel changes scheduling. Tests must cover close, + reset, context cancellation, and concurrent requests. diff --git a/docs/plans/mcp-bridge-reliability/task-01-stabilize-internal-mcp-bridge.md b/docs/plans/mcp-bridge-reliability/task-01-stabilize-internal-mcp-bridge.md new file mode 100644 index 0000000000..812eeb15b2 --- /dev/null +++ b/docs/plans/mcp-bridge-reliability/task-01-stabilize-internal-mcp-bridge.md @@ -0,0 +1,104 @@ +--- +id: "01-stabilize-internal-mcp-bridge" +title: "Stabilize the internal MCP bridge" +status: done +wave: 1 +depends_on: [] +plan: "plan.md" +requirements: + - REQ-AGENTS-MCP-BRIDGE-RELIABILITY-001 +acceptance_criteria: + - AC-AGENTS-MCP-BRIDGE-RELIABILITY-001.1 + - AC-AGENTS-MCP-BRIDGE-RELIABILITY-001.2 + - AC-AGENTS-MCP-BRIDGE-RELIABILITY-001.3 + - AC-AGENTS-MCP-BRIDGE-RELIABILITY-001.4 + - AC-AGENTS-MCP-BRIDGE-RELIABILITY-001.5 + - AC-AGENTS-MCP-BRIDGE-RELIABILITY-001.6 + - AC-AGENTS-MCP-BRIDGE-RELIABILITY-001.7 +system_design: + - ../../specs/agents/system-design/mcp-bridge-reliability.md +--- + +# Task 01: Stabilize the Internal MCP Bridge + +## Summary + +Add failing regression tests for the startup race and lost-response paths. +Then make every bridge request receive a result or a transport error. + +## In scope + +- Add synchronized, request-time dispatcher resolution to `StreamManager`. +- Preserve server-derived execution and principal scope. +- Return correlated errors for an unavailable handler and an empty response. +- Make the request channel unbuffered. +- Bind pending requests to the backend stream that sends them. +- Release pending requests after write errors and stream disconnects. +- Add safe default-level diagnostic events. + +## Out of scope + +- Add a common deadline to all backend handlers. +- Change user-question or parent-question wait behavior. +- Add user-interface behavior. +- Change public MCP protocol shapes. + +## Acceptance + +- A recovered stream created before dispatcher setup uses the dispatcher after + setup without a reconnect. +- Each missing-delivery path returns one correlated error and records a safe + diagnostic event. +- Existing successful, cancellation, reset, close, and long-wait paths pass + without behavior changes. + +## Verification + +```bash +cd apps/backend && go test ./internal/mcp/server ./internal/agent/runtime/agentctl ./internal/agentctl/server/api ./internal/agent/runtime/lifecycle -count=1 +cd apps/backend && go test -race ./internal/mcp/server ./internal/agent/runtime/agentctl ./internal/agentctl/server/api ./internal/agent/runtime/lifecycle -run 'MCP|ChannelBackendClient|StreamUpdates' -count=1 +``` + +## Files likely touched + +- `apps/backend/internal/mcp/server/backend_client.go` +- `apps/backend/internal/mcp/server/backend_client_test.go` +- `apps/backend/internal/agentctl/server/api/agent.go` +- `apps/backend/internal/agentctl/server/api/agent_mcp_stream_test.go` +- `apps/backend/internal/agent/runtime/agentctl/agent.go` +- `apps/backend/internal/agent/runtime/agentctl/agent_mcp_stream_test.go` +- `apps/backend/internal/agent/runtime/lifecycle/manager.go` +- `apps/backend/internal/agent/runtime/lifecycle/mcp_identity.go` +- `apps/backend/internal/agent/runtime/lifecycle/mcp_handler_binding_test.go` + +## Dependencies + +None. + +## Risks + +- Old and replacement streams can overlap during reconnect. +- A completion path can race with response delivery, reset, context + cancellation, or client close. + +## Parallelism + +`sequential` + +## Inputs + +- [MCP bridge reliability requirements](../../specs/agents/requirements/mcp-bridge-reliability.md) +- [MCP bridge reliability system design](../../specs/agents/system-design/mcp-bridge-reliability.md) +- [GitHub issue 3364](https://github.com/kdlbs/kandev/issues/3364) + +## Results + +Implemented synchronized request-time dispatcher resolution, correlated error +responses, unbuffered request publication, and per-stream pending-request +ownership. Added regression tests for startup wiring, absent consumers, +unavailable or empty dispatch, write failure, disconnect, and replacement +stream isolation. Terminal agentctl failures now preserve the configured +session correlation without recording request payloads. + +The targeted suite passed 3,560 tests in four packages. The race-focused suite +passed 109 tests in the same packages. diff --git a/docs/specs/agents/README.md b/docs/specs/agents/README.md index 9e6377f022..42785e8481 100644 --- a/docs/specs/agents/README.md +++ b/docs/specs/agents/README.md @@ -57,6 +57,7 @@ surface shared by task and Office consumers. - [Agent Roles — Security, QA, and DevOps](requirements/roles.md) - [Managed npm runtime recovery](requirements/managed-npm-runtime-recovery.md) - [Managed Agent Runtime Versions and Updates](requirements/runtime-updates.md) +- [MCP Bridge Reliability](requirements/mcp-bridge-reliability.md) - [MCP Protocol Compatibility](requirements/mcp-protocol-compatibility.md) - [Simplify the agent settings profile layout](requirements/settings-profile-layout.md) - [Spawn Session Effective Agent Profile](requirements/spawn-session-effective-profile.md) @@ -78,6 +79,7 @@ surface shared by task and Office consumers. - [Managed Agent Runtime Versions and Updates System Design Part 1](system-design/runtime-updates-01.md) - [Managed Agent Runtime Versions and Updates System Design Part 2](system-design/runtime-updates-02.md) - [Managed npm runtime recovery](system-design/managed-npm-runtime-recovery.md) +- [MCP Bridge Reliability](system-design/mcp-bridge-reliability.md) - [MCP Protocol Compatibility](system-design/mcp-protocol-compatibility.md) - [Subagent context persistence System Design Part 1](system-design/subagent-context-persistence-01.md) - [Subagent context persistence System Design Part 2](system-design/subagent-context-persistence-02.md) diff --git a/docs/specs/agents/requirements/mcp-bridge-reliability.md b/docs/specs/agents/requirements/mcp-bridge-reliability.md new file mode 100644 index 0000000000..ffec64e639 --- /dev/null +++ b/docs/specs/agents/requirements/mcp-bridge-reliability.md @@ -0,0 +1,63 @@ +--- +status: active +system: agents +created: 2026-09-04 +owners: + - Kandev +--- + +# MCP Bridge Reliability Requirements + +## Overview + +Kandev agents use a local MCP server in agentctl. Backend-backed tools cross an +internal WebSocket bridge before the backend handles them. + +The agent system owns this contract because the bridge is part of every +Kandev-managed agent runtime. Task and integration systems own the behavior of +their tools after the bridge delivers a request. + +## Terminology + +- **MCP bridge:** The request and response path between the agentctl MCP server + and the Kandev backend. +- **Backend stream:** The WebSocket connection that carries MCP requests from + agentctl to the backend and returns correlated responses. + +## Requirements + +### REQ-AGENTS-MCP-BRIDGE-RELIABILITY-001: Reliable backend-backed MCP delivery + +**Intent:** An agent must receive a result or a descriptive bridge error when +it calls a Kandev tool. A lost internal request must not consume the outer MCP +client timeout without diagnostic evidence. + +#### Acceptance criteria + +- **AC-AGENTS-MCP-BRIDGE-RELIABILITY-001.1:** When a new or recovered agent + stream receives an MCP request, Kandev shall use the current configured + dispatcher. Startup order shall not leave the stream bound to an earlier + empty dispatcher. +- **AC-AGENTS-MCP-BRIDGE-RELIABILITY-001.2:** When no backend stream consumes + an MCP request, agentctl shall return a descriptive error within five + seconds and shall record the action at the default log level. +- **AC-AGENTS-MCP-BRIDGE-RELIABILITY-001.3:** When the backend stream has no + MCP dispatcher, Kandev shall return a correlated error response and shall + record the request identity at the default log level. +- **AC-AGENTS-MCP-BRIDGE-RELIABILITY-001.4:** When a dispatcher returns no + response for a request, Kandev shall return a correlated error response. It + shall not silently discard the request. +- **AC-AGENTS-MCP-BRIDGE-RELIABILITY-001.5:** When the backend stream closes + before a response arrives, agentctl shall fail each request owned by that + stream with a descriptive disconnect error. +- **AC-AGENTS-MCP-BRIDGE-RELIABILITY-001.6:** Successful tool results and + intentional user or parent question waits shall keep their current behavior. +- **AC-AGENTS-MCP-BRIDGE-RELIABILITY-001.7:** Default logs shall identify the + action, request, and session for accepted requests and terminal bridge + errors. These logs shall not contain tool arguments. + +## Out of scope + +- Deadlines for task, integration, plugin, or agent-launch business operations. +- A new user-interface alert for a stalled MCP request. +- Traffic between an agent and a third-party MCP server. diff --git a/docs/specs/agents/system-design/mcp-bridge-reliability.md b/docs/specs/agents/system-design/mcp-bridge-reliability.md new file mode 100644 index 0000000000..a6e8c241f6 --- /dev/null +++ b/docs/specs/agents/system-design/mcp-bridge-reliability.md @@ -0,0 +1,158 @@ +--- +status: current +system: agents +requirements: + - REQ-AGENTS-MCP-BRIDGE-RELIABILITY-001 +created: 2026-09-04 +owners: + - Kandev +--- + +# MCP Bridge Reliability System Design + +## Purpose and boundaries + +This design makes the internal MCP bridge fail with a correlated error when +delivery is not possible. It also removes a startup race for recovered agent +streams. + +The bridge does not set a common deadline for backend operations. Some tools +wait for a person, and agent launch can use a 15-minute budget. Each business +operation keeps its own context and deadline. + +## Requirement mapping + +| Requirement | Design sections | +| --- | --- | +| `REQ-AGENTS-MCP-BRIDGE-RELIABILITY-001` | [Current dispatcher binding](#current-dispatcher-binding), [Request ownership](#request-ownership), [Failure behavior](#failure-behavior), [Observability](#observability) | + +## Confirmed fault + +`provideAgentRuntime` calls `lifecycle.Manager.Start` before route registration +calls `SetMCPHandler`. Startup recovery starts reconnect work during this gap. + +`StreamManager.connectUpdatesStream` passes the current `mcpHandler` value to +`Client.StreamUpdates`. The read loop keeps that interface value for the life +of the stream. A stream that starts during the gap therefore keeps a `nil` +handler after startup finishes. + +The read loop silently ignores a request when its handler is `nil`. Agentctl +then waits on `ChannelBackendClient.pending` until the calling MCP client ends +the request. + +Two other bridge states can lose a response. The buffered request channel can +accept a request without a stream consumer. A stream can also close after its +writer removes a request from the channel. + +## Components and responsibilities + +- `internal/mcp/server.ChannelBackendClient` owns agentctl request state and + correlation. +- `internal/agentctl/server/api.Server` owns the agentctl side of each backend + stream. +- `internal/agent/runtime/agentctl.Client` reads backend requests and writes + their responses. +- `internal/agent/runtime/lifecycle.StreamManager` supplies the dispatcher and + server-derived execution scope for each backend stream. + +## Current dispatcher binding + +`StreamManager` stores the dispatcher and its scope functions behind one +read-write mutex. Manager setter methods update that state through a +`StreamManager` method. + +`mcpHandlerFor` returns a stable execution-bound proxy. The proxy reads the +current dispatcher and scope functions for each request. Thus a stream that +starts before route registration can use the dispatcher after registration. + +If no dispatcher exists when a request arrives, the proxy returns a typed +unavailable error. `dispatchMCPRequest` converts that error to a correlated +WebSocket error response. + +`readUpdatesStream` also handles a direct `nil` handler defensively. It writes +a correlated error response instead of skipping the request. This path covers +callers that do not use `StreamManager`. + +If a dispatcher returns both a `nil` response and a `nil` error for a request, +`dispatchMCPRequest` writes a correlated internal error. WebSocket +notifications remain separate from request messages and do not require a +response. + +## Request ownership + +`ChannelBackendClient.requestCh` becomes unbuffered. A send succeeds only when +an active stream writer accepts the request. The existing five-second send +limit therefore detects the absence of a consumer. + +Each pending request stores a result channel and an optional backend-stream +identifier. The stream writer binds the request to its identifier after a +successful WebSocket write. The stream handler waits for the writer goroutine +before it starts disconnect cleanup. Therefore, binding happens before cleanup +can inspect the pending requests. + +If the write fails, the writer completes that request with a delivery error. +When the stream ends, the API server completes only requests bound to that +stream. A replacement stream can overlap with an old stream without losing its +requests. + +Session reset still completes all pending requests with the existing reset +error. Client close still rejects new requests and releases all current waits. + +## Control flow + +1. The local MCP handler registers a pending request before it sends the + request to `requestCh`. +2. An active stream writer accepts the unbuffered request. +3. The writer sends the request and binds it to the stream identifier. +4. The backend read loop records receipt and resolves the current dispatcher. +5. The dispatcher returns one correlated response or error. +6. Agentctl completes the matching pending request. + +If steps 2 through 5 cannot finish, the owning component completes the pending +request with the applicable transport error. + +## Failure behavior + +| Condition | Result | +| --- | --- | +| No active stream writer | The existing send timer returns an error within five seconds. | +| No configured dispatcher | The backend returns a correlated unavailable error. | +| Dispatcher returns no response | The backend returns a correlated internal error. | +| WebSocket write fails | Agentctl completes that request with a delivery error. | +| Owning backend stream closes | Agentctl completes requests bound to that stream with a disconnect error. | +| Request context ends | `RequestPayload` returns the context error. | +| Session resets | Pending requests return the existing session-reset error. | + +This design does not add a common response deadline. Such a deadline can stop +valid clarification waits and long launch operations. Transport ownership and +caller contexts provide the required termination signals. + +## Observability + +The backend records one request-received event at `Info`. The event includes +the action, request ID, session ID, and pending ID. + +Unavailable dispatch, empty response, write failure, send timeout, and stream +disconnect events use `Warn` or `Error`. Accepted requests and terminal bridge +errors include the action, request ID, and server-configured session ID. The +bridge does not record request payloads or tool arguments at these levels. + +Existing trace spans continue to measure dispatcher duration and response +outcome. + +## Testing strategy + +Unit tests cover these cases: + +- A handler proxy created before `SetMCPHandler` uses the later dispatcher and + retains execution scope. +- A direct stream with a `nil` handler returns a correlated error frame. +- A dispatcher with no response returns a correlated error frame. +- A request without a stream consumer cannot enter the request channel. +- A stream write error releases the matching request. +- A stream disconnect releases only requests owned by that stream. +- Successful responses, context cancellation, reset, close, and clarification + waits keep their current behavior. + +Race-enabled tests cover concurrent dispatcher updates, request completion, +stream replacement, reset, and close.