feat: remove blocking http call when creating the agent (#214)

apps/backend/internal/agent/lifecycle/manager.go

@@ -541,6 +541,7 @@ func (m *Manager) Start(ctx context.Context) error {
 				agentctl:             ri.Client,
 				standaloneInstanceID: ri.StandaloneInstanceID,
 				standalonePort:       ri.StandalonePort,
+				promptDoneCh:         make(chan PromptCompletionSignal, 1),
 			}
 			m.executionStore.Add(execution)
 
@@ -1329,6 +1330,11 @@ func (m *Manager) initializeACPSession(ctx context.Context, execution *AgentExec
 
 // handleAgentEvent processes incoming agent events from the agent
 func (m *Manager) handleAgentEvent(execution *AgentExecution, event agentctl.AgentEvent) {
+	// Update last activity timestamp for stall detection
+	execution.lastActivityAtMu.Lock()
+	execution.lastActivityAt = time.Now()
+	execution.lastActivityAtMu.Unlock()
+
 	// Log all incoming events for debugging
 	m.logger.Debug("handleAgentEvent entry",
 		zap.String("execution_id", execution.ID),
@@ -1538,6 +1544,26 @@ func (m *Manager) handleAgentEvent(execution *AgentExecution, event agentctl.Age
 			}
 		}
 
+		// Signal the completion channel so SendPrompt unblocks
+		stopReason := "end_turn"
+		errorMsg := ""
+		if isError {
+			stopReason = "error"
+			errorMsg = "agent error completion"
+			if event.Error != "" {
+				errorMsg = event.Error
+			}
+		}
+		select {
+		case execution.promptDoneCh <- PromptCompletionSignal{
+			StopReason: stopReason,
+			IsError:    isError,
+			Error:      errorMsg,
+		}:
+		default:
+			// Channel full or no one waiting — that's fine (e.g., initial prompt in goroutine)
+		}
+
 	case "permission_request":
 		m.logger.Debug("permission request received",
 			zap.String("execution_id", execution.ID),
@@ -1842,7 +1868,7 @@ func (m *Manager) PromptAgent(ctx context.Context, executionID string, prompt st
 	if !exists {
 		return nil, fmt.Errorf("execution %q not found: %w", executionID, ErrExecutionNotFound)
 	}
-	return m.sessionManager.SendPrompt(ctx, execution, prompt, true, attachments, m.MarkReady)
+	return m.sessionManager.SendPrompt(ctx, execution, prompt, true, attachments)
 }
 
 // CancelAgent interrupts the current agent turn without terminating the process,

apps/backend/internal/agent/lifecycle/manager_test.go

@@ -879,11 +879,12 @@ func createTestManagerWithTracking() (*Manager, *MockEventBusWithTracking) {
 // createTestExecution creates a test execution with proper initialization
 func createTestExecution(id, taskID, sessionID string) *AgentExecution {
 	return &AgentExecution{
-		ID:        id,
-		TaskID:    taskID,
-		SessionID: sessionID,
-		Status:    v1.AgentStatusRunning,
-		StartedAt: time.Now(),
+		ID:           id,
+		TaskID:       taskID,
+		SessionID:    sessionID,
+		Status:       v1.AgentStatusRunning,
+		StartedAt:    time.Now(),
+		promptDoneCh: make(chan PromptCompletionSignal, 1),
 	}
 }
 
@@ -1250,3 +1251,116 @@ func TestHandleAgentEvent_MultipleToolCalls(t *testing.T) {
 		}
 	}
 }
+
+// TestHandleAgentEvent_CompleteSignalsPromptDoneCh verifies that the complete event
+// signals the promptDoneCh channel with the correct stop reason.
+func TestHandleAgentEvent_CompleteSignalsPromptDoneCh(t *testing.T) {
+	mgr, _ := createTestManagerWithTracking()
+	execution := createTestExecution("exec-1", "task-1", "session-1")
+	mgr.executionStore.Add(execution)
+
+	// Send a normal complete event
+	mgr.handleAgentEvent(execution, agentctl.AgentEvent{
+		Type: "complete",
+	})
+
+	// promptDoneCh should have a signal
+	select {
+	case signal := <-execution.promptDoneCh:
+		if signal.IsError {
+			t.Error("expected non-error signal for normal complete")
+		}
+		if signal.StopReason != "end_turn" {
+			t.Errorf("expected stop_reason 'end_turn', got %q", signal.StopReason)
+		}
+	default:
+		t.Error("expected signal on promptDoneCh, got none")
+	}
+}
+
+// TestHandleAgentEvent_ErrorCompleteSignalsPromptDoneCh verifies that an error completion
+// signals the promptDoneCh channel with IsError=true and the error message.
+func TestHandleAgentEvent_ErrorCompleteSignalsPromptDoneCh(t *testing.T) {
+	mgr, _ := createTestManagerWithTracking()
+	execution := createTestExecution("exec-1", "task-1", "session-1")
+	mgr.executionStore.Add(execution)
+
+	// Send an error complete event
+	mgr.handleAgentEvent(execution, agentctl.AgentEvent{
+		Type:  "complete",
+		Error: "something went wrong",
+		Data:  map[string]interface{}{"is_error": true},
+	})
+
+	// promptDoneCh should have an error signal
+	select {
+	case signal := <-execution.promptDoneCh:
+		if !signal.IsError {
+			t.Error("expected error signal for error complete")
+		}
+		if signal.StopReason != "error" {
+			t.Errorf("expected stop_reason 'error', got %q", signal.StopReason)
+		}
+		if signal.Error != "something went wrong" {
+			t.Errorf("expected error 'something went wrong', got %q", signal.Error)
+		}
+	default:
+		t.Error("expected signal on promptDoneCh, got none")
+	}
+}
+
+// TestHandleAgentEvent_UpdatesLastActivityAt verifies that every agent event
+// updates the lastActivityAt timestamp for stall detection.
+func TestHandleAgentEvent_UpdatesLastActivityAt(t *testing.T) {
+	mgr, _ := createTestManagerWithTracking()
+	execution := createTestExecution("exec-1", "task-1", "session-1")
+	mgr.executionStore.Add(execution)
+
+	// Set lastActivityAt to a known old time
+	oldTime := time.Now().Add(-10 * time.Minute)
+	execution.lastActivityAtMu.Lock()
+	execution.lastActivityAt = oldTime
+	execution.lastActivityAtMu.Unlock()
+
+	// Send any event
+	mgr.handleAgentEvent(execution, agentctl.AgentEvent{
+		Type: "message_chunk",
+		Text: "hello\n",
+	})
+
+	// lastActivityAt should be updated to approximately now
+	execution.lastActivityAtMu.Lock()
+	elapsed := time.Since(execution.lastActivityAt)
+	execution.lastActivityAtMu.Unlock()
+
+	if elapsed > 1*time.Second {
+		t.Errorf("lastActivityAt not updated: elapsed %v since last event", elapsed)
+	}
+}
+
+// TestHandleAgentEvent_PromptDoneChDoesNotBlockWhenFull verifies that signaling
+// promptDoneCh with a full channel (no receiver) doesn't block the event handler.
+func TestHandleAgentEvent_PromptDoneChDoesNotBlockWhenFull(t *testing.T) {
+	mgr, _ := createTestManagerWithTracking()
+	execution := createTestExecution("exec-1", "task-1", "session-1")
+	mgr.executionStore.Add(execution)
+
+	// Pre-fill the channel
+	execution.promptDoneCh <- PromptCompletionSignal{StopReason: "stale"}
+
+	// Send complete event — should not block
+	done := make(chan struct{})
+	go func() {
+		mgr.handleAgentEvent(execution, agentctl.AgentEvent{
+			Type: "complete",
+		})
+		close(done)
+	}()
+
+	select {
+	case <-done:
+		// Good — didn't block
+	case <-time.After(2 * time.Second):
+		t.Fatal("handleAgentEvent blocked when promptDoneCh was full")
+	}
+}

apps/backend/internal/agent/lifecycle/runtime.go

@@ -125,5 +125,6 @@ func (ri *RuntimeInstance) ToAgentExecution(req *RuntimeCreateRequest) *AgentExe
 		agentctl:             ri.Client,
 		standaloneInstanceID: ri.StandaloneInstanceID,
 		standalonePort:       ri.StandalonePort,
+		promptDoneCh:         make(chan PromptCompletionSignal, 1),
 	}
 }

apps/backend/internal/agent/lifecycle/session.go

@@ -335,7 +335,7 @@ func (sm *SessionManager) InitializeAndPrompt(
 			defer cancel()
 
 			// Initial prompt has no attachments
-			_, err := sm.SendPrompt(promptCtx, execution, effectivePrompt, false, nil, markReady)
+			_, err := sm.SendPrompt(promptCtx, execution, effectivePrompt, false, nil)
 			if err != nil {
 				sm.logger.Error("initial prompt failed",
 					zap.String("execution_id", execution.ID),
@@ -372,13 +372,13 @@ func (sm *SessionManager) InitializeAndPrompt(
 // For initial prompts, pass validateStatus=false. For follow-up prompts, pass validateStatus=true.
 // Attachments (images) are passed to the agent if provided.
 // Returns the prompt result containing the stop reason and agent message.
+// Note: MarkReady is handled by handleAgentEvent(complete), not by this method.
 func (sm *SessionManager) SendPrompt(
 	ctx context.Context,
 	execution *AgentExecution,
 	prompt string,
 	validateStatus bool,
 	attachments []v1.MessageAttachment,
-	markReady func(executionID string) error,
 ) (*PromptResult, error) {
 	if execution.agentctl == nil {
 		return nil, fmt.Errorf("execution %q has no agentctl client", execution.ID)
@@ -449,45 +449,71 @@ func (sm *SessionManager) SendPrompt(
 		}
 	}
 
-	// Prompt is synchronous - blocks until agent completes
-	resp, err := execution.agentctl.Prompt(ctx, effectivePrompt, attachments)
-	if err != nil {
-		sm.logger.Error("ACP prompt failed",
+	// Initialize activity timestamp for stall detection
+	execution.lastActivityAtMu.Lock()
+	execution.lastActivityAt = time.Now()
+	execution.lastActivityAtMu.Unlock()
+
+	// Drain any stale signal from a previous prompt cycle
+	select {
+	case <-execution.promptDoneCh:
+	default:
+	}
+
+	// Fire the prompt (returns immediately now — completion comes via WebSocket complete event)
+	if err := execution.agentctl.Prompt(ctx, effectivePrompt, attachments); err != nil {
+		sm.logger.Error("failed to trigger prompt",
 			zap.String("execution_id", execution.ID),
 			zap.Error(err))
-		return nil, fmt.Errorf("prompt failed: %w", err)
+		return nil, fmt.Errorf("failed to trigger prompt: %w", err)
 	}
 
-	// Extract accumulated content from buffers for the return value.
-	// All adapters now emit complete events via the stream, so handleAgentEvent will
-	// handle the buffer content when the complete event arrives. We just peek at the
-	// buffer here to return the message content to the caller.
-	stopReason := ""
-	if resp != nil {
-		stopReason = string(resp.StopReason)
+	// Wait for completion signal from handleAgentEvent(complete) or stream disconnect.
+	// Periodically check for stall conditions (no events for an extended period).
+	stallTicker := time.NewTicker(30 * time.Second)
+	defer stallTicker.Stop()
+
+	for {
+		select {
+		case signal := <-execution.promptDoneCh:
+			if signal.IsError {
+				sm.logger.Error("prompt completed with error",
+					zap.String("execution_id", execution.ID),
+					zap.String("error", signal.Error))
+				return nil, fmt.Errorf("agent error: %s", signal.Error)
+			}
+
+			// Peek at buffer for return value
+			execution.messageMu.Lock()
+			agentMessage := execution.messageBuffer.String()
+			execution.messageMu.Unlock()
+
+			sm.logger.Info("prompt completed",
+				zap.String("execution_id", execution.ID),
+				zap.String("stop_reason", signal.StopReason),
+				zap.Int("message_length", len(agentMessage)))
+
+			// Note: markReady is NOT called here — handleAgentEvent(complete) already handles it.
+			return &PromptResult{
+				StopReason:   signal.StopReason,
+				AgentMessage: agentMessage,
+			}, nil
+
+		case <-ctx.Done():
+			return nil, ctx.Err()
+
+		case <-stallTicker.C:
+			execution.lastActivityAtMu.Lock()
+			elapsed := time.Since(execution.lastActivityAt)
+			lastActivity := execution.lastActivityAt
+			execution.lastActivityAtMu.Unlock()
+
+			if elapsed > 5*time.Minute {
+				sm.logger.Warn("agent stall detected: no events received",
+					zap.String("execution_id", execution.ID),
+					zap.Duration("elapsed_since_last_event", elapsed),
+					zap.Time("last_activity", lastActivity))
+			}
+		}
 	}
-
-	// Peek at the message for the return value, but don't clear the buffer.
-	// The complete event handler (handleAgentEvent) will clear the buffer.
-	execution.messageMu.Lock()
-	bufLen := execution.messageBuffer.Len()
-	agentMessage := execution.messageBuffer.String()
-	execution.messageMu.Unlock()
-
-	sm.logger.Info("prompt completed",
-		zap.String("execution_id", execution.ID),
-		zap.String("stop_reason", stopReason),
-		zap.Int("buffer_length", bufLen))
-
-	// Mark as READY for next prompt
-	if err := markReady(execution.ID); err != nil {
-		sm.logger.Error("failed to mark execution as ready after prompt",
-			zap.String("execution_id", execution.ID),
-			zap.Error(err))
-	}
-
-	return &PromptResult{
-		StopReason:   stopReason,
-		AgentMessage: agentMessage,
-	}, nil
 }

apps/backend/internal/agent/lifecycle/streams.go

@@ -96,7 +96,19 @@ func (sm *StreamManager) connectUpdatesStream(execution *AgentExecution, ready c
 		if sm.callbacks.OnAgentEvent != nil {
 			sm.callbacks.OnAgentEvent(execution, event)
 		}
-	}, sm.mcpHandler)
+	}, sm.mcpHandler, func(disconnectErr error) {
+		// WebSocket dropped — signal promptDoneCh so SendPrompt doesn't hang forever.
+		// Only signal on unexpected errors (not normal close).
+		if disconnectErr != nil {
+			select {
+			case execution.promptDoneCh <- PromptCompletionSignal{
+				IsError: true,
+				Error:   "agent stream disconnected: " + disconnectErr.Error(),
+			}:
+			default:
+			}
+		}
+	})
 
 	// Signal that the stream connection attempt is complete (success or failure)
 	// StreamUpdates returns immediately after establishing the WebSocket connection

apps/backend/internal/agent/lifecycle/types.go

@@ -70,6 +70,21 @@ type AgentExecution struct {
 	// Available commands from the agent (for slash command menu)
 	availableCommands   []streams.AvailableCommand
 	availableCommandsMu sync.RWMutex
+
+	// Channel signaled by handleAgentEvent(complete) or stream disconnect to unblock SendPrompt.
+	// Buffered (size 1) so the sender never blocks.
+	promptDoneCh chan PromptCompletionSignal
+
+	// Last time an agent event was received (for stall detection)
+	lastActivityAt   time.Time
+	lastActivityAtMu sync.Mutex
+}
+
+// PromptCompletionSignal carries the result from a complete event or disconnect.
+type PromptCompletionSignal struct {
+	StopReason string
+	IsError    bool
+	Error      string
 }
 
 // GetAgentCtlClient returns the agentctl client for this execution

apps/backend/internal/agentctl/client/agent.go

@@ -8,11 +8,9 @@ import (
 	"net/http"
 	"sync"
 
-	"github.com/coder/acp-go-sdk"
 	"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/constants"
 	v1 "github.com/kandev/kandev/pkg/api/v1"
 	ws "github.com/kandev/kandev/pkg/websocket"
 	"go.uber.org/zap"
@@ -193,17 +191,10 @@ func (c *Client) LoadSession(ctx context.Context, sessionID string) error {
 	return nil
 }
 
-// PromptResponse contains the response from a prompt call
-type PromptResponse struct {
-	Success    bool           `json:"success"`
-	StopReason acp.StopReason `json:"stop_reason,omitempty"`
-	Error      string         `json:"error,omitempty"`
-}
-
-// Prompt sends a prompt to the agent and returns when the agent completes
-// The StopReason indicates why the agent stopped (e.g., "end_turn", "needs_input")
-// Attachments (images) are passed to the agent if provided
-func (c *Client) Prompt(ctx context.Context, text string, attachments []v1.MessageAttachment) (*PromptResponse, error) {
+// Prompt sends a fire-and-forget prompt to the agent.
+// The agentctl server returns 202 immediately; completion is signaled via the WebSocket complete event.
+// Attachments (images) are passed to the agent if provided.
+func (c *Client) Prompt(ctx context.Context, text string, attachments []v1.MessageAttachment) error {
 	reqBody := struct {
 		Text        string                 `json:"text"`
 		Attachments []v1.MessageAttachment `json:"attachments,omitempty"`
@@ -211,23 +202,18 @@ func (c *Client) Prompt(ctx context.Context, text string, attachments []v1.Messa
 
 	body, err := json.Marshal(reqBody)
 	if err != nil {
-		return nil, err
-	}
-
-	// Use a longer timeout for prompt - agent may take time
-	client := &http.Client{
-		Timeout: constants.PromptTimeout,
+		return err
 	}
 
 	req, err := http.NewRequestWithContext(ctx, "POST", c.baseURL+"/api/v1/agent/prompt", bytes.NewReader(body))
 	if err != nil {
-		return nil, err
+		return err
 	}
 	req.Header.Set("Content-Type", "application/json")
 
-	resp, err := client.Do(req)
+	resp, err := c.httpClient.Do(req)
 	if err != nil {
-		return nil, err
+		return err
 	}
 	defer func() {
 		if err := resp.Body.Close(); err != nil {
@@ -237,25 +223,29 @@ func (c *Client) Prompt(ctx context.Context, text string, attachments []v1.Messa
 
 	respBody, err := readResponseBody(resp)
 	if err != nil {
-		return nil, fmt.Errorf("failed to read response body: %w", err)
+		return fmt.Errorf("failed to read response body: %w", err)
 	}
 
-	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
-		return nil, fmt.Errorf("prompt request failed with status %d: %s", resp.StatusCode, string(respBody))
+	// 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))
 	}
 
-	var result PromptResponse
+	var result struct {
+		Success bool   `json:"success"`
+		Error   string `json:"error,omitempty"`
+	}
 	if err := json.Unmarshal(respBody, &result); err != nil {
-		return nil, fmt.Errorf("failed to parse prompt response (status %d, body: %s): %w", resp.StatusCode, truncateBody(respBody), err)
+		return fmt.Errorf("failed to parse prompt response (status %d, body: %s): %w", resp.StatusCode, truncateBody(respBody), 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)))
-		return nil, fmt.Errorf("prompt failed: %s", result.Error)
+		return fmt.Errorf("prompt failed: %s", result.Error)
 	}
-	return &result, nil
+	return nil
 }
 
 // Note: AgentEvent, PermissionNotification, PermissionOption, and
@@ -270,7 +260,8 @@ type MCPHandler interface {
 // StreamUpdates opens a WebSocket connection for streaming agent events.
 // Events include message chunks, reasoning, tool calls, plan updates, and completion/error.
 // If mcpHandler is provided, MCP requests from agentctl will be dispatched to it and responses sent back.
-func (c *Client) StreamUpdates(ctx context.Context, handler func(AgentEvent), mcpHandler MCPHandler) error {
+// If onDisconnect is provided, it is called when the WebSocket read goroutine exits (e.g., on error or close).
+func (c *Client) StreamUpdates(ctx context.Context, handler func(AgentEvent), mcpHandler MCPHandler, onDisconnect func(err error)) error {
 	wsURL := "ws" + c.baseURL[4:] + "/api/v1/agent/stream"
 
 	conn, _, err := websocket.DefaultDialer.DialContext(ctx, wsURL, nil)
@@ -294,6 +285,7 @@ func (c *Client) StreamUpdates(ctx context.Context, handler func(AgentEvent), mc
 
 	// Read messages in a goroutine
 	go func() {
+		var lastErr error
 		defer func() {
 			c.mu.Lock()
 			c.agentStreamConn = nil
@@ -301,6 +293,10 @@ func (c *Client) StreamUpdates(ctx context.Context, handler func(AgentEvent), mc
 			if err := conn.Close(); err != nil {
 				c.logger.Debug("failed to close updates websocket", zap.Error(err))
 			}
+			// Signal disconnect to caller
+			if onDisconnect != nil {
+				onDisconnect(lastErr)
+			}
 		}()
 
 		for {
@@ -308,8 +304,10 @@ func (c *Client) StreamUpdates(ctx context.Context, handler func(AgentEvent), mc
 			if err != nil {
 				if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway) {
 					c.logger.Info("updates stream closed normally")
+					// Normal close — don't report as disconnect error
 				} else {
 					c.logger.Debug("updates stream error", zap.Error(err))
+					lastErr = err
 				}
 				return
 			}

apps/backend/internal/agentctl/client/launcher/launcher.go

@@ -321,10 +321,15 @@ func (l *Launcher) waitForHealthy(ctx context.Context) error {
 }
 
 // pipeOutput reads from a scanner and logs each line.
+// stderr is logged at WARN level for visibility; stdout remains at DEBUG.
 func (l *Launcher) pipeOutput(name string, scanner *bufio.Scanner) {
 	for scanner.Scan() {
 		line := scanner.Text()
-		l.logger.Debug(line, zap.String("stream", name))
+		if name == "stderr" {
+			l.logger.Warn(line, zap.String("stream", name))
+		} else {
+			l.logger.Debug(line, zap.String("stream", name))
+		}
 	}
 }
 
@@ -339,9 +344,11 @@ func (l *Launcher) monitorExit() {
 	if err != nil && !stopping {
 		l.logger.Error("agentctl exited unexpectedly",
 			zap.Error(err),
+			zap.Int("pid", l.cmd.Process.Pid),
 			zap.Int("exit_code", l.cmd.ProcessState.ExitCode()))
 	} else if !stopping {
 		l.logger.Info("agentctl exited",
+			zap.Int("pid", l.cmd.Process.Pid),
 			zap.Int("exit_code", l.cmd.ProcessState.ExitCode()))
 	}
 

apps/backend/internal/agentctl/server/api/agent.go

@@ -256,25 +256,20 @@ func (s *Server) handleAgentPrompt(c *gin.Context) {
 		return
 	}
 
-	// Use a long timeout for prompt - agent may take time to complete
-	ctx, cancel := context.WithTimeout(c.Request.Context(), constants.PromptTimeout)
-	defer cancel()
+	// 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))
+		}
+	}()
 
-	err := adapter.Prompt(ctx, req.Text, req.Attachments)
-	if err != nil {
-		s.logger.Error("prompt failed", zap.Error(err))
-		c.JSON(http.StatusInternalServerError, PromptResponse{
-			Success: false,
-			Error:   err.Error(),
-		})
-		return
-	}
+	s.logger.Info("prompt accepted (async)", zap.Int("attachments", len(req.Attachments)))
 
-	s.logger.Info("prompt completed", zap.Int("attachments", len(req.Attachments)))
-
-	c.JSON(http.StatusOK, PromptResponse{
-		Success: true,
-	})
+	// Return immediately — completion comes via WebSocket complete event
+	c.JSON(http.StatusAccepted, PromptResponse{Success: true})
 }
 
 // handleAgentStreamWS streams agent session notifications via WebSocket.