mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-21 18:26:51 +00:00
refactor(cli)!: remove automatic gateway→embedded fallback from openclaw agent (#112074)
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.
This commit is contained in:
Peter Steinberger
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
<AccordionGroup>
|
||||
<Accordion title="Isolated run hardening">
|
||||
|
||||
+6
-7
@@ -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 <key>` 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:<agent-id>:<session-key>`, 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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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() };
|
||||
|
||||
|
||||
@@ -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<ReturnType<(typeof import("../embedded-agent.js"))["runEmbeddedAgent"]>>;
|
||||
type DurableSendResult = Awaited<ReturnType<typeof sendDurableMessageBatch>>;
|
||||
@@ -98,7 +98,7 @@ type AgentCommandDeliveryStatus = {
|
||||
/** Agent command result after payload normalization and optional delivery. */
|
||||
type AgentCommandDeliveryResult = {
|
||||
payloads: ReturnType<typeof projectOutboundPayloadPlanForJson>;
|
||||
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,
|
||||
|
||||
@@ -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<void>;
|
||||
/** 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;
|
||||
|
||||
@@ -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 });
|
||||
|
||||
@@ -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<ReturnType<typeof AgentCommand>>) => 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<ReturnType<typeof AgentCommand>>);
|
||||
|
||||
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<string, unknown> })
|
||||
.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<ReturnType<typeof AgentCommand>>;
|
||||
});
|
||||
|
||||
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 () => {
|
||||
|
||||
@@ -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<T>(
|
||||
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<string | undefined> {
|
||||
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)) {
|
||||
|
||||
Reference in New Issue
Block a user