diff --git a/apps/backend/internal/agent/lifecycle/runtime_standalone.go b/apps/backend/internal/agent/lifecycle/runtime_standalone.go index 04b3eb989b..63e482f8af 100644 --- a/apps/backend/internal/agent/lifecycle/runtime_standalone.go +++ b/apps/backend/internal/agent/lifecycle/runtime_standalone.go @@ -3,6 +3,7 @@ package lifecycle import ( "context" "fmt" + "time" "go.uber.org/zap" @@ -37,7 +38,38 @@ func (r *StandaloneRuntime) HealthCheck(ctx context.Context) error { return r.ctl.Health(ctx) } +func (r *StandaloneRuntime) waitForReady(ctx context.Context) error { + if err := r.ctl.Health(ctx); err == nil { + return nil + } + + waitCtx := ctx + if _, hasDeadline := ctx.Deadline(); !hasDeadline { + var cancel context.CancelFunc + waitCtx, cancel = context.WithTimeout(ctx, 10*time.Second) + defer cancel() + } + + ticker := time.NewTicker(500 * time.Millisecond) + defer ticker.Stop() + + for { + select { + case <-waitCtx.Done(): + return fmt.Errorf("agentctl not ready: %w", waitCtx.Err()) + case <-ticker.C: + if err := r.ctl.Health(waitCtx); err == nil { + return nil + } + } + } +} + func (r *StandaloneRuntime) CreateInstance(ctx context.Context, req *RuntimeCreateRequest) (*RuntimeInstance, error) { + if err := r.waitForReady(ctx); err != nil { + return nil, err + } + // Build environment variables env := req.Env if env == nil { diff --git a/apps/backend/internal/gateway/websocket/task_notifications.go b/apps/backend/internal/gateway/websocket/task_notifications.go index dbe16780fa..054287c201 100644 --- a/apps/backend/internal/gateway/websocket/task_notifications.go +++ b/apps/backend/internal/gateway/websocket/task_notifications.go @@ -50,6 +50,7 @@ func RegisterTaskNotifications(ctx context.Context, eventBus bus.EventBus, hub * b.subscribe(eventBus, events.EnvironmentCreated, ws.ActionEnvironmentCreated) b.subscribe(eventBus, events.EnvironmentUpdated, ws.ActionEnvironmentUpdated) b.subscribe(eventBus, events.EnvironmentDeleted, ws.ActionEnvironmentDeleted) + b.subscribe(eventBus, events.TaskSessionStateChanged, ws.ActionSessionStateChanged) b.subscribe(eventBus, events.MessageAdded, ws.ActionSessionMessageAdded) b.subscribe(eventBus, events.MessageUpdated, ws.ActionSessionMessageUpdated) b.subscribe(eventBus, events.AgentctlStarting, ws.ActionSessionAgentctlStarting) @@ -81,7 +82,7 @@ func (b *TaskEventBroadcaster) subscribe(eventBus bus.EventBus, subject, action return nil } switch action { - case ws.ActionSessionAgentctlStarting, ws.ActionSessionAgentctlReady, ws.ActionSessionAgentctlError: + case ws.ActionSessionAgentctlStarting, ws.ActionSessionAgentctlReady, ws.ActionSessionAgentctlError, ws.ActionSessionStateChanged: if sessionID, ok := event.Data["session_id"].(string); ok && sessionID != "" { b.hub.BroadcastToSession(sessionID, msg) return nil diff --git a/apps/backend/internal/orchestrator/executor/executor.go b/apps/backend/internal/orchestrator/executor/executor.go index 302de8053c..44d1bf166a 100644 --- a/apps/backend/internal/orchestrator/executor/executor.go +++ b/apps/backend/internal/orchestrator/executor/executor.go @@ -862,6 +862,7 @@ func (e *Executor) RespondToPermission(ctx context.Context, sessionID, pendingID // GetExecution returns the current execution state for a task func (e *Executor) GetExecution(taskID string) (*TaskExecution, bool) { ctx := context.Background() + const startupGracePeriod = 30 * time.Second e.mu.RLock() execution, exists := e.executions[taskID] @@ -870,23 +871,9 @@ func (e *Executor) GetExecution(taskID string) (*TaskExecution, bool) { if exists { // Verify the agent is actually running by probing agentctl if execution.SessionID == "" || !e.agentManager.IsAgentRunningForSession(ctx, execution.SessionID) { - // Agent stopped - clean up in-memory state - e.mu.Lock() - delete(e.executions, taskID) - e.mu.Unlock() - - // Also update DB if we have a session - if execution.SessionID != "" { - _ = e.repo.UpdateTaskSessionState(ctx, execution.SessionID, models.TaskSessionStateCancelled, "agent process stopped") - } - - // Clean up the stale execution from the agent manager - if execution.SessionID != "" { - if err := e.agentManager.CleanupStaleExecutionBySessionID(ctx, execution.SessionID); err != nil { - e.logger.Warn("failed to cleanup stale agent execution", - zap.String("session_id", execution.SessionID), - zap.Error(err)) - } + if time.Since(execution.StartedAt) < startupGracePeriod { + execCopy := *execution + return &execCopy, true } return nil, false } @@ -905,19 +892,14 @@ func (e *Executor) GetExecution(taskID string) (*TaskExecution, bool) { // Verify the agent is actually running by probing the lifecycle manager // This handles the case where backend was restarted and DB has stale "running" sessions if session.ID == "" || !e.agentManager.IsAgentRunningForSession(ctx, session.ID) { - // Agent is not running - mark the session as stopped in DB and return false - e.logger.Info("stale agent session detected - agent not running", - zap.String("task_id", taskID), - zap.String("session_id", session.ID)) - _ = e.repo.UpdateTaskSessionState(ctx, session.ID, models.TaskSessionStateCancelled, "agent not running after backend restart") - - // Clean up the stale execution from the agent manager - if session.ID != "" { - if err := e.agentManager.CleanupStaleExecutionBySessionID(ctx, session.ID); err != nil { - e.logger.Warn("failed to cleanup stale agent execution", - zap.String("session_id", session.ID), - zap.Error(err)) - } + if (session.State == models.TaskSessionStateStarting || session.State == models.TaskSessionStateRunning) && + time.Since(session.UpdatedAt) < startupGracePeriod { + execution = FromTaskSession(session) + e.mu.Lock() + e.executions[taskID] = execution + e.mu.Unlock() + execCopy := *execution + return &execCopy, true } return nil, false } diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index b2c14b9c5d..739d228465 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -486,7 +486,6 @@ func (s *Service) GetTaskSessionStatus(ctx context.Context, taskID, sessionID st resp.IsAgentRunning = false resp.IsResumable = false resp.NeedsResume = false - resp.Error = "session has no resume token" return resp, nil } diff --git a/apps/web/app/task/[id]/page-client.tsx b/apps/web/app/task/[id]/page-client.tsx index d76b570eb4..81d75012b9 100644 --- a/apps/web/app/task/[id]/page-client.tsx +++ b/apps/web/app/task/[id]/page-client.tsx @@ -69,13 +69,15 @@ export default function TaskPage({ const initialSessionId = sessionId ?? taskSessionId ?? null; const effectiveSessionId = activeSessionId ?? initialSessionId; + const storeSessionState = useAppStore((state) => + effectiveSessionId ? state.taskSessions.items[effectiveSessionId]?.state ?? null : null + ); const agentctlStatus = useSessionAgentctl(effectiveSessionId); // Session resumption hook - handles auto-resume on page reload const { resumptionState, error: resumptionError, - taskSessionState: resumedSessionState, worktreePath: resumedWorktreePath, worktreeBranch: resumedWorktreeBranch, } = useSessionResumption(task?.id ?? null, effectiveSessionId); @@ -86,7 +88,7 @@ export default function TaskPage({ // Both hooks read taskSessionState from the store, so prefer resumedSessionState as it's always // updated when we have a sessionId from URL. For worktree, use resumed values when resuming/resumed. - const taskSessionState = resumedSessionState ?? agentSessionState; + const taskSessionState = storeSessionState ?? agentSessionState; const worktreePath = effectiveSessionId ? (resumedWorktreePath ?? agentWorktreePath) : agentWorktreePath; const worktreeBranch = effectiveSessionId ? (resumedWorktreeBranch ?? agentWorktreeBranch) : agentWorktreeBranch; diff --git a/apps/web/components/task/shell-terminal.tsx b/apps/web/components/task/shell-terminal.tsx index bec6b2d236..25d3931510 100644 --- a/apps/web/components/task/shell-terminal.tsx +++ b/apps/web/components/task/shell-terminal.tsx @@ -16,6 +16,7 @@ export function ShellTerminal() { const lastOutputLengthRef = useRef(0); const subscriptionIdRef = useRef(0); const onDataDisposableRef = useRef<{ dispose: () => void } | null>(null); + const retryTimeoutRef = useRef | null>(null); const storeApi = useAppStoreApi(); const sessionId = useAppStore((state) => state.tasks.activeSessionId); @@ -139,21 +140,46 @@ export function ShellTerminal() { const client = getWebSocketClient(); if (!client) return; - client - .request<{ success: boolean; buffer?: string }>('shell.subscribe', { session_id: sessionId }) - .then((response) => { - if (subscriptionIdRef.current !== currentSubscriptionId) return; - if (response.buffer) { - storeApi.getState().appendShellOutput(sessionId, response.buffer); - } - }) - .catch((err) => { - if (subscriptionIdRef.current !== currentSubscriptionId) return; - console.error('Failed to subscribe to shell:', err); - }); + if (retryTimeoutRef.current) { + clearTimeout(retryTimeoutRef.current); + retryTimeoutRef.current = null; + } + + let cancelled = false; + + const attemptSubscribe = () => { + client + .request<{ success: boolean; buffer?: string }>('shell.subscribe', { session_id: sessionId }) + .then((response) => { + if (cancelled || subscriptionIdRef.current !== currentSubscriptionId) return; + if (response.buffer) { + storeApi.getState().appendShellOutput(sessionId, response.buffer); + } + }) + .catch((err) => { + if (cancelled || subscriptionIdRef.current !== currentSubscriptionId) return; + const message = err instanceof Error ? err.message : String(err); + if (message.includes('no agent running')) { + retryTimeoutRef.current = setTimeout(() => { + if (!cancelled && subscriptionIdRef.current === currentSubscriptionId) { + attemptSubscribe(); + } + }, 1000); + return; + } + console.error('Failed to subscribe to shell:', err); + }); + }; + + attemptSubscribe(); return () => { subscriptionIdRef.current += 1; + cancelled = true; + if (retryTimeoutRef.current) { + clearTimeout(retryTimeoutRef.current); + retryTimeoutRef.current = null; + } }; }, [taskId, sessionId, storeApi, canSubscribe]); diff --git a/apps/web/components/task/task-session-switcher.tsx b/apps/web/components/task/task-session-switcher.tsx index 4f8bf4e502..607f7aa5bc 100644 --- a/apps/web/components/task/task-session-switcher.tsx +++ b/apps/web/components/task/task-session-switcher.tsx @@ -170,12 +170,13 @@ export function TaskSessionSwitcher({ { if (!taskId) return; onSelectSession(taskId, session.id); }} - className="group flex w-full cursor-pointer items-center gap-3 px-2 py-1.5 rounded-sm text-left transition-colors data-[selected=true]:bg-muted" + className={`group flex w-full cursor-pointer items-center gap-3 px-2 py-1.5 rounded-sm text-left transition-colors ${ + isActive ? 'bg-muted' : 'hover:bg-muted/30' + }`} > {formatSessionLabel(index, sortedSessions.length)} diff --git a/apps/web/lib/http/index.ts b/apps/web/lib/http/index.ts index f13190fb71..173f779141 100644 --- a/apps/web/lib/http/index.ts +++ b/apps/web/lib/http/index.ts @@ -35,7 +35,8 @@ function resolveUrl(pathOrUrl: string, baseUrl: string) { async function fetchJson(pathOrUrl: string, options?: ApiRequestOptions): Promise { const baseUrl = options?.baseUrl ?? getBackendConfig().apiBaseUrl; - const response = await fetch(resolveUrl(pathOrUrl, baseUrl), { + const url = resolveUrl(pathOrUrl, baseUrl); + const response = await fetch(url, { ...options?.init, cache: options?.cache, headers: {