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..49dab407c 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 () => {}; }, + subscribeAllTaskRuntime() { + 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 () => {}; }, + subscribeAllTaskRuntime() { + 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 () => {}; }, + subscribeAllTaskRuntime() { + 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 () => {}; }, + subscribeAllTaskRuntime() { + return () => {}; + }, }, "@/services/agentInputService": { submitTextInput() { @@ -5145,6 +5216,9 @@ test("shell-ball replays delivery.ready notifications that arrive before submit subscribeTaskUpdated() { return () => {}; }, + subscribeAllTaskRuntime() { + return () => {}; + }, }, "@/services/agentInputService": { submitTextInput() { @@ -5358,6 +5432,9 @@ test("shell-ball replays task.updated notifications that arrive before submit re taskUpdatedListener = callback; return () => {}; }, + subscribeAllTaskRuntime() { + return () => {}; + }, }, "@/services/agentInputService": { submitTextInput() { @@ -5476,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; @@ -5528,6 +5782,9 @@ test("shell-ball ignores untracked approval.pending notifications without a pend subscribeTaskUpdated() { return () => {}; }, + subscribeAllTaskRuntime() { + return () => {}; + }, }, "@/services/agentInputService": { submitTextInput() { @@ -5714,6 +5971,9 @@ test("shell-ball approval responses do not overwrite newer task subscription sta taskUpdatedListener = callback; return () => {}; }, + subscribeAllTaskRuntime() { + return () => {}; + }, }, "@/services/agentInputService": { submitTextInput() { @@ -5913,6 +6173,9 @@ test("shell-ball delivery.ready auto-opens tracked formal delivery results", asy deliveryReadyListener = callback; return () => {}; }, + subscribeAllTaskRuntime() { + return () => {}; + }, }, "@/features/dashboard/tasks/taskOutput.service": { openTaskDeliveryForTask(taskId: string) { @@ -6099,6 +6362,9 @@ test("shell-ball task-detail deliveries auto-open the dashboard detail view", as subscribeDeliveryReady() { return () => {}; }, + subscribeAllTaskRuntime() { + return () => {}; + }, }, "@/features/dashboard/tasks/taskOutput.service": { openTaskDeliveryForTask(taskId: string) { @@ -6895,13 +7161,16 @@ 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, /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, /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/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 372c6a7ef..381183ccd 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 { subscribeAllTaskRuntime, subscribeApprovalPending, subscribeDeliveryReady, subscribeTaskUpdated } from "@/rpc/subscriptions"; import { submitTextInput } from "@/services/agentInputService"; import { SHELL_BALL_PINNED_BUBBLE_WINDOW_FRAME, @@ -78,7 +87,13 @@ type QueuedDeliveryReadyNotification = { taskId: string; }; +type QueuedRuntimeNotification = { + payload: ShellBallRuntimeNotification; + taskId: string; +}; + type QueuedTaskUpdatedNotification = TaskUpdatedNotification; +type ShellBallRuntimeNotification = TaskRuntimeNotification | TaskSteeredNotification; type ShellBallTaskOutputServiceModule = { openTaskDeliveryForTask: (taskId: string, artifactId: string | undefined, source?: "rpc" | "mock") => Promise; performTaskOpenExecution: ( @@ -414,6 +429,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 @@ -706,6 +749,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. @@ -718,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. @@ -863,6 +911,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, @@ -897,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; @@ -915,6 +1006,7 @@ export function useShellBallCoordinator(input: ShellBallCoordinatorInput) { queuedApprovalPendingNotificationsRef.current.clear(); queuedTaskUpdatedNotificationsRef.current.clear(); queuedDeliveryReadyNotificationsRef.current.clear(); + queuedRuntimeNotificationsRef.current.clear(); } }; }, []); @@ -1876,6 +1968,26 @@ export function useShellBallCoordinator(input: ShellBallCoordinatorInput) { }); }, [appendDeliveryReadyBubble]); + useEffect(() => { + 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; + } + + appendRuntimeObservationBubble(payload.task_id, payload); + }); + }, [appendRuntimeObservationBubble]); + useEffect(() => { const currentWindow = getCurrentWindow(); const latestSnapshot = snapshot; 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); } })