diff --git a/qa/scenarios/channels/channel-participant-identity-inspection.yaml b/qa/scenarios/channels/channel-participant-identity-inspection.yaml index 8e8d1ec16f9b..d9d295ff35d9 100644 --- a/qa/scenarios/channels/channel-participant-identity-inspection.yaml +++ b/qa/scenarios/channels/channel-participant-identity-inspection.yaml @@ -34,7 +34,7 @@ scenario: - Direct and group runs project a person invoker while room and route identifiers never become principals. - Senderless input is admitted with an unknown invoker, and an allowlist rejection creates no run, identity context, or decision fact. - Same and mixed participants remain correctly distinguished across collect-configured QA Channel ingress. - - JSON and human CLI inspection remain stable after a Gateway replacement restart. + - JSON and human CLI inspection remain stable after a Gateway replacement restart without exposing private generic decision facts as trusted owner evidence. docsRefs: - docs/gateway/audit.md - docs/cli/audit.md @@ -134,8 +134,8 @@ flow: - 60000 - 250 - assert: - expr: "groupInspect.identity.context.invoker.state === 'present' && groupInspect.decisions.some((receipt) => receipt.action.family === 'channel' && receipt.enforcement.coverageState === 'enforced') && !JSON.stringify(groupInspect).includes('qa-group-room')" - message: group allowlist admission must record enforced participant evidence without the room as principal + expr: "groupInspect.identity.context.invoker.state === 'present' && groupInspect.coverage.state === 'unknown' && groupInspect.coverage.missingEvidence.includes('decision.display_provenance') && groupInspect.decisionDisplays.some((receipt) => receipt.action.family === 'decision' && receipt.provenance.state === 'unverified') && !JSON.stringify(groupInspect).includes('qa-group-room')" + message: group inspection must retain participant identity without presenting private generic facts or the room as trusted owner evidence - set: senderlessRunsBefore value: expr: "[...new Set((await runQaCli(env, ['audit', '--kind', 'agent_run', '--limit', '500', '--json'], { timeoutMs: 60000, json: true })).events.map((event) => event.runId).filter((id) => typeof id === 'string'))]" @@ -164,7 +164,7 @@ flow: - 60000 - 250 - assert: - expr: "senderlessInspect.identity.context.invoker.state === 'unknown' && senderlessInspect.decisions.some((receipt) => receipt.action.family === 'channel' && receipt.enforcement.coverageState === 'unknown')" + expr: "senderlessInspect.identity.context.invoker.state === 'unknown' && senderlessInspect.coverage.state === 'unknown' && senderlessInspect.decisionDisplays.some((receipt) => receipt.action.family === 'decision' && receipt.provenance.state === 'unverified')" message: senderless admission must stay unknown detailsExpr: "`dm=${dmRunId}; group=${groupRunId}; senderless=${senderlessRunId}`" diff --git a/src/agents/tools/in-process-gateway.test.ts b/src/agents/tools/in-process-gateway.test.ts index 6104eeb33e6c..427c859076a2 100644 --- a/src/agents/tools/in-process-gateway.test.ts +++ b/src/agents/tools/in-process-gateway.test.ts @@ -71,6 +71,35 @@ describe("trusted in-process Gateway session creation", () => { ); }); + it("uses an explicitly bound Gateway when worker creation has no ambient request scope", async () => { + mocks.hasContext = false; + const admitted = {} as GatewayRequestContext; + const resolveGatewayContext = () => admitted; + const sessionMutationCommitGuard = vi.fn(); + const creation = { + via: "spawn" as const, + actor: { type: "agent" as const, id: "main" }, + requesterSessionKey: "agent:main:dashboard:worker", + inheritedToolPolicy: { version: 1 as const, allow: ["sessions_spawn"], deny: [] }, + }; + + await callInProcessGatewayToolWithCreation("sessions.create", { agentId: "main" }, creation, { + resolveGatewayContext, + sessionMutationCommitGuard, + }); + + expect(mocks.dispatch).toHaveBeenCalledWith( + "sessions.create", + { agentId: "main" }, + expect.objectContaining({ + resolveGatewayContext: expect.any(Function), + sessionMutationCommitGuard, + sessionCreation: creation, + }), + ); + expect(mocks.callGatewayTool).not.toHaveBeenCalled(); + }); + it("carries visible-spawn policy through signed identity on fallback dispatch", async () => { mocks.hasContext = false; const inheritedToolPolicy = { diff --git a/src/agents/tools/in-process-gateway.ts b/src/agents/tools/in-process-gateway.ts index e9b91aec4011..7c8c952b426b 100644 --- a/src/agents/tools/in-process-gateway.ts +++ b/src/agents/tools/in-process-gateway.ts @@ -21,6 +21,7 @@ import { callGatewayTool } from "./gateway.js"; type InProcessGatewayCallOptions = { resolveGatewayContext?: GatewayContextResolver; + sessionMutationCommitGuard?: () => void; }; export type InProcessGatewayCaller = >( @@ -198,11 +199,16 @@ export const callAgentToolGatewayRequest: AgentToolGatewayRequestCaller = async ); }; -export const callInProcessGatewayTool: InProcessGatewayCaller = async ( +async function callInProcessGatewayToolBound( method: string, params: Record, - options: InProcessGatewayCallOptions = {}, -): Promise => { + options: InProcessGatewayCallOptions & { + sessionCreation?: TrustedSessionCreation; + signal?: AbortSignal; + timeoutMs?: number | null; + }, + fallback: (scopes: ReturnType) => Promise, +): Promise { const scopes = resolveLeastPrivilegeOperatorScopesForMethod(method, params); const resolveGatewayContext = callerGatewayContextResolver(options.resolveGatewayContext); const boundGateway = resolveGatewayContext @@ -215,35 +221,10 @@ export const callInProcessGatewayTool: InProcessGatewayCaller = async ( await dispatchGatewayMethodInProcess(method, params, { forceSyntheticClient: true, syntheticScopes: scopes, - ...(boundResolver ? { resolveGatewayContext: boundResolver } : {}), - }), - ); - } - if (boundGateway) { - throw new Error(`Gateway instance unavailable for ${method}`); - } - return await callGatewayTool(method, {}, params, { scopes }); -}; - -export async function callInProcessGatewayToolWithCreation>( - method: string, - params: Record, - creation: TrustedSessionCreation, - options: { signal?: AbortSignal; timeoutMs?: number | null } = {}, -): Promise { - const scopes = resolveLeastPrivilegeOperatorScopesForMethod(method, params); - const resolveGatewayContext = callerGatewayContextResolver(); - const boundGateway = resolveGatewayContext - ? bindInProcessGatewayContext(method, resolveGatewayContext) - : undefined; - if (hasInProcessGatewayContext(boundGateway?.resolve)) { - return await runBoundInProcessGatewayCall( - boundGateway, - async (boundResolver) => - await dispatchGatewayMethodInProcess(method, params, { - forceSyntheticClient: true, - sessionCreation: creation, - syntheticScopes: scopes, + ...(options.sessionCreation ? { sessionCreation: options.sessionCreation } : {}), + ...(options.sessionMutationCommitGuard + ? { sessionMutationCommitGuard: options.sessionMutationCommitGuard } + : {}), ...(options.signal ? { signal: options.signal } : {}), ...(options.timeoutMs !== undefined && options.timeoutMs !== null ? { timeoutMs: options.timeoutMs } @@ -255,28 +236,59 @@ export async function callInProcessGatewayToolWithCreation(method, {}, params, { - scopes, - ...(options.signal ? { signal: options.signal } : {}), - ...(options.timeoutMs !== undefined ? { timeoutMs: options.timeoutMs } : {}), - }); - } - return await runWithGatewaySessionSpawnContext( - { - ...(creation.completionOwnerSessionKey - ? { completionOwnerSessionKey: creation.completionOwnerSessionKey } - : {}), - inheritedToolPolicy: creation.inheritedToolPolicy, + return await fallback(scopes); +} + +export const callInProcessGatewayTool: InProcessGatewayCaller = async ( + method: string, + params: Record, + options: InProcessGatewayCallOptions = {}, +): Promise => { + return await callInProcessGatewayToolBound(method, params, options, async (scopes) => + callGatewayTool(method, {}, params, { scopes }), + ); +}; + +export async function callInProcessGatewayToolWithCreation>( + method: string, + params: Record, + creation: TrustedSessionCreation, + options: { + resolveGatewayContext?: GatewayContextResolver; + sessionMutationCommitGuard?: () => void; + signal?: AbortSignal; + timeoutMs?: number | null; + } = {}, +): Promise { + return await callInProcessGatewayToolBound( + method, + params, + { ...options, sessionCreation: creation }, + async (scopes) => { + // The fallback is a real local Gateway request. Carry spawn policy only in + // the signed agent-runtime identity token, never in model-authored params. + if (creation.via !== "spawn" || !creation.inheritedToolPolicy) { + return await callGatewayTool(method, {}, params, { + scopes, + ...(options.signal ? { signal: options.signal } : {}), + ...(options.timeoutMs !== undefined ? { timeoutMs: options.timeoutMs } : {}), + }); + } + return await runWithGatewaySessionSpawnContext( + { + ...(creation.completionOwnerSessionKey + ? { completionOwnerSessionKey: creation.completionOwnerSessionKey } + : {}), + inheritedToolPolicy: creation.inheritedToolPolicy, + }, + () => + callGatewayTool(method, {}, params, { + scopes, + requireAgentRuntimeIdentity: true, + ...(options.signal ? { signal: options.signal } : {}), + ...(options.timeoutMs !== undefined ? { timeoutMs: options.timeoutMs } : {}), + }), + ); }, - () => - callGatewayTool(method, {}, params, { - scopes, - requireAgentRuntimeIdentity: true, - ...(options.signal ? { signal: options.signal } : {}), - ...(options.timeoutMs !== undefined ? { timeoutMs: options.timeoutMs } : {}), - }), ); } diff --git a/src/gateway/server-in-process-dispatch.ts b/src/gateway/server-in-process-dispatch.ts index 9938cc552ef3..5b7b92d7a1b1 100644 --- a/src/gateway/server-in-process-dispatch.ts +++ b/src/gateway/server-in-process-dispatch.ts @@ -22,6 +22,7 @@ type InProcessGatewayDispatchOptions = { onAccepted?: (payload: unknown) => void; onSignalAbort?: () => Promise | void; requestIdPrefix?: string; + sessionMutationCommitGuard?: () => void; timeoutMs?: number; signal?: AbortSignal; }; @@ -182,6 +183,7 @@ export async function dispatchGatewayRequestInProcessRaw( }, context: options.context, methodRegistry: options.methodRegistry, + sessionMutationCommitGuard: options.sessionMutationCommitGuard, ...(options.signal ? { signal: options.signal } : {}), }) .then(() => { diff --git a/src/gateway/server-methods.ts b/src/gateway/server-methods.ts index 0b73977ebc10..4242c7474ead 100644 --- a/src/gateway/server-methods.ts +++ b/src/gateway/server-methods.ts @@ -702,6 +702,9 @@ export async function handleGatewayRequest( respond, context, ...(signal ? { signal } : {}), + ...(opts.sessionMutationCommitGuard + ? { sessionMutationCommitGuard: opts.sessionMutationCommitGuard } + : {}), ...(authorization.sessionMutationAuthorization ? { sessionMutationAuthorization: authorization.sessionMutationAuthorization } : {}), diff --git a/src/gateway/server-methods/sessions-create.ts b/src/gateway/server-methods/sessions-create.ts index c33134dfe240..0062209635c7 100644 --- a/src/gateway/server-methods/sessions-create.ts +++ b/src/gateway/server-methods/sessions-create.ts @@ -67,6 +67,7 @@ export const sessionCreateHandlers: GatewayRequestHandlers = { context, client, isWebchatConnect, + sessionMutationCommitGuard, sessionMutationAuthorization, }) => { if (!assertValidParams(params, validateSessionsCreateParams, "sessions.create", respond)) { @@ -78,8 +79,9 @@ export const sessionCreateHandlers: GatewayRequestHandlers = { const cfg = context.getRuntimeConfig(); const authority = createAgentRuntimeAuthorityGuard(client, context, respond); const commitGuard = - authority.commitGuard || sessionMutationAuthorization + authority.commitGuard || sessionMutationCommitGuard || sessionMutationAuthorization ? () => { + sessionMutationCommitGuard?.(); authority.commitGuard?.(); sessionMutationAuthorization?.assertCurrent(); } diff --git a/src/gateway/server-methods/shared-types.ts b/src/gateway/server-methods/shared-types.ts index 720541ce2d5c..6ff4608334d5 100644 --- a/src/gateway/server-methods/shared-types.ts +++ b/src/gateway/server-methods/shared-types.ts @@ -451,6 +451,8 @@ export type GatewayRequestOptions = { respond: RespondFn; context: GatewayRequestContext; methodRegistry?: GatewayMethodRegistryView; + /** In-process Gateway lifetime guard composed into durable session mutations. */ + sessionMutationCommitGuard?: () => void; /** In-process caller lifetime; never serialized into a Gateway request frame. */ signal?: AbortSignal; }; @@ -469,6 +471,7 @@ export type GatewayRequestHandlerOptions = { isWebchatConnect: (params: ConnectParams | null | undefined) => boolean; respond: RespondFn; context: GatewayRequestContext; + sessionMutationCommitGuard?: () => void; sessionMutationAuthorization?: SessionMutationAuthorization; /** In-process caller lifetime; absent for ordinary transport requests. */ signal?: AbortSignal; diff --git a/src/gateway/server-plugin-in-process-dispatch.authorization.test.ts b/src/gateway/server-plugin-in-process-dispatch.authorization.test.ts index 2297c662b9a3..13af63ee665e 100644 --- a/src/gateway/server-plugin-in-process-dispatch.authorization.test.ts +++ b/src/gateway/server-plugin-in-process-dispatch.authorization.test.ts @@ -8,6 +8,7 @@ import { PROTOCOL_VERSION } from "../../packages/gateway-protocol/src/version.js import { createOperationalRunInstanceRef } from "../agents/admitted-run-context.js"; import { upsertSessionEntryCore } from "../config/sessions/session-accessor.js"; import { withPluginRuntimeGatewayRequestScope } from "../plugins/runtime/gateway-request-scope.js"; +import { createDeferredCore } from "../shared/deferred.js"; import { withOpenClawTestState } from "../test-utils/openclaw-test-state.js"; import { createGatewayMethodRegistry } from "./methods/registry.js"; import { resolveNodeInvokeRuntimeAuthorityError } from "./server-methods/nodes.invoke-authority.js"; @@ -192,6 +193,79 @@ describe("typed in-process agent authorization", () => { ); }); + it("carries an explicit Gateway binding into the session mutation commit guard", async () => { + const admitted = createContext(); + const replacement = createContext(); + let current = admitted; + let committed = false; + const setupStarted = createDeferredCore(); + const releaseSetup = createDeferredCore(); + admitted.getGatewayMethodRegistry = () => + createGatewayMethodRegistry([ + { + name: "sessions.create", + scope: "operator.write", + owner: { kind: "core", area: "sessions" }, + handler: async ({ + respond, + sessionMutationCommitGuard, + }: GatewayRequestHandlerOptions) => { + setupStarted.resolve(); + await releaseSetup.promise; + sessionMutationCommitGuard?.(); + committed = true; + respond(true, { key: "agent:main:dashboard:child" }); + }, + }, + ]); + + const dispatch = dispatchGatewayMethodInProcess( + "sessions.create", + { agentId: "main" }, + { + forceSyntheticClient: true, + resolveGatewayContext: () => current, + syntheticScopes: ["operator.write"], + }, + ); + await setupStarted.promise; + current = replacement; + releaseSetup.resolve(); + + await expect(dispatch).rejects.toThrow("current gateway instance binding"); + expect(committed).toBe(false); + }); + + it("composes caller authority into the session mutation commit guard", async () => { + const admitted = createContext(); + const assertCallerCurrent = vi.fn(); + admitted.getGatewayMethodRegistry = () => + createGatewayMethodRegistry([ + { + name: "sessions.create", + scope: "operator.write", + owner: { kind: "core", area: "sessions" }, + handler: ({ respond, sessionMutationCommitGuard }: GatewayRequestHandlerOptions) => { + sessionMutationCommitGuard?.(); + respond(true, { key: "agent:main:dashboard:child" }); + }, + }, + ]); + + await dispatchGatewayMethodInProcess( + "sessions.create", + { agentId: "main" }, + { + forceSyntheticClient: true, + resolveGatewayContext: () => admitted, + sessionMutationCommitGuard: assertCallerCurrent, + syntheticScopes: ["operator.write"], + }, + ); + + expect(assertCallerCurrent).toHaveBeenCalledOnce(); + }); + it("preserves the scoped operator identity across synthetic model-initiated session creation", async () => { const owner = createOperatorClient({ profileId: "model-spawn-owner", diff --git a/src/gateway/server-plugin-in-process-dispatch.ts b/src/gateway/server-plugin-in-process-dispatch.ts index 826eff7f2ad2..034a9d4a2c98 100644 --- a/src/gateway/server-plugin-in-process-dispatch.ts +++ b/src/gateway/server-plugin-in-process-dispatch.ts @@ -84,6 +84,7 @@ type DispatchGatewayMethodInProcessOptions = { timeoutMs?: number; signal?: AbortSignal; resolveGatewayContext?: GatewayContextResolver; + sessionMutationCommitGuard?: () => void; }; type ResolvedInProcessGatewayDispatch = { @@ -275,23 +276,37 @@ export async function dispatchGatewayMethodInProcessRaw( params: unknown, options?: DispatchGatewayMethodInProcessOptions, ): Promise { - return await withInProcessGatewayDispatch( - method, - options, - async (resolved) => - await dispatchGatewayRequestInProcessRaw(method, params, { - client: resolved.client, - context: resolved.context, - expectFinal: options?.expectFinal, - isWebchatConnect: resolved.isWebchatConnect, - methodRegistry: resolved.context.getGatewayMethodRegistry?.(), - onAccepted: options?.onAccepted, - onSignalAbort: options?.onSignalAbort, - requestIdPrefix: "plugin-subagent", - timeoutMs: options?.timeoutMs, - ...(options?.signal ? { signal: options.signal } : {}), - }), - ); + return await withInProcessGatewayDispatch(method, options, async (resolved) => { + const assertGatewayContextCurrent = options?.resolveGatewayContext + ? () => { + if (options.resolveGatewayContext?.() !== resolved.context) { + throw new Error( + `In-process gateway dispatch requires a current gateway instance binding (method: ${method}).`, + ); + } + } + : undefined; + const sessionMutationCommitGuard = + assertGatewayContextCurrent || options?.sessionMutationCommitGuard + ? () => { + assertGatewayContextCurrent?.(); + options?.sessionMutationCommitGuard?.(); + } + : undefined; + return await dispatchGatewayRequestInProcessRaw(method, params, { + client: resolved.client, + context: resolved.context, + expectFinal: options?.expectFinal, + isWebchatConnect: resolved.isWebchatConnect, + methodRegistry: resolved.context.getGatewayMethodRegistry?.(), + onAccepted: options?.onAccepted, + onSignalAbort: options?.onSignalAbort, + requestIdPrefix: "plugin-subagent", + ...(sessionMutationCommitGuard ? { sessionMutationCommitGuard } : {}), + timeoutMs: options?.timeoutMs, + ...(options?.signal ? { signal: options.signal } : {}), + }); + }); } /** Live request context for trusted built-in tools that need direct runtime state. */ diff --git a/src/gateway/server-runtime-state-prepare.ts b/src/gateway/server-runtime-state-prepare.ts index 37ef8eba1820..6a6e9361204e 100644 --- a/src/gateway/server-runtime-state-prepare.ts +++ b/src/gateway/server-runtime-state-prepare.ts @@ -157,6 +157,7 @@ export async function prepareGatewayKernelState(params: { return await workerModule.createGatewayWorkerEnvironmentRuntime({ getPluginRegistry: () => pluginRuntime.registry, getPortalRuntime: () => pluginGatewayContext.current, + resolveGatewayContext: () => pluginGatewayContext.current, desktopSessionRegistry, ...(nodeDesktopStreamBroker ? { nodeDesktopStreamBroker } : {}), startup: workerEnvironmentStartup, diff --git a/src/gateway/server-worker-environment-startup.test.ts b/src/gateway/server-worker-environment-startup.test.ts index 5a5d984c95dc..97101bb9d6ba 100644 --- a/src/gateway/server-worker-environment-startup.test.ts +++ b/src/gateway/server-worker-environment-startup.test.ts @@ -47,6 +47,7 @@ describe("gateway worker environment startup", () => { const runtime = await createGatewayWorkerEnvironmentRuntime({ getPluginRegistry: () => ({ workerProviders: new Map() }), getPortalRuntime: () => undefined, + resolveGatewayContext: () => undefined, desktopSessionRegistry: createDesktopSessionRegistry({ lingerMs: 1 }), startup, log: { child: () => ({ warn: () => {} }) }, @@ -108,6 +109,7 @@ describe("gateway worker environment startup", () => { const runtime = await createGatewayWorkerEnvironmentRuntime({ getPluginRegistry: () => ({ workerProviders: new Map() }), getPortalRuntime: () => undefined, + resolveGatewayContext: () => undefined, desktopSessionRegistry: createDesktopSessionRegistry({ lingerMs: 1 }), startup, log: { child: () => ({ warn: () => {} }) }, @@ -205,6 +207,7 @@ describe("gateway worker environment startup", () => { const runtime = await createGatewayWorkerEnvironmentRuntime({ getPluginRegistry: () => ({ workerProviders: new Map() }), getPortalRuntime: () => undefined, + resolveGatewayContext: () => undefined, desktopSessionRegistry: createDesktopSessionRegistry({ lingerMs: 1 }), nodeDesktopStreamBroker: createNodeDesktopStreamBroker(), startup, diff --git a/src/gateway/server-worker-environment-startup.ts b/src/gateway/server-worker-environment-startup.ts index fc5fda848596..46a10237c740 100644 --- a/src/gateway/server-worker-environment-startup.ts +++ b/src/gateway/server-worker-environment-startup.ts @@ -13,7 +13,7 @@ import type { NodeDesktopStreamBroker } from "./desktop/node-stream-broker.js"; import type { DesktopSessionRegistry } from "./desktop/session-registry.js"; import type { GitHubPublicationCoordinator } from "./github-publication.js"; import type { NodeWorkerSupervisorTransport } from "./node-registry-private.js"; -import type { GatewayRequestContext } from "./server-methods/types.js"; +import type { GatewayContextResolver, GatewayRequestContext } from "./server-methods/types.js"; import type { WorkerBundleProducer, WorkerNpmArtifact } from "./worker-environments/bundle.js"; import { bindDeviceWorkerAvailability, @@ -110,6 +110,7 @@ export async function loadGatewayWorkerEnvironmentStartupState(): Promise Pick; getPortalRuntime: () => Pick | undefined; + resolveGatewayContext: GatewayContextResolver; desktopSessionRegistry: DesktopSessionRegistry; nodeDesktopStreamBroker?: NodeDesktopStreamBroker; startup: GatewayWorkerEnvironmentStartupState; @@ -382,6 +383,7 @@ export async function createGatewayWorkerEnvironmentRuntime(params: { return environmentIds; }); executeSessionTool = createWorkerSessionToolExecutor({ + resolveGatewayContext: params.resolveGatewayContext, placements: params.startup.placementStore, environments: workerEnvironmentService, dispatchChild: (request) => dispatchChild(request), diff --git a/src/gateway/server.sessions.create.test.ts b/src/gateway/server.sessions.create.test.ts index fefef568734a..20b0c55c1753 100644 --- a/src/gateway/server.sessions.create.test.ts +++ b/src/gateway/server.sessions.create.test.ts @@ -39,7 +39,10 @@ import { openOpenClawAgentDatabase, resolveIncognitoOpenClawAgentSqlitePath, } from "../state/openclaw-agent-db.js"; -import { closeOpenClawStateDatabaseForTest } from "../state/openclaw-state-db.js"; +import { + closeOpenClawStateDatabaseForTest, + openOpenClawStateDatabase, +} from "../state/openclaw-state-db.js"; import { ensureProfileForEmail, setUserProfileRole } from "../state/user-profiles.js"; import { createOpenClawTestState } from "../test-utils/openclaw-test-state.js"; import { GATEWAY_CLIENT_MODES, GATEWAY_CLIENT_NAMES } from "../utils/message-channel.js"; @@ -76,6 +79,7 @@ import { seedSessionTranscript, threadBindingMocks, } from "./test/server-sessions.test-helpers.js"; +import { createWorkerSessionPlacementStore } from "./worker-environments/placement-store.js"; type EnsureSessionDiffBaseline = (typeof import("../sessions/session-diff-baseline.js"))["ensureSessionDiffBaseline"]; @@ -3457,6 +3461,126 @@ test("sessions.create commits no session after delegated authority closes", asyn ).toBeUndefined(); }); +test("sessions.create commits no child after its bound Gateway is replaced", async () => { + const { storePath } = await createSessionStoreDir(); + const sessionKey = "agent:main:dashboard:gateway-replacement-race"; + const admitted = {}; + const replacement = {}; + let current = admitted; + let guardCalls = 0; + const firstGuard = createDeferredCore(); + const writerEntered = createDeferredCore(); + const releaseWriter = createDeferredCore(); + const resolvedStore = resolveSqliteStoreScope(storePath, { agentId: "main" }); + const heldWriter = runExclusiveSqliteSessionWrite(resolvedStore, async () => { + writerEntered.resolve(); + await releaseWriter.promise; + }); + await writerEntered.promise; + const creating = directSessionReq( + "sessions.create", + { agentId: "main", key: sessionKey }, + { + sessionMutationAuthorization: { + assertCurrent: () => { + if (current !== admitted) { + throw new Error("current gateway instance binding was replaced"); + } + guardCalls += 1; + if (guardCalls === 1) { + firstGuard.resolve(); + } + }, + assertTargetCurrent: vi.fn(), + }, + }, + ); + + await firstGuard.promise; + current = replacement; + releaseWriter.resolve(); + await heldWriter; + + await expect(creating).rejects.toThrow("current gateway instance binding was replaced"); + expect(loadSessionEntry({ agentId: "main", sessionKey, storePath })).toBeUndefined(); +}); + +test("sessions.create commits no child after its worker turn closes", async () => { + const { storePath } = await createSessionStoreDir(); + const sessionKey = "agent:main:dashboard:worker-turn-race"; + const placements = createWorkerSessionPlacementStore({ + database: openOpenClawStateDatabase(), + }); + let placement = placements.startDispatch({ + agentId: "main", + sessionId: "worker-source-session", + sessionKey: "agent:main:dashboard:worker-source", + }); + for (const [from, to, patch] of [ + ["requested", "provisioning", { environmentId: "worker-environment" }], + ["provisioning", "syncing", { workerBundleHash: "a".repeat(64) }], + [ + "syncing", + "starting", + { remoteWorkspaceDir: "/workspace/source", workspaceBaseManifestRef: "manifest-source" }, + ], + ["starting", "active", { activeOwnerEpoch: 7 }], + ] as const) { + placement = placements.transition({ + sessionId: placement.sessionId, + from, + to, + expectedGeneration: placement.generation, + patch, + }); + } + const turnClaim = placements.claimTurn({ + agentId: placement.agentId, + sessionId: placement.sessionId, + sessionKey: placement.sessionKey, + claimId: "worker-claim", + runId: "worker-run", + owner: { kind: "worker", environmentId: "worker-environment", ownerEpoch: 7 }, + }); + let guardCalls = 0; + const firstGuard = createDeferredCore(); + const writerEntered = createDeferredCore(); + const releaseWriter = createDeferredCore(); + const heldWriter = runExclusiveSqliteSessionWrite( + resolveSqliteStoreScope(storePath, { agentId: "main" }), + async () => { + writerEntered.resolve(); + await releaseWriter.promise; + }, + ); + await writerEntered.promise; + const creating = directSessionReq( + "sessions.create", + { agentId: "main", key: sessionKey }, + { + sessionMutationAuthorization: { + assertCurrent: () => { + if (!placements.validateTurnClaim(turnClaim)) { + throw new Error("worker turn authority changed"); + } + if (++guardCalls === 1) { + firstGuard.resolve(); + } + }, + assertTargetCurrent: vi.fn(), + }, + }, + ); + + await firstGuard.promise; + placements.releaseTurn(turnClaim); + releaseWriter.resolve(); + await heldWriter; + + await expect(creating).rejects.toThrow("worker turn authority changed"); + expect(loadSessionEntry({ agentId: "main", sessionKey, storePath })).toBeUndefined(); +}); + test("sessions.create starts no initial turn when authority closes after session commit", async () => { await createSessionStoreDir(); const sessionKey = "agent:main:dashboard:authority-post-commit"; diff --git a/src/gateway/worker-environments/worker-session-tool-executor.test.ts b/src/gateway/worker-environments/worker-session-tool-executor.test.ts index 89738583a053..9f8defeb7a85 100644 --- a/src/gateway/worker-environments/worker-session-tool-executor.test.ts +++ b/src/gateway/worker-environments/worker-session-tool-executor.test.ts @@ -33,6 +33,7 @@ const dispatchChild = vi.hoisted(() => vi.fn()); const spawnCallerIdentity = vi.hoisted(() => vi.fn()); const spawnArgs = vi.hoisted(() => vi.fn()); const githubPublicationRequest = vi.hoisted(() => vi.fn()); +const resolveGatewayContext = () => undefined; const scopedSessionAccess = vi.hoisted(() => vi.fn(async (params: { run: () => Promise }) => await params.run()), ); @@ -97,7 +98,8 @@ vi.mock("../../agents/tools/in-process-gateway.js", () => ({ method: string, params: Record, creation: unknown, - ) => gatewayCreate({ creation, method, params }), + options: unknown, + ) => gatewayCreate({ creation, method, options, params }), withAgentToolGatewayRuntimeIdentity: (request: unknown, identity: unknown) => { gatewayRuntimeIdentity(request, identity); return request; @@ -246,6 +248,7 @@ describe("worker session tool topology", () => { }, ); execute = createWorkerSessionToolExecutor({ + resolveGatewayContext, placements, dispatchChild, githubPublication: { requestForClaim: githubPublicationRequest }, @@ -452,6 +455,11 @@ describe("worker session tool topology", () => { via: "spawn", }), method: "sessions.create", + options: { + resolveGatewayContext, + sessionMutationCommitGuard: expect.any(Function), + timeoutMs: null, + }, params: expect.not.objectContaining({ task: expect.anything() }), }), ); diff --git a/src/gateway/worker-environments/worker-session-tool-executor.ts b/src/gateway/worker-environments/worker-session-tool-executor.ts index 2d8c6fcca937..236a3cb9982a 100644 --- a/src/gateway/worker-environments/worker-session-tool-executor.ts +++ b/src/gateway/worker-environments/worker-session-tool-executor.ts @@ -26,6 +26,7 @@ import { sha256Base64Url, sha256HexPrefixCore } from "../../infra/crypto-digest. import { normalizeAgentId } from "../../routing/session-key.js"; import { WORKER_TOOL_NAMES } from "../../worker/tool-authority.js"; import type { GitHubPublicationCoordinator } from "../github-publication.js"; +import type { GatewayContextResolver } from "../server-methods/types.js"; import { loadGatewaySessionEntryReadOnly } from "../session-utils.js"; import type { WorkerConnectionIdentity } from "./connection-identity.js"; import type { WorkerSessionPlacementStore } from "./placement-store.js"; @@ -76,19 +77,15 @@ function operationKey(operationSeed: string, purpose: string): string { return sha256Base64Url(`openclaw.worker-session-tool-operation.v1\0${operationSeed}\0${purpose}`); } -function throwIfAborted(signal: AbortSignal | undefined): void { - signal?.throwIfAborted(); -} - -function childSessionKey(params: { operationSeed: string; targetAgentId: string }): string { - const suffix = sha256HexPrefixCore( - `openclaw.worker-session-tool-operation.v1\0${params.operationSeed}\0child-session`, +function childSessionKey(operationSeed: string, targetAgentId: string): string { + return `agent:${targetAgentId}:dashboard:cloud-${sha256HexPrefixCore( + `openclaw.worker-session-tool-operation.v1\0${operationSeed}\0child-session`, 32, - ); - return `agent:${params.targetAgentId}:dashboard:cloud-${suffix}`; + )}`; } export function createWorkerSessionToolExecutor(params: { + resolveGatewayContext: GatewayContextResolver; placements: WorkerSessionPlacementStore; environments: Pick; dispatchChild: WorkerPlacementDispatchContract["dispatch"]; @@ -106,14 +103,13 @@ export function createWorkerSessionToolExecutor(params: { childSessionKey: string; signal?: AbortSignal; }) => { - throwIfAborted(operation.signal); + operation.signal?.throwIfAborted(); const sourceEnvironment = params.environments.get(operation.identity.environmentId); if ( !sourceEnvironment || sourceEnvironment.state !== "attached" || sourceEnvironment.ownerEpoch !== operation.identity.ownerEpoch || - sourceEnvironment.attachedSessionIds.length !== 1 || - sourceEnvironment.attachedSessionIds[0] !== operation.source.sessionId + !isDeepStrictEqual(sourceEnvironment.attachedSessionIds, [operation.source.sessionId]) ) { throw new Error("Worker source environment changed before child spawn"); } @@ -138,7 +134,7 @@ export function createWorkerSessionToolExecutor(params: { timeoutMs: null, }); } - throwIfAborted(operation.signal); + operation.signal?.throwIfAborted(); exactSource({ identity: operation.identity, placements: params.placements }); let loaded = loadGatewaySessionEntryReadOnly(operation.childSessionKey, { agentId: targetAgentId, @@ -183,6 +179,9 @@ export function createWorkerSessionToolExecutor(params: { inheritedToolPolicy: { version: 1, allow: authorizedTools, deny: [] }, }, { + resolveGatewayContext: params.resolveGatewayContext, + sessionMutationCommitGuard: () => + exactSource({ identity: operation.identity, placements: params.placements }), ...(operation.signal ? { signal: operation.signal } : {}), timeoutMs: null, }, @@ -244,7 +243,7 @@ export function createWorkerSessionToolExecutor(params: { } }; const childPlacement = params.placements.get(childSessionId); - throwIfAborted(operation.signal); + operation.signal?.throwIfAborted(); exactSource({ identity: operation.identity, placements: params.placements }); if (childPlacement?.state !== "active") { try { @@ -269,7 +268,7 @@ export function createWorkerSessionToolExecutor(params: { } assertActiveChildPlacement(); exactSource({ identity: operation.identity, placements: params.placements }); - throwIfAborted(operation.signal); + operation.signal?.throwIfAborted(); assertExactChild({ childSessionKey: operation.childSessionKey, childSessionId, @@ -306,7 +305,7 @@ export function createWorkerSessionToolExecutor(params: { let sendResult: Record | undefined; for (let attempt = 0; attempt < 2; attempt += 1) { try { - throwIfAborted(operation.signal); + operation.signal?.throwIfAborted(); exactSource({ identity: operation.identity, placements: params.placements }); assertExactChild({ childSessionKey: operation.childSessionKey, @@ -424,6 +423,7 @@ export function createWorkerSessionToolExecutor(params: { { agentId: identity.agentId, sessionKey: identity.sessionKey, + gatewayContextResolver: params.resolveGatewayContext, operationalRunInstance: identity.operationalRunInstance, executionIdentityToken: identity.executionIdentityToken, receiptAuthority: identity.receiptAuthority, @@ -436,7 +436,14 @@ export function createWorkerSessionToolExecutor(params: { workerIdentity = undefined; } }) - : await executeSpawn(); + : await withGatewayToolCallerIdentity( + { + agentId: operation.source.agentId, + sessionKey: operation.source.sessionKey, + gatewayContextResolver: params.resolveGatewayContext, + }, + executeSpawn, + ); }; const send = async (operation: { @@ -447,7 +454,7 @@ export function createWorkerSessionToolExecutor(params: { idempotencyKey: string; signal?: AbortSignal; }) => { - throwIfAborted(operation.signal); + operation.signal?.throwIfAborted(); exactSource({ identity: operation.identity, placements: params.placements }); const config = getRuntimeConfig(); const executeFencedSend = async () => { @@ -480,7 +487,7 @@ export function createWorkerSessionToolExecutor(params: { }); for (let attempt = 0; attempt < 2; attempt += 1) { try { - throwIfAborted(operation.signal); + operation.signal?.throwIfAborted(); exactSource({ identity: operation.identity, placements: params.placements }); assertCurrentTarget(); return await tool.execute(operation.request.toolCallId, { @@ -528,7 +535,7 @@ export function createWorkerSessionToolExecutor(params: { } }; assertPublicationAuthority(); - throwIfAborted(request.signal); + request.signal?.throwIfAborted(); const publication = await params.githubPublication.requestForClaim({ claim: source.turnClaim, sessionKey: source.sessionKey, @@ -618,10 +625,7 @@ export function createWorkerSessionToolExecutor(params: { let childKey = started.childSessionKey; if (request.toolName === "sessions_spawn" && !childKey) { const targetAgentId = normalizeAgentId(request.request.agentId ?? source.agentId); - childKey = childSessionKey({ - operationSeed: started.operationSeed, - targetAgentId, - }); + childKey = childSessionKey(started.operationSeed, targetAgentId); if ( !params.placements.bindWorkerSessionToolOperationChild({ sourceSessionId: source.sessionId,