fix(core): retry busy write admission

CHANGELOG.md

@@ -17,6 +17,10 @@ published version with a date and open a fresh empty `[Unreleased]` above it.
 - See `docs/migration/file-native-writeback.md` for the new read/edit/create/delete contract, schema discovery flow, writeback status surface, and per-adapter draft filename examples.
 - See `docs/migration/new-json-callers.md` for the Cloud and demo caller scan that identifies `new.json` write paths to migrate.
 
+### Fixed
+
+- Direct HTTP write admission now honors `Retry-After` up to 30 seconds for `workspace_busy` / `write_admission_limit` responses in the SDK's existing four-attempt retry layer, while preserving the prior two-second cap for other retryable responses. Implicit defaults are three seconds for receipt polling and 90 seconds for admission; an explicit `writebackTimeoutMs` bounds each phase independently, with `0` leaving admission unbounded and receipt polling disabled.
+
 ### Added
 
 - GitHub pull indexes now surface `merged` and `mergedAt` across webhook, direct, and bulk ingestion paths so time-windowed consumers can identify merged pull requests without opening every `meta.json`.

packages/core/src/vfs-client/index.ts

@@ -85,6 +85,33 @@ export class RelayfileWritebackPendingError extends RelayfileWritebackError {
   }
 }
 
+export interface RelayfileWritebackAdmissionTimeoutErrorOptions {
+  provider: string;
+  operation: string;
+  path: string;
+  timeoutMs: number;
+}
+
+/** Direct HTTP admission never minted an op before the caller's admission deadline. */
+export class RelayfileWritebackAdmissionTimeoutError extends RelayfileWritebackError {
+  readonly path: string;
+  readonly timeoutMs: number;
+
+  constructor(options: RelayfileWritebackAdmissionTimeoutErrorOptions) {
+    super({
+      provider: options.provider,
+      operation: options.operation,
+      cause: new Error(
+        `writeback_admission_timeout: no operation admitted for ${options.path} after ${options.timeoutMs}ms`
+      ),
+      retryable: true
+    });
+    this.name = "RelayfileWritebackAdmissionTimeoutError";
+    this.path = options.path;
+    this.timeoutMs = options.timeoutMs;
+  }
+}
+
 export interface RelayfileWritebackTerminalErrorOptions {
   provider: string;
   operation: string;
@@ -148,7 +175,12 @@ export interface IntegrationClientOptions {
   /**
    * Max wait, in ms, for the Relayfile writeback worker to emit a receipt onto
    * the just-written draft. Defaults to 3000ms. `0` means fire-and-forget — the
-   * client returns immediately without a receipt.
+   * client returns immediately without a receipt. In direct HTTP mode, an
+   * explicit value also bounds write admission as an independent phase. When
+   * omitted, receipt waiting defaults to 3s while admission defaults to 90s.
+   * Advertised delays are honored while they fit inside that admission budget;
+   * after three consecutive 30s delays, the deadline wins at t+90s before a
+   * fourth request.
    */
   writebackTimeoutMs?: number;
   /** Poll interval while waiting for a receipt. Default 250ms. */
@@ -197,6 +229,9 @@ export interface WritebackResult {
 }
 
 const DEFAULT_WRITEBACK_TIMEOUT_MS = 3_000;
+const SDK_LEGACY_RETRY_MAX_DELAY_MS = 2_000;
+const WORKSPACE_BUSY_RETRY_MAX_DELAY_MS = 30_000;
+const DEFAULT_WRITEBACK_ADMISSION_TIMEOUT_MS = WORKSPACE_BUSY_RETRY_MAX_DELAY_MS * 3;
 
 function isRecord(value: unknown): value is Record<string, unknown> {
   return typeof value === "object" && value !== null && !Array.isArray(value);
@@ -207,6 +242,85 @@ function nonEmpty(value: string | undefined): string | undefined {
   return trimmed ? trimmed : undefined;
 }
 
+function isWorkspaceBusyAdmission(value: unknown): boolean {
+  if (!isRecord(value) || value.code !== "workspace_busy") return false;
+  const reason = value.reason ?? (isRecord(value.details) ? value.details.reason : undefined);
+  return reason === "write_admission_limit";
+}
+
+function isWorkspaceBusyAdmissionError(value: unknown): value is RelayFileApiError {
+  return value instanceof RelayFileApiError && value.status === 429 && isWorkspaceBusyAdmission(value);
+}
+
+function retryAfterDelayMs(value: string): number | undefined {
+  const trimmed = value.trim();
+  if (/^\d+$/.test(trimmed)) {
+    const seconds = Number(trimmed);
+    if (Number.isFinite(seconds)) return seconds * 1_000;
+  }
+  const timestamp = Date.parse(trimmed);
+  return Number.isNaN(timestamp) ? undefined : Math.max(0, timestamp - Date.now());
+}
+
+async function responseIsWorkspaceBusyAdmission(response: Response): Promise<boolean> {
+  if (response.status !== 429) return false;
+  try {
+    return isWorkspaceBusyAdmission(await response.clone().json());
+  } catch {
+    return false;
+  }
+}
+
+function responseWithRetryAfter(response: Response, retryAfter: string): Response {
+  const headers = new Headers(response.headers);
+  headers.set("Retry-After", retryAfter);
+  return new Response(response.body, {
+    status: response.status,
+    statusText: response.statusText,
+    headers
+  });
+}
+
+function isDirectWriteAdmissionRequest(input: RequestInfo | URL, init?: RequestInit): boolean {
+  const method = (init?.method ?? (input instanceof Request ? input.method : "GET")).toUpperCase();
+  const url = input instanceof Request ? input.url : String(input);
+  try {
+    return method === "PUT" && new URL(url, "http://relayfile.invalid").pathname.endsWith("/fs/file");
+  } catch {
+    return false;
+  }
+}
+
+/**
+ * The SDK owns the only retry loop. Raising its max delay lets workspace write
+ * admission honor Retry-After; this adapter keeps the SDK's previous 2s cap for
+ * every other retryable response so unrelated 429/5xx behavior does not move.
+ */
+function directRetryFetch(fetchImpl: typeof fetch): typeof fetch {
+  return async (input, init) => {
+    const response = await fetchImpl(input, init);
+    if (response.status !== 429 && (response.status < 500 || response.status > 599)) {
+      return response;
+    }
+    const retryAfter = response.headers.get("Retry-After");
+    if (!retryAfter) return response;
+    const delayMs = retryAfterDelayMs(retryAfter);
+    if (
+      isDirectWriteAdmissionRequest(input, init) &&
+      (await responseIsWorkspaceBusyAdmission(response))
+    ) {
+      return delayMs === undefined
+        ? response
+        : responseWithRetryAfter(response, String(Math.ceil(delayMs / 1_000)));
+    }
+
+    if (delayMs === undefined || delayMs <= SDK_LEGACY_RETRY_MAX_DELAY_MS) {
+      return response;
+    }
+    return responseWithRetryAfter(response, String(SDK_LEGACY_RETRY_MAX_DELAY_MS / 1_000));
+  };
+}
+
 function mountRootCandidate(value: string | undefined): string | undefined {
   const trimmed = value?.trim();
   return trimmed ? trimmed : undefined;
@@ -304,12 +418,16 @@ function directClientConfig(
     nonEmpty(process.env.RELAYFILE_WORKSPACE) ??
     nonEmpty(process.env.RELAY_WORKSPACE_ID);
   if (!baseUrl || !token || !workspaceId) return undefined;
+  const fetchImpl = directRetryFetch(client.fetchImpl ?? globalThis.fetch);
   return {
     workspaceId,
     relayfile: new RelayFileClient({
       baseUrl,
       token,
-      fetchImpl: client.fetchImpl
+      fetchImpl,
+      retry: {
+        maxDelayMs: WORKSPACE_BUSY_RETRY_MAX_DELAY_MS
+      }
     })
   };
 }
@@ -423,30 +541,53 @@ async function writeJsonFileViaRelayfileApi(
   body: unknown,
   direct: { workspaceId: string; relayfile: RelayFileClient }
 ): Promise<WritebackResult> {
-  const queued = await direct.relayfile.writeFile({
-    workspaceId: direct.workspaceId,
-    path: relayPath,
-    baseRevision: "*",
-    contentType: "application/json",
-    content: `${JSON.stringify(body, null, 2)}\n`
-  });
-  if (queued.writeback && !nonEmpty(queued.opId)) {
-    throw new RelayfileWritebackReceiptError({
-      provider,
-      operation,
-      opId: "(missing)",
-      reason: "queued writeback response did not include opId"
+  const timeoutMs = client.writebackTimeoutMs ?? DEFAULT_WRITEBACK_ADMISSION_TIMEOUT_MS;
+  const controller = timeoutMs > 0 ? new AbortController() : undefined;
+  const deadlineTimer = controller
+    ? setTimeout(() => controller.abort(), timeoutMs)
+    : undefined;
+  let admitted = false;
+  try {
+    const queued = await direct.relayfile.writeFile({
+      workspaceId: direct.workspaceId,
+      path: relayPath,
+      baseRevision: "*",
+      contentType: "application/json",
+      content: `${JSON.stringify(body, null, 2)}\n`,
+      signal: controller?.signal
     });
+    admitted = true;
+    if (deadlineTimer !== undefined) clearTimeout(deadlineTimer);
+    if (queued.writeback && !nonEmpty(queued.opId)) {
+      throw new RelayfileWritebackReceiptError({
+        provider,
+        operation,
+        opId: "(missing)",
+        reason: "queued writeback response did not include opId"
+      });
+    }
+    const receipt = queued.opId
+      ? await waitForOperationReceipt(client, provider, operation, direct, queued.opId, relayPath)
+      : undefined;
+    return {
+      path: relayPath,
+      absolutePath: relayPath,
+      opId: queued.opId,
+      ...(receipt ? { receipt } : {})
+    };
+  } catch (error) {
+    if (controller?.signal.aborted && !admitted && !(error instanceof RelayfileWritebackError)) {
+      throw new RelayfileWritebackAdmissionTimeoutError({
+        provider,
+        operation,
+        path: relayPath,
+        timeoutMs
+      });
+    }
+    throw error;
+  } finally {
+    if (deadlineTimer !== undefined) clearTimeout(deadlineTimer);
   }
-  const receipt = queued.opId
-    ? await waitForOperationReceipt(client, provider, operation, direct, queued.opId, relayPath)
-    : undefined;
-  return {
-    path: relayPath,
-    absolutePath: relayPath,
-    opId: queued.opId,
-    ...(receipt ? { receipt } : {})
-  };
 }
 
 function toAbsolutePath(client: IntegrationClientOptions, relayPath: string): string {
@@ -612,7 +753,12 @@ export async function writeJsonFile(
     if (cause instanceof RelayfileWritebackError) {
       throw cause;
     }
-    throw new RelayfileWritebackError({ provider, operation, cause, retryable: false });
+    throw new RelayfileWritebackError({
+      provider,
+      operation,
+      cause,
+      retryable: isWorkspaceBusyAdmissionError(cause)
+    });
   }
 }