diff --git a/docs/cli/worker.md b/docs/cli/worker.md index cd33ffe2f4ad..e065d668a0d7 100644 --- a/docs/cli/worker.md +++ b/docs/cli/worker.md @@ -21,13 +21,17 @@ admits as the dedicated `worker` role. The command reads exactly one bounded JSON launch envelope from standard input. The envelope carries the local socket location, minted worker credential, bundle -and protocol identity, owner epoch, and the single assigned session and turn. +and protocol identity, owner epoch, the single assigned session and turn, and the +exact worker-local tool names authorized for that turn. The Gateway resolves this +final tool set from current policy before handoff; raw config and scheduled-owner +identity never enter the worker envelope. The credential is never accepted through command-line arguments, and this page intentionally provides no credential or hand-authored envelope example. Admission fails closed if the envelope is invalid, the credential is rejected, the bundle or protocol features do not match, or the session and owner epoch are -no longer current. Operators should start workers through the cloud worker +no longer current. Missing, duplicate, or unknown tool names also invalidate the +envelope. Operators should start workers through the cloud worker orchestrator rather than invoke this entry point directly. ## Runtime boundary @@ -35,7 +39,8 @@ orchestrator rather than invoke this entry point directly. The process runs the normal embedded agent loop with a restricted backend: - The `read`, `write`, `edit`, `apply_patch`, `exec`, and `process` coding tools - run locally in the worker workspace. + run locally in the worker workspace when present in the Gateway-issued turn + authority. An empty authority runs the model with no tools. - Model calls use the gateway inference proxy. No local model auth profile is loaded. - Transcript writes use the gateway transcript-commit RPC. diff --git a/docs/gateway/cloud-workers.md b/docs/gateway/cloud-workers.md index 1cf3e717b1c7..9607ed764469 100644 --- a/docs/gateway/cloud-workers.md +++ b/docs/gateway/cloud-workers.md @@ -119,6 +119,7 @@ Placement moves through a durable state machine (`local → requested → provis ## Security model - **Closed worker ingress.** Workers speak a dedicated protocol on the tunneled socket with a closed method allowlist — a worker cannot call operator RPCs. +- **Gateway-owned tool authority.** Before every turn, the Gateway projects current profile, provider, agent, group, sender, sandbox, delegation, inherited, and runtime-cap policy over the worker's fixed coding-tool catalog. The launch envelope carries only that final closed-vocabulary subset. Explicitly capped scheduled turns reuse their trusted owner-group context without sending that identity to the box or reapplying a fresh sender overlay. Tools outside the worker catalog remain unavailable; an empty result runs with no tools. - **Minted credentials, hashed at rest.** Each dispatch mints a worker credential; the Gateway stores only its hash. Credential rotation and owner-epoch fencing guarantee at most one live owner per session — a stale worker that reconnects is fenced, never merged. - **Host-key pinning.** The provider must surface the box's SSH host key at provision time; bootstrap connects with strict pinning and fails closed without it. - **No standing model, forge, or cloud credentials on the box.** Model auth stays on the Gateway (inference travels by `{provider, model}` reference), workspace git commits are authored without forge credentials, and Crabbox AWS lease metadata is checked authoritatively for an instance role before setup. Keep setup commands credential-free too. diff --git a/src/agents/agent-tools.policy.ts b/src/agents/agent-tools.policy.ts index 6047a2f8622b..04c4d0a529da 100644 --- a/src/agents/agent-tools.policy.ts +++ b/src/agents/agent-tools.policy.ts @@ -21,7 +21,6 @@ import { } from "../sessions/session-key-utils.js"; import { normalizeMessageChannel } from "../utils/message-channel.js"; import { resolveAgentConfig, resolveAgentIdFromSessionKey } from "./agent-scope.js"; -import type { AnyAgentTool } from "./agent-tools.types.js"; import { resolveProviderToolPolicy } from "./provider-tool-policy.js"; import { pickSandboxToolPolicy } from "./sandbox-tool-policy.js"; import type { SandboxToolPolicy } from "./sandbox.js"; @@ -142,7 +141,10 @@ export function resolveInheritedToolPolicyForSession( } /** Filter runtime tools by sandbox allow/deny policy. */ -export function filterToolsByPolicy(tools: AnyAgentTool[], policy?: SandboxToolPolicy) { +export function filterToolsByPolicy( + tools: TTool[], + policy?: SandboxToolPolicy, +): TTool[] { if (!policy) { return tools; } diff --git a/src/agents/agent-tools.ts b/src/agents/agent-tools.ts index f688f19fc947..435fdebc7815 100644 --- a/src/agents/agent-tools.ts +++ b/src/agents/agent-tools.ts @@ -70,6 +70,10 @@ import { type ResolvedConversationCapabilityProfile, } from "./conversation-capability-profile.js"; import type { ConversationRecallContext } from "./conversation-recall.types.js"; +import { + buildConversationToolPolicyPipelineSteps, + resolveConversationToolPolicies, +} from "./conversation-tool-policy-pipeline.js"; import type { OpenClawCodingToolConstructionPlan } from "./core-tool-factory-descriptors.js"; import { applyDelegationCapability, type DelegationCapability } from "./delegation-capability.js"; import { resolveImageSanitizationLimits } from "./image-sanitization.js"; @@ -92,14 +96,10 @@ import { createToolFsPolicy, resolveToolFsConfig } from "./tool-fs-policy.js"; import { resolveToolLoopDetectionConfig } from "./tool-loop-detection-config.js"; import { buildDeclaredToolAllowlistContext } from "./tool-policy-declared-context.js"; import { isToolAllowedByPolicies } from "./tool-policy-match.js"; -import { - applyToolPolicyPipeline, - buildDefaultToolPolicyPipelineSteps, -} from "./tool-policy-pipeline.js"; +import { applyToolPolicyPipeline } from "./tool-policy-pipeline.js"; import { expandToolGroups, hasRestrictiveAllowPolicy, - mergeAlsoAllowPolicy, normalizeToolName, replaceWithEffectiveToolAllowlist, } from "./tool-policy.js"; @@ -529,25 +529,7 @@ function createOpenClawCodingToolsInternal(options?: OpenClawCodingToolsOptions) trustedInternalHandoff: options?.trustedInternalHandoff, scheduledToolPolicy: options?.scheduledToolPolicy, }); - const { - agentId, - globalPolicy, - globalProviderPolicy, - agentPolicy, - agentProviderPolicy, - profile, - providerProfile, - profilePolicy, - providerProfilePolicy, - profileAlsoAllow, - providerProfileAlsoAllow, - groupPolicy, - senderPolicy, - subagentPolicy, - inheritedToolPolicy, - runtimePluginToolGrant, - runtimeToolPolicyForInheritance, - } = capabilityProfile.policy; + const { agentId, runtimePluginToolGrant } = capabilityProfile.policy; const enableHeartbeatTool = options?.enableHeartbeatTool === true || @@ -565,9 +547,6 @@ function createOpenClawCodingToolsInternal(options?: OpenClawCodingToolsOptions) TOOL_CALL_RAW_TOOL_NAME, ] : []; - const mergeToolSearchControlAllowlist = ( - policy: TPolicy | undefined, - ) => mergeAlsoAllowPolicy(policy, toolSearchControlAllowlist); const runtimeToolAllowlistIncludesMessage = expandToolGroups( options?.runtimeToolAllowlist ?? [], ).some((toolName) => { @@ -587,14 +566,11 @@ function createOpenClawCodingToolsInternal(options?: OpenClawCodingToolsOptions) ...(forceHeartbeatTool ? [HEARTBEAT_RESPONSE_TOOL_NAME] : []), ...toolSearchControlAllowlist, ]; - const profilePolicyWithAlsoAllow = mergeAlsoAllowPolicy(profilePolicy, [ - ...(profileAlsoAllow ?? []), - ...runtimeProfileAlsoAllow, - ]); - const providerProfilePolicyWithAlsoAllow = mergeAlsoAllowPolicy(providerProfilePolicy, [ - ...(providerProfileAlsoAllow ?? []), - ...runtimeProfileAlsoAllow, - ]); + const conversationToolPolicies = resolveConversationToolPolicies({ + capabilityProfile, + additionalProfileAllow: runtimeProfileAlsoAllow, + additionalPolicyAllow: toolSearchControlAllowlist, + }); // Prefer sessionKey for process isolation scope to prevent cross-session process visibility/killing. // Fallback to agentId if no sessionKey is available (e.g. legacy or global contexts). const scopeKey = resolveProcessToolScopeKey({ @@ -603,29 +579,18 @@ function createOpenClawCodingToolsInternal(options?: OpenClawCodingToolsOptions) sessionId: options?.sessionId, agentId, }); - const globalPolicyWithToolSearchControls = mergeToolSearchControlAllowlist(globalPolicy); - const globalProviderPolicyWithToolSearchControls = - mergeToolSearchControlAllowlist(globalProviderPolicy); - const agentPolicyWithToolSearchControls = mergeToolSearchControlAllowlist(agentPolicy); - const agentProviderPolicyWithToolSearchControls = - mergeToolSearchControlAllowlist(agentProviderPolicy); - const groupPolicyWithToolSearchControls = mergeToolSearchControlAllowlist(groupPolicy); - const senderPolicyWithToolSearchControls = mergeToolSearchControlAllowlist(senderPolicy); - const sandboxToolPolicyWithToolSearchControls = - mergeToolSearchControlAllowlist(sandboxToolPolicy); - const subagentPolicyWithToolSearchControls = mergeToolSearchControlAllowlist(subagentPolicy); const allowBackground = isToolAllowedByPolicies("process", [ - profilePolicyWithAlsoAllow, - providerProfilePolicyWithAlsoAllow, - globalPolicyWithToolSearchControls, - globalProviderPolicyWithToolSearchControls, - agentPolicyWithToolSearchControls, - agentProviderPolicyWithToolSearchControls, - groupPolicyWithToolSearchControls, - senderPolicyWithToolSearchControls, - sandboxToolPolicyWithToolSearchControls, - subagentPolicyWithToolSearchControls, - inheritedToolPolicy, + conversationToolPolicies.profilePolicy, + conversationToolPolicies.providerProfilePolicy, + conversationToolPolicies.globalPolicy, + conversationToolPolicies.globalProviderPolicy, + conversationToolPolicies.agentPolicy, + conversationToolPolicies.agentProviderPolicy, + conversationToolPolicies.groupPolicy, + conversationToolPolicies.senderPolicy, + conversationToolPolicies.sandboxPolicy, + conversationToolPolicies.subagentPolicy, + conversationToolPolicies.inheritedToolPolicy, ]); options?.recordToolPrepStage?.("tool-policy"); const execConfig = resolveExecToolConfig({ cfg: options?.config, agentId }); @@ -1095,45 +1060,19 @@ function createOpenClawCodingToolsInternal(options?: OpenClawCodingToolsOptions) tools: toolsForModelProvider, toolMeta: (tool) => getPluginToolMeta(tool), warn: logWarn, - steps: [ - ...buildDefaultToolPolicyPipelineSteps({ - profilePolicy: profilePolicyWithAlsoAllow, - profile, - profileUnavailableCoreWarningAllowlist: profilePolicy?.allow, - providerProfilePolicy: providerProfilePolicyWithAlsoAllow, - providerProfile, - providerProfileUnavailableCoreWarningAllowlist: providerProfilePolicy?.allow, - globalPolicy: globalPolicyWithToolSearchControls, - globalProviderPolicy: globalProviderPolicyWithToolSearchControls, - agentPolicy: agentPolicyWithToolSearchControls, - agentProviderPolicy: agentProviderPolicyWithToolSearchControls, - groupPolicy: groupPolicyWithToolSearchControls, - senderPolicy: senderPolicyWithToolSearchControls, - agentId, - unavailableCoreToolReason, - }), - { - policy: sandboxToolPolicyWithToolSearchControls, - label: "sandbox tools.allow", - unavailableCoreToolReason, - }, - { - policy: ownerOnlyCoreToolPolicy, - label: "gateway sender owner-only tools", - unavailableCoreToolReason, - }, - { - policy: subagentPolicyWithToolSearchControls, - label: "subagent tools.allow", - unavailableCoreToolReason, - }, - { - policy: runtimeToolPolicyForInheritance, - label: "runtime tools.allow", - unavailableCoreToolReason, - }, - { policy: inheritedToolPolicy, label: "inherited tools", unavailableCoreToolReason }, - ], + steps: buildConversationToolPolicyPipelineSteps({ + capabilityProfile, + policies: conversationToolPolicies, + additionalStepsAfterSandbox: [ + { + policy: ownerOnlyCoreToolPolicy, + label: "gateway sender owner-only tools", + unavailableCoreToolReason, + }, + ], + includeRuntimeToolPolicy: true, + unavailableCoreToolReason, + }), auditLogLevel: options?.toolPolicyAuditLogLevel, declaredToolAllowlist: buildDeclaredToolAllowlistContext({ config: options?.config, diff --git a/src/agents/conversation-tool-policy-pipeline.ts b/src/agents/conversation-tool-policy-pipeline.ts new file mode 100644 index 000000000000..f8a08e233b18 --- /dev/null +++ b/src/agents/conversation-tool-policy-pipeline.ts @@ -0,0 +1,145 @@ +import type { ResolvedConversationCapabilityProfile } from "./conversation-capability-profile.js"; +import { + applyToolPolicyPipeline, + buildDefaultToolPolicyPipelineSteps, + type ToolPolicyPipelineStep, +} from "./tool-policy-pipeline.js"; +import { mergeAlsoAllowPolicy, type ToolPolicyLike } from "./tool-policy.js"; + +export type ResolvedConversationToolPolicies = { + profilePolicy?: ToolPolicyLike; + providerProfilePolicy?: ToolPolicyLike; + globalPolicy?: ToolPolicyLike; + globalProviderPolicy?: ToolPolicyLike; + agentPolicy?: ToolPolicyLike; + agentProviderPolicy?: ToolPolicyLike; + groupPolicy?: ToolPolicyLike; + senderPolicy?: ToolPolicyLike; + sandboxPolicy?: ToolPolicyLike; + subagentPolicy?: ToolPolicyLike; + runtimeToolPolicy?: ToolPolicyLike; + inheritedToolPolicy?: ToolPolicyLike; +}; + +function mergePolicyAllowlist( + policy: TPolicy | undefined, + alsoAllow: readonly string[] | undefined, +): TPolicy | undefined { + return mergeAlsoAllowPolicy(policy, alsoAllow ? [...alsoAllow] : undefined); +} + +/** + * Resolves the shared policy layers once so local and remote fixed tool surfaces cannot + * diverge on profile `alsoAllow`, sender, sandbox, delegation, or runtime-cap semantics. + */ +export function resolveConversationToolPolicies(params: { + capabilityProfile: ResolvedConversationCapabilityProfile; + additionalProfileAllow?: readonly string[]; + additionalPolicyAllow?: readonly string[]; +}): ResolvedConversationToolPolicies { + const policy = params.capabilityProfile.policy; + const profileAllow = [ + ...(policy.profileAlsoAllow ?? []), + ...(params.additionalProfileAllow ?? []), + ]; + const providerProfileAllow = [ + ...(policy.providerProfileAlsoAllow ?? []), + ...(params.additionalProfileAllow ?? []), + ]; + return { + profilePolicy: mergePolicyAllowlist(policy.profilePolicy, profileAllow), + providerProfilePolicy: mergePolicyAllowlist(policy.providerProfilePolicy, providerProfileAllow), + globalPolicy: mergePolicyAllowlist(policy.globalPolicy, params.additionalPolicyAllow), + globalProviderPolicy: mergePolicyAllowlist( + policy.globalProviderPolicy, + params.additionalPolicyAllow, + ), + agentPolicy: mergePolicyAllowlist(policy.agentPolicy, params.additionalPolicyAllow), + agentProviderPolicy: mergePolicyAllowlist( + policy.agentProviderPolicy, + params.additionalPolicyAllow, + ), + groupPolicy: mergePolicyAllowlist(policy.groupPolicy, params.additionalPolicyAllow), + senderPolicy: mergePolicyAllowlist(policy.senderPolicy, params.additionalPolicyAllow), + sandboxPolicy: mergePolicyAllowlist(policy.sandboxPolicy, params.additionalPolicyAllow), + subagentPolicy: mergePolicyAllowlist(policy.subagentPolicy, params.additionalPolicyAllow), + runtimeToolPolicy: policy.runtimeToolPolicyForInheritance, + inheritedToolPolicy: policy.inheritedToolPolicy, + }; +} + +/** Builds the canonical ordered policy pipeline for a resolved conversation. */ +export function buildConversationToolPolicyPipelineSteps(params: { + capabilityProfile: ResolvedConversationCapabilityProfile; + policies: ResolvedConversationToolPolicies; + additionalStepsAfterSandbox?: ToolPolicyPipelineStep[]; + includeRuntimeToolPolicy: boolean; + unavailableCoreToolReason?: string; +}): ToolPolicyPipelineStep[] { + const profile = params.capabilityProfile.policy; + return [ + ...buildDefaultToolPolicyPipelineSteps({ + profilePolicy: params.policies.profilePolicy, + profile: profile.profile, + profileUnavailableCoreWarningAllowlist: profile.profilePolicy?.allow, + providerProfilePolicy: params.policies.providerProfilePolicy, + providerProfile: profile.providerProfile, + providerProfileUnavailableCoreWarningAllowlist: profile.providerProfilePolicy?.allow, + globalPolicy: params.policies.globalPolicy, + globalProviderPolicy: params.policies.globalProviderPolicy, + agentPolicy: params.policies.agentPolicy, + agentProviderPolicy: params.policies.agentProviderPolicy, + groupPolicy: params.policies.groupPolicy, + senderPolicy: params.policies.senderPolicy, + agentId: profile.agentId, + unavailableCoreToolReason: params.unavailableCoreToolReason, + }), + { + policy: params.policies.sandboxPolicy, + label: "sandbox tools.allow", + unavailableCoreToolReason: params.unavailableCoreToolReason, + }, + ...(params.additionalStepsAfterSandbox ?? []), + { + policy: params.policies.subagentPolicy, + label: "subagent tools.allow", + unavailableCoreToolReason: params.unavailableCoreToolReason, + }, + ...(params.includeRuntimeToolPolicy + ? [ + { + policy: params.policies.runtimeToolPolicy, + label: "runtime tools.allow", + unavailableCoreToolReason: params.unavailableCoreToolReason, + }, + ] + : []), + { + policy: params.policies.inheritedToolPolicy, + label: "inherited tools", + unavailableCoreToolReason: params.unavailableCoreToolReason, + }, + ]; +} + +/** Projects a fixed runtime catalog through the exact conversation policy pipeline. */ +export function projectConversationToolNames(params: { + capabilityProfile: ResolvedConversationCapabilityProfile; + toolNames: readonly TName[]; + warn: (message: string) => void; +}): TName[] { + const policies = resolveConversationToolPolicies({ + capabilityProfile: params.capabilityProfile, + }); + const tools = params.toolNames.map((name) => ({ name })); + return applyToolPolicyPipeline({ + tools, + toolMeta: () => undefined, + warn: params.warn, + steps: buildConversationToolPolicyPipelineSteps({ + capabilityProfile: params.capabilityProfile, + policies, + includeRuntimeToolPolicy: true, + }), + }).map((tool) => tool.name); +} diff --git a/src/agents/embedded-agent-runner/effective-tool-policy.ts b/src/agents/embedded-agent-runner/effective-tool-policy.ts index 774ee1d3a983..e5f4f3925294 100644 --- a/src/agents/embedded-agent-runner/effective-tool-policy.ts +++ b/src/agents/embedded-agent-runner/effective-tool-policy.ts @@ -5,14 +5,17 @@ import type { OpenClawConfig } from "../../config/types.openclaw.js"; import type { PluginMetadataSnapshot } from "../../plugins/plugin-metadata-snapshot.types.js"; import { getPluginToolMeta } from "../../plugins/tools.js"; import type { ResolvedConversationCapabilityProfile } from "../conversation-capability-profile.js"; +import { + buildConversationToolPolicyPipelineSteps, + resolveConversationToolPolicies, +} from "../conversation-tool-policy-pipeline.js"; import { buildDeclaredToolAllowlistContext } from "../tool-policy-declared-context.js"; import { applyToolPolicyPipeline, - buildDefaultToolPolicyPipelineSteps, type ToolPolicyFilterEvent, type ToolPolicyPipelineStep, } from "../tool-policy-pipeline.js"; -import { collectExplicitDenylist, mergeAlsoAllowPolicy } from "../tool-policy.js"; +import { collectExplicitDenylist } from "../tool-policy.js"; import type { AnyAgentTool } from "../tools/common.js"; /** @@ -54,29 +57,7 @@ export function applyFinalEffectiveToolPolicy( "effective tool policy: dropping caller-provided groupId that does not match session-derived group context", ); } - const { - agentId, - globalPolicy, - globalProviderPolicy, - agentPolicy, - agentProviderPolicy, - profile, - providerProfile, - profilePolicy, - providerProfilePolicy, - profileAlsoAllow, - providerProfileAlsoAllow, - groupPolicy, - senderPolicy, - sandboxPolicy, - subagentPolicy, - inheritedToolPolicy, - } = capabilityProfile.policy; - const profilePolicyWithAlsoAllow = mergeAlsoAllowPolicy(profilePolicy, profileAlsoAllow); - const providerProfilePolicyWithAlsoAllow = mergeAlsoAllowPolicy( - providerProfilePolicy, - providerProfileAlsoAllow, - ); + const policies = resolveConversationToolPolicies({ capabilityProfile }); // Suppress unavailable-core-tool warnings on every step of this pass. // `applyToolPolicyPipeline` infers `coreToolNames` from the `tools` array // it's filtering, and this pass only sees the bundled MCP/LSP subset. @@ -87,26 +68,11 @@ export function applyFinalEffectiveToolPolicy( // real diagnostics from the shared warning cache. Genuinely unknown // entries (typos) still surface through the `otherEntries` path in // `applyToolPolicyPipeline`. - const pipelineSteps: ToolPolicyPipelineStep[] = [ - ...buildDefaultToolPolicyPipelineSteps({ - profilePolicy: profilePolicyWithAlsoAllow, - profile, - profileUnavailableCoreWarningAllowlist: profilePolicy?.allow, - providerProfilePolicy: providerProfilePolicyWithAlsoAllow, - providerProfile, - providerProfileUnavailableCoreWarningAllowlist: providerProfilePolicy?.allow, - globalPolicy, - globalProviderPolicy, - agentPolicy, - agentProviderPolicy, - groupPolicy, - senderPolicy, - agentId, - }), - { policy: sandboxPolicy, label: "sandbox tools.allow" }, - { policy: subagentPolicy, label: "subagent tools.allow" }, - { policy: inheritedToolPolicy, label: "inherited tools" }, - ].map((step) => Object.assign({}, step, { suppressUnavailableCoreToolWarning: true })); + const pipelineSteps: ToolPolicyPipelineStep[] = buildConversationToolPolicyPipelineSteps({ + capabilityProfile, + policies, + includeRuntimeToolPolicy: false, + }).map((step) => Object.assign({}, step, { suppressUnavailableCoreToolWarning: true })); return applyToolPolicyPipeline({ tools: params.bundledTools, toolMeta: (tool) => getPluginToolMeta(tool), diff --git a/src/agents/tool-policy-pipeline.ts b/src/agents/tool-policy-pipeline.ts index 0c38889e5ed5..dc3bf9d8f7ca 100644 --- a/src/agents/tool-policy-pipeline.ts +++ b/src/agents/tool-policy-pipeline.ts @@ -46,11 +46,11 @@ export type ToolPolicyPipelineStep = { }; /** One policy application, exposed for diagnostics that need exclusion provenance. */ -export type ToolPolicyFilterEvent = { +export type ToolPolicyFilterEvent = { step: ToolPolicyPipelineStep; policy: ToolPolicyLike; - before: readonly AnyAgentTool[]; - after: readonly AnyAgentTool[]; + before: readonly TTool[]; + after: readonly TTool[]; }; /** Builds the default profile, provider, agent, group, and sender policy layers. */ @@ -132,15 +132,15 @@ export function buildDefaultToolPolicyPipelineSteps(params: { } /** Applies configured policy layers to a tool list and emits deduped warnings/audit events. */ -export function applyToolPolicyPipeline(params: { - tools: AnyAgentTool[]; - toolMeta: (tool: AnyAgentTool) => { pluginId: string } | undefined; +export function applyToolPolicyPipeline(params: { + tools: TTool[]; + toolMeta: (tool: TTool) => { pluginId: string } | undefined; warn: (message: string) => void; steps: ToolPolicyPipelineStep[]; auditLogLevel?: ToolPolicyAuditLogLevel; declaredToolAllowlist?: DeclaredToolAllowlistContext; - onFilter?: (event: ToolPolicyFilterEvent) => void; -}): AnyAgentTool[] { + onFilter?: (event: ToolPolicyFilterEvent) => void; +}): TTool[] { const coreToolNames = new Set( params.tools .filter((tool) => !params.toolMeta(tool)) diff --git a/src/gateway/server-worker-placement-startup.ts b/src/gateway/server-worker-placement-startup.ts index 77c037d9c423..e5a517276d7a 100644 --- a/src/gateway/server-worker-placement-startup.ts +++ b/src/gateway/server-worker-placement-startup.ts @@ -468,6 +468,7 @@ export function createGatewayWorkerPlacementRuntime(params: GatewayWorkerPlaceme environments: params.environments, placements: params.placements, admitNewPlacements: params.admitNewPlacements, + resolveWorkspacePath, redispatchReclaimed: createReclaimedPlacementRedispatch({ environments: params.environments, dispatch: dispatchService.dispatch, diff --git a/src/gateway/worker-environments/live-events.test.ts b/src/gateway/worker-environments/live-events.test.ts index 2b7fc0419c42..534c13d8c189 100644 --- a/src/gateway/worker-environments/live-events.test.ts +++ b/src/gateway/worker-environments/live-events.test.ts @@ -15,6 +15,7 @@ import { getAgentEventLifecycleGeneration, getAgentRunContext, onAgentRuntimeEvent, + releaseAgentRunContext, sweepStaleRunContexts, type AgentEventRuntimePayload as Event, } from "../../infra/agent-events.js"; @@ -316,6 +317,27 @@ describe("worker live events", () => { expect(deltas()).toEqual(["first", "second", "new", "current"]); }); + it("retires completed process fences when a new turn reuses its durable run id", () => { + ack(live(1, lifecycle({ phase: "start", startedAt: 100 }))); + ack(live(2, lifecycle({ phase: "end", startedAt: 100, endedAt: 200 }))); + const credentialHash = ["next", "process", "credential"].join("-"); + + expect( + rx.rotateCredential({ + credentialHash, + environmentId: ID.environmentId, + newProcessTurn: true, + previousCredentialHash: ID.credentialHash, + runEpoch: EPOCH, + sessionId: SID, + }), + ).toBe(true); + + const nextProcess = { ...ID, credentialHash }; + ack(live(3, lifecycle({ phase: "start", startedAt: 300 })), 3, nextProcess); + expect(events.map((event) => event.data.phase)).toEqual(["start", "end", "start"]); + }); + it("ACKs before buffered failure", () => { const first = msg(1, "first", 0, "run-prefix"); const second = msg(2, "second", 0, "run-buffered"); @@ -579,6 +601,51 @@ describe("worker live events", () => { expect(events[0]?.controlUiVisible).toBe(true); }); + it("shares a compatible non-exclusive Gateway run owner", () => { + const lifecycleGeneration = getAgentEventLifecycleGeneration(); + const gatewayClaim = claimAgentRunContext( + RUN, + { + sessionId: LOCAL.sessionId, + sessionKey: LOCAL.sessionKey, + isControlUiVisible: false, + lifecycleGeneration, + }, + { ownsContext: true, trackOwner: true }, + ); + expect(gatewayClaim).toBeDefined(); + + ack(live(1, lifecycle({ phase: "start", startedAt: 100 }))); + + expect(getAgentRunContext(RUN)).toMatchObject({ + ...LOCAL, + isControlUiVisible: false, + lifecycleGeneration, + projectSessionActive: true, + }); + expect(events).toHaveLength(1); + expect(events[0]?.controlUiVisible).toBe(false); + + rx.clear(); + expect(getAgentRunContext(RUN)).toBeDefined(); + releaseAgentRunContext(RUN, gatewayClaim); + }); + + it("rejects a compatible context held by an exclusive Gateway owner", () => { + const lifecycleGeneration = getAgentEventLifecycleGeneration(); + const gatewayClaim = claimAgentRunContext( + RUN, + { ...LOCAL, lifecycleGeneration }, + { exclusive: true, ownsContext: true, trackOwner: true }, + ); + expect(gatewayClaim).toBeDefined(); + + fail(msg(1, "blocked"), "invalid-event"); + expect(events).toEqual([]); + + releaseAgentRunContext(RUN, gatewayClaim); + }); + it("rejects pre-registered gateway run contexts with mismatched identity", () => { const lifecycleGeneration = getAgentEventLifecycleGeneration(); const mismatches: Array<{ diff --git a/src/gateway/worker-environments/live-events.ts b/src/gateway/worker-environments/live-events.ts index bf1f6682297b..bfcdb2aaabe7 100644 --- a/src/gateway/worker-environments/live-events.ts +++ b/src/gateway/worker-environments/live-events.ts @@ -11,6 +11,7 @@ import { import type { OpenClawConfig } from "../../config/types.openclaw.js"; import { claimAgentRunContext, + emitAgentEventIfCurrent, emitAgentEventForOwner, getAgentEventLifecycleGeneration, getAgentRunContext, @@ -43,6 +44,7 @@ type PendingLiveEvent = { type OwnedLiveRun = { claimId: string; controlUiVisible: boolean; + emissionMode: "exclusive" | "shared"; lifecycleGeneration: string; trajectoryRecorder: WorkerLiveTrajectoryRecorder; }; @@ -60,6 +62,7 @@ type BoundLiveSession = WorkerLiveSessionBinding & { target: LiveEventTarget }; type WorkerLiveCredentialRotation = Readonly<{ credentialHash: string; environmentId: string; + newProcessTurn?: boolean; previousCredentialHash: string; runEpoch: number; sessionId: string; @@ -233,6 +236,18 @@ export function createWorkerLiveEventReceiver(options: WorkerLiveEventReceiverOp window.environmentId === rotation.environmentId && window.runEpoch === rotation.runEpoch ) { + if (rotation.newProcessTurn === true) { + // A per-turn credential is an unforgeable process boundary. Retire only + // the prior process's transient run claims/fences while preserving the + // durable ACK cursor; cron may intentionally reuse its durable run id. + for (const [runId, owned] of window.activeRuns) { + releaseAgentRunContext(runId, owned.claimId); + } + window.activeRuns.clear(); + window.pending.clear(); + window.pendingBytes = 0; + window.terminalRuns.clear(); + } window.credentialHash = rotation.credentialHash; return true; } @@ -552,12 +567,14 @@ export function createWorkerLiveEventReceiver(options: WorkerLiveEventReceiverOp existingContext && (existingContext.sessionId !== window.sessionId || existingContext.sessionKey !== window.target.sessionKey || - existingContext.agentId !== window.target.agentId || + (existingContext.agentId !== undefined && + existingContext.agentId !== window.target.agentId) || existingContext.lifecycleGeneration !== lifecycleGeneration) ) { return invalidEvent(); } - const claimId = claimAgentRunContext( + let emissionMode: OwnedLiveRun["emissionMode"] = "exclusive"; + let claimId = claimAgentRunContext( runId, { ...(window.target.agentId ? { agentId: window.target.agentId } : {}), @@ -579,12 +596,40 @@ export function createWorkerLiveEventReceiver(options: WorkerLiveEventReceiverOp trackOwner: true, }, ); + if (!claimId && existingContext) { + // Cron and other Gateway-owned handoffs retain a non-exclusive claim while the + // assigned worker runs. Share only that corroborated identity; exclusive owners + // still reject this claim and prevent a foreign execution from joining the run. + claimId = claimAgentRunContext( + runId, + { + ...(window.target.agentId ? { agentId: window.target.agentId } : {}), + isControlUiVisible: controlUiVisible, + lifecycleGeneration, + projectSessionActive: true, + sessionId: window.sessionId, + sessionKey: window.target.sessionKey, + }, + { + exclusive: false, + onClearRequested: (clearedClaimId) => { + if (window.activeRuns.get(runId)?.claimId === clearedClaimId) { + fenceReleasedRun(window, runId); + } + }, + ownsContext: false, + trackOwner: true, + }, + ); + emissionMode = "shared"; + } if (!claimId) { return invalidEvent(); } const claimed = { claimId, controlUiVisible, + emissionMode, lifecycleGeneration, trajectoryRecorder: createWorkerLiveTrajectoryRecorder({ runId, target: window.target }), }; @@ -607,14 +652,21 @@ export function createWorkerLiveEventReceiver(options: WorkerLiveEventReceiverOp // Fence first so terminal delivery cannot reopen the run ID. window.terminalRuns.set(request.runId, request.seq); } - emitAgentEventForOwner( - { - runId: request.runId, - stream: request.event.kind, - data: prepareWorkerLiveEventData(request.event), - }, - owned.claimId, - ); + const event = { + runId: request.runId, + stream: request.event.kind, + data: prepareWorkerLiveEventData(request.event), + }; + if (owned.emissionMode === "shared") { + if (!emitAgentEventIfCurrent(event)) { + if (definitiveTerminal) { + window.terminalRuns.delete(request.runId); + } + return invalidEvent(); + } + } else { + emitAgentEventForOwner(event, owned.claimId); + } recordWorkerLiveTrajectoryEvent(owned.trajectoryRecorder, request.event); // Gateway handler owns cleanup so detach can revoke deferred terminal delivery. return undefined; diff --git a/src/gateway/worker-environments/service.ts b/src/gateway/worker-environments/service.ts index 0c34a9333cdd..3017b128c74b 100644 --- a/src/gateway/worker-environments/service.ts +++ b/src/gateway/worker-environments/service.ts @@ -1528,6 +1528,7 @@ export function createWorkerEnvironmentService(options: WorkerEnvironmentService options.liveEvents?.rotateCredential({ credentialHash: minted.credentialHash, environmentId: binding.environmentId, + newProcessTurn: true, previousCredentialHash: previous.credentialHash, runEpoch: binding.ownerEpoch, sessionId: binding.sessionId, diff --git a/src/gateway/worker-environments/worker-tool-authority.test.ts b/src/gateway/worker-environments/worker-tool-authority.test.ts new file mode 100644 index 000000000000..bf9acfb3fa37 --- /dev/null +++ b/src/gateway/worker-environments/worker-tool-authority.test.ts @@ -0,0 +1,111 @@ +import { describe, expect, it } from "vitest"; +import type { SessionPlacementTurnParams } from "../../agents/session-placement-admission.js"; +import { resolveWorkerToolAuthority } from "./worker-tool-authority.js"; + +function turn(overrides: Partial = {}): SessionPlacementTurnParams { + return { + sessionId: "session-worker-authority", + sessionKey: "agent:main:cron:job:run:session", + sessionFile: "/tmp/session.jsonl", + workspaceDir: "/tmp/workspace", + prompt: "run", + timeoutMs: 1_000, + runId: "run-worker-authority", + provider: "openai", + model: "gpt-test", + agentId: "main", + ...overrides, + } as SessionPlacementTurnParams; +} + +function authority(overrides: Partial = {}) { + return resolveWorkerToolAuthority({ + modelRef: { provider: "openai", model: "gpt-test" }, + turn: turn(overrides), + }).allowedToolNames; +} + +describe("resolveWorkerToolAuthority", () => { + it("keeps the deterministic complete worker surface when no policy narrows it", () => { + expect(authority()).toEqual(["read", "write", "edit", "apply_patch", "exec", "process"]); + }); + + it("projects runtime caps with canonical write-to-apply_patch semantics", () => { + expect(authority({ toolsAllow: ["write"] })).toEqual(["write", "apply_patch"]); + expect(authority({ toolsAllow: [] })).toEqual([]); + expect(authority({ toolsAllow: ["web_search"] })).toEqual([]); + }); + + it("uses scheduled owner group policy without reapplying fresh sender overlays", () => { + const config = { + tools: { + deny: ["exec"], + toolsBySender: { "*": { deny: ["write", "apply_patch"] } }, + }, + channels: { + whatsapp: { + groups: { + team: { + tools: { allow: ["read", "write", "exec"] }, + toolsBySender: { "*": { deny: ["write", "apply_patch"] } }, + }, + }, + }, + }, + } as SessionPlacementTurnParams["config"]; + + expect( + authority({ + config, + messageProvider: "whatsapp", + senderId: "guest", + toolsAllow: ["read", "write", "exec"], + scheduledToolPolicy: { ownerSessionKey: "agent:main:whatsapp:group:team" }, + }), + ).toEqual(["read", "write", "apply_patch"]); + expect( + authority({ + config, + messageProvider: "whatsapp", + senderId: "guest", + toolsAllow: ["read", "write", "exec"], + }), + ).toEqual(["read"]); + }); + + it("re-resolves current owner-group restrictions for every scheduled turn", () => { + expect( + authority({ + config: { + channels: { + whatsapp: { + groups: { team: { tools: { deny: ["write", "apply_patch"] } } }, + }, + }, + }, + messageProvider: "whatsapp", + toolsAllow: ["write"], + scheduledToolPolicy: { ownerSessionKey: "agent:main:whatsapp:group:team" }, + }), + ).toEqual([]); + }); + + it("applies sandbox tool policy when the session is configured for sandboxing", () => { + expect( + authority({ + sessionKey: "agent:main:worker-sandboxed", + config: { + agents: { defaults: { sandbox: { mode: "all" } } }, + tools: { sandbox: { tools: { allow: ["read"] } } }, + }, + }), + ).toEqual(["read"]); + }); + + it.each([{ disableTools: true }, { modelRun: true }, { promptMode: "none" as const }])( + "exposes no tools for non-tool run mode %#", + (overrides) => { + expect(authority(overrides)).toEqual([]); + }, + ); +}); diff --git a/src/gateway/worker-environments/worker-tool-authority.ts b/src/gateway/worker-environments/worker-tool-authority.ts new file mode 100644 index 000000000000..8ff8009b1872 --- /dev/null +++ b/src/gateway/worker-environments/worker-tool-authority.ts @@ -0,0 +1,90 @@ +import { resolveConversationCapabilityProfile } from "../../agents/conversation-capability-profile.js"; +import { projectConversationToolNames } from "../../agents/conversation-tool-policy-pipeline.js"; +import { applyEmbeddedAttemptToolsAllow } from "../../agents/embedded-agent-runner/run/attempt-tool-construction-plan.js"; +import { resolveSandboxRuntimeStatus } from "../../agents/sandbox/runtime-status.js"; +import type { SessionPlacementTurnParams } from "../../agents/session-placement-admission.js"; +import { logWarn } from "../../logger.js"; +import { + WORKER_LOCAL_TOOL_NAMES, + type WorkerLocalToolName, + type WorkerToolAuthority, +} from "../../worker/tool-authority.js"; + +function resolveWorkerCapabilityProfile(params: { + modelRef: { provider: string; model: string }; + turn: SessionPlacementTurnParams; +}) { + const turn = params.turn; + const sandboxSessionKey = + turn.sandboxSessionKey?.trim() || turn.sessionKey?.trim() || turn.sessionId; + const sandbox = resolveSandboxRuntimeStatus({ + cfg: turn.config, + sessionKey: sandboxSessionKey, + agentId: turn.agentId, + }); + return resolveConversationCapabilityProfile({ + config: turn.config, + sessionKey: sandboxSessionKey, + runSessionKey: + turn.sessionKey && turn.sessionKey !== sandboxSessionKey ? turn.sessionKey : undefined, + sessionId: turn.sessionId, + runId: turn.runId, + agentId: turn.agentId, + agentDir: turn.agentDir, + agentAccountId: turn.agentAccountId, + messageProvider: turn.messageProvider, + messageChannel: turn.messageChannel, + chatType: turn.chatType, + messageTo: turn.messageTo, + messageThreadId: turn.messageThreadId, + currentChannelId: turn.currentChannelId, + currentMessagingTarget: turn.currentMessagingTarget, + currentThreadTs: turn.currentThreadTs, + currentMessageId: turn.currentMessageId, + groupId: turn.groupId, + groupChannel: turn.groupChannel, + groupSpace: turn.groupSpace, + memberRoleIds: turn.memberRoleIds, + spawnedBy: turn.spawnedBy, + senderId: turn.senderId, + senderName: turn.senderName, + senderUsername: turn.senderUsername, + senderE164: turn.senderE164, + senderIsOwner: turn.senderIsOwner, + modelProvider: params.modelRef.provider, + modelId: params.modelRef.model, + workspaceDir: turn.workspaceDir, + cwd: turn.cwd, + isCanonicalWorkspace: turn.isCanonicalWorkspace, + promptMode: turn.promptMode, + skillsSnapshot: turn.skillsSnapshot, + sandboxToolPolicy: sandbox.sandboxed ? sandbox.toolPolicy : undefined, + runtimeToolAllowlist: turn.toolsAllow, + inheritRuntimeToolAllowlist: true, + runtimePluginToolGrant: turn.runtimePluginToolGrant, + inputProvenance: turn.inputProvenance, + trustedInternalHandoff: turn.trustedInternalHandoff, + scheduledToolPolicy: turn.scheduledToolPolicy, + }); +} + +/** Resolves the final fixed worker surface at the trusted Gateway handoff boundary. */ +export function resolveWorkerToolAuthority(params: { + modelRef: { provider: string; model: string }; + turn: SessionPlacementTurnParams; +}): WorkerToolAuthority { + const turn = params.turn; + if (turn.disableTools === true || turn.modelRun === true || turn.promptMode === "none") { + return { allowedToolNames: [] }; + } + const runtimeCappedTools = applyEmbeddedAttemptToolsAllow( + WORKER_LOCAL_TOOL_NAMES.map((name) => ({ name })), + turn.toolsAllow, + ); + const projected: WorkerLocalToolName[] = projectConversationToolNames({ + capabilityProfile: resolveWorkerCapabilityProfile(params), + toolNames: runtimeCappedTools.map((tool) => tool.name), + warn: logWarn, + }); + return { allowedToolNames: projected }; +} diff --git a/src/gateway/worker-environments/worker-turn-launcher.test.ts b/src/gateway/worker-environments/worker-turn-launcher.test.ts index de5880be19d1..ac2a0a95a4b8 100644 --- a/src/gateway/worker-environments/worker-turn-launcher.test.ts +++ b/src/gateway/worker-environments/worker-turn-launcher.test.ts @@ -30,9 +30,9 @@ import { } from "./placement-store.js"; import { createWorkerSessionPlacementGate } from "./placement-worker-gate.js"; import type { WorkerTunnelHandle } from "./tunnel-contract.js"; -import { createWorkerSessionTurnPlacementProvider } from "./worker-turn-launcher.js"; +import { createWorkerSessionTurnPlacementProvider as createRawWorkerSessionTurnPlacementProvider } from "./worker-turn-launcher.js"; -type WorkerTurnLauncherOptions = Parameters[0]; +type WorkerTurnLauncherOptions = Parameters[0]; type WorkerTurnEnvironmentService = WorkerTurnLauncherOptions["environments"]; const SESSION_ID = "session-worker-turn"; @@ -74,6 +74,16 @@ describe("worker turn launcher", () => { await fs.rm(root, { recursive: true, force: true }); }); + function createWorkerSessionTurnPlacementProvider( + options: Omit & + Partial>, + ) { + return createRawWorkerSessionTurnPlacementProvider({ + resolveWorkspacePath: async () => root, + ...options, + }); + } + function seedActivePlacement(): void { let placement = placements.startDispatch({ sessionId: SESSION_ID, @@ -594,7 +604,12 @@ describe("worker turn launcher", () => { stopTunnel: vi.fn(async () => {}), destroy: vi.fn(async () => attachedEnvironment()), }; - const provider = createWorkerSessionTurnPlacementProvider({ environments, placements }); + const resolveWorkspacePath = vi.fn(async () => root); + const provider = createWorkerSessionTurnPlacementProvider({ + environments, + placements, + resolveWorkspacePath, + }); const runLocal = vi.fn(async () => ({ meta: { durationMs: 1 } })); const onAgentEvent = vi.fn(() => { throw new Error("supplemental event failed"); @@ -607,11 +622,25 @@ describe("worker turn launcher", () => { agentId: "main", runId: "run-worker-turn", }, - { ...turn(), transcriptPrompt: "Canonical transcript request", onAgentEvent }, + { + ...turn(), + workspaceDir: path.join(root, "stale-caller-workspace"), + transcriptPrompt: "Canonical transcript request", + onAgentEvent, + }, runLocal, ); expect(runLocal).not.toHaveBeenCalled(); + expect(resolveWorkspacePath).toHaveBeenCalledWith({ + sessionId: SESSION_ID, + sessionKey: SESSION_KEY, + agentId: "main", + runId: "run-worker-turn", + }); + expect(tunnel.reconcileWorkspace).toHaveBeenCalledWith( + expect.objectContaining({ localPath: root }), + ); const conflictSummary = "Cloud result applied with 1 conflict(s); kept local versions: src/local.ts. Cloud versions staged at refs/openclaw/worker-results/"; expect(result.payloads).toEqual([ @@ -639,6 +668,15 @@ describe("worker turn launcher", () => { ).toBe(true); expect(descriptor?.assignment.prompt).toBe("Inspect this workspace"); expect(descriptor?.assignment.suppressPromptTranscript).toBe(true); + expect(descriptor?.version).toBe(2); + expect(descriptor?.assignment.toolAuthority.allowedToolNames).toEqual([ + "read", + "write", + "edit", + "apply_patch", + "exec", + "process", + ]); expect(descriptor?.assignment.initialMessages).toEqual([ { role: "user", diff --git a/src/gateway/worker-environments/worker-turn-launcher.ts b/src/gateway/worker-environments/worker-turn-launcher.ts index 96ea479a264f..aaf75d68b59f 100644 --- a/src/gateway/worker-environments/worker-turn-launcher.ts +++ b/src/gateway/worker-environments/worker-turn-launcher.ts @@ -18,6 +18,7 @@ import type { WorkerSessionTurnClaim, } from "./placement-store.js"; import type { WorkerEnvironmentService } from "./service.js"; +import { resolveWorkerToolAuthority } from "./worker-tool-authority.js"; import { claimWorkerTurn, latestDurableWorkspaceConflict, @@ -70,6 +71,7 @@ type WorkerTurnLauncherOptions = { admitNewPlacements?: boolean; environments: WorkerTurnEnvironmentService; placements: WorkerSessionPlacementStore; + resolveWorkspacePath: (claim: LocalTurnPlacementClaim) => Promise; workspaceOperations?: WorkerWorkspaceOperationCoordinator; redispatchReclaimed?: (placement: ReclaimedWorkerPlacement) => Promise; }; @@ -172,6 +174,7 @@ async function executeWorkerTurn(params: { workspaceOperations: WorkerWorkspaceOperationCoordinator; turn: SessionPlacementTurnParams; turnClaim: WorkerSessionTurnClaim; + localWorkspaceDir: string; }) { const { placement, turn } = params; const modelRef = assertSupportedTurn(turn); @@ -216,7 +219,10 @@ async function executeWorkerTurn(params: { } const pending = journal.load(); if (pending) { - await recoverWorkerWorkspaceReconciliation({ root: turn.workspaceDir, journal: pending }); + await recoverWorkerWorkspaceReconciliation({ + root: params.localWorkspaceDir, + journal: pending, + }); journal.abort(); } }); @@ -244,7 +250,7 @@ async function executeWorkerTurn(params: { let baseLeafId = manager.getLeafId(); if (!userMessageAlreadyPersisted) { const persisted = turn.userTurnTranscriptRecorder - ? await turn.userTurnTranscriptRecorder.persistApproved({ cwd: turn.workspaceDir }) + ? await turn.userTurnTranscriptRecorder.persistApproved({ cwd: params.localWorkspaceDir }) : undefined; if (persisted) { baseLeafId = persisted.messageId; @@ -285,10 +291,11 @@ async function executeWorkerTurn(params: { timeoutMs: turn.timeoutMs, }); const reasoning = mapThinkingLevelForProvider(turn.thinkLevel); + const toolAuthority = resolveWorkerToolAuthority({ modelRef, turn }); const descriptor = fitLaunchDescriptor( (windowedMessages) => parseWorkerLaunchDescriptor({ - version: 1, + version: 2, socketPath: tunnel.remoteSocketPath, admission: { environmentId: placement.environmentId, @@ -316,6 +323,7 @@ async function executeWorkerTurn(params: { ackedSeq: placement.lastLiveEventAckCursor ?? 0, nextSeq: (placement.lastLiveEventAckCursor ?? 0) + 1, }, + toolAuthority, }, }), initialMessages, @@ -426,7 +434,7 @@ async function executeWorkerTurn(params: { try { const stagedResultRef = workerWorkspaceResultRef(params.turnClaim.claimId); const reconciliation = await tunnel.reconcileWorkspace({ - localPath: turn.workspaceDir, + localPath: params.localWorkspaceDir, remoteWorkspaceDir: currentPlacement.remoteWorkspaceDir, baseManifestRef: currentPlacement.workspaceBaseManifestRef, journal, @@ -457,7 +465,7 @@ async function executeWorkerTurn(params: { conflictPaths: applied?.conflictPaths ?? [], priorConflict: priorWorkspaceConflict, stagedResultRef: recordedStagedResultRef, - root: turn.workspaceDir, + root: params.localWorkspaceDir, report: async (report) => { if ("cleared" in report) { SessionManager.open(turn.sessionFile).appendCustomMessageEntry( @@ -490,7 +498,7 @@ async function executeWorkerTurn(params: { await settleStagedWorkspaceResult({ placements: params.placements, turnClaim: params.turnClaim, - root: turn.workspaceDir, + root: params.localWorkspaceDir, stagedResultRef: recordedStagedResultRef, conflictRetained: finalized.conflictRetained, reclaim: false, @@ -585,6 +593,9 @@ export function createWorkerSessionTurnPlacementProvider( } const identity = resolvePlacementIdentity(claim, routablePlacement); let placement = requireActivePlacement(routablePlacement); + // The placement owns the managed worktree. Callers can carry a default or stale + // workspace path, but remote results must only reconcile into that canonical root. + const localWorkspaceDir = await options.resolveWorkspacePath(claim); const admitted = await claimWorkerTurn({ placements: options.placements, identity, @@ -603,6 +614,7 @@ export function createWorkerSessionTurnPlacementProvider( }, placement, placements: options.placements, + localWorkspaceDir, workspaceOperations, turn, turnClaim, diff --git a/src/gateway/worker-environments/worker-turn-payload.test.ts b/src/gateway/worker-environments/worker-turn-payload.test.ts index 1fef16a08110..8d3ef81ba812 100644 --- a/src/gateway/worker-environments/worker-turn-payload.test.ts +++ b/src/gateway/worker-environments/worker-turn-payload.test.ts @@ -3,8 +3,8 @@ import type { SessionPlacementTurnParams } from "../../agents/session-placement- import { assertSupportedTurn } from "./worker-turn-payload.js"; describe("assertSupportedTurn", () => { - it("rejects scheduled authority before cloud-worker handoff", () => { - expect(() => + it("accepts scheduled authority for the worker launch envelope", () => { + expect( assertSupportedTurn({ sessionId: "session-1", sessionFile: "/tmp/session.jsonl", @@ -14,9 +14,16 @@ describe("assertSupportedTurn", () => { runId: "run-1", provider: "openai", model: "gpt-5.4", + config: { + agents: { + defaults: { + models: { "openai/gpt-5.4": { agentRuntime: { id: "openclaw" } } }, + }, + }, + }, toolsAllow: ["write"], scheduledToolPolicy: { ownerSessionKey: "agent:main:discord:group:ops" }, } as SessionPlacementTurnParams), - ).toThrow("Cloud worker turns do not yet preserve scheduled tool policy"); + ).toEqual({ provider: "openai", model: "gpt-5.4" }); }); }); diff --git a/src/gateway/worker-environments/worker-turn-payload.ts b/src/gateway/worker-environments/worker-turn-payload.ts index ef75ab76daaa..24e3ff515f42 100644 --- a/src/gateway/worker-environments/worker-turn-payload.ts +++ b/src/gateway/worker-environments/worker-turn-payload.ts @@ -176,9 +176,6 @@ export function assertSupportedTurn(params: SessionPlacementTurnParams): { if (params.clientTools?.length) { throw new Error("Cloud worker turns do not support client-provided tools"); } - if (params.scheduledToolPolicy) { - throw new Error("Cloud worker turns do not yet preserve scheduled tool policy"); - } const modelRef = resolveTurnModelRef(params); const explicitRuntime = normalizeOptionalAgentRuntimeId(params.agentHarnessId) ?? diff --git a/src/worker/embedded-agent.runtime.ts b/src/worker/embedded-agent.runtime.ts index 9a750433fdb1..a2b89adfc0b3 100644 --- a/src/worker/embedded-agent.runtime.ts +++ b/src/worker/embedded-agent.runtime.ts @@ -27,17 +27,9 @@ import { toAgentMessage, toWorkerInferenceContext, } from "./embedded-agent-transcript.runtime.js"; +import { WORKER_LOCAL_TOOL_NAMES, type WorkerLocalToolName } from "./tool-authority.js"; import { toWorkerTranscriptMessage } from "./transcript-message.js"; -const LOCAL_WORKER_TOOL_NAMES = [ - "read", - "write", - "edit", - "apply_patch", - "exec", - "process", -] as const; - function toError(value: unknown, fallback: string): Error { return value instanceof Error ? value : new Error(fallback, { cause: value }); } @@ -78,6 +70,7 @@ type RunWorkerEmbeddedTurnParams = { suppressPromptTranscript?: boolean; systemPrompt?: string; inferenceOptions?: WorkerInferenceOptions; + allowedToolNames: readonly WorkerLocalToolName[]; signal?: AbortSignal; }; @@ -127,7 +120,9 @@ export async function runWorkerEmbeddedTurn( onMessagePersisted: transcriptRuntime.onMessagePersisted, }); - const toolNameSet = new Set(LOCAL_WORKER_TOOL_NAMES); + const allowedToolNameSet = new Set(params.allowedToolNames); + const activeToolNames = WORKER_LOCAL_TOOL_NAMES.filter((name) => allowedToolNameSet.has(name)); + const localToolNameSet = new Set(WORKER_LOCAL_TOOL_NAMES); const localTools = createOpenClawCodingTools({ cwd: params.cwd, workspaceDir: params.cwd, @@ -138,7 +133,7 @@ export async function runWorkerEmbeddedTurn( oneShotCliRun: true, senderIsOwner: true, disableMessageTool: true, - runtimeToolAllowlist: [...LOCAL_WORKER_TOOL_NAMES], + runtimeToolAllowlist: [...WORKER_LOCAL_TOOL_NAMES], modelProvider: params.modelRef.provider, modelId: params.modelRef.model, modelApi: model.api, @@ -152,9 +147,9 @@ export async function runWorkerEmbeddedTurn( includeOpenClawTools: false, includePluginTools: false, }, - }).filter((tool) => toolNameSet.has(tool.name)); + }).filter((tool) => localToolNameSet.has(tool.name)); const discoveredToolNames = new Set(localTools.map((tool) => tool.name)); - for (const toolName of LOCAL_WORKER_TOOL_NAMES) { + for (const toolName of WORKER_LOCAL_TOOL_NAMES) { if (!discoveredToolNames.has(toolName)) { throw new Error(`Worker coding tool unavailable: ${toolName}`); } @@ -167,8 +162,8 @@ export async function runWorkerEmbeddedTurn( modelRegistry, model, thinkingLevel: "medium", - tools: [...LOCAL_WORKER_TOOL_NAMES], - customTools: toToolDefinitions(localTools), + tools: [...activeToolNames], + customTools: toToolDefinitions(localTools.filter((tool) => allowedToolNameSet.has(tool.name))), noTools: "all", sessionManager, settingsManager, @@ -176,7 +171,7 @@ export async function runWorkerEmbeddedTurn( withSessionWriteLock: transcriptRuntime.withSessionWriteLock, }); session.agent.sessionId = params.sessionId; - session.setActiveToolsByName([...LOCAL_WORKER_TOOL_NAMES]); + session.setActiveToolsByName([...activeToolNames]); session.agent.streamFn = (_model, context, options) => params.inference.stream({ modelRef: params.modelRef, diff --git a/src/worker/launch-descriptor.test.ts b/src/worker/launch-descriptor.test.ts index 35c0f4a09927..e05eb318416c 100644 --- a/src/worker/launch-descriptor.test.ts +++ b/src/worker/launch-descriptor.test.ts @@ -10,7 +10,7 @@ import { buildWorkerConnectParams, parseWorkerLaunchDescriptor } from "./launch- function launchDescriptor(): WorkerLaunchDescriptor { return { - version: 1, + version: 2, socketPath: "/tmp/openclaw-worker/gateway.sock", admission: { environmentId: "environment-1", @@ -41,6 +41,7 @@ function launchDescriptor(): WorkerLaunchDescriptor { ], transcript: { baseLeafId: "leaf-7", nextSeq: 8 }, liveEvents: { ackedSeq: 12, nextSeq: 13 }, + toolAuthority: { allowedToolNames: ["read", "exec"] }, }, }; } @@ -97,6 +98,13 @@ describe("worker launch descriptor", () => { liveEvents: { ...descriptor.assignment.liveEvents, unexpected: true }, }, }, + { + ...descriptor, + assignment: { + ...descriptor.assignment, + toolAuthority: { ...descriptor.assignment.toolAuthority, unexpected: true }, + }, + }, ]; for (const candidate of cases) { @@ -106,6 +114,38 @@ describe("worker launch descriptor", () => { } }); + it("requires a unique closed worker tool authority", () => { + const descriptor = launchDescriptor(); + const { toolAuthority: _missing, ...assignmentWithoutAuthority } = descriptor.assignment; + const cases: unknown[] = [ + { ...descriptor, version: 1 }, + { ...descriptor, assignment: assignmentWithoutAuthority }, + { + ...descriptor, + assignment: { + ...descriptor.assignment, + toolAuthority: { allowedToolNames: ["read", "read"] }, + }, + }, + { + ...descriptor, + assignment: { + ...descriptor.assignment, + toolAuthority: { allowedToolNames: ["read", "gateway"] }, + }, + }, + ]; + + for (const candidate of cases) { + expect(() => parseWorkerLaunchDescriptor(candidate)).toThrow( + "invalid worker launch descriptor", + ); + } + + descriptor.assignment.toolAuthority.allowedToolNames = []; + expect(parseWorkerLaunchDescriptor(structuredClone(descriptor))).toEqual(descriptor); + }); + it("rejects non-absolute paths, unattached sessions, and discontinuous event sequences", () => { const descriptor = launchDescriptor(); const cases: unknown[] = [ diff --git a/src/worker/launch-descriptor.ts b/src/worker/launch-descriptor.ts index ee3eb50728fb..cfe0799f98f8 100644 --- a/src/worker/launch-descriptor.ts +++ b/src/worker/launch-descriptor.ts @@ -23,9 +23,10 @@ import { WorkerInferenceOptionsSchema, } from "../../packages/gateway-protocol/src/schema/worker-inference.js"; import { PROTOCOL_VERSION } from "../../packages/gateway-protocol/src/version.js"; +import { isWorkerLocalToolName, type WorkerToolAuthority } from "./tool-authority.js"; import { isWorkerTranscriptMessageFrameSafe } from "./transcript-message.js"; -const LAUNCH_VERSION = 1; +const LAUNCH_VERSION = 2; type WorkerLaunchAssignment = { runId: string; @@ -45,6 +46,7 @@ type WorkerLaunchAssignment = { ackedSeq: number; nextSeq: number; }; + toolAuthority: WorkerToolAuthority; }; type WorkerLaunchAdmission = Omit & { @@ -52,7 +54,7 @@ type WorkerLaunchAdmission = Omit & { }; export type WorkerLaunchDescriptor = { - version: 1; + version: 2; socketPath: string; admission: WorkerLaunchAdmission; assignment: WorkerLaunchAssignment; @@ -86,6 +88,19 @@ function isInferenceOptions(value: unknown): value is WorkerInferenceOptions { return Value.Check(WorkerInferenceOptionsSchema, value); } +function parseToolAuthority(value: unknown): WorkerToolAuthority | undefined { + if ( + !isRecord(value) || + !hasExactKeys(value, ["allowedToolNames"]) || + !Array.isArray(value.allowedToolNames) || + !value.allowedToolNames.every(isWorkerLocalToolName) || + new Set(value.allowedToolNames).size !== value.allowedToolNames.length + ) { + return undefined; + } + return { allowedToolNames: [...value.allowedToolNames] }; +} + function parseAssignment(value: unknown): WorkerLaunchAssignment | undefined { if ( !isRecord(value) || @@ -102,6 +117,7 @@ function parseAssignment(value: unknown): WorkerLaunchAssignment | undefined { "initialMessages", "transcript", "liveEvents", + "toolAuthority", ], ["systemPrompt"], ) @@ -122,6 +138,10 @@ function parseAssignment(value: unknown): WorkerLaunchAssignment | undefined { ) { return undefined; } + const toolAuthority = parseToolAuthority(value.toolAuthority); + if (!toolAuthority) { + return undefined; + } if ( !Value.Check(WorkerInferenceModelRefSchema, value.modelRef) || !isInferenceOptions(value.inferenceOptions) @@ -145,7 +165,7 @@ function parseAssignment(value: unknown): WorkerLaunchAssignment | undefined { ) { return undefined; } - return value as WorkerLaunchAssignment; + return { ...value, toolAuthority } as WorkerLaunchAssignment; } export function buildWorkerConnectParams( diff --git a/src/worker/tool-authority.ts b/src/worker/tool-authority.ts new file mode 100644 index 000000000000..ab0dd1c8ad60 --- /dev/null +++ b/src/worker/tool-authority.ts @@ -0,0 +1,20 @@ +export const WORKER_LOCAL_TOOL_NAMES = [ + "read", + "write", + "edit", + "apply_patch", + "exec", + "process", +] as const; + +export type WorkerLocalToolName = (typeof WORKER_LOCAL_TOOL_NAMES)[number]; + +const WORKER_LOCAL_TOOL_NAME_SET = new Set(WORKER_LOCAL_TOOL_NAMES); + +export function isWorkerLocalToolName(value: unknown): value is WorkerLocalToolName { + return typeof value === "string" && WORKER_LOCAL_TOOL_NAME_SET.has(value); +} + +export type WorkerToolAuthority = { + allowedToolNames: WorkerLocalToolName[]; +}; diff --git a/src/worker/worker.fault-injection.test.ts b/src/worker/worker.fault-injection.test.ts index 3765521d67af..95fe3f9c1035 100644 --- a/src/worker/worker.fault-injection.test.ts +++ b/src/worker/worker.fault-injection.test.ts @@ -332,7 +332,7 @@ class ComposedGatewayHarness { const epoch = params.epoch ?? this.epoch; const credential = params.admissionProof ?? CREDENTIAL; const descriptor: WorkerLaunchDescriptor = { - version: 1, + version: 2, socketPath: this.socketPath, admission: { environmentId: ENVIRONMENT_ID, @@ -356,6 +356,9 @@ class ComposedGatewayHarness { ackedSeq: params.initialAckedSeq ?? 0, nextSeq: (params.initialAckedSeq ?? 0) + 1, }, + toolAuthority: { + allowedToolNames: ["read", "write", "edit", "apply_patch", "exec", "process"], + }, }, }; const connection = createWorkerConnection({ diff --git a/src/worker/worker.runtime.test.ts b/src/worker/worker.runtime.test.ts index 73a49db0ccf7..fbee955c4115 100644 --- a/src/worker/worker.runtime.test.ts +++ b/src/worker/worker.runtime.test.ts @@ -706,7 +706,7 @@ class FakeWorkerGateway { function descriptor(socketPath: string, workspaceDir: string): WorkerLaunchDescriptor { return { - version: 1, + version: 2, socketPath, admission: { environmentId: "worker-environment", @@ -731,6 +731,9 @@ function descriptor(socketPath: string, workspaceDir: string): WorkerLaunchDescr initialMessages: [], transcript: { baseLeafId: "leaf-base", nextSeq: 3 }, liveEvents: { ackedSeq: 0, nextSeq: 1 }, + toolAuthority: { + allowedToolNames: ["read", "write", "edit", "apply_patch", "exec", "process"], + }, }, }; } @@ -808,6 +811,27 @@ describe("worker runtime", () => { }); }); + it("exposes exactly the Gateway-authorized worker tools", async () => { + const { gateway, launch } = await setup(); + launch.assignment.toolAuthority.allowedToolNames = ["read", "exec"]; + + await expect(runWorkerDescriptor(launch)).resolves.toMatchObject({ status: "completed" }); + + expect(gateway.inferenceRequests[0]?.context.tools?.map((tool) => tool.name)).toEqual([ + "read", + "exec", + ]); + }); + + it("runs with no tools when the Gateway authority is empty", async () => { + const { gateway, launch } = await setup(); + launch.assignment.toolAuthority.allowedToolNames = []; + + await expect(runWorkerDescriptor(launch)).resolves.toMatchObject({ status: "completed" }); + + expect(gateway.inferenceRequests[0]?.context.tools ?? []).toEqual([]); + }); + it("fail-stops a stale mid-run transcript without duplicating or rebasing the paid tail", async () => { const { gateway, launch } = await setup({ transcriptFailureAtRequest: 2 }); diff --git a/src/worker/worker.runtime.ts b/src/worker/worker.runtime.ts index 7dd3ae538292..0d5c22dbdbdd 100644 --- a/src/worker/worker.runtime.ts +++ b/src/worker/worker.runtime.ts @@ -132,6 +132,7 @@ export async function runWorkerDescriptor( ? {} : { systemPrompt: descriptor.assignment.systemPrompt }), inferenceOptions: descriptor.assignment.inferenceOptions, + allowedToolNames: descriptor.assignment.toolAuthority.allowedToolNames, inference: { stream }, transcript: { commit: async (messages) => {