feat: add runner syncing, simplify generation selection and stream handling

packages/giselle/src/react/generations/generate-content-runner.tsx

@@ -61,10 +61,31 @@ export function GenerateContentRunner({
 	const upsertMessage = useGenerationStore((s) => s.upsertMessage);
 	const updateGeneration = useGenerationStore((s) => s.updateGeneration);
 	const { updateGenerationStatusToComplete } = useGenerationRunnerSystem();
-	const didRun = useRef(false);
+	const didPerformingContentGeneration = useRef(false);
+	const didListeningContentGeneration = useRef(false);
 	const reachedStreamEnd = useRef(false);
 	const messageUpdateQueue = useRef<Map<UIMessage["id"], UIMessage>>(new Map());
 	const pendingUpdate = useRef<number | null>(null);
+	const prevGenerationId = useRef(generation.id);
+
+	// Reset lifecycle refs when generation changes
+	useEffect(() => {
+		if (prevGenerationId.current === generation.id) {
+			return;
+		}
+		didPerformingContentGeneration.current = false;
+		didListeningContentGeneration.current = false;
+		reachedStreamEnd.current = false;
+
+		// Clear message queue to prevent stale messages from previous generation
+		messageUpdateQueue.current.clear();
+
+		// Cancel pending animation frame to prevent applying stale updates
+		if (pendingUpdate.current !== null) {
+			cancelAnimationFrame(pendingUpdate.current);
+			pendingUpdate.current = null;
+		}
+	}, [generation.id]);
 
 	const flushMessageUpdates = useCallback(() => {
 		if (messageUpdateQueue.current.size === 0) return;
@@ -160,20 +181,32 @@ export function GenerateContentRunner({
 	]);
 
 	useEffect(() => {
-		if (didRun.current) {
+		if (didPerformingContentGeneration.current) {
 			return;
 		}
 		if (generation.status !== "queued") {
 			return;
 		}
-		didRun.current = true;
+		didPerformingContentGeneration.current = true;
 		client
 			.startContentGeneration({ generation })
 			.then(({ generation: runningGeneration }) => {
 				onStart?.(runningGeneration);
 				updateGeneration(runningGeneration);
-				processStream();
 			});
-	}, [generation, client, processStream, updateGeneration, onStart]);
+	}, [generation, client, updateGeneration, onStart]);
+
+	useEffect(() => {
+		if (didListeningContentGeneration.current) {
+			return;
+		}
+		if (generation.status !== "running") {
+			return;
+		}
+		didListeningContentGeneration.current = true;
+
+		processStream();
+	}, [generation, processStream]);
+
 	return null;
 }

packages/giselle/src/react/generations/hooks/use-node-generations.ts

@@ -1,5 +1,5 @@
 import type { NodeId } from "@giselle-sdk/data-type";
-import { useCallback, useMemo } from "react";
+import { useCallback, useEffect, useMemo } from "react";
 import useSWR from "swr";
 import { useShallow } from "zustand/shallow";
 import type {
@@ -30,12 +30,13 @@ export function useNodeGenerations({
 		startGenerationRunner,
 		createAndStartGenerationRunner,
 		stopGenerationRunner: stopGenerationSystem,
+		addGenerationRunner,
 	} = useGenerationRunnerSystem();
 	const client = useGiselleEngine();
 	const { experimental_storage } = useFeatureFlag();
 
 	/** @todo fetch on server */
-	const { data } = useSWR(
+	const { data, isLoading } = useSWR(
 		{
 			api: "node-generations",
 			origin,
@@ -49,35 +50,33 @@ export function useNodeGenerations({
 			revalidateOnReconnect: false,
 		},
 	);
+
+	useEffect(() => {
+		if (isLoading || data === undefined) {
+			return;
+		}
+		addGenerationRunner(data);
+	}, [isLoading, data, addGenerationRunner]);
+
 	const currentGeneration = useMemo<Generation>(() => {
-		const fetchGenerations = data ?? [];
-		const createdGenerations = generations.filter(
-			(generation) =>
-				generation.context.operationNode.id === nodeId &&
-				generation.context.origin.type === origin.type &&
-				(origin.type === "studio"
-					? generation.context.origin.type === "studio" &&
-						generation.context.origin.workspaceId === origin.workspaceId
-					: generation.context.origin.type !== "studio" &&
-						generation.context.origin.actId === origin.actId),
-		);
-		// Deduplicate generations by filtering out fetched generations from created ones
-		const deduplicatedCreatedGenerations = createdGenerations.filter(
-			(created) =>
-				!fetchGenerations.some((fetched) => fetched.id === created.id),
-		);
-		// Filter out cancelled generations from both sources after deduplication
-		const allGenerations = [
-			...fetchGenerations,
-			...deduplicatedCreatedGenerations,
-		]
-			.filter((generation) => generation.status !== "cancelled")
+		const filteredGenerations = generations
+			.filter(
+				(generation) =>
+					generation.status !== "cancelled" &&
+					generation.context.operationNode.id === nodeId &&
+					generation.context.origin.type === origin.type &&
+					(origin.type === "studio"
+						? generation.context.origin.type === "studio" &&
+							generation.context.origin.workspaceId === origin.workspaceId
+						: generation.context.origin.type !== "studio" &&
+							generation.context.origin.actId === origin.actId),
+			)
 			.sort(
 				(a, b) =>
 					new Date(b.createdAt).getTime() - new Date(a.createdAt).getTime(),
 			);
-		return allGenerations[0];
-	}, [generations, data, nodeId, origin]);
+		return filteredGenerations[0];
+	}, [generations, nodeId, origin]);
 
 	const isGenerating = useMemo(
 		() =>

packages/giselle/src/react/generations/store.ts

@@ -24,8 +24,12 @@ export const useGenerationStore = create<GenerationStore>((set) => ({
 	addGenerationRunner: (generations) =>
 		set((state) => {
 			const arr = Array.isArray(generations) ? generations : [generations];
+			const incomingIds = new Set(arr.map((g) => g.id));
+			const filteredExisting = state.generations.filter(
+				(g) => !incomingIds.has(g.id),
+			);
 			return {
-				generations: [...state.generations, ...arr],
+				generations: [...filteredExisting, ...arr],
 			};
 		}),
 	updateGeneration: (generation) =>