diff --git a/src/auto-reply/reply/agent-runner-steer-adoption.ts b/src/auto-reply/reply/agent-runner-steer-adoption.ts index 41a5d6316b98..c2fbbb840124 100644 --- a/src/auto-reply/reply/agent-runner-steer-adoption.ts +++ b/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) | 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), diff --git a/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts b/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts index 9d281af51aee..76631b5b1285 100644 --- a/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts +++ b/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 | 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();