diff --git a/src/tui/commands.test.ts b/src/tui/commands.test.ts index f8fa432061e8..96d6c4677a5d 100644 --- a/src/tui/commands.test.ts +++ b/src/tui/commands.test.ts @@ -153,7 +153,7 @@ describe("getSlashCommands", () => { expect(names).toEqual( expect.not.arrayContaining(["commands", "status", "compact", "context", "tools"]), ); - expect(names).toEqual(expect.arrayContaining(["goal", "btw", "side", "stop", "t"])); + expect(names).toEqual(expect.arrayContaining(["goal", "btw", "side", "queue", "stop", "t"])); }); }); diff --git a/src/tui/commands.ts b/src/tui/commands.ts index 53a1327d98c3..8e83b9c53d94 100644 --- a/src/tui/commands.ts +++ b/src/tui/commands.ts @@ -41,7 +41,7 @@ const COMMAND_ALIASES: Record = { // These shared commands have explicit local TUI routing but no same-named // built-in autocomplete entry. Other shared commands require the Gateway and // must stay out of local autocomplete and model prompts. -const LOCAL_TUI_ROUTED_SHARED_COMMANDS = new Set(["btw", "goal", "stop"]); +const LOCAL_TUI_ROUTED_SHARED_COMMANDS = new Set(["btw", "goal", "queue", "stop"]); function createLevelCompletion( levels: string[], diff --git a/src/tui/embedded-backend.test.ts b/src/tui/embedded-backend.test.ts index 4e9b4076af67..e4032dca619c 100644 --- a/src/tui/embedded-backend.test.ts +++ b/src/tui/embedded-backend.test.ts @@ -15,6 +15,8 @@ import { notifyListeners } from "../shared/listeners.js"; import { withEnvAsync } from "../test-utils/env.js"; const agentCommandFromIngressMock = vi.fn(); +const queueEmbeddedAgentMessageWithOutcomeAsyncMock = vi.fn(); +const resolveActiveEmbeddedRunSessionIdMock = vi.fn(); const runBtwSideQuestionMock = vi.fn(); const updateSessionStoreMock = vi.fn(); const applySessionPatchProjectionMock = vi.fn(); @@ -83,6 +85,13 @@ vi.mock("../agents/agent-command.js", () => ({ agentCommandFromIngress: (...args: unknown[]) => agentCommandFromIngressMock(...args), })); +vi.mock("../agents/embedded-agent-runner/runs.js", () => ({ + queueEmbeddedAgentMessageWithOutcomeAsync: (...args: unknown[]) => + queueEmbeddedAgentMessageWithOutcomeAsyncMock(...args), + resolveActiveEmbeddedRunSessionId: (...args: unknown[]) => + resolveActiveEmbeddedRunSessionIdMock(...args), +})); + vi.mock("../agents/btw.js", () => ({ runBtwSideQuestion: (...args: unknown[]) => runBtwSideQuestionMock(...args), })); @@ -276,6 +285,9 @@ describe("EmbeddedTuiBackend", () => { vi.useFakeTimers(); vi.setSystemTime(embeddedEventTimestamp); agentCommandFromIngressMock.mockReset(); + queueEmbeddedAgentMessageWithOutcomeAsyncMock.mockReset(); + resolveActiveEmbeddedRunSessionIdMock.mockReset(); + resolveActiveEmbeddedRunSessionIdMock.mockReturnValue(undefined); runBtwSideQuestionMock.mockReset(); updateSessionStoreMock.mockReset(); updateSessionStoreMock.mockImplementation( @@ -1279,6 +1291,345 @@ describe("EmbeddedTuiBackend", () => { }); }); + it("steers same-session sends into the active local run", async () => { + const { EmbeddedTuiBackend } = await import("./embedded-backend.js"); + const first = deferred<{ + payloads: Array<{ text: string }>; + meta: Record; + }>(); + agentCommandFromIngressMock.mockReturnValueOnce(first.promise); + resolveActiveEmbeddedRunSessionIdMock.mockReturnValue("active-session"); + loadSessionEntryMock.mockImplementation((sessionKey: string) => ({ + cfg: { messages: { queue: { debounceMs: 125 } } }, + canonicalKey: sessionKey, + storePath: "/tmp/openclaw-sessions.json", + store: {}, + entry: {}, + })); + queueEmbeddedAgentMessageWithOutcomeAsyncMock.mockResolvedValue({ + queued: true, + sessionId: "active-session", + target: "embedded_run", + gatewayHealth: "live", + }); + + const backend = new EmbeddedTuiBackend(); + backend.start(); + await backend.sendChat({ + sessionKey: "agent:main:main", + message: "first", + runId: "run-local-first", + }); + + const result = await backend.sendChat({ + sessionKey: "agent:main:main", + message: "steer this turn", + runId: "run-local-second", + }); + + expect(result).toEqual({ runId: "run-local-first" }); + expect(queueEmbeddedAgentMessageWithOutcomeAsyncMock).toHaveBeenCalledWith( + "active-session", + "steer this turn", + { steeringMode: "all", debounceMs: 125 }, + ); + expect(agentCommandFromIngressMock).toHaveBeenCalledTimes(1); + + first.resolve({ payloads: [{ text: "done" }], meta: {} }); + await flushMicrotasks(); + }); + + it("queues local sends when active-runtime steering rejects them", async () => { + const { EmbeddedTuiBackend } = await import("./embedded-backend.js"); + const first = deferred<{ + payloads: Array<{ text: string }>; + meta: Record; + }>(); + const second = deferred<{ + payloads: Array<{ text: string }>; + meta: Record; + }>(); + agentCommandFromIngressMock + .mockReturnValueOnce(first.promise) + .mockReturnValueOnce(second.promise); + resolveActiveEmbeddedRunSessionIdMock.mockReturnValue("active-session"); + loadSessionEntryMock.mockImplementation((sessionKey: string) => ({ + cfg: { messages: { queue: { debounceMs: 0 } } }, + canonicalKey: sessionKey, + storePath: "/tmp/openclaw-sessions.json", + store: {}, + entry: {}, + })); + queueEmbeddedAgentMessageWithOutcomeAsyncMock.mockResolvedValue({ + queued: false, + sessionId: "active-session", + reason: "runtime_rejected", + gatewayHealth: "live", + }); + + const backend = new EmbeddedTuiBackend(); + backend.start(); + await backend.sendChat({ + sessionKey: "agent:main:main", + message: "first", + runId: "run-local-first", + }); + await backend.sendChat({ + sessionKey: "agent:main:main", + message: "queue on rejection", + runId: "run-local-second", + }); + + expect(agentCommandFromIngressMock).toHaveBeenCalledTimes(1); + first.resolve({ payloads: [{ text: "first done" }], meta: {} }); + await vi.waitFor(() => { + expect(agentCommandFromIngressMock).toHaveBeenCalledTimes(2); + }); + second.resolve({ payloads: [{ text: "second done" }], meta: {} }); + await flushMicrotasks(); + }); + + it("honors a persisted local followup queue override", async () => { + const { EmbeddedTuiBackend } = await import("./embedded-backend.js"); + const first = deferred<{ + payloads: Array<{ text: string }>; + meta: Record; + }>(); + const second = deferred<{ + payloads: Array<{ text: string }>; + meta: Record; + }>(); + agentCommandFromIngressMock + .mockReturnValueOnce(first.promise) + .mockReturnValueOnce(second.promise); + loadSessionEntryMock.mockImplementation((sessionKey: string) => ({ + cfg: { messages: { queue: { mode: "steer" } } }, + canonicalKey: sessionKey, + storePath: "/tmp/openclaw-sessions.json", + store: {}, + entry: { queueMode: "followup", queueDebounceMs: 0 }, + })); + resolveActiveEmbeddedRunSessionIdMock.mockReturnValue("active-session"); + + const backend = new EmbeddedTuiBackend(); + backend.start(); + await backend.sendChat({ + sessionKey: "agent:main:main", + message: "first", + runId: "run-local-first", + }); + await backend.sendChat({ + sessionKey: "agent:main:main", + message: "follow up later", + runId: "run-local-second", + }); + + expect(resolveActiveEmbeddedRunSessionIdMock).not.toHaveBeenCalled(); + expect(queueEmbeddedAgentMessageWithOutcomeAsyncMock).not.toHaveBeenCalled(); + expect(agentCommandFromIngressMock).toHaveBeenCalledTimes(1); + first.resolve({ payloads: [{ text: "first done" }], meta: {} }); + await vi.waitFor(() => { + expect(agentCommandFromIngressMock).toHaveBeenCalledTimes(2); + }); + second.resolve({ payloads: [{ text: "second done" }], meta: {} }); + await flushMicrotasks(); + }); + + it("collects pending local messages into one followup turn", async () => { + const { EmbeddedTuiBackend } = await import("./embedded-backend.js"); + const first = deferred<{ + payloads: Array<{ text: string }>; + meta: Record; + }>(); + const collected = deferred<{ + payloads: Array<{ text: string }>; + meta: Record; + }>(); + agentCommandFromIngressMock + .mockReturnValueOnce(first.promise) + .mockReturnValueOnce(collected.promise); + loadSessionEntryMock.mockImplementation((sessionKey: string) => ({ + cfg: { messages: { queue: { mode: "collect", debounceMs: 0 } } }, + canonicalKey: sessionKey, + storePath: "/tmp/openclaw-sessions.json", + store: {}, + entry: {}, + })); + + const backend = new EmbeddedTuiBackend(); + backend.start(); + await backend.sendChat({ + sessionKey: "agent:main:main", + message: "first", + runId: "run-local-first", + }); + const second = await backend.sendChat({ + sessionKey: "agent:main:main", + message: "collect alpha", + runId: "run-local-second", + }); + const third = await backend.sendChat({ + sessionKey: "agent:main:main", + message: "collect beta", + runId: "run-local-third", + }); + + expect(second).toEqual({ runId: "run-local-second" }); + expect(third).toEqual({ runId: "run-local-second" }); + expect(agentCommandFromIngressMock).toHaveBeenCalledTimes(1); + first.resolve({ payloads: [{ text: "first done" }], meta: {} }); + await vi.waitFor(() => { + expect(agentCommandFromIngressMock).toHaveBeenCalledTimes(2); + }); + const collectedCall = agentCommandFromIngressMock.mock.calls[1]; + if (!collectedCall) { + throw new Error("expected collected local followup call"); + } + const collectedPrompt = (collectedCall[0] as { message: string }).message; + expect(collectedPrompt).toContain("[Queued messages while agent was busy]"); + expect(collectedPrompt).toContain("collect alpha"); + expect(collectedPrompt).toContain("collect beta"); + collected.resolve({ payloads: [{ text: "collected done" }], meta: {} }); + await flushMicrotasks(); + }); + + it("applies the local queue cap and drop-new policy", async () => { + const { EmbeddedTuiBackend } = await import("./embedded-backend.js"); + const first = deferred<{ + payloads: Array<{ text: string }>; + meta: Record; + }>(); + const second = deferred<{ + payloads: Array<{ text: string }>; + meta: Record; + }>(); + agentCommandFromIngressMock + .mockReturnValueOnce(first.promise) + .mockReturnValueOnce(second.promise); + loadSessionEntryMock.mockImplementation((sessionKey: string) => ({ + cfg: { + messages: { queue: { mode: "followup", debounceMs: 0, cap: 1, drop: "new" } }, + }, + canonicalKey: sessionKey, + storePath: "/tmp/openclaw-sessions.json", + store: {}, + entry: {}, + })); + + const backend = new EmbeddedTuiBackend(); + backend.start(); + await backend.sendChat({ + sessionKey: "agent:main:main", + message: "first", + runId: "run-local-first", + }); + await backend.sendChat({ + sessionKey: "agent:main:main", + message: "kept followup", + runId: "run-local-second", + }); + const dropped = await backend.sendChat({ + sessionKey: "agent:main:main", + message: "dropped followup", + runId: "run-local-third", + }); + + expect(dropped).toEqual({ runId: "run-local-second" }); + first.resolve({ payloads: [{ text: "first done" }], meta: {} }); + await vi.waitFor(() => { + expect(agentCommandFromIngressMock).toHaveBeenCalledTimes(2); + }); + expect(agentCommandFromIngressMock.mock.calls[1]?.[0]).toEqual( + expect.objectContaining({ message: "kept followup" }), + ); + second.resolve({ payloads: [{ text: "second done" }], meta: {} }); + await flushMicrotasks(); + }); + + it("interrupts the active local run before starting its replacement", async () => { + const { EmbeddedTuiBackend } = await import("./embedded-backend.js"); + const first = deferred<{ + payloads: Array<{ text: string }>; + meta: Record; + }>(); + const firstAbortListener = vi.fn(() => { + first.resolve({ payloads: [{ text: "first aborted" }], meta: {} }); + }); + agentCommandFromIngressMock + .mockImplementationOnce((opts: { abortSignal?: AbortSignal }) => { + opts.abortSignal?.addEventListener("abort", firstAbortListener); + return first.promise; + }) + .mockResolvedValueOnce({ payloads: [{ text: "replacement done" }], meta: {} }); + loadSessionEntryMock.mockImplementation((sessionKey: string) => ({ + cfg: { messages: { queue: { mode: "interrupt" } } }, + canonicalKey: sessionKey, + storePath: "/tmp/openclaw-sessions.json", + store: {}, + entry: {}, + })); + + const backend = new EmbeddedTuiBackend(); + backend.start(); + await backend.sendChat({ + sessionKey: "agent:main:main", + message: "first", + runId: "run-local-first", + }); + await backend.sendChat({ + sessionKey: "agent:main:main", + message: "replace it", + runId: "run-local-second", + }); + + expect(firstAbortListener).toHaveBeenCalledTimes(1); + await vi.waitFor(() => { + expect(agentCommandFromIngressMock).toHaveBeenCalledTimes(2); + }); + }); + + it("does not inject local queue directives into an active run", async () => { + const { EmbeddedTuiBackend } = await import("./embedded-backend.js"); + const first = deferred<{ + payloads: Array<{ text: string }>; + meta: Record; + }>(); + const firstAbortListener = vi.fn(); + agentCommandFromIngressMock + .mockImplementationOnce((opts: { abortSignal?: AbortSignal }) => { + opts.abortSignal?.addEventListener("abort", firstAbortListener); + return first.promise; + }) + .mockResolvedValueOnce({ payloads: [{ text: "queue updated" }], meta: {} }); + loadSessionEntryMock.mockImplementation((sessionKey: string) => ({ + cfg: {}, + canonicalKey: sessionKey, + storePath: "/tmp/openclaw-sessions.json", + store: {}, + entry: { queueMode: "interrupt" }, + })); + resolveActiveEmbeddedRunSessionIdMock.mockReturnValue("active-session"); + + const backend = new EmbeddedTuiBackend(); + backend.start(); + await backend.sendChat({ + sessionKey: "agent:main:main", + message: "first", + runId: "run-local-first", + }); + await backend.sendChat({ + sessionKey: "agent:main:main", + message: "/queue followup", + runId: "run-local-queue", + }); + + expect(queueEmbeddedAgentMessageWithOutcomeAsyncMock).not.toHaveBeenCalled(); + expect(firstAbortListener).not.toHaveBeenCalled(); + expect(agentCommandFromIngressMock).toHaveBeenCalledTimes(2); + first.resolve({ payloads: [{ text: "first done" }], meta: {} }); + await flushMicrotasks(); + }); + it("does not queue stop commands behind active local runs", async () => { const { EmbeddedTuiBackend } = await import("./embedded-backend.js"); const first = deferred<{ diff --git a/src/tui/embedded-backend.ts b/src/tui/embedded-backend.ts index e64fe1eabe71..5f7ba1865436 100644 --- a/src/tui/embedded-backend.ts +++ b/src/tui/embedded-backend.ts @@ -10,6 +10,10 @@ import { } from "../agents/agent-scope.js"; import { ensureContextWindowCacheLoaded } from "../agents/context.js"; import { DEFAULT_PROVIDER } from "../agents/defaults.js"; +import { + queueEmbeddedAgentMessageWithOutcomeAsync, + resolveActiveEmbeddedRunSessionId, +} from "../agents/embedded-agent-runner/runs.js"; import { buildAllowedModelSet, buildConfiguredModelCatalog, @@ -17,7 +21,15 @@ import { } from "../agents/model-selection.js"; import { ensureRuntimePluginsLoaded } from "../agents/runtime-plugins.js"; import { readToolValidationErrorSummary } from "../agents/tool-error-summary.js"; +import { resolveTextCommand } from "../auto-reply/commands-registry.js"; import { parseGoalCommand } from "../auto-reply/reply/commands-goal.js"; +import { resolveQueueSettings } from "../auto-reply/reply/queue/settings.js"; +import { + DEFAULT_QUEUE_CAP, + DEFAULT_QUEUE_DEBOUNCE_MS, + DEFAULT_QUEUE_DROP, +} from "../auto-reply/reply/queue/state.js"; +import type { QueueSettings } from "../auto-reply/reply/queue/types.js"; import { createDefaultDeps } from "../cli/deps.js"; import { getRuntimeConfig } from "../config/config.js"; import { @@ -80,6 +92,12 @@ import { logInfo, logWarn } from "../logger.js"; import { normalizeAgentId } from "../routing/session-key.js"; import { defaultRuntime } from "../runtime.js"; import { INTERNAL_MESSAGE_CHANNEL } from "../utils/message-channel.js"; +import { + applyQueueDropPolicy, + buildCollectPrompt, + previewQueueSummaryPrompt, + waitForQueueDebounce, +} from "../utils/queue-helpers.js"; import { resolveLocalRunShutdownGraceMs } from "./local-run-shutdown.js"; import type { ChatSendOptions, @@ -107,15 +125,31 @@ type LocalRunState = { toolErrorSummary?: string; finalSent: boolean; registered: boolean; + pendingQueue?: { + mode: "followup" | "collect"; + messages: string[]; + debounceMs: number; + lastEnqueuedAt: number; + dropPolicy: NonNullable; + droppedCount: number; + summaryLines: string[]; + }; queuedRunReady: Promise; markQueuedRunReady: () => void; }; type QueuedSessionRun = { + runId: string; run: LocalRunState; promise: Promise; }; +type LocalPendingMessage = { + run: LocalRunState; + messageIndex: number; + message: string; +}; + const LIFECYCLE_ERROR_RETRY_GRACE_MS = 15_000; const silentRuntime = { @@ -187,6 +221,22 @@ function resolveBtwQuestion(message: string): string | undefined { return question ? question : undefined; } +function buildLocalQueuedPrompt(queue: NonNullable): string { + const summary = previewQueueSummaryPrompt({ + state: queue, + noun: "message", + }); + const prompt = + queue.mode === "collect" && queue.messages.length > 1 + ? buildCollectPrompt({ + title: "[Queued messages while agent was busy]", + items: queue.messages, + renderItem: (message, index) => `---\nQueued #${index + 1}\n${message}`, + }) + : (queue.messages[0] ?? ""); + return [summary, prompt].filter(Boolean).join("\n\n"); +} + function payloadText(parts: unknown): string { if (!Array.isArray(parts)) { return ""; @@ -399,6 +449,7 @@ export class EmbeddedTuiBackend implements TuiBackend { await this.ready; const runId = opts.runId ?? randomUUID(); const question = resolveBtwQuestion(opts.message); + const isQueueCommand = resolveTextCommand(opts.message)?.command.key === "queue"; const runScope = { sessionKey: opts.sessionKey, agentId: opts.agentId, @@ -406,11 +457,62 @@ export class EmbeddedTuiBackend implements TuiBackend { const abortableSessionRun = this.hasAbortableSessionRun(runScope); const stopCommand = abortableSessionRun && isChatStopCommandText(opts.message); const queuedAfter = - question || stopCommand ? undefined : this.findQueuedSessionRunPromise(runScope); + question || stopCommand || isQueueCommand + ? undefined + : this.findQueuedSessionRunPromise(runScope); if (stopCommand) { this.abortSessionRuns(runScope); return { runId }; } + let pendingQueue: LocalRunState["pendingQueue"]; + if (queuedAfter) { + const loadOptions = opts.agentId ? { agentId: opts.agentId } : undefined; + const { cfg, canonicalKey, entry } = loadSessionEntry(opts.sessionKey, loadOptions); + const queueSettings = resolveQueueSettings({ + cfg, + channel: INTERNAL_MESSAGE_CHANNEL, + sessionEntry: entry, + }); + if (queueSettings.mode === "steer") { + const activeSessionId = resolveActiveEmbeddedRunSessionId(canonicalKey); + if (activeSessionId) { + const outcome = await queueEmbeddedAgentMessageWithOutcomeAsync( + activeSessionId, + opts.message, + { + steeringMode: "all", + debounceMs: queueSettings.debounceMs ?? DEFAULT_QUEUE_DEBOUNCE_MS, + }, + ).catch(() => undefined); + if (outcome?.queued) { + return { runId: queuedAfter.runId }; + } + } + const queued = this.enqueuePendingLocalMessage({ + runScope, + message: opts.message, + settings: { ...queueSettings, mode: "followup" }, + fallbackRunId: queuedAfter.runId, + }); + if (queued.kind === "handled") { + return { runId: queued.runId }; + } + pendingQueue = queued.queue; + } else if (queueSettings.mode === "interrupt") { + this.abortSessionRuns(runScope); + } else { + const queued = this.enqueuePendingLocalMessage({ + runScope, + message: opts.message, + settings: queueSettings, + fallbackRunId: queuedAfter.runId, + }); + if (queued.kind === "handled") { + return { runId: queued.runId }; + } + pendingQueue = queued.queue; + } + } const controller = new AbortController(); const queuedRunReadiness = createQueuedRunReadiness(); this.runs.set(runId, { @@ -424,6 +526,7 @@ export class EmbeddedTuiBackend implements TuiBackend { lifecycleEnded: false, finalSent: false, registered: false, + ...(pendingQueue ? { pendingQueue } : {}), queuedRunReady: queuedRunReadiness.promise, markQueuedRunReady: queuedRunReadiness.markReady, }); @@ -444,6 +547,12 @@ export class EmbeddedTuiBackend implements TuiBackend { this.runPromises.delete(runId); }); + if (isQueueCommand) { + // Queue directives are control-plane mutations. Complete them before + // admitting another local prompt so later sends cannot overtake the new mode. + await runPromise; + } + return { runId }; } @@ -896,6 +1005,110 @@ export class EmbeddedTuiBackend implements TuiBackend { } } + private enqueuePendingLocalMessage(params: { + runScope: { sessionKey: string; agentId?: string }; + message: string; + settings: QueueSettings; + fallbackRunId: string; + }): + | { kind: "handled"; runId: string } + | { kind: "enqueue"; queue: NonNullable } { + const pendingMessages = this.listPendingLocalMessages(params.runScope); + const overflowQueue = { + items: [...pendingMessages], + cap: params.settings.cap ?? DEFAULT_QUEUE_CAP, + dropPolicy: params.settings.dropPolicy ?? DEFAULT_QUEUE_DROP, + droppedCount: 0, + summaryLines: [] as string[], + }; + const admitted = applyQueueDropPolicy({ + queue: overflowQueue, + summarize: (item) => item.message, + }); + if (!admitted) { + return { kind: "handled", runId: params.fallbackRunId }; + } + + const retained = new Set(overflowQueue.items); + const droppedByRun = new Map(); + for (const dropped of pendingMessages) { + if (retained.has(dropped)) { + continue; + } + const indices = droppedByRun.get(dropped.run) ?? []; + indices.push(dropped.messageIndex); + droppedByRun.set(dropped.run, indices); + } + const inheritedSummaryLines: string[] = []; + for (const [run, indices] of droppedByRun) { + for (const index of indices.toSorted((a, b) => b - a)) { + run.pendingQueue?.messages.splice(index, 1); + } + if (run.pendingQueue?.messages.length === 0) { + inheritedSummaryLines.push(...run.pendingQueue.summaryLines); + overflowQueue.droppedCount += run.pendingQueue.droppedCount; + run.controller.abort(); + } + } + overflowQueue.summaryLines.unshift(...inheritedSummaryLines); + if (overflowQueue.summaryLines.length > overflowQueue.cap) { + overflowQueue.summaryLines.splice(0, overflowQueue.summaryLines.length - overflowQueue.cap); + } + + const enqueuedAt = Date.now(); + for (const run of this.runs.values()) { + if (!this.isSameRunScope(run, params.runScope) || !run.pendingQueue) { + continue; + } + run.pendingQueue.lastEnqueuedAt = enqueuedAt; + run.pendingQueue.debounceMs = params.settings.debounceMs ?? DEFAULT_QUEUE_DEBOUNCE_MS; + } + + if (params.settings.mode === "collect") { + const target = [...this.runs.entries()].findLast( + ([, run]) => this.isSameRunScope(run, params.runScope) && run.pendingQueue, + ); + const targetQueue = target?.[1].pendingQueue; + if (target && targetQueue?.mode === "collect" && !target[1].controller.signal.aborted) { + const [targetRunId] = target; + targetQueue.messages.push(params.message); + targetQueue.dropPolicy = params.settings.dropPolicy ?? DEFAULT_QUEUE_DROP; + targetQueue.droppedCount += overflowQueue.droppedCount; + targetQueue.summaryLines.push(...overflowQueue.summaryLines); + return { kind: "handled", runId: targetRunId }; + } + } + + return { + kind: "enqueue", + queue: { + mode: params.settings.mode === "collect" ? "collect" : "followup", + messages: [params.message], + debounceMs: params.settings.debounceMs ?? DEFAULT_QUEUE_DEBOUNCE_MS, + lastEnqueuedAt: enqueuedAt, + dropPolicy: params.settings.dropPolicy ?? DEFAULT_QUEUE_DROP, + droppedCount: overflowQueue.droppedCount, + summaryLines: overflowQueue.summaryLines, + }, + }; + } + + private listPendingLocalMessages(params: { + sessionKey: string; + agentId?: string; + }): LocalPendingMessage[] { + const pending: LocalPendingMessage[] = []; + for (const run of this.runs.values()) { + if (!this.isSameRunScope(run, params) || !run.pendingQueue) { + continue; + } + run.pendingQueue.messages.forEach((message, messageIndex) => { + pending.push({ run, messageIndex, message }); + }); + } + return pending; + } + private findQueuedSessionRunPromise(params: { sessionKey: string; agentId?: string; @@ -905,7 +1118,7 @@ export class EmbeddedTuiBackend implements TuiBackend { if (this.isSameRunScope(run, params) && !run.isBtw) { const promise = this.runPromises.get(runId); if (promise) { - queuedAfter = { run, promise }; + queuedAfter = { runId, run, promise }; } } } @@ -1224,6 +1437,16 @@ export class EmbeddedTuiBackend implements TuiBackend { } } const activeRun = this.runs.get(params.runId); + let message = params.message; + if (activeRun?.pendingQueue) { + await waitForQueueDebounce(activeRun.pendingQueue, params.controller.signal); + if (params.controller.signal.aborted) { + this.emitChatAborted(params.runId, activeRun); + return; + } + message = buildLocalQueuedPrompt(activeRun.pendingQueue); + delete activeRun.pendingQueue; + } if (activeRun?.isBtw && activeRun.question) { const result = await this.runBtwTurn({ runId: params.runId, @@ -1260,7 +1483,7 @@ export class EmbeddedTuiBackend implements TuiBackend { // boundary (normalizeMessagesForLlmBoundary) from each message's own // timestamp, so the current turn and historical turns carry identical // bytes on the wire. See: https://github.com/openclaw/openclaw/issues/3658 - message: params.message, + message, sessionKey: canonicalKey, ...(params.agentId ? { agentId: params.agentId } : {}), ...(entry?.sessionId ? { sessionId: entry.sessionId } : {}), diff --git a/src/tui/tui-command-handlers.test.ts b/src/tui/tui-command-handlers.test.ts index a90fdde6b113..a97748aa24c4 100644 --- a/src/tui/tui-command-handlers.test.ts +++ b/src/tui/tui-command-handlers.test.ts @@ -1827,6 +1827,19 @@ describe("tui command handlers", () => { expectSendChatFields(sendChat, { message: "/queue:followup" }); }); + it("routes /queue directives through the local backend", async () => { + const { handleCommand, sendChat, addSystem } = createHarness({ + opts: { local: true }, + activeChatRunId: "run-active", + activityStatus: "streaming", + }); + + await handleCommand("/queue followup"); + + expectSendChatFields(sendChat, { message: "/queue followup" }); + expect(addSystem).not.toHaveBeenCalledWith("/queue is unavailable in local mode"); + }); + it("blocks /queue while optimistic user message is pending", async () => { const { handleCommand, sendChat, addSystem } = createHarness({ activeChatRunId: "run-active", diff --git a/src/tui/tui-command-handlers.ts b/src/tui/tui-command-handlers.ts index 53d517474cab..cd24f91dc0a8 100644 --- a/src/tui/tui-command-handlers.ts +++ b/src/tui/tui-command-handlers.ts @@ -490,6 +490,9 @@ export function createCommandHandlers(context: CommandHandlerContext) { chatLog.addSystem("Usage: /btw [side question]"); } break; + case "queue": + await sendMessage(raw); + break; case "openclaw": chatLog.addSystem( args ? `returning to OpenClaw with request: ${args}` : "returning to OpenClaw", diff --git a/src/tui/tui-pty-local.e2e.test.ts b/src/tui/tui-pty-local.e2e.test.ts index 3d1a29e74af4..2a0f57aa65b1 100644 --- a/src/tui/tui-pty-local.e2e.test.ts +++ b/src/tui/tui-pty-local.e2e.test.ts @@ -514,7 +514,11 @@ function buildLocalModeConfig(params: { async function startLocalModeTui( registerCleanup: CleanupRegistrar, - opts: { invalidEditLoop?: boolean } = {}, + opts: { + invalidEditLoop?: boolean; + holdFirstResponse?: boolean; + followupReplyText?: string; + } = {}, ) { const replyText = "LOCAL_PTY_RESPONSE"; const tempDir = await mkdtemp(path.join(tmpdir(), "openclaw-tui-pty-local-")); @@ -527,6 +531,8 @@ async function startLocalModeTui( const configPath = path.join(tempDir, "openclaw.json"); const mockModel = await startMockModelServer(replyText, { invalidEditLoop: opts.invalidEditLoop, + holdFirstResponse: opts.holdFirstResponse, + followupReplyText: opts.followupReplyText, }); const config = buildLocalModeConfig({ workspaceDir, @@ -881,6 +887,68 @@ describe("TUI PTY real backends", () => { LOCAL_TEST_TIMEOUT_MS, ); + it( + "steers an active real local session in the same turn", + async ({ onTestFinished }) => { + const fixture = await startLocalModeTui(onTestFinished, { + holdFirstResponse: true, + followupReplyText: "LOCAL_STEER_COMPLETE", + }); + try { + await fixture.run.waitForOutput("local ready", LOCAL_STARTUP_TIMEOUT_MS); + await fixture.run.write("slow local parent\r"); + await waitFor({ + timeoutMs: LOCAL_OUTPUT_TIMEOUT_MS, + read: () => (fixture.mockModel.requests().length === 1 ? true : null), + onTimeout: () => + new Error(`first prompt did not reach the model\n${fixture.run.output()}`), + }); + + const steerOffset = fixture.run.output().length; + await fixture.run.write("steer the active local turn\r"); + await waitForOutputAfter(fixture.run, "steer the active local turn", steerOffset); + await sleep(SUBMISSION_SETTLE_MS); + fixture.mockModel.releaseFirstResponse("gpt-5.5"); + + await waitFor({ + timeoutMs: LOCAL_OUTPUT_TIMEOUT_MS, + read: () => (fixture.mockModel.requests().length === 2 ? true : null), + onTimeout: () => + new Error( + `steered prompt did not reach the active local session\nrequests=${JSON.stringify( + fixture.mockModel.requests(), + null, + 2, + )}\n${fixture.run.output()}`, + ), + }); + expect(JSON.stringify(fixture.mockModel.requests()[1]?.body)).toContain( + "steer the active local turn", + ); + await fixture.run.waitForOutput("LOCAL_STEER_COMPLETE"); + if (process.env.OPENCLAW_BEHAVIOR_EVIDENCE === "1") { + console.info( + "[behavior-evidence] local-steer", + JSON.stringify({ + providerRequestCount: fixture.mockModel.requests().length, + secondRequestHasDynamicPrompt: ( + JSON.stringify(fixture.mockModel.requests()[1]?.body) ?? "" + ).includes("steer the active local turn"), + renderedCompletion: fixture.run.output().includes("LOCAL_STEER_COMPLETE"), + secondPromptEchoedBeforeRelease: true, + }), + ); + } + + await fixture.run.write("/exit\r", { delay: false }); + expect((await fixture.run.waitForExit()).exitCode).toBe(0); + } finally { + await fixture.cleanup(); + } + }, + LOCAL_TEST_TIMEOUT_MS, + ); + function registerValidationLoopTest(mode: "gateway" | "local") { it( `renders safe validation-loop abort diagnostics through the real ${mode} backend`,