diff --git a/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift b/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift index 1416d7864bfc..d0279a32026f 100644 --- a/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift +++ b/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift @@ -12998,6 +12998,7 @@ public struct ChatSendParams: Codable, Sendable { public let fastmodevalue: AnyCodable? public var fastmode: Bool? { fastmodevalue?.value as? Bool } public let fastautoonseconds: Int? + public let queuemode: String? public let deliver: Bool? public let originatingchannel: String? public let originatingto: String? @@ -13019,6 +13020,7 @@ public struct ChatSendParams: Codable, Sendable { thinking: String? = nil, fastmodevalue: AnyCodable? = nil, fastautoonseconds: Int? = nil, + queuemode: String? = nil, deliver: Bool? = nil, originatingchannel: String? = nil, originatingto: String? = nil, @@ -13039,6 +13041,7 @@ public struct ChatSendParams: Codable, Sendable { self.thinking = thinking self.fastmodevalue = fastmodevalue self.fastautoonseconds = fastautoonseconds + self.queuemode = queuemode self.deliver = deliver self.originatingchannel = originatingchannel self.originatingto = originatingto @@ -13060,6 +13063,7 @@ public struct ChatSendParams: Codable, Sendable { message: String, thinking: String? = nil, fastmode: Bool?, + queuemode: String? = nil, deliver: Bool? = nil, originatingchannel: String? = nil, originatingto: String? = nil, @@ -13081,6 +13085,7 @@ public struct ChatSendParams: Codable, Sendable { thinking: thinking, fastmodevalue: fastmode.map { AnyCodable($0) }, fastautoonseconds: nil, + queuemode: queuemode, deliver: deliver, originatingchannel: originatingchannel, originatingto: originatingto, @@ -13103,6 +13108,7 @@ public struct ChatSendParams: Codable, Sendable { case thinking case fastmodevalue = "fastMode" case fastautoonseconds = "fastAutoOnSeconds" + case queuemode = "queueMode" case deliver case originatingchannel = "originatingChannel" case originatingto = "originatingTo" diff --git a/docs/gateway/protocol.md b/docs/gateway/protocol.md index c18905079b57..26fbd3bd0809 100644 --- a/docs/gateway/protocol.md +++ b/docs/gateway/protocol.md @@ -539,7 +539,7 @@ methods. Treat this as feature discovery, not a full enumeration of - Chat execution still uses `chat.history`, `chat.send`, `chat.abort`, and `chat.inject`. `chat.history` is display-normalized for UI clients: inline directive tags are stripped from visible text, plain-text tool-call XML payloads (`...`, `...`, `...`, `...`, and truncated tool-call blocks) and leaked ASCII/full-width model control tokens are stripped, pure silent-token assistant rows (exact `NO_REPLY` / `no_reply`) are omitted, and oversized rows can be replaced with placeholders. - `chat.message.get` is the additive bounded full-message reader for a single visible transcript entry. Pass `sessionKey`, optional `agentId` when session selection is agent-scoped, and a transcript `messageId` previously surfaced through `chat.history`; the gateway returns the same display-normalized projection without the lightweight history truncation cap when the stored entry is still available and not oversized. - `chat.toolTitles` returns short purpose titles for tool calls rendered in the Control UI (batched, max 24 items with bounded inputs). The feature is opt-in via `gateway.controlUi.toolTitles` (default off); disabled gateways answer `{ titles: {}, disabled: true }` with no model call so clients stop asking. When enabled, titles use standard utility-model routing: an explicitly configured `utilityModel` (an operator decision that, like all utility tasks, may send bounded task content to the chosen provider), else the session provider's declared small-model default so no new egress destination appears implicitly; an empty `utilityModel` disables them entirely. Titles never fall back to the primary model. Results cache in the per-agent state database keyed by tool name + input, so repeated views never re-bill the same calls. - - `chat.send` accepts one-turn `fastMode: "auto"` to use fast mode for model calls started before the auto cutoff, then start later retry, fallback, tool-result, or continuation calls without fast mode. The cutoff defaults to 60 seconds (`DEFAULT_FAST_MODE_AUTO_ON_SECONDS`) and can be configured per model with `agents.defaults.models["/"].params.fastAutoOnSeconds`. A `chat.send` caller can pass one-turn `fastAutoOnSeconds` to override the cutoff for that request. + - `chat.send` accepts one-turn `fastMode: "auto"` to use fast mode for model calls started before the auto cutoff, then start later retry, fallback, tool-result, or continuation calls without fast mode. The cutoff defaults to 60 seconds (`DEFAULT_FAST_MODE_AUTO_ON_SECONDS`) and can be configured per model with `agents.defaults.models["/"].params.fastAutoOnSeconds`. A `chat.send` caller can pass one-turn `fastAutoOnSeconds` to override the cutoff for that request. Pass `queueMode` (`steer`, `followup`, `collect`, or `interrupt`) to override the stored queue mode for this request only; explicit Control UI steer actions use `queueMode: "steer"`. diff --git a/packages/gateway-protocol/src/index.test.ts b/packages/gateway-protocol/src/index.test.ts index 66c7faa2f019..5f7ea6b52de4 100644 --- a/packages/gateway-protocol/src/index.test.ts +++ b/packages/gateway-protocol/src/index.test.ts @@ -952,6 +952,19 @@ describe("validateChatSendParams", () => { expect(validateChatSendParams({ ...base, fastAutoOnSeconds: 2 })).toBe(true); expect(validateChatSendParams({ ...base, fastAutoOnSeconds: 0 })).toBe(false); }); + + it("accepts one-turn queue mode overrides", () => { + const base = { + sessionKey: "agent:main:main", + message: "hello", + idempotencyKey: "run-1", + }; + + for (const queueMode of ["steer", "followup", "collect", "interrupt"] as const) { + expect(validateChatSendParams({ ...base, queueMode })).toBe(true); + } + expect(validateChatSendParams({ ...base, queueMode: "invalid" })).toBe(false); + }); }); describe("validateModelsListParams", () => { diff --git a/packages/gateway-protocol/src/schema/logs-chat.ts b/packages/gateway-protocol/src/schema/logs-chat.ts index a7e13201c3b7..b596ad244cd7 100644 --- a/packages/gateway-protocol/src/schema/logs-chat.ts +++ b/packages/gateway-protocol/src/schema/logs-chat.ts @@ -95,6 +95,8 @@ export const ChatSendParamsSchema = closedObject({ fastMode: Type.Optional(Type.Union([Type.Boolean(), Type.Literal("auto")])), // One-turn override for auto fast-mode cutoff seconds. fastAutoOnSeconds: Type.Optional(Type.Integer({ minimum: 1 })), + // One-turn override for active-run queue admission. + queueMode: Type.Optional(Type.String({ enum: ["steer", "followup", "collect", "interrupt"] })), deliver: Type.Optional(Type.Boolean()), originatingChannel: Type.Optional(Type.String()), originatingTo: Type.Optional(Type.String()), diff --git a/src/auto-reply/reply/get-reply-run.media-only.test.ts b/src/auto-reply/reply/get-reply-run.media-only.test.ts index a3ef2d39fef8..6644c5f09349 100644 --- a/src/auto-reply/reply/get-reply-run.media-only.test.ts +++ b/src/auto-reply/reply/get-reply-run.media-only.test.ts @@ -796,6 +796,38 @@ describe("runPreparedReply media-only handling", () => { expect(call?.followupRun.originatingChannel).toBe(channel); }); + it("prefers a one-turn queue override over the stored session mode", async () => { + const queueSettings = await import("./queue/settings-runtime.js"); + const embeddedAgentRuntime = await import("../../agents/embedded-agent.runtime.js"); + vi.mocked(queueSettings.resolveQueueSettings).mockImplementationOnce((params) => ({ + mode: params.inlineMode ?? params.sessionEntry?.queueMode ?? "steer", + })); + vi.mocked(embeddedAgentRuntime.resolveActiveEmbeddedRunSessionId) + .mockReturnValueOnce("active-session") + .mockReturnValueOnce("active-session"); + vi.mocked(embeddedAgentRuntime.isEmbeddedAgentRunActive).mockReturnValueOnce(true); + vi.mocked(embeddedAgentRuntime.isEmbeddedAgentRunStreaming).mockReturnValueOnce(true); + + await runPreparedReply( + baseParams({ + sessionEntry: { + sessionId: "active-session", + updatedAt: Date.now(), + queueMode: "followup", + }, + opts: { queueModeOverride: "steer" }, + }), + ); + + expect(queueSettings.resolveQueueSettings).toHaveBeenCalledWith( + expect.objectContaining({ inlineMode: "steer" }), + ); + expect(requireLastRunReplyAgentCall()).toMatchObject({ + shouldSteer: true, + resolvedQueue: { mode: "steer" }, + }); + }); + it("keeps thread history context on follow-up turns", async () => { const result = await runPreparedReply( baseParams({ diff --git a/src/auto-reply/reply/get-reply-run.ts b/src/auto-reply/reply/get-reply-run.ts index ef37fa5ed17b..6092acd33343 100644 --- a/src/auto-reply/reply/get-reply-run.ts +++ b/src/auto-reply/reply/get-reply-run.ts @@ -538,11 +538,12 @@ export async function runPreparedReply( const isHeartbeat = opts?.isHeartbeat === true; const heartbeatRunScope = resolveHeartbeatRunScope(opts); const explicitThinkingLevelOverride = normalizeThinkLevel(opts?.thinkingLevelOverride); + const effectiveQueueMode = opts?.queueModeOverride ?? perMessageQueueMode; const traceAttributes = { provider, hasSessionKey: Boolean(sessionKey), isHeartbeat, - queueMode: perMessageQueueMode ?? "configured", + queueMode: effectiveQueueMode ?? "configured", }; const traceRunPhase = (name: string, run: () => Promise | T): Promise => measureDiagnosticsTimelineSpan(name, run, { @@ -1138,7 +1139,7 @@ export async function runPreparedReply( cfg, channel: sessionCtx.Provider, sessionEntry, - inlineMode: perMessageQueueMode, + inlineMode: effectiveQueueMode, inlineOptions: perMessageQueueOptions, }); const embeddedAgentRuntime = useFastReplyRuntime diff --git a/src/auto-reply/reply/get-reply.types.ts b/src/auto-reply/reply/get-reply.types.ts index 07fbd45155c0..9d13204d818e 100644 --- a/src/auto-reply/reply/get-reply.types.ts +++ b/src/auto-reply/reply/get-reply.types.ts @@ -4,6 +4,7 @@ import type { ReplyOptionsWithHeartbeatRunScope } from "../../infra/heartbeat-ru import type { GetReplyOptions } from "../get-reply-options.types.js"; import type { ReplyPayload } from "../reply-payload.js"; import type { MsgContext } from "../templating.js"; +import type { QueueMode } from "./queue/types.js"; export type ReplySessionBinding = { sessionKey?: string; @@ -21,6 +22,8 @@ type InternalReplySessionOptions = { sessionPromptSourceReplyDeliveryMode?: GetReplyOptions["sourceReplyDeliveryMode"]; /** Marks when this reply is waiting to own its session's reply lane. */ onReplyAdmissionWaitChange?: (waiting: boolean) => void; + /** Overrides persisted queue mode for this reply only. */ + queueModeOverride?: QueueMode; }; export type InternalGetReplyOptions = GetReplyOptions & diff --git a/src/gateway/server-methods/chat-send-request.ts b/src/gateway/server-methods/chat-send-request.ts index d70fc9c52c3a..ce708e3694ca 100644 --- a/src/gateway/server-methods/chat-send-request.ts +++ b/src/gateway/server-methods/chat-send-request.ts @@ -10,6 +10,7 @@ import { validateChatSendParams, } from "../../../packages/gateway-protocol/src/index.js"; import { isBtwRequestText } from "../../auto-reply/reply/btw-command.js"; +import type { QueueMode } from "../../auto-reply/reply/queue/types.js"; import type { InputProvenance } from "../../sessions/input-provenance.js"; import { normalizeInputProvenance } from "../../sessions/input-provenance.js"; import { isOperatorUiClient } from "../../utils/message-channel.js"; @@ -34,6 +35,7 @@ type ChatSendRequestParams = { thinking?: string; fastMode?: FastMode; fastAutoOnSeconds?: number; + queueMode?: QueueMode; deliver?: boolean; originatingChannel?: string; originatingTo?: string; diff --git a/src/gateway/server-methods/chat.ts b/src/gateway/server-methods/chat.ts index 2b446aa2f98a..884f294d8c02 100644 --- a/src/gateway/server-methods/chat.ts +++ b/src/gateway/server-methods/chat.ts @@ -1317,6 +1317,7 @@ export const chatHandlers: GatewayRequestHandlers = { imageOrder: imageOrder.length > 0 ? imageOrder : undefined, thinkingLevelOverride: p.thinking, fastModeOverride: p.fastMode, + queueModeOverride: p.queueMode, userTurnTranscriptRecorder: userTurnRecorder, ...(restartSafeAdmission ? { suppressNextUserMessagePersistence: true } : {}), fastModeAutoOnSecondsOverride: p.fastAutoOnSeconds, diff --git a/src/gateway/server.chat.gateway-server-chat-b.test.ts b/src/gateway/server.chat.gateway-server-chat-b.test.ts index dcd639dc98ea..2c451234c5c1 100644 --- a/src/gateway/server.chat.gateway-server-chat-b.test.ts +++ b/src/gateway/server.chat.gateway-server-chat-b.test.ts @@ -5247,6 +5247,47 @@ describe("gateway server chat", () => { ); }); + test("chat.send forwards one-turn queue mode overrides internally", async () => { + await withGatewayChatHarness( + async ({ ws, createSessionDir }) => { + const spy = getReplyFromConfig; + await connectOk(ws, { + client: { + id: GATEWAY_CLIENT_NAMES.CONTROL_UI, + version: "1.0.0", + platform: "web", + mode: GATEWAY_CLIENT_MODES.WEBCHAT, + }, + }); + + await createSessionDir(); + await writeMainSessionStore(); + let capturedOpts: InternalGetReplyOptions | undefined; + mockGetReplyFromConfigOnce(async (_ctx, opts) => { + capturedOpts = opts; + return undefined; + }); + + const sendRes = await rpcReq(ws, "chat.send", { + sessionKey: "main", + message: "steer this turn", + queueMode: "steer", + idempotencyKey: "idem-queue-mode-override", + }); + expect(sendRes.ok).toBe(true); + + await vi.waitFor(() => { + expect(spy.mock.calls.length).toBeGreaterThan(0); + }, FAST_WAIT_OPTS); + + expect(capturedOpts).toMatchObject({ queueModeOverride: "steer" }); + }, + { + headers: { origin: `http://127.0.0.1:${harness.port}` }, + }, + ); + }); + test("chat.history hard-caps single oversized nested payloads", async () => { await withGatewayChatHarness(async ({ ws, createSessionDir }) => { const sessionDir = await prepareMainHistoryHarness({ ws, createSessionDir }); diff --git a/ui/src/pages/chat/chat-command-executor.test.ts b/ui/src/pages/chat/chat-command-executor.test.ts index 9ac63c03159a..ff7e5c2b8e79 100644 --- a/ui/src/pages/chat/chat-command-executor.test.ts +++ b/ui/src/pages/chat/chat-command-executor.test.ts @@ -1273,6 +1273,7 @@ describe("executeSlashCommand /steer (soft inject)", () => { expect(chatSend.payload.sessionKey).toBe("agent:main:main"); expect(chatSend.payload.message).toBe("try a different approach"); expect(chatSend.payload.deliver).toBe(false); + expect(chatSend.payload.queueMode).toBe("steer"); }); it("uses canonical active-run state when the session row only reports hasActiveRun", async () => { diff --git a/ui/src/pages/chat/chat-command-executor.ts b/ui/src/pages/chat/chat-command-executor.ts index f55b833e394e..fc9755911fd1 100644 --- a/ui/src/pages/chat/chat-command-executor.ts +++ b/ui/src/pages/chat/chat-command-executor.ts @@ -738,6 +738,7 @@ async function executeSteer( ...selectedGlobalScope(resolved.key, context), message: resolved.message, deliver: false, + queueMode: "steer", idempotencyKey: generateUUID(), }), ); diff --git a/ui/src/pages/chat/chat-send.test.ts b/ui/src/pages/chat/chat-send.test.ts index 917fbe863848..c61493d88f6b 100644 --- a/ui/src/pages/chat/chat-send.test.ts +++ b/ui/src/pages/chat/chat-send.test.ts @@ -3272,6 +3272,7 @@ describe("handleSendChat", () => { sessionKey: "agent:main:main", message: "tighten the plan", deliver: false, + queueMode: "steer", }), ), ); @@ -6765,6 +6766,7 @@ describe("handleSendChat", () => { sessionKey: "agent:main:main", message: "tighten the plan", deliver: false, + queueMode: "steer", idempotencyKey, attachments: undefined, }); diff --git a/ui/src/pages/chat/chat-send.ts b/ui/src/pages/chat/chat-send.ts index 11829edd47a3..3fa3074ccacc 100644 --- a/ui/src/pages/chat/chat-send.ts +++ b/ui/src/pages/chat/chat-send.ts @@ -254,6 +254,7 @@ async function requestChatSend( runId: string; sessionKey?: string; agentId?: string; + queueMode?: "steer"; }, ): Promise { const routing = resolveChatSendRouting(state, params); @@ -269,6 +270,7 @@ async function requestChatSend( ...(controlUiReconnectResume ? { __controlUiReconnectResume: true } : {}), message: params.message, deliver: false, + ...(params.queueMode ? { queueMode: params.queueMode } : {}), idempotencyKey: params.runId, attachments: buildChatApiAttachments(params.attachments), }); @@ -352,8 +354,11 @@ async function sendChatMessageWithGeneratedRunId( state: ChatState, message: string, attachments?: ChatAttachment[], - canApplyError: () => boolean = () => true, - runIdOverride?: string, + options: { + canApplyError?: () => boolean; + queueMode?: "steer"; + runId?: string; + } = {}, ): Promise { if (!state.client || !state.connected) { return null; @@ -363,12 +368,18 @@ async function sendChatMessageWithGeneratedRunId( if (!msg && !hasAttachments) { return null; } + const canApplyError = options.canApplyError ?? (() => true); if (canApplyError()) { setChatError(state, null); } - const runId = runIdOverride ?? generateUUID(); + const runId = options.runId ?? generateUUID(); try { - return await requestChatSend(state, { message: msg, attachments, runId }); + return await requestChatSend(state, { + message: msg, + attachments, + runId, + ...(options.queueMode ? { queueMode: options.queueMode } : {}), + }); } catch (err) { if (canApplyError()) { setChatError(state, formatConnectError(err)); @@ -383,7 +394,7 @@ async function sendDetachedChatMessage( attachments?: ChatAttachment[], runId?: string, ): Promise { - return sendChatMessageWithGeneratedRunId(state, message, attachments, () => true, runId); + return sendChatMessageWithGeneratedRunId(state, message, attachments, { runId }); } function isChatResetCommand(text: string) { @@ -1364,7 +1375,10 @@ export async function steerQueuedChatMessage(host: ChatHost, id: string) { host as unknown as ChatState, message, hasAttachments ? attachments : undefined, - () => visibleSessionMatches(host, itemSessionKey, item.agentId), + { + canApplyError: () => visibleSessionMatches(host, itemSessionKey, item.agentId), + queueMode: "steer", + }, ); const pendingStillVisible = activeRunId ? host.chatQueue.some((entry) => entry.id === id && entry.pendingRunId === activeRunId)