diff --git a/docs/pages/deploying-in-production.mdx b/docs/pages/deploying-in-production.mdx index 64949546c..a4ad76ee1 100644 --- a/docs/pages/deploying-in-production.mdx +++ b/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: diff --git a/docs/public/md/deploying-in-production.md b/docs/public/md/deploying-in-production.md index 64949546c..a4ad76ee1 100644 --- a/docs/public/md/deploying-in-production.md +++ b/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: diff --git a/services/api-rs/crates/centaur-api-server/src/routes.rs b/services/api-rs/crates/centaur-api-server/src/routes.rs index 37cfc0ec2..5ed90ad68 100644 --- a/services/api-rs/crates/centaur-api-server/src/routes.rs +++ b/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, diff --git a/services/api-rs/crates/centaur-api-server/src/types.rs b/services/api-rs/crates/centaur-api-server/src/types.rs index 6721ee5f3..0a10478af 100644 --- a/services/api-rs/crates/centaur-api-server/src/types.rs +++ b/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, + /// 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)] diff --git a/services/api-rs/crates/centaur-session-cli/src/main.rs b/services/api-rs/crates/centaur-session-cli/src/main.rs index 79c7b3988..c4a86016b 100644 --- a/services/api-rs/crates/centaur-session-cli/src/main.rs +++ b/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?; diff --git a/services/api-rs/crates/centaur-session-runtime/src/lib.rs b/services/api-rs/crates/centaur-session-runtime/src/lib.rs index 7d58ce718..ab1268d4e 100644 --- a/services/api-rs/crates/centaur-session-runtime/src/lib.rs +++ b/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, 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"); diff --git a/services/api-rs/crates/centaur-workflows/src/lib.rs b/services/api-rs/crates/centaur-workflows/src/lib.rs index bb4792672..70488f89e 100644 --- a/services/api-rs/crates/centaur-workflows/src/lib.rs +++ b/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 diff --git a/services/slackbotv2/src/index.ts b/services/slackbotv2/src/index.ts index 90aebc0bc..27c7daaa4 100644 --- a/services/slackbotv2/src/index.ts +++ b/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 const DEFAULT_MESSAGE_OVERRIDES_STRATEGY = createFlagMessageOverridesStrategy() +const queuedExecutionChains = new Map>() 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> + /** 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, + message: ChatMessage, + input: SyncThreadMessageInput, + resolvedMessageOverrides: Awaited>, + 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, message: SlackbotV2ApiMessage, diff --git a/services/slackbotv2/src/message-overrides-strategy.ts b/services/slackbotv2/src/message-overrides-strategy.ts index ea4edad1b..78360794d 100644 --- a/services/slackbotv2/src/message-overrides-strategy.ts +++ b/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 } diff --git a/services/slackbotv2/src/overrides.ts b/services/slackbotv2/src/overrides.ts index 636a42ea2..f489236cb 100644 --- a/services/slackbotv2/src/overrides.ts +++ b/services/slackbotv2/src/overrides.ts @@ -6,6 +6,7 @@ * --model (or --model=) pick the model within that harness * -rsn (or -rsn=) 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,6 +213,7 @@ export function validateStrategyOverrides( harness?: unknown model?: unknown provider?: unknown + queue?: unknown reasoning?: unknown } | null | undefined ): HarnessOverrides { @@ -209,6 +221,7 @@ export function validateStrategyOverrides( 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 } } /** diff --git a/services/slackbotv2/src/session-api.ts b/services/slackbotv2/src/session-api.ts index 3c0ef7f30..88041ec44 100644 --- a/services/slackbotv2/src/session-api.ts +++ b/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 { 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, diff --git a/services/slackbotv2/src/types.ts b/services/slackbotv2/src/types.ts index 26b380339..1060608be 100644 --- a/services/slackbotv2/src/types.ts +++ b/services/slackbotv2/src/types.ts @@ -69,6 +69,8 @@ export type SlackbotV2SessionMessage = { export type SlackbotV2AppendMessagesRequest = { messages: SlackbotV2SessionMessage[] + /** Defaults to true. False persists the messages without steering an active turn. */ + steer_active_execution?: boolean } export type SlackbotV2CreateSessionRequest = { @@ -270,6 +272,8 @@ export type ForwardSessionInput = { metadataModel?: string /** Effective model provider selected by sticky thread flags (--bedrock); codex only. */ provider?: string + /** Whether appending these messages may steer the currently active execution. */ + steerActiveExecution?: boolean /** Per-turn reasoning effort parsed from the `-rsn` flag (codex only). */ reasoning?: string onEventId(eventId: number): void diff --git a/services/slackbotv2/test/chat-sdk-emulate.test.ts b/services/slackbotv2/test/chat-sdk-emulate.test.ts index d46624fa4..6dfa8e7eb 100644 --- a/services/slackbotv2/test/chat-sdk-emulate.test.ts +++ b/services/slackbotv2/test/chat-sdk-emulate.test.ts @@ -2237,6 +2237,80 @@ describe('slackbotv2', () => { await Promise.all(firstWaits) }) + it('queues a --queue mention as the next turn without steering the active turn', async () => { + codexApi.autoRespond = false + + const parent = await postUserMessage('Context before queued execution.') + const firstMention = await postUserMessage(`<@${BOT_USER_ID}> start a long run`, parent.ts) + const firstWaits: Promise[] = [] + const firstResponse = await bot.app.request( + '/api/webhooks/slack', + signedSlackEvent({ + event_id: 'Ev-slackbotv2-queue-first', + event: { + type: 'app_mention', + user: USER_ID, + channel: CHANNEL_ID, + team: TEAM_ID, + ts: firstMention.ts, + thread_ts: parent.ts, + text: `<@${BOT_USER_ID}> start a long run` + } + }), + {}, + waitUntilContext(firstWaits) + ) + expect(firstResponse.status).toBe(200) + await waitFor(() => codexApi.executes.length === 1) + await waitFor(() => codexApi.eventRequests.length === 1) + + const queuedText = `<@${BOT_USER_ID}> --queue run this after the current turn` + const queuedMention = await postUserMessage(queuedText, parent.ts) + const queuedWaits: Promise[] = [] + const queuedResponse = await bot.app.request( + '/api/webhooks/slack', + signedSlackEvent({ + event_id: 'Ev-slackbotv2-queue-second', + event: { + type: 'app_mention', + user: USER_ID, + channel: CHANNEL_ID, + team: TEAM_ID, + ts: queuedMention.ts, + thread_ts: parent.ts, + text: queuedText + } + }), + {}, + waitUntilContext(queuedWaits) + ) + expect(queuedResponse.status).toBe(200) + await waitFor(() => codexApi.appends.length === 2) + expect(codexApi.executes).toHaveLength(1) + expect(codexApi.appends[1]!.body.steer_active_execution).toBe(false) + expect(sessionMessageTexts(codexApi.appends[1]!.body.messages).at(-1)).toBe( + `@${BOT_USER_ID} run this after the current turn` + ) + + const firstExecutionId = codexApi.eventRequests[0]!.executionId + codexApi.emitOutputLines( + threadKey(parent.ts), + sampleCodexOutputLines('First turn complete.'), + firstExecutionId + ) + await waitFor(() => codexApi.executes.length === 2) + await waitFor(() => codexApi.eventRequests.length === 2) + + const secondExecutionId = codexApi.eventRequests[1]!.executionId + codexApi.emitOutputLines( + threadKey(parent.ts), + sampleCodexOutputLines('Queued turn complete.'), + secondExecutionId + ) + await Promise.all([...firstWaits, ...queuedWaits]) + expect(await threadText(parent.ts)).toContain('Queued turn complete.') + }) + it('renders raw turn.failed session output as visible final text', async () => { codexApi.autoRespond = false diff --git a/services/slackbotv2/test/overrides.test.ts b/services/slackbotv2/test/overrides.test.ts index 175ad09a0..57a3e045e 100644 --- a/services/slackbotv2/test/overrides.test.ts +++ b/services/slackbotv2/test/overrides.test.ts @@ -173,6 +173,20 @@ describe('extractMessageOverrides', () => { expect(extractMessageOverrides('--ampere hi').harnessType).toBeUndefined() }) + test('parses --queue as a non-sticky delivery flag and strips it', () => { + expect(extractMessageOverrides('--queue do this next')).toEqual({ + cleanedText: 'do this next', + harnessType: undefined, + model: undefined, + provider: undefined, + queue: true, + reasoning: undefined + }) + expect(extractMessageOverrides('do this --queue next').cleanedText).toBe('do this next') + expect(extractMessageOverrides('document the --queue flag').queue).toBe(true) + expect(extractMessageOverrides('pre--queue remains literal').queue).toBeUndefined() + }) + test('flag-only message cleans to empty text', () => { expect(extractMessageOverrides('--claude')).toEqual({ cleanedText: '', diff --git a/services/slackbotv2/test/session-api.test.ts b/services/slackbotv2/test/session-api.test.ts index 798c227e4..91438b35f 100644 --- a/services/slackbotv2/test/session-api.test.ts +++ b/services/slackbotv2/test/session-api.test.ts @@ -515,6 +515,22 @@ describe('Slack attachment serialization', () => { }) describe('forwardToSessionApi overrides', () => { + test('can append a queued message without steering an active execution', async () => { + const { fetchFn, requests } = fakeApi() + await forwardToSessionApi( + options(fetchFn), + forwardInput(apiMessage('do this next'), { + executeMessage: undefined, + steerActiveExecution: false + }) + ) + const append = requests.find(request => request.url.endsWith('/messages')) + expect((append?.body as { steer_active_execution?: boolean }).steer_active_execution).toBe( + false + ) + expect(requests.some(request => request.url.endsWith('/execute'))).toBe(false) + }) + test('creates session with default codex harness', async () => { const { fetchFn, requests } = fakeApi() await forwardToSessionApi(options(fetchFn), forwardInput(apiMessage('hi')))