Merge pull request #1261 from kubet/fix/stream-restore-on-thread-navigation

refactor: improve thread and stream handling on navigation

Author: kubet

frontend/src/app/(dashboard)/projects/[projectId]/thread/_hooks/useThreadData.ts

@@ -145,23 +145,7 @@ export function useThreadData(threadId: string, projectId: string): UseThreadDat
         if (threadQuery.data && messagesQuery.data && agentRunsQuery.data) {
           initialLoadCompleted.current = true;
           setIsLoading(false);
-          
-          // Final safety check: if no recent active runs found, ensure status is idle
-          if (agentRunsCheckedRef.current) {
-            const thirtySecondsAgo = new Date(Date.now() - 30 * 1000);
-            const hasRecentActiveRun = agentRunsQuery.data.find((run) => {
-              const runCreatedAt = new Date(run.started_at || 0);
-              return run.status === 'running' && runCreatedAt > thirtySecondsAgo;
-            });
-            
-            if (!hasRecentActiveRun) {
-              console.log('[PAGE] Final check: No recent active runs, ensuring idle status');
-              if (isMounted) {
-                setAgentStatus('idle');
-                setAgentRunId(null);
-              }
-            }
-          }
+          // Removed time-based final check to avoid incorrectly forcing idle while a stream is active
         }
 
       } catch (err) {

frontend/src/hooks/useAgentStream.ts

@@ -116,8 +116,19 @@ export function useAgentStream(
     textContentRef.current = textContent;
   }, [textContent]);
 
-  // Update refs if threadId changes (no persistence across navigation)
+  // On thread change, ensure any existing stream is cleaned up to avoid stale subscriptions
   useEffect(() => {
+    const previousThreadId = threadIdRef.current;
+    if (previousThreadId && previousThreadId !== threadId && streamCleanupRef.current) {
+      // Close the existing stream for the previous thread
+      streamCleanupRef.current();
+      streamCleanupRef.current = null;
+      setStatus('idle');
+      setTextContent([]);
+      setToolCall(null);
+      setAgentRunId(null);
+      currentRunIdRef.current = null;
+    }
     threadIdRef.current = threadId;
   }, [threadId]);
 
@@ -547,12 +558,38 @@ export function useAgentStream(
           `[useAgentStream] Agent run ${runId} confirmed running. Setting up EventSource.`,
         );
         const cleanup = streamAgent(runId, {
-          onMessage: handleStreamMessage,
-          onError: handleStreamError,
-          onClose: handleStreamClose,
+          onMessage: (data) => {
+            // Ignore messages if threadId changed while the EventSource stayed open
+            if (threadIdRef.current !== threadId) return;
+            handleStreamMessage(data);
+          },
+          onError: (err) => {
+            if (threadIdRef.current !== threadId) return;
+            handleStreamError(err);
+          },
+          onClose: () => {
+            if (threadIdRef.current !== threadId) return;
+            handleStreamClose();
+          },
         });
         streamCleanupRef.current = cleanup;
         // Status will be updated to 'streaming' by the first message received in handleStreamMessage
+        // If for some reason no message arrives shortly, verify liveness again to avoid zombie state
+        setTimeout(async () => {
+          if (!isMountedRef.current) return;
+          if (currentRunIdRef.current !== runId) return; // Another run started
+          if (statusRef.current === 'streaming') return; // Already streaming
+          try {
+            const latest = await getAgentStatus(runId);
+            if (!isMountedRef.current) return;
+            if (currentRunIdRef.current !== runId) return;
+            if (latest.status !== 'running') {
+              finalizeStream(mapAgentStatus(latest.status) || 'agent_not_running', runId);
+            }
+          } catch {
+            // ignore
+          }
+        }, 1500);
       } catch (err) {
         if (!isMountedRef.current) return; // Check mount status after async call