From c5254f13ee221096149c8871e8ff4de98820fa23 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Mon, 20 Jul 2026 23:30:37 -0700 Subject: [PATCH] =?UTF-8?q?refactor(cli)!:=20remove=20automatic=20gateway?= =?UTF-8?q?=E2=86=92embedded=20fallback=20from=20openclaw=20agent=20(#1120?= =?UTF-8?q?74)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A Gateway timeout or closed connection now fails the command with an actionable stderr hint instead of silently re-running the whole turn embedded under a fresh gateway-fallback-* session. The silent fallback could double-execute side effects (the Gateway may still finish an accepted turn), returned context-free answers to --session-key callers, and ran with the CLI host's local config. --local remains the only embedded execution path. Also deletes the resultMetaOverrides plumbing (the fallback was its only writer) and the fallback marker fields added in #111645. --- .../.generated/plugin-sdk-api-baseline.sha256 | 2 +- docs/automation/cron-jobs.md | 2 +- docs/cli/agent.md | 13 +- docs/tools/agent-send.md | 10 +- src/agents/command/delivery.test.ts | 42 -- src/agents/command/delivery.ts | 30 +- src/agents/command/types.ts | 18 +- src/cli/run-main.exit.test.ts | 4 +- src/commands/agent-via-gateway.test.ts | 420 ++---------------- src/commands/agent-via-gateway.ts | 133 ++---- 10 files changed, 97 insertions(+), 577 deletions(-) diff --git a/docs/.generated/plugin-sdk-api-baseline.sha256 b/docs/.generated/plugin-sdk-api-baseline.sha256 index f65782c72b11..cca2d42b40c7 100644 --- a/docs/.generated/plugin-sdk-api-baseline.sha256 +++ b/docs/.generated/plugin-sdk-api-baseline.sha256 @@ -6,7 +6,7 @@ e5e67ddf3cab38fcbf9220bc3160715897e2709d9a9ff6ff36f1ecc9453c2367 module/agent-c 74daa746deb548379d3f0d6eac3c4d082df1034c4360cc03bf51fee0f10a2e4d module/agent-harness c67e52e5bc6d6917f9a13f4133e7b18e111dbd33b66b6a5eae79623d9ff6e918 module/agent-harness-runtime 5168648cd946abad8a92822889f13ceacc87ed502314a66190d0b1eb8ebe76ea module/agent-media-payload -2c60564ae141dc249234e409994498f4143943d36c7860fa5759a63e4375e377 module/agent-runtime +0133dcb10ac42e89a5cac9ffe96504b9b3f4d747e70e0eeb1769fc9202c2e7c8 module/agent-runtime dd9282f1eeadf44db2887599b52d80db7f5fb99c6d9eac720dbf1b77065f2145 module/allow-from 55cea5390d68839ca7768b4a0cc570b17b65fa0fa3bc4d76130ef0f16cb79ede module/allowlist-config-edit 7ddd81bd5f55de9adf64bf4d92d012f24b37b6da0a72805a3a220d8feff24ca3 module/approval-auth-runtime diff --git a/docs/automation/cron-jobs.md b/docs/automation/cron-jobs.md index 4f586034c244..ac9d712ab502 100644 --- a/docs/automation/cron-jobs.md +++ b/docs/automation/cron-jobs.md @@ -45,7 +45,7 @@ Cron is the Gateway's built-in scheduler. It persists jobs, wakes the agent at t - One-shot jobs (`--at`) auto-delete after success by default; pass `--keep-after-run` to keep them. - Per-run wall-clock budget: `--timeout-seconds` when set. Otherwise, isolated/detached agent-turn jobs are bounded by cron's own 60-minute watchdog before the underlying agent-turn timeout (`agents.defaults.timeoutSeconds`, default 48 hours) would ever apply; command jobs default to 10 minutes, and script payloads default to 5 minutes. - On Gateway startup, overdue isolated agent-turn jobs are rescheduled instead of replayed immediately, keeping model/tool bootstrap work out of the channel-connect window. -- If you drive `openclaw agent` from system cron or another external scheduler, wrap it with a hard-kill escalation even though the CLI already handles `SIGTERM`/`SIGINT`. Gateway-backed runs ask the Gateway to abort accepted runs; local and embedded fallback runs get the same abort signal. For GNU `timeout`, prefer `timeout -k 60 600 openclaw agent ...` over plain `timeout 600 ...` — the `-k` value is the backstop if the process cannot drain in time. For systemd units, use a `SIGTERM` stop signal with a grace window (`TimeoutStopSec`) before the final kill. Reusing a `--run-id` while the original Gateway run is still active reports the duplicate as in-flight instead of starting a second run. +- If you drive `openclaw agent` from system cron or another external scheduler, wrap it with a hard-kill escalation even though the CLI already handles `SIGTERM`/`SIGINT`. Gateway-backed runs ask the Gateway to abort accepted runs; `--local` runs get the same abort signal. For GNU `timeout`, prefer `timeout -k 60 600 openclaw agent ...` over plain `timeout 600 ...` — the `-k` value is the backstop if the process cannot drain in time. For systemd units, use a `SIGTERM` stop signal with a grace window (`TimeoutStopSec`) before the final kill. Reusing a `--run-id` while the original Gateway run is still active reports the duplicate as in-flight instead of starting a second run. diff --git a/docs/cli/agent.md b/docs/cli/agent.md index 971246267a72..671a957a5d75 100644 --- a/docs/cli/agent.md +++ b/docs/cli/agent.md @@ -7,7 +7,7 @@ title: "Agent" # `openclaw agent` -Run one agent turn through the Gateway. Falls back to the embedded agent if the Gateway request fails; pass `--local` to force embedded execution up front. +Run one agent turn through the Gateway. The explicit `--local` flag is the only embedded execution path. Pass at least one session selector: `--to`, `--session-key`, `--session-id`, or `--agent`. @@ -52,14 +52,13 @@ 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` and embedded fallback 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. -- Standalone embedded execution (`--local` and transport fallback) 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. +- `--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. +- 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. -- `--json` reserves stdout for the JSON response; Gateway, plugin, and embedded-fallback diagnostics go to stderr so scripts can parse stdout directly. -- Embedded fallback JSON includes `meta.transport: "embedded"` and `meta.fallbackFrom: "gateway"` so scripts can detect a fallback run. -- If the Gateway accepts a run but the CLI times out waiting for the final reply, embedded fallback uses a fresh `gateway-fallback-*` session/run id and reports `meta.fallbackReason: "gateway_timeout"` plus the fallback session fields, instead of racing the Gateway-owned transcript or silently replacing the original session. -- `SIGTERM`/`SIGINT` interrupt a waiting Gateway-backed request; if the Gateway already accepted the run, the CLI also sends `chat.abort` for that run id before exiting. `--local` and embedded fallback runs receive the same signal but do not send `chat.abort`. If the internal run-dedup key already has an active run for this session, the response reports `status: "in_flight"` and the non-JSON CLI prints a stderr diagnostic instead of an empty reply. For external cron/systemd wrappers, keep a hard-kill backstop such as `timeout -k 60 600 openclaw agent ...` so the supervisor can reap the process if shutdown cannot drain. +- `--json` reserves stdout for the JSON response; Gateway, plugin, and `--local` diagnostics go to stderr so scripts can parse stdout directly. +- After transient handshake retries are exhausted, a Gateway timeout or closed connection fails the command; the CLI never silently reruns the turn embedded. Transport loss is ambiguous — the Gateway may have accepted and may still finish the turn — so the stderr hint says to check `openclaw gateway status` and the session transcript before retrying or rerunning with `--local`, to avoid executing the turn twice. +- `SIGTERM`/`SIGINT` interrupt a waiting Gateway-backed request; if the Gateway already accepted the run, the CLI also sends `chat.abort` for that run id before exiting. `--local` runs receive the same signal but do not send `chat.abort`. If the internal run-dedup key already has an active run for this session, the response reports `status: "in_flight"` and the non-JSON CLI prints a stderr diagnostic instead of an empty reply. For external cron/systemd wrappers, keep a hard-kill backstop such as `timeout -k 60 600 openclaw agent ...` so the supervisor can reap the process if shutdown cannot drain. - When this command triggers `models.json` regeneration, SecretRef-managed provider credentials are persisted as non-secret markers (for example env var names, `secretref-env:ENV_VAR_NAME`, or `secretref-managed`), never resolved secret plaintext. Marker writes come from the active source config snapshot, not from resolved runtime secret values. ## JSON delivery status diff --git a/docs/tools/agent-send.md b/docs/tools/agent-send.md index 7c021ec81944..002b18e18127 100644 --- a/docs/tools/agent-send.md +++ b/docs/tools/agent-send.md @@ -91,9 +91,10 @@ programmatic delivery. Full flag and behavior reference: - Pass exactly one of `--message` or `--message-file`. File messages preserve multiline content after removing an optional UTF-8 BOM. Files larger than 4 MiB are rejected before dispatch. -- If the Gateway request fails, the CLI **falls back** to the local embedded - run; a Gateway timeout falls back with a fresh session instead of racing the - original transcript. +- After transient handshake retries, a Gateway timeout or closed connection + fails the command with a stderr hint; the CLI never silently reruns the turn + embedded. The Gateway may still finish an accepted turn, so verify Gateway + and session state before retrying or rerunning with `--local`. - Session selection: `--to` derives the session key (group/channel targets preserve isolation; direct chats collapse to `main`). With `--agent`, `--channel`, and `--to` together, routing follows the channel's canonical @@ -105,8 +106,7 @@ programmatic delivery. Full flag and behavior reference: supplied; for example, `--agent ops --session-key incident-42` routes to `agent:ops:incident-42`. Without `--agent`, bare non-sentinel keys are scoped to the configured default agent. Literal `global` and `unknown` remain - unscoped only when no `--agent` is supplied; the embedded fallback path - resolves those sentinel sessions to the configured default agent. + unscoped only when no `--agent` is supplied. - `--reply-channel` and `--reply-account` affect delivery only. - Thinking and verbose flags persist into the session store. - Output: plain text by default, or `--json` for structured payload + metadata. diff --git a/src/agents/command/delivery.test.ts b/src/agents/command/delivery.test.ts index df1784d53001..8f0684f822bb 100644 --- a/src/agents/command/delivery.test.ts +++ b/src/agents/command/delivery.test.ts @@ -722,48 +722,6 @@ describe("deliverAgentCommandResult payload normalization", () => { ); }); - it("merges result metadata overrides into JSON output and returned results", async () => { - const runtime = { - log: vi.fn(), - writeStdout: vi.fn(), - writeJson: vi.fn(), - }; - - const delivered = await deliverAgentCommandResult({ - cfg: {} as OpenClawConfig, - deps: {} as CliDeps, - runtime: runtime as never, - opts: { - message: "test", - json: true, - resultMetaOverrides: { - transport: "embedded", - fallbackFrom: "gateway", - }, - } as AgentCommandOpts, - outboundSession: undefined, - sessionEntry: undefined, - payloads: [{ text: "local" }], - result: createResult(), - }); - - expect(runtime.log).not.toHaveBeenCalled(); - expect(runtime.writeJson).toHaveBeenCalledWith( - { - payloads: [{ text: "local", mediaUrl: null }], - meta: { - durationMs: 1, - transport: "embedded", - fallbackFrom: "gateway", - }, - }, - 2, - ); - expect(delivered.meta.durationMs).toBe(1); - expect(delivered.meta.transport).toBe("embedded"); - expect(delivered.meta.fallbackFrom).toBe("gateway"); - }); - it("preserves committed message-tool delivery evidence when automatic delivery is disabled", async () => { const runtime = { log: vi.fn(), error: vi.fn() }; diff --git a/src/agents/command/delivery.ts b/src/agents/command/delivery.ts index 308c1d72596e..ad2738743fdd 100644 --- a/src/agents/command/delivery.ts +++ b/src/agents/command/delivery.ts @@ -50,7 +50,7 @@ import type { MessagingToolSend } from "../embedded-agent-messaging.types.js"; import type { EmbeddedAgentRunMeta } from "../embedded-agent-runner/types.js"; import { isNestedAgentLane } from "../lanes.js"; import { isAgentRunRestartAbortReason } from "../run-termination.js"; -import type { AgentCommandOpts, AgentCommandResultMetaOverrides } from "./types.js"; +import type { AgentCommandOpts } from "./types.js"; type RunResult = Awaited>; type DurableSendResult = Awaited>; @@ -98,7 +98,7 @@ type AgentCommandDeliveryStatus = { /** Agent command result after payload normalization and optional delivery. */ type AgentCommandDeliveryResult = { payloads: ReturnType; - meta: EmbeddedAgentRunMeta & AgentCommandResultMetaOverrides; + meta: EmbeddedAgentRunMeta; didSendViaMessagingTool?: boolean; messagingToolSentTexts?: string[]; messagingToolSentMediaUrls?: string[]; @@ -184,19 +184,6 @@ function logNestedOutput( } } -function mergeResultMetaOverrides( - meta: EmbeddedAgentRunMeta, - overrides: AgentCommandResultMetaOverrides | undefined, -): EmbeddedAgentRunMeta & AgentCommandResultMetaOverrides { - if (!overrides) { - return meta; - } - return { - ...meta, - ...overrides, - }; -} - function hasNonEmptyString(value: unknown): value is string { return typeof value === "string" && value.trim().length > 0; } @@ -852,7 +839,6 @@ export async function deliverAgentCommandResult( params.assertDeliveryCurrent?.(); const outboundPayloadPlan = createOutboundPayloadPlan(mediaNormalizedReplyPayloads); const normalizedPayloads = projectOutboundPayloadPlanForJson(outboundPayloadPlan); - const resultMeta = mergeResultMetaOverrides(result.meta, opts.resultMetaOverrides); const captureDeliveryResult = ( deliveryResult: AgentCommandDeliveryResult, ): AgentCommandDeliveryResult => { @@ -866,7 +852,7 @@ export async function deliverAgentCommandResult( writeRuntimeJson(runtime, { ...buildOutboundResultEnvelope({ payloads: normalizedPayloads, - meta: resultMeta, + meta: result.meta, }), ...(status ? { deliveryStatus: status } : {}), }); @@ -876,7 +862,7 @@ export async function deliverAgentCommandResult( captureDeliveryResult( buildDeliveryResult({ payloads: normalizedPayloads, - meta: resultMeta, + meta: result.meta, result, deliveryStatus, }), @@ -892,7 +878,7 @@ export async function deliverAgentCommandResult( return captureDeliveryResult( buildDeliveryResult({ payloads: normalizedPayloads, - meta: resultMeta, + meta: result.meta, result, deliverySucceeded, deliveryStatus, @@ -921,7 +907,7 @@ export async function deliverAgentCommandResult( } emitJsonEnvelope(); return captureDeliveryResult( - buildDeliveryResult({ payloads: normalizedPayloads, meta: resultMeta, result }), + buildDeliveryResult({ payloads: normalizedPayloads, meta: result.meta, result }), ); } if (deliver && deliveryChannel && !isInternalMessageChannel(deliveryChannel)) { @@ -959,7 +945,7 @@ export async function deliverAgentCommandResult( captureDeliveryResult( buildDeliveryResult({ payloads: normalizedPayloads, - meta: resultMeta, + meta: result.meta, result, deliverySucceeded: false, deliveryStatus, @@ -989,7 +975,7 @@ export async function deliverAgentCommandResult( return captureDeliveryResult( buildDeliveryResult({ payloads: normalizedPayloads, - meta: resultMeta, + meta: result.meta, result, deliverySucceeded, deliveryStatus, diff --git a/src/agents/command/types.ts b/src/agents/command/types.ts index 43ae01ba268f..d5a0a40f7a02 100644 --- a/src/agents/command/types.ts +++ b/src/agents/command/types.ts @@ -28,20 +28,6 @@ export type ImageContent = { mimeType: string; }; -/** Metadata overrides for trusted internal agent command callers. */ -export type AgentCommandResultMetaOverrides = { - transport?: "embedded"; - fallbackFrom?: "gateway"; - fallbackReason?: "gateway_timeout" | "gateway_closed"; - fallbackSessionId?: string; - fallbackSessionKey?: string; - fallback?: { - reason: "gateway_timeout" | "gateway_closed"; - requestedSessionKey: string | null; - sessionKey: string; - }; -}; - /** ACP turn source markers accepted by trusted command callsites. */ type AcpTurnSource = "manual_spawn"; @@ -189,8 +175,6 @@ export type AgentCommandOpts = { mainRestartRecoveryOwnerLease?: MainSessionRecoveryOwnerLease; /** Gateway already consumed this automatic recovery run's durable reservation. */ mainRestartRecoveryAdmitted?: boolean; - /** Internal local CLI callers can annotate result metadata before JSON/text output. */ - resultMetaOverrides?: AgentCommandResultMetaOverrides; /** Called when the actual run model is selected, including fallback retries. */ onActiveModelSelected?: (ctx: { provider: string; model: string }) => void | Promise; /** Called when compaction rotates the active run onto a successor session. */ @@ -210,7 +194,7 @@ export type AgentCommandOpts = { /** Restricted option surface for external ingress callsites. */ export type AgentCommandIngressOpts = Omit< AgentCommandOpts, - "senderIsOwner" | "allowModelOverride" | "resultMetaOverrides" + "senderIsOwner" | "allowModelOverride" > & { /** Trusted sender identity bit for command/channel-action auth; defaults false for ingress. */ senderIsOwner?: boolean; diff --git a/src/cli/run-main.exit.test.ts b/src/cli/run-main.exit.test.ts index b6f0debd26c1..3cecb9703918 100644 --- a/src/cli/run-main.exit.test.ts +++ b/src/cli/run-main.exit.test.ts @@ -1560,7 +1560,7 @@ describe("runCli exit behavior", () => { }); it("drops gateway.env selectors when the default state dotenv selects a custom state", async () => { - const homeDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-gateway-fallback-hop-")); + const homeDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-gateway-env-hop-")); const defaultStateDir = path.join(homeDir, ".openclaw"); const selectedStateDir = path.join(homeDir, "selected-state"); const gatewayEnvDir = path.join(homeDir, ".config", "openclaw"); @@ -1612,7 +1612,7 @@ describe("runCli exit behavior", () => { }); it("preserves gateway.env selectors when the compatibility fallback selects the target", async () => { - const homeDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-gateway-fallback-select-")); + const homeDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-gateway-env-select-")); const selectedStateDir = path.join(homeDir, "selected-state"); const gatewayEnvDir = path.join(homeDir, ".config", "openclaw"); await fs.mkdir(selectedStateDir, { recursive: true }); diff --git a/src/commands/agent-via-gateway.test.ts b/src/commands/agent-via-gateway.test.ts index 5480097cc924..5cf7c910de70 100644 --- a/src/commands/agent-via-gateway.test.ts +++ b/src/commands/agent-via-gateway.test.ts @@ -1519,34 +1519,7 @@ describe("agentCliCommand", () => { }); }); - it("exits for embedded fallback runs that resolve after SIGTERM aborts them", async () => { - await withTempStore(async () => { - const signals = createSignalProcess(); - callGateway.mockRejectedValueOnce(createGatewayClosedError()); - let resolveFallback: ((value: Awaited>) => void) | undefined; - agentCommand.mockImplementationOnce(async (_opts: { abortSignal?: AbortSignal }) => { - return await new Promise((resolve) => { - resolveFallback = resolve; - }); - }); - - const run = agentCliCommand({ message: "hi", to: "+1555" }, runtime, { - process: signals.processLike, - }); - await waitForAgentCommandCall(); - signals.emit("SIGTERM"); - resolveFallback?.({ - payloads: [], - meta: { aborted: true }, - } as unknown as Awaited>); - - await expect(run).resolves.toBeUndefined(); - expect(callGateway).toHaveBeenCalledTimes(1); - expect(runtime.exit).toHaveBeenCalledWith(143); - }); - }); - - it("does not route abort errors through embedded gateway fallback classification", async () => { + it("does not classify abort errors as gateway transport failures", async () => { await withTempStore(async () => { const err = new Error("gateway request aborted for agent"); err.name = "AbortError"; @@ -1714,59 +1687,37 @@ describe("agentCliCommand", () => { }); }); - it("falls back to embedded agent when gateway fails", async () => { + it("rejects gateway timeout errors unchanged with a local retry hint", async () => { await withTempStore(async () => { - callGateway.mockRejectedValue(createGatewayClosedError()); - mockLocalAgentReply(); + const error = createGatewayTimeoutError(); + callGateway.mockRejectedValue(error); - await agentCliCommand({ message: "hi", to: "+1555" }, runtime); + await expect(agentCliCommand({ message: "hi", to: "+1555" }, runtime)).rejects.toBe(error); expect(callGateway).toHaveBeenCalledTimes(1); - expect(agentCommand).toHaveBeenCalledTimes(1); - const fallbackOpts = requireRecord( - requireFirstCallArg(agentCommand, "embedded agent"), - "embedded agent options", - ); - const resultMetaOverrides = requireRecord( - fallbackOpts.resultMetaOverrides, - "fallback metadata", - ); - expect(resultMetaOverrides.transport).toBe("embedded"); - expect(resultMetaOverrides.fallbackFrom).toBe("gateway"); - expect(resultMetaOverrides.fallbackReason).toBe("gateway_closed"); + expect(agentCommand).not.toHaveBeenCalled(); expect( - mockMessages(runtime.error).some((message) => - message.includes( - "Gateway agent connection closed; continuity was intentionally not preserved", - ), + mockMessages(runtime.error).some( + (message) => + message.includes("Gateway agent call timed out") && message.includes("--local"), ), ).toBe(true); - expect(runtime.log).toHaveBeenCalledWith("local"); }); }); - it("retries transient normal gateway closes before embedded fallback", async () => { + it("retries transient normal gateway closes before failing", async () => { vi.useFakeTimers(); try { await withTempStore(async () => { - callGateway - .mockRejectedValueOnce(createGatewayNormalCloseError()) - .mockRejectedValueOnce(createGatewayNormalCloseError()) - .mockResolvedValue({ - runId: "idem-1", - status: "ok", - result: { - payloads: [{ text: "remote" }], - meta: { stub: true }, - }, - }); + const error = createGatewayNormalCloseError(); + callGateway.mockRejectedValue(error); const command = agentCliCommand({ message: "hi", to: "+1555" }, runtime); - await vi.advanceTimersByTimeAsync(1_000); - await vi.advanceTimersByTimeAsync(2_000); - await command; + const rejection = expect(command).rejects.toBe(error); + await vi.advanceTimersByTimeAsync(33_000); + await rejection; - expect(callGateway).toHaveBeenCalledTimes(3); + expect(callGateway).toHaveBeenCalledTimes(6); const idempotencyKeys = callGateway.mock.calls.map( ([call]) => (call as { params?: { idempotencyKey?: unknown } }).params?.idempotencyKey, ); @@ -1776,313 +1727,53 @@ describe("agentCliCommand", () => { mockMessages(runtime.error).filter((message) => message.includes("Gateway agent connection closed during handshake"), ), - ).toHaveLength(2); - expect(runtime.log).toHaveBeenCalledWith("remote"); + ).toHaveLength(5); + expect( + mockMessages(runtime.error).some( + (message) => + message.includes("Gateway agent call connection closed") && + message.includes("--local"), + ), + ).toBe(true); }); } finally { vi.useRealTimers(); } }); - it("uses a fresh embedded session when an accepted gateway turn closes abnormally", async () => { + it("rejects transport-closed errors unchanged with a local retry hint", async () => { await withTempStore(async () => { - callGateway.mockImplementationOnce(async (requestValue: unknown) => { - const request = requireRecord(requestValue, "gateway request"); - const onAccepted = request.onAccepted as ((payload: unknown) => void) | undefined; - onAccepted?.({ - status: "accepted", - runId: "accepted-run", - sessionKey: "agent:main:incident-42", - }); - throw createGatewayClosedError(); - }); - mockLocalAgentReply(); + const error = createGatewayClosedError(); + callGateway.mockRejectedValue(error); - await agentCliCommand({ message: "hi", sessionKey: "agent:main:incident-42" }, runtime); - - expect(callGateway).toHaveBeenCalledTimes(1); - expect(agentCommand).toHaveBeenCalledTimes(1); - const fallbackOpts = requireRecord( - requireFirstCallArg(agentCommand, "embedded agent"), - "embedded agent options", - ); - const fallbackSessionId = String(fallbackOpts.sessionId); - const fallbackSessionKey = String(fallbackOpts.sessionKey); - expect(fallbackSessionId).toMatch(/^gateway-fallback-/); - expect(fallbackSessionKey).toBe(`agent:main:explicit:${fallbackSessionId}`); - expect(fallbackSessionKey).not.toBe("agent:main:incident-42"); - expect(fallbackOpts.runId).toBe(fallbackSessionId); - expect(fallbackOpts.resultMetaOverrides).toMatchObject({ - transport: "embedded", - fallbackFrom: "gateway", - fallbackReason: "gateway_closed", - fallbackSessionId, - fallbackSessionKey, - }); - }); - }); - - it("does not pass a harness-owned session key into closed gateway fallback", async () => { - await withTempStore(async () => { - const sessionKey = "agent:main:harness:codex:supervision:missing-fallback"; - callGateway.mockRejectedValue(createGatewayClosedError()); - mockLocalAgentReply(); - - await agentCliCommand({ message: "hi", sessionKey }, runtime); - - expect(callGateway).toHaveBeenCalledOnce(); - expect(agentCommand).toHaveBeenCalledOnce(); - const fallbackOpts = requireRecord( - requireFirstCallArg(agentCommand, "embedded agent"), - "options", - ); - const fallbackSessionId = String(fallbackOpts.sessionId); - expect(fallbackOpts.sessionKey).toBe(`agent:main:explicit:${fallbackSessionId}`); - expect(fallbackOpts.sessionKey).not.toBe(sessionKey); - }); - }); - - it("does not fall back to embedded agent for gateway request errors", async () => { - await withTempStore(async () => { - callGateway.mockRejectedValue( - Object.assign(new Error("missing scope: operator.admin"), { - name: "GatewayClientRequestError", - gatewayCode: "INVALID_REQUEST", - }), - ); - - await expect(agentCliCommand({ message: "hi", to: "+1555" }, runtime)).rejects.toThrow( - "missing scope: operator.admin", - ); + await expect(agentCliCommand({ message: "hi", to: "+1555" }, runtime)).rejects.toBe(error); expect(callGateway).toHaveBeenCalledTimes(1); expect(agentCommand).not.toHaveBeenCalled(); expect( - mockMessages(runtime.error).some((message) => message.includes("EMBEDDED FALLBACK")), - ).toBe(false); + mockMessages(runtime.error).some( + (message) => + message.includes("Gateway agent call connection closed") && message.includes("--local"), + ), + ).toBe(true); }); }); - it("uses a fresh embedded session when gateway agent times out", async () => { + it("rejects non-transport errors without a local retry hint", async () => { await withTempStore(async () => { - callGateway.mockRejectedValue(createGatewayTimeoutError()); - mockLocalAgentReply(); + const error = Object.assign(new Error("missing scope: operator.admin"), { + name: "GatewayClientRequestError", + gatewayCode: "INVALID_REQUEST", + }); + callGateway.mockRejectedValue(error); - await agentCliCommand( - { - message: "hi", - sessionId: "locked-session", - runId: "locked-run", - }, - runtime, - ); + await expect(agentCliCommand({ message: "hi", to: "+1555" }, runtime)).rejects.toBe(error); expect(callGateway).toHaveBeenCalledTimes(1); - expect(agentCommand).toHaveBeenCalledTimes(1); - const fallbackOpts = requireRecord( - requireFirstCallArg(agentCommand, "embedded agent"), - "embedded agent options", + expect(agentCommand).not.toHaveBeenCalled(); + expect(mockMessages(runtime.error).some((message) => message.includes("--local"))).toBe( + false, ); - const fallbackSessionId = String(fallbackOpts.sessionId); - const fallbackSessionKey = String(fallbackOpts.sessionKey); - expect(fallbackSessionId).toMatch(/^gateway-fallback-/); - expect(fallbackSessionId).not.toBe("locked-session"); - expect(fallbackSessionKey).toBe(`agent:main:explicit:${fallbackSessionId}`); - expect(fallbackOpts.runId).toBe(fallbackSessionId); - const resultMetaOverrides = requireRecord( - fallbackOpts.resultMetaOverrides, - "fallback metadata", - ); - expect(resultMetaOverrides.transport).toBe("embedded"); - expect(resultMetaOverrides.fallbackFrom).toBe("gateway"); - expect(resultMetaOverrides.fallbackReason).toBe("gateway_timeout"); - expect(resultMetaOverrides.fallbackSessionId).toBe(fallbackSessionId); - expect(resultMetaOverrides.fallbackSessionKey).toBe(fallbackSessionKey); - expect( - mockMessages(runtime.error).some((message) => - message.includes("Gateway agent timed out; continuity was intentionally not preserved"), - ), - ).toBe(true); - expect(runtime.log).toHaveBeenCalledWith("local"); - }); - }); - - it("uses the explicit session key agent for timeout fallback sessions", async () => { - await withTempStore(async () => { - callGateway.mockRejectedValue(createGatewayTimeoutError()); - mockLocalAgentReply(); - - await agentCliCommand({ message: "hi", sessionKey: "agent:ops:incident-42" }, runtime); - - expect(agentCommand).toHaveBeenCalledTimes(1); - const fallbackOpts = requireFirstCallArg(agentCommand, "embedded agent") as { - sessionId?: string; - sessionKey?: string; - }; - expect(fallbackOpts.sessionId).toMatch(/^gateway-fallback-/); - expect(fallbackOpts.sessionKey).toBe(`agent:ops:explicit:${fallbackOpts.sessionId}`); - }); - }); - - it("uses the default-scoped legacy session key agent for timeout fallback sessions", async () => { - await withTempStore( - async () => { - callGateway.mockRejectedValue(createGatewayTimeoutError()); - mockLocalAgentReply(); - - await agentCliCommand({ message: "hi", sessionKey: "incident-42" }, runtime); - - expect(agentCommand).toHaveBeenCalledTimes(1); - const fallbackOpts = requireFirstCallArg(agentCommand, "embedded agent") as { - sessionId?: string; - sessionKey?: string; - }; - expect(fallbackOpts.sessionId).toMatch(/^gateway-fallback-/); - expect(fallbackOpts.sessionKey).toBe(`agent:ops:explicit:${fallbackOpts.sessionId}`); - expect(loadConfig.mock.calls).toEqual([[], []]); - }, - { agents: { list: [{ id: "ops", default: true }, { id: "main" }] } }, - ); - }); - - it("uses the default agent for timeout fallback with unscoped global session keys", async () => { - await withTempStore( - async () => { - callGateway.mockRejectedValue(createGatewayTimeoutError()); - mockLocalAgentReply(); - - await agentCliCommand({ message: "hi", sessionKey: "global" }, runtime); - - expect(agentCommand).toHaveBeenCalledTimes(1); - const fallbackOpts = requireFirstCallArg(agentCommand, "embedded agent") as { - sessionId?: string; - sessionKey?: string; - }; - expect(fallbackOpts.sessionId).toMatch(/^gateway-fallback-/); - expect(fallbackOpts.sessionKey).toBe(`agent:ops:explicit:${fallbackOpts.sessionId}`); - }, - { agents: { list: [{ id: "ops", default: true }, { id: "main" }] } }, - ); - }); - - it("keeps timeout fallback from replacing the routed conversation session key", async () => { - await withTempStore(async () => { - callGateway.mockRejectedValue(createGatewayTimeoutError()); - mockLocalAgentReply(); - - await agentCliCommand( - { - message: "hi", - to: "+1555", - }, - runtime, - ); - - const fallbackOpts = requireFirstCallArg(agentCommand, "embedded agent") as { - sessionId?: string; - sessionKey?: string; - to?: string; - }; - expect(fallbackOpts.to).toBe("+1555"); - expect(fallbackOpts.sessionId).toMatch(/^gateway-fallback-/); - expect(fallbackOpts.sessionKey).toBe(`agent:main:explicit:${fallbackOpts.sessionId}`); - expect(fallbackOpts.sessionKey).not.toBe("agent:main:+1555"); - }); - }); - - it.each([ - { - label: "connection-closed", - createError: createGatewayClosedError, - reason: "gateway_closed" as const, - }, - { - label: "timeout", - createError: createGatewayTimeoutError, - reason: "gateway_timeout" as const, - }, - ])("surfaces $label session divergence in JSON fallback output", async (testCase) => { - await withTempStore(async () => { - const requestedSessionKey = "agent:main:incident-42"; - callGateway.mockRejectedValue(testCase.createError()); - agentCommand.mockImplementationOnce(async (opts, rt) => { - expect(loggingState.forceConsoleToStderr).toBe(true); - const resultMetaOverrides = (opts as { resultMetaOverrides?: Record }) - .resultMetaOverrides; - const meta = { - durationMs: 1, - agentMeta: { sessionId: "s", provider: "p", model: "m" }, - ...resultMetaOverrides, - }; - rt?.log?.( - JSON.stringify( - { - payloads: [{ text: "local" }], - meta, - }, - null, - 2, - ), - ); - return { - payloads: [{ text: "local" }], - meta, - } as unknown as Awaited>; - }); - - const result = await agentCliCommand( - { message: "hi", sessionKey: requestedSessionKey, json: true }, - jsonRuntime, - ); - - expect(agentCommand).toHaveBeenCalledTimes(1); - const fallbackOpts = requireRecord( - requireFirstCallArg(agentCommand, "embedded agent"), - "embedded agent options", - ); - const resultMetaOverrides = requireRecord( - fallbackOpts.resultMetaOverrides, - "fallback metadata", - ); - const fallbackSessionKey = String(fallbackOpts.sessionKey); - expect(resultMetaOverrides.transport).toBe("embedded"); - expect(resultMetaOverrides.fallbackFrom).toBe("gateway"); - expect(resultMetaOverrides.fallback).toEqual({ - reason: testCase.reason, - requestedSessionKey, - sessionKey: fallbackSessionKey, - }); - expect( - mockMessages(jsonRuntime.error).some((message) => - message.includes( - "continuity was intentionally not preserved. Running embedded agent with fresh session", - ), - ), - ).toBe(true); - expect(loggingState.forceConsoleToStderr).toBe(true); - expect(jsonRuntime.log).toHaveBeenCalledTimes(1); - const jsonPayload = requireFirstCallArg(jsonRuntime.log, "json runtime log"); - const payload = requireRecord(JSON.parse(String(jsonPayload)), "json log payload"); - expect(payload.payloads).toEqual([{ text: "local" }]); - const payloadMeta = requireRecord(payload.meta, "json log metadata"); - expect(payloadMeta.durationMs).toBe(1); - expect(payloadMeta.transport).toBe("embedded"); - expect(payloadMeta.fallbackFrom).toBe("gateway"); - expect(payloadMeta.fallback).toEqual({ - reason: testCase.reason, - requestedSessionKey, - sessionKey: fallbackSessionKey, - }); - const resultRecord = requireRecord(result, "command result"); - const resultMeta = requireRecord(resultRecord.meta, "command result metadata"); - expect(resultMeta.durationMs).toBe(1); - expect(resultMeta.transport).toBe("embedded"); - expect(resultMeta.fallbackFrom).toBe("gateway"); - expect(resultMeta.fallback).toEqual({ - reason: testCase.reason, - requestedSessionKey, - sessionKey: fallbackSessionKey, - }); }); }); @@ -2108,7 +1799,6 @@ describe("agentCliCommand", () => { expect(localOpts.cleanupBundleMcpOnRunEnd).toBe(true); expect(localOpts.cleanupCliLiveSessionOnRunEnd).toBe(true); expect(localOpts.oneShotCliRun).toBe(true); - expect(localOpts).not.toHaveProperty("resultMetaOverrides"); expect(runtime.log).toHaveBeenCalledWith("local"); }); }); @@ -2200,7 +1890,7 @@ describe("agentCliCommand", () => { }); }); - it("rejects malformed agent-prefixed session keys before gateway or local fallback", async () => { + it("rejects malformed agent-prefixed session keys before gateway or local dispatch", async () => { await withTempStore(async () => { await expect( agentCliCommand({ message: "hi", sessionKey: "agent:main" }, runtime), @@ -2227,24 +1917,6 @@ describe("agentCliCommand", () => { }); }); - it("forces bundle MCP cleanup on embedded fallback", async () => { - await withTempStore(async () => { - callGateway.mockRejectedValue(createGatewayClosedError()); - mockLocalAgentReply(); - - await agentCliCommand({ message: "hi", to: "+1555" }, runtime); - - expect(agentCommand).toHaveBeenCalledTimes(1); - const fallbackOpts = requireRecord( - requireFirstCallArg(agentCommand, "embedded agent"), - "embedded agent options", - ); - expect(fallbackOpts.cleanupBundleMcpOnRunEnd).toBe(true); - expect(fallbackOpts.cleanupCliLiveSessionOnRunEnd).toBe(true); - expect(fallbackOpts.oneShotCliRun).toBe(false); - }); - }); - for (const message of [ "/compact", "/compact Keep recent decisions.", @@ -2264,13 +1936,12 @@ describe("agentCliCommand", () => { // The slash-command handler rejects CLI senders, so a /compact turn would // otherwise fall through to a normal turn and exit 0 without compacting. - // It must fail loudly before touching the gateway or a fresh embedded session. + // It must fail loudly before touching the gateway or local agent. expect(callGateway).not.toHaveBeenCalled(); expect(agentCommand).not.toHaveBeenCalled(); expect(runtime.exit).toHaveBeenCalledWith(1); const errorMessages = mockMessages(runtime.error); expect(errorMessages.some((m) => m.includes("openclaw sessions compact"))).toBe(true); - expect(errorMessages.some((m) => m.includes("EMBEDDED FALLBACK"))).toBe(false); }); } @@ -2291,7 +1962,6 @@ describe("agentCliCommand", () => { expect(runtime.exit).toHaveBeenCalledWith(1); const errorMessages = mockMessages(runtime.error); expect(errorMessages.some((m) => m.includes("openclaw sessions compact"))).toBe(true); - expect(errorMessages.some((m) => m.includes("EMBEDDED FALLBACK"))).toBe(false); }); it("does not mistake a /compacting-prefixed message for the /compact control command", async () => { diff --git a/src/commands/agent-via-gateway.ts b/src/commands/agent-via-gateway.ts index 086581dc7b91..ed8a172ea383 100644 --- a/src/commands/agent-via-gateway.ts +++ b/src/commands/agent-via-gateway.ts @@ -1,5 +1,4 @@ -import { randomUUID } from "node:crypto"; -// Gateway-first agent CLI implementation with embedded fallback for local/runtime failures. +// Gateway-first agent CLI implementation with explicit --local embedded execution. import fs from "node:fs/promises"; import { TextDecoder } from "node:util"; import { resolveTimerTimeoutMs } from "@openclaw/normalization-core/number-coercion"; @@ -61,11 +60,6 @@ type GatewayAgentResponse = { }; const NO_GATEWAY_TIMEOUT_MS = 2_147_000_000; -const EMBEDDED_FALLBACK_META = { - transport: "embedded", - fallbackFrom: "gateway", -} as const; -const GATEWAY_FALLBACK_SESSION_PREFIX = "gateway-fallback-"; const GATEWAY_TRANSIENT_CONNECT_RETRY_DELAYS_MS = [1_000, 2_000, 5_000, 10_000, 15_000] as const; type AgentCliOpts = { @@ -334,15 +328,13 @@ function shouldRetryGatewayDispatchWithShellEnvFallback(err: unknown): boolean { ); } -function resolveGatewayAgentEmbeddedFallbackReason( +function resolveGatewayAgentFailureHint( err: unknown, -): "gateway_timeout" | "gateway_closed" | undefined { +): "timed out" | "connection closed" | undefined { if (isGatewayAgentTimeoutError(err)) { - return "gateway_timeout"; + return "timed out"; } - // GatewayTransportErrorKind is exactly "closed" | "timeout", so this branch - // pair covers every transport error; non-transport errors never fell back. - return isGatewayTransportError(err) && err.kind === "closed" ? "gateway_closed" : undefined; + return isGatewayTransportError(err) && err.kind === "closed" ? "connection closed" : undefined; } function isTransientGatewayAgentConnectClose(err: unknown): boolean { @@ -645,52 +637,6 @@ function returnAfterSignalExit( return exitForReceivedSignal(signal, runtime) ? undefined : value; } -function createGatewayFallbackSessionId(): string { - return `${GATEWAY_FALLBACK_SESSION_PREFIX}${randomUUID()}`; -} - -function createGatewayFallbackSession(agentId?: string): { - sessionId: string; - sessionKey: string; -} { - const sessionId = createGatewayFallbackSessionId(); - return { - sessionId, - sessionKey: `agent:${normalizeAgentId(agentId)}:explicit:${sessionId.trim()}`, - }; -} - -async function resolveAgentIdForGatewayFallback( - opts: AgentDispatchOpts, -): Promise { - const explicitSessionKey = opts.sessionKey?.trim(); - if (classifySessionKeyShape(explicitSessionKey) === "agent") { - return resolveAgentIdFromSessionKey(explicitSessionKey); - } - if (isUnscopedSessionKeySentinel(explicitSessionKey)) { - return resolveDefaultAgentId(await getGatewayDispatchConfig()); - } - - const agentIdRaw = opts.agent?.trim(); - if (agentIdRaw) { - return normalizeAgentId(agentIdRaw); - } - - if (!opts.to && !opts.sessionId) { - return undefined; - } - const cfg = await getGatewayDispatchConfig(); - const { resolveSessionKeyForRequest } = await loadAgentSessionModule(); - const resolvedSessionKey = resolveSessionKeyForRequest({ - cfg, - to: opts.to, - sessionId: opts.sessionId, - }).sessionKey; - return classifySessionKeyShape(resolvedSessionKey) === "agent" - ? resolveAgentIdFromSessionKey(resolvedSessionKey) - : undefined; -} - function buildGatewayJsonResponse(response: GatewayAgentResponse): GatewayAgentResponse { const deliveryStatus = response.result?.deliveryStatus; if (deliveryStatus === undefined) { @@ -934,7 +880,7 @@ async function agentViaGatewayCommandWithTransientRetries( throw err; } runtime.error?.( - `Gateway agent connection closed during handshake; retrying in ${retryDelayMs}ms before embedded fallback.`, + `Gateway agent connection closed during handshake; retrying in ${retryDelayMs}ms before failing.`, ); await delayMs(retryDelayMs, signalBridge.signal); } @@ -960,26 +906,28 @@ export async function agentCliCommand( runtime.exit(1); return undefined; } - const requestedSessionKey = messageOpts.sessionKey?.trim() || null; const dispatchOpts = await normalizeSessionKeyOptsForDispatch(messageOpts); validateExplicitSessionKeyForDispatch(dispatchOpts); const gatewayDispatchOpts = dispatchOpts.runId ? dispatchOpts : { ...dispatchOpts, runId: randomIdempotencyKey() }; const signalBridge = createAgentCliSignalBridge(resolveAgentCliProcessLike(deps)); - const localOpts = { - ...gatewayDispatchOpts, - agentId: gatewayDispatchOpts.agent, - replyAccountId: gatewayDispatchOpts.replyAccount, - cleanupBundleMcpOnRunEnd: true, - cleanupCliLiveSessionOnRunEnd: true, - oneShotCliRun: dispatchOpts.local === true, - abortSignal: signalBridge.signal, - }; try { if (dispatchOpts.local === true) { const agentCommand = await loadEmbeddedAgentCommand(); - const result = await agentCommand(localOpts, runtime, deps); + const result = await agentCommand( + { + ...gatewayDispatchOpts, + agentId: gatewayDispatchOpts.agent, + replyAccountId: gatewayDispatchOpts.replyAccount, + cleanupBundleMcpOnRunEnd: true, + cleanupCliLiveSessionOnRunEnd: true, + oneShotCliRun: true, + abortSignal: signalBridge.signal, + }, + runtime, + deps, + ); return returnAfterSignalExit(result, signalBridge.getReceivedSignal(), runtime); } @@ -997,41 +945,16 @@ export async function agentCliCommand( } throw err; } - const fallbackReason = resolveGatewayAgentEmbeddedFallbackReason(err); - if (!fallbackReason) { - throw err; + const failureHint = resolveGatewayAgentFailureHint(err); + if (failureHint) { + // Transport loss is ambiguous: the Gateway may have accepted and may still + // finish this turn. Recommending a blind retry or --local here could + // double-execute the message, so point at verification first. + runtime.error?.( + `Gateway agent call ${failureHint}; the Gateway may still be running this turn. Check \`openclaw gateway status\` and the session transcript before retrying or rerunning with --local, so the turn does not execute twice.`, + ); } - - const fallbackAgentId = await resolveAgentIdForGatewayFallback(dispatchOpts); - const fallbackSession = createGatewayFallbackSession(fallbackAgentId); - // Transport loss is ambiguous: the Gateway may still own or recover the original turn. - // Keep embedded work on a separate session so both processes cannot write one transcript. - runtime.error?.( - `EMBEDDED FALLBACK: Gateway agent ${fallbackReason === "gateway_timeout" ? "timed out" : "connection closed"}; continuity was intentionally not preserved. Running embedded agent with fresh session ${fallbackSession.sessionId} to avoid double-driving the original session: ${String(err)}`, - ); - const agentCommand = await loadEmbeddedAgentCommand(); - const result = await agentCommand( - { - ...localOpts, - sessionId: fallbackSession.sessionId, - sessionKey: fallbackSession.sessionKey, - runId: fallbackSession.sessionId, - resultMetaOverrides: { - ...EMBEDDED_FALLBACK_META, - fallbackReason, - fallbackSessionId: fallbackSession.sessionId, - fallbackSessionKey: fallbackSession.sessionKey, - fallback: { - reason: fallbackReason, - requestedSessionKey, - sessionKey: fallbackSession.sessionKey, - }, - }, - }, - runtime, - deps, - ); - return returnAfterSignalExit(result, signalBridge.getReceivedSignal(), runtime); + throw err; } } catch (err) { if (isAbortError(err) && exitForReceivedSignal(signalBridge.getReceivedSignal(), runtime)) {