feat: add Slack --queue flag

docs/pages/deploying-in-production.mdx

@@ -347,6 +347,8 @@ reply with exactly PONG
 
 Slack messages without a harness flag use Codex. Use `--amp`, `--claude`,
 `--codex`, or `--pi` only when you want to select a specific harness.
+While a turn is running, add `--queue` to a message to run it as the next turn
+instead of steering the active turn.
 
 Inspect sandbox pods with the labels Centaur actually sets:
 

docs/public/md/deploying-in-production.md

@@ -347,6 +347,8 @@ reply with exactly PONG
 
 Slack messages without a harness flag use Codex. Use `--amp`, `--claude`,
 `--codex`, or `--pi` only when you want to select a specific harness.
+While a turn is running, add `--queue` to a message to run it as the next turn
+instead of steering the active turn.
 
 Inspect sandbox pods with the labels Centaur actually sets:
 

services/api-rs/crates/centaur-api-server/src/routes.rs

@@ -609,7 +609,11 @@ async fn append_messages(
     let thread_key = ThreadKey::try_from(raw_thread_key)?;
     ensure_session_resource_authorized(&runtime, &thread_key, &authorization).await?;
     let message_ids = runtime
-        .append_messages(&thread_key, &request.messages)
+        .append_messages(
+            &thread_key,
+            &request.messages,
+            request.steer_active_execution,
+        )
         .await?;
     Ok(Json(AppendMessagesResponse {
         ok: true,

services/api-rs/crates/centaur-api-server/src/types.rs

@@ -88,6 +88,33 @@ pub struct GithubThreadContext {
 #[derive(Clone, Debug, Deserialize, Serialize)]
 pub struct AppendMessagesRequest {
     pub messages: Vec<SessionMessageInput>,
+    /// Whether user messages should be delivered to an active harness turn as steering.
+    /// Existing clients default to steering; callers can disable it to queue a later turn.
+    #[serde(default = "default_true")]
+    pub steer_active_execution: bool,
+}
+
+fn default_true() -> bool {
+    true
+}
+
+#[cfg(test)]
+mod append_messages_request_tests {
+    use super::AppendMessagesRequest;
+
+    #[test]
+    fn append_messages_steers_by_default_but_can_be_queued() {
+        let default_request: AppendMessagesRequest =
+            serde_json::from_value(serde_json::json!({ "messages": [] })).unwrap();
+        assert!(default_request.steer_active_execution);
+
+        let queued_request: AppendMessagesRequest = serde_json::from_value(serde_json::json!({
+            "messages": [],
+            "steer_active_execution": false
+        }))
+        .unwrap();
+        assert!(!queued_request.steer_active_execution);
+    }
 }
 
 #[derive(Clone, Debug, Deserialize, Serialize)]

services/api-rs/crates/centaur-session-cli/src/main.rs

@@ -227,6 +227,7 @@ pub(crate) async fn append_user_message(
                         "source": "centaur-session-cli",
                     }),
                 }],
+                steer_active_execution: true,
             },
         )
         .await?;

services/api-rs/crates/centaur-session-runtime/src/lib.rs

@@ -1692,6 +1692,7 @@ impl SessionRuntime {
         &self,
         thread_key: &ThreadKey,
         messages: &[SessionMessageInput],
+        steer_active_execution: bool,
     ) -> Result<Vec<String>, SessionRuntimeError> {
         let span = info_span!(
             "centaur.api_rs.session.messages.append",
@@ -1757,8 +1758,10 @@ impl SessionRuntime {
                 return Err(error);
             }
         };
-        self.forward_messages_to_active_execution(thread_key, messages, &message_ids)
-            .await;
+        if steer_active_execution {
+            self.forward_messages_to_active_execution(thread_key, messages, &message_ids)
+                .await;
+        }
         self.spawn_session_title_generation(thread_key);
         Ok(message_ids)
     }
@@ -9376,6 +9379,7 @@ mod adoption_tests {
                     ],
                     metadata: json!({}),
                 }],
+                true,
             ),
         )
         .await
@@ -9401,6 +9405,7 @@ mod adoption_tests {
                     parts: vec![json!({"type": "text", "text": "add more logging"})],
                     metadata: json!({}),
                 }],
+                true,
             )
             .await
             .expect("append burst message");
