feat(agent): add active-run steering and queue controls (#3270)

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);
+  });
+});

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<string, ArtifactSnapshot[]>;
-  /** Per-thread run queue status. Updated from queue_status RPC responses. */
-  queueStatusByThread: Record<string, QueueStatus>;
   sessionTokenUsage: SessionTokenUsage;
+  queueStatusByThread: Record<string, QueueStatus>;
 }
 
 /** 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: {},
 };
 
 /**

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,
         }
     }

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()),