diff --git a/src/agents/subagent-spawn.in-process-gateway.test.ts b/src/agents/subagent-spawn.in-process-gateway.test.ts index 0385916a0e23..d6b4a7dc043f 100644 --- a/src/agents/subagent-spawn.in-process-gateway.test.ts +++ b/src/agents/subagent-spawn.in-process-gateway.test.ts @@ -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); diff --git a/src/gateway/agent-turn/io.ts b/src/gateway/agent-turn/io.ts new file mode 100644 index 000000000000..6c359b26c6ee --- /dev/null +++ b/src/gateway/agent-turn/io.ts @@ -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[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 }; +} diff --git a/src/gateway/agent-turn/types.ts b/src/gateway/agent-turn/types.ts index d1caaa914f59..dc8825a7c9df 100644 --- a/src/gateway/agent-turn/types.ts +++ b/src/gateway/agent-turn/types.ts @@ -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[0], + payload: Parameters[1], + error: Parameters[2], +]; + +type AgentTurnAcceptance = AgentTurnFrame; +type AgentTurnFinal = AgentTurnFrame; + +export type AgentTurnIo = { + emitAcceptance: (acceptance: AgentTurnAcceptance, meta?: Parameters[3]) => void; + emitFinal: (final: AgentTurnFinal, meta?: Parameters[3]) => void; +}; export type AgentTurnPrincipal = Pick< GatewayClient, diff --git a/src/gateway/server-methods/agent-admission-controller.ts b/src/gateway/server-methods/agent-admission-controller.ts index 40177b7ef129..3ea9b692796c 100644 --- a/src/gateway/server-methods/agent-admission-controller.ts +++ b/src/gateway/server-methods/agent-admission-controller.ts @@ -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; diff --git a/src/gateway/server-methods/agent-dedupe-lifecycle.ts b/src/gateway/server-methods/agent-dedupe-lifecycle.ts index f26988b539df..c638ae9e61cb 100644 --- a/src/gateway/server-methods/agent-dedupe-lifecycle.ts +++ b/src/gateway/server-methods/agent-dedupe-lifecycle.ts @@ -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; @@ -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; diff --git a/src/gateway/server-methods/agent-request-preflight.test.ts b/src/gateway/server-methods/agent-request-preflight.test.ts index cc72308eba60..2b57b325407a 100644 --- a/src/gateway/server-methods/agent-request-preflight.test.ts +++ b/src/gateway/server-methods/agent-request-preflight.test.ts @@ -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(), diff --git a/src/gateway/server-methods/agent-request-preflight.ts b/src/gateway/server-methods/agent-request-preflight.ts index b999ef9a2425..b293314369fe 100644 --- a/src/gateway/server-methods/agent-request-preflight.ts +++ b/src/gateway/server-methods/agent-request-preflight.ts @@ -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 & { - 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; } diff --git a/src/gateway/server-methods/agent-run-admission-phase.ts b/src/gateway/server-methods/agent-run-admission-phase.ts index d92c7ecd85d1..0bfdd1871d21 100644 --- a/src/gateway/server-methods/agent-run-admission-phase.ts +++ b/src/gateway/server-methods/agent-run-admission-phase.ts @@ -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; 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, diff --git a/src/gateway/server-methods/agent-run-dispatch.ts b/src/gateway/server-methods/agent-run-dispatch.ts index 1f092900f816..34ce7ddc47a8 100644 --- a/src/gateway/server-methods/agent-run-dispatch.ts +++ b/src/gateway/server-methods/agent-run-dispatch.ts @@ -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; restoreAdmittedRecovery?: () => Promise; @@ -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) }), }); diff --git a/src/gateway/server-methods/agent-run-execution-phase.ts b/src/gateway/server-methods/agent-run-execution-phase.ts index d39894ff91fd..43e2d34df407 100644 --- a/src/gateway/server-methods/agent-run-execution-phase.ts +++ b/src/gateway/server-methods/agent-run-execution-phase.ts @@ -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), }); diff --git a/src/gateway/server-methods/agent-run-handler.ts b/src/gateway/server-methods/agent-run-handler.ts index eb21fde5fa84..25835c1c6b12 100644 --- a/src/gateway/server-methods/agent-run-handler.ts +++ b/src/gateway/server-methods/agent-run-handler.ts @@ -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; diff --git a/src/gateway/server-methods/agent.abort-integration.test-utils.ts b/src/gateway/server-methods/agent.abort-integration.test-utils.ts index ed95a50fa14c..f17dc889adf7 100644 --- a/src/gateway/server-methods/agent.abort-integration.test-utils.ts +++ b/src/gateway/server-methods/agent.abort-integration.test-utils.ts @@ -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(); diff --git a/src/gateway/server-methods/agent.sessions-and-models.test-utils.ts b/src/gateway/server-methods/agent.sessions-and-models.test-utils.ts index 509d7ac003d5..ea4bba5c0bcc 100644 --- a/src/gateway/server-methods/agent.sessions-and-models.test-utils.ts +++ b/src/gateway/server-methods/agent.sessions-and-models.test-utils.ts @@ -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,