From 18f4b4e6fbf79d85018d6f12f87e5f1f44b6e441 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Salavisa?= Date: Sun, 18 Jan 2026 15:58:56 +0000 Subject: [PATCH] Cleanup deprecated code and fix session recovery (#37) --- apps/backend/cmd/agentctl/main.go | 7 +- .../internal/agent/lifecycle/command.go | 3 +- .../internal/agent/lifecycle/event_types.go | 23 ++-- .../internal/agent/lifecycle/events.go | 23 ++-- .../internal/agent/lifecycle/manager.go | 26 ++++ .../internal/agent/lifecycle/session.go | 19 +-- .../internal/agent/registry/defaults.go | 6 + .../internal/agent/registry/registry.go | 15 +++ .../agentctl/server/api/control_server.go | 5 +- .../internal/agentctl/server/config/config.go | 6 +- .../agentctl/server/instance/manager.go | 5 - .../internal/agentctl/types/streams/agent.go | 15 +++ apps/backend/internal/orchestrator/service.go | 116 +++++++++++++++--- .../task/handlers/message_handlers.go | 14 ++- .../task/chat/messages/status-message.tsx | 26 +++- 15 files changed, 219 insertions(+), 90 deletions(-) diff --git a/apps/backend/cmd/agentctl/main.go b/apps/backend/cmd/agentctl/main.go index 5c938a8520..49be04afbd 100644 --- a/apps/backend/cmd/agentctl/main.go +++ b/apps/backend/cmd/agentctl/main.go @@ -32,7 +32,6 @@ var ( agentCommandFlag = flag.String("agent-command", "", "Command to run the agent") workDirFlag = flag.String("workdir", "", "Working directory for the agent") portFlag = flag.Int("port", 0, "HTTP server port") - maxInstancesFlag = flag.Int("max-instances", 0, "Maximum number of concurrent instances") ) func main() { @@ -54,9 +53,6 @@ func main() { if *portFlag != 0 { cfg.Port = *portFlag } - if *maxInstancesFlag != 0 { - cfg.MaxInstances = *maxInstancesFlag - } // Initialize logger log, err := logger.NewLogger(logger.LoggingConfig{ @@ -73,8 +69,7 @@ func main() { }() log.Info("starting agentctl", - zap.Int("port", cfg.Port), - zap.Int("max_instances", cfg.MaxInstances)) + zap.Int("port", cfg.Port)) run(cfg, log) } diff --git a/apps/backend/internal/agent/lifecycle/command.go b/apps/backend/internal/agent/lifecycle/command.go index 7e3bfc8e53..39f23f1187 100644 --- a/apps/backend/internal/agent/lifecycle/command.go +++ b/apps/backend/internal/agent/lifecycle/command.go @@ -40,7 +40,8 @@ func (cb *CommandBuilder) BuildCommand(agentConfig *registry.AgentTypeConfig, op // 1. Session ID is provided // 2. Agent does NOT use ACP for session resumption (uses CLI flag instead) // 3. Agent has a ResumeFlag configured - if opts.SessionID != "" && !agentConfig.SessionConfig.ResumeViaACP && agentConfig.SessionConfig.ResumeFlag != "" { + // 4. Agent supports session recovery (CanRecover is not false) + if opts.SessionID != "" && !agentConfig.SessionConfig.ResumeViaACP && agentConfig.SessionConfig.ResumeFlag != "" && agentConfig.SessionConfig.SupportsRecovery() { cmd = append(cmd, agentConfig.SessionConfig.ResumeFlag, opts.SessionID) } diff --git a/apps/backend/internal/agent/lifecycle/event_types.go b/apps/backend/internal/agent/lifecycle/event_types.go index 98ff0cc59c..df60558234 100644 --- a/apps/backend/internal/agent/lifecycle/event_types.go +++ b/apps/backend/internal/agent/lifecycle/event_types.go @@ -34,17 +34,18 @@ type ACPSessionCreatedPayload struct { // AgentStreamEventData contains the nested event data within AgentStreamEventPayload. type AgentStreamEventData struct { - Type string `json:"type"` - ACPSessionID string `json:"acp_session_id,omitempty"` - Text string `json:"text,omitempty"` - ToolCallID string `json:"tool_call_id,omitempty"` - ToolName string `json:"tool_name,omitempty"` - ToolTitle string `json:"tool_title,omitempty"` - ToolStatus string `json:"tool_status,omitempty"` - ToolArgs map[string]interface{} `json:"tool_args,omitempty"` - ToolResult interface{} `json:"tool_result,omitempty"` - Error string `json:"error,omitempty"` - Data interface{} `json:"data,omitempty"` + Type string `json:"type"` + ACPSessionID string `json:"acp_session_id,omitempty"` + Text string `json:"text,omitempty"` + ToolCallID string `json:"tool_call_id,omitempty"` + ToolName string `json:"tool_name,omitempty"` + ToolTitle string `json:"tool_title,omitempty"` + ToolStatus string `json:"tool_status,omitempty"` + ToolArgs map[string]interface{} `json:"tool_args,omitempty"` + ToolResult interface{} `json:"tool_result,omitempty"` + Error string `json:"error,omitempty"` + SessionStatus string `json:"session_status,omitempty"` // "resumed" or "new" for session_status events + Data interface{} `json:"data,omitempty"` } // AgentStreamEventPayload is the payload for agent stream events (WebSocket streaming). diff --git a/apps/backend/internal/agent/lifecycle/events.go b/apps/backend/internal/agent/lifecycle/events.go index 35203cfc13..f7b6e67365 100644 --- a/apps/backend/internal/agent/lifecycle/events.go +++ b/apps/backend/internal/agent/lifecycle/events.go @@ -114,17 +114,18 @@ func (p *EventPublisher) PublishAgentStreamEvent(execution *AgentExecution, even // Build the nested event data // event.SessionID is the ACP session ID (internal agent protocol session) eventData := &AgentStreamEventData{ - Type: event.Type, - ACPSessionID: event.SessionID, - Text: event.Text, - ToolCallID: event.ToolCallID, - ToolName: event.ToolName, - ToolTitle: event.ToolTitle, - ToolStatus: event.ToolStatus, - ToolArgs: event.ToolArgs, - ToolResult: event.ToolResult, - Error: event.Error, - Data: event.Data, + Type: event.Type, + ACPSessionID: event.SessionID, + Text: event.Text, + ToolCallID: event.ToolCallID, + ToolName: event.ToolName, + ToolTitle: event.ToolTitle, + ToolStatus: event.ToolStatus, + ToolArgs: event.ToolArgs, + ToolResult: event.ToolResult, + Error: event.Error, + SessionStatus: event.SessionStatus, + Data: event.Data, } // Build agent event message payload diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index ed928eca5e..643174509c 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -17,6 +17,7 @@ import ( "github.com/kandev/kandev/internal/agent/registry" "github.com/kandev/kandev/internal/agent/worktree" agentctl "github.com/kandev/kandev/internal/agentctl/client" + "github.com/kandev/kandev/internal/agentctl/types/streams" "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/events" "github.com/kandev/kandev/internal/events/bus" @@ -574,6 +575,8 @@ func (m *Manager) StartAgentProcess(ctx context.Context, executionID string) err return fmt.Errorf("failed to initialize ACP: %w", err) } + m.emitSessionStatusEvent(execution, agentConfig) + return nil } @@ -707,6 +710,29 @@ func (m *Manager) initializeACPSession(ctx context.Context, execution *AgentExec return m.sessionManager.InitializeAndPrompt(ctx, execution, agentConfig, taskDescription, m.MarkReady) } +// emitSessionStatusEvent emits a session_status event indicating whether the session was resumed or new. +func (m *Manager) emitSessionStatusEvent(execution *AgentExecution, agentConfig *registry.AgentTypeConfig) { + wasResumed := false + if execution.ACPSessionID != "" { + if agentConfig.SessionConfig.ResumeViaACP { + wasResumed = true + } else if agentConfig.SessionConfig.ResumeFlag != "" && agentConfig.SessionConfig.SupportsRecovery() { + wasResumed = true + } + } + + sessionStatus := streams.SessionStatusNew + if wasResumed { + sessionStatus = streams.SessionStatusResumed + } + + m.eventPublisher.PublishAgentStreamEvent(execution, streams.AgentEvent{ + Type: streams.EventTypeSessionStatus, + SessionID: execution.ACPSessionID, + SessionStatus: sessionStatus, + }) +} + // handlePermissionRequestEvent processes permission requests from the unified agent event stream func (m *Manager) handlePermissionRequestEvent(execution *AgentExecution, event agentctl.AgentEvent) { m.eventPublisher.PublishPermissionRequest(execution, event) diff --git a/apps/backend/internal/agent/lifecycle/session.go b/apps/backend/internal/agent/lifecycle/session.go index 2f8c788f3f..21d0d03fee 100644 --- a/apps/backend/internal/agent/lifecycle/session.go +++ b/apps/backend/internal/agent/lifecycle/session.go @@ -101,8 +101,7 @@ func (sm *SessionManager) InitializeSession( return result, nil } -// createOrLoadSession determines whether to create a new session or load an existing one -// based on the agent configuration and whether an existing session ID is provided. +// createOrLoadSession creates a new session or loads an existing one based on agent config. func (sm *SessionManager) createOrLoadSession( ctx context.Context, client *agentctl.Client, @@ -110,23 +109,9 @@ func (sm *SessionManager) createOrLoadSession( existingSessionID string, workspacePath string, ) (string, error) { - // Use session/load only when: - // 1. ResumeViaACP is true (agent supports ACP-based resume) - // 2. An existing session ID is provided - shouldLoadSession := agentConfig.SessionConfig.ResumeViaACP && existingSessionID != "" - - if shouldLoadSession { + if agentConfig.SessionConfig.ResumeViaACP && existingSessionID != "" { return sm.loadSession(ctx, client, agentConfig, existingSessionID) } - - // Log why we're creating a new session when an existing session ID was provided - if existingSessionID != "" && !agentConfig.SessionConfig.ResumeViaACP { - sm.logger.Info("skipping ACP session/load (resume handled via CLI)", - zap.String("agent_type", agentConfig.ID), - zap.String("session_id", existingSessionID), - zap.String("resume_flag", agentConfig.SessionConfig.ResumeFlag)) - } - return sm.createNewSession(ctx, client, agentConfig, workspacePath) } diff --git a/apps/backend/internal/agent/registry/defaults.go b/apps/backend/internal/agent/registry/defaults.go index 93c056ccd8..dea29b6bb0 100644 --- a/apps/backend/internal/agent/registry/defaults.go +++ b/apps/backend/internal/agent/registry/defaults.go @@ -2,6 +2,11 @@ package registry import "github.com/kandev/kandev/pkg/agent" +// boolPtr returns a pointer to a bool value +func boolPtr(v bool) *bool { + return &v +} + // DefaultAgents returns the default agent configurations func DefaultAgents() []*AgentTypeConfig { return []*AgentTypeConfig{ @@ -31,6 +36,7 @@ func DefaultAgents() []*AgentTypeConfig { SessionConfig: SessionConfig{ ResumeViaACP: false, // Auggie handles resume via CLI flag ResumeFlag: "--resume", + CanRecover: boolPtr(false), // Auggie cannot recover sessions after backend restart SessionDirTemplate: "{home}/.augment/sessions", SessionDirTarget: "/root/.augment/sessions", }, diff --git a/apps/backend/internal/agent/registry/registry.go b/apps/backend/internal/agent/registry/registry.go index 499354b315..a50a1bc575 100644 --- a/apps/backend/internal/agent/registry/registry.go +++ b/apps/backend/internal/agent/registry/registry.go @@ -55,6 +55,12 @@ type SessionConfig struct { // Example: "--resume" for auggie, empty for agents that use ACP session/load. ResumeFlag string `json:"resume_flag,omitempty"` + // CanRecover indicates whether this agent supports session recovery after a backend restart. + // When false, sessions for this agent cannot be resumed after the backend restarts. + // On recovery, a new session will be started instead of trying to resume. + // Default is true (agent supports recovery). + CanRecover *bool `json:"can_recover,omitempty"` + // SessionDirTemplate is a template for the host directory where session data is stored. // Supports variables: {home} (user home directory). // Example: "{home}/.augment/sessions" for auggie. @@ -65,6 +71,15 @@ type SessionConfig struct { SessionDirTarget string `json:"session_dir_target,omitempty"` } +// SupportsRecovery returns whether the agent supports session recovery after backend restart. +// Returns true by default if CanRecover is not explicitly set. +func (c SessionConfig) SupportsRecovery() bool { + if c.CanRecover == nil { + return true + } + return *c.CanRecover +} + // PermissionConfig defines how tool permissions are requested for an agent type type PermissionConfig struct { // PermissionFlag is the CLI flag format for requesting tool permissions. diff --git a/apps/backend/internal/agentctl/server/api/control_server.go b/apps/backend/internal/agentctl/server/api/control_server.go index 89445c3f98..c352be2be2 100644 --- a/apps/backend/internal/agentctl/server/api/control_server.go +++ b/apps/backend/internal/agentctl/server/api/control_server.go @@ -56,9 +56,8 @@ func (m *ControlServer) setupRoutes() { func (m *ControlServer) handleHealth(c *gin.Context) { c.JSON(http.StatusOK, gin.H{ - "status": "ok", - "timestamp": time.Now().UTC().Format(time.RFC3339), - "max_instances": m.cfg.MaxInstances, + "status": "ok", + "timestamp": time.Now().UTC().Format(time.RFC3339), }) } diff --git a/apps/backend/internal/agentctl/server/config/config.go b/apps/backend/internal/agentctl/server/config/config.go index 6317c457ef..62bab93c54 100644 --- a/apps/backend/internal/agentctl/server/config/config.go +++ b/apps/backend/internal/agentctl/server/config/config.go @@ -28,9 +28,6 @@ type Config struct { // Ports configures the port range for instance allocation Ports PortConfig - // MaxInstances is the maximum number of concurrent agent instances. - MaxInstances int - // Defaults provides default values for new instances Defaults InstanceDefaults @@ -117,8 +114,7 @@ type InstanceConfig struct { // Load loads the configuration from environment variables. func Load() *Config { return &Config{ - Port: getEnvInt("AGENTCTL_PORT", 9999), - MaxInstances: getEnvInt("AGENTCTL_MAX_INSTANCES", 10), + Port: getEnvInt("AGENTCTL_PORT", 9999), Ports: PortConfig{ Base: getEnvInt("AGENTCTL_INSTANCE_PORT_BASE", 10001), Max: getEnvInt("AGENTCTL_INSTANCE_PORT_MAX", 10100), diff --git a/apps/backend/internal/agentctl/server/instance/manager.go b/apps/backend/internal/agentctl/server/instance/manager.go index 93be0cde5c..dee142c0ef 100644 --- a/apps/backend/internal/agentctl/server/instance/manager.go +++ b/apps/backend/internal/agentctl/server/instance/manager.go @@ -53,11 +53,6 @@ func (m *Manager) CreateInstance(ctx context.Context, req *CreateRequest) (*Crea m.mu.Lock() defer m.mu.Unlock() - // Check if max instances reached - if len(m.instances) >= m.config.MaxInstances { - return nil, fmt.Errorf("maximum number of instances (%d) reached", m.config.MaxInstances) - } - // Generate ID if not provided id := req.ID if id == "" { diff --git a/apps/backend/internal/agentctl/types/streams/agent.go b/apps/backend/internal/agentctl/types/streams/agent.go index 18f951b97c..e3bda6db3f 100644 --- a/apps/backend/internal/agentctl/types/streams/agent.go +++ b/apps/backend/internal/agentctl/types/streams/agent.go @@ -25,6 +25,15 @@ const ( // EventTypePermissionRequest indicates the agent is requesting permission. EventTypePermissionRequest = "permission_request" + + // EventTypeSessionStatus indicates a session status update (resumed or new). + EventTypeSessionStatus = "session_status" +) + +// Session status constants for EventTypeSessionStatus events. +const ( + SessionStatusResumed = "resumed" + SessionStatusNew = "new" ) // AgentEvent is the message type streamed from the agent process. @@ -110,6 +119,12 @@ type AgentEvent struct { // ActionDetails contains structured details about the action. ActionDetails map[string]interface{} `json:"action_details,omitempty"` + // --- Session status fields (for "session_status" type) --- + + // SessionStatus indicates whether the session was resumed or new. + // Use SessionStatus* constants: "resumed", "new". + SessionStatus string `json:"session_status,omitempty"` + // --- Extension fields --- // Data contains raw protocol-specific extensions. diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 24f5389d24..da42fe509b 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -257,19 +257,14 @@ func (s *Service) resumeExecutorsOnStartup(ctx context.Context) { continue } - if !running.Resumable { - s.logger.Warn("executor not resumable; marking session as failed", - zap.String("session_id", sessionID)) - _ = s.repo.UpdateTaskSessionState(ctx, sessionID, models.TaskSessionStateFailed, "executor not resumable") - continue - } - - // When the executors start, the session changes to RUNNING state. - // After resuming, we need to restore the previous state. previousState := session.State + + // Handle sessions in terminal states - clean up executor record switch previousState { case models.TaskSessionStateCompleted, models.TaskSessionStateFailed, models.TaskSessionStateCancelled: - // Session was already in a terminal state; no need to resume. + s.logger.Info("session in terminal state; cleaning up executor record", + zap.String("session_id", sessionID), + zap.String("state", string(previousState))) if err := s.repo.DeleteExecutorRunningBySessionID(ctx, sessionID); err != nil { s.logger.Warn("failed to remove executor record for terminal session", zap.String("session_id", sessionID), @@ -277,32 +272,88 @@ func (s *Service) resumeExecutorsOnStartup(ctx context.Context) { } continue } - // validate the environment (e.g., worktrees) before resuming + + // Handle sessions that never started - skip without failure + if previousState == models.TaskSessionStateCreated { + s.logger.Info("session was never started; skipping recovery", + zap.String("session_id", sessionID)) + if err := s.repo.DeleteExecutorRunningBySessionID(ctx, sessionID); err != nil { + s.logger.Warn("failed to remove executor record for unstarted session", + zap.String("session_id", sessionID), + zap.Error(err)) + } + continue + } + + // For sessions that were STARTING, RUNNING, or WAITING_FOR_INPUT: + // We need a resume token to recover them + if running.ResumeToken == "" { + s.logger.Warn("no resume token available; marking session as failed", + zap.String("session_id", sessionID), + zap.String("previous_state", string(previousState))) + _ = s.repo.UpdateTaskSessionState(ctx, sessionID, models.TaskSessionStateFailed, + "backend restarted and session cannot be resumed (no resume token)") + _ = s.repo.DeleteExecutorRunningBySessionID(ctx, sessionID) + continue + } + + // Check if executor supports resume + if !running.Resumable { + s.logger.Warn("executor not resumable; marking session as failed", + zap.String("session_id", sessionID)) + _ = s.repo.UpdateTaskSessionState(ctx, sessionID, models.TaskSessionStateFailed, + "executor does not support session resume") + _ = s.repo.DeleteExecutorRunningBySessionID(ctx, sessionID) + continue + } + + // Validate worktree exists before resuming if err := validateSessionWorktrees(session); err != nil { s.logger.Warn("worktree validation failed; marking session as failed", zap.String("session_id", sessionID), zap.Error(err)) _ = s.repo.UpdateTaskSessionState(ctx, sessionID, models.TaskSessionStateFailed, err.Error()) + _ = s.repo.DeleteExecutorRunningBySessionID(ctx, sessionID) continue } - startAgent := running.ResumeToken != "" - if _, err := s.executor.ResumeSession(ctx, session, startAgent); err != nil { + s.logger.Info("resuming session on startup", + zap.String("session_id", sessionID), + zap.String("task_id", running.TaskID), + zap.String("previous_state", string(previousState)), + zap.String("resume_token", running.ResumeToken)) + + // Resume the session - this will: + // 1. Create a new agentctl instance + // 2. Start the agent process + // 3. Initialize and load the session via ACP/Codex protocol + // + // ResumeSession will temporarily set state to STARTING during launch. + // After resume, the agent is immediately ready for input (session is loaded, + // no prompt is being processed). We set WAITING_FOR_INPUT directly because: + // - Session resume just loads conversation history, doesn't process a prompt + // - Agents don't send a "complete" event after session load (only after prompt completion) + // - The agent is ready for the next user prompt immediately after session loads + if _, err := s.executor.ResumeSession(ctx, session, true); err != nil { s.logger.Warn("failed to resume executor on startup", zap.String("session_id", sessionID), zap.Error(err)) _ = s.repo.UpdateTaskSessionState(ctx, sessionID, models.TaskSessionStateFailed, err.Error()) + _ = s.repo.DeleteExecutorRunningBySessionID(ctx, sessionID) continue } - // Restore previous state if not CREATED or STARTING - if previousState != models.TaskSessionStateCreated && previousState != models.TaskSessionStateStarting { - if err := s.repo.UpdateTaskSessionState(ctx, sessionID, previousState, ""); err != nil { - s.logger.Warn("failed to restore session state after resume", - zap.String("session_id", sessionID), - zap.Error(err)) - } + // Set state to WAITING_FOR_INPUT - agent is ready for the next prompt + if err := s.repo.UpdateTaskSessionState(ctx, sessionID, models.TaskSessionStateWaitingForInput, ""); err != nil { + s.logger.Warn("failed to set session state to WAITING_FOR_INPUT after resume", + zap.String("session_id", sessionID), + zap.Error(err)) } + + s.logger.Info("session resumed successfully", + zap.String("session_id", sessionID), + zap.String("task_id", running.TaskID), + zap.String("state", string(models.TaskSessionStateWaitingForInput))) } } @@ -881,6 +932,31 @@ func (s *Service) handleAgentStreamEvent(ctx context.Context, payload *lifecycle zap.Error(err)) } } + + case "session_status": + // Handle session status events (resumed vs new session) + if sessionID != "" && s.messageCreator != nil { + var statusMsg string + if payload.Data.SessionStatus == "resumed" { + statusMsg = "Session resumed" + } else { + statusMsg = "New session started" + } + if err := s.messageCreator.CreateSessionMessage( + ctx, + taskID, + statusMsg, + sessionID, + string(v1.MessageTypeStatus), + nil, + false, + ); err != nil { + s.logger.Error("failed to create session status message", + zap.String("task_id", taskID), + zap.String("session_id", sessionID), + zap.Error(err)) + } + } } // Broadcast to all registered handlers (WebSocket streaming) diff --git a/apps/backend/internal/task/handlers/message_handlers.go b/apps/backend/internal/task/handlers/message_handlers.go index 5920f417e0..66b55a11a3 100644 --- a/apps/backend/internal/task/handlers/message_handlers.go +++ b/apps/backend/internal/task/handlers/message_handlers.go @@ -147,10 +147,9 @@ func (h *MessageHandlers) wsAddMessage(ctx context.Context, msg *ws.Message) (*w return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to get task", nil) } - // If task is in REVIEW state, transition back to IN_PROGRESS + // If task is in REVIEW state, transition back to IN_PROGRESS (also moves to matching column) if task.State == v1.TaskStateReview { - nextState := v1.TaskStateInProgress - if _, err := h.taskController.UpdateTask(ctx, dto.UpdateTaskRequest{ID: req.TaskID, State: &nextState}); err != nil { + if _, err := h.taskController.UpdateTaskState(ctx, dto.UpdateTaskStateRequest{ID: req.TaskID, State: v1.TaskStateInProgress}); err != nil { h.logger.Error("failed to transition task from REVIEW to IN_PROGRESS", zap.String("task_id", req.TaskID), zap.Error(err)) @@ -173,11 +172,14 @@ func (h *MessageHandlers) wsAddMessage(ctx context.Context, msg *ws.Message) (*w } // Auto-forward message as prompt to running agent if orchestrator is available + // Use context.WithoutCancel so the prompt continues even if the WebSocket client disconnects. + // The user's message is already saved, and agent responses are broadcast via notifications. if h.orchestrator != nil { - _, err := h.orchestrator.PromptTask(ctx, req.TaskID, req.TaskSessionID, req.Content) + promptCtx := context.WithoutCancel(ctx) + _, err := h.orchestrator.PromptTask(promptCtx, req.TaskID, req.TaskSessionID, req.Content) if err != nil { if errors.Is(err, executor.ErrExecutionNotFound) { - if resumeErr := h.orchestrator.ResumeTaskSession(ctx, req.TaskID, req.TaskSessionID); resumeErr != nil { + if resumeErr := h.orchestrator.ResumeTaskSession(promptCtx, req.TaskID, req.TaskSessionID); resumeErr != nil { h.logger.Warn("failed to resume task session for prompt", zap.String("task_id", req.TaskID), zap.String("session_id", req.TaskSessionID), @@ -185,7 +187,7 @@ func (h *MessageHandlers) wsAddMessage(ctx context.Context, msg *ws.Message) (*w } else { for attempt := 0; attempt < 3; attempt++ { time.Sleep(500 * time.Millisecond) - _, err = h.orchestrator.PromptTask(ctx, req.TaskID, req.TaskSessionID, req.Content) + _, err = h.orchestrator.PromptTask(promptCtx, req.TaskID, req.TaskSessionID, req.Content) if err == nil { break } diff --git a/apps/web/components/task/chat/messages/status-message.tsx b/apps/web/components/task/chat/messages/status-message.tsx index ce2320f85c..9c03b83460 100644 --- a/apps/web/components/task/chat/messages/status-message.tsx +++ b/apps/web/components/task/chat/messages/status-message.tsx @@ -1,6 +1,6 @@ 'use client'; -import { IconAlertTriangle, IconInfoCircle } from '@tabler/icons-react'; +import { IconAlertTriangle, IconInfoCircle, IconPointFilled } from '@tabler/icons-react'; import { cn } from '@/lib/utils'; import type { Message } from '@/lib/types/http'; import type { StatusMetadata } from '@/components/task/chat/types'; @@ -11,21 +11,37 @@ export function StatusMessage({ comment }: { comment: Message }) { typeof metadata?.progress === 'number' ? Math.min(Math.max(metadata.progress, 0), 100) : null; const statusLine = metadata?.stage || metadata?.status; const message = metadata?.message || comment.content || statusLine || 'Status update'; - const Icon = comment.type === 'error' ? IconAlertTriangle : IconInfoCircle; - const iconClass = comment.type === 'error' ? 'text-red-500' : 'text-muted-foreground'; + const isError = comment.type === 'error'; + + // Simple system message: no metadata, no progress, just a short message + const isSimpleStatus = !isError && progress === null && !statusLine && !metadata?.message; + + if (isSimpleStatus) { + return ( +
+
+ + {message} +
+
+ ); + } + + const Icon = isError ? IconAlertTriangle : IconInfoCircle; + const iconClass = isError ? 'text-red-500' : 'text-muted-foreground'; return (
- {comment.type === 'error' ? 'Error' : 'Status'} + {isError ? 'Error' : 'Status'}
{message}
{progress !== null && (