refactor(gateway): extract agent turn service (#121215)

2026-08-12 21:53:00 -06:00 · 2026-08-09 12:45:38 -07:00
parent 1d271907f3
commit 0b1fc0e279
10 changed files with 839 additions and 696 deletions
@@ -334,7 +334,7 @@ describe("spawnSubagentDirect in-process Gateway collector launch", () => {

        const externalRespond = vi.fn();
        const externalPreflight = prepareAgentRequestPreflight({
-          params,
+          request: params,
          io: createAgentTurnIo(externalRespond),
          context: gatewayContext,
          client: externalCliClient(),
@@ -345,7 +345,7 @@ describe("spawnSubagentDirect in-process Gateway collector launch", () => {
          ? createSyntheticPluginRuntimeClient({ scopes: options.syntheticScopes })
          : externalCliClient();
        const hostPreflight = prepareAgentRequestPreflight({
-          params,
+          request: params,
          io: createAgentTurnIo(hostRespond),
          context: gatewayContext,
          client,
@@ -0,0 +1,683 @@
+import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
+import {
+  ErrorCodes,
+  errorShape,
+  type AgentWaitParams,
+} from "../../../packages/gateway-protocol/src/index.js";
+import { scheduleMainSessionRecoveryPendingTarget } from "../../agents/main-session-recovery-owner-release.js";
+import {
+  releaseMainSessionRecoveryOwner,
+  type MainSessionRecoveryOwnerLease,
+} from "../../agents/main-session-recovery-store.js";
+import { mergeSessionEntry, type SessionEntry } from "../../config/sessions.js";
+import type { OpenClawConfig } from "../../config/types.openclaw.js";
+import { getAgentEventLifecycleGeneration } from "../../infra/agent-events.js";
+import { normalizeDeliveryContext } from "../../utils/delivery-context.shared.js";
+import { createAgentAdmissionController } from "../server-methods/agent-admission-controller.js";
+import { prepareAgentContentPhase } from "../server-methods/agent-content-phase.js";
+import { createCronContinuationController } from "../server-methods/agent-cron-continuation.js";
+import { createAgentDedupeLifecycle } from "../server-methods/agent-dedupe-lifecycle.js";
+import {
+  isAcceptedAgentDedupePayload,
+  readGatewayDedupeEntry,
+} from "../server-methods/agent-dedupe.js";
+import { resolveAgentDeliveryPhase } from "../server-methods/agent-delivery-phase.js";
+import type { RestoredCronContinuation } from "../server-methods/agent-handler-helpers.js";
+import { waitForAgentJob } from "../server-methods/agent-job.js";
+import type { AgentRequestPreflight } from "../server-methods/agent-request-preflight.js";
+import { prepareAgentRequestRouting } from "../server-methods/agent-request-routing.js";
+import { runAgentResetPhase } from "../server-methods/agent-reset-phase.js";
+import { prepareAgentRunDispatch } from "../server-methods/agent-run-admission-phase.js";
+import { startAgentRunExecution } from "../server-methods/agent-run-execution-phase.js";
+import { buildAgentSessionPatch } from "../server-methods/agent-session-patch.js";
+import { persistAgentSessionPhase } from "../server-methods/agent-session-persist.js";
+import { prepareAgentSession } from "../server-methods/agent-session-prepare.js";
+import { handleChatAbortRequest } from "../server-methods/chat-abort-handler.js";
+import { resolveAgentRunSessionCreation } from "../server-methods/session-creation-provenance.js";
+import type { GatewayRequestHandlerOptions, RespondFn } from "../server-methods/shared-types.js";
+import { authorizeResolvedSessionMutation } from "../session-sharing.js";
+import { formatForLog } from "../ws-log.js";
+import type { AgentTurnIo, AgentTurnPrincipal } from "./types.js";
+
+type AgentTurnStartRequest = {
+  preflight: AgentRequestPreflight;
+  principal: AgentTurnPrincipal | null;
+  io: AgentTurnIo;
+  onRunObserved?: (runId: string) => void;
+};
+
+function createAcceptanceRespond(io: AgentTurnIo): RespondFn {
+  return (ok, payload, error, meta) => io.emitAcceptance([ok, payload, error], meta);
+}
+
+function replayAgentTurnIfCached(params: {
+  preflight: AgentRequestPreflight;
+  context: GatewayRequestHandlerOptions["context"];
+  io: AgentTurnIo;
+}): boolean {
+  const { agentDedupeKeys, runId } = params.preflight;
+  const cached = readGatewayDedupeEntry({
+    dedupe: params.context.dedupe,
+    keys: agentDedupeKeys,
+  });
+  if (!cached) {
+    return false;
+  }
+  if (cached.ok && isAcceptedAgentDedupePayload(cached.payload)) {
+    const cachedRunId =
+      typeof cached.payload.runId === "string" && cached.payload.runId.trim()
+        ? cached.payload.runId.trim()
+        : runId;
+    const cachedSessionKey =
+      typeof cached.payload.sessionKey === "string" && cached.payload.sessionKey.trim()
+        ? cached.payload.sessionKey.trim()
+        : undefined;
+    const cachedAgentId =
+      cachedSessionKey === "global" &&
+      typeof cached.payload.agentId === "string" &&
+      cached.payload.agentId.trim()
+        ? cached.payload.agentId.trim()
+        : undefined;
+    params.io.emitAcceptance(
+      [
+        true,
+        {
+          runId: cachedRunId,
+          status: "in_flight" as const,
+          ...(cachedSessionKey ? { sessionKey: cachedSessionKey } : {}),
+          ...(cachedAgentId ? { agentId: cachedAgentId } : {}),
+        },
+        undefined,
+      ],
+      { cached: true, runId: cachedRunId },
+    );
+  } else {
+    params.io.emitAcceptance([cached.ok, cached.payload, cached.error], { cached: true });
+  }
+  return true;
+}
+
+export function createAgentTurnService({
+  context,
+  isWebchatConnect,
+}: Pick<GatewayRequestHandlerOptions, "context" | "isWebchatConnect">) {
+  const startTurn = async ({
+    preflight,
+    principal,
+    io,
+    onRunObserved,
+  }: AgentTurnStartRequest): Promise<void> => {
+    if (replayAgentTurnIfCached({ preflight, context, io })) {
+      return;
+    }
+    const respond = createAcceptanceRespond(io);
+    const {
+      request,
+      cfg,
+      runId,
+      allowModelOverride,
+      canUseInternalRuntimeHandoff,
+      canUseCronRunContinuation,
+      expectedSession,
+      expectedExistingSessionId,
+      providerOverride,
+      modelOverride,
+      execApprovalFollowupApprovalId,
+      normalizedSpawned,
+      inputProvenance,
+      isRestartRecoveryResumeRun,
+      preserveUserFacingSessionModelState,
+      sessionEffects,
+      suppressVisibleSessionEffects,
+      requestedPromptPersistenceSuppression,
+      isOneShotModelRun,
+      isRawModelRun,
+      agentDedupeKeys,
+    } = preflight;
+    // Cached replay returns before a new lifecycle generation is observed, matching
+    // the idempotency path that preceded this service extraction.
+    const lifecycleGeneration = getAgentEventLifecycleGeneration();
+    const idem = runId;
+    let resolvedGroupId: string | undefined = normalizedSpawned.groupId;
+    let resolvedGroupChannel: string | undefined = normalizedSpawned.groupChannel;
+    let resolvedGroupSpace: string | undefined = normalizedSpawned.groupSpace;
+    let spawnedByValue: string | undefined;
+    const ownerConnId = typeof principal?.connId === "string" ? principal.connId : undefined;
+    const ownerDeviceId =
+      typeof principal?.connect?.device?.id === "string" ? principal.connect.device.id : undefined;
+    const dedupeLifecycle = createAgentDedupeLifecycle({
+      cfg,
+      request,
+      runId,
+      lifecycleGeneration,
+      agentDedupeKeys,
+      suppressVisibleSessionEffects,
+      ownerConnId,
+      ownerDeviceId,
+      context,
+      io,
+    });
+    const reservePreAcceptedAgentDedupe = dedupeLifecycle.reserve;
+    const clearUnacceptedAgentDedupe = dedupeLifecycle.clearUnaccepted;
+    const abortForLifecycleRotation = dedupeLifecycle.abortForLifecycleRotation;
+    const routing = await prepareAgentRequestRouting({
+      request,
+      cfg,
+      expectedSession,
+      isRawModelRun,
+      execApprovalFollowupApprovalId,
+      runId,
+      agentDedupeKeys,
+      context,
+      respond,
+      reserveDedupe: reservePreAcceptedAgentDedupe,
+      clearDedupe: clearUnacceptedAgentDedupe,
+    });
+    if (!routing) {
+      return;
+    }
+    const {
+      normalizedAttachments,
+      requestedBestEffortDeliver,
+      knownAgents,
+      requestedSessionId,
+      requestedToRaw,
+      sessionKeyFromTo,
+      requestedSessionKeyRaw,
+      explicitRecipientSession,
+      preAcceptedReservedSessionKey,
+      preAttachmentSession,
+    } = routing;
+    let agentId = routing.agentId;
+    let requestedSessionKey = routing.requestedSessionKey;
+    let gatewayAdmissionTransferred = false;
+    let mainRestartRecoveryOwnerLease: MainSessionRecoveryOwnerLease | undefined;
+    let releaseGatewayAdmission = () => {};
+    const cronContinuation = createCronContinuationController({
+      runId,
+      lifecycleGeneration,
+      context,
+    });
+    const releaseCronContinuationClaimWithRecovery = cronContinuation.releaseWithRecovery;
+    try {
+      const content = await prepareAgentContentPhase({
+        request,
+        cfg,
+        context,
+        respond,
+        isRawModelRun,
+        inputProvenance,
+        normalizedAttachments,
+        requestedSessionKeyRaw,
+        requestedSessionKey,
+        requestedSessionId,
+        requestedToRaw,
+        sessionKeyFromTo,
+        agentId,
+        providerOverride,
+        modelOverride,
+        explicitRecipientSession,
+        knownAgents,
+      });
+      if (!content) {
+        return;
+      }
+      agentId = content.agentId;
+      requestedSessionKey = content.requestedSessionKey;
+      // Participation is authorized below against the canonical session the run
+      // actually targets (see prepareAgentSession). A keyless request resolves its
+      // default/effective session there, so authorizing only an explicit key here
+      // would let a non-member drive a restricted default session.
+      let effectiveTranscriptInputText = content.effectiveTranscriptInputText;
+      let message = content.message;
+      const {
+        images,
+        imageOrder,
+        media,
+        replyTo,
+        recipientChannel,
+        recipientAccountId,
+        recipientThreadId,
+        to,
+      } = content;
+      let resolvedSessionId = requestedSessionId;
+      let sessionEntry: SessionEntry | undefined;
+      let effectiveBootstrapContextRunKind = request.bootstrapContextRunKind;
+      let restoredCronContinuation: RestoredCronContinuation | undefined;
+      let restoredCronContinuationIdentity:
+        | Pick<RestoredCronContinuation, "lifecycleRevision" | "sessionId">
+        | undefined;
+      let sessionPersistedBeforeGatewayAdmission = false;
+      let bestEffortDeliver = requestedBestEffortDeliver ?? false;
+      let cfgForAgent: OpenClawConfig | undefined;
+      let resolvedSessionKey = requestedSessionKey;
+      let resolvedSessionAgentId: string | undefined;
+      let isNewSession = false;
+      let supersededSessionId: string | undefined;
+      let skipAgentInitialSessionTouch = false;
+      let pendingChatRun: { sessionKey: string; agentId?: string } | undefined;
+      let admittedSessionId = resolvedSessionId ?? runId;
+      const admissionController = createAgentAdmissionController({
+        cfg,
+        runId,
+        lifecycleGeneration,
+        agentDedupeKeys,
+        preAcceptedReservedSessionKey,
+        expectedSession,
+        context,
+        io,
+        dedupeLifecycle,
+        getRequestedSessionKey: () => requestedSessionKey,
+        getResolvedSessionKey: () => resolvedSessionKey,
+        getResolvedSessionId: () => resolvedSessionId,
+        getResolvedSessionAgentId: () => resolvedSessionAgentId,
+        getAgentId: () => agentId,
+        getCfgForAgent: () => cfgForAgent,
+        getSessionPersisted: () => sessionPersistedBeforeGatewayAdmission,
+        getSupersededSessionId: () => supersededSessionId,
+        setAdmittedSessionId: (sessionId) => {
+          admittedSessionId = sessionId;
+        },
+      });
+      const admissionAgentId = admissionController.admissionAgentId;
+      const assertGatewayWorkAdmissionAllowed = admissionController.assertAllowed;
+      const acquireGatewayWorkAdmission = admissionController.acquire;
+      const respondToGatewayAdmissionOutcome = admissionController.respondToOutcome;
+      releaseGatewayAdmission = admissionController.release;
+      const resetPhase = await runAgentResetPhase({
+        request,
+        cfg,
+        requestedSessionKey,
+        resolvedSessionId,
+        effectiveTranscriptInputText,
+        message,
+        agentId,
+        sessionKeyFromTo,
+        lifecycleGeneration,
+        runId,
+        agentDedupeKeys,
+        client: principal,
+        context,
+        respond,
+        abortForLifecycleRotation,
+        setCommittedResetCompletion: dedupeLifecycle.setCommittedResetCompletion,
+      });
+      requestedSessionKey = resetPhase.requestedSessionKey;
+      resolvedSessionId = resetPhase.resolvedSessionId;
+      effectiveTranscriptInputText = resetPhase.effectiveTranscriptInputText;
+      message = resetPhase.message;
+      if (resetPhase.accepted) {
+        dedupeLifecycle.markAccepted(true);
+      }
+      if (resetPhase.stop) {
+        return;
+      }
+
+      if (requestedSessionKey) {
+        const preparedSession = prepareAgentSession({
+          requestedSessionKey,
+          requestedSessionId,
+          expectedExistingSessionId,
+          agentId,
+          recipientChannel,
+          request,
+          canUseCronRunContinuation,
+          lifecycleGeneration,
+          effectiveBootstrapContextRunKind,
+          preAttachmentSession,
+          respond,
+        });
+        if (!preparedSession) {
+          return;
+        }
+        const {
+          cfg: cfgLocal,
+          storePath,
+          entry,
+          canonicalKey,
+          storeKeys,
+          maintenanceConfig: sessionMaintenanceConfig,
+          canonicalSessionAgentId,
+          resetPolicy,
+          now,
+          visibleRequest,
+          mainSessionKey: mainSessionKeyForRequest,
+          isSystemGatewayRun,
+          sessionId,
+          touchInteraction,
+          failedSessionTranscriptMissing: resolveFailedSessionTranscriptMissingForEntry,
+        } = preparedSession;
+        cfgForAgent = cfgLocal;
+        // Authorize the canonical session the run will actually target — covering
+        // keyless requests whose default/effective session is resolved only here —
+        // before any run side effects (admission, dispatch).
+        const sharingError = authorizeResolvedSessionMutation({
+          cfg: cfgLocal,
+          client: principal,
+          sessionKey: canonicalKey,
+          agentId: canonicalSessionAgentId,
+        });
+        if (sharingError) {
+          io.emitAcceptance([false, undefined, sharingError]);
+          return;
+        }
+        effectiveBootstrapContextRunKind = preparedSession.effectiveBootstrapContextRunKind;
+        restoredCronContinuationIdentity = preparedSession.restoredCronContinuationIdentity;
+        sessionPersistedBeforeGatewayAdmission =
+          preparedSession.sessionPersistedBeforeGatewayAdmission;
+        isNewSession = preparedSession.isNewSession;
+        const sessionAgent = canonicalSessionAgentId;
+        const requestDeliveryHint = normalizeDeliveryContext({
+          channel: recipientChannel?.trim(),
+          to,
+          accountId: recipientAccountId?.trim(),
+          // Pass threadId directly — normalizeDeliveryContext handles both
+          // string and numeric threadIds (e.g., Matrix uses integers).
+          threadId: recipientThreadId,
+        });
+        const buildSessionPatch = (freshEntry: SessionEntry | undefined) =>
+          buildAgentSessionPatch({
+            freshEntry,
+            initialEntry: entry,
+            cfg: cfgLocal,
+            sessionAgentId: sessionAgent,
+            canonicalSessionKey: canonicalKey,
+            storePath,
+            normalizedSpawned,
+            requestDeliveryHint,
+            requestLabel: request.label,
+            pluginOwnerId:
+              freshEntry === undefined
+                ? normalizeOptionalString(principal?.internal?.pluginRuntimeOwnerId)
+                : undefined,
+            expectedExistingSessionId,
+            hasRestoredCronContinuation: restoredCronContinuationIdentity !== undefined,
+            resetPolicy,
+            now,
+            requestedSessionId,
+            isSystemGatewayRun,
+            visibleRequest,
+            fallbackSessionId: sessionId,
+            touchInteraction,
+            failedSessionTranscriptMissing: resolveFailedSessionTranscriptMissingForEntry,
+          });
+        const patchBuild = buildSessionPatch(entry);
+        isNewSession = patchBuild.isNewSession;
+        sessionEntry = mergeSessionEntry(entry, patchBuild.patch);
+        resolvedSessionId = sessionEntry?.sessionId ?? sessionId;
+        admittedSessionId = resolvedSessionId ?? runId;
+        const canonicalSessionKey = canonicalKey;
+        resolvedSessionKey = canonicalSessionKey;
+        const sessionAgentId = canonicalSessionAgentId;
+        resolvedSessionAgentId = sessionAgentId;
+        const mainSessionKey = mainSessionKeyForRequest;
+        try {
+          await acquireGatewayWorkAdmission(storePath ?? `agent:${sessionAgentId}`);
+        } catch (err) {
+          io.emitAcceptance([
+            false,
+            undefined,
+            errorShape(ErrorCodes.INVALID_REQUEST, formatForLog(err)),
+          ]);
+          return;
+        }
+        if (respondToGatewayAdmissionOutcome()) {
+          return;
+        }
+        const persistedSession = await persistAgentSessionPhase({
+          request,
+          cfg: cfgLocal,
+          storePath,
+          storeKeys,
+          entry,
+          canonicalSessionKey,
+          sessionAgentId,
+          mainSessionKey,
+          creation: resolveAgentRunSessionCreation(principal),
+          lifecycleGeneration,
+          isRestartRecoveryResumeRun,
+          runId,
+          agentId,
+          suppressVisibleSessionEffects,
+          restoredCronContinuationIdentity,
+          initialPatchBuild: patchBuild,
+          buildSessionPatch,
+          initialSessionEntry: sessionEntry,
+          initialResolvedSessionId: resolvedSessionId,
+          initialSessionPersistedBeforeGatewayAdmission: sessionPersistedBeforeGatewayAdmission,
+          initialSupersededSessionId: supersededSessionId,
+          touchInteraction,
+          requestedBestEffortDeliver,
+          bestEffortDeliver,
+          expectedSession,
+          maintenanceConfig: sessionMaintenanceConfig,
+          abortForLifecycleRotation,
+          assertGatewayWorkAdmissionAllowed,
+          respondToGatewayAdmissionOutcome,
+          updateAdmissionState: (state) => {
+            resolvedSessionId = state.resolvedSessionId;
+            admittedSessionId = state.admittedSessionId;
+            supersededSessionId = state.supersededSessionId;
+            sessionPersistedBeforeGatewayAdmission = state.sessionPersistedBeforeGatewayAdmission;
+          },
+          getAdmittedSessionId: () => admittedSessionId,
+          setCronContinuationClaim: cronContinuation.setClaim,
+          setMainRestartRecoveryOwnerLease: (lease) => {
+            mainRestartRecoveryOwnerLease = lease;
+          },
+          respond,
+        });
+        if (!persistedSession) {
+          return;
+        }
+        sessionEntry = persistedSession.sessionEntry;
+        resolvedSessionId = persistedSession.resolvedSessionId;
+        sessionPersistedBeforeGatewayAdmission =
+          persistedSession.sessionPersistedBeforeGatewayAdmission;
+        supersededSessionId = persistedSession.supersededSessionId;
+        admittedSessionId = persistedSession.admittedSessionId;
+        skipAgentInitialSessionTouch = persistedSession.skipAgentInitialSessionTouch;
+        isNewSession = persistedSession.isNewSession;
+        spawnedByValue = persistedSession.spawnedBy;
+        resolvedGroupId = persistedSession.groupId;
+        resolvedGroupChannel = persistedSession.groupChannel;
+        resolvedGroupSpace = persistedSession.groupSpace;
+        pendingChatRun = persistedSession.pendingChatRun;
+        bestEffortDeliver = persistedSession.bestEffortDeliver;
+        restoredCronContinuation = persistedSession.restoredCronContinuation;
+      }
+
+      const delivery = await resolveAgentDeliveryPhase({
+        request,
+        cfg,
+        cfgForAgent,
+        sessionEntry,
+        resolvedSessionKey,
+        resolvedSessionAgentId,
+        agentId,
+        replyTo,
+        to,
+        recipientChannel,
+        recipientAccountId,
+        recipientThreadId,
+        bestEffortDeliver,
+        runId,
+        client: principal,
+        context,
+        respond,
+        isWebchatConnect,
+        onRunObserved,
+      });
+      if (!delivery) {
+        return;
+      }
+      const { activeSessionAgentId } = delivery;
+
+      const preparedDispatch = await prepareAgentRunDispatch({
+        request,
+        cfg,
+        cfgForAgent,
+        sessionEntry,
+        resolvedSessionKey,
+        requestedSessionKey,
+        preAcceptedReservedSessionKey,
+        activeSessionAgentId,
+        delivery,
+        restoredCronContinuationIdentity,
+        restoredCronContinuation,
+        providerOverride,
+        modelOverride,
+        allowModelOverride,
+        lifecycleGeneration,
+        getAdmittedSessionId: () => admittedSessionId,
+        ownerConnId,
+        ownerDeviceId,
+        suppressVisibleSessionEffects,
+        pendingChatRun,
+        inputProvenance,
+        isOneShotModelRun,
+        isRestartRecoveryResumeRun,
+        runId,
+        agentDedupeKeys,
+        context,
+        client: principal,
+        io,
+        abortForLifecycleRotation,
+        acquireGatewayWorkAdmission,
+        assertGatewayWorkAdmissionAllowed,
+        hasGatewayAdmissionOutcome: admissionController.hasOutcome,
+        respondToGatewayAdmissionOutcome,
+        admissionAgentId,
+        getGatewayWorkAdmission: admissionController.getAdmission,
+        setAdmittedRunAbort: admissionController.setAdmittedRunAbort,
+        getAdmittedRunAbort: admissionController.getAdmittedRunAbort,
+        markAgentRunAccepted: dedupeLifecycle.markAccepted,
+      });
+      if (!preparedDispatch) {
+        return;
+      }
+      resolvedSessionId = admittedSessionId;
+      gatewayAdmissionTransferred = true;
+      // This captures ambient root admission synchronously, then settles the final
+      // frame on the existing detached chain after the router returns its acceptance.
+      startAgentRunExecution({
+        prepared: preparedDispatch,
+        mainRestartRecoveryOwnerLease,
+        request,
+        cfg,
+        cfgForAgent,
+        sessionEntry,
+        resolvedSessionKey,
+        requestedSessionKey,
+        requestedSessionKeyRaw,
+        resolvedSessionId,
+        agentId,
+        activeSessionAgentId,
+        delivery,
+        isNewSession,
+        isRawModelRun,
+        isOneShotModelRun,
+        isRestartRecoveryResumeRun,
+        suppressVisibleSessionEffects,
+        message,
+        images,
+        imageOrder,
+        media,
+        effectiveTranscriptInputText,
+        inputProvenance,
+        runId,
+        idempotencyKey: idem,
+        agentDedupeKeys,
+        spawnedBy: spawnedByValue,
+        groupId: resolvedGroupId,
+        groupChannel: resolvedGroupChannel,
+        groupSpace: resolvedGroupSpace,
+        bestEffortDeliver,
+        lifecycleGeneration,
+        effectiveBootstrapContextRunKind,
+        requestedPromptPersistenceSuppression,
+        preserveUserFacingSessionModelState,
+        sessionEffects,
+        skipAgentInitialSessionTouch,
+        restoredCronContinuation,
+        canUseInternalRuntimeHandoff,
+        execApprovalFollowupApprovalId,
+        client: principal,
+        context,
+        io,
+        releaseCronContinuationClaimWithRecovery,
+      });
+      mainRestartRecoveryOwnerLease = undefined;
+    } finally {
+      try {
+        if (!gatewayAdmissionTransferred) {
+          let pendingRecovery: Awaited<ReturnType<typeof releaseMainSessionRecoveryOwner>> =
+            undefined;
+          try {
+            pendingRecovery = await releaseMainSessionRecoveryOwner(mainRestartRecoveryOwnerLease);
+          } finally {
+            try {
+              releaseGatewayAdmission();
+            } finally {
+              try {
+                await releaseCronContinuationClaimWithRecovery();
+              } finally {
+                scheduleMainSessionRecoveryPendingTarget(pendingRecovery);
+              }
+            }
+          }
+        }
+      } finally {
+        clearUnacceptedAgentDedupe();
+      }
+    }
+  };
+
+  const waitForTurn = async (params: AgentWaitParams) => {
+    const runId = (params.runId ?? "").trim();
+    const timeoutMs =
+      typeof params.timeoutMs === "number" && Number.isFinite(params.timeoutMs)
+        ? Math.max(0, Math.floor(params.timeoutMs))
+        : 30_000;
+    // Keep the captured entry across the wait so timeout attribution uses the
+    // same owner snapshot that selected the chat-vs-agent observation source.
+    const activeChatEntry = context.chatAbortControllers.get(runId);
+    const hasActiveChatRun = activeChatEntry !== undefined && activeChatEntry.kind !== "agent";
+    const snapshot = await waitForAgentJob({
+      runId,
+      timeoutMs,
+      ...(hasActiveChatRun ? { source: "chat" } : {}),
+    });
+    if (!snapshot) {
+      const activeRunRegistered = activeChatEntry !== undefined;
+      return {
+        runId,
+        status: "timeout" as const,
+        timeoutPhase: activeRunRegistered ? ("gateway_draining" as const) : ("queue" as const),
+        ...(activeRunRegistered ? {} : { providerStarted: false }),
+      };
+    }
+    return {
+      runId,
+      status: snapshot.status,
+      startedAt: snapshot.startedAt,
+      endedAt: snapshot.endedAt,
+      error: snapshot.error,
+      stopReason: snapshot.stopReason,
+      livenessState: snapshot.livenessState,
+      yielded: snapshot.yielded,
+      pendingError: snapshot.pendingError,
+      timeoutPhase: snapshot.timeoutPhase,
+      providerStarted: snapshot.providerStarted,
+      ...(snapshot.terminalDelivery ? { terminalDelivery: snapshot.terminalDelivery } : {}),
+      terminalReceipt: snapshot.terminalReceipt,
+      terminalReply: snapshot.terminalReply,
+    };
+  };
+
+  const abortTurn = async (options: GatewayRequestHandlerOptions): Promise<void> => {
+    await handleChatAbortRequest({ ...options, context, isWebchatConnect });
+  };
+
+  return { startTurn, waitForTurn, abortTurn };
+}
@@ -41,5 +41,4 @@ export type AgentTurnContext = Pick<
  | "loadGatewayModelCatalog"
  | "loadGatewayModelCatalogSnapshot"
  | "logGateway"
-  | "registerToolEventRecipient"
 >;
@@ -1,8 +1,4 @@
 import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
-import {
-  GATEWAY_CLIENT_CAPS,
-  hasGatewayClientCap,
-} from "../../../packages/gateway-protocol/src/client-info.js";
 import { ErrorCodes, errorShape } from "../../../packages/gateway-protocol/src/index.js";
 import { resolveDefaultAgentId } from "../../agents/agent-scope.js";
 import { resolveAgentIdFromSessionKey, type SessionEntry } from "../../config/sessions.js";
@@ -58,6 +54,7 @@ export async function resolveAgentDeliveryPhase(params: {
  context: AgentTurnContext;
  respond: GatewayRequestHandlerOptions["respond"];
  isWebchatConnect: GatewayRequestHandlerOptions["isWebchatConnect"];
+  onRunObserved?: (runId: string) => void;
 }): Promise<AgentDeliveryPhaseResult | undefined> {
  const activeSessionAgentId =
    params.resolvedSessionKey === "global" && params.resolvedSessionAgentId
@@ -66,18 +63,14 @@ export async function resolveAgentDeliveryPhase(params: {
        ? resolveAgentIdFromSessionKey(params.resolvedSessionKey)
        : (params.agentId ?? resolveDefaultAgentId(params.cfgForAgent ?? params.cfg));

-  const connId = typeof params.client?.connId === "string" ? params.client.connId : undefined;
-  if (
-    connId &&
-    hasGatewayClientCap(params.client?.connect?.caps, GATEWAY_CLIENT_CAPS.TOOL_EVENTS)
-  ) {
-    params.context.registerToolEventRecipient(params.runId, connId);
+  if (params.onRunObserved) {
+    params.onRunObserved(params.runId);
    for (const [activeRunId, active] of params.context.chatAbortControllers) {
      const sameSession = active.sessionKey === params.resolvedSessionKey;
      const sameSelectedGlobalAgent =
        params.resolvedSessionKey === "global" ? active.agentId === activeSessionAgentId : true;
      if (activeRunId !== params.runId && sameSession && sameSelectedGlobalAgent) {
-        params.context.registerToolEventRecipient(activeRunId, connId);
+        params.onRunObserved(activeRunId);
      }
    }
  }
@@ -1,6 +1,7 @@
 import { beforeEach, describe, expect, it, vi } from "vitest";
 import { subagentRuns } from "../../agents/subagent-registry-memory.js";
 import * as sessionAccessor from "../../config/sessions/session-accessor.js";
+import { createAgentTurnService } from "../agent-turn/agent-turn-service.js";
 import { createAgentTurnIo } from "../agent-turn/io.js";
 import { prepareAgentRequestPreflight } from "./agent-request-preflight.js";

@@ -49,44 +50,57 @@ function runPreflight(
    });
  }
  const respond = vi.fn();
+  const context = {
+    getRuntimeConfig: () =>
+      options?.requesterOnlyEnabled
+        ? {
+            agents: {
+              list: [{ id: "main", tools: { swarm: true } }, { id: "worker" }],
+            },
+          }
+        : options?.enabled
+          ? { tools: { swarm: true } }
+          : {},
+    dedupe: options?.cached
+      ? new Map([
+          [
+            "agent:collector-run",
+            {
+              ts: 1,
+              ok: true,
+              payload: { status: "accepted", runId: "gateway-run", sessionKey },
+            },
+          ],
+        ])
+      : new Map(),
+  };
+  const client = options?.backend
+    ? { connect: { client: { mode: "backend" }, scopes: ["operator.write"] } }
+    : undefined;
+  const io = createAgentTurnIo(respond);
  const result = prepareAgentRequestPreflight({
-    params: {
+    request: {
      message: "collect",
      sessionKey,
      idempotencyKey: options?.idempotencyKey ?? "collector-run",
      lane: "subagent",
      ...(options?.includeCollectorFields === false ? {} : { swarmCollector, swarmOutputSchema }),
    },
-    io: createAgentTurnIo(respond),
-    context: {
-      getRuntimeConfig: () =>
-        options?.requesterOnlyEnabled
-          ? {
-              agents: {
-                list: [{ id: "main", tools: { swarm: true } }, { id: "worker" }],
-              },
-            }
-          : options?.enabled
-            ? { tools: { swarm: true } }
-            : {},
-      dedupe: options?.cached
-        ? new Map([
-            [
-              "agent:collector-run",
-              {
-                ts: 1,
-                ok: true,
-                payload: { status: "accepted", runId: "gateway-run", sessionKey },
-              },
-            ],
-          ])
-        : new Map(),
-    },
-    client: options?.backend
-      ? { connect: { client: { mode: "backend" }, scopes: ["operator.write"] } }
-      : undefined,
+    io,
+    context,
+    client,
  } as never);
-  return { respond, result };
+  const replay = async () => {
+    if (!result) {
+      return;
+    }
+    await createAgentTurnService({ context, isWebchatConnect: () => false } as never).startTurn({
+      preflight: result,
+      principal: client ?? null,
+      io,
+    } as never);
+  };
+  return { respond, result, replay };
 }

 describe("agent request Swarm preflight", () => {
@@ -237,7 +251,7 @@ describe("agent request Swarm preflight", () => {
    );
  });

-  it("allows an accepted collector launch identity to replay only from Gateway dedupe", () => {
+  it("allows an accepted collector launch identity to replay only from Gateway dedupe", async () => {
    const rejected = runPreflight({ type: "object" }, true, {
      enabled: true,
      backend: true,
@@ -259,7 +273,8 @@ describe("agent request Swarm preflight", () => {
      launchPending: false,
      cached: true,
    });
-    expect(replayed.result).toBeUndefined();
+    expect(replayed.result).toBeDefined();
+    await replayed.replay();
    expect(replayed.respond).toHaveBeenCalledWith(
      true,
      expect.objectContaining({ runId: "gateway-run", status: "in_flight" }),
@@ -268,14 +283,15 @@ describe("agent request Swarm preflight", () => {
    );
  });

-  it("allows an exact cached collector replay after Swarm is disabled", () => {
+  it("allows an exact cached collector replay after Swarm is disabled", async () => {
    const replayed = runPreflight({ type: "object" }, true, {
      backend: true,
      register: true,
      launchPending: false,
      cached: true,
    });
-    expect(replayed.result).toBeUndefined();
+    expect(replayed.result).toBeDefined();
+    await replayed.replay();
    expect(replayed.respond).toHaveBeenCalledWith(
      true,
      expect.objectContaining({ runId: "gateway-run", status: "in_flight" }),
@@ -299,7 +315,7 @@ describe("agent request Swarm preflight", () => {
    );
  });

-  it("keeps completed collector sessions closed while allowing their exact cached replay", () => {
+  it("keeps completed collector sessions closed while allowing their exact cached replay", async () => {
    const ordinary = runPreflight({ type: "object" }, true, {
      enabled: true,
      backend: true,
@@ -324,7 +340,8 @@ describe("agent request Swarm preflight", () => {
      completed: true,
      cached: true,
    });
-    expect(replayed.result).toBeUndefined();
+    expect(replayed.result).toBeDefined();
+    await replayed.replay();
    expect(replayed.respond).toHaveBeenCalledWith(
      true,
      expect.objectContaining({ runId: "gateway-run", status: "in_flight" }),
@@ -366,7 +383,7 @@ describe("agent request restart recovery preflight", () => {
  ) {
    const respond = vi.fn();
    const result = prepareAgentRequestPreflight({
-      params: {
+      request: {
        message: "continue",
        idempotencyKey: "restart-recovery-run",
        forceRestartSafeTools: true,
@@ -1,10 +1,6 @@
 import path from "node:path";
 import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
-import {
-  ErrorCodes,
-  errorShape,
-  validateAgentParams,
-} from "../../../packages/gateway-protocol/src/index.js";
+import { ErrorCodes, errorShape } from "../../../packages/gateway-protocol/src/index.js";
 import { resolveDefaultAgentId } from "../../agents/agent-scope.js";
 import { parseExecApprovalFollowupApprovalId } from "../../agents/bash-tools.exec-approval-followup-state.js";
 import { normalizeSpawnedRunMetadata } from "../../agents/spawned-context.js";
@@ -16,7 +12,6 @@ import { resolveSwarmConfig } from "../../agents/swarm-config.js";
 import { validateStructuredOutputSchema } from "../../agents/swarm-output-schema.js";
 import { resolveAgentIdFromSessionKey, resolveStorePath } from "../../config/sessions.js";
 import { loadSessionEntry } from "../../config/sessions/session-accessor.js";
-import { getAgentEventLifecycleGeneration } from "../../infra/agent-events.js";
 import {
  isMainSessionRestartRecoveryInputProvenance,
  normalizeInputProvenance,
@@ -24,11 +19,7 @@ import {
 } from "../../sessions/input-provenance.js";
 import { isSubagentSessionKey } from "../../sessions/session-key-utils.js";
 import type { AgentTurnContext, AgentTurnIo, AgentTurnPrincipal } from "../agent-turn/types.js";
-import {
-  isAcceptedAgentDedupePayload,
-  readGatewayDedupeEntry,
-  resolveAgentDedupeKeys,
-} from "./agent-dedupe.js";
+import { readGatewayDedupeEntry, resolveAgentDedupeKeys } from "./agent-dedupe.js";
 import {
  resolveExpectedExistingSessionConstraint,
  type ExpectedExistingSessionConstraint,
@@ -39,13 +30,11 @@ import {
  resolveCanUseInternalRuntimeHandoff,
 } from "./agent-handler-helpers.js";
 import type { AgentRunRequest } from "./agent-request-types.js";
-import { assertValidParams } from "./validation.js";

-type AgentRequestPreflight = {
+export type AgentRequestPreflight = {
  request: AgentRunRequest;
  cfg: ReturnType<AgentTurnContext["getRuntimeConfig"]>;
  runId: string;
-  lifecycleGeneration: string;
  allowModelOverride: boolean;
  canUseInternalRuntimeHandoff: boolean;
  canUseCronRunContinuation: boolean;
@@ -67,19 +56,12 @@ type AgentRequestPreflight = {
 };

 export function prepareAgentRequestPreflight(params: {
-  params: unknown;
+  request: AgentRunRequest;
  context: AgentTurnContext;
  client: AgentTurnPrincipal | null;
  io: AgentTurnIo;
 }): AgentRequestPreflight | undefined {
-  if (
-    !assertValidParams(params.params, validateAgentParams, "agent", (ok, payload, error, meta) =>
-      params.io.emitAcceptance([ok, payload, error], meta),
-    )
-  ) {
-    return undefined;
-  }
-  const request = params.params as AgentRunRequest;
+  const { request } = params;
  const cfg = params.context.getRuntimeConfig();
  const canUseInternalRuntimeHandoff = resolveCanUseInternalRuntimeHandoff(params.client);
  const requestSessionKey = request.sessionKey?.trim();
@@ -272,46 +254,10 @@ export function prepareAgentRequestPreflight(params: {
    idempotencyKey: runId,
    execApprovalFollowupApprovalId,
  });
-  const cached = readGatewayDedupeEntry({ dedupe: params.context.dedupe, keys: agentDedupeKeys });
-  if (cached) {
-    if (cached.ok && isAcceptedAgentDedupePayload(cached.payload)) {
-      const cachedRunId =
-        typeof cached.payload.runId === "string" && cached.payload.runId.trim()
-          ? cached.payload.runId.trim()
-          : runId;
-      const cachedSessionKey =
-        typeof cached.payload.sessionKey === "string" && cached.payload.sessionKey.trim()
-          ? cached.payload.sessionKey.trim()
-          : undefined;
-      const cachedAgentId =
-        cachedSessionKey === "global" &&
-        typeof cached.payload.agentId === "string" &&
-        cached.payload.agentId.trim()
-          ? cached.payload.agentId.trim()
-          : undefined;
-      params.io.emitAcceptance(
-        [
-          true,
-          {
-            runId: cachedRunId,
-            status: "in_flight" as const,
-            ...(cachedSessionKey ? { sessionKey: cachedSessionKey } : {}),
-            ...(cachedAgentId ? { agentId: cachedAgentId } : {}),
-          },
-          undefined,
-        ],
-        { cached: true, runId: cachedRunId },
-      );
-    } else {
-      params.io.emitAcceptance([cached.ok, cached.payload, cached.error], { cached: true });
-    }
-    return undefined;
-  }
  return {
    request,
    cfg,
    runId,
-    lifecycleGeneration: getAgentEventLifecycleGeneration(),
    allowModelOverride,
    canUseInternalRuntimeHandoff,
    canUseCronRunContinuation,
@@ -1,32 +1,29 @@
-import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
-import { ErrorCodes, errorShape } from "../../../packages/gateway-protocol/src/index.js";
-import { scheduleMainSessionRecoveryPendingTarget } from "../../agents/main-session-recovery-owner-release.js";
 import {
-  releaseMainSessionRecoveryOwner,
-  type MainSessionRecoveryOwnerLease,
-} from "../../agents/main-session-recovery-store.js";
-import { mergeSessionEntry, type SessionEntry } from "../../config/sessions.js";
-import type { OpenClawConfig } from "../../config/types.openclaw.js";
-import { normalizeDeliveryContext } from "../../utils/delivery-context.shared.js";
+  GATEWAY_CLIENT_CAPS,
+  hasGatewayClientCap,
+} from "../../../packages/gateway-protocol/src/client-info.js";
+import { validateAgentParams } from "../../../packages/gateway-protocol/src/index.js";
+import { createAgentTurnService } from "../agent-turn/agent-turn-service.js";
 import { createAgentTurnIo } from "../agent-turn/io.js";
-import { authorizeResolvedSessionMutation } from "../session-sharing.js";
-import { formatForLog } from "../ws-log.js";
-import { createAgentAdmissionController } from "./agent-admission-controller.js";
-import { prepareAgentContentPhase } from "./agent-content-phase.js";
-import { createCronContinuationController } from "./agent-cron-continuation.js";
-import { createAgentDedupeLifecycle } from "./agent-dedupe-lifecycle.js";
-import { resolveAgentDeliveryPhase } from "./agent-delivery-phase.js";
-import type { RestoredCronContinuation } from "./agent-handler-helpers.js";
+import type { AgentTurnPrincipal } from "../agent-turn/types.js";
 import { prepareAgentRequestPreflight } from "./agent-request-preflight.js";
-import { prepareAgentRequestRouting } from "./agent-request-routing.js";
-import { runAgentResetPhase } from "./agent-reset-phase.js";
-import { prepareAgentRunDispatch } from "./agent-run-admission-phase.js";
-import { startAgentRunExecution } from "./agent-run-execution-phase.js";
-import { buildAgentSessionPatch } from "./agent-session-patch.js";
-import { persistAgentSessionPhase } from "./agent-session-persist.js";
-import { prepareAgentSession } from "./agent-session-prepare.js";
-import { resolveAgentRunSessionCreation } from "./session-creation-provenance.js";
-import type { GatewayRequestHandlers } from "./types.js";
+import type { AgentRunRequest } from "./agent-request-types.js";
+import type { GatewayClient, GatewayRequestHandlers } from "./types.js";
+import { assertValidParams } from "./validation.js";
+
+function captureAgentTurnPrincipal(client: GatewayClient | null): AgentTurnPrincipal | null {
+  if (!client) {
+    return null;
+  }
+  return {
+    authenticatedUserId: client.authenticatedUserId,
+    authenticatedUserProfile: client.authenticatedUserProfile,
+    connId: client.connId,
+    connect: client.connect,
+    internal: client.internal,
+    isDeviceTokenAuth: client.isDeviceTokenAuth,
+  };
+}

 export const agentRunHandler: GatewayRequestHandlers["agent"] = async ({
  params,
@@ -36,523 +33,28 @@ export const agentRunHandler: GatewayRequestHandlers["agent"] = async ({
  isWebchatConnect,
 }) => {
  const io = createAgentTurnIo(respond);
-  const preflight = prepareAgentRequestPreflight({ params, context, client, io });
+  if (
+    !assertValidParams(params, validateAgentParams, "agent", (ok, payload, error, meta) =>
+      io.emitAcceptance([ok, payload, error], meta),
+    )
+  ) {
+    return;
+  }
+  const request = params as AgentRunRequest;
+  const principal = captureAgentTurnPrincipal(client);
+  const preflight = prepareAgentRequestPreflight({ request, context, client: principal, io });
  if (!preflight) {
    return;
  }
-  const {
-    request,
-    cfg,
-    runId,
-    lifecycleGeneration,
-    allowModelOverride,
-    canUseInternalRuntimeHandoff,
-    canUseCronRunContinuation,
-    expectedSession,
-    expectedExistingSessionId,
-    providerOverride,
-    modelOverride,
-    execApprovalFollowupApprovalId,
-    normalizedSpawned,
-    inputProvenance,
-    isRestartRecoveryResumeRun,
-    preserveUserFacingSessionModelState,
-    sessionEffects,
-    suppressVisibleSessionEffects,
-    requestedPromptPersistenceSuppression,
-    isOneShotModelRun,
-    isRawModelRun,
-    agentDedupeKeys,
-  } = preflight;
-  const idem = runId;
-  let resolvedGroupId: string | undefined = normalizedSpawned.groupId;
-  let resolvedGroupChannel: string | undefined = normalizedSpawned.groupChannel;
-  let resolvedGroupSpace: string | undefined = normalizedSpawned.groupSpace;
-  let spawnedByValue: string | undefined;
-  const ownerConnId = typeof client?.connId === "string" ? client.connId : undefined;
-  const ownerDeviceId =
-    typeof client?.connect?.device?.id === "string" ? client.connect.device.id : undefined;
-  const dedupeLifecycle = createAgentDedupeLifecycle({
-    cfg,
-    request,
-    runId,
-    lifecycleGeneration,
-    agentDedupeKeys,
-    suppressVisibleSessionEffects,
-    ownerConnId,
-    ownerDeviceId,
-    context,
+  const connId = principal?.connId;
+  const onRunObserved =
+    connId && hasGatewayClientCap(principal?.connect?.caps, GATEWAY_CLIENT_CAPS.TOOL_EVENTS)
+      ? (runId: string) => context.registerToolEventRecipient(runId, connId)
+      : undefined;
+  await createAgentTurnService({ context, isWebchatConnect }).startTurn({
+    preflight,
+    principal,
    io,
+    onRunObserved,
  });
-  const reservePreAcceptedAgentDedupe = dedupeLifecycle.reserve;
-  const clearUnacceptedAgentDedupe = dedupeLifecycle.clearUnaccepted;
-  const abortForLifecycleRotation = dedupeLifecycle.abortForLifecycleRotation;
-  const routing = await prepareAgentRequestRouting({
-    request,
-    cfg,
-    expectedSession,
-    isRawModelRun,
-    execApprovalFollowupApprovalId,
-    runId,
-    agentDedupeKeys,
-    context,
-    respond,
-    reserveDedupe: reservePreAcceptedAgentDedupe,
-    clearDedupe: clearUnacceptedAgentDedupe,
-  });
-  if (!routing) {
-    return;
-  }
-  const {
-    normalizedAttachments,
-    requestedBestEffortDeliver,
-    knownAgents,
-    requestedSessionId,
-    requestedToRaw,
-    sessionKeyFromTo,
-    requestedSessionKeyRaw,
-    explicitRecipientSession,
-    preAcceptedReservedSessionKey,
-    preAttachmentSession,
-  } = routing;
-  let agentId = routing.agentId;
-  let requestedSessionKey = routing.requestedSessionKey;
-  let gatewayAdmissionTransferred = false;
-  let mainRestartRecoveryOwnerLease: MainSessionRecoveryOwnerLease | undefined;
-  let releaseGatewayAdmission = () => {};
-  const cronContinuation = createCronContinuationController({
-    runId,
-    lifecycleGeneration,
-    context,
-  });
-  const releaseCronContinuationClaimWithRecovery = cronContinuation.releaseWithRecovery;
-  try {
-    const content = await prepareAgentContentPhase({
-      request,
-      cfg,
-      context,
-      respond,
-      isRawModelRun,
-      inputProvenance,
-      normalizedAttachments,
-      requestedSessionKeyRaw,
-      requestedSessionKey,
-      requestedSessionId,
-      requestedToRaw,
-      sessionKeyFromTo,
-      agentId,
-      providerOverride,
-      modelOverride,
-      explicitRecipientSession,
-      knownAgents,
-    });
-    if (!content) {
-      return;
-    }
-    agentId = content.agentId;
-    requestedSessionKey = content.requestedSessionKey;
-    // Participation is authorized below against the canonical session the run
-    // actually targets (see prepareAgentSession). A keyless request resolves its
-    // default/effective session there, so authorizing only an explicit key here
-    // would let a non-member drive a restricted default session.
-    let effectiveTranscriptInputText = content.effectiveTranscriptInputText;
-    let message = content.message;
-    const {
-      images,
-      imageOrder,
-      media,
-      replyTo,
-      recipientChannel,
-      recipientAccountId,
-      recipientThreadId,
-      to,
-    } = content;
-    let resolvedSessionId = requestedSessionId;
-    let sessionEntry: SessionEntry | undefined;
-    let effectiveBootstrapContextRunKind = request.bootstrapContextRunKind;
-    let restoredCronContinuation: RestoredCronContinuation | undefined;
-    let restoredCronContinuationIdentity:
-      | Pick<RestoredCronContinuation, "lifecycleRevision" | "sessionId">
-      | undefined;
-    let sessionPersistedBeforeGatewayAdmission = false;
-    let bestEffortDeliver = requestedBestEffortDeliver ?? false;
-    let cfgForAgent: OpenClawConfig | undefined;
-    let resolvedSessionKey = requestedSessionKey;
-    let resolvedSessionAgentId: string | undefined;
-    let isNewSession = false;
-    let supersededSessionId: string | undefined;
-    let skipAgentInitialSessionTouch = false;
-    let pendingChatRun: { sessionKey: string; agentId?: string } | undefined;
-    let admittedSessionId = resolvedSessionId ?? runId;
-    const admissionController = createAgentAdmissionController({
-      cfg,
-      runId,
-      lifecycleGeneration,
-      agentDedupeKeys,
-      preAcceptedReservedSessionKey,
-      expectedSession,
-      context,
-      io,
-      dedupeLifecycle,
-      getRequestedSessionKey: () => requestedSessionKey,
-      getResolvedSessionKey: () => resolvedSessionKey,
-      getResolvedSessionId: () => resolvedSessionId,
-      getResolvedSessionAgentId: () => resolvedSessionAgentId,
-      getAgentId: () => agentId,
-      getCfgForAgent: () => cfgForAgent,
-      getSessionPersisted: () => sessionPersistedBeforeGatewayAdmission,
-      getSupersededSessionId: () => supersededSessionId,
-      setAdmittedSessionId: (sessionId) => {
-        admittedSessionId = sessionId;
-      },
-    });
-    const admissionAgentId = admissionController.admissionAgentId;
-    const assertGatewayWorkAdmissionAllowed = admissionController.assertAllowed;
-    const acquireGatewayWorkAdmission = admissionController.acquire;
-    const respondToGatewayAdmissionOutcome = admissionController.respondToOutcome;
-    releaseGatewayAdmission = admissionController.release;
-    const resetPhase = await runAgentResetPhase({
-      request,
-      cfg,
-      requestedSessionKey,
-      resolvedSessionId,
-      effectiveTranscriptInputText,
-      message,
-      agentId,
-      sessionKeyFromTo,
-      lifecycleGeneration,
-      runId,
-      agentDedupeKeys,
-      client,
-      context,
-      respond,
-      abortForLifecycleRotation,
-      setCommittedResetCompletion: dedupeLifecycle.setCommittedResetCompletion,
-    });
-    requestedSessionKey = resetPhase.requestedSessionKey;
-    resolvedSessionId = resetPhase.resolvedSessionId;
-    effectiveTranscriptInputText = resetPhase.effectiveTranscriptInputText;
-    message = resetPhase.message;
-    if (resetPhase.accepted) {
-      dedupeLifecycle.markAccepted(true);
-    }
-    if (resetPhase.stop) {
-      return;
-    }
-
-    if (requestedSessionKey) {
-      const preparedSession = prepareAgentSession({
-        requestedSessionKey,
-        requestedSessionId,
-        expectedExistingSessionId,
-        agentId,
-        recipientChannel,
-        request,
-        canUseCronRunContinuation,
-        lifecycleGeneration,
-        effectiveBootstrapContextRunKind,
-        preAttachmentSession,
-        respond,
-      });
-      if (!preparedSession) {
-        return;
-      }
-      const {
-        cfg: cfgLocal,
-        storePath,
-        entry,
-        canonicalKey,
-        storeKeys,
-        maintenanceConfig: sessionMaintenanceConfig,
-        canonicalSessionAgentId,
-        resetPolicy,
-        now,
-        visibleRequest,
-        mainSessionKey: mainSessionKeyForRequest,
-        isSystemGatewayRun,
-        sessionId,
-        touchInteraction,
-        failedSessionTranscriptMissing: resolveFailedSessionTranscriptMissingForEntry,
-      } = preparedSession;
-      cfgForAgent = cfgLocal;
-      // Authorize the canonical session the run will actually target — covering
-      // keyless requests whose default/effective session is resolved only here —
-      // before any run side effects (admission, dispatch).
-      const sharingError = authorizeResolvedSessionMutation({
-        cfg: cfgLocal,
-        client,
-        sessionKey: canonicalKey,
-        agentId: canonicalSessionAgentId,
-      });
-      if (sharingError) {
-        io.emitAcceptance([false, undefined, sharingError]);
-        return;
-      }
-      effectiveBootstrapContextRunKind = preparedSession.effectiveBootstrapContextRunKind;
-      restoredCronContinuationIdentity = preparedSession.restoredCronContinuationIdentity;
-      sessionPersistedBeforeGatewayAdmission =
-        preparedSession.sessionPersistedBeforeGatewayAdmission;
-      isNewSession = preparedSession.isNewSession;
-      const sessionAgent = canonicalSessionAgentId;
-      const requestDeliveryHint = normalizeDeliveryContext({
-        channel: recipientChannel?.trim(),
-        to,
-        accountId: recipientAccountId?.trim(),
-        // Pass threadId directly — normalizeDeliveryContext handles both
-        // string and numeric threadIds (e.g., Matrix uses integers).
-        threadId: recipientThreadId,
-      });
-      const buildSessionPatch = (freshEntry: SessionEntry | undefined) =>
-        buildAgentSessionPatch({
-          freshEntry,
-          initialEntry: entry,
-          cfg: cfgLocal,
-          sessionAgentId: sessionAgent,
-          canonicalSessionKey: canonicalKey,
-          storePath,
-          normalizedSpawned,
-          requestDeliveryHint,
-          requestLabel: request.label,
-          pluginOwnerId:
-            freshEntry === undefined
-              ? normalizeOptionalString(client?.internal?.pluginRuntimeOwnerId)
-              : undefined,
-          expectedExistingSessionId,
-          hasRestoredCronContinuation: restoredCronContinuationIdentity !== undefined,
-          resetPolicy,
-          now,
-          requestedSessionId,
-          isSystemGatewayRun,
-          visibleRequest,
-          fallbackSessionId: sessionId,
-          touchInteraction,
-          failedSessionTranscriptMissing: resolveFailedSessionTranscriptMissingForEntry,
-        });
-      const patchBuild = buildSessionPatch(entry);
-      isNewSession = patchBuild.isNewSession;
-      sessionEntry = mergeSessionEntry(entry, patchBuild.patch);
-      resolvedSessionId = sessionEntry?.sessionId ?? sessionId;
-      admittedSessionId = resolvedSessionId ?? runId;
-      const canonicalSessionKey = canonicalKey;
-      resolvedSessionKey = canonicalSessionKey;
-      const sessionAgentId = canonicalSessionAgentId;
-      resolvedSessionAgentId = sessionAgentId;
-      const mainSessionKey = mainSessionKeyForRequest;
-      try {
-        await acquireGatewayWorkAdmission(storePath ?? `agent:${sessionAgentId}`);
-      } catch (err) {
-        io.emitAcceptance([
-          false,
-          undefined,
-          errorShape(ErrorCodes.INVALID_REQUEST, formatForLog(err)),
-        ]);
-        return;
-      }
-      if (respondToGatewayAdmissionOutcome()) {
-        return;
-      }
-      const persistedSession = await persistAgentSessionPhase({
-        request,
-        cfg: cfgLocal,
-        storePath,
-        storeKeys,
-        entry,
-        canonicalSessionKey,
-        sessionAgentId,
-        mainSessionKey,
-        creation: resolveAgentRunSessionCreation(client),
-        lifecycleGeneration,
-        isRestartRecoveryResumeRun,
-        runId,
-        agentId,
-        suppressVisibleSessionEffects,
-        restoredCronContinuationIdentity,
-        initialPatchBuild: patchBuild,
-        buildSessionPatch,
-        initialSessionEntry: sessionEntry,
-        initialResolvedSessionId: resolvedSessionId,
-        initialSessionPersistedBeforeGatewayAdmission: sessionPersistedBeforeGatewayAdmission,
-        initialSupersededSessionId: supersededSessionId,
-        touchInteraction,
-        requestedBestEffortDeliver,
-        bestEffortDeliver,
-        expectedSession,
-        maintenanceConfig: sessionMaintenanceConfig,
-        abortForLifecycleRotation,
-        assertGatewayWorkAdmissionAllowed,
-        respondToGatewayAdmissionOutcome,
-        updateAdmissionState: (state) => {
-          resolvedSessionId = state.resolvedSessionId;
-          admittedSessionId = state.admittedSessionId;
-          supersededSessionId = state.supersededSessionId;
-          sessionPersistedBeforeGatewayAdmission = state.sessionPersistedBeforeGatewayAdmission;
-        },
-        getAdmittedSessionId: () => admittedSessionId,
-        setCronContinuationClaim: cronContinuation.setClaim,
-        setMainRestartRecoveryOwnerLease: (lease) => {
-          mainRestartRecoveryOwnerLease = lease;
-        },
-        respond,
-      });
-      if (!persistedSession) {
-        return;
-      }
-      sessionEntry = persistedSession.sessionEntry;
-      resolvedSessionId = persistedSession.resolvedSessionId;
-      sessionPersistedBeforeGatewayAdmission =
-        persistedSession.sessionPersistedBeforeGatewayAdmission;
-      supersededSessionId = persistedSession.supersededSessionId;
-      admittedSessionId = persistedSession.admittedSessionId;
-      skipAgentInitialSessionTouch = persistedSession.skipAgentInitialSessionTouch;
-      isNewSession = persistedSession.isNewSession;
-      spawnedByValue = persistedSession.spawnedBy;
-      resolvedGroupId = persistedSession.groupId;
-      resolvedGroupChannel = persistedSession.groupChannel;
-      resolvedGroupSpace = persistedSession.groupSpace;
-      pendingChatRun = persistedSession.pendingChatRun;
-      bestEffortDeliver = persistedSession.bestEffortDeliver;
-      restoredCronContinuation = persistedSession.restoredCronContinuation;
-    }
-
-    const delivery = await resolveAgentDeliveryPhase({
-      request,
-      cfg,
-      cfgForAgent,
-      sessionEntry,
-      resolvedSessionKey,
-      resolvedSessionAgentId,
-      agentId,
-      replyTo,
-      to,
-      recipientChannel,
-      recipientAccountId,
-      recipientThreadId,
-      bestEffortDeliver,
-      runId,
-      client,
-      context,
-      respond,
-      isWebchatConnect,
-    });
-    if (!delivery) {
-      return;
-    }
-    const { activeSessionAgentId } = delivery;
-
-    const preparedDispatch = await prepareAgentRunDispatch({
-      request,
-      cfg,
-      cfgForAgent,
-      sessionEntry,
-      resolvedSessionKey,
-      requestedSessionKey,
-      preAcceptedReservedSessionKey,
-      activeSessionAgentId,
-      delivery,
-      restoredCronContinuationIdentity,
-      restoredCronContinuation,
-      providerOverride,
-      modelOverride,
-      allowModelOverride,
-      lifecycleGeneration,
-      getAdmittedSessionId: () => admittedSessionId,
-      ownerConnId,
-      ownerDeviceId,
-      suppressVisibleSessionEffects,
-      pendingChatRun,
-      inputProvenance,
-      isOneShotModelRun,
-      isRestartRecoveryResumeRun,
-      runId,
-      agentDedupeKeys,
-      context,
-      client,
-      io,
-      abortForLifecycleRotation,
-      acquireGatewayWorkAdmission,
-      assertGatewayWorkAdmissionAllowed,
-      hasGatewayAdmissionOutcome: admissionController.hasOutcome,
-      respondToGatewayAdmissionOutcome,
-      admissionAgentId,
-      getGatewayWorkAdmission: admissionController.getAdmission,
-      setAdmittedRunAbort: admissionController.setAdmittedRunAbort,
-      getAdmittedRunAbort: admissionController.getAdmittedRunAbort,
-      markAgentRunAccepted: dedupeLifecycle.markAccepted,
-    });
-    if (!preparedDispatch) {
-      return;
-    }
-    resolvedSessionId = admittedSessionId;
-    gatewayAdmissionTransferred = true;
-    startAgentRunExecution({
-      prepared: preparedDispatch,
-      mainRestartRecoveryOwnerLease,
-      request,
-      cfg,
-      cfgForAgent,
-      sessionEntry,
-      resolvedSessionKey,
-      requestedSessionKey,
-      requestedSessionKeyRaw,
-      resolvedSessionId,
-      agentId,
-      activeSessionAgentId,
-      delivery,
-      isNewSession,
-      isRawModelRun,
-      isOneShotModelRun,
-      isRestartRecoveryResumeRun,
-      suppressVisibleSessionEffects,
-      message,
-      images,
-      imageOrder,
-      media,
-      effectiveTranscriptInputText,
-      inputProvenance,
-      runId,
-      idempotencyKey: idem,
-      agentDedupeKeys,
-      spawnedBy: spawnedByValue,
-      groupId: resolvedGroupId,
-      groupChannel: resolvedGroupChannel,
-      groupSpace: resolvedGroupSpace,
-      bestEffortDeliver,
-      lifecycleGeneration,
-      effectiveBootstrapContextRunKind,
-      requestedPromptPersistenceSuppression,
-      preserveUserFacingSessionModelState,
-      sessionEffects,
-      skipAgentInitialSessionTouch,
-      restoredCronContinuation,
-      canUseInternalRuntimeHandoff,
-      execApprovalFollowupApprovalId,
-      client,
-      context,
-      io,
-      releaseCronContinuationClaimWithRecovery,
-    });
-    mainRestartRecoveryOwnerLease = undefined;
-  } finally {
-    try {
-      if (!gatewayAdmissionTransferred) {
-        let pendingRecovery: Awaited<ReturnType<typeof releaseMainSessionRecoveryOwner>> =
-          undefined;
-        try {
-          pendingRecovery = await releaseMainSessionRecoveryOwner(mainRestartRecoveryOwnerLease);
-        } finally {
-          try {
-            releaseGatewayAdmission();
-          } finally {
-            try {
-              await releaseCronContinuationClaimWithRecovery();
-            } finally {
-              scheduleMainSessionRecoveryPendingTarget(pendingRecovery);
-            }
-          }
-        }
-      }
-    } finally {
-      clearUnacceptedAgentDedupe();
-    }
-  }
 };
@@ -5,7 +5,10 @@ import type { DedupeEntry } from "../server-shared.js";
 import { setGatewayDedupeEntry } from "./agent-job.js";
 import { agentHandlers } from "./agent.js";

-function waitThroughGateway(params: { runId: string; timeoutMs: number }) {
+function waitThroughGateway(
+  params: { runId: string; timeoutMs: number },
+  activeKind?: "agent" | "chat",
+) {
  const respond = vi.fn();
  const handler = expectDefined(
    agentHandlers["agent.wait"],
@@ -15,7 +18,11 @@ function waitThroughGateway(params: { runId: string; timeoutMs: number }) {
    handler({
      params,
      respond,
-      context: { chatAbortControllers: new Map() },
+      context: {
+        chatAbortControllers: activeKind
+          ? new Map([[params.runId, { kind: activeKind }]])
+          : new Map(),
+      },
    } as unknown as Parameters<typeof handler>[0]),
  );
  return { promise, respond };
@@ -51,6 +58,32 @@ afterEach(() => {
 });

 describe("agent.wait gateway dedupe observations", () => {
+  it.each([
+    ["agent", "timeout"],
+    ["chat", "ok"],
+  ] as const)("uses the %s abort entry to select the run observation", async (kind, status) => {
+    const runId = `run-kind-${kind}`;
+    const dedupe = new Map<string, DedupeEntry>();
+    setGatewayDedupeEntry({
+      dedupe,
+      key: `agent:${runId}`,
+      entry: {
+        ts: 100,
+        ok: false,
+        payload: { runId, status: "timeout", endedAt: 100, timeoutPhase: "provider" },
+      },
+    });
+    setGatewayDedupeEntry({
+      dedupe,
+      key: `chat:${runId}`,
+      entry: { ts: 200, ok: true, payload: { runId, status: "ok", endedAt: 200 } },
+    });
+
+    const waiter = waitThroughGateway({ runId, timeoutMs: 0 }, kind);
+    await waiter.promise;
+    expect(waiter.respond).toHaveBeenCalledWith(true, expect.objectContaining({ runId, status }));
+  });
+
  it("resolves concurrent waiters when the terminal dedupe entry lands", async () => {
    const runId = "run-public-concurrent-waiters";
    const dedupe = new Map<string, DedupeEntry>();
@@ -1,5 +1,8 @@
-import { validateAgentWaitParams } from "../../../packages/gateway-protocol/src/index.js";
-import { waitForAgentJob } from "./agent-job.js";
+import {
+  validateAgentWaitParams,
+  type AgentWaitParams,
+} from "../../../packages/gateway-protocol/src/index.js";
+import { createAgentTurnService } from "../agent-turn/agent-turn-service.js";
 import type { GatewayRequestHandlers } from "./types.js";
 import { assertValidParams } from "./validation.js";

@@ -7,48 +10,13 @@ export const agentWaitHandler: GatewayRequestHandlers["agent.wait"] = async ({
  params,
  respond,
  context,
+  isWebchatConnect,
 }) => {
  if (!assertValidParams(params, validateAgentWaitParams, "agent.wait", respond)) {
    return;
  }
-  const runId = (params.runId ?? "").trim();
-  const timeoutMs =
-    typeof params.timeoutMs === "number" && Number.isFinite(params.timeoutMs)
-      ? Math.max(0, Math.floor(params.timeoutMs))
-      : 30_000;
-  // `hasActiveChatRun` must exclude agent-kind abort entries so wait snapshot
-  // preference continues to distinguish chat.send from agent RPC runs.
-  const activeChatEntry = context.chatAbortControllers.get(runId);
-  const hasActiveChatRun = activeChatEntry !== undefined && activeChatEntry.kind !== "agent";
-  const snapshot = await waitForAgentJob({
-    runId,
-    timeoutMs,
-    ...(hasActiveChatRun ? { source: "chat" } : {}),
-  });
-  if (!snapshot) {
-    const activeRunRegistered = activeChatEntry !== undefined;
-    respond(true, {
-      runId,
-      status: "timeout",
-      timeoutPhase: activeRunRegistered ? "gateway_draining" : "queue",
-      ...(activeRunRegistered ? {} : { providerStarted: false }),
-    });
-    return;
-  }
-  respond(true, {
-    runId,
-    status: snapshot.status,
-    startedAt: snapshot.startedAt,
-    endedAt: snapshot.endedAt,
-    error: snapshot.error,
-    stopReason: snapshot.stopReason,
-    livenessState: snapshot.livenessState,
-    yielded: snapshot.yielded,
-    pendingError: snapshot.pendingError,
-    timeoutPhase: snapshot.timeoutPhase,
-    providerStarted: snapshot.providerStarted,
-    ...(snapshot.terminalDelivery ? { terminalDelivery: snapshot.terminalDelivery } : {}),
-    terminalReceipt: snapshot.terminalReceipt,
-    terminalReply: snapshot.terminalReply,
-  });
+  const result = await createAgentTurnService({ context, isWebchatConnect }).waitForTurn(
+    params as AgentWaitParams,
+  );
+  respond(true, result);
 };
@@ -9,6 +9,7 @@ import { resolveSessionAgentId } from "../../agents/agent-scope.js";
 import { resolveSessionWorkStartError } from "../../config/sessions.js";
 import type { OpenClawConfig } from "../../config/types.openclaw.js";
 import { beginSessionWorkAdmission } from "../../sessions/session-lifecycle-admission.js";
+import { createAgentTurnService } from "../agent-turn/agent-turn-service.js";
 import {
  projectChatDisplayMessage,
  resolveEffectiveChatHistoryMaxChars,
@@ -19,7 +20,6 @@ import {
  resolveSessionModelRef,
 } from "../session-utils.js";
 import { formatForLog } from "../ws-log.js";
-import { handleChatAbortRequest } from "./chat-abort-handler.js";
 import { sendGlobalAwareNodeChatPayload } from "./chat-broadcast.js";
 import { chatHistoryHandlers } from "./chat-history-handler.js";
 import { chatMessageGetHandlers } from "./chat-message-get-handler.js";
@@ -100,7 +100,9 @@ export const chatHandlers: GatewayRequestHandlers = {
    });
    respond(true, { titles });
  },
-  "chat.abort": handleChatAbortRequest,
+  "chat.abort": async (options) => {
+    await createAgentTurnService(options).abortTurn(options);
+  },
  "chat.send": handleDirectExternalChatSend,
  "chat.inject": async ({ params, respond, context }) => {
    if (!assertValidParams(params, validateChatInjectParams, "chat.inject", respond)) {