fix: local chat honors queue steering (#109689)

* fix(tui): honor local queue steering

* fix(tui): narrow pending collect queue

* test(tui): validate collected call safely

* test(tui): emit local steering evidence

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"]));
   });
 });
 

src/tui/commands.ts

@@ -41,7 +41,7 @@ const COMMAND_ALIASES: Record<string, string> = {
 // 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[],

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<string, unknown>;
+    }>();
+    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<string, unknown>;
+    }>();
+    const second = deferred<{
+      payloads: Array<{ text: string }>;
+      meta: Record<string, unknown>;
+    }>();
+    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<string, unknown>;
+    }>();
+    const second = deferred<{
+      payloads: Array<{ text: string }>;
+      meta: Record<string, unknown>;
+    }>();
+    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<string, unknown>;
+    }>();
+    const collected = deferred<{
+      payloads: Array<{ text: string }>;
+      meta: Record<string, unknown>;
+    }>();
+    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<string, unknown>;
+    }>();
+    const second = deferred<{
+      payloads: Array<{ text: string }>;
+      meta: Record<string, unknown>;
+    }>();
+    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<string, unknown>;
+    }>();
+    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<string, unknown>;
+    }>();
+    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<{

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<QueueSettings["dropPolicy"]>;
+    droppedCount: number;
+    summaryLines: string[];
+  };
   queuedRunReady: Promise<void>;
   markQueuedRunReady: () => void;
 };
 
 type QueuedSessionRun = {
+  runId: string;
   run: LocalRunState;
   promise: Promise<void>;
 };
 
+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<LocalRunState["pendingQueue"]>): 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<LocalRunState["pendingQueue"]> } {
+    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<LocalRunState, number[]>();
+    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 } : {}),

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

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

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