diff --git a/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift b/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift index 7bec7d75e0a8..be2c7babbbb0 100644 --- a/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift +++ b/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift @@ -15742,21 +15742,25 @@ public struct ApprovalResolveParams: Codable, Sendable { public let id: String public let kind: ApprovalKind public let decision: ApprovalDecision + public let reviewer: [String: AnyCodable]? public init( id: String, kind: ApprovalKind, - decision: ApprovalDecision) + decision: ApprovalDecision, + reviewer: [String: AnyCodable]? = nil) { self.id = id self.kind = kind self.decision = decision + self.reviewer = reviewer } private enum CodingKeys: String, CodingKey { case id case kind case decision + case reviewer } } @@ -16143,18 +16147,22 @@ public struct ExecApprovalRequestParams: Codable, Sendable { public struct ExecApprovalResolveParams: Codable, Sendable { public let id: String public let decision: String + public let reviewer: [String: AnyCodable]? public init( id: String, - decision: String) + decision: String, + reviewer: [String: AnyCodable]? = nil) { self.id = id self.decision = decision + self.reviewer = reviewer } private enum CodingKeys: String, CodingKey { case id case decision + case reviewer } } @@ -16513,18 +16521,22 @@ public struct PluginApprovalRequestParams: Codable, Sendable { public struct PluginApprovalResolveParams: Codable, Sendable { public let id: String public let decision: String + public let reviewer: [String: AnyCodable]? public init( id: String, - decision: String) + decision: String, + reviewer: [String: AnyCodable]? = nil) { self.id = id self.decision = decision + self.reviewer = reviewer } private enum CodingKeys: String, CodingKey { case id case decision + case reviewer } } diff --git a/src/gateway/server-worker-placement-startup.ts b/src/gateway/server-worker-placement-startup.ts index 04653fed3742..1a0f6b7ef42f 100644 --- a/src/gateway/server-worker-placement-startup.ts +++ b/src/gateway/server-worker-placement-startup.ts @@ -56,6 +56,58 @@ class WorkerDispatchTargetChangedError extends Error { readonly code = "invalid_state"; } +type WorkerPlacementSessionRuntime = Awaited< + ReturnType +>; +type WorkerPlacementSessionTarget = ReturnType< + WorkerPlacementSessionRuntime["resolveGatewaySessionStoreTargetWithStore"] +>; + +/** Keeps store identity, session incarnation, canonical ownership, and the live worktree + * in one cross-phase fence. Initial resolution throws normally; barrier revalidation + * supplies expectedTarget and yields an invalid_state retry when the target changed. */ +function resolveWorkerPlacementSessionTarget(params: { + sessionRuntime: WorkerPlacementSessionRuntime; + config: ReturnType; + sessionId: string; + sessionKey: string; + agentId: string; + expectedTarget?: WorkerPlacementSessionTarget; + errorMessage: string; +}) { + const target = params.sessionRuntime.resolveGatewaySessionStoreTargetWithStore({ + cfg: params.config, + key: params.sessionKey, + agentId: params.agentId, + clone: false, + }); + const entry = params.sessionRuntime.resolveCanonicalSessionEntryFromStoreKeys( + target.store, + target.storeKeys, + ); + const worktree = params.sessionRuntime.managedWorktrees.findLiveByOwner( + "session", + target.canonicalKey, + ); + const expected = params.expectedTarget; + if ( + (expected && + (target.storePath !== expected.storePath || + target.canonicalKey !== expected.canonicalKey || + target.agentId !== expected.agentId)) || + entry?.sessionId !== params.sessionId || + !entry.worktree?.id || + !worktree || + worktree.id !== entry.worktree.id || + worktree.ownerId !== target.canonicalKey + ) { + throw expected + ? new WorkerDispatchTargetChangedError(params.errorMessage) + : new Error(params.errorMessage); + } + return { config: params.config, target, entry, worktree }; +} + /** Serializes reconciliation sweeps against in-flight dispatches so a sweep never * observes a placement mid-transition. Dispatches wait out any pending sweep. */ export function coordinateWorkerPlacementDispatch( @@ -200,28 +252,15 @@ export function createGatewayWorkerPlacementRuntime(params: GatewayWorkerPlaceme sessionKey: string; agentId: string; }): Promise => { - const { - managedWorktrees, - resolveCanonicalSessionEntryFromStoreKeys, - resolveGatewaySessionStoreTargetWithStore, - } = await loadWorkerPlacementSessionRuntimeModule(); - const target = resolveGatewaySessionStoreTargetWithStore({ - cfg: getRuntimeConfig(), - key: sessionKey, + const sessionRuntime = await loadWorkerPlacementSessionRuntimeModule(); + const { worktree } = resolveWorkerPlacementSessionTarget({ + sessionRuntime, + config: getRuntimeConfig(), + sessionId, + sessionKey, agentId, - clone: false, + errorMessage: `Session ${sessionKey} dispatch requires a session-owned managed worktree`, }); - const sessionEntry = resolveCanonicalSessionEntryFromStoreKeys(target.store, target.storeKeys); - const worktree = managedWorktrees.findLiveByOwner("session", target.canonicalKey); - if ( - sessionEntry?.sessionId !== sessionId || - !sessionEntry.worktree?.id || - !worktree || - worktree.id !== sessionEntry.worktree.id || - worktree.ownerId !== target.canonicalKey - ) { - throw new Error(`Session ${sessionKey} dispatch requires a session-owned managed worktree`); - } return worktree.path; }; const dispatchService = coordinateWorkerPlacementDispatch( @@ -230,13 +269,12 @@ export function createGatewayWorkerPlacementRuntime(params: GatewayWorkerPlaceme environments: params.environments, ...workspaceConflictHandlers, runLocalBarrier: async ({ sessionId, sessionKey, agentId, startDispatch }) => { + const sessionRuntime = await loadWorkerPlacementSessionRuntimeModule(); const { isWorkerPlacementSessionRuntimeSupported, - managedWorktrees, - resolveCanonicalSessionEntryFromStoreKeys, resolveGatewaySessionStoreTargetWithStore, resolveWorkerPlacementSessionRuntime, - } = await loadWorkerPlacementSessionRuntimeModule(); + } = sessionRuntime; const target = resolveGatewaySessionStoreTargetWithStore({ cfg: getRuntimeConfig(), key: sessionKey, @@ -254,35 +292,20 @@ export function createGatewayWorkerPlacementRuntime(params: GatewayWorkerPlaceme scope: target.storePath, identities: lifecycleIdentities, prepare: async () => { - const currentConfig = getRuntimeConfig(); - const currentTarget = resolveGatewaySessionStoreTargetWithStore({ - cfg: currentConfig, - key: sessionKey, + const { + config: currentConfig, + target: currentTarget, + entry: currentEntry, + worktree, + } = resolveWorkerPlacementSessionTarget({ + sessionRuntime, + config: getRuntimeConfig(), + sessionId, + sessionKey, agentId, - clone: false, + expectedTarget: target, + errorMessage: `Session ${sessionKey} changed before cloud worker dispatch. Retry.`, }); - const currentEntry = resolveCanonicalSessionEntryFromStoreKeys( - currentTarget.store, - currentTarget.storeKeys, - ); - const worktree = managedWorktrees.findLiveByOwner( - "session", - currentTarget.canonicalKey, - ); - if ( - currentTarget.storePath !== target.storePath || - currentTarget.canonicalKey !== target.canonicalKey || - currentTarget.agentId !== target.agentId || - currentEntry?.sessionId !== sessionId || - !currentEntry.worktree?.id || - !worktree || - worktree.id !== currentEntry.worktree.id || - worktree.ownerId !== currentTarget.canonicalKey - ) { - throw new WorkerDispatchTargetChangedError( - `Session ${sessionKey} changed before cloud worker dispatch. Retry.`, - ); - } if (currentEntry.archivedAt !== undefined) { throw new WorkerDispatchTargetChangedError( `Session ${sessionKey} was archived before cloud worker dispatch. Retry.`, @@ -334,13 +357,12 @@ export function createGatewayWorkerPlacementRuntime(params: GatewayWorkerPlaceme return placement; }, runActivationBarrier: async ({ sessionId, sessionKey, agentId, activate }) => { + const sessionRuntime = await loadWorkerPlacementSessionRuntimeModule(); const { isWorkerPlacementSessionRuntimeSupported, - managedWorktrees, - resolveCanonicalSessionEntryFromStoreKeys, resolveGatewaySessionStoreTargetWithStore, resolveWorkerPlacementSessionRuntime, - } = await loadWorkerPlacementSessionRuntimeModule(); + } = sessionRuntime; const target = resolveGatewaySessionStoreTargetWithStore({ cfg: getRuntimeConfig(), key: sessionKey, @@ -358,35 +380,19 @@ export function createGatewayWorkerPlacementRuntime(params: GatewayWorkerPlaceme scope: target.storePath, identities: lifecycleIdentities, run: async () => { - const currentConfig = getRuntimeConfig(); - const currentTarget = resolveGatewaySessionStoreTargetWithStore({ - cfg: currentConfig, - key: sessionKey, + const { + config: currentConfig, + target: currentTarget, + entry: currentEntry, + } = resolveWorkerPlacementSessionTarget({ + sessionRuntime, + config: getRuntimeConfig(), + sessionId, + sessionKey, agentId, - clone: false, + expectedTarget: target, + errorMessage: `Session ${sessionKey} changed before cloud worker activation. Retry.`, }); - const currentEntry = resolveCanonicalSessionEntryFromStoreKeys( - currentTarget.store, - currentTarget.storeKeys, - ); - const worktree = managedWorktrees.findLiveByOwner( - "session", - currentTarget.canonicalKey, - ); - if ( - currentTarget.storePath !== target.storePath || - currentTarget.canonicalKey !== target.canonicalKey || - currentTarget.agentId !== target.agentId || - currentEntry?.sessionId !== sessionId || - !currentEntry.worktree?.id || - !worktree || - worktree.id !== currentEntry.worktree.id || - worktree.ownerId !== currentTarget.canonicalKey - ) { - throw new WorkerDispatchTargetChangedError( - `Session ${sessionKey} changed before cloud worker activation. Retry.`, - ); - } if (currentEntry.archivedAt !== undefined) { throw new WorkerDispatchTargetChangedError( `Session ${sessionKey} was archived before cloud worker activation. Retry.`, @@ -412,11 +418,8 @@ export function createGatewayWorkerPlacementRuntime(params: GatewayWorkerPlaceme return activePlacement; }, runReclaimBarrier: async ({ sessionId, sessionKey, agentId, reclaim }) => { - const { - managedWorktrees, - resolveCanonicalSessionEntryFromStoreKeys, - resolveGatewaySessionStoreTargetWithStore, - } = await loadWorkerPlacementSessionRuntimeModule(); + const sessionRuntime = await loadWorkerPlacementSessionRuntimeModule(); + const { resolveGatewaySessionStoreTargetWithStore } = sessionRuntime; const target = resolveGatewaySessionStoreTargetWithStore({ cfg: getRuntimeConfig(), key: sessionKey, @@ -435,34 +438,15 @@ export function createGatewayWorkerPlacementRuntime(params: GatewayWorkerPlaceme scope: target.storePath, identities: lifecycleIdentities, prepare: async () => { - const currentTarget = resolveGatewaySessionStoreTargetWithStore({ - cfg: getRuntimeConfig(), - key: sessionKey, + const { worktree } = resolveWorkerPlacementSessionTarget({ + sessionRuntime, + config: getRuntimeConfig(), + sessionId, + sessionKey, agentId, - clone: false, + expectedTarget: target, + errorMessage: `Session ${sessionKey} changed before cloud worker stop. Retry.`, }); - const currentEntry = resolveCanonicalSessionEntryFromStoreKeys( - currentTarget.store, - currentTarget.storeKeys, - ); - const worktree = managedWorktrees.findLiveByOwner( - "session", - currentTarget.canonicalKey, - ); - if ( - currentTarget.storePath !== target.storePath || - currentTarget.canonicalKey !== target.canonicalKey || - currentTarget.agentId !== target.agentId || - currentEntry?.sessionId !== sessionId || - !currentEntry.worktree?.id || - !worktree || - worktree.id !== currentEntry.worktree.id || - worktree.ownerId !== currentTarget.canonicalKey - ) { - throw new WorkerDispatchTargetChangedError( - `Session ${sessionKey} changed before cloud worker stop. Retry.`, - ); - } const placement = params.placements.get(sessionId); if (placement?.state !== "active" || placement.turnClaim) { throw new Error( diff --git a/src/gateway/worker-environments/placement-dispatch-failure.ts b/src/gateway/worker-environments/placement-dispatch-failure.ts index 576fba13d3d3..8fc8a2fb3713 100644 --- a/src/gateway/worker-environments/placement-dispatch-failure.ts +++ b/src/gateway/worker-environments/placement-dispatch-failure.ts @@ -232,11 +232,6 @@ export function createPlacementFailureActions(deps: { environment: ReturnType, claimedTurnError: Error, ): Promise => { - if (placement.turnClaim) { - const draining = startDrain(placement); - await failDraining(draining, claimedTurnError, { forceClaimFence: true }); - return; - } const draining = startDrain(placement); if (draining.turnClaim) { await failDraining(draining, claimedTurnError, { forceClaimFence: true });