diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go b/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go index 6f67becfb4..3a3f6e2f62 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go @@ -1878,6 +1878,10 @@ func (m *Manager) RecoverAgentPromptStream(ctx context.Context, sessionID string } if client.HasAgentStream() { releaseClient() + if execution.Status == v1.AgentStatusFailed && + execution.isSessionInitialized() && execution.ACPSessionID != "" { + return m.restoreRecoveredFailedExecution(ctx, execution) + } return nil } releaseClient() diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_interaction_test.go b/apps/backend/internal/agent/runtime/lifecycle/manager_interaction_test.go index ef31d75b1d..59606f4b5e 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_interaction_test.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_interaction_test.go @@ -1539,6 +1539,44 @@ func TestRecoverAgentPromptStream(t *testing.T) { require.ErrorContains(t, err, "stream manager is not configured") }) + t.Run("restores stale failed status when the recovered stream is already connected", func(t *testing.T) { + mock := newMockAgentServer(t) + t.Cleanup(mock.Close) + + client := createTestClient(t, mock.server.URL) + t.Cleanup(client.Close) + + streamCtx, cancelStream := context.WithCancel(context.Background()) + t.Cleanup(cancelStream) + require.NoError(t, client.StreamUpdates(streamCtx, func(agentctl.AgentEvent) {}, nil, nil)) + select { + case <-mock.wsConnected: + case <-time.After(2 * time.Second): + t.Fatal("mock server did not see preconnected updates stream") + } + + mgr := newTestManager(t) + exec := &AgentExecution{ + ID: "exec-preconnected-recover", + SessionID: "session-preconnected-recover", + ACPSessionID: "acp-session-1", + Status: v1.AgentStatusFailed, + agentctl: client, + promptDoneCh: make(chan PromptCompletionSignal, 1), + sessionInitialized: true, + } + require.NoError(t, mgr.executionStore.Add(exec)) + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + t.Cleanup(cancel) + require.NoError(t, mgr.RecoverAgentPromptStream(ctx, exec.SessionID)) + + updated, ok := mgr.executionStore.Get(exec.ID) + require.True(t, ok) + require.Equal(t, v1.AgentStatusReady, updated.Status, + "a remote refresh may reconnect the stream before prompt recovery repairs the disconnect status") + }) + t.Run("reconnects stream and restores stale failed status", func(t *testing.T) { mock := newMockAgentServer(t) t.Cleanup(mock.Close)