@@ -9420,6 +9425,7 @@ mod adoption_tests {
                     parts: vec![json!({"type": "text", "text": "add more logging"})],
                     metadata: json!({}),
                 }],
+                true,
             )
             .await
             .expect("append second message");

services/api-rs/crates/centaur-workflows/src/lib.rs

@@ -4266,6 +4266,7 @@ async fn run_agent_session_turn(
                 parts: parts.clone(),
                 metadata: message_metadata,
             }],
+            true,
         )
         .await?;
     let execution = session_runtime

services/slackbotv2/src/index.ts

@@ -145,6 +145,8 @@ const LATE_SLACK_FILE_PENDING_TTL_MS = 60_000
 const LATE_SLACK_FILE_CONSUMED_TTL_MS = 5 * 60_000
 const LATE_SLACK_FILE_IDLE_WAIT_MS = 90_000
 const LATE_SLACK_FILE_IDLE_POLL_MS = 500
+const QUEUED_EXECUTION_WAIT_MS = 4 * 60 * 60 * 1000
+const QUEUED_EXECUTION_POLL_MS = 500
 const LATE_SLACK_FILE_MESSAGE_TEXT = 'Late Slack file attachment for the previous message.'
 const SLACK_BLOCK_ACTION_DEDUPE_TTL_MS = 24 * 60 * 60 * 1000
 const SLACK_BLOCK_ACTION_LEASE_TTL_MS = 60 * 1000
@@ -160,6 +162,7 @@ type PendingLateSlackFileMention = {
 
 type StickyThreadOverrides = Pick<SlackbotV2ThreadState, 'harnessType' | 'model' | 'provider'>
 const DEFAULT_MESSAGE_OVERRIDES_STRATEGY = createFlagMessageOverridesStrategy()
+const queuedExecutionChains = new Map<string, Promise<void>>()
 
 export async function messageOverridesForText(
   options: SlackbotV2Options,
@@ -821,6 +824,8 @@ type SyncThreadMessageInput = {
   retryAttempt?: number
   /** Resolved once per local handoff chain so retryable failures stay idempotent. */
   resolvedMessageOverrides?: Awaited<ReturnType<typeof messageOverridesForText>>
+  /** True for the detached worker that is waiting to start a --queue message. */
+  queueWorker?: boolean
   state: StateAdapter
 }
 
@@ -945,6 +950,8 @@ async function syncThreadMessageToSession(
     setMessageText(serializedMessage, messageOverrides.cleanedText)
   }
   const overrides = messageOverrides.overrides
+  const shouldQueueBehindActiveExecution =
+    input.mode === 'execute' && overrides.queue === true && state.activeExecution === true
   const stickyOverridesUpdate = stickyThreadOverrideUpdate(overrides)
   const effectiveOverrides = resolveStickyThreadOverrides(state, stickyOverridesUpdate)
   // Slack-only "Open chat in Console" link on the FIRST assistant message in
@@ -986,11 +993,18 @@ async function syncThreadMessageToSession(
         model: effectiveModel
       })
     : undefined
-  if (overrides.harnessType || overrides.model || overrides.provider || overrides.reasoning) {
+  if (
+    overrides.harnessType ||
+    overrides.model ||
+    overrides.provider ||
+    overrides.queue ||
+    overrides.reasoning
+  ) {
     traceLog(input.options, 'slackbotv2_forward_overrides_parsed', trace, {
       harness_type: overrides.harnessType,
       model: overrides.model,
       provider: overrides.provider,
+      queue: overrides.queue,
       reasoning: overrides.reasoning
     })
   }
@@ -1057,6 +1071,7 @@ async function syncThreadMessageToSession(
     metadataModel: shouldStartExecution ? effectiveModel : undefined,
     provider: shouldStartExecution ? resolvedProvider : undefined,
     reasoning: resolvedReasoning,
+    steerActiveExecution: !shouldQueueBehindActiveExecution,
     onEventId: eventId => {
       lastEventId = Math.max(lastEventId, eventId)
     },
@@ -1174,6 +1189,9 @@ async function syncThreadMessageToSession(
     traceLog(input.options, 'slackbotv2_forward_complete', trace)
     recordForward(input.mode, 'complete', traceStartedAtMs)
     if (input.retryAttempt) slackbotMetrics.handoffRetries.inc({ outcome: 'succeeded' })
+    if (shouldQueueBehindActiveExecution && !input.queueWorker) {
+      scheduleQueuedExecution(thread, message, input, messageOverrides, trace)
+    }
     return
   }
 
@@ -1254,6 +1272,67 @@ async function syncThreadMessageToSession(
   }
 }
 
+function scheduleQueuedExecution(
+  thread: Thread<SlackbotV2ThreadState>,
+  message: ChatMessage,
+  input: SyncThreadMessageInput,
+  resolvedMessageOverrides: Awaited<ReturnType<typeof messageOverridesForText>>,
+  trace: SlackbotV2Trace
+): void {
+  traceLog(input.options, 'slackbotv2_queued_execution_scheduled', trace)
+  const previous = queuedExecutionChains.get(thread.id) ?? Promise.resolve()
+  const promise = previous.catch(() => undefined).then(async () => {
+    const startedAtMs = nowMs()
+    while (elapsedMs(startedAtMs) < QUEUED_EXECUTION_WAIT_MS) {
+      const latest = (await thread.state) ?? {}
+      if (latest.executedMessageIds?.includes(message.id)) {
+        traceLog(input.options, 'slackbotv2_queued_execution_already_started', trace)
+        return
+      }
+      if (latest.activeExecution === true) {
+        await sleep(QUEUED_EXECUTION_POLL_MS)
+        continue
+      }
+
+      const assistantStatusVisible = await setInitialAssistantStatus(
+        thread,
+        input.options,
+        trace
+      ).catch(() => false)
+      await syncThreadMessageToSession(thread, message, {
+        initialAssistantStatusRequested: true,
+        initialAssistantStatusVisible: assistantStatusVisible,
+        mode: 'execute',
+        options: input.options,
+        queueWorker: true,
+        resolvedMessageOverrides,
+        state: input.state
+      })
+
+      const afterAttempt = (await thread.state) ?? {}
+      if (afterAttempt.executedMessageIds?.includes(message.id)) {
+        traceLog(input.options, 'slackbotv2_queued_execution_started', trace, {
+          waited_ms: elapsedMs(startedAtMs)
+        })
+        return
+      }
+      await sleep(QUEUED_EXECUTION_POLL_MS)
+    }
+    traceWarn(input.options, 'slackbotv2_queued_execution_wait_timeout', trace, {
+      waited_ms: elapsedMs(startedAtMs)
+    })
+  }).catch(error => {
+    traceWarn(input.options, 'slackbotv2_queued_execution_failed', trace, {
+      error: errorMessage(error)
+    })
+  })
+  queuedExecutionChains.set(thread.id, promise)
+  void promise.finally(() => {
+    if (queuedExecutionChains.get(thread.id) === promise) queuedExecutionChains.delete(thread.id)
+  })
+  backgroundWaitUntil(promise)
+}
+
 function scheduleExecutionRender(
   thread: Thread<SlackbotV2ThreadState>,
   message: SlackbotV2ApiMessage,

services/slackbotv2/src/message-overrides-strategy.ts

@@ -16,6 +16,7 @@ const SYSTEM_PROMPT = [
   'Allowed harness values: codex, claudecode, amp.',
   'Allowed provider values: responses, amazon-bedrock, openrouter.',
   'Allowed reasoning values: none, minimal, low, medium, high, xhigh, max.',
+  'Set queue to true only when the message contains the literal --queue flag; otherwise false.',
   'Map fuzzy effort words to the nearest reasoning value by magnitude. Examples: tiny/cheap/fast -> low or minimal; normal/default -> medium; deep/strong/intense -> high or xhigh; maximum/superduper/biggest -> max.',
   'Return reasoning even when the requested model is not Codex; validation will ignore reasoning that cannot apply.',
   'Map OpenAI model aliases to canonical IDs: sol -> gpt-5.6-sol, terra -> gpt-5.6-terra, luna -> gpt-5.6-luna, 5.5 -> gpt-5.5, 5.5 pro -> gpt-5.5-pro, 5.4 -> gpt-5.4, 5.4 pro -> gpt-5.4-pro, 5.4 mini -> gpt-5.4-mini, 5.4 nano -> gpt-5.4-nano.',
@@ -60,12 +61,15 @@ const MESSAGE_OVERRIDES_SCHEMA = {
       enum: ['responses', 'amazon-bedrock', 'openrouter', null],
       type: ['string', 'null']
     },
+    queue: {
+      type: 'boolean'
+    },
     reasoning: {
       enum: ['none', 'minimal', 'low', 'medium', 'high', 'xhigh', 'max', null],
       type: ['string', 'null']
     }
   },
-  required: ['harness', 'model', 'provider', 'reasoning'],
+  required: ['harness', 'model', 'provider', 'queue', 'reasoning'],
   type: 'object'
 }
 
@@ -83,6 +87,7 @@ type OpenAiMessageOverridesStrategyOutput = {
   harness?: unknown
   model?: unknown
   provider?: unknown
+  queue?: unknown
   reasoning?: unknown
 }
 

services/slackbotv2/src/overrides.ts

@@ -6,6 +6,7 @@
  *   --model <name> (or --model=<name>)           pick the model within that harness
  *   -rsn <effort> (or -rsn=<effort>)             per-turn reasoning effort (codex)
  *   --fable | --opus | --sonnet | --haiku        model shortcuts (imply claude-code)
+ *   --queue                                       run after the active turn instead of steering it
  *
  * Flags are stripped from the text before it reaches the agent. The harness
  * applies at session creation — an explicit harness flag on a thread pinned to
@@ -29,6 +30,8 @@ export type HarnessOverrides = {
   harnessType?: string
   model?: string
   provider?: string
+  /** Per-message delivery behavior; never persisted as a sticky thread override. */
+  queue?: boolean
   reasoning?: string
 }
 
@@ -141,8 +144,15 @@ export function extractMessageOverrides(text: string): MessageOverrides {
   let model: string | undefined
   let modelAliasHarness: string | undefined
   let provider: string | undefined
+  let queue: boolean | undefined
   let reasoning: string | undefined
 
+  const queueMatch = flagPattern('queue').exec(cleaned)
+  if (queueMatch) {
+    queue = true
+    cleaned = stripMatch(cleaned, queueMatch)
+  }
+
   const modelMatch = MODEL_FLAG_PATTERN.exec(cleaned)
   if (modelMatch) {
     const value = modelMatch[1]!
@@ -193,6 +203,7 @@ export function extractMessageOverrides(text: string): MessageOverrides {
     harnessType,
     model,
     provider,
+    ...(queue ? { queue } : {}),
     reasoning
   }
 }
@@ -202,13 +213,15 @@ export function validateStrategyOverrides(
     harness?: unknown
     model?: unknown
     provider?: unknown
+    queue?: unknown
     reasoning?: unknown
   } | null | undefined
 ): HarnessOverrides {
   if (!raw || typeof raw !== 'object') return {}
   let harnessType: string | undefined
   let model: string | undefined
   let provider: string | undefined
+  const queue = raw.queue === true ? true : undefined
   let reasoning: string | undefined
 
   const harnessRaw = cleanString(raw.harness)
@@ -243,7 +256,7 @@ export function validateStrategyOverrides(
     reasoning = harnessType === undefined || harnessType === 'codex' ? normalized : undefined
   }
 
-  return { harnessType, model, provider, reasoning }
+  return { harnessType, model, provider, ...(queue ? { queue } : {}), reasoning }
 }
 
 /**

services/slackbotv2/src/session-api.ts

@@ -492,7 +492,14 @@ export async function forwardToSessionApi(
     const appendStartedAtMs = nowMs()
     await recordSessionApiOperation(
       'append_messages',
-      () => appendSessionMessages(options, input.threadId, input.messages, !input.executeMessage),
+      () =>
+        appendSessionMessages(
+          options,
+          input.threadId,
+          input.messages,
+          !input.executeMessage,
+          input.steerActiveExecution
+        ),
       sessionApiTimeoutMs(options),
       'append session messages'
     )
@@ -1205,13 +1212,15 @@ async function appendSessionMessages(
   options: SlackbotV2Options,
   threadId: string,
   messages: SlackbotV2ApiMessage[],
-  includeRequesterContext = false
+  includeRequesterContext = false,
+  steerActiveExecution = true
 ): Promise<void> {
   const fetchFn = options.fetch ?? fetch
   const body: SlackbotV2AppendMessagesRequest = {
     messages: await Promise.all(
       messages.map(message => toSessionMessage(options, message, includeRequesterContext))
-    )
+    ),
+    steer_active_execution: steerActiveExecution
   }
   const response = await fetchWithTimeout(
     fetchFn,