openclaw · steipete · Jul 5, 2026 · Jul 5, 2026
diff --git a/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift b/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift
@@ -7778,21 +7778,25 @@ public struct ChatAbortParams: Codable, Sendable {
     public let sessionkey: String
     public let agentid: String?
     public let runid: String?
+    public let preservesideruns: Bool?
 
     public init(
         sessionkey: String,
         agentid: String? = nil,
-        runid: String?)
+        runid: String?,
+        preservesideruns: Bool? = nil)
     {
         self.sessionkey = sessionkey
         self.agentid = agentid
         self.runid = runid
+        self.preservesideruns = preservesideruns
     }
 
     private enum CodingKeys: String, CodingKey {
         case sessionkey = "sessionKey"
         case agentid = "agentId"
         case runid = "runId"
+        case preservesideruns = "preserveSideRuns"
     }
 }
 

diff --git a/docs/concepts/queue.md b/docs/concepts/queue.md
@@ -118,6 +118,28 @@ keys.
 - Applies to auto-reply agent runs across all inbound channels that use the gateway reply pipeline (WhatsApp web, Telegram, Slack, Discord, Signal, iMessage, webchat, etc.).
 - Default lane (`main`) is process-wide for inbound + main heartbeats; set `agents.defaults.maxConcurrent` to allow multiple sessions in parallel.
 - Additional lanes may exist (e.g. `cron`, `cron-nested`, `nested`, `subagent`) so background jobs can run in parallel without blocking inbound replies. Isolated cron agent turns hold a `cron` slot while their inner agent execution uses `cron-nested`; both use `cron.maxConcurrentRuns`. Shared non-cron `nested` flows keep their own lane behavior. These detached runs are tracked as [background tasks](/automation/tasks).
+
+## Queued-turn cancellation
+
+When Gateway admits a prompt into the followup/collect queue (for example a TUI
+or webchat `chat.send` while another turn is active), it keeps a **Gateway-owned
+cancel identity** for that client `runId` until the queued content runs or is
+dropped. The identity follows content folded into an overflow summary.
+
+- `chat.abort` with a specific `runId` cancels that turn while it is still queued,
+  if the requester is authorized (same ownership rules as active runs).
+- `chat.abort` for a session without `runId` cancels **authorized queued turns
+  first**, then aborts authorized active runs. That order prevents queue drain
+  from promoting work into a half-stopped session.
+- Clearing the entire session queue without per-requester checks is not the stop
+  path for multi-owner sessions.
+- Queued waits are not projected as active agent runs for `sessions.list` and do
+  not own active-run timeout semantics; only the active phase does.
+
+Clients (including the TUI) forward mid-run prompts and let Gateway apply the
+queue mode. Esc/`/stop` uses a session-scoped abort so lost local handles cannot
+leave a still-queued prompt running.
+
 - Per-session lanes guarantee that only one agent run touches a given session at a time.
 - No external dependencies or background worker threads; pure TypeScript + promises.
 

diff --git a/docs/docs_map.md b/docs/docs_map.md
@@ -2757,6 +2757,7 @@ Do not edit it by hand; run `pnpm docs:map:gen`.
   - H2: Precedence
   - H2: Per-session overrides
   - H2: Scope and guarantees
+  - H2: Queued-turn cancellation
   - H2: Troubleshooting
   - H2: Related
 

diff --git a/packages/gateway-protocol/src/index.test.ts b/packages/gateway-protocol/src/index.test.ts
@@ -118,6 +118,7 @@ describe("lazy protocol validators", () => {
         sessionKey: "global",
         agentId: "work",
         runId: "run-global-work",
+        preserveSideRuns: true,
       }),
     ).toBe(true);
     expect(

diff --git a/packages/gateway-protocol/src/schema/logs-chat.ts b/packages/gateway-protocol/src/schema/logs-chat.ts
@@ -107,6 +107,7 @@ export const ChatAbortParamsSchema = Type.Object(
     sessionKey: NonEmptyString,
     agentId: Type.Optional(NonEmptyString),
     runId: Type.Optional(NonEmptyString),
+    preserveSideRuns: Type.Optional(Type.Boolean()),
   },
   { additionalProperties: false },
 );

diff --git a/scripts/protocol-gen-swift.ts b/scripts/protocol-gen-swift.ts
@@ -64,7 +64,7 @@ const DEFAULTED_OPTIONAL_INIT_PARAM_ENTRIES: readonly [string, readonly string[]
   ["SessionsUsageParams", ["agentId", "agentScope"]],
   ["ChatHistoryParams", ["agentId", "offset"]],
   ["ChatSendParams", ["agentId"]],
-  ["ChatAbortParams", ["agentId"]],
+  ["ChatAbortParams", ["agentId", "preserveSideRuns"]],
   ["ChatInjectParams", ["agentId"]],
   ["ChatDeltaEvent", ["agentId"]],
   ["ChatFinalEvent", ["agentId"]],
@@ -73,7 +73,7 @@ const DEFAULTED_OPTIONAL_INIT_PARAM_ENTRIES: readonly [string, readonly string[]
   ["ArtifactsListParams", ["agentId"]],
   ["ArtifactsGetParams", ["agentId"]],
   ["ArtifactsDownloadParams", ["agentId"]],
-  ["ChatAbortParams", ["agentId"]],
+  ["ChatAbortParams", ["agentId", "preserveSideRuns"]],
   ["ChatAbortedEvent", ["agentId"]],
   ["ChatDeltaEvent", ["agentId"]],
   ["ChatErrorEvent", ["agentId"]],

diff --git a/src/auto-reply/get-reply-options.types.ts b/src/auto-reply/get-reply-options.types.ts
@@ -43,7 +43,14 @@ export type QueuedReplyDeliveryCorrelation = {
 
 /** Lifecycle hooks for queued follow-up replies. */
 export type QueuedReplyLifecycle = {
-  onEnqueued?: () => void;
+  /** Stable cancellation owner used to keep collect-mode batches authorization-safe. */
+  ownerKey?: string;
+  /** Return false when the external owner rejects this queue identity. */
+  onEnqueued?: () => boolean | void;
+  /** Retires this source's cancellation ownership while retaining its live identity. */
+  onCancellationRetired?: () => void;
+  /** Called after the queued turn owns the reply lane, before model/tool execution. */
+  onAdmitted?: () => void;
   onComplete?: () => void;
 };
 

diff --git a/src/auto-reply/reply/dispatch-from-config.test.ts b/src/auto-reply/reply/dispatch-from-config.test.ts
@@ -2198,6 +2198,48 @@ describe("dispatchReplyFromConfig", () => {
     }
   });
 
+  it("lets Gateway-owned turns reach queue resolution while a reply operation is active", async () => {
+    setNoAbort();
+    const sessionKey = "agent:main:main";
+    const activeOperation = createReplyOperation({
+      sessionKey,
+      sessionId: "active-session",
+      resetTriggered: false,
+    });
+    activeOperation.setPhase("running");
+    const dispatcher = createDispatcher();
+    const replyResolver = vi.fn(async () => undefined);
+
+    try {
+      const result = await dispatchReplyFromConfig({
+        ctx: buildTestCtx({
+          Provider: "webchat",
+          Surface: "webchat",
+          SessionKey: sessionKey,
+          BodyForAgent: "queue this turn",
+        }),
+        cfg: emptyConfig,
+        dispatcher,
+        replyOptions: {
+          queuedFollowupLifecycle: {
+            onEnqueued: vi.fn(),
+            onComplete: vi.fn(),
+          },
+        },
+        replyResolver,
+      });
+
+      expect(result).toMatchObject({
+        queuedFinal: false,
+        counts: { tool: 0, block: 0, final: 0 },
+      });
+      expect(replyResolver).toHaveBeenCalledTimes(1);
+      expect(replyRunRegistry.get(sessionKey)).toBe(activeOperation);
+    } finally {
+      activeOperation.complete();
+    }
+  });
+
   it("clears stale active reply operations for terminal sessions and retries admission", async () => {
     setNoAbort();
     const sessionKey = "agent:main:telegram:group:-1003774691294";

diff --git a/src/auto-reply/reply/dispatch-from-config.ts b/src/auto-reply/reply/dispatch-from-config.ts
@@ -62,11 +62,11 @@ import {
   toPluginMessageContext,
   toPluginMessageReceivedEvent,
 } from "../../hooks/message-hook-mappers.js";
+import { isAbortError } from "../../infra/abort-signal.js";
 import { isDiagnosticsEnabled } from "../../infra/diagnostic-events.js";
 import { measureDiagnosticsTimelineSpan } from "../../infra/diagnostics-timeline.js";
 import { formatErrorMessage } from "../../infra/errors.js";
 import { getSessionBindingService } from "../../infra/outbound/session-binding-service.js";
-import { isAbortError } from "../../infra/abort-signal.js";
 import type { StuckSessionRecoveryOutcome } from "../../logging/diagnostic-session-recovery.js";
 import {
   logMessageDispatchCompleted,
@@ -1452,6 +1452,16 @@ export async function dispatchReplyFromConfig(
       crypto.randomUUID();
     const replyTurnKind = resolveReplyTurnKind(params.replyOptions);
     const allowActivePreDispatch = phase === "pre_dispatch" && replyTurnKind === "visible";
+    const allowGatewayQueueResolution =
+      phase === "dispatch" &&
+      replyTurnKind === "visible" &&
+      params.replyOptions?.queuedFollowupLifecycle !== undefined &&
+      replyRunRegistry.get(dispatchOperationSessionKey) !== undefined;
+    if (allowGatewayQueueResolution) {
+      // Gateway turns need to reach getReplyFromConfig while the owner is active;
+      // that layer applies the session's steer/followup/collect/drop policy.
+      return { status: "ready" };
+    }
     const allowSlackRoutedThreadBypass =
       phase === "dispatch" &&
       shouldLetSlackRoutedThreadBypassBusyReplyOperation({

diff --git a/src/auto-reply/reply/followup-runner.test.ts b/src/auto-reply/reply/followup-runner.test.ts
@@ -618,6 +618,32 @@ function createQueuedRun(
 }
 
 describe("createFollowupRunner reply-lane admission", () => {
+  it("notifies queued owners after admission and before model execution", async () => {
+    const events: string[] = [];
+    runEmbeddedAgentMock.mockImplementationOnce(async () => {
+      events.push("run");
+      return { payloads: [], meta: {} };
+    });
+    const runner = createFollowupRunner({
+      typing: createMockTypingController(),
+      typingMode: "instant",
+      sessionKey: "main",
+      defaultModel: "anthropic/claude",
+    });
+
+    await runner(
+      createQueuedRun({
+        queuedLifecycle: {
+          onAdmitted: () => events.push("admitted"),
+          onComplete: () => events.push("complete"),
+        },
+        run: { provider: "anthropic", model: "claude" },
+      }),
+    );
+
+    expect(events).toEqual(["admitted", "run", "complete"]);
+  });
+
   it("passes prepared media user turns to embedded runtime dispatch", async () => {
     const preparedUserTurnMessage = {
       role: "user",
@@ -888,6 +914,7 @@ describe("createFollowupRunner reply-lane admission", () => {
 
   it("preserves non-compaction preflight failures for queued followup runs", async () => {
     runPreflightCompactionIfNeededMock.mockRejectedValueOnce(new Error("session load failed"));
+    const onComplete = vi.fn();
     const runner = createFollowupRunner({
       typing: createMockTypingController(),
       typingMode: "instant",
@@ -906,12 +933,14 @@ describe("createFollowupRunner reply-lane admission", () => {
             model: "claude",
             sessionKey: "main",
           },
+          queuedLifecycle: { onComplete },
         }),
       ),
     ).rejects.toThrow("session load failed");
 
     expect(runEmbeddedAgentMock).not.toHaveBeenCalled();
     expect(routeReplyMock).not.toHaveBeenCalled();
+    expect(onComplete).not.toHaveBeenCalled();
   });
 });
 

diff --git a/src/auto-reply/reply/followup-runner.ts b/src/auto-reply/reply/followup-runner.ts
@@ -521,6 +521,7 @@ export function createFollowupRunner(params: {
     const queuedImageOrder = queued.imageOrder ?? opts?.imageOrder;
     let replyOperation: ReplyOperation | undefined;
     let deferred = false;
+    let failed = false;
 
     try {
       queued.run.config = await resolveQueuedReplyExecutionConfig(queued.run.config, {
@@ -626,6 +627,9 @@ export function createFollowupRunner(params: {
         return;
       }
       replyOperation = admission.operation;
+      // Multi-source collected turns become atomic at reply-lane admission.
+      // Their queue owner uses this boundary to retire source cancellation ids.
+      effectiveQueued.queuedLifecycle?.onAdmitted?.();
       if (replyOperation.sessionId !== run.sessionId) {
         run = { ...run, sessionId: replyOperation.sessionId };
         effectiveQueued = { ...effectiveQueued, run };
@@ -1579,6 +1583,9 @@ export function createFollowupRunner(params: {
         },
         { runId },
       );
+    } catch (err) {
+      failed = true;
+      throw err;
     } finally {
       for (const end of endDeliveryCorrelations.toReversed()) {
         try {
@@ -1589,7 +1596,9 @@ export function createFollowupRunner(params: {
           );
         }
       }
-      if (!deferred) {
+      // A thrown attempt stays in the drain queue for retry. Its lifecycle
+      // identity remains live until the drain later consumes or drops it.
+      if (!deferred && !failed) {
         completeFollowupRunLifecycle(queued);
       }
       replyOperation?.complete();

diff --git a/src/auto-reply/reply/get-reply-run.ts b/src/auto-reply/reply/get-reply-run.ts
@@ -1267,8 +1267,15 @@ export async function runPreparedReply(
       extractedFileImages: opts?.extractedFileImages,
     }),
   );
+  // Abort-signal attachment for queued followups:
+  // - room_event: always inherit (source admission fence / ambient cancel).
+  // - Gateway-owned lifecycle (chat.send): always inherit so Esc can cancel a
+  //   turn after chat.send terminalizes while still queued.
+  // - plain user_request without lifecycle: deliberately detach from the
+  //   source/active-lane signal so a superseded parent abort does not cancel a
+  //   still-valid queued user turn.
   const queuedFollowupAbortSignal =
-    inboundEventKind === "room_event"
+    opts?.queuedFollowupLifecycle || inboundEventKind === "room_event"
       ? (opts?.queuedFollowupAbortSignal ?? opts?.abortSignal)
       : undefined;
   const userTurnMediaForPersistence = buildPersistedUserTurnMediaInputsFromFields(ctx);