diff --git a/wework/DESIGN.md b/wework/DESIGN.md index bad1498d1b..0920385832 100644 --- a/wework/DESIGN.md +++ b/wework/DESIGN.md @@ -474,6 +474,10 @@ The active-conversation capture is also normative: multiple markers may be active when content from multiple turns is visible. - The bottom Composer shares the thread column and stays visible. It uses the same input hierarchy as home but without the home project-selector layer. +- Runtime stream lifecycle events update only the affected task's in-memory + status. They must not refresh the whole sidebar work list. A generated task + title is authoritative over older list requests already in flight until the + local executor confirms the same title. - When opening, closing, or resizing a side panel reflows conversation content, preserve the reader's visible message or content anchor. Continue following the bottom only when the reader was already at the bottom before the reflow. diff --git a/wework/src/features/workbench/WorkbenchProvider.test.tsx b/wework/src/features/workbench/WorkbenchProvider.test.tsx index be4673c386..2bd51aa40c 100644 --- a/wework/src/features/workbench/WorkbenchProvider.test.tsx +++ b/wework/src/features/workbench/WorkbenchProvider.test.tsx @@ -2047,7 +2047,7 @@ describe('WorkbenchProvider runtime tasks', () => { await waitFor(() => expect(getComposerApps()).toEqual([])) }) - test('keeps a background runtime task settled when its terminal refresh is stale', async () => { + test('settles a background runtime task while rejecting a stale running snapshot', async () => { let backgroundStreamHandlers: ChatStreamHandlers | null = null const subscribe = vi.fn((handlers: ChatStreamHandlers) => { if (hasRuntimeStreamHandler(handlers)) { @@ -2085,12 +2085,7 @@ describe('WorkbenchProvider runtime tasks', () => { ], totalTasks: 1, }) - const staleTerminalRefresh = deferred() - const listRuntimeWork = vi - .fn() - .mockResolvedValue(runningRuntimeWork) - .mockResolvedValueOnce(runningRuntimeWork) - .mockImplementationOnce(() => staleTerminalRefresh.promise) + const listRuntimeWork = vi.fn().mockResolvedValue(runningRuntimeWork) const runtimeWorkApi = createRuntimeWorkApiMock({ listRuntimeWork }) const services = createWorkbenchServices({ runtimeWorkApi: runtimeWorkApi as WorkbenchServices['runtimeWorkApi'], @@ -2115,14 +2110,10 @@ describe('WorkbenchProvider runtime tasks', () => { }) }) - await waitFor(() => expect(listRuntimeWork).toHaveBeenCalledTimes(2)) - await act(async () => { - staleTerminalRefresh.resolve(runningRuntimeWork) - await staleTerminalRefresh.promise - }) await waitFor(() => expect(screen.getByTestId('runtime-running-task-ids')).toHaveTextContent('none') ) + expect(listRuntimeWork).toHaveBeenCalledTimes(2) }) test('settles guidance applied while its runtime pane is in the background', async () => { @@ -2667,6 +2658,83 @@ describe('WorkbenchProvider runtime tasks', () => { ) }) + test('does not let a stale cloud refresh roll back a generated runtime task title', async () => { + let streamHandlers: ChatStreamHandlers = {} + const subscribe = vi.fn((handlers: ChatStreamHandlers) => { + if (hasRuntimeStreamHandler(handlers)) streamHandlers = handlers + return vi.fn() + }) + const cloudRuntimeWork = deferred() + const localRuntimeWork = createRuntimeWork({ + projects: [ + { + project: { id: 7, name: 'Wegent' }, + deviceWorkspaces: [ + { + id: 22, + projectId: 7, + deviceId: 'device-1', + deviceName: 'Project Device', + deviceStatus: 'online', + workspacePath: '/workspace/project-alpha', + mapped: true, + available: true, + tasks: [ + { + taskId: 'runtime-a', + workspacePath: '/workspace/project-alpha', + title: '解决冲突', + runtime: 'codex', + }, + ], + }, + ], + totalTasks: 1, + }, + ], + totalTasks: 1, + }) + const services = createWorkbenchServices({ + runtimeWorkApi: createRuntimeWorkApiMock({ + listRuntimeWork: vi.fn().mockResolvedValue(localRuntimeWork), + }), + chatStream: { + subscribe, + } as unknown as WorkbenchServices['chatStream'], + cloudBackgroundApi: { + listTeams: vi.fn().mockResolvedValue([]), + listDevices: vi.fn().mockResolvedValue([]), + listRuntimeWork: vi.fn(() => cloudRuntimeWork.promise), + }, + }) + + renderWorkbench(, services) + + await waitFor(() => + expect(screen.getByTestId('runtime-task-titles')).toHaveTextContent('解决冲突') + ) + await waitFor(() => expect(streamHandlers.onRuntimeTaskTitleUpdated).toBeDefined()) + + act(() => { + streamHandlers.onRuntimeTaskTitleUpdated?.({ + taskId: 'runtime-a', + subtaskId: 'friendly-title', + deviceId: 'device-1', + title: '解决分支冲突', + }) + }) + expect(screen.getByTestId('runtime-task-titles')).toHaveTextContent('解决分支冲突') + + await act(async () => { + cloudRuntimeWork.resolve(localRuntimeWork) + }) + + await waitFor(() => + expect(screen.getByTestId('runtime-task-titles')).toHaveTextContent('解决分支冲突') + ) + expect(screen.getByTestId('runtime-task-titles')).not.toHaveTextContent('解决冲突') + }) + test('cancels an in-flight cloud sync before a manual device refresh', async () => { const runtimeWork = deferred() const manualDevices = deferred() @@ -8976,7 +9044,7 @@ describe('WorkbenchProvider runtime tasks', () => { expect(sendRuntimeMessage).not.toHaveBeenCalled() }) - test('refreshes runtime work when the current runtime task starts streaming', async () => { + test('marks the current runtime task running without refreshing runtime work', async () => { let streamHandlers: Parameters[0] | null = null const subscribe = vi.fn(handlers => { if (hasRuntimeStreamHandler(handlers)) streamHandlers = handlers @@ -9010,7 +9078,10 @@ describe('WorkbenchProvider runtime tasks', () => { }) }) - await waitFor(() => expect(listRuntimeWork).toHaveBeenCalledTimes(callsBeforeStart + 1)) + await waitFor(() => + expect(screen.getByTestId('current-runtime-task-running')).toHaveTextContent('running') + ) + expect(listRuntimeWork).toHaveBeenCalledTimes(callsBeforeStart) }) test('hides the runtime goal when the settled task reports the goal complete', async () => { @@ -9649,37 +9720,84 @@ describe('WorkbenchProvider runtime tasks', () => { }) const updateTaskTrackingStatus = vi.fn().mockResolvedValue(null) const updateTaskTrackingTitle = vi.fn().mockResolvedValue(null) - const runtimeWorkApi = createRuntimeWorkApiMock({ - listRuntimeWork: vi.fn().mockResolvedValue( - createRuntimeWork({ - projects: [ + const initialRuntimeWork = createRuntimeWork({ + projects: [ + { + project: { id: 7, name: 'Wegent' }, + deviceWorkspaces: [ { - project: { id: 7, name: 'Wegent' }, - deviceWorkspaces: [ + deviceId: 'device-1', + deviceName: 'Project Device', + deviceStatus: 'online', + workspacePath: '/workspace/project-alpha', + mapped: true, + available: true, + tasks: [ { - deviceId: 'device-1', - deviceName: 'Project Device', - deviceStatus: 'online', + taskId: 'runtime-a', workspacePath: '/workspace/project-alpha', - mapped: true, - available: true, - tasks: [ - { - taskId: 'runtime-a', - workspacePath: '/workspace/project-alpha', - title: 'Runtime A', - runtime: 'codex', - running: false, - status: 'active', - }, - ], + title: 'Runtime A', + runtime: 'codex', + running: false, + status: 'active', + }, + { + taskId: 'runtime-b', + workspacePath: '/workspace/project-alpha', + title: 'Runtime B', + runtime: 'codex', + running: false, + status: 'done', }, ], }, ], - totalTasks: 1, - }) - ), + }, + ], + totalTasks: 2, + }) + const settledRuntimeWork = createRuntimeWork({ + projects: [ + { + project: { id: 7, name: 'Wegent' }, + deviceWorkspaces: [ + { + deviceId: 'device-1', + deviceName: 'Project Device', + deviceStatus: 'online', + workspacePath: '/workspace/project-alpha', + mapped: true, + available: true, + tasks: [ + { + taskId: 'runtime-a', + workspacePath: '/workspace/project-alpha', + title: 'Runtime A', + runtime: 'codex', + running: false, + status: 'done', + }, + { + taskId: 'runtime-b', + workspacePath: '/workspace/project-alpha', + title: 'Stale Runtime B', + runtime: 'codex', + running: false, + status: 'done', + }, + ], + }, + ], + }, + ], + totalTasks: 2, + }) + const listRuntimeWork = vi + .fn() + .mockResolvedValueOnce(initialRuntimeWork) + .mockResolvedValue(settledRuntimeWork) + const runtimeWorkApi = createRuntimeWorkApiMock({ + listRuntimeWork, }) const services = createWorkbenchServices({ runtimeWorkApi: runtimeWorkApi as WorkbenchServices['runtimeWorkApi'], @@ -9691,8 +9809,15 @@ describe('WorkbenchProvider runtime tasks', () => { } as unknown as WorkbenchServices['projectSpaceApis'], }) - renderWorkbench(, services) + renderWorkbench( + <> + + + , + services + ) await waitFor(() => expect(streamHandlers.onChatStart).toBeDefined()) + await waitFor(() => expect(listRuntimeWork).toHaveBeenCalledTimes(1)) act(() => { streamHandlers.onChatStart?.({ @@ -9710,34 +9835,39 @@ describe('WorkbenchProvider runtime tasks', () => { ) act(() => { - streamHandlers.onChatDone?.({ + streamHandlers.onRuntimeTaskTitleUpdated?.({ taskId: 'runtime-a', - subtaskId: '101', + subtaskId: 'friendly-title', deviceId: 'device-1', - result: { value: 'done' }, + title: '修复登录回调', }) }) await waitFor(() => - expect(updateTaskTrackingStatus).toHaveBeenCalledWith( + expect(updateTaskTrackingTitle).toHaveBeenCalledWith( expect.objectContaining({ deviceId: 'device-1', taskId: 'runtime-a' }), - 'succeeded' + '修复登录回调' ) ) act(() => { - streamHandlers.onRuntimeTaskTitleUpdated?.({ + streamHandlers.onChatDone?.({ taskId: 'runtime-a', - subtaskId: 'friendly-title', + subtaskId: '101', deviceId: 'device-1', - title: '修复登录回调', + result: { value: 'done' }, }) }) await waitFor(() => - expect(updateTaskTrackingTitle).toHaveBeenCalledWith( + expect(updateTaskTrackingStatus).toHaveBeenCalledWith( expect.objectContaining({ deviceId: 'device-1', taskId: 'runtime-a' }), - '修复登录回调' + 'succeeded' ) ) + await waitFor(() => expect(listRuntimeWork).toHaveBeenCalledTimes(2)) + expect(screen.getByTestId('runtime-local-task-titles')).toHaveTextContent( + '修复登录回调|Runtime B' + ) + expect(screen.getByTestId('runtime-local-task-titles')).not.toHaveTextContent('Stale Runtime B') }) test('sends queued runtime messages when the task becomes idle', async () => { @@ -9861,7 +9991,7 @@ describe('WorkbenchProvider runtime tasks', () => { await act(async () => { streamHandlers.onChatDone?.({ taskId: 'runtime-a', - subtaskId: '101', + subtaskId: 'provider-turn-101', deviceId: 'device-1', result: { value: 'done' }, }) diff --git a/wework/src/features/workbench/WorkbenchProvider.tsx b/wework/src/features/workbench/WorkbenchProvider.tsx index d4a9eaf751..5e8e9db69e 100644 --- a/wework/src/features/workbench/WorkbenchProvider.tsx +++ b/wework/src/features/workbench/WorkbenchProvider.tsx @@ -762,7 +762,11 @@ export function WorkbenchProvider({ markRuntimeProjectRemoved, clearRuntimeProjectRemoval, refreshWorkLists, + refreshRuntimeTask, refreshDevices, + updateLocalRuntimeTaskExecution, + updateLocalRuntimeTaskSnapshot, + updateLocalRuntimeTaskTitle, getRemoteDeviceStartupCommand, } = useWorkbenchDataRefresh({ user, @@ -1514,14 +1518,21 @@ export function WorkbenchProvider({ ) => settleRuntimeConversationGuidance(address, payload) ) const stableRefreshWorkLists = useStableEvent(refreshWorkLists) - const refreshRuntimeWorkLists = useStableEvent((address: RuntimeTaskAddress) => { - void stableRefreshWorkLists().catch(error => { - console.warn('[Wework] Runtime work list refresh failed', { - deviceId: address.deviceId, - taskId: address.taskId, - error, + const syncRuntimeTaskSnapshot = useStableEvent((address: RuntimeTaskAddress) => { + const expectedLifecycle = lifecycleStore.getTask(address) + void refreshRuntimeTask(address) + .then(task => { + if (task && lifecycleStore.syncRuntimeTask(address, task, expectedLifecycle)) { + updateLocalRuntimeTaskSnapshot(address, task) + } + }) + .catch(error => { + console.warn('[Wework] Runtime task snapshot sync failed', { + deviceId: address.deviceId, + taskId: address.taskId, + error, + }) }) - }) }) const syncRuntimeTaskTitle = useStableEvent((address: RuntimeTaskAddress, title: string) => { const normalizedTitle = title.trim() @@ -1571,6 +1582,13 @@ export function WorkbenchProvider({ settleRuntimeConversationAcceptedMessage(address) markRuntimeConversationAssistantStarted(address) lifecycleStore.turnStarted(address, turnId) + updateLocalRuntimeTaskExecution(address, true, 'active') + dispatch({ + type: 'runtime_task_execution_updated', + address, + running: true, + status: 'active', + }) aiGenerationTelemetry.onAssistantStart(address, turnId) }, onAssistantFirstToken: (address, turnId) => { @@ -1581,16 +1599,36 @@ export function WorkbenchProvider({ }, onAssistantSettled: (address, turnId, outcome) => { settleRuntimeConversationSubagents(address) - lifecycleStore.turnSettled(address, turnId, outcome) + // Runtime providers may replace the provisional subtask ID with their canonical + // turn ID while streaming. A terminal event is already scoped to one task, so it + // must settle that task even when the provider-facing ID changed. + lifecycleStore.turnSettled(address, null, outcome) + const running = lifecycleStore.getTask(address)?.derived.isRunning ?? false + const status = running + ? 'active' + : outcome === 'succeeded' + ? 'done' + : outcome === 'failed' + ? 'failed' + : 'cancelled' + updateLocalRuntimeTaskExecution(address, running, status) + dispatch({ + type: 'runtime_task_execution_updated', + address, + running, + status, + }) aiGenerationTelemetry.onAssistantSettled( address, turnId, outcome === 'succeeded' ? 'success' : outcome === 'failed' ? 'failure' : 'cancelled' ) + syncRuntimeTaskSnapshot(address) }, onContextUsageUpdated: updateCanonicalRuntimeContextUsage, onSubagentActivity: applyRuntimeConversationSubagentActivity, onRuntimeTaskTitleUpdated: (address, payload) => { + updateLocalRuntimeTaskTitle(address, payload.title) dispatch({ type: 'runtime_task_title_updated', address, @@ -1602,34 +1640,36 @@ export function WorkbenchProvider({ const goal = payload.goal ?? null setRuntimeConversationGoal(address, goal) lifecycleStore.goalStatusReceived(address, goal?.status ?? null) - refreshRuntimeWorkLists(address) + syncRuntimeTaskSnapshot(address) }, onRuntimeGoalCleared: address => { setRuntimeConversationGoal(address, null) lifecycleStore.goalStatusReceived(address, null) - refreshRuntimeWorkLists(address) + syncRuntimeTaskSnapshot(address) }, onRuntimeSupervisorUpdated: address => { - refreshRuntimeWorkLists(address) + syncRuntimeTaskSnapshot(address) }, onRuntimeGoalContinuation: (address, payload) => { applyRuntimeConversationGoalContinuation(address, payload) - refreshRuntimeWorkLists(address) + syncRuntimeTaskSnapshot(address) }, onRuntimePlanUpdated: setRuntimeConversationTaskPlan, onRuntimeTransportReplaced: publishRuntimeTransportReplaced, - onRefreshWorkLists: refreshRuntimeWorkLists, }) ), [ aiGenerationTelemetry, applyCanonicalRuntimeAction, lifecycleStore, - refreshRuntimeWorkLists, resolvedServices.chatStream, settleCanonicalRuntimeGuidance, + syncRuntimeTaskSnapshot, syncRuntimeTaskTitle, updateCanonicalRuntimeContextUsage, + updateLocalRuntimeTaskExecution, + updateLocalRuntimeTaskSnapshot, + updateLocalRuntimeTaskTitle, ] ) diff --git a/wework/src/features/workbench/runtimePaneMessages.test.ts b/wework/src/features/workbench/runtimePaneMessages.test.ts index c7de381523..a9dfaccf77 100644 --- a/wework/src/features/workbench/runtimePaneMessages.test.ts +++ b/wework/src/features/workbench/runtimePaneMessages.test.ts @@ -165,11 +165,9 @@ describe('createRuntimeTaskStreamHandlers', () => { deviceId: 'device-1', taskId: 'runtime-task-1', } - const onRefreshWorkLists = vi.fn() const onRuntimeTaskTitleUpdated = vi.fn() const handlers = createRuntimeTaskStreamHandlers(address, { onMessageAction: vi.fn(), - onRefreshWorkLists, onRuntimeTaskTitleUpdated, }) @@ -180,7 +178,6 @@ describe('createRuntimeTaskStreamHandlers', () => { title: '测试标题生成功能', }) - expect(onRefreshWorkLists).not.toHaveBeenCalled() expect(onRuntimeTaskTitleUpdated).toHaveBeenCalledWith({ taskId: 'runtime-task-1', subtaskId: 'friendly-title-turn', @@ -506,18 +503,16 @@ describe('createRuntimeTaskStreamHandlers', () => { ) }) - test('passes context compaction through regular block created actions', () => { + test('passes context compaction through regular block created actions without refreshing work', () => { const address: RuntimeTaskAddress = { deviceId: 'device-1', taskId: 'runtime-task-1', } const actions: RuntimePaneMessageAction[] = [] const onAssistantSettled = vi.fn() - const onRefreshWorkLists = vi.fn() const handlers = createRuntimeTaskStreamHandlers(address, { onMessageAction: action => actions.push(action), onAssistantSettled, - onRefreshWorkLists, }) handlers.onBlockCreated?.({ @@ -548,7 +543,6 @@ describe('createRuntimeTaskStreamHandlers', () => { subtaskId: 'runtime-task-1-context-compact', }) expect(onAssistantSettled).toHaveBeenCalledTimes(1) - expect(onRefreshWorkLists).toHaveBeenCalledTimes(1) }) test('passes reclassified assistant text identity to the conversation reducer', () => { @@ -596,11 +590,9 @@ describe('createRuntimeTaskStreamHandlers', () => { } const actions: RuntimePaneMessageAction[] = [] const onAssistantSettled = vi.fn() - const onRefreshWorkLists = vi.fn() const handlers = createRuntimeTaskStreamHandlers(address, { onMessageAction: action => actions.push(action), onAssistantSettled, - onRefreshWorkLists, }) handlers.onBlockCreated?.({ @@ -628,7 +620,6 @@ describe('createRuntimeTaskStreamHandlers', () => { }, }) expect(onAssistantSettled).not.toHaveBeenCalled() - expect(onRefreshWorkLists).not.toHaveBeenCalled() }) test('preserves request user input render payload on block created events', () => { diff --git a/wework/src/features/workbench/runtimePaneMessages.ts b/wework/src/features/workbench/runtimePaneMessages.ts index d3a7041f71..ac799ced08 100644 --- a/wework/src/features/workbench/runtimePaneMessages.ts +++ b/wework/src/features/workbench/runtimePaneMessages.ts @@ -43,7 +43,6 @@ export interface RuntimeTaskStreamHandlers { onAssistantFirstToken?: (turnId: string) => void onAssistantResponseSize?: (turnId: string, responseSizeBytes: number) => void onAssistantSettled?: (turnId: string, outcome: 'succeeded' | 'failed' | 'cancelled') => void - onRefreshWorkLists?: () => void onContextUsageUpdated?: (usage: RuntimeContextUsage) => void onSubagentActivity?: (payload: RuntimeSubagentActivityPayload) => void onRuntimeTaskTitleUpdated?: (payload: RuntimeTaskTitleUpdatedPayload) => void @@ -70,7 +69,6 @@ export interface RuntimeConversationStreamHandlers { turnId: string, outcome: 'succeeded' | 'failed' | 'cancelled' ) => void - onRefreshWorkLists?: (address: RuntimeTaskAddress) => void onContextUsageUpdated?: (address: RuntimeTaskAddress, usage: RuntimeContextUsage) => void onSubagentActivity?: ( address: RuntimeTaskAddress, @@ -124,7 +122,6 @@ export function createRuntimeConversationStreamHandlers( handlers.onAssistantResponseSize?.(address, turnId, responseSizeBytes), onAssistantSettled: (turnId, outcome) => handlers.onAssistantSettled?.(address, turnId, outcome), - onRefreshWorkLists: () => handlers.onRefreshWorkLists?.(address), onContextUsageUpdated: usage => handlers.onContextUsageUpdated?.(address, usage), onSubagentActivity: payload => handlers.onSubagentActivity?.(address, payload), onRuntimeTaskTitleUpdated: payload => handlers.onRuntimeTaskTitleUpdated?.(address, payload), @@ -201,7 +198,6 @@ export function createRuntimeTaskStreamHandlers( clientUserMessageId: payload.clientUserMessageId, shellType: payload.shellType, }) - handlers.onRefreshWorkLists?.() }, onChatChunk: payload => { if (!isRuntimeTaskStreamPayload(address, payload)) return @@ -314,7 +310,6 @@ export function createRuntimeTaskStreamHandlers( ) } handlers.onAssistantSettled?.(identity.subtaskId, 'succeeded') - handlers.onRefreshWorkLists?.() }, onChatError: payload => { if (!isRuntimeTaskStreamPayload(address, payload)) { @@ -358,7 +353,6 @@ export function createRuntimeTaskStreamHandlers( } handlers.onAssistantSettled?.(identity.subtaskId, cancelled ? 'cancelled' : 'failed') streamedFileChanges.delete(identity.subtaskId) - handlers.onRefreshWorkLists?.() }, onBlockCreated: payload => { if (!isRuntimeTaskStreamPayload(address, payload)) return @@ -395,7 +389,6 @@ export function createRuntimeTaskStreamHandlers( subtaskId: identity.subtaskId, }) handlers.onAssistantSettled?.(identity.subtaskId, 'succeeded') - handlers.onRefreshWorkLists?.() } }, onBlockUpdated: payload => { diff --git a/wework/src/features/workbench/runtimeTaskLifecycle/RuntimeTaskLifecycleStore.test.ts b/wework/src/features/workbench/runtimeTaskLifecycle/RuntimeTaskLifecycleStore.test.ts index 645e51ca14..8a151ebb0a 100644 --- a/wework/src/features/workbench/runtimeTaskLifecycle/RuntimeTaskLifecycleStore.test.ts +++ b/wework/src/features/workbench/runtimeTaskLifecycle/RuntimeTaskLifecycleStore.test.ts @@ -102,6 +102,41 @@ describe('RuntimeTaskLifecycleStore', () => { expect(store.getTask(address)?.turn.outcome).toBeNull() }) + test('rejects a stale running snapshot after a terminal event without a matching start', () => { + const store = new RuntimeTaskLifecycleStore('test') + store.syncRuntimeWork(runtimeWork(task({ running: true }))) + + store.turnSettled(address, null, 'succeeded') + const accepted = store.syncRuntimeTask(address, task({ running: true })) + + expect(accepted).toBe(false) + expect(store.getTask(address)?.execution.running).toBe(false) + }) + + test('keeps terminal executor snapshots idle even when running remains true', () => { + const store = new RuntimeTaskLifecycleStore('test') + const terminalTask = task({ running: true, status: 'complete' }) + + store.syncRuntimeWork(runtimeWork(terminalTask)) + store.syncRuntimeWork(runtimeWork(terminalTask)) + + expect(store.getTask(address)?.execution.running).toBe(false) + expect(store.getTask(address)?.turn.phase).toBe('idle') + }) + + test('rejects a snapshot when a newer send starts before the response arrives', () => { + const store = new RuntimeTaskLifecycleStore('test') + store.syncRuntimeWork(runtimeWork(task({ running: true }))) + store.turnSettled(address, null, 'succeeded') + const expectedSnapshot = store.getTask(address) + + store.sendRequested(address) + const accepted = store.syncRuntimeTask(address, task({ running: true }), expectedSnapshot) + + expect(accepted).toBe(false) + expect(store.getTask(address)?.turn.phase).toBe('submitting') + }) + test('does not regress a stream that starts before the create acknowledgement', () => { const store = new RuntimeTaskLifecycleStore('test') @@ -376,6 +411,22 @@ describe('RuntimeTaskLifecycleStore', () => { expect(store.getTask(address)?.derived.shouldShowUnread).toBe(false) }) + test.each(['paused', 'blocked', 'usageLimited', 'budgetLimited', 'complete'] as const)( + 'settles execution when the Goal reports %s', + goalStatus => { + const store = new RuntimeTaskLifecycleStore('test') + store.syncRuntimeWork(runtimeWork(task({ running: true, goalStatus: 'active' }))) + store.turnStarted(address, 'goal-turn') + store.turnSettled(address, 'goal-turn') + + store.goalStatusReceived(address, goalStatus) + + expect(store.getTask(address)?.goalStatus).toBe(goalStatus) + expect(store.getTask(address)?.execution.running).toBe(false) + expect(store.getTask(address)?.turn.phase).toBe('idle') + } + ) + test('migrates optimistic lifecycle state when the executor resolves a new task identity', () => { const store = new RuntimeTaskLifecycleStore('test') const resolved = { ...address, taskId: 'resolved-task' } diff --git a/wework/src/features/workbench/runtimeTaskLifecycle/RuntimeTaskLifecycleStore.ts b/wework/src/features/workbench/runtimeTaskLifecycle/RuntimeTaskLifecycleStore.ts index f3d5961e44..33c31a9735 100644 --- a/wework/src/features/workbench/runtimeTaskLifecycle/RuntimeTaskLifecycleStore.ts +++ b/wework/src/features/workbench/runtimeTaskLifecycle/RuntimeTaskLifecycleStore.ts @@ -67,6 +67,24 @@ export class RuntimeTaskLifecycleStore { if (changed) this.publish() } + syncRuntimeTask( + address: RuntimeTaskAddress, + task: RuntimeTaskSummary, + expectedSnapshot?: RuntimeTaskLifecycleSnapshot | null + ): boolean { + if (expectedSnapshot !== undefined && this.getTask(address) !== expectedSnapshot) { + return false + } + const changed = this.reduceMachine(address, { + type: 'executor_snapshot_received', + address, + task, + }) + if (changed) this.publish() + const executionRunning = this.getTask(address)?.execution.running + return typeof task.running !== 'boolean' || task.running === executionRunning + } + setCurrentTask(address: RuntimeTaskAddress | null | undefined): void { const nextKey = address ? getRuntimeTaskLifecycleKey(address) : null if (nextKey === this.currentTaskKey) return diff --git a/wework/src/features/workbench/runtimeTaskLifecycle/reducer.ts b/wework/src/features/workbench/runtimeTaskLifecycle/reducer.ts index 376b3ac475..c976b6c787 100644 --- a/wework/src/features/workbench/runtimeTaskLifecycle/reducer.ts +++ b/wework/src/features/workbench/runtimeTaskLifecycle/reducer.ts @@ -9,23 +9,22 @@ export function reduceRuntimeTaskLifecycle( const snapshotRunning = typeof event.task.running === 'boolean' ? event.task.running : null const expectedRunning = state.expectedExecutorRunning const hasIdentifiedActiveTurn = state.turnPhase === 'streaming' && state.activeTurnId !== null + const terminalStatus = isTerminalTaskStatus(event.task.status) const shouldIgnoreStaleSnapshot = snapshotRunning !== null && expectedRunning !== null && snapshotRunning !== expectedRunning && - (!isTerminalTaskStatus(event.task.status) || hasIdentifiedActiveTurn) + (!terminalStatus || hasIdentifiedActiveTurn) + if (shouldIgnoreStaleSnapshot) return state + const executionPhase = - snapshotRunning === null - ? state.executionPhase - : shouldIgnoreStaleSnapshot - ? state.executionPhase - : snapshotRunning - ? 'running' - : 'idle' - const turnPhase = - snapshotRunning === false && !shouldIgnoreStaleSnapshot ? 'idle' : state.turnPhase - const activeTurnId = - snapshotRunning === false && !shouldIgnoreStaleSnapshot ? null : state.activeTurnId + terminalStatus || snapshotRunning === false + ? 'idle' + : snapshotRunning === true + ? 'running' + : state.executionPhase + const turnPhase = terminalStatus || snapshotRunning === false ? 'idle' : state.turnPhase + const activeTurnId = terminalStatus || snapshotRunning === false ? null : state.activeTurnId return { ...state, @@ -37,10 +36,9 @@ export function reduceRuntimeTaskLifecycle( goalStatus: event.task.goalStatus === undefined ? state.goalStatus : event.task.goalStatus, continuable: event.task.continuable !== false, expectedExecutorRunning: - !shouldIgnoreStaleSnapshot && snapshotRunning !== null && event.task.optimistic !== true && - (snapshotRunning === expectedRunning || isTerminalTaskStatus(event.task.status)) + (snapshotRunning === expectedRunning || terminalStatus) ? null : expectedRunning, } @@ -156,10 +154,19 @@ export function reduceRuntimeTaskLifecycle( : state case 'goal_status_received': - return { - ...state, - goalStatus: event.goalStatus, - } + return event.goalStatus !== null && event.goalStatus !== 'active' + ? { + ...state, + executionPhase: 'idle', + turnPhase: 'idle', + activeTurnId: null, + goalStatus: event.goalStatus, + expectedExecutorRunning: false, + } + : { + ...state, + goalStatus: event.goalStatus, + } case 'marked_read': return state.unread ? { ...state, unread: false } : state diff --git a/wework/src/features/workbench/useWorkbenchDataRefresh.ts b/wework/src/features/workbench/useWorkbenchDataRefresh.ts index 4aa32f701b..e556e22590 100644 --- a/wework/src/features/workbench/useWorkbenchDataRefresh.ts +++ b/wework/src/features/workbench/useWorkbenchDataRefresh.ts @@ -6,6 +6,7 @@ import type { DeviceInfo, ProjectWithTasks, RuntimeTaskAddress, + RuntimeTaskSummary, RuntimeWorkListResponse, User, } from '@/types/api' @@ -32,10 +33,13 @@ import { import type { WorkbenchAction } from './workbenchReducer' import { debugRuntimeSidebarState, summarizeRuntimeWorkTaskIds } from './runtimeSidebarDiagnostics' import { + findRuntimeTask, getRememberedStandaloneDeviceId, getRuntimeTaskRouteKey, removeRuntimeTasks, runtimeWorkContainsTask, + updateRuntimeWorkTask, + updateRuntimeWorkTaskTitle, } from './workbenchRuntimeHelpers' import type { WorkbenchServices } from './workbenchServices' import type { RefreshWorkLists } from './workbenchContextTypes' @@ -191,6 +195,9 @@ export function useWorkbenchDataRefresh({ const cloudBackgroundRequestControllerRef = useRef(null) const runtimeWorkRef = useRef(state.runtimeWork) const localRuntimeWorkRef = useRef(null) + const runtimeTaskTitleOverridesRef = useRef( + new Map() + ) const devicesRef = useRef(state.devices) const archivedRuntimeTaskAddressesRef = useRef([]) const removedRuntimeProjectsRef = useRef< @@ -215,6 +222,22 @@ export function useWorkbenchDataRefresh({ setCloudRuntimeState(next) }, []) + const applyRuntimeTaskTitleOverrides = useCallback( + (runtimeWork: RuntimeWorkListResponse, confirmExecutorTitles = false) => { + let next = runtimeWork + runtimeTaskTitleOverridesRef.current.forEach((override, key) => { + const task = findRuntimeTask(runtimeWork, override.address) + if (confirmExecutorTitles && task?.title === override.title) { + runtimeTaskTitleOverridesRef.current.delete(key) + return + } + next = updateRuntimeWorkTaskTitle(next, override.address, override.title) ?? next + }) + return next + }, + [] + ) + useEffect(() => { cloudBackgroundApiRef.current = services.cloudBackgroundApi runtimeWorkRef.current = state.runtimeWork @@ -226,6 +249,7 @@ export function useWorkbenchDataRefresh({ userId: user.id, runtimeWork: initialCachedRemoteRuntimeWork, } + runtimeTaskTitleOverridesRef.current.clear() archivedRuntimeTaskAddressesRef.current = [] removedRuntimeProjectsRef.current = [] // eslint-disable-next-line react-hooks/set-state-in-effect -- Cached runtime work must switch atomically with the authenticated user. @@ -415,12 +439,16 @@ export function useWorkbenchDataRefresh({ return } - const latestLocalRuntimeWork = localRuntimeWorkRef.current ?? baseRuntimeWork + const latestLocalRuntimeWork = applyRuntimeTaskTitleOverrides( + localRuntimeWorkRef.current ?? baseRuntimeWork + ) const filteredRuntimeWorkResult = runtimeWorkResult?.status === 'fulfilled' ? { status: 'fulfilled' as const, - value: filterRemovedRuntimeProjects(runtimeWorkResult.value), + value: applyRuntimeTaskTitleOverrides( + filterRemovedRuntimeProjects(runtimeWorkResult.value) + ), } : runtimeWorkResult if (filteredRuntimeWorkResult?.status === 'fulfilled') { @@ -493,6 +521,7 @@ export function useWorkbenchDataRefresh({ } }, [ + applyRuntimeTaskTitleOverrides, dispatch, filterRemovedRuntimeProjects, selectVisibleRuntimeWork, @@ -552,7 +581,10 @@ export function useWorkbenchDataRefresh({ if (cancelled) return const runtimeWork = runtimeWorkResult.status === 'fulfilled' - ? filterRemovedRuntimeProjects(runtimeWorkResult.value) + ? applyRuntimeTaskTitleOverrides( + filterRemovedRuntimeProjects(runtimeWorkResult.value), + true + ) : EMPTY_RUNTIME_WORK if (runtimeWorkResult.status === 'fulfilled') { localRuntimeWorkRef.current = runtimeWork @@ -590,6 +622,7 @@ export function useWorkbenchDataRefresh({ window.clearTimeout(slowTimer) } }, [ + applyRuntimeTaskTitleOverrides, dispatch, executorClient, filterRemovedRuntimeProjects, @@ -614,7 +647,7 @@ export function useWorkbenchDataRefresh({ selectVisibleDevices(devices, cloudRuntimeStateRef.current) ) const filteredRuntimeWorkResult = runtimeWorkResult - ? filterRemovedRuntimeProjects(runtimeWorkResult) + ? applyRuntimeTaskTitleOverrides(filterRemovedRuntimeProjects(runtimeWorkResult), true) : undefined if (filteredRuntimeWorkResult) { localRuntimeWorkRef.current = filteredRuntimeWorkResult @@ -655,6 +688,7 @@ export function useWorkbenchDataRefresh({ } }, [ + applyRuntimeTaskTitleOverrides, dispatch, executorClient, filterRemovedRuntimeProjects, @@ -764,6 +798,57 @@ export function useWorkbenchDataRefresh({ ] ) + const updateLocalRuntimeTaskTitle = useCallback((address: RuntimeTaskAddress, title: string) => { + runtimeTaskTitleOverridesRef.current.set(getRuntimeTaskRouteKey(address), { + address, + title, + }) + localRuntimeWorkRef.current = updateRuntimeWorkTaskTitle( + localRuntimeWorkRef.current, + address, + title + ) + }, []) + + const updateLocalRuntimeTaskExecution = useCallback( + (address: RuntimeTaskAddress, running: boolean, status: string) => { + localRuntimeWorkRef.current = updateRuntimeWorkTask(localRuntimeWorkRef.current, address, { + running, + status, + optimistic: true, + }) + }, + [] + ) + + const refreshRuntimeTask = useCallback( + async (address: RuntimeTaskAddress) => { + const runtimeWork = applyRuntimeTaskTitleOverrides( + filterRemovedRuntimeProjects(await executorClient.runtime.listRuntimeWork()), + true + ) + const task = findRuntimeTask(runtimeWork, address) + return task ? { ...task, optimistic: false } : null + }, + [applyRuntimeTaskTitleOverrides, executorClient, filterRemovedRuntimeProjects] + ) + + const updateLocalRuntimeTaskSnapshot = useCallback( + (address: RuntimeTaskAddress, task: RuntimeTaskSummary) => { + localRuntimeWorkRef.current = updateRuntimeWorkTask( + localRuntimeWorkRef.current, + address, + task + ) + dispatch({ + type: 'runtime_task_snapshot_updated', + address, + task, + }) + }, + [dispatch] + ) + const getRemoteDeviceStartupCommand = useCallback(async (): Promise => { const createCommand = services.deviceApi.createDockerRemoteDeviceCommand @@ -779,7 +864,11 @@ export function useWorkbenchDataRefresh({ markRuntimeProjectRemoved, clearRuntimeProjectRemoval, refreshWorkLists, + refreshRuntimeTask, refreshDevices, + updateLocalRuntimeTaskExecution, + updateLocalRuntimeTaskSnapshot, + updateLocalRuntimeTaskTitle, getRemoteDeviceStartupCommand, } } diff --git a/wework/src/features/workbench/workbenchReducer.ts b/wework/src/features/workbench/workbenchReducer.ts index 3ca6bf16a8..be3a0c5888 100644 --- a/wework/src/features/workbench/workbenchReducer.ts +++ b/wework/src/features/workbench/workbenchReducer.ts @@ -23,6 +23,8 @@ import { getRuntimeTaskWorkspacePath, mergeRuntimeTaskHandles, removeRuntimeTasks, + updateRuntimeWorkTask, + updateRuntimeWorkTaskTitle, } from './workbenchRuntimeHelpers' import { debugRuntimeSidebarState, summarizeRuntimeWorkTaskIds } from './runtimeSidebarDiagnostics' @@ -131,6 +133,17 @@ export type WorkbenchAction = address: RuntimeTaskAddress title: string } + | { + type: 'runtime_task_execution_updated' + address: RuntimeTaskAddress + running: boolean + status: string + } + | { + type: 'runtime_task_snapshot_updated' + address: RuntimeTaskAddress + task: RuntimeTaskSummary + } | { type: 'runtime_tasks_archived'; addresses: RuntimeTaskAddress[] } | { type: 'current_task_cleared' } | { type: 'error_set'; error: string | null } @@ -179,35 +192,6 @@ function updateRuntimeWorkDeviceStatus( } } -function updateRuntimeWorkTaskTitle( - runtimeWork: RuntimeWorkListResponse | null | undefined, - address: RuntimeTaskAddress, - title: string -): RuntimeWorkListResponse | null { - if (!runtimeWork) return null - - const updateWorkspace = (workspace: RuntimeDeviceWorkspace): RuntimeDeviceWorkspace => { - if (workspace.deviceId !== address.deviceId && workspace.remoteHostId !== address.deviceId) { - return workspace - } - - const tasks = workspace.tasks.map(task => - task.taskId === address.taskId ? { ...task, title } : task - ) - if (tasks.every((task, index) => task === workspace.tasks[index])) return workspace - return { ...workspace, tasks } - } - - return { - ...runtimeWork, - projects: runtimeWork.projects.map(project => ({ - ...project, - deviceWorkspaces: project.deviceWorkspaces.map(updateWorkspace), - })), - chats: runtimeWork.chats.map(updateWorkspace), - } -} - function mergeRuntimeWorkPreservingTaskOrder( current: RuntimeWorkListResponse | null | undefined, next: RuntimeWorkListResponse | null @@ -1215,6 +1199,20 @@ export function workbenchReducer(state: WorkbenchState, action: WorkbenchAction) ...state, runtimeWork: updateRuntimeWorkTaskTitle(state.runtimeWork, action.address, action.title), } + case 'runtime_task_execution_updated': + return { + ...state, + runtimeWork: updateRuntimeWorkTask(state.runtimeWork, action.address, { + running: action.running, + status: action.status, + optimistic: true, + }), + } + case 'runtime_task_snapshot_updated': + return { + ...state, + runtimeWork: updateRuntimeWorkTask(state.runtimeWork, action.address, action.task), + } case 'runtime_tasks_archived': return { ...state, diff --git a/wework/src/features/workbench/workbenchRuntimeHelpers.test.ts b/wework/src/features/workbench/workbenchRuntimeHelpers.test.ts index 9b36ec15c6..58086393fd 100644 --- a/wework/src/features/workbench/workbenchRuntimeHelpers.test.ts +++ b/wework/src/features/workbench/workbenchRuntimeHelpers.test.ts @@ -9,6 +9,7 @@ import { readLastProjectId, removeRuntimeTasks, truncateRuntimeTaskTitle, + updateRuntimeWorkTaskTitle, writeLastProjectId, } from './workbenchRuntimeHelpers' import type { RuntimeWorkListResponse } from '@/types/api' @@ -156,6 +157,43 @@ describe('workbenchRuntimeHelpers', () => { ).toBe(0) }) + test('updates a runtime task title through its remote host identity', () => { + const runtimeWork: RuntimeWorkListResponse = { + projects: [], + chats: [ + { + deviceId: 'local-device', + remoteHostId: 'device-1', + workspacePath: '/workspace/project-alpha', + available: true, + tasks: [ + { + taskId: 'runtime-a', + workspacePath: '/workspace/project-alpha', + title: '解决冲突', + runtime: 'codex', + }, + { + taskId: 'runtime-b', + workspacePath: '/workspace/project-alpha', + title: '未修改', + runtime: 'codex', + }, + ], + }, + ], + totalTasks: 2, + } + + const updated = updateRuntimeWorkTaskTitle( + runtimeWork, + { deviceId: 'device-1', taskId: 'runtime-a' }, + '解决分支冲突' + ) + + expect(updated?.chats[0]?.tasks.map(task => task.title)).toEqual(['解决分支冲突', '未修改']) + }) + test('stores the last project per user and ignores invalid values', () => { writeLastProjectId(7, 42) diff --git a/wework/src/features/workbench/workbenchRuntimeHelpers.ts b/wework/src/features/workbench/workbenchRuntimeHelpers.ts index 550b9491a3..2afd17024d 100644 --- a/wework/src/features/workbench/workbenchRuntimeHelpers.ts +++ b/wework/src/features/workbench/workbenchRuntimeHelpers.ts @@ -203,6 +203,43 @@ export function removeRuntimeTasks( } } +export function updateRuntimeWorkTask( + runtimeWork: RuntimeWorkListResponse | null | undefined, + address: RuntimeTaskAddress, + updates: Partial +): RuntimeWorkListResponse | null { + if (!runtimeWork) return null + + const updateWorkspace = (workspace: RuntimeDeviceWorkspace): RuntimeDeviceWorkspace => { + if (workspace.deviceId !== address.deviceId && workspace.remoteHostId !== address.deviceId) { + return workspace + } + + const tasks = workspace.tasks.map(task => + task.taskId === address.taskId ? { ...task, ...updates } : task + ) + if (tasks.every((task, index) => task === workspace.tasks[index])) return workspace + return { ...workspace, tasks } + } + + return { + ...runtimeWork, + projects: runtimeWork.projects.map(project => ({ + ...project, + deviceWorkspaces: project.deviceWorkspaces.map(updateWorkspace), + })), + chats: runtimeWork.chats.map(updateWorkspace), + } +} + +export function updateRuntimeWorkTaskTitle( + runtimeWork: RuntimeWorkListResponse | null | undefined, + address: RuntimeTaskAddress, + title: string +): RuntimeWorkListResponse | null { + return updateRuntimeWorkTask(runtimeWork, address, { title }) +} + export function runtimeWorkContainsTask( runtimeWork: RuntimeWorkListResponse, address: RuntimeTaskAddress