refactor(steering): gateway-owned atomic start-or-steer (#125808)

* refactor(steering): gateway-owned start-or-steer via one captured-operation lifecycle chat.send with queue mode steer now atomically captures the selected session's current direct reply operation under the writer barrier and injects into exactly that operation; with no direct owner it starts a new run instead of failing with active-leaf-changed. Client-supplied expectedRunId keeps exact-match semantics; the transcript-branch CAS (expectedLeafEntryId) now guards only non-steer sends. All three steering paths (gateway chat.send, channel queue steering, server /steer) share one begin/finalize lifecycle on the reply-run registry, including the captured-instance abort from 5a15e1a39c. Deleted: the duplicate accepted/rejected/unconfirmed policy machine in steer adoption, the raw embedded /steer runtime path, the leaf-bound injection identity and reject-before-ack compat, and the dead target tool-authority fingerprint. messageInjectionAttempted becomes a typed messageInjectionDisposition (none | accepted | rejected); rejected injections take one visible followup fallback, so non-injectable runtimes queue instead of silently dropping. Deliberate semantics: a registry-less active embedded run is no longer steered by raw session id (correlation-only authority) - the message becomes a visible followup; /steer injects under its own command authorization instead of the inbound tool-authority gate it could never satisfy. * docs(protocol): describe gateway start-or-steer contract for chat.send Targetless steer is no longer a leaf-bound compatibility path that can reject with active-leaf-changed; it targets the selected session's current state (inject into the direct run, else start a turn). expectedRunId stays an exact-run fence; expectedLeafEntryId is the non-steer transcript-branch CAS. * fix(steering): require matching tool authority for /steer injection The authorized-sender command gate is weaker than tool-authority equality. Make /steer and gateway injection present the same projected evidence as channel steering so mismatches fall back to a normal prompt under the sender’s own authority.
2026-08-19 09:01:39 -06:00 · 2026-08-18 07:33:46 -07:00
parent 774d9e6c35
commit 6515f6a255
25 changed files with 879 additions and 777 deletions
@@ -663,7 +663,7 @@ methods. Treat this as feature discovery, not a full enumeration of
      Tail responses can include an opaque `deltaCursor`. Pass it back as `cursor` to `chat.history` or `chat.startup` instead of `offset` or `messageId`. A successful catch-up returns `{ kind: "delta", messages, deltaCursor, sessionInfo }`; replay each `messages` entry through the same reducer as a live `session.message` payload. `{ kind: "reset" }` means the cursor is invalid, stale, belongs to another session, crossed a reset or compaction, or is too far behind; fetch a normal tail page. Catch-up never returns a partial page or continuation: more than 200 raw events or the 1 MB payload budget resets to a tail fetch.
    - `chat.message.get` is the additive bounded full-message reader for a single visible transcript entry. Pass `sessionKey`, optional `agentId` when session selection is agent-scoped, and a transcript `messageId` previously surfaced through `chat.history`; the gateway returns the same display-normalized projection without the lightweight history truncation cap when the stored entry is still available and not oversized.
    - `chat.toolTitles` returns short purpose titles for tool calls rendered in the Control UI (batched, max 24 items with bounded inputs). The feature is opt-in via `gateway.controlUi.toolTitles` (default off); disabled gateways answer `{ titles: {}, disabled: true }` with no model call so clients stop asking. When enabled, titles use standard utility-model routing: an explicitly configured `utilityModel` (an operator decision that, like all utility tasks, may send bounded task content to the chosen provider), else the session provider's declared small-model default so no new egress destination appears implicitly; an empty `utilityModel` disables them entirely. Titles never fall back to the primary model. Results cache in the per-agent state database keyed by tool name + input, so repeated views never re-bill the same calls.
-    - `chat.send` accepts one-turn `fastMode: "auto"` to use fast mode for model calls started before the auto cutoff, then start later retry, fallback, tool-result, or continuation calls without fast mode. The cutoff defaults to 60 seconds (`DEFAULT_FAST_MODE_AUTO_ON_SECONDS`) and can be configured per model with `agents.defaults.models["<provider>/<model>"].params.fastAutoOnSeconds`. A `chat.send` caller can pass one-turn `fastAutoOnSeconds` to override the cutoff for that request. Pass `queueMode` (`steer`, `followup`, `collect`, or `interrupt`) to override the stored queue mode for this request only; explicit Control UI steer actions use `queueMode: "steer"`. Modern clients, especially clients that persist or retry a steer, should also pass the active `expectedRunId`; the Gateway binds it to one exact run so a retry cannot reach a successor. Older targetless `queueMode: "steer"` requests remain accepted only as a leaf-bound compatibility path: they must pass the active operation's immutable `expectedLeafEntryId` (or deliberate `null` for an authoritative empty transcript), and can reject with `details.reason: "active-leaf-changed"` when the leaf, owner, freshness, or injection capability cannot be proven. Other interactive sends may pass `expectedLeafEntryId` to reject if another client switched transcript branches first.
+    - `chat.send` accepts one-turn `fastMode: "auto"` to use fast mode for model calls started before the auto cutoff, then start later retry, fallback, tool-result, or continuation calls without fast mode. The cutoff defaults to 60 seconds (`DEFAULT_FAST_MODE_AUTO_ON_SECONDS`) and can be configured per model with `agents.defaults.models["<provider>/<model>"].params.fastAutoOnSeconds`. A `chat.send` caller can pass one-turn `fastAutoOnSeconds` to override the cutoff for that request. Pass `queueMode` (`steer`, `followup`, `collect`, or `interrupt`) to override the stored queue mode for this request only; explicit Control UI steer actions use `queueMode: "steer"`. A steer send targets the selected session's current state: the Gateway atomically injects the message into that session's direct active run, or starts a new turn when the session is idle. Activity in descendant subagent sessions never makes the selected session busy for this decision. A client that captured an exact run id may additionally pass `expectedRunId` as a fence; the Gateway binds the injection to that exact run and rejects on mismatch instead of redirecting to a successor. `expectedLeafEntryId` is an independent transcript-branch compare-and-swap for non-steer interactive sends: pass the displayed branch leaf (or deliberate `null` for an authoritative empty transcript) and the send rejects with `details.reason: "active-leaf-changed"` if another client switched transcript branches first; steer sends ignore it.

  </Accordion>

@@ -164,11 +164,12 @@ export const ChatSendParamsSchema = closedObject({
  systemInputProvenance: Type.Optional(InputProvenanceSchema),
  systemProvenanceReceipt: Type.Optional(Type.String()),
  suppressCommandInterpretation: Type.Optional(Type.Boolean()),
-  // Client's believed active-branch leaf entry id. Legacy targetless steering
-  // requires this immutable fence and may reject; null means an authoritative empty transcript.
+  // Transcript-branch CAS for non-steer interactive sends: the client's displayed
+  // branch leaf (null = authoritative empty transcript). Steer sends ignore it;
+  // the Gateway steers the session's direct run or starts a turn when idle.
  expectedLeafEntryId: Type.Optional(Type.Union([NonEmptyString, Type.Null()])),
-  // Optional for wire compatibility. Modern/durable steer clients should always
-  // send this exact run precondition so a retry cannot move to a successor run.
+  // Optional exact-run fence for steer sends; on mismatch the send rejects
+  // instead of reaching a same-key successor run.
  expectedRunId: Type.Optional(NonEmptyString),
  expectedSessionRoutingContract: Type.Optional(NonEmptyString),
  idempotencyKey: NonEmptyString,
@@ -134,8 +134,8 @@ export type GetReplyOptions = {
  turnAdoptionLifecycle?: TurnAdoptionLifecycle;
  /** Shared lifecycle owner for the current user-turn transcript append. */
  userTurnTranscriptRecorder?: UserTurnTranscriptRecorder;
-  /** Gateway already attempted exact active-run injection for this turn. */
-  messageInjectionAttempted?: true;
+  /** Gateway-owned start-or-steer decision for this turn. */
+  messageInjectionDisposition?: "none" | "accepted" | "rejected";
  /** Current user turn is already durable; replay it without appending another copy. */
  suppressNextUserMessagePersistence?: boolean;
  onReplyStart?: () => Promise<void> | void;
@@ -137,6 +137,7 @@ export async function runReplyAgent(
    });
  const effectiveShouldSteer = !isHeartbeat && !effectiveResetTriggered && shouldSteer;
  const effectiveShouldFollowup = !effectiveResetTriggered && shouldFollowup;
+  const messageInjectionDisposition = opts?.messageInjectionDisposition ?? "none";
  const incomingToolAuthorityFingerprint = resolveFollowupRunToolAuthorityFingerprint(followupRun);
  const activeReplyOperation = sessionKey
    ? (replyRunRegistry.get(sessionKey) ?? providedReplyOperation)
@@ -251,11 +252,20 @@ export async function runReplyAgent(
    toolProgressDetail,
  });

+  if (messageInjectionDisposition === "accepted") {
+    if (replyOperationRunState) {
+      replyOperationRunState.admission = { status: "accepted", mode: "steer" };
+    }
+    releaseAdmissionTicket();
+    typing.cleanup();
+    return undefined;
+  }
+
  if (
    effectiveShouldSteer &&
    isActive &&
    !shouldQueueAuthorityMismatch &&
-    opts?.messageInjectionAttempted !== true
+    messageInjectionDisposition === "none"
  ) {
    replyRunState.bindQueueDispositionToRunState(followupRun, replyOperationRunState);
    await runActiveReplySteer({
@@ -1,12 +1,8 @@
 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,
-} from "../../agents/embedded-agent-runner/runs.js";
 import { isIngressAdoptionLostError } from "../../channels/message/ingress-drain.js";
 import { logVerbose } from "../../globals.js";
+import { formatErrorMessage } from "../../infra/errors.js";
 import {
  scheduleFollowupDrainAfterReplyOperationClear,
  type RunReplyAgentParams,
@@ -19,7 +15,12 @@ import {
  type FollowupRun,
 } from "./queue.js";
 import type { ReplyOperationRunState } from "./reply-operation-run-state.js";
-import { type ReplyOperation, replyRunRegistry } from "./reply-run-registry.js";
+import {
+  beginReplyMessageInjectionTarget,
+  finalizeReplyMessageInjectionAttempt,
+  type ReplyOperation,
+  replyRunRegistry,
+} from "./reply-run-registry.js";
 import { refreshReplyOperationTyping } from "./reply-run-typing.js";
 import { buildChannelSourceTurnId } from "./source-turn-id.js";
 import type { TypingSignaler } from "./typing-mode.js";
@@ -66,56 +67,6 @@ function resolveAcceptedSteerRunId(params: ActiveReplySteerParams): string {
  );
 }

-async function finalizeAcceptedSteer(params: {
-  activeReplyOperation: ReplyOperation | undefined;
-  activeEmbeddedRunAbort: (() => void) | undefined;
-  cleanupTyping: () => void;
-  errorMessage: string | undefined;
-  onAdopted: (() => void | Promise<void>) | undefined;
-  replyOperationRunState: ReplyOperationRunState | undefined;
-  steerSessionId: string;
-  transcriptCommit: "unconfirmed" | undefined;
-}): Promise<"continue" | "stop"> {
-  const transcriptCommitUnconfirmed = params.transcriptCommit === "unconfirmed";
-  if (params.replyOperationRunState) {
-    // Harness acceptance has transferred this turn to the active session.
-    // Replay after an uncertain receipt could run the same user side effects twice.
-    params.replyOperationRunState.admission = { status: "accepted", mode: "steer" };
-  }
-  params.activeReplyOperation?.recordActivity();
-  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.
-    abortActiveRun();
-    logVerbose(
-      `queue: active session ${params.steerSessionId} accepted steering without transcript confirmation; aborting active run without ingress replay (${params.errorMessage ?? "unknown receipt failure"})`,
-    );
-  }
-  const adoptionBoundary = transcriptCommitUnconfirmed ? "harness acceptance" : "transcript commit";
-  try {
-    await params.onAdopted?.();
-  } catch (error) {
-    if (isIngressAdoptionLostError(error)) {
-      abortActiveRun();
-      logVerbose(
-        `queue: active session ${params.steerSessionId} adoption lost after ${adoptionBoundary} (${error.code}); aborting steered turn without ingress replay`,
-      );
-      params.cleanupTyping();
-      return "stop";
-    }
-    logVerbose(
-      `queue: active session ${params.steerSessionId} adoption finalizer failed after ${adoptionBoundary}: ${String(error)}`,
-    );
-  }
-  if (transcriptCommitUnconfirmed) {
-    params.cleanupTyping();
-    return "stop";
-  }
-  return "continue";
-}
-
 export async function runActiveReplySteer(params: ActiveReplySteerParams): Promise<"handled"> {
  const {
    followupRun,
@@ -139,6 +90,11 @@ export async function runActiveReplySteer(params: ActiveReplySteerParams): Promi
      ? params.providedReplyOperation
      : (registeredReplyOperation ?? params.providedReplyOperation);
  const steerSessionId = activeReplyOperation?.sessionId ?? followupRun.run.sessionId;
+  // Capture exact injection authority before parking or awaiting admission.
+  // A same-key successor must never inherit this turn's steer or abort.
+  const injectionTarget = replyRunRegistry.resolveCurrentMessageInjectionTarget(
+    activeReplyOperation?.key ?? queueKey,
+  );
  const parked = parkSteerCandidate(queueKey, followupRun, resolvedQueue, runFollowup);
  if (!parked) {
    releaseAdmissionTicket();
@@ -159,6 +115,18 @@ export async function runActiveReplySteer(params: ActiveReplySteerParams): Promi
  };
  scheduleParkedFallback();
  releaseAdmissionTicket();
+  const fallback = async (reason?: string): Promise<"handled"> => {
+    parked.fallback();
+    if (replyOperationRunState) {
+      replyOperationRunState.admission = { status: "accepted", mode: "followup" };
+    }
+    if (reason) {
+      logVerbose(`queue: active session ${steerSessionId} rejected steering (${reason})`);
+    }
+    await touchActiveSessionEntry();
+    typing.cleanup();
+    return "handled";
+  };
  try {
    const admission = await parked.admit();
    if (admission === "cancelled") {
@@ -167,71 +135,69 @@ export async function runActiveReplySteer(params: ActiveReplySteerParams): Promi
      return "handled";
    }
    if (admission === "fallback") {
-      parked.fallback();
-      if (replyOperationRunState) {
-        replyOperationRunState.admission = { status: "accepted", mode: "followup" };
-      }
-      await touchActiveSessionEntry();
-      typing.cleanup();
-      return "handled";
+      return await fallback();
    }
-    // 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,
-      {
-        steeringMode: "all",
-        isInboundUserMessage: true,
-        toolAuthorityFingerprint: params.toolAuthorityFingerprint,
-        ...(params.pendingInputAuthorityFingerprint
-          ? { pendingInputAuthorityFingerprint: params.pendingInputAuthorityFingerprint }
-          : {}),
-        ...(followupRun.images?.length ? { images: followupRun.images } : {}),
-        ...(followupRun.imageOrder?.length ? { imageOrder: followupRun.imageOrder } : {}),
-        ...(followupRun.media?.length ? { media: followupRun.media } : {}),
-        waitForTranscriptCommit: true,
-        queueIdentity: resolveAcceptedSteerRunId(params),
-        abortSignal: resolveFollowupAbortSignal(followupRun),
-        onQueueAccepted: parked.accepted,
-        ...(resolvedQueue.debounceMs !== undefined ? { debounceMs: resolvedQueue.debounceMs } : {}),
-        ...(followupRun.run.sourceReplyDeliveryMode
-          ? { sourceReplyDeliveryMode: followupRun.run.sourceReplyDeliveryMode }
-          : {}),
-        taskSuggestionDeliveryMode: followupRun.run.taskSuggestionDeliveryMode,
-        ...(followupRun.userTurnTranscriptRecorder
-          ? { userTurnTranscriptRecorder: followupRun.userTurnTranscriptRecorder }
-          : {}),
-      },
-    );
-    if (!steerOutcome.queued) {
-      parked.fallback();
-      if (replyOperationRunState) {
-        replyOperationRunState.admission = { status: "accepted", mode: "followup" };
-      }
-      const summary = formatEmbeddedAgentQueueFailureSummary(steerOutcome);
-      logVerbose(`queue: active session ${steerSessionId} rejected steering injection: ${summary}`);
-      await touchActiveSessionEntry();
-      typing.cleanup();
-      return "handled";
+    if (!injectionTarget) {
+      return await fallback("no injectable reply operation");
    }
-    const adoptionDisposition = await finalizeAcceptedSteer({
-      activeEmbeddedRunAbort: activeEmbeddedRun ? () => activeEmbeddedRun.abort() : undefined,
-      activeReplyOperation,
-      cleanupTyping: () => typing.cleanup(),
-      errorMessage: steerOutcome.errorMessage,
-      onAdopted: () => admitFollowupRunLifecycle(followupRun),
-      replyOperationRunState,
-      steerSessionId,
-      transcriptCommit: steerOutcome.transcriptCommit,
+    const injectionAttempt = beginReplyMessageInjectionTarget(injectionTarget, followupRun.prompt, {
+      steeringMode: "all",
+      isInboundUserMessage: true,
+      toolAuthorityFingerprint: params.toolAuthorityFingerprint,
+      ...(params.pendingInputAuthorityFingerprint
+        ? { pendingInputAuthorityFingerprint: params.pendingInputAuthorityFingerprint }
+        : {}),
+      ...(followupRun.images?.length ? { images: followupRun.images } : {}),
+      ...(followupRun.imageOrder?.length ? { imageOrder: followupRun.imageOrder } : {}),
+      ...(followupRun.media?.length ? { media: followupRun.media } : {}),
+      waitForTranscriptCommit: true,
+      queueIdentity: resolveAcceptedSteerRunId(params),
+      abortSignal: resolveFollowupAbortSignal(followupRun),
+      onQueueAccepted: parked.accepted,
+      ...(resolvedQueue.debounceMs !== undefined ? { debounceMs: resolvedQueue.debounceMs } : {}),
+      ...(followupRun.run.sourceReplyDeliveryMode
+        ? { sourceReplyDeliveryMode: followupRun.run.sourceReplyDeliveryMode }
+        : {}),
+      taskSuggestionDeliveryMode: followupRun.run.taskSuggestionDeliveryMode,
+      ...(followupRun.userTurnTranscriptRecorder
+        ? { userTurnTranscriptRecorder: followupRun.userTurnTranscriptRecorder }
+        : {}),
    });
+    const finalization = await finalizeReplyMessageInjectionAttempt({
+      attempt: injectionAttempt,
+      target: injectionTarget,
+      inboundAudio: followupRun.currentInboundAudio === true,
+      onAccepted: () => {
+        if (replyOperationRunState) {
+          replyOperationRunState.admission = { status: "accepted", mode: "steer" };
+        }
+      },
+      onAdopted: () => admitFollowupRunLifecycle(followupRun),
+      shouldAbortOnAdoptionError: isIngressAdoptionLostError,
+    });
+    if (finalization.status === "rejected") {
+      return await fallback(finalization.outcome.reason);
+    }
    parked.consume();
-    if (adoptionDisposition === "stop") {
+    const transcriptCommitUnconfirmed =
+      finalization.outcome.result?.transcriptCommit === "unconfirmed";
+    if (finalization.aborted) {
+      if (replyOperationRunState) {
+        replyOperationRunState.messageInjectionAborted = true;
+      }
+      const reason = transcriptCommitUnconfirmed
+        ? (finalization.outcome.result?.errorMessage ?? "transcript commitment unconfirmed")
+        : `adoption lost: ${formatErrorMessage(finalization.adoptionError)}`;
+      logVerbose(
+        `queue: active session ${steerSessionId} aborted exact steered target without replay (${reason})`,
+      );
+      typing.cleanup();
      return "handled";
    }
-    if (followupRun.currentInboundAudio === true) {
-      activeReplyOperation?.markAcceptedSteeredInboundAudio();
+    if (finalization.adoptionError) {
+      logVerbose(
+        `queue: active session ${steerSessionId} adoption finalizer failed: ${formatErrorMessage(finalization.adoptionError)}`,
+      );
    }
    if (activeReplyOperation) {
      await refreshReplyOperationTyping(activeReplyOperation, {
@@ -7,7 +7,10 @@ import type { EmbeddedAgentQueueMessageOutcome } from "../../agents/embedded-age
 import type { OpenClawConfig } from "../../config/types.openclaw.js";
 import type { TemplateContext } from "../templating.js";
 import type { FollowupRun, QueueSettings } from "./queue.js";
-import { createReplyOperation as createRegisteredReplyOperation } from "./reply-run-registry.js";
+import {
+  createReplyOperation as createRegisteredReplyOperation,
+  type ReplyOperation,
+} from "./reply-run-registry.js";
 import { resolveFollowupRunToolAuthorityFingerprint } from "./reply-tool-authority.js";
 import {
  createMockFollowupRun,
@@ -53,6 +56,7 @@ const resolveOutboundAttachmentFromUrlMock = vi.fn();
 const createReplyMediaContextRuntimeMock = vi.fn();
 const EXPECTED_STEER_QUEUE_IDENTITY =
  "channel-user:v1:6f3f31084a7a2a6ff17176c0c16682e64d9f21301f64ff7e5bf1173b54fadc33";
+const registeredOperations: ReplyOperation[] = [];
 vi.mock("../../agents/model-fallback-runner.js", () => ({
  runWithModelFallback: (params: {
    provider: string;
@@ -311,12 +315,53 @@ function makeRunReplyAgentParams(
        workspaceDir,
      },
    });
-  const replyOperation = overrides.replyOperation ?? createMockReplyOperation().replyOperation;
+  const replyOperation =
+    overrides.replyOperation ??
+    (overrides.isActive === true
+      ? createRegisteredReplyOperation({
+          sessionKey: overrides.sessionKey ?? "main",
+          sessionId: followupRun.run.sessionId,
+          resetTriggered: false,
+        })
+      : createMockReplyOperation().replyOperation);
+  if (overrides.isActive === true) {
+    registeredOperations.push(replyOperation);
+    if (!overrides.replyOperation) {
+      replyOperation.setPhase("running");
+    }
+  }
  if (overrides.isActive === true && !replyOperation.toolAuthorityFingerprint) {
    replyOperation.bindToolAuthorityFingerprint(
      resolveFollowupRunToolAuthorityFingerprint(followupRun),
    );
  }
+  if (overrides.isActive === true) {
+    replyOperation.attachBackend({
+      kind: "embedded",
+      cancel: vi.fn(),
+      supportsQueueMessageImages: true,
+      taskSuggestionDeliveryMode: followupRun.run.taskSuggestionDeliveryMode,
+      messageInjection: {
+        isAvailable: () => true,
+        queueMessage: async (text, options) => {
+          const outcome = await queueEmbeddedAgentMessageWithOutcomeAsyncMock(
+            replyOperation.sessionId,
+            text,
+            options,
+          );
+          if (!outcome.queued) {
+            throw new Error(outcome.reason);
+          }
+          return outcome.transcriptCommit === "unconfirmed"
+            ? {
+                transcriptCommit: outcome.transcriptCommit,
+                errorMessage: outcome.errorMessage ?? "commit unconfirmed",
+              }
+            : undefined;
+        },
+      },
+    });
+  }

  return {
    commandBody: prompt,
@@ -414,6 +459,9 @@ describe("runReplyAgent media path normalization", () => {
  });

  afterEach(() => {
+    for (const operation of registeredOperations.splice(0)) {
+      operation.complete();
+    }
    vi.useRealTimers();
    const paths = cleanupPaths.splice(0);
    return Promise.all(paths.map((entry) => rm(entry, { recursive: true, force: true })));
@@ -480,7 +528,7 @@ describe("runReplyAgent media path normalization", () => {
        isInboundUserMessage: true,
        waitForTranscriptCommit: true,
        queueIdentity: EXPECTED_STEER_QUEUE_IDENTITY,
-        onQueueAccepted: parkedSteerAcceptedMock,
+        onQueueAccepted: expect.any(Function),
        taskSuggestionDeliveryMode: "gateway",
        toolAuthorityFingerprint: resolveFollowupRunToolAuthorityFingerprint(followupRun),
      },
@@ -527,7 +575,7 @@ describe("runReplyAgent media path normalization", () => {
        isInboundUserMessage: true,
        waitForTranscriptCommit: true,
        queueIdentity: EXPECTED_STEER_QUEUE_IDENTITY,
-        onQueueAccepted: parkedSteerAcceptedMock,
+        onQueueAccepted: expect.any(Function),
        images,
        media: followupRun.media,
        taskSuggestionDeliveryMode: undefined,
@@ -613,7 +661,7 @@ describe("runReplyAgent media path normalization", () => {
        isInboundUserMessage: true,
        waitForTranscriptCommit: true,
        queueIdentity: EXPECTED_STEER_QUEUE_IDENTITY,
-        onQueueAccepted: parkedSteerAcceptedMock,
+        onQueueAccepted: expect.any(Function),
        taskSuggestionDeliveryMode: undefined,
        toolAuthorityFingerprint: operation.toolAuthorityFingerprint,
      },
@@ -88,6 +88,7 @@ const state = vi.hoisted(() => ({
  getChannelPluginMock: vi.fn(),
  materializeMcpAppChannelPresentationMock: vi.fn(),
  queueEmbeddedAgentMessageMock: vi.fn(),
+  activeBackendCancelMock: vi.fn(),
  runEmbeddedAgentMock: vi.fn(),
 }));
 const parkedSteer = vi.hoisted(() => {
@@ -323,6 +324,7 @@ beforeEach(() => {
    meta: { agentMeta: { usage: { input: 1, output: 1 } } },
  });
  state.queueEmbeddedAgentMessageMock.mockReset();
+  state.activeBackendCancelMock.mockReset();
  state.beforeAgentReplyHasHooksMock.mockReset().mockReturnValue(false);
  state.beforeAgentReplyRunMock.mockReset();
  state.queueEmbeddedAgentMessageMock.mockReturnValue(false);
@@ -365,6 +367,7 @@ function createMinimalRun(params?: {
  sourceTurnId?: string;
  runOverrides?: Partial<FollowupRun["run"]>;
  bindActiveAuthority?: boolean;
+  attachSteerBackend?: boolean;
 }) {
  const typing = createMockTypingController();
  const opts = params?.opts;
@@ -433,6 +436,47 @@ function createMinimalRun(params?: {
    opts,
    run: async () => {
      const runReplyAgent = await getRunReplyAgent();
+      const operation = replyRunRegistry.get(sessionKey);
+      if (operation && params?.attachSteerBackend !== false) {
+        operation.attachBackend({
+          kind: "embedded",
+          cancel: state.activeBackendCancelMock,
+          claimPendingUserInputAnswer: async (prompt, options) => {
+            const result = state.queueEmbeddedAgentMessageMock(
+              operation.sessionId,
+              prompt,
+              options,
+            ) as boolean | { queued: boolean };
+            return result === true || (typeof result === "object" && result.queued);
+          },
+          messageInjection: {
+            isAvailable: () => true,
+            queueMessage: async (prompt, options) => {
+              const result = state.queueEmbeddedAgentMessageMock(
+                operation.sessionId,
+                prompt,
+                options,
+              ) as
+                | boolean
+                | {
+                    queued: boolean;
+                    reason?: string;
+                    transcriptCommit?: "unconfirmed";
+                    errorMessage?: string;
+                  };
+              if (result === false || (typeof result === "object" && !result.queued)) {
+                throw new Error(typeof result === "object" ? result.reason : "queue rejected");
+              }
+              return typeof result === "object" && result.transcriptCommit === "unconfirmed"
+                ? {
+                    transcriptCommit: result.transcriptCommit,
+                    errorMessage: result.errorMessage ?? "commit unconfirmed",
+                  }
+                : undefined;
+            },
+          },
+        });
+      }
      return runReplyAgent({
        commandBody: "hello",
        followupRun,
@@ -768,6 +812,12 @@ describe("runReplyAgent active steering", () => {
  });

  it("replays a declined steer without dispatching its hook twice", async () => {
+    const active = createReplyOperation({
+      sessionKey: "main",
+      sessionId: "session",
+      resetTriggered: false,
+    });
+    active.setPhase("running");
    state.beforeAgentReplyHasHooksMock.mockImplementation(
      (hookName) => hookName === "before_agent_reply",
    );
@@ -793,19 +843,20 @@ describe("runReplyAgent active steering", () => {
    expect(state.queueEmbeddedAgentMessageMock).toHaveBeenCalledOnce();
    expect(parkedSteer.fallback).toHaveBeenCalledOnce();
    expect(parkedSteer.consume).not.toHaveBeenCalled();
+    active.complete();
    await requireScheduledFollowupRunner()(followupRun);
    expect(state.runEmbeddedAgentMock).toHaveBeenCalledOnce();
    expect(state.beforeAgentReplyRunMock).toHaveBeenCalledOnce();
  });

-  it("runs normal reply hooks once after Gateway already attempted injection", async () => {
+  it("runs one normal fallback after Gateway rejects injection", async () => {
    state.beforeAgentReplyHasHooksMock.mockImplementation(
      (hookName) => hookName === "before_agent_reply",
    );
    state.beforeAgentReplyRunMock.mockResolvedValue(undefined);
    state.runEmbeddedAgentMock.mockImplementationOnce(runHookBackedEmbeddedAgent);
    const { run } = createMinimalRun({
-      opts: { messageInjectionAttempted: true },
+      opts: { messageInjectionDisposition: "rejected" },
      isActive: true,
      shouldSteer: true,
      resolvedQueueMode: "steer",
@@ -825,7 +876,58 @@ describe("runReplyAgent active steering", () => {
    expect(state.runEmbeddedAgentMock).toHaveBeenCalledOnce();
  });

+  it("does not steer, enqueue, or start a second run after accepted Gateway injection", async () => {
+    const runState: ReplyOperationRunState = {};
+    const { run } = createMinimalRun({
+      opts: {
+        messageInjectionDisposition: "accepted",
+        [REPLY_OPERATION_RUN_STATE]: runState,
+      },
+      isActive: true,
+      shouldSteer: true,
+      shouldFollowup: true,
+      resolvedQueueMode: "steer",
+    });
+
+    await expect(run()).resolves.toBeUndefined();
+
+    expect(runState.admission).toEqual({ status: "accepted", mode: "steer" });
+    expect(state.queueEmbeddedAgentMessageMock).not.toHaveBeenCalled();
+    expect(vi.mocked(enqueueFollowupRun)).not.toHaveBeenCalled();
+    expect(state.runEmbeddedAgentMock).not.toHaveBeenCalled();
+  });
+
+  it("falls back visibly when the active CLI backend cannot accept injection", async () => {
+    const active = createReplyOperation({
+      sessionKey: "main",
+      sessionId: "session",
+      resetTriggered: false,
+    });
+    active.setPhase("running");
+    active.attachBackend({ kind: "cli", cancel: vi.fn() });
+    state.runEmbeddedAgentMock.mockImplementationOnce(runHookBackedEmbeddedAgent);
+    const { followupRun, run } = createMinimalRun({
+      attachSteerBackend: false,
+      isActive: true,
+      shouldSteer: true,
+      shouldFollowup: true,
+      resolvedQueueMode: "steer",
+    });
+
+    await expect(run()).resolves.toBeUndefined();
+    expect(parkedSteer.fallback).toHaveBeenCalledOnce();
+    active.complete();
+    await requireScheduledFollowupRunner()(followupRun);
+    expect(state.runEmbeddedAgentMock).toHaveBeenCalledOnce();
+  });
+
  it("carries the prepared user-turn recorder into the embedded queue", async () => {
+    const active = createReplyOperation({
+      sessionKey: "main",
+      sessionId: "session",
+      resetTriggered: false,
+    });
+    active.setPhase("running");
    state.queueEmbeddedAgentMessageMock.mockReturnValueOnce(true);
    const recorder = createUserTurnTranscriptRecorder({
      input: {
@@ -851,6 +953,7 @@ describe("runReplyAgent active steering", () => {
        userTurnTranscriptRecorder: recorder,
      }),
    );
+    active.complete();
  });

  it("steers against the session's registered run owner, not a source-keyed reservation", async () => {
@@ -889,6 +992,12 @@ describe("runReplyAgent active steering", () => {
  });

  it("waits for transcript commit and keeps a rejected adoption finalizer irrevocably adopted", async () => {
+    const active = createReplyOperation({
+      sessionKey: "main",
+      sessionId: "session",
+      resetTriggered: false,
+    });
+    active.setPhase("running");
    const finalizerError = new Error("dedupe finalizer failed");
    const events: string[] = [];
    state.queueEmbeddedAgentMessageMock.mockImplementationOnce(
@@ -918,6 +1027,7 @@ describe("runReplyAgent active steering", () => {
    expect(onAdopted).toHaveBeenCalledTimes(1);
    expect(parkedSteer.consume).toHaveBeenCalledOnce();
    expect(parkedSteer.fallback).not.toHaveBeenCalled();
+    active.complete();
    expect(state.queueEmbeddedAgentMessageMock).toHaveBeenCalledTimes(1);
    expect(vi.mocked(enqueueFollowupRun)).not.toHaveBeenCalled();
    expect(state.runEmbeddedAgentMock).not.toHaveBeenCalled();
@@ -970,17 +1080,11 @@ describe("runReplyAgent active steering", () => {

  it("adopts and consumes unconfirmed steering without replay", async () => {
    const runState: ReplyOperationRunState = {};
-    const cancel = vi.fn();
    const active = createReplyOperation({
      sessionKey: "main",
      sessionId: "session",
      resetTriggered: false,
    });
-    active.attachBackend({
-      kind: "embedded",
-      cancel,
-      isStreaming: () => true,
-    });
    active.setPhase("running");
    state.queueEmbeddedAgentMessageMock.mockReturnValueOnce({
      queued: true,
@@ -1005,8 +1109,9 @@ describe("runReplyAgent active steering", () => {
    await expect(run()).resolves.toBeUndefined();

    expect(onAdopted).toHaveBeenCalledOnce();
-    expect(cancel).toHaveBeenCalledOnce();
+    expect(state.activeBackendCancelMock).toHaveBeenCalledOnce();
    expect(runState.admission).toEqual({ status: "accepted", mode: "steer" });
+    expect(runState.messageInjectionAborted).toBe(true);
    expect(parkedSteer.consume).toHaveBeenCalledOnce();
    expect(parkedSteer.fallback).not.toHaveBeenCalled();
    expect(state.runEmbeddedAgentMock).not.toHaveBeenCalled();
@@ -1,7 +0,0 @@
-// Runtime barrel for embedded-agent steering helpers used by auto-reply commands.
-export {
-  formatEmbeddedAgentQueueFailureSummary,
-  isEmbeddedAgentRunActive,
-  queueEmbeddedAgentMessageWithOutcomeAsync,
-  resolveActiveEmbeddedRunSessionId,
-} from "../../agents/embedded-agent-runner/runs.js";
@@ -1,17 +1,15 @@
-// Tests steer command persistence and retrieval for session guidance.
-import { beforeEach, describe, expect, it, vi } from "vitest";
+// Tests /steer target capture, accepted delivery, and visible fallback.
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import type { ChatType } from "../../channels/chat-type.js";
 import type { OpenClawConfig } from "../../config/types.openclaw.js";
 import { buildCommandTestParams } from "./commands.test-harness.js";
-
-const steerRuntimeMocks = vi.hoisted(() => ({
-  formatEmbeddedAgentQueueFailureSummary: vi.fn(),
-  isEmbeddedAgentRunActive: vi.fn(),
-  queueEmbeddedAgentMessageWithOutcomeAsync: vi.fn(),
-  resolveActiveEmbeddedRunSessionId: vi.fn(),
-  resolveActiveEmbeddedRunSessionIdBySessionFile: vi.fn(),
-}));
-
-vi.mock("./commands-steer.runtime.js", () => steerRuntimeMocks);
+import type { ReplyBackendQueueMessageOptions, ReplyOperation } from "./reply-run-registry.js";
+import { createReplyOperation } from "./reply-run-registry.js";
+import {
+  createFollowupRunToolAuthorityProjector,
+  resolveFollowupRunToolAuthorityFingerprint,
+} from "./reply-tool-authority.js";
+import { createMockFollowupRun } from "./test-helpers.js";

 const { handleSteerCommand } = await import("./commands-steer.js");

@@ -19,186 +17,176 @@ const baseCfg = {
  commands: { text: true },
  session: { mainKey: "main", scope: "per-sender" },
 } as OpenClawConfig;
+const queueMessage = vi.fn(
+  async (_text: string, _options?: ReplyBackendQueueMessageOptions) => undefined,
+);
+const operations: ReplyOperation[] = [];

 function buildParams(commandBody: string) {
  return buildCommandTestParams(commandBody, baseCfg);
 }

-describe("handleSteerCommand", () => {
-  beforeEach(() => {
-    steerRuntimeMocks.formatEmbeddedAgentQueueFailureSummary
-      .mockReset()
-      .mockReturnValue(
-        "queue_message_failed reason=not_streaming sessionId=session-active gatewayHealth=live",
-      );
-    steerRuntimeMocks.isEmbeddedAgentRunActive.mockReset().mockReturnValue(false);
-    steerRuntimeMocks.queueEmbeddedAgentMessageWithOutcomeAsync.mockReset().mockResolvedValue({
-      queued: true,
+function beginActiveOperation(
+  sessionKey: string,
+  sessionId = "session-active",
+  taskSuggestionDeliveryMode?: "gateway",
+  authorityRun = createMockFollowupRun({ run: { sessionId, sessionKey } }),
+) {
+  const operation = createReplyOperation({ sessionKey, sessionId, resetTriggered: false });
+  const authorityRoute = {
+    provider: authorityRun.run.provider,
+    model: authorityRun.run.model,
+  };
+  const toolAuthorityFingerprint = resolveFollowupRunToolAuthorityFingerprint(
+    authorityRun,
+    authorityRoute,
+  );
+  operation.bindToolAuthorityProjector(createFollowupRunToolAuthorityProjector(authorityRun));
+  operation.bindToolAuthorityRoute(authorityRoute);
+  operation.bindToolAuthorityFingerprint(toolAuthorityFingerprint);
+  operation.setPhase("running");
+  operation.attachBackend({
+    kind: "embedded",
+    cancel: vi.fn(),
+    taskSuggestionDeliveryMode,
+    messageInjection: { isAvailable: () => true, queueMessage },
+  });
+  operations.push(operation);
+  return { operation, toolAuthorityFingerprint };
+}
+
+function createCommandAuthorityRun(params: ReturnType<typeof buildParams>) {
+  return createMockFollowupRun({
+    originatingChannel: params.ctx.OriginatingChannel,
+    toolsAllow: params.opts?.toolsAllow,
+    disableTools: params.opts?.disableTools,
+    run: {
+      agentId: params.agentId ?? "main",
+      agentDir: params.agentDir ?? "/tmp/agent",
      sessionId: "session-active",
-      target: "embedded_run",
-      gatewayHealth: "live",
-    });
-    steerRuntimeMocks.resolveActiveEmbeddedRunSessionId.mockReset().mockReturnValue(undefined);
-    steerRuntimeMocks.resolveActiveEmbeddedRunSessionIdBySessionFile
-      .mockReset()
-      .mockReturnValue(undefined);
+      sessionKey: params.sessionKey,
+      messageProvider: params.ctx.OriginatingChannel ?? params.ctx.Provider ?? params.ctx.Surface,
+      chatType: params.ctx.ChatType as ChatType | undefined,
+      agentAccountId: params.ctx.AccountId,
+      conversationToolPolicy: params.ctx.ConversationToolPolicy,
+      groupId: undefined,
+      groupChannel: undefined,
+      groupSpace: undefined,
+      memberRoleIds: params.ctx.MemberRoleIds,
+      spawnedBy: params.sessionEntry?.spawnedBy,
+      senderId: params.ctx.SenderId,
+      senderName: params.ctx.SenderName,
+      senderUsername: params.ctx.SenderUsername,
+      senderE164: params.ctx.SenderE164,
+      senderIsOwner: params.command.senderIsOwner,
+      traceAuthorized:
+        params.command.senderIsOwner ||
+        (params.ctx.GatewayClientScopes ?? []).includes("operator.admin"),
+      approvalReviewerDeviceId: params.ctx.ApprovalReviewerDeviceId,
+      clientCaps: params.ctx.GatewayClientCaps,
+      toolBindings: params.ctx.GatewayRunToolBindings,
+      inputProvenance: params.ctx.InputProvenance,
+      workspaceDir: params.workspaceDir,
+      config: params.cfg,
+      toolOverrides: params.sessionEntry?.toolOverrides,
+      provider: params.provider,
+      model: params.model,
+    },
+  });
+}
+
+describe("handleSteerCommand", () => {
+  beforeEach(() => queueMessage.mockReset().mockResolvedValue(undefined));
+
+  afterEach(() => {
+    for (const operation of operations.splice(0)) {
+      operation.complete();
+    }
  });

-  it("queues steering for the active current text-command session", async () => {
-    steerRuntimeMocks.resolveActiveEmbeddedRunSessionId.mockReturnValue("session-active");
+  it("matching authority /steer injects into the captured operation", async () => {
+    const params = buildParams("/steer keep going");
+    params.opts = { toolsAllow: ["read"] };
+    const { toolAuthorityFingerprint } = beginActiveOperation(
+      "agent:main:main",
+      "session-active",
+      undefined,
+      createCommandAuthorityRun(params),
+    );

-    const result = await handleSteerCommand(buildParams("/steer keep going"), true);
+    const result = await handleSteerCommand(params, true);

    expect(result).toEqual({
      shouldContinue: false,
      reply: { text: "steered current session." },
    });
-    expect(steerRuntimeMocks.resolveActiveEmbeddedRunSessionId).toHaveBeenCalledWith(
+    expect(queueMessage).toHaveBeenCalledWith("keep going", {
+      steeringMode: "all",
+      isInboundUserMessage: true,
+      toolAuthorityFingerprint,
+      debounceMs: 0,
+      taskSuggestionDeliveryMode: undefined,
+      onQueueAccepted: expect.any(Function),
+    });
+  });
+
+  it("authorized sender with mismatched tool authority cannot inject via /steer", async () => {
+    const activeParams = buildParams("/steer keep going");
+    activeParams.opts = { toolsAllow: ["exec"] };
+    beginActiveOperation(
      "agent:main:main",
-    );
-    expect(steerRuntimeMocks.queueEmbeddedAgentMessageWithOutcomeAsync).toHaveBeenCalledWith(
      "session-active",
-      "keep going",
-      {
-        steeringMode: "all",
-        isInboundUserMessage: true,
-        debounceMs: 0,
-        taskSuggestionDeliveryMode: undefined,
-      },
+      undefined,
+      createCommandAuthorityRun(activeParams),
    );
+    const params = buildParams("/steer keep going");
+    params.opts = { toolsAllow: ["read"] };
+
+    const result = await handleSteerCommand(params, true);
+
+    expect(result).toEqual({ shouldContinue: true });
+    expect(params.ctx.BodyForAgent).toBe("keep going");
+    expect(params.command.commandBodyNormalized).toBe("keep going");
+    expect(queueMessage).not.toHaveBeenCalled();
  });

  it("passes the initiating surface task capability into steering", async () => {
-    steerRuntimeMocks.resolveActiveEmbeddedRunSessionId.mockReturnValue("session-active");
+    beginActiveOperation("agent:main:main", "session-active", "gateway");
    const params = buildParams("/steer keep going");
    params.opts = { taskSuggestionDeliveryMode: "gateway" };

    await handleSteerCommand(params, true);

-    expect(steerRuntimeMocks.queueEmbeddedAgentMessageWithOutcomeAsync).toHaveBeenCalledWith(
-      "session-active",
+    expect(queueMessage).toHaveBeenCalledWith(
      "keep going",
-      {
-        steeringMode: "all",
-        isInboundUserMessage: true,
-        debounceMs: 0,
-        taskSuggestionDeliveryMode: "gateway",
-      },
+      expect.objectContaining({ taskSuggestionDeliveryMode: "gateway" }),
    );
  });

-  it("prefers the native command target session key over the slash-command session", async () => {
-    steerRuntimeMocks.resolveActiveEmbeddedRunSessionId.mockReturnValue("session-target");
-
+  it("prefers the native command target over the slash-command source", async () => {
+    beginActiveOperation("agent:main:discord:direct:target", "session-target");
    const params = buildParams("/steer check the target");
    params.ctx.CommandSource = "native";
    params.ctx.CommandTargetSessionKey = "agent:main:discord:direct:target";
    params.sessionKey = "agent:main:discord:slash:user";

-    await handleSteerCommand(params, true);
+    const result = await handleSteerCommand(params, true);

-    expect(steerRuntimeMocks.resolveActiveEmbeddedRunSessionId).toHaveBeenCalledWith(
-      "agent:main:discord:direct:target",
-    );
-    expect(steerRuntimeMocks.queueEmbeddedAgentMessageWithOutcomeAsync).toHaveBeenCalledWith(
-      "session-target",
-      "check the target",
-      {
-        steeringMode: "all",
-        isInboundUserMessage: true,
-        debounceMs: 0,
-        taskSuggestionDeliveryMode: undefined,
-      },
-    );
+    expect(result).toEqual({
+      shouldContinue: false,
+      reply: { text: "steered current session." },
+    });
+    expect(queueMessage).toHaveBeenCalledWith("check the target", expect.any(Object));
  });

-  it("falls back to the stored session id when it is still active", async () => {
-    steerRuntimeMocks.isEmbeddedAgentRunActive.mockReturnValue(true);
-
-    const params = buildParams("/tell continue from state");
-    params.sessionEntry = { sessionId: "stored-session-id", updatedAt: Date.now() };
-
-    await handleSteerCommand(params, true);
-
-    expect(steerRuntimeMocks.resolveActiveEmbeddedRunSessionId).toHaveBeenCalledWith(
-      "agent:main:main",
-    );
-    expect(steerRuntimeMocks.isEmbeddedAgentRunActive).toHaveBeenCalledWith("stored-session-id");
-    expect(steerRuntimeMocks.queueEmbeddedAgentMessageWithOutcomeAsync).toHaveBeenCalledWith(
-      "stored-session-id",
-      "continue from state",
-      {
-        steeringMode: "all",
-        isInboundUserMessage: true,
-        debounceMs: 0,
-        taskSuggestionDeliveryMode: undefined,
-      },
-    );
-  });
-
-  it("resolves an active run from the target session key before stored session id fallback", async () => {
-    steerRuntimeMocks.resolveActiveEmbeddedRunSessionId.mockReturnValue("session-key-active");
-
-    const params = buildParams("/steer check the active file");
-    params.ctx.CommandSource = "native";
-    params.ctx.CommandTargetSessionKey = "agent:main:telegram:topic:5907";
-    params.sessionKey = "agent:main:telegram:control";
-    params.sessionStore = {
-      "agent:main:telegram:topic:5907": {
-        sessionId: "stored-session-id",
-        updatedAt: Date.now(),
-      },
-    };
-
-    await handleSteerCommand(params, true);
-
-    expect(steerRuntimeMocks.resolveActiveEmbeddedRunSessionId).toHaveBeenCalledWith(
-      "agent:main:telegram:topic:5907",
-    );
-    expect(steerRuntimeMocks.resolveActiveEmbeddedRunSessionIdBySessionFile).not.toHaveBeenCalled();
-    expect(steerRuntimeMocks.isEmbeddedAgentRunActive).not.toHaveBeenCalledWith(
-      "stored-session-id",
-    );
-    expect(steerRuntimeMocks.queueEmbeddedAgentMessageWithOutcomeAsync).toHaveBeenCalledWith(
-      "session-key-active",
-      "check the active file",
-      {
-        steeringMode: "all",
-        isInboundUserMessage: true,
-        debounceMs: 0,
-        taskSuggestionDeliveryMode: undefined,
-      },
-    );
-  });
-
-  it("falls back from a slash-lane command session to an active direct sibling", async () => {
-    steerRuntimeMocks.resolveActiveEmbeddedRunSessionId.mockImplementation((key: string) =>
-      key === "agent:main:telegram:direct:123" ? "session-direct-active" : undefined,
-    );
-
+  it("maps a text slash source lane to its active direct conversation", async () => {
+    beginActiveOperation("agent:main:telegram:direct:123", "session-direct-active");
    const params = buildParams("/steer use the active direct lane");
    params.sessionKey = "agent:main:telegram:slash:123";

    await handleSteerCommand(params, true);

-    expect(steerRuntimeMocks.resolveActiveEmbeddedRunSessionId).toHaveBeenNthCalledWith(
-      1,
-      "agent:main:telegram:slash:123",
-    );
-    expect(steerRuntimeMocks.resolveActiveEmbeddedRunSessionId).toHaveBeenNthCalledWith(
-      2,
-      "agent:main:telegram:direct:123",
-    );
-    expect(steerRuntimeMocks.queueEmbeddedAgentMessageWithOutcomeAsync).toHaveBeenCalledWith(
-      "session-direct-active",
-      "use the active direct lane",
-      {
-        steeringMode: "all",
-        isInboundUserMessage: true,
-        debounceMs: 0,
-        taskSuggestionDeliveryMode: undefined,
-      },
-    );
+    expect(queueMessage).toHaveBeenCalledWith("use the active direct lane", expect.any(Object));
  });

  it("returns usage for an empty steer command", async () => {
@@ -208,79 +196,29 @@ describe("handleSteerCommand", () => {
      shouldContinue: false,
      reply: { text: "Usage: /steer <message>" },
    });
-    expect(steerRuntimeMocks.queueEmbeddedAgentMessageWithOutcomeAsync).not.toHaveBeenCalled();
+    expect(queueMessage).not.toHaveBeenCalled();
  });

-  it("continues as a normal prompt when no current session run is active", async () => {
+  it("continues visibly as a normal prompt when no direct owner is active", async () => {
    const params = buildParams("/steer keep going");
    const result = await handleSteerCommand(params, true);

-    expect(result).toEqual({
-      shouldContinue: true,
-    });
+    expect(result).toEqual({ shouldContinue: true });
    expect(params.ctx.Body).toBe("keep going");
    expect(params.ctx.BodyForAgent).toBe("keep going");
-    expect((params.ctx as Record<string, unknown>).BodyStripped).toBe("keep going");
    expect(params.command.commandBodyNormalized).toBe("keep going");
-    expect(steerRuntimeMocks.queueEmbeddedAgentMessageWithOutcomeAsync).not.toHaveBeenCalled();
+    expect(queueMessage).not.toHaveBeenCalled();
  });

-  it("continues as a normal prompt when the active run rejects steering injection", async () => {
-    steerRuntimeMocks.resolveActiveEmbeddedRunSessionId.mockReturnValue("session-active");
-    steerRuntimeMocks.queueEmbeddedAgentMessageWithOutcomeAsync.mockResolvedValue({
-      queued: false,
-      sessionId: "session-active",
-      reason: "not_streaming",
-      gatewayHealth: "live",
-    });
-
+  it("continues visibly as a normal prompt when captured injection rejects", async () => {
+    beginActiveOperation("agent:main:main");
+    queueMessage.mockRejectedValueOnce(new Error("runtime rejected"));
    const params = buildParams("/steer keep going");
+
    const result = await handleSteerCommand(params, true);

-    expect(result).toEqual({
-      shouldContinue: true,
-    });
+    expect(result).toEqual({ shouldContinue: true });
    expect(params.ctx.BodyForAgent).toBe("keep going");
    expect(params.command.commandBodyNormalized).toBe("keep going");
-    expect(steerRuntimeMocks.formatEmbeddedAgentQueueFailureSummary).toHaveBeenCalledWith({
-      queued: false,
-      sessionId: "session-active",
-      reason: "not_streaming",
-      gatewayHealth: "live",
-    });
-  });
-
-  it("continues as a normal prompt when steering throws", async () => {
-    steerRuntimeMocks.resolveActiveEmbeddedRunSessionId.mockReturnValue("session-active");
-    steerRuntimeMocks.queueEmbeddedAgentMessageWithOutcomeAsync.mockRejectedValue(
-      new Error("socket closed"),
-    );
-
-    const params = buildParams("/steer keep going");
-    const result = await handleSteerCommand(params, true);
-
-    expect(result).toEqual({
-      shouldContinue: true,
-    });
-    expect(params.ctx.BodyForAgent).toBe("keep going");
-    expect(params.command.commandBodyNormalized).toBe("keep going");
-  });
-
-  it("continues as a normal prompt when the active run is compacting", async () => {
-    steerRuntimeMocks.resolveActiveEmbeddedRunSessionId.mockReturnValue("session-active");
-    steerRuntimeMocks.queueEmbeddedAgentMessageWithOutcomeAsync.mockResolvedValue({
-      queued: false,
-      sessionId: "session-active",
-      reason: "compacting",
-      gatewayHealth: "live",
-    });
-
-    const params = buildParams("/steer keep going");
-    const result = await handleSteerCommand(params, true);
-
-    expect(result).toEqual({
-      shouldContinue: true,
-    });
-    expect(params.ctx.BodyForAgent).toBe("keep going");
  });
 });
@@ -4,23 +4,23 @@ import {
  resolveInternalSessionKey,
  resolveMainSessionAlias,
 } from "../../agents/tools/sessions-helpers.js";
-import type { SessionEntry } from "../../config/sessions.js";
 import { logVerbose } from "../../globals.js";
 import { formatErrorMessage } from "../../infra/errors.js";
 import { isNativeCommandTurn, resolveCommandTurnContext } from "../command-turn-context.js";
 import { applyCommandTextToParams } from "./command-context-rewrite.js";
 import { commandReply, defineAuthorizedTextCommand } from "./command-gates.js";
-import {
-  formatEmbeddedAgentQueueFailureSummary,
-  isEmbeddedAgentRunActive,
-  queueEmbeddedAgentMessageWithOutcomeAsync,
-  resolveActiveEmbeddedRunSessionId,
-} from "./commands-steer.runtime.js";
 import type {
  CommandHandler,
  CommandHandlerResult,
  HandleCommandsParams,
 } from "./commands-types.js";
+import {
+  beginReplyMessageInjectionTarget,
+  finalizeReplyMessageInjectionAttempt,
+  replyRunRegistry,
+  type ReplyMessageInjectionTarget,
+} from "./reply-run-registry.js";
+import { resolveInboundReplyToolAuthorityOverlay } from "./reply-tool-authority.js";

 const STEER_USAGE = "Usage: /steer <message>";

@@ -46,21 +46,10 @@ function resolveSteerTargetSessionKey(params: HandleCommandsParams): string | un
  return resolveInternalSessionKey({ key: raw, alias, mainKey });
 }

-function resolveStoredSessionEntry(
-  params: HandleCommandsParams,
-  targetSessionKey: string,
-): SessionEntry | undefined {
-  if (params.sessionStore?.[targetSessionKey]) {
-    return params.sessionStore[targetSessionKey];
-  }
-  if (params.sessionKey === targetSessionKey) {
-    return params.sessionEntry;
-  }
-  return undefined;
-}
-
 function listSteerCandidateSessionKeys(targetSessionKey: string): string[] {
  const candidates = [targetSessionKey];
+  // Text slash turns still arrive on a source-only :slash: lane while the
+  // direct conversation owns the reply operation (#104844, #116763).
  if (targetSessionKey.includes(":slash:")) {
    candidates.push(
      targetSessionKey.replace(":slash:", ":direct:"),
@@ -70,23 +59,17 @@ function listSteerCandidateSessionKeys(targetSessionKey: string): string[] {
  return [...new Set(candidates)];
 }

-function resolveSteerSessionId(params: {
-  commandParams: HandleCommandsParams;
-  targetSessionKey: string;
-}): string | undefined {
-  const candidateKeys = listSteerCandidateSessionKeys(params.targetSessionKey);
+function resolveSteerTarget(
+  targetSessionKey: string,
+): { sessionId: string; sessionKey: string; target: ReplyMessageInjectionTarget } | undefined {
+  const candidateKeys = listSteerCandidateSessionKeys(targetSessionKey);
  for (const candidateKey of candidateKeys) {
-    const activeSessionId = resolveActiveEmbeddedRunSessionId(candidateKey);
-    if (activeSessionId) {
-      return activeSessionId;
-    }
-  }
-
-  for (const candidateKey of candidateKeys) {
-    const entry = resolveStoredSessionEntry(params.commandParams, candidateKey);
-    const sessionId = normalizeOptionalString(entry?.sessionId);
-    if (sessionId && isEmbeddedAgentRunActive(sessionId)) {
-      return sessionId;
+    const operation = replyRunRegistry.get(candidateKey);
+    const target = operation
+      ? replyRunRegistry.resolveCurrentMessageInjectionTarget(candidateKey)
+      : undefined;
+    if (operation && target) {
+      return { sessionId: operation.sessionId, sessionKey: candidateKey, target };
    }
  }

@@ -119,8 +102,8 @@ export const handleSteerCommand: CommandHandler = defineAuthorizedTextCommand(
      );
    }

-    const sessionId = resolveSteerSessionId({ commandParams: params, targetSessionKey });
-    if (!sessionId) {
+    const steerTarget = resolveSteerTarget(targetSessionKey);
+    if (!steerTarget) {
      return continueWithSteerFallback(
        params,
        message,
@@ -128,30 +111,41 @@ export const handleSteerCommand: CommandHandler = defineAuthorizedTextCommand(
      );
    }

-    const queueOutcome = await queueEmbeddedAgentMessageWithOutcomeAsync(sessionId, message, {
-      steeringMode: "all",
-      isInboundUserMessage: true,
-      debounceMs: 0,
-      ...(params.opts?.sourceReplyDeliveryMode
-        ? { sourceReplyDeliveryMode: params.opts.sourceReplyDeliveryMode }
-        : {}),
-      taskSuggestionDeliveryMode: params.opts?.taskSuggestionDeliveryMode,
+    const finalization = await finalizeReplyMessageInjectionAttempt({
+      target: steerTarget.target,
+      attempt: beginReplyMessageInjectionTarget(steerTarget.target, message, {
+        steeringMode: "all",
+        isInboundUserMessage: true,
+        toolAuthorityOverlay: resolveInboundReplyToolAuthorityOverlay({
+          ctx: params.ctx,
+          sessionEntry:
+            params.sessionStore?.[steerTarget.sessionKey] ??
+            (params.sessionKey === steerTarget.sessionKey ? params.sessionEntry : undefined),
+          senderIsOwner: params.command.senderIsOwner,
+          toolsAllow: params.opts?.toolsAllow,
+          disableTools: params.opts?.disableTools === true,
+        }),
+        debounceMs: 0,
+        ...(params.opts?.sourceReplyDeliveryMode
+          ? { sourceReplyDeliveryMode: params.opts.sourceReplyDeliveryMode }
+          : {}),
+        taskSuggestionDeliveryMode: params.opts?.taskSuggestionDeliveryMode,
+      }),
    }).catch((err: unknown): CommandHandlerResult => {
      return continueWithSteerFallback(
        params,
        message,
-        `steer: active session ${sessionId} threw while steering: ${formatErrorMessage(err)}; continuing with /steer payload as a normal prompt`,
+        `steer: active session ${steerTarget.sessionId} threw while steering: ${formatErrorMessage(err)}; continuing with /steer payload as a normal prompt`,
      );
    });
-    if ("shouldContinue" in queueOutcome) {
-      return queueOutcome;
+    if ("shouldContinue" in finalization) {
+      return finalization;
    }
-    if (!queueOutcome.queued) {
-      const summary = formatEmbeddedAgentQueueFailureSummary(queueOutcome);
+    if (finalization.status === "rejected") {
      return continueWithSteerFallback(
        params,
        message,
-        `steer: active session ${sessionId} rejected steering injection: ${summary}; continuing with /steer payload as a normal prompt`,
+        `steer: active session ${steerTarget.sessionId} rejected steering injection (${finalization.outcome.reason}); continuing with /steer payload as a normal prompt`,
      );
    }

@@ -54,6 +54,7 @@ import {
 } from "./dispatch-from-config.test-harness.js";
 import { getPreparedReplyDispatchRuntime } from "./prepared-reply-dispatch-context.js";
 import { createReplyDispatcher } from "./reply-dispatcher.js";
+import { resolveReplyOperationRunState } from "./reply-operation-run-state.js";
 import { admitReplyTurn } from "./reply-turn-admission.js";
 import { buildChannelSourceTurnId } from "./source-turn-id.js";
 import { buildTestCtx } from "./test-ctx.js";
@@ -343,6 +344,42 @@ describe("dispatchReplyFromConfig", () => {
    activeOperation.complete();
  });

+  it.each([
+    ["confirmed", false, "succeeded", "completed", "active_run_injected"],
+    ["unconfirmed", true, "blocked", "skipped", "reply_operation_aborted"],
+  ])(
+    "audits %s shared steering finalization with the Gateway terminal",
+    async (_name, aborted, status, outcome, reasonCode) => {
+      setNoAbort();
+      const dispatcher = createDispatcher();
+      const replyResolver = vi.fn(async (_ctx: MsgContext, opts?: GetReplyOptions) => {
+        const runState = resolveReplyOperationRunState(opts);
+        if (!runState) {
+          throw new Error("expected reply operation run state");
+        }
+        runState.admission = { status: "accepted", mode: "steer" };
+        runState.messageInjectionAborted = aborted ? true : undefined;
+        return undefined;
+      });
+
+      const result = await dispatchReplyFromConfig({
+        ctx: buildTestCtx({
+          Provider: "telegram",
+          Surface: "telegram",
+          SessionKey: "agent:main:telegram:direct:steer-audit",
+        }),
+        cfg: automaticDirectReplyConfig,
+        dispatcher,
+        replyResolver,
+      });
+
+      expect(result.deferredToActiveRun).toBe("steer");
+      expect(messageAuditEvents()).toContainEqual(
+        expect.objectContaining({ status, outcome, reasonCode }),
+      );
+    },
+  );
+
  it("skips a Telegram topic heartbeat turn while a reply operation is active", async () => {
    setNoAbort();
    const sessionKey = "agent:main:telegram:group:-1003774691294:topic:3731";
@@ -358,12 +358,17 @@ export async function finalizeDispatchAndAudit(state: ExecuteDispatchReadyState)
  counts.final += routedFinalCount;
  const agentRunTerminalOutcome = state.getAgentRunTerminalOutcome();
  state.commitInboundDedupeIfClaimed();
-  const dispatchOutcome = queueCapRejected ? "skipped" : "completed";
+  const messageInjectionAborted = state.replyOperationRunState.messageInjectionAborted === true;
+  const dispatchOutcome = queueCapRejected || messageInjectionAborted ? "skipped" : "completed";
  const dispatchReason = queueCapRejected
    ? "queue-cap"
-    : channelTransformSuppressed
-      ? "channel_transform"
-      : state.bindingState.pluginFallbackReason;
+    : messageInjectionAborted
+      ? "reply_operation_aborted"
+      : replyAdmission?.status === "accepted" && replyAdmission.mode === "steer"
+        ? "active_run_injected"
+        : channelTransformSuppressed
+          ? "channel_transform"
+          : state.bindingState.pluginFallbackReason;
  state.recordAgentDispatchCompleted(
    dispatchOutcome,
    dispatchReason ? { reason: dispatchReason } : undefined,
@@ -10,6 +10,7 @@ type ReplyOperationAdmissionSnapshot =

 export type ReplyOperationRunState = {
  admission?: ReplyOperationAdmissionSnapshot;
+  messageInjectionAborted?: true;
 };

 // Carries this invocation's admission decision through reply option spreads so
@@ -119,6 +119,11 @@ export type ReplyBackendHandle = {
  readonly taskSuggestionDeliveryMode?: TaskSuggestionDeliveryMode;
  /** True only when queueMessage preserves images supplied in its options. */
  readonly supportsQueueMessageImages?: boolean;
+  claimPendingUserInputAnswer?: (
+    text: string,
+    options?: ReplyBackendQueueMessageOptions,
+  ) => Promise<boolean>;
+  cancelPendingUserInput?: (resolvedBy: string) => Promise<boolean>;
  cancel(reason?: ReplyBackendCancelReason): void;
  readonly messageInjection?: ReplyBackendMessageInjection;
  /** @deprecated Compatibility for shipped embedded handles. Use messageInjection. */
@@ -140,19 +145,14 @@ export type ReplyBackendHandle = {
 export const replyMessageInjectionTargetOperation = Symbol("replyMessageInjectionTargetOperation");
 export type ReplyMessageInjectionTarget = {
  readonly [replyMessageInjectionTargetOperation]: ReplyOperation;
-  /** Legacy targets stay leaf-bound even when their backend exposes a run id. */
-  readonly identity: "leaf" | "run";
+  readonly identity: "operation" | "run";
  readonly runId?: string;
-  readonly originatingLeafEntryId: string | null | undefined;
-  /** Tool authority captured with the exact active operation. */
-  readonly toolAuthorityFingerprint?: string;
 };

 type ReplyMessageInjectionRejectionReason =
  | "no_active_run"
  | "not_running"
  | "stale_run"
-  | "leaf_mismatch"
  | "run_mismatch"
  | "injection_unavailable"
  | ReplyBackendQueueMessageMismatch
@@ -165,8 +165,6 @@ export type ReplyMessageInjectionOutcome =
 export type ReplyMessageInjectionAttempt = {
  /** Native run identity captured with the opaque operation target. */
  targetRunId: string | undefined;
-  /** Leaf-bound compatibility must reject before ACK instead of falling through. */
-  rejectBeforeAck?: true;
  /** Settles once the runtime accepts or rejects ownership of this exact message. */
  acceptance: Promise<boolean>;
  /** Settles after the backend confirms or rejects this exact injection. */
@@ -327,9 +325,12 @@ export type ReplyRunRegistry = {
  isActive(sessionKey: string): boolean;
  resolveMessageInjectionTarget(params: {
    sessionKey: string;
+    /** Retained in the internal call shape until expected-leaf removal; injection ignores it. */
    originatingLeafEntryId: string | null | undefined;
    expectedRunId?: string;
  }): ReplyMessageInjectionTarget | undefined;
+  /** Captures the current direct owner without requiring client-supplied run identity. */
+  resolveCurrentMessageInjectionTarget(sessionKey: string): ReplyMessageInjectionTarget | undefined;
  abort(sessionKey: string): boolean;
  waitForIdle(
    sessionKey: string,
@@ -28,7 +28,6 @@ type ReplyMessageInjectionRejectionReason =
  | "no_active_run"
  | "not_running"
  | "stale_run"
-  | "leaf_mismatch"
  | "run_mismatch"
  | "injection_unavailable"
  | ReplyBackendQueueMessageMismatch
@@ -101,11 +100,14 @@ function resolveReplyBackendMessageInjection(

 export function resolveReplyMessageInjectionRejection(params: {
  operation: ReplyOperation | undefined;
-  originatingLeafEntryId: string | null | undefined;
  expectedRunId?: string;
  options?: ReplyBackendQueueMessageOptions;
 }):
-  | { reason: ReplyMessageInjectionRejectionReason; errorMessage?: string }
+  | {
+      reason: ReplyMessageInjectionRejectionReason;
+      errorMessage?: string;
+      backend?: ReplyBackendHandle;
+    }
  | { backend: ReplyBackendHandle; injection: ReplyBackendMessageInjection } {
  const { operation } = params;
  if (!operation || replyRunState.activeRunsByKey.get(operation.key) !== operation) {
@@ -117,9 +119,6 @@ export function resolveReplyMessageInjectionRejection(params: {
  const expectedRunId = normalizeOptionalString(params.expectedRunId);
  // Exact run identity supersedes the operation's immutable origin leaf. The
  // same run advances its transcript leaf during ordinary tool/output progress.
-  if (!expectedRunId && operation.originatingLeafEntryId !== params.originatingLeafEntryId) {
-    return { reason: "leaf_mismatch" };
-  }
  if (isReplyRunEvidenceStale(operation)) {
    return { reason: "stale_run" };
  }
@@ -139,16 +138,31 @@ export function resolveReplyMessageInjectionRejection(params: {
    return { reason: "injection_unavailable", errorMessage: String(error) };
  }
  const mismatch = resolveReplyBackendQueueMessageMismatch(backend, params.options, operation);
-  return mismatch ? { reason: mismatch } : { backend, injection };
-}
-
-function isLeafOwnershipRejection(reason: ReplyMessageInjectionRejectionReason): boolean {
-  return (
-    reason === "no_active_run" ||
-    reason === "not_running" ||
-    reason === "stale_run" ||
-    reason === "leaf_mismatch"
+  const activeFingerprint = normalizeOptionalString(
+    backend.toolAuthorityFingerprint ?? operation.toolAuthorityFingerprint,
  );
+  const pendingInputAuthorityProven =
+    activeFingerprint !== undefined &&
+    normalizeOptionalString(params.options?.pendingInputAuthorityFingerprint) === activeFingerprint;
+  if (
+    mismatch === "tool_authority_mismatch" &&
+    pendingInputAuthorityProven &&
+    !params.options?.images?.length &&
+    backend.claimPendingUserInputAnswer
+  ) {
+    return {
+      backend,
+      injection: {
+        isAvailable: () => true,
+        queueMessage: async (text, options) => {
+          if (!(await backend.claimPendingUserInputAnswer?.(text, options))) {
+            throw new Error("pending user input was not accepted");
+          }
+        },
+      },
+    };
+  }
+  return mismatch ? { reason: mismatch, backend } : { backend, injection };
 }

 export function beginReplyMessageInjectionTarget(
@@ -171,19 +185,28 @@ export function beginReplyMessageInjectionTarget(
    : undefined;
  const resolved = resolveReplyMessageInjectionRejection({
    operation,
-    originatingLeafEntryId: target.originatingLeafEntryId,
    expectedRunId: target.identity === "run" ? target.runId : undefined,
    options: queueOptions,
  });
  if (!("injection" in resolved)) {
-    const immediateRejection = { status: "rejected" as const, ...resolved };
+    const immediateRejection = {
+      status: "rejected" as const,
+      reason: resolved.reason,
+      ...(resolved.errorMessage ? { errorMessage: resolved.errorMessage } : {}),
+    };
+    const cancelPendingImage =
+      options?.isInboundUserMessage === true &&
+      Boolean(options.images?.length) &&
+      (resolved.reason === "tool_authority_mismatch" ||
+        resolved.reason === "image_input_unsupported")
+        ? resolved.backend?.cancelPendingUserInput
+        : undefined;
    return {
      targetRunId: target.runId,
-      ...(target.identity === "leaf" && isLeafOwnershipRejection(resolved.reason)
-        ? { rejectBeforeAck: true as const }
-        : {}),
      acceptance: Promise.resolve(false),
-      outcome: Promise.resolve(immediateRejection),
+      outcome: cancelPendingImage
+        ? Promise.resolve(cancelPendingImage("image-reply")).then(() => immediateRejection)
+        : Promise.resolve(immediateRejection),
    };
  }
  const targetRunId = normalizeOptionalString(resolved.backend.runId);
@@ -256,13 +279,53 @@ export function beginReplyMessageInjectionTarget(
  };
 }

+/** Finalize adoption and cleanup on the captured operation without rediscovery. */
+export async function finalizeReplyMessageInjectionAttempt(params: {
+  attempt: ReplyMessageInjectionAttempt;
+  target: ReplyMessageInjectionTarget;
+  inboundAudio?: boolean;
+  onAccepted?: () => void;
+  onAdopted?: () => void | Promise<void>;
+  shouldAbortOnAdoptionError?: (error: unknown) => boolean;
+}) {
+  const outcome = await params.attempt.outcome;
+  if (outcome.status === "rejected") {
+    return { status: "rejected" as const, outcome, targetRunId: params.attempt.targetRunId };
+  }
+  recordAcceptedReplyMessageInjectionTarget(params.target, {
+    inboundAudio: params.inboundAudio,
+  });
+  params.onAccepted?.();
+  let aborted = outcome.result?.transcriptCommit === "unconfirmed";
+  if (aborted) {
+    abortReplyMessageInjectionTarget(params.target);
+  }
+  let adoptionError: unknown;
+  try {
+    await params.onAdopted?.();
+  } catch (error) {
+    adoptionError = error;
+    if (params.shouldAbortOnAdoptionError?.(error)) {
+      abortReplyMessageInjectionTarget(params.target);
+      aborted = true;
+    }
+  }
+  return {
+    status: "accepted" as const,
+    outcome,
+    targetRunId: params.attempt.targetRunId,
+    aborted,
+    ...(adoptionError === undefined ? {} : { adoptionError }),
+  };
+}
+
 /** Abort only the operation captured by this target; never a same-key successor. */
-export function abortReplyMessageInjectionTarget(target: ReplyMessageInjectionTarget): boolean {
+function abortReplyMessageInjectionTarget(target: ReplyMessageInjectionTarget): boolean {
  return target[replyMessageInjectionTargetOperation].abortByUser();
 }

 /** Record accepted input on the exact operation without rediscovering its session slot. */
-export function recordAcceptedReplyMessageInjectionTarget(
+function recordAcceptedReplyMessageInjectionTarget(
  target: ReplyMessageInjectionTarget,
  options?: { inboundAudio?: boolean },
 ): void {
@@ -108,11 +108,10 @@ export const replyRunRegistry: ReplyRunRegistry = {
    }
    return replyRunState.activeRunsByKey.has(normalizedSessionKey);
  },
-  resolveMessageInjectionTarget({ sessionKey, originatingLeafEntryId, expectedRunId }) {
+  resolveMessageInjectionTarget({ sessionKey, expectedRunId }) {
    const operation = this.get(sessionKey);
    const resolved = resolveReplyMessageInjectionRejection({
      operation,
-      originatingLeafEntryId,
      expectedRunId,
    });
    if (!("injection" in resolved)) {
@@ -120,15 +119,25 @@ export const replyRunRegistry: ReplyRunRegistry = {
    }
    const target: ReplyMessageInjectionTarget = {
      [replyMessageInjectionTargetOperation]: operation!,
-      identity: normalizeOptionalString(expectedRunId) ? "run" : "leaf",
+      identity: normalizeOptionalString(expectedRunId) ? "run" : "operation",
      ...(resolved.backend.runId ? { runId: resolved.backend.runId } : {}),
-      originatingLeafEntryId,
-      ...(operation?.toolAuthorityFingerprint
-        ? { toolAuthorityFingerprint: operation.toolAuthorityFingerprint }
-        : {}),
    };
    return target;
  },
+  resolveCurrentMessageInjectionTarget(sessionKey) {
+    const operation = this.get(sessionKey);
+    const resolved = resolveReplyMessageInjectionRejection({
+      operation,
+    });
+    if (!operation || !("injection" in resolved)) {
+      return undefined;
+    }
+    return {
+      [replyMessageInjectionTargetOperation]: operation,
+      identity: "operation",
+      ...(resolved.backend.runId ? { runId: resolved.backend.runId } : {}),
+    };
+  },
  abort(sessionKey) {
    const operation = this.get(sessionKey);
    if (!operation) {
@@ -16,9 +16,9 @@ import { enqueueCommandInLane, setCommandLaneConcurrency } from "../../process/c
 import { resetCommandQueueStateForTest } from "../../process/command-queue.test-support.js";
 import { createQueueTestRun } from "./queue.test-helpers.js";
 import { beginReplyOperationFinalizationWork } from "./reply-run-finalization-lease.js";
+import type { ReplyToolAuthorityOverlay } from "./reply-run-registry.contracts.js";
 import {
  abortActiveReplyRuns,
-  abortReplyMessageInjectionTarget,
  beginReplyMessageInjectionTarget,
  createReplyOperation,
  expireStaleReplyOperation,
@@ -33,7 +33,6 @@ import {
  REPLY_RUN_TERMINAL_SETTLE_TIMEOUT_MS,
  registerReplyOperationSuccessorBarrier,
  type ReplyBackendQueueMessageOptions,
-  type ReplyToolAuthorityOverlay,
  ReplyRunAlreadyActiveError,
  ReplyRunSuccessorAdmissionBlockedError,
  replyRunRegistry,
@@ -250,79 +249,6 @@ describe("reply run registry", () => {
    expect(isReplyRunAbortableForCompaction("session-compact")).toBe(true);
  });

-  it("binds modern targets by run while preserving leaf-only legacy targeting", async () => {
-    const operation = createTestReplyOperation({ originatingLeafEntryId: "leaf-a" });
-    let stopped = false;
-    const queueMessage = vi.fn(async () => {});
-    operation.setPhase("running");
-    operation.attachBackend({
-      kind: "embedded",
-      runId: "run-a",
-      cancel: () => {},
-      messageInjection: { isAvailable: () => !stopped, queueMessage },
-    });
-
-    const target = replyRunRegistry.resolveMessageInjectionTarget({
-      sessionKey: "agent:main:main",
-      originatingLeafEntryId: "leaf-b",
-      expectedRunId: "run-a",
-    });
-    expect(target).toMatchObject({ identity: "run", runId: "run-a" });
-    const legacyTarget = replyRunRegistry.resolveMessageInjectionTarget({
-      sessionKey: "agent:main:main",
-      originatingLeafEntryId: "leaf-a",
-    });
-    expect(legacyTarget).toMatchObject({ identity: "leaf", runId: "run-a" });
-    expect(
-      replyRunRegistry.resolveMessageInjectionTarget({
-        sessionKey: "agent:main:main",
-        originatingLeafEntryId: "leaf-b",
-      }),
-    ).toBeUndefined();
-    await expect(
-      queueReplyMessageInjectionTarget(target!, "steer during tool work"),
-    ).resolves.toEqual({ status: "accepted" });
-    await expect(queueReplyMessageInjectionTarget(legacyTarget!, "legacy steer")).resolves.toEqual({
-      status: "accepted",
-    });
-    expect(queueMessage).toHaveBeenCalledWith(
-      "steer during tool work",
-      expect.objectContaining({ onQueueAccepted: expect.any(Function) }),
-    );
-    expect(queueMessage).toHaveBeenCalledWith(
-      "legacy steer",
-      expect.objectContaining({ onQueueAccepted: expect.any(Function) }),
-    );
-    stopped = true;
-    await expect(queueReplyMessageInjectionTarget(target!, "late steer")).resolves.toEqual({
-      status: "rejected",
-      reason: "injection_unavailable",
-    });
-  });
-
-  it("requires an explicit legacy leaf while preserving deliberate null", () => {
-    const operation = createTestReplyOperation({ originatingLeafEntryId: null });
-    operation.setPhase("running");
-    operation.attachBackend({
-      kind: "embedded",
-      cancel: vi.fn(),
-      messageInjection: { isAvailable: () => true, queueMessage: vi.fn(async () => {}) },
-    });
-
-    expect(
-      replyRunRegistry.resolveMessageInjectionTarget({
-        sessionKey: operation.key,
-        originatingLeafEntryId: undefined,
-      }),
-    ).toBeUndefined();
-    expect(
-      replyRunRegistry.resolveMessageInjectionTarget({
-        sessionKey: operation.key,
-        originatingLeafEntryId: null,
-      }),
-    ).toMatchObject({ identity: "leaf", originatingLeafEntryId: null });
-  });
-
  it("records reply-operation progress without claiming embedded-run activity", () => {
    const operation = createTestReplyOperation({
      sessionKey: "agent:main:telegram:direct:chat-1",
@@ -2292,36 +2218,6 @@ describe("reply run registry", () => {
    expect(successorQueue).not.toHaveBeenCalled();
  });

-  it("exact-target abort cannot abort a same-key successor", () => {
-    const first = createTestReplyOperation({ originatingLeafEntryId: "leaf-a" });
-    first.setPhase("running");
-    first.attachBackend({
-      kind: "embedded",
-      runId: "run-a",
-      cancel: vi.fn(),
-      messageInjection: { isAvailable: () => true, queueMessage: vi.fn(async () => {}) },
-    });
-    const target = replyRunRegistry.resolveMessageInjectionTarget({
-      sessionKey: first.key,
-      originatingLeafEntryId: "leaf-a",
-      expectedRunId: "run-a",
-    })!;
-    first.complete();
-    const successorCancel = vi.fn();
-    const successor = createTestReplyOperation({ originatingLeafEntryId: "leaf-a" });
-    successor.setPhase("running");
-    successor.attachBackend({
-      kind: "embedded",
-      runId: "run-b",
-      cancel: successorCancel,
-      messageInjection: { isAvailable: () => true, queueMessage: vi.fn(async () => {}) },
-    });
-
-    expect(abortReplyMessageInjectionTarget(target)).toBe(false);
-    expect(successor.result).toBeNull();
-    expect(successorCancel).not.toHaveBeenCalled();
-  });
-
  it("uses a replacement backend on the same operation", async () => {
    const operation = createTestReplyOperation({ originatingLeafEntryId: "leaf-a" });
    operation.setPhase("running");
@@ -11,17 +11,14 @@ export type {
  ReplyBackendQueueMessageOptions,
  ReplyBackendQueueMessageResult,
  ReplyMessageInjectionAttempt,
-  ReplyMessageInjectionOutcome,
  ReplyMessageInjectionTarget,
  ReplyOperation,
  ReplyOperationPhase,
-  ReplyToolAuthorityOverlay,
  ReplyTurnKind,
 } from "./reply-run-registry.contracts.js";
 export {
-  abortReplyMessageInjectionTarget,
  beginReplyMessageInjectionTarget,
-  recordAcceptedReplyMessageInjectionTarget,
+  finalizeReplyMessageInjectionAttempt,
  resolveReplyBackendQueueMessageMismatch,
 } from "./reply-run-registry.message-injection.js";
 export {
@@ -1,8 +1,14 @@
 import { createHash } from "node:crypto";
 import { stableStringify } from "@openclaw/normalization-core";
+import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
 import { resolveConversationCapabilityProfile } from "../../agents/conversation-capability-profile.js";
 import { resolveSandboxRuntimeStatus } from "../../agents/sandbox/runtime-status.js";
 import { readToolAllowlistIntersection } from "../../agents/tool-policy.js";
+import { normalizeChatType } from "../../channels/chat-type.js";
+import type { SessionEntry } from "../../config/sessions.js";
+import { resolveGroupSessionKey } from "../../config/sessions/group.js";
+import type { RuntimeMsgContext } from "../templating.js";
+import { resolveOriginMessageProvider } from "./origin-routing.js";
 import type { FollowupRun } from "./queue.js";
 import type {
  ReplyToolAuthorityOverlay,
@@ -18,6 +24,53 @@ type ReplyToolAuthoritySnapshot = {
  run: FollowupRun["run"];
 };

+/** Projects current inbound facts against the active run's frozen authority snapshot. */
+export function resolveInboundReplyToolAuthorityOverlay(params: {
+  ctx: RuntimeMsgContext;
+  sessionEntry?: Pick<SessionEntry, "spawnedBy">;
+  senderIsOwner: boolean;
+  toolsAllow?: string[];
+  disableTools: boolean;
+}): ReplyToolAuthorityOverlay {
+  const { ctx } = params;
+  return {
+    originatingChannel: ctx.OriginatingChannel,
+    messageProvider: resolveOriginMessageProvider({
+      originatingChannel: ctx.OriginatingChannel,
+      provider: ctx.Provider ?? ctx.Surface,
+    }),
+    chatType: normalizeChatType(ctx.ChatType),
+    agentAccountId: ctx.AccountId,
+    conversationToolPolicy: ctx.ConversationToolPolicy,
+    groupId: resolveGroupSessionKey(ctx)?.id,
+    groupChannel:
+      normalizeOptionalString(ctx.GroupChannel) ?? normalizeOptionalString(ctx.GroupSubject),
+    groupSpace: normalizeOptionalString(ctx.GroupSpace),
+    memberRoleIds: Array.isArray(ctx.MemberRoleIds)
+      ? ctx.MemberRoleIds.map((roleId) => normalizeOptionalString(roleId)).filter(
+          (roleId): roleId is string => Boolean(roleId),
+        )
+      : undefined,
+    spawnedBy: normalizeOptionalString(params.sessionEntry?.spawnedBy),
+    senderId: normalizeOptionalString(ctx.SenderId),
+    senderName: normalizeOptionalString(ctx.SenderName),
+    senderUsername: normalizeOptionalString(ctx.SenderUsername),
+    senderE164: normalizeOptionalString(ctx.SenderE164),
+    senderIsOwner: params.senderIsOwner,
+    inputProvenance: ctx.InputProvenance,
+    trustedInternalHandoff: undefined,
+    scheduledToolPolicy: undefined,
+    runtimePluginToolGrant: undefined,
+    toolsAllow: params.toolsAllow,
+    disableTools: params.disableTools,
+    traceAuthorized:
+      params.senderIsOwner || (ctx.GatewayClientScopes ?? []).includes("operator.admin"),
+    approvalReviewerDeviceId: normalizeOptionalString(ctx.ApprovalReviewerDeviceId),
+    clientCaps: ctx.GatewayClientCaps,
+    toolBindings: ctx.GatewayRunToolBindings,
+  };
+}
+
 function snapshotFollowupRunToolAuthority(run: FollowupRun): ReplyToolAuthoritySnapshot {
  return {
    originatingChannel: run.originatingChannel,
@@ -27,7 +27,6 @@ import {
  hasRestartRecoveryTerminalRun,
  isRetryableUnadoptedChatClaim,
  resolveRestartSafeChatAdmission,
-  terminalizeRestartSafeChatAdmission,
 } from "./chat-restart-recovery.js";
 import {
  ACTIVE_LEAF_CHANGED_ERROR_REASON,
@@ -195,30 +194,22 @@ export async function admitChatSend(params: {
    if (entry && !latestEntry) {
      throw new Error(`Session "${sessionKey}" was deleted while starting work. Retry.`);
    }
-    // An active owner can advance this branch while a steer is being composed.
-    // The lifecycle admission keeps branch identity fixed after this check; if
-    // the owner clears, acceptance-aware dispatch preserves this turn as follow-up.
-    const hasSteerIdentity = expectedRunId !== undefined || expectedLeafEntryId !== undefined;
+    // Capture the exact direct owner under the writer barrier. If it clears
+    // later, the opaque target rejects instead of resolving a successor.
    const resolvedInjectionTarget =
-      p.queueMode === "steer" && hasSteerIdentity
-        ? replyRunRegistry.resolveMessageInjectionTarget({
-            sessionKey: activeRunScopeKey,
-            originatingLeafEntryId: expectedLeafEntryId,
-            expectedRunId,
-          })
-        : undefined;
+      p.queueMode !== "steer"
+        ? undefined
+        : expectedRunId !== undefined
+          ? replyRunRegistry.resolveMessageInjectionTarget({
+              sessionKey: activeRunScopeKey,
+              originatingLeafEntryId: expectedLeafEntryId,
+              expectedRunId,
+            })
+          : replyRunRegistry.resolveCurrentMessageInjectionTarget(activeRunScopeKey);
    if (commitOutcome && resolvedInjectionTarget) {
      messageInjectionTarget = resolvedInjectionTarget;
    }
-    if (
-      commitOutcome &&
-      p.queueMode === "steer" &&
-      expectedRunId === undefined &&
-      !resolvedInjectionTarget
-    ) {
-      throw new Error(ACTIVE_LEAF_CHANGED_ERROR_REASON);
-    }
-    if (commitOutcome && expectedLeafEntryId !== undefined && !resolvedInjectionTarget) {
+    if (commitOutcome && p.queueMode !== "steer" && expectedLeafEntryId !== undefined) {
      // Runtime session identity resolves through the canonical SQLite accessor;
      // legacy/reset-archive files are read-only history fallbacks, never send targets.
      const activePathRelation = latestEntry?.sessionId
@@ -474,25 +465,6 @@ export async function admitChatSend(params: {
    discardAbandonedPreparedMedia?.();
    discardAbandonedPreparedMedia = undefined;
  };
-  const rejectActiveLeafChanged = async () => {
-    if (
-      restartSafeAdmission &&
-      !(await terminalizeRestartSafeChatAdmission({
-        admittedSessionId,
-        clientRunId,
-        sessionKey,
-        startedAt: now,
-        status: "failed",
-        storePath,
-        retryable: true,
-      }))
-    ) {
-      throw new Error("chat admission ownership changed before terminalization");
-    }
-    cleanupAdmittedRun({ force: true });
-    clearAgentRunContext(clientRunId, lifecycleGeneration);
-    respondChatActiveLeafChanged(respond);
-  };
  const rejectSessionRoutingChanged = () => {
    cleanupAdmittedRun({ force: true });
    clearAgentRunContext(clientRunId, lifecycleGeneration);
@@ -529,7 +501,6 @@ export async function admitChatSend(params: {
      lifecycleGeneration,
      messageInjectionTarget,
      originatingRoute,
-      rejectActiveLeafChanged,
      rejectSessionRoutingChanged,
      retainGatewayWorkAdmission,
      restartSafeAdmission,
@@ -118,7 +118,6 @@ export function startChatDispatch(params: StartChatDispatchParams): void {
    clientRunId,
    entry,
    expectedLeafEntryId,
-    expectedRunId,
    requestedSessionId,
    resolvedSessionModel,
    selectedAgent,
@@ -224,19 +223,18 @@ export function startChatDispatch(params: StartChatDispatchParams): void {
            messageInjectionAttempt = beginCapturedMessageInjection();
          }
          if (messageInjectionAttempt) {
-            const outcome = await messageInjectionAttempt.outcome;
-            if (outcome.status === "accepted") {
-              acceptedMessageInjection = true;
+            if (
              await finalizeAcceptedChatSendMessageInjection({
+                attempt: messageInjectionAttempt,
                context,
                ctx,
-                outcome,
                persistUserTurnTranscriptBestEffort: persistGatewayUserTurnTranscriptBestEffort,
                session,
                startedAt: admissionStartedAt,
                target: messageInjectionTarget!,
-                targetRunId: messageInjectionAttempt.targetRunId,
-              });
+              })
+            ) {
+              acceptedMessageInjection = true;
              return {
                queuedFinal: false,
                counts: { tool: 0, block: 0, final: 0 },
@@ -304,9 +302,8 @@ export function startChatDispatch(params: StartChatDispatchParams): void {
                fastModeOverride: p.fastMode,
                queueModeOverride: p.queueMode,
                userTurnTranscriptRecorder: userTurnRecorder,
-                ...((messageInjectionTarget && !isInternalTextSlashCommandTurn) ||
-                (p.queueMode === "steer" && expectedRunId !== undefined)
-                  ? { messageInjectionAttempted: true as const }
+                ...(p.queueMode === "steer"
+                  ? { messageInjectionDisposition: "rejected" as const }
                  : {}),
                ...(restartSafeAdmission ? { suppressNextUserMessagePersistence: true } : {}),
                fastModeAutoOnSecondsOverride: p.fastAutoOnSeconds,
@@ -212,7 +212,6 @@ async function handleChatSendWithOptions(
      attempt: messageInjectionAttempt,
      isAborted: () => activeRunAbort.controller.signal.aborted,
      sessionRoutingChanged: () => sessionRoutingChanged(context.getRuntimeConfig()),
-      onActiveLeafChanged: admitted.value.rejectActiveLeafChanged,
      onAborted: finishAbortedChatSend,
      onSessionRoutingChanged: admitted.value.rejectSessionRoutingChanged,
    });
@@ -2,9 +2,7 @@
 import { beforeEach, describe, expect, it, vi } from "vitest";
 import { emitInboundMessageAuditTerminal } from "../../auto-reply/reply/dispatch-from-config.audit.js";
 import {
-  abortReplyMessageInjectionTarget,
-  recordAcceptedReplyMessageInjectionTarget,
-  type ReplyMessageInjectionOutcome,
+  finalizeReplyMessageInjectionAttempt,
  type ReplyMessageInjectionTarget,
 } from "../../auto-reply/reply/reply-run-registry.js";
 import { updateSessionEntry } from "../../config/sessions/session-accessor.js";
@@ -16,9 +14,8 @@ vi.mock("../../auto-reply/reply/dispatch-from-config.audit.js", () => ({
  emitInboundMessageAuditTerminal: vi.fn(),
 }));
 vi.mock("../../auto-reply/reply/reply-run-registry.js", () => ({
-  abortReplyMessageInjectionTarget: vi.fn(() => true),
  beginReplyMessageInjectionTarget: vi.fn(),
-  recordAcceptedReplyMessageInjectionTarget: vi.fn(),
+  finalizeReplyMessageInjectionAttempt: vi.fn(),
 }));
 vi.mock("../../auto-reply/reply/message-received-hooks.js", () => ({
  emitMessageReceivedHooks: vi.fn(),
@@ -40,7 +37,7 @@ vi.mock("../agent-turn/agent-job.js", () => ({
  setGatewayDedupeEntry: vi.fn(),
 }));

-function makeParams(outcome: Extract<ReplyMessageInjectionOutcome, { status: "accepted" }>) {
+function makeParams() {
  const context = {
    logGateway: { warn: vi.fn() },
    chatRunState: { hasAbortMarker: () => true },
@@ -49,7 +46,7 @@ function makeParams(outcome: Extract<ReplyMessageInjectionOutcome, { status: "ac
  return {
    context,
    ctx: { Provider: "dashboard", From: "user", To: "user", Body: "steer" },
-    outcome,
+    attempt: {},
    persistUserTurnTranscriptBestEffort: vi.fn(async () => undefined),
    session: {
      agentId: "main",
@@ -61,7 +58,6 @@ function makeParams(outcome: Extract<ReplyMessageInjectionOutcome, { status: "ac
    },
    startedAt: Date.now(),
    target: {} as ReplyMessageInjectionTarget,
-    targetRunId: "run-1",
  } as unknown as Parameters<typeof finalizeAcceptedChatSendMessageInjection>[0];
 }

@@ -71,12 +67,14 @@ beforeEach(() => {

 describe("finalizeAcceptedChatSendMessageInjection", () => {
  it("audits a confirmed steer as completed active_run_injected", async () => {
-    await finalizeAcceptedChatSendMessageInjection(
-      makeParams({ status: "accepted", result: undefined } as never),
-    );
+    vi.mocked(finalizeReplyMessageInjectionAttempt).mockResolvedValueOnce({
+      status: "accepted",
+      outcome: { status: "accepted" },
+      targetRunId: "run-1",
+      aborted: false,
+    });
+    await finalizeAcceptedChatSendMessageInjection(makeParams());

-    expect(abortReplyMessageInjectionTarget).not.toHaveBeenCalled();
-    expect(recordAcceptedReplyMessageInjectionTarget).toHaveBeenCalledOnce();
    expect(logMessageProcessed).toHaveBeenCalledWith(
      expect.objectContaining({ outcome: "completed", reason: "active_run_injected" }),
    );
@@ -89,14 +87,17 @@ describe("finalizeAcceptedChatSendMessageInjection", () => {
  });

  it("audits an unconfirmed-transcript steer abort as skipped, not completed", async () => {
-    await finalizeAcceptedChatSendMessageInjection(
-      makeParams({
+    vi.mocked(finalizeReplyMessageInjectionAttempt).mockResolvedValueOnce({
+      status: "accepted",
+      outcome: {
        status: "accepted",
        result: { transcriptCommit: "unconfirmed", errorMessage: "commit timeout" },
-      } as never),
-    );
+      },
+      targetRunId: "run-1",
+      aborted: true,
+    });
+    await finalizeAcceptedChatSendMessageInjection(makeParams());

-    expect(abortReplyMessageInjectionTarget).toHaveBeenCalledOnce();
    expect(logMessageProcessed).toHaveBeenCalledWith(
      expect.objectContaining({ outcome: "skipped", reason: "reply_operation_aborted" }),
    );
@@ -1,27 +1,19 @@
-import {
-  normalizeLowercaseStringOrEmpty,
-  normalizeOptionalString,
-} from "@openclaw/normalization-core/string-coerce";
+import { normalizeLowercaseStringOrEmpty } from "@openclaw/normalization-core/string-coerce";
 import { resolveCommandAuthorization } from "../../auto-reply/command-auth.js";
 import { emitInboundMessageAuditTerminal } from "../../auto-reply/reply/dispatch-from-config.audit.js";
 import { finalizeInboundContext } from "../../auto-reply/reply/inbound-context.js";
 import { hasInboundAudio } from "../../auto-reply/reply/inbound-media.js";
 import { emitMessageReceivedHooks } from "../../auto-reply/reply/message-received-hooks.js";
-import { resolveOriginMessageProvider } from "../../auto-reply/reply/origin-routing.js";
 import { resolveQueueSettings } from "../../auto-reply/reply/queue/settings-runtime.js";
 import {
-  abortReplyMessageInjectionTarget,
  beginReplyMessageInjectionTarget,
-  recordAcceptedReplyMessageInjectionTarget,
+  finalizeReplyMessageInjectionAttempt,
  type ReplyBackendQueueMessageOptions,
  type ReplyMessageInjectionAttempt,
-  type ReplyMessageInjectionOutcome,
  type ReplyMessageInjectionTarget,
-  type ReplyToolAuthorityOverlay,
 } from "../../auto-reply/reply/reply-run-registry.js";
+import { resolveInboundReplyToolAuthorityOverlay } from "../../auto-reply/reply/reply-tool-authority.js";
 import type { RuntimeMsgContext } from "../../auto-reply/templating.js";
-import { normalizeChatType } from "../../channels/chat-type.js";
-import { resolveGroupSessionKey } from "../../config/sessions/group.js";
 import { updateSessionEntry } from "../../config/sessions/session-accessor.js";
 import { isDiagnosticsEnabled } from "../../infra/diagnostic-events.js";
 import { logMessageProcessed, logMessageReceived } from "../../logging/diagnostic.js";
@@ -34,54 +26,6 @@ import type { PreparedChatSendSession } from "./chat-send-session.js";
 import type { prepareChatSendUserTurn } from "./chat-send-user-turn.js";
 import type { GatewayRequestContext } from "./types.js";

-function resolveChatSendToolAuthorityOverlay(params: {
-  ctx: RuntimeMsgContext;
-  session: Pick<PreparedChatSendSession, "cfg" | "entry">;
-}): ReplyToolAuthorityOverlay {
-  const { ctx, session } = params;
-  const authorization = resolveCommandAuthorization({
-    ctx,
-    cfg: session.cfg,
-    commandAuthorized: ctx.CommandAuthorized === true,
-  });
-  const senderIsOwner = authorization.senderIsOwner;
-  return {
-    originatingChannel: ctx.OriginatingChannel,
-    messageProvider: resolveOriginMessageProvider({
-      originatingChannel: ctx.OriginatingChannel,
-      provider: ctx.Provider,
-    }),
-    chatType: normalizeChatType(ctx.ChatType),
-    agentAccountId: ctx.AccountId,
-    conversationToolPolicy: ctx.ConversationToolPolicy,
-    groupId: resolveGroupSessionKey(ctx)?.id,
-    groupChannel:
-      normalizeOptionalString(ctx.GroupChannel) ?? normalizeOptionalString(ctx.GroupSubject),
-    groupSpace: normalizeOptionalString(ctx.GroupSpace),
-    memberRoleIds: Array.isArray(ctx.MemberRoleIds)
-      ? ctx.MemberRoleIds.map((roleId) => normalizeOptionalString(roleId)).filter(
-          (roleId): roleId is string => Boolean(roleId),
-        )
-      : undefined,
-    spawnedBy: session.entry?.spawnedBy,
-    senderId: normalizeOptionalString(ctx.SenderId),
-    senderName: normalizeOptionalString(ctx.SenderName),
-    senderUsername: normalizeOptionalString(ctx.SenderUsername),
-    senderE164: normalizeOptionalString(ctx.SenderE164),
-    senderIsOwner,
-    inputProvenance: ctx.InputProvenance,
-    trustedInternalHandoff: undefined,
-    scheduledToolPolicy: undefined,
-    runtimePluginToolGrant: undefined,
-    toolsAllow: undefined,
-    disableTools: false,
-    traceAuthorized: senderIsOwner || (ctx.GatewayClientScopes ?? []).includes("operator.admin"),
-    approvalReviewerDeviceId: normalizeOptionalString(ctx.ApprovalReviewerDeviceId),
-    clientCaps: ctx.GatewayClientCaps,
-    toolBindings: ctx.GatewayRunToolBindings,
-  };
-}
-
 /** Captures the prepared request data used by both pre-ACK and detached injection attempts. */
 export function createChatSendMessageInjectionStarter(params: {
  target: ReplyMessageInjectionTarget | undefined;
@@ -107,6 +51,11 @@ export function createChatSendMessageInjectionStarter(params: {
      inlineMode: p.queueMode,
    });
    const text = ctx.BodyForAgent ?? ctx.Body ?? rawMessage;
+    const authorization = resolveCommandAuthorization({
+      ctx,
+      cfg,
+      commandAuthorized: ctx.CommandAuthorized === true,
+    });
    const attempt = beginReplyMessageInjectionTarget(
      params.target,
      p.replyToId
@@ -115,7 +64,12 @@ export function createChatSendMessageInjectionStarter(params: {
      {
        steeringMode: "all",
        isInboundUserMessage: true,
-        toolAuthorityOverlay: resolveChatSendToolAuthorityOverlay({ ctx, session: params.session }),
+        toolAuthorityOverlay: resolveInboundReplyToolAuthorityOverlay({
+          ctx,
+          sessionEntry: entry,
+          senderIsOwner: authorization.senderIsOwner,
+          disableTools: false,
+        }),
        ...(replyOptionImages?.length ? { images: replyOptionImages } : {}),
        ...(params.imageOrder?.length ? { imageOrder: params.imageOrder } : {}),
        ...(replyOptionMedia?.length ? { media: replyOptionMedia } : {}),
@@ -138,14 +92,9 @@ export async function settleChatSendPreAckMessageInjection(params: {
  attempt: ReplyMessageInjectionAttempt | undefined;
  isAborted: () => boolean;
  sessionRoutingChanged: () => boolean;
-  onActiveLeafChanged: () => Promise<void>;
  onAborted: () => void;
  onSessionRoutingChanged: () => void;
 }): Promise<PreAckMessageInjectionResult> {
-  if (params.attempt?.rejectBeforeAck) {
-    await params.onActiveLeafChanged();
-    return { status: "handled" };
-  }
  if (!params.attempt || (await params.attempt.acceptance)) {
    return { status: "continue", attempt: params.attempt };
  }
@@ -160,11 +109,11 @@ export async function settleChatSendPreAckMessageInjection(params: {
  return { status: "continue", attempt: undefined };
 }

-/** Finish an irrevocably accepted steer without entering reply dispatch. */
+/** Finish an accepted steer without entering reply dispatch, or return false for fallback. */
 export async function finalizeAcceptedChatSendMessageInjection(params: {
+  attempt: ReplyMessageInjectionAttempt;
  context: GatewayRequestContext;
  ctx: RuntimeMsgContext;
-  outcome: Extract<ReplyMessageInjectionOutcome, { status: "accepted" }>;
  persistUserTurnTranscriptBestEffort: () => Promise<void>;
  session: Pick<
    PreparedChatSendSession,
@@ -172,11 +121,18 @@ export async function finalizeAcceptedChatSendMessageInjection(params: {
  >;
  startedAt: number;
  target: ReplyMessageInjectionTarget;
-  targetRunId: string | undefined;
-}): Promise<void> {
-  const { context, ctx, outcome, session, target } = params;
+}): Promise<boolean> {
+  const { context, ctx, session } = params;
  const { agentId, cfg, clientRunId, entry, sessionKey, storePath } = session;
  const finalizedCtx = finalizeInboundContext(ctx);
+  const finalization = await finalizeReplyMessageInjectionAttempt({
+    attempt: params.attempt,
+    target: params.target,
+    inboundAudio: hasInboundAudio(finalizedCtx),
+  });
+  if (finalization.status === "rejected") {
+    return false;
+  }
  const channel = normalizeLowercaseStringOrEmpty(
    finalizedCtx.Surface ?? finalizedCtx.Provider ?? "unknown",
  );
@@ -186,17 +142,10 @@ export async function finalizeAcceptedChatSendMessageInjection(params: {
    finalizedCtx.MessageSid ??
    finalizedCtx.MessageSidFirst ??
    finalizedCtx.MessageSidLast;
-  recordAcceptedReplyMessageInjectionTarget(target, {
-    inboundAudio: hasInboundAudio(finalizedCtx),
-  });
-  // An unconfirmed transcript commit aborts the exact target without replay:
-  // the steer did not take effect, so diagnostics and audit must record the
-  // abort, not a completed injection.
-  const steerAborted = outcome.result?.transcriptCommit === "unconfirmed";
+  const steerAborted = finalization.aborted;
  if (steerAborted) {
-    abortReplyMessageInjectionTarget(target);
    context.logGateway.warn(
-      `active run ${params.targetRunId ?? "unknown"} accepted chat steering without transcript confirmation; aborted exact target without replay`,
+      `active run ${finalization.targetRunId ?? "unknown"} accepted chat steering without transcript confirmation; aborted exact target without replay`,
    );
  }
  await params.persistUserTurnTranscriptBestEffort();
@@ -260,4 +209,5 @@ export async function finalizeAcceptedChatSendMessageInjection(params: {
    });
    broadcastChatFinal({ context, runId: clientRunId, sessionKey, agentId });
  }
+  return true;
 }
@@ -145,7 +145,7 @@ const mockState = vi.hoisted(() => ({
  lastDispatchThinkingLevelOverride: undefined as string | undefined,
  lastDispatchOriginatingLeafEntryId: undefined as string | null | undefined,
  lastTaskSuggestionDeliveryMode: undefined as "gateway" | undefined,
-  lastMessageInjectionAttempted: undefined as true | undefined,
+  lastMessageInjectionDisposition: undefined as "none" | "accepted" | "rejected" | undefined,
  lastDispatchUserTurnInput: undefined as unknown,
  modelCatalog: null as ModelCatalogEntry[] | null,
  emittedTranscriptUpdates: [] as Array<{
@@ -372,7 +372,7 @@ dispatchInboundMessageMock.mockImplementation(
        imageOrder?: string[];
        thinkingLevelOverride?: string;
        taskSuggestionDeliveryMode?: "gateway";
-        messageInjectionAttempted?: true;
+        messageInjectionDisposition?: "none" | "accepted" | "rejected";
        turnAdoptionLifecycle?: {
          originatingLeafEntryId?: string | null;
        };
@@ -385,7 +385,7 @@ dispatchInboundMessageMock.mockImplementation(
      mockState.lastDispatchOriginatingLeafEntryId =
        params.replyOptions?.turnAdoptionLifecycle?.originatingLeafEntryId;
      mockState.lastTaskSuggestionDeliveryMode = params.replyOptions?.taskSuggestionDeliveryMode;
-      mockState.lastMessageInjectionAttempted = params.replyOptions?.messageInjectionAttempted;
+      mockState.lastMessageInjectionDisposition = params.replyOptions?.messageInjectionDisposition;
      await mockState.cronAuthorityProbe?.(
        params.replyOptions?.runId,
        params.replyOptions?.cronCreatorAuthorityCapability,
@@ -1332,7 +1332,7 @@ describe("chat directive tag stripping for non-streaming final payloads", () =>
    mockState.lastDispatchThinkingLevelOverride = undefined;
    mockState.lastDispatchOriginatingLeafEntryId = undefined;
    mockState.lastTaskSuggestionDeliveryMode = undefined;
-    mockState.lastMessageInjectionAttempted = undefined;
+    mockState.lastMessageInjectionDisposition = undefined;
    mockState.lastDispatchUserTurnInput = undefined;
    mockState.modelCatalog = null;
    mockState.emittedTranscriptUpdates = [];
@@ -1634,7 +1634,7 @@ describe("chat directive tag stripping for non-streaming final payloads", () =>
    expect(context.addChatRun).toHaveBeenCalledTimes(1);
  });

-  it("rejects targetless steer when no leaf-bound owner exists", async () => {
+  it("starts one normal turn when steer admission finds no direct owner", async () => {
    await createGatewayUserTurnSqliteFixture("openclaw-chat-send-stale-steer-no-owner-");
    await appendTranscriptMessage(transcriptScope(), {
      eventId: "current-leaf",
@@ -1653,15 +1653,17 @@ describe("chat directive tag stripping for non-streaming final payloads", () =>
      waitFor: "none",
    });

-    expect(lastRespondCall(respond)).toEqual([
-      false,
+    expect(respond).toHaveBeenCalledWith(
+      true,
+      expect.objectContaining({ status: "started" }),
      undefined,
-      expect.objectContaining({ details: { reason: "active-leaf-changed" } }),
-    ]);
-    expect(context.addChatRun).not.toHaveBeenCalled();
+      expect.any(Object),
+    );
+    await waitForAssertion(() => expect(mockState.lastDispatchCtx?.BodyForAgent).toBe("hello"));
+    expect(context.addChatRun).toHaveBeenCalledOnce();
  });

-  it("rejects targetless steer without a supplied leaf before dispatch", async () => {
+  it("injects targetless steer into the exact direct owner without a client run id", async () => {
    await createGatewayUserTurnSqliteFixture("openclaw-chat-send-targetless-no-leaf-");
    const { context, respond, send } = createChatRequestFixture();
    const queueMessage = vi.fn(async () => {});
@@ -1688,16 +1690,118 @@ describe("chat directive tag stripping for non-streaming final payloads", () =>
      operation.complete();
    }

-    expect(lastRespondCall(respond)).toEqual([
-      false,
+    expect(respond).toHaveBeenCalledWith(
+      true,
+      expect.objectContaining({ status: "started" }),
      undefined,
-      expect.objectContaining({ details: { reason: "active-leaf-changed" } }),
-    ]);
-    expect(queueMessage).not.toHaveBeenCalled();
-    expect(context.addChatRun).not.toHaveBeenCalled();
+      expect.any(Object),
+    );
+    expect(queueMessage).toHaveBeenCalledOnce();
+    expect(context.addChatRun).toHaveBeenCalledOnce();
    expect(mockState.lastDispatchCtx).toBeUndefined();
  });

+  it.each([
+    ["active descendant", "agent:main:main:subagent:child", "descendant"],
+    ["different session", "agent:main:telegram:direct:other", "other"],
+  ])(
+    "starts the selected idle session despite %s activity",
+    async (_name, activeSessionKey, slug) => {
+      await createGatewayUserTurnSqliteFixture(`openclaw-chat-send-idle-${slug}-`);
+      const { context, respond, send } = createChatRequestFixture();
+      const unrelated = replyRunRegistry.begin({
+        sessionKey: activeSessionKey,
+        sessionId: `${mockState.sessionId}-${slug}`,
+        resetTriggered: false,
+      });
+      unrelated.setPhase("running");
+
+      try {
+        await send({
+          idempotencyKey: `idem-idle-${slug}`,
+          requestParams: { queueMode: "steer" },
+        });
+      } finally {
+        unrelated.complete();
+      }
+
+      expect(respond).toHaveBeenCalledWith(
+        true,
+        expect.objectContaining({ status: "started" }),
+        undefined,
+        expect.any(Object),
+      );
+      expect(context.addChatRun).toHaveBeenCalledOnce();
+      expect(unrelated.result).toEqual({ kind: "completed" });
+      expect(mockState.lastMessageInjectionDisposition).toBe("rejected");
+    },
+  );
+
+  it("falls back visibly when the direct owner has a non-injectable CLI backend", async () => {
+    await createGatewayUserTurnSqliteFixture("openclaw-chat-send-cli-steer-fallback-");
+    const { context, respond, send } = createChatRequestFixture();
+    const operation = replyRunRegistry.begin({
+      sessionKey: "agent:main:main",
+      sessionId: mockState.sessionId,
+      resetTriggered: false,
+    });
+    operation.setPhase("running");
+    operation.attachBackend({ kind: "cli", runId: "cli-run", cancel: vi.fn() });
+
+    try {
+      await send({
+        idempotencyKey: "idem-cli-steer-fallback",
+        requestParams: { queueMode: "steer" },
+      });
+    } finally {
+      operation.complete();
+    }
+
+    expect(respond).toHaveBeenCalledWith(
+      true,
+      expect.objectContaining({ status: "started" }),
+      undefined,
+      expect.any(Object),
+    );
+    expect(context.addChatRun).toHaveBeenCalledOnce();
+    expect(mockState.lastDispatchCtx?.BodyForAgent).toBe("hello");
+    expect(mockState.lastMessageInjectionDisposition).toBe("rejected");
+  });
+
+  it("deduplicates a retried targetless steer before a second injection", async () => {
+    await createGatewayUserTurnSqliteFixture("openclaw-chat-send-steer-idempotent-");
+    const { context, send } = createChatRequestFixture();
+    const queueMessage = vi.fn(async () => {});
+    const operation = replyRunRegistry.begin({
+      sessionKey: "agent:main:main",
+      sessionId: mockState.sessionId,
+      resetTriggered: false,
+    });
+    bindTestToolAuthority(operation);
+    operation.setPhase("running");
+    operation.attachBackend({
+      kind: "embedded",
+      cancel: vi.fn(),
+      messageInjection: { isAvailable: () => true, queueMessage },
+    });
+
+    try {
+      await send({
+        idempotencyKey: "idem-steer-idempotent",
+        requestParams: { queueMode: "steer" },
+      });
+      await send({
+        idempotencyKey: "idem-steer-idempotent",
+        requestParams: { queueMode: "steer" },
+      });
+    } finally {
+      operation.complete();
+    }
+
+    expect(queueMessage).toHaveBeenCalledOnce();
+    expect(context.addChatRun).toHaveBeenCalledOnce();
+  });
+
  it("injects a matching leaf-bound targetless steer through the legacy backend seam", async () => {
    await createGatewayUserTurnSqliteFixture("openclaw-chat-send-targetless-steer-");
    await appendTranscriptMessage(transcriptScope(), {
@@ -1809,51 +1913,7 @@ describe("chat directive tag stripping for non-streaming final payloads", () =>
    }
  });

-  it("rejects targetless steer when the owner immutable leaf differs", async () => {
-    await createGatewayUserTurnSqliteFixture("openclaw-chat-send-targetless-leaf-mismatch-");
-    await appendTranscriptMessage(transcriptScope(), {
-      eventId: "current-leaf",
-      message: { role: "assistant", content: "working" },
-      now: 1,
-      parentId: null,
-    });
-    const { context, respond, send } = createChatRequestFixture();
-    const queueMessage = vi.fn(async () => {});
-    const operation = replyRunRegistry.begin({
-      sessionKey: "agent:main:main",
-      sessionId: mockState.sessionId,
-      resetTriggered: false,
-      originatingLeafEntryId: "different-owner-leaf",
-    });
-    bindTestToolAuthority(operation);
-    operation.setPhase("running");
-    operation.attachBackend({
-      kind: "embedded",
-      cancel: () => {},
-      messageInjection: { isAvailable: () => true, queueMessage },
-    });
-
-    try {
-      await send({
-        idempotencyKey: "idem-targetless-leaf-mismatch",
-        requestParams: { expectedLeafEntryId: "current-leaf", queueMode: "steer" },
-        waitFor: "none",
-      });
-    } finally {
-      operation.complete();
-    }
-
-    expect(lastRespondCall(respond)).toEqual([
-      false,
-      undefined,
-      expect.objectContaining({ details: { reason: "active-leaf-changed" } }),
-    ]);
-    expect(queueMessage).not.toHaveBeenCalled();
-    expect(context.addChatRun).not.toHaveBeenCalled();
-    expect(mockState.lastDispatchCtx).toBeUndefined();
-  });
-
-  it("rejects a captured targetless steer when a successor replaces its operation", async () => {
+  it("falls back once without touching a successor when the captured owner ends", async () => {
    await createGatewayUserTurnSqliteFixture("openclaw-chat-send-targetless-operation-aba-");
    await appendTranscriptMessage(transcriptScope(), {
      eventId: "current-leaf",
@@ -1864,6 +1924,8 @@ describe("chat directive tag stripping for non-streaming final payloads", () =>
    const { context, respond } = createChatRequestFixture();
    const originalQueue = vi.fn(async () => {});
    const successorQueue = vi.fn(async () => {});
+    const successorCancel = vi.fn();
+    const dispatchCallsBefore = dispatchInboundMessageMock.mock.calls.length;
    const original = replyRunRegistry.begin({
      sessionKey: "agent:main:main",
      sessionId: mockState.sessionId,
@@ -1917,7 +1979,7 @@ describe("chat directive tag stripping for non-streaming final payloads", () =>
          successor.setPhase("running");
          successor.attachBackend({
            kind: "embedded",
-            cancel: () => {},
+            cancel: successorCancel,
            messageInjection: { isAvailable: () => true, queueMessage: successorQueue },
          });
          return true;
@@ -1928,15 +1990,20 @@ describe("chat directive tag stripping for non-streaming final payloads", () =>
      successor?.complete();
    }

-    expect(lastRespondCall(respond)).toEqual([
-      false,
+    expect(respond).toHaveBeenCalledWith(
+      true,
+      expect.objectContaining({ status: "started" }),
      undefined,
-      expect.objectContaining({ details: { reason: "active-leaf-changed" } }),
-    ]);
+      expect.any(Object),
+    );
+    await waitForAssertion(() =>
+      expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(dispatchCallsBefore + 1),
+    );
    expect(originalQueue).not.toHaveBeenCalled();
    expect(successorQueue).not.toHaveBeenCalled();
-    expect(context.addChatRun).not.toHaveBeenCalled();
-    expect(mockState.lastDispatchCtx).toBeUndefined();
+    expect(successorCancel).not.toHaveBeenCalled();
+    expect(context.addChatRun).toHaveBeenCalledOnce();
+    expect(mockState.lastMessageInjectionDisposition).toBe("rejected");
  });

  it("allows an exact-run steer after the active transcript leaf advances", async () => {
@@ -2350,7 +2417,7 @@ describe("chat directive tag stripping for non-streaming final payloads", () =>
    expect(successorCancel).not.toHaveBeenCalled();
    expect(mockState.replyContextCalls).toBe(1);
    expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(dispatchCallsBefore + 1);
-    expect(mockState.lastMessageInjectionAttempted).toBe(true);
+    expect(mockState.lastMessageInjectionDisposition).toBe("rejected");
    expect(readPersistedUserMessages()).toHaveLength(1);
    expect(
      (readPersistedUserMessages()[0]?.["__openclaw"] as Record<string, unknown> | undefined)
@@ -2618,7 +2685,7 @@ describe("chat directive tag stripping for non-streaming final payloads", () =>
    );
    expect(context.addChatRun).toHaveBeenCalledOnce();
    expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(dispatchCallsBefore + 1);
-    expect(mockState.lastMessageInjectionAttempted).toBe(true);
+    expect(mockState.lastMessageInjectionDisposition).toBe("rejected");
    expect(successorQueue).not.toHaveBeenCalled();
    expect(successorCancel).not.toHaveBeenCalled();
    expect(readPersistedUserMessages()).toHaveLength(1);
@@ -2688,7 +2755,7 @@ describe("chat directive tag stripping for non-streaming final payloads", () =>
    );
    expect(context.addChatRun).toHaveBeenCalledOnce();
    expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(dispatchCallsBefore + 1);
-    expect(mockState.lastMessageInjectionAttempted).toBe(true);
+    expect(mockState.lastMessageInjectionDisposition).toBe("rejected");
    expect(staleQueue).not.toHaveBeenCalled();
    expect(staleCancel).not.toHaveBeenCalled();
    expect(readPersistedUserMessages()).toHaveLength(1);