diff --git a/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift b/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift
index 3ec67a88bf43..53624c0d2bfd 100644
--- a/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift
+++ b/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift
@@ -7374,21 +7374,25 @@ public struct SessionsAbortParams: Codable, Sendable {
public let key: String?
public let runid: String?
public let agentid: String?
+ public let clearqueued: Bool?
public init(
key: String? = nil,
runid: String? = nil,
- agentid: String? = nil)
+ agentid: String? = nil,
+ clearqueued: Bool? = nil)
{
self.key = key
self.runid = runid
self.agentid = agentid
+ self.clearqueued = clearqueued
}
private enum CodingKeys: String, CodingKey {
case key
case runid = "runId"
case agentid = "agentId"
+ case clearqueued = "clearQueued"
}
}
diff --git a/docs/gateway/protocol.md b/docs/gateway/protocol.md
index a9ebefa2b2f4..fc5c9f746129 100644
--- a/docs/gateway/protocol.md
+++ b/docs/gateway/protocol.md
@@ -577,7 +577,7 @@ methods. Treat this as feature discovery, not a full enumeration of
- `sessions.groups.list`, `sessions.groups.put`, `sessions.groups.rename`, and `sessions.groups.delete` manage the gateway-owned custom session group catalog (names + display order). Membership stays on each session's `category` field; rename and delete update member sessions server-side.
- `sessions.send` sends a message into an existing session.
- `sessions.steer` is the interrupt-and-steer variant for an active session.
- - `sessions.abort` aborts active work for a session. Pass `key` plus optional `runId`, or `runId` alone for active runs the gateway can resolve to a session.
+ - `sessions.abort` aborts active work for a session. Pass `key` plus optional `runId`, or `runId` alone for active runs the gateway can resolve to a session. Supplying `runId` keeps cancellation scoped to that run. Set `clearQueued: true` on a key-only non-global request to also discard followup and lane queues owned by that session. Existing callers that omit `clearQueued` preserve those queues. The literal `global` key keeps the existing agent-qualified `chat.abort` ownership rules and does not perform non-global followup or lane cleanup.
- `sessions.patch` updates session metadata/overrides and reports the resolved canonical model plus effective `agentRuntime`. Spawn lineage (`spawnedBy`, `spawnedWorkspaceDir`, `spawnedCwd`, `spawnDepth`, `subagentRole`, `subagentControlScope`) is no longer publicly patchable; those facts are written once by trusted creation paths, and requests that still send them are rejected.
- `sessions.reset`, `sessions.delete`, and `sessions.compact` perform session maintenance.
- `sessions.get` returns the full stored session row.
diff --git a/docs/web/control-ui.md b/docs/web/control-ui.md
index 43b9d2214530..c57f3cc59773 100644
--- a/docs/web/control-ui.md
+++ b/docs/web/control-ui.md
@@ -460,11 +460,11 @@ The macOS app keeps its native link-browser sidebar for links clicked in the das
- - Click **Stop** (calls `chat.abort`).
+ - Click **Stop**. Runs with an exact local run ID call `chat.abort`; when selected-session state reports active work but the Control UI has no local run ID, it calls `sessions.abort` instead. For non-global sessions, that selected-session path also discards queued follow-ups so they cannot restart work after the stop.
- While a run is active, normal follow-ups use the Gateway's effective `messages.queue` mode. `steer` injects into the running turn; other modes keep the browser's durable queued delivery. Steering rejection also falls back to that queue. Click **Steer** on a queued message to inject it manually.
- **Settings → Appearance → Chat → Follow-ups while the agent is working** can override that server default for the current browser. The page marks an override explicitly and offers **Reset to server default**. `Steer into the active run` sends follow-ups immediately, while `Queue until the run ends` holds them until the run finishes.
- Type `/stop` (or standalone abort phrases like `stop`, `stop action`, `stop run`, `stop openclaw`, `please stop`) to abort out-of-band.
- - `chat.abort` supports `{ sessionKey }` (no `runId`) to abort all active runs for that session.
+ - `chat.abort` supports `{ sessionKey }` (no `runId`) to abort all active runs for that session. The Control UI uses `sessions.abort` when it has no local run ID.
@@ -484,7 +484,8 @@ realtime/session actions pause until the connection returns; **Retry now** in th
immediate attempt. Chat remains editable: ordinary text and attachment sends are kept in the
current tab's gateway/session-scoped browser storage, shown as waiting for reconnect, and sent
automatically when the Gateway returns. Live controls and slash commands remain unavailable while
-offline.
+offline, except that **Stop** can queue an exact local run ID for replay. A session-only stop
+is not replayed because newer work may start in that session before the connection returns.
When this browser already holds credentials (a configured token/password or an approved device
token), first opens and reloads show a small animated OpenClaw mark while the connection is
diff --git a/extensions/codex/src/app-server/run-attempt-active-turn.ts b/extensions/codex/src/app-server/run-attempt-active-turn.ts
index 1742d9b73c55..5bc3a0d8f7a7 100644
--- a/extensions/codex/src/app-server/run-attempt-active-turn.ts
+++ b/extensions/codex/src/app-server/run-attempt-active-turn.ts
@@ -184,6 +184,7 @@ export async function activateCodexAttemptTurn(
await activeSteeringQueue.queue(text, optionsLocal);
},
isStreaming: () => !state.completed && !runAbortController.signal.aborted,
+ isAborted: () => runAbortController.signal.aborted,
isStopped: () => state.completed || state.timedOut || runAbortController.signal.aborted,
isAbortable: () =>
!terminalState.terminalOutcomeFrozen || terminalState.sharedAbortAllowedAfterTerminalOutcome,
diff --git a/extensions/codex/src/app-server/run-attempt.steering.test.ts b/extensions/codex/src/app-server/run-attempt.steering.test.ts
index 974fa6f3761f..a6dc1d909eb0 100644
--- a/extensions/codex/src/app-server/run-attempt.steering.test.ts
+++ b/extensions/codex/src/app-server/run-attempt.steering.test.ts
@@ -107,6 +107,40 @@ async function waitAndQueueActiveRunMessage(
}
describe("runCodexAppServerAttempt steering", () => {
+ it("marks the active run aborted before asynchronous cleanup releases its handle", async () => {
+ const { requests, waitForMethod } = createStartedThreadHarness();
+ const params = createSteeringParams();
+ activeRunRegistrationMocks.setActiveEmbeddedRun.mockClear();
+ activeRunRegistrationMocks.clearActiveEmbeddedRun.mockClear();
+
+ const run = runCodexAppServerAttempt(params);
+ await waitForMethod("turn/start");
+
+ let handle: { abort: () => void; isAborted?: () => boolean } | undefined;
+ await vi.waitFor(() => {
+ handle = activeRunRegistrationMocks.setActiveEmbeddedRun.mock.calls.findLast(
+ (call) => call[0] === params.sessionId,
+ )?.[1] as typeof handle;
+ expect(handle).toBeDefined();
+ }, fastWait);
+ expect(handle?.isAborted?.()).toBe(false);
+
+ handle?.abort();
+ expect(handle?.isAborted?.()).toBe(true);
+ expect(activeRunRegistrationMocks.clearActiveEmbeddedRun).not.toHaveBeenCalled();
+ await expect(run).resolves.toMatchObject({ aborted: true });
+ expect(activeRunRegistrationMocks.clearActiveEmbeddedRun).toHaveBeenCalledWith(
+ params.sessionId,
+ handle,
+ params.sessionKey,
+ params.sessionFile,
+ );
+ expect(requests).toContainEqual({
+ method: "turn/interrupt",
+ params: { threadId: "thread-1", turnId: "turn-1" },
+ });
+ });
+
it("accepts Gateway transcript-backed steering for the active Codex turn", async () => {
const { requests, waitForMethod, completeTurn, notify } = createStartedThreadHarness();
const params = createSteeringParams();
diff --git a/extensions/copilot/src/attempt.test.ts b/extensions/copilot/src/attempt.test.ts
index 945c7c4afd8e..b32c2b07f624 100644
--- a/extensions/copilot/src/attempt.test.ts
+++ b/extensions/copilot/src/attempt.test.ts
@@ -43,6 +43,7 @@ const gatewayQuestionMock = vi.hoisted(() => ({
waiters: new Map void>(),
cancelError: undefined as Error | undefined,
warn: vi.fn(),
+ setActiveEmbeddedRun: vi.fn(),
}));
vi.mock("openclaw/plugin-sdk/agent-harness-runtime", async (importOriginal) => {
@@ -81,6 +82,12 @@ vi.mock("openclaw/plugin-sdk/agent-harness-runtime", async (importOriginal) => {
}
return await actual.callGatewayTool(...args);
},
+ setActiveEmbeddedRun: (
+ ...args: Parameters
+ ): ReturnType => {
+ gatewayQuestionMock.setActiveEmbeddedRun(...args);
+ return actual.setActiveEmbeddedRun(...args);
+ },
};
});
@@ -1361,14 +1368,12 @@ describe("runCopilotAttempt", () => {
});
it("active-run abort path marks the attempt as externally aborted", async () => {
+ gatewayQuestionMock.setActiveEmbeddedRun.mockClear();
const sendDeferred = createDeferred();
const sessionCreated = createDeferred();
const sdk = makeFakeSdk({
onCreateSession: (session) => {
session.sendAndWait.mockReturnValue(sendDeferred.promise);
- session.abort.mockImplementationOnce(async () => {
- sendDeferred.resolve(undefined);
- });
sessionCreated.resolve(session);
},
});
@@ -1381,12 +1386,21 @@ describe("runCopilotAttempt", () => {
});
const session = await sessionCreated.promise;
await vi.waitFor(() => expect(session.sendAndWait).toHaveBeenCalledTimes(1));
+ const activeRunHandle = expectDefined(
+ gatewayQuestionMock.setActiveEmbeddedRun.mock.calls.findLast(
+ ([sessionId]) => sessionId === "session-1",
+ )?.[1] as { isAborted?: () => boolean } | undefined,
+ "active Copilot run handle",
+ );
+ expect(activeRunHandle.isAborted?.()).toBe(false);
gatewayQuestionMock.cancelError = new Error("gateway unavailable");
expect(abortAgentHarnessRun("session-1")).toBe(true);
+ expect(activeRunHandle.isAborted?.()).toBe(true);
+ expect(session.abort).toHaveBeenCalledTimes(1);
+ sendDeferred.resolve(undefined);
const result = await runPromise;
- expect(session.abort).toHaveBeenCalledTimes(1);
expect(result.terminal).toMatchObject({ kind: "aborted", source: "external" });
await vi.waitFor(() =>
expect(gatewayQuestionMock.warn).toHaveBeenCalledWith(
diff --git a/extensions/copilot/src/attempt.ts b/extensions/copilot/src/attempt.ts
index 6ecf7752491a..40297b534ed8 100644
--- a/extensions/copilot/src/attempt.ts
+++ b/extensions/copilot/src/attempt.ts
@@ -1070,6 +1070,7 @@ export async function runCopilotAttempt(
throw new Error("Copilot runtime is not waiting for user input.");
},
isStreaming: () => !settled && !aborted,
+ isAborted: () => aborted,
isCompacting: () => bridge?.isCompacting() ?? false,
sourceReplyDeliveryMode: input.sourceReplyDeliveryMode,
cancel: () => {
diff --git a/packages/gateway-protocol/src/schema/sessions.ts b/packages/gateway-protocol/src/schema/sessions.ts
index d98093311eae..019d2743785b 100644
--- a/packages/gateway-protocol/src/schema/sessions.ts
+++ b/packages/gateway-protocol/src/schema/sessions.ts
@@ -432,6 +432,8 @@ export const SessionsAbortParamsSchema = closedObject({
key: Type.Optional(NonEmptyString),
runId: Type.Optional(NonEmptyString),
agentId: Type.Optional(NonEmptyString),
+ /** Also discard followup and lane queues for a key-only non-global session abort. */
+ clearQueued: Type.Optional(Type.Boolean()),
});
/** Mutable per-session preferences and routing metadata. */
diff --git a/packages/sdk/src/index.test.ts b/packages/sdk/src/index.test.ts
index 58b1b512f116..18202e2c3990 100644
--- a/packages/sdk/src/index.test.ts
+++ b/packages/sdk/src/index.test.ts
@@ -1414,6 +1414,23 @@ describe("OpenClaw SDK", () => {
]);
});
+ it("keeps key-only Session.abort compatible by omitting clearQueued", async () => {
+ const transport = new FakeTransport({
+ "sessions.create": { key: "session-main", label: "Main" },
+ "sessions.abort": { ok: true, abortedRunId: null, status: "no-active-run" },
+ });
+ const oc = new OpenClaw({ transport });
+
+ const session = await oc.sessions.create({ key: "session-main" });
+ await session.abort();
+
+ expect(transport.calls.at(-1)).toEqual({
+ method: "sessions.abort",
+ options: undefined,
+ params: { key: "session-main" },
+ });
+ });
+
it("normalizes Gateway agent stream events into SDK events", () => {
const ts = 1_777_000_000_000;
diff --git a/src/agents/embedded-agent-runner/compact.queued.ts b/src/agents/embedded-agent-runner/compact.queued.ts
index 2b99aa35184c..ecb73804b251 100644
--- a/src/agents/embedded-agent-runner/compact.queued.ts
+++ b/src/agents/embedded-agent-runner/compact.queued.ts
@@ -262,6 +262,7 @@ export async function compactEmbeddedAgentSession(
kind: "embedded",
queueMessage: async () => {},
isStreaming: () => true,
+ isAborted: () => abortSignal.aborted,
isCompacting: () => true,
abort: (reason) => controller.abort(reason ?? "user_abort"),
cancel: (reason) => controller.abort(reason ?? "user_abort"),
diff --git a/src/agents/embedded-agent-runner/run-state.ts b/src/agents/embedded-agent-runner/run-state.ts
index 0c401c7d8f59..236cb5b79067 100644
--- a/src/agents/embedded-agent-runner/run-state.ts
+++ b/src/agents/embedded-agent-runner/run-state.ts
@@ -30,6 +30,8 @@ export type EmbeddedAgentQueueHandle = {
queueMessage: (text: string, options?: EmbeddedAgentQueueMessageOptions) => Promise;
isStreaming: () => boolean;
isStopped?: () => boolean;
+ /** True after this handle has accepted an abort, even while cleanup retains it. */
+ isAborted?: () => boolean;
isAbortable?: () => boolean;
isCompacting: () => boolean;
supportsTranscriptCommitWait?: boolean;
diff --git a/src/agents/embedded-agent-runner/run/attempt-stream-prepare.test.ts b/src/agents/embedded-agent-runner/run/attempt-stream-prepare.test.ts
index be60a8d0d93a..50b02751962f 100644
--- a/src/agents/embedded-agent-runner/run/attempt-stream-prepare.test.ts
+++ b/src/agents/embedded-agent-runner/run/attempt-stream-prepare.test.ts
@@ -24,9 +24,21 @@ vi.mock("./tool-activity-heartbeat.js", () => ({
}));
import { prepareEmbeddedAttemptStream } from "./attempt-stream-prepare.js";
+import { SESSIONS_YIELD_ABORT_REASON } from "./attempt.sessions-yield.js";
-function prepareCatalogExecutor(projections: ToolSearchTargetTranscriptProjection[]) {
- const runAbortController = new AbortController();
+function prepareCatalogExecutor(
+ projections: ToolSearchTargetTranscriptProjection[],
+ options?: {
+ getRunState?: () => {
+ aborted: boolean;
+ promptError: unknown;
+ timedOut: boolean;
+ yieldDetected: boolean;
+ };
+ runAbortController?: AbortController;
+ },
+) {
+ const runAbortController = options?.runAbortController ?? new AbortController();
return prepareEmbeddedAttemptStream({
attempt: {
runId: "run-output-schema",
@@ -43,12 +55,14 @@ function prepareCatalogExecutor(projections: ToolSearchTargetTranscriptProjectio
runAbortController,
abortRun: vi.fn(),
markExternalAbort: vi.fn(),
- getRunState: () => ({
- aborted: false,
- promptError: undefined,
- timedOut: false,
- yieldDetected: false,
- }),
+ getRunState:
+ options?.getRunState ??
+ (() => ({
+ aborted: false,
+ promptError: undefined,
+ timedOut: false,
+ yieldDetected: false,
+ })),
hasDeliveredSourceReply: () => false,
markSourceReplyDelivered: vi.fn(),
onBlockReply: vi.fn(),
@@ -150,4 +164,29 @@ describe("prepareEmbeddedAttemptStream", () => {
expect(Object.isFrozen(returned)).toBe(true);
expect(Object.isFrozen(returned.details)).toBe(true);
});
+
+ it("distinguishes an accepted abort from normal steering closure and sessions_yield", () => {
+ const runAbortController = new AbortController();
+ let aborted = false;
+ const prepared = prepareCatalogExecutor([], {
+ runAbortController,
+ getRunState: () => ({
+ aborted,
+ promptError: undefined,
+ timedOut: false,
+ yieldDetected: false,
+ }),
+ });
+
+ expect(prepared.queueHandle.isAborted?.()).toBe(false);
+ prepared.stopAcceptingSteerMessages();
+ expect(prepared.queueHandle.isStopped?.()).toBe(true);
+ expect(prepared.queueHandle.isAborted?.()).toBe(false);
+
+ runAbortController.abort(SESSIONS_YIELD_ABORT_REASON);
+ expect(prepared.queueHandle.isAborted?.()).toBe(false);
+
+ aborted = true;
+ expect(prepared.queueHandle.isAborted?.()).toBe(true);
+ });
});
diff --git a/src/agents/embedded-agent-runner/run/attempt-stream-prepare.ts b/src/agents/embedded-agent-runner/run/attempt-stream-prepare.ts
index 6c403ae4794d..2fde18f2b7cf 100644
--- a/src/agents/embedded-agent-runner/run/attempt-stream-prepare.ts
+++ b/src/agents/embedded-agent-runner/run/attempt-stream-prepare.ts
@@ -366,6 +366,7 @@ export function prepareEmbeddedAttemptStream(input: {
);
},
isStreaming: () => input.activeSession.isStreaming,
+ isAborted: () => input.getRunState().aborted,
isStopped: () =>
!acceptingSteerMessages ||
input.getRunState().aborted ||
diff --git a/src/agents/embedded-agent-runner/runs.ts b/src/agents/embedded-agent-runner/runs.ts
index 989957e6b73e..b205f5526f05 100644
--- a/src/agents/embedded-agent-runner/runs.ts
+++ b/src/agents/embedded-agent-runner/runs.ts
@@ -619,6 +619,35 @@ export function isEmbeddedAgentRunActive(sessionId: string): boolean {
return active;
}
+/**
+ * Returns whether a registry-owned run is still doing user-visible work.
+ * Terminal reply operations and aborted handles retain their lane for cleanup,
+ * but must not keep session activity projections in the running state.
+ */
+export function isEmbeddedAgentRunInProgress(sessionId: string): boolean {
+ const replyPhase = resolveReplyRunPhaseForSessionId(sessionId);
+ const replyInProgress =
+ replyPhase !== undefined &&
+ replyPhase !== "completed" &&
+ replyPhase !== "failed" &&
+ replyPhase !== "aborted";
+ const handle = ACTIVE_EMBEDDED_RUNS.get(sessionId);
+ let handleInProgress = handle !== undefined;
+ if (handle?.isAborted) {
+ try {
+ if (handle.isAborted()) {
+ handleInProgress = false;
+ }
+ } catch {
+ // A failed optional status probe cannot prove that live work has ended.
+ handleInProgress = true;
+ }
+ }
+ // Reply operations and embedded handles are independent lifecycle owners.
+ // A retained terminal owner must not hide a newer live owner for the session.
+ return replyInProgress || handleInProgress;
+}
+
export function resolveEmbeddedAgentReplyRunPhase(
sessionId: string,
): ReplyOperationPhase | undefined {
diff --git a/src/gateway/server-methods/chat-abort-authorization.ts b/src/gateway/server-methods/chat-abort-authorization.ts
index 98f462ed5015..24e1913e3e07 100644
--- a/src/gateway/server-methods/chat-abort-authorization.ts
+++ b/src/gateway/server-methods/chat-abort-authorization.ts
@@ -160,6 +160,7 @@ export function readPreRegisteredRun(params: {
key: string;
entry: GatewayRequestContext["dedupe"] extends Map ? T | undefined : never;
keyPrefix: string;
+ includeHidden?: boolean;
}): PreRegisteredAgentRun | undefined {
if (!params.key.startsWith(params.keyPrefix) || !params.entry?.ok) {
return undefined;
@@ -168,7 +169,7 @@ export function readPreRegisteredRun(params: {
if (payload?.status !== "accepted") {
return undefined;
}
- if (payload.controlUiVisible === false) {
+ if (!params.includeHidden && payload.controlUiVisible === false) {
return undefined;
}
const runId =
@@ -300,15 +301,19 @@ export function resolveAuthorizedPreRegisteredRunsForSessionKeys(params: {
),
);
const authorizedByRunId = new Map();
- let matchedSessionRuns = 0;
+ let hasUnauthorizedRuns = false;
+ let hasUnauthorizedProtectedRuns = false;
+ let hasProtectedRuns = false;
for (const [key, entry] of params.context.dedupe) {
- const run = readPreRegisteredRun({ key, entry, keyPrefix: params.keyPrefix });
+ const run = readPreRegisteredRun({
+ key,
+ entry,
+ keyPrefix: params.keyPrefix,
+ includeHidden: true,
+ });
if (!run) {
continue;
}
- if (params.preserveSideRuns && normalizeUnknownText(run.payload.turnKind) === "btw") {
- continue;
- }
const runSessionKeys = [
run.sessionKey,
...(Array.isArray(run.payload.sessionKeyAliases)
@@ -330,16 +335,34 @@ export function resolveAuthorizedPreRegisteredRunsForSessionKeys(params: {
params.defaultAgentId,
) !== agentId
) {
+ // Global keys are shared across agent stores; another agent's run is
+ // outside the selected global-agent scope.
continue;
}
- matchedSessionRuns += 1;
- if (canRequesterAbortPreRegisteredRun(run.payload, params.requester)) {
+ const requesterCanAbort = canRequesterAbortPreRegisteredRun(run.payload, params.requester);
+ const isProtected =
+ run.payload.controlUiVisible === false ||
+ (params.preserveSideRuns && normalizeUnknownText(run.payload.turnKind) === "btw");
+ if (isProtected) {
+ // Broad lifecycle cleanup still needs ownership, while ordinary chat.abort
+ // must keep treating hidden or preserved work as a non-match.
+ hasProtectedRuns = true;
+ if (!requesterCanAbort) {
+ hasUnauthorizedProtectedRuns = true;
+ }
+ continue;
+ }
+ if (requesterCanAbort) {
authorizedByRunId.set(run.runId, run);
+ } else {
+ hasUnauthorizedRuns = true;
}
}
return {
- matchedSessionRuns,
authorizedRuns: [...authorizedByRunId.values()],
+ hasUnauthorizedRuns,
+ hasUnauthorizedProtectedRuns,
+ hasProtectedRuns,
};
}
@@ -364,14 +387,11 @@ export function resolveAuthorizedRunsForSessionKeys(params: {
);
const agentId = normalizeOptionalText(params.agentId)?.toLowerCase();
const authorizedRuns: Array<{ runId: string; sessionKey: string }> = [];
- let matchedSessionRuns = 0;
+ const matchedRunIds: string[] = [];
+ let hasUnauthorizedRuns = false;
+ let hasUnauthorizedProtectedRuns = false;
+ let hasProtectedRuns = false;
for (const [runId, active] of params.chatAbortControllers) {
- if (active.controlUiVisible === false) {
- continue;
- }
- if (params.preserveSideRuns && active.turnKind === "btw") {
- continue;
- }
if (!sessionKeys.has(active.sessionKey) && !sessionIds.has(active.sessionId)) {
continue;
}
@@ -380,16 +400,35 @@ export function resolveAuthorizedRunsForSessionKeys(params: {
active.sessionKey === "global" &&
resolveStoredGlobalRunAgentId(active.agentId, params.defaultAgentId) !== agentId
) {
+ // Global keys are shared across agent stores; another agent's run is
+ // outside the selected global-agent scope.
continue;
}
- matchedSessionRuns += 1;
- if (canRequesterAbortChatRun(active, params.requester)) {
+ matchedRunIds.push(runId);
+ const requesterCanAbort = canRequesterAbortChatRun(active, params.requester);
+ const isProtected =
+ active.controlUiVisible === false || (params.preserveSideRuns && active.turnKind === "btw");
+ if (isProtected) {
+ // Broad lifecycle cleanup still needs ownership, while ordinary chat.abort
+ // must keep treating hidden or preserved work as a non-match.
+ hasProtectedRuns = true;
+ if (!requesterCanAbort) {
+ hasUnauthorizedProtectedRuns = true;
+ }
+ continue;
+ }
+ if (requesterCanAbort) {
authorizedRuns.push({ runId, sessionKey: active.sessionKey });
+ } else {
+ hasUnauthorizedRuns = true;
}
}
return {
- matchedSessionRuns,
authorizedRuns,
+ matchedRunIds,
+ hasUnauthorizedRuns,
+ hasUnauthorizedProtectedRuns,
+ hasProtectedRuns,
};
}
diff --git a/src/gateway/server-methods/chat-abort-handler.ts b/src/gateway/server-methods/chat-abort-handler.ts
index c3f08f15c175..68f59510d6be 100644
--- a/src/gateway/server-methods/chat-abort-handler.ts
+++ b/src/gateway/server-methods/chat-abort-handler.ts
@@ -38,12 +38,14 @@ import {
} from "./chat-text-normalization.js";
import type { GatewayRequestContext, GatewayRequestHandlerOptions } from "./types.js";
-export async function handleChatAbortRequest({
- params,
- respond,
- context,
- client,
-}: GatewayRequestHandlerOptions): Promise {
+type ChatAbortLifecycle = {
+ onAuthorizedAfterQueuedAbort?: () => boolean;
+};
+
+export async function handleChatAbortRequestWithLifecycle(
+ { params, respond, context, client }: GatewayRequestHandlerOptions,
+ lifecycle: ChatAbortLifecycle = {},
+): Promise {
if (!validateChatAbortParams(params)) {
respond(
false,
@@ -124,6 +126,7 @@ export async function handleChatAbortRequest({
stopReason: "rpc",
requester,
preserveSideRuns,
+ onAuthorizedAfterQueuedAbort: lifecycle.onAuthorizedAfterQueuedAbort,
});
if (res.unauthorized) {
respond(false, undefined, errorShape(ErrorCodes.INVALID_REQUEST, "unauthorized"));
@@ -320,3 +323,7 @@ export async function handleChatAbortRequest({
}
respondWithWorkerRuns(res.aborted ? [runId] : [], active.sessionId);
}
+
+export async function handleChatAbortRequest(options: GatewayRequestHandlerOptions): Promise {
+ await handleChatAbortRequestWithLifecycle(options);
+}
diff --git a/src/gateway/server-methods/chat-abort-runtime.ts b/src/gateway/server-methods/chat-abort-runtime.ts
index b62b5bc269dd..78a7bcf0cd7c 100644
--- a/src/gateway/server-methods/chat-abort-runtime.ts
+++ b/src/gateway/server-methods/chat-abort-runtime.ts
@@ -119,19 +119,14 @@ export function ensureChatQueuedTurns(context: GatewayRequestContext): QueuedCha
return context.chatQueuedTurns;
}
-/**
- * Cancel authorized queued turns for a session BEFORE active-run abort so
- * drain cannot promote work into a half-aborted session.
- */
-function abortAuthorizedQueuedTurnsForSession(params: {
+function resolveAuthorizedQueuedTurnsForSession(params: {
context: GatewayRequestContext;
sessionKeys: string[];
sessionId?: string;
agentId?: string;
defaultAgentId: string;
requester: ChatAbortRequester;
- stopReason?: string;
-}): { runIds: string[]; matched: number; unauthorizedOnly: boolean } {
+}) {
const chatQueuedTurns = ensureChatQueuedTurns(params.context);
const matches = listQueuedChatTurnsForSession({
chatQueuedTurns,
@@ -141,16 +136,15 @@ function abortAuthorizedQueuedTurnsForSession(params: {
defaultAgentId: params.defaultAgentId,
});
if (matches.length === 0) {
- return { runIds: [], matched: 0, unauthorizedOnly: false };
+ return { authorized: [], hasUnauthorizedRuns: false };
}
const authorized = matches.filter((m) =>
canRequesterAbortQueuedChatTurn(m.entry, params.requester),
);
- if (authorized.length === 0) {
- return { runIds: [], matched: matches.length, unauthorizedOnly: true };
- }
- const runIds = abortQueuedChatTurns(chatQueuedTurns, authorized, params.stopReason);
- return { runIds, matched: matches.length, unauthorizedOnly: false };
+ return {
+ authorized,
+ hasUnauthorizedRuns: authorized.length < matches.length,
+ };
}
export function cancelWorkerInferenceForSession(params: {
@@ -183,20 +177,25 @@ export async function abortChatRunsForSessionKeyWithPartials(params: {
stopReason?: string;
requester: ChatAbortRequester;
preserveSideRuns?: boolean;
+ /** Internal session-wide cleanup after exact resolution and all matching owner checks. */
+ onAuthorizedAfterQueuedAbort?: () => boolean;
}): Promise<{ aborted: boolean; runIds: string[]; unauthorized: boolean }> {
const sessionKeys = [params.sessionKey, ...(params.sessionKeyAliases ?? [])];
- // Queued-turn cancel MUST run before active abort so followup drain cannot
- // promote cancelled work into the gap between active stop and queue clear.
- const queuedAbort = abortAuthorizedQueuedTurnsForSession({
+ const queuedPlan = resolveAuthorizedQueuedTurnsForSession({
context: params.context,
sessionKeys,
sessionId: params.sessionId,
agentId: params.agentId,
defaultAgentId: params.defaultAgentId,
requester: params.requester,
- stopReason: params.stopReason,
});
- const { matchedSessionRuns, authorizedRuns } = resolveAuthorizedRunsForSessionKeys({
+ const {
+ authorizedRuns,
+ matchedRunIds: matchedActiveRunIds,
+ hasUnauthorizedRuns: hasUnauthorizedActiveRuns,
+ hasUnauthorizedProtectedRuns: hasUnauthorizedProtectedActiveRuns,
+ hasProtectedRuns: hasProtectedActiveRuns,
+ } = resolveAuthorizedRunsForSessionKeys({
chatAbortControllers: params.context.chatAbortControllers,
sessionKeys,
sessionIds: [params.sessionId],
@@ -206,8 +205,10 @@ export async function abortChatRunsForSessionKeyWithPartials(params: {
preserveSideRuns: params.preserveSideRuns,
});
const {
- matchedSessionRuns: matchedPendingAgentRuns,
authorizedRuns: authorizedPendingAgentRuns,
+ hasUnauthorizedRuns: hasUnauthorizedPendingAgentRuns,
+ hasUnauthorizedProtectedRuns: hasUnauthorizedProtectedPendingAgentRuns,
+ hasProtectedRuns: hasProtectedPendingAgentRuns,
} = resolveAuthorizedPreRegisteredRunsForSessionKeys({
context: params.context,
sessionKeys,
@@ -217,43 +218,81 @@ export async function abortChatRunsForSessionKeyWithPartials(params: {
keyPrefix: "agent:",
preserveSideRuns: params.preserveSideRuns,
});
- const { matchedSessionRuns: matchedPendingChatRuns, authorizedRuns: authorizedPendingChatRuns } =
- resolveAuthorizedPreRegisteredRunsForSessionKeys({
- context: params.context,
- sessionKeys,
- agentId: params.agentId,
- defaultAgentId: params.defaultAgentId,
- requester: params.requester,
- keyPrefix: PENDING_CHAT_SEND_DEDUPE_PREFIX,
- preserveSideRuns: params.preserveSideRuns,
- });
- const unauthorizedOnly =
- matchedSessionRuns > 0 ||
- matchedPendingAgentRuns > 0 ||
- matchedPendingChatRuns > 0 ||
- queuedAbort.unauthorizedOnly;
- if (
- authorizedRuns.length === 0 &&
- authorizedPendingAgentRuns.length === 0 &&
- authorizedPendingChatRuns.length === 0 &&
- queuedAbort.runIds.length === 0
- ) {
- if (unauthorizedOnly) {
+ const {
+ authorizedRuns: authorizedPendingChatRuns,
+ hasUnauthorizedRuns: hasUnauthorizedPendingChatRuns,
+ hasUnauthorizedProtectedRuns: hasUnauthorizedProtectedPendingChatRuns,
+ hasProtectedRuns: hasProtectedPendingChatRuns,
+ } = resolveAuthorizedPreRegisteredRunsForSessionKeys({
+ context: params.context,
+ sessionKeys,
+ agentId: params.agentId,
+ defaultAgentId: params.defaultAgentId,
+ requester: params.requester,
+ keyPrefix: PENDING_CHAT_SEND_DEDUPE_PREFIX,
+ preserveSideRuns: params.preserveSideRuns,
+ });
+ const hasAuthorizedGatewayRuns =
+ authorizedRuns.length > 0 ||
+ authorizedPendingAgentRuns.length > 0 ||
+ authorizedPendingChatRuns.length > 0 ||
+ queuedPlan.authorized.length > 0;
+ const workerService = asWorkerInferenceControl(params.context.workerEnvironmentService);
+ const workerSessionId = params.sessionId;
+ const hasWorkerRun = Boolean(
+ workerSessionId &&
+ (!hasAuthorizedGatewayRuns || params.onAuthorizedAfterQueuedAbort) &&
+ workerService?.hasInferenceForSession(workerSessionId),
+ );
+ // The worker manager admits at most one active inference per session, and a
+ // worker-backed turn shares its controller's runId. One exact match therefore
+ // represents the only worker owner instead of inventing a second owner.
+ const hasControllerRepresentedWorkerRun = Boolean(
+ hasWorkerRun &&
+ workerSessionId &&
+ workerService &&
+ matchedActiveRunIds.some((runId) =>
+ workerService.hasInferenceForSession(workerSessionId, runId),
+ ),
+ );
+ const hasUnauthorizedOwner =
+ hasUnauthorizedActiveRuns ||
+ hasUnauthorizedPendingAgentRuns ||
+ hasUnauthorizedPendingChatRuns ||
+ queuedPlan.hasUnauthorizedRuns ||
+ (hasWorkerRun && !hasControllerRepresentedWorkerRun && !params.requester.isAdmin);
+ const hasProtectedLifecycleRuns =
+ hasProtectedActiveRuns || hasProtectedPendingAgentRuns || hasProtectedPendingChatRuns;
+ const hasUnauthorizedProtectedOwner =
+ hasUnauthorizedProtectedActiveRuns ||
+ hasUnauthorizedProtectedPendingAgentRuns ||
+ hasUnauthorizedProtectedPendingChatRuns;
+ const hasUnauthorizedLifecycleOwner =
+ Boolean(params.onAuthorizedAfterQueuedAbort) && hasUnauthorizedProtectedOwner;
+ const canRunLifecycleCleanup = !hasUnauthorizedOwner && !hasProtectedLifecycleRuns;
+ // Keep ordinary chat.abort's admin worker behavior; only the injected broad
+ // lifecycle path must preserve hidden or explicitly preserved Gateway runs.
+ const canCancelWorkerSession = !params.onAuthorizedAfterQueuedAbort || !hasProtectedLifecycleRuns;
+ if (!hasAuthorizedGatewayRuns) {
+ // The injected lifecycle callback must not turn a persisted session id into
+ // a bypass around a matching connection or protected run owner.
+ if (hasUnauthorizedOwner || hasUnauthorizedLifecycleOwner) {
return { aborted: false, runIds: [], unauthorized: true };
}
- const workerService = asWorkerInferenceControl(params.context.workerEnvironmentService);
- if (!params.sessionId || !workerService?.hasInferenceForSession(params.sessionId)) {
- return { aborted: false, runIds: [], unauthorized: false };
- }
- if (!params.requester.isAdmin) {
- return { aborted: false, runIds: [], unauthorized: true };
+ // With no owned Gateway run, the exact persisted session is the boundary,
+ // matching sessions.steer's operator.write behavior for ownerless work.
+ const additionalAborted = canRunLifecycleCleanup
+ ? (params.onAuthorizedAfterQueuedAbort?.() ?? false)
+ : false;
+ if (!hasWorkerRun || !workerSessionId || !params.requester.isAdmin || !canCancelWorkerSession) {
+ return { aborted: additionalAborted, runIds: [], unauthorized: false };
}
const workerRunIds = cancelWorkerInferenceForSession({
context: params.context,
- sessionId: params.sessionId,
+ sessionId: workerSessionId,
});
return {
- aborted: workerRunIds.length > 0,
+ aborted: additionalAborted || workerRunIds.length > 0,
runIds: workerRunIds,
unauthorized: false,
};
@@ -265,8 +304,18 @@ export async function abortChatRunsForSessionKeyWithPartials(params: {
runIds: authorizedRunIdSet,
abortOrigin: params.abortOrigin,
});
- // Queued cancellations already applied above; keep them first in the response.
- const runIds: string[] = [...queuedAbort.runIds];
+ // Abort queued owners before any active-work signal can promote a successor.
+ // Keep them first in the response to preserve the established runIds ordering.
+ const runIds: string[] = abortQueuedChatTurns(
+ ensureChatQueuedTurns(params.context),
+ queuedPlan.authorized,
+ params.stopReason,
+ );
+ // Hidden and preserved side runs must also block broad cleanup: authorization
+ // alone must not let the callback abort work intentionally excluded above.
+ const additionalAborted = canRunLifecycleCleanup
+ ? (params.onAuthorizedAfterQueuedAbort?.() ?? false)
+ : false;
for (const { runId, sessionKey } of authorizedRuns) {
const res = abortChatRunById(params.ops, {
runId,
@@ -300,7 +349,7 @@ export async function abortChatRunsForSessionKeyWithPartials(params: {
});
runIds.push(runId);
}
- if (params.requester.isAdmin) {
+ if (params.requester.isAdmin && canCancelWorkerSession) {
for (const runId of cancelWorkerInferenceForSession({
context: params.context,
sessionId: params.sessionId,
@@ -310,7 +359,7 @@ export async function abortChatRunsForSessionKeyWithPartials(params: {
}
}
}
- const res = { aborted: runIds.length > 0, runIds, unauthorized: false };
+ const res = { aborted: additionalAborted || runIds.length > 0, runIds, unauthorized: false };
if (res.aborted && snapshots.length > 0) {
const abortedRunIds = new Set(runIds);
await persistAbortedPartials({
diff --git a/src/gateway/server-methods/chat.abort-authorization.test.ts b/src/gateway/server-methods/chat.abort-authorization.test.ts
index 13df04e60079..af9c127c6f89 100644
--- a/src/gateway/server-methods/chat.abort-authorization.test.ts
+++ b/src/gateway/server-methods/chat.abort-authorization.test.ts
@@ -4,6 +4,7 @@
import { expectDefined } from "@openclaw/normalization-core";
import { describe, expect, it, vi } from "vitest";
import { createChatRunState } from "../server-chat-state.js";
+import { handleChatAbortRequestWithLifecycle } from "./chat-abort-handler.js";
import {
createActiveRun,
createChatAbortContext,
@@ -32,6 +33,7 @@ async function invokeAbort({
deviceId,
preserveSideRuns,
scopes = ["operator.write"],
+ onAuthorizedAfterQueuedAbort,
}: {
context: ReturnType;
sessionKey?: string;
@@ -40,9 +42,12 @@ async function invokeAbort({
deviceId: string;
preserveSideRuns?: boolean;
scopes?: string[];
+ onAuthorizedAfterQueuedAbort?: () => boolean;
}) {
return await invokeChatAbortHandler({
- handler: expectDefined(chatHandlers["chat.abort"], 'chatHandlers["chat.abort"] test invariant'),
+ handler: onAuthorizedAfterQueuedAbort
+ ? (options) => handleChatAbortRequestWithLifecycle(options, { onAuthorizedAfterQueuedAbort })
+ : expectDefined(chatHandlers["chat.abort"], 'chatHandlers["chat.abort"] test invariant'),
context,
request: {
sessionKey,
@@ -182,6 +187,7 @@ describe("chat.abort authorization", () => {
});
it("does not abort hidden internal runs by visible session key", async () => {
+ const onAuthorizedAfterQueuedAbort = vi.fn(() => true);
const context = createChatAbortContext({
chatAbortControllers: new Map([
["run-hidden", createActiveRun("main", { controlUiVisible: false })],
@@ -192,15 +198,40 @@ describe("chat.abort authorization", () => {
context,
connId: "conn-owner",
deviceId: "dev-owner",
+ onAuthorizedAfterQueuedAbort,
});
const [ok, payload] = requireLastRespondCall(respond);
expect(ok).toBe(true);
expectAbortPayload(payload, { aborted: false, runIds: [] });
+ expect(onAuthorizedAfterQueuedAbort).not.toHaveBeenCalled();
+ expect(context.chatAbortControllers.has("run-hidden")).toBe(true);
+ });
+
+ it("does not reveal a foreign hidden run to ordinary session aborts", async () => {
+ const hidden = createActiveRun("main", {
+ controlUiVisible: false,
+ owner: { connId: "conn-hidden", deviceId: "dev-hidden" },
+ });
+ const context = createChatAbortContext({
+ chatAbortControllers: new Map([["run-hidden", hidden]]),
+ });
+
+ const respond = await invokeAbort({
+ context,
+ connId: "conn-other",
+ deviceId: "dev-other",
+ });
+
+ const [ok, payload] = requireLastRespondCall(respond);
+ expect(ok).toBe(true);
+ expectAbortPayload(payload, { aborted: false, runIds: [] });
+ expect(hidden.controller.signal.aborted).toBe(false);
expect(context.chatAbortControllers.has("run-hidden")).toBe(true);
});
it("preserves BTW runs for TUI session stops", async () => {
+ const onAuthorizedAfterQueuedAbort = vi.fn(() => true);
const main = createActiveRun("main", {
owner: { connId: "conn-owner", deviceId: "dev-owner" },
});
@@ -220,6 +251,7 @@ describe("chat.abort authorization", () => {
connId: "conn-owner",
deviceId: "dev-owner",
preserveSideRuns: true,
+ onAuthorizedAfterQueuedAbort,
});
const [ok, payload] = requireLastRespondCall(respond);
@@ -227,10 +259,12 @@ describe("chat.abort authorization", () => {
expectAbortPayload(payload, { aborted: true, runIds: ["run-main"] });
expect(main.controller.signal.aborted).toBe(true);
expect(btw.controller.signal.aborted).toBe(false);
+ expect(onAuthorizedAfterQueuedAbort).not.toHaveBeenCalled();
expect(context.chatAbortControllers.has("run-btw")).toBe(true);
});
it("preserves BTW runs waiting for chat admission", async () => {
+ const onAuthorizedAfterQueuedAbort = vi.fn(() => true);
const context = createChatAbortContext();
context.dedupe.set("pending-chat:run-btw", {
ts: Date.now(),
@@ -250,11 +284,13 @@ describe("chat.abort authorization", () => {
connId: "conn-owner",
deviceId: "dev-owner",
preserveSideRuns: true,
+ onAuthorizedAfterQueuedAbort,
});
const [ok, payload] = requireLastRespondCall(respond);
expect(ok).toBe(true);
expectAbortPayload(payload, { aborted: false, runIds: [] });
+ expect(onAuthorizedAfterQueuedAbort).not.toHaveBeenCalled();
expect(context.dedupe.get("pending-chat:run-btw")).toEqual(
expect.objectContaining({
payload: expect.objectContaining({ status: "accepted", turnKind: "btw" }),
@@ -337,6 +373,449 @@ describe("chat.abort authorization", () => {
});
describe("chat.abort queued-turn contract", () => {
+ it("cancels queued turns before session cleanup and the active run", async () => {
+ const order: string[] = [];
+ const queuedController = new AbortController();
+ queuedController.signal.addEventListener("abort", () => order.push("queued-abort"));
+ const active = createActiveRun("main", {
+ owner: { connId: "conn-owner", deviceId: "dev-owner" },
+ });
+ active.controller.signal.addEventListener("abort", () => order.push("active-abort"));
+ const context = createChatAbortContext({
+ chatAbortControllers: new Map([["active-1", active]]),
+ chatQueuedTurns: new Map([
+ [
+ "queued-1",
+ {
+ controller: queuedController,
+ sessionId: "main-session",
+ sessionKey: "main",
+ ownerConnId: "conn-owner",
+ ownerDeviceId: "dev-owner",
+ },
+ ],
+ ]),
+ });
+
+ const respond = await invokeAbort({
+ context,
+ connId: "conn-owner",
+ deviceId: "dev-owner",
+ onAuthorizedAfterQueuedAbort: () => {
+ order.push("session-cleanup");
+ return true;
+ },
+ });
+
+ expect(requireLastRespondCall(respond)[0]).toBe(true);
+ expect(order).toEqual(["queued-abort", "session-cleanup", "active-abort"]);
+ });
+
+ it("cancels a queued-only turn before session cleanup", async () => {
+ const order: string[] = [];
+ const queuedController = new AbortController();
+ queuedController.signal.addEventListener("abort", () => order.push("queued-abort"));
+ const context = createChatAbortContext({
+ chatQueuedTurns: new Map([
+ [
+ "queued-1",
+ {
+ controller: queuedController,
+ sessionId: "main-session",
+ sessionKey: "main",
+ ownerConnId: "conn-owner",
+ ownerDeviceId: "dev-owner",
+ },
+ ],
+ ]),
+ });
+
+ const respond = await invokeAbort({
+ context,
+ connId: "conn-owner",
+ deviceId: "dev-owner",
+ onAuthorizedAfterQueuedAbort: () => {
+ order.push("session-cleanup");
+ return true;
+ },
+ });
+
+ expect(requireLastRespondCall(respond)[0]).toBe(true);
+ expect(order).toEqual(["queued-abort", "session-cleanup"]);
+ });
+
+ it("does not let session cleanup bypass a foreign chat owner", async () => {
+ const onAuthorizedAfterQueuedAbort = vi.fn(() => false);
+ const context = createSingleAbortContext();
+
+ const respond = await invokeAbort({
+ context,
+ connId: "conn-other",
+ deviceId: "dev-other",
+ onAuthorizedAfterQueuedAbort,
+ });
+
+ const call = requireLastRespondCall(respond);
+ expect(call[0]).toBe(false);
+ expect(call[2]?.message).toBe("unauthorized");
+ expect(onAuthorizedAfterQueuedAbort).not.toHaveBeenCalled();
+ expect(context.chatAbortControllers.has("run-1")).toBe(true);
+ });
+
+ it("allows operator.write session cleanup when no chat run is registered", async () => {
+ const onAuthorizedAfterQueuedAbort = vi.fn(() => true);
+ const respond = await invokeAbort({
+ context: createChatAbortContext(),
+ connId: "conn-owner",
+ deviceId: "dev-owner",
+ onAuthorizedAfterQueuedAbort,
+ });
+
+ expect(onAuthorizedAfterQueuedAbort).toHaveBeenCalledTimes(1);
+ expectAbortPayload(requireLastRespondCall(respond)[1], { aborted: true, runIds: [] });
+ });
+
+ it("does not count a controller-represented worker run as a second owner", async () => {
+ const onAuthorizedAfterQueuedAbort = vi.fn(() => true);
+ const cancelInferenceForSession = vi.fn(() => ["run-1"]);
+ const context = createSingleAbortContext();
+ context.workerEnvironmentService = {
+ cancelInferenceForSession,
+ hasInferenceForSession: (sessionId: string, runId?: string) =>
+ sessionId === "main-session" && (!runId || runId === "run-1"),
+ } as never;
+
+ const respond = await invokeAbort({
+ context,
+ connId: "conn-owner",
+ deviceId: "dev-owner",
+ onAuthorizedAfterQueuedAbort,
+ });
+
+ expectAbortPayload(requireLastRespondCall(respond)[1], {
+ aborted: true,
+ runIds: ["run-1"],
+ });
+ expect(onAuthorizedAfterQueuedAbort).toHaveBeenCalledTimes(1);
+ expect(cancelInferenceForSession).not.toHaveBeenCalled();
+ });
+
+ it("does not let session cleanup bypass a worker run", async () => {
+ const onAuthorizedAfterQueuedAbort = vi.fn(() => false);
+ const cancelInferenceForSession = vi.fn(() => ["worker-run"]);
+ const context = createChatAbortContext({
+ workerEnvironmentService: {
+ cancelInferenceForSession,
+ hasInferenceForSession: () => true,
+ },
+ });
+
+ const respond = await invokeAbort({
+ context,
+ connId: "conn-other",
+ deviceId: "dev-other",
+ onAuthorizedAfterQueuedAbort,
+ });
+
+ const call = requireLastRespondCall(respond);
+ expect(call[0]).toBe(false);
+ expect(call[2]?.message).toBe("unauthorized");
+ expect(onAuthorizedAfterQueuedAbort).not.toHaveBeenCalled();
+ expect(cancelInferenceForSession).not.toHaveBeenCalled();
+ });
+
+ it("protects hidden worker runs only from injected lifecycle cleanup", async () => {
+ const onAuthorizedAfterQueuedAbort = vi.fn(() => true);
+ const cancelInferenceForSession = vi.fn(() => ["run-hidden"]);
+ const hidden = createActiveRun("main", {
+ controlUiVisible: false,
+ owner: { connId: "conn-owner", deviceId: "dev-owner" },
+ });
+ const context = createChatAbortContext({
+ chatAbortControllers: new Map([["run-hidden", hidden]]),
+ workerEnvironmentService: {
+ cancelInferenceForSession,
+ hasInferenceForSession: (sessionId: string, runId?: string) =>
+ sessionId === "main-session" && (!runId || runId === "run-hidden"),
+ },
+ });
+
+ const lifecycleRespond = await invokeAbort({
+ context,
+ connId: "conn-owner",
+ deviceId: "dev-owner",
+ onAuthorizedAfterQueuedAbort,
+ });
+
+ expectAbortPayload(requireLastRespondCall(lifecycleRespond)[1], {
+ aborted: false,
+ runIds: [],
+ });
+ expect(onAuthorizedAfterQueuedAbort).not.toHaveBeenCalled();
+ expect(cancelInferenceForSession).not.toHaveBeenCalled();
+ expect(hidden.controller.signal.aborted).toBe(false);
+
+ const ordinaryRespond = await invokeAbort({
+ context,
+ connId: "conn-admin",
+ deviceId: "dev-admin",
+ scopes: ["operator.admin"],
+ });
+ expectAbortPayload(requireLastRespondCall(ordinaryRespond)[1], {
+ aborted: true,
+ runIds: ["run-hidden"],
+ });
+ expect(cancelInferenceForSession).toHaveBeenCalledWith({ sessionId: "main-session" });
+ });
+
+ it("aborts only the requester runs without session cleanup in a mixed-owner session", async () => {
+ const onAuthorizedAfterQueuedAbort = vi.fn(() => true);
+ const mine = createActiveRun("main", {
+ owner: { connId: "conn-owner", deviceId: "dev-owner" },
+ });
+ const foreign = createActiveRun("main", {
+ owner: { connId: "conn-other", deviceId: "dev-other" },
+ });
+ const context = createChatAbortContext({
+ chatAbortControllers: new Map([
+ ["run-mine", mine],
+ ["run-foreign", foreign],
+ ]),
+ });
+
+ const respond = await invokeAbort({
+ context,
+ connId: "conn-owner",
+ deviceId: "dev-owner",
+ onAuthorizedAfterQueuedAbort,
+ });
+
+ expectAbortPayload(requireLastRespondCall(respond)[1], {
+ aborted: true,
+ runIds: ["run-mine"],
+ });
+ expect(onAuthorizedAfterQueuedAbort).not.toHaveBeenCalled();
+ expect(mine.controller.signal.aborted).toBe(true);
+ expect(foreign.controller.signal.aborted).toBe(false);
+ expect(context.chatAbortControllers.has("run-foreign")).toBe(true);
+ });
+
+ it("does not let session cleanup bypass a foreign hidden owner", async () => {
+ const onAuthorizedAfterQueuedAbort = vi.fn(() => true);
+ const hidden = createActiveRun("main", {
+ controlUiVisible: false,
+ owner: { connId: "conn-hidden", deviceId: "dev-hidden" },
+ });
+ const context = createChatAbortContext({
+ chatAbortControllers: new Map([["run-hidden", hidden]]),
+ });
+
+ const respond = await invokeAbort({
+ context,
+ connId: "conn-other",
+ deviceId: "dev-other",
+ onAuthorizedAfterQueuedAbort,
+ });
+
+ const call = requireLastRespondCall(respond);
+ expect(call[0]).toBe(false);
+ expect(call[2]?.message).toBe("unauthorized");
+ expect(onAuthorizedAfterQueuedAbort).not.toHaveBeenCalled();
+ expect(hidden.controller.signal.aborted).toBe(false);
+ expect(context.chatAbortControllers.has("run-hidden")).toBe(true);
+ });
+
+ it("aborts an owned run without cleanup when a foreign hidden run shares the session", async () => {
+ const onAuthorizedAfterQueuedAbort = vi.fn(() => true);
+ const mine = createActiveRun("main", {
+ owner: { connId: "conn-owner", deviceId: "dev-owner" },
+ });
+ const hidden = createActiveRun("main", {
+ controlUiVisible: false,
+ owner: { connId: "conn-hidden", deviceId: "dev-hidden" },
+ });
+ const context = createChatAbortContext({
+ chatAbortControllers: new Map([
+ ["run-mine", mine],
+ ["run-hidden", hidden],
+ ]),
+ });
+
+ const respond = await invokeAbort({
+ context,
+ connId: "conn-owner",
+ deviceId: "dev-owner",
+ onAuthorizedAfterQueuedAbort,
+ });
+
+ expectAbortPayload(requireLastRespondCall(respond)[1], {
+ aborted: true,
+ runIds: ["run-mine"],
+ });
+ expect(onAuthorizedAfterQueuedAbort).not.toHaveBeenCalled();
+ expect(mine.controller.signal.aborted).toBe(true);
+ expect(hidden.controller.signal.aborted).toBe(false);
+ expect(context.chatAbortControllers.has("run-hidden")).toBe(true);
+ });
+
+ it("allows cleanup for duplicate pending identities owned by the requester", async () => {
+ const onAuthorizedAfterQueuedAbort = vi.fn(() => true);
+ const pending = {
+ ts: Date.now(),
+ ok: true,
+ payload: {
+ runId: "run-pending",
+ sessionKey: "main",
+ status: "accepted",
+ ownerConnId: "conn-owner",
+ ownerDeviceId: "dev-owner",
+ dedupeKeys: ["agent:run-pending-alias"],
+ },
+ };
+ const context = createChatAbortContext({
+ dedupe: new Map([
+ ["agent:run-pending", pending],
+ ["agent:run-pending-alias", pending],
+ ]),
+ });
+
+ const respond = await invokeAbort({
+ context,
+ connId: "conn-owner",
+ deviceId: "dev-owner",
+ onAuthorizedAfterQueuedAbort,
+ });
+
+ expectAbortPayload(requireLastRespondCall(respond)[1], {
+ aborted: true,
+ runIds: ["run-pending"],
+ });
+ expect(onAuthorizedAfterQueuedAbort).toHaveBeenCalledTimes(1);
+ });
+
+ it("does not run session cleanup around hidden pending work", async () => {
+ const onAuthorizedAfterQueuedAbort = vi.fn(() => true);
+ const context = createChatAbortContext();
+ context.dedupe.set("agent:run-hidden", {
+ ts: Date.now(),
+ ok: true,
+ payload: {
+ runId: "run-hidden",
+ sessionKey: "main",
+ status: "accepted",
+ controlUiVisible: false,
+ ownerConnId: "conn-owner",
+ ownerDeviceId: "dev-owner",
+ },
+ });
+
+ const respond = await invokeAbort({
+ context,
+ connId: "conn-owner",
+ deviceId: "dev-owner",
+ onAuthorizedAfterQueuedAbort,
+ });
+
+ expectAbortPayload(requireLastRespondCall(respond)[1], {
+ aborted: false,
+ runIds: [],
+ });
+ expect(onAuthorizedAfterQueuedAbort).not.toHaveBeenCalled();
+ expect(context.dedupe.get("agent:run-hidden")).toEqual(
+ expect.objectContaining({
+ payload: expect.objectContaining({ status: "accepted", controlUiVisible: false }),
+ }),
+ );
+ });
+
+ it("protects foreign hidden pending work across ordinary and lifecycle aborts", async () => {
+ const context = createChatAbortContext();
+ context.dedupe.set("agent:run-hidden", {
+ ts: Date.now(),
+ ok: true,
+ payload: {
+ runId: "run-hidden",
+ sessionKey: "main",
+ status: "accepted",
+ controlUiVisible: false,
+ ownerConnId: "conn-hidden",
+ ownerDeviceId: "dev-hidden",
+ },
+ });
+
+ const respond = await invokeAbort({
+ context,
+ connId: "conn-other",
+ deviceId: "dev-other",
+ });
+
+ const [ok, payload] = requireLastRespondCall(respond);
+ expect(ok).toBe(true);
+ expectAbortPayload(payload, { aborted: false, runIds: [] });
+
+ const onAuthorizedAfterQueuedAbort = vi.fn(() => true);
+ const lifecycleRespond = await invokeAbort({
+ context,
+ connId: "conn-other",
+ deviceId: "dev-other",
+ onAuthorizedAfterQueuedAbort,
+ });
+ const lifecycleCall = requireLastRespondCall(lifecycleRespond);
+ expect(lifecycleCall[0]).toBe(false);
+ expect(lifecycleCall[2]?.message).toBe("unauthorized");
+ expect(onAuthorizedAfterQueuedAbort).not.toHaveBeenCalled();
+ expect(context.dedupe.get("agent:run-hidden")).toEqual(
+ expect.objectContaining({
+ payload: expect.objectContaining({ status: "accepted", controlUiVisible: false }),
+ }),
+ );
+ });
+
+ it("skips session cleanup when a pending run has a foreign owner", async () => {
+ const onAuthorizedAfterQueuedAbort = vi.fn(() => true);
+ const context = createChatAbortContext();
+ context.dedupe.set("agent:run-mine", {
+ ts: Date.now(),
+ ok: true,
+ payload: {
+ runId: "run-mine",
+ sessionKey: "main",
+ status: "accepted",
+ ownerConnId: "conn-owner",
+ ownerDeviceId: "dev-owner",
+ },
+ });
+ context.dedupe.set("agent:run-foreign", {
+ ts: Date.now(),
+ ok: true,
+ payload: {
+ runId: "run-foreign",
+ sessionKey: "main",
+ status: "accepted",
+ ownerConnId: "conn-other",
+ ownerDeviceId: "dev-other",
+ },
+ });
+
+ const respond = await invokeAbort({
+ context,
+ connId: "conn-owner",
+ deviceId: "dev-owner",
+ onAuthorizedAfterQueuedAbort,
+ });
+
+ expectAbortPayload(requireLastRespondCall(respond)[1], {
+ aborted: true,
+ runIds: ["run-mine"],
+ });
+ expect(onAuthorizedAfterQueuedAbort).not.toHaveBeenCalled();
+ expect(context.dedupe.get("agent:run-foreign")).toEqual(
+ expect.objectContaining({
+ payload: expect.objectContaining({ status: "accepted" }),
+ }),
+ );
+ });
+
it("aborts a queued turn by runId after active registration is gone", async () => {
const controller = new AbortController();
const context = createChatAbortContext({
@@ -471,6 +950,7 @@ describe("chat.abort queued-turn contract", () => {
});
it("session abort does not clear another owner's queued turns", async () => {
+ const onAuthorizedAfterQueuedAbort = vi.fn(() => true);
const foreign = new AbortController();
const context = createChatAbortContext({
chatQueuedTurns: new Map([
@@ -491,10 +971,58 @@ describe("chat.abort queued-turn contract", () => {
context,
connId: "conn-other",
deviceId: "dev-other",
+ onAuthorizedAfterQueuedAbort,
});
const call = requireLastRespondCall(respond);
// unauthorized when only foreign queued matches
expect(call[0]).toBe(false);
+ expect(onAuthorizedAfterQueuedAbort).not.toHaveBeenCalled();
+ expect(foreign.signal.aborted).toBe(false);
+ expect(context.chatQueuedTurns.has("queued-foreign")).toBe(true);
+ });
+
+ it("aborts only requester queues without session cleanup in a mixed-owner session", async () => {
+ const onAuthorizedAfterQueuedAbort = vi.fn(() => true);
+ const mine = new AbortController();
+ const foreign = new AbortController();
+ const context = createChatAbortContext({
+ chatQueuedTurns: new Map([
+ [
+ "queued-mine",
+ {
+ controller: mine,
+ sessionId: "main-session",
+ sessionKey: "main",
+ ownerConnId: "conn-owner",
+ ownerDeviceId: "dev-owner",
+ },
+ ],
+ [
+ "queued-foreign",
+ {
+ controller: foreign,
+ sessionId: "main-session",
+ sessionKey: "main",
+ ownerConnId: "conn-other",
+ ownerDeviceId: "dev-other",
+ },
+ ],
+ ]),
+ });
+
+ const respond = await invokeAbort({
+ context,
+ connId: "conn-owner",
+ deviceId: "dev-owner",
+ onAuthorizedAfterQueuedAbort,
+ });
+
+ expectAbortPayload(requireLastRespondCall(respond)[1], {
+ aborted: true,
+ runIds: ["queued-mine"],
+ });
+ expect(onAuthorizedAfterQueuedAbort).not.toHaveBeenCalled();
+ expect(mine.signal.aborted).toBe(true);
expect(foreign.signal.aborted).toBe(false);
expect(context.chatQueuedTurns.has("queued-foreign")).toBe(true);
});
diff --git a/src/gateway/server-methods/session-active-runs.test.ts b/src/gateway/server-methods/session-active-runs.test.ts
index 52d5a1e878ec..47905e2715c4 100644
--- a/src/gateway/server-methods/session-active-runs.test.ts
+++ b/src/gateway/server-methods/session-active-runs.test.ts
@@ -1,5 +1,13 @@
// Tests gateway active-run matching by logical session key and backing id.
import { expect, it } from "vitest";
+import type { EmbeddedAgentQueueHandle } from "../../agents/embedded-agent-runner/run-state.js";
+import {
+ abortEmbeddedAgentRun,
+ clearActiveEmbeddedRun,
+ isEmbeddedAgentRunActive,
+ setActiveEmbeddedRun,
+} from "../../agents/embedded-agent-runner/runs.js";
+import { createReplyOperation } from "../../auto-reply/reply/reply-run-registry.js";
import { clearAgentRunContext, registerAgentRunContext } from "../../infra/agent-events.js";
import {
hasVisibleActiveSessionRun,
@@ -69,3 +77,129 @@ it("projects a lifecycle-owned worker run without widening event visibility", ()
clearAgentRunContext("worker-run");
}
});
+
+it("does not project a terminal reply operation retained for settlement as active", () => {
+ const sessionKey = "agent:main:reply-settling";
+ const sessionId = "reply-settling-session";
+ const operation = createReplyOperation({ sessionKey, sessionId, resetTriggered: false });
+ const replacementHandle: EmbeddedAgentQueueHandle = {
+ abort: () => undefined,
+ isAborted: () => false,
+ isCompacting: () => false,
+ isStreaming: () => true,
+ queueMessage: async () => undefined,
+ };
+ try {
+ expect(
+ resolveVisibleActiveSessionRunState({
+ context: {},
+ requestedKey: sessionKey,
+ canonicalKey: sessionKey,
+ sessionId,
+ }),
+ ).toEqual({ active: true, runIds: [] });
+
+ operation.setPhase("running");
+ expect(operation.abortByUser()).toBe(true);
+ expect(isEmbeddedAgentRunActive(sessionId)).toBe(true);
+ expect(
+ resolveVisibleActiveSessionRunState({
+ context: {},
+ requestedKey: sessionKey,
+ canonicalKey: sessionKey,
+ sessionId,
+ }),
+ ).toEqual({ active: false, runIds: [] });
+
+ setActiveEmbeddedRun(sessionId, replacementHandle, sessionKey);
+ expect(
+ resolveVisibleActiveSessionRunState({
+ context: {},
+ requestedKey: sessionKey,
+ canonicalKey: sessionKey,
+ sessionId,
+ }),
+ ).toEqual({ active: true, runIds: [] });
+ } finally {
+ clearActiveEmbeddedRun(sessionId, replacementHandle, sessionKey);
+ operation.complete();
+ }
+});
+
+it("preserves an independent lifecycle-owned worker while a reply operation settles", () => {
+ const sessionKey = "agent:main:worker-overlap";
+ const sessionId = "worker-overlap-session";
+ const operation = createReplyOperation({ sessionKey, sessionId, resetTriggered: false });
+ registerAgentRunContext("worker-overlap-run", {
+ projectSessionActive: true,
+ sessionId,
+ sessionKey,
+ });
+ try {
+ expect(operation.abortByUser()).toBe(true);
+ expect(
+ resolveVisibleActiveSessionRunState({
+ context: {},
+ requestedKey: sessionKey,
+ canonicalKey: sessionKey,
+ sessionId,
+ }),
+ ).toEqual({ active: true, runIds: [] });
+ } finally {
+ operation.complete();
+ clearAgentRunContext("worker-overlap-run");
+ }
+});
+
+it("does not project an aborted embedded handle retained for cleanup as active", () => {
+ const sessionKey = "agent:main:handle-settling";
+ const sessionId = "handle-settling-session";
+ let aborted = false;
+ const handle: EmbeddedAgentQueueHandle = {
+ abort: () => {
+ aborted = true;
+ },
+ isAborted: () => aborted,
+ isCompacting: () => false,
+ // Prompt completion closes steering before post-turn finalization. That
+ // state alone must not make a normally finishing run disappear.
+ isStopped: () => true,
+ isStreaming: () => false,
+ queueMessage: async () => undefined,
+ };
+ setActiveEmbeddedRun(sessionId, handle, sessionKey);
+ try {
+ expect(
+ resolveVisibleActiveSessionRunState({
+ context: {},
+ requestedKey: sessionKey,
+ canonicalKey: sessionKey,
+ sessionId,
+ }),
+ ).toEqual({ active: true, runIds: [] });
+
+ expect(abortEmbeddedAgentRun(sessionId)).toBe(true);
+ expect(isEmbeddedAgentRunActive(sessionId)).toBe(true);
+ expect(
+ resolveVisibleActiveSessionRunState({
+ context: {},
+ requestedKey: sessionKey,
+ canonicalKey: sessionKey,
+ sessionId,
+ }),
+ ).toEqual({ active: false, runIds: [] });
+
+ expect(
+ resolveVisibleActiveSessionRunState({
+ context: {
+ chatAbortControllers: new Map([["new-run", { sessionId, sessionKey }]]),
+ } as never,
+ requestedKey: sessionKey,
+ canonicalKey: sessionKey,
+ sessionId,
+ }),
+ ).toEqual({ active: true, runIds: ["new-run"] });
+ } finally {
+ clearActiveEmbeddedRun(sessionId, handle, sessionKey);
+ }
+});
diff --git a/src/gateway/server-methods/session-active-runs.ts b/src/gateway/server-methods/session-active-runs.ts
index e04a0d5b9d46..d87ff3376ad3 100644
--- a/src/gateway/server-methods/session-active-runs.ts
+++ b/src/gateway/server-methods/session-active-runs.ts
@@ -1,4 +1,4 @@
-import { isEmbeddedAgentRunActive } from "../../agents/embedded-agent-runner/runs.js";
+import { isEmbeddedAgentRunInProgress } from "../../agents/embedded-agent-runner/runs.js";
import { hasProjectedAgentRunForSession } from "../../infra/agent-events.js";
import { normalizeAgentId } from "../../routing/session-key.js";
import type { GatewayRequestContext } from "./types.js";
@@ -116,11 +116,11 @@ export function resolveVisibleActiveSessionRunState(params: {
sessionKeys: [params.requestedKey, params.canonicalKey],
...(sessionId ? { sessionId } : {}),
});
+ const embeddedRunInProgress = sessionId !== undefined && isEmbeddedAgentRunInProgress(sessionId);
+ // Connection, worker-lifecycle, and embedded registries are independent owners.
+ // Settlement in one must not hide live work owned by another.
return {
- active:
- runIds.length > 0 ||
- hasProjectedRun ||
- (sessionId !== undefined && isEmbeddedAgentRunActive(sessionId)),
+ active: runIds.length > 0 || hasProjectedRun || embeddedRunInProgress,
runIds,
};
}
diff --git a/src/gateway/server-methods/sessions-abort.ts b/src/gateway/server-methods/sessions-abort.ts
index a5b06c80c364..f6fc6e61e947 100644
--- a/src/gateway/server-methods/sessions-abort.ts
+++ b/src/gateway/server-methods/sessions-abort.ts
@@ -1,5 +1,4 @@
// Session active-run cancellation and agent-scope resolution.
-import { expectDefined } from "@openclaw/normalization-core";
import {
normalizeOptionalString,
readStringValue,
@@ -10,6 +9,8 @@ import {
validateSessionsAbortParams,
} from "../../../packages/gateway-protocol/src/index.js";
import { resolveDefaultAgentId } from "../../agents/agent-scope.js";
+import { abortEmbeddedAgentRun } from "../../agents/embedded-agent-runner/runs.js";
+import { clearSessionQueues } from "../../auto-reply/reply/queue/cleanup.js";
import {
isConfiguredSessionStoreAgentId,
resolveExistingAgentSessionStoreTargetsSync,
@@ -28,7 +29,7 @@ import { loadSessionEntry } from "../session-utils.js";
import { asWorkerInferenceControl } from "../worker-environments/inference-control.js";
import { resolveWorkerSessionTarget } from "../worker-environments/session-target.js";
import { setGatewayDedupeEntry } from "./agent-job.js";
-import { chatHandlers } from "./chat.js";
+import { handleChatAbortRequestWithLifecycle } from "./chat-abort-handler.js";
import { emitSessionsChanged } from "./session-change-event.js";
import { requireSessionKey } from "./sessions-shared.js";
import type { GatewayRequestContext, GatewayRequestHandlers } from "./types.js";
@@ -125,6 +126,7 @@ export const sessionAbortHandlers: GatewayRequestHandlers = {
const requestedRunId = readStringValue(p.runId);
const requestedKey = normalizeOptionalString(p.key);
const requestedParamAgentId = normalizeOptionalString(p.agentId);
+ const clearQueued = p.clearQueued === true;
const workerRunSessionId = requestedRunId
? asWorkerInferenceControl(context.workerEnvironmentService)?.resolveInferenceSessionForRunId(
requestedRunId,
@@ -221,14 +223,18 @@ export const sessionAbortHandlers: GatewayRequestHandlers = {
}
// An exact live controller is already authoritative. Avoid opening the fallback store when
// neither config nor persistence owns it; that edge is the only one that could create state.
- const canonicalKey =
+ const loadedSession =
configuredTarget || existingTargets.length > 0
- ? loadSessionEntry(key, { agentId: requestedGlobalAgentId }).canonicalKey
- : resolveSessionStoreKey({
- cfg,
- sessionKey: key,
- ...(requestedGlobalAgentId ? { storeAgentId: requestedGlobalAgentId } : {}),
- });
+ ? loadSessionEntry(key, { agentId: requestedGlobalAgentId })
+ : undefined;
+ const canonicalKey =
+ loadedSession?.canonicalKey ??
+ resolveSessionStoreKey({
+ cfg,
+ sessionKey: key,
+ ...(requestedGlobalAgentId ? { storeAgentId: requestedGlobalAgentId } : {}),
+ });
+ const sessionEntry = loadedSession?.entry;
const requestedKeyAliases =
requestedKey &&
requestedKey !== key &&
@@ -260,68 +266,105 @@ export const sessionAbortHandlers: GatewayRequestHandlers = {
}
}
let abortedRunId: string | null = null;
- await expectDefined(
- chatHandlers["chat.abort"],
- "chat.abort handler",
- )({
- req,
- params: {
- sessionKey: abortSessionKey,
- runId: requestedRunId,
- ...(abortAgentId ? { agentId: abortAgentId } : {}),
- },
- respond: (ok, payload, error, meta) => {
- if (!ok) {
- respond(ok, payload, error, meta);
- return;
- }
- const runIds =
- payload &&
- typeof payload === "object" &&
- Array.isArray((payload as { runIds?: unknown[] }).runIds)
- ? (payload as { runIds: unknown[] }).runIds.filter((value): value is string =>
- Boolean(normalizeOptionalString(value)),
- )
- : [];
- const firstAbortedRunId = runIds[0] ?? null;
- abortedRunId = firstAbortedRunId;
- const workerOnly = Boolean(workerRunSessionId && !activeRun);
- if (firstAbortedRunId && !workerOnly) {
- const endedAt = Date.now();
- const runKind = preAbortRunKinds.get(firstAbortedRunId);
- const dedupePrefix = runKind === "agent" ? "agent" : "chat";
- setGatewayDedupeEntry({
- dedupe: context.dedupe,
- key: `${dedupePrefix}:${firstAbortedRunId}`,
- entry: {
- ts: endedAt,
- ok: true,
- payload: {
- status: "timeout",
- runId: firstAbortedRunId,
- ...(abortAgentId ? { agentId: abortAgentId } : {}),
- stopReason: "rpc",
- endedAt,
+ let aborted = false;
+ let chatAbortSucceeded = false;
+ let responseMeta: Record | undefined;
+ const persistedSessionId = sessionEntry?.sessionId;
+ const onAuthorizedAfterQueuedAbort =
+ !requestedRunId && canonicalKey !== "global" && (clearQueued || persistedSessionId)
+ ? () => {
+ let queueCleared = false;
+ if (clearQueued) {
+ // Explicit full-session stops clear first so an aborting run cannot
+ // promote queued work. Ordinary sessions.abort calls preserve it.
+ const cleared = clearSessionQueues([
+ key,
+ ...(requestedKeyAliases ?? []),
+ canonicalKey,
+ ...(persistedSessionId ? [persistedSessionId] : []),
+ ]);
+ queueCleared = cleared.followupCleared > 0 || cleared.laneCleared > 0;
+ }
+ // Persisted channel replies are active session work even when they
+ // have no connection-owned chat controller.
+ const embeddedAborted = persistedSessionId
+ ? abortEmbeddedAgentRun(persistedSessionId)
+ : false;
+ return embeddedAborted || queueCleared;
+ }
+ : undefined;
+ await handleChatAbortRequestWithLifecycle(
+ {
+ req,
+ params: {
+ sessionKey: abortSessionKey,
+ runId: requestedRunId,
+ ...(abortAgentId ? { agentId: abortAgentId } : {}),
+ },
+ respond: (ok, payload, error, meta) => {
+ if (!ok) {
+ respond(ok, payload, error, meta);
+ return;
+ }
+ chatAbortSucceeded = true;
+ responseMeta = meta;
+ const runIds =
+ payload &&
+ typeof payload === "object" &&
+ Array.isArray((payload as { runIds?: unknown[] }).runIds)
+ ? (payload as { runIds: unknown[] }).runIds.filter((value): value is string =>
+ Boolean(normalizeOptionalString(value)),
+ )
+ : [];
+ const firstAbortedRunId = runIds[0] ?? null;
+ abortedRunId = firstAbortedRunId;
+ aborted =
+ firstAbortedRunId !== null ||
+ (payload !== null &&
+ typeof payload === "object" &&
+ (payload as { aborted?: unknown }).aborted === true);
+ const workerOnly = Boolean(workerRunSessionId && !activeRun);
+ if (firstAbortedRunId && !workerOnly) {
+ const endedAt = Date.now();
+ const runKind = preAbortRunKinds.get(firstAbortedRunId);
+ const dedupePrefix = runKind === "agent" ? "agent" : "chat";
+ setGatewayDedupeEntry({
+ dedupe: context.dedupe,
+ key: `${dedupePrefix}:${firstAbortedRunId}`,
+ entry: {
+ ts: endedAt,
+ ok: true,
+ payload: {
+ status: "timeout",
+ runId: firstAbortedRunId,
+ ...(abortAgentId ? { agentId: abortAgentId } : {}),
+ stopReason: "rpc",
+ endedAt,
+ },
},
- },
- });
- }
- respond(
- true,
- {
- ok: true,
- abortedRunId,
- status: abortedRunId ? "aborted" : "no-active-run",
- },
- undefined,
- meta,
- );
+ });
+ }
+ },
+ context,
+ client,
+ isWebchatConnect,
},
- context,
- client,
- isWebchatConnect,
- });
- if (abortedRunId) {
+ onAuthorizedAfterQueuedAbort ? { onAuthorizedAfterQueuedAbort } : {},
+ );
+ if (!chatAbortSucceeded) {
+ return;
+ }
+ respond(
+ true,
+ {
+ ok: true,
+ abortedRunId,
+ status: aborted ? "aborted" : "no-active-run",
+ },
+ undefined,
+ responseMeta,
+ );
+ if (aborted) {
emitSessionsChanged(context, {
sessionKey: canonicalKey,
...(canonicalKey === "global" && abortAgentId ? { agentId: abortAgentId } : {}),
diff --git a/src/gateway/server-methods/sessions.abort-agent-scope.test.ts b/src/gateway/server-methods/sessions.abort-agent-scope.test.ts
index b5cb425a12ae..10ebd9fb2683 100644
--- a/src/gateway/server-methods/sessions.abort-agent-scope.test.ts
+++ b/src/gateway/server-methods/sessions.abort-agent-scope.test.ts
@@ -4,16 +4,20 @@
import { expectDefined } from "@openclaw/normalization-core";
import { beforeEach, describe, expect, it, vi } from "vitest";
+import { createReplyOperation } from "../../auto-reply/reply/reply-run-registry.js";
import type { GatewayClient, GatewayRequestContext, RespondFn } from "./types.js";
const chatAbortMock = vi.fn();
const resolveSessionKeyForRunMock = vi.fn();
const listSessionsFromStoreAsyncMock = vi.fn();
const loadCombinedSessionStoreForGatewayMock = vi.fn();
-const isEmbeddedAgentRunActiveMock = vi.fn();
+const isEmbeddedAgentRunInProgressMock = vi.fn();
+const abortEmbeddedAgentRunMock = vi.fn();
+const clearSessionQueuesMock = vi.fn();
const loadSessionEntryMock = vi.fn((sessionKey: string, _opts?: { agentId?: string }) => ({
canonicalKey: sessionKey,
}));
+const loadGatewaySessionRowMock = vi.fn();
vi.mock("../server-session-key.js", () => ({
resolveSessionKeyForRun: (...args: unknown[]) => resolveSessionKeyForRunMock(...args),
@@ -25,6 +29,10 @@ vi.mock("./chat.js", () => ({
},
}));
+vi.mock("./chat-abort-handler.js", () => ({
+ handleChatAbortRequestWithLifecycle: (...args: unknown[]) => chatAbortMock(...args),
+}));
+
vi.mock("../worker-environments/session-target.js", () => ({
resolveWorkerSessionTarget: () => ({
agentId: "work",
@@ -43,6 +51,7 @@ vi.mock("../session-utils.js", async () => {
loadSessionEntryMock(...(args as [string, { agentId?: string }?])),
loadSessionEntryReadOnly: (...args: unknown[]) =>
loadSessionEntryMock(...(args as [string, { agentId?: string }?])),
+ loadGatewaySessionRow: (...args: unknown[]) => loadGatewaySessionRowMock(...args),
};
});
@@ -52,10 +61,18 @@ vi.mock("../../agents/embedded-agent-runner/runs.js", async () => {
);
return {
...actual,
- isEmbeddedAgentRunActive: (...args: unknown[]) => isEmbeddedAgentRunActiveMock(...args),
+ abortEmbeddedAgentRun: (sessionId: string) => {
+ abortEmbeddedAgentRunMock(sessionId);
+ return actual.abortEmbeddedAgentRun(sessionId);
+ },
+ isEmbeddedAgentRunInProgress: (...args: unknown[]) => isEmbeddedAgentRunInProgressMock(...args),
};
});
+vi.mock("../../auto-reply/reply/queue/cleanup.js", () => ({
+ clearSessionQueues: (...args: unknown[]) => clearSessionQueuesMock(...args),
+}));
+
import { sessionsHandlers } from "./sessions.js";
function createActiveRun(sessionKey: string, params: { agentId?: string } = {}) {
@@ -150,7 +167,15 @@ function expectRespondErrorMessage(respond: RespondFn, message: string): void {
}
function mockChatSuccess(mock: typeof chatAbortMock, payload: Record): void {
- mock.mockImplementationOnce(({ respond }: { respond: RespondFn }) => respond(true, payload));
+ mock.mockImplementationOnce(
+ (
+ { respond }: { respond: RespondFn },
+ lifecycle?: { onAuthorizedAfterQueuedAbort?: () => boolean },
+ ) => {
+ const additionalAborted = lifecycle?.onAuthorizedAfterQueuedAbort?.() ?? false;
+ respond(true, additionalAborted ? { ...payload, aborted: true } : payload);
+ },
+ );
}
function expectSessionsListActiveRun(respond: RespondFn, hasActiveRun: boolean): void {
@@ -199,8 +224,13 @@ describe("sessions.abort agent scope", () => {
store: {},
});
loadSessionEntryMock.mockClear();
- isEmbeddedAgentRunActiveMock.mockReset();
- isEmbeddedAgentRunActiveMock.mockReturnValue(false);
+ loadGatewaySessionRowMock.mockReset();
+ loadGatewaySessionRowMock.mockReturnValue(null);
+ isEmbeddedAgentRunInProgressMock.mockReset();
+ isEmbeddedAgentRunInProgressMock.mockReturnValue(false);
+ abortEmbeddedAgentRunMock.mockReset();
+ clearSessionQueuesMock.mockReset();
+ clearSessionQueuesMock.mockReturnValue({ followupCleared: 0, laneCleared: 0, keys: [] });
});
it("does not abort an active run whose session key belongs to another requested agent", async () => {
@@ -230,7 +260,7 @@ describe("sessions.abort agent scope", () => {
listSessionsFromStoreAsyncMock.mockResolvedValue({
sessions: [{ key: "agent:main:openclaw-weixin:direct:user", sessionId: "sess-weixin" }],
});
- isEmbeddedAgentRunActiveMock.mockImplementation(
+ isEmbeddedAgentRunInProgressMock.mockImplementation(
(sessionId: string) => sessionId === "sess-weixin",
);
@@ -240,7 +270,7 @@ describe("sessions.abort agent scope", () => {
{ context, reqId: "req-channel-active" },
);
- expect(isEmbeddedAgentRunActiveMock).toHaveBeenCalledWith("sess-weixin");
+ expect(isEmbeddedAgentRunInProgressMock).toHaveBeenCalledWith("sess-weixin");
expect(respond).toHaveBeenCalledWith(
true,
expect.objectContaining({
@@ -354,6 +384,287 @@ describe("sessions.abort agent scope", () => {
);
});
+ it("reports reply-only aborts as aborted without a fabricated run id", async () => {
+ const broadcastToConnIds = vi.fn();
+ const weixinOperation = createReplyOperation({
+ sessionKey: "agent:main:openclaw-weixin:direct:wechat-user",
+ sessionId: "weixin-session",
+ resetTriggered: false,
+ });
+ const telegramOperation = createReplyOperation({
+ sessionKey: "agent:main:telegram:direct:telegram-user",
+ sessionId: "telegram-session",
+ resetTriggered: false,
+ });
+ mockChatSuccess(chatAbortMock, { ok: true, aborted: false, runIds: [] });
+ loadSessionEntryMock.mockImplementationOnce((sessionKey: string) => ({
+ canonicalKey: sessionKey,
+ entry: { sessionId: "weixin-session" },
+ }));
+ loadGatewaySessionRowMock.mockReturnValue({
+ key: "agent:main:openclaw-weixin:direct:wechat-user",
+ kind: "direct",
+ sessionId: "weixin-session",
+ updatedAt: null,
+ });
+ const context = createContext({
+ extra: {
+ getSessionEventSubscriberConnIds: () => new Set(["conn-1"]),
+ broadcastToConnIds,
+ dedupe: new Map(),
+ },
+ });
+
+ try {
+ const respond = await callSessions(
+ "sessions.abort",
+ { key: "agent:main:openclaw-weixin:direct:wechat-user" },
+ { context, reqId: "req-reply-only-abort" },
+ );
+
+ expect(respond).toHaveBeenCalledWith(
+ true,
+ { ok: true, abortedRunId: null, status: "aborted" },
+ undefined,
+ undefined,
+ );
+ expect(clearSessionQueuesMock).not.toHaveBeenCalled();
+ expect(abortEmbeddedAgentRunMock).toHaveBeenCalledWith("weixin-session");
+ expect(weixinOperation.abortSignal.aborted).toBe(true);
+ expect(telegramOperation.abortSignal.aborted).toBe(false);
+ expect(broadcastToConnIds).toHaveBeenCalledWith(
+ "sessions.changed",
+ expect.objectContaining({
+ hasActiveRun: false,
+ sessionKey: "agent:main:openclaw-weixin:direct:wechat-user",
+ reason: "abort",
+ }),
+ new Set(["conn-1"]),
+ {
+ dropIfSlow: true,
+ sessionKeys: ["agent:main:openclaw-weixin:direct:wechat-user"],
+ },
+ );
+ } finally {
+ weixinOperation.complete();
+ telegramOperation.complete();
+ }
+ });
+
+ it("preserves queued work while also aborting the exact active reply run", async () => {
+ const weixinOperation = createReplyOperation({
+ sessionKey: "agent:main:openclaw-weixin:direct:wechat-user",
+ sessionId: "weixin-session",
+ resetTriggered: false,
+ });
+ const telegramOperation = createReplyOperation({
+ sessionKey: "agent:main:telegram:direct:telegram-user",
+ sessionId: "telegram-session",
+ resetTriggered: false,
+ });
+ mockChatSuccess(chatAbortMock, { ok: true, aborted: true, runIds: ["visible-run"] });
+ loadSessionEntryMock.mockImplementationOnce((sessionKey: string) => ({
+ canonicalKey: sessionKey,
+ entry: { sessionId: "weixin-session" },
+ }));
+ const context = createContext({
+ extra: {
+ dedupe: new Map(),
+ getSessionEventSubscriberConnIds: () => new Set(),
+ },
+ });
+
+ try {
+ const respond = await callSessions(
+ "sessions.abort",
+ { key: "agent:main:openclaw-weixin:direct:wechat-user" },
+ { context, reqId: "req-visible-and-reply-abort" },
+ );
+
+ expect(clearSessionQueuesMock).not.toHaveBeenCalled();
+ expect(abortEmbeddedAgentRunMock).toHaveBeenCalledWith("weixin-session");
+ expect(weixinOperation.abortSignal.aborted).toBe(true);
+ expect(telegramOperation.abortSignal.aborted).toBe(false);
+ expect(respond).toHaveBeenCalledWith(
+ true,
+ { ok: true, abortedRunId: "visible-run", status: "aborted" },
+ undefined,
+ undefined,
+ );
+ } finally {
+ weixinOperation.complete();
+ telegramOperation.complete();
+ }
+ });
+
+ it("clears queued session work even when no embedded run remains active", async () => {
+ mockChatSuccess(chatAbortMock, { ok: true, aborted: false, runIds: [] });
+ loadSessionEntryMock.mockImplementationOnce((sessionKey: string) => ({
+ canonicalKey: sessionKey,
+ entry: { sessionId: "queued-session" },
+ }));
+ clearSessionQueuesMock.mockReturnValueOnce({
+ followupCleared: 1,
+ laneCleared: 0,
+ keys: ["queued-session"],
+ });
+ const context = createContext({
+ extra: {
+ getSessionEventSubscriberConnIds: () => new Set(),
+ },
+ });
+
+ const respond = await callSessions(
+ "sessions.abort",
+ {
+ key: "agent:main:openclaw-weixin:direct:queued-user",
+ clearQueued: true,
+ },
+ { context, reqId: "req-queued-only-abort" },
+ );
+
+ expect(clearSessionQueuesMock).toHaveBeenCalledWith([
+ "agent:main:openclaw-weixin:direct:queued-user",
+ "agent:main:openclaw-weixin:direct:queued-user",
+ "queued-session",
+ ]);
+ expect(abortEmbeddedAgentRunMock).toHaveBeenCalledWith("queued-session");
+ expect(respond).toHaveBeenCalledWith(
+ true,
+ { ok: true, abortedRunId: null, status: "aborted" },
+ undefined,
+ undefined,
+ );
+ });
+
+ it("clears key-addressed queues without requiring a persisted session id", async () => {
+ const sessionKey = "agent:main:openclaw-weixin:direct:queued-without-entry";
+ mockChatSuccess(chatAbortMock, { ok: true, aborted: false, runIds: [] });
+ loadSessionEntryMock.mockImplementationOnce(() => ({ canonicalKey: sessionKey }));
+ clearSessionQueuesMock.mockReturnValueOnce({
+ followupCleared: 1,
+ laneCleared: 0,
+ keys: [sessionKey],
+ });
+ const context = createContext({
+ extra: {
+ getSessionEventSubscriberConnIds: () => new Set(),
+ },
+ });
+
+ const respond = await callSessions(
+ "sessions.abort",
+ { key: sessionKey, clearQueued: true },
+ { context, reqId: "req-key-only-queue-abort" },
+ );
+
+ expect(clearSessionQueuesMock).toHaveBeenCalledWith([sessionKey, sessionKey]);
+ expect(abortEmbeddedAgentRunMock).not.toHaveBeenCalled();
+ expect(respond).toHaveBeenCalledWith(
+ true,
+ { ok: true, abortedRunId: null, status: "aborted" },
+ undefined,
+ undefined,
+ );
+ });
+
+ it("keeps explicit runId aborts targeted instead of clearing the whole session", async () => {
+ mockChatSuccess(chatAbortMock, { ok: true, aborted: false, runIds: [] });
+ loadSessionEntryMock.mockImplementationOnce((sessionKey: string) => ({
+ canonicalKey: sessionKey,
+ entry: { sessionId: "persisted-session" },
+ }));
+ const context = createContext();
+
+ const respond = await callSessions(
+ "sessions.abort",
+ {
+ key: "agent:main:openclaw-weixin:direct:wechat-user",
+ runId: "missing-run",
+ },
+ { context, reqId: "req-targeted-run-abort" },
+ );
+
+ expect(clearSessionQueuesMock).not.toHaveBeenCalled();
+ expect(abortEmbeddedAgentRunMock).not.toHaveBeenCalled();
+ expect(respond).toHaveBeenCalledWith(
+ true,
+ { ok: true, abortedRunId: null, status: "no-active-run" },
+ undefined,
+ undefined,
+ );
+ });
+
+ it("clears legacy aliases only when they belong to the selected agent", async () => {
+ loadSessionEntryMock.mockImplementationOnce((sessionKey: string) => ({
+ canonicalKey: sessionKey,
+ entry: { sessionId: "work-session" },
+ }));
+ mockChatSuccess(chatAbortMock, { ok: true, aborted: false, runIds: [] });
+
+ await callSessions(
+ "sessions.abort",
+ { key: "main", agentId: "work", clearQueued: true },
+ {
+ context: createContext({
+ agents: [{ id: "work", default: true }, { id: "main" }],
+ }),
+ reqId: "req-owned-legacy-alias-abort",
+ },
+ );
+
+ expect(clearSessionQueuesMock).toHaveBeenLastCalledWith([
+ "agent:work:main",
+ "main",
+ "agent:work:main",
+ "work-session",
+ ]);
+
+ clearSessionQueuesMock.mockClear();
+ loadSessionEntryMock.mockImplementationOnce((sessionKey: string) => ({
+ canonicalKey: sessionKey,
+ entry: { sessionId: "work-session" },
+ }));
+ mockChatSuccess(chatAbortMock, { ok: true, aborted: false, runIds: [] });
+
+ await callSessions(
+ "sessions.abort",
+ { key: "main", agentId: "work", clearQueued: true },
+ { context: createContext(), reqId: "req-foreign-legacy-alias-abort" },
+ );
+
+ expect(clearSessionQueuesMock).toHaveBeenLastCalledWith([
+ "agent:work:main",
+ "agent:work:main",
+ "work-session",
+ ]);
+ });
+
+ it("leaves global-scope cleanup on chat.abort without an agent-qualified queue key", async () => {
+ mockChatSuccess(chatAbortMock, { ok: true, aborted: false, runIds: [] });
+ loadSessionEntryMock.mockImplementationOnce(() => ({
+ canonicalKey: "global",
+ entry: { sessionId: "work-global-session" },
+ }));
+ const context = createContext({ globalScope: true });
+
+ const respond = await callSessions(
+ "sessions.abort",
+ { key: "global", agentId: "work", clearQueued: true },
+ { context, reqId: "req-scoped-global-queue-abort" },
+ );
+
+ expectChatAbortParams({ sessionKey: "global", runId: undefined, agentId: "work" });
+ expect(clearSessionQueuesMock).not.toHaveBeenCalled();
+ expect(abortEmbeddedAgentRunMock).not.toHaveBeenCalled();
+ expect(respond).toHaveBeenCalledWith(
+ true,
+ { ok: true, abortedRunId: null, status: "no-active-run" },
+ undefined,
+ undefined,
+ );
+ });
+
it("forwards selected-agent scope for key-based global aborts", async () => {
const context = createContext({ globalScope: true });
diff --git a/src/gateway/server-session-events.test.ts b/src/gateway/server-session-events.test.ts
index bce109e78e17..7ef3bc9b6f2d 100644
--- a/src/gateway/server-session-events.test.ts
+++ b/src/gateway/server-session-events.test.ts
@@ -13,7 +13,7 @@ const sessionRow = vi.hoisted(() => ({
thinkingDefault: "medium",
agentRuntime: { id: "openclaw", source: "model" },
}));
-const isEmbeddedAgentRunActiveMock = vi.hoisted(() => vi.fn());
+const isEmbeddedAgentRunInProgressMock = vi.hoisted(() => vi.fn());
vi.mock("../config/io.js", () => ({ getRuntimeConfig: () => ({}) }));
vi.mock("./chat-display-projection.js", () => ({
@@ -31,7 +31,7 @@ vi.mock("../agents/embedded-agent-runner/runs.js", async () => {
);
return {
...actual,
- isEmbeddedAgentRunActive: (...args: unknown[]) => isEmbeddedAgentRunActiveMock(...args),
+ isEmbeddedAgentRunInProgress: (...args: unknown[]) => isEmbeddedAgentRunInProgressMock(...args),
};
});
@@ -79,7 +79,7 @@ async function emitAssistantTranscriptUpdate(
describe("createTranscriptUpdateBroadcastHandler", () => {
beforeEach(() => {
vi.clearAllMocks();
- isEmbeddedAgentRunActiveMock.mockReturnValue(false);
+ isEmbeddedAgentRunInProgressMock.mockReturnValue(false);
sessionRow.thinkingLevel = "ultra";
});
@@ -128,14 +128,14 @@ describe("createTranscriptUpdateBroadcastHandler", () => {
});
it("keeps transcript snapshots active for embedded or channel reply runs", async () => {
- isEmbeddedAgentRunActiveMock.mockImplementation((sessionId) => sessionId === "sess-main");
+ isEmbeddedAgentRunInProgressMock.mockImplementation((sessionId) => sessionId === "sess-main");
await expect(emitAssistantTranscriptUpdate(false)).resolves.toMatchObject({
sessionKey: "agent:main:main",
hasActiveRun: true,
session: { key: "agent:main:main", sessionId: "sess-main", hasActiveRun: true },
});
- expect(isEmbeddedAgentRunActiveMock).toHaveBeenCalledWith("sess-main");
+ expect(isEmbeddedAgentRunInProgressMock).toHaveBeenCalledWith("sess-main");
});
it("broadcasts user idempotency keys in session.message metadata", async () => {
diff --git a/src/gateway/test-helpers.mocks.ts b/src/gateway/test-helpers.mocks.ts
index cebe0fe770b8..9efc0128aa5c 100644
--- a/src/gateway/test-helpers.mocks.ts
+++ b/src/gateway/test-helpers.mocks.ts
@@ -22,6 +22,7 @@ function createEmbeddedRunMockExports() {
compactEmbeddedAgentSession: (...args: unknown[]) =>
embeddedRunMock.compactEmbeddedAgentSession(...args),
isEmbeddedAgentRunActive: (sessionId: string) => embeddedRunMock.activeIds.has(sessionId),
+ isEmbeddedAgentRunInProgress: (sessionId: string) => embeddedRunMock.activeIds.has(sessionId),
abortEmbeddedAgentRun: (sessionId: string) => {
embeddedRunMock.abortCalls.push(sessionId);
return embeddedRunMock.activeIds.has(sessionId);
diff --git a/ui/src/e2e/chat-flow.e2e.test.ts b/ui/src/e2e/chat-flow.e2e.test.ts
index 4a87f3f9188f..e9331f0d4464 100644
--- a/ui/src/e2e/chat-flow.e2e.test.ts
+++ b/ui/src/e2e/chat-flow.e2e.test.ts
@@ -31,6 +31,12 @@ const managedImageCacheProofDir = path.join(
"control-ui-e2e",
"managed-image-cache",
);
+const channelStopProofDir = path.join(
+ process.cwd(),
+ ".artifacts",
+ "control-ui-e2e",
+ "channel-stop",
+);
let server: ControlUiE2eServer;
// Browser contexts preserve test isolation; keep one process warm for this file.
@@ -655,6 +661,87 @@ describeControlUiE2e("Control UI mocked Gateway E2E", () => {
}
});
+ it("sends /stop to the exact selected channel session and clears its working indicator", async () => {
+ const context = await newBrowserContext({
+ locale: "en-US",
+ serviceWorkers: "block",
+ viewport: { height: 900, width: 1280 },
+ });
+ const page = await context.newPage();
+ const channelSessionKey = "agent:main:openclaw-weixin:direct:wechat-user";
+ const gateway = await installMockGateway(page, {
+ sessionKey: channelSessionKey,
+ methodResponses: {
+ "sessions.abort": { abortedRunId: null, ok: true, status: "aborted" },
+ "sessions.list": chatSessionListResponse([
+ {
+ hasActiveRun: true,
+ key: channelSessionKey,
+ kind: "direct",
+ label: "WeChat user",
+ status: "running",
+ updatedAt: Date.now(),
+ },
+ ]),
+ },
+ });
+
+ try {
+ await page.goto(`${server.baseUrl}chat`);
+ const composer = page.locator(".agent-chat__composer-combobox textarea");
+ await composer.waitFor({ state: "visible", timeout: 10_000 });
+ await gateway.waitForRequest("sessions.list");
+ const workingIndicator = page.locator(".chat-working-indicator");
+ await workingIndicator.waitFor({ state: "visible", timeout: 10_000 });
+
+ await composer.fill("/stop");
+ await page.getByRole("option", { name: /\/stop/ }).waitFor();
+ await composer.press("Enter");
+
+ const abortRequest = await gateway.waitForRequest("sessions.abort");
+ expect(requireRecord(abortRequest.params)).toEqual({
+ key: channelSessionKey,
+ clearQueued: true,
+ });
+ await gateway.setMethodResponse(
+ "sessions.list",
+ chatSessionListResponse([
+ {
+ activeRunIds: [],
+ hasActiveRun: false,
+ key: channelSessionKey,
+ kind: "direct",
+ label: "WeChat user",
+ status: "running",
+ updatedAt: Date.now(),
+ },
+ ]),
+ );
+ await gateway.emitGatewayEvent("sessions.changed", {
+ activeRunIds: [],
+ hasActiveRun: false,
+ reason: "abort",
+ sessionKey: channelSessionKey,
+ status: "running",
+ updatedAt: Date.now(),
+ });
+ await workingIndicator.waitFor({ state: "detached", timeout: 10_000 });
+ await expectRequestCountStable(gateway, "chat.abort", 0);
+ await expectRequestCountStable(gateway, "chat.send", 0);
+ await expect.poll(() => page.getByRole("listbox").count()).toBe(0);
+ expect(await composer.inputValue()).toBe("");
+ if (captureUiProofEnabled) {
+ await mkdir(channelStopProofDir, { recursive: true });
+ await page.screenshot({
+ path: path.join(channelStopProofDir, "stopped.png"),
+ fullPage: true,
+ });
+ }
+ } finally {
+ await closeBrowserContext(context);
+ }
+ });
+
it("persists the chat send shortcut and keeps multiline and IME input safe", async () => {
const context = await newBrowserContext({
locale: "en-US",
diff --git a/ui/src/pages/chat/chat-pane-lifecycle.test.ts b/ui/src/pages/chat/chat-pane-lifecycle.test.ts
index 4b08c9db2c8c..c3096b1508e9 100644
--- a/ui/src/pages/chat/chat-pane-lifecycle.test.ts
+++ b/ui/src/pages/chat/chat-pane-lifecycle.test.ts
@@ -10,8 +10,10 @@ import type {
} from "../../../../packages/gateway-protocol/src/index.js";
import { GatewayRequestError, type GatewayBrowserClient } from "../../api/gateway.ts";
import type { GatewaySessionRow } from "../../api/types.ts";
+import type { ApplicationContext } from "../../app/context.ts";
import type { SessionCapability } from "../../lib/sessions/index.ts";
import { createTestChatPane } from "./chat-pane.test-support.ts";
+import type { ChatPageHost } from "./chat-state.ts";
import {
dismissConfirmedActionPopovers,
openChatRewindConfirmation,
@@ -697,3 +699,52 @@ describe("chat pane presentation teardown", () => {
}
});
});
+
+describe("chat pane connection lifecycle", () => {
+ it("replays a pending exact-run stop when the gateway reconnects", async () => {
+ const request = vi.fn((method: string) =>
+ method === "chat.abort" ? Promise.resolve({ aborted: true }) : new Promise(() => {}),
+ );
+ const client = { request } as unknown as GatewayBrowserClient;
+ const { pane, state } = createTestChatPane({ client, sessions: {} as SessionCapability });
+ const sessionKey = "agent:main";
+ pane.context = {
+ ...pane.context,
+ config: {
+ current: {
+ assistantIdentity: { name: "Assistant" },
+ terminalEnabled: false,
+ },
+ },
+ } as unknown as ApplicationContext;
+ state.loadAssistantIdentity = vi.fn(async () => {});
+ state.realtimeTalkInputLevel = {
+ set: vi.fn(),
+ } as unknown as ChatPageHost["realtimeTalkInputLevel"];
+ state.resetToolStream = vi.fn();
+ const snapshot = {
+ ...pane.context.gateway.snapshot,
+ client,
+ assistantAgentId: "main",
+ };
+
+ pane.applyGatewaySnapshot({ ...snapshot, phase: "reconnecting", hello: null });
+ state.pendingAbort = { sourceClient: client, runId: "run-main", sessionKey };
+
+ pane.applyGatewaySnapshot({
+ ...snapshot,
+ phase: "connected",
+ });
+
+ await vi.waitFor(() =>
+ expect(request).toHaveBeenCalledWith("chat.abort", {
+ sessionKey,
+ runId: "run-main",
+ }),
+ );
+ expect(state.pendingAbort).toBeNull();
+
+ pane.applyGatewaySnapshot({ ...snapshot, phase: "connected" });
+ expect(request.mock.calls.filter(([method]) => method === "chat.abort")).toHaveLength(1);
+ });
+});
diff --git a/ui/src/pages/chat/chat-pane.ts b/ui/src/pages/chat/chat-pane.ts
index 9a84aa73b58b..0e664f9c1fca 100644
--- a/ui/src/pages/chat/chat-pane.ts
+++ b/ui/src/pages/chat/chat-pane.ts
@@ -263,6 +263,7 @@ import { admitInitialUserMessageHandoff } from "./initial-turn-handoff.ts";
import {
hasAbortableSessionRun,
reconcileStaleChatRunAfterSessionStatePublication,
+ replayPendingChatAbort,
} from "./run-lifecycle.ts";
import { scheduleChatScroll } from "./scroll.ts";
import {
@@ -3267,6 +3268,9 @@ class ChatPane extends OpenClawLightDomElement {
state.client = snapshot.client;
state.connected = snapshot.phase === "connected";
state.connectionEpoch = this.connectionGeneration;
+ if (state.connected && state.pendingAbort) {
+ void replayPendingChatAbort(state).finally(() => state.requestUpdate?.());
+ }
state.hello = snapshot.hello;
if (sourceChanged && state.sidebarContent?.kind === "session-discussion") {
// A reconnect may point at a different gateway/provider; an open panel
diff --git a/ui/src/pages/chat/chat-send.test.ts b/ui/src/pages/chat/chat-send.test.ts
index 66fccffa07c8..962544e188bc 100644
--- a/ui/src/pages/chat/chat-send.test.ts
+++ b/ui/src/pages/chat/chat-send.test.ts
@@ -7960,6 +7960,87 @@ describe("handleAbortChat", () => {
expect(host.chatRunId).toBe("run-main");
});
+ it("aborts the exact selected session when no browser run id exists", async () => {
+ const request = vi.fn(async () => ({ abortedRunId: null, status: "aborted" }));
+ const sessionKey = "agent:main:openclaw-weixin:direct:wechat-user";
+ const host = makeHost({
+ client: { request } as unknown as ChatHost["client"],
+ chatRunId: null,
+ chatMessage: "/stop",
+ sessionKey,
+ sessionsResult: createSessionsResult([
+ row(sessionKey, { hasActiveRun: true, status: "running" }),
+ ]),
+ });
+
+ await handleAbortChat(host);
+
+ expect(request).toHaveBeenCalledWith("sessions.abort", {
+ key: sessionKey,
+ clearQueued: true,
+ });
+ expect(request).not.toHaveBeenCalledWith("chat.abort", expect.anything());
+ expect(host.chatMessage).toBe("");
+ });
+
+ it("keeps selected global aborts on the compatible key-only request", async () => {
+ const request = vi.fn(async () => ({ abortedRunId: null, status: "aborted" }));
+ const host = makeHost({
+ client: { request } as unknown as ChatHost["client"],
+ chatRunId: null,
+ chatMessage: "/stop",
+ sessionKey: "global",
+ assistantAgentId: "work",
+ agentsList: { defaultId: "main" },
+ sessionsResult: createSessionsResult([
+ row("global", { hasActiveRun: true, agentId: "work" } as Partial),
+ ]),
+ });
+
+ await handleAbortChat(host);
+
+ expect(request).toHaveBeenCalledWith("sessions.abort", {
+ key: "global",
+ agentId: "work",
+ });
+ });
+
+ it.each([
+ {
+ name: "clears queues for a per-sender agent main session",
+ scope: "per-sender",
+ expected: {
+ key: "agent:work:main",
+ agentId: "work",
+ clearQueued: true,
+ },
+ },
+ {
+ name: "keeps a global-scope agent main alias on the compatible request",
+ scope: "global",
+ expected: {
+ key: "agent:work:main",
+ agentId: "work",
+ },
+ },
+ ])("$name", async ({ scope, expected }) => {
+ const request = vi.fn(async () => ({ abortedRunId: null, status: "aborted" }));
+ const sessionKey = "agent:work:main";
+ const host = makeHost({
+ client: { request } as unknown as ChatHost["client"],
+ chatRunId: null,
+ sessionKey,
+ agentsList: { defaultId: "main", mainKey: "main", scope },
+ sessionsResult: createSessionsResult([
+ row(sessionKey, { hasActiveRun: true, status: "running" }),
+ ]),
+ });
+
+ await handleAbortChat(host);
+
+ expect(request).toHaveBeenCalledWith("sessions.abort", expected);
+ });
+
it.each(["/stop", "stop", "esc", "abort", "wait", "exit"])(
"clears the typed stop command %s after aborting the active run",
async (message) => {
@@ -7983,7 +8064,9 @@ describe("handleAbortChat", () => {
);
it("queues the active run abort while disconnected", async () => {
+ const client = { request: vi.fn() } as unknown as NonNullable;
const host = makeHost({
+ client,
connected: false,
chatRunId: "run-main",
chatMessage: "draft",
@@ -7992,13 +8075,19 @@ describe("handleAbortChat", () => {
await handleAbortChat(host);
- expect(host.pendingAbort).toEqual({ runId: "run-main", sessionKey: "agent:main" });
+ expect(host.pendingAbort).toEqual({
+ sourceClient: client,
+ runId: "run-main",
+ sessionKey: "agent:main",
+ });
expect(host.chatMessage).toBe("");
expect(host.chatRunId).toBe("run-main");
});
it("preserves the draft when queueing a toolbar abort while disconnected", async () => {
+ const client = { request: vi.fn() } as unknown as NonNullable;
const host = makeHost({
+ client,
connected: false,
chatRunId: "run-main",
chatMessage: "draft",
@@ -8007,31 +8096,43 @@ describe("handleAbortChat", () => {
await handleAbortChat(host, { preserveDraft: true });
- expect(host.pendingAbort).toEqual({ runId: "run-main", sessionKey: "agent:main" });
+ expect(host.pendingAbort).toEqual({
+ sourceClient: client,
+ runId: "run-main",
+ sessionKey: "agent:main",
+ });
expect(host.chatMessage).toBe("draft");
expect(host.chatRunId).toBe("run-main");
});
- it("queues a session-scoped abort while disconnected after active run state is recovered", async () => {
+ it("does not queue an unversioned session stop while disconnected", async () => {
+ const request = vi.fn();
+ const client = { request } as unknown as NonNullable;
+ const sessionKey = "agent:main:telegram:direct:queued-user";
const host = makeHost({
+ client,
connected: false,
chatRunId: null,
chatMessage: "draft",
- sessionKey: "agent:main",
+ sessionKey,
sessionsResult: createSessionsResult([
- row("agent:main", { hasActiveRun: true }),
+ row(sessionKey, { hasActiveRun: true }),
row("agent:other", { hasActiveRun: true }),
]),
});
await handleAbortChat(host);
- expect(host.pendingAbort).toEqual({ runId: null, sessionKey: "agent:main" });
- expect(host.chatMessage).toBe("");
+ expect(host.pendingAbort).toBeUndefined();
+ expect(host.chatMessage).toBe("draft");
+ expect(request).not.toHaveBeenCalled();
});
- it("queues selected-agent global aborts with agent scope while disconnected", async () => {
+ it("does not queue an unversioned global stop while disconnected", async () => {
+ const request = vi.fn();
+ const client = { request } as unknown as NonNullable;
const host = makeHost({
+ client,
connected: false,
chatRunId: null,
chatMessage: "draft",
@@ -8045,12 +8146,9 @@ describe("handleAbortChat", () => {
await handleAbortChat(host);
- expect(host.pendingAbort).toEqual({
- runId: null,
- sessionKey: "global",
- agentId: "work",
- });
- expect(host.chatMessage).toBe("");
+ expect(host.pendingAbort).toBeUndefined();
+ expect(host.chatMessage).toBe("draft");
+ expect(request).not.toHaveBeenCalled();
});
it.each([
diff --git a/ui/src/pages/chat/chat-state.ts b/ui/src/pages/chat/chat-state.ts
index 9271e28dd6ee..51bae385b5a8 100644
--- a/ui/src/pages/chat/chat-state.ts
+++ b/ui/src/pages/chat/chat-state.ts
@@ -130,6 +130,7 @@ import {
reconcileChatRunFromSessionRow,
reconcileChatRunLifecycle,
reconcileStaleChatRunAfterSessionStatePublication,
+ type PendingChatAbort,
} from "./run-lifecycle.ts";
import {
cancelChatScroll,
@@ -240,7 +241,7 @@ export type ChatPageHost = ChatHost &
agentsList: AgentsListResult | null;
agentsSelectedId: string | null;
refreshSessionsAfterChat: Map;
- pendingAbort: { runId?: string | null; sessionKey: string; agentId?: string } | null;
+ pendingAbort: PendingChatAbort | null;
pendingSessionMessageReloadSessionKey: string | null;
chatSubmitGuards: Map>;
chatSendTimingsByRun: Map;
diff --git a/ui/src/pages/chat/run-lifecycle.test.ts b/ui/src/pages/chat/run-lifecycle.test.ts
index b6dd6e301e37..565fe579c133 100644
--- a/ui/src/pages/chat/run-lifecycle.test.ts
+++ b/ui/src/pages/chat/run-lifecycle.test.ts
@@ -1,6 +1,7 @@
// @vitest-environment node
// Control UI tests cover run lifecycle behavior.
import { describe, expect, it, vi } from "vitest";
+import type { GatewayBrowserClient } from "../../api/gateway.ts";
import type { SessionsListResult } from "../../api/types.ts";
import { isSessionRunActive } from "../../lib/session-run-state.ts";
import {
@@ -10,6 +11,7 @@ import {
reconcileChatRunFromSessionRow,
reconcileChatRunLifecycle,
reconcileStaleChatRunAfterSessionStatePublication,
+ replayPendingChatAbort,
} from "./run-lifecycle.ts";
type ReconcileHost = Parameters[0];
@@ -39,6 +41,93 @@ describe("hasAbortableSessionRun", () => {
});
});
+type AbortHost = Parameters[0];
+
+function makeAbortHost(over: Partial = {}): AbortHost {
+ return {
+ client: null,
+ connected: true,
+ sessionKey: "agent:main",
+ chatRunId: null,
+ chatLoading: false,
+ chatMessage: "",
+ chatMessages: [],
+ chatLocalInputHistoryBySession: {},
+ chatInputHistorySessionKey: null,
+ chatInputHistoryItems: null,
+ chatInputHistoryIndex: -1,
+ chatDraftBeforeHistory: null,
+ hello: null,
+ ...over,
+ };
+}
+
+describe("replayPendingChatAbort", () => {
+ it("dispatches a queued exact browser run stop through chat.abort", async () => {
+ const request = vi.fn(async () => ({ aborted: true }));
+ const client = { request } as unknown as GatewayBrowserClient;
+ const host = makeAbortHost({
+ client,
+ pendingAbort: {
+ sourceClient: client,
+ runId: "run-main",
+ sessionKey: "global",
+ agentId: "work",
+ },
+ });
+
+ await expect(replayPendingChatAbort(host)).resolves.toBe(true);
+
+ expect(request).toHaveBeenCalledWith("chat.abort", {
+ sessionKey: "global",
+ agentId: "work",
+ runId: "run-main",
+ });
+ expect(host.pendingAbort).toBeNull();
+ });
+
+ it("consumes an ambiguously failed exact-run stop without retrying it", async () => {
+ const request = vi.fn(async () => {
+ throw new Error("gateway closed before acknowledgement");
+ });
+ const client = { request } as unknown as GatewayBrowserClient;
+ const host = makeAbortHost({
+ client,
+ pendingAbort: {
+ sourceClient: client,
+ runId: "run-main",
+ sessionKey: "agent:main:telegram:direct:queued-user",
+ },
+ });
+
+ await expect(replayPendingChatAbort(host)).resolves.toBe(false);
+
+ expect(request).toHaveBeenCalledOnce();
+ expect(host.pendingAbort).toBeNull();
+ expect(host.chatError).toBe("gateway closed before acknowledgement");
+ expect(host.lastError).toBe("gateway closed before acknowledgement");
+ });
+
+ it("discards a queued stop when the reconnect uses a replacement client", async () => {
+ const sourceClient = { request: vi.fn() } as unknown as GatewayBrowserClient;
+ const replacementRequest = vi.fn();
+ const host = makeAbortHost({
+ client: { request: replacementRequest } as unknown as GatewayBrowserClient,
+ pendingAbort: {
+ sourceClient,
+ runId: "run-main",
+ sessionKey: "agent:main:telegram:direct:queued-user",
+ },
+ });
+
+ await expect(replayPendingChatAbort(host)).resolves.toBe(false);
+
+ expect(replacementRequest).not.toHaveBeenCalled();
+ expect(host.pendingAbort).toBeNull();
+ expect(host.chatError ?? null).toBeNull();
+ });
+});
+
function makeHost(over: Partial = {}): ReconcileHost {
return {
sessionKey: "s1",
diff --git a/ui/src/pages/chat/run-lifecycle.ts b/ui/src/pages/chat/run-lifecycle.ts
index a1a3c813fec3..fd1052e8d99e 100644
--- a/ui/src/pages/chat/run-lifecycle.ts
+++ b/ui/src/pages/chat/run-lifecycle.ts
@@ -10,6 +10,9 @@ import {
} from "../../lib/sessions/index.ts";
import {
areUiSessionKeysEquivalent,
+ isUiGlobalScopeConfigured,
+ isUiGlobalSessionKey,
+ resolveUiGlobalAliasAgentId,
uiSessionRowMatchesSelectedChat,
} from "../../lib/sessions/session-key.ts";
import { normalizeLowercaseStringOrEmpty } from "../../lib/string-coerce.ts";
@@ -98,9 +101,28 @@ type ChatAbortRunState = SessionScopeHost & {
chatError?: string | null;
};
+type ChatAbortIntentBase = {
+ sourceClient: GatewayBrowserClient;
+ sessionKey: string;
+ agentId?: string;
+};
+
+export type PendingChatAbort = ChatAbortIntentBase & {
+ // Session-key-only stops can become stale and target a newer run after reconnect.
+ // Only an exact run identity is safe to replay.
+ runId: string;
+};
+
+type ChatAbortIntent =
+ | PendingChatAbort
+ | (ChatAbortIntentBase & {
+ runId: null;
+ clearQueued?: true;
+ });
+
type ChatAbortHost = ChatAbortRunState &
ChatInputHistoryState & {
- pendingAbort?: { runId?: string | null; sessionKey: string; agentId?: string } | null;
+ pendingAbort?: PendingChatAbort | null;
sessionsResult?: SessionsListResult | null;
};
@@ -140,42 +162,114 @@ export function isChatStopCommand(text: string) {
return CHAT_STOP_COMMANDS.has(normalizeLowercaseStringOrEmpty(text.trim()));
}
+function queuedSessionAbortParams(
+ host: SessionScopeHost,
+ sessionKey: string,
+): { clearQueued?: true } {
+ // Agent main aliases reach the global stream only in global scope.
+ // Per-sender main sessions own queues that a full stop must clear explicitly.
+ const isGlobalSession =
+ isUiGlobalSessionKey(sessionKey) ||
+ (isUiGlobalScopeConfigured(host) && resolveUiGlobalAliasAgentId(host, sessionKey) !== null);
+ return isGlobalSession ? {} : { clearQueued: true };
+}
+
type ChatAbortOptions = { preserveDraft?: boolean };
-async function abortChatRun(state: ChatAbortRunState): Promise {
- if (!state.client || !state.connected) {
- return false;
- }
- const runId = state.chatRunId;
+async function requestChatAbort(
+ client: GatewayBrowserClient,
+ intent: ChatAbortIntent,
+): Promise<{ ok: true } | { ok: false; error: unknown }> {
try {
- await state.client.request("chat.abort", {
- sessionKey: state.sessionKey,
- ...scopedAgentParamsForSession(state, state.sessionKey),
- ...(runId ? { runId } : {}),
- });
- return true;
+ if (intent.runId !== null) {
+ await client.request("chat.abort", {
+ sessionKey: intent.sessionKey,
+ ...(intent.agentId ? { agentId: intent.agentId } : {}),
+ runId: intent.runId,
+ });
+ } else {
+ // A channel reply can be active without a browser-local chat run ID.
+ // Session abort resolves the selected persisted session's exact run.
+ await client.request("sessions.abort", {
+ key: intent.sessionKey,
+ ...(intent.agentId ? { agentId: intent.agentId } : {}),
+ ...(intent.clearQueued ? { clearQueued: true } : {}),
+ });
+ }
+ return { ok: true };
} catch (err) {
- setChatError(state, formatConnectError(err));
- return false;
+ return { ok: false, error: err };
}
}
+function currentChatAbortIntent(
+ state: ChatAbortRunState,
+ sourceClient: GatewayBrowserClient,
+): ChatAbortIntent {
+ const runId = state.chatRunId ?? null;
+ const base = {
+ sourceClient,
+ sessionKey: state.sessionKey,
+ ...scopedAgentParamsForSession(state, state.sessionKey),
+ };
+ return runId
+ ? { ...base, runId }
+ : {
+ ...base,
+ runId: null,
+ ...queuedSessionAbortParams(state, state.sessionKey),
+ };
+}
+
+async function abortChatRun(state: ChatAbortRunState): Promise {
+ const client = state.client;
+ if (!client || !state.connected) {
+ return false;
+ }
+ const result = await requestChatAbort(client, currentChatAbortIntent(state, client));
+ if (!result.ok) {
+ setChatError(state, formatConnectError(result.error));
+ }
+ return result.ok;
+}
+
+export async function replayPendingChatAbort(host: ChatAbortHost): Promise {
+ const intent = host.pendingAbort;
+ const client = host.client;
+ if (!intent || !client || !host.connected) {
+ return false;
+ }
+ // Consume before sending so repeated connected snapshots cannot duplicate
+ // the exact-run request.
+ host.pendingAbort = null;
+ // Automatic reconnects retain the browser client. A replacement client may
+ // target another Gateway, where the same session key can name unrelated work.
+ if (intent.sourceClient !== client) {
+ return false;
+ }
+ const result = await requestChatAbort(client, intent);
+ if (result.ok) {
+ return true;
+ }
+ setChatError(host, formatConnectError(result.error));
+ return false;
+}
+
export async function handleAbortChat(host: ChatAbortHost, opts?: ChatAbortOptions) {
- const activeRunId = host.chatRunId;
- const queueAbort = !host.connected && hasAbortableSessionRun(host);
- if (!host.connected && !queueAbort) {
+ const disconnectedClient = host.connected ? null : host.client;
+ const disconnectedIntent = disconnectedClient
+ ? currentChatAbortIntent(host, disconnectedClient)
+ : null;
+ const pendingAbort = disconnectedIntent?.runId ? disconnectedIntent : null;
+ if (!host.connected && !pendingAbort) {
return;
}
if (!opts?.preserveDraft) {
host.chatMessage = "";
resetChatInputHistoryNavigation(host);
}
- if (queueAbort) {
- host.pendingAbort = {
- runId: activeRunId,
- sessionKey: host.sessionKey,
- ...scopedAgentParamsForSession(host, host.sessionKey),
- };
+ if (pendingAbort) {
+ host.pendingAbort = pendingAbort;
return;
}
await abortChatRun(host);