fix(auto-reply): abort only the captured steer operation (#125763)

Accepted-unconfirmed steer cleanup re-resolved the reply operation by
session key after unbounded awaits, so a same-key successor admitted in
the window could be aborted instead of the run that actually accepted
the injection. Capture the exact reply operation and embedded-run abort
authority before queueing and abort only those, per the
delegated-run-authority invariant (correlation ids never carry abort
authority across awaits).

Regression: successor admitted mid-await is never aborted; the captured
operation is.

src/auto-reply/reply/agent-runner-steer-adoption.ts

@@ -1,5 +1,6 @@
 import { expectDefined } from "@openclaw/normalization-core";
 import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
+import { ACTIVE_EMBEDDED_RUNS } from "../../agents/embedded-agent-runner/run-state.js";
 import {
   formatEmbeddedAgentQueueFailureSummary,
   queueEmbeddedAgentMessageWithOutcomeAsync,
@@ -67,7 +68,7 @@ function resolveAcceptedSteerRunId(params: ActiveReplySteerParams): string {
 
 async function finalizeAcceptedSteer(params: {
   activeReplyOperation: ReplyOperation | undefined;
-  abortKey: string | undefined;
+  activeEmbeddedRunAbort: (() => void) | undefined;
   cleanupTyping: () => void;
   errorMessage: string | undefined;
   onAdopted: (() => void | Promise<void>) | undefined;
@@ -82,11 +83,8 @@ async function finalizeAcceptedSteer(params: {
     params.replyOperationRunState.admission = { status: "accepted", mode: "steer" };
   }
   params.activeReplyOperation?.recordActivity();
-  const abortActiveRun = () => {
-    if (params.abortKey) {
-      replyRunRegistry.abort(params.abortKey);
-    }
-  };
+  const abortActiveRun = () =>
+    params.activeReplyOperation?.abortByUser() ?? params.activeEmbeddedRunAbort?.();
   if (transcriptCommitUnconfirmed) {
     // The runtime accepted this message, but exact cancellation could not find it.
     // Preserve at-most-once delivery: abort the uncertain owner without replaying.
@@ -179,6 +177,7 @@ export async function runActiveReplySteer(params: ActiveReplySteerParams): Promi
     }
     // Channel dispatch normally stamps the route-scoped source id. Internal
     // callers can derive the same per-message identity from the prepared turn.
+    const activeEmbeddedRun = ACTIVE_EMBEDDED_RUNS.get(steerSessionId);
     const steerOutcome = await queueEmbeddedAgentMessageWithOutcomeAsync(
       steerSessionId,
       followupRun.prompt,
@@ -218,8 +217,8 @@ export async function runActiveReplySteer(params: ActiveReplySteerParams): Promi
       return "handled";
     }
     const adoptionDisposition = await finalizeAcceptedSteer({
+      activeEmbeddedRunAbort: activeEmbeddedRun ? () => activeEmbeddedRun.abort() : undefined,
       activeReplyOperation,
-      abortKey: sessionKey ?? queueKey,
       cleanupTyping: () => typing.cleanup(),
       errorMessage: steerOutcome.errorMessage,
       onAdopted: () => admitFollowupRunLifecycle(followupRun),

src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts

@@ -3,7 +3,16 @@ import { tmpdir } from "node:os";
 import { join } from "node:path";
 // E2E tests for run-reply-agent execution and generated session artifacts.
 import { createRequireRecord } from "openclaw/plugin-sdk/test-fixtures";
-import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+import {
+  afterEach,
+  beforeAll,
+  beforeEach,
+  describe,
+  expect,
+  it,
+  vi,
+  type MockInstance,
+} from "vitest";
 import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js";
 import {
   GENERIC_EXTERNAL_RUN_FAILURE_TEXT,
@@ -1005,6 +1014,53 @@ describe("runReplyAgent active steering", () => {
     active.complete();
   });
 
+  it("unconfirmed steer commit aborts only the captured operation, never a same-key successor", async () => {
+    const active = createReplyOperation({
+      sessionKey: "main",
+      sessionId: "session-a",
+      resetTriggered: false,
+    });
+    active.setPhase("running");
+    const activeAbortByUser = vi.spyOn(active, "abortByUser");
+    let successor: ReplyOperation | undefined;
+    let successorAbortByUser: MockInstance<ReplyOperation["abortByUser"]> | undefined;
+    state.queueEmbeddedAgentMessageMock.mockImplementationOnce(() => {
+      active.complete();
+      successor = createReplyOperation({
+        sessionKey: "main",
+        sessionId: "session-b",
+        resetTriggered: false,
+      });
+      successor.setPhase("running");
+      successorAbortByUser = vi.spyOn(successor, "abortByUser");
+      return {
+        queued: true,
+        sessionId: "session-a",
+        target: "embedded_run",
+        gatewayHealth: "live",
+        transcriptCommit: "unconfirmed",
+        errorMessage: "receipt unavailable",
+      };
+    });
+    const { run } = createMinimalRun({
+      isActive: true,
+      shouldSteer: true,
+      shouldFollowup: true,
+      resolvedQueueMode: "steer",
+    });
+
+    await expect(run()).resolves.toBeUndefined();
+    if (!successor || !successorAbortByUser) {
+      throw new Error("expected same-key successor operation");
+    }
+    try {
+      expect(successorAbortByUser).not.toHaveBeenCalled();
+      expect(activeAbortByUser).toHaveBeenCalledOnce();
+    } finally {
+      successor.complete();
+    }
+  });
+
   it("admits an ordinary rejected steering turn with durable recovery state", async () => {
     const { sessionEntry, sessionStore, storePath } = await makeSessionFixture();
     const onAdopted = vi.fn();