vrr/openclaw
2
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-11-04 06:35:52 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 23

feat: enforce required worker policy before local inference (#166611)

Browse source 
* fix: source workspace workers load from external working directories

* fix: stop device preparation when dispatch authority ends

* fix: retain session custody until dispatch work settles

* refactor: introduce custody export with its production caller

* test: persist worker custody fixtures and drain databases

* feat: enforce required worker policy before local inference

* fix: fence provider auth with live worker authority

* test: prove live worker authority at provider I/O

* test(gateway): align inherited fixtures with current projections
This commit is contained in:
Dallin Romney 2026-10-07 13:17:34 -07:00 • committed by GitHub
parent e73028f81c
commit c18456cce9
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
26 changed files with 1404 additions and 205 deletions
Download patch file Download diff file Expand all files Collapse all files

2
docs/.generated/config-baseline.counts.json
View file

@ -1,5 +1,5 @@
{
"core": 2494,
"core": 2495,
"channel": 3874,
"plugin": 4524
}

4
docs/.generated/config-baseline.sha256
View file

@ -1,4 +1,4 @@
588149ef6dc6dfe173b73439a77255d7c6a49caeaef21e184f0fcbd421c77885 config-baseline.json
f88e493e5d7fdbf2796a8640e01abfdbb636a418dfa8b93efc3eb232b057456e config-baseline.core.json
c0078523400115899089a9419d62462e256fa71e5f87a642ab1e5f508b8ff103 config-baseline.json
cedc5254457691642cb4d00ebdff5096e3732b53ef4a3862d56106c410243b20 config-baseline.core.json
f38b3030f28f66f1ad93eebe364915623144de9bf08d98a8c6ef7eeb5f73ea7c config-baseline.channel.json
af1e981255ae5fc1c346e2f8e3ec432ec14196b6c76c2d625ce2815b17207389 config-baseline.plugin.json

38
src/agents/embedded-agent-runner/run-entry.ts
View file

@ -1,3 +1,4 @@
import { assertRequiredWorkerSelection } from "../../config/required-worker-profile.js";
import type { OpenClawConfig } from "../../config/types.openclaw.js";
import type { ContextEngineHostSupport } from "../../context-engine/host-compat.js";
import {
@ -43,7 +44,11 @@ import type { ModelManifestNormalizationContext } from "../model-ref-shared.js";
import { modelKey } from "../model-ref-shared.js";
import { settleFailedRequesterRun, settleRequesterRun } from "../requester-run-settlement.js";
import { resolveAgentRunAbortLifecycleFields } from "../run-termination.js";
import { resolveSessionPlacementRuntimeOverride } from "../session-placement-admission.js";
import {
resolveSessionPlacementRuntimeOverride,
sessionPlacementUsesWorkerInference,
withRequiredSessionPlacement,
} from "../session-placement-admission.js";
import {
didEmbeddedCyberFailoverTargetCommitWork,
EMBEDDED_CYBER_FAILOVER_TRIGGER_CODE,
@ -164,7 +169,21 @@ export async function runEmbeddedAgentEntry<T extends EmbeddedAgentRunResult>(
abortSignal: params.abortSignal,
};
try {
const result = await runEmbeddedAgentEntryInternal(params);
assertRequiredWorkerSelection(params.selection.cfg, {
agentRuntime: params.harness.resolveRuntimeOverride(
params.selection.provider,
params.selection.model,
),
});
const result = await withRequiredSessionPlacement(
params.identity,
{
config: params.selection.cfg,
assertCurrent: () => admission?.assertSourceCurrent(),
signal: params.abortSignal,
},
() => runEmbeddedAgentEntryInternal(params),
);
// Placement and asynchronous terminal cleanup have finished. Only this
// accepted logical result may release children retained across candidates.
await settleRequesterRun(requester, result.result, () => admission?.assertSourceCurrent());
@ -193,12 +212,16 @@ async function runEmbeddedAgentEntryInternal<T extends EmbeddedAgentRunResult>(
});
const resolveRuntimeOverride = (provider: string, model: string) => {
const requestedRuntime = params.harness.resolveRuntimeOverride(provider, model);
if (requestedRuntime || !placementRuntime) {
return requestedRuntime;
}
assertRequiredWorkerSelection(params.selection.cfg, { agentRuntime: requestedRuntime });
const policy = resolveAgentHarnessPolicy(harnessContext(provider, model));
// Explicit runtime choices still reach placement's compatibility check.
return policy.runtimeSource === "implicit" ? placementRuntime : undefined;
if (params.selection.cfg.cloudWorkers?.requiredProfile) {
return policy.runtime;
}
return requestedRuntime || !placementRuntime
? requestedRuntime
: policy.runtimeSource === "implicit"
? placementRuntime
: undefined;
};
const clearObservedModel = () => {
const event = {
@ -288,6 +311,7 @@ async function runEmbeddedAgentEntryInternal<T extends EmbeddedAgentRunResult>(
...selection,
...params.identity,
operatorAuthority,
skipAuthProfileRuntime: sessionPlacementUsesWorkerInference(params.identity),
abortSignal: params.abortSignal,
resolveAgentHarnessRuntimeOverride: resolveRuntimeOverride,
prepareCandidateChain: async (candidates) => {

62
src/agents/embedded-agent-runner/run-orchestrator.required-profile.test.ts Normal file
View file

@ -0,0 +1,62 @@
import { expect, it, vi } from "vitest";
import { upsertSessionEntryCore } from "../../config/sessions/session-accessor.js";
import type { OpenClawConfig } from "../../config/types.openclaw.js";
import { createOpenClawTestState } from "../../test-utils/openclaw-test-state.js";
import { prepareSystemAgentRunAdmission } from "../admitted-run-context.js";
import { installSessionPlacementAdmissionProvider } from "../session-placement-admission.js";
import { runEmbeddedAgent } from "./run-orchestrator.js";
it("resolves the admitted canonical agent before mandatory placement preparation", async () => {
const state = await createOpenClawTestState({ label: "required-worker-canonical-target" });
const agentId = "worker-agent";
const sessionKey = "agent:worker-agent:direct:required";
const sessionId = "canonical-required-session";
const cfg: OpenClawConfig = {
agents: { entries: { [agentId]: { workspace: state.workspaceDir } } },
cloudWorkers: { requiredProfile: "dedicated" },
};
await state.writeConfig(cfg);
await upsertSessionEntryCore({ agentId, sessionKey }, { sessionId, updatedAt: Date.now() });
const reached = new Error("placement owner reached before model execution");
const prepare = vi.fn(
async (identity: { agentId?: string; sessionKey?: string; sessionId: string }) => {
expect(identity).toMatchObject({ agentId, sessionKey, sessionId });
throw reached;
},
);
const local = vi.fn(async () => ({ meta: { durationMs: 0 } }));
const uninstall = installSessionPlacementAdmissionProvider({
withRequiredSession: prepare,
assertCompactionSuccessorAllowed: () => {},
executeLocalTurn: async (_claim, run) => await run(),
executeTurn: local,
});
const admission = prepareSystemAgentRunAdmission(
cfg,
"canonical-required-run",
agentId,
"required-worker-test",
);
try {
await expect(
runEmbeddedAgent({
// Public run inputs can omit agentId when the session key already identifies it.
sessionKey,
sessionId,
sessionFile: sessionKey,
workspaceDir: state.workspaceDir,
config: cfg,
prompt: "Use the dedicated worker",
runId: "canonical-required-run",
timeoutMs: 1000,
preparedRunAdmission: admission,
}),
).rejects.toBe(reached);
expect(prepare).toHaveBeenCalledOnce();
expect(local).not.toHaveBeenCalled();
} finally {
admission.close();
uninstall();
await state.cleanup();
}
});

58
src/agents/embedded-agent-runner/run-orchestrator.ts
View file

@ -1,5 +1,4 @@
import { AsyncLocalStorage } from "node:async_hooks";
import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
import {
createAgentLifecycleTerminalBackstop,
resolveAgentLifecycleTerminalMetadata,
@ -11,7 +10,6 @@ import { revokeMessageActionTurnCapability } from "../../gateway/message-action-
import {
assertAgentRunLifecycleGenerationCurrent,
captureAgentRunLifecycleGeneration,
getAgentEventLifecycleGeneration,
withAgentRunLifecycleGeneration,
} from "../../infra/agent-events.js";
import {
@ -59,11 +57,8 @@ import {
} from "../prepared-model-runtime.js";
import { prepareAgentPromptProjects } from "../prompt-projects.js";
import { settleFailedRequesterRun, settleRequesterRun } from "../requester-run-settlement.js";
import {
applyAgentRunSessionTargetIdentity,
resolveAgentRunSessionTarget,
} from "../run-session-target.js";
import { resolveAgentRunErrorLifecycleFields } from "../run-termination.js";
import { withRequiredSessionPlacement } from "../session-placement-admission.js";
import { resolveSessionPlacementTurnSettlementAssertion } from "../session-placement-forced-terminal-settlement.js";
import {
resolveSessionSuspensionTarget,
@ -96,7 +91,7 @@ import {
import { createEmbeddedRunProgressController } from "./run/progress-controller.js";
import { createRecoveryMessageActionTurnCapability } from "./run/recovery-message-action-capability.js";
import { resolveInitialEmbeddedRunModel } from "./run/runtime-resolution.js";
import { assertAgentHarnessRunAdmission, backfillSessionKey } from "./run/session-bootstrap.js";
import { prepareEmbeddedRunSession } from "./run/session-bootstrap.js";
import type { EmbeddedAgentRunResult } from "./types.js";
import {
createUsageAccumulator,
@ -134,32 +129,32 @@ export function runEmbeddedAgent(
async function runEmbeddedAgentInternal(
paramsInput: RunEmbeddedAgentInternalParams,
): Promise<EmbeddedAgentRunResult> {
const contextEngineAgentId =
normalizeOptionalString(paramsInput.sessionTarget?.agentId) ??
normalizeOptionalString(paramsInput.agentId);
const paramsBase = applyAgentRunSessionTargetIdentity(paramsInput);
const prepared = await prepareEmbeddedRunSession(paramsInput);
return await withRequiredSessionPlacement(
prepared.runSessionTarget,
{
config: prepared.params.config,
assertCurrent: () => prepared.params.preparedRunAdmission?.assertSourceCurrent(),
signal: prepared.params.abortSignal,
},
() => runEmbeddedAgentForSession(prepared),
);
}
async function runEmbeddedAgentForSession(
prepared: Awaited<ReturnType<typeof prepareEmbeddedRunSession>>,
): Promise<EmbeddedAgentRunResult> {
const {
params: paramsBase,
runSessionTarget,
sessionAdmission,
contextEngineAgentId,
queuedLifecycleGeneration,
} = prepared;
const skillWorkshopProposalMutationBudget = paramsBase.skillWorkshopProposalOnly
? (paramsBase.skillWorkshopProposalMutationBudget ?? { remaining: 1 })
: undefined;
let lifecycleGeneration = paramsBase.lifecycleGeneration!;
const queuedLifecycleGeneration = getAgentEventLifecycleGeneration();
// Resolve sessionKey early so all downstream consumers (hooks, LCM, compaction)
// receive a non-null key even when callers omit it. See #60552.
const effectiveSessionKey = backfillSessionKey({
config: paramsBase.config,
sessionId: paramsBase.sessionId,
sessionKey: paramsBase.sessionKey,
agentId: paramsBase.agentId,
});
const sessionAdmission = await assertAgentHarnessRunAdmission({
...paramsBase,
sessionKey: effectiveSessionKey,
});
const runSessionTarget = await resolveAgentRunSessionTarget({
...paramsBase,
missingSessionKey: "create",
sessionKey: effectiveSessionKey,
});
let params: RunEmbeddedAgentParamsWithSessionFile = withExecutionPhaseDiagnostics({
...paramsBase,
// Establish one detached transcript owner for CLI dispatch and every retry.
@ -168,11 +163,6 @@ async function runEmbeddedAgentInternal(
(paramsBase.sessionPersistence === "detached"
? SessionManager.inMemory(paramsBase.cwd ?? paramsBase.workspaceDir)
: undefined),
agentId: runSessionTarget.agentId,
sessionId: runSessionTarget.sessionId,
sessionKey: runSessionTarget.sessionKey,
sessionTarget: runSessionTarget,
sessionFile: runSessionTarget.sessionKey,
skillWorkshopProposalMutationBudget,
});
const sessionLane = resolveSessionLane(params.sessionKey?.trim() || params.sessionId);

75
src/agents/embedded-agent-runner/run/session-bootstrap.ts
View file

@ -1,6 +1,7 @@
import path from "node:path";
import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
import { sanitizeForLog } from "../../../../packages/terminal-core/src/ansi.js";
import { assertRequiredWorkerSelection } from "../../../config/required-worker-profile.js";
import {
resolveSessionStorePathCore,
SESSION_TOTAL_TOKENS_VERSION,
@ -22,7 +23,10 @@ import {
} from "../../../config/sessions/transcript-write-context.js";
import type { InternalSessionEntry } from "../../../config/sessions/types.js";
import type { ContextEngineSessionTarget } from "../../../context-engine/types.js";
import { emitAgentEventIfCurrent } from "../../../infra/agent-events.js";
import {
emitAgentEventIfCurrent,
getAgentEventLifecycleGeneration,
} from "../../../infra/agent-events.js";
import { getAgentRunContext } from "../../../infra/agent-run-registry.js";
import { formatErrorMessage } from "../../../infra/errors.js";
import { parseAgentSessionKey } from "../../../routing/session-key.js";
@ -35,6 +39,10 @@ import {
resolveSessionKeyForRequestCore,
resolveStoredSessionKeyForSessionId,
} from "../../command/session.js";
import {
applyAgentRunSessionTargetIdentity,
resolveAgentRunSessionTarget,
} from "../../run-session-target.js";
import {
AGENT_RUN_SUPERSEDED_ERROR,
AGENT_RUN_SUPERSEDED_STOP_REASON,
@ -42,6 +50,10 @@ import {
import { redactRunIdentifier } from "../../workspace-run.js";
import { log } from "../logger.js";
import { supersedeEmbeddedAgentRunByRunId } from "../runs.js";
import type {
RunEmbeddedAgentInternalParams,
RunEmbeddedAgentParamsWithSessionFile,
} from "./internal-params.js";
import type { RunEmbeddedAgentParams } from "./params.js";
import { resolveAgentHarnessRunAdmissionError } from "./setup.js";
@ -208,19 +220,19 @@ export async function resetNoRealConversationTokenSnapshot(params: {
}
}
/** Best-effort read-only session-key lookup for callers that only provide sessionId. */
export function backfillSessionKey(params: {
/** Best-effort identity lookup retains the agent that owns an unqualified stored key. */
function backfillSessionIdentity(params: {
config: RunEmbeddedAgentParams["config"];
sessionId: string;
sessionKey?: string;
agentId?: string;
}): string | undefined {
}): Pick<RunEmbeddedAgentInternalParams, "agentId" | "sessionKey"> {
const trimmed = normalizeOptionalString(params.sessionKey);
if (trimmed) {
return trimmed;
return { sessionKey: trimmed };
}
if (!params.config || !params.sessionId) {
return undefined;
return {};
}
try {
const resolved = normalizeOptionalString(params.agentId)
@ -233,15 +245,60 @@ export function backfillSessionKey(params: {
cfg: params.config,
sessionId: params.sessionId,
});
return normalizeOptionalString(resolved.sessionKey);
return {
sessionKey: normalizeOptionalString(resolved.sessionKey),
agentId: normalizeOptionalString(params.agentId) ?? normalizeOptionalString(resolved.agentId),
};
} catch (err) {
log.warn(
`[backfillSessionKey] Failed to resolve sessionKey for sessionId=${redactRunIdentifier(sanitizeForLog(params.sessionId))}: ${formatErrorMessage(err)}`,
`[backfillSessionIdentity] Failed to resolve sessionKey for sessionId=${redactRunIdentifier(sanitizeForLog(params.sessionId))}: ${formatErrorMessage(err)}`,
);
return undefined;
return {};
}
}
/** Prepare canonical session identity without acquiring execution or placement ownership. */
export async function prepareEmbeddedRunSession(paramsInput: RunEmbeddedAgentInternalParams) {
const contextEngineAgentId =
normalizeOptionalString(paramsInput.sessionTarget?.agentId) ??
normalizeOptionalString(paramsInput.agentId);
const queuedLifecycleGeneration = getAgentEventLifecycleGeneration();
const supplied = applyAgentRunSessionTargetIdentity(paramsInput);
// Carry the lookup's owner into every admission; a bare stored key cannot encode it.
const paramsBase = {
...supplied,
...backfillSessionIdentity({
config: supplied.config,
sessionId: supplied.sessionId,
sessionKey: supplied.sessionKey,
agentId: supplied.agentId,
}),
};
const sessionAdmission = await assertAgentHarnessRunAdmission(paramsBase);
assertRequiredWorkerSelection(paramsBase.config ?? {}, {
agentRuntime: paramsBase.agentHarnessId ?? paramsBase.agentHarnessRuntimeOverride,
});
const runSessionTarget = await resolveAgentRunSessionTarget({
...paramsBase,
missingSessionKey: "create",
});
const params: RunEmbeddedAgentParamsWithSessionFile = {
...paramsBase,
agentId: runSessionTarget.agentId,
sessionId: runSessionTarget.sessionId,
sessionKey: runSessionTarget.sessionKey,
sessionTarget: runSessionTarget,
sessionFile: runSessionTarget.sessionKey,
};
return {
params,
runSessionTarget,
sessionAdmission,
contextEngineAgentId,
queuedLifecycleGeneration,
};
}
/** Reserves only a missing row's first writer; no row or claim exists until lazy persistence. */
export async function prepareInitialSessionWriter(params: {
runParams: RunEmbeddedAgentParams;

5
src/agents/harness/policy.ts
View file

@ -30,6 +30,11 @@ export function resolveAgentHarnessPolicy(
// Configured selectors can reuse a normalized lookup without losing their route input.
configured = resolveModelRuntimePolicy(params),
): AgentHarnessPolicy {
// The required worker owns runtime selection, before implicit provider harness routing.
// Explicit session overrides are rejected at session mutation and run admission.
if (params.config?.cloudWorkers?.requiredProfile) {
return { runtime: "openclaw", runtimeSource: "implicit" };
}
const configuredRuntime = normalizeOptionalAgentRuntimeId(configured.policy?.id);
const runtime =
configuredRuntime && configuredRuntime !== "default"

35
src/agents/isolated-completion.test.ts
View file

@ -33,6 +33,41 @@ const { resolveIsolatedCompletionRuntime } = await import("./isolated-completion
beforeEach(resetIsolatedCompletionTestState);
describe("runIsolatedCompletion", () => {
it.each(["cli", "host-v2", "harness-v2", "v1"] as const)(
"blocks required worker policy before isolated %s execution using the admitted config",
async (route) => {
const dispatch = vi.fn(async () => ({
assistant: isolatedAssistant([{ type: "text", text: "must not run" }]),
}));
if (route === "cli") {
mocks.isCliRuntimeAliasForProvider.mockReturnValue(true);
mocks.runCliAgent.mockResolvedValue({ payloads: [{ text: "must not run" }] });
} else {
registerIsolatedHarness({
authBootstrap: route === "harness-v2" ? "harness" : undefined,
...(route === "v1"
? { runIsolatedCompletion: dispatch }
: { runIsolatedCompletionV2: dispatch }),
});
}
mocks.acquireAgentRunPreparedModelRuntime.mockImplementationOnce(async () => {
Object.assign(preparedModelRuntime, {
config: { cloudWorkers: { requiredProfile: "dedicated" } },
});
return { snapshot: preparedModelRuntime, [Symbol.asyncDispose]: releaseRuntimeLease };
});
await expect(runIsolatedCompletion(isolatedRequest())).rejects.toMatchObject({
code: "unsupported",
message: expect.stringContaining("requiredProfile"),
});
expect(mocks.runCliAgent).not.toHaveBeenCalled();
expect(dispatch).not.toHaveBeenCalled();
expect(mocks.prepareSimpleCompletionModel).not.toHaveBeenCalled();
expect(mocks.resolveModelAsync).not.toHaveBeenCalled();
expect(releaseRuntimeLease).toHaveBeenCalledOnce();
},
);
function prepareQuotaProfiles() {
const profileIds = ["openai:first", "openai:backup"];
mocks.ensureAuthProfileStore.mockReturnValue({

13
src/agents/isolated-completion.ts
View file

@ -1,3 +1,7 @@
import { randomUUID } from "node:crypto";
import path from "node:path";
import { withTempWorkspace } from "@openclaw/fs-safe/temp";
import type { ThinkLevel } from "../auto-reply/thinking.js";
/**
* Fresh, prompt-only inference through the selected runtime.
*
@ -6,10 +10,7 @@
* literal empty native tool surface or fail before inference starts, except
* Agents API: its restricted sessions may retain service-owned helpers.
*/
import { randomUUID } from "node:crypto";
import path from "node:path";
import { withTempWorkspace } from "@openclaw/fs-safe/temp";
import type { ThinkLevel } from "../auto-reply/thinking.js";
import { requiredWorkerHelperError } from "../config/required-worker-profile.js";
import type { OpenClawConfig } from "../config/types.openclaw.js";
import { resolvePreferredOpenClawTmpDir } from "../infra/tmp-openclaw-dir.js";
import type { Model } from "../llm/types.js";
@ -361,6 +362,10 @@ async function runIsolatedCompletionOwned(
workspaceDir: lease.snapshot.workspaceDir ?? requestedWorkspaceDir,
};
const { config, agentDir, workspaceDir } = context;
const blocked = requiredWorkerHelperError(config);
if (blocked) {
throw new IsolatedCompletionError("unsupported", blocked.error);
}
const request = { ...input, ...context, assertCurrent };
await ensureSelectedAgentHarnessPlugin({
provider,

92
src/agents/session-placement-admission.ts
View file

@ -1,4 +1,7 @@
import { AsyncLocalStorage } from "node:async_hooks";
import { registerReplyOperationSuccessorBarrier } from "../auto-reply/reply/reply-run-registry.js";
import { getRuntimeConfig } from "../config/config.js";
import { assertRequiredWorkerLocalExecution } from "../config/required-worker-profile.js";
import type { SessionTranscriptRuntimeTarget } from "../config/sessions/session-accessor.js";
import { composeSessionSourceAssertion } from "../config/sessions/session-source-authority.js";
import type { OpenClawConfig } from "../config/types.openclaw.js";
@ -40,7 +43,7 @@ export type LocalTurnPlacementClaim = {
export type SessionPlacementTurnParams = RunEmbeddedAgentInternalParams & { sessionFile: string };
type SessionPlacementSandboxParams = {
export type SessionPlacementSandboxParams = {
agentId: string;
config?: OpenClawConfig;
sessionId: string;
@ -54,6 +57,14 @@ export type PreparedSessionPlacementSandbox = Disposable & {
};
export type SessionPlacementAdmissionProvider = {
withRequiredSession?: <T>(
identity: Omit<LocalTurnPlacementClaim, "runId">,
task: (assertPlacementCurrent: () => void) => Promise<T>,
assertCurrent?: () => void,
signal?: AbortSignal,
preparation?: { waitForReady: false },
) => Promise<T>;
usesWorkerInference?: (identity: Omit<LocalTurnPlacementClaim, "runId">) => boolean;
resolveRuntimeOverride?: (
identity: Omit<LocalTurnPlacementClaim, "runId">,
) => Promise<string | undefined>;
@ -118,6 +129,78 @@ export async function resolveSessionPlacementRuntimeOverride(
return runtime;
}
const requiredPlacementScope = resolveGlobalSingleton(
Symbol.for("openclaw.requiredSessionPlacementScope"),
() =>
new AsyncLocalStorage<{
identity: Omit<LocalTurnPlacementClaim, "runId">;
provider: SessionPlacementAdmissionProvider;
assertPlacementCurrent: () => void;
}>(),
);
/** Nested entry points consume this run's owner-held admission, never a cached permission. */
export async function withRequiredSessionPlacement<T>(
identity: Omit<LocalTurnPlacementClaim, "runId">,
options: { config?: OpenClawConfig; assertCurrent?: () => void; signal?: AbortSignal },
task: () => Promise<T>,
): Promise<T> {
const inherited = requiredPlacementScope.getStore();
if (
inherited &&
inherited.identity.sessionId === identity.sessionId &&
inherited.identity.agentId === identity.agentId &&
inherited.identity.sessionKey === identity.sessionKey
) {
options.signal?.throwIfAborted();
options.assertCurrent?.();
if (state.provider !== inherited.provider) {
throw createAbortError("session placement owner changed during required worker preparation");
}
inherited.assertPlacementCurrent();
return await task();
}
const required =
getRuntimeConfig().cloudWorkers?.requiredProfile ??
options.config?.cloudWorkers?.requiredProfile;
if (!required) {
return await task();
}
const provider = state.provider;
if (!identity.sessionKey?.trim() || !provider?.withRequiredSession) {
throw new Error(
"Required worker execution needs a real session and an available Gateway placement owner; sessionless model helpers are unsupported.",
);
}
const assertCurrent = () => {
options.signal?.throwIfAborted();
options.assertCurrent?.();
if (state.provider !== provider) {
throw createAbortError("session placement owner changed during required worker preparation");
}
};
assertCurrent();
return await provider.withRequiredSession(
identity,
async (assertPlacementCurrent) => {
assertCurrent();
assertPlacementCurrent();
return await requiredPlacementScope.run(
{ identity: { ...identity }, provider, assertPlacementCurrent },
task,
);
},
assertCurrent,
options.signal,
);
}
export function sessionPlacementUsesWorkerInference(
identity: Omit<LocalTurnPlacementClaim, "runId">,
): boolean {
return state.provider?.usesWorkerInference?.(identity) === true;
}
/** Captures the exact placement owner, including standalone absence, before awaited work. */
export function captureSessionPlacementCompactionSuccessorAssertion(): SessionPlacementAdmissionProvider["assertCompactionSuccessorAllowed"] {
const provider = state.provider;
@ -161,6 +244,8 @@ export async function withSessionPlacementTurnAdmission(
// Providers may execute locally or remotely; both must release queue ownership
// only when their actual execution path has acquired its placement claim.
const runAdmittedLocalTurn = async () => {
assertRequiredWorkerLocalExecution(getRuntimeConfig(), "Gateway");
assertRequiredWorkerLocalExecution(params.config ?? {}, "Gateway");
const settle = resolveSessionPlacementForcedTerminalSettlement();
const assertCurrent = resolveSessionPlacementTurnSettlementAssertion();
if (params.replyOperation && settle) {
@ -231,6 +316,10 @@ export async function withLocalSessionPlacementTurnSettlement(
| "isFinalFallbackAttempt"
> = {},
): Promise<EmbeddedAgentRunResult> {
const assertLocalAllowed = () => {
assertRequiredWorkerLocalExecution(getRuntimeConfig(), "Local CLI");
};
assertLocalAllowed();
const provider = state.provider;
const lifecycleGeneration =
options.lifecycleGeneration ?? captureAgentRunLifecycleGeneration(claim.runId);
@ -265,6 +354,7 @@ export async function withLocalSessionPlacementTurnSettlement(
async (taskMarker) => {
assertCurrent();
const runLocal = async () => {
assertLocalAllowed();
// Placement admission can itself await work. A cancelled or replaced
// queue owner must never execute through the captured provider.
assertCurrent();

219
src/agents/session-placement-required-profile.test.ts Normal file
View file

@ -0,0 +1,219 @@
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import type { OpenClawConfig } from "../config/types.openclaw.js";
import { createDeferredCore } from "../shared/deferred.js";
import {
installSessionPlacementAdmissionProvider,
withRequiredSessionPlacement,
withLocalSessionPlacementTurnSettlement,
withSessionPlacementTurnAdmission,
type SessionPlacementAdmissionProvider,
} from "./session-placement-admission.js";
const state = vi.hoisted(() => ({ config: {} as OpenClawConfig }));
vi.mock("../config/config.js", async (importOriginal) => ({
...(await importOriginal<typeof import("../config/config.js")>()),
getRuntimeConfig: () => state.config,
}));
const identity = {
sessionId: "required-session",
sessionKey: "agent:main:required",
agentId: "main",
};
const turn = {
...identity,
runId: "required-run",
sessionFile: identity.sessionKey,
workspaceDir: "/workspace",
prompt: "test",
timeoutMs: 1000,
};
let uninstall: (() => void) | undefined;
const install = (
withRequiredSession?: SessionPlacementAdmissionProvider["withRequiredSession"],
) => {
uninstall = installSessionPlacementAdmissionProvider({
withRequiredSession,
assertCompactionSuccessorAllowed: () => {},
executeLocalTurn: async (_claim, run) => await run(),
executeTurn: async (_claim, _turn, run) => await run(),
});
};
beforeEach(() => {
state.config = { cloudWorkers: { requiredProfile: "remote" } };
});
afterEach(() => {
uninstall?.();
uninstall = undefined;
state.config = {};
});
describe("required worker run admission", () => {
it("rejects unavailable ownership and sessionless helpers without creating a session", async () => {
await expect(withRequiredSessionPlacement(identity, {}, async () => undefined)).rejects.toThrow(
"available Gateway placement owner",
);
let preparations = 0;
const prepare: NonNullable<SessionPlacementAdmissionProvider["withRequiredSession"]> = async (
_identity,
task,
) => {
preparations += 1;
return await task(() => {});
};
install(prepare);
await expect(
withRequiredSessionPlacement(
{ sessionId: "helper", agentId: "main" },
{},
async () => undefined,
),
).rejects.toThrow("sessionless model helpers");
expect(preparations).toBe(0);
});
it.each(["no provider", "local provider"])("never runs a local turn with %s", async (mode) => {
if (mode === "local provider") {
install();
}
const run = vi.fn(async () => ({ meta: { durationMs: 1 } }));
await expect(withSessionPlacementTurnAdmission(turn, turn, run)).rejects.toThrow(
"Gateway execution is disabled",
);
await expect(withLocalSessionPlacementTurnSettlement(turn, run)).rejects.toThrow(
"Local CLI execution is disabled",
);
expect(run).not.toHaveBeenCalled();
});
it("rechecks the exact provider after awaited preparation", async () => {
const started = createDeferredCore();
const finish = createDeferredCore();
install(async (_identity, task, assertCurrent) => {
assertCurrent?.();
started.resolve();
await finish.promise;
return await task(() => assertCurrent?.());
});
const pending = withRequiredSessionPlacement(identity, {}, async () => undefined);
const outcome = pending.catch((error: unknown) => error);
await started.promise;
uninstall?.();
finish.resolve();
expect(await outcome).toMatchObject({
message: "session placement owner changed during required worker preparation",
});
});
it("rechecks the exact caller after awaited preparation", async () => {
const started = createDeferredCore();
const finish = createDeferredCore();
let callerCurrent = true;
const assertCaller = vi.fn(() => {
if (!callerCurrent) {
throw new Error("caller changed during required worker preparation");
}
});
install(async (_identity, task, assertCurrent) => {
assertCurrent?.();
started.resolve();
await finish.promise;
return await task(() => assertCurrent?.());
});
const pending = withRequiredSessionPlacement(
identity,
{ assertCurrent: assertCaller },
async () => undefined,
);
const outcome = pending.catch((error: unknown) => error);
await started.promise;
callerCurrent = false;
finish.resolve();
expect(await outcome).toMatchObject({
message: "caller changed during required worker preparation",
});
});
it("rejects a local CLI task when mandatory placement is enabled during admission", async () => {
state.config = {};
const entered = createDeferredCore();
const resume = createDeferredCore();
const run = vi.fn(async () => ({ meta: { durationMs: 1 } }));
uninstall = installSessionPlacementAdmissionProvider({
assertCompactionSuccessorAllowed: () => {},
executeLocalTurn: async (_claim, runLocal) => {
entered.resolve();
await resume.promise;
return await runLocal();
},
executeTurn: async (_claim, _params, runLocal) => await runLocal(),
});
const pending = withLocalSessionPlacementTurnSettlement(turn, run);
const outcome = pending.catch((error: unknown) => error);
try {
await entered.promise;
state.config = { cloudWorkers: { requiredProfile: "remote" } };
} finally {
resume.resolve();
}
expect(await outcome).toMatchObject({
message: expect.stringContaining("Local CLI execution is disabled"),
});
expect(run).not.toHaveBeenCalled();
});
it("retains one owner admission across nested execution and fences it after release", async () => {
let released = false;
let current = true;
let preparations = 0;
const prepare: NonNullable<SessionPlacementAdmissionProvider["withRequiredSession"]> = async (
_identity,
task,
) => {
preparations += 1;
try {
return await task(() => {
if (released || !current) {
throw new Error("required placement authority ended");
}
});
} finally {
released = true;
}
};
install(prepare);
let retainedUse: (() => Promise<void>) | undefined;
await withRequiredSessionPlacement(identity, {}, async () => {
await withRequiredSessionPlacement(identity, {}, async () => undefined);
expect(released).toBe(false);
current = false;
await expect(
withRequiredSessionPlacement(identity, {}, async () => undefined),
).rejects.toThrow("required placement authority ended");
current = true;
// Async work can retain the scope after the awaited turn ends; it cannot retain authority.
const deferred = createDeferredCore();
retainedUse = () => {
deferred.resolve();
return outcome;
};
const outcome = (async () => {
await deferred.promise;
await withRequiredSessionPlacement(identity, {}, async () => undefined);
})();
});
expect(preparations).toBe(1);
expect(released).toBe(true);
await expect(retainedUse!()).rejects.toThrow("required placement authority ended");
});
it("preserves standalone local execution when the requirement is absent", async () => {
state.config = {};
await expect(
withRequiredSessionPlacement({ sessionId: "helper" }, {}, async () => undefined),
).resolves.toBeUndefined();
const result = { meta: { durationMs: 1 } };
await expect(withSessionPlacementTurnAdmission(turn, turn, async () => result)).resolves.toBe(
result,
);
});
});

59
src/agents/simple-completion-runtime.generation.test.ts
View file

@ -3,6 +3,7 @@ import { createApiRegistry } from "@openclaw/ai";
import { beforeEach, expect, it, vi } from "vitest";
import type { Model } from "../llm/types.js";
import { createPluginMetadataSnapshotFixture } from "../plugins/plugin-metadata.test-support.js";
import { resolveSecretSentinel } from "../secrets/sentinel.js";
import { AsyncWorkScope } from "../shared/async-work-scope.js";
import { createDeferredCore } from "../shared/deferred.js";
import type { PreparedModelRuntimeSnapshot } from "./prepared-model-runtime.js";
@ -316,3 +317,61 @@ it.each(["/", "entry"])(
expect(release).toHaveBeenCalledOnce();
},
);
it.each(["current", "credential lookup", "provider auth exchange"] as const)(
"keeps required-worker authority live through %s",
async (revokedDuring) => {
const setRuntimeApiKey = vi.spyOn(
preparedModelRuntime.createStores().authStorage,
"setRuntimeApiKey",
);
let current = true;
const revoked = new Error("worker claim revoked");
const required = { cloudWorkers: { requiredProfile: "required" } };
preparedModelRuntime = { ...preparedModelRuntime, config: required };
mocks.getApiKeyForModel.mockImplementation(async () => {
await Promise.resolve();
if (revokedDuring === "credential lookup") {
current = false;
}
return { apiKey: "worker-source-token", source: "worker fixture", mode: "token" };
});
mocks.prepareProviderRuntimeAuth.mockImplementation(async () => {
await Promise.resolve();
if (revokedDuring === "provider auth exchange") {
current = false;
}
return { apiKey: "worker-runtime-token" };
});
const preparing = prepareSimpleCompletionModel({
preparedModelRuntime,
cfg: required,
provider: "ollama",
modelId: "fixture-model",
modelResolver: createOllamaModelResolver(),
workerInferenceAuthority: {
assertCurrent: () => {
if (!current) {
throw revoked;
}
},
},
});
if (revokedDuring === "current") {
expect(await preparing).not.toHaveProperty("error");
expect(mocks.prepareProviderRuntimeAuth).toHaveBeenCalledOnce();
expect(setRuntimeApiKey).toHaveBeenCalledOnce();
const [provider, credential] = setRuntimeApiKey.mock.calls[0] as [string, string];
expect(provider).toBe("ollama");
expect(resolveSecretSentinel(credential)).toBe("worker-runtime-token");
} else {
await expect(preparing).rejects.toBe(revoked);
if (revokedDuring === "credential lookup") {
expect(mocks.prepareProviderRuntimeAuth).not.toHaveBeenCalled();
} else {
expect(mocks.prepareProviderRuntimeAuth).toHaveBeenCalledOnce();
}
expect(setRuntimeApiKey).not.toHaveBeenCalled();
}
},
);

20
src/agents/simple-completion-runtime.test.ts
View file

@ -223,6 +223,26 @@ function createOpenAIRouteModelResolver(params: {
}
describe("prepareSimpleCompletionModel", () => {
it.each([undefined, {}])(
"blocks helper inference under the admitted required policy with caller config %j",
async (cfg) => {
preparedModelRuntime = {
...preparedModelRuntime,
config: { cloudWorkers: { requiredProfile: "required" } },
};
const result = await prepareSimpleCompletionModel({
preparedModelRuntime,
cfg,
provider: "anthropic",
modelId: "claude-opus-4-6",
modelResolver: hoisted.resolveModelAsyncMock as SimpleCompletionModelResolver,
});
expect(result).toMatchObject({ error: expect.stringContaining("Sessionless model helpers") });
expect(hoisted.resolveModelAsyncMock).not.toHaveBeenCalled();
expect(hoisted.getApiKeyForModelMock).not.toHaveBeenCalled();
},
);
it("captures the exact locked auth owner used by a bound completion", async () => {
const credential = {
type: "api_key" as const,

174
src/agents/simple-completion-runtime.ts
View file

@ -1,4 +1,5 @@
import { prepareModelForSimpleCompletion } from "@openclaw/ai/transports";
import { requiredWorkerHelperError } from "../config/required-worker-profile.js";
import type { OpenClawConfig } from "../config/types.openclaw.js";
import { formatErrorMessage } from "../infra/errors.js";
import { bindModelLlmRuntime } from "../llm/model-runtime-binding.js";
@ -13,16 +14,10 @@ import { prepareProviderRuntimeAuth } from "../plugins/provider-runtime.runtime.
import { withPluginRuntimeGenerationScope } from "../plugins/runtime/generation-scope.js";
import { runWithAsyncWorkResources } from "../shared/async-work-resources.js";
import { createDeferredCore } from "../shared/deferred.js";
import {
resolveAgentDir,
resolveNativeModelPrimary,
resolveAgentWorkspaceDir,
resolveDefaultAgentId,
} from "./agent-scope.js";
import { resolveAgentDir, resolveAgentWorkspaceDir, resolveDefaultAgentId } from "./agent-scope.js";
import { ensureAuthProfileStore } from "./auth-profiles/store-runtime.js";
import type { AuthProfileStore } from "./auth-profiles/types.js";
import { reconcileAuthProfileQuotaBlocks } from "./auth-profiles/usage.js";
import { DEFAULT_PROVIDER } from "./defaults.js";
import {
fingerprintAuthProfileCredential,
fingerprintResolvedProviderAuth,
@ -37,13 +32,8 @@ import {
getApiKeyForModelCore,
type ResolvedProviderAuth,
} from "./model-auth.js";
import { splitTrailingAuthProfile } from "./model-ref-profile.js";
import { resolveModelRouteIntent } from "./model-runtime-policy.js";
import {
buildModelAliasIndex,
resolveDefaultModelForAgent,
resolveModelRefFromString,
} from "./model-selection.js";
import { resolveDefaultModelForAgent } from "./model-selection.js";
import { resolveOpenAIModelRoutes } from "./openai-model-routes.js";
import {
acquireAgentRunPreparedModelRuntime,
@ -64,13 +54,14 @@ import {
type PreparedSimpleCompletionResolverContext,
type SimpleCompletionModelResolver,
} from "./simple-completion-scope.js";
import { resolveSimpleCompletionSelectionRequest } from "./simple-completion-selection.js";
import type {
AgentSimpleCompletionSelection,
PreparedSimpleCompletionModel,
PreparedSimpleCompletionModelForAgent,
PrepareSimpleCompletionModelForAgentParams,
} from "./simple-completion.types.js";
import { resolveUtilityModelRefForAgent } from "./utility-model.js";
export { resolveSimpleCompletionSelectionForAgent } from "./simple-completion-selection.js";
type AllowedMissingApiKeyMode = ResolvedProviderAuth["mode"];
@ -81,95 +72,6 @@ type PreparedStreamCompletionModel =
>;
})
| Extract<PreparedSimpleCompletionModel, { error: string }>;
type SimpleCompletionSelectionParams = {
cfg: OpenClawConfig;
agentId: string;
agentDir?: string;
modelRef?: string;
useUtilityModel?: boolean | "required";
manifestPlugins?:
| PluginMetadataSnapshot["plugins"]
| Pick<PluginMetadataSnapshot, "plugins" | "owners">;
};
type SimpleCompletionSelectionRequest = {
selection: AgentSimpleCompletionSelection;
shorthandModelId?: string;
};
function resolveSimpleCompletionSelectionRequest(
params: SimpleCompletionSelectionParams,
): SimpleCompletionSelectionRequest | null {
const fallbackRef = resolveDefaultModelForAgent({
cfg: params.cfg,
agentId: params.agentId,
manifestPlugins: params.manifestPlugins,
});
// Utility routing derives a provider-declared small model when unset and
// optional routing treats an empty utilityModel as "use the primary" (disabled).
const modelRef =
params.modelRef?.trim() ||
(params.useUtilityModel
? resolveUtilityModelRefForAgent({
cfg: params.cfg,
agentId: params.agentId,
primaryProvider: fallbackRef.provider,
...(params.manifestPlugins
? {
metadataSnapshot:
"plugins" in params.manifestPlugins
? params.manifestPlugins
: { plugins: params.manifestPlugins },
}
: {}),
})
: undefined) ||
(params.useUtilityModel === "required"
? undefined
: resolveNativeModelPrimary(params.cfg, params.agentId));
const split = modelRef ? splitTrailingAuthProfile(modelRef) : null;
const aliasIndex = buildModelAliasIndex({
cfg: params.cfg,
agentId: params.agentId,
defaultProvider: fallbackRef.provider || DEFAULT_PROVIDER,
manifestPlugins: params.manifestPlugins,
});
const resolved = split
? resolveModelRefFromString({
cfg: params.cfg,
agentId: params.agentId,
raw: split.model,
defaultProvider: fallbackRef.provider || DEFAULT_PROVIDER,
aliasIndex,
manifestPlugins: params.manifestPlugins,
})
: null;
if (params.useUtilityModel === "required" && !resolved) {
return null;
}
const provider = resolved?.ref.provider ?? fallbackRef.provider;
const modelId = resolved?.ref.model ?? fallbackRef.model;
if (!provider || !modelId) {
return null;
}
return {
selection: {
provider,
modelId,
profileId: split?.profile || undefined,
agentDir: params.agentDir?.trim() || resolveAgentDir(params.cfg, params.agentId),
},
...(split && !split.model.includes("/") ? { shorthandModelId: split.model } : {}),
};
}
export function resolveSimpleCompletionSelectionForAgent(
params: SimpleCompletionSelectionParams,
): AgentSimpleCompletionSelection | null {
return resolveSimpleCompletionSelectionRequest(params)?.selection ?? null;
}
export type PrepareSimpleCompletionModelParams = {
cfg: OpenClawConfig | undefined;
agentId?: string;
@ -191,6 +93,8 @@ export type PrepareSimpleCompletionModelParams = {
agentRuntimeId?: string;
/** Internal stream callers own provider transport construction and embedded policy. */
transport?: "simple-completion" | "provider-stream";
/** Internal worker RPC owner; sessionless helper callers cannot supply this authority. */
workerInferenceAuthority?: { assertCurrent: () => void };
};
/** Prepares a model within the exact generation already held by its caller. */
@ -202,6 +106,15 @@ export async function prepareSimpleCompletionModel(
): Promise<PreparedStreamCompletionModel> {
params.signal?.throwIfAborted();
const config = params.cfg ?? {};
const blocked =
requiredWorkerHelperError(
params.preparedModelRuntime.config,
Boolean(params.workerInferenceAuthority),
) ?? requiredWorkerHelperError(config, Boolean(params.workerInferenceAuthority));
if (blocked) {
return blocked;
}
params.workerInferenceAuthority?.assertCurrent();
const preparedModelRuntime = params.preparedModelRuntime;
const context = createPreparedSimpleCompletionResolverContext({
preparedModelRuntime,
@ -216,9 +129,13 @@ export async function prepareSimpleCompletionModel(
prepareSimpleCompletionModelCore(
{ ...params, agentDir: preparedModelRuntime.agentDir },
context,
assertCurrent,
() => {
assertCurrent?.();
params.workerInferenceAuthority?.assertCurrent();
},
),
);
params.workerInferenceAuthority?.assertCurrent();
params.signal?.throwIfAborted();
return prepared;
}
@ -228,6 +145,15 @@ async function prepareSimpleCompletionModelCore(
context: PreparedSimpleCompletionResolverContext,
assertCurrent?: () => void,
): Promise<PreparedStreamCompletionModel> {
const blocked =
requiredWorkerHelperError(
context.preparedModelRuntime.config,
Boolean(params.workerInferenceAuthority),
) ?? requiredWorkerHelperError(params.cfg ?? {}, Boolean(params.workerInferenceAuthority));
if (blocked) {
return blocked;
}
params.workerInferenceAuthority?.assertCurrent();
const { modelResolver, workspaceDir } = context;
const resolved = await modelResolver(
params.provider,
@ -405,6 +331,8 @@ async function prepareSimpleCompletionModelCore(
error: `Auth lookup failed for provider "${initialModel.provider}": ${formatErrorMessage(err)}`,
};
}
assertCurrent?.();
params.signal?.throwIfAborted();
const rawApiKey = auth.apiKey?.trim();
if (!rawApiKey && !params.allowMissingApiKeyModes?.includes(auth.mode)) {
return {
@ -415,25 +343,29 @@ async function prepareSimpleCompletionModelCore(
let authValue = rawApiKey;
if (rawApiKey) {
const preparedAuth = protectPreparedProviderRuntimeAuth({
const runtimeAuth = await prepareProviderRuntimeAuth({
provider: resolvedModel.provider,
preparedAuth: await prepareProviderRuntimeAuth({
provider: resolvedModel.provider,
config: params.cfg,
workspaceDir,
env: process.env,
assertCurrent,
context: {
config: params.cfg,
workspaceDir,
env: process.env,
context: {
config: params.cfg,
workspaceDir,
env: process.env,
provider: resolvedModel.provider,
modelId: resolvedModel.id,
model: resolvedModel,
apiKey: rawApiKey,
authMode: auth.mode,
profileId: auth.profileId,
},
}),
provider: resolvedModel.provider,
modelId: resolvedModel.id,
model: resolvedModel,
apiKey: rawApiKey,
authMode: auth.mode,
profileId: auth.profileId,
},
});
assertCurrent?.();
params.signal?.throwIfAborted();
const preparedAuth = protectPreparedProviderRuntimeAuth({
provider: resolvedModel.provider,
preparedAuth: runtimeAuth,
});
authValue = preparedAuth?.apiKey?.trim() || rawApiKey;
resolved.authStorage.setRuntimeApiKey(resolvedModel.provider, authValue);
@ -535,6 +467,10 @@ export async function acquireSimpleCompletionModelWithSelection(
shorthandModelId?: string;
} | null,
): Promise<AcquiredSimpleCompletionModelForAgent> {
const blocked = requiredWorkerHelperError(params.cfg);
if (blocked) {
return blocked;
}
const agentId = params.agentId ?? resolveDefaultAgentId(params.cfg);
const agentDir = params.agentDir?.trim() || resolveAgentDir(params.cfg, agentId);
const tentativeRequest = resolveRequest();

100
src/agents/simple-completion-selection.ts Normal file
View file

@ -0,0 +1,100 @@
import type { OpenClawConfig } from "../config/types.openclaw.js";
import type { PluginMetadataSnapshot } from "../plugins/plugin-metadata-snapshot.types.js";
import { resolveAgentDir, resolveNativeModelPrimary } from "./agent-scope.js";
import { DEFAULT_PROVIDER } from "./defaults.js";
import { splitTrailingAuthProfile } from "./model-ref-profile.js";
import {
buildModelAliasIndex,
resolveDefaultModelForAgent,
resolveModelRefFromString,
} from "./model-selection.js";
import type { AgentSimpleCompletionSelection } from "./simple-completion.types.js";
import { resolveUtilityModelRefForAgent } from "./utility-model.js";
export type SimpleCompletionSelectionParams = {
cfg: OpenClawConfig;
agentId: string;
agentDir?: string;
modelRef?: string;
useUtilityModel?: boolean | "required";
manifestPlugins?:
| PluginMetadataSnapshot["plugins"]
| Pick<PluginMetadataSnapshot, "plugins" | "owners">;
};
type SimpleCompletionSelectionRequest = {
selection: AgentSimpleCompletionSelection;
shorthandModelId?: string;
};
export function resolveSimpleCompletionSelectionRequest(
params: SimpleCompletionSelectionParams,
): SimpleCompletionSelectionRequest | null {
const fallbackRef = resolveDefaultModelForAgent({
cfg: params.cfg,
agentId: params.agentId,
manifestPlugins: params.manifestPlugins,
});
// Utility routing derives a provider-declared small model when unset and
// optional routing treats an empty utilityModel as "use the primary" (disabled).
const modelRef =
params.modelRef?.trim() ||
(params.useUtilityModel
? resolveUtilityModelRefForAgent({
cfg: params.cfg,
agentId: params.agentId,
primaryProvider: fallbackRef.provider,
...(params.manifestPlugins
? {
metadataSnapshot:
"plugins" in params.manifestPlugins
? params.manifestPlugins
: { plugins: params.manifestPlugins },
}
: {}),
})
: undefined) ||
(params.useUtilityModel === "required"
? undefined
: resolveNativeModelPrimary(params.cfg, params.agentId));
const split = modelRef ? splitTrailingAuthProfile(modelRef) : null;
const aliasIndex = buildModelAliasIndex({
cfg: params.cfg,
agentId: params.agentId,
defaultProvider: fallbackRef.provider || DEFAULT_PROVIDER,
manifestPlugins: params.manifestPlugins,
});
const resolved = split
? resolveModelRefFromString({
cfg: params.cfg,
agentId: params.agentId,
raw: split.model,
defaultProvider: fallbackRef.provider || DEFAULT_PROVIDER,
aliasIndex,
manifestPlugins: params.manifestPlugins,
})
: null;
if (params.useUtilityModel === "required" && !resolved) {
return null;
}
const provider = resolved?.ref.provider ?? fallbackRef.provider;
const modelId = resolved?.ref.model ?? fallbackRef.model;
if (!provider || !modelId) {
return null;
}
return {
selection: {
provider,
modelId,
profileId: split?.profile || undefined,
agentDir: params.agentDir?.trim() || resolveAgentDir(params.cfg, params.agentId),
},
...(split && !split.model.includes("/") ? { shorthandModelId: split.model } : {}),
};
}
export function resolveSimpleCompletionSelectionForAgent(
params: SimpleCompletionSelectionParams,
): AgentSimpleCompletionSelection | null {
return resolveSimpleCompletionSelectionRequest(params)?.selection ?? null;
}

65
src/config/required-worker-profile.ts Normal file
View file

@ -0,0 +1,65 @@
import type { OpenClawConfig } from "./types.openclaw.js";
class RequiredWorkerProfileError extends Error {
readonly code = "invalid_profile";
}
/** Configuration is intent; only the placement owner can attest an admitted destination. */
export function assertRequiredWorkerSelection(
config: Pick<OpenClawConfig, "cloudWorkers">,
selection: {
profileId?: string;
deviceId?: string;
autoDevice?: boolean;
machineClass?: string;
os?: string;
agentRuntime?: string | null;
execNode?: string | null;
catalogId?: string;
},
): void {
const required = config.cloudWorkers?.requiredProfile;
if (!required) {
return;
}
if (
(selection.profileId !== undefined && selection.profileId !== required) ||
selection.catalogId !== undefined ||
selection.deviceId !== undefined ||
selection.autoDevice === true ||
selection.machineClass !== undefined ||
selection.os !== undefined ||
(selection.execNode !== undefined && selection.execNode !== null) ||
(selection.agentRuntime !== undefined &&
selection.agentRuntime !== null &&
selection.agentRuntime !== "openclaw" &&
selection.agentRuntime !== "auto" &&
selection.agentRuntime !== "default")
) {
throw new RequiredWorkerProfileError(
`Gateway policy requires worker profile "${required}" with the OpenClaw runtime; session execution overrides are not allowed.`,
);
}
}
/** Sessionless helpers cannot acquire provider auth under a mandatory placement policy. */
export function requiredWorkerHelperError(
config: Pick<OpenClawConfig, "cloudWorkers">,
workerAdmitted = false,
): { error: string } | undefined {
return config.cloudWorkers?.requiredProfile && !workerAdmitted
? {
error:
"Sessionless model helpers are unsupported when cloudWorkers.requiredProfile is configured; run the task in a worker-backed session.",
}
: undefined;
}
export function assertRequiredWorkerLocalExecution(
config: Pick<OpenClawConfig, "cloudWorkers">,
origin: "Gateway" | "Local CLI" | "Local" = "Local",
): void {
if (config.cloudWorkers?.requiredProfile) {
throw new Error(`${origin} execution is disabled by the required worker profile policy.`);
}
}

15
src/config/zod-schema.cloud-workers.test.ts
View file

@ -6,6 +6,21 @@ function cloudProfile(profile: Record<string, unknown>) {
}
describe("OpenClawSchema cloudWorkers config", () => {
it("retains a required profile before enrollment without blocking startup", () => {
expect(
OpenClawSchema.parse({ cloudWorkers: { requiredProfile: "not-enrolled-yet" } }).cloudWorkers,
).toEqual({
requiredProfile: "not-enrolled-yet",
});
});
it.each(["", " ", " worker", "worker ", null, true, 3])(
"rejects invalid required profile %j",
(requiredProfile) => {
expect(OpenClawSchema.safeParse({ cloudWorkers: { requiredProfile } }).success).toBe(false);
},
);
it("accepts normalized per-project default profiles", () => {
const projectProfiles = { "github.com/acme/app": "development" };
expect(OpenClawSchema.parse({ cloudWorkers: { projectProfiles } }).cloudWorkers).toStrictEqual({

4
src/config/zod-schema.cloud-workers.ts
View file

@ -70,6 +70,10 @@ const CloudWorkerProjectProfileSchema = CloudWorkerProfileIdSchema.register(conf
export const CloudWorkersConfigSchema = z
.strictObject({
requiredProfile: CloudWorkerProfileIdSchema.optional().register(configUiMetadata, {
label: "Required Worker Profile",
help: "Require every session to execute on this worker profile. Sessions are placed automatically and cannot fall back to Gateway execution. Unavailable workers block turns, not Gateway startup.",
}),
desktop: z.boolean().optional().register(configUiMetadata, {
label: "Cloud Worker Desktop (Labs)",
help: "Enables the experimental worker.desktop.observe surface and Control UI Desktop panel for desktop-capable cloud worker environments.",

14
src/gateway/agent-turn/agent-request-routing.session-id.test.ts
View file

@ -1,6 +1,6 @@
import { expect, it, vi } from "vitest";
import * as commandSession from "../../agents/command/session.js";
import { backfillSessionKey } from "../../agents/embedded-agent-runner/run/session-bootstrap.js";
import { prepareEmbeddedRunSession } from "../../agents/embedded-agent-runner/run/session-bootstrap.js";
import {
clearRuntimeConfigSnapshot,
setRuntimeConfigSnapshot,
@ -108,7 +108,17 @@ it.each([
expect(respond).not.toHaveBeenCalled();
expect(routing).toMatchObject({ agentId: "ops", requestedSessionKey: sessionKey });
} else {
expect(backfillSessionKey({ config: cfg, sessionId: "ops-session" })).toBe(sessionKey);
const prepared = await prepareEmbeddedRunSession({
config: cfg,
sessionId: "ops-session",
sessionFile: sessionKey,
workspaceDir: state.workspaceDir,
prompt: "resume",
runId: "lookup",
timeoutMs: 1000,
});
expect(prepared.params.sessionKey).toBe(sessionKey);
expect(prepared.params.agentId).toBe("ops");
}
// The legacy listing still reads on the host; only the added inspection is removed.
expect(opens).toContain("listing");

346
src/gateway/worker-environments/inference-authority.e2e.test.ts Normal file
View file

@ -0,0 +1,346 @@
import fs from "node:fs";
import http from "node:http";
import { expect, it, vi } from "vitest";
import {
awaitGateBeforeSettlement,
createDeferred,
withinTest,
} from "../../../test/helpers/promise.js";
import { runQaGatewayFixture } from "../../../test/helpers/qa-gateway-cleanup.js";
import { acquireAgentRunPreparedModelRuntime } from "../../agents/prepared-model-runtime.js";
import { resetPreparedModelRuntimeSnapshotsForTest } from "../../agents/prepared-model-runtime.test-support.js";
import { AuthStorage } from "../../agents/sessions/auth-storage.js";
import { upsertSessionEntryCore } from "../../config/sessions/session-accessor.js";
import type { OpenClawConfig } from "../../config/types.openclaw.js";
import {
claimAgentRunDelegatedAuthority,
registerAgentRunContext,
releaseAgentRunDelegatedAuthority,
validateAgentRunDelegatedAuthority,
resetAgentRunRegistryForTest,
} from "../../infra/agent-run-registry.js";
import { clearPluginMetadataLifecycleCaches } from "../../plugins/plugin-metadata-lifecycle.js";
import { clearActivePluginRegistry } from "../../plugins/runtime.js";
import { createColdPluginFixture } from "../../plugins/test-helpers/cold-plugin-fixtures.js";
import { openOpenClawStateDatabase } from "../../state/openclaw-state-db.js";
import { closeStateDatabaseForTest } from "../../test-utils/database-cleanup.js";
import { withEnvAsync } from "../../test-utils/env.js";
import { withOpenClawTestState } from "../../test-utils/openclaw-test-state.js";
import { reserveTestPortListener } from "../../test-utils/port-claims.js";
import { executeWorkerInference } from "./inference-runtime.js";
import { createWorkerSessionPlacementStore } from "./placement-store.js";
import { advancePlacementFixtureToActive } from "./placement-test-fixtures.js";
import { bindWorkerTurnOwner } from "./placement-turn-claim-events.js";
// This release-tier composition protects the authority-to-provider boundary that
// isolated auth-hook tests cannot prove. No runtime, credentials, or claim owner is mocked.
it("fences real admitted worker credential exchange and inference I/O", async ({ signal }) => {
await withOpenClawTestState({ label: "worker-auth-boundary" }, async (state) => {
const provider = "authority-proof";
const key = "synthetic-admission-key";
const exchanged = "synthetic-exchanged-key";
const effects: { path: string; authorized: boolean }[] = [];
let exchangeArrived = createDeferred();
let finishExchange = createDeferred();
let holdExchange = false;
const handleProviderRequest = async (
request: http.IncomingMessage,
response: http.ServerResponse,
) => {
effects.push({
path: request.url ?? "",
authorized:
request.headers.authorization === `Bearer ${request.url === "/auth" ? key : exchanged}`,
});
if (request.url === "/auth") {
exchangeArrived.resolve();
if (holdExchange) {
await finishExchange.promise;
}
response.writeHead(200, { "content-type": "application/json" });
response.end(JSON.stringify({ apiKey: exchanged }));
return;
}
response.writeHead(200, { "content-type": "text/event-stream" });
const chunk = {
id: "synthetic",
object: "chat.completion.chunk",
created: 1,
model: "model",
};
response.end(
[
`data: ${JSON.stringify({ ...chunk, choices: [{ index: 0, delta: { role: "assistant", content: "allowed worker response" }, finish_reason: null }] })}\n\n`,
`data: ${JSON.stringify({ ...chunk, choices: [{ index: 0, delta: {}, finish_reason: "stop" }], usage: { prompt_tokens: 1, completion_tokens: 1, total_tokens: 2 } })}\n\n`,
"data: [DONE]\n\n",
].join(""),
);
};
const server = http.createServer((request, response) => {
void handleProviderRequest(request, response).catch((error: unknown) => {
response.destroy(error instanceof Error ? error : new Error("Synthetic provider failed"));
});
});
const reservation = await reserveTestPortListener({
offsets: [0],
createListener: () => server,
});
await runQaGatewayFixture(
async () => {
const address = server.address();
if (!address || typeof address === "string") {
throw new Error("Missing loopback provider address");
}
const endpoint = `http://127.0.0.1:${address.port}`;
const pluginDir = state.path("provider");
fs.mkdirSync(pluginDir);
const plugin = createColdPluginFixture({
rootDir: pluginDir,
pluginId: provider,
providerId: provider,
manifest: {
providers: [provider],
channels: [],
channelConfigs: {},
providerAuthChoices: [],
},
});
fs.writeFileSync(
plugin.runtimeSource,
`module.exports = { id: ${JSON.stringify(provider)}, register(api) {
api.registerProvider({ id: ${JSON.stringify(provider)}, label: "Synthetic boundary provider", auth: [],
async prepareRuntimeAuth(context) {
const response = await fetch(${JSON.stringify(`${endpoint}/auth`)}, { method: "POST", headers: { authorization: "Bearer " + context.apiKey } });
return await response.json();
}
});
} };`,
);
const config: OpenClawConfig = {
cloudWorkers: { requiredProfile: "development" },
agents: {
defaults: {
workspace: state.workspaceDir,
model: `${provider}/model`,
models: { [`${provider}/model`]: {} },
},
entries: { main: {} },
},
models: {
mode: "replace",
providers: {
[provider]: {
api: "openai-completions",
apiKey: key,
baseUrl: `${endpoint}/v1`,
models: [
{
id: "model",
name: "Boundary model",
reasoning: false,
input: ["text"],
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
contextWindow: 8192,
maxTokens: 128,
},
],
},
},
},
plugins: {
allow: [provider],
load: { paths: [pluginDir] },
entries: { [provider]: { enabled: true } },
slots: { memory: "none" },
},
};
await withEnvAsync({ OPENCLAW_DISABLE_BUNDLED_PLUGINS: "1" }, async () => {
const database = openOpenClawStateDatabase();
const placements = createWorkerSessionPlacementStore({ database });
await resetPreparedModelRuntimeSnapshotsForTest();
clearPluginMetadataLifecycleCaches();
try {
await using lease = await acquireAgentRunPreparedModelRuntime({
config,
agentId: "main",
agentDir: state.agentDir(),
workspaceDir: state.workspaceDir,
});
for (const mode of [
"allowed",
"wrong-session",
"revoked-claim",
"revoked-run",
"revoked-exchange",
] as const) {
effects.length = 0;
exchangeArrived = createDeferred();
finishExchange = createDeferred();
holdExchange = mode === "revoked-exchange";
const session = {
sessionId: `session-${mode}`,
agentId: "main",
sessionKey: `agent:main:${mode}`,
};
const target = {
...session,
storePath: state.statePath("agents", "main", "sessions", "sessions.sqlite"),
};
await upsertSessionEntryCore(target, { sessionId: session.sessionId, updatedAt: 1 });
const active = await advancePlacementFixtureToActive(placements, database, {
...session,
executionMode: "worker-turn",
});
const claim = await placements.claimTurn({
...session,
claimId: `claim-${mode}`,
runId: `run-${mode}`,
owner: {
kind: "worker",
environmentId: active.environmentId,
ownerEpoch: active.activeOwnerEpoch,
},
});
const claimAuthority = await placements.prepareTurnClaimAuthority(claim);
const instance = { runId: claim.runId, instanceId: `instance-${mode}` };
registerAgentRunContext(claim.runId, session);
const runAuthority = claimAgentRunDelegatedAuthority(instance, () => {
if (!claimAuthority.isCurrent()) {
throw new Error("Persisted worker claim revoked");
}
});
await bindWorkerTurnOwner(
placements,
claim,
undefined,
instance,
target,
() => {
if (!validateAgentRunDelegatedAuthority(runAuthority)) {
throw new Error("Admitted run revoked");
}
},
undefined,
undefined,
undefined,
{ boundaryCount: 0 },
);
if (mode === "revoked-claim") {
await placements.releaseTurn(claim);
} else if (mode === "revoked-run") {
releaseAgentRunDelegatedAuthority(runAuthority);
}
const installed = vi.spyOn(AuthStorage.prototype, "setRuntimeApiKey");
try {
const pending = executeWorkerInference({
identity: {
environmentId: active.environmentId,
credentialHash: "synthetic-worker-hash",
bundleHash: "synthetic-bundle-hash",
sessionId: session.sessionId,
runId: claim.runId,
turnClaim: claim,
ownerEpoch: active.activeOwnerEpoch,
rpcSetVersion: 1,
protocolFeatures: ["worker-inference-v1"],
credentialExpiresAtMs: Date.now() + 60_000,
},
request: {
sessionId: mode === "wrong-session" ? "foreign-session" : session.sessionId,
runId: claim.runId,
turnId: `turn-${mode}`,
runEpoch: active.activeOwnerEpoch,
modelRef: { provider, model: "model" },
context: { messages: [{ role: "user", content: "Say allowed", timestamp: 1 }] },
options: { maxTokens: 16 },
},
config,
sessionTarget: target,
signal,
isCurrent: () =>
claimAuthority.isCurrent() && validateAgentRunDelegatedAuthority(runAuthority),
emit: () => {},
});
// Attach immediately so a rejected asynchronous exchange cannot become unhandled.
const settled = pending.then(
(value) => ({ value }),
(error: unknown) => ({ error }),
);
if (holdExchange) {
await withinTest(
awaitGateBeforeSettlement(
exchangeArrived.promise,
settled,
"Worker did not reach real credential exchange",
),
signal,
);
await placements.releaseTurn(claim);
expect(claimAuthority.isCurrent()).toBe(false);
finishExchange.resolve();
}
const outcome = await withinTest(settled, signal);
// Claim/run denial must not be mistaken for an expired model generation.
expect(lease.snapshot.isCurrent()).toBe(true);
if (mode === "allowed") {
expect(outcome).toMatchObject({
value: {
type: "done",
message: { content: [{ type: "text", text: "allowed worker response" }] },
},
});
expect(effects).toEqual([
{ path: "/auth", authorized: true },
{ path: "/v1/chat/completions", authorized: true },
]);
expect(installed.mock.calls.some(([id]) => id === provider)).toBe(true);
} else {
if (holdExchange) {
expect(outcome).toMatchObject({
error: { message: "Worker inference source is no longer current" },
});
} else {
expect(outcome).toMatchObject({
value: {
type: "error",
reason: mode === "wrong-session" ? "session-not-attached" : "cancelled",
},
});
}
expect(effects).toEqual(
holdExchange ? [{ path: "/auth", authorized: true }] : [],
);
expect(installed.mock.calls.length).toBe(0);
}
console.info(
`authority-boundary ${mode}: auth=${effects.filter((effect) => effect.path === "/auth").length} inference=${effects.filter((effect) => effect.path !== "/auth").length} installed=${installed.mock.calls.length} ${mode === "allowed" ? "success" : "rejected"}`,
);
} finally {
finishExchange.resolve();
installed.mockRestore();
releaseAgentRunDelegatedAuthority(runAuthority);
claimAuthority.release();
if (placements.get(session.sessionId)?.turnClaim) {
await placements.releaseTurn(claim);
}
}
}
} finally {
vi.restoreAllMocks();
await resetPreparedModelRuntimeSnapshotsForTest();
await clearActivePluginRegistry();
clearPluginMetadataLifecycleCaches();
resetAgentRunRegistryForTest();
await closeStateDatabaseForTest();
}
});
},
async () => {
finishExchange.resolve();
server.closeAllConnections();
await reservation.releaseListener();
},
reservation.claim.release,
);
});
});

92
src/gateway/worker-environments/inference-model.test.ts
View file

@ -1,9 +1,11 @@
import assert from "node:assert/strict";
import { describe, expect, it, vi } from "vitest";
import { createDeferred } from "../../../test/helpers/promise.js";
import { prepareSimpleCompletionModel } from "../../agents/simple-completion-runtime.js";
import type { OpenClawConfig } from "../../config/types.openclaw.js";
import type { Model } from "../../llm/types.js";
import { createEmptyPluginMetadataSnapshot } from "../../plugins/plugin-metadata-empty.test-support.js";
import { resolveApprovedWorkerLocalModel, resolveApprovedWorkerModel } from "./inference-model.js";
import { resolveApprovedWorkerModel } from "./inference-model.js";
import {
config,
logicalModel,
@ -14,6 +16,9 @@ import {
setup,
WORKSPACE,
} from "./inference-runtime.test-support.js";
import { prepareWorkerTurnModel } from "./worker-turn-model.js";
const prepareModelWithPolicy = prepareSimpleCompletionModel;
const model: Model = {
...logicalModel,
@ -36,9 +41,86 @@ const configured: OpenClawConfig = {
};
describe("worker prompt credential-owned transcript policy", () => {
it("forwards current required worker authority through real model preparation", async () => {
const required = { ...config, cloudWorkers: { requiredProfile: "required" } };
const runtime = setup(sessionEntry, { config: required });
const reached = new Error("admitted selected-model resolver reached");
const resolver = vi.fn(async () => {
throw reached;
});
runtime.prepareModel.mockImplementation((input, assertCurrent) =>
prepareModelWithPolicy({ ...input, modelResolver: resolver }, assertCurrent),
);
await using lease = await runtime.acquireRuntimeLease({
config: required,
agentId: "runtime-agent",
agentDir: "/gateway-agent",
workspaceDir: WORKSPACE,
});
const assertCurrent = vi.fn();
await expect(
resolveApprovedWorkerModel({
target: { ...params(request(), vi.fn()).sessionTarget, sessionEntry },
modelRef: request().modelRef,
runtimeSnapshot: lease.snapshot,
assertCurrent,
}),
).rejects.toBe(reached);
expect(resolver).toHaveBeenCalledOnce();
expect(runtime.prepareModel).toHaveBeenCalledWith(
expect.objectContaining({
workerInferenceAuthority: { assertCurrent },
}),
);
});
it.each(["wrong session", "revoked selection"] as const)(
"rejects required worker %s before provider preparation",
async (state) => {
const required = { ...config, cloudWorkers: { requiredProfile: "required" } };
const runtime = setup(sessionEntry, { config: required });
const selected = createDeferred();
let current = true;
if (state === "revoked selection") {
runtime.resolveAuthSelection.mockImplementation(async () => {
await selected.promise;
return undefined;
});
}
await using lease = await runtime.acquireRuntimeLease({
config: required,
agentId: "runtime-agent",
agentDir: "/gateway-agent",
workspaceDir: WORKSPACE,
});
const pending = resolveApprovedWorkerModel({
target: {
...params(request(), vi.fn()).sessionTarget,
sessionEntry:
state === "wrong session" ? { ...sessionEntry, sessionId: "other" } : sessionEntry,
},
modelRef: request().modelRef,
runtimeSnapshot: lease.snapshot,
assertCurrent: () => {
if (!current) {
throw new Error("worker claim revoked");
}
},
});
if (state === "wrong session") {
expect(await pending).toBeUndefined();
} else {
current = false;
selected.resolve();
await expect(pending).rejects.toThrow("worker claim revoked");
}
expect(runtime.prepareModel).not.toHaveBeenCalled();
},
);
it("approves a worker-local configured model without resolving Gateway credentials", async () => {
const runtime = setup(sessionEntry, {
config: structuredClone(configured),
config: { ...structuredClone(configured), cloudWorkers: { requiredProfile: "required" } },
configuredRuntimeModel: model,
});
await using lease = await runtime.acquireRuntimeLease({
@ -47,10 +129,12 @@ describe("worker prompt credential-owned transcript policy", () => {
agentDir: "/gateway-agent",
workspaceDir: WORKSPACE,
});
const approved = await resolveApprovedWorkerLocalModel({
target: { ...params(request(), vi.fn()).sessionTarget, sessionEntry },
const approved = await prepareWorkerTurnModel({
target: params(request(), vi.fn()).sessionTarget,
modelRef,
runtimeSnapshot: lease.snapshot,
inferencePlacement: "worker",
turn: { config: lease.snapshot.config, workspaceDir: WORKSPACE },
assertCurrent: () => undefined,
});
assert(approved && !("error" in approved));

2
src/gateway/worker-environments/inference-model.ts
View file

@ -179,6 +179,8 @@ export async function resolveApprovedWorkerModel(params: ResolveApprovedWorkerMo
// Route projection and credential selection are one decision. Pin even an
// automatic profile so generic auth fallback cannot cross to another route.
const prepared = await prepareSimpleCompletionModel({
// This session-bound worker owner revalidates the exact live turn after awaited work.
workerInferenceAuthority: { assertCurrent: params.assertCurrent },
cfg: modelConfig,
transport: "provider-stream",
agentId: target.agentId,

60
src/plugins/provider-runtime-auth.test.ts Normal file
View file

@ -0,0 +1,60 @@
import { beforeEach, expect, it, vi } from "vitest";
import { makeProviderModelFixture } from "../agents/test-helpers/provider-model-fixture.js";
const mocks = vi.hoisted(() => ({ resolveProviderRuntimePlugin: vi.fn() }));
// mock-isolation: Keep plugin registry discovery outside the credential exchange boundary.
vi.mock("./provider-hook-runtime.js", () => ({
resolveProviderRuntimePlugin: mocks.resolveProviderRuntimePlugin,
}));
// mock-isolation: Load only the real auth owner while exercising the unchanged lazy runtime facade.
vi.mock("./provider-runtime.js", async () => ({
prepareProviderRuntimeAuth: (await import("./provider-runtime-auth.js"))
.prepareProviderRuntimeAuth,
}));
const MODEL = makeProviderModelFixture({
provider: "demo",
id: "fixture-model",
api: "openai-responses",
baseUrl: "https://provider.example/v1",
});
beforeEach(() => mocks.resolveProviderRuntimePlugin.mockReset());
it.each([false, true])(
"revalidates auth exchange after lazy provider loading (revoked: %s)",
async (revoke) => {
const lazyRuntime = await import("./provider-runtime.runtime.js");
const prepareRuntimeAuth = vi.fn(async () => ({ apiKey: "runtime-token" }));
mocks.resolveProviderRuntimePlugin.mockReturnValue({ prepareRuntimeAuth });
let current = true;
const revoked = new Error("worker claim revoked during provider loading");
const pending = lazyRuntime.prepareProviderRuntimeAuth({
provider: "demo",
assertCurrent: () => {
if (!current) {
throw revoked;
}
},
context: {
env: process.env,
provider: "demo",
modelId: MODEL.id,
model: MODEL,
apiKey: "provider-source-token",
authMode: "token",
},
});
expect(prepareRuntimeAuth).not.toHaveBeenCalled();
current = !revoke;
if (revoke) {
await expect(pending).rejects.toBe(revoked);
expect(prepareRuntimeAuth).not.toHaveBeenCalled();
} else {
await expect(pending).resolves.toEqual({ apiKey: "runtime-token" });
expect(prepareRuntimeAuth).toHaveBeenCalledOnce();
}
},
);

30
src/plugins/provider-runtime-auth.ts Normal file
View file

@ -0,0 +1,30 @@
import { unwrapSecretSentinelsForProviderEgress } from "../agents/provider-secret-egress.js";
import { resolveProviderRuntimePlugin } from "./provider-hook-runtime.js";
import type { ProviderPrepareRuntimeAuthContext } from "./types.js";
export async function prepareProviderRuntimeAuth(
params: Pick<
Parameters<typeof resolveProviderRuntimePlugin>[0],
"provider" | "config" | "workspaceDir" | "env"
> & {
assertCurrent?: () => void;
context: ProviderPrepareRuntimeAuthContext;
},
) {
const prepareRuntimeAuth = resolveProviderRuntimePlugin(params)?.prepareRuntimeAuth;
if (!prepareRuntimeAuth) {
return undefined;
}
// The lazy runtime import may outlive the caller's authority to exchange credentials.
params.assertCurrent?.();
// Secret material crosses into provider code only when that provider owns an
// auth hook. Callers can safely pass sentinels without probing plugin state.
const preparedInput = unwrapSecretSentinelsForProviderEgress(
params.context.apiKey,
"provider runtime auth exchange",
);
return await prepareRuntimeAuth({
...params.context,
apiKey: preparedInput,
});
}

23
src/plugins/provider-runtime.ts
View file

@ -12,7 +12,6 @@ import {
applyPluginTextReplacements,
mergePluginTextTransforms,
} from "../agents/plugin-text-transforms.js";
import { unwrapSecretSentinelsForProviderEgress } from "../agents/provider-secret-egress.js";
import type { StreamFn } from "../agents/runtime/index.js";
import type { ProviderSystemPromptContribution } from "../agents/system-prompt-contribution.js";
import type { ModelProviderConfig } from "../config/types.js";
@ -70,7 +69,6 @@ import type {
ProviderNormalizeTransportContext,
ProviderPreferRuntimeResolvedModelContext,
ProviderPlugin,
ProviderPrepareRuntimeAuthContext,
ProviderResolveConfigApiKeyContext,
ProviderResolveTransportTurnStateContext,
ProviderSystemPromptContributionContext,
@ -484,26 +482,7 @@ export function resolveProviderTransportTurnStateWithPlugin(
};
}
export async function prepareProviderRuntimeAuth(
params: ProviderRuntimeLookup & {
context: ProviderPrepareRuntimeAuthContext;
},
) {
const prepareRuntimeAuth = resolveProviderRuntimePlugin(params)?.prepareRuntimeAuth;
if (!prepareRuntimeAuth) {
return undefined;
}
// Secret material crosses into provider code only when that provider owns an
// auth hook. Callers can safely pass sentinels without probing plugin state.
const preparedInput = unwrapSecretSentinelsForProviderEgress(
params.context.apiKey,
"provider runtime auth exchange",
);
return await prepareRuntimeAuth({
...params.context,
apiKey: preparedInput,
});
}
export { prepareProviderRuntimeAuth } from "./provider-runtime-auth.js";
const resolveUsageAuth = asyncRuntimeHook("resolveUsageAuth");
export const resolveProviderUsageAuthWithPlugin = async (

2
test/vitest/vitest.database-worker-core-paths.mjs
View file

@ -1,5 +1,6 @@
// These consumers need the host-owned SQLite broker, which runs in forked processes.
export const databaseWorkerCoreTestFiles = [
"src/gateway/agent-turn/agent-request-routing.session-id.test.ts",
"src/gateway/conversation-route-ownership.worker.test.ts",
"src/gateway/conversation-list.admission.test.ts",
"src/gateway/conversation-list.account-removal.test.ts",
@ -98,6 +99,7 @@ export const databaseWorkerCoreTestFiles = [
"src/agents/embedded-agent-runner/run-entry.cleanup.test.ts",
"src/agents/embedded-agent-runner/run-entry.cyber-failover.runner.test.ts",
"src/agents/embedded-agent-runner/run-orchestrator.rooted-proof.test.ts",
"src/agents/embedded-agent-runner/run-orchestrator.required-profile.test.ts",
"src/agents/embedded-agent-runner/run-orchestrator.suspension.test.ts",
"src/agents/embedded-agent-runner/run.cli-dispatch-lane.test.ts",
"src/agents/embedded-agent-runner.run-embedded-agent.finalization-scope.test.ts",