diff --git a/app/src/store/chatRuntimeSlice.test.ts b/app/src/store/chatRuntimeSlice.test.ts new file mode 100644 index 000000000..8a8b8448d --- /dev/null +++ b/app/src/store/chatRuntimeSlice.test.ts @@ -0,0 +1,98 @@ +import { configureStore } from '@reduxjs/toolkit'; +import { describe, expect, it } from 'vitest'; + +import chatRuntimeReducer, { + clearAllChatRuntime, + clearQueueStatusForThread, + clearRuntimeForThread, + type QueueStatus, + setQueueStatusForThread, +} from './chatRuntimeSlice'; + +function makeStore() { + return configureStore({ reducer: { chatRuntime: chatRuntimeReducer } }); +} + +describe('chatRuntimeSlice queue status', () => { + it('sets queue status for a thread', () => { + const store = makeStore(); + const status: QueueStatus = { active: true, steers: 1, followups: 2, collects: 0, total: 3 }; + store.dispatch(setQueueStatusForThread({ threadId: 't1', status })); + expect(store.getState().chatRuntime.queueStatusByThread['t1']).toEqual(status); + }); + + it('clears queue status for a thread', () => { + const store = makeStore(); + const status: QueueStatus = { active: true, steers: 1, followups: 0, collects: 0, total: 1 }; + store.dispatch(setQueueStatusForThread({ threadId: 't1', status })); + store.dispatch(clearQueueStatusForThread({ threadId: 't1' })); + expect(store.getState().chatRuntime.queueStatusByThread['t1']).toBeUndefined(); + }); + + it('clearRuntimeForThread removes queue status', () => { + const store = makeStore(); + const status: QueueStatus = { active: true, steers: 1, followups: 0, collects: 0, total: 1 }; + store.dispatch(setQueueStatusForThread({ threadId: 't1', status })); + store.dispatch(clearRuntimeForThread({ threadId: 't1' })); + expect(store.getState().chatRuntime.queueStatusByThread['t1']).toBeUndefined(); + }); + + it('clearAllChatRuntime removes all queue statuses', () => { + const store = makeStore(); + store.dispatch( + setQueueStatusForThread({ + threadId: 't1', + status: { active: true, steers: 1, followups: 0, collects: 0, total: 1 }, + }) + ); + store.dispatch( + setQueueStatusForThread({ + threadId: 't2', + status: { active: true, steers: 0, followups: 1, collects: 0, total: 1 }, + }) + ); + store.dispatch(clearAllChatRuntime()); + expect(store.getState().chatRuntime.queueStatusByThread).toEqual({}); + }); + + it('updates queue status when set again', () => { + const store = makeStore(); + store.dispatch( + setQueueStatusForThread({ + threadId: 't1', + status: { active: true, steers: 1, followups: 0, collects: 0, total: 1 }, + }) + ); + store.dispatch( + setQueueStatusForThread({ + threadId: 't1', + status: { active: true, steers: 0, followups: 0, collects: 0, total: 0 }, + }) + ); + expect(store.getState().chatRuntime.queueStatusByThread['t1']).toEqual({ + active: true, + steers: 0, + followups: 0, + collects: 0, + total: 0, + }); + }); + + it('isolates queue status across threads', () => { + const store = makeStore(); + store.dispatch( + setQueueStatusForThread({ + threadId: 't1', + status: { active: true, steers: 1, followups: 0, collects: 0, total: 1 }, + }) + ); + store.dispatch( + setQueueStatusForThread({ + threadId: 't2', + status: { active: true, steers: 0, followups: 2, collects: 0, total: 2 }, + }) + ); + expect(store.getState().chatRuntime.queueStatusByThread['t1']?.steers).toBe(1); + expect(store.getState().chatRuntime.queueStatusByThread['t2']?.followups).toBe(2); + }); +}); diff --git a/app/src/store/chatRuntimeSlice.ts b/app/src/store/chatRuntimeSlice.ts index 12992cc11..e1c77de6a 100644 --- a/app/src/store/chatRuntimeSlice.ts +++ b/app/src/store/chatRuntimeSlice.ts @@ -222,6 +222,9 @@ export interface ArtifactSnapshot { updatedAt: number; } +/** Queue behavior when a turn is already in flight for a thread. */ +export type QueueMode = 'interrupt' | 'steer' | 'followup' | 'collect'; + /** * Per-thread UI state for an in-flight agent turn (socket events while the user * may navigate away from Conversations). The thread slice keeps `activeThreadId` @@ -242,9 +245,8 @@ interface ChatRuntimeState { * download / retry affordances (#2779). */ artifactsByThread: Record; - /** Per-thread run queue status. Updated from queue_status RPC responses. */ - queueStatusByThread: Record; sessionTokenUsage: SessionTokenUsage; + queueStatusByThread: Record; } /** Snapshot of the active-run queue depth per lane. */ @@ -264,8 +266,8 @@ const initialState: ChatRuntimeState = { inferenceTurnLifecycleByThread: {}, pendingApprovalByThread: {}, artifactsByThread: {}, - queueStatusByThread: {}, sessionTokenUsage: { inputTokens: 0, outputTokens: 0, turns: 0, lastUpdated: 0 }, + queueStatusByThread: {}, }; /** diff --git a/src/core/event_bus/events.rs b/src/core/event_bus/events.rs index 2f88e0815..a1b60abd8 100644 --- a/src/core/event_bus/events.rs +++ b/src/core/event_bus/events.rs @@ -1205,6 +1205,10 @@ impl DomainEvent { | Self::ChannelDisconnected { channel, .. } => Some(channel.as_str()), Self::ToolExecutionStarted { tool_name, .. } | Self::ToolExecutionCompleted { tool_name, .. } => Some(tool_name.as_str()), + Self::RunQueueMessageQueued { thread_id, .. } + | Self::RunQueueMessageDelivered { thread_id, .. } + | Self::RunQueueFollowupDispatched { thread_id, .. } + | Self::RunQueueInterrupted { thread_id, .. } => Some(thread_id.as_str()), _ => None, } } diff --git a/src/openhuman/channels/providers/web.rs b/src/openhuman/channels/providers/web.rs index e7fe22625..4a8581252 100644 --- a/src/openhuman/channels/providers/web.rs +++ b/src/openhuman/channels/providers/web.rs @@ -644,7 +644,10 @@ pub async fn start_chat( { let mut in_flight = IN_FLIGHT.lock().await; + + // Interrupt path: abort any in-flight turn (existing behavior). if let Some(existing) = in_flight.remove(&map_key) { + let cancelled_id = existing.request_id.clone(); existing.handle.abort(); log::info!( "[web-channel] interrupted in-flight turn thread_id={} cancelled_request_id={}", @@ -659,7 +662,7 @@ pub async fn start_chat( event: "chat_error".to_string(), client_id: client_id.clone(), thread_id: thread_id.clone(), - request_id: existing.request_id, + request_id: cancelled_id, full_response: None, message: Some("Cancelled by newer request".to_string()), error_type: Some("cancelled".to_string()),