diff --git a/CHANGELOG.md b/CHANGELOG.md index e310125df27b..d12b184746c9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -42,6 +42,7 @@ Docs: https://docs.openclaw.ai ### Fixes +- **Reef startup reconciliation:** contain retryable relay failures during startup without supervisor restart loops, while preserving definitive-error and cancellation handling. Thanks @Yigtwxx. - **Codex stale-session replies:** stop model fallback after another gateway supersedes a Codex session generation and deliver a safe retry notice instead of abandoning the message silently. - **Bounded input and provider responses:** cap pasted auth/config input and enforce wall-clock deadlines across generated-media downloads, polling JSON, and failed response details so oversized or slow-drip streams cannot exceed resource budgets (thanks @Pick-cat). - **Cloud worker derived workspace caches:** exclude Python caches, dependency trees, and macOS metadata symmetrically from outbound sync and inbound reconciliation so local cache rewrites cannot fence later cloud results or worker reclaim. diff --git a/extensions/reef/src/channel-lifecycle.ts b/extensions/reef/src/channel-lifecycle.ts index e8d5970fc8e2..829151b496a7 100644 --- a/extensions/reef/src/channel-lifecycle.ts +++ b/extensions/reef/src/channel-lifecycle.ts @@ -1,6 +1,27 @@ import { abortableSleep } from "./transport.js"; const REEF_RECONCILE_INTERVAL_MS = 30_000; +const CONTINUE_AFTER_RECONCILE_ERROR = () => true; +const STOP_AFTER_RECONCILE_ERROR = () => false; + +async function runReconcileStep(params: { + reconcile: () => Promise; + onReconcileError: (error: unknown) => void; + shouldContinueAfterError: (error: unknown) => boolean; + signal: AbortSignal; +}): Promise { + try { + await params.reconcile(); + } catch (error) { + if (params.signal.aborted) { + return; + } + if (!params.shouldContinueAfterError(error)) { + throw error; + } + params.onReconcileError(error); + } +} // One abort scope owns both account loops. If either branch throws, the inbox // loop must be torn down and awaited before startAccount settles: a leaked @@ -11,6 +32,12 @@ export async function runReefChannelLifecycle(params: { startInbox: (signal: AbortSignal) => Promise; reconcile: () => Promise; onReconcileError: (error: unknown) => void; + // Startup may continue only for errors the channel classifies as retryable. + // Periodic reconcile remains best-effort once the account is already active. + shouldContinueAfterStartupReconcileError?: (error: unknown) => boolean; + // Runs after the startup reconcile either refreshes peer keys or reports a + // classified retryable failure, before the inbox can dispatch a turn. + onReady?: () => Promise; reconcileIntervalMs?: number; }): Promise { const lifecycle = new AbortController(); @@ -27,18 +54,33 @@ export async function runReefChannelLifecycle(params: { if (lifecycle.signal.aborted) { return; } - try { - await params.reconcile(); - } catch (error) { - // Transient relay failures (429, network) must not crash the channel: - // the crash-restart cycle re-registers the inbox connection and is - // itself what escalates relay rate limiting. - params.onReconcileError(error); - } + await runReconcileStep({ + ...params, + shouldContinueAfterError: CONTINUE_AFTER_RECONCILE_ERROR, + signal: lifecycle.signal, + }); } }; - const inboxTask = params.startInbox(lifecycle.signal); + // Declared outside the try so the finally can await it even when the startup + // steps below throw before the inbox is started. + let inboxTask: Promise | undefined; try { + if (!lifecycle.signal.aborted) { + await runReconcileStep({ + ...params, + shouldContinueAfterError: + params.shouldContinueAfterStartupReconcileError ?? STOP_AFTER_RECONCILE_ERROR, + signal: lifecycle.signal, + }); + } + if (lifecycle.signal.aborted) { + return; + } + await params.onReady?.(); + if (lifecycle.signal.aborted) { + return; + } + inboxTask = params.startInbox(lifecycle.signal); await Promise.all([inboxTask, reconciliationLoop()]); } finally { lifecycle.abort(); diff --git a/extensions/reef/src/channel.test.ts b/extensions/reef/src/channel.test.ts index 16b00ac2e1e2..db3c694077bf 100644 --- a/extensions/reef/src/channel.test.ts +++ b/extensions/reef/src/channel.test.ts @@ -17,6 +17,16 @@ import { resolveReefInboundDispatchContent } from "./inbound.js"; import { setReefRuntime } from "./runtime.js"; import { openReefTrustStore } from "./trust-store.js"; +function deferred() { + let resolve!: () => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, resolve, reject }; +} + describe("Reef inbound dispatch content", () => { it("keeps provenance model-visible without storing it in the transcript body", () => { const content = resolveReefInboundDispatchContent({ @@ -126,6 +136,162 @@ describe("Reef channel lifecycle", () => { return { startInbox, seen, isSettled: () => settled }; } + it("activates and starts the inbox when the startup reconcile fails", async () => { + const parent = new AbortController(); + const inbox = hangingInbox(); + const errors: unknown[] = []; + let reconciles = 0; + // Captured inside onReady so the assertion pins the startup reconcile + // specifically, not "some reconcile eventually failed" once the periodic + // loop has had a chance to run. + let reconcilesAtActivation = -1; + let errorsAtActivation = -1; + const lifecycle = runReefChannelLifecycle({ + parentSignal: parent.signal, + startInbox: inbox.startInbox, + reconcile: async () => { + reconciles += 1; + throw new Error("rate_limited"); + }, + onReconcileError: (error) => errors.push(error), + shouldContinueAfterStartupReconcileError: () => true, + onReady: async () => { + reconcilesAtActivation = reconciles; + errorsAtActivation = errors.length; + }, + reconcileIntervalMs: 5, + }); + await vi.waitFor(() => { + expect(reconcilesAtActivation).toBe(1); + }); + // A relay 429 at startup must not escape startAccount: the supervisor would + // restart the account, and that restart cycle is what escalates the rate + // limiting in the first place. + expect(errorsAtActivation).toBe(1); + expect(inbox.seen).toHaveLength(1); + expect(inbox.isSettled()).toBe(false); + parent.abort(); + await lifecycle; + expect(inbox.isSettled()).toBe(true); + }); + + it("refreshes peer keys before activating and before the inbox starts", async () => { + const parent = new AbortController(); + const inbox = hangingInbox(); + const order: string[] = []; + const lifecycle = runReefChannelLifecycle({ + parentSignal: parent.signal, + startInbox: (signal) => { + order.push("inbox"); + return inbox.startInbox(signal); + }, + reconcile: async () => { + order.push("reconcile"); + }, + onReconcileError: () => {}, + onReady: async () => { + order.push("ready"); + }, + reconcileIntervalMs: 5_000, + }); + await vi.waitFor(() => { + expect(order).toEqual(["reconcile", "ready", "inbox"]); + }); + parent.abort(); + await lifecycle; + }); + + it("rejects startup when the reconcile error is not retryable", async () => { + const parent = new AbortController(); + const inbox = hangingInbox(); + const onReady = vi.fn(async () => {}); + const error = new Error("approval store unavailable"); + await expect( + runReefChannelLifecycle({ + parentSignal: parent.signal, + startInbox: inbox.startInbox, + reconcile: async () => { + throw error; + }, + onReconcileError: () => {}, + shouldContinueAfterStartupReconcileError: () => false, + onReady, + }), + ).rejects.toBe(error); + expect(onReady).not.toHaveBeenCalled(); + expect(inbox.seen).toHaveLength(0); + }); + + it("does not activate when the parent aborts during startup reconcile", async () => { + const parent = new AbortController(); + const inbox = hangingInbox(); + const reconcileStarted = deferred(); + const finishReconcile = deferred(); + const onReady = vi.fn(async () => {}); + const lifecycle = runReefChannelLifecycle({ + parentSignal: parent.signal, + startInbox: inbox.startInbox, + reconcile: async () => { + reconcileStarted.resolve(); + await finishReconcile.promise; + }, + onReconcileError: () => {}, + onReady, + }); + await reconcileStarted.promise; + parent.abort(); + finishReconcile.resolve(); + await lifecycle; + expect(onReady).not.toHaveBeenCalled(); + expect(inbox.seen).toHaveLength(0); + }); + + it("does not reject when startup reconcile fails after the parent aborts", async () => { + const parent = new AbortController(); + const inbox = hangingInbox(); + const reconcileStarted = deferred(); + const finishReconcile = deferred(); + const onReady = vi.fn(async () => {}); + const lifecycle = runReefChannelLifecycle({ + parentSignal: parent.signal, + startInbox: inbox.startInbox, + reconcile: async () => { + reconcileStarted.resolve(); + await finishReconcile.promise; + }, + onReconcileError: () => {}, + onReady, + }); + await reconcileStarted.promise; + parent.abort(); + finishReconcile.reject(new DOMException("aborted", "AbortError")); + await expect(lifecycle).resolves.toBeUndefined(); + expect(onReady).not.toHaveBeenCalled(); + expect(inbox.seen).toHaveLength(0); + }); + + it("does not start the inbox when the parent aborts during activation", async () => { + const parent = new AbortController(); + const inbox = hangingInbox(); + const activationStarted = deferred(); + const finishActivation = deferred(); + const lifecycle = runReefChannelLifecycle({ + parentSignal: parent.signal, + startInbox: inbox.startInbox, + reconcile: async () => {}, + onReconcileError: () => {}, + onReady: async () => { + activationStarted.resolve(); + await finishActivation.promise; + }, + }); + await activationStarted.promise; + parent.abort(); + finishActivation.resolve(); + await lifecycle; + expect(inbox.seen).toHaveLength(0); + }); + it("keeps running when a periodic reconcile fails", async () => { const parent = new AbortController(); const inbox = hangingInbox(); @@ -136,13 +302,15 @@ describe("Reef channel lifecycle", () => { startInbox: inbox.startInbox, reconcile: async () => { reconciles += 1; - throw new Error("rate_limited"); + if (reconciles > 1) { + throw new Error("rate_limited"); + } }, onReconcileError: (error) => errors.push(error), reconcileIntervalMs: 5, }); await vi.waitFor(() => { - expect(reconciles).toBeGreaterThanOrEqual(2); + expect(reconciles).toBeGreaterThanOrEqual(3); }); expect(errors.length).toBeGreaterThanOrEqual(2); expect(inbox.isSettled()).toBe(false); @@ -155,11 +323,18 @@ describe("Reef channel lifecycle", () => { const parent = new AbortController(); const inbox = hangingInbox(); // Simulate a non-transport crash escaping the lifecycle (reconcile errors - // are contained, so throw from the error hook itself). + // are contained, so throw from the error hook itself). The startup + // reconcile succeeds so the failure lands on the periodic loop, with the + // inbox already running and therefore able to leak. + let reconciles = 0; const lifecycle = runReefChannelLifecycle({ parentSignal: parent.signal, startInbox: inbox.startInbox, reconcile: async () => { + reconciles += 1; + if (reconciles === 1) { + return; + } throw new Error("boom"); }, onReconcileError: () => { @@ -194,6 +369,6 @@ describe("Reef channel lifecycle abort inheritance", () => { onReconcileError: () => {}, reconcileIntervalMs: 5, }); - expect(seen[0]?.aborted).toBe(true); + expect(seen).toHaveLength(0); }); }); diff --git a/extensions/reef/src/channel.ts b/extensions/reef/src/channel.ts index 9ea923988f29..2b3bfbc0dfb5 100644 --- a/extensions/reef/src/channel.ts +++ b/extensions/reef/src/channel.ts @@ -33,7 +33,12 @@ import { isRephrasedReefResend } from "./rejection-resend.js"; import { getActiveReef, getOptionalReefRuntime, getReefRuntime, setActiveReef } from "./runtime.js"; import { reefSetupAdapter, reefSetupWizard } from "./setup.js"; import { assertReefIdentityBinding, loadKeys, openStores, ReefInboxCursorStore } from "./state.js"; -import { ReefInboxConnection, ReefTransportClient, createReefWebSocket } from "./transport.js"; +import { + ReefInboxConnection, + ReefTransportClient, + createReefWebSocket, + isRetryableReefRelayFailure, +} from "./transport.js"; import { isReefPairingApprovalToken, openReefTrustStore } from "./trust-store.js"; import type { ReefAccount, ReefIngressMessage } from "./types.js"; @@ -401,12 +406,17 @@ export const reefPlugin: ChannelPlugin = { }); }); }; - // Refresh peer keys before recovery can dispatch an agent turn. Activate - // only after reconciliation, but before that turn can use Reef outbound. - await reconcile(); - setActiveReef({ flow, friends, reviews }); - await receiptNotifier.notifyRejections(trust.pendingOutboundRejections()); - ctx.setStatus({ accountId: "default", running: true, connected: false }); + // Attempt the peer-key refresh before recovery can dispatch an agent + // turn. The lifecycle activates only after that attempt is classified. + // The lifecycle owns both the ordering and the reconcile failure policy. + const activate = async () => { + await receiptNotifier.notifyRejections(trust.pendingOutboundRejections()); + if (ctx.abortSignal.aborted) { + return; + } + setActiveReef({ flow, friends, reviews }); + ctx.setStatus({ accountId: "default", running: true, connected: false }); + }; const inbox = new ReefInboxConnection( transport, (entries) => @@ -472,6 +482,8 @@ export const reefPlugin: ChannelPlugin = { }, onReconcileError: (error) => ctx.log?.error?.(`reef friend reconcile failed: ${String(error)}`), + shouldContinueAfterStartupReconcileError: isRetryableReefRelayFailure, + onReady: activate, }); } finally { ctx.setStatus({ accountId: "default", running: false, connected: false }); diff --git a/extensions/reef/src/transport.test.ts b/extensions/reef/src/transport.test.ts index bab989979f91..d953db3bf1af 100644 --- a/extensions/reef/src/transport.test.ts +++ b/extensions/reef/src/transport.test.ts @@ -10,6 +10,7 @@ import { ReefRelayError, ReefTransportClient, createReefWebSocket, + isRetryableReefRelayFailure, type WebSocketLike, } from "./transport.js"; import type { InboxEntry, ReefKeys, RelayFriend } from "./types.js"; @@ -34,6 +35,62 @@ afterEach(() => { vi.useRealTimers(); }); +describe("isRetryableReefRelayFailure", () => { + it("accepts transient relay responses and timeouts", () => { + expect(isRetryableReefRelayFailure(new ReefRelayError(408, "timeout"))).toBe(true); + expect(isRetryableReefRelayFailure(new ReefRelayError(429, "rate_limited"))).toBe(true); + expect(isRetryableReefRelayFailure(new ReefRelayError(503, "unavailable"))).toBe(true); + expect( + isRetryableReefRelayFailure(Object.assign(new Error("timed out"), { name: "TimeoutError" })), + ).toBe(true); + }); + + it("rejects definitive relay and local failures", () => { + expect(isRetryableReefRelayFailure(new ReefRelayError(401, "unauthorized"))).toBe(false); + expect(isRetryableReefRelayFailure(new Error("approval store unavailable"))).toBe(false); + }); +}); + +describe("ReefTransportClient network failures", () => { + it("normalizes fetch failures without swallowing the cause", async () => { + const cause = new TypeError("fetch failed"); + const client = new ReefTransportClient("https://relay.example", "alice", keys, async () => { + throw cause; + }); + + const error = await client.listFriends().catch((failure: unknown) => failure); + expect(error).toMatchObject({ + name: "ReefRelayUnavailableError", + message: "fetch failed", + cause, + }); + expect(isRetryableReefRelayFailure(error)).toBe(true); + }); + + it("normalizes connection loss while reading a successful response body", async () => { + const cause = new TypeError("terminated"); + const response = new Response( + new ReadableStream({ + start(controller) { + controller.error(cause); + }, + }), + ); + const client = new ReefTransportClient( + "https://relay.example", + "alice", + keys, + async () => response, + ); + + await expect(client.listFriends()).rejects.toMatchObject({ + name: "ReefRelayUnavailableError", + message: "terminated", + cause, + }); + }); +}); + function verifyRelaySignature( signature: string, input: { method: string; path: string; ts: number; bodySha256: string }, diff --git a/extensions/reef/src/transport.ts b/extensions/reef/src/transport.ts index a397fca065f6..4cc4c9b22abe 100644 --- a/extensions/reef/src/transport.ts +++ b/extensions/reef/src/transport.ts @@ -36,6 +36,13 @@ export class ReefRelayError extends Error { } } +class ReefRelayUnavailableError extends Error { + constructor(cause: unknown) { + super(cause instanceof Error ? cause.message : String(cause), { cause }); + this.name = "ReefRelayUnavailableError"; + } +} + export function isDefinitiveReefRegistrationFailure(error: unknown): boolean { return ( error instanceof ReefRelayError && @@ -46,10 +53,38 @@ export function isDefinitiveReefRegistrationFailure(error: unknown): boolean { ); } +export function isRetryableReefRelayFailure(error: unknown): boolean { + if (error instanceof ReefRelayError) { + return error.status === 408 || error.status === 429 || error.status >= 500; + } + return ( + error instanceof ReefRelayUnavailableError || + (error instanceof Error && error.name === "TimeoutError") + ); +} + export function isReefOwnershipRejection(error: unknown): boolean { return error instanceof ReefRelayError && error.message === "unknown_handle"; } +async function readReefRelaySuccessJson(response: Response, signal?: AbortSignal): Promise { + try { + return await readProviderJsonResponse(response, "reef.relay", { + maxBytes: REEF_RELAY_JSON_MAX_BYTES, + }); + } catch (error) { + if (signal?.aborted) { + throw signal.reason; + } + // Undici surfaces socket loss during response-body consumption as a + // TypeError even though fetch already resolved with response headers. + if (error instanceof TypeError) { + throw new ReefRelayUnavailableError(error); + } + throw error; + } +} + export class ReefTransportClient { // Ed25519 is deterministic: identical (method, path, ts, body) requests produce // identical signatures, which collide with the relay's replay key. Keep ts @@ -194,12 +229,20 @@ export class ReefTransportClient { url, }); try { - const response = await this.fetcher(url, { - method, - headers: { ...headers, ...(bytes.length ? { "content-type": "application/json" } : {}) }, - ...(bytes.length ? { body: bytes as BodyInit } : {}), - signal: timeout.signal, - }); + let response: Response; + try { + response = await this.fetcher(url, { + method, + headers: { ...headers, ...(bytes.length ? { "content-type": "application/json" } : {}) }, + ...(bytes.length ? { body: bytes as BodyInit } : {}), + signal: timeout.signal, + }); + } catch (error) { + if (timeout.signal?.aborted) { + throw timeout.signal.reason; + } + throw new ReefRelayUnavailableError(error); + } if (!response.ok) { let message = `relay HTTP ${response.status}`; try { @@ -223,9 +266,7 @@ export class ReefTransportClient { if (response.status === 204) { return undefined as T; } - return await readProviderJsonResponse(response, "reef.relay", { - maxBytes: REEF_RELAY_JSON_MAX_BYTES, - }); + return await readReefRelaySuccessJson(response, timeout.signal); } finally { timeout.cleanup(); }