接入DirectProject前端线程订阅

新增独立raw event reducer并按item与turn事件重建运行态\n页面进入时注册notify订阅并通过consume推进Rust内部游标\n队列过期后保留旧状态并重新bootstrap

apps/ai-game-creator-shell/src/App.tsx

@@ -222,6 +222,12 @@ import {
 } from './features/project-workspace/agentRunTrace';
 import { DeveloperProjectPanels } from './features/project-workspace/DeveloperProjectPanels';
 import { DeveloperRuntimePanels } from './features/project-workspace/DeveloperRuntimePanels';
+import {
+  type DirectThreadConsumeResult,
+  type DirectThreadSubscriptionBootstrap,
+  emptyDirectThreadReducerState,
+  reduceDirectThreadEvents,
+} from './features/project-workspace/directThreadEvents';
 import type { DirectCodexUserContentPart } from './features/project-workspace/generated';
 import {
   appendMemoryContent,
@@ -771,6 +777,8 @@ export function App({
     lastSequence: number;
     receivedDirectUpdate: boolean;
   } | null>(null);
+  const directThreadSubscriptionIdRef = useRef<string | null>(null);
+  const directThreadReducerStateRef = useRef(emptyDirectThreadReducerState());
   const lastDirectCodexActivityRef = useRef<string | null>(null);
   const directCodexConversationTurnSequenceRef = useRef(0);
   const [projectSupervisorSessionId, setProjectSupervisorSessionId] = useState<
@@ -1905,6 +1913,118 @@ export function App({
     return () => window.clearInterval(timer);
   }, [chatAgentBusy, directCodexProductRuntime]);
 
+  useEffect(() => {
+    const projectPath = localProject?.projectPath ?? null;
+    const directInvoke = resolveTauriInvoke();
+    if (!directCodexProductRuntime || !projectPath || !directInvoke) {
+      directThreadSubscriptionIdRef.current = null;
+      directThreadReducerStateRef.current = emptyDirectThreadReducerState();
+      return;
+    }
+    let disposed = false;
+    let cleanup: (() => void) | null = null;
+
+    const applyReducerState = (
+      state: ReturnType<typeof emptyDirectThreadReducerState>,
+    ) => {
+      if (disposed) return;
+      directThreadReducerStateRef.current = state;
+      const running =
+        state.status === 'accepted' ||
+        state.status === 'running' ||
+        state.status === 'streaming' ||
+        state.status === 'finalizing';
+      setChatAgentBusy(running);
+      setDirectCodexStatus(state.status);
+      setDirectCodexProgress(state.progress);
+      setDirectCodexProgressUpdatedAt(Date.now());
+      if (state.accumulatedText) {
+        setDirectCodexTransientReply(state.accumulatedText);
+        directCodexTransientReplyRef.current = state.accumulatedText;
+        setDirectCodexTransientReplyUpdatedAt(Date.now());
+      }
+    };
+
+    const bootstrap = async () => {
+      const result = await directInvoke<DirectThreadSubscriptionBootstrap>(
+        'subscribe_direct_project_thread',
+        { projectPath },
+      );
+      if (disposed) return;
+      directThreadSubscriptionIdRef.current = result.subscriptionId;
+      const state = reduceDirectThreadEvents(
+        result.events,
+        emptyDirectThreadReducerState(),
+      );
+      applyReducerState(state);
+      if (state.turnId && !activeDirectCodexTurnRef.current) {
+        activeDirectCodexTurnRef.current = {
+          projectPath,
+          turnId: state.turnId,
+          lastSequence: state.lastSeq,
+          receivedDirectUpdate: true,
+        };
+        setDirectCodexProcessKey(`${projectPath}\u0000${state.turnId}`);
+      }
+    };
+
+    const consume = async () => {
+      const subscriptionId = directThreadSubscriptionIdRef.current;
+      if (!subscriptionId || disposed) return;
+      try {
+        const result = await directInvoke<DirectThreadConsumeResult>(
+          'consume_direct_project_thread',
+          { subscriptionId },
+        );
+        if (disposed) return;
+        const state = reduceDirectThreadEvents(
+          result.events,
+          directThreadReducerStateRef.current,
+        );
+        applyReducerState(state);
+      } catch (error) {
+        if (String(error).includes('SUBSCRIPTION_EXPIRED')) {
+          directThreadSubscriptionIdRef.current = null;
+          try {
+            await bootstrap();
+          } catch {
+            // A later project activation or notification will retry bootstrap.
+          }
+        }
+      }
+    };
+
+    const setup = async () => {
+      try {
+        const unlisten = await subscribeTauriEvent<{ subscriptionId: string }>(
+          'game-creator-direct-thread-notify',
+          (event) => {
+            if (
+              event.payload.subscriptionId ===
+              directThreadSubscriptionIdRef.current
+            ) {
+              void consume();
+            }
+          },
+        );
+        if (disposed) {
+          unlisten();
+          return;
+        }
+        cleanup = unlisten;
+        await bootstrap();
+      } catch {
+        // The history view remains usable when the runtime subscription is unavailable.
+      }
+    };
+    void setup();
+    return () => {
+      disposed = true;
+      cleanup?.();
+      directThreadSubscriptionIdRef.current = null;
+    };
+  }, [directCodexProductRuntime, localProject?.projectPath]);
+
   useEffect(() => {
     if (projectSupervisorOnly && !directCodexProductRuntime) {
       return;

apps/ai-game-creator-shell/src/features/project-workspace/directThreadEvents.ts

@@ -0,0 +1,120 @@
+export type DirectThreadRawEvent = {
+  seq: number;
+  type: string;
+  turnId: string;
+  itemId?: string;
+  payload: Record<string, unknown>;
+};
+
+export type DirectThreadSubscriptionBootstrap = {
+  subscriptionId: string;
+  lastCompletedItemId: string | null;
+  events: DirectThreadRawEvent[];
+};
+
+export type DirectThreadConsumeResult = {
+  events: DirectThreadRawEvent[];
+};
+
+export type DirectThreadReducerState = {
+  lastSeq: number;
+  turnId: string | null;
+  status:
+    | 'accepted'
+    | 'running'
+    | 'streaming'
+    | 'finalizing'
+    | 'completed'
+    | 'failed'
+    | null;
+  activeItemIds: Set<string>;
+  accumulatedText: string;
+  progress: string;
+};
+
+export const emptyDirectThreadReducerState = (): DirectThreadReducerState => ({
+  lastSeq: 0,
+  turnId: null,
+  status: null,
+  activeItemIds: new Set(),
+  accumulatedText: '',
+  progress: '',
+});
+
+function itemType(event: DirectThreadRawEvent) {
+  const value = event.payload.itemType;
+  return typeof value === 'string' ? value : '';
+}
+
+function activityDetail(type: string) {
+  switch (type) {
+    case 'mcpToolCall':
+      return '正在调用工具';
+    case 'commandExecution':
+      return '正在执行命令';
+    case 'fileChange':
+      return '正在写入文件';
+    case 'webSearch':
+      return '正在搜索资料';
+    case 'contextCompaction':
+      return '正在整理上下文';
+    default:
+      return '正在处理';
+  }
+}
+
+export function reduceDirectThreadEvent(
+  state: DirectThreadReducerState,
+  event: DirectThreadRawEvent,
+): DirectThreadReducerState {
+  if (!Number.isSafeInteger(event.seq) || event.seq <= state.lastSeq) {
+    return state;
+  }
+  const next: DirectThreadReducerState = {
+    ...state,
+    lastSeq: event.seq,
+    turnId: event.turnId || state.turnId,
+    activeItemIds: new Set(state.activeItemIds),
+  };
+  switch (event.type) {
+    case 'turn.started':
+      next.status = 'running';
+      next.progress = '正在处理';
+      break;
+    case 'item.started':
+      if (event.itemId) next.activeItemIds.add(event.itemId);
+      next.status = 'running';
+      next.progress = activityDetail(itemType(event));
+      break;
+    case 'item.delta': {
+      const delta = event.payload.delta;
+      if (typeof delta === 'string' && delta) {
+        next.accumulatedText += delta;
+        next.status = 'streaming';
+        next.progress = '正在生成回复';
+      }
+      break;
+    }
+    case 'item.completed':
+      if (event.itemId) next.activeItemIds.delete(event.itemId);
+      if (next.status === null) next.status = 'running';
+      break;
+    case 'turn.completed': {
+      const status = event.payload.status;
+      next.status = status === 'completed' ? 'completed' : 'failed';
+      next.progress =
+        status === 'completed' ? '正在提交回复' : '正在记录失败原因';
+      break;
+    }
+    default:
+      break;
+  }
+  return next;
+}
+
+export function reduceDirectThreadEvents(
+  events: DirectThreadRawEvent[],
+  initial = emptyDirectThreadReducerState(),
+) {
+  return events.reduce(reduceDirectThreadEvent, initial);
+}