From 0fd544aa67ceb2784358bdab46cb0b894f33ace5 Mon Sep 17 00:00:00 2001 From: gdemonc <1502689916@qq.com> Date: Mon, 27 Apr 2026 22:15:20 +0800 Subject: [PATCH 1/2] fix(shell-ball): listen for runtime task notifications --- .../shell-ball/shellBall.contract.test.ts | 94 ++++++++++++++++++- .../shell-ball/useShellBallCoordinator.ts | 91 +++++++++++++++++- 2 files changed, 181 insertions(+), 4 deletions(-) diff --git a/apps/desktop/src/features/shell-ball/shellBall.contract.test.ts b/apps/desktop/src/features/shell-ball/shellBall.contract.test.ts index 6e11fdcd8..c05e4082c 100644 --- a/apps/desktop/src/features/shell-ball/shellBall.contract.test.ts +++ b/apps/desktop/src/features/shell-ball/shellBall.contract.test.ts @@ -94,6 +94,7 @@ import { import { applyShellBallBubbleAction, createShellBallAgentBubbleItem, + createShellBallRuntimeObservationReply, shouldAutoOpenShellBallDeliveryResult, sortShellBallBubbleItemsByTimestamp, } from "./useShellBallCoordinator"; @@ -3987,6 +3988,64 @@ test("shell-ball auto-open helper only targets formal delivery types with native assert.equal(shouldAutoOpenShellBallDeliveryResult({ type: "reveal_in_folder" } as any), true); assert.equal(shouldAutoOpenShellBallDeliveryResult({ type: "result_page" } as any), true); }); + +test("shell-ball runtime observation helper keeps runtime hints lightweight", () => { + assert.equal( + createShellBallRuntimeObservationReply({ + task_id: "task-runtime-observation", + message: "Added another instruction.", + }), + "Added another instruction.", + ); + assert.equal( + createShellBallRuntimeObservationReply({ + task_id: "task-runtime-observation", + event: { + event_id: "evt_loop_retry_1", + run_id: "run-runtime", + task_id: "task-runtime-observation", + type: "loop.retrying", + level: "warn", + payload_json: "{}", + created_at: "2026-04-27T08:00:00.000Z", + }, + stop_reason: "network timeout", + }), + "Retrying the current task step after network timeout.", + ); + assert.equal( + createShellBallRuntimeObservationReply({ + task_id: "task-runtime-observation", + event: { + event_id: "evt_loop_failed_1", + run_id: "run-runtime", + task_id: "task-runtime-observation", + type: "loop.failed", + level: "error", + payload_json: "{}", + created_at: "2026-04-27T08:01:00.000Z", + }, + stop_reason: "rate limit", + }), + "Task runtime failed: rate limit. Open task detail for more context.", + ); + assert.equal( + createShellBallRuntimeObservationReply({ + task_id: "task-runtime-observation", + event: { + event_id: "evt_loop_started_1", + run_id: "run-runtime", + task_id: "task-runtime-observation", + type: "loop.started", + level: "info", + payload_json: "{}", + created_at: "2026-04-27T08:02:00.000Z", + }, + }), + null, + ); +}); + test("shell-ball coordinator bubble actions restore unpinned bubbles by timestamp then id", () => { const sourceItems: ShellBallBubbleItem[] = [ { @@ -4248,6 +4307,9 @@ test("shell-ball selected-text prompt stays below an existing intent bubble even subscribeDeliveryReady() { return () => {}; }, + subscribeTaskRuntime() { + return () => {}; + }, }, "../../platform/shellBallWindowController": { SHELL_BALL_PINNED_BUBBLE_WINDOW_FRAME: { width: 240, height: 140 }, @@ -4551,6 +4613,9 @@ test("shell-ball submit auto-opens formal delivery results through the shared de subscribeDeliveryReady() { return () => {}; }, + subscribeTaskRuntime() { + return () => {}; + }, }, "@/features/dashboard/tasks/taskOutput.service": { openTaskDeliveryForTask(taskId: string) { @@ -4750,6 +4815,9 @@ test("shell-ball voice submit reuses task tracking and task-detail auto-open flo subscribeDeliveryReady() { return () => {}; }, + subscribeTaskRuntime() { + return () => {}; + }, }, "@/features/dashboard/tasks/taskOutput.service": { openTaskDeliveryForTask(taskId: string) { @@ -4955,6 +5023,9 @@ test("shell-ball replays approval.pending notifications that arrive before submi subscribeTaskUpdated() { return () => {}; }, + subscribeTaskRuntime() { + return () => {}; + }, }, "@/services/agentInputService": { submitTextInput() { @@ -5145,6 +5216,9 @@ test("shell-ball replays delivery.ready notifications that arrive before submit subscribeTaskUpdated() { return () => {}; }, + subscribeTaskRuntime() { + return () => {}; + }, }, "@/services/agentInputService": { submitTextInput() { @@ -5358,6 +5432,9 @@ test("shell-ball replays task.updated notifications that arrive before submit re taskUpdatedListener = callback; return () => {}; }, + subscribeTaskRuntime() { + return () => {}; + }, }, "@/services/agentInputService": { submitTextInput() { @@ -5528,6 +5605,9 @@ test("shell-ball ignores untracked approval.pending notifications without a pend subscribeTaskUpdated() { return () => {}; }, + subscribeTaskRuntime() { + return () => {}; + }, }, "@/services/agentInputService": { submitTextInput() { @@ -5714,6 +5794,9 @@ test("shell-ball approval responses do not overwrite newer task subscription sta taskUpdatedListener = callback; return () => {}; }, + subscribeTaskRuntime() { + return () => {}; + }, }, "@/services/agentInputService": { submitTextInput() { @@ -5913,6 +5996,9 @@ test("shell-ball delivery.ready auto-opens tracked formal delivery results", asy deliveryReadyListener = callback; return () => {}; }, + subscribeTaskRuntime() { + return () => {}; + }, }, "@/features/dashboard/tasks/taskOutput.service": { openTaskDeliveryForTask(taskId: string) { @@ -6099,6 +6185,9 @@ test("shell-ball task-detail deliveries auto-open the dashboard detail view", as subscribeDeliveryReady() { return () => {}; }, + subscribeTaskRuntime() { + return () => {}; + }, }, "@/features/dashboard/tasks/taskOutput.service": { openTaskDeliveryForTask(taskId: string) { @@ -6895,13 +6984,14 @@ test("shell-ball window command routes through the formal screen task path", () assert.doesNotMatch(coordinatorSource, /createShellBallWindowContextReply/); }); -test("shell-ball coordinator subscribes to formal task and approval updates", () => { +test("shell-ball coordinator subscribes to formal task, approval, and runtime updates", () => { const coordinatorSource = readFileSync(resolve(desktopRoot, "src/features/shell-ball/useShellBallCoordinator.ts"), "utf8"); assert.match(coordinatorSource, /subscribeTaskUpdated\(\(payload\) => \{/); assert.match(coordinatorSource, /subscribeApprovalPending\(\(payload\) => \{/); + assert.match(coordinatorSource, /subscribeTaskRuntime\(activeShellBallTaskId, \(payload\) => \{/); assert.match(coordinatorSource, /syncShellBallVisualStateFromTaskStatus\(payload\.status\)/); - assert.match(coordinatorSource, /createShellBallApprovalPendingReply\(payload\.approval_request\)/); + assert.match(coordinatorSource, /approvalRequest: payload\.approval_request/); }); test("desktop tauri setup enables mouse activity tracking for dwell context", () => { diff --git a/apps/desktop/src/features/shell-ball/useShellBallCoordinator.ts b/apps/desktop/src/features/shell-ball/useShellBallCoordinator.ts index 372c6a7ef..2fd4b6be2 100644 --- a/apps/desktop/src/features/shell-ball/useShellBallCoordinator.ts +++ b/apps/desktop/src/features/shell-ball/useShellBallCoordinator.ts @@ -1,8 +1,17 @@ -import type { ApprovalDecision, ApprovalRequest, BubbleMessage, DeliveryResult, InputContext, TaskUpdatedNotification } from "@cialloclaw/protocol"; +import type { + ApprovalDecision, + ApprovalRequest, + BubbleMessage, + DeliveryResult, + InputContext, + TaskRuntimeNotification, + TaskSteeredNotification, + TaskUpdatedNotification, +} from "@cialloclaw/protocol"; import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { getCurrentWindow } from "@tauri-apps/api/window"; import { respondSecurityDetailed } from "@/rpc/methods"; -import { subscribeApprovalPending, subscribeDeliveryReady, subscribeTaskUpdated } from "@/rpc/subscriptions"; +import { subscribeApprovalPending, subscribeDeliveryReady, subscribeTaskRuntime, subscribeTaskUpdated } from "@/rpc/subscriptions"; import { submitTextInput } from "@/services/agentInputService"; import { SHELL_BALL_PINNED_BUBBLE_WINDOW_FRAME, @@ -79,6 +88,7 @@ type QueuedDeliveryReadyNotification = { }; type QueuedTaskUpdatedNotification = TaskUpdatedNotification; +type ShellBallRuntimeNotification = TaskRuntimeNotification | TaskSteeredNotification; type ShellBallTaskOutputServiceModule = { openTaskDeliveryForTask: (taskId: string, artifactId: string | undefined, source?: "rpc" | "mock") => Promise; performTaskOpenExecution: ( @@ -414,6 +424,34 @@ function createShellBallApprovalPendingReply(approvalRequest: ApprovalRequest) { return "Waiting for approval before the task can continue."; } +/** + * Runtime notifications stay observation-only in shell-ball. The formal task + * status still comes from task.updated, while selected runtime events become + * lightweight local bubbles for the current task conversation. + */ +export function createShellBallRuntimeObservationReply(payload: ShellBallRuntimeNotification) { + if ("message" in payload) { + const message = payload.message.trim(); + return message === "" ? null : message; + } + + const stopReason = payload.stop_reason?.trim(); + + if (payload.event.type === "loop.retrying") { + return stopReason === undefined || stopReason === "" + ? "Retrying the current task step." + : `Retrying the current task step after ${stopReason}.`; + } + + if (payload.event.type === "loop.failed") { + return stopReason === undefined || stopReason === "" + ? "Task runtime failed. Open task detail for more context." + : `Task runtime failed: ${stopReason}. Open task detail for more context.`; + } + + return null; +} + /** * Pending approval bubbles keep one approval id in shell-ball-local state so * the floating surface can submit the formal decision RPC without inventing a @@ -671,6 +709,7 @@ export function applyShellBallBubbleAction( export function useShellBallCoordinator(input: ShellBallCoordinatorInput) { const [bubbleItems, setBubbleItems] = useState(() => sortShellBallBubbleItemsByTimestamp(cloneShellBallBubbleItems(SHELL_BALL_LOCAL_BUBBLE_ITEMS))); + const [activeShellBallTaskId, setActiveShellBallTaskId] = useState(null); const appendedVoiceBubbleSequenceRef = useRef(0); const handledFinalizedSpeechPayloadRef = useRef(null); const bubbleTurnIndexRef = useRef(0); @@ -706,6 +745,7 @@ export function useShellBallCoordinator(input: ShellBallCoordinatorInput) { const detachedPinnedBubbleIdsRef = useRef(new Set()); const deliveryReadyBubbleKeysRef = useRef(new Set()); const approvalPendingBubbleKeysRef = useRef(new Set()); + const runtimeObservationBubbleKeysRef = useRef(new Set()); // Approval notifications can win the race against `agent.input.submit`. // Keep them task-scoped until the submit result binds the formal task id to // this shell-ball turn, then replay them into the local bubble timeline. @@ -863,6 +903,42 @@ export function useShellBallCoordinator(input: ShellBallCoordinatorInput) { void autoOpenShellBallDeliveryResultRef.current(input.taskId, input.deliveryResult); }, [allocateBubbleTurnIndex, bindTaskToBubbleTurn, getTaskBubbleTurnIndex]); + const appendRuntimeObservationBubble = useCallback((taskId: string, payload: ShellBallRuntimeNotification) => { + const bubbleText = createShellBallRuntimeObservationReply(payload); + if (bubbleText === null) { + return; + } + + const bubbleKey = "message" in payload + ? `${taskId}:task.steered:${bubbleText}` + : `${taskId}:${payload.event.event_id}`; + + if (runtimeObservationBubbleKeysRef.current.has(bubbleKey)) { + return; + } + + runtimeObservationBubbleKeysRef.current.add(bubbleKey); + + const turnIndex = getTaskBubbleTurnIndex(taskId) ?? allocateBubbleTurnIndex(); + bindTaskToBubbleTurn(taskId, turnIndex); + + setBubbleItems((currentItems) => + sortShellBallBubbleItemsByTimestamp([ + ...currentItems, + createShellBallTextBubbleItem({ + role: "agent", + text: bubbleText, + bubbleType: "status", + createdAt: "message" in payload ? new Date().toISOString() : payload.event.created_at, + taskId, + turnIndex, + turnPhase: 2, + }), + ]), + ); + revealBubbleRegionRef.current(); + }, [allocateBubbleTurnIndex, bindTaskToBubbleTurn, getTaskBubbleTurnIndex]); + const registerShellBallTask = useCallback(( taskId: string, turnIndex?: number, @@ -870,6 +946,7 @@ export function useShellBallCoordinator(input: ShellBallCoordinatorInput) { ) => { shellBallTaskIdsRef.current.add(taskId); activeShellBallTaskIdRef.current = taskId; + setActiveShellBallTaskId((currentTaskId) => currentTaskId === taskId ? currentTaskId : taskId); if (turnIndex !== undefined) { shellBallTaskTurnIndexRef.current.set(taskId, turnIndex); @@ -1876,6 +1953,16 @@ export function useShellBallCoordinator(input: ShellBallCoordinatorInput) { }); }, [appendDeliveryReadyBubble]); + useEffect(() => { + if (activeShellBallTaskId === null) { + return; + } + + return subscribeTaskRuntime(activeShellBallTaskId, (payload) => { + appendRuntimeObservationBubble(activeShellBallTaskId, payload); + }); + }, [activeShellBallTaskId, appendRuntimeObservationBubble]); + useEffect(() => { const currentWindow = getCurrentWindow(); const latestSnapshot = snapshot; From 3d36dbe89f1fde3dfa90ac66db2d2b190fbdbc81 Mon Sep 17 00:00:00 2001 From: xgopilot Date: Tue, 28 Apr 2026 05:08:18 +0000 Subject: [PATCH 2/2] fix(shell-ball): buffer early runtime notifications Generated with [codeagent](https://github.com/qbox/codeagent) Co-authored-by: gdemonc <146809967+gdemonc@users.noreply.github.com> --- .../shell-ball/shellBall.contract.test.ts | 201 +++++++++++++++++- .../shell-ball/test-stubs/rpcSubscriptions.ts | 4 + .../shell-ball/useShellBallCoordinator.ts | 45 +++- apps/desktop/src/rpc/subscriptions.ts | 13 +- 4 files changed, 241 insertions(+), 22 deletions(-) diff --git a/apps/desktop/src/features/shell-ball/shellBall.contract.test.ts b/apps/desktop/src/features/shell-ball/shellBall.contract.test.ts index c05e4082c..49dab407c 100644 --- a/apps/desktop/src/features/shell-ball/shellBall.contract.test.ts +++ b/apps/desktop/src/features/shell-ball/shellBall.contract.test.ts @@ -4307,7 +4307,7 @@ test("shell-ball selected-text prompt stays below an existing intent bubble even subscribeDeliveryReady() { return () => {}; }, - subscribeTaskRuntime() { + subscribeAllTaskRuntime() { return () => {}; }, }, @@ -4613,7 +4613,7 @@ test("shell-ball submit auto-opens formal delivery results through the shared de subscribeDeliveryReady() { return () => {}; }, - subscribeTaskRuntime() { + subscribeAllTaskRuntime() { return () => {}; }, }, @@ -4815,7 +4815,7 @@ test("shell-ball voice submit reuses task tracking and task-detail auto-open flo subscribeDeliveryReady() { return () => {}; }, - subscribeTaskRuntime() { + subscribeAllTaskRuntime() { return () => {}; }, }, @@ -5023,7 +5023,7 @@ test("shell-ball replays approval.pending notifications that arrive before submi subscribeTaskUpdated() { return () => {}; }, - subscribeTaskRuntime() { + subscribeAllTaskRuntime() { return () => {}; }, }, @@ -5216,7 +5216,7 @@ test("shell-ball replays delivery.ready notifications that arrive before submit subscribeTaskUpdated() { return () => {}; }, - subscribeTaskRuntime() { + subscribeAllTaskRuntime() { return () => {}; }, }, @@ -5432,7 +5432,7 @@ test("shell-ball replays task.updated notifications that arrive before submit re taskUpdatedListener = callback; return () => {}; }, - subscribeTaskRuntime() { + subscribeAllTaskRuntime() { return () => {}; }, }, @@ -5553,6 +5553,183 @@ test("shell-ball replays task.updated notifications that arrive before submit re assert.equal(useShellBallStore.getState().visualState, "waiting_auth"); }); +test("shell-ball replays runtime notifications that arrive before submit resolves", async () => { + const coordinatorSource = readFileSync(resolve(desktopRoot, "src/features/shell-ball/useShellBallCoordinator.ts"), "utf8"); + let runtimeListener: ((payload: { + task_id: string; + message: string; + }) => void) | null = null; + let resolveSubmit: ((value: { + task: { + task_id: string; + status: "processing"; + }; + bubble_message: null; + delivery_result: null; + }) => void) | null = null; + const reactRuntime = createImmediateShellBallReactRuntime(); + + await withSourceModuleRuntime( + resolve(desktopRoot, "src/features/shell-ball/useShellBallCoordinator.ts"), + { + react: reactRuntime.react, + "@tauri-apps/api/window": { + getCurrentWindow() { + return { + label: shellBallWindowLabels.ball, + listen() { + return Promise.resolve(() => {}); + }, + onMoved() { + return Promise.resolve(() => {}); + }, + onResized() { + return Promise.resolve(() => {}); + }, + outerPosition() { + return Promise.resolve({ toLogical: () => ({ x: 0, y: 0 }) }); + }, + outerSize() { + return Promise.resolve({ toLogical: () => ({ width: 124, height: 104 }) }); + }, + scaleFactor() { + return Promise.resolve(1); + }, + }; + }, + }, + "@/rpc/subscriptions": { + subscribeAllTaskRuntime(callback: typeof runtimeListener) { + runtimeListener = callback; + return () => {}; + }, + subscribeApprovalPending() { + return () => {}; + }, + subscribeDeliveryReady() { + return () => {}; + }, + subscribeTaskUpdated() { + return () => {}; + }, + }, + "@/services/agentInputService": { + submitTextInput() { + return new Promise((resolve) => { + resolveSubmit = resolve as typeof resolveSubmit; + }); + }, + }, + "@/features/dashboard/tasks/taskOutput.service": { + openTaskDeliveryForTask() { + return Promise.resolve(null); + }, + resolveTaskOpenExecutionPlan(): { + feedback: string; + mode: "task_detail" | "open_url" | "open_local_path" | "reveal_local_path" | "copy_path"; + path: string | null; + taskId: string | null; + url: string | null; + } { + return { + feedback: "open task detail", + mode: "task_detail" as const, + path: null, + taskId: null, + url: null, + }; + }, + performTaskOpenExecution() { + return Promise.resolve("open task detail"); + }, + }, + "@/features/dashboard/shared/dashboardTaskDetailNavigation": { + requestDashboardTaskDetailOpen() { + return Promise.resolve(); + }, + }, + "../../platform/shellBallWindowController": { + SHELL_BALL_PINNED_BUBBLE_WINDOW_FRAME: { width: 240, height: 140 }, + closeShellBallPinnedBubbleWindow() { + return Promise.resolve(); + }, + emitToShellBallWindowLabel() { + return Promise.resolve(); + }, + getShellBallPinnedBubbleIdFromLabel(): string | null { + return null; + }, + getShellBallPinnedBubbleWindowAnchor() { + return { x: 0, y: 0 }; + }, + getShellBallPinnedBubbleWindowLabel(bubbleId: string) { + return `shell-ball-bubble-pinned-${bubbleId}`; + }, + openShellBallPinnedBubbleWindow() { + return Promise.resolve(); + }, + setShellBallPinnedBubbleWindowVisible() { + return Promise.resolve(); + }, + shellBallWindowLabels, + }, + "./useShellBallWindowMetrics": { + getShellBallBubbleAnchor() { + return { x: 0, y: 0 }; + }, + }, + }, + async (moduleExports) => { + const { useShellBallCoordinator } = moduleExports as { + useShellBallCoordinator: typeof import("./useShellBallCoordinator").useShellBallCoordinator; + }; + + const { handlePrimaryAction } = useShellBallCoordinator({ + visualState: "hover_input", + regionActive: false, + inputValue: "截屏", + inputFocused: true, + finalizedSpeechPayload: null, + voicePreview: null, + voiceHintMode: "hidden", + setInputValue: () => {}, + onFinalizedSpeechHandled: () => {}, + onRegionEnter: () => {}, + onRegionLeave: () => {}, + onInputHoverChange: () => {}, + onInputFocusChange: () => {}, + onSubmitText: async () => null, + onAttachFile: () => {}, + onPrimaryClick: () => {}, + }); + + const submitPromise = handlePrimaryAction("submit"); + await flushAsyncEffects(); + + runtimeListener?.({ + task_id: "task-runtime-race", + message: "Added another instruction.", + }); + + resolveSubmit?.({ + task: { + task_id: "task-runtime-race", + status: "processing", + }, + bubble_message: null, + delivery_result: null, + }); + + await submitPromise; + await flushAsyncEffects(); + }, + ); + + assert.match(coordinatorSource, /const queuedRuntimeNotificationsRef = useRef\(new Map\(\)\);/); + const runtimeBubble = reactRuntime.getBubbleItems().find((item) => item.bubble.task_id === "task-runtime-race" && item.bubble.text === "Added another instruction."); + assert.ok(runtimeBubble); +}); + test("shell-ball ignores untracked approval.pending notifications without a pending submit", async () => { let approvalPendingListener: ((payload: { task_id: string; @@ -5605,7 +5782,7 @@ test("shell-ball ignores untracked approval.pending notifications without a pend subscribeTaskUpdated() { return () => {}; }, - subscribeTaskRuntime() { + subscribeAllTaskRuntime() { return () => {}; }, }, @@ -5794,7 +5971,7 @@ test("shell-ball approval responses do not overwrite newer task subscription sta taskUpdatedListener = callback; return () => {}; }, - subscribeTaskRuntime() { + subscribeAllTaskRuntime() { return () => {}; }, }, @@ -5996,7 +6173,7 @@ test("shell-ball delivery.ready auto-opens tracked formal delivery results", asy deliveryReadyListener = callback; return () => {}; }, - subscribeTaskRuntime() { + subscribeAllTaskRuntime() { return () => {}; }, }, @@ -6185,7 +6362,7 @@ test("shell-ball task-detail deliveries auto-open the dashboard detail view", as subscribeDeliveryReady() { return () => {}; }, - subscribeTaskRuntime() { + subscribeAllTaskRuntime() { return () => {}; }, }, @@ -6989,7 +7166,9 @@ test("shell-ball coordinator subscribes to formal task, approval, and runtime up assert.match(coordinatorSource, /subscribeTaskUpdated\(\(payload\) => \{/); assert.match(coordinatorSource, /subscribeApprovalPending\(\(payload\) => \{/); - assert.match(coordinatorSource, /subscribeTaskRuntime\(activeShellBallTaskId, \(payload\) => \{/); + assert.match(coordinatorSource, /subscribeAllTaskRuntime\(\(payload\) => \{/); + assert.match(coordinatorSource, /const queuedRuntimeNotificationsRef = useRef\(new Map\(\)\);/); + assert.match(coordinatorSource, /queuedRuntimeNotifications\.forEach\(\(notification\) => \{\s*appendRuntimeObservationBubble\(notification\.taskId, notification\.payload\);/); assert.match(coordinatorSource, /syncShellBallVisualStateFromTaskStatus\(payload\.status\)/); assert.match(coordinatorSource, /approvalRequest: payload\.approval_request/); }); diff --git a/apps/desktop/src/features/shell-ball/test-stubs/rpcSubscriptions.ts b/apps/desktop/src/features/shell-ball/test-stubs/rpcSubscriptions.ts index b0db7dfe5..322bd3b94 100644 --- a/apps/desktop/src/features/shell-ball/test-stubs/rpcSubscriptions.ts +++ b/apps/desktop/src/features/shell-ball/test-stubs/rpcSubscriptions.ts @@ -34,3 +34,7 @@ export function subscribeTaskUpdated(_listener?: (payload: TaskUpdatedNotificati export function subscribeTaskRuntime(_taskId?: string, _listener?: (payload: TaskSteeredNotification | TaskRuntimeNotification) => void) { return noopUnsubscribe(); } + +export function subscribeAllTaskRuntime(_listener?: (payload: TaskSteeredNotification | TaskRuntimeNotification) => void) { + return noopUnsubscribe(); +} diff --git a/apps/desktop/src/features/shell-ball/useShellBallCoordinator.ts b/apps/desktop/src/features/shell-ball/useShellBallCoordinator.ts index 2fd4b6be2..381183ccd 100644 --- a/apps/desktop/src/features/shell-ball/useShellBallCoordinator.ts +++ b/apps/desktop/src/features/shell-ball/useShellBallCoordinator.ts @@ -11,7 +11,7 @@ import type { import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { getCurrentWindow } from "@tauri-apps/api/window"; import { respondSecurityDetailed } from "@/rpc/methods"; -import { subscribeApprovalPending, subscribeDeliveryReady, subscribeTaskRuntime, subscribeTaskUpdated } from "@/rpc/subscriptions"; +import { subscribeAllTaskRuntime, subscribeApprovalPending, subscribeDeliveryReady, subscribeTaskUpdated } from "@/rpc/subscriptions"; import { submitTextInput } from "@/services/agentInputService"; import { SHELL_BALL_PINNED_BUBBLE_WINDOW_FRAME, @@ -87,6 +87,11 @@ type QueuedDeliveryReadyNotification = { taskId: string; }; +type QueuedRuntimeNotification = { + payload: ShellBallRuntimeNotification; + taskId: string; +}; + type QueuedTaskUpdatedNotification = TaskUpdatedNotification; type ShellBallRuntimeNotification = TaskRuntimeNotification | TaskSteeredNotification; type ShellBallTaskOutputServiceModule = { @@ -709,7 +714,6 @@ export function applyShellBallBubbleAction( export function useShellBallCoordinator(input: ShellBallCoordinatorInput) { const [bubbleItems, setBubbleItems] = useState(() => sortShellBallBubbleItemsByTimestamp(cloneShellBallBubbleItems(SHELL_BALL_LOCAL_BUBBLE_ITEMS))); - const [activeShellBallTaskId, setActiveShellBallTaskId] = useState(null); const appendedVoiceBubbleSequenceRef = useRef(0); const handledFinalizedSpeechPayloadRef = useRef(null); const bubbleTurnIndexRef = useRef(0); @@ -758,6 +762,10 @@ export function useShellBallCoordinator(input: ShellBallCoordinatorInput) { // the formal task id locally. Buffer them with the same task-scoped replay // path so shell-ball still shows the result bubble and open flow. const queuedDeliveryReadyNotificationsRef = useRef(new Map()); + // Runtime notifications can also race ahead of the submit response. Keep + // them task-scoped and replay them once shell-ball has registered the formal + // task id for the active conversation turn. + const queuedRuntimeNotificationsRef = useRef(new Map()); // Only shell-ball submissions that are still waiting for their formal task id // are allowed to buffer approval notifications. This keeps unrelated desktop // approvals from lingering in shell-ball memory forever. @@ -946,7 +954,6 @@ export function useShellBallCoordinator(input: ShellBallCoordinatorInput) { ) => { shellBallTaskIdsRef.current.add(taskId); activeShellBallTaskIdRef.current = taskId; - setActiveShellBallTaskId((currentTaskId) => currentTaskId === taskId ? currentTaskId : taskId); if (turnIndex !== undefined) { shellBallTaskTurnIndexRef.current.set(taskId, turnIndex); @@ -974,7 +981,14 @@ export function useShellBallCoordinator(input: ShellBallCoordinatorInput) { queuedDeliveryNotifications.forEach((notification) => { appendDeliveryReadyBubble(notification); }); - }, [appendApprovalPendingBubble, appendDeliveryReadyBubble]); + + const queuedRuntimeNotifications = queuedRuntimeNotificationsRef.current.get(taskId) ?? []; + queuedRuntimeNotificationsRef.current.delete(taskId); + + queuedRuntimeNotifications.forEach((notification) => { + appendRuntimeObservationBubble(notification.taskId, notification.payload); + }); + }, [appendApprovalPendingBubble, appendDeliveryReadyBubble, appendRuntimeObservationBubble]); const beginPendingShellBallTaskRegistration = useCallback(() => { pendingShellBallTaskRegistrationsRef.current += 1; @@ -992,6 +1006,7 @@ export function useShellBallCoordinator(input: ShellBallCoordinatorInput) { queuedApprovalPendingNotificationsRef.current.clear(); queuedTaskUpdatedNotificationsRef.current.clear(); queuedDeliveryReadyNotificationsRef.current.clear(); + queuedRuntimeNotificationsRef.current.clear(); } }; }, []); @@ -1954,14 +1969,24 @@ export function useShellBallCoordinator(input: ShellBallCoordinatorInput) { }, [appendDeliveryReadyBubble]); useEffect(() => { - if (activeShellBallTaskId === null) { - return; - } + return subscribeAllTaskRuntime((payload) => { + if (!shellBallTaskIdsRef.current.has(payload.task_id)) { + if (pendingShellBallTaskRegistrationsRef.current === 0) { + return; + } + + const queuedNotifications = queuedRuntimeNotificationsRef.current.get(payload.task_id) ?? []; + queuedNotifications.push({ + payload, + taskId: payload.task_id, + }); + queuedRuntimeNotificationsRef.current.set(payload.task_id, queuedNotifications); + return; + } - return subscribeTaskRuntime(activeShellBallTaskId, (payload) => { - appendRuntimeObservationBubble(activeShellBallTaskId, payload); + appendRuntimeObservationBubble(payload.task_id, payload); }); - }, [activeShellBallTaskId, appendRuntimeObservationBubble]); + }, [appendRuntimeObservationBubble]); useEffect(() => { const currentWindow = getCurrentWindow(); diff --git a/apps/desktop/src/rpc/subscriptions.ts b/apps/desktop/src/rpc/subscriptions.ts index a9554487d..25bc5e63d 100644 --- a/apps/desktop/src/rpc/subscriptions.ts +++ b/apps/desktop/src/rpc/subscriptions.ts @@ -193,6 +193,17 @@ export function subscribeDeliveryReady(onMessage: (payload: DeliveryReadyNotific // loop.* notifications without promoting these runtime payloads into formal // frontend state. Callers still re-read the task-centric queries after each hit. export function subscribeTaskRuntime(taskId: string, onMessage: (payload: TaskSteeredNotification | TaskRuntimeNotification) => void) { + return subscribeAllTaskRuntime((payload) => { + if (payload.task_id === taskId) { + onMessage(payload); + } + }); +} + +// subscribeAllTaskRuntime is the global runtime-notification stream for +// frontend surfaces that must buffer task-scoped observations before the +// formal task id is locally registered. +export function subscribeAllTaskRuntime(onMessage: (payload: TaskSteeredNotification | TaskRuntimeNotification) => void) { const bridge = window.__CIALLOCLAW_NAMED_PIPE__; if (!bridge?.subscribe) { @@ -221,7 +232,7 @@ export function subscribeTaskRuntime(taskId: string, onMessage: (payload: TaskSt } const message = payload as { params?: TaskSteeredNotification | TaskRuntimeNotification }; - if (message.params?.task_id === taskId) { + if (message.params) { onMessage(message.params); } })