From 4e1951aa412f75d3a6d51ef405b50e4904df573e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Mon, 9 Feb 2026 21:58:15 +0000 Subject: [PATCH] feat: remove blocking http call when creating the agent (#214) --- .../internal/agent/lifecycle/manager.go | 28 +++- .../internal/agent/lifecycle/manager_test.go | 124 +++++++++++++++++- .../internal/agent/lifecycle/runtime.go | 1 + .../internal/agent/lifecycle/session.go | 102 ++++++++------ .../internal/agent/lifecycle/streams.go | 14 +- .../backend/internal/agent/lifecycle/types.go | 15 +++ .../backend/internal/agentctl/client/agent.go | 58 ++++---- .../agentctl/client/launcher/launcher.go | 9 +- .../internal/agentctl/server/api/agent.go | 29 ++-- 9 files changed, 287 insertions(+), 93 deletions(-) diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index 65c87da32d..5cf058763f 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/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, diff --git a/apps/backend/internal/agent/lifecycle/manager_test.go b/apps/backend/internal/agent/lifecycle/manager_test.go index d7ca7c8678..979bc78884 100644 --- a/apps/backend/internal/agent/lifecycle/manager_test.go +++ b/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") + } +} diff --git a/apps/backend/internal/agent/lifecycle/runtime.go b/apps/backend/internal/agent/lifecycle/runtime.go index be93c84923..5e198db339 100644 --- a/apps/backend/internal/agent/lifecycle/runtime.go +++ b/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), } } diff --git a/apps/backend/internal/agent/lifecycle/session.go b/apps/backend/internal/agent/lifecycle/session.go index 353c452324..eca17d0a7a 100644 --- a/apps/backend/internal/agent/lifecycle/session.go +++ b/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 } diff --git a/apps/backend/internal/agent/lifecycle/streams.go b/apps/backend/internal/agent/lifecycle/streams.go index 0e834091a0..e5b3e707c0 100644 --- a/apps/backend/internal/agent/lifecycle/streams.go +++ b/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 diff --git a/apps/backend/internal/agent/lifecycle/types.go b/apps/backend/internal/agent/lifecycle/types.go index c92a6ed21a..24adcedfce 100644 --- a/apps/backend/internal/agent/lifecycle/types.go +++ b/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 diff --git a/apps/backend/internal/agentctl/client/agent.go b/apps/backend/internal/agentctl/client/agent.go index 9102a0ddb2..18cb9714cb 100644 --- a/apps/backend/internal/agentctl/client/agent.go +++ b/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 } diff --git a/apps/backend/internal/agentctl/client/launcher/launcher.go b/apps/backend/internal/agentctl/client/launcher/launcher.go index 53d7aafc6e..e378e607cf 100644 --- a/apps/backend/internal/agentctl/client/launcher/launcher.go +++ b/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())) } diff --git a/apps/backend/internal/agentctl/server/api/agent.go b/apps/backend/internal/agentctl/server/api/agent.go index 7d5975b1a1..875214165d 100644 --- a/apps/backend/internal/agentctl/server/api/agent.go +++ b/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.