fix(reef): contain startup friend reconcile failures (#110918)

* fix(reef): contain startup friend reconcile failures

The periodic reconcile treats a transient relay failure as non-fatal, but
the startup reconcile was a bare await outside the lifecycle. A relay 429
there rejects startAccount, and the supervisor restart that follows is
itself what escalates relay rate limiting -- so Reef gives up permanently
after MAX_RESTARTS while the relay is still throttling.

Move the startup reconcile into runReefChannelLifecycle so both paths
share one failure policy, and activate through an onReady hook to preserve
the existing ordering: refresh peer keys, activate, then start the inbox.

* fix(reef): harden reconcile recovery

Co-authored-by: Yigtwxx <yigiterdogan023@gmail.com>

* docs(changelog): credit Reef reconcile fix

* test(reef): satisfy exact-head validation

* test(reef): satisfy lint contract

---------

Co-authored-by: Peter Steinberger <steipete@gmail.com>

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<void>;
+  onReconcileError: (error: unknown) => void;
+  shouldContinueAfterError: (error: unknown) => boolean;
+  signal: AbortSignal;
+}): Promise<void> {
+  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<void>;
   reconcile: () => Promise<void>;
   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<void>;
   reconcileIntervalMs?: number;
 }): Promise<void> {
   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<void> | 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();

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<void>((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);
   });
 });

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<ReefAccount> = {
           });
         });
       };
-      // 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<ReefAccount> = {
           },
           onReconcileError: (error) =>
             ctx.log?.error?.(`reef friend reconcile failed: ${String(error)}`),
+          shouldContinueAfterStartupReconcileError: isRetryableReefRelayFailure,
+          onReady: activate,
         });
       } finally {
         ctx.setStatus({ accountId: "default", running: false, connected: false });

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 },

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<T>(response: Response, signal?: AbortSignal): Promise<T> {
+  try {
+    return await readProviderJsonResponse<T>(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<T>(response, "reef.relay", {
-        maxBytes: REEF_RELAY_JSON_MAX_BYTES,
-      });
+      return await readReefRelaySuccessJson<T>(response, timeout.signal);
     } finally {
       timeout.cleanup();
     }