fix(ai): name the real cause of failed briefings and stop retrying what cannot heal

The nightly briefing warm completed every night and read "all-failed",
because one run event held every account's annotations and the last
account won. Each candidate now runs in its own wide event and the run
reports every outcome through `reportJobRun`, so ok, dead sign-in,
withheld and failed accounts show side by side.

A dead Codex refresh token threw a bare error with no status, so the chain
read it as a network failure: the health ledger recorded hard_failed, no
"reconnect" hint appeared, and every night tried the same token. The
refresh now throws `CodexRefreshError` with a 401 for a dead grant
(invalid_grant / invalid_refresh_token on a 400 included), as do a missing
or undecryptable stored credential. The ledger records auth_failed, the
chain marks the primary credential expired, and the nightly pass reports
the account as auth_failed without queuing a retry.

A briefing the outbound screen withheld (a causal claim) failed outright,
and the 45-minute provider retry then asked the same question eight times
a night. The generator now makes one bounded repair that names the broken
rule and keeps the repaired payload only if it passes the same parse,
grounding and screen; a withheld outcome queues no retry.

A provider that keeps failing is paused: from five consecutive hard
failures the skip window starts at 30 minutes and doubles per further
failure, capped at a day. A paused provider is walked last and tried again
when the window lifts; one success resumes it. The operator card shows how
many accounts it is paused for and until when. After a 5xx from a
non-canonical OpenAI-compatible endpoint the client asks the free model
listing, and a configured model the endpoint no longer offers is named in
the error, on the hop line and on the card.

Also:
- The tiered context's live day band read `measured_at AT TIME ZONE tz`,
  which treats the stored UTC as local time; a far-from-UTC reader's
  evening readings landed on the next day. It converts from UTC first now.
- Google Health intraday heart rate re-reads two hours per sync instead of
  a day; daily summaries and sleep keep the day (#1023).
- The first filtered daily read that answers 400 and triggers the
  snake_case retry is recorded as a note, not an error (#1023).
- The single derived-score read goes through the derived cache with
  stale-while-revalidate, like the batch beside it.
This commit is contained in:
Marc-André Bombeck committed 2026-10-08 09:06:09 +02:00
1 parent 2210f823d6
commit 27342ee734
34 files changed
+2309 -261

No files matched your search

+65
View File
@@ -0,0 +1,65 @@
{
"en": {
"admin": {
"providerHealth": {
"statusPaused": "Paused",
"pausedHint": "Paused after repeated failures for {count} users until {at}. It is tried again automatically then.",
"modelNotServed": "The endpoint no longer offers the model {model}. Choose a model it lists in the AI settings."
}
}
},
"de": {
"admin": {
"providerHealth": {
"statusPaused": "Pausiert",
"pausedHint": "Nach wiederholten Fehlern für {count} Nutzer pausiert bis {at}. Danach wird er automatisch wieder versucht.",
"modelNotServed": "Der Endpunkt bietet das Modell {model} nicht mehr an. Wähle in den KI-Einstellungen ein Modell, das er anbietet."
}
}
},
"es": {
"admin": {
"providerHealth": {
"statusPaused": "En pausa",
"pausedHint": "En pausa tras fallos repetidos para {count} usuarios hasta {at}. Después se vuelve a intentar automáticamente.",
"modelNotServed": "El endpoint ya no ofrece el modelo {model}. Elige en los ajustes de IA un modelo que ofrezca."
}
}
},
"fr": {
"admin": {
"providerHealth": {
"statusPaused": "En pause",
"pausedHint": "En pause après des échecs répétés pour {count} utilisateurs jusqu’au {at}. Il sera ensuite réessayé automatiquement.",
"modelNotServed": "Le point d’accès ne propose plus le modèle {model}. Choisissez dans les réglages IA un modèle qu’il propose."
}
}
},
"it": {
"admin": {
"providerHealth": {
"statusPaused": "In pausa",
"pausedHint": "In pausa dopo errori ripetuti per {count} utenti fino al {at}. Poi viene ritentato automaticamente.",
"modelNotServed": "L’endpoint non offre più il modello {model}. Scegli nelle impostazioni IA un modello che offre."
}
}
},
"ko": {
"admin": {
"providerHealth": {
"statusPaused": "일시 중지됨",
"pausedHint": "반복된 오류로 사용자 {count}명에 대해 {at}까지 일시 중지되었습니다. 그 후 자동으로 다시 시도합니다.",
"modelNotServed": "엔드포인트가 더 이상 {model} 모델을 제공하지 않습니다. AI 설정에서 제공되는 모델을 선택하세요."
}
}
},
"pl": {
"admin": {
"providerHealth": {
"statusPaused": "Wstrzymany",
"pausedHint": "Wstrzymany po powtarzających się błędach dla {count} użytkowników do {at}. Potem zostanie automatycznie ponowiony.",
"modelNotServed": "Punkt końcowy nie oferuje już modelu {model}. Wybierz w ustawieniach AI model, który oferuje."
}
}
}
}
@@ -972,7 +972,15 @@ describe("derived response consumer guard", () => {
`derived apiSuccess branch ${index + 1} must preserve producer lineage`,
).toEqual(expectedLineage);
}
expect(variableCall(ROUTE_FILE, "derived")).toBe("computeDerivedMetric");
// v1.42 — the single read goes through the derived cache; the value is
// still the compute's, built inside the cache's builder.
const derivedProducer = variableCall(ROUTE_FILE, "derived");
expect(["computeDerivedMetric", "cachedSwr"]).toContain(derivedProducer);
if (derivedProducer === "cachedSwr") {
expect(readFileSync(ROUTE_FILE, "utf8")).toMatch(
/const derived = await cachedSwr\([\s\S]*?computeDerivedMetric\(/,
);
}
expect(variableCall(ROUTE_FILE, "assessment")).toBe(
"resolveDerivedAssessment",
);
@@ -17,13 +17,24 @@ vi.mock("@/lib/db", () => ({
prisma: {
providerHealth: {
groupBy: vi.fn(),
// v1.42 — the paused rows.
findMany: vi.fn(async () => []),
},
coachUsage: {
aggregate: vi.fn(),
},
// v1.42 — the operator key's model check reads the admin settings.
appSettings: { findUnique: vi.fn(async () => null) },
},
}));
// v1.42 — the model check probes the operator endpoint's listing.
const probeModelListing = vi.fn();
vi.mock("@/lib/ai/model-availability", () => ({
probeModelListing: (...a: unknown[]) => probeModelListing(...a),
}));
vi.mock("@/lib/crypto", () => ({ decrypt: (v: string) => v }));
vi.mock("@/lib/api-handler", async () => {
const actual =
await vi.importActual<typeof import("@/lib/api-handler")>(
@@ -115,6 +126,9 @@ describe("GET /api/admin/provider-health", () => {
lastOkAt: "2026-08-27T07:00:00.000Z",
lastFailureAt: "2026-08-27T06:00:00.000Z",
lastFailureStatus: 503,
pausedUsers: 0,
pausedUntil: null,
modelListing: null,
},
]);
});
@@ -257,3 +271,89 @@ describe("GET /api/admin/provider-health", () => {
expect(body.data.providers).toEqual([]);
});
});
describe("GET /api/admin/provider-health — pause and model check (v1.42)", () => {
const findMany = vi.mocked(prisma.providerHealth.findMany);
const settings = vi.mocked(prisma.appSettings.findUnique);
beforeEach(() => {
vi.clearAllMocks();
vi.mocked(requireAdmin).mockResolvedValue({
user: { id: "admin" },
} as never);
aggregate.mockResolvedValue({
_sum: { totalTokens: 0, operatorTokens: 0 },
} as never);
findMany.mockResolvedValue([] as never);
settings.mockResolvedValue(null as never);
});
it("counts the users a provider is paused for and when the last pause lifts", async () => {
groupBy.mockResolvedValue([
group("admin-openai", "hard_failed", 2, {
consecutiveFailures: 1766,
lastFailureAt: new Date("2026-10-08T04:31:00Z"),
lastStatus: 500,
}),
] as never);
findMany.mockResolvedValue([
{
providerType: "admin-openai",
nextRetryAt: new Date("2026-10-08T05:01:00Z"),
},
{
providerType: "admin-openai",
nextRetryAt: new Date("2026-10-09T04:31:00Z"),
},
] as never);
const body = await (await GET()).json();
expect(body.data.providers[0]).toMatchObject({
pausedUsers: 2,
pausedUntil: "2026-10-09T04:31:00.000Z",
});
// Only rows past the pause threshold, inside their window, are read.
expect(findMany.mock.calls[0][0]).toMatchObject({
where: {
lastResult: "hard_failed",
consecutiveFailures: { gte: 5 },
},
});
});
it("names a model the operator's endpoint no longer lists", async () => {
groupBy.mockResolvedValue([
group("admin-openai", "hard_failed", 1, {
consecutiveFailures: 9,
lastFailureAt: new Date("2026-10-08T04:31:00Z"),
lastStatus: 500,
}),
] as never);
settings.mockResolvedValue({
adminAiModel: "gpt-5.4-mini",
adminAiBaseUrl: "http://openai-oauth-proxy:10531/v1",
adminAiKeyEncrypted: "key",
} as never);
probeModelListing.mockResolvedValue("not_listed");
const body = await (await GET()).json();
expect(body.data.providers[0].modelListing).toEqual({
model: "gpt-5.4-mini",
listing: "not_listed",
});
expect(probeModelListing).toHaveBeenCalledWith(
expect.objectContaining({ operatorTrusted: true, model: "gpt-5.4-mini" }),
);
});
it("does not probe while the operator's key is healthy", async () => {
groupBy.mockResolvedValue([
group("admin-openai", "ok", 4, {
lastOkAt: new Date("2026-10-08T04:31:00Z"),
}),
] as never);
const body = await (await GET()).json();
expect(body.data.providers[0].modelListing).toBeNull();
expect(probeModelListing).not.toHaveBeenCalled();
});
});
@@ -4,6 +4,14 @@ import { prisma } from "@/lib/db";
import { apiSuccess } from "@/lib/api-response";
import { annotate } from "@/lib/logging/context";
import { OPERATOR_HELD_PROVIDER_TYPES } from "@/lib/ai/provider-egress";
import { PROVIDER_PAUSE_THRESHOLD } from "@/lib/ai/provider-health-ledger";
import {
probeModelListing,
type ModelListing,
} from "@/lib/ai/model-availability";
import { decrypt } from "@/lib/crypto";
import { isAnthropicBaseUrl } from "@/lib/ai/provider";
import { isCanonicalOpenAIEndpoint } from "@/lib/ai/openai-capabilities";
export const dynamic = "force-dynamic";
@@ -42,6 +50,20 @@ interface ProviderHealthSummary {
* success clears the status along with the rest of the failure state.
*/
lastFailureStatus: number | null;
/**
* v1.42 — users whose chain has this provider PAUSED right now (a run of
* hard failures past the pause threshold, inside its window). The chain
* walks a paused provider last and tries it again once the window lifts.
*/
pausedUsers: number;
/** When the latest of those pauses lifts. Null when none is paused. */
pausedUntil: string | null;
/**
* v1.42 — for the operator's own key while it is failing: whether its
* endpoint still lists the configured model. `not_listed` is the fix the
* card names; null when there is nothing to check.
*/
modelListing: { model: string; listing: ModelListing } | null;
}
/**
@@ -71,6 +93,19 @@ export const GET = apiHandler(async () => {
},
});
// v1.42 — paused providers: a hard-failure run at or past the threshold,
// inside its window. Read as rows (one per user and type, two columns) so
// the count and the latest lift time are exact, not folded from a group.
const now = new Date();
const pausedRows = await prisma.providerHealth.findMany({
where: {
lastResult: "hard_failed",
consecutiveFailures: { gte: PROVIDER_PAUSE_THRESHOLD },
nextRetryAt: { gt: now },
},
select: { providerType: true, nextRetryAt: true },
});
const byType = new Map<string, ProviderHealthSummary>();
for (const g of groups) {
const row = byType.get(g.providerType) ?? {
@@ -81,6 +116,9 @@ export const GET = apiHandler(async () => {
lastOkAt: null,
lastFailureAt: null,
lastFailureStatus: null,
pausedUsers: 0,
pausedUntil: null,
modelListing: null,
};
row.tracked += g._count._all;
if (g.lastResult !== "ok") {
@@ -103,6 +141,21 @@ export const GET = apiHandler(async () => {
byType.set(g.providerType, row);
}
for (const paused of pausedRows) {
const row = byType.get(paused.providerType);
if (!row || paused.nextRetryAt === null) continue;
row.pausedUsers += 1;
const until = paused.nextRetryAt.toISOString();
if (!row.pausedUntil || until > row.pausedUntil) row.pausedUntil = until;
}
// The operator's own key, while it is failing: does its endpoint still list
// the configured model? A free GET, cached; only on this admin read.
const adminRow = byType.get("admin-openai");
if (adminRow && adminRow.failing > 0) {
adminRow.modelListing = await adminModelListing();
}
const providers = [...byType.values()].sort((a, b) => {
const ca = CENTRAL_TYPES.indexOf(a.providerType);
const cb = CENTRAL_TYPES.indexOf(b.providerType);
@@ -136,3 +189,42 @@ export const GET = apiHandler(async () => {
},
});
});
/**
* The model check for the operator's key. Never fails the readout: a missing
* settings row, a key that does not decrypt, or a probe error is simply no
* answer.
*/
async function adminModelListing(): Promise<{
model: string;
listing: ModelListing;
} | null> {
try {
const settings = await prisma.appSettings.findUnique({
where: { id: "singleton" },
select: {
adminAiModel: true,
adminAiBaseUrl: true,
adminAiKeyEncrypted: true,
},
});
if (!settings?.adminAiKeyEncrypted) return null;
const baseUrl = settings.adminAiBaseUrl ?? "https://api.openai.com/v1";
// The same scope as the client's own check: an OpenAI-compatible
// endpoint the operator typed. `api.openai.com` names a missing model in
// its own error; Anthropic lists models on a different contract.
if (isCanonicalOpenAIEndpoint(baseUrl) || isAnthropicBaseUrl(baseUrl)) {
return null;
}
const model = settings.adminAiModel ?? "gpt-4o";
const listing = await probeModelListing({
baseUrl,
apiKey: decrypt(settings.adminAiKeyEncrypted),
model,
operatorTrusted: true,
});
return { model, listing };
} catch {
return null;
}
}
@@ -90,6 +90,7 @@ import { getSession } from "@/lib/auth/session";
import { prisma } from "@/lib/db";
import { computeDerivedMetric } from "@/lib/insights/derived";
import { DERIVED_MAX_WINDOW_DAYS } from "@/lib/insights/derived/types";
import { __resetAllCachesForTests } from "@/lib/cache/server-cache";
const SESSION_OK = {
session: { id: "sess-1", expiresAt: new Date(Date.now() + 3_600_000) },
@@ -119,6 +120,8 @@ function makeReqWith(params: Record<string, string>): NextRequest {
beforeEach(() => {
vi.clearAllMocks();
// v1.42 — the route reads through the derived cache now.
__resetAllCachesForTests();
vi.mocked(prisma.appSettings.findUnique).mockResolvedValue(null as never);
vi.mocked(prisma.user.findUnique).mockResolvedValue({
dateOfBirth: new Date("1986-01-01"),
@@ -270,3 +273,34 @@ describe("GET /api/insights/derived — windowDays", () => {
}
});
});
// v1.42 — the single-score read was the one derived surface with no cache:
// every open of a score's detail recomputed it over the rollup tier.
describe("GET /api/insights/derived — read-through cache (v1.42)", () => {
it("serves a repeat read from the cache, per metric, type and window", async () => {
vi.mocked(getSession).mockResolvedValue(SESSION_OK as never);
await callGet(makeReq("VITALS_BASELINE", "RESTING_HEART_RATE"));
await callGet(makeReq("VITALS_BASELINE", "RESTING_HEART_RATE"));
expect(computeDerivedMetric).toHaveBeenCalledTimes(1);
await callGet(makeReq("VITALS_BASELINE", "WEIGHT"));
await callGet(
makeReqWith({
metric: "VITALS_BASELINE",
type: "RESTING_HEART_RATE",
windowDays: "7",
}),
);
expect(computeDerivedMetric).toHaveBeenCalledTimes(3);
});
it("recomputes after the account's measurements change", async () => {
vi.mocked(getSession).mockResolvedValue(SESSION_OK as never);
const { invalidateUserMeasurements } =
await import("@/lib/cache/invalidate");
await callGet(makeReq("VITALS_BASELINE", "RESTING_HEART_RATE"));
invalidateUserMeasurements("user-1");
await callGet(makeReq("VITALS_BASELINE", "RESTING_HEART_RATE"));
expect(computeDerivedMetric).toHaveBeenCalledTimes(2);
});
});
+28 -9
View File
@@ -12,7 +12,8 @@
* The `data` payload is the flat `Derived<T>` union so iOS can decode +
* combine values across metrics: `{ metric, status, value?, coverage,
* confidence?, provenance, reason? }`. Numbers are pure compute over the
* rollup tier (Tier-1, no narrative) — no provider call, no cache table.
* rollup tier (Tier-1, no narrative) — no provider call, no cache table;
* since v1.42 the in-process derived cache, as the batch route uses.
*
* Wave 1 implements `VITALS_BASELINE` end-to-end; other registered ids
* return a `not_implemented` insufficient until their compute lands in a
@@ -36,6 +37,7 @@ import { DERIVED_MAX_WINDOW_DAYS } from "@/lib/insights/derived/types";
import { resolveDerivedAssessment } from "@/lib/insights/derived/derived-assessment-ai";
import { resolveServerLocale } from "@/lib/i18n/server-locale";
import { aiCapabilityToServe } from "@/lib/ai/capabilities/gate";
import { cachedSwr, caches, type ServerCache } from "@/lib/cache/server-cache";
export const dynamic = "force-dynamic";
@@ -122,14 +124,31 @@ export const GET = apiHandler(async (request: NextRequest) => {
if (!gate.enabled) return gate.response;
}
const profile = await loadBaselineProfile(prisma, user.id);
const derived = await computeDerivedMetric({
metric,
userId: user.id,
profile,
type: parsed.data.type ?? null,
windowDays: parsed.data.windowDays,
});
// v1.42 — read-through the per-user derived cache with
// stale-while-revalidate, like the batch grid beside it. This single-score
// read was the one derived surface that recomputed on every request: the
// score detail sheet asks for it on each open, and a composite walks the
// rollup tier once per input. Same bucket, so the invalidation the batch
// relies on covers it unchanged (measurement writes evict or mark the
// `${userId}|` prefix stale, mood writes mark it stale). The module gate
// above and the AI assessment below stay per request.
const type = parsed.data.type ?? null;
const windowDays = parsed.data.windowDays;
const derived = await cachedSwr(
caches.insightsDerived as ServerCache<
Awaited<ReturnType<typeof computeDerivedMetric>>
>,
`${user.id}|single|${metric}|${type ?? ""}|${windowDays ?? ""}`,
async () =>
computeDerivedMetric({
metric,
userId: user.id,
profile: await loadBaselineProfile(prisma, user.id),
type,
windowDays,
}),
annotate,
);
annotate({
action: { name: "insights.derived" },
@@ -48,6 +48,14 @@ interface ProviderHealthRow {
lastOkAt: string | null;
lastFailureAt: string | null;
lastFailureStatus: number | null;
/** v1.42 — users this provider is paused for, and when the last lifts. */
pausedUsers?: number;
pausedUntil?: string | null;
/** v1.42 — the operator key's model check, while it is failing. */
modelListing?: {
model: string;
listing: "listed" | "not_listed" | "unknown";
} | null;
}
export function ProviderHealthSection() {
@@ -151,11 +159,33 @@ export function ProviderHealthSection() {
})}`
: ""}
</span>
{p.pausedUsers && p.pausedUntil ? (
<span
className="text-muted-foreground text-xs"
data-slot="provider-paused"
>
{t("admin.providerHealth.pausedHint", {
count: p.pausedUsers,
at: formatDateTime(p.pausedUntil),
})}
</span>
) : null}
{p.modelListing?.listing === "not_listed" ? (
<span className="text-sm" data-slot="provider-model-missing">
{t("admin.providerHealth.modelNotServed", {
model: p.modelListing.model,
})}
</span>
) : null}
</div>
{p.failing === 0 ? (
<Badge variant="secondary">
{t("admin.providerHealth.statusHealthy")}
</Badge>
) : p.pausedUsers ? (
<Badge variant="destructive">
{t("admin.providerHealth.statusPaused")}
</Badge>
) : (
<Badge variant="destructive">
{t("admin.providerHealth.statusFailing", {
+72
View File
@@ -11,6 +11,8 @@ import {
requestDeviceCode,
pollDeviceCode,
refreshDeviceTokens,
CodexRefreshError,
codexReconnectRequired,
encryptCodexCreds,
decryptCodexCreds,
getCodexClientId,
@@ -172,6 +174,76 @@ describe("codex-oauth", () => {
});
});
describe("refreshDeviceTokens failures", () => {
function failWith(status: number, body: string) {
global.fetch = vi.fn(
async () =>
new Response(body, {
status,
headers: { "content-type": "application/json" },
}),
) as never;
}
async function refreshError(): Promise<CodexRefreshError> {
try {
await refreshDeviceTokens("refresh-dead");
} catch (err) {
return err as CodexRefreshError;
}
throw new Error("refresh unexpectedly succeeded");
}
it("reads a dead refresh token as an expired credential with an auth status", async () => {
// The production shape: 401 with `invalid_refresh_token`.
failWith(401, JSON.stringify({ error: "invalid_refresh_token" }));
const err = await refreshError();
expect(err).toBeInstanceOf(CodexRefreshError);
expect(err.httpStatus).toBe(401);
expect(err.credentialExpired).toBe(true);
expect(err.oauthError).toBe("invalid_refresh_token");
});
it("classifies invalid_grant on a 400 as a dead credential (401), not a request bug", async () => {
failWith(
400,
JSON.stringify({
error: { code: "invalid_grant", message: "Refresh token revoked" },
}),
);
const err = await refreshError();
expect(err.httpStatus).toBe(401);
expect(err.credentialExpired).toBe(true);
expect(err.oauthError).toBe("invalid_grant");
});
it("keeps the status of a refusal that is not about the credential", async () => {
failWith(503, "upstream unavailable");
const err = await refreshError();
expect(err.httpStatus).toBe(503);
expect(err.credentialExpired).toBe(false);
expect(err.oauthError).toBeNull();
});
it("never quotes the response body in the message", async () => {
failWith(
401,
JSON.stringify({
error: "invalid_refresh_token",
error_description: "token rt-secret-value was revoked",
}),
);
const err = await refreshError();
expect(err.message).not.toContain("rt-secret-value");
});
it("codexReconnectRequired carries the same auth contract", () => {
const err = codexReconnectRequired("no refresh token");
expect(err.httpStatus).toBe(401);
expect(err.credentialExpired).toBe(true);
});
});
describe("encryptCodexCreds / decryptCodexCreds", () => {
it("round-trips access token, refresh token, and account id", () => {
const expiresAt = new Date(Date.now() + 3600_000);
@@ -0,0 +1,160 @@
/**
* v1.42 — a provider that stopped offering the configured model says so.
*
* The operator's OAuth proxy reinstalled itself at a newer version, dropped
* the configured model name, and answered every call with a bare HTTP 500.
* The chain recorded a generic failure 1 766 times in a row. These cases pin
* the diagnosis: after a 5xx from a non-canonical endpoint the client asks
* the endpoint's free `/models` listing, and a listing without the model
* turns the error into one that names the cause (`modelNotServed`), which
* the chain's hop line carries as `model_not_served`.
*/
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { OpenAIClient } from "../openai-client";
import {
clearModelListingCache,
probeModelListing,
} from "../model-availability";
import { singleUserTurn } from "../types";
// safeFetch's requirePublicHost path runs through undici's own `fetch`;
// delegate it to the global stub so the interception still applies.
vi.mock("undici", async (importOriginal) => {
const actual = await importOriginal<typeof import("undici")>();
return {
...actual,
fetch: (input: unknown, init?: unknown) =>
(globalThis.fetch as unknown as (i: unknown, n?: unknown) => unknown)(
input,
init,
),
};
});
vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() }));
const emitSignal = vi.fn();
vi.mock("@/lib/logging/signal", () => ({
emitSignal: (...a: unknown[]) => emitSignal(...a),
}));
const PROXY = "https://oauth-proxy.example.com/v1";
function listing(ids: string[]) {
return {
ok: true,
status: 200,
json: () => Promise.resolve({ data: ids.map((id) => ({ id })) }),
};
}
function serverError() {
return {
ok: false,
status: 500,
text: () => Promise.resolve("Internal Server Error"),
};
}
/** Route by method: the completion is a POST, the listing a GET. */
function stubEndpoint(models: string[] | "broken") {
const fetchMock = vi.fn(async (_url: unknown, init?: { method?: string }) =>
init?.method === "GET"
? models === "broken"
? { ok: false, status: 404, json: () => Promise.resolve({}) }
: listing(models)
: serverError(),
);
vi.stubGlobal("fetch", fetchMock);
return fetchMock;
}
const turn = () => singleUserTurn({ system: "s", user: "u" });
beforeEach(() => {
clearModelListingCache();
emitSignal.mockClear();
});
afterEach(() => {
vi.unstubAllGlobals();
});
describe("probeModelListing", () => {
it("answers listed, not_listed, or unknown, and never guesses", async () => {
stubEndpoint(["gpt-6", "gpt-5.6-mini"]);
await expect(
probeModelListing({ baseUrl: PROXY, apiKey: "k", model: "gpt-6" }),
).resolves.toBe("listed");
await expect(
probeModelListing({ baseUrl: PROXY, apiKey: "k", model: "gpt-5.4-mini" }),
).resolves.toBe("not_listed");
clearModelListingCache();
stubEndpoint("broken");
await expect(
probeModelListing({ baseUrl: PROXY, apiKey: "k", model: "gpt-5.4-mini" }),
).resolves.toBe("unknown");
});
it("asks the endpoint once per model for a burst of failures", async () => {
const fetchMock = stubEndpoint(["gpt-6"]);
for (let i = 0; i < 3; i += 1) {
await probeModelListing({ baseUrl: PROXY, apiKey: "k", model: "x" });
}
expect(fetchMock).toHaveBeenCalledTimes(1);
});
});
describe("OpenAIClient — a 500 from an endpoint that dropped the model", () => {
it("names the missing model and marks the error modelNotServed", async () => {
stubEndpoint(["gpt-6", "gpt-5.6-mini"]);
const client = new OpenAIClient({
apiKey: "k",
model: "gpt-5.4-mini",
baseUrl: PROXY,
providerType: "admin-key",
operatorTrusted: true,
});
await expect(client.generateCompletion(turn())).rejects.toMatchObject({
httpStatus: 500,
modelNotServed: true,
message: expect.stringContaining('model "gpt-5.4-mini" is not offered'),
});
expect(emitSignal).toHaveBeenCalledWith(
expect.objectContaining({
action: "ai.provider.model_not_served",
level: "error",
}),
);
});
it("leaves an ordinary 500 alone when the model is still listed", async () => {
stubEndpoint(["gpt-5.4-mini"]);
const client = new OpenAIClient({
apiKey: "k",
model: "gpt-5.4-mini",
baseUrl: PROXY,
providerType: "admin-key",
operatorTrusted: true,
});
const error = await client.generateCompletion(turn()).catch((e) => e);
expect(error.httpStatus).toBe(500);
expect(error.modelNotServed).toBeUndefined();
});
it("never probes the canonical OpenAI endpoint", async () => {
const fetchMock = stubEndpoint(["gpt-6"]);
const client = new OpenAIClient({
apiKey: "k",
model: "gpt-5.4-mini",
baseUrl: "https://api.openai.com/v1",
providerType: "admin-key",
});
await client.generateCompletion(turn()).catch(() => undefined);
expect(
fetchMock.mock.calls.filter(
(c) => (c[1] as { method?: string })?.method === "GET",
),
).toHaveLength(0);
});
});
@@ -26,7 +26,11 @@ import {
AUTH_FAILURE_COOLDOWN_MS,
HARD_FAILURE_COOLDOWN_MS,
HARD_FAILURE_SKIP_THRESHOLD,
PROVIDER_PAUSE_BASE_MS,
PROVIDER_PAUSE_MAX_MS,
PROVIDER_PAUSE_THRESHOLD,
classifyFailure,
hardFailureCooldownMs,
createInMemoryProviderHealthLedger,
findCredentialExpiredProviders,
postgresProviderHealthLedger,
@@ -197,3 +201,68 @@ describe("createInMemoryProviderHealthLedger", () => {
expect((await ledger.getSkipHints("u1")).size).toBe(0);
});
});
// v1.42 — a provider that keeps failing is paused with a pause that grows,
// and resumes by itself. Before, the five-minute backoff lifted after every
// failure and put a gateway that answered 500 to everything back at the front
// of the chain: 1 766 failures in a row.
describe("provider pause (v1.42)", () => {
afterEach(() => {
vi.useRealTimers();
});
it("keeps the short backoff below the threshold and doubles from it, capped at a day", () => {
expect(hardFailureCooldownMs(1)).toBe(HARD_FAILURE_COOLDOWN_MS);
expect(hardFailureCooldownMs(PROVIDER_PAUSE_THRESHOLD - 1)).toBe(
HARD_FAILURE_COOLDOWN_MS,
);
expect(hardFailureCooldownMs(PROVIDER_PAUSE_THRESHOLD)).toBe(
PROVIDER_PAUSE_BASE_MS,
);
expect(hardFailureCooldownMs(PROVIDER_PAUSE_THRESHOLD + 1)).toBe(
2 * PROVIDER_PAUSE_BASE_MS,
);
expect(hardFailureCooldownMs(PROVIDER_PAUSE_THRESHOLD + 40)).toBe(
PROVIDER_PAUSE_MAX_MS,
);
// Conservative: the pause never starts before the skip threshold.
expect(PROVIDER_PAUSE_THRESHOLD).toBeGreaterThan(
HARD_FAILURE_SKIP_THRESHOLD,
);
});
it("pauses after N hard failures, keeps it paused past the old five minutes, and resumes after the window", async () => {
vi.useFakeTimers();
vi.setSystemTime(new Date("2026-10-08T04:30:00Z"));
const ledger = createInMemoryProviderHealthLedger();
for (let i = 0; i < PROVIDER_PAUSE_THRESHOLD; i += 1) {
await ledger.recordFailure("u1", "admin-openai", 500);
}
expect(ledger.inspect("u1").get("admin-openai")?.reason).toBe("paused");
// Past the old five-minute backoff it is still paused.
vi.advanceTimersByTime(HARD_FAILURE_COOLDOWN_MS + 1_000);
expect(ledger.inspect("u1").get("admin-openai")?.reason).toBe("paused");
// Past the pause it is tried again.
vi.advanceTimersByTime(PROVIDER_PAUSE_BASE_MS);
expect(ledger.inspect("u1").has("admin-openai")).toBe(false);
// One more failure doubles the pause; a success clears it.
await ledger.recordFailure("u1", "admin-openai", 500);
const retryAt = ledger.inspect("u1").get("admin-openai")?.retryAt;
expect(retryAt!.getTime() - Date.now()).toBe(2 * PROVIDER_PAUSE_BASE_MS);
await ledger.recordSuccess("u1", "admin-openai");
expect(ledger.inspect("u1").has("admin-openai")).toBe(false);
});
it("arms the escalating pause in the Postgres upsert itself", async () => {
await postgresProviderHealthLedger.recordFailure("u1", "admin-openai", 500);
const sql = (
vi.mocked(prisma.$executeRaw).mock.calls[0][0] as unknown as string[]
).join("?");
expect(sql).toContain("power(");
expect(sql).toContain("LEAST(");
expect(sql).toMatch(/consecutive_failures \+ 1 >=/);
});
});
@@ -26,6 +26,12 @@ vi.mock("../coach/budget", async () => {
};
});
// v1.42 — the per-hop failure lines.
const emitSignal = vi.hoisted(() => vi.fn());
vi.mock("@/lib/logging/signal", () => ({
emitSignal: (...a: unknown[]) => emitSignal(...a),
}));
vi.mock("../provider-health-ledger", async () => {
const actual = await vi.importActual<
typeof import("../provider-health-ledger")
@@ -52,8 +58,11 @@ import {
} from "../provider-runner";
import {
AUTH_FAILURE_COOLDOWN_MS,
PROVIDER_PAUSE_BASE_MS,
PROVIDER_PAUSE_THRESHOLD,
createInMemoryProviderHealthLedger,
} from "../provider-health-ledger";
import { CodexRefreshError } from "../codex-oauth";
const VALID_RESPONSE = JSON.stringify({
summary: "ok",
@@ -113,6 +122,7 @@ function err(status: number, msg = "boom"): Error & { httpStatus: number } {
beforeEach(() => {
clearLastWorkingProviderCache();
emitSignal.mockClear();
budgetState.spent = 0;
budgetState.operatorSpent = 0;
});
@@ -946,3 +956,145 @@ describe("runRawCompletionWithFallback — operator-cost cap at hop time", () =>
expect(outcome.workingProvider.providerType).toBe("openai");
});
});
describe("v1.42 — failed hops get their own line", () => {
function linkSignals() {
return emitSignal.mock.calls
.map(
(c) =>
c[0] as {
action: string;
level: string;
meta: Record<string, unknown>;
},
)
.filter((x) => x.action === "ai.chain.link_failed");
}
it("logs a hop the next provider covered at warn, with a stable cause", async () => {
const admin = new ScriptedProvider({
type: "admin-key",
script: [{ ok: false, error: err(500) }],
});
const codex = new ScriptedProvider({
type: "codex",
script: [{ ok: true }],
});
await runRawCompletionWithFallback({
userId: "u-signal-warn",
surface: "job",
providers: [
{ providerType: "admin-openai", instance: admin },
{ providerType: "codex", instance: codex },
],
params: singleUserTurn({ system: "s", user: "u" }),
});
expect(linkSignals()).toEqual([
expect.objectContaining({
level: "warn",
meta: expect.objectContaining({
provider: "admin-openai",
status: 500,
cause: "upstream_5xx",
recovered: true,
}),
}),
]);
});
it("logs every hop at error when nothing answered", async () => {
const a = new ScriptedProvider({
type: "codex",
script: [
{ ok: false, error: new CodexRefreshError(401, "invalid_grant") },
],
});
const b = new ScriptedProvider({
type: "admin-key",
script: [
{
ok: false,
error: Object.assign(err(500), { modelNotServed: true }),
},
],
});
await expect(
runRawCompletionWithFallback({
userId: "u-signal-error",
surface: "job",
providers: [
{ providerType: "codex", instance: a },
{ providerType: "admin-openai", instance: b },
],
params: singleUserTurn({ system: "s", user: "u" }),
}),
).rejects.toMatchObject({ primaryCredentialExpired: true });
const signals = linkSignals();
expect(signals.map((x) => x.level)).toEqual(["error", "error"]);
expect(signals.map((x) => x.meta.cause)).toEqual([
"credential_expired",
"model_not_served",
]);
});
it("records a dead Codex refresh token as an auth failure, not a network one", async () => {
const ledger = createInMemoryProviderHealthLedger();
const codex = new ScriptedProvider({
type: "codex",
script: [
{ ok: false, error: new CodexRefreshError(400, "invalid_grant") },
],
});
await expect(
runRawCompletionWithFallback({
userId: "u-dead-refresh",
surface: "job",
providers: [{ providerType: "codex", instance: codex }],
params: singleUserTurn({ system: "s", user: "u" }),
ledger,
}),
).rejects.toBeInstanceOf(AllProvidersFailedError);
expect(ledger.inspect("u-dead-refresh").get("codex")?.reason).toBe(
"credential_expired",
);
});
it("walks a paused provider last, so the healthy one answers without the failing hop", async () => {
vi.useFakeTimers();
const ledger = createInMemoryProviderHealthLedger();
for (let i = 0; i < PROVIDER_PAUSE_THRESHOLD; i += 1) {
await ledger.recordFailure("u-paused", "admin-openai", 500);
}
const admin = new ScriptedProvider({
type: "admin-key",
script: [{ ok: true }],
});
const codex = new ScriptedProvider({
type: "codex",
script: [{ ok: true }],
});
const run = () =>
runRawCompletionWithFallback({
userId: "u-paused",
surface: "job",
providers: [
{ providerType: "admin-openai", instance: admin },
{ providerType: "codex", instance: codex },
],
params: singleUserTurn({ system: "s", user: "u" }),
ledger,
});
// Well past the old five-minute backoff: still paused, never asked.
vi.advanceTimersByTime(20 * 60 * 1000);
clearLastWorkingProviderCache();
expect((await run()).workingProvider.providerType).toBe("codex");
expect(admin.callCount).toBe(0);
// Once the pause lifts it leads the chain again and answers.
vi.advanceTimersByTime(PROVIDER_PAUSE_BASE_MS);
clearLastWorkingProviderCache();
expect((await run()).workingProvider.providerType).toBe("admin-openai");
expect(admin.callCount).toBe(1);
});
});
+97 -3
View File
@@ -288,14 +288,108 @@ export async function refreshDeviceTokens(
});
if (!res.ok) {
const body = await res.text().catch(() => "");
throw new Error(
`Codex device-token refresh failed (${res.status}): ${body}`,
);
throw codexRefreshError(res.status, body);
}
const tokens = (await res.json()) as RawTokenResponse;
return buildCreds(tokens, refreshToken);
}
/**
* OAuth error codes on the refresh grant that mean the stored refresh token
* is dead for good: revoked, expired, already rotated, or issued to another
* client. Retrying cannot help; only signing in again can.
*/
const DEAD_REFRESH_TOKEN_CODES: ReadonlySet<string> = new Set([
"invalid_grant",
"invalid_refresh_token",
"refresh_token_expired",
"refresh_token_reused",
"refresh_token_invalidated",
"invalid_token",
]);
/**
* A failed refresh of the Codex OAuth token.
*
* The refresh used to throw a bare `Error` with the status only in its
* message. The chain runner reads `httpStatus` to classify a failure, so a
* dead refresh token read as a network error: the health ledger recorded
* `hard_failed` instead of `auth_failed`, nothing told the person to
* reconnect, and every night tried the same dead token again.
*
* `httpStatus` is 401 whenever the grant itself is dead (whatever status the
* token endpoint chose to send it with, 400 for `invalid_grant` included), so
* every classifier downstream (`classifyFailure`, `AllProvidersFailedError`,
* the briefing failure marker) reads it as an expired credential. Any other
* refusal keeps the endpoint's status. The body is not quoted: only the
* OAuth error code, which is the endpoint's vocabulary, rides on the error.
*/
export class CodexRefreshError extends Error {
readonly httpStatus: number;
readonly upstream = "codex";
/** True when the person has to sign in to ChatGPT again. */
readonly credentialExpired: boolean;
/** The OAuth `error` code from the response body, when it had one. */
readonly oauthError: string | null;
constructor(status: number, oauthError: string | null) {
const credentialExpired =
status === 401 ||
status === 403 ||
(oauthError !== null && DEAD_REFRESH_TOKEN_CODES.has(oauthError));
super(
credentialExpired
? `Codex sign-in expired (${status}${oauthError ? ` ${oauthError}` : ""}); reconnect ChatGPT`
: `Codex device-token refresh failed (${status}${oauthError ? ` ${oauthError}` : ""})`,
);
this.name = "CodexRefreshError";
this.credentialExpired = credentialExpired;
this.oauthError = oauthError;
this.httpStatus = credentialExpired && status !== 403 ? 401 : status;
}
}
/**
* The stored Codex credential cannot be used at all (missing refresh token,
* or storage that no longer decrypts). Same contract as a dead refresh token:
* the person has to reconnect.
*/
export function codexReconnectRequired(reason: string): CodexRefreshError {
const err = new CodexRefreshError(401, null);
err.message = `Codex sign-in unusable (${reason}); reconnect ChatGPT`;
return err;
}
/** Read the OAuth error code from a token-endpoint error body. */
function readOAuthErrorCode(body: string): string | null {
try {
const parsed = JSON.parse(body) as {
error?: unknown;
code?: unknown;
};
const error = parsed.error;
if (typeof error === "string") return normaliseCode(error);
if (error && typeof error === "object") {
const code = (error as { code?: unknown; type?: unknown }).code;
if (typeof code === "string") return normaliseCode(code);
}
if (typeof parsed.code === "string") return normaliseCode(parsed.code);
} catch {
// Not JSON: no code to read. The status still classifies the failure.
}
return null;
}
/** Codes are the endpoint's own vocabulary; keep them short and plain. */
function normaliseCode(code: string): string | null {
const trimmed = code.trim().toLowerCase();
return /^[a-z0-9_.-]{1,64}$/.test(trimmed) ? trimmed : null;
}
function codexRefreshError(status: number, body: string): CodexRefreshError {
return new CodexRefreshError(status, readOAuthErrorCode(body));
}
// ─── Encrypted-storage codec ────────────────────────────────────────
//
// We re-purpose `codexAccessTokenEncrypted` to store the entire creds
+91
View File
@@ -0,0 +1,91 @@
import { safeFetch } from "@/lib/safe-fetch";
import { aiEgressPolicyFor } from "./local-host-allowlist";
/**
* Does an OpenAI-compatible endpoint still offer the configured model?
*
* The operator's provider pointed at an OAuth proxy that installs
* `openai-oauth@latest` on every restart. One restart later the proxy listed
* only newer model names, the configured one was gone, and every call
* answered HTTP 500 with a body that named nothing. From the outside that
* reads as "the provider is down", and the fix (pick a model the endpoint
* offers) is invisible.
*
* `GET {baseUrl}/models` answers the question for free: it is the standard
* listing every OpenAI-compatible server implements, it bills nothing, and
* it needs the same bearer as a completion. The answer is cached for a few
* minutes per endpoint and model, so a burst of failures probes once.
*
* Tri-state on purpose. `unknown` (no listing route, a non-JSON answer, the
* probe itself failing) never claims the model is missing: only an endpoint
* that answered with a list that does not contain the name earns
* `not_listed`.
*/
export type ModelListing = "listed" | "not_listed" | "unknown";
const PROBE_TIMEOUT_MS = 5_000;
const CACHE_TTL_MS = 10 * 60 * 1000;
const cache = new Map<string, { at: number; value: ModelListing }>();
/** Test-only: forget every cached answer. */
export function clearModelListingCache(): void {
cache.clear();
}
export async function probeModelListing(input: {
baseUrl: string;
apiKey: string;
model: string;
/** See `OpenAIClient`'s `operatorTrusted`. */
operatorTrusted?: boolean;
now?: number;
}): Promise<ModelListing> {
const now = input.now ?? Date.now();
const key = `${input.baseUrl}\u0000${input.model}`;
const hit = cache.get(key);
if (hit && now - hit.at < CACHE_TTL_MS) return hit.value;
const value = await probeOnce(input);
cache.set(key, { at: now, value });
return value;
}
async function probeOnce(input: {
baseUrl: string;
apiKey: string;
model: string;
operatorTrusted?: boolean;
}): Promise<ModelListing> {
const url = `${input.baseUrl.replace(/\/$/, "")}/models`;
const headers: Record<string, string> = { Accept: "application/json" };
const key = input.apiKey.trim();
if (key) headers.Authorization = `Bearer ${key}`;
try {
const res = await safeFetch(
url,
{ method: "GET", headers },
{
timeoutMs: PROBE_TIMEOUT_MS,
...aiEgressPolicyFor(url, { operatorTrusted: input.operatorTrusted }),
},
);
if (!res.ok) return "unknown";
const json = (await res.json().catch(() => null)) as {
data?: unknown;
} | null;
if (!json || !Array.isArray(json.data)) return "unknown";
const ids = json.data
.map((entry) =>
entry && typeof entry === "object"
? (entry as { id?: unknown }).id
: undefined,
)
.filter((id): id is string => typeof id === "string");
if (ids.length === 0) return "unknown";
return ids.includes(input.model) ? "listed" : "not_listed";
} catch {
// The probe is a diagnosis, never a gate: a failed probe says nothing.
return "unknown";
}
}
+38 -1
View File
@@ -1,6 +1,7 @@
import { safeFetch } from "@/lib/safe-fetch";
import { callTimeoutMs } from "./effective-timeout";
import { annotate } from "@/lib/logging/context";
import { emitSignal } from "@/lib/logging/signal";
import type { AIProvider, CompletionParams, CompletionResult } from "./types";
import {
isCanonicalOpenAIEndpoint,
@@ -28,6 +29,7 @@ import {
rememberJsonModeDialect,
} from "./json-dialect";
import { aiEgressPolicyFor } from "./local-host-allowlist";
import { probeModelListing } from "./model-availability";
import type { ReasoningEffort } from "./reasoning-effort";
import {
buildOpenAIMessages,
@@ -113,6 +115,33 @@ export class OpenAIClient implements AIProvider {
this.type = config.providerType ?? "admin-key";
}
/**
* v1.42 — after a server error from an operator- or person-configured
* endpoint (never the canonical `api.openai.com`, which names a missing
* model in its own 404), check whether it still lists the configured
* model. A 4xx already carries its reason in the body; a 5xx from a proxy
* often carries none. See `model-availability.ts`.
*/
private async modelNoLongerServed(status: number): Promise<boolean> {
if (status < 500) return false;
if (this.type === "codex") return false;
if (isCanonicalOpenAIEndpoint(this.config.baseUrl)) return false;
const listing = await probeModelListing({
baseUrl: this.config.baseUrl,
apiKey: this.config.apiKey,
model: this.config.model,
operatorTrusted: this.config.operatorTrusted,
});
if (listing === "not_listed") {
emitSignal({
action: "ai.provider.model_not_served",
level: "error",
meta: { provider: this.type, model: this.config.model, status },
});
}
return listing === "not_listed";
}
/**
* True only for the user-configured gateway. Read it as "the base URL came
* from a person, not from this repository" — every behaviour that must not
@@ -306,8 +335,15 @@ export class OpenAIClient implements AIProvider {
});
return this.generateCompletion(params);
}
// v1.42 — a proxy that stopped offering the configured model answers
// with a bare 500 that names nothing. Ask its model listing (free, no
// completion) and say so plainly when the name is gone, so the chain,
// the health card and the log name the actual fix.
const modelNotServed = await this.modelNoLongerServed(res.status);
const err = new Error(
`${this.isGateway ? "OpenAI-compatible gateway" : "OpenAI"} request failed (${res.status})`,
modelNotServed
? `${this.isGateway ? "OpenAI-compatible gateway" : "OpenAI"} request failed (${res.status}): model "${this.config.model}" is not offered by this endpoint`
: `${this.isGateway ? "OpenAI-compatible gateway" : "OpenAI"} request failed (${res.status})`,
);
// Keep the body in a structured field rather than the message so
// Error.message stays short and the excerpt lands in dedicated log
@@ -317,6 +353,7 @@ export class OpenAIClient implements AIProvider {
upstream: this.isGateway ? "openai-compatible" : "openai",
model: this.config.model,
bodyExcerpt,
...(modelNotServed ? { modelNotServed: true } : {}),
});
throw err;
}
+83 -8
View File
@@ -52,13 +52,59 @@ export const HARD_FAILURE_COOLDOWN_MS = 5 * 60 * 1000; // 5min
* never productive. */
export const HARD_FAILURE_SKIP_THRESHOLD = 3;
/**
* v1.42 — a provider that keeps failing is PAUSED, with a pause that grows.
*
* The backoff above lifts five minutes after each failure, and a lifted
* provider goes back to the front of the chain. An operator gateway that
* answered HTTP 500 to every call (it no longer offered the configured model)
* was therefore tried first again every five minutes: 1 766 failures in a row,
* and a 500 in front of 61 chains that the next provider then answered.
*
* From `PROVIDER_PAUSE_THRESHOLD` consecutive hard failures on, each failure
* re-arms the skip window at `PROVIDER_PAUSE_BASE_MS` doubled per further
* failure, capped at `PROVIDER_PAUSE_MAX_MS`. Paused is the same skip as
* backoff (the provider moves to the tail of the chain, behind everything
* healthy, and is still walked as a last resort, so a chain never loses its
* only provider). When the window lifts, the next call tries it again: one
* success clears the count and resumes it in place, one more failure doubles
* the pause. Nothing needs to be reset by hand. The admin provider-health card
* shows a paused provider and until when.
*
* Conservative on purpose: five failures in a row is far past a brown-out,
* the first pause is half an hour, and the ceiling is a day, so a provider
* that recovers is tried again within a day even without a success elsewhere.
*/
export const PROVIDER_PAUSE_THRESHOLD = 5;
export const PROVIDER_PAUSE_BASE_MS = 30 * 60 * 1000;
export const PROVIDER_PAUSE_MAX_MS = 24 * 60 * 60 * 1000;
/**
* The skip window a hard failure arms, given the consecutive-failure count
* INCLUDING that failure. Below the pause threshold it is the short backoff;
* from the threshold on it doubles per failure up to the ceiling.
*/
export function hardFailureCooldownMs(consecutiveFailures: number): number {
if (consecutiveFailures < PROVIDER_PAUSE_THRESHOLD) {
return HARD_FAILURE_COOLDOWN_MS;
}
const doublings = Math.min(
consecutiveFailures - PROVIDER_PAUSE_THRESHOLD,
16,
);
return Math.min(
PROVIDER_PAUSE_BASE_MS * 2 ** doublings,
PROVIDER_PAUSE_MAX_MS,
);
}
export type ProviderHealthResult = "ok" | "hard_failed" | "auth_failed";
/** A provider the runner should skip (negative cache) on this call,
* with the reason so the caller can surface it. */
export interface ProviderSkipHint {
providerType: ProviderChainType;
reason: "credential_expired" | "backoff";
reason: "credential_expired" | "backoff" | "paused";
/** When the negative cache lifts (ledger `nextRetryAt`). */
retryAt: Date | null;
}
@@ -132,7 +178,10 @@ export const postgresProviderHealthLedger: ProviderHealthLedger = {
} else if (row.consecutiveFailures >= HARD_FAILURE_SKIP_THRESHOLD) {
hints.set(provider, {
providerType: provider,
reason: "backoff",
reason:
row.consecutiveFailures >= PROVIDER_PAUSE_THRESHOLD
? "paused"
: "backoff",
retryAt: row.nextRetryAt,
});
}
@@ -172,11 +221,18 @@ export const postgresProviderHealthLedger: ProviderHealthLedger = {
const { result, cooldownMs } = classifyFailure(httpStatus);
const interval = `${cooldownMs} milliseconds`;
const status = httpStatus ?? null;
// v1.42 — the escalating pause for a hard failure, computed in SQL from
// the count the same statement writes, so two workers recording at once
// still arm the pause the combined count deserves. Mirrors
// `hardFailureCooldownMs`; an auth failure keeps its fixed cooldown.
const pauseFrom = PROVIDER_PAUSE_THRESHOLD;
const pauseBase = `${PROVIDER_PAUSE_BASE_MS} milliseconds`;
const pauseMax = `${PROVIDER_PAUSE_MAX_MS} milliseconds`;
try {
// Atomic upsert. `consecutive_failures` accumulates across workers;
// an auth failure forces the cooldown regardless of count, a hard
// failure extends it. The fixed cooldown anchors on NOW() so a
// fresh failure always re-arms the skip window.
// failure extends it. The cooldown anchors on NOW() so a fresh failure
// always re-arms the skip window.
await prisma.$executeRaw`
INSERT INTO provider_health
(id, user_id, provider_type, last_result, last_status,
@@ -190,7 +246,18 @@ export const postgresProviderHealthLedger: ProviderHealthLedger = {
last_status = ${status},
consecutive_failures = provider_health.consecutive_failures + 1,
last_failure_at = NOW(),
next_retry_at = NOW() + ${interval}::interval,
next_retry_at = NOW() + CASE
WHEN ${result} = 'hard_failed'
AND provider_health.consecutive_failures + 1 >= ${pauseFrom}::int
THEN LEAST(
${pauseMax}::interval,
${pauseBase}::interval * power(
2,
LEAST(provider_health.consecutive_failures + 1 - ${pauseFrom}::int, 16)
)
)
ELSE ${interval}::interval
END,
updated_at = NOW()
`;
} catch {
@@ -265,7 +332,10 @@ export function createInMemoryProviderHealthLedger(): ProviderHealthLedger & {
) {
out.set(provider, {
providerType: provider,
reason: "backoff",
reason:
row.consecutiveFailures >= PROVIDER_PAUSE_THRESHOLD
? "paused"
: "backoff",
retryAt: new Date(row.nextRetryAt),
});
}
@@ -289,11 +359,16 @@ export function createInMemoryProviderHealthLedger(): ProviderHealthLedger & {
const prev = rowsFor(userId).get(providerType);
const priorCount =
prev && prev.result !== "ok" ? prev.consecutiveFailures : 0;
const count = priorCount + 1;
rowsFor(userId).set(providerType, {
result,
status: httpStatus,
consecutiveFailures: priorCount + 1,
nextRetryAt: Date.now() + cooldownMs,
consecutiveFailures: count,
nextRetryAt:
Date.now() +
(result === "hard_failed"
? hardFailureCooldownMs(count)
: cooldownMs),
});
},
inspect(userId) {
+71
View File
@@ -6,6 +6,7 @@ import {
type ProviderSkipHint,
} from "./provider-health-ledger";
import { annotate } from "@/lib/logging/context";
import { emitSignal } from "@/lib/logging/signal";
import {
isOperatorFundedProvider,
readDailySpend,
@@ -395,6 +396,67 @@ export function classifyErrorBody(body: unknown): string | null {
return `unclassified:${body.length}chars`;
}
/**
* A short, stable cause for one failed hop, for the `ai.chain.link_failed`
* line. Our own vocabulary only: the provider's words never leave.
*/
function hopCause(
error: unknown,
status: number | null,
bodyClass: string | null,
): string {
const err = error as {
credentialExpired?: unknown;
modelNotServed?: unknown;
} | null;
if (err?.modelNotServed === true) return "model_not_served";
if (err?.credentialExpired === true) return "credential_expired";
if (bodyClass?.startsWith("classified:")) {
return bodyClass.slice("classified:".length);
}
if (status === null || status <= 0) return "network";
if (status === 401 || status === 403) return "auth_rejected";
if (status === 429) return "rate_limited";
if (status >= 500) return "upstream_5xx";
return "upstream_4xx";
}
interface HopSignal {
provider: ProviderChainType;
attempt: number;
status: number | null;
cause: string;
}
/**
* v1.42 — one `ai.chain.link_failed` line per failed hop, written once the
* chain has settled: `warn` when a later provider answered (the fallback
* covered it), `error` when nothing did. The hop details used to live only as
* `ai_chain_hop_*` meta keys on the caller's line at level `info`, where a
* query for failures could not see them and the last caller of a batch
* overwrote the others.
*/
function signalFailedHops(
hops: readonly HopSignal[],
recovered: boolean,
surface: BudgetSurface,
): void {
for (const hop of hops) {
emitSignal({
action: "ai.chain.link_failed",
level: recovered ? "warn" : "error",
meta: {
provider: hop.provider,
attempt: hop.attempt,
status: hop.status,
cause: hop.cause,
recovered,
surface,
},
});
}
}
/**
* Shared chain walker for the two RAW runners (streaming + buffered). Owns the
* one set of chain / fallback / health-ledger semantics — empty-chain guard,
@@ -431,6 +493,7 @@ async function runRawChain(
const ordered = await resolveChainOrder(userId, providers, ledger);
const hops: FallbackHop[] = [];
const failedHops: HopSignal[] = [];
for (let i = 0; i < ordered.length; i += 1) {
const candidate = ordered[i];
@@ -510,6 +573,7 @@ async function runRawChain(
: {}),
},
});
signalFailedHops(failedHops, true, surface);
return {
result,
workingProvider: candidate,
@@ -520,6 +584,12 @@ async function runRawChain(
throw error;
}
const summary = summariseError(error);
failedHops.push({
provider: candidate.providerType,
attempt: i + 1,
status: summary.status,
cause: hopCause(error, summary.status, summary.bodyExcerpt),
});
void ledger.recordFailure(userId, candidate.providerType, summary.status);
const hop: FallbackHop = {
providerType: candidate.providerType,
@@ -545,6 +615,7 @@ async function runRawChain(
ai_chain_fallback_count: hops.length,
},
});
signalFailedHops(failedHops, false, surface);
throw new AllProvidersFailedError(hops);
}
+5 -6
View File
@@ -9,6 +9,7 @@ import { AnthropicClient } from "./anthropic-client";
import { LocalOpenAICompatibleClient } from "./local-client";
import {
refreshDeviceTokens,
codexReconnectRequired,
encryptCodexCreds,
decryptCodexCreds,
encryptAdminCodexCreds,
@@ -328,14 +329,14 @@ async function resolveCodexProvider(
!freshUser?.codexAccessTokenEncrypted ||
!freshUser.codexRefreshTokenEncrypted
) {
throw new Error("No refresh token available");
throw codexReconnectRequired("no refresh token");
}
const decoded = decryptCodexCreds({
accessEncrypted: freshUser.codexAccessTokenEncrypted,
refreshEncrypted: freshUser.codexRefreshTokenEncrypted,
});
if (!decoded) {
throw new Error("Codex token storage corrupt; user must re-link");
throw codexReconnectRequired("token storage corrupt");
}
const fresh = await refreshDeviceTokens(decoded.refreshToken);
const enc = encryptCodexCreds(fresh);
@@ -444,7 +445,7 @@ async function resolveAdminCodexProvider(): Promise<AIProvider | null> {
!fresh.adminCodexRefreshTokenEncrypted ||
!fresh.adminCodexAccountIdEncrypted
) {
throw new Error("No central-codex refresh token available");
throw codexReconnectRequired("no central refresh token");
}
const decoded = decryptAdminCodexCreds({
accessEncrypted: fresh.adminCodexAccessTokenEncrypted,
@@ -453,9 +454,7 @@ async function resolveAdminCodexProvider(): Promise<AIProvider | null> {
expiresAt: fresh.adminCodexTokenExpiresAt,
});
if (!decoded) {
throw new Error(
"Central-codex token storage corrupt; re-link required",
);
throw codexReconnectRequired("central token storage corrupt");
}
const refreshed = await refreshDeviceTokens(decoded.refreshToken);
const enc = encryptAdminCodexCreds(refreshed);
@@ -19,6 +19,8 @@ vi.mock("@/lib/safe-fetch", async (importOriginal) => {
});
import { GOOGLE_HEALTH_DATA_TYPES, fetchDataPoints } from "../client";
import { WideEventBuilder } from "@/lib/logging/event-builder";
import { eventStorage } from "@/lib/logging/context";
import { GoogleHealthApiError } from "../response-classifier";
function jsonResponse(status: number, body: unknown): Response {
@@ -177,3 +179,82 @@ describe("fetchDataPoints — daily-summary filter style", () => {
expect(capturedFilter(0)).toBeNull();
});
});
// v1.42 (#1023) — the first-request 400 that triggers the snake_case retry is
// expected, and the 200 right after it is the answer. It was logged as an
// error beside that 200; a self-hoster read it as a failing sync.
describe("fetchDataPoints — the fallback's 400 is a note, not an error", () => {
async function callsFor(run: () => Promise<unknown>) {
const evt = new WideEventBuilder("background");
await eventStorage.run(evt, () => run().catch(() => undefined));
return evt.toJSON().external_calls ?? [];
}
it("records the rejected camelCase probe without an error", async () => {
safeFetchMock
.mockResolvedValueOnce(
jsonResponse(400, {
error: { status: "INVALID_ARGUMENT", message: "Invalid filter" },
}),
)
.mockResolvedValueOnce(jsonResponse(200, { dataPoints: [] }));
const calls = await callsFor(() =>
fetchDataPoints(
GOOGLE_HEALTH_DATA_TYPES.restingHeartRate,
"t",
"fetchRhr",
{
start,
},
),
);
expect(calls).toHaveLength(2);
expect(calls[0]).toMatchObject({
status: 400,
note: "date_filter_rejected_retrying_snake_case",
});
expect(calls[0].error).toBeUndefined();
expect(calls[1].error).toBeUndefined();
});
it("still records the error when the snake_case retry is refused too", async () => {
safeFetchMock.mockResolvedValue(
jsonResponse(400, {
error: { status: "INVALID_ARGUMENT", message: "Invalid filter" },
}),
);
const calls = await callsFor(() =>
fetchDataPoints(
GOOGLE_HEALTH_DATA_TYPES.restingHeartRate,
"t",
"fetchRhr",
{
start,
},
),
);
expect(calls[1].error).toMatch(/400/);
});
it("still records a 400 on a sample read, which has no fallback", async () => {
safeFetchMock.mockResolvedValue(
jsonResponse(400, {
error: { status: "INVALID_ARGUMENT", message: "Invalid filter" },
}),
);
const calls = await callsFor(() =>
fetchDataPoints(
GOOGLE_HEALTH_DATA_TYPES.heartRate,
"t",
"fetchHeartRate",
{
start,
},
),
);
expect(calls).toHaveLength(1);
expect(calls[0].error).toBeDefined();
expect(calls[0].note).toBeUndefined();
});
});
@@ -0,0 +1,88 @@
/**
* v1.42 (#1023) — intraday heart rate re-reads two hours, not a day.
*
* Every hourly sync re-read the last 24 hours of every Google Health type,
* because daily summaries and sleep are re-scored after the fact. A
* per-minute heart-rate stream does not change once written, and re-reading
* a whole day of it each hour was 38 to 40 pages per sync on a slow host.
* Heart rate now uses the short intraday overlap; daily summaries keep the
* full day.
*/
import { beforeEach, describe, expect, it, vi } from "vitest";
const { safeFetchMock } = vi.hoisted(() => ({ safeFetchMock: vi.fn() }));
vi.mock("@/lib/safe-fetch", async (importOriginal) => {
const actual = await importOriginal<typeof import("@/lib/safe-fetch")>();
return { ...actual, safeFetch: safeFetchMock };
});
vi.mock("../sync-core", async (importOriginal) => {
const actual = await importOriginal<typeof import("../sync-core")>();
return {
...actual,
getValidToken: vi.fn(async () => ({ accessToken: "token" })),
handleCollectionFetchError: vi.fn(async () => 0),
upsertGoogleHealthMeasurements: vi.fn(async () => ({
imported: 0,
touched: [],
inserted: [],
})),
};
});
vi.mock("@/lib/db", () => ({
prisma: { user: { findUnique: vi.fn(), update: vi.fn() } },
}));
vi.mock("@/lib/tz/resolver", () => ({
resolveUserTimezone: vi.fn(async () => "UTC"),
}));
import {
GOOGLE_HEALTH_DEFAULT_OVERLAP_MS,
GOOGLE_HEALTH_INTRADAY_OVERLAP_MS,
incrementalStart,
} from "../sync-core";
import { syncUserMetrics } from "../sync-metrics";
const LAST_SYNC = new Date("2026-10-08T03:00:00.000Z");
/** The `filter` the request for one data-type path carried. */
function filterFor(path: string): string | null {
const call = safeFetchMock.mock.calls.find((c) =>
new URL(c[0] as string).pathname.endsWith(`/dataTypes/${path}/dataPoints`),
);
return call ? new URL(call[0] as string).searchParams.get("filter") : null;
}
beforeEach(() => {
safeFetchMock.mockReset().mockImplementation(async () => ({
status: 200,
ok: true,
json: async () => ({ dataPoints: [] }),
}));
});
describe("Google Health incremental overlap", () => {
it("keeps a day for summaries and narrows intraday samples to two hours", () => {
expect(GOOGLE_HEALTH_DEFAULT_OVERLAP_MS).toBe(24 * 60 * 60 * 1000);
expect(GOOGLE_HEALTH_INTRADAY_OVERLAP_MS).toBe(2 * 60 * 60 * 1000);
});
it("reads heart rate from the intraday start and resting heart rate from the day start", async () => {
const start = incrementalStart(LAST_SYNC);
const intradayStart = incrementalStart(LAST_SYNC, {
overlapMs: GOOGLE_HEALTH_INTRADAY_OVERLAP_MS,
});
await syncUserMetrics("u1", { start, intradayStart });
expect(filterFor("heart-rate")).toBe(
'heart_rate.sample_time.physical_time >= "2026-10-08T01:00:00.000Z"',
);
expect(filterFor("daily-resting-heart-rate")).toBe(
'dailyRestingHeartRate.date >= "2026-10-07"',
);
});
});
@@ -135,6 +135,7 @@ vi.mock("@/lib/insights/comprehensive-generate", () => ({
vi.mock("../sync-core", () => ({
GOOGLE_HEALTH_INTEGRATION_KEY: "google-health",
incrementalStart: vi.fn(() => undefined),
GOOGLE_HEALTH_INTRADAY_OVERLAP_MS: 2 * 60 * 60 * 1000,
markSynced: vi.fn(async () => {}),
runWithGoogleHealthSyncCycle: runCycle,
}));
+21 -6
View File
@@ -829,17 +829,32 @@ export async function forEachDataPointPage(
verdict.classification === "success"
? undefined
: extractGoogleApiErrorDetail(json);
// v1.42 (#1023) — the first filtered daily-summary read answering 400
// is the expected trigger for the snake_case retry below, not a
// failure: some accounts reject the documented camelCase prefix on
// some types, and the very next request answers 200. It was logged as
// an error next to the 200 that followed it. It is recorded as a note
// now; any other 400 still carries its error.
const expectedFallback =
canFallBack &&
dateStyle === "camel" &&
requestCount === 1 &&
res.status === 400;
getEvent()?.addExternalCall({
service: "google-health",
method: `${verb}(page=${pageCount})`,
duration_ms: Math.round(performance.now() - pageStart),
status: res.status,
error:
verdict.classification === "success"
? undefined
: apiErrorDetail
? `${verdict.reason} ${apiErrorDetail}`
: verdict.reason,
...(expectedFallback
? { note: "date_filter_rejected_retrying_snake_case" }
: {
error:
verdict.classification === "success"
? undefined
: apiErrorDetail
? `${verdict.reason} ${apiErrorDetail}`
: verdict.reason,
}),
});
if (verdict.classification !== "success") {
throw new GoogleHealthApiError({
+16
View File
@@ -59,6 +59,16 @@ const TOKEN_REFRESH_BUFFER_MS = 5 * 60 * 1000;
*/
export const GOOGLE_HEALTH_DEFAULT_OVERLAP_MS = 24 * 60 * 60 * 1000;
/**
* v1.42 (#1023) — the overlap for intraday samples (heart rate, one point a
* minute). The 24 h overlap exists for daily summaries and sleep, which
* Google re-scores after the fact; a minute sample is final when it is
* written. Re-reading a whole day of it every hour meant 38 to 40 pages per
* sync on a per-minute stream, which a slow host felt. Two hours still
* covers a watch that syncs to the phone late.
*/
export const GOOGLE_HEALTH_INTRADAY_OVERLAP_MS = 2 * 60 * 60 * 1000;
export interface GoogleHealthTokenInfo {
accessToken: string;
connection: { id: string; googleUserId: string };
@@ -354,6 +364,12 @@ export interface GoogleHealthResourceSyncOptions {
fullSync?: boolean;
/** The incremental lower bound, snapshotted once by the orchestrator. */
start?: Date;
/**
* v1.42 — the incremental lower bound for intraday samples, from the same
* snapshot with `GOOGLE_HEALTH_INTRADAY_OVERLAP_MS`. Undefined on a full
* run, like `start`; a resource that does not set it falls back to `start`.
*/
intradayStart?: Date;
/**
* When true, `upsertGoogleHealthMeasurements` writes the rows but SKIPS the
* inline per-(type,day) DAY-rollup recompute + status-insight invalidate,
+10 -1
View File
@@ -65,6 +65,11 @@ interface MetricResource {
verb: string;
/** Page ceiling for the walk; the client default when omitted. */
maxPages?: number;
/**
* v1.42 — a minute-level sample stream, final when written, so it
* re-reads only the short intraday overlap instead of a whole day.
*/
intraday?: boolean;
}
/** The launch metric resources (Measurement-producing). Height handled separately. */
@@ -104,6 +109,7 @@ const METRIC_RESOURCES: MetricResource[] = [
map: mapHeartRate,
verb: "fetchHeartRate",
maxPages: GOOGLE_HEALTH_DENSE_MAX_PAGES,
intraday: true,
},
{
dataType: GOOGLE_HEALTH_DATA_TYPES.bloodGlucose,
@@ -166,7 +172,10 @@ export async function syncUserMetrics(
resource.dataType,
tokenInfo.accessToken,
resource.verb,
{ start, maxPages: resource.maxPages },
{
start: resource.intraday ? (opts.intradayStart ?? start) : start,
maxPages: resource.maxPages,
},
async (points) => {
const readings: GoogleHealthMeasurementUpsert[] = [];
for (const point of points) {
+6
View File
@@ -19,6 +19,7 @@ import { syncUserWorkout } from "./sync-workout";
import {
GOOGLE_HEALTH_INTEGRATION_KEY,
incrementalStart,
GOOGLE_HEALTH_INTRADAY_OVERLAP_MS,
markSynced,
runWithGoogleHealthSyncCycle,
type GoogleHealthResourceSyncOptions,
@@ -173,9 +174,14 @@ export async function syncUserGoogleHealth(
const start = incrementalStart(connection.lastSyncedAt, {
fullSync: opts.fullSync,
});
const intradayStart = incrementalStart(connection.lastSyncedAt, {
fullSync: opts.fullSync,
overlapMs: GOOGLE_HEALTH_INTRADAY_OVERLAP_MS,
});
const resourceOpts: GoogleHealthResourceSyncOptions = {
fullSync: opts.fullSync,
start,
intradayStart,
deferRollup: opts.fullSync === true,
};
const resources = [
@@ -0,0 +1,177 @@
/**
* v1.42 — one bounded repair after the outbound safety screen withholds a
* generated briefing.
*
* A screened generation used to fail outright, and the nightly pass then
* queued its 45-minute provider retry, which asked the same model the same
* question and was screened again (eight times a night on one account). Now
* the generator asks once more with the broken contract named, and persists
* the repair only when it passes the same parse, grounding and screen as a
* first pass. These tests pin the repair, its correction text, and that a
* repair which is screened again still persists nothing.
*/
import { describe, it, expect, vi, beforeEach } from "vitest";
const findUnique = vi.fn();
const userUpdate = vi.fn();
const resolveProviderChain = vi.fn();
const resolveProvider = vi.fn();
const runRawCompletionWithFallback = vi.fn();
const extractFeatures = vi.fn();
vi.mock("@/lib/db", () => ({
prisma: {
// The briefing path now reserves against the day's token ledger before
// egress and reconciles after (`reserveBudget` / `reconcileSpend`), both
// over raw SQL. A zero prior total keeps every generation under the cap,
// so these suites keep testing what they were written to test.
$queryRaw: vi.fn(async () => [{ total_tokens: 0, operator_tokens: 0 }]),
$executeRaw: vi.fn(async () => 0),
user: {
findUnique: (...a: unknown[]) => findUnique(...a),
update: (...a: unknown[]) => userUpdate(...a),
// v1.32.22 (M6) — the cache commit is now a privacy-mode-guarded
// updateMany; record it on the same spy but return a matched count so the
// guard proceeds (mode unchanged in these tests).
updateMany: (...a: unknown[]) => {
userUpdate(...a);
return Promise.resolve({ count: 1 });
},
},
auditLog: { deleteMany: vi.fn() },
// v1.41 — the operator's reasoning controls; an untouched instance.
appSettings: { findUnique: vi.fn(async () => null) },
},
}));
// The `briefing` capability is available in these fixtures; the refusal
// path has its own test in comprehensive-generate-force.test.ts.
// The wire re-check (`aiEgressRefusal`) passes in these fixtures; its refusal
// arm has its own test in comprehensive-budget-refusal.test.ts.
vi.mock("@/lib/ai/capabilities/egress", () => ({
aiEgressRefusal: vi.fn(async () => null),
}));
vi.mock("@/lib/ai/capabilities/gate", () => ({
aiCapabilityForRecord: vi.fn(async () => ({
available: true,
reason: null,
onDeviceAllowed: true,
})),
aiCapabilityToServe: async () => ({
available: true,
reason: null,
onDeviceAllowed: true,
}),
}));
vi.mock("@/lib/ai/provider", () => ({
resolveProviderChain: (...a: unknown[]) => resolveProviderChain(...a),
resolveProvider: (...a: unknown[]) => resolveProvider(...a),
}));
vi.mock("@/lib/ai/provider-runner", () => ({
AllProvidersFailedError: class extends Error {},
runRawCompletionWithFallback: (...a: unknown[]) =>
runRawCompletionWithFallback(...a),
}));
vi.mock("@/lib/insights/features", () => ({
FeaturesPayloadTooLargeError: class extends Error {
sizeBytes = 0;
},
extractFeatures: (...a: unknown[]) => extractFeatures(...a),
BRIEFING_FEATURE_WINDOW_DAYS: 400,
}));
vi.mock("@/lib/insights/illness-cycle-briefing", () => ({
buildBriefingIllnessCycleContext: vi.fn().mockResolvedValue(null),
buildBriefingIllnessCyclePrompt: vi.fn().mockReturnValue(""),
}));
vi.mock("@/lib/insights/glp1-plateau", () => ({
detectGlp1Plateau: vi.fn(async () => null),
buildGlp1PlateauPrompt: vi.fn(() => ""),
}));
vi.mock("@/lib/ai/coach/about-me", () => ({
getSelfContextTextForUser: vi.fn(async () => null),
buildAboutMeInsightBlock: vi.fn(() => ""),
}));
vi.mock("@/lib/cache/invalidate", () => ({
invalidateUserInsights: vi.fn(),
}));
import { generateComprehensiveInsight } from "../comprehensive-generate";
const FEATURES = { weight: { count: 12, latest: 81.4, mean30: 82.1 } };
beforeEach(() => {
vi.clearAllMocks();
// A once-queued reply must not leak from one case into the next.
runRawCompletionWithFallback.mockReset();
resolveProviderChain.mockResolvedValue([
{ providerType: "openai", instance: {} },
]);
resolveProvider.mockResolvedValue({ type: "none" });
extractFeatures.mockResolvedValue(FEATURES);
userUpdate.mockResolvedValue({});
// No cached text / hash → always runs a full generation (no gate hit).
findUnique.mockResolvedValue({
insightsPrivacyMode: "aggregated",
insightsCachedAt: null,
insightsCachedText: null,
insightsExcludeMetrics: [],
insightsSnapshotHash: null,
insightsBriefingRerollDate: null,
});
});
const CAUSAL = JSON.stringify({
summary: "Your weight fell because you slept more.",
dailyBriefing: { paragraph: "ok" },
});
const CLEAN = JSON.stringify({
summary: "Your weight was lower on days when you slept longer.",
dailyBriefing: { paragraph: "ok" },
});
function reply(content: string) {
return {
result: { content, tokensUsed: 10, model: "m" },
workingProvider: { providerType: "openai" },
fallbackHops: [],
};
}
describe("comprehensive outbound-screen repair", () => {
it("persists a repair that passes the screen, after one corrective call", async () => {
runRawCompletionWithFallback
.mockResolvedValueOnce(reply(CAUSAL))
.mockResolvedValueOnce(reply(CLEAN));
const outcome = await generateComprehensiveInsight("u1", { locale: "en" });
expect(outcome).toEqual({ status: "generated", providerType: "openai" });
expect(runRawCompletionWithFallback).toHaveBeenCalledTimes(2);
const repairPrompt =
runRawCompletionWithFallback.mock.calls[1][0].params.messages[0].content;
expect(repairPrompt).toContain("withheld by the safety screen");
expect(repairPrompt).toContain("CAUSES");
// The repaired text is what lands in the cache, never the screened one.
const written = JSON.stringify(userUpdate.mock.calls.at(-1));
expect(written).toContain("on days when you slept longer");
expect(written).not.toContain("because you slept more");
});
it("stays withheld when the repair is screened again, and asks only once", async () => {
runRawCompletionWithFallback.mockResolvedValue(reply(CAUSAL));
const outcome = await generateComprehensiveInsight("u1", { locale: "en" });
expect(outcome).toEqual({ status: "failed", reason: "outbound-screened" });
expect(runRawCompletionWithFallback).toHaveBeenCalledTimes(2);
});
it("stays withheld when the repair call itself fails", async () => {
runRawCompletionWithFallback
.mockResolvedValueOnce(reply(CAUSAL))
.mockRejectedValueOnce(new Error("upstream down"));
const outcome = await generateComprehensiveInsight("u1", { locale: "en" });
expect(outcome).toEqual({ status: "failed", reason: "outbound-screened" });
});
});
+1 -1
View File
@@ -57,7 +57,7 @@ import { resolveJobReasoning } from "@/lib/ai/reasoning/controls";
* correction did not).
*/
export type BriefingCompletionStage =
"generate" | "json-retry" | "grounding-retry" | "reroll";
"generate" | "json-retry" | "grounding-retry" | "screen-retry" | "reroll";
/**
* Thrown when the reservation would push the user past the day's ceiling.
+96 -4
View File
@@ -41,6 +41,7 @@ import {
fenceBlock,
} from "@/lib/ai/coach/data-fence";
import { screenInsightPayloadProse } from "@/lib/ai/safety/insight-payload-screen";
import type { OutboundReason } from "@/lib/ai/safety/outbound-screen";
import { computeCitationCoverage } from "@/lib/ai/citation-coverage";
import { applyInsightsExcludeFilter } from "@/lib/insights/exclude-filter";
import { getCachedFeatures } from "@/lib/insights/feature-cache";
@@ -166,7 +167,17 @@ export type GenerateOutcome =
| "scope-changed"
| "profile-scope-changed";
}
| { status: "failed"; reason: string };
| {
status: "failed";
reason: string;
/**
* v1.42 — the chain's first provider refused the credential (a dead
* Codex sign-in, a revoked key). The nightly pass reports the account
* as `auth_failed` and does not queue a retry that cannot succeed until
* the person reconnects.
*/
credentialExpired?: boolean;
};
/** Higher, seedless temperature for the daily-briefing phrasing re-roll. */
const BRIEFING_REROLL_TEMPERATURE = 0.6;
@@ -1259,7 +1270,11 @@ export async function generateComprehensiveInsight(
locale,
httpStatus: typeof err.httpStatus === "number" ? err.httpStatus : null,
});
return { status: "failed", reason };
const credentialExpired =
e instanceof AllProvidersFailedError && e.primaryCredentialExpired;
return credentialExpired
? { status: "failed", reason, credentialExpired }
: { status: "failed", reason };
}
// Anthropic + local have no native JSON mode, so a ```json-fenced or
@@ -1492,14 +1507,72 @@ export async function generateComprehensiveInsight(
// persists NOTHING -- the previous cached payload stays, exactly as a
// provider outage or an unparseable response already behaves. The next run
// regenerates from scratch.
//
// v1.42 — one bounded repair before giving up, the way the grounding gate
// above gets one corrective retry. A withheld generation used to fail
// outright, and the nightly pass then queued its 45-minute provider retry,
// which asked the same model the same question and was screened again: on
// one account that was eight wasted generations a night over a single
// causal sentence. The repair names the contract that tripped and asks for
// the whole object again; the repaired payload has to pass the same JSON
// parse, the same number grounding and the same screen as a first pass, or
// nothing is persisted, exactly as before. A screened outcome is final for
// the caller: it is not a provider fault, so nothing retries it.
const screened = screenInsightPayloadProse(insights, locale);
if (screened) {
annotate({
action: { name: "insights.generate.outbound_blocked" },
meta: { locale, reason: screened },
});
void recordBriefingFailure({ userId, reason: "outbound-screened", locale });
return { status: "failed", reason: "outbound-screened" };
let repaired = false;
try {
const retry = await runBriefingCompletion({
userId,
chain,
systemPrompt,
userPrompt: `${userPrompt}\n\n${buildOutboundScreenCorrection(screened)}`,
temperature: AI_BUDGETS.comprehensive.temperature,
maxTokens: resolveInsightsMaxTokens(),
timeoutMs: effectiveTimeoutMs,
stage: "screen-retry",
});
const retryInsights = parseComprehensiveResult(retry.result.content);
const retryUngrounded =
retryInsights === null
? []
: findUngroundedBriefingNumbers(
readBriefingBlock(retryInsights),
features.signalsOfDay ?? null,
features,
groundingExtra,
);
if (
retryInsights !== null &&
retryUngrounded.length === 0 &&
screenInsightPayloadProse(retryInsights, locale) === null
) {
insights = retryInsights;
result = retry.result;
workingProviderType = retry.workingProvider.providerType;
briefingStrippedHard = false;
repaired = true;
}
} catch {
// A transport failure or the day's budget on the repair: the first
// pass stays withheld, nothing else changes.
}
annotate({
action: { name: "insights.generate.outbound_repair" },
meta: { locale, reason: screened, repaired },
});
if (!repaired) {
void recordBriefingFailure({
userId,
reason: "outbound-screened",
locale,
});
return { status: "failed", reason: "outbound-screened" };
}
}
// v1.9.0 — the caller abandoned this generation (its bounded timeout
@@ -1556,3 +1629,22 @@ export async function generateComprehensiveInsight(
return { status: "generated", providerType: workingProviderType };
}
/**
* The correction the one screen repair is sent with. Names the contract the
* first reply broke, never the sentence that broke it: the model gets the
* rule, not its own words back.
*/
export function buildOutboundScreenCorrection(reason: OutboundReason): string {
const rule =
reason === "causal_claim"
? 'It stated that one measurement or habit CAUSES another. Describe co-occurrence and timing only ("was higher on days when", "tends to move together"); never say that something causes, drives, leads to or results in something else.'
: reason === "dose_prescription"
? "It told the reader to take, change, stop or dose a medication. Never give a dose or a medication instruction; at most suggest discussing medication with their clinician."
: "It gave a risk score, a probability of disease or a diagnosis. Never estimate risk or name a diagnosis; describe the readings and their trend only.";
return `
Your previous reply was withheld by the safety screen. ${rule}
Re-write the whole JSON object with the same structure and the same numbers, changing only the wording that broke this rule. Return the full JSON object again.
`;
}
@@ -1508,3 +1508,131 @@ describe("runInsightPregenerate — warm budget honours the response timeout (v1
}
});
});
// v1.42 — the run used to annotate one wide event per candidate and the last
// write won, so a night where one account worked and the last one failed read
// as `all-failed`. The run now reports every candidate's outcome together, a
// withheld (screened) briefing and a dead sign-in queue no provider retry,
// and a time-budget cut-off reports the unreached accounts as deferred.
describe("runInsightPregenerate — one outcome per candidate (v1.42)", () => {
type Reported = {
queue: string;
runId: string;
candidates: Array<{ key: string; outcome: string; cause?: string }>;
};
function harness(users: Array<{ id: string; locale: string }>) {
const { prisma } = makePrisma(users);
const reportRun = vi.fn(async (input: Reported) => ({
total: input.candidates.length,
attempted: 0,
failed: 0,
allFailed: input.candidates.every((c) => c.outcome === "failed"),
partial: false,
}));
const enqueueRetry = vi.fn().mockResolvedValue(undefined);
const statusGenerators = Array.from({ length: 7 }, () =>
vi.fn().mockResolvedValue({ hasProvider: true, cached: true }),
);
const warmGenericMetrics = vi.fn().mockResolvedValue(0);
return {
prisma,
reportRun,
enqueueRetry,
statusGenerators,
warmGenericMetrics,
};
}
it("reports ok, auth_failed and screened candidates side by side, and retries none of the last two", async () => {
const h = harness([
{ id: "ok-user", locale: "de" },
{ id: "dead-codex", locale: "de" },
{ id: "screened", locale: "de" },
]);
const generate = vi
.fn()
.mockResolvedValueOnce({ status: "generated", providerType: "codex" })
.mockResolvedValueOnce({
status: "failed",
reason: "all-providers-failed",
credentialExpired: true,
})
.mockResolvedValueOnce({ status: "failed", reason: "outbound-screened" });
const result = await runInsightPregenerate(h.prisma as never, {
generate,
statusGenerators: h.statusGenerators,
warmGenericMetrics: h.warmGenericMetrics,
enqueueRetry: h.enqueueRetry,
reportRun: h.reportRun as never,
});
expect(h.reportRun).toHaveBeenCalledTimes(1);
const reported = h.reportRun.mock.calls[0][0];
expect(reported.queue).toBe(INSIGHT_PREGENERATE_QUEUE);
expect(reported.candidates.map((c) => c.outcome)).toEqual([
"ok",
"auth_failed",
"screened",
]);
expect(reported.candidates[1].cause).toBe("all-providers-failed");
expect(reported.candidates[2].cause).toBe("outbound-screened");
// Keys never carry the account id.
for (const c of reported.candidates) {
expect(c.key).toMatch(/^[0-9a-f]{8}$/);
expect(JSON.stringify(c)).not.toContain("user");
}
expect(h.enqueueRetry).not.toHaveBeenCalled();
expect(result.allFailed).toBe(false);
});
it("still queues the bounded retry for an ordinary provider failure", async () => {
const h = harness([{ id: "u1", locale: "de" }]);
const generate = vi
.fn()
.mockResolvedValue({ status: "failed", reason: "all-providers-failed" });
const result = await runInsightPregenerate(h.prisma as never, {
generate,
statusGenerators: h.statusGenerators,
warmGenericMetrics: h.warmGenericMetrics,
enqueueRetry: h.enqueueRetry,
reportRun: h.reportRun as never,
});
expect(h.enqueueRetry).toHaveBeenCalledTimes(1);
expect(h.reportRun.mock.calls[0][0].candidates).toEqual([
expect.objectContaining({
outcome: "failed",
cause: "all-providers-failed",
}),
]);
expect(result.allFailed).toBe(true);
});
it("reports the candidates a time-budget stop never reached as deferred", async () => {
const h = harness([
{ id: "a", locale: "de" },
{ id: "b", locale: "de" },
{ id: "c", locale: "de" },
]);
const generate = vi
.fn()
.mockResolvedValue({ status: "generated", providerType: "x" });
let calls = 0;
await runInsightPregenerate(h.prisma as never, {
generate,
statusGenerators: h.statusGenerators,
warmGenericMetrics: h.warmGenericMetrics,
enqueueRetry: h.enqueueRetry,
reportRun: h.reportRun as never,
shouldStop: () => ++calls > 1,
});
expect(
h.reportRun.mock.calls[0][0].candidates.map((c) => c.outcome),
).toEqual(["ok", "deferred", "deferred"]);
});
});
+7 -3
View File
@@ -16,6 +16,7 @@
import type { SupportedLocale } from "@/lib/insights/status-shared";
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
import { annotate } from "@/lib/logging/context";
import { logCaught } from "@/lib/logging/signal";
export const INSIGHT_PREGENERATE_QUEUE = "insight-pregenerate";
@@ -111,7 +112,8 @@ export async function enqueueForceWarm(payload: {
action: { name: "insights.pregenerate.force.enqueued" },
meta: { locale: payload.locale },
});
} catch {
} catch (err) {
logCaught("insights.pregenerate.enqueue_failed", err, { kind: "force" });
// Enqueue is best-effort; a failure here just means the caches stay
// cold until the next poll / nightly cron warms them.
}
@@ -169,7 +171,8 @@ export async function enqueueTodayBriefingWarm(payload: {
action: { name: "insights.pregenerate.today.enqueued" },
meta: { locale: payload.locale, immediate: payload.immediate },
});
} catch {
} catch (err) {
logCaught("insights.pregenerate.enqueue_failed", err, { kind: "today" });
// Best-effort, like every warm enqueue: the next read asks again.
}
}
@@ -240,7 +243,8 @@ export async function enqueuePregenerateFailureRetry(payload: {
delay_seconds: PREGENERATE_RETRY_DELAY_SECONDS,
},
});
} catch {
} catch (err) {
logCaught("insights.pregenerate.enqueue_failed", err, { kind: "retry" });
// Best-effort: a failed enqueue leaves the nightly tick as the
// catch-net, exactly the pre-retry behaviour.
}
+317 -216
View File
@@ -48,12 +48,20 @@
* Recurring pg-boss task — never runs inside an HTTP request and never
* shells out to `tsx` (CLAUDE.md DO-NOTs).
*/
import { createHash, randomUUID } from "node:crypto";
import pLimit from "p-limit";
import type { PrismaClient } from "@/generated/prisma/client";
import { checkRateLimit } from "@/lib/rate-limit";
import { getAssistantFlags } from "@/lib/feature-flags";
import { annotate } from "@/lib/logging/context";
import { withBackgroundEvent } from "@/lib/logging/background";
import { logCaught } from "@/lib/logging/signal";
import {
reportJobRun,
type JobRunCandidate,
type JobRunOutcome,
} from "@/lib/jobs/job-run-report";
import { aiCapabilityForJob } from "@/lib/ai/capabilities/gate";
import { userIdsWithModuleOff } from "@/lib/jobs/ai-job-candidates";
import type { SupportedLocale } from "@/lib/insights/status-shared";
@@ -279,6 +287,13 @@ export interface PregenerateRunResult {
budgetBlocked: number;
/** The job's time budget ran out before every candidate was reached. */
stoppedEarly: boolean;
/**
* v1.42 — every candidate the run attempted failed (see `reportJobRun`).
* Written into the job's facts as `all_failed`, which is how a completed
* night with nothing generated still counts toward the repeated-failure
* streak.
*/
allFailed: boolean;
/**
* Count of per-metric assessment caches written warm across the whole
* run. The comprehensive generator evicts the seven `*-status`
@@ -342,15 +357,11 @@ async function warmPerStatusCaches(
} catch (err) {
// A single card's generation failing must not abort the rest of
// the warm pass for this user, nor the cron's user loop — but it
// must not be invisible either: annotate so a failing card is
// diagnosable from one grep instead of a silent zero.
annotate({
action: { name: "insights.pregenerate.warm_failed" },
meta: {
stage: "status-card",
locale,
message: describeError(err),
},
// must not be invisible either: a warn line under one action, so a
// failing card is diagnosable from one grep instead of a silent zero.
logCaught("insights.pregenerate.warm_failed", err, {
stage: "status-card",
locale,
});
return 0;
}
@@ -391,13 +402,9 @@ async function warmStatusBatch(
} catch (err) {
// The batch must never throw the cron's user loop off course — but
// a swallowed batch failure left ~7 cards cold with zero signal.
annotate({
action: { name: "insights.pregenerate.warm_failed" },
meta: {
stage: "status-batch",
locale,
message: describeError(err),
},
logCaught("insights.pregenerate.warm_failed", err, {
stage: "status-batch",
locale,
});
}
}
@@ -440,9 +447,8 @@ async function warmGenericMetricCaches(
} catch (err) {
// Best-effort discovery — but a silently-failing grouped read used to
// zero out the whole generic warm with no trace.
annotate({
action: { name: "insights.pregenerate.warm_failed" },
meta: { stage: "metric-discovery", message: describeError(err) },
logCaught("insights.pregenerate.warm_failed", err, {
stage: "metric-discovery",
});
return 0;
}
@@ -478,14 +484,10 @@ async function warmGenericMetricCaches(
? 1
: 0;
} catch (err) {
annotate({
action: { name: "insights.pregenerate.warm_failed" },
meta: {
stage: "metric-card",
metric,
locale,
message: describeError(err),
},
logCaught("insights.pregenerate.warm_failed", err, {
stage: "metric-card",
metric,
locale,
});
return 0;
}
@@ -496,6 +498,15 @@ async function warmGenericMetricCaches(
return results.reduce((sum: number, n) => sum + n, 0);
}
/**
* A stable, non-identifying key for one candidate on the run's log line: a
* short hash of the account id, enough to follow one account across nights
* without naming it.
*/
function candidateKey(userId: string): string {
return createHash("sha256").update(userId).digest("hex").slice(0, 8);
}
interface PregenerateCandidate {
id: string;
locale: string | null;
@@ -583,12 +594,15 @@ export async function runInsightPregenerate(
* not reach are the stalest again tomorrow, so they lead the next pass.
*/
shouldStop?: () => boolean;
/** Injected for the test — defaults to the real run report. */
reportRun?: typeof reportJobRun;
} = {},
): Promise<PregenerateRunResult> {
const now = options.now ?? new Date();
const cap = options.cap ?? PREGENERATE_BATCH_CAP;
const generate = options.generate ?? generateComprehensiveInsight;
const enqueueRetry = options.enqueueRetry ?? enqueuePregenerateFailureRetry;
const reportRun = options.reportRun ?? reportJobRun;
// v1.18.7 (HIGH-1) — when the caller injects its own `statusGenerators`
// (the unit tests do), keep the per-card warm so those assertions hold;
// production injects nothing, so the seven warm calls collapse into ONE
@@ -615,6 +629,7 @@ export async function runInsightPregenerate(
failed: 0,
budgetBlocked: 0,
stoppedEarly: false,
allFailed: false,
assessmentsWarmed: 0,
metricAssessmentsWarmed: 0,
};
@@ -627,197 +642,35 @@ export async function runInsightPregenerate(
const candidates = await findPregenerateCandidates(prisma, now, cap);
result.total = candidates.length;
for (const candidate of candidates) {
// v1.42 — one outcome per candidate, reported together at the end of the
// run (`reportJobRun`). Each candidate also runs inside its own wide event,
// so what one account's generation annotates (the provider chain's hops,
// the failure cause) lands on that account's line instead of overwriting
// the run's: the run used to read `all-failed` whenever the LAST account
// failed, on nights where the others had worked.
const runId = randomUUID();
const runOutcomes: JobRunCandidate[] = [];
for (const [index, candidate] of candidates.entries()) {
if (options.shouldStop?.()) {
result.stoppedEarly = true;
// The rest are the stalest again tomorrow; they lead the next pass.
for (const rest of candidates.slice(index)) {
runOutcomes.push({
key: candidateKey(rest.id),
outcome: "deferred",
cause: "job_budget",
});
}
break;
}
// v1.18.11 P3 — one feature-cache scope per candidate, so the
// comprehensive briefing's bounded feature read is computed once and any
// sibling consumer in this candidate's warm reuses it. Scoped per user so
// no cross-user object can leak.
await withFeatureCacheScope(async () => {
// The two capabilities this pass serves, resolved before any snapshot
// or rate-limit write: the briefing for the comprehensive insight, the
// status notes for the per-metric warm. Each half runs only when its
// own capability is available; the chokepoints re-check at the wire.
const [briefing, statusText] = await Promise.all([
aiCapabilityForJob(candidate.id, "briefing"),
aiCapabilityForJob(candidate.id, "statusText"),
]);
if (!briefing.available && !statusText.available) {
result.skipped++;
annotate({
action: { name: "insights.pregenerate.skipped" },
meta: {
briefing: briefing.reason,
status_text: statusText.reason,
},
});
return;
}
// Budget gate — one COMPREHENSIVE pre-generation per user per 20 h.
// The route's on-demand path uses a different bucket
// (`insights:${userId}`), so this never starves a user's manual
// regenerate quota. The gate bounds the comprehensive cost only: a
// blocked user still gets the refill-only status warm below, because
// the 02:xx status crons already skipped every pregenerate candidate
// on the assumption that THIS pass warms their cards — exiting early
// here left those cards cold until the first on-visit generation.
const budget = briefing.available
? await checkRateLimit(
`insight-pregenerate:${candidate.id}`,
1,
PREGENERATE_BUDGET_WINDOW_MS,
)
: null;
// The stored locale, then the operator default: a NULL column used to
// resolve to English here, so a German-language instance warmed its
// briefings in English every night.
const locale = await resolveJobLocale(candidate.locale);
if (budget === null) {
result.skipped++;
annotate({
action: { name: "insights.pregenerate.skipped" },
meta: { briefing: briefing.reason },
});
} else if (!budget.allowed) {
result.budgetBlocked++;
} else {
// Force a fresh generation: the discovery window (20 h) is shorter
// than the generator's 24 h cache TTL, so without `force` a user
// whose cache is 20–24 h old would be discovered, consume the
// budget bucket, then short-circuit to `cached` with no actual
// pre-generation — defeating the "warm the cache before the user's
// morning visit" intent for the common case. The per-user budget
// bucket (above) bounds attempts, and the generator's content-hash
// gate (v1.16.8) turns a same-data force into a timestamp refresh
// (`unchanged`) with no provider call.
//
// Bounded budget + abort (v1.16.1): one stalled provider must not
// pin the whole batch behind a single candidate, and `withTimeout`
// alone cannot cancel the detached generation — without the abort a
// late resolve would still write a cache row + timestamp after the
// loop moved on (the same race the forced single-user warm closes).
const controller = new AbortController();
// v1.25.3 — size the outer abort from the candidate's resolved
// response-timeout so a raised setting extends the warm budget
// instead of being clipped by a fixed cap.
const warmBudgetMs = comprehensiveWarmBudgetMs(
candidate.aiResponseTimeoutSeconds,
);
// Capture the rejection so the failure annotation can carry the
// actual error text — `withTimeout` folds a rejection into a bare
// `errored: true` envelope, which made the cause undiagnosable.
let thrown: unknown = null;
const bounded = await withTimeout(
() =>
generate(candidate.id, {
locale,
force: true,
signal: controller.signal,
}).catch((err: unknown) => {
thrown = err;
throw err;
}),
warmBudgetMs,
null,
() => controller.abort(),
);
if (bounded.timedOut || bounded.errored || bounded.value === null) {
result.failed++;
// v1.21.5 — make the warm failure queryable. The nightly loop used to
// bump `failed` with no annotation, so a candidate whose comprehensive
// generation timed out (e.g. the briefing clipped at the old 45 s
// bound) left the cached block empty AND emitted nothing greppable.
// v1.28.30 — carry the stage + error text so ONE grep on
// `comprehensive_failed` names the failing catch and its cause.
annotate({
action: { name: "insights.pregenerate.comprehensive_failed" },
meta: {
locale,
stage: "nightly.bound",
cause: bounded.timedOut
? "timeout"
: bounded.errored
? "error"
: "null",
budget_ms: warmBudgetMs,
message: thrown === null ? null : describeError(thrown),
},
});
// v1.28.30 — bounded intra-day retry: one delayed forced warm for
// just this user, so a transient 04:30 provider hiccup heals
// before the morning visit instead of waiting for the next night.
await enqueueRetry({ userId: candidate.id, locale });
} else {
const outcome = bounded.value;
switch (outcome.status) {
case "generated":
result.generated++;
break;
case "cached":
result.cached++;
break;
case "unchanged":
result.unchanged++;
break;
case "skipped":
result.skipped++;
break;
case "failed":
result.failed++;
// v1.28.30 — THE previously-silent failure path. A generator
// that resolves with `{ status: "failed" }` (features error,
// oversize payload, invalid JSON after retry, abort) bumped
// the tally with no annotation at all — the nightly summary
// showed `failed: 1` and nothing else was greppable. Reuse
// the `comprehensive_failed` shape with the generator's own
// reason as the cause.
annotate({
action: { name: "insights.pregenerate.comprehensive_failed" },
meta: {
locale,
stage: "nightly.generator",
cause: `generator:${outcome.reason}`,
budget_ms: warmBudgetMs,
message: null,
},
});
await enqueueRetry({ userId: candidate.id, locale });
break;
}
}
}
// Warm the per-metric assessment caches — refill-only, single locale
// (v1.16.8). The comprehensive write no longer evicts the per-status
// rows, so there is nothing to force-refill: a card already generated
// today is a cheap cache read, a cold card runs its generator, and the
// generator's content-hash gate turns a same-data regeneration into a
// timestamp refresh with no provider call. The warm runs on every
// comprehensive outcome (budget-blocked, failed, timed-out,
// consent-skipped included) because the 02:xx status crons skip every
// pregenerate candidate on the assumption that THIS pass covers them.
//
// Nothing to warm when the status notes themselves are unavailable.
if (!statusText.available) {
annotate({
action: { name: "insights.pregenerate.status_skipped" },
meta: { reason: statusText.reason },
});
return;
}
result.assessmentsWarmed += await warmStatus(candidate.id, [locale]);
// v1.8.7.1 — warm the generic per-HealthKit-metric caches too, for
// the user's data-bearing metrics only (the helper filters via one
// grouped count, so an empty metric never reaches the provider).
result.metricAssessmentsWarmed += await warmGenericMetrics(candidate.id, [
locale,
]);
});
const key = candidateKey(candidate.id);
const outcome = await withBackgroundEvent(
"job.insight_pregenerate.candidate",
(evt) =>
withFeatureCacheScope(() => pregenerateCandidate(candidate, key, evt)),
);
runOutcomes.push(outcome);
}
// Wide-event tally so the nightly dashboard can track how many
@@ -837,8 +690,254 @@ export async function runInsightPregenerate(
metric_assessments_warmed: result.metricAssessmentsWarmed,
},
});
const verdict = await reportRun({
queue: INSIGHT_PREGENERATE_QUEUE,
runId,
candidates: runOutcomes,
});
result.allFailed = verdict.allFailed;
return result;
/**
* One candidate's whole pass: capabilities, the budget gate, the bounded
* comprehensive generation and the refill-only card warm. Returns how the
* candidate ended for the run report; never throws for a provider failure
* (those are outcomes), only for something unexpected, which aborts the
* run as it always did.
*/
async function pregenerateCandidate(
candidate: PregenerateCandidate,
key: string,
evt: { elevateLevel(level: "warn"): unknown },
): Promise<JobRunCandidate> {
annotate({ meta: { candidate: key, run_id: runId } });
const state: { outcome: JobRunOutcome; cause?: string } = {
outcome: "ok",
};
const settle = (outcome: JobRunOutcome, cause?: string) => {
state.outcome = outcome;
state.cause = cause;
};
// The two capabilities this pass serves, resolved before any snapshot
// or rate-limit write: the briefing for the comprehensive insight, the
// status notes for the per-metric warm. Each half runs only when its
// own capability is available; the chokepoints re-check at the wire.
const [briefing, statusText] = await Promise.all([
aiCapabilityForJob(candidate.id, "briefing"),
aiCapabilityForJob(candidate.id, "statusText"),
]);
if (!briefing.available && !statusText.available) {
result.skipped++;
annotate({
action: { name: "insights.pregenerate.skipped" },
meta: {
briefing: briefing.reason,
status_text: statusText.reason,
},
});
return {
key,
outcome: "skipped",
cause: briefing.reason ?? "unavailable",
};
}
// Budget gate — one COMPREHENSIVE pre-generation per user per 20 h.
// The route's on-demand path uses a different bucket
// (`insights:${userId}`), so this never starves a user's manual
// regenerate quota. The gate bounds the comprehensive cost only: a
// blocked user still gets the refill-only status warm below, because
// the 02:xx status crons already skipped every pregenerate candidate
// on the assumption that THIS pass warms their cards — exiting early
// here left those cards cold until the first on-visit generation.
const budget = briefing.available
? await checkRateLimit(
`insight-pregenerate:${candidate.id}`,
1,
PREGENERATE_BUDGET_WINDOW_MS,
)
: null;
// The stored locale, then the operator default: a NULL column used to
// resolve to English here, so a German-language instance warmed its
// briefings in English every night.
const locale = await resolveJobLocale(candidate.locale);
if (budget === null) {
result.skipped++;
settle("skipped", briefing.reason ?? "briefing_unavailable");
annotate({
action: { name: "insights.pregenerate.skipped" },
meta: { briefing: briefing.reason },
});
} else if (!budget.allowed) {
result.budgetBlocked++;
settle("skipped", "budget_blocked");
} else {
// Force a fresh generation: the discovery window (20 h) is shorter
// than the generator's 24 h cache TTL, so without `force` a user
// whose cache is 20–24 h old would be discovered, consume the
// budget bucket, then short-circuit to `cached` with no actual
// pre-generation — defeating the "warm the cache before the user's
// morning visit" intent for the common case. The per-user budget
// bucket (above) bounds attempts, and the generator's content-hash
// gate (v1.16.8) turns a same-data force into a timestamp refresh
// (`unchanged`) with no provider call.
//
// Bounded budget + abort (v1.16.1): one stalled provider must not
// pin the whole batch behind a single candidate, and `withTimeout`
// alone cannot cancel the detached generation — without the abort a
// late resolve would still write a cache row + timestamp after the
// loop moved on (the same race the forced single-user warm closes).
const controller = new AbortController();
// v1.25.3 — size the outer abort from the candidate's resolved
// response-timeout so a raised setting extends the warm budget
// instead of being clipped by a fixed cap.
const warmBudgetMs = comprehensiveWarmBudgetMs(
candidate.aiResponseTimeoutSeconds,
);
// Capture the rejection so the failure annotation can carry the
// actual error text — `withTimeout` folds a rejection into a bare
// `errored: true` envelope, which made the cause undiagnosable.
let thrown: unknown = null;
const bounded = await withTimeout(
() =>
generate(candidate.id, {
locale,
force: true,
signal: controller.signal,
}).catch((err: unknown) => {
thrown = err;
throw err;
}),
warmBudgetMs,
null,
() => controller.abort(),
);
if (bounded.timedOut || bounded.errored || bounded.value === null) {
result.failed++;
const boundCause = bounded.timedOut
? "timeout"
: bounded.errored
? "error"
: "null";
settle("failed", boundCause);
// v1.21.5 — make the warm failure queryable. The nightly loop used to
// bump `failed` with no annotation, so a candidate whose comprehensive
// generation timed out (e.g. the briefing clipped at the old 45 s
// bound) left the cached block empty AND emitted nothing greppable.
// v1.28.30 — carry the stage + error text so ONE grep on
// `comprehensive_failed` names the failing catch and its cause.
annotate({
action: { name: "insights.pregenerate.comprehensive_failed" },
meta: {
locale,
stage: "nightly.bound",
cause: boundCause,
budget_ms: warmBudgetMs,
message: thrown === null ? null : describeError(thrown),
},
});
// v1.28.30 — bounded intra-day retry: one delayed forced warm for
// just this user, so a transient 04:30 provider hiccup heals
// before the morning visit instead of waiting for the next night.
await enqueueRetry({ userId: candidate.id, locale });
} else {
const generated = bounded.value;
switch (generated.status) {
case "generated":
case "rerolled":
result.generated++;
break;
case "cached":
result.cached++;
break;
case "unchanged":
result.unchanged++;
break;
case "skipped":
result.skipped++;
settle("skipped", generated.reason);
break;
case "failed":
result.failed++;
// v1.28.30 — THE previously-silent failure path. A generator
// that resolves with `{ status: "failed" }` (features error,
// oversize payload, invalid JSON after retry, abort) bumped
// the tally with no annotation at all — the nightly summary
// showed `failed: 1` and nothing else was greppable. Reuse
// the `comprehensive_failed` shape with the generator's own
// reason as the cause.
annotate({
action: { name: "insights.pregenerate.comprehensive_failed" },
meta: {
locale,
stage: "nightly.generator",
cause: `generator:${generated.reason}`,
budget_ms: warmBudgetMs,
message: null,
},
});
if (generated.reason === "outbound-screened") {
// v1.42 — the model answered and the safety screen withheld
// it, after the generator's own repair. Asking the same model
// again in 45 minutes is not a provider retry; it is the same
// question, and it was screened eight times a night.
settle("screened", generated.reason);
} else if (generated.credentialExpired) {
// v1.42 — a dead sign-in (a Codex refresh token the issuer
// revoked) does not heal in 45 minutes either; the person has
// to reconnect, and the ledger already tells them so.
settle("auth_failed", generated.reason);
} else {
settle("failed", generated.reason);
await enqueueRetry({ userId: candidate.id, locale });
}
break;
}
}
}
// Warm the per-metric assessment caches — refill-only, single locale
// (v1.16.8). The comprehensive write no longer evicts the per-status
// rows, so there is nothing to force-refill: a card already generated
// today is a cheap cache read, a cold card runs its generator, and the
// generator's content-hash gate turns a same-data regeneration into a
// timestamp refresh with no provider call. The warm runs on every
// comprehensive outcome (budget-blocked, failed, timed-out,
// consent-skipped included) because the 02:xx status crons skip every
// pregenerate candidate on the assumption that THIS pass covers them.
//
// Nothing to warm when the status notes themselves are unavailable.
if (!statusText.available) {
annotate({
action: { name: "insights.pregenerate.status_skipped" },
meta: { reason: statusText.reason },
});
} else {
result.assessmentsWarmed += await warmStatus(candidate.id, [locale]);
// v1.8.7.1 — warm the generic per-HealthKit-metric caches too, for
// the user's data-bearing metrics only (the helper filters via one
// grouped count, so an empty metric never reaches the provider).
result.metricAssessmentsWarmed += await warmGenericMetrics(candidate.id, [
locale,
]);
}
const final = state.outcome;
if (final === "failed" || final === "auth_failed" || final === "screened") {
evt.elevateLevel("warn");
}
annotate({
meta: { outcome: final, ...(state.cause ? { cause: state.cause } : {}) },
});
return {
key,
outcome: final,
...(state.cause ? { cause: state.cause } : {}),
};
}
}
/**
@@ -1001,7 +1100,8 @@ export async function forceWarmUser(
aiResponseTimeoutSeconds: true,
},
});
} catch {
} catch (err) {
logCaught("insights.pregenerate.force.preflight_failed", err);
// Best-effort pre-flight — on a read failure fall through to the
// budgeted generation attempt below.
}
@@ -1119,7 +1219,8 @@ export async function forceWarmUser(
data: { insightsWarmFailedAt: null },
});
}
} catch {
} catch (err) {
logCaught("insights.pregenerate.force.marker_failed", err);
// Marker maintenance is best-effort; never fail the warm on it.
}
}
@@ -381,6 +381,9 @@ export async function handleInsightPregenerateJob(
failed: summary.failed,
budget_blocked: summary.budgetBlocked,
stopped_early: summary.stoppedEarly,
// v1.42 — read back by `readConsecutiveRunFailures`, so a night that
// completed with nothing generated counts toward the failure streak.
all_failed: summary.allFailed,
assessments_warmed: summary.assessmentsWarmed,
metric_assessments_warmed: summary.metricAssessmentsWarmed,
});
+12 -2
View File
@@ -311,6 +311,16 @@ const BAND_TRUNC_UNIT: Record<
*
* `tzKey` (the far-from-UTC day-grain guard) buckets on the user's LOCAL
* calendar via `AT TIME ZONE`; parameter-bound, never spliced.
*
* v1.42 — `measured_at` is a `timestamp` without time zone holding UTC wall
* time, so the local wall clock is `(measured_at AT TIME ZONE 'UTC') AT TIME
* ZONE tz`: the first step says what the value is, the second converts it.
* The single `measured_at AT TIME ZONE tz` this used read the UTC value AS IF
* it were local time and converted the wrong way, so a far-from-UTC reader's
* readings were shifted by twice their offset (sixteen hours in Los Angeles):
* an evening reading landed on the next day's bucket. Every other local-day
* reader
* (`day-aggregates.ts`, `day-mean.ts`) already used the two-step form.
*/
async function readBandLive(
userId: string,
@@ -337,7 +347,7 @@ async function readBandLive(
}>
>`
SELECT
date_trunc(${unit}, m."measured_at" AT TIME ZONE ${tzKey}) AS bucket_start,
date_trunc(${unit}, (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${tzKey}) AS bucket_start,
COUNT(*)::int AS count,
AVG(m."value")::double precision AS mean,
MIN(m."value")::double precision AS min_value,
@@ -419,7 +429,7 @@ async function readHourlyMeanBandLive(
}),
);
const bucket = tzKey
? Prisma.sql`date_trunc(${unit}, m."measured_at" AT TIME ZONE ${tzKey})`
? Prisma.sql`date_trunc(${unit}, (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${tzKey})`
: Prisma.sql`date_trunc(${unit}, m."measured_at")`;
try {
const rows = await prisma.$queryRaw<
+80
View File
@@ -0,0 +1,80 @@
/**
* v1.42 — the escalating provider pause, against the real upsert.
*
* The pause is computed inside the one `INSERT … ON CONFLICT DO UPDATE` that
* records a failure, from the count that same statement writes, so two
* workers failing at once still arm the pause their combined count deserves.
* Only Postgres can show the CASE arithmetic is right; the unit test can only
* show the statement carries it.
*/
import { beforeEach, describe, expect, it, vi } from "vitest";
import { getPrismaClient, truncateAllTables } from "./setup";
vi.mock("@/lib/db-compat", () => ({
ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
}));
import {
HARD_FAILURE_COOLDOWN_MS,
PROVIDER_PAUSE_BASE_MS,
PROVIDER_PAUSE_THRESHOLD,
postgresProviderHealthLedger as ledger,
} from "@/lib/ai/provider-health-ledger";
const prisma = getPrismaClient();
async function windowMs(userId: string): Promise<number> {
const row = await prisma.providerHealth.findFirstOrThrow({
where: { userId, providerType: "admin-openai" },
select: { nextRetryAt: true, lastFailureAt: true },
});
return row.nextRetryAt!.getTime() - row.lastFailureAt!.getTime();
}
describe("provider pause in the Postgres ledger", () => {
let userId: string;
beforeEach(async () => {
await truncateAllTables(prisma);
const user = await prisma.user.create({
data: {
email: "pause@example.test",
username: "pause",
passwordHash: "x",
},
});
userId = user.id;
});
it("backs off briefly, then pauses from the threshold, doubling, and a success resumes", async () => {
for (let i = 1; i < PROVIDER_PAUSE_THRESHOLD; i += 1) {
await ledger.recordFailure(userId, "admin-openai", 500);
}
expect(await windowMs(userId)).toBe(HARD_FAILURE_COOLDOWN_MS);
expect(
(await ledger.getSkipHints(userId)).get("admin-openai")?.reason,
).toBe("backoff");
await ledger.recordFailure(userId, "admin-openai", 500);
expect(await windowMs(userId)).toBe(PROVIDER_PAUSE_BASE_MS);
expect(
(await ledger.getSkipHints(userId)).get("admin-openai")?.reason,
).toBe("paused");
await ledger.recordFailure(userId, "admin-openai", 500);
expect(await windowMs(userId)).toBe(2 * PROVIDER_PAUSE_BASE_MS);
await ledger.recordSuccess(userId, "admin-openai");
expect((await ledger.getSkipHints(userId)).has("admin-openai")).toBe(false);
});
it("keeps the fixed cooldown for a dead credential, however long the run", async () => {
for (let i = 0; i < PROVIDER_PAUSE_THRESHOLD + 2; i += 1) {
await ledger.recordFailure(userId, "admin-openai", 401);
}
expect(
(await ledger.getSkipHints(userId)).get("admin-openai")?.reason,
).toBe("credential_expired");
});
});
@@ -0,0 +1,69 @@
/**
* v1.42 — the tiered context's live day band buckets on the reader's own
* calendar day, including on a daylight-saving changeover.
*
* For a reader far from UTC the 14–30-day band is read live, bucketed by
* `date_trunc('day', <local wall clock>)`. `measured_at` is a `timestamp`
* without time zone holding UTC, and the band read it with a single
* `AT TIME ZONE tz`, which interprets the value as local time and converts
* the wrong way: in Los Angeles every reading moved sixteen hours, and an
* evening reading was summarised as the next day. The Coach and the briefing
* then quoted a day's mean built from two different days.
*
* 2026-03-08 is the US changeover (02:00 PST becomes 03:00 PDT). Two weight
* readings on that local day, 08:00 PDT (15:00Z) and 23:30 PDT (06:30Z on
* the 9th), must form one bucket for the 8th.
*/
import { beforeAll, describe, expect, it, vi } from "vitest";
import { getPrismaClient, truncateAllTables } from "./setup";
vi.mock("@/lib/db-compat", () => ({
ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
}));
import { buildTieredSeries } from "@/lib/rollups/tiered-context";
const prisma = getPrismaClient();
const USER = "tiered-context-local-day";
const TZ = "America/Los_Angeles";
/** "Now" for the band arithmetic: the 8th sits inside the 14–30-day band. */
const NOW = new Date("2026-03-25T12:00:00Z").getTime();
beforeAll(async () => {
await truncateAllTables(prisma);
await prisma.user.create({
data: { id: USER, username: USER, timezone: TZ },
});
await prisma.measurement.createMany({
data: [
{ at: "2026-03-08T15:00:00Z", value: 70 },
{ at: "2026-03-09T06:30:00Z", value: 80 },
// A reading on the local 7th, late evening PST, for contrast.
{ at: "2026-03-08T07:30:00Z", value: 90 },
].map(({ at, value }) => ({
userId: USER,
type: "WEIGHT" as const,
value,
unit: "kg",
measuredAt: new Date(at),
source: "MANUAL" as const,
})),
});
});
describe("tiered context — live day band on the reader's calendar", () => {
it("puts an evening reading on its own local day across the DST changeover", async () => {
const series = await buildTieredSeries(USER, "WEIGHT", {
tz: TZ,
now: NOW,
skipEnsureFresh: true,
});
const byDay = Object.fromEntries(
series.dayBand.map((b) => [b.bucketStart.slice(0, 10), b]),
);
expect(Object.keys(byDay).sort()).toEqual(["2026-03-07", "2026-03-08"]);
expect(byDay["2026-03-08"]).toMatchObject({ count: 2, mean: 75 });
expect(byDay["2026-03-07"]).toMatchObject({ count: 1, mean: 90 });
});
});