diff --git a/packages/ai/CHANGELOG.md b/packages/ai/CHANGELOG.md
index e6eaa2df6..467f5fcec 100644
--- a/packages/ai/CHANGELOG.md
+++ b/packages/ai/CHANGELOG.md
@@ -1,18 +1,20 @@
 # Changelog
 
 ## [Unreleased]
-
 ### Breaking Changes
 
+- Changed `AuthBrokerClient.fetchSnapshot()` to return status-based results (`200` or `304`) instead of always returning a raw snapshot body, so callers now need to branch on `status`
 - Renamed public schema utilities in `@oh-my-pi/pi-ai/utils/schema` by replacing `sanitizeSchemaForGoogle`, `sanitizeSchemaForCCA`, `prepareSchemaForCCA`, and `sanitizeSchemaForMCP` with `normalizeSchemaForGoogle`, `normalizeSchemaForCCA`, and `normalizeSchemaForMCP`
 - Added MCP schema normalization via `normalizeSchemaForMCP` for compatibility checks
 - Removed the `StringEnum` helper from `@oh-my-pi/pi-ai/utils/schema`. Use `z.enum([...])` directly; Zod's emitted JSON Schema is already wire-compatible with Google and other providers.
 - Renamed the concrete SQLite credential store class from `AuthCredentialStore` to `SqliteAuthCredentialStore`. `AuthCredentialStore` is now the persistence interface implemented by both the SQLite store and the new `RemoteAuthCredentialStore`. Update `new AuthCredentialStore(db)` / `AuthCredentialStore.open(...)` call-sites to `SqliteAuthCredentialStore`; type-position uses (`store: AuthCredentialStore`) continue to work unchanged.
 
 ### Added
+
+- Added `onAuthError` to `StreamOptions` and wired `streamSimple()` to retry once with a replacement API key when the first provider response is a 401 before any assistant events are emitted
+- Added generation-aware snapshot metadata (`generation`, `serverNowMs`, `refresher`, and `rotatesInMs`) to auth-broker snapshot responses to support client-side credential-rotation planning
 - Added `transport: "pi-native"` on `Model` and the matching `streamPiNative` client. When `model.transport === "pi-native"`, `streamSimple` short-circuits the per-provider dispatch and POSTs the canonical `Context` to the auth-gateway's `POST /v1/pi/stream` endpoint. The response is SSE-framed `AssistantMessageEvent`s parsed by `readSseJson` and pushed verbatim into the local `AssistantMessageEventStream` — no wire-format translation, no partial-stripping reconstruction. Used by containerized omp installs (robomp slots, swarm extension, etc.) to route every LLM call through a credential-holding sidecar; the slot itself never sees the real provider tokens. Server-controlled fields (`apiKey`, `signal`, `fetch`, lifecycle callbacks, the provider-session map) are stripped from the wire body — `apiKey` rides in the `Authorization` header as the gateway bearer.
 - Added `POST /v1/pi/stream` to the auth-gateway. Same auth + abort + model-resolution + codex-compat + prefix-cache plumbing as the foreign-wire routes; only the wire-format translation is skipped. Request body is `{ modelId, context, options?, stream? }` where `context` is the canonical pi-ai `Context` and `options` is `SimpleStreamOptions` with non-serializable fields stripped. Response is SSE-framed `AssistantMessageEvent` (terminated by `data: [DONE]`) when streaming, or `{ message: AssistantMessage }` JSON when `stream: false`.
-
 - Added Vertex AI authentication via Google Application Default Credentials from `GOOGLE_APPLICATION_CREDENTIALS`, `~/.config/gcloud/application_default_credentials.json`, or metadata server tokens, with token caching and refresh skew control via `GOOGLE_VERTEX_REFRESH_SKEW_MS`
 - Added support for Anthropic image message parts with `type: "url"` and `type: "file"` sources
 - Added `stopSequences` and `frequencyPenalty` to shared stream options and wired them through to OpenAI request translation
@@ -42,6 +44,7 @@
 
 ### Changed
 
+- Changed `GET /v1/snapshot` to support generation-based polling with `If-None-Match` and `wait` for long-poll updates and to return `304` when no snapshot changes are available
 - Changed Bedrock credential resolution for streaming calls to prefer environment keys, AWS profile/SSO credentials, and IMDSv2 fallback when available
 - Changed auth-gateway parsing for OpenAI chat-completions and Responses to ignore unsupported SDK-only fields instead of rejecting requests
 - Changed auth-gateway protocol handling to include CORS headers on responses and support browser-origin requests
@@ -57,6 +60,8 @@
 
 ### Fixed
 
+- Fixed OAuth credential refresh flow so concurrent manual and background refreshes now share one in-flight attempt per credential, and `RemoteAuthCredentialStore` now re-synchronizes before using near-expiring OAuth credentials
+- Fixed stale-credential handling after auth failures by waiting for updated broker snapshots and refreshing suspect credentials through broker endpoints before continuing
 - Fixed Google Generative AI startup behavior to throw a clear API-key-required error when no key is configured
 - Fixed AWS Bedrock image message serialization to preserve base64 `source.bytes` payloads instead of decoding and rebuilding them
 - Fixed Google provider error handling to extract the API-reported `error.message` from JSON response bodies when available
diff --git a/packages/ai/src/auth-broker/client.ts b/packages/ai/src/auth-broker/client.ts
index 4b93a3d99..2b8fd6167 100644
--- a/packages/ai/src/auth-broker/client.ts
+++ b/packages/ai/src/auth-broker/client.ts
@@ -50,6 +50,28 @@ export class AuthBrokerError extends Error {
 	}
 }
 
+export interface FetchSnapshotOptions {
+	ifGenerationGt?: number;
+	waitMs?: number;
+	signal?: AbortSignal;
+}
+
+export type FetchSnapshotResult =
+	| { status: 200; snapshot: SnapshotResponse; generation: number }
+	| { status: 304; generation: number };
+
+function parseGenerationTag(header: string | null): number | undefined {
+	if (!header) return undefined;
+	let value = header.trim();
+	if (value.startsWith("W/")) value = value.slice(2).trim();
+	if (value.startsWith('"') && value.endsWith('"') && value.length >= 2) {
+		value = value.slice(1, -1);
+	}
+	const generation = Number(value);
+	if (!Number.isInteger(generation) || generation < 0) return undefined;
+	return generation;
+}
+
 const DEFAULT_TIMEOUT_MS = 10_000;
 const DEFAULT_MAX_RETRIES = 1;
 
@@ -72,13 +94,38 @@ export class AuthBrokerClient {
 		return this.#request("GET", "/v1/healthz", { schema: healthzResponseSchema, auth: false, signal });
 	}
 
-	fetchSnapshot(signal?: AbortSignal): Promise<SnapshotResponse> {
-		// `snapshotResponseSchema` narrows `refresh` to the sentinel literal where
-		// the public type uses plain `string`; the wire shape is identical.
-		return this.#request("GET", "/v1/snapshot", {
-			schema: snapshotResponseSchema,
-			signal,
-		}) as Promise<SnapshotResponse>;
+	async fetchSnapshot(opts: FetchSnapshotOptions = {}): Promise<FetchSnapshotResult> {
+		return this.#fetchSnapshotResult(opts);
+	}
+	async #fetchSnapshotResult(opts: FetchSnapshotOptions): Promise<FetchSnapshotResult> {
+		const query = new URLSearchParams();
+		if (opts.waitMs !== undefined) query.set("wait", String(opts.waitMs));
+		const path = `/v1/snapshot${query.size > 0 ? `?${query.toString()}` : ""}`;
+		const headers: Record<string, string> = {};
+		if (opts.ifGenerationGt !== undefined) headers["If-None-Match"] = `"${opts.ifGenerationGt}"`;
+		const timeoutMs =
+			opts.waitMs !== undefined && opts.waitMs > 0 ? Math.max(this.#timeoutMs, opts.waitMs + 1000) : undefined;
+		const response = await this.#fetchRaw("GET", path, {
+			auth: true,
+			headers,
+			signal: opts.signal,
+			timeoutMs,
+		});
+		const etagGeneration = parseGenerationTag(response.headers.get("etag"));
+		if (response.status === 304) {
+			return { status: 304, generation: etagGeneration ?? opts.ifGenerationGt ?? 0 };
+		}
+		const text = await response.text();
+		const raw = this.#parseJson(text, response.status);
+		const validated = snapshotResponseSchema.safeParse(raw);
+		if (!validated.success) {
+			throw new AuthBrokerError("Auth broker response failed schema validation", {
+				status: response.status,
+				body: validated.error.message,
+			});
+		}
+		const snapshot = validated.data as SnapshotResponse;
+		return { status: 200, snapshot, generation: etagGeneration ?? snapshot.generation };
 	}
 
 	fetchUsage(signal?: AbortSignal): Promise<UsageResponse> {
@@ -123,9 +170,45 @@ export class AuthBrokerClient {
 		path: string,
 		opts: { schema: TSchema; auth?: boolean; body?: unknown; signal?: AbortSignal },
 	): Promise<zInfer<TSchema>> {
+		const response = await this.#fetchRaw(method, path, opts);
+		const text = await response.text();
+		const raw = this.#parseJson(text, response.status);
+		const validated = opts.schema.safeParse(raw);
+		if (!validated.success) {
+			throw new AuthBrokerError("Auth broker response failed schema validation", {
+				status: response.status,
+				body: validated.error.message,
+			});
+		}
+		return validated.data;
+	}
+
+	#parseJson(text: string, status: number): unknown {
+		try {
+			return text.length === 0 ? null : JSON.parse(text);
+		} catch (parseError) {
+			throw new AuthBrokerError("Auth broker returned malformed JSON", {
+				status,
+				body: text,
+				cause: parseError,
+			});
+		}
+	}
+
+	async #fetchRaw(
+		method: "GET" | "POST",
+		path: string,
+		opts: {
+			auth?: boolean;
+			body?: unknown;
+			signal?: AbortSignal;
+			headers?: Record<string, string>;
+			timeoutMs?: number;
+		},
+	): Promise<Response> {
 		const auth = opts.auth ?? true;
 		const url = `${this.#baseUrl}${path}`;
-		const headers: Record<string, string> = { Accept: "application/json" };
+		const headers: Record<string, string> = { Accept: "application/json", ...(opts.headers ?? {}) };
 		if (auth) headers.Authorization = `Bearer ${this.#token}`;
 		let payload: string | undefined;
 		if (opts.body !== undefined) {
@@ -141,10 +224,7 @@ export class AuthBrokerClient {
 
 		let lastError: unknown;
 		for (let attempt = 0; attempt <= this.#maxRetries; attempt += 1) {
-			// Compose caller's signal with the per-attempt timeout so either
-			// source can cancel the in-flight fetch. `AbortSignal.any` is the
-			// supported merge primitive in Bun ≥ 1.0 / Node ≥ 20.
-			const timeoutSignal = AbortSignal.timeout(this.#timeoutMs);
+			const timeoutSignal = AbortSignal.timeout(opts.timeoutMs ?? this.#timeoutMs);
 			const signal = opts.signal ? AbortSignal.any([opts.signal, timeoutSignal]) : timeoutSignal;
 			try {
 				const response = await this.#fetch(url, {
@@ -153,31 +233,14 @@ export class AuthBrokerClient {
 					body: payload,
 					signal,
 				});
-				const text = await response.text();
-				if (!response.ok) {
+				if (!response.ok && response.status !== 304) {
+					const text = await response.text();
 					throw new AuthBrokerError(`Auth broker request failed: ${response.status} ${response.statusText}`, {
 						status: response.status,
 						body: text,
 					});
 				}
-				let raw: unknown;
-				try {
-					raw = text.length === 0 ? null : JSON.parse(text);
-				} catch (parseError) {
-					throw new AuthBrokerError("Auth broker returned malformed JSON", {
-						status: response.status,
-						body: text,
-						cause: parseError,
-					});
-				}
-				const validated = opts.schema.safeParse(raw);
-				if (!validated.success) {
-					throw new AuthBrokerError("Auth broker response failed schema validation", {
-						status: response.status,
-						body: validated.error.message,
-					});
-				}
-				return validated.data;
+				return response;
 			} catch (error) {
 				lastError = error;
 				// Caller-driven abort wins over retry — the caller said stop.
diff --git a/packages/ai/src/auth-broker/refresher.ts b/packages/ai/src/auth-broker/refresher.ts
index 7b9e9947b..6c226dd5a 100644
--- a/packages/ai/src/auth-broker/refresher.ts
+++ b/packages/ai/src/auth-broker/refresher.ts
@@ -2,9 +2,9 @@
  * Background OAuth refresh loop for the auth-broker server.
  *
  * Iterates active OAuth credentials at `refreshIntervalMs` cadence, refreshing
- * any whose `expires - Date.now() < refreshSkewMs`. Single-flighted per
- * credential id so a long refresh can't be retriggered until it settles.
- *
+ * any whose `expires - Date.now() < refreshSkewMs`. Refresh single-flight
+ * lives in {@link AuthStorage} so manual and background refreshes share the
+ * same upstream attempt.
  * Definitively-failed credentials (invalid_grant / 401 not from network blip)
  * are disabled via {@link AuthStorage.disableCredentialById} so the next
  * snapshot pull surfaces a clean delete on the client.
@@ -33,26 +33,34 @@ function isDefinitiveFailure(errorMsg: string): boolean {
 	return false;
 }
 
+export interface AuthBrokerRefresherSchedule {
+	enabled: boolean;
+	intervalMs: number;
+	skewMs: number;
+	nextSweepAt: number;
+}
+
 export class AuthBrokerRefresher {
 	readonly #storage: AuthStorage;
 	readonly #refreshSkewMs: number;
 	readonly #refreshIntervalMs: number;
 	readonly #now: () => number;
-	readonly #inFlight: Map<number, Promise<void>> = new Map();
 	#timer: NodeJS.Timeout | undefined;
 	#running = false;
-
+	#nextSweepAt: number;
 	constructor(opts: AuthBrokerRefresherOptions) {
 		this.#storage = opts.storage;
 		this.#refreshSkewMs = opts.refreshSkewMs ?? DEFAULT_REFRESH_SKEW_MS;
 		this.#refreshIntervalMs = opts.refreshIntervalMs ?? DEFAULT_REFRESH_INTERVAL_MS;
 		this.#now = opts.now ?? Date.now;
+		this.#nextSweepAt = this.#now();
 	}
 
 	start(): void {
 		if (this.#timer !== undefined) return;
 		// Refresh sweep is best-effort; kick once immediately so freshly-booted
 		// brokers don't hand out near-expired tokens for the first interval.
+		this.#nextSweepAt = this.#now();
 		void this.tick();
 		this.#timer = setInterval(() => {
 			void this.tick();
@@ -66,10 +74,20 @@ export class AuthBrokerRefresher {
 		}
 	}
 
+	getSchedule(): AuthBrokerRefresherSchedule {
+		return {
+			enabled: true,
+			intervalMs: this.#refreshIntervalMs,
+			skewMs: this.#refreshSkewMs,
+			nextSweepAt: this.#nextSweepAt,
+		};
+	}
+
 	/** Run one sweep. Exposed for tests. */
 	async tick(): Promise<void> {
 		if (this.#running) return;
 		this.#running = true;
+		this.#nextSweepAt = this.#now();
 		try {
 			await this.#storage.reload();
 			const snapshot = this.#storage.exportSnapshot();
@@ -86,31 +104,24 @@ export class AuthBrokerRefresher {
 			await Promise.all(targets.map(id => this.#refreshOne(id)));
 		} finally {
 			this.#running = false;
+			this.#nextSweepAt = this.#now() + this.#refreshIntervalMs;
 		}
 	}
 
-	#refreshOne(id: number): Promise<void> {
-		const existing = this.#inFlight.get(id);
-		if (existing) return existing;
-		const promise = (async () => {
-			try {
-				await this.#storage.forceRefreshCredentialById(id);
-			} catch (error) {
-				const errorMsg = String(error);
-				if (isDefinitiveFailure(errorMsg)) {
-					logger.warn("auth-broker refresh failed definitively; disabling credential", {
-						id,
-						error: errorMsg,
-					});
-					this.#storage.disableCredentialById(id, `auth-broker refresh failed: ${errorMsg}`);
-				} else {
-					logger.debug("auth-broker refresh failed (transient)", { id, error: errorMsg });
-				}
-			} finally {
-				this.#inFlight.delete(id);
+	async #refreshOne(id: number): Promise<void> {
+		try {
+			await this.#storage.refreshCredentialById(id);
+		} catch (error) {
+			const errorMsg = String(error);
+			if (isDefinitiveFailure(errorMsg)) {
+				logger.warn("auth-broker refresh failed definitively; disabling credential", {
+					id,
+					error: errorMsg,
+				});
+				this.#storage.disableCredentialById(id, `auth-broker refresh failed: ${errorMsg}`);
+			} else {
+				logger.debug("auth-broker refresh failed (transient)", { id, error: errorMsg });
 			}
-		})();
-		this.#inFlight.set(id, promise);
-		return promise;
+		}
 	}
 }
diff --git a/packages/ai/src/auth-broker/remote-store.ts b/packages/ai/src/auth-broker/remote-store.ts
index 386749b8b..35515d540 100644
--- a/packages/ai/src/auth-broker/remote-store.ts
+++ b/packages/ai/src/auth-broker/remote-store.ts
@@ -7,10 +7,10 @@
  * usage reports cache TTL is 5 minutes per credential, so durability across
  * runs isn't required.
  */
+import { scheduler } from "node:timers/promises";
 import { logger } from "@oh-my-pi/pi-utils";
 import {
 	type AuthCredential,
-	type AuthCredentialSnapshot,
 	type AuthCredentialStore,
 	type OAuthCredential,
 	REMOTE_REFRESH_SENTINEL,
@@ -20,6 +20,7 @@ import type { Provider } from "../types";
 import type { UsageReport } from "../usage";
 import type { OAuthCredentials } from "../utils/oauth/types";
 import type { AuthBrokerClient } from "./client";
+import type { SnapshotResponse } from "./types";
 
 /**
  * Client-side TTL for the aggregate `/v1/usage` response. Set below the
@@ -28,6 +29,26 @@ import type { AuthBrokerClient } from "./client";
  * the parallel fan-out from `#rankOAuthSelections` into a single round-trip.
  */
 const USAGE_CACHE_TTL_MS = 15_000;
+const WAIT_THRESHOLD_MS = 1_000;
+const MAX_WAIT_MS = 5_000;
+const BACKGROUND_WAIT_MS = 30_000;
+const BACKGROUND_BACKOFF_INITIAL_MS = 500;
+const BACKGROUND_BACKOFF_MAX_MS = 30_000;
+
+function emptySnapshot(): SnapshotResponse {
+	return {
+		generation: 0,
+		generatedAt: 0,
+		serverNowMs: 0,
+		refresher: {
+			enabled: false,
+			intervalMs: 0,
+			skewMs: 0,
+			nextSweepInMs: Number.MAX_SAFE_INTEGER,
+		},
+		credentials: [],
+	};
+}
 
 interface CacheEntry {
 	value: string;
@@ -45,12 +66,15 @@ export interface RemoteAuthCredentialStoreOptions {
 	 * Initial snapshot. When omitted, callers must call
 	 * {@link RemoteAuthCredentialStore.refreshSnapshot} before the first read.
 	 */
-	initialSnapshot?: AuthCredentialSnapshot;
+	initialSnapshot?: SnapshotResponse;
 }
 
 export class RemoteAuthCredentialStore implements AuthCredentialStore {
 	readonly #client: AuthBrokerClient;
-	#snapshot: AuthCredentialSnapshot;
+	#snapshot: SnapshotResponse = emptySnapshot();
+	#snapshotReceivedAt = Date.now();
+	#generation = 0;
+	#backgroundAbort = new AbortController();
 	#cache: Map<string, CacheEntry> = new Map();
 	#usageCache?: UsageCacheEntry;
 	#usageInflight?: Promise<UsageReport[] | null>;
@@ -58,20 +82,48 @@ export class RemoteAuthCredentialStore implements AuthCredentialStore {
 
 	constructor(opts: RemoteAuthCredentialStoreOptions) {
 		this.#client = opts.client;
-		this.#snapshot = opts.initialSnapshot ?? { generatedAt: 0, credentials: [] };
+		this.#applySnapshot(opts.initialSnapshot ?? emptySnapshot(), opts.initialSnapshot?.generation ?? 0);
+		void this.#runBackgroundLongPoll();
 	}
 
 	get client(): AuthBrokerClient {
 		return this.#client;
 	}
 
-	get snapshot(): AuthCredentialSnapshot {
+	get snapshot(): SnapshotResponse {
 		return this.#snapshot;
 	}
 
+	#applySnapshot(snapshot: SnapshotResponse, generation: number): void {
+		this.#snapshot = snapshot;
+		this.#generation = generation;
+		this.#snapshotReceivedAt = Date.now();
+	}
+
+	async #runBackgroundLongPoll(): Promise<void> {
+		let backoffMs = BACKGROUND_BACKOFF_INITIAL_MS;
+		while (!this.#closed && !this.#backgroundAbort.signal.aborted) {
+			try {
+				const result = await this.#client.fetchSnapshot({
+					ifGenerationGt: this.#generation,
+					waitMs: BACKGROUND_WAIT_MS,
+					signal: this.#backgroundAbort.signal,
+				});
+				if (result.status === 200) this.#applySnapshot(result.snapshot, result.generation);
+				backoffMs = BACKGROUND_BACKOFF_INITIAL_MS;
+			} catch (error) {
+				if (this.#closed || this.#backgroundAbort.signal.aborted) break;
+				logger.debug("auth-broker background snapshot sync failed", { error: String(error) });
+				await scheduler.wait(backoffMs, { signal: this.#backgroundAbort.signal }).catch(() => {});
+				backoffMs = Math.min(BACKGROUND_BACKOFF_MAX_MS, backoffMs * 2);
+			}
+		}
+	}
+
 	/** Re-hydrate the in-memory snapshot from the broker. */
-	async refreshSnapshot(): Promise<AuthCredentialSnapshot> {
-		this.#snapshot = await this.#client.fetchSnapshot();
+	async refreshSnapshot(): Promise<SnapshotResponse> {
+		const result = await this.#client.fetchSnapshot();
+		if (result.status === 200) this.#applySnapshot(result.snapshot, result.generation);
 		return this.#snapshot;
 	}
 
@@ -118,6 +170,28 @@ export class RemoteAuthCredentialStore implements AuthCredentialStore {
 		return true;
 	}
 
+	async waitForFreshSnapshot(maxWaitMs: number, opts: { signal?: AbortSignal } = {}): Promise<void> {
+		const result = await this.#client.fetchSnapshot({
+			ifGenerationGt: this.#generation,
+			waitMs: maxWaitMs,
+			signal: opts.signal,
+		});
+		if (result.status === 200) this.#applySnapshot(result.snapshot, result.generation);
+	}
+
+	async prepareForRequest(credentialId: number, opts: { signal?: AbortSignal } = {}): Promise<void> {
+		const entry = this.#snapshot.credentials.find(candidate => candidate.id === credentialId);
+		if (!entry || entry.credential.type !== "oauth" || entry.rotatesInMs === null) return;
+		const remainingMs = this.#snapshotReceivedAt + entry.rotatesInMs - Date.now();
+		if (remainingMs > WAIT_THRESHOLD_MS) return;
+		await this.waitForFreshSnapshot(MAX_WAIT_MS, opts);
+	}
+
+	async markCredentialSuspect(credentialId: number, opts: { signal?: AbortSignal } = {}): Promise<void> {
+		await this.#client.refreshCredential(credentialId, opts.signal);
+		await this.waitForFreshSnapshot(MAX_WAIT_MS, opts);
+	}
+
 	replaceAuthCredentialsForProvider(_provider: string, _credentials: AuthCredential[]): StoredAuthCredential[] {
 		throw new Error(
 			"RemoteAuthCredentialStore is read-only on the client. Use `omp auth-broker login <provider>` to mutate credentials.",
@@ -170,6 +244,9 @@ export class RemoteAuthCredentialStore implements AuthCredentialStore {
 		signal?: AbortSignal,
 	): Promise<OAuthCredentials> {
 		const { entry } = await this.#client.refreshCredential(credentialId, signal);
+		await this.refreshSnapshot().catch(error => {
+			logger.debug("auth-broker snapshot refresh after credential refresh failed", { error: String(error) });
+		});
 		if (entry.credential.type !== "oauth") {
 			throw new Error(`Broker returned non-OAuth credential for id=${credentialId}`);
 		}
@@ -268,6 +345,7 @@ export class RemoteAuthCredentialStore implements AuthCredentialStore {
 	close(): void {
 		if (this.#closed) return;
 		this.#closed = true;
+		this.#backgroundAbort.abort();
 		this.#cache.clear();
 	}
 }
diff --git a/packages/ai/src/auth-broker/server.ts b/packages/ai/src/auth-broker/server.ts
index cc1de341d..e1fe23da5 100644
--- a/packages/ai/src/auth-broker/server.ts
+++ b/packages/ai/src/auth-broker/server.ts
@@ -12,12 +12,14 @@
 import { logger } from "@oh-my-pi/pi-utils";
 import type { AuthStorage } from "../auth-storage";
 import { parseBind } from "../utils/parse-bind";
-import { AuthBrokerRefresher } from "./refresher";
+import { AuthBrokerRefresher, type AuthBrokerRefresherSchedule } from "./refresher";
 import type {
 	CredentialDisableResponse,
 	CredentialRefreshResponse,
 	CredentialUploadResponse,
 	HealthzResponse,
+	RefresherSchedule,
+	SnapshotEntry,
 	SnapshotResponse,
 } from "./types";
 import { DEFAULT_AUTH_BROKER_BIND, DEFAULT_REFRESH_INTERVAL_MS, DEFAULT_REFRESH_SKEW_MS } from "./types";
@@ -48,13 +50,17 @@ export interface AuthBrokerServerHandle {
 	close(): Promise<void>;
 }
 
-function json(status: number, body: unknown): Response {
+function json(status: number, body: unknown, headers?: Record<string, string>): Response {
 	return new Response(JSON.stringify(body), {
 		status,
-		headers: { "Content-Type": "application/json" },
+		headers: { "Content-Type": "application/json", ...(headers ?? {}) },
 	});
 }
 
+function empty(status: number, headers?: Record<string, string>): Response {
+	return new Response(null, { status, headers });
+}
+
 function isAuthorized(req: Request, tokens: ReadonlySet<string>): boolean {
 	if (tokens.size === 0) return true;
 	const header = req.headers.get("authorization");
@@ -99,6 +105,211 @@ async function parseBody<T>(
 const REFRESH_ROUTE = /^\/v1\/credential\/(\d+)\/refresh$/;
 const DISABLE_ROUTE = /^\/v1\/credential\/(\d+)\/disable$/;
 
+const MAX_SNAPSHOT_WAIT_MS = 30_000;
+const DISABLED_NEXT_SWEEP_IN_MS = Number.MAX_SAFE_INTEGER;
+
+function snapshotHeaders(generation: number): Record<string, string> {
+	return {
+		ETag: `"${generation}"`,
+		"Cache-Control": "no-store",
+	};
+}
+
+function parseGenerationTag(header: string | null): number | undefined {
+	if (!header) return undefined;
+	let value = header.trim();
+	if (value.startsWith("W/")) value = value.slice(2).trim();
+	if (value.startsWith('"') && value.endsWith('"') && value.length >= 2) {
+		value = value.slice(1, -1);
+	}
+	const generation = Number(value);
+	if (!Number.isInteger(generation) || generation < 0) return undefined;
+	return generation;
+}
+
+function parseWaitMs(url: URL): number {
+	const raw = url.searchParams.get("wait");
+	if (raw === null) return 0;
+	const parsed = Number(raw);
+	if (!Number.isFinite(parsed)) return 0;
+	return Math.max(0, Math.min(MAX_SNAPSHOT_WAIT_MS, Math.trunc(parsed)));
+}
+
+function delayResult(ms: number): { promise: Promise<"timeout">; cancel: () => void } {
+	const done = Promise.withResolvers<"timeout">();
+	const timer = setTimeout(() => done.resolve("timeout"), ms);
+	timer.unref?.();
+	return {
+		promise: done.promise,
+		cancel: () => clearTimeout(timer),
+	};
+}
+
+class GenerationGate {
+	readonly #storage: AuthStorage;
+	readonly #unsubscribe: () => void;
+	#waiters: Map<number, Set<() => void>> = new Map();
+
+	constructor(storage: AuthStorage) {
+		this.#storage = storage;
+		this.#unsubscribe = storage.onGenerationChanged(generation => this.#wake(generation));
+	}
+
+	waitForChange(afterGeneration: number, signal: AbortSignal): Promise<"changed" | "aborted"> {
+		if (this.#storage.getGeneration() !== afterGeneration) return Promise.resolve("changed");
+		if (signal.aborted) return Promise.resolve("aborted");
+
+		const done = Promise.withResolvers<"changed" | "aborted">();
+		let settled = false;
+		const waiters = this.#waiters.get(afterGeneration) ?? new Set<() => void>();
+		this.#waiters.set(afterGeneration, waiters);
+
+		const cleanup = (): void => {
+			signal.removeEventListener("abort", onAbort);
+			waiters.delete(resolveChanged);
+			if (waiters.size === 0) this.#waiters.delete(afterGeneration);
+		};
+		const settle = (result: "changed" | "aborted"): void => {
+			if (settled) return;
+			settled = true;
+			cleanup();
+			done.resolve(result);
+		};
+		const resolveChanged = (): void => settle("changed");
+		const onAbort = (): void => settle("aborted");
+
+		waiters.add(resolveChanged);
+		signal.addEventListener("abort", onAbort, { once: true });
+		return done.promise;
+	}
+
+	close(): void {
+		this.#unsubscribe();
+		for (const waiters of this.#waiters.values()) {
+			for (const resolve of waiters) resolve();
+		}
+		this.#waiters.clear();
+	}
+
+	#wake(generation: number): void {
+		for (const [waitingFor, waiters] of [...this.#waiters]) {
+			if (generation <= waitingFor) continue;
+			for (const resolve of [...waiters]) resolve();
+		}
+	}
+}
+
+function resolveRefresherSchedule(
+	refresher: AuthBrokerRefresher | undefined,
+	serverNowMs: number,
+): { wire: RefresherSchedule; nextSweepAt: number } {
+	if (!refresher) {
+		return {
+			wire: {
+				enabled: false,
+				intervalMs: 0,
+				skewMs: 0,
+				nextSweepInMs: DISABLED_NEXT_SWEEP_IN_MS,
+			},
+			nextSweepAt: DISABLED_NEXT_SWEEP_IN_MS,
+		};
+	}
+	const schedule: AuthBrokerRefresherSchedule = refresher.getSchedule();
+	return {
+		wire: {
+			enabled: schedule.enabled,
+			intervalMs: schedule.intervalMs,
+			skewMs: schedule.skewMs,
+			nextSweepInMs: Math.max(0, schedule.nextSweepAt - serverNowMs),
+		},
+		nextSweepAt: schedule.nextSweepAt,
+	};
+}
+
+function computeRotatesInMs(
+	entry: { credential: { type: string; expires?: number } },
+	schedule: RefresherSchedule,
+	nextSweepAt: number,
+	serverNowMs: number,
+): number | null {
+	if (!schedule.enabled || entry.credential.type !== "oauth") return null;
+	const expires = entry.credential.expires;
+	if (typeof expires !== "number" || !Number.isFinite(expires)) return null;
+	if (!Number.isFinite(nextSweepAt) || !Number.isFinite(schedule.intervalMs) || schedule.intervalMs <= 0) return null;
+
+	const dueAt = expires - schedule.skewMs;
+	const eligibleAt = Math.max(serverNowMs, dueAt);
+	if (dueAt <= serverNowMs && nextSweepAt <= serverNowMs) return 0;
+	if (nextSweepAt >= eligibleAt) return Math.max(0, nextSweepAt - serverNowMs);
+	const steps = Math.ceil((eligibleAt - nextSweepAt) / schedule.intervalMs);
+	const rotatesAt = nextSweepAt + steps * schedule.intervalMs;
+	return Math.max(0, rotatesAt - serverNowMs);
+}
+
+function buildSnapshot(storage: AuthStorage, refresher: AuthBrokerRefresher | undefined): SnapshotResponse {
+	const serverNowMs = Date.now();
+	const base = storage.exportSnapshot();
+	const { wire, nextSweepAt } = resolveRefresherSchedule(refresher, serverNowMs);
+	const credentials: SnapshotEntry[] = base.credentials.map(entry => ({
+		...entry,
+		rotatesInMs: computeRotatesInMs(entry, wire, nextSweepAt, serverNowMs),
+	}));
+	return {
+		generation: base.generation,
+		generatedAt: base.generatedAt,
+		serverNowMs,
+		refresher: wire,
+		credentials,
+	};
+}
+
+async function serveSnapshot(
+	req: Request,
+	url: URL,
+	storage: AuthStorage,
+	gate: GenerationGate,
+	refresher: AuthBrokerRefresher | undefined,
+	peer: string,
+): Promise<Response> {
+	await storage.reload();
+	let currentGeneration = storage.getGeneration();
+	const clientGeneration = parseGenerationTag(req.headers.get("if-none-match"));
+	const waitMs = parseWaitMs(url);
+
+	if (clientGeneration === undefined || currentGeneration !== clientGeneration || waitMs <= 0) {
+		const body = buildSnapshot(storage, refresher);
+		logger.info("auth-broker snapshot served", {
+			peer,
+			credentials: body.credentials.length,
+			generation: body.generation,
+		});
+		return json(200, body, snapshotHeaders(body.generation));
+	}
+
+	const delay = delayResult(waitMs);
+	const waitController = new AbortController();
+	const waitSignal = AbortSignal.any([req.signal, waitController.signal]);
+	const result = await Promise.race([gate.waitForChange(clientGeneration, waitSignal), delay.promise]);
+	delay.cancel();
+	waitController.abort();
+	if (result === "aborted" || req.signal.aborted) return empty(499, snapshotHeaders(currentGeneration));
+
+	await storage.reload();
+	currentGeneration = storage.getGeneration();
+	if (currentGeneration !== clientGeneration) {
+		const body = buildSnapshot(storage, refresher);
+		logger.info("auth-broker snapshot long-poll changed", {
+			peer,
+			credentials: body.credentials.length,
+			generation: body.generation,
+		});
+		return json(200, body, snapshotHeaders(body.generation));
+	}
+
+	logger.info("auth-broker snapshot long-poll unchanged", { peer, generation: currentGeneration });
+	return empty(304, snapshotHeaders(currentGeneration));
+}
+
 /** Boot the broker. Caller owns lifecycle; `handle.close()` to stop. */
 export function startAuthBroker(opts: AuthBrokerServerOptions): AuthBrokerServerHandle {
 	const bind = parseBind(opts.bind ?? DEFAULT_AUTH_BROKER_BIND);
@@ -113,6 +324,7 @@ export function startAuthBroker(opts: AuthBrokerServerOptions): AuthBrokerServer
 				refreshIntervalMs: opts.refreshIntervalMs ?? DEFAULT_REFRESH_INTERVAL_MS,
 			});
 	refresher?.start();
+	const generationGate = new GenerationGate(opts.storage);
 
 	const server = Bun.serve({
 		hostname: bind.hostname,
@@ -132,10 +344,7 @@ export function startAuthBroker(opts: AuthBrokerServerOptions): AuthBrokerServer
 					return json(401, { error: "unauthorized" });
 				}
 				if (req.method === "GET" && pathname === "/v1/snapshot") {
-					await opts.storage.reload();
-					const body: SnapshotResponse = opts.storage.exportSnapshot();
-					logger.info("auth-broker snapshot served", { peer, credentials: body.credentials.length });
-					return json(200, body);
+					return serveSnapshot(req, url, opts.storage, generationGate, refresher, peer);
 				}
 				if (req.method === "GET" && pathname === "/v1/usage") {
 					try {
@@ -161,7 +370,7 @@ export function startAuthBroker(opts: AuthBrokerServerOptions): AuthBrokerServer
 				if (refreshMatch) {
 					const id = Number.parseInt(refreshMatch[1], 10);
 					try {
-						const entry = await opts.storage.forceRefreshCredentialById(id, req.signal);
+						const entry = await opts.storage.refreshCredentialById(id, req.signal);
 						const body: CredentialRefreshResponse = { entry };
 						logger.info("auth-broker credential refreshed", {
 							id,
@@ -238,6 +447,7 @@ export function startAuthBroker(opts: AuthBrokerServerOptions): AuthBrokerServer
 		hostname: boundHost,
 		close: async () => {
 			refresher?.stop();
+			generationGate.close();
 			server.stop(true);
 		},
 	};
diff --git a/packages/ai/src/auth-broker/types.ts b/packages/ai/src/auth-broker/types.ts
index bcabd4c7c..5deb01c74 100644
--- a/packages/ai/src/auth-broker/types.ts
+++ b/packages/ai/src/auth-broker/types.ts
@@ -15,8 +15,23 @@ export interface HealthzResponse {
 	version?: string;
 }
 
+export interface RefresherSchedule {
+	enabled: boolean;
+	intervalMs: number;
+	skewMs: number;
+	nextSweepInMs: number;
+}
+
+export type SnapshotEntry = AuthCredentialSnapshotEntry & {
+	rotatesInMs: number | null;
+};
+
 /** GET /v1/snapshot response body. */
-export type SnapshotResponse = AuthCredentialSnapshot;
+export interface SnapshotResponse extends Omit<AuthCredentialSnapshot, "credentials"> {
+	serverNowMs: number;
+	refresher: RefresherSchedule;
+	credentials: SnapshotEntry[];
+}
 
 /** GET /v1/usage response body — matches the local `AuthStorage.fetchUsageReports` shape. */
 export interface UsageResponse {
diff --git a/packages/ai/src/auth-broker/wire-schemas.ts b/packages/ai/src/auth-broker/wire-schemas.ts
index 34411f0fa..8fe755c5b 100644
--- a/packages/ai/src/auth-broker/wire-schemas.ts
+++ b/packages/ai/src/auth-broker/wire-schemas.ts
@@ -66,7 +66,7 @@ export const snapshotCredentialSchema = z.discriminatedUnion("type", [
 
 // ─── Snapshot ──────────────────────────────────────────────────────────────
 
-export const snapshotEntrySchema = z
+export const credentialSnapshotEntrySchema = z
 	.object({
 		id: z.number().int(),
 		provider: z.string().min(1),
@@ -75,9 +75,27 @@ export const snapshotEntrySchema = z
 	})
 	.strict();
 
+export const snapshotEntrySchema = credentialSnapshotEntrySchema
+	.extend({
+		rotatesInMs: z.number().nullable(),
+	})
+	.strict();
+
+export const refresherScheduleSchema = z
+	.object({
+		enabled: z.boolean(),
+		intervalMs: z.number(),
+		skewMs: z.number(),
+		nextSweepInMs: z.number(),
+	})
+	.strict();
+
 export const snapshotResponseSchema = z
 	.object({
+		generation: z.number().int(),
 		generatedAt: z.number(),
+		serverNowMs: z.number(),
+		refresher: refresherScheduleSchema,
 		credentials: z.array(snapshotEntrySchema),
 	})
 	.strict();
@@ -110,7 +128,7 @@ export const usageResponseSchema = z
 
 export const credentialRefreshResponseSchema = z
 	.object({
-		entry: snapshotEntrySchema,
+		entry: credentialSnapshotEntrySchema,
 	})
 	.strict();
 
@@ -139,6 +157,6 @@ export const credentialUploadRequestSchema = z
 
 export const credentialUploadResponseSchema = z
 	.object({
-		entries: z.array(snapshotEntrySchema),
+		entries: z.array(credentialSnapshotEntrySchema),
 	})
 	.strict();
diff --git a/packages/ai/src/auth-storage.ts b/packages/ai/src/auth-storage.ts
index de038c3e7..34fdfe2ea 100644
--- a/packages/ai/src/auth-storage.ts
+++ b/packages/ai/src/auth-storage.ts
@@ -110,6 +110,7 @@ export interface AuthCredentialSnapshotEntry {
  * served by the auth-broker server on `GET /v1/snapshot`.
  */
 export interface AuthCredentialSnapshot {
+	generation: number;
 	generatedAt: number;
 	credentials: AuthCredentialSnapshotEntry[];
 }
@@ -156,6 +157,13 @@ export interface AuthCredentialStore {
 		credential: OAuthCredential,
 		signal?: AbortSignal,
 	): Promise<OAuthCredentials>;
+	/**
+	 * Optional async pre-read hook invoked after AuthStorage selects a stored
+	 * credential but before it returns that credential for an outbound request.
+	 * Remote broker stores use this to wait out imminent rotations and refresh
+	 * their local snapshot before the caller sees a stale access token.
+	 */
+	prepareForRequest?(credentialId: number, opts?: { signal?: AbortSignal }): Promise<void>;
 	/**
 	 * Optional store-supplied aggregate usage fetch. When present, `AuthStorage`
 	 * routes `fetchUsageReports()` here instead of fanning out per-credential.
@@ -183,6 +191,13 @@ export interface AuthCredentialStore {
 	 * `signal` propagates the agent's cancel down to the broker fetch.
 	 */
 	getUsageReport?(provider: Provider, credential: OAuthCredential, signal?: AbortSignal): Promise<UsageReport | null>;
+	/**
+	 * Optional store hook to invalidate a specific credential after the upstream
+	 * provider returned 401 on a supposedly-fresh key. Remote stores force the
+	 * broker to re-issue the row; local stores can leave it undefined and let
+	 * {@link AuthStorage.invalidateCredentialMatching} fall back to `reload()`.
+	 */
+	markCredentialSuspect?(credentialId: number, opts?: { signal?: AbortSignal }): Promise<void>;
 }
 
 // ─────────────────────────────────────────────────────────────────────────────
@@ -419,6 +434,50 @@ function raceUsageWithSignal<T>(promise: Promise<T>, signal: AbortSignal | undef
 	});
 }
 
+function raceCredentialRefreshWithSignal<T>(
+	promise: Promise<T>,
+	signal: AbortSignal | undefined,
+	message = "credential refresh aborted",
+): Promise<T> {
+	if (!signal) return promise;
+	if (signal.aborted) return Promise.reject(new Error(message));
+	const abort = Promise.withResolvers<never>();
+	const onAbort = (): void => abort.reject(new Error(message));
+	signal.addEventListener("abort", onAbort, { once: true });
+	return Promise.race([promise, abort.promise]).finally(() => {
+		signal.removeEventListener("abort", onAbort);
+	});
+}
+
+function authCredentialEquals(left: AuthCredential, right: AuthCredential): boolean {
+	if (left.type !== right.type) return false;
+	if (left.type === "api_key") {
+		return right.type === "api_key" && left.key === right.key;
+	}
+	if (right.type !== "oauth") return false;
+	return (
+		left.access === right.access &&
+		left.refresh === right.refresh &&
+		left.expires === right.expires &&
+		left.accountId === right.accountId &&
+		left.email === right.email &&
+		left.projectId === right.projectId &&
+		left.enterpriseUrl === right.enterpriseUrl
+	);
+}
+
+function storedCredentialArraysEqual(left: StoredCredential[], right: StoredCredential[]): boolean {
+	if (left.length !== right.length) return false;
+	for (let index = 0; index < left.length; index += 1) {
+		const leftEntry = left[index];
+		const rightEntry = right[index];
+		if (!leftEntry || !rightEntry) return false;
+		if (leftEntry.id !== rightEntry.id) return false;
+		if (!authCredentialEquals(leftEntry.credential, rightEntry.credential)) return false;
+	}
+	return true;
+}
+
 // ─────────────────────────────────────────────────────────────────────────────
 // Usage Cache (backed by AuthCredentialStore)
 // ─────────────────────────────────────────────────────────────────────────────
@@ -502,6 +561,9 @@ export class AuthStorage {
 	 * but a process that runs without subscribers for a long time shouldn't grow this unboundedly).
 	 */
 	#pendingDisabledEvents: CredentialDisabledEvent[] = [];
+	#generation = 1;
+	#generationListeners: Set<(generation: number) => void> = new Set();
+	#oauthRefreshInFlight: Map<number, Promise<AuthCredentialSnapshotEntry>> = new Map();
 	#closed = false;
 
 	constructor(store: AuthCredentialStore, options: AuthStorageOptions = {}) {
@@ -549,6 +611,32 @@ export class AuthStorage {
 		this.#store.close();
 	}
 
+	getGeneration(): number {
+		return this.#generation;
+	}
+
+	onGenerationChanged(listener: (generation: number) => void): () => void {
+		this.#generationListeners.add(listener);
+		return () => {
+			this.#generationListeners.delete(listener);
+		};
+	}
+
+	offGenerationChanged(listener: (generation: number) => void): void {
+		this.#generationListeners.delete(listener);
+	}
+
+	#bumpGeneration(reason: string): void {
+		this.#generation += 1;
+		for (const listener of [...this.#generationListeners]) {
+			try {
+				listener(this.#generation);
+			} catch (error) {
+				logger.debug("AuthStorage generation listener failed", { reason, error: String(error) });
+			}
+		}
+	}
+
 	/**
 	 * Subscribe to {@link CredentialDisabledEvent}s. Multiple subscribers are supported and
 	 * each fires for every disable event; subscribers are invoked in registration order with
@@ -653,7 +741,15 @@ export class AuthStorage {
 				dedupedGrouped.set(provider, deduped);
 			}
 		}
-		this.#data = dedupedGrouped;
+
+		const removedProviders = new Set(this.#data.keys());
+		for (const [provider, entries] of dedupedGrouped) {
+			this.#setStoredCredentials(provider, entries);
+			removedProviders.delete(provider);
+		}
+		for (const provider of removedProviders) {
+			this.#setStoredCredentials(provider, []);
+		}
 	}
 
 	/**
@@ -672,11 +768,14 @@ export class AuthStorage {
 	 * @param credentials - Array of stored credentials to cache
 	 */
 	#setStoredCredentials(provider: string, credentials: StoredCredential[]): void {
+		const current = this.#data.get(provider) ?? [];
+		if (storedCredentialArraysEqual(current, credentials)) return;
 		if (credentials.length === 0) {
 			this.#data.delete(provider);
 		} else {
 			this.#data.set(provider, credentials);
 		}
+		this.#bumpGeneration("credentials");
 	}
 
 	#resolveOAuthDedupeIdentityKey(provider: string, credential: OAuthCredential): string | null {
@@ -999,7 +1098,7 @@ export class AuthStorage {
 	 */
 	async remove(provider: string): Promise<void> {
 		this.#store.deleteAuthCredentialsForProvider(provider, "deleted by user");
-		this.#data.delete(provider);
+		this.#setStoredCredentials(provider, []);
 		this.#resetProviderAssignments(provider);
 	}
 
@@ -2086,7 +2185,7 @@ export class AuthStorage {
 		// timer so the timeout doesn't pin the process and clear it on the happy
 		// path so memory drops immediately.
 		const timer = setTimeout(() => timeoutSignal.resolve(null), usageTimeout);
-		(timer as { unref?: () => void }).unref?.();
+		timer.unref?.();
 		const usageResults = await Promise.race([usagePromise, timeoutSignal.promise]).then(result => {
 			clearTimeout(timer);
 			return (
@@ -2295,28 +2394,54 @@ export class AuthStorage {
 		// take priority over the floor timeout.
 		let timeout: NodeJS.Timeout | undefined;
 		let onAbort: (() => void) | undefined;
-		const cancellationPromise = new Promise<never>((_, reject) => {
-			timeout = setTimeout(
-				() => reject(new Error(`OAuth token refresh timed out for provider: ${provider}`)),
-				DEFAULT_OAUTH_REFRESH_TIMEOUT_MS,
-			);
-			if (signal) {
-				if (signal.aborted) {
-					reject(new Error("OAuth token refresh aborted by caller"));
-					return;
-				}
-				onAbort = () => reject(new Error("OAuth token refresh aborted by caller"));
+		const cancellation = Promise.withResolvers<never>();
+		timeout = setTimeout(
+			() => cancellation.reject(new Error(`OAuth token refresh timed out for provider: ${provider}`)),
+			DEFAULT_OAUTH_REFRESH_TIMEOUT_MS,
+		);
+		if (signal) {
+			if (signal.aborted) {
+				cancellation.reject(new Error("OAuth token refresh aborted by caller"));
+			} else {
+				onAbort = () => cancellation.reject(new Error("OAuth token refresh aborted by caller"));
 				signal.addEventListener("abort", onAbort, { once: true });
 			}
-		});
+		}
 		try {
-			return await Promise.race([refreshPromise, cancellationPromise]);
+			return await Promise.race([refreshPromise, cancellation.promise]);
 		} finally {
 			if (timeout) clearTimeout(timeout);
 			if (signal && onAbort) signal.removeEventListener("abort", onAbort);
 		}
 	}
 
+	async #prepareOAuthCredentialForRequest(
+		provider: string,
+		selection: { credential: OAuthCredential; index: number },
+		options: AuthApiKeyOptions | undefined,
+	): Promise<boolean> {
+		const prepare = this.#store.prepareForRequest?.bind(this.#store);
+		if (!prepare) return true;
+		const stored = this.#getStoredCredentials(provider);
+		const selected = stored[selection.index];
+		if (!selected || selected.credential.type !== "oauth") return false;
+
+		await prepare(selected.id, { signal: options?.signal });
+
+		const latestRows = this.#store.listAuthCredentials(provider);
+		this.#setStoredCredentials(
+			provider,
+			latestRows.map(row => ({ id: row.id, credential: row.credential })),
+		);
+		const latestIndex = latestRows.findIndex(row => row.id === selected.id);
+		if (latestIndex === -1) return false;
+		const latest = latestRows[latestIndex];
+		if (!latest || latest.credential.type !== "oauth") return false;
+		selection.index = latestIndex;
+		selection.credential = latest.credential;
+		return true;
+	}
+
 	/** Attempts to use a single OAuth credential, checking usage and refreshing token. */
 	async #tryOAuthCredential(
 		provider: Provider,
@@ -2343,6 +2468,10 @@ export class AuthStorage {
 			return undefined;
 		}
 
+		if (!(await this.#prepareOAuthCredentialForRequest(provider, selection, options))) {
+			return undefined;
+		}
+
 		const requiresProModel = requiresOpenAICodexProModel(provider, options?.modelId);
 		const applyProFilter = enforceProRequirement ?? requiresProModel;
 		let usage: UsageReport | null = null;
@@ -2589,6 +2718,54 @@ export class AuthStorage {
 		return this.#fallbackResolver?.(provider) ?? undefined;
 	}
 
+	#extractStructuredApiKeyToken(apiKey: string): string | undefined {
+		if (!apiKey.startsWith("{")) return undefined;
+		try {
+			const parsed = JSON.parse(apiKey) as { token?: unknown };
+			return typeof parsed.token === "string" ? parsed.token : undefined;
+		} catch {
+			return undefined;
+		}
+	}
+
+	async #credentialMatchesApiKey(credential: AuthCredential, apiKey: string): Promise<boolean> {
+		if (credential.type === "api_key") {
+			return (await this.#configValueResolver(credential.key)) === apiKey;
+		}
+		if (credential.access === apiKey) return true;
+		return this.#extractStructuredApiKeyToken(apiKey) === credential.access;
+	}
+
+	async invalidateCredentialMatching(provider: string, apiKey: string, signal?: AbortSignal): Promise<boolean> {
+		const stored = this.#getStoredCredentials(provider);
+		let matchedId: number | undefined;
+		for (const entry of stored) {
+			if (await this.#credentialMatchesApiKey(entry.credential, apiKey)) {
+				matchedId = entry.id;
+				break;
+			}
+		}
+
+		if (matchedId === undefined) {
+			await this.reload();
+			return false;
+		}
+
+		const markSuspect = this.#store.markCredentialSuspect?.bind(this.#store);
+		if (markSuspect) {
+			await markSuspect(matchedId, { signal });
+		} else {
+			await this.reload();
+		}
+
+		const latestRows = this.#store.listAuthCredentials(provider);
+		this.#setStoredCredentials(
+			provider,
+			latestRows.map(row => ({ id: row.id, credential: row.credential })),
+		);
+		return true;
+	}
+
 	// ─── Auth Broker integration ────────────────────────────────────────────
 
 	/**
@@ -2614,7 +2791,32 @@ export class AuthStorage {
 				});
 			}
 		}
-		return { generatedAt: Date.now(), credentials: entries };
+		return { generation: this.#generation, generatedAt: Date.now(), credentials: entries };
+	}
+
+	/**
+	 * Refresh the OAuth credential with the given id through a per-credential
+	 * single-flight. Concurrent callers for the same row await the same upstream
+	 * refresh attempt, which is required for providers that rotate refresh tokens
+	 * on every successful refresh.
+	 */
+	async refreshCredentialById(id: number, signal?: AbortSignal): Promise<AuthCredentialSnapshotEntry> {
+		const existing = this.#oauthRefreshInFlight.get(id);
+		if (existing) return raceCredentialRefreshWithSignal(existing, signal);
+
+		const promise = (async () => {
+			this.#bumpGeneration("credential-refresh-start");
+			try {
+				return await this.#forceRefreshCredentialByIdUnshared(id, signal);
+			} catch (error) {
+				this.#bumpGeneration("credential-refresh-failure");
+				throw error;
+			} finally {
+				this.#oauthRefreshInFlight.delete(id);
+			}
+		})();
+		this.#oauthRefreshInFlight.set(id, promise);
+		return raceCredentialRefreshWithSignal(promise, signal);
 	}
 
 	/**
@@ -2626,6 +2828,10 @@ export class AuthStorage {
 	 * Throws when no OAuth credential with that id is loaded.
 	 */
 	async forceRefreshCredentialById(id: number, signal?: AbortSignal): Promise<AuthCredentialSnapshotEntry> {
+		return this.refreshCredentialById(id, signal);
+	}
+
+	async #forceRefreshCredentialByIdUnshared(id: number, signal?: AbortSignal): Promise<AuthCredentialSnapshotEntry> {
 		for (const [provider, entries] of this.#data) {
 			const index = entries.findIndex(entry => entry.id === id);
 			if (index === -1) continue;
diff --git a/packages/ai/src/stream.ts b/packages/ai/src/stream.ts
index 93c72dcb6..91dc4ecf4 100644
--- a/packages/ai/src/stream.ts
+++ b/packages/ai/src/stream.ts
@@ -1,7 +1,7 @@
 import * as fs from "node:fs";
 import * as os from "node:os";
 import * as path from "node:path";
-import { $env, $pickenv } from "@oh-my-pi/pi-utils";
+import { $env, $pickenv, extractHttpStatusFromError } from "@oh-my-pi/pi-utils";
 import { getCustomApi } from "./api-registry";
 import type { Effort } from "./model-thinking";
 import {
@@ -45,7 +45,6 @@ import { isSyntheticModel, streamSynthetic } from "./providers/synthetic";
 import type {
 	Api,
 	AssistantMessage,
-	AssistantMessageEventStream,
 	Context,
 	Model,
 	OptionsForApi,
@@ -54,6 +53,7 @@ import type {
 	ThinkingBudgets,
 	ToolChoice,
 } from "./types";
+import { AssistantMessageEventStream } from "./utils/event-stream";
 import { isFoundryEnabled } from "./utils/foundry";
 
 let cachedVertexAdcCredentialsExists: boolean | null = null;
@@ -310,36 +310,87 @@ export function streamSimple<TApi extends Api>(
 		throw new Error(`No API key for provider: ${model.provider}`);
 	}
 
-	// GitLab Duo - wraps Anthropic/OpenAI behind GitLab AI Gateway direct access tokens
-	if (isGitLabDuoModel(model)) {
-		return streamGitLabDuo(model, context, {
-			...options,
-			apiKey,
-		});
+	// Dispatch the actual provider stream. `onAuthError` is consumed here and
+	// stripped from downstream options so the inner recursion can't retry
+	// again (one-shot per outer call).
+	const dispatch = (effectiveKey: string): AssistantMessageEventStream => {
+		const downstreamOptions = options ? { ...options, apiKey: effectiveKey, onAuthError: undefined } : undefined;
+
+		// GitLab Duo - wraps Anthropic/OpenAI behind GitLab AI Gateway direct access tokens
+		if (isGitLabDuoModel(model)) {
+			return streamGitLabDuo(model, context, { ...downstreamOptions, apiKey: effectiveKey });
+		}
+
+		// Kimi Code - route to dedicated handler that wraps OpenAI or Anthropic API
+		if (isKimiModel(model)) {
+			return streamKimi(model as Model<"openai-completions">, context, {
+				...downstreamOptions,
+				apiKey: effectiveKey,
+				format: options?.kimiApiFormat ?? "anthropic",
+			});
+		}
+
+		// Synthetic - route to dedicated handler that wraps OpenAI or Anthropic API
+		if (isSyntheticModel(model)) {
+			return streamSynthetic(model as Model<"openai-completions">, context, {
+				...downstreamOptions,
+				apiKey: effectiveKey,
+				format: options?.syntheticApiFormat ?? "openai", // Default to OpenAI format
+			});
+		}
+
+		const providerOptions = mapOptionsForApi(model, downstreamOptions, effectiveKey);
+		return stream(model, context, providerOptions);
+	};
+
+	if (!options?.onAuthError) {
+		return dispatch(apiKey);
 	}
 
-	// Kimi Code - route to dedicated handler that wraps OpenAI or Anthropic API
-	if (isKimiModel(model)) {
-		// Pass raw SimpleStreamOptions - streamKimi handles mapping internally
-		return streamKimi(model as Model<"openai-completions">, context, {
-			...options,
-			apiKey,
-			format: options?.kimiApiFormat ?? "anthropic",
-		});
-	}
-
-	// Synthetic - route to dedicated handler that wraps OpenAI or Anthropic API
-	if (isSyntheticModel(model)) {
-		// Pass raw SimpleStreamOptions - streamSynthetic handles mapping internally
-		return streamSynthetic(model as Model<"openai-completions">, context, {
-			...options,
-			apiKey,
-			format: options?.syntheticApiFormat ?? "openai", // Default to OpenAI format
-		});
-	}
-
-	const providerOptions = mapOptionsForApi(model, options, apiKey);
-	return stream(model, context, providerOptions);
+	// One-shot 401 recovery. Forward events from the inner stream; on a 401
+	// before any event has fired, ask `onAuthError` for a new key and re-dispatch
+	// once. After the first event the request is committed — we cannot replay
+	// partial assistant content, so we surface the error normally.
+	const onAuthError = options.onAuthError;
+	const outer = new AssistantMessageEventStream();
+	const inner = dispatch(apiKey);
+	let emitted = false;
+	void (async () => {
+		try {
+			for await (const event of inner) {
+				emitted = true;
+				outer.push(event);
+				if (outer.done) return;
+			}
+			if (!outer.done) outer.end(await inner.result());
+		} catch (error) {
+			if (emitted || extractHttpStatusFromError(error) !== 401) {
+				outer.fail(error);
+				return;
+			}
+			let nextKey: string | undefined;
+			try {
+				nextKey = await onAuthError(model.provider, apiKey, error);
+			} catch {
+				nextKey = undefined;
+			}
+			if (!nextKey || nextKey === apiKey) {
+				outer.fail(error);
+				return;
+			}
+			try {
+				const retried = dispatch(nextKey);
+				for await (const event of retried) {
+					outer.push(event);
+					if (outer.done) return;
+				}
+				if (!outer.done) outer.end(await retried.result());
+			} catch (retryError) {
+				outer.fail(retryError);
+			}
+		}
+	})();
+	return outer;
 }
 
 export async function completeSimple<TApi extends Api>(
diff --git a/packages/ai/src/types.ts b/packages/ai/src/types.ts
index 51a295257..4d8e82e1e 100644
--- a/packages/ai/src/types.ts
+++ b/packages/ai/src/types.ts
@@ -235,6 +235,11 @@ export interface StreamOptions {
 	maxTokens?: number;
 	signal?: AbortSignal;
 	apiKey?: string;
+	/**
+	 * Called when a provider returns 401 before any assistant event has been
+	 * emitted. Returning a different key retries the provider request once.
+	 */
+	onAuthError?: (provider: string, apiKey: string, error: unknown) => Promise<string | undefined>;
 	cacheRetention?: CacheRetention;
 	/**
 	 * Additional headers to include in provider requests.
diff --git a/packages/ai/test/auth-broker-wire.test.ts b/packages/ai/test/auth-broker-wire.test.ts
index d3d3d57c1..5cbd92383 100644
--- a/packages/ai/test/auth-broker-wire.test.ts
+++ b/packages/ai/test/auth-broker-wire.test.ts
@@ -76,7 +76,9 @@ describe("auth-broker wire surface", () => {
 		expect(unauthorized.status).toBe(401);
 
 		const client = new AuthBrokerClient({ url: handle!.url, token });
-		const snapshot = await client.fetchSnapshot();
+		const snapshotResult = await client.fetchSnapshot();
+		if (snapshotResult.status !== 200) throw new Error("expected snapshot");
+		const snapshot = snapshotResult.snapshot;
 		expect(snapshot.credentials).toHaveLength(1);
 		const entry = snapshot.credentials[0];
 		expect(entry.provider).toBe("anthropic");
@@ -98,8 +100,9 @@ describe("auth-broker wire surface", () => {
 		};
 		vi.spyOn(oauthUtils, "refreshOAuthToken").mockResolvedValue(refreshed);
 
-		const initialSnapshot = await new AuthBrokerClient({ url: handle!.url, token }).fetchSnapshot();
-		const id = initialSnapshot.credentials[0].id;
+		const initialResult = await new AuthBrokerClient({ url: handle!.url, token }).fetchSnapshot();
+		if (initialResult.status !== 200) throw new Error("expected snapshot");
+		const id = initialResult.snapshot.credentials[0].id;
 
 		const client = new AuthBrokerClient({ url: handle!.url, token });
 		const result = await client.refreshCredential(id);
@@ -117,14 +120,16 @@ describe("auth-broker wire surface", () => {
 
 	test("POST /v1/credential/:id/disable soft-deletes the credential and surfaces 404 thereafter", async () => {
 		const client = new AuthBrokerClient({ url: handle!.url, token });
-		const initialSnapshot = await client.fetchSnapshot();
-		const id = initialSnapshot.credentials[0].id;
+		const initialResult = await client.fetchSnapshot();
+		if (initialResult.status !== 200) throw new Error("expected snapshot");
+		const id = initialResult.snapshot.credentials[0].id;
 
 		const result = await client.disableCredential(id, "revoked by user");
 		expect(result.ok).toBe(true);
 
-		const after = await client.fetchSnapshot();
-		expect(after.credentials).toHaveLength(0);
+		const afterResult = await client.fetchSnapshot();
+		if (afterResult.status !== 200) throw new Error("expected snapshot");
+		expect(afterResult.snapshot.credentials).toHaveLength(0);
 
 		await expect(client.refreshCredential(id)).rejects.toThrow();
 	});
diff --git a/packages/ai/test/remote-auth-store.test.ts b/packages/ai/test/remote-auth-store.test.ts
index ae22162cf..4003535c4 100644
--- a/packages/ai/test/remote-auth-store.test.ts
+++ b/packages/ai/test/remote-auth-store.test.ts
@@ -71,7 +71,9 @@ describe("RemoteAuthCredentialStore + AuthStorage integration", () => {
 		const refreshSpy = vi.spyOn(oauthUtils, "refreshOAuthToken").mockResolvedValue(rotated);
 
 		const brokerClient = new AuthBrokerClient({ url: handle!.url, token });
-		const initialSnapshot = await brokerClient.fetchSnapshot();
+		const initialResult = await brokerClient.fetchSnapshot();
+		if (initialResult.status !== 200) throw new Error("expected snapshot");
+		const initialSnapshot = initialResult.snapshot;
 		expect(initialSnapshot.credentials).toHaveLength(1);
 
 		const remoteStore = new RemoteAuthCredentialStore({
@@ -120,7 +122,13 @@ describe("RemoteAuthCredentialStore + AuthStorage integration", () => {
 		const brokerClient = new AuthBrokerClient({ url: handle!.url, token });
 		const remoteStore = new RemoteAuthCredentialStore({
 			client: brokerClient,
-			initialSnapshot: { generatedAt: 0, credentials: [] },
+			initialSnapshot: {
+				generation: 0,
+				generatedAt: 0,
+				serverNowMs: 0,
+				refresher: { enabled: false, intervalMs: 0, skewMs: 0, nextSweepInMs: Number.MAX_SAFE_INTEGER },
+				credentials: [],
+			},
 		});
 
 		const reportForA = {
diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md
index 618941d53..3f3909deb 100644
--- a/packages/coding-agent/CHANGELOG.md
+++ b/packages/coding-agent/CHANGELOG.md
@@ -1,6 +1,7 @@
 # Changelog
 
 ## [Unreleased]
+
 ### Breaking Changes
 
 - Renamed the embedded-documentation internal URL scheme from `pi://` to `omp://`. `OmpProtocolHandler` replaces `PiProtocolHandler`; update any external references accordingly.
@@ -33,6 +34,7 @@
 
 ### Fixed
 
+- Fixed `auth-broker` migration, `auth-gateway` startup, and `discoverAuthStorage` to fail fast with a clear error when the broker snapshot endpoint returns a non-200 response
 - Fixed `omp auth-broker migrate` to skip local placeholder `<authenticated>` API credentials (not real keys) when exporting to a remote broker
 - Fixed `auth-gateway` token initialization to avoid clobbering an existing token when multiple processes initialize it concurrently
 - Fixed `omp auth-gateway` request handling to reject unsupported OpenAI/Anthropic protocol controls with 400 instead of accepting and ignoring them, propagate upstream error/abort terminal states as failures, preserve Responses reasoning and completed text items, accept string/system Responses messages, and keep Anthropic tool-result ordering valid.
diff --git a/packages/coding-agent/src/cli/auth-broker-cli.ts b/packages/coding-agent/src/cli/auth-broker-cli.ts
index b3c3bc963..651d139b9 100644
--- a/packages/coding-agent/src/cli/auth-broker-cli.ts
+++ b/packages/coding-agent/src/cli/auth-broker-cli.ts
@@ -542,8 +542,9 @@ async function runMigrate(flags: AuthBrokerCommandArgs["flags"]): Promise<void>
 	}
 
 	const client = new AuthBrokerClient({ url: brokerConfig.url, token: brokerConfig.token });
-	const snapshot = await client.fetchSnapshot();
-	const existing = indexBrokerSnapshot(snapshot);
+	const snapshotResult = await client.fetchSnapshot();
+	if (snapshotResult.status !== 200) throw new Error("Auth broker returned no snapshot");
+	const existing = indexBrokerSnapshot(snapshotResult.snapshot);
 
 	const plan: MigratePlanEntry[] = [];
 	const skipped: MigrateSkip[] = [];
diff --git a/packages/coding-agent/src/cli/auth-gateway-cli.ts b/packages/coding-agent/src/cli/auth-gateway-cli.ts
index cf800eeca..5ffb98c68 100644
--- a/packages/coding-agent/src/cli/auth-gateway-cli.ts
+++ b/packages/coding-agent/src/cli/auth-gateway-cli.ts
@@ -123,7 +123,9 @@ function createBrokerClient(brokerConfig: AuthBrokerClientConfig): AuthBrokerCli
 }
 
 async function fetchBrokerSnapshot(client: AuthBrokerClient): Promise<SnapshotResponse> {
-	return client.fetchSnapshot();
+	const result = await client.fetchSnapshot();
+	if (result.status !== 200) throw new Error("Auth broker returned no initial snapshot");
+	return result.snapshot;
 }
 
 async function runServe(flags: AuthGatewayCommandArgs["flags"]): Promise<void> {
diff --git a/packages/coding-agent/src/eval/py/kernel.ts b/packages/coding-agent/src/eval/py/kernel.ts
index dfd43bbf3..cdeee7722 100644
--- a/packages/coding-agent/src/eval/py/kernel.ts
+++ b/packages/coding-agent/src/eval/py/kernel.ts
@@ -626,7 +626,7 @@ export class PythonKernel {
 		const exitedPromise = this.#exitedPromise;
 		const timeout = new Promise<null>(resolve => {
 			const timer = setTimeout(() => resolve(null), Math.max(0, timeoutMs));
-			(timer as { unref?: () => void }).unref?.();
+			timer.unref?.();
 		});
 		return Promise.race([exitedPromise.then(code => code as number | null), timeout]);
 	}
diff --git a/packages/coding-agent/src/sdk.ts b/packages/coding-agent/src/sdk.ts
index 2573a8a73..54b6b3c58 100644
--- a/packages/coding-agent/src/sdk.ts
+++ b/packages/coding-agent/src/sdk.ts
@@ -332,8 +332,9 @@ export async function discoverAuthStorage(agentDir: string = getDefaultAgentDir(
 	const brokerConfig = await resolveAuthBrokerConfig();
 	if (brokerConfig) {
 		const client = new AuthBrokerClient({ url: brokerConfig.url, token: brokerConfig.token });
-		const initialSnapshot = await client.fetchSnapshot();
-		const store = new RemoteAuthCredentialStore({ client, initialSnapshot });
+		const initialResult = await client.fetchSnapshot();
+		if (initialResult.status !== 200) throw new Error("Auth broker returned no initial snapshot");
+		const store = new RemoteAuthCredentialStore({ client, initialSnapshot: initialResult.snapshot });
 		// Refresh + usage hooks live on RemoteAuthCredentialStore; AuthStorage
 		// discovers them automatically when no explicit option overrides them.
 		const storage = new AuthStorage(store, {