From e2ba35b3bd60f59481b62adeacc5b6aaffc5dc04 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Salavisa?= Date: Fri, 13 Feb 2026 09:05:40 +0000 Subject: [PATCH] feat: migrate agent operations from HTTP to WebSocket (#218) --- .../agent/credentials/manager_test.go | 3 +- .../agent/handlers/shell_handlers_test.go | 16 +- .../internal/agent/lifecycle/manager_test.go | 34 +- .../agent/lifecycle/profile_resolver_test.go | 56 +- .../agent/lifecycle/runtime_registry_test.go | 13 +- .../internal/agent/lifecycle/session.go | 28 +- .../internal/agent/lifecycle/session_test.go | 503 +++++++++++ .../internal/agent/registry/registry_test.go | 3 +- .../backend/internal/agentctl/client/agent.go | 323 +++---- .../internal/agentctl/client/agent_test.go | 834 ++++++++++++++++++ .../internal/agentctl/client/client.go | 104 ++- .../adapter/transport/codex/errors_test.go | 28 +- .../internal/agentctl/server/api/agent.go | 509 ++++++----- .../agentctl/server/api/agent_test.go | 492 +++++++++++ .../internal/agentctl/server/api/server.go | 10 +- .../server/process/interactive_runner_test.go | 19 +- .../internal/orchestrator/queue/queue_test.go | 3 +- .../orchestrator/watcher/watcher_test.go | 3 +- .../internal/worktree/recreator_test.go | 3 +- apps/backend/pkg/claudecode/types_test.go | 11 +- 20 files changed, 2400 insertions(+), 595 deletions(-) create mode 100644 apps/backend/internal/agent/lifecycle/session_test.go create mode 100644 apps/backend/internal/agentctl/client/agent_test.go create mode 100644 apps/backend/internal/agentctl/server/api/agent_test.go diff --git a/apps/backend/internal/agent/credentials/manager_test.go b/apps/backend/internal/agent/credentials/manager_test.go index 724a337971..5d53a57f3b 100644 --- a/apps/backend/internal/agent/credentials/manager_test.go +++ b/apps/backend/internal/agent/credentials/manager_test.go @@ -25,8 +25,7 @@ func TestNewManager(t *testing.T) { if mgr == nil { t.Fatal("expected non-nil manager") - } - if len(mgr.providers) != 0 { + } else if len(mgr.providers) != 0 { t.Errorf("expected no providers, got %d", len(mgr.providers)) } } diff --git a/apps/backend/internal/agent/handlers/shell_handlers_test.go b/apps/backend/internal/agent/handlers/shell_handlers_test.go index 6b12e57add..60a2686a2e 100644 --- a/apps/backend/internal/agent/handlers/shell_handlers_test.go +++ b/apps/backend/internal/agent/handlers/shell_handlers_test.go @@ -84,12 +84,13 @@ func TestNewShellHandlers(t *testing.T) { if handlers == nil { t.Fatal("expected non-nil handlers") - } - if handlers.lifecycleMgr != nil { - t.Error("expected nil lifecycleMgr when nil passed") - } - if handlers.logger == nil { - t.Error("expected non-nil logger") + } else { + if handlers.lifecycleMgr != nil { + t.Error("expected nil lifecycleMgr when nil passed") + } + if handlers.logger == nil { + t.Error("expected non-nil logger") + } } } @@ -267,8 +268,7 @@ func TestNewShellHandlers_WithManager(t *testing.T) { if handlers == nil { t.Fatal("expected non-nil handlers") - } - if handlers.lifecycleMgr != mgr { + } else if handlers.lifecycleMgr != mgr { t.Error("expected lifecycleMgr to be set to provided manager") } } diff --git a/apps/backend/internal/agent/lifecycle/manager_test.go b/apps/backend/internal/agent/lifecycle/manager_test.go index 979bc78884..bd1c69b0a4 100644 --- a/apps/backend/internal/agent/lifecycle/manager_test.go +++ b/apps/backend/internal/agent/lifecycle/manager_test.go @@ -688,15 +688,16 @@ func TestFinalizeBootMessage_Success(t *testing.T) { lastMsg := bootSvc.getLastUpdatedMessage() if lastMsg == nil { t.Fatal("expected boot message to be updated") - } - if lastMsg.Metadata["status"] != "exited" { - t.Errorf("expected status 'exited', got %v", lastMsg.Metadata["status"]) - } - if lastMsg.Metadata["exit_code"] != 0 { - t.Errorf("expected exit_code 0, got %v", lastMsg.Metadata["exit_code"]) - } - if lastMsg.Metadata["completed_at"] == nil { - t.Error("expected completed_at to be set") + } else { + if lastMsg.Metadata["status"] != "exited" { + t.Errorf("expected status 'exited', got %v", lastMsg.Metadata["status"]) + } + if lastMsg.Metadata["exit_code"] != 0 { + t.Errorf("expected exit_code 0, got %v", lastMsg.Metadata["exit_code"]) + } + if lastMsg.Metadata["completed_at"] == nil { + t.Error("expected completed_at to be set") + } } } @@ -716,13 +717,14 @@ func TestFinalizeBootMessage_Failed(t *testing.T) { lastMsg := bootSvc.getLastUpdatedMessage() if lastMsg == nil { t.Fatal("expected boot message to be updated") - } - if lastMsg.Metadata["status"] != "failed" { - t.Errorf("expected status 'failed', got %v", lastMsg.Metadata["status"]) - } - // Failed status should NOT have exit_code - if _, ok := lastMsg.Metadata["exit_code"]; ok { - t.Error("expected no exit_code for failed status") + } else { + if lastMsg.Metadata["status"] != "failed" { + t.Errorf("expected status 'failed', got %v", lastMsg.Metadata["status"]) + } + // Failed status should NOT have exit_code + if _, ok := lastMsg.Metadata["exit_code"]; ok { + t.Error("expected no exit_code for failed status") + } } } diff --git a/apps/backend/internal/agent/lifecycle/profile_resolver_test.go b/apps/backend/internal/agent/lifecycle/profile_resolver_test.go index d93cd15206..2396ebe541 100644 --- a/apps/backend/internal/agent/lifecycle/profile_resolver_test.go +++ b/apps/backend/internal/agent/lifecycle/profile_resolver_test.go @@ -100,8 +100,7 @@ func TestNewStoreProfileResolver(t *testing.T) { if resolver == nil { t.Fatal("expected non-nil resolver") - } - if resolver.store != mockRepo { + } else if resolver.store != mockRepo { t.Error("expected resolver to use the provided store") } } @@ -136,27 +135,28 @@ func TestStoreProfileResolver_ResolveProfile_Success(t *testing.T) { } if info == nil { t.Fatal("expected non-nil profile info") - } - if info.ProfileID != "profile-123" { - t.Errorf("expected ProfileID 'profile-123', got '%s'", info.ProfileID) - } - if info.ProfileName != "My Profile" { - t.Errorf("expected ProfileName 'My Profile', got '%s'", info.ProfileName) - } - if info.AgentID != "agent-456" { - t.Errorf("expected AgentID 'agent-456', got '%s'", info.AgentID) - } - if info.AgentName != "claude" { - t.Errorf("expected AgentName 'claude', got '%s'", info.AgentName) - } - if info.Model != "claude-3.5-sonnet" { - t.Errorf("expected Model 'claude-3.5-sonnet', got '%s'", info.Model) - } - if info.AutoApprove != true { - t.Error("expected AutoApprove to be true") - } - if info.DangerouslySkipPermissions != false { - t.Error("expected DangerouslySkipPermissions to be false") + } else { + if info.ProfileID != "profile-123" { + t.Errorf("expected ProfileID 'profile-123', got '%s'", info.ProfileID) + } + if info.ProfileName != "My Profile" { + t.Errorf("expected ProfileName 'My Profile', got '%s'", info.ProfileName) + } + if info.AgentID != "agent-456" { + t.Errorf("expected AgentID 'agent-456', got '%s'", info.AgentID) + } + if info.AgentName != "claude" { + t.Errorf("expected AgentName 'claude', got '%s'", info.AgentName) + } + if info.Model != "claude-3.5-sonnet" { + t.Errorf("expected Model 'claude-3.5-sonnet', got '%s'", info.Model) + } + if info.AutoApprove != true { + t.Error("expected AutoApprove to be true") + } + if info.DangerouslySkipPermissions != false { + t.Error("expected DangerouslySkipPermissions to be false") + } } } @@ -270,9 +270,8 @@ func TestStoreProfileResolver_ResolveProfile_FallbackToRegistryDefaultModel(t *t } if info == nil { t.Fatal("expected non-nil profile info") - } - // Should have fallback to registry's default model - if info.Model != "claude-sonnet-4-20250514" { + } else if info.Model != "claude-sonnet-4-20250514" { + // Should have fallback to registry's default model t.Errorf("expected Model 'claude-sonnet-4-20250514' (from registry), got '%s'", info.Model) } } @@ -306,9 +305,8 @@ func TestStoreProfileResolver_ResolveProfile_EmptyModelNoRegistry(t *testing.T) } if info == nil { t.Fatal("expected non-nil profile info") - } - // Model should remain empty since no registry fallback available - if info.Model != "" { + } else if info.Model != "" { + // Model should remain empty since no registry fallback available t.Errorf("expected empty Model when no registry fallback, got '%s'", info.Model) } } diff --git a/apps/backend/internal/agent/lifecycle/runtime_registry_test.go b/apps/backend/internal/agent/lifecycle/runtime_registry_test.go index 991446876b..4f9533112b 100644 --- a/apps/backend/internal/agent/lifecycle/runtime_registry_test.go +++ b/apps/backend/internal/agent/lifecycle/runtime_registry_test.go @@ -47,12 +47,13 @@ func TestNewRuntimeRegistry(t *testing.T) { if registry == nil { t.Fatal("expected non-nil registry") - } - if registry.runtimes == nil { - t.Error("expected runtimes map to be initialized") - } - if len(registry.runtimes) != 0 { - t.Errorf("expected empty runtimes map, got %d entries", len(registry.runtimes)) + } else { + if registry.runtimes == nil { + t.Error("expected runtimes map to be initialized") + } + if len(registry.runtimes) != 0 { + t.Errorf("expected empty runtimes map, got %d entries", len(registry.runtimes)) + } } } diff --git a/apps/backend/internal/agent/lifecycle/session.go b/apps/backend/internal/agent/lifecycle/session.go index eca17d0a7a..633380f97e 100644 --- a/apps/backend/internal/agent/lifecycle/session.go +++ b/apps/backend/internal/agent/lifecycle/session.go @@ -266,6 +266,20 @@ func (sm *SessionManager) InitializeAndPrompt( zap.String("existing_acp_session_id", execution.ACPSessionID), zap.Bool("native_session_resume", agentConfig.SessionConfig.NativeSessionResume)) + // Connect WebSocket streams FIRST — agent operations now go over the stream + if sm.streamManager != nil { + updatesReady := make(chan struct{}) + sm.streamManager.ConnectAll(execution, updatesReady) + + // Wait for the updates stream to connect — required for agent operations + select { + case <-updatesReady: + sm.logger.Debug("updates stream ready") + case <-time.After(10 * time.Second): + return fmt.Errorf("timeout waiting for agent stream to connect") + } + } + // Use InitializeSession for configuration-driven session initialization result, err := sm.InitializeSession( ctx, @@ -295,20 +309,6 @@ func (sm *SessionManager) InitializeAndPrompt( sm.eventPublisher.PublishACPSessionCreated(execution, result.SessionID) } - // Set up WebSocket streams - if sm.streamManager != nil { - updatesReady := make(chan struct{}) - sm.streamManager.ConnectAll(execution, updatesReady) - - // Wait for the updates stream to connect before sending prompt - select { - case <-updatesReady: - sm.logger.Debug("updates stream ready") - case <-time.After(5 * time.Second): - sm.logger.Warn("timeout waiting for updates stream to connect, proceeding anyway") - } - } - // Send the task prompt if provided - run asynchronously so orchestrator.start returns quickly. // The agent will process the prompt and emit events via the WebSocket stream. // diff --git a/apps/backend/internal/agent/lifecycle/session_test.go b/apps/backend/internal/agent/lifecycle/session_test.go new file mode 100644 index 0000000000..3233e21178 --- /dev/null +++ b/apps/backend/internal/agent/lifecycle/session_test.go @@ -0,0 +1,503 @@ +package lifecycle + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/gorilla/websocket" + "github.com/kandev/kandev/internal/agent/registry" + agentctl "github.com/kandev/kandev/internal/agentctl/client" + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/pkg/agent" + ws "github.com/kandev/kandev/pkg/websocket" +) + +func newSessionTestLogger() *logger.Logger { + log, _ := logger.NewLogger(logger.LoggingConfig{ + Level: "error", + Format: "json", + }) + return log +} + +// mockAgentServer creates a test WebSocket server simulating agentctl. +// It responds to agent stream requests and tracks which actions were called and in what order. +type mockAgentServer struct { + server *httptest.Server + mu sync.Mutex + actionLog []string // ordered log of actions received + upgrader websocket.Upgrader + handler func(msg ws.Message) *ws.Message + wsConnected chan struct{} // closed when WS stream connects +} + +func newMockAgentServer(t *testing.T) *mockAgentServer { + t.Helper() + m := &mockAgentServer{ + wsConnected: make(chan struct{}), + upgrader: websocket.Upgrader{ + CheckOrigin: func(r *http.Request) bool { return true }, + }, + } + + mux := http.NewServeMux() + + // Agent stream WebSocket endpoint + mux.HandleFunc("/api/v1/agent/stream", func(w http.ResponseWriter, r *http.Request) { + conn, err := m.upgrader.Upgrade(w, r, nil) + if err != nil { + return + } + defer func() { _ = conn.Close() }() + + // Signal that WS is connected + select { + case <-m.wsConnected: + default: + close(m.wsConnected) + } + + for { + _, message, err := conn.ReadMessage() + if err != nil { + return + } + + var msg ws.Message + if err := json.Unmarshal(message, &msg); err != nil { + continue + } + + if msg.Type == ws.MessageTypeRequest { + m.mu.Lock() + m.actionLog = append(m.actionLog, msg.Action) + m.mu.Unlock() + + var resp *ws.Message + if m.handler != nil { + resp = m.handler(msg) + } else { + resp = m.defaultHandler(msg) + } + + if resp != nil { + data, _ := json.Marshal(resp) + if err := conn.WriteMessage(websocket.TextMessage, data); err != nil { + return + } + } + } + } + }) + + // Workspace stream WebSocket endpoint (required by ConnectAll) + mux.HandleFunc("/api/v1/workspace/stream", func(w http.ResponseWriter, r *http.Request) { + conn, err := m.upgrader.Upgrade(w, r, nil) + if err != nil { + return + } + defer func() { _ = conn.Close() }() + + // Send connected message + connMsg := map[string]string{"type": "connected"} + data, _ := json.Marshal(connMsg) + _ = conn.WriteMessage(websocket.TextMessage, data) + + // Keep alive + for { + if _, _, err := conn.ReadMessage(); err != nil { + return + } + } + }) + + m.server = httptest.NewServer(mux) + return m +} + +func (m *mockAgentServer) defaultHandler(msg ws.Message) *ws.Message { + switch msg.Action { + case "agent.initialize": + resp, _ := ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "success": true, + "agent_info": map[string]string{ + "name": "test-agent", + "version": "1.0.0", + }, + }) + return resp + case "agent.session.new": + resp, _ := ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "success": true, + "session_id": "test-session-123", + }) + return resp + case "agent.session.load": + resp, _ := ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "success": true, + "session_id": "loaded-session", + }) + return resp + case "agent.prompt": + resp, _ := ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "success": true, + }) + return resp + default: + resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeUnknownAction, "unknown action", nil) + return resp + } +} + +func (m *mockAgentServer) getActionLog() []string { + m.mu.Lock() + defer m.mu.Unlock() + result := make([]string, len(m.actionLog)) + copy(result, m.actionLog) + return result +} + +func (m *mockAgentServer) Close() { + m.server.Close() +} + +// createTestClient creates an agentctl.Client pointed at the mock server. +func createTestClient(t *testing.T, serverURL string) *agentctl.Client { + t.Helper() + // Parse the test server URL (http://127.0.0.1:PORT) + url := strings.TrimPrefix(serverURL, "http://") + parts := strings.Split(url, ":") + host := parts[0] + var port int + _, _ = fmt.Sscanf(parts[1], "%d", &port) + + log := newSessionTestLogger() + return agentctl.NewClient(host, port, log) +} + +// --- Tests --- + +func TestInitializeAndPrompt_StreamBeforeInitialize(t *testing.T) { + // This test verifies the critical ordering: stream connects BEFORE initialize is called. + mock := newMockAgentServer(t) + defer mock.Close() + + log := newSessionTestLogger() + sm := NewSessionManager(log, make(chan struct{})) + + // Set up real stream manager with callbacks + streamMgr := NewStreamManager(log, StreamCallbacks{ + OnAgentEvent: func(execution *AgentExecution, event agentctl.AgentEvent) {}, + }, nil) + sm.SetDependencies(nil, streamMgr, nil, nil) + + client := createTestClient(t, mock.server.URL) + defer client.Close() + + execution := &AgentExecution{ + ID: "test-exec", + TaskID: "test-task", + SessionID: "test-session", + WorkspacePath: "/workspace", + agentctl: client, + promptDoneCh: make(chan PromptCompletionSignal, 1), + } + + agentConfig := ®istry.AgentTypeConfig{ + ID: "test-agent", + Protocol: agent.ProtocolACP, + SessionConfig: registry.SessionConfig{ + NativeSessionResume: false, + }, + } + + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + err := sm.InitializeAndPrompt(ctx, execution, agentConfig, "", nil, func(executionID string) error { + return nil + }) + if err != nil { + t.Fatalf("InitializeAndPrompt failed: %v", err) + } + + // Wait for stream to connect (it should have connected before initialize) + select { + case <-mock.wsConnected: + case <-time.After(5 * time.Second): + t.Fatal("stream never connected") + } + + // Verify the ordering: initialize and session.new should have been called + // (the stream was connected first, then these went through it) + actions := mock.getActionLog() + if len(actions) < 2 { + t.Fatalf("expected at least 2 actions, got %d: %v", len(actions), actions) + } + if actions[0] != "agent.initialize" { + t.Errorf("expected first action to be 'agent.initialize', got %q", actions[0]) + } + if actions[1] != "agent.session.new" { + t.Errorf("expected second action to be 'agent.session.new', got %q", actions[1]) + } + + // The session ID should be set + if execution.ACPSessionID != "test-session-123" { + t.Errorf("expected ACPSessionID 'test-session-123', got %q", execution.ACPSessionID) + } +} + +func TestInitializeAndPrompt_StreamTimeout(t *testing.T) { + // This test verifies that InitializeAndPrompt returns an error if + // the stream fails to connect within the timeout. + log := newSessionTestLogger() + sm := NewSessionManager(log, make(chan struct{})) + + // Create a stream manager that will try to connect to a server that doesn't exist + streamMgr := NewStreamManager(log, StreamCallbacks{ + OnAgentEvent: func(execution *AgentExecution, event agentctl.AgentEvent) {}, + }, nil) + sm.SetDependencies(nil, streamMgr, nil, nil) + + // Point client at a port that doesn't exist + badClient := agentctl.NewClient("127.0.0.1", 1, log) + defer badClient.Close() + + execution := &AgentExecution{ + ID: "test-exec", + TaskID: "test-task", + SessionID: "test-session", + WorkspacePath: "/workspace", + agentctl: badClient, + promptDoneCh: make(chan PromptCompletionSignal, 1), + } + + agentConfig := ®istry.AgentTypeConfig{ + ID: "test-agent", + Protocol: agent.ProtocolACP, + } + + // Use short timeout so test doesn't take 10s + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + err := sm.InitializeAndPrompt(ctx, execution, agentConfig, "", nil, func(executionID string) error { + return nil + }) + + // Should fail because stream couldn't connect and Initialize fails + if err == nil { + t.Fatal("expected error when stream cannot connect") + } + // The error could be either timeout waiting for stream or initialize failure + // (since stream failed, initialize over WS will also fail) +} + +func TestInitializeAndPrompt_WithTaskDescription(t *testing.T) { + // Test that when a task description is provided, prompt is sent after initialization. + mock := newMockAgentServer(t) + defer mock.Close() + + log := newSessionTestLogger() + sm := NewSessionManager(log, make(chan struct{})) + + streamMgr := NewStreamManager(log, StreamCallbacks{ + OnAgentEvent: func(execution *AgentExecution, event agentctl.AgentEvent) {}, + }, nil) + sm.SetDependencies(nil, streamMgr, nil, nil) + + client := createTestClient(t, mock.server.URL) + defer client.Close() + + execution := &AgentExecution{ + ID: "test-exec", + TaskID: "test-task", + SessionID: "test-session", + WorkspacePath: "/workspace", + agentctl: client, + promptDoneCh: make(chan PromptCompletionSignal, 1), + } + + agentConfig := ®istry.AgentTypeConfig{ + ID: "test-agent", + Protocol: agent.ProtocolACP, + SessionConfig: registry.SessionConfig{ + NativeSessionResume: false, + }, + } + + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + err := sm.InitializeAndPrompt(ctx, execution, agentConfig, "Build a feature", nil, func(executionID string) error { + return nil + }) + if err != nil { + t.Fatalf("InitializeAndPrompt failed: %v", err) + } + + // Wait for the prompt to be sent asynchronously + time.Sleep(500 * time.Millisecond) + + actions := mock.getActionLog() + + // Should have: initialize, session.new, prompt + if len(actions) < 3 { + t.Fatalf("expected at least 3 actions (initialize, session.new, prompt), got %d: %v", len(actions), actions) + } + if actions[0] != "agent.initialize" { + t.Errorf("expected first action 'agent.initialize', got %q", actions[0]) + } + if actions[1] != "agent.session.new" { + t.Errorf("expected second action 'agent.session.new', got %q", actions[1]) + } + if actions[2] != "agent.prompt" { + t.Errorf("expected third action 'agent.prompt', got %q", actions[2]) + } +} + +func TestInitializeAndPrompt_NoStreamManager(t *testing.T) { + // When streamManager is nil, InitializeAndPrompt should still work + // (though in practice this means sendStreamRequest will fail). + // This tests that the nil guard for streamManager works. + log := newSessionTestLogger() + sm := NewSessionManager(log, make(chan struct{})) + // No dependencies set — streamManager is nil + + // We can't actually call InitializeAndPrompt without a stream because + // the client's sendStreamRequest will fail. But we can verify the code path + // doesn't panic when streamManager is nil. + mock := newMockAgentServer(t) + defer mock.Close() + + client := createTestClient(t, mock.server.URL) + defer client.Close() + + execution := &AgentExecution{ + ID: "test-exec", + TaskID: "test-task", + SessionID: "test-session", + WorkspacePath: "/workspace", + agentctl: client, + promptDoneCh: make(chan PromptCompletionSignal, 1), + } + + agentConfig := ®istry.AgentTypeConfig{ + ID: "test-agent", + Protocol: agent.ProtocolACP, + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + // This will fail because the stream isn't connected (sendStreamRequest returns error). + // But it should NOT panic due to nil streamManager. + err := sm.InitializeAndPrompt(ctx, execution, agentConfig, "", nil, func(executionID string) error { + return nil + }) + + // Expect error because Initialize call over WS will fail (stream not connected) + if err == nil { + t.Fatal("expected error when stream is not connected") + } +} + +func TestInitializeSession_CreatesNewSession(t *testing.T) { + mock := newMockAgentServer(t) + defer mock.Close() + + log := newSessionTestLogger() + sm := NewSessionManager(log, make(chan struct{})) + + // Create client and manually connect stream + client := createTestClient(t, mock.server.URL) + defer client.Close() + + ctx := context.Background() + err := client.StreamUpdates(ctx, func(event agentctl.AgentEvent) {}, nil, nil) + if err != nil { + t.Fatalf("failed to connect stream: %v", err) + } + time.Sleep(100 * time.Millisecond) // let goroutine start + + agentConfig := ®istry.AgentTypeConfig{ + ID: "test-agent", + Protocol: agent.ProtocolACP, + SessionConfig: registry.SessionConfig{ + NativeSessionResume: false, + }, + } + + result, err := sm.InitializeSession(ctx, client, agentConfig, "", "/workspace", nil) + if err != nil { + t.Fatalf("InitializeSession failed: %v", err) + } + + if result.AgentName != "test-agent" { + t.Errorf("expected agent name 'test-agent', got %q", result.AgentName) + } + if result.SessionID != "test-session-123" { + t.Errorf("expected session ID 'test-session-123', got %q", result.SessionID) + } +} + +func TestInitializeSession_LoadsExistingSession(t *testing.T) { + mock := newMockAgentServer(t) + defer mock.Close() + + log := newSessionTestLogger() + sm := NewSessionManager(log, make(chan struct{})) + + client := createTestClient(t, mock.server.URL) + defer client.Close() + + ctx := context.Background() + err := client.StreamUpdates(ctx, func(event agentctl.AgentEvent) {}, nil, nil) + if err != nil { + t.Fatalf("failed to connect stream: %v", err) + } + time.Sleep(100 * time.Millisecond) + + agentConfig := ®istry.AgentTypeConfig{ + ID: "test-agent", + Protocol: agent.ProtocolACP, + SessionConfig: registry.SessionConfig{ + NativeSessionResume: true, + }, + } + + result, err := sm.InitializeSession(ctx, client, agentConfig, "existing-session", "/workspace", nil) + if err != nil { + t.Fatalf("InitializeSession failed: %v", err) + } + + if result.SessionID != "existing-session" { + t.Errorf("expected session ID 'existing-session', got %q", result.SessionID) + } + + // Verify that session.load was called (not session.new) + actions := mock.getActionLog() + hasLoad := false + hasNew := false + for _, a := range actions { + if a == "agent.session.load" { + hasLoad = true + } + if a == "agent.session.new" { + hasNew = true + } + } + if !hasLoad { + t.Error("expected agent.session.load to be called") + } + if hasNew { + t.Error("did not expect agent.session.new to be called when loading existing session") + } +} diff --git a/apps/backend/internal/agent/registry/registry_test.go b/apps/backend/internal/agent/registry/registry_test.go index 3a524a4ba2..d3acaa1f33 100644 --- a/apps/backend/internal/agent/registry/registry_test.go +++ b/apps/backend/internal/agent/registry/registry_test.go @@ -41,8 +41,7 @@ func TestNewRegistry(t *testing.T) { if reg == nil { t.Fatal("expected non-nil registry") - } - if len(reg.agents) != 0 { + } else if len(reg.agents) != 0 { t.Errorf("expected empty agents map, got %d", len(reg.agents)) } } diff --git a/apps/backend/internal/agentctl/client/agent.go b/apps/backend/internal/agentctl/client/agent.go index 18cb9714cb..d7dac026e0 100644 --- a/apps/backend/internal/agentctl/client/agent.go +++ b/apps/backend/internal/agentctl/client/agent.go @@ -1,12 +1,9 @@ package client import ( - "bytes" "context" "encoding/json" "fmt" - "net/http" - "sync" "github.com/gorilla/websocket" "github.com/kandev/kandev/internal/agentctl/types" @@ -38,9 +35,9 @@ type InitializeResponse struct { Error string `json:"error,omitempty"` } -// Initialize sends the ACP initialize request +// Initialize sends the ACP initialize request via the agent WebSocket stream. func (c *Client) Initialize(ctx context.Context, clientName, clientVersion string) (*AgentInfo, error) { - reqBody := struct { + payload := struct { ClientName string `json:"client_name"` ClientVersion string `json:"client_version"` }{ @@ -48,39 +45,22 @@ func (c *Client) Initialize(ctx context.Context, clientName, clientVersion strin ClientVersion: clientVersion, } - body, err := json.Marshal(reqBody) + resp, err := c.sendStreamRequest(ctx, "agent.initialize", payload) if err != nil { - return nil, err + return nil, fmt.Errorf("initialize request failed: %w", err) } - req, err := http.NewRequestWithContext(ctx, "POST", c.baseURL+"/api/v1/agent/initialize", bytes.NewReader(body)) - if err != nil { - return nil, err - } - req.Header.Set("Content-Type", "application/json") - - resp, err := c.httpClient.Do(req) - if err != nil { - return nil, err - } - defer func() { - if err := resp.Body.Close(); err != nil { - c.logger.Debug("failed to close initialize response body", zap.Error(err)) + if resp.Type == ws.MessageTypeError { + var errPayload ws.ErrorPayload + if err := resp.ParsePayload(&errPayload); err != nil { + return nil, fmt.Errorf("initialize failed: unable to parse error") } - }() - - respBody, err := readResponseBody(resp) - if err != nil { - return nil, fmt.Errorf("failed to read response body: %w", err) - } - - if resp.StatusCode < 200 || resp.StatusCode >= 300 { - return nil, fmt.Errorf("initialize request failed with status %d: %s", resp.StatusCode, string(respBody)) + return nil, fmt.Errorf("initialize failed: %s", errPayload.Message) } var result InitializeResponse - if err := json.Unmarshal(respBody, &result); err != nil { - return nil, fmt.Errorf("failed to parse initialize response (status %d, body: %s): %w", resp.StatusCode, truncateBody(respBody), err) + if err := resp.ParsePayload(&result); err != nil { + return nil, fmt.Errorf("failed to parse initialize response: %w", err) } if !result.Success { return nil, fmt.Errorf("initialize failed: %s", result.Error) @@ -95,46 +75,29 @@ type NewSessionResponse struct { Error string `json:"error,omitempty"` } -// NewSession creates a new ACP session +// NewSession creates a new ACP session via the agent WebSocket stream. func (c *Client) NewSession(ctx context.Context, cwd string, mcpServers []types.McpServer) (string, error) { - reqBody := struct { + payload := struct { Cwd string `json:"cwd"` McpServers []types.McpServer `json:"mcp_servers,omitempty"` }{Cwd: cwd, McpServers: mcpServers} - body, err := json.Marshal(reqBody) + resp, err := c.sendStreamRequest(ctx, "agent.session.new", payload) if err != nil { - return "", err + return "", fmt.Errorf("new session request failed: %w", err) } - req, err := http.NewRequestWithContext(ctx, "POST", c.baseURL+"/api/v1/agent/session/new", bytes.NewReader(body)) - if err != nil { - return "", err - } - req.Header.Set("Content-Type", "application/json") - - resp, err := c.httpClient.Do(req) - if err != nil { - return "", err - } - defer func() { - if err := resp.Body.Close(); err != nil { - c.logger.Debug("failed to close session response body", zap.Error(err)) + if resp.Type == ws.MessageTypeError { + var errPayload ws.ErrorPayload + if err := resp.ParsePayload(&errPayload); err != nil { + return "", fmt.Errorf("new session failed: unable to parse error") } - }() - - respBody, err := readResponseBody(resp) - if err != nil { - return "", fmt.Errorf("failed to read response body: %w", err) - } - - if resp.StatusCode < 200 || resp.StatusCode >= 300 { - return "", fmt.Errorf("new session request failed with status %d: %s", resp.StatusCode, string(respBody)) + return "", fmt.Errorf("new session failed: %s", errPayload.Message) } var result NewSessionResponse - if err := json.Unmarshal(respBody, &result); err != nil { - return "", fmt.Errorf("failed to parse new session response (status %d, body: %s): %w", resp.StatusCode, truncateBody(respBody), err) + if err := resp.ParsePayload(&result); err != nil { + return "", fmt.Errorf("failed to parse new session response: %w", err) } if !result.Success { return "", fmt.Errorf("new session failed: %s", result.Error) @@ -142,48 +105,31 @@ func (c *Client) NewSession(ctx context.Context, cwd string, mcpServers []types. return result.SessionID, nil } -// LoadSession resumes an existing ACP session +// LoadSession resumes an existing ACP session via the agent WebSocket stream. func (c *Client) LoadSession(ctx context.Context, sessionID string) error { - reqBody := struct { + payload := struct { SessionID string `json:"session_id"` }{SessionID: sessionID} - body, err := json.Marshal(reqBody) + resp, err := c.sendStreamRequest(ctx, "agent.session.load", payload) if err != nil { - return err + return fmt.Errorf("load session request failed: %w", err) } - req, err := http.NewRequestWithContext(ctx, "POST", c.baseURL+"/api/v1/agent/session/load", bytes.NewReader(body)) - if err != nil { - return err - } - req.Header.Set("Content-Type", "application/json") - - resp, err := c.httpClient.Do(req) - if err != nil { - return err - } - defer func() { - if err := resp.Body.Close(); err != nil { - c.logger.Debug("failed to close load session response body", zap.Error(err)) + if resp.Type == ws.MessageTypeError { + var errPayload ws.ErrorPayload + if err := resp.ParsePayload(&errPayload); err != nil { + return fmt.Errorf("load session failed: unable to parse error") } - }() - - respBody, err := readResponseBody(resp) - if err != nil { - return fmt.Errorf("failed to read response body: %w", err) - } - - if resp.StatusCode < 200 || resp.StatusCode >= 300 { - return fmt.Errorf("load session request failed with status %d: %s", resp.StatusCode, string(respBody)) + return fmt.Errorf("load session failed: %s", errPayload.Message) } var result struct { Success bool `json:"success"` Error string `json:"error,omitempty"` } - if err := json.Unmarshal(respBody, &result); err != nil { - return fmt.Errorf("failed to parse load session response (status %d, body: %s): %w", resp.StatusCode, truncateBody(respBody), err) + if err := resp.ParsePayload(&result); err != nil { + return fmt.Errorf("failed to parse load session response: %w", err) } if !result.Success { return fmt.Errorf("load session failed: %s", result.Error) @@ -191,58 +137,37 @@ func (c *Client) LoadSession(ctx context.Context, sessionID string) error { return nil } -// Prompt sends a fire-and-forget prompt to the agent. -// The agentctl server returns 202 immediately; completion is signaled via the WebSocket complete event. +// Prompt sends a fire-and-forget prompt to the agent via the agent WebSocket stream. +// The server returns an accepted response immediately; completion is signaled via stream events. // Attachments (images) are passed to the agent if provided. func (c *Client) Prompt(ctx context.Context, text string, attachments []v1.MessageAttachment) error { - reqBody := struct { + payload := struct { Text string `json:"text"` Attachments []v1.MessageAttachment `json:"attachments,omitempty"` }{Text: text, Attachments: attachments} - body, err := json.Marshal(reqBody) + resp, err := c.sendStreamRequest(ctx, "agent.prompt", payload) if err != nil { - return err + return fmt.Errorf("prompt request failed: %w", err) } - req, err := http.NewRequestWithContext(ctx, "POST", c.baseURL+"/api/v1/agent/prompt", bytes.NewReader(body)) - if err != nil { - return err - } - req.Header.Set("Content-Type", "application/json") - - resp, err := c.httpClient.Do(req) - if err != nil { - return err - } - defer func() { - if err := resp.Body.Close(); err != nil { - c.logger.Debug("failed to close prompt response body", zap.Error(err)) + if resp.Type == ws.MessageTypeError { + var errPayload ws.ErrorPayload + if err := resp.ParsePayload(&errPayload); err != nil { + return fmt.Errorf("prompt failed: unable to parse error") } - }() - - respBody, err := readResponseBody(resp) - if err != nil { - return fmt.Errorf("failed to read response body: %w", err) - } - - // Accept both 200 (backwards compat) and 202 (async) - if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusAccepted { - return fmt.Errorf("prompt request failed with status %d: %s", resp.StatusCode, string(respBody)) + return fmt.Errorf("prompt failed: %s", errPayload.Message) } var result struct { Success bool `json:"success"` Error string `json:"error,omitempty"` } - if err := json.Unmarshal(respBody, &result); err != nil { - return fmt.Errorf("failed to parse prompt response (status %d, body: %s): %w", resp.StatusCode, truncateBody(respBody), err) + if err := resp.ParsePayload(&result); err != nil { + return fmt.Errorf("failed to parse prompt response: %w", err) } if !result.Success { - c.logger.Warn("prompt returned failure response", - zap.String("error", result.Error), - zap.Int("status_code", resp.StatusCode), - zap.String("response_body", truncateBody(respBody))) + c.logger.Warn("prompt returned failure response", zap.String("error", result.Error)) return fmt.Errorf("prompt failed: %s", result.Error) } return nil @@ -275,11 +200,10 @@ func (c *Client) StreamUpdates(ctx context.Context, handler func(AgentEvent), mc c.logger.Info("connected to updates stream", zap.String("url", wsURL)) - // Use mutex for writing responses - var writeMu sync.Mutex + // writeMessage uses the shared stream write mutex for thread-safe writes writeMessage := func(data []byte) error { - writeMu.Lock() - defer writeMu.Unlock() + c.streamWriteMu.Lock() + defer c.streamWriteMu.Unlock() return conn.WriteMessage(websocket.TextMessage, data) } @@ -287,6 +211,9 @@ func (c *Client) StreamUpdates(ctx context.Context, handler func(AgentEvent), mc go func() { var lastErr error defer func() { + // Clean up pending requests before signaling disconnect + c.cleanupPendingRequests() + c.mu.Lock() c.agentStreamConn = nil c.mu.Unlock() @@ -312,33 +239,40 @@ func (c *Client) StreamUpdates(ctx context.Context, handler func(AgentEvent), mc return } - // Try to parse as ws.Message to check if it's an MCP request + // Try to parse as ws.Message to check message type var wsMsg ws.Message - if err := json.Unmarshal(message, &wsMsg); err == nil && wsMsg.Type == ws.MessageTypeRequest { - // This is an MCP request - dispatch it - if mcpHandler != nil { - go func(msg ws.Message) { - resp, err := mcpHandler.Dispatch(ctx, &msg) - if err != nil { - c.logger.Error("MCP dispatch error", zap.Error(err)) - resp, _ = ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) - } - if resp != nil { - data, err := json.Marshal(resp) - if err != nil { - c.logger.Error("failed to marshal MCP response", zap.Error(err)) - return - } - if err := writeMessage(data); err != nil { - c.logger.Debug("failed to write MCP response", zap.Error(err)) - } - } - }(wsMsg) + if err := json.Unmarshal(message, &wsMsg); err == nil { + // Check if this is a response/error to a pending request + if (wsMsg.Type == ws.MessageTypeResponse || wsMsg.Type == ws.MessageTypeError) && c.resolvePendingRequest(&wsMsg) { + continue + } + + // This is an MCP request - dispatch it + if wsMsg.Type == ws.MessageTypeRequest { + if mcpHandler != nil { + go func(msg ws.Message) { + resp, err := mcpHandler.Dispatch(ctx, &msg) + if err != nil { + c.logger.Error("MCP dispatch error", zap.Error(err)) + resp, _ = ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) + } + if resp != nil { + data, err := json.Marshal(resp) + if err != nil { + c.logger.Error("failed to marshal MCP response", zap.Error(err)) + return + } + if err := writeMessage(data); err != nil { + c.logger.Debug("failed to write MCP response", zap.Error(err)) + } + } + }(wsMsg) + } + continue } - continue } - // Not an MCP request, parse as agent event + // Not a ws.Message request/response, parse as agent event var event AgentEvent if err := json.Unmarshal(message, &event); err != nil { c.logger.Warn("failed to parse agent event", zap.Error(err)) @@ -371,109 +305,78 @@ type CancelResponse struct { Error string `json:"error,omitempty"` } -// Cancel interrupts the current agent turn. +// Cancel interrupts the current agent turn via the agent WebSocket stream. func (c *Client) Cancel(ctx context.Context) error { - req, err := http.NewRequestWithContext(ctx, "POST", c.baseURL+"/api/v1/agent/cancel", nil) + resp, err := c.sendStreamRequest(ctx, "agent.cancel", nil) if err != nil { - return err + return fmt.Errorf("cancel request failed: %w", err) } - req.Header.Set("Content-Type", "application/json") - resp, err := c.httpClient.Do(req) - if err != nil { - return err - } - defer func() { - if err := resp.Body.Close(); err != nil { - c.logger.Debug("failed to close cancel response body", zap.Error(err)) + if resp.Type == ws.MessageTypeError { + var errPayload ws.ErrorPayload + if err := resp.ParsePayload(&errPayload); err != nil { + return fmt.Errorf("cancel failed: unable to parse error") } - }() - - respBody, err := readResponseBody(resp) - if err != nil { - return fmt.Errorf("failed to read response body: %w", err) - } - - if resp.StatusCode < 200 || resp.StatusCode >= 300 { - return fmt.Errorf("cancel request failed with status %d: %s", resp.StatusCode, string(respBody)) + return fmt.Errorf("cancel failed: %s", errPayload.Message) } var result CancelResponse - if err := json.Unmarshal(respBody, &result); err != nil { - return fmt.Errorf("failed to parse cancel response (status %d, body: %s): %w", resp.StatusCode, truncateBody(respBody), err) + if err := resp.ParsePayload(&result); err != nil { + return fmt.Errorf("failed to parse cancel response: %w", err) } if !result.Success { return fmt.Errorf("cancel failed: %s", result.Error) } return nil } -// GetAgentStderr returns recent stderr lines from the agent process. +// GetAgentStderr returns recent stderr lines from the agent process via the agent WebSocket stream. func (c *Client) GetAgentStderr(ctx context.Context) ([]string, error) { - req, err := http.NewRequestWithContext(ctx, "GET", c.baseURL+"/api/v1/agent/stderr", nil) + resp, err := c.sendStreamRequest(ctx, "agent.stderr", nil) if err != nil { - return nil, err + return nil, fmt.Errorf("agent stderr request failed: %w", err) } - resp, err := c.httpClient.Do(req) - if err != nil { - return nil, err - } - defer func() { _ = resp.Body.Close() }() - - if resp.StatusCode < 200 || resp.StatusCode >= 300 { - return nil, fmt.Errorf("agent stderr request failed with status %d", resp.StatusCode) + if resp.Type == ws.MessageTypeError { + var errPayload ws.ErrorPayload + if err := resp.ParsePayload(&errPayload); err != nil { + return nil, fmt.Errorf("agent stderr failed: unable to parse error") + } + return nil, fmt.Errorf("agent stderr failed: %s", errPayload.Message) } var result struct { Lines []string `json:"lines"` } - if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + if err := resp.ParsePayload(&result); err != nil { return nil, fmt.Errorf("failed to parse agent stderr response: %w", err) } return result.Lines, nil } -// RespondToPermission sends a response to a permission request +// RespondToPermission sends a response to a permission request via the agent WebSocket stream. func (c *Client) RespondToPermission(ctx context.Context, pendingID, optionID string, cancelled bool) error { - reqBody := PermissionRespondRequest{ + payload := PermissionRespondRequest{ PendingID: pendingID, OptionID: optionID, Cancelled: cancelled, } - body, err := json.Marshal(reqBody) + resp, err := c.sendStreamRequest(ctx, "agent.permissions.respond", payload) if err != nil { - return err + return fmt.Errorf("permission response request failed: %w", err) } - req, err := http.NewRequestWithContext(ctx, "POST", c.baseURL+"/api/v1/agent/permissions/respond", bytes.NewReader(body)) - if err != nil { - return err - } - req.Header.Set("Content-Type", "application/json") - - resp, err := c.httpClient.Do(req) - if err != nil { - return err - } - defer func() { - if err := resp.Body.Close(); err != nil { - c.logger.Debug("failed to close permission response body", zap.Error(err)) + if resp.Type == ws.MessageTypeError { + var errPayload ws.ErrorPayload + if err := resp.ParsePayload(&errPayload); err != nil { + return fmt.Errorf("permission response failed: unable to parse error") } - }() - - respBody, err := readResponseBody(resp) - if err != nil { - return fmt.Errorf("failed to read response body: %w", err) - } - - if resp.StatusCode < 200 || resp.StatusCode >= 300 { - return fmt.Errorf("permission response request failed with status %d: %s", resp.StatusCode, string(respBody)) + return fmt.Errorf("permission response failed: %s", errPayload.Message) } var result PermissionRespondResponse - if err := json.Unmarshal(respBody, &result); err != nil { - return fmt.Errorf("failed to parse permission response (status %d, body: %s): %w", resp.StatusCode, truncateBody(respBody), err) + if err := resp.ParsePayload(&result); err != nil { + return fmt.Errorf("failed to parse permission response: %w", err) } if !result.Success { return fmt.Errorf("permission response failed: %s", result.Error) diff --git a/apps/backend/internal/agentctl/client/agent_test.go b/apps/backend/internal/agentctl/client/agent_test.go new file mode 100644 index 0000000000..2a8172f6b3 --- /dev/null +++ b/apps/backend/internal/agentctl/client/agent_test.go @@ -0,0 +1,834 @@ +package client + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/gorilla/websocket" + "github.com/kandev/kandev/internal/common/logger" + ws "github.com/kandev/kandev/pkg/websocket" +) + +func newTestLogger() *logger.Logger { + log, _ := logger.NewLogger(logger.LoggingConfig{ + Level: "error", + Format: "json", + }) + return log +} + +// wsTestServer creates a test WebSocket server that echoes back response messages. +// The handler receives each request and should return a response message. +type wsTestServer struct { + server *httptest.Server + handler func(msg ws.Message) *ws.Message +} + +func newWSTestServer(t *testing.T, handler func(msg ws.Message) *ws.Message) *wsTestServer { + t.Helper() + upgrader := websocket.Upgrader{ + CheckOrigin: func(r *http.Request) bool { return true }, + } + ts := &wsTestServer{handler: handler} + + ts.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + t.Logf("upgrade error: %v", err) + return + } + defer func() { _ = conn.Close() }() + + for { + _, message, err := conn.ReadMessage() + if err != nil { + return + } + + var msg ws.Message + if err := json.Unmarshal(message, &msg); err != nil { + continue + } + + if msg.Type == ws.MessageTypeRequest && ts.handler != nil { + resp := ts.handler(msg) + if resp != nil { + data, _ := json.Marshal(resp) + if err := conn.WriteMessage(websocket.TextMessage, data); err != nil { + return + } + } + } + } + })) + + return ts +} + +func (ts *wsTestServer) Close() { + ts.server.Close() +} + +// newTestClientWithStream creates a Client connected to a test WebSocket server. +// The handler processes incoming requests and returns responses. +func newTestClientWithStream(t *testing.T, handler func(msg ws.Message) *ws.Message) (*Client, *wsTestServer) { + t.Helper() + + ts := newWSTestServer(t, handler) + + // Parse the test server URL to get host and port + // httptest URL is like http://127.0.0.1:PORT + url := ts.server.URL + log := newTestLogger() + + // Create client with the test server URL + c := &Client{ + baseURL: url, + httpClient: &http.Client{Timeout: 5 * time.Second}, + logger: log, + pendingRequests: make(map[string]chan *ws.Message), + } + + // Connect the stream + ctx := context.Background() + err := c.StreamUpdates(ctx, func(event AgentEvent) { + // no-op handler for agent events + }, nil, nil) + if err != nil { + ts.Close() + t.Fatalf("failed to connect stream: %v", err) + } + + // Give the goroutine time to start + time.Sleep(50 * time.Millisecond) + + return c, ts +} + +// --- Tests for infrastructure methods --- + +func TestSendStreamRequest_NotConnected(t *testing.T) { + log := newTestLogger() + c := &Client{ + baseURL: "http://localhost:0", + logger: log, + pendingRequests: make(map[string]chan *ws.Message), + } + + ctx := context.Background() + _, err := c.sendStreamRequest(ctx, "test.action", nil) + if err == nil { + t.Fatal("expected error when stream not connected") + } + if !strings.Contains(err.Error(), "agent stream not connected") { + t.Fatalf("expected 'agent stream not connected' error, got: %v", err) + } +} + +func TestSendStreamRequest_ContextCancelled(t *testing.T) { + // Server that never responds + c, ts := newTestClientWithStream(t, func(msg ws.Message) *ws.Message { + return nil // don't respond + }) + defer ts.Close() + defer c.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + + _, err := c.sendStreamRequest(ctx, "test.action", nil) + if err == nil { + t.Fatal("expected error on timeout") + } + if !strings.Contains(err.Error(), "context deadline exceeded") { + t.Fatalf("expected deadline exceeded error, got: %v", err) + } +} + +func TestResolvePendingRequest_NoMatch(t *testing.T) { + log := newTestLogger() + c := &Client{ + logger: log, + pendingRequests: make(map[string]chan *ws.Message), + } + + msg := &ws.Message{ID: "nonexistent", Type: ws.MessageTypeResponse} + if c.resolvePendingRequest(msg) { + t.Fatal("expected false for unmatched request") + } +} + +func TestResolvePendingRequest_EmptyID(t *testing.T) { + log := newTestLogger() + c := &Client{ + logger: log, + pendingRequests: make(map[string]chan *ws.Message), + } + + msg := &ws.Message{ID: "", Type: ws.MessageTypeResponse} + if c.resolvePendingRequest(msg) { + t.Fatal("expected false for empty ID") + } +} + +func TestResolvePendingRequest_Match(t *testing.T) { + log := newTestLogger() + c := &Client{ + logger: log, + pendingRequests: make(map[string]chan *ws.Message), + } + + ch := make(chan *ws.Message, 1) + c.pendingMu.Lock() + c.pendingRequests["test-id"] = ch + c.pendingMu.Unlock() + + msg := &ws.Message{ID: "test-id", Type: ws.MessageTypeResponse, Action: "test.action"} + if !c.resolvePendingRequest(msg) { + t.Fatal("expected true for matched request") + } + + select { + case resp := <-ch: + if resp.ID != "test-id" { + t.Fatalf("expected response ID 'test-id', got %q", resp.ID) + } + default: + t.Fatal("expected response in channel") + } +} + +func TestCleanupPendingRequests(t *testing.T) { + log := newTestLogger() + c := &Client{ + logger: log, + pendingRequests: make(map[string]chan *ws.Message), + } + + ch1 := make(chan *ws.Message, 1) + ch2 := make(chan *ws.Message, 1) + c.pendingMu.Lock() + c.pendingRequests["req-1"] = ch1 + c.pendingRequests["req-2"] = ch2 + c.pendingMu.Unlock() + + c.cleanupPendingRequests() + + c.pendingMu.Lock() + count := len(c.pendingRequests) + c.pendingMu.Unlock() + + if count != 0 { + t.Fatalf("expected 0 pending requests after cleanup, got %d", count) + } + + // Channels should be closed + _, ok := <-ch1 + if ok { + t.Fatal("expected ch1 to be closed") + } + _, ok = <-ch2 + if ok { + t.Fatal("expected ch2 to be closed") + } +} + +func TestCleanupPendingRequests_ConcurrentSafe(t *testing.T) { + log := newTestLogger() + c := &Client{ + logger: log, + pendingRequests: make(map[string]chan *ws.Message), + } + + // Add many pending requests + for i := 0; i < 50; i++ { + ch := make(chan *ws.Message, 1) + c.pendingMu.Lock() + c.pendingRequests[strings.Repeat("x", i+1)] = ch + c.pendingMu.Unlock() + } + + // Cleanup concurrently + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + c.cleanupPendingRequests() + }() + wg.Wait() + + c.pendingMu.Lock() + count := len(c.pendingRequests) + c.pendingMu.Unlock() + + if count != 0 { + t.Fatalf("expected 0 pending requests, got %d", count) + } +} + +// --- Tests for the 7 migrated methods --- + +func TestInitialize_Success(t *testing.T) { + c, ts := newTestClientWithStream(t, func(msg ws.Message) *ws.Message { + if msg.Action != "agent.initialize" { + t.Errorf("expected action 'agent.initialize', got %q", msg.Action) + } + resp, _ := ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "success": true, + "agent_info": map[string]string{ + "name": "test-agent", + "version": "1.0.0", + }, + }) + return resp + }) + defer ts.Close() + defer c.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + info, err := c.Initialize(ctx, "kandev", "1.0.0") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if info == nil { + t.Fatal("expected agent info, got nil") + } else { + if info.Name != "test-agent" { + t.Errorf("expected name 'test-agent', got %q", info.Name) + } + if info.Version != "1.0.0" { + t.Errorf("expected version '1.0.0', got %q", info.Version) + } + } +} + +func TestInitialize_ServerError(t *testing.T) { + c, ts := newTestClientWithStream(t, func(msg ws.Message) *ws.Message { + resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "agent not running", nil) + return resp + }) + defer ts.Close() + defer c.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + _, err := c.Initialize(ctx, "kandev", "1.0.0") + if err == nil { + t.Fatal("expected error") + } + if !strings.Contains(err.Error(), "agent not running") { + t.Fatalf("expected 'agent not running' error, got: %v", err) + } +} + +func TestInitialize_SuccessFalse(t *testing.T) { + c, ts := newTestClientWithStream(t, func(msg ws.Message) *ws.Message { + resp, _ := ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "success": false, + "error": "init failed internally", + }) + return resp + }) + defer ts.Close() + defer c.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + _, err := c.Initialize(ctx, "kandev", "1.0.0") + if err == nil { + t.Fatal("expected error") + } + if !strings.Contains(err.Error(), "init failed internally") { + t.Fatalf("expected 'init failed internally' error, got: %v", err) + } +} + +func TestNewSession_Success(t *testing.T) { + c, ts := newTestClientWithStream(t, func(msg ws.Message) *ws.Message { + if msg.Action != "agent.session.new" { + t.Errorf("expected action 'agent.session.new', got %q", msg.Action) + } + resp, _ := ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "success": true, + "session_id": "sess-123", + }) + return resp + }) + defer ts.Close() + defer c.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + sessionID, err := c.NewSession(ctx, "/workspace", nil) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if sessionID != "sess-123" { + t.Errorf("expected session ID 'sess-123', got %q", sessionID) + } +} + +func TestNewSession_Error(t *testing.T) { + c, ts := newTestClientWithStream(t, func(msg ws.Message) *ws.Message { + resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "agent not running", nil) + return resp + }) + defer ts.Close() + defer c.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + _, err := c.NewSession(ctx, "/workspace", nil) + if err == nil { + t.Fatal("expected error") + } + if !strings.Contains(err.Error(), "agent not running") { + t.Fatalf("expected 'agent not running' error, got: %v", err) + } +} + +func TestLoadSession_Success(t *testing.T) { + c, ts := newTestClientWithStream(t, func(msg ws.Message) *ws.Message { + if msg.Action != "agent.session.load" { + t.Errorf("expected action 'agent.session.load', got %q", msg.Action) + } + resp, _ := ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "success": true, + "session_id": "sess-456", + }) + return resp + }) + defer ts.Close() + defer c.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + err := c.LoadSession(ctx, "sess-456") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } +} + +func TestLoadSession_Error(t *testing.T) { + c, ts := newTestClientWithStream(t, func(msg ws.Message) *ws.Message { + resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Method not found", nil) + return resp + }) + defer ts.Close() + defer c.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + err := c.LoadSession(ctx, "sess-456") + if err == nil { + t.Fatal("expected error") + } + if !strings.Contains(err.Error(), "Method not found") { + t.Fatalf("expected 'Method not found' error, got: %v", err) + } +} + +func TestPrompt_Success(t *testing.T) { + c, ts := newTestClientWithStream(t, func(msg ws.Message) *ws.Message { + if msg.Action != "agent.prompt" { + t.Errorf("expected action 'agent.prompt', got %q", msg.Action) + } + // Verify payload + var payload struct { + Text string `json:"text"` + } + if err := msg.ParsePayload(&payload); err != nil { + t.Errorf("failed to parse payload: %v", err) + } + if payload.Text != "hello agent" { + t.Errorf("expected text 'hello agent', got %q", payload.Text) + } + resp, _ := ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "success": true, + }) + return resp + }) + defer ts.Close() + defer c.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + err := c.Prompt(ctx, "hello agent", nil) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } +} + +func TestPrompt_NoActiveSession(t *testing.T) { + c, ts := newTestClientWithStream(t, func(msg ws.Message) *ws.Message { + resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "no active session - call new_session first", nil) + return resp + }) + defer ts.Close() + defer c.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + err := c.Prompt(ctx, "hello agent", nil) + if err == nil { + t.Fatal("expected error") + } + if !strings.Contains(err.Error(), "no active session") { + t.Fatalf("expected 'no active session' error, got: %v", err) + } +} + +func TestCancel_Success(t *testing.T) { + c, ts := newTestClientWithStream(t, func(msg ws.Message) *ws.Message { + if msg.Action != "agent.cancel" { + t.Errorf("expected action 'agent.cancel', got %q", msg.Action) + } + resp, _ := ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "success": true, + }) + return resp + }) + defer ts.Close() + defer c.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + err := c.Cancel(ctx) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } +} + +func TestCancel_Error(t *testing.T) { + c, ts := newTestClientWithStream(t, func(msg ws.Message) *ws.Message { + resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "agent not running", nil) + return resp + }) + defer ts.Close() + defer c.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + err := c.Cancel(ctx) + if err == nil { + t.Fatal("expected error") + } + if !strings.Contains(err.Error(), "agent not running") { + t.Fatalf("expected 'agent not running' error, got: %v", err) + } +} + +func TestGetAgentStderr_Success(t *testing.T) { + c, ts := newTestClientWithStream(t, func(msg ws.Message) *ws.Message { + if msg.Action != "agent.stderr" { + t.Errorf("expected action 'agent.stderr', got %q", msg.Action) + } + resp, _ := ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "lines": []string{"error line 1", "error line 2"}, + }) + return resp + }) + defer ts.Close() + defer c.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + lines, err := c.GetAgentStderr(ctx) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(lines) != 2 { + t.Fatalf("expected 2 lines, got %d", len(lines)) + } + if lines[0] != "error line 1" { + t.Errorf("expected 'error line 1', got %q", lines[0]) + } +} + +func TestGetAgentStderr_NotConnected(t *testing.T) { + log := newTestLogger() + c := &Client{ + baseURL: "http://localhost:0", + logger: log, + pendingRequests: make(map[string]chan *ws.Message), + } + + ctx := context.Background() + _, err := c.GetAgentStderr(ctx) + if err == nil { + t.Fatal("expected error when stream not connected") + } + if !strings.Contains(err.Error(), "agent stream not connected") { + t.Fatalf("expected 'agent stream not connected' error, got: %v", err) + } +} + +func TestRespondToPermission_Success(t *testing.T) { + c, ts := newTestClientWithStream(t, func(msg ws.Message) *ws.Message { + if msg.Action != "agent.permissions.respond" { + t.Errorf("expected action 'agent.permissions.respond', got %q", msg.Action) + } + // Verify payload + var payload struct { + PendingID string `json:"pending_id"` + OptionID string `json:"option_id"` + Cancelled bool `json:"cancelled"` + } + if err := msg.ParsePayload(&payload); err != nil { + t.Errorf("failed to parse payload: %v", err) + } + if payload.PendingID != "perm-1" { + t.Errorf("expected pending_id 'perm-1', got %q", payload.PendingID) + } + if payload.OptionID != "allow" { + t.Errorf("expected option_id 'allow', got %q", payload.OptionID) + } + resp, _ := ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "success": true, + }) + return resp + }) + defer ts.Close() + defer c.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + err := c.RespondToPermission(ctx, "perm-1", "allow", false) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } +} + +func TestRespondToPermission_NotFound(t *testing.T) { + c, ts := newTestClientWithStream(t, func(msg ws.Message) *ws.Message { + resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeNotFound, "permission request not found", nil) + return resp + }) + defer ts.Close() + defer c.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + err := c.RespondToPermission(ctx, "nonexistent", "allow", false) + if err == nil { + t.Fatal("expected error") + } + if !strings.Contains(err.Error(), "permission request not found") { + t.Fatalf("expected 'permission request not found' error, got: %v", err) + } +} + +// --- Tests for StreamUpdates response routing --- + +func TestStreamUpdates_RoutesResponsesToPending(t *testing.T) { + // This test verifies that StreamUpdates correctly routes response messages + // to pending requests instead of treating them as agent events. + var receivedEvents []AgentEvent + var mu sync.Mutex + + upgrader := websocket.Upgrader{ + CheckOrigin: func(r *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 { + return + } + defer func() { _ = conn.Close() }() + + for { + _, message, err := conn.ReadMessage() + if err != nil { + return + } + var msg ws.Message + if err := json.Unmarshal(message, &msg); err != nil { + continue + } + if msg.Type == ws.MessageTypeRequest { + // Send response back + resp, _ := ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "success": true, + }) + data, _ := json.Marshal(resp) + _ = conn.WriteMessage(websocket.TextMessage, data) + + // Also send an agent event + event := AgentEvent{Type: "message_chunk", Text: "hello"} + eventData, _ := json.Marshal(event) + _ = conn.WriteMessage(websocket.TextMessage, eventData) + } + } + })) + defer server.Close() + + log := newTestLogger() + c := &Client{ + baseURL: server.URL, + httpClient: &http.Client{Timeout: 5 * time.Second}, + logger: log, + pendingRequests: make(map[string]chan *ws.Message), + } + + ctx := context.Background() + err := c.StreamUpdates(ctx, func(event AgentEvent) { + mu.Lock() + receivedEvents = append(receivedEvents, event) + mu.Unlock() + }, nil, nil) + if err != nil { + t.Fatalf("failed to connect: %v", err) + } + + time.Sleep(50 * time.Millisecond) + + // Send a request and get a response + reqCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + resp, err := c.sendStreamRequest(reqCtx, "test.action", nil) + if err != nil { + t.Fatalf("sendStreamRequest failed: %v", err) + } + if resp.Type != ws.MessageTypeResponse { + t.Errorf("expected response type, got %q", resp.Type) + } + + // Wait for the agent event to be processed + time.Sleep(100 * time.Millisecond) + + mu.Lock() + eventCount := len(receivedEvents) + mu.Unlock() + + if eventCount != 1 { + t.Errorf("expected 1 agent event, got %d", eventCount) + } + + c.Close() +} + +func TestStreamUpdates_DisconnectCleansPending(t *testing.T) { + // This test verifies that pending requests are cleaned up on stream disconnect. + upgrader := websocket.Upgrader{ + CheckOrigin: func(r *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 { + return + } + // Close immediately to trigger disconnect + time.Sleep(100 * time.Millisecond) + _ = conn.Close() + })) + defer server.Close() + + log := newTestLogger() + c := &Client{ + baseURL: server.URL, + httpClient: &http.Client{Timeout: 5 * time.Second}, + logger: log, + pendingRequests: make(map[string]chan *ws.Message), + } + + disconnected := make(chan struct{}) + ctx := context.Background() + err := c.StreamUpdates(ctx, func(event AgentEvent) {}, nil, func(err error) { + close(disconnected) + }) + if err != nil { + t.Fatalf("failed to connect: %v", err) + } + + time.Sleep(50 * time.Millisecond) + + // Add a pending request + ch := make(chan *ws.Message, 1) + c.pendingMu.Lock() + c.pendingRequests["pending-test"] = ch + c.pendingMu.Unlock() + + // Wait for disconnect + select { + case <-disconnected: + case <-time.After(5 * time.Second): + t.Fatal("timeout waiting for disconnect") + } + + // The channel should be closed + _, ok := <-ch + if ok { + t.Fatal("expected channel to be closed after disconnect") + } + + c.pendingMu.Lock() + count := len(c.pendingRequests) + c.pendingMu.Unlock() + if count != 0 { + t.Fatalf("expected 0 pending requests after disconnect, got %d", count) + } +} + +// --- Concurrent request tests --- + +func TestConcurrentRequests(t *testing.T) { + // Multiple concurrent requests should all resolve correctly. + c, ts := newTestClientWithStream(t, func(msg ws.Message) *ws.Message { + // Simulate some processing time + time.Sleep(10 * time.Millisecond) + resp, _ := ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "success": true, + "action": msg.Action, + }) + return resp + }) + defer ts.Close() + defer c.Close() + + var wg sync.WaitGroup + errors := make(chan error, 10) + + for i := 0; i < 10; i++ { + wg.Add(1) + go func() { + defer wg.Done() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + _, err := c.sendStreamRequest(ctx, "test.concurrent", nil) + if err != nil { + errors <- err + } + }() + } + + wg.Wait() + close(errors) + + for err := range errors { + t.Errorf("concurrent request failed: %v", err) + } +} diff --git a/apps/backend/internal/agentctl/client/client.go b/apps/backend/internal/agentctl/client/client.go index a9d6baecc1..701c4667dd 100644 --- a/apps/backend/internal/agentctl/client/client.go +++ b/apps/backend/internal/agentctl/client/client.go @@ -11,14 +11,16 @@ import ( "sync" "time" + "github.com/google/uuid" "github.com/gorilla/websocket" "github.com/kandev/kandev/internal/agentctl/types" "github.com/kandev/kandev/internal/agentctl/types/streams" "github.com/kandev/kandev/internal/common/logger" + ws "github.com/kandev/kandev/pkg/websocket" "go.uber.org/zap" ) -// Client communicates with agentctl via HTTP +// Client communicates with agentctl via HTTP and WebSocket type Client struct { baseURL string httpClient *http.Client @@ -28,6 +30,13 @@ type Client struct { agentStreamConn *websocket.Conn workspaceStreamConn *websocket.Conn mu sync.RWMutex + + // Shared write mutex for agent stream (used by StreamUpdates and sendStreamRequest) + streamWriteMu sync.Mutex + + // Pending request/response tracking for agent stream + pendingRequests map[string]chan *ws.Message + pendingMu sync.Mutex } // StatusResponse from agentctl @@ -49,7 +58,8 @@ func NewClient(host string, port int, log *logger.Logger) *Client { httpClient: &http.Client{ Timeout: 60 * time.Second, }, - logger: log.WithFields(zap.String("component", "agentctl-client")), + logger: log.WithFields(zap.String("component", "agentctl-client")), + pendingRequests: make(map[string]chan *ws.Message), } } @@ -503,6 +513,96 @@ func (c *Client) DeleteFile(ctx context.Context, path string) (*FileDeleteRespon return &response, nil } +// sendStreamRequest sends a request over the agent WebSocket stream and waits for a response. +// It creates a ws.Message with a UUID, registers a pending response channel, +// writes the message to the stream, and blocks until a response arrives or context is cancelled. +func (c *Client) sendStreamRequest(ctx context.Context, action string, payload interface{}) (*ws.Message, error) { + c.mu.RLock() + conn := c.agentStreamConn + c.mu.RUnlock() + + if conn == nil { + return nil, fmt.Errorf("agent stream not connected") + } + + reqID := uuid.New().String() + msg, err := ws.NewRequest(reqID, action, payload) + if err != nil { + return nil, fmt.Errorf("failed to create request message: %w", err) + } + + // Register pending request + respCh := make(chan *ws.Message, 1) + c.pendingMu.Lock() + c.pendingRequests[reqID] = respCh + c.pendingMu.Unlock() + + // Clean up on exit + defer func() { + c.pendingMu.Lock() + delete(c.pendingRequests, reqID) + c.pendingMu.Unlock() + }() + + // Serialize write to stream + data, err := json.Marshal(msg) + if err != nil { + return nil, fmt.Errorf("failed to marshal request: %w", err) + } + + c.streamWriteMu.Lock() + writeErr := conn.WriteMessage(websocket.TextMessage, data) + c.streamWriteMu.Unlock() + if writeErr != nil { + return nil, fmt.Errorf("failed to write request to stream: %w", writeErr) + } + + // Wait for response or context cancellation + select { + case resp := <-respCh: + if resp == nil { + return nil, fmt.Errorf("agent stream disconnected while waiting for response") + } + return resp, nil + case <-ctx.Done(): + return nil, ctx.Err() + } +} + +// resolvePendingRequest matches a response message to a pending request by ID. +// Returns true if the message was matched to a pending request. +func (c *Client) resolvePendingRequest(msg *ws.Message) bool { + if msg.ID == "" { + return false + } + + c.pendingMu.Lock() + ch, ok := c.pendingRequests[msg.ID] + c.pendingMu.Unlock() + + if !ok { + return false + } + + // Send response to waiting caller (non-blocking since channel is buffered) + select { + case ch <- msg: + default: + } + return true +} + +// cleanupPendingRequests unblocks all pending requests with nil (signaling disconnect). +func (c *Client) cleanupPendingRequests() { + c.pendingMu.Lock() + defer c.pendingMu.Unlock() + + for id, ch := range c.pendingRequests { + close(ch) + delete(c.pendingRequests, id) + } +} + // Close closes all connections func (c *Client) Close() { c.CloseUpdatesStream() diff --git a/apps/backend/internal/agentctl/server/adapter/transport/codex/errors_test.go b/apps/backend/internal/agentctl/server/adapter/transport/codex/errors_test.go index 57f4d31d99..e9ffc4b53e 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/codex/errors_test.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/codex/errors_test.go @@ -129,18 +129,18 @@ func TestParseCodexStderrError(t *testing.T) { if result == nil { t.Fatal("expected non-nil result, got nil") - } + } else { + if result.HTTPError != tt.wantHTTPError { + t.Errorf("HTTPError = %q, want %q", result.HTTPError, tt.wantHTTPError) + } - if result.HTTPError != tt.wantHTTPError { - t.Errorf("HTTPError = %q, want %q", result.HTTPError, tt.wantHTTPError) - } + if tt.wantErrorType != "" && result.ErrorType != tt.wantErrorType { + t.Errorf("ErrorType = %q, want %q", result.ErrorType, tt.wantErrorType) + } - if tt.wantErrorType != "" && result.ErrorType != tt.wantErrorType { - t.Errorf("ErrorType = %q, want %q", result.ErrorType, tt.wantErrorType) - } - - if !strings.Contains(result.Message, tt.wantMsgContains) { - t.Errorf("Message = %q, want to contain %q", result.Message, tt.wantMsgContains) + if !strings.Contains(result.Message, tt.wantMsgContains) { + t.Errorf("Message = %q, want to contain %q", result.Message, tt.wantMsgContains) + } } if tt.wantRawJSON && result.RawJSON == nil { @@ -221,9 +221,7 @@ func TestParseCodexStderrLines(t *testing.T) { if result == nil { t.Fatal("expected non-nil result, got nil") - } - - if !strings.Contains(result.Message, tt.wantMsgContains) { + } else if !strings.Contains(result.Message, tt.wantMsgContains) { t.Errorf("Message = %q, want to contain %q", result.Message, tt.wantMsgContains) } }) @@ -237,9 +235,7 @@ func TestParseCodexStderrError_ResetsInSeconds(t *testing.T) { result := ParseCodexStderrError(input) if result == nil { t.Fatal("expected non-nil result") - } - - if result.ResetsInSeconds != 3600 { + } else if result.ResetsInSeconds != 3600 { t.Errorf("ResetsInSeconds = %d, want 3600", result.ResetsInSeconds) } } diff --git a/apps/backend/internal/agentctl/server/api/agent.go b/apps/backend/internal/agentctl/server/api/agent.go index 875214165d..3a29950ffb 100644 --- a/apps/backend/internal/agentctl/server/api/agent.go +++ b/apps/backend/internal/agentctl/server/api/agent.go @@ -4,7 +4,6 @@ import ( "context" "encoding/json" "fmt" - "net/http" "sync" "time" @@ -36,53 +35,6 @@ type InitializeResponse struct { Error string `json:"error,omitempty"` } -func (s *Server) handleAgentInitialize(c *gin.Context) { - var req InitializeRequest - if err := c.ShouldBindJSON(&req); err != nil { - c.JSON(http.StatusBadRequest, InitializeResponse{ - Success: false, - Error: "invalid request: " + err.Error(), - }) - return - } - - ctx, cancel := context.WithTimeout(c.Request.Context(), 60*time.Second) - defer cancel() - - adapter := s.procMgr.GetAdapter() - if adapter == nil { - c.JSON(http.StatusServiceUnavailable, InitializeResponse{ - Success: false, - Error: "agent not running", - }) - return - } - - err := adapter.Initialize(ctx) - if err != nil { - s.logger.Error("initialize failed", zap.Error(err)) - c.JSON(http.StatusInternalServerError, InitializeResponse{ - Success: false, - Error: err.Error(), - }) - return - } - - // Get agent info after successful initialization - var agentInfoResp *AgentInfoResponse - if info := adapter.GetAgentInfo(); info != nil { - agentInfoResp = &AgentInfoResponse{ - Name: info.Name, - Version: info.Version, - } - } - - c.JSON(http.StatusOK, InitializeResponse{ - Success: true, - AgentInfo: agentInfoResp, - }) -} - // NewSessionRequest is a request to create a new ACP session type NewSessionRequest struct { Cwd string `json:"cwd"` // Working directory for the session @@ -96,69 +48,6 @@ type NewSessionResponse struct { Error string `json:"error,omitempty"` } -func (s *Server) handleAgentNewSession(c *gin.Context) { - var req NewSessionRequest - if err := c.ShouldBindJSON(&req); err != nil { - c.JSON(http.StatusBadRequest, NewSessionResponse{ - Success: false, - Error: "invalid request: " + err.Error(), - }) - return - } - - ctx, cancel := context.WithTimeout(c.Request.Context(), 30*time.Second) - defer cancel() - - adapter := s.procMgr.GetAdapter() - if adapter == nil { - c.JSON(http.StatusServiceUnavailable, NewSessionResponse{ - Success: false, - Error: "agent not running", - }) - return - } - - // If MCP server is enabled, prepend the local kandev MCP server to the list. - // This replaces the old external MCP server URL (http://localhost:9090/sse) with - // the local agentctl MCP server (http://localhost:{port}/sse). - mcpServers := req.McpServers - if s.mcpServer != nil { - localKandevMcp := types.McpServer{ - Name: "kandev", - Type: "sse", - URL: fmt.Sprintf("http://localhost:%d/sse", s.cfg.Port), - } - // Filter out any existing kandev server from the list (to avoid duplicates) - filtered := make([]types.McpServer, 0, len(mcpServers)+1) - filtered = append(filtered, localKandevMcp) - for _, srv := range mcpServers { - if srv.Name != "kandev" { - filtered = append(filtered, srv) - } - } - mcpServers = filtered - s.logger.Debug("injected local kandev MCP server", - zap.String("url", localKandevMcp.URL), - zap.Int("total_servers", len(mcpServers))) - } - - sessionID, err := adapter.NewSession(ctx, mcpServers) - if err != nil { - s.logger.Error("new session failed", zap.Error(err)) - c.JSON(http.StatusInternalServerError, NewSessionResponse{ - Success: false, - Error: err.Error(), - }) - return - } - - // Session ID is stored in the adapter - no need to duplicate in Manager - c.JSON(http.StatusOK, NewSessionResponse{ - Success: true, - SessionID: sessionID, - }) -} - // LoadSessionRequest is a request to load an existing ACP session type LoadSessionRequest struct { SessionID string `json:"session_id"` @@ -171,50 +60,6 @@ type LoadSessionResponse struct { Error string `json:"error,omitempty"` } -func (s *Server) handleAgentLoadSession(c *gin.Context) { - var req LoadSessionRequest - if err := c.ShouldBindJSON(&req); err != nil { - c.JSON(http.StatusBadRequest, LoadSessionResponse{ - Success: false, - Error: "invalid request: " + err.Error(), - }) - return - } - if req.SessionID == "" { - c.JSON(http.StatusBadRequest, LoadSessionResponse{ - Success: false, - Error: "session_id is required", - }) - return - } - - ctx, cancel := context.WithTimeout(c.Request.Context(), 30*time.Second) - defer cancel() - - adapter := s.procMgr.GetAdapter() - if adapter == nil { - c.JSON(http.StatusServiceUnavailable, LoadSessionResponse{ - Success: false, - Error: "agent not running", - }) - return - } - - if err := adapter.LoadSession(ctx, req.SessionID); err != nil { - s.logger.Error("load session failed", zap.Error(err)) - c.JSON(http.StatusInternalServerError, LoadSessionResponse{ - Success: false, - Error: err.Error(), - }) - return - } - - c.JSON(http.StatusOK, LoadSessionResponse{ - Success: true, - SessionID: req.SessionID, - }) -} - // PromptRequest is a request to send a prompt to the agent type PromptRequest struct { Text string `json:"text"` // Simple text prompt @@ -228,54 +73,34 @@ type PromptResponse struct { Error string `json:"error,omitempty"` } -func (s *Server) handleAgentPrompt(c *gin.Context) { - var req PromptRequest - if err := c.ShouldBindJSON(&req); err != nil { - c.JSON(http.StatusBadRequest, PromptResponse{ - Success: false, - Error: "invalid request: " + err.Error(), - }) - return - } +// PermissionRespondRequest is a request to respond to a permission request +type PermissionRespondRequest struct { + PendingID string `json:"pending_id"` + OptionID string `json:"option_id,omitempty"` + Cancelled bool `json:"cancelled,omitempty"` +} - adapter := s.procMgr.GetAdapter() - if adapter == nil { - c.JSON(http.StatusServiceUnavailable, PromptResponse{ - Success: false, - Error: "agent not running", - }) - return - } +// PermissionRespondResponse is the response to a permission respond call +type PermissionRespondResponse struct { + Success bool `json:"success"` + Error string `json:"error,omitempty"` +} - sessionID := s.procMgr.GetSessionID() - if sessionID == "" { - c.JSON(http.StatusBadRequest, PromptResponse{ - Success: false, - Error: "no active session - call new_session first", - }) - return - } +// AgentStderrResponse contains recent stderr lines from the agent process. +type AgentStderrResponse struct { + Lines []string `json:"lines"` +} - // Start prompt processing asynchronously. - // Completion is signaled via the WebSocket complete event, not the HTTP response. - go func() { - ctx, cancel := context.WithTimeout(context.Background(), constants.PromptTimeout) - defer cancel() - if err := adapter.Prompt(ctx, req.Text, req.Attachments); err != nil { - s.logger.Error("async prompt failed", zap.Error(err)) - } - }() - - s.logger.Info("prompt accepted (async)", zap.Int("attachments", len(req.Attachments))) - - // Return immediately — completion comes via WebSocket complete event - c.JSON(http.StatusAccepted, PromptResponse{Success: true}) +// CancelResponse is the response from a cancel request. +type CancelResponse struct { + Success bool `json:"success"` + Error string `json:"error,omitempty"` } // handleAgentStreamWS streams agent session notifications via WebSocket. // This is a bidirectional stream: // - agentctl -> backend: agent events, MCP requests -// - backend -> agentctl: MCP responses +// - backend -> agentctl: MCP responses, agent operation requests func (s *Server) handleAgentStreamWS(c *gin.Context) { conn, err := s.upgrader.Upgrade(c.Writer, c.Request, nil) if err != nil { @@ -308,7 +133,7 @@ func (s *Server) handleAgentStreamWS(c *gin.Context) { // WaitGroup for cleanup var wg sync.WaitGroup - // Read goroutine: reads MCP responses from the backend + // Read goroutine: reads MCP responses and agent operation requests from the backend wg.Add(1) go func() { defer wg.Done() @@ -324,13 +149,31 @@ func (s *Server) handleAgentStreamWS(c *gin.Context) { return } - // Parse as WebSocket message (MCP response) + // Parse as WebSocket message var msg ws.Message if err := json.Unmarshal(message, &msg); err != nil { s.logger.Warn("failed to parse message", zap.Error(err)) continue } + // Handle agent operation requests (type=request) + if msg.Type == ws.MessageTypeRequest { + go func(reqMsg ws.Message) { + resp := s.handleAgentStreamRequest(ctx, &reqMsg) + if resp != nil { + data, err := json.Marshal(resp) + if err != nil { + s.logger.Error("failed to marshal WS response", zap.Error(err)) + return + } + if err := writeMessage(data); err != nil { + s.logger.Debug("failed to write WS response", zap.Error(err)) + } + } + }(msg) + continue + } + // Forward MCP response to the backend client if s.mcpBackendClient != nil && (msg.Type == ws.MessageTypeResponse || msg.Type == ws.MessageTypeError) { s.mcpBackendClient.HandleResponse(&msg) @@ -387,27 +230,213 @@ func (s *Server) handleAgentStreamWS(c *gin.Context) { wg.Wait() } -// PermissionRespondRequest is a request to respond to a permission request -type PermissionRespondRequest struct { - PendingID string `json:"pending_id" binding:"required"` - OptionID string `json:"option_id,omitempty"` - Cancelled bool `json:"cancelled,omitempty"` +// handleAgentStreamRequest dispatches agent operation requests received on the WebSocket stream. +func (s *Server) handleAgentStreamRequest(ctx context.Context, msg *ws.Message) *ws.Message { + switch msg.Action { + case "agent.initialize": + return s.handleWSInitialize(ctx, msg) + case "agent.session.new": + return s.handleWSNewSession(ctx, msg) + case "agent.session.load": + return s.handleWSLoadSession(ctx, msg) + case "agent.prompt": + return s.handleWSPrompt(ctx, msg) + case "agent.cancel": + return s.handleWSCancel(ctx, msg) + case "agent.permissions.respond": + return s.handleWSPermissionRespond(ctx, msg) + case "agent.stderr": + return s.handleWSStderr(ctx, msg) + default: + resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeUnknownAction, fmt.Sprintf("unknown action: %s", msg.Action), nil) + return resp + } } -// PermissionRespondResponse is the response to a permission respond call -type PermissionRespondResponse struct { - Success bool `json:"success"` - Error string `json:"error,omitempty"` +func (s *Server) handleWSInitialize(ctx context.Context, msg *ws.Message) *ws.Message { + var req InitializeRequest + if err := msg.ParsePayload(&req); err != nil { + resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "invalid request: "+err.Error(), nil) + return resp + } + + ctx, cancel := context.WithTimeout(ctx, 60*time.Second) + defer cancel() + + adapter := s.procMgr.GetAdapter() + if adapter == nil { + resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "agent not running", nil) + return resp + } + + if err := adapter.Initialize(ctx); err != nil { + s.logger.Error("initialize failed", zap.Error(err)) + resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) + return resp + } + + // Get agent info after successful initialization + var agentInfoResp *AgentInfoResponse + if info := adapter.GetAgentInfo(); info != nil { + agentInfoResp = &AgentInfoResponse{ + Name: info.Name, + Version: info.Version, + } + } + + resp, _ := ws.NewResponse(msg.ID, msg.Action, InitializeResponse{ + Success: true, + AgentInfo: agentInfoResp, + }) + return resp } -func (s *Server) handlePermissionRespond(c *gin.Context) { +func (s *Server) handleWSNewSession(ctx context.Context, msg *ws.Message) *ws.Message { + var req NewSessionRequest + if err := msg.ParsePayload(&req); err != nil { + resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "invalid request: "+err.Error(), nil) + return resp + } + + ctx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + + adapter := s.procMgr.GetAdapter() + if adapter == nil { + resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "agent not running", nil) + return resp + } + + // If MCP server is enabled, prepend the local kandev MCP server to the list. + mcpServers := req.McpServers + if s.mcpServer != nil { + localKandevMcp := types.McpServer{ + Name: "kandev", + Type: "sse", + URL: fmt.Sprintf("http://localhost:%d/sse", s.cfg.Port), + } + filtered := make([]types.McpServer, 0, len(mcpServers)+1) + filtered = append(filtered, localKandevMcp) + for _, srv := range mcpServers { + if srv.Name != "kandev" { + filtered = append(filtered, srv) + } + } + mcpServers = filtered + s.logger.Debug("injected local kandev MCP server", + zap.String("url", localKandevMcp.URL), + zap.Int("total_servers", len(mcpServers))) + } + + sessionID, err := adapter.NewSession(ctx, mcpServers) + if err != nil { + s.logger.Error("new session failed", zap.Error(err)) + resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) + return resp + } + + resp, _ := ws.NewResponse(msg.ID, msg.Action, NewSessionResponse{ + Success: true, + SessionID: sessionID, + }) + return resp +} + +func (s *Server) handleWSLoadSession(ctx context.Context, msg *ws.Message) *ws.Message { + var req LoadSessionRequest + if err := msg.ParsePayload(&req); err != nil { + resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "invalid request: "+err.Error(), nil) + return resp + } + if req.SessionID == "" { + resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "session_id is required", nil) + return resp + } + + ctx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + + adapter := s.procMgr.GetAdapter() + if adapter == nil { + resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "agent not running", nil) + return resp + } + + if err := adapter.LoadSession(ctx, req.SessionID); err != nil { + s.logger.Error("load session failed", zap.Error(err)) + resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) + return resp + } + + resp, _ := ws.NewResponse(msg.ID, msg.Action, LoadSessionResponse{ + Success: true, + SessionID: req.SessionID, + }) + return resp +} + +func (s *Server) handleWSPrompt(ctx context.Context, msg *ws.Message) *ws.Message { + var req PromptRequest + if err := msg.ParsePayload(&req); err != nil { + resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "invalid request: "+err.Error(), nil) + return resp + } + + adapter := s.procMgr.GetAdapter() + if adapter == nil { + resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "agent not running", nil) + return resp + } + + sessionID := s.procMgr.GetSessionID() + if sessionID == "" { + resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "no active session - call new_session first", nil) + return resp + } + + // Start prompt processing asynchronously. + // Completion is signaled via the WebSocket complete event, not this response. + go func() { + promptCtx, cancel := context.WithTimeout(context.Background(), constants.PromptTimeout) + defer cancel() + if err := adapter.Prompt(promptCtx, req.Text, req.Attachments); err != nil { + s.logger.Error("async prompt failed", zap.Error(err)) + } + }() + + s.logger.Info("prompt accepted (async)", zap.Int("attachments", len(req.Attachments))) + + // Return immediately — completion comes via WebSocket complete event + resp, _ := ws.NewResponse(msg.ID, msg.Action, PromptResponse{Success: true}) + return resp +} + +func (s *Server) handleWSCancel(ctx context.Context, msg *ws.Message) *ws.Message { + ctx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + + adapter := s.procMgr.GetAdapter() + if adapter == nil { + resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "agent not running", nil) + return resp + } + + if err := adapter.Cancel(ctx); err != nil { + s.logger.Error("cancel failed", zap.Error(err)) + resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) + return resp + } + + s.logger.Info("cancel completed") + resp, _ := ws.NewResponse(msg.ID, msg.Action, CancelResponse{Success: true}) + return resp +} + +func (s *Server) handleWSPermissionRespond(_ context.Context, msg *ws.Message) *ws.Message { var req PermissionRespondRequest - if err := c.ShouldBindJSON(&req); err != nil { - c.JSON(http.StatusBadRequest, PermissionRespondResponse{ - Success: false, - Error: "invalid request: " + err.Error(), - }) - return + if err := msg.ParsePayload(&req); err != nil { + resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "invalid request: "+err.Error(), nil) + return resp } s.logger.Info("received permission response", @@ -417,56 +446,16 @@ func (s *Server) handlePermissionRespond(c *gin.Context) { if err := s.procMgr.RespondToPermission(req.PendingID, req.OptionID, req.Cancelled); err != nil { s.logger.Error("failed to respond to permission", zap.Error(err)) - c.JSON(http.StatusNotFound, PermissionRespondResponse{ - Success: false, - Error: err.Error(), - }) - return + resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeNotFound, err.Error(), nil) + return resp } - c.JSON(http.StatusOK, PermissionRespondResponse{ - Success: true, - }) -} -// AgentStderrResponse contains recent stderr lines from the agent process. -type AgentStderrResponse struct { - Lines []string `json:"lines"` + resp, _ := ws.NewResponse(msg.ID, msg.Action, PermissionRespondResponse{Success: true}) + return resp } -func (s *Server) handleAgentStderr(c *gin.Context) { +func (s *Server) handleWSStderr(_ context.Context, msg *ws.Message) *ws.Message { lines := s.procMgr.GetRecentStderr() - c.JSON(http.StatusOK, AgentStderrResponse{Lines: lines}) -} - -// CancelResponse is the response from a cancel request. -type CancelResponse struct { - Success bool `json:"success"` - Error string `json:"error,omitempty"` -} - -// handleAgentCancel interrupts the current agent turn. -func (s *Server) handleAgentCancel(c *gin.Context) { - ctx, cancel := context.WithTimeout(c.Request.Context(), 30*time.Second) - defer cancel() - - adapter := s.procMgr.GetAdapter() - if adapter == nil { - c.JSON(http.StatusServiceUnavailable, CancelResponse{ - Success: false, - Error: "agent not running", - }) - return - } - - if err := adapter.Cancel(ctx); err != nil { - s.logger.Error("cancel failed", zap.Error(err)) - c.JSON(http.StatusInternalServerError, CancelResponse{ - Success: false, - Error: err.Error(), - }) - return - } - - s.logger.Info("cancel completed") - c.JSON(http.StatusOK, CancelResponse{Success: true}) + resp, _ := ws.NewResponse(msg.ID, msg.Action, AgentStderrResponse{Lines: lines}) + return resp } diff --git a/apps/backend/internal/agentctl/server/api/agent_test.go b/apps/backend/internal/agentctl/server/api/agent_test.go new file mode 100644 index 0000000000..a160070da9 --- /dev/null +++ b/apps/backend/internal/agentctl/server/api/agent_test.go @@ -0,0 +1,492 @@ +package api + +import ( + "context" + "encoding/json" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/gorilla/websocket" + "github.com/kandev/kandev/internal/agentctl/server/config" + "github.com/kandev/kandev/internal/agentctl/server/process" + "github.com/kandev/kandev/internal/common/logger" + ws "github.com/kandev/kandev/pkg/websocket" +) + +func newTestLogger() *logger.Logger { + log, _ := logger.NewLogger(logger.LoggingConfig{ + Level: "error", + Format: "json", + }) + return log +} + +// newTestServer creates a minimal Server with a process.Manager (no adapter). +// Adapter is nil so all handlers that need it return "agent not running". +func newTestServer(t *testing.T) *Server { + t.Helper() + log := newTestLogger() + cfg := &config.InstanceConfig{ + Port: 0, + WorkDir: "/tmp/test", + } + procMgr := process.NewManager(cfg, log) + return NewServer(cfg, procMgr, nil, nil, log) +} + +// dialTestWS connects a WebSocket client to the test server's /api/v1/agent/stream endpoint. +func dialTestWS(t *testing.T, server *httptest.Server) *websocket.Conn { + t.Helper() + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/api/v1/agent/stream" + conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil) + if err != nil { + t.Fatalf("failed to dial WebSocket: %v", err) + } + return conn +} + +// sendWSRequest sends a ws.Message request and reads the response. +func sendWSRequest(t *testing.T, conn *websocket.Conn, action string, payload interface{}) *ws.Message { + t.Helper() + msg, err := ws.NewRequest("test-req-id", action, payload) + if err != nil { + t.Fatalf("failed to create request: %v", err) + } + data, err := json.Marshal(msg) + if err != nil { + t.Fatalf("failed to marshal request: %v", err) + } + if err := conn.WriteMessage(websocket.TextMessage, data); err != nil { + t.Fatalf("failed to write request: %v", err) + } + + // Read response with timeout + _ = conn.SetReadDeadline(time.Now().Add(5 * time.Second)) + _, respData, err := conn.ReadMessage() + if err != nil { + t.Fatalf("failed to read response: %v", err) + } + + var resp ws.Message + if err := json.Unmarshal(respData, &resp); err != nil { + t.Fatalf("failed to parse response: %v", err) + } + return &resp +} + +// --- Tests for handleAgentStreamRequest dispatcher --- + +func TestHandleAgentStreamRequest_UnknownAction(t *testing.T) { + s := newTestServer(t) + ctx := context.Background() + + msg, _ := ws.NewRequest("req-1", "unknown.action", nil) + resp := s.handleAgentStreamRequest(ctx, msg) + + if resp == nil { + t.Fatal("expected response") + } else if resp.Type != ws.MessageTypeError { + t.Errorf("expected error type, got %q", resp.Type) + } + + var errPayload ws.ErrorPayload + if err := resp.ParsePayload(&errPayload); err != nil { + t.Fatalf("failed to parse error payload: %v", err) + } + if errPayload.Code != ws.ErrorCodeUnknownAction { + t.Errorf("expected UNKNOWN_ACTION code, got %q", errPayload.Code) + } + if !strings.Contains(errPayload.Message, "unknown.action") { + t.Errorf("expected error message to contain action name, got %q", errPayload.Message) + } +} + +func TestHandleAgentStreamRequest_DispatchesCorrectActions(t *testing.T) { + s := newTestServer(t) + ctx := context.Background() + + // All these should dispatch to their handlers (returning "agent not running" since no adapter) + actions := []string{ + "agent.initialize", + "agent.session.new", + "agent.session.load", + "agent.prompt", + "agent.cancel", + "agent.permissions.respond", + "agent.stderr", + } + + for _, action := range actions { + t.Run(action, func(t *testing.T) { + msg, _ := ws.NewRequest("req-"+action, action, map[string]string{}) + resp := s.handleAgentStreamRequest(ctx, msg) + + if resp == nil { + t.Fatal("expected response") + } else if resp.ID != "req-"+action { + t.Errorf("expected response ID 'req-%s', got %q", action, resp.ID) + } + // stderr doesn't need an adapter, so it should succeed + if action == "agent.stderr" { + if resp.Type != ws.MessageTypeResponse { + t.Errorf("expected response type for stderr, got %q", resp.Type) + } + } + }) + } +} + +// --- Tests for individual WS handlers with no adapter (agent not running) --- + +func TestHandleWSInitialize_NoAdapter(t *testing.T) { + s := newTestServer(t) + ctx := context.Background() + + msg, _ := ws.NewRequest("req-1", "agent.initialize", map[string]string{ + "client_name": "test", + "client_version": "1.0.0", + }) + resp := s.handleWSInitialize(ctx, msg) + + if resp.Type != ws.MessageTypeError { + t.Errorf("expected error type, got %q", resp.Type) + } + var errPayload ws.ErrorPayload + if err := resp.ParsePayload(&errPayload); err != nil { + t.Fatalf("failed to parse error: %v", err) + } + if !strings.Contains(errPayload.Message, "agent not running") { + t.Errorf("expected 'agent not running', got %q", errPayload.Message) + } +} + +func TestHandleWSNewSession_NoAdapter(t *testing.T) { + s := newTestServer(t) + ctx := context.Background() + + msg, _ := ws.NewRequest("req-1", "agent.session.new", map[string]string{ + "cwd": "/workspace", + }) + resp := s.handleWSNewSession(ctx, msg) + + if resp.Type != ws.MessageTypeError { + t.Errorf("expected error type, got %q", resp.Type) + } + var errPayload ws.ErrorPayload + if err := resp.ParsePayload(&errPayload); err != nil { + t.Fatalf("failed to parse error: %v", err) + } + if !strings.Contains(errPayload.Message, "agent not running") { + t.Errorf("expected 'agent not running', got %q", errPayload.Message) + } +} + +func TestHandleWSLoadSession_NoAdapter(t *testing.T) { + s := newTestServer(t) + ctx := context.Background() + + msg, _ := ws.NewRequest("req-1", "agent.session.load", map[string]string{ + "session_id": "sess-123", + }) + resp := s.handleWSLoadSession(ctx, msg) + + if resp.Type != ws.MessageTypeError { + t.Errorf("expected error type, got %q", resp.Type) + } + var errPayload ws.ErrorPayload + if err := resp.ParsePayload(&errPayload); err != nil { + t.Fatalf("failed to parse error: %v", err) + } + if !strings.Contains(errPayload.Message, "agent not running") { + t.Errorf("expected 'agent not running', got %q", errPayload.Message) + } +} + +func TestHandleWSLoadSession_MissingSessionID(t *testing.T) { + s := newTestServer(t) + ctx := context.Background() + + msg, _ := ws.NewRequest("req-1", "agent.session.load", map[string]string{}) + resp := s.handleWSLoadSession(ctx, msg) + + if resp.Type != ws.MessageTypeError { + t.Errorf("expected error type, got %q", resp.Type) + } + var errPayload ws.ErrorPayload + if err := resp.ParsePayload(&errPayload); err != nil { + t.Fatalf("failed to parse error: %v", err) + } + if errPayload.Code != ws.ErrorCodeBadRequest { + t.Errorf("expected BAD_REQUEST code, got %q", errPayload.Code) + } + if !strings.Contains(errPayload.Message, "session_id is required") { + t.Errorf("expected 'session_id is required', got %q", errPayload.Message) + } +} + +func TestHandleWSPrompt_NoAdapter(t *testing.T) { + s := newTestServer(t) + ctx := context.Background() + + msg, _ := ws.NewRequest("req-1", "agent.prompt", map[string]string{ + "text": "hello", + }) + resp := s.handleWSPrompt(ctx, msg) + + if resp.Type != ws.MessageTypeError { + t.Errorf("expected error type, got %q", resp.Type) + } + var errPayload ws.ErrorPayload + if err := resp.ParsePayload(&errPayload); err != nil { + t.Fatalf("failed to parse error: %v", err) + } + if !strings.Contains(errPayload.Message, "agent not running") { + t.Errorf("expected 'agent not running', got %q", errPayload.Message) + } +} + +func TestHandleWSCancel_NoAdapter(t *testing.T) { + s := newTestServer(t) + ctx := context.Background() + + msg, _ := ws.NewRequest("req-1", "agent.cancel", nil) + resp := s.handleWSCancel(ctx, msg) + + if resp.Type != ws.MessageTypeError { + t.Errorf("expected error type, got %q", resp.Type) + } + var errPayload ws.ErrorPayload + if err := resp.ParsePayload(&errPayload); err != nil { + t.Fatalf("failed to parse error: %v", err) + } + if !strings.Contains(errPayload.Message, "agent not running") { + t.Errorf("expected 'agent not running', got %q", errPayload.Message) + } +} + +func TestHandleWSStderr_Empty(t *testing.T) { + s := newTestServer(t) + ctx := context.Background() + + msg, _ := ws.NewRequest("req-1", "agent.stderr", nil) + resp := s.handleWSStderr(ctx, msg) + + if resp.Type != ws.MessageTypeResponse { + t.Errorf("expected response type, got %q", resp.Type) + } + + var result AgentStderrResponse + if err := resp.ParsePayload(&result); err != nil { + t.Fatalf("failed to parse response: %v", err) + } + if len(result.Lines) != 0 { + t.Errorf("expected empty lines, got %d", len(result.Lines)) + } +} + +func TestHandleWSPermissionRespond_BadPayload(t *testing.T) { + s := newTestServer(t) + ctx := context.Background() + + // pending_id is required but since it's not gin binding, ParsePayload will succeed + // with empty values. The actual error comes from RespondToPermission not finding it. + msg, _ := ws.NewRequest("req-1", "agent.permissions.respond", map[string]string{ + "pending_id": "nonexistent", + "option_id": "allow", + }) + resp := s.handleWSPermissionRespond(ctx, msg) + + if resp.Type != ws.MessageTypeError { + t.Errorf("expected error type, got %q", resp.Type) + } + var errPayload ws.ErrorPayload + if err := resp.ParsePayload(&errPayload); err != nil { + t.Fatalf("failed to parse error: %v", err) + } + if errPayload.Code != ws.ErrorCodeNotFound { + t.Errorf("expected NOT_FOUND code, got %q", errPayload.Code) + } +} + +func TestHandleWSInitialize_BadPayload(t *testing.T) { + s := newTestServer(t) + ctx := context.Background() + + // Send invalid payload (non-JSON raw message) + msg := &ws.Message{ + ID: "req-1", + Type: ws.MessageTypeRequest, + Action: "agent.initialize", + Payload: json.RawMessage(`invalid json`), + } + resp := s.handleWSInitialize(ctx, msg) + + if resp.Type != ws.MessageTypeError { + t.Errorf("expected error type, got %q", resp.Type) + } + var errPayload ws.ErrorPayload + if err := resp.ParsePayload(&errPayload); err != nil { + t.Fatalf("failed to parse error: %v", err) + } + if errPayload.Code != ws.ErrorCodeBadRequest { + t.Errorf("expected BAD_REQUEST code, got %q", errPayload.Code) + } +} + +// --- WebSocket integration tests --- + +func TestAgentStreamWS_RequestResponseFlow(t *testing.T) { + s := newTestServer(t) + server := httptest.NewServer(s.router) + defer server.Close() + + conn := dialTestWS(t, server) + defer func() { _ = conn.Close() }() + + // Send an stderr request (doesn't need adapter) + resp := sendWSRequest(t, conn, "agent.stderr", nil) + if resp.Type != ws.MessageTypeResponse { + t.Errorf("expected response type, got %q", resp.Type) + } + if resp.ID != "test-req-id" { + t.Errorf("expected response ID 'test-req-id', got %q", resp.ID) + } +} + +func TestAgentStreamWS_UnknownActionReturnsError(t *testing.T) { + s := newTestServer(t) + server := httptest.NewServer(s.router) + defer server.Close() + + conn := dialTestWS(t, server) + defer func() { _ = conn.Close() }() + + resp := sendWSRequest(t, conn, "nonexistent.action", nil) + if resp.Type != ws.MessageTypeError { + t.Errorf("expected error type, got %q", resp.Type) + } + var errPayload ws.ErrorPayload + if err := resp.ParsePayload(&errPayload); err != nil { + t.Fatalf("failed to parse error: %v", err) + } + if errPayload.Code != ws.ErrorCodeUnknownAction { + t.Errorf("expected UNKNOWN_ACTION, got %q", errPayload.Code) + } +} + +func TestAgentStreamWS_AgentNotRunningError(t *testing.T) { + s := newTestServer(t) + server := httptest.NewServer(s.router) + defer server.Close() + + conn := dialTestWS(t, server) + defer func() { _ = conn.Close() }() + + resp := sendWSRequest(t, conn, "agent.initialize", map[string]string{ + "client_name": "test", + "client_version": "1.0.0", + }) + if resp.Type != ws.MessageTypeError { + t.Errorf("expected error type, got %q", resp.Type) + } + var errPayload ws.ErrorPayload + if err := resp.ParsePayload(&errPayload); err != nil { + t.Fatalf("failed to parse error: %v", err) + } + if !strings.Contains(errPayload.Message, "agent not running") { + t.Errorf("expected 'agent not running', got %q", errPayload.Message) + } +} + +func TestAgentStreamWS_MultipleRequests(t *testing.T) { + s := newTestServer(t) + server := httptest.NewServer(s.router) + defer server.Close() + + conn := dialTestWS(t, server) + defer func() { _ = conn.Close() }() + + // Send multiple requests sequentially + actions := []string{"agent.stderr", "agent.stderr", "agent.cancel"} + for i, action := range actions { + msg, _ := ws.NewRequest("req-"+string(rune('a'+i)), action, nil) + data, _ := json.Marshal(msg) + if err := conn.WriteMessage(websocket.TextMessage, data); err != nil { + t.Fatalf("failed to write request %d: %v", i, err) + } + } + + // Read all responses + for i := 0; i < len(actions); i++ { + _ = conn.SetReadDeadline(time.Now().Add(5 * time.Second)) + _, respData, err := conn.ReadMessage() + if err != nil { + t.Fatalf("failed to read response %d: %v", i, err) + } + var resp ws.Message + if err := json.Unmarshal(respData, &resp); err != nil { + t.Fatalf("failed to parse response %d: %v", i, err) + } + // All should have some response (either success or error) + if resp.Type != ws.MessageTypeResponse && resp.Type != ws.MessageTypeError { + t.Errorf("response %d: expected response or error type, got %q", i, resp.Type) + } + } +} + +func TestAgentStreamWS_ResponsePreservesMessageID(t *testing.T) { + s := newTestServer(t) + server := httptest.NewServer(s.router) + defer server.Close() + + conn := dialTestWS(t, server) + defer func() { _ = conn.Close() }() + + requestID := "unique-req-id-12345" + msg, _ := ws.NewRequest(requestID, "agent.stderr", nil) + data, _ := json.Marshal(msg) + if err := conn.WriteMessage(websocket.TextMessage, data); err != nil { + t.Fatalf("failed to write: %v", err) + } + + _ = conn.SetReadDeadline(time.Now().Add(5 * time.Second)) + _, respData, err := conn.ReadMessage() + if err != nil { + t.Fatalf("failed to read: %v", err) + } + + var resp ws.Message + if err := json.Unmarshal(respData, &resp); err != nil { + t.Fatalf("failed to parse: %v", err) + } + + if resp.ID != requestID { + t.Errorf("expected response ID %q, got %q", requestID, resp.ID) + } + if resp.Action != "agent.stderr" { + t.Errorf("expected action 'agent.stderr', got %q", resp.Action) + } +} + +func TestAgentStreamWS_MalformedMessage(t *testing.T) { + s := newTestServer(t) + server := httptest.NewServer(s.router) + defer server.Close() + + conn := dialTestWS(t, server) + defer func() { _ = conn.Close() }() + + // Send malformed JSON - should not crash the server + if err := conn.WriteMessage(websocket.TextMessage, []byte(`not json`)); err != nil { + t.Fatalf("failed to write: %v", err) + } + + // Server should still be alive - send a valid request + time.Sleep(50 * time.Millisecond) + resp := sendWSRequest(t, conn, "agent.stderr", nil) + if resp.Type != ws.MessageTypeResponse { + t.Errorf("expected response after malformed message, got %q", resp.Type) + } +} diff --git a/apps/backend/internal/agentctl/server/api/server.go b/apps/backend/internal/agentctl/server/api/server.go index 718abc8ba0..59aad112cd 100644 --- a/apps/backend/internal/agentctl/server/api/server.go +++ b/apps/backend/internal/agentctl/server/api/server.go @@ -74,15 +74,9 @@ func (s *Server) setupRoutes() { api.POST("/start", s.handleStart) api.POST("/stop", s.handleStop) - // Agent session/prompt methods (works with all adapters: ACP, Codex, ClaudeCode, OpenCode) - api.POST("/agent/initialize", s.handleAgentInitialize) - api.POST("/agent/session/new", s.handleAgentNewSession) - api.POST("/agent/session/load", s.handleAgentLoadSession) - api.POST("/agent/prompt", s.handleAgentPrompt) - api.POST("/agent/cancel", s.handleAgentCancel) + // Agent stream: bidirectional WebSocket for agent events, MCP, and agent operations + // (initialize, session/new, session/load, prompt, cancel, stderr, permissions/respond) api.GET("/agent/stream", s.handleAgentStreamWS) - api.GET("/agent/stderr", s.handleAgentStderr) - api.POST("/agent/permissions/respond", s.handlePermissionRespond) // Unified workspace stream (git status, files, shell) api.GET("/workspace/stream", s.handleWorkspaceStreamWS) diff --git a/apps/backend/internal/agentctl/server/process/interactive_runner_test.go b/apps/backend/internal/agentctl/server/process/interactive_runner_test.go index 7a681dab4b..19610797e9 100644 --- a/apps/backend/internal/agentctl/server/process/interactive_runner_test.go +++ b/apps/backend/internal/agentctl/server/process/interactive_runner_test.go @@ -779,15 +779,16 @@ func TestInteractiveRunner_RegisterScriptShell(t *testing.T) { } if scriptShell == nil { t.Fatal("script shell not found in list") - } - if scriptShell.Label != "npm start" { - t.Errorf("script shell Label = %q, want 'npm start'", scriptShell.Label) - } - if scriptShell.InitialCommand != "npm run start" { - t.Errorf("script shell InitialCommand = %q, want 'npm run start'", scriptShell.InitialCommand) - } - if !scriptShell.Closable { - t.Error("script shell should be closable") + } else { + if scriptShell.Label != "npm start" { + t.Errorf("script shell Label = %q, want 'npm start'", scriptShell.Label) + } + if scriptShell.InitialCommand != "npm run start" { + t.Errorf("script shell InitialCommand = %q, want 'npm run start'", scriptShell.InitialCommand) + } + if !scriptShell.Closable { + t.Error("script shell should be closable") + } } if scriptShell.ProcessID != "" { t.Errorf("script shell should have empty ProcessID before WebSocket connect, got %q", scriptShell.ProcessID) diff --git a/apps/backend/internal/orchestrator/queue/queue_test.go b/apps/backend/internal/orchestrator/queue/queue_test.go index 882488aa5a..1f5336e015 100644 --- a/apps/backend/internal/orchestrator/queue/queue_test.go +++ b/apps/backend/internal/orchestrator/queue/queue_test.go @@ -79,8 +79,7 @@ func TestDequeue(t *testing.T) { if dequeued == nil { t.Fatal("Dequeue returned nil") - } - if dequeued.TaskID != task.ID { + } else if dequeued.TaskID != task.ID { t.Errorf("expected TaskID = %s, got %s", task.ID, dequeued.TaskID) } if q.Len() != 0 { diff --git a/apps/backend/internal/orchestrator/watcher/watcher_test.go b/apps/backend/internal/orchestrator/watcher/watcher_test.go index 886b53ffb3..4898e7be50 100644 --- a/apps/backend/internal/orchestrator/watcher/watcher_test.go +++ b/apps/backend/internal/orchestrator/watcher/watcher_test.go @@ -138,8 +138,7 @@ func TestNewWatcher(t *testing.T) { w := NewWatcher(eventBus, handlers, log) if w == nil { t.Fatal("NewWatcher returned nil") - } - if w.running { + } else if w.running { t.Error("watcher should not be running initially") } } diff --git a/apps/backend/internal/worktree/recreator_test.go b/apps/backend/internal/worktree/recreator_test.go index cb4d4542b1..933348f5f9 100644 --- a/apps/backend/internal/worktree/recreator_test.go +++ b/apps/backend/internal/worktree/recreator_test.go @@ -18,8 +18,7 @@ func TestNewRecreator(t *testing.T) { recreator := NewRecreator(mgr) if recreator == nil { t.Fatal("expected non-nil recreator") - } - if recreator.manager != mgr { + } else if recreator.manager != mgr { t.Error("recreator manager not set correctly") } } diff --git a/apps/backend/pkg/claudecode/types_test.go b/apps/backend/pkg/claudecode/types_test.go index e3cd108692..0a57418e45 100644 --- a/apps/backend/pkg/claudecode/types_test.go +++ b/apps/backend/pkg/claudecode/types_test.go @@ -43,13 +43,10 @@ func TestCLIMessage_GetResultData(t *testing.T) { if got != nil { t.Errorf("GetResultData() = %v, want nil", got) } - } else { - if got == nil { - t.Fatalf("GetResultData() = nil, want non-nil") - } - if got.Text != tt.wantText { - t.Errorf("GetResultData().Text = %q, want %q", got.Text, tt.wantText) - } + } else if got == nil { + t.Fatalf("GetResultData() = nil, want non-nil") + } else if got.Text != tt.wantText { + t.Errorf("GetResultData().Text = %q, want %q", got.Text, tt.wantText) } }) }