From 8ede4046e2267cff666519fd9395a29328e62889 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 8 Aug 2026 23:59:50 -0700 Subject: [PATCH] fix(cli): guard embedded agent state ownership (#120896) --- docs/cli/agent.md | 1 + .../gateway-cli/run.supervised-lock.test.ts | 24 ++++ src/commands/agent-via-gateway.test.ts | 109 +++++++++++++++++- src/commands/agent-via-gateway.ts | 77 ++++++++++--- src/infra/gateway-lock.roles.test.ts | 64 +++++++++- src/infra/gateway-lock.ts | 25 +++- 6 files changed, 277 insertions(+), 23 deletions(-) diff --git a/docs/cli/agent.md b/docs/cli/agent.md index 98d0c5eb7354..8af5d55057d4 100644 --- a/docs/cli/agent.md +++ b/docs/cli/agent.md @@ -165,6 +165,7 @@ openclaw agent --agent ops --message "Run locally" --local - Pass exactly one of `--message` or `--message-file`. `--message-file` strips a leading UTF-8 BOM and preserves multiline content; it rejects files that are not valid UTF-8. Files larger than 4 MiB are rejected before dispatch. - Slash commands (for example `/compact`) cannot run through `--message`. The CLI rejects them and points you at the first-class command instead (`openclaw sessions compact ` for compaction). - `--local` runs are one-shot: bundled MCP loopback resources and warm Claude stdio sessions opened for the run are retired after the reply, so scripted invocations do not leave local child processes running. Gateway-backed runs keep Gateway-owned MCP loopback resources under the running Gateway process instead. +- `--local` requires exclusive ownership of the configured state directory. It refuses to start while a Gateway or another `agent --local` run owns that directory, then holds the same state lock for the full embedded turn. Run without `--local` to use the active Gateway, or stop it first with `openclaw gateway stop`. - Standalone embedded execution with `--local` refuses to reuse an existing main session while restart recovery is pending. Run the turn through a healthy Gateway, or reset it there with `/new` or `/reset`; an independent embedded process cannot safely coordinate that recovery owner with the Gateway scanner. - With `--agent`, `--channel` and `--to` together, session routing follows the channel's canonical recipient and `session.dmScope`. Channels with a stable outbound-only recipient identity use a provider-owned session isolated from the agent's main session. `--reply-channel` and `--reply-account` affect delivery only. - `--session-key` selects an explicit session key. Agent-prefixed keys must use `agent::`, and `--agent` must match the key's agent id when both are given. Bare non-sentinel keys scope to `--agent` when supplied, or to the configured default agent otherwise; for example `--agent ops --session-key incident-42` routes to `agent:ops:incident-42`. The literal keys `global` and `unknown` stay unscoped only when no `--agent` is supplied. diff --git a/src/cli/gateway-cli/run.supervised-lock.test.ts b/src/cli/gateway-cli/run.supervised-lock.test.ts index 91d07cc6ed32..a2a2d4ba5ac9 100644 --- a/src/cli/gateway-cli/run.supervised-lock.test.ts +++ b/src/cli/gateway-cli/run.supervised-lock.test.ts @@ -102,6 +102,30 @@ describe("supervised gateway lock recovery", () => { expect(testing.resolveGatewayLockErrorExitCode(failure)).toBe(78); }); + it("preserves an agent-embedded owner error under a supervisor", async () => { + const err = new GatewayLockError( + "another openclaw agent --local run is active (pid 123); lock timeout after 5000ms", + ); + const startLoop = vi.fn(async () => { + throw err; + }); + const probeHealth = vi.fn(async () => true); + + await expect( + testing.runGatewayLoopWithSupervisedLockRecovery({ + startLoop, + supervisor: "systemd", + port: 18789, + healthHost: "127.0.0.1", + log: createLogger(), + probeHealth, + }), + ).rejects.toBe(err); + + expect(startLoop).toHaveBeenCalledTimes(1); + expect(probeHealth).not.toHaveBeenCalled(); + }); + it("bounds supervised retries when the existing gateway stays unhealthy", async () => { let now = 0; const startLoop = vi.fn(async () => { diff --git a/src/commands/agent-via-gateway.test.ts b/src/commands/agent-via-gateway.test.ts index 738ad6c480b7..365542477f93 100644 --- a/src/commands/agent-via-gateway.test.ts +++ b/src/commands/agent-via-gateway.test.ts @@ -11,6 +11,7 @@ import { hasExecutionIdentityAdmissionSink, } from "../audit/execution-identity-admission.js"; import type { OpenClawConfig } from "../config/config.js"; +import { acquireGatewayLock, type GatewayLockOptions } from "../infra/gateway-lock.js"; import { loggingState } from "../logging/state.js"; import type { RuntimeEnv } from "../runtime.js"; import { AGENT_HARNESS_SESSION_KEY_RESERVED_MESSAGE } from "../sessions/agent-harness-session-key.js"; @@ -121,6 +122,23 @@ function mockLocalAgentReply(text = "local") { }); } +function createLocalGatewayLockOptions( + stateDir: string, + overrides: Partial = {}, +): GatewayLockOptions { + return { + allowInTests: true, + env: { + ...process.env, + OPENCLAW_CONFIG_PATH: path.join(stateDir, "openclaw.json"), + OPENCLAW_STATE_DIR: stateDir, + }, + lockDir: path.join(stateDir, "gateway-locks"), + timeoutMs: 100, + ...overrides, + }; +} + function requireFirstCallArg(mock: { mock: { calls: unknown[][] } }, label: string): unknown { const [call] = mock.mock.calls; if (!call) { @@ -464,6 +482,81 @@ describe("agentCliCommand", () => { }); }); + it("refuses --local before embedded startup when a live Gateway owns the state directory", async () => { + await withTempStore(async ({ dir }) => { + const lockOptions = createLocalGatewayLockOptions(dir, { + readProcessStartTime: () => 123_456, + }); + const gatewayLock = await acquireGatewayLock({ + ...lockOptions, + port: 28789, + }); + expect(gatewayLock).not.toBeNull(); + if (!gatewayLock) { + throw new Error("Expected live Gateway fixture lock"); + } + + try { + await expect( + agentCliCommand({ message: "hi", to: "+1555", local: true, json: true }, jsonRuntime, { + localGatewayLockOptions: lockOptions, + }), + ).rejects.toThrow( + `A Gateway is running for this state directory (pid ${process.pid}, port 28789). Run without --local to use it, or stop the Gateway first (openclaw gateway stop).`, + ); + expect(agentCommand).not.toHaveBeenCalled(); + expect(startOneShotDiagnosticsExporters).not.toHaveBeenCalled(); + expect(jsonRuntime.writeJson).not.toHaveBeenCalled(); + } finally { + await gatewayLock.release(); + } + }); + }); + + it("holds one agent-embedded state lock for the run and rejects a concurrent --local run", async () => { + await withTempStore(async ({ dir }) => { + const lockOptions = createLocalGatewayLockOptions(dir); + let finishFirstRun: ((value: Awaited>) => void) | undefined; + agentCommand.mockImplementationOnce( + async () => + await new Promise>>((resolve) => { + finishFirstRun = resolve; + }), + ); + + const firstRun = agentCliCommand({ message: "first", to: "+1555", local: true }, runtime, { + localGatewayLockOptions: lockOptions, + }); + await waitForAgentCommandCall(); + + const stateLockPath = path.join(lockOptions.lockDir!, "gateway.state.lock"); + const payload = JSON.parse(fs.readFileSync(stateLockPath, "utf8")) as { + pid?: number; + role?: string; + }; + expect(payload).toMatchObject({ pid: process.pid, role: "agent-embedded" }); + + await expect( + agentCliCommand({ message: "second", to: "+1555", local: true }, runtime, { + localGatewayLockOptions: { ...lockOptions, pollIntervalMs: 2, timeoutMs: 15 }, + }), + ).rejects.toThrow( + `another openclaw agent --local run is active (pid ${process.pid}); lock timeout after 15ms`, + ); + expect(agentCommand).toHaveBeenCalledTimes(1); + + if (!finishFirstRun) { + throw new Error("Expected first embedded run to start"); + } + finishFirstRun({ + payloads: [{ text: "done" }], + meta: { durationMs: 1 }, + } as Awaited>); + await firstRun; + expect(fs.existsSync(stateLockPath)).toBe(false); + }); + }); + it("rejects inline and file messages together", async () => { await expect( agentCliCommand( @@ -1537,9 +1630,13 @@ describe("agentCliCommand", () => { }); }); - it("passes SIGTERM abort signals into local agent runs", async () => { - await withTempStore(async () => { + it.each([ + ["SIGTERM", 143], + ["SIGINT", 130], + ] as const)("releases the local state lock after %s aborts the run", async (signal, exitCode) => { + await withTempStore(async ({ dir }) => { const signals = createSignalProcess(); + const lockOptions = createLocalGatewayLockOptions(dir); agentCommand.mockImplementationOnce(async (opts: { abortSignal?: AbortSignal }) => { expect(opts.abortSignal).toBeInstanceOf(AbortSignal); return await new Promise((_, reject) => { @@ -1557,13 +1654,17 @@ describe("agentCliCommand", () => { const run = agentCliCommand({ message: "hi", to: "+1555", local: true }, runtime, { process: signals.processLike, + localGatewayLockOptions: lockOptions, }); await waitForAgentCommandCall(); - signals.emit("SIGTERM"); + const stateLockPath = path.join(lockOptions.lockDir!, "gateway.state.lock"); + expect(fs.existsSync(stateLockPath)).toBe(true); + signals.emit(signal); await run; + expect(fs.existsSync(stateLockPath)).toBe(false); expect(callGateway).not.toHaveBeenCalled(); - expect(runtime.exit).toHaveBeenCalledWith(143); + expect(runtime.exit).toHaveBeenCalledWith(exitCode); expect(signals.listenerCount("SIGTERM")).toBe(0); expect(signals.listenerCount("SIGINT")).toBe(0); }); diff --git a/src/commands/agent-via-gateway.ts b/src/commands/agent-via-gateway.ts index c77be1d2e034..a5bc7bae8bb9 100644 --- a/src/commands/agent-via-gateway.ts +++ b/src/commands/agent-via-gateway.ts @@ -30,6 +30,7 @@ import { isGatewaySecretRefUnavailableError } from "../gateway/credentials.js"; import { ADMIN_SCOPE } from "../gateway/operator-scopes.js"; import { createAbortError } from "../infra/abort-signal.js"; import { readFileDescriptorBounded } from "../infra/boundary-file-read.js"; +import type { GatewayLockIdentity, GatewayLockOptions } from "../infra/gateway-lock.js"; import { parseStrictNonNegativeInteger } from "../infra/parse-finite-number.js"; import { routeLogsToStderr } from "../logging/console.js"; import { @@ -103,6 +104,7 @@ type AgentCliProcessLike = { }; type AgentCliDeps = CliDeps & { process?: AgentCliProcessLike; + localGatewayLockOptions?: GatewayLockOptions; }; type AgentGatewayCallIdentity = Pick< Parameters[0], @@ -136,6 +138,9 @@ const agentSessionModuleCache = createLazyPromiseLoader(() => agentSessionModule const runtimeConfigModuleLoader = createLazyPromiseLoader(() => import("../config/io.js"), { cacheRejections: true, }); +const gatewayLockModuleLoader = createLazyPromiseLoader(() => import("../infra/gateway-lock.js"), { + cacheRejections: true, +}); const replyPayloadModuleLoader = createLazyPromiseLoader( () => import("openclaw/plugin-sdk/reply-payload"), { cacheRejections: true }, @@ -213,6 +218,42 @@ async function loadRuntimeConfig(): Promise { return getRuntimeConfig(); } +function formatActiveGatewayLocalRefusal(identity: GatewayLockIdentity): string { + return `A Gateway is running for this state directory (pid ${identity.pid}, port ${identity.port}). Run without --local to use it, or stop the Gateway first (${formatCliCommand("openclaw gateway stop")}).`; +} + +async function acquireEmbeddedAgentStateLock( + options: GatewayLockOptions | undefined, + signal: AbortSignal, +) { + const { acquireGatewayLock, GatewayLockError, readActiveGatewayLockIdentity } = + await gatewayLockModuleLoader.load(); + const env = options?.env ?? process.env; + if (options?.allowInTests !== true && (env.VITEST || env.NODE_ENV === "test")) { + return null; + } + const activeGateway = await readActiveGatewayLockIdentity(options); + if (activeGateway) { + throw new GatewayLockError(formatActiveGatewayLocalRefusal(activeGateway)); + } + try { + return await acquireGatewayLock({ + ...options, + role: "agent-embedded", + sleep: options?.sleep ?? (async (ms) => await delayMs(ms, signal)), + }); + } catch (error) { + if (!(error instanceof GatewayLockError)) { + throw error; + } + const racedGateway = await readActiveGatewayLockIdentity(options); + if (racedGateway) { + throw new GatewayLockError(formatActiveGatewayLocalRefusal(racedGateway), error); + } + throw error; + } +} + const loadReplyPayloadModule = replyPayloadModuleLoader.load; /** Test-only hooks for resetting lazy imports and shortening retry timing. */ @@ -222,6 +263,7 @@ export const agentViaGatewayTesting = { localAuditModuleLoader.clear(); agentSessionModuleCache.clear(); runtimeConfigModuleLoader.clear(); + gatewayLockModuleLoader.clear(); replyPayloadModuleLoader.clear(); agentSessionModuleLoader = defaultAgentSessionModuleLoader; }, @@ -989,20 +1031,29 @@ export async function agentCliCommand( const signalBridge = createAgentCliSignalBridge(resolveAgentCliProcessLike(deps)); try { if (dispatchOpts.local === true) { - const result = await runEmbeddedAgentCommand( - { - ...gatewayDispatchOpts, - agentId: gatewayDispatchOpts.agent, - replyAccountId: gatewayDispatchOpts.replyAccount, - cleanupBundleMcpOnRunEnd: true, - cleanupCliLiveSessionOnRunEnd: true, - oneShotCliRun: true, - abortSignal: signalBridge.signal, - }, - runtime, - deps, - { suppressStdoutDiagnosticLogs: dispatchOpts.json === true }, + const stateLock = await acquireEmbeddedAgentStateLock( + deps?.localGatewayLockOptions, + signalBridge.signal, ); + let result: Awaited>; + try { + result = await runEmbeddedAgentCommand( + { + ...gatewayDispatchOpts, + agentId: gatewayDispatchOpts.agent, + replyAccountId: gatewayDispatchOpts.replyAccount, + cleanupBundleMcpOnRunEnd: true, + cleanupCliLiveSessionOnRunEnd: true, + oneShotCliRun: true, + abortSignal: signalBridge.signal, + }, + runtime, + deps, + { suppressStdoutDiagnosticLogs: dispatchOpts.json === true }, + ); + } finally { + await stateLock?.release(); + } return returnAfterSignalExit(result, signalBridge.getReceivedSignal(), runtime); } diff --git a/src/infra/gateway-lock.roles.test.ts b/src/infra/gateway-lock.roles.test.ts index 1226578d107a..9f1401d16ae1 100644 --- a/src/infra/gateway-lock.roles.test.ts +++ b/src/infra/gateway-lock.roles.test.ts @@ -4,7 +4,12 @@ import path from "node:path"; import { setTimeout as nativeSleep } from "node:timers/promises"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { createSuiteTempRootTracker } from "../test-helpers/temp-dir.js"; -import { acquireGatewayLock, GatewayLockError, readActiveGatewayLockPort } from "./gateway-lock.js"; +import { + acquireGatewayLock, + GatewayLockError, + readActiveGatewayLockIdentity, + readActiveGatewayLockPort, +} from "./gateway-lock.js"; const fixtureRootTracker = createSuiteTempRootTracker({ prefix: "openclaw-gateway-lock-workshop-", @@ -118,4 +123,61 @@ describe("Gateway lock roles", () => { await lock.release(); } }); + + it("keeps agent-embedded ownership distinct from a running Gateway", async () => { + const stateDir = await fixtureRootTracker.make("agent-embedded-role"); + const lockDir = path.join(fixtureRoot, "__locks"); + const configPath = path.join(stateDir, "openclaw.json"); + await fs.writeFile(configPath, "{}", "utf8"); + const env = { + ...process.env, + OPENCLAW_CONFIG_PATH: configPath, + OPENCLAW_STATE_DIR: stateDir, + }; + const readProcessCmdline = () => ["openclaw", "agent", "--local", "--message", "hello"]; + const lock = await acquireGatewayLock({ + allowInTests: true, + env, + lockDir, + platform: "darwin", + port: 28789, + readProcessCmdline, + readProcessStartTime: () => null, + role: "agent-embedded", + timeoutMs: 30, + }); + expect(lock).not.toBeNull(); + if (!lock) { + throw new Error("Expected embedded agent Gateway lock"); + } + + try { + await expect( + readActiveGatewayLockIdentity({ + env, + lockDir, + platform: "darwin", + readProcessCmdline, + readProcessStartTime: () => null, + }), + ).resolves.toBeUndefined(); + await expect( + acquireGatewayLock({ + allowInTests: true, + env, + lockDir, + platform: "darwin", + pollIntervalMs: 2, + readProcessCmdline, + readProcessStartTime: () => null, + sleep: nativeSleep, + timeoutMs: 15, + }), + ).rejects.toThrow( + `another openclaw agent --local run is active (pid ${process.pid}); lock timeout after 15ms`, + ); + } finally { + await lock.release(); + } + }); }); diff --git a/src/infra/gateway-lock.ts b/src/infra/gateway-lock.ts index e658d9752881..6a72aa6a8097 100644 --- a/src/infra/gateway-lock.ts +++ b/src/infra/gateway-lock.ts @@ -44,7 +44,9 @@ const LockPayloadSchema = z.object({ createdAt: z.string(), configPath: z.string(), port: z.number().int().min(1).max(65_535).optional(), - role: z.enum(["gateway", "skill-workshop-apply", "sqlite-maintenance"]).optional(), + role: z + .enum(["gateway", "agent-embedded", "skill-workshop-apply", "sqlite-maintenance"]) + .optional(), stateDir: z.string().optional(), startTime: z.number().optional(), }) as z.ZodType; @@ -56,7 +58,7 @@ type GatewayLockHandle = { release: () => Promise; }; -type GatewayLockRole = "gateway" | "skill-workshop-apply" | "sqlite-maintenance"; +type GatewayLockRole = "gateway" | "agent-embedded" | "skill-workshop-apply" | "sqlite-maintenance"; export type GatewayLockIdentity = { pid: number; @@ -198,11 +200,18 @@ async function resolveGatewayOwnerStatus( } const readFn = readCmdline ?? ((p: number) => defaultReadProcessCmdline(p, platform)); - if (role === "sqlite-maintenance" || role === "skill-workshop-apply") { + if ( + role === "agent-embedded" || + role === "sqlite-maintenance" || + role === "skill-workshop-apply" + ) { const args = readFn(pid); if (!args) { return "unknown"; } + if (role === "agent-embedded") { + return isOpenClawCommandArgv(args, "agent") && args.includes("--local") ? "alive" : "dead"; + } const command = role === "sqlite-maintenance" ? "doctor" : "skills"; return isOpenClawCommandArgv(args, command) ? "alive" : "dead"; } @@ -545,6 +554,12 @@ async function acquireLockFile( await sleep(Math.min(pollIntervalMs, remainingMs)); } - const owner = lastPayload?.pid ? ` (pid ${lastPayload.pid})` : ""; - throw new GatewayLockError(`gateway already running${owner}; lock timeout after ${timeoutMs}ms`); + const ownerPid = lastPayload?.pid ? ` (pid ${lastPayload.pid})` : ""; + const owner = + lastPayload?.role === "agent-embedded" + ? `another openclaw agent --local run is active${ownerPid}` + : lastPayload?.role && lastPayload.role !== "gateway" + ? `state directory is locked by ${lastPayload.role}${ownerPid}` + : `gateway already running${ownerPid}`; + throw new GatewayLockError(`${owner}; lock timeout after ${timeoutMs}ms`); }