Skip to content

Commit 4b7f193

Browse files
joshavantjoshavant
authored
fix: gateway-owned work fails through same-host relay (#108423)
* fix(gateway): dispatch owned runtime work in process * fix(gateway): publish canonical approval resolutions internally * fix(gateway): seal owned approval runtime lifecycle * fix(gateway): keep instance runtime contracts acyclic * fix(gateway): remove stale runtime type imports * fix(agents): reconcile ambiguous recovery timeouts * refactor(gateway): constrain internal approval methods * fix(agents): route orphan recovery in process * fix(agents): preserve aborted restores for recovery * fix(agents): keep lifecycle waits in process * test(agents): type lifecycle runtime mock * test(agents): complete restored session fixture
1 parent de3f792 commit 4b7f193

49 files changed

Lines changed: 1807 additions & 450 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎config/max-lines-baseline.txt‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -876,7 +876,6 @@ src/gateway/server-methods/usage.ts
876876
src/gateway/server-node-events.test.ts
877877
src/gateway/server-node-events.ts
878878
src/gateway/server-plugins.test.ts
879-
src/gateway/server-plugins.ts
880879
src/gateway/server-reload-handlers.test.ts
881880
src/gateway/server-reload-handlers.ts
882881
src/gateway/server-restart-sentinel.test.ts

‎packages/gateway-client/src/client.ts‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,11 @@ import {
4444
type GatewayProtocolSocketHandlers,
4545
} from "./protocol-client.js";
4646
import { shouldPauseGatewayReconnect } from "./reconnect-policy.js";
47-
import { resolveConnectChallengeTimeoutMs, resolveSafeTimeoutDelayMs } from "./timeouts.js";
47+
import {
48+
DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS,
49+
resolveConnectChallengeTimeoutMs,
50+
resolveSafeTimeoutDelayMs,
51+
} from "./timeouts.js";
4852
import { rawDataToString } from "./websocket-data.js";
4953

5054
export type DeviceIdentity = {
@@ -407,7 +411,7 @@ export class GatewayClient {
407411
this.requestTimeoutMs =
408412
typeof opts.requestTimeoutMs === "number" && Number.isFinite(opts.requestTimeoutMs)
409413
? resolveSafeTimeoutDelayMs(opts.requestTimeoutMs, { minMs: 0 })
410-
: 30_000;
414+
: DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS;
411415
this.protocol = new GatewayProtocolClient<AssembledConnect>({
412416
createSocket: (handlers) => this.createSocket(handlers),
413417
createRequestId: randomUUID,

‎packages/gateway-client/src/timeouts.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,8 @@ function parseStrictPositiveInteger(value: string): number | undefined {
1212
export const MAX_SAFE_TIMEOUT_DELAY_MS = 2_147_483_647;
1313
/** Default server-side window for gateway preauth handshakes. */
1414
export const DEFAULT_PREAUTH_HANDSHAKE_TIMEOUT_MS = 15_000;
15+
/** Default deadline for a single non-streaming Gateway request. */
16+
export const DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS = 30_000;
1517
/** Minimum client watchdog delay for connect challenge setup. */
1618
export const MIN_CONNECT_CHALLENGE_TIMEOUT_MS = 250;
1719
/** Default maximum client watchdog delay, aligned with the preauth server timeout. */

‎src/agents/main-session-restart-dispatch.ts‎

Lines changed: 25 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@ import type { SessionEntry } from "../config/sessions.js";
55
import { buildRestartRecoveryClaimCleanupPatch } from "../config/sessions/restart-recovery-state.js";
66
import { applySessionEntryReplacements } from "../config/sessions/session-accessor.js";
77
import type { OpenClawConfig } from "../config/types.openclaw.js";
8-
import { callGateway } from "../gateway/call.js";
8+
import type { GatewayRecoveryRuntime } from "../gateway/server-instance-runtime.types.js";
99
import { createSubsystemLogger } from "../logging/subsystem.js";
1010
import { CommandLane } from "../process/lanes.js";
1111
import { resolveSendPolicy } from "../sessions/send-policy.js";
@@ -86,13 +86,13 @@ function normalizeRestartRecoveryTerminalStatus(
8686

8787
async function probeRestartRecoveryTerminalStatus(
8888
runId: string,
89+
gatewayRuntime: GatewayRecoveryRuntime,
8990
): Promise<RestartRecoveryTerminalStatus | undefined> {
9091
try {
91-
const result = await callGateway<{ endedAt?: unknown; status?: unknown }>({
92-
method: "agent.wait",
93-
params: { runId, timeoutMs: 0 },
94-
timeoutMs: 2_000,
95-
});
92+
const result = await gatewayRuntime.waitForAgent<{ endedAt?: unknown; status?: unknown }>(
93+
{ runId, timeoutMs: 0 },
94+
2_000,
95+
);
9696
const status = normalizeRestartRecoveryTerminalStatus(result.status);
9797
// A zero-time wait also reports timeout for active or unknown work.
9898
return status === "timeout" && typeof result.endedAt !== "number" ? undefined : status;
@@ -189,6 +189,7 @@ export async function resumeMainSession(params: {
189189
pendingFinalDeliveryText?: string | null;
190190
forceRestartSafeTools?: boolean;
191191
sessionWorkAdmissionHandoffId?: string;
192+
gatewayRuntime: GatewayRecoveryRuntime;
192193
}): Promise<boolean> {
193194
const sanitizedPendingText =
194195
typeof params.pendingFinalDeliveryText === "string"
@@ -205,6 +206,7 @@ export async function resumeMainSession(params: {
205206
const reusingRecoveryRunId = recoveryRunId === claimedRunId;
206207
const dispatchSessionKey = params.canonicalSessionKey ?? params.sessionKey;
207208
const recoverySessionKeys = Array.from(new Set([dispatchSessionKey, params.sessionKey]));
209+
let dispatchOutcomeUnknown = false;
208210
try {
209211
// Persist one stable RPC id before dispatch. A transport rejection is
210212
// ambiguous; retries must reuse this id so accepted work cannot duplicate.
@@ -267,14 +269,20 @@ export async function resumeMainSession(params: {
267269
if (params.forceRestartSafeTools) {
268270
log.info(`dispatching restart-safe recovery for ${params.sessionKey}`);
269271
}
270-
const dispatchResult = await callGateway<{ runId: string; status?: unknown }>({
271-
method: "agent",
272-
params: agentParams,
273-
timeoutMs: 10_000,
274-
});
272+
// Once dispatch starts, any rejection is ambiguous because the stable RPC
273+
// may still have been accepted; a successful return resolves that ambiguity.
274+
dispatchOutcomeUnknown = true;
275+
const dispatchResult = await params.gatewayRuntime.dispatchAgent<{
276+
runId: string;
277+
status?: unknown;
278+
}>(agentParams, 10_000);
279+
dispatchOutcomeUnknown = false;
275280
let terminalStatus = normalizeRestartRecoveryTerminalStatus(dispatchResult.status);
276281
if (!terminalStatus && reusingRecoveryRunId && dispatchResult.status === "accepted") {
277-
terminalStatus = await probeRestartRecoveryTerminalStatus(recoveryRunId);
282+
terminalStatus = await probeRestartRecoveryTerminalStatus(
283+
recoveryRunId,
284+
params.gatewayRuntime,
285+
);
278286
}
279287
await settleRestartRecoveryDispatch({
280288
expectedRecoveryRunId: recoveryRunId,
@@ -292,12 +300,11 @@ export async function resumeMainSession(params: {
292300
);
293301
return true;
294302
} catch (error) {
295-
if (
296-
reusingRecoveryRunId &&
297-
error instanceof Error &&
298-
error.name === "GatewayClientRequestError"
299-
) {
300-
const terminalStatus = await probeRestartRecoveryTerminalStatus(recoveryRunId);
303+
if (reusingRecoveryRunId && dispatchOutcomeUnknown) {
304+
const terminalStatus = await probeRestartRecoveryTerminalStatus(
305+
recoveryRunId,
306+
params.gatewayRuntime,
307+
);
301308
if (terminalStatus) {
302309
await settleRestartRecoveryDispatch({
303310
expectedRecoveryRunId: recoveryRunId,

‎src/agents/main-session-restart-recovery.test.ts‎

Lines changed: 117 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ import {
1313
replaceSessionEntry,
1414
} from "../config/sessions/session-accessor.js";
1515
import { callGateway } from "../gateway/call.js";
16+
import type { GatewayRecoveryRuntime } from "../gateway/server-instance-runtime.types.js";
1617
import {
1718
getAgentEventLifecycleGeneration,
1819
registerAgentRunContext,
@@ -38,10 +39,10 @@ import {
3839
markRestartAbortedMainSessions,
3940
markRestartAbortedMainSessionsFromLocks,
4041
markStartupOrphanedMainSessionsForRecovery,
41-
recoverStartupOrphanedMainSessions,
42-
recoverRestartAbortedMainSessions,
43-
retryRestartAbortedMainSessionRecovery,
44-
scheduleRestartAbortedMainSessionRecovery,
42+
recoverStartupOrphanedMainSessions as recoverStartupOrphanedMainSessionsBase,
43+
recoverRestartAbortedMainSessions as recoverRestartAbortedMainSessionsBase,
44+
retryRestartAbortedMainSessionRecovery as retryRestartAbortedMainSessionRecoveryBase,
45+
scheduleRestartAbortedMainSessionRecovery as scheduleRestartAbortedMainSessionRecoveryBase,
4546
} from "./main-session-restart-recovery.js";
4647
import type { SessionLockInspection } from "./session-write-lock.js";
4748

@@ -53,6 +54,32 @@ vi.mock("../gateway/call.js", () => ({
5354
callGateway: vi.fn(async () => ({ runId: "run-resumed" })),
5455
}));
5556

57+
const mockRecoveryRuntime = {
58+
dispatchAgent: async <T>(params: Record<string, unknown>, timeoutMs?: number) =>
59+
(await callGateway({ method: "agent", params, timeoutMs })) as T,
60+
waitForAgent: async <T>(params: Record<string, unknown>, timeoutMs?: number) =>
61+
(await callGateway({ method: "agent.wait", params, timeoutMs })) as T,
62+
sendRecoveryNotice: async <T>(params: Record<string, unknown>, timeoutMs?: number) =>
63+
(await callGateway({ method: "message.action", params, timeoutMs })) as T,
64+
};
65+
66+
type RecoveryParams<T extends { gatewayRuntime: unknown }> = Omit<T, "gatewayRuntime"> &
67+
Partial<Pick<T, "gatewayRuntime">>;
68+
69+
const recoverRestartAbortedMainSessions = (
70+
params: RecoveryParams<Parameters<typeof recoverRestartAbortedMainSessionsBase>[0]>,
71+
) => recoverRestartAbortedMainSessionsBase({ gatewayRuntime: mockRecoveryRuntime, ...params });
72+
const recoverStartupOrphanedMainSessions = (
73+
params: RecoveryParams<Parameters<typeof recoverStartupOrphanedMainSessionsBase>[0]>,
74+
) => recoverStartupOrphanedMainSessionsBase({ gatewayRuntime: mockRecoveryRuntime, ...params });
75+
const retryRestartAbortedMainSessionRecovery = (
76+
params: RecoveryParams<Parameters<typeof retryRestartAbortedMainSessionRecoveryBase>[0]>,
77+
) => retryRestartAbortedMainSessionRecoveryBase({ gatewayRuntime: mockRecoveryRuntime, ...params });
78+
const scheduleRestartAbortedMainSessionRecovery = (
79+
params: RecoveryParams<Parameters<typeof scheduleRestartAbortedMainSessionRecoveryBase>[0]>,
80+
) =>
81+
scheduleRestartAbortedMainSessionRecoveryBase({ gatewayRuntime: mockRecoveryRuntime, ...params });
82+
5683
vi.mock("../config/sessions/transcript.js", async (importOriginal) => {
5784
const actual = await importOriginal<typeof import("../config/sessions/transcript.js")>();
5885
transcriptMocks.appendAssistantMessageToSessionTranscript.mockImplementation(
@@ -1024,6 +1051,50 @@ describe("main-session-restart-recovery", () => {
10241051
expect(settled?.restartRecoveryDeliverySourceRunId).toBeUndefined();
10251052
});
10261053

1054+
it("settles a reused recovery RPC after its dispatch wait times out", async () => {
1055+
const sessionsDir = await makeSessionsDir();
1056+
const storePath = path.join(sessionsDir, "sessions.json");
1057+
await writeStore(sessionsDir, {
1058+
"agent:main:main": {
1059+
sessionId: "main-session",
1060+
updatedAt: Date.now() - 10_000,
1061+
status: "running",
1062+
abortedLastRun: true,
1063+
restartRecoveryDeliveryRunId: "recovery-run",
1064+
restartRecoveryDeliverySourceRunId: "control-ui-run",
1065+
},
1066+
});
1067+
await writeTranscript(sessionsDir, "main-session", [
1068+
{ role: "user", content: "run the tool" },
1069+
{ role: "assistant", content: [{ type: "toolCall", id: "call-1", name: "exec" }] },
1070+
{ role: "toolResult", content: "done" },
1071+
]);
1072+
vi.mocked(callGateway)
1073+
.mockRejectedValueOnce(new Error("gateway request timeout for agent"))
1074+
.mockResolvedValueOnce({
1075+
runId: "recovery-run",
1076+
status: "ok",
1077+
endedAt: Date.now(),
1078+
});
1079+
1080+
await expect(recoverRestartAbortedMainSessions({ stateDir: tmpDir })).resolves.toEqual({
1081+
recovered: 1,
1082+
failed: 0,
1083+
skipped: 0,
1084+
});
1085+
1086+
expect(firstGatewayParams().idempotencyKey).toBe("recovery-run");
1087+
expect(vi.mocked(callGateway).mock.calls[1]?.[0]).toMatchObject({
1088+
method: "agent.wait",
1089+
params: { runId: "recovery-run", timeoutMs: 0 },
1090+
});
1091+
expect(loadSessionEntry({ sessionKey: "agent:main:main", storePath })).toMatchObject({
1092+
abortedLastRun: false,
1093+
restartRecoveryTerminalRunIds: ["control-ui-run"],
1094+
status: "done",
1095+
});
1096+
});
1097+
10271098
it("does not deliver restart recovery when session send policy denies sends", async () => {
10281099
const sessionsDir = await makeSessionsDir();
10291100
await writeStore(sessionsDir, {
@@ -1627,6 +1698,48 @@ describe("main-session-restart-recovery", () => {
16271698
});
16281699
});
16291700

1701+
it("dispatches an abandoned durable claim through its owning Gateway instance", async () => {
1702+
const sessionsDir = await makeSessionsDir();
1703+
const storePath = path.join(sessionsDir, "sessions.json");
1704+
await writeStore(sessionsDir, {
1705+
"agent:main:main": {
1706+
sessionId: "main-session",
1707+
updatedAt: Date.now() - 10_000,
1708+
status: "running",
1709+
abortedLastRun: true,
1710+
restartRecoveryDeliveryRunId: "recovery-main",
1711+
restartRecoveryDeliverySourceRunId: "source-main",
1712+
},
1713+
});
1714+
await writeTranscript(sessionsDir, "main-session", [
1715+
{ role: "user", content: "recover without a socket" },
1716+
]);
1717+
const dispatchAgent = vi.fn(async () => ({ runId: "recovery-main", status: "accepted" }));
1718+
1719+
const result = await retryRestartAbortedMainSessionRecovery({
1720+
expectedRecoveryRunId: "recovery-main",
1721+
expectedRecoverySourceRunId: "source-main",
1722+
expectedSessionId: "main-session",
1723+
sessionKey: "agent:main:main",
1724+
storePath,
1725+
gatewayRuntime: {
1726+
dispatchAgent: dispatchAgent as GatewayRecoveryRuntime["dispatchAgent"],
1727+
waitForAgent: vi.fn(),
1728+
sendRecoveryNotice: vi.fn(),
1729+
},
1730+
});
1731+
1732+
expect(result).toEqual({ recovered: 1, failed: 0, skipped: 0 });
1733+
expect(dispatchAgent).toHaveBeenCalledWith(
1734+
expect.objectContaining({
1735+
idempotencyKey: "recovery-main",
1736+
sessionKey: "agent:main:main",
1737+
}),
1738+
10_000,
1739+
);
1740+
expect(callGateway).not.toHaveBeenCalled();
1741+
});
1742+
16301743
it("targets a legacy durable row through its canonical agent session key", async () => {
16311744
const sessionsDir = await makeSessionsDir();
16321745
const storePath = path.join(sessionsDir, "sessions.json");
@@ -2208,13 +2321,9 @@ describe("main-session-restart-recovery", () => {
22082321
| {
22092322
method?: string;
22102323
params?: Record<string, unknown>;
2211-
clientName?: string;
2212-
mode?: string;
22132324
}
22142325
| undefined;
22152326
expect(gatewayCall?.method).toBe("message.action");
2216-
expect(gatewayCall?.clientName).toBe("gateway-client");
2217-
expect(gatewayCall?.mode).toBe("backend");
22182327
expect(gatewayCall?.params).toMatchObject({
22192328
channel: "discord",
22202329
action: "send",

0 commit comments

Comments
 (0)