From 0b1fc0e27990bb1a7ff547a234b1785433949cdc Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sun, 9 Aug 2026 12:45:38 -0700 Subject: [PATCH] refactor(gateway): extract agent turn service (#121215) --- .../subagent-spawn.in-process-gateway.test.ts | 4 +- src/gateway/agent-turn/agent-turn-service.ts | 683 ++++++++++++++++++ src/gateway/agent-turn/types.ts | 1 - .../server-methods/agent-delivery-phase.ts | 15 +- .../agent-request-preflight.test.ts | 91 ++- .../server-methods/agent-request-preflight.ts | 64 +- .../server-methods/agent-run-handler.ts | 582 ++------------- .../server-methods/agent-wait-dedupe.test.ts | 37 +- src/gateway/server-methods/agent-wait.ts | 52 +- src/gateway/server-methods/chat.ts | 6 +- 10 files changed, 839 insertions(+), 696 deletions(-) create mode 100644 src/gateway/agent-turn/agent-turn-service.ts diff --git a/src/agents/subagent-spawn.in-process-gateway.test.ts b/src/agents/subagent-spawn.in-process-gateway.test.ts index d6b4a7dc043f..23a39ebd8714 100644 --- a/src/agents/subagent-spawn.in-process-gateway.test.ts +++ b/src/agents/subagent-spawn.in-process-gateway.test.ts @@ -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, diff --git a/src/gateway/agent-turn/agent-turn-service.ts b/src/gateway/agent-turn/agent-turn-service.ts new file mode 100644 index 000000000000..fa829795e79f --- /dev/null +++ b/src/gateway/agent-turn/agent-turn-service.ts @@ -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) { + const startTurn = async ({ + preflight, + principal, + io, + onRunObserved, + }: AgentTurnStartRequest): Promise => { + 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 + | 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> = + 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 => { + await handleChatAbortRequest({ ...options, context, isWebchatConnect }); + }; + + return { startTurn, waitForTurn, abortTurn }; +} diff --git a/src/gateway/agent-turn/types.ts b/src/gateway/agent-turn/types.ts index dc8825a7c9df..b2bda7a3e7f8 100644 --- a/src/gateway/agent-turn/types.ts +++ b/src/gateway/agent-turn/types.ts @@ -41,5 +41,4 @@ export type AgentTurnContext = Pick< | "loadGatewayModelCatalog" | "loadGatewayModelCatalogSnapshot" | "logGateway" - | "registerToolEventRecipient" >; diff --git a/src/gateway/server-methods/agent-delivery-phase.ts b/src/gateway/server-methods/agent-delivery-phase.ts index c53a7296f179..8133052c029c 100644 --- a/src/gateway/server-methods/agent-delivery-phase.ts +++ b/src/gateway/server-methods/agent-delivery-phase.ts @@ -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 { 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); } } } diff --git a/src/gateway/server-methods/agent-request-preflight.test.ts b/src/gateway/server-methods/agent-request-preflight.test.ts index 2b57b325407a..b8b46fb38d8b 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 { 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, diff --git a/src/gateway/server-methods/agent-request-preflight.ts b/src/gateway/server-methods/agent-request-preflight.ts index b293314369fe..3fe57edb4764 100644 --- a/src/gateway/server-methods/agent-request-preflight.ts +++ b/src/gateway/server-methods/agent-request-preflight.ts @@ -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; 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, diff --git a/src/gateway/server-methods/agent-run-handler.ts b/src/gateway/server-methods/agent-run-handler.ts index 25835c1c6b12..9a04ee81f660 100644 --- a/src/gateway/server-methods/agent-run-handler.ts +++ b/src/gateway/server-methods/agent-run-handler.ts @@ -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 - | 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> = - undefined; - try { - pendingRecovery = await releaseMainSessionRecoveryOwner(mainRestartRecoveryOwnerLease); - } finally { - try { - releaseGatewayAdmission(); - } finally { - try { - await releaseCronContinuationClaimWithRecovery(); - } finally { - scheduleMainSessionRecoveryPendingTarget(pendingRecovery); - } - } - } - } - } finally { - clearUnacceptedAgentDedupe(); - } - } }; diff --git a/src/gateway/server-methods/agent-wait-dedupe.test.ts b/src/gateway/server-methods/agent-wait-dedupe.test.ts index f4ba3a21cc13..364a582be3bc 100644 --- a/src/gateway/server-methods/agent-wait-dedupe.test.ts +++ b/src/gateway/server-methods/agent-wait-dedupe.test.ts @@ -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[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(); + 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(); diff --git a/src/gateway/server-methods/agent-wait.ts b/src/gateway/server-methods/agent-wait.ts index 0a9dc185a272..6d6c1d1d6675 100644 --- a/src/gateway/server-methods/agent-wait.ts +++ b/src/gateway/server-methods/agent-wait.ts @@ -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); }; diff --git a/src/gateway/server-methods/chat.ts b/src/gateway/server-methods/chat.ts index 48595251f48b..6fcb083c950a 100644 --- a/src/gateway/server-methods/chat.ts +++ b/src/gateway/server-methods/chat.ts @@ -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)) {