refactor(gateway): narrow agent turn response sink (#121189)

* refactor(gateway): narrow agent turn response sink * fix(gateway): keep turn frame aliases internal
2026-08-25 20:05:46 -06:00 · 2026-08-09 11:48:46 -07:00
parent 40dbc7a65a
commit cdc5dd24b9
13 changed files with 195 additions and 129 deletions
@@ -7,6 +7,7 @@ import {
  clearRuntimeConfigSnapshot,
  getRuntimeConfig,
 } from "../config/config.js";
+import { createAgentTurnIo } from "../gateway/agent-turn/io.js";
 import { prepareAgentRequestPreflight } from "../gateway/server-methods/agent-request-preflight.js";
 import type {
  GatewayRequestContext,
@@ -334,7 +335,7 @@ describe("spawnSubagentDirect in-process Gateway collector launch", () => {
        const externalRespond = vi.fn();
        const externalPreflight = prepareAgentRequestPreflight({
          params,
-          respond: externalRespond,
+          io: createAgentTurnIo(externalRespond),
          context: gatewayContext,
          client: externalCliClient(),
        } as never);
@@ -345,7 +346,7 @@ describe("spawnSubagentDirect in-process Gateway collector launch", () => {
          : externalCliClient();
        const hostPreflight = prepareAgentRequestPreflight({
          params,
-          respond: hostRespond,
+          io: createAgentTurnIo(hostRespond),
          context: gatewayContext,
          client,
        } as never);
@@ -0,0 +1,15 @@
+import type { RespondFn } from "../server-methods/shared-types.js";
+import type { AgentTurnFrame, AgentTurnIo } from "./types.js";
+
+export function createAgentTurnIo(respond: RespondFn): AgentTurnIo {
+  const emit = (frame: AgentTurnFrame, meta?: Parameters<RespondFn>[3]) => {
+    // Response order is positional, and final responses may outlive the handler.
+    // Delegate synchronously while preserving the original three- or four-argument call.
+    if (meta === undefined) {
+      respond(...frame);
+      return;
+    }
+    respond(...frame, meta);
+  };
+  return { emitAcceptance: emit, emitFinal: emit };
+}
@@ -1,4 +1,22 @@
-import type { GatewayClient, GatewayRequestContext } from "../server-methods/shared-types.js";
+import type {
+  GatewayClient,
+  GatewayRequestContext,
+  RespondFn,
+} from "../server-methods/shared-types.js";
+
+export type AgentTurnFrame = readonly [
+  ok: Parameters<RespondFn>[0],
+  payload: Parameters<RespondFn>[1],
+  error: Parameters<RespondFn>[2],
+];
+
+type AgentTurnAcceptance = AgentTurnFrame;
+type AgentTurnFinal = AgentTurnFrame;
+
+export type AgentTurnIo = {
+  emitAcceptance: (acceptance: AgentTurnAcceptance, meta?: Parameters<RespondFn>[3]) => void;
+  emitFinal: (final: AgentTurnFinal, meta?: Parameters<RespondFn>[3]) => void;
+};

 export type AgentTurnPrincipal = Pick<
  GatewayClient,
@@ -11,7 +11,7 @@ import {
  beginSessionWorkAdmission,
  type SessionWorkAdmissionLease,
 } from "../../sessions/session-lifecycle-admission.js";
-import type { AgentTurnContext } from "../agent-turn/types.js";
+import type { AgentTurnContext, AgentTurnIo } from "../agent-turn/types.js";
 import { registerChatAbortController } from "../chat-abort.js";
 import { loadSessionEntry } from "../session-utils.js";
 import type { AgentDedupeLifecycle } from "./agent-dedupe-lifecycle.js";
@@ -26,7 +26,6 @@ import {
  consumeExpectedSessionWorkAdmission,
  type ExpectedExistingSessionConstraint,
 } from "./agent-expected-session.js";
-import type { GatewayRequestHandlerOptions } from "./types.js";

 export function createAgentAdmissionController(params: {
  cfg: OpenClawConfig;
@@ -36,7 +35,7 @@ export function createAgentAdmissionController(params: {
  preAcceptedReservedSessionKey?: string;
  expectedSession?: ExpectedExistingSessionConstraint;
  context: AgentTurnContext;
-  respond: GatewayRequestHandlerOptions["respond"];
+  io: AgentTurnIo;
  dedupeLifecycle: AgentDedupeLifecycle;
  getRequestedSessionKey: () => string | undefined;
  getResolvedSessionKey: () => string | undefined;
@@ -232,19 +231,24 @@ export function createAgentAdmissionController(params: {
    if (postAdmissionAbort) {
      admission?.release();
      params.dedupeLifecycle.markAccepted(true);
-      params.respond(postAdmissionAbort.ok, postAdmissionAbort.payload, postAdmissionAbort.error, {
-        cached: true,
-        runId: params.runId,
-      });
+      params.io.emitAcceptance(
+        [postAdmissionAbort.ok, postAdmissionAbort.payload, postAdmissionAbort.error],
+        {
+          cached: true,
+          runId: params.runId,
+        },
+      );
      return true;
    }
    if (postAdmissionTimeout || postAdmissionSuperseded) {
      admission?.release();
      params.dedupeLifecycle.markAccepted(true);
-      params.respond(
-        true,
-        postAdmissionTimeout ?? { runId: params.runId, status: "in_flight" as const },
-        undefined,
+      params.io.emitAcceptance(
+        [
+          true,
+          postAdmissionTimeout ?? { runId: params.runId, status: "in_flight" as const },
+          undefined,
+        ],
        { cached: true, runId: params.runId },
      );
      return true;
@@ -2,7 +2,7 @@ import { randomUUID } from "node:crypto";
 import { AGENT_RUN_RESTART_ABORT_STOP_REASON } from "../../agents/run-termination.js";
 import { resolveAgentTimeoutMs } from "../../agents/timeout.js";
 import { getAgentEventLifecycleGeneration } from "../../infra/agent-events.js";
-import type { AgentTurnContext } from "../agent-turn/types.js";
+import type { AgentTurnContext, AgentTurnIo } from "../agent-turn/types.js";
 import { resolveAgentRunExpiresAtMs } from "../chat-abort.js";
 import { resolveSessionStoreKey } from "../session-utils.js";
 import {
@@ -21,7 +21,6 @@ import {
  sessionResetAckText,
 } from "./agent-session-reset.js";
 import { emitSessionsChanged } from "./session-change-event.js";
-import type { GatewayRequestHandlerOptions } from "./types.js";

 export type AgentDedupeLifecycle = ReturnType<typeof createAgentDedupeLifecycle>;

@@ -35,7 +34,7 @@ export function createAgentDedupeLifecycle(params: {
  ownerConnId?: string;
  ownerDeviceId?: string;
  context: AgentTurnContext;
-  respond: GatewayRequestHandlerOptions["respond"];
+  io: AgentTurnIo;
 }) {
  let reserved = false;
  let accepted = false;
@@ -126,7 +125,7 @@ export function createAgentDedupeLifecycle(params: {
        keys: params.agentDedupeKeys,
        entry: { ts: Date.now(), ok: true, payload: responsePayload },
      });
-      params.respond(true, responsePayload, undefined, { runId: params.runId });
+      params.io.emitAcceptance([true, responsePayload, undefined], { runId: params.runId });
      emitSessionsChanged(params.context, {
        sessionKey: completion.sessionKey,
        ...(completion.sessionKey === "global" && completion.agentId
@@ -145,17 +144,19 @@ export function createAgentDedupeLifecycle(params: {
      runId: params.runId,
      stopReason: AGENT_RUN_RESTART_ABORT_STOP_REASON,
    });
-    params.respond(
-      true,
-      {
-        runId: params.runId,
-        status: "timeout" as const,
-        summary: "aborted",
-        stopReason: AGENT_RUN_RESTART_ABORT_STOP_REASON,
-        timeoutPhase: "queue" as const,
-        providerStarted: false,
-      },
-      undefined,
+    params.io.emitAcceptance(
+      [
+        true,
+        {
+          runId: params.runId,
+          status: "timeout" as const,
+          summary: "aborted",
+          stopReason: AGENT_RUN_RESTART_ABORT_STOP_REASON,
+          timeoutPhase: "queue" as const,
+          providerStarted: false,
+        },
+        undefined,
+      ],
      { runId: params.runId },
    );
    return true;
@@ -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 { createAgentTurnIo } from "../agent-turn/io.js";
 import { prepareAgentRequestPreflight } from "./agent-request-preflight.js";

 function runPreflight(
@@ -56,7 +57,7 @@ function runPreflight(
      lane: "subagent",
      ...(options?.includeCollectorFields === false ? {} : { swarmCollector, swarmOutputSchema }),
    },
-    respond,
+    io: createAgentTurnIo(respond),
    context: {
      getRuntimeConfig: () =>
        options?.requesterOnlyEnabled
@@ -377,7 +378,7 @@ describe("agent request restart recovery preflight", () => {
          sourceTool,
        },
      },
-      respond,
+      io: createAgentTurnIo(respond),
      context: {
        getRuntimeConfig: () => ({}),
        dedupe: new Map(),
@@ -23,7 +23,7 @@ import {
  shouldPreserveUserFacingSessionStateForInputProvenance,
 } from "../../sessions/input-provenance.js";
 import { isSubagentSessionKey } from "../../sessions/session-key-utils.js";
-import type { AgentTurnContext, AgentTurnPrincipal } from "../agent-turn/types.js";
+import type { AgentTurnContext, AgentTurnIo, AgentTurnPrincipal } from "../agent-turn/types.js";
 import {
  isAcceptedAgentDedupePayload,
  readGatewayDedupeEntry,
@@ -39,7 +39,6 @@ import {
  resolveCanUseInternalRuntimeHandoff,
 } from "./agent-handler-helpers.js";
 import type { AgentRunRequest } from "./agent-request-types.js";
-import type { GatewayRequestHandlerOptions } from "./types.js";
 import { assertValidParams } from "./validation.js";

 type AgentRequestPreflight = {
@@ -67,13 +66,17 @@ type AgentRequestPreflight = {
  agentDedupeKeys: string[];
 };

-export function prepareAgentRequestPreflight(
-  params: Pick<GatewayRequestHandlerOptions, "params" | "respond"> & {
-    context: AgentTurnContext;
-    client: AgentTurnPrincipal | null;
-  },
-): AgentRequestPreflight | undefined {
-  if (!assertValidParams(params.params, validateAgentParams, "agent", params.respond)) {
+export function prepareAgentRequestPreflight(params: {
+  params: unknown;
+  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;
@@ -103,14 +106,14 @@ export function prepareAgentRequestPreflight(
      ? validateStructuredOutputSchema(request.swarmOutputSchema)
      : undefined;
    if (request.swarmCollector !== true || schemaError) {
-      params.respond(
+      params.io.emitAcceptance([
        false,
        undefined,
        errorShape(
          ErrorCodes.INVALID_REQUEST,
          schemaError ?? "active swarm collector sessions require swarmCollector=true",
        ),
-      );
+      ]);
      return undefined;
    }
    const registeredCollector = findAuthorizedSwarmCollectorRequest({
@@ -142,31 +145,31 @@ export function prepareAgentRequestPreflight(
      !registeredCollector ||
      (!pendingCollectorLaunch && !collectorDedupe)
    ) {
-      params.respond(
+      params.io.emitAcceptance([
        false,
        undefined,
        errorShape(
          ErrorCodes.INVALID_REQUEST,
          "swarm collector fields require an enabled, host-registered collector run",
        ),
-      );
+      ]);
      return undefined;
    }
  }
  if (request.cwd && !path.isAbsolute(request.cwd)) {
-    params.respond(
+    params.io.emitAcceptance([
      false,
      undefined,
      errorShape(ErrorCodes.INVALID_REQUEST, "cwd must be absolute"),
-    );
+    ]);
    return undefined;
  }
  if (request.cwd && !normalizeOptionalString(params.client?.internal?.pluginRuntimeOwnerId)) {
-    params.respond(
+    params.io.emitAcceptance([
      false,
      undefined,
      errorShape(ErrorCodes.INVALID_REQUEST, "cwd is reserved for plugin-owned subagent runs"),
-    );
+    ]);
    return undefined;
  }
  const allowModelOverride = resolveAllowModelOverrideFromClient(params.client);
@@ -177,11 +180,11 @@ export function prepareAgentRequestPreflight(
    internalRuntimeHandoffId: request.internalRuntimeHandoffId,
  });
  if (!expectedSessionResult.ok) {
-    params.respond(
+    params.io.emitAcceptance([
      false,
      undefined,
      errorShape(ErrorCodes.INVALID_REQUEST, expectedSessionResult.error),
-    );
+    ]);
    return undefined;
  }
  const requestedPromptPersistenceSuppression = request.suppressPromptPersistence === true;
@@ -190,77 +193,77 @@ export function prepareAgentRequestPreflight(
  const isOneShotModelRun = request.modelRun === true;
  const isRawModelRun = isOneShotModelRun || request.promptMode === "none";
  if (request.promptMode === "none" && !isOneShotModelRun) {
-    params.respond(
+    params.io.emitAcceptance([
      false,
      undefined,
      errorShape(
        ErrorCodes.INVALID_REQUEST,
        'promptMode="none" requires modelRun=true so the run cannot mutate a durable session.',
      ),
-    );
+    ]);
    return undefined;
  }
  if (requestedModelOverride && !allowModelOverride) {
-    params.respond(
+    params.io.emitAcceptance([
      false,
      undefined,
      errorShape(
        ErrorCodes.INVALID_REQUEST,
        "provider/model overrides are not authorized for this caller.",
      ),
-    );
+    ]);
    return undefined;
  }
  if (
    (requestedInternalSessionEffects || requestedPromptPersistenceSuppression) &&
    !canUseInternalRuntimeHandoff
  ) {
-    params.respond(
+    params.io.emitAcceptance([
      false,
      undefined,
      errorShape(
        ErrorCodes.INVALID_REQUEST,
        "internal session-effect controls are reserved for backend callers.",
      ),
-    );
+    ]);
    return undefined;
  }
  const runId = request.idempotencyKey;
  const execApprovalFollowupApprovalId = parseExecApprovalFollowupApprovalId(runId);
  if (execApprovalFollowupApprovalId && !canUseInternalRuntimeHandoff) {
-    params.respond(
+    params.io.emitAcceptance([
      false,
      undefined,
      errorShape(
        ErrorCodes.INVALID_REQUEST,
        "exec approval followup idempotency keys are reserved for backend callers.",
      ),
-    );
+    ]);
    return undefined;
  }
  const inputProvenance = normalizeInputProvenance(request.inputProvenance);
  const isRestartRecoveryResumeRun =
    canUseInternalRuntimeHandoff && isMainSessionRestartRecoveryInputProvenance(inputProvenance);
  if (request.internalExecutionIdentityRetry !== undefined && !isRestartRecoveryResumeRun) {
-    params.respond(
+    params.io.emitAcceptance([
      false,
      undefined,
      errorShape(
        ErrorCodes.INVALID_REQUEST,
        "internal execution identity retry mode is reserved for main-session restart recovery.",
      ),
-    );
+    ]);
    return undefined;
  }
  if (request.forceCodeModeTools === true && !isRestartRecoveryResumeRun) {
-    params.respond(
+    params.io.emitAcceptance([
      false,
      undefined,
      errorShape(
        ErrorCodes.INVALID_REQUEST,
        "forceCodeModeTools is reserved for main-session restart recovery.",
      ),
-    );
+    ]);
    return undefined;
  }
  const sessionEffects =
@@ -286,19 +289,21 @@ export function prepareAgentRequestPreflight(
        cached.payload.agentId.trim()
          ? cached.payload.agentId.trim()
          : undefined;
-      params.respond(
-        true,
-        {
-          runId: cachedRunId,
-          status: "in_flight" as const,
-          ...(cachedSessionKey ? { sessionKey: cachedSessionKey } : {}),
-          ...(cachedAgentId ? { agentId: cachedAgentId } : {}),
-        },
-        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.respond(cached.ok, cached.payload, cached.error, { cached: true });
+      params.io.emitAcceptance([cached.ok, cached.payload, cached.error], { cached: true });
    }
    return undefined;
  }
@@ -22,7 +22,7 @@ import type { OpenClawConfig } from "../../config/types.openclaw.js";
 import { claimAgentRunContext } from "../../infra/agent-run-registry.js";
 import type { InputProvenance } from "../../sessions/input-provenance.js";
 import type { SessionWorkAdmissionLease } from "../../sessions/session-lifecycle-admission.js";
-import type { AgentTurnContext, AgentTurnPrincipal } from "../agent-turn/types.js";
+import type { AgentTurnContext, AgentTurnIo, AgentTurnPrincipal } from "../agent-turn/types.js";
 import { registerChatAbortController, resolveAgentRunExpiresAtMs } from "../chat-abort.js";
 import { loadSessionEntry, resolveSessionModelRef } from "../session-utils.js";
 import { consumeSubagentCompletionToolHandoff } from "../subagent-completion-tool-handoff.js";
@@ -45,7 +45,6 @@ import {
  resolveGatewayCronCreatorAuthorityAdmission,
  type GatewayCronCreatorAuthorityAdmission,
 } from "./cron-creator-authority-admission.js";
-import type { GatewayRequestHandlerOptions } from "./types.js";

 export type PreparedAgentRunDispatch = {
  activeGatewayWorkAdmission: SessionWorkAdmissionLease;
@@ -96,7 +95,7 @@ export async function prepareAgentRunDispatch(params: {
  agentDedupeKeys: readonly string[];
  context: AgentTurnContext;
  client: AgentTurnPrincipal | null;
-  respond: GatewayRequestHandlerOptions["respond"];
+  io: AgentTurnIo;
  abortForLifecycleRotation: (target?: { sessionKey?: string; agentId?: string }) => boolean;
  acquireGatewayWorkAdmission: (scope: string) => Promise<void>;
  assertGatewayWorkAdmissionAllowed: () => void;
@@ -121,7 +120,7 @@ export async function prepareAgentRunDispatch(params: {
    })
  ) {
    params.markAgentRunAccepted(true);
-    params.respond(true, preRegistrationAbort?.payload, undefined, {
+    params.io.emitAcceptance([true, preRegistrationAbort?.payload, undefined], {
      cached: true,
      runId: params.runId,
    });
@@ -136,11 +135,11 @@ export async function prepareAgentRunDispatch(params: {
    return undefined;
  }
  if (params.restoredCronContinuationIdentity && !params.restoredCronContinuation) {
-    params.respond(
+    params.io.emitAcceptance([
      false,
      undefined,
      errorShape(ErrorCodes.UNAVAILABLE, "cron run continuation could not be restored"),
-    );
+    ]);
    return undefined;
  }

@@ -223,7 +222,11 @@ export async function prepareAgentRunDispatch(params: {
      );
    }
  } catch (err) {
-    params.respond(false, undefined, errorShape(ErrorCodes.INVALID_REQUEST, formatForLog(err)));
+    params.io.emitAcceptance([
+      false,
+      undefined,
+      errorShape(ErrorCodes.INVALID_REQUEST, formatForLog(err)),
+    ]);
    return undefined;
  }
  if (params.respondToGatewayAdmissionOutcome()) {
@@ -231,31 +234,34 @@ export async function prepareAgentRunDispatch(params: {
  }
  const activeGatewayWorkAdmission = params.getGatewayWorkAdmission();
  if (!activeGatewayWorkAdmission) {
-    params.respond(
+    params.io.emitAcceptance([
      false,
      undefined,
      errorShape(ErrorCodes.UNAVAILABLE, "agent run admission failed"),
-    );
+    ]);
    return undefined;
  }
  const activeRunAbort = params.getAdmittedRunAbort();
  if (!activeRunAbort) {
    activeGatewayWorkAdmission.release();
-    params.respond(
+    params.io.emitAcceptance([
      false,
      undefined,
      errorShape(ErrorCodes.UNAVAILABLE, "agent run admission failed"),
-    );
+    ]);
    return undefined;
  }
  const existingRunAbort = params.context.chatAbortControllers.get(params.runId);
  if (!activeRunAbort.registered && existingRunAbort) {
    activeGatewayWorkAdmission.release();
    params.markAgentRunAccepted(existingRunAbort.kind === "agent");
-    params.respond(true, { runId: params.runId, status: "in_flight" as const }, undefined, {
-      cached: true,
-      runId: params.runId,
-    });
+    params.io.emitAcceptance(
+      [true, { runId: params.runId, status: "in_flight" as const }, undefined],
+      {
+        cached: true,
+        runId: params.runId,
+      },
+    );
    return undefined;
  }
  if (!activeRunAbort.registered) {
@@ -332,14 +338,14 @@ export async function prepareAgentRunDispatch(params: {
      );
      activeRunAbort.cleanup({ force: true });
      activeGatewayWorkAdmission.release();
-      params.respond(
+      params.io.emitAcceptance([
        false,
        undefined,
        errorShape(
          ErrorCodes.UNAVAILABLE,
          `plugin subagent registry persistence failed; run was not started: ${formatForLog(err)}`,
        ),
-      );
+      ]);
      return undefined;
    }
  }
@@ -351,11 +357,11 @@ export async function prepareAgentRunDispatch(params: {
    if (!recoverySessionKey) {
      activeRunAbort.cleanup({ force: true });
      activeGatewayWorkAdmission.release();
-      params.respond(
+      params.io.emitAcceptance([
        false,
        undefined,
        errorShape(ErrorCodes.UNAVAILABLE, "restart recovery session target is unavailable"),
-      );
+      ]);
      return undefined;
    }
    try {
@@ -408,7 +414,11 @@ export async function prepareAgentRunDispatch(params: {
    } catch (err) {
      activeRunAbort.cleanup({ force: true });
      activeGatewayWorkAdmission.release();
-      params.respond(false, undefined, errorShape(ErrorCodes.UNAVAILABLE, formatForLog(err)));
+      params.io.emitAcceptance([
+        false,
+        undefined,
+        errorShape(ErrorCodes.UNAVAILABLE, formatForLog(err)),
+      ]);
      return undefined;
    }
  }
@@ -436,7 +446,7 @@ export async function prepareAgentRunDispatch(params: {
      },
    },
  });
-  params.respond(true, accepted, undefined, { runId: params.runId });
+  params.io.emitAcceptance([true, accepted, undefined], { runId: params.runId });
  const cronCreatorAuthority = resolveGatewayCronCreatorAuthorityAdmission({
    runId: params.runId,
    resolvedSessionKey: params.resolvedSessionKey,
@@ -17,7 +17,7 @@ import { defaultRuntime } from "../../runtime.js";
 import { createRunningTaskRun } from "../../tasks/detached-task-runtime.js";
 import { mapAgentRunTerminalOutcomeToTaskStatus } from "../../tasks/task-registry-common.js";
 import { normalizeDeliveryContext } from "../../utils/delivery-context.shared.js";
-import type { AgentTurnContext } from "../agent-turn/types.js";
+import type { AgentTurnContext, AgentTurnIo } from "../agent-turn/types.js";
 import type { ChatAbortControllerEntry } from "../chat-abort.js";
 import { formatForLog } from "../ws-log.js";
 import { setGatewayDedupeEntries } from "./agent-dedupe.js";
@@ -26,7 +26,6 @@ import {
  type GatewayAgentTaskTrackingMode,
 } from "./agent-task-tracking.js";
 import type { GatewayCronCreatorAuthorityAdmission } from "./cron-creator-authority-admission.js";
-import type { GatewayRequestHandlerOptions } from "./types.js";

 function resolveResolvedAgentTimeoutStopReason(
  meta: unknown,
@@ -116,7 +115,7 @@ export function dispatchAgentRunFromGateway(params: {
   */
  abortController: AbortController;
  cleanupAbortController: () => void;
-  respond: GatewayRequestHandlerOptions["respond"];
+  io: AgentTurnIo;
  context: AgentTurnContext;
  taskTrackingMode: Exclude<GatewayAgentTaskTrackingMode, "plugin_subagent">;
  restoreAdmittedRecovery?: () => Promise<MainSessionRecoveryPendingTarget | undefined>;
@@ -264,13 +263,16 @@ export function dispatchAgentRunFromGateway(params: {
          keys: params.dedupeKeys,
          entry: { ts: Date.now(), ok: false, payload: failedPayload, error },
        });
-        params.respond(false, failedPayload, error, { runId: params.runId, error: summary });
+        params.io.emitFinal([false, failedPayload, error], {
+          runId: params.runId,
+          error: summary,
+        });
        return;
      }
      persistTerminalDedupe();
      // Send a second res frame (same id) so TS clients with expectFinal can wait.
      // Swift clients will typically treat the first res as the result and ignore this.
-      params.respond(true, payload, undefined, { runId: params.runId });
+      params.io.emitFinal([true, payload, undefined], { runId: params.runId });
    })
    .catch(async (err: unknown) => {
      const aborted = isGatewayAgentAbortRejection(err, params.abortController.signal);
@@ -327,7 +329,7 @@ export function dispatchAgentRunFromGateway(params: {
        onRecovered: () => persistTerminalDedupe(true),
      });
      persistTerminalDedupe(settled);
-      params.respond(aborted && settled, payload, aborted && settled ? undefined : error, {
+      params.io.emitFinal([aborted && settled, payload, aborted && settled ? undefined : error], {
        runId: params.runId,
        ...(aborted ? {} : { error: formatForLog(err) }),
      });
@@ -40,7 +40,7 @@ import {
  buildRunUserTurnIdempotencyKey,
  createUserTurnTranscriptRecorder,
 } from "../../sessions/user-turn-transcript.js";
-import type { AgentTurnContext, AgentTurnPrincipal } from "../agent-turn/types.js";
+import type { AgentTurnContext, AgentTurnIo, AgentTurnPrincipal } from "../agent-turn/types.js";
 import { reactivateCompletedSubagentSession } from "../session-subagent-reactivation.js";
 import { loadSessionEntry } from "../session-utils.js";
 import { formatForLog } from "../ws-log.js";
@@ -66,7 +66,6 @@ import { createAgentRunModelSelectionHandler } from "./agent-run-model-selection
 import { resolveSessionRuntimeCwd } from "./agent-session-reset.js";
 import { gatewayClientSenderFields } from "./gateway-client-identity.js";
 import { emitSessionsChanged } from "./session-change-event.js";
-import type { GatewayRequestHandlerOptions } from "./types.js";

 export function startAgentRunExecution(params: {
  prepared: PreparedAgentRunDispatch;
@@ -112,7 +111,7 @@ export function startAgentRunExecution(params: {
  execApprovalFollowupApprovalId?: string;
  client: AgentTurnPrincipal | null;
  context: AgentTurnContext;
-  respond: GatewayRequestHandlerOptions["respond"];
+  io: AgentTurnIo;
  releaseCronContinuationClaimWithRecovery: (
    outcome?: { terminalOutcome: AgentRunTerminalOutcome },
    onRecovered?: () => void,
@@ -143,17 +142,19 @@ export function startAgentRunExecution(params: {
          runId: params.runId,
          stopReason,
        });
-        params.respond(
-          true,
-          {
-            runId: params.runId,
-            status: "timeout" as const,
-            summary: "aborted",
-            stopReason,
-            timeoutPhase: "queue" as const,
-            providerStarted: false,
-          },
-          undefined,
+        params.io.emitFinal(
+          [
+            true,
+            {
+              runId: params.runId,
+              status: "timeout" as const,
+              summary: "aborted",
+              stopReason,
+              timeoutPhase: "queue" as const,
+              providerStarted: false,
+            },
+            undefined,
+          ],
          { runId: params.runId },
        );
        return;
@@ -492,7 +493,7 @@ export function startAgentRunExecution(params: {
                onRecovered,
              )
          : undefined,
-        respond: params.respond,
+        io: params.io,
        context: params.context,
        taskTrackingMode: prepared.dispatchTaskTrackingMode,
        restoreAdmittedRecovery: prepared.restoreAdmittedRestartRecoveryInterrupted,
@@ -514,7 +515,7 @@ export function startAgentRunExecution(params: {
        keys: params.agentDedupeKeys,
        entry: { ts: Date.now(), ok: false, payload, error },
      });
-      params.respond(false, payload, error, {
+      params.io.emitFinal([false, payload, error], {
        runId: params.runId,
        error: formatForLog(err),
      });
@@ -8,6 +8,7 @@ import {
 import { mergeSessionEntry, type SessionEntry } from "../../config/sessions.js";
 import type { OpenClawConfig } from "../../config/types.openclaw.js";
 import { normalizeDeliveryContext } from "../../utils/delivery-context.shared.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";
@@ -34,7 +35,8 @@ export const agentRunHandler: GatewayRequestHandlers["agent"] = async ({
  client,
  isWebchatConnect,
 }) => {
-  const preflight = prepareAgentRequestPreflight({ params, respond, context, client });
+  const io = createAgentTurnIo(respond);
+  const preflight = prepareAgentRequestPreflight({ params, context, client, io });
  if (!preflight) {
    return;
  }
@@ -80,7 +82,7 @@ export const agentRunHandler: GatewayRequestHandlers["agent"] = async ({
    ownerConnId,
    ownerDeviceId,
    context,
-    respond,
+    io,
  });
  const reservePreAcceptedAgentDedupe = dedupeLifecycle.reserve;
  const clearUnacceptedAgentDedupe = dedupeLifecycle.clearUnaccepted;
@@ -190,7 +192,7 @@ export const agentRunHandler: GatewayRequestHandlers["agent"] = async ({
      preAcceptedReservedSessionKey,
      expectedSession,
      context,
-      respond,
+      io,
      dedupeLifecycle,
      getRequestedSessionKey: () => requestedSessionKey,
      getResolvedSessionKey: () => resolvedSessionKey,
@@ -283,7 +285,7 @@ export const agentRunHandler: GatewayRequestHandlers["agent"] = async ({
        agentId: canonicalSessionAgentId,
      });
      if (sharingError) {
-        respond(false, undefined, sharingError);
+        io.emitAcceptance([false, undefined, sharingError]);
        return;
      }
      effectiveBootstrapContextRunKind = preparedSession.effectiveBootstrapContextRunKind;
@@ -339,7 +341,11 @@ export const agentRunHandler: GatewayRequestHandlers["agent"] = async ({
      try {
        await acquireGatewayWorkAdmission(storePath ?? `agent:${sessionAgentId}`);
      } catch (err) {
-        respond(false, undefined, errorShape(ErrorCodes.INVALID_REQUEST, formatForLog(err)));
+        io.emitAcceptance([
+          false,
+          undefined,
+          errorShape(ErrorCodes.INVALID_REQUEST, formatForLog(err)),
+        ]);
        return;
      }
      if (respondToGatewayAdmissionOutcome()) {
@@ -461,7 +467,7 @@ export const agentRunHandler: GatewayRequestHandlers["agent"] = async ({
      agentDedupeKeys,
      context,
      client,
-      respond,
+      io,
      abortForLifecycleRotation,
      acquireGatewayWorkAdmission,
      assertGatewayWorkAdmissionAllowed,
@@ -522,7 +528,7 @@ export const agentRunHandler: GatewayRequestHandlers["agent"] = async ({
      execApprovalFollowupApprovalId,
      client,
      context,
-      respond,
+      io,
      releaseCronContinuationClaimWithRecovery,
    });
    mainRestartRecoveryOwnerLease = undefined;
@@ -4,6 +4,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { registerExecApprovalFollowupRuntimeHandoff } from "../../agents/bash-tools.exec-approval-followup-state.js";
 import type { InternalSessionEntry as SessionEntry } from "../../config/sessions.js";
 import { runExclusiveSessionLifecycleMutation } from "../../sessions/session-lifecycle-admission.js";
+import { createAgentTurnIo } from "../agent-turn/io.js";
 import { resolveAgentRunExpiresAtMs } from "../chat-abort.js";
 import { setGatewayDedupeEntry } from "./agent-job.js";
 import { prepareAgentRunDispatch } from "./agent-run-admission-phase.js";
@@ -1654,7 +1655,7 @@ describe("gateway agent handler chat.abort integration", () => {
        agentDedupeKeys: [`agent:${runId}`],
        context,
        client: null,
-        respond: vi.fn(),
+        io: createAgentTurnIo(vi.fn()),
        abortForLifecycleRotation: () => false,
        acquireGatewayWorkAdmission: async () => {
          markAcquireStarted();
@@ -18,6 +18,7 @@ import {
 } from "../../tasks/task-registry.js";
 import { setDetachedTaskLifecycleRuntime } from "../../tasks/task-runtime.test-helpers.js";
 import { withTempDir } from "../../test-helpers/temp-dir.js";
+import { createAgentTurnIo } from "../agent-turn/io.js";
 import { dispatchAgentRunFromGateway } from "./agent-run-dispatch.js";
 import { registerPluginSubagentRunFromGateway } from "./agent-task-tracking.js";
 import {
@@ -736,7 +737,7 @@ describe("gateway agent handler", () => {
      dedupeKeys: ["agent:agent-run-tool-use-deadline"],
      abortController,
      cleanupAbortController: vi.fn(),
-      respond,
+      io: createAgentTurnIo(respond),
      context,
      taskTrackingMode: "none",
      onSettled,
@@ -795,7 +796,7 @@ describe("gateway agent handler", () => {
      dedupeKeys: ["agent:agent-run-provider-timeout-result"],
      abortController: new AbortController(),
      cleanupAbortController: vi.fn(),
-      respond,
+      io: createAgentTurnIo(respond),
      context,
      taskTrackingMode: "none",
      onSettled,
@@ -849,7 +850,7 @@ describe("gateway agent handler", () => {
      dedupeKeys: ["agent:agent-run-resolved-error"],
      abortController: new AbortController(),
      cleanupAbortController: vi.fn(),
-      respond,
+      io: createAgentTurnIo(respond),
      context,
      taskTrackingMode: "none",
      onSettled,
@@ -952,7 +953,7 @@ describe("gateway agent handler", () => {
      dedupeKeys: [`agent:deadline-control-${stopReason}-${durationMs}`],
      abortController,
      cleanupAbortController: vi.fn(),
-      respond,
+      io: createAgentTurnIo(respond),
      context,
      taskTrackingMode: "none",
    });
@@ -1312,7 +1313,7 @@ describe("gateway agent handler", () => {
      dedupeKeys: [`agent:${runId}`],
      abortController: new AbortController(),
      cleanupAbortController: vi.fn(),
-      respond,
+      io: createAgentTurnIo(respond),
      context,
      taskTrackingMode: "none",
      onSettled,
@@ -1349,7 +1350,7 @@ describe("gateway agent handler", () => {
      dedupeKeys: ["agent:agent-run-provider-error-settlement"],
      abortController: new AbortController(),
      cleanupAbortController: vi.fn(),
-      respond,
+      io: createAgentTurnIo(respond),
      context,
      taskTrackingMode: "none",
      onSettled,