From 73e98c6ef5dbfd0efdd40107acade51ccd094fc5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sat, 24 Oct 2026 09:23:20 +0200 Subject: [PATCH] fix(insights): give the briefing warm every call of its generation The out-of-band warm the Insights page relies on bounded one comprehensive generation at the account's response timeout plus 30 s. A generation is up to five model calls in a row: the generation, the JSON retry, the grounding retry, the outbound-screen repair, and the one repeat a thinking model gets when it answered with reasoning alone. Each call honoured the setting, but a local model needing 300 s per answer under a 600 s setting met the 630 s bound in the middle of its first corrective retry, and the briefing was dropped. The bound is now the per-call timeout times COMPREHENSIVE_MODEL_CALLS plus the headroom, for the nightly pass and the forced warm alike. A test drives the real warm, generator, chain runner, Local client, safeFetch and pinned dispatcher against a loopback server that answers after 300 s; under the old bound it reports a timeout on the third answer. The response-timeout guard now names every outer bound around model work and holds both briefing generators' stages to the call counts the outer waits multiply by. Refs #1274 --- src/__tests__/ai-call-timeout-guard.test.ts | 105 ++++- .../baselines/source-file-lines.json | 4 +- .../baselines/version-stamp-comments.json | 4 +- src/lib/ai/ai-budgets.ts | 18 +- .../briefing-warm-slow-model.test.ts | 371 ++++++++++++++++++ .../__tests__/insight-pregenerate.test.ts | 72 ++-- src/lib/jobs/insight-pregenerate.ts | 82 ++-- src/lib/jobs/queue-runtime.ts | 2 +- 8 files changed, 562 insertions(+), 96 deletions(-) create mode 100644 src/lib/jobs/__tests__/briefing-warm-slow-model.test.ts diff --git a/src/__tests__/ai-call-timeout-guard.test.ts b/src/__tests__/ai-call-timeout-guard.test.ts index c8f9b097e..1dbad7b14 100644 --- a/src/__tests__/ai-call-timeout-guard.test.ts +++ b/src/__tests__/ai-call-timeout-guard.test.ts @@ -19,7 +19,14 @@ * 3. a client is constructed outside `provider.ts`, where no resolver binds * it (the offline evaluation harness is the one named exception); * 4. a surface opts out with `timeoutPolicy: "surface-ceiling"` without - * being named below with its reason. + * being named below with its reason; + * 5. an outer bound waits on several model calls in a row but is sized as + * one (#1274: the briefing warm gave a five-call generation one call + * plus 30 s, so a local model answering in 300 s under a 600 s setting + * lost its briefing on the first corrective retry). Every `withTimeout` + * around model work is named below with what it must allow, and the + * briefing generators' stages are held to the call counts the outer + * waits multiply by. * * Its limit: it reads source text. A provider object built by hand in * production code that dials the network itself would slip all four matchers, @@ -29,9 +36,10 @@ * * Mutation check: put `params.timeoutMs ?? 60_000` back into any client, drop * `bindResponseTimeout` from `resolveProviderChain`, construct a - * `LocalOpenAICompatibleClient` in a route, or add - * `timeoutPolicy: "surface-ceiling"` to the document summary: each one goes - * red here. + * `LocalOpenAICompatibleClient` in a route, add + * `timeoutPolicy: "surface-ceiling"` to the document summary, wrap a model + * call in a new `withTimeout`, or add a sixth `runBriefingCompletion` stage to + * the comprehensive generator: each one goes red here. */ import { readFileSync, readdirSync } from "node:fs"; import { join } from "node:path"; @@ -187,3 +195,92 @@ describe("surface ceilings are frozen", () => { expect(sites.sort()).toEqual(Object.keys(SURFACE_CEILING_SITES).sort()); }); }); + +/** + * Every `withTimeout` around model work. It cannot stop the calls it waits + * for, so it is a backstop against a wedged run and has to sit above the + * longest run those calls allow: the per-call timeout times the calls made + * one after another. `bound` is what each call site's budget must read. + */ +const OUTER_BOUND_SITES: Record< + string, + { sites: number; bound: RegExp; sizedBy?: RegExp; why: string } +> = { + "lib/jobs/insight-pregenerate.ts": { + sites: 2, + bound: /\bwarmBudgetMs\b/, + sizedBy: /\bwarmBudgetMs\s*=\s*comprehensiveWarmBudgetMs\s*\(/g, + why: "The comprehensive warm, nightly and forced: the per-call timeout times COMPREHENSIVE_MODEL_CALLS, plus headroom.", + }, + "lib/insights/status-provider.ts": { + sites: 1, + bound: /\beffectiveTimeoutMs\b/, + why: "One short status note per chain run, bounded at the per-call timeout on purpose: the card has deterministic fallback copy and the next read retries.", + }, +}; + +describe("outer bounds around model work allow every call they wait on", () => { + const files = walkSourceFiles(SRC, { floor: 3000 }) + .filter((p) => !p.startsWith("generated/")) + .filter((p) => !p.includes("__tests__")) + .filter((p) => !p.endsWith(".test.ts") && !p.endsWith(".test.tsx")) + .filter((p) => p !== "lib/insights/with-timeout.ts"); + const callsIn = (rel: string): string[] => { + const source = code(rel); + return [...source.matchAll(/\bwithTimeout\s*\(/g)].map((m) => + source.slice(m.index, m.index + 900), + ); + }; + const sites = files.filter((rel) => callsIn(rel).length > 0); + + it("names every file that bounds model work from outside", () => { + expect(sites.length).toBeGreaterThan(0); + expect(sites.sort()).toEqual(Object.keys(OUTER_BOUND_SITES).sort()); + }); + + it.each(Object.entries(OUTER_BOUND_SITES))( + "%s sizes each bound from the setting", + (rel, { sites: count, bound, sizedBy }) => { + const calls = callsIn(rel); + expect(calls).toHaveLength(count); + for (const call of calls) expect(call).toMatch(bound); + if (sizedBy) { + expect(code(rel).match(sizedBy) ?? []).toHaveLength(count); + } + }, + ); + + /** The `stage` of every briefing completion a file starts. */ + function stagesIn(rel: string): string[] { + return [...code(rel).matchAll(/\bstage:\s*"([a-z-]+)",/g)].map((m) => m[1]); + } + + it("the comprehensive generator makes no more calls in a row than the warm waits for", () => { + // `reroll` replaces `generate` on an unchanged snapshot; the other four + // run one after another, and the Local and gateway clients may ask the + // first again once. A new stage means a longer generation: recount + // COMPREHENSIVE_MODEL_CALLS before adding it here. + expect(stagesIn("lib/insights/comprehensive-generate.ts").sort()).toEqual([ + "generate", + "grounding-retry", + "json-retry", + "reroll", + "screen-retry", + ]); + expect(code("lib/ai/ai-budgets.ts")).toMatch( + /\bCOMPREHENSIVE_MODEL_CALLS\s*=\s*5\b/, + ); + }); + + it("the on-demand regenerate makes no more calls in a row than the browser waits for", () => { + // The generation and its grounding retry, and the one repeat inside the + // client: FORCE_ADVISOR_MODEL_CALLS in the advisor hook. + expect([ + ...stagesIn("lib/insights/briefing-request/completion.ts"), + ...stagesIn("lib/insights/briefing-request/grounding.ts"), + ]).toEqual(["generate", "grounding-retry"]); + expect(code("components/insights/use-insights-advisor.ts")).toMatch( + /\bFORCE_ADVISOR_MODEL_CALLS\s*=\s*3\b/, + ); + }); +}); diff --git a/src/__tests__/baselines/source-file-lines.json b/src/__tests__/baselines/source-file-lines.json index f03269e86..a63b419a0 100644 --- a/src/__tests__/baselines/source-file-lines.json +++ b/src/__tests__/baselines/source-file-lines.json @@ -1,7 +1,7 @@ { "$comment": "Production files over 800 lines and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.", "count": 89, - "total": 99593, + "total": 99565, "entries": { "app/medications/page-client.tsx": 839, "components/charts/mood-chart.tsx": 1166, @@ -68,7 +68,7 @@ "lib/insights/status-cache.ts": 812, "lib/integrations/status.ts": 1445, "lib/jobs/coach-nudge.ts": 1084, - "lib/jobs/insight-pregenerate.ts": 1259, + "lib/jobs/insight-pregenerate.ts": 1231, "lib/jobs/medication-intake-import.ts": 819, "lib/jobs/offhost-backup.ts": 1147, "lib/jobs/reaction-line.ts": 824, diff --git a/src/__tests__/baselines/version-stamp-comments.json b/src/__tests__/baselines/version-stamp-comments.json index 33e262ea2..41f17cc58 100644 --- a/src/__tests__/baselines/version-stamp-comments.json +++ b/src/__tests__/baselines/version-stamp-comments.json @@ -1,7 +1,7 @@ { "$comment": "Comment lines that open with a release version, per file under src/. Maintained by src/__tests__/version-stamp-ratchet.test.ts.", "count": 2897, - "total": 9285, + "total": 9283, "entries": { "__tests__/account-payload-consumer-guard.test.ts": 2, "__tests__/acting-account-boundary-guard.test.ts": 7, @@ -2324,7 +2324,7 @@ "lib/jobs/google-health-oauth-state-cleanup.ts": 1, "lib/jobs/google-health-sleep-repair.ts": 1, "lib/jobs/insight-pregenerate-shared.ts": 1, - "lib/jobs/insight-pregenerate.ts": 39, + "lib/jobs/insight-pregenerate.ts": 37, "lib/jobs/insight-status-generate-shared.ts": 3, "lib/jobs/insight-status-generate.ts": 3, "lib/jobs/intake-auto-skip.ts": 8, diff --git a/src/lib/ai/ai-budgets.ts b/src/lib/ai/ai-budgets.ts index 5c033e8ba..fbf586b68 100644 --- a/src/lib/ai/ai-budgets.ts +++ b/src/lib/ai/ai-budgets.ts @@ -76,7 +76,8 @@ export const AI_BUDGETS = { * per-user `aiResponseTimeoutSeconds` setting still overrides this default * for operators on even slower backends; this only lifts the floor for * accounts that never set it. The worker cap in `insight-pregenerate.ts` - * derives from this value plus fixed headroom, so it tracks automatically. + * derives from this value times `COMPREHENSIVE_MODEL_CALLS` plus fixed + * headroom, so it tracks automatically. */ comprehensive: { temperature: 0.3, timeoutMs: 180_000 }, @@ -249,6 +250,21 @@ export const AI_BUDGETS = { documentChat: { temperature: 0.3, maxTokens: 600 }, } as const satisfies Record; +/** + * The most model calls one comprehensive generation makes one after another: + * the generation itself, the JSON retry when its reply does not parse, the + * grounding retry when the briefing quotes a figure the server never + * computed and the repair when the outbound screen withholds the payload, + * plus the one repeat the Local and gateway clients send when a thinking + * model answered with reasoning alone. Every one of them has the person's + * response timeout (or `comprehensive.timeoutMs`) to itself, so whatever + * waits on the whole generation from outside (the warm's abort in + * `insight-pregenerate.ts`, the Insights page's poll for the warm) waits for + * this many (#1274). `ai-call-timeout-guard.test.ts` holds the generator's + * stages to this count. + */ +export const COMPREHENSIVE_MODEL_CALLS = 5; + /** Bounds + default for the comprehensive / briefing output-token ceiling. */ const INSIGHTS_MAX_TOKENS_DEFAULT = 2500; const INSIGHTS_MAX_TOKENS_MIN = 500; diff --git a/src/lib/jobs/__tests__/briefing-warm-slow-model.test.ts b/src/lib/jobs/__tests__/briefing-warm-slow-model.test.ts new file mode 100644 index 000000000..3cad7c6f9 --- /dev/null +++ b/src/lib/jobs/__tests__/briefing-warm-slow-model.test.ts @@ -0,0 +1,371 @@ +/** + * #1274 — the out-of-band briefing warm against a local model that needs + * 300 s per answer, with the response timeout at 600 s. + * + * The warm is what the Insights page relies on: the page's read serves the + * cache and enqueues it, and the regenerate is only the explicit button. Its + * outer abort was sized as one model call plus headroom, while one + * comprehensive generation makes up to four calls one after another: the + * generation (asked once more inside the client when a thinking model answered + * with reasoning alone), the JSON retry and the grounding retry. Every single + * call honoured the 600 s; three 300 s answers in a row did not fit the 630 s + * the warm allowed, and the briefing was dropped. + * + * Driven end to end: the real forced warm, its real outer abort, the real + * comprehensive generator with its retries, the real budgeted chain runner, + * the real Local client with the setting bound by `bindResponseTimeout`, the + * real `safeFetch` and pinned dispatcher, and a real HTTP server on loopback + * that answers each request after 300 s. Time is fake for the timers the warm + * and the server use and real for the sockets; the database, the feature read + * and the capability gates are stand-ins. + */ +import http from "node:http"; +import type { AddressInfo } from "node:net"; +import { + afterAll, + afterEach, + beforeAll, + beforeEach, + describe, + expect, + it, + vi, +} from "vitest"; + +const findUnique = vi.hoisted(() => vi.fn()); +const updateMany = vi.hoisted(() => vi.fn()); +const chain = vi.hoisted(() => ({ current: [] as ProviderChainResolved[] })); + +const prismaMock = vi.hoisted(() => ({ + $queryRaw: vi.fn(async () => [{ total_tokens: 0, operator_tokens: 0 }]), + $executeRaw: vi.fn(async () => 0), + user: { + findUnique: (...a: unknown[]) => findUnique(...a), + update: vi.fn(async () => ({})), + updateMany: (...a: unknown[]) => updateMany(...a), + }, + auditLog: { + create: vi.fn(async () => ({})), + deleteMany: vi.fn(async () => ({ count: 0 })), + }, +})); + +vi.mock("@/lib/db", () => ({ prisma: prismaMock })); +vi.mock("@/lib/rate-limit", () => ({ + checkRateLimit: vi.fn(async () => ({ allowed: true })), +})); +vi.mock("@/lib/ai/capabilities/gate", () => { + const available = async () => ({ + available: true, + reason: null, + onDeviceAllowed: true, + }); + return { + aiCapabilityForJob: available, + aiCapabilityForRecord: available, + aiCapabilityToServe: available, + }; +}); +vi.mock("@/lib/ai/capabilities/egress", () => ({ + aiEgressRefusal: vi.fn(async () => null), +})); +vi.mock("@/lib/ai/provider", () => ({ + resolveProviderChain: vi.fn(async () => chain.current), + resolveProvider: vi.fn(async () => ({ type: "none" })), +})); +vi.mock("@/lib/ai/provider-health-ledger", async () => { + const actual = await vi.importActual< + typeof import("@/lib/ai/provider-health-ledger") + >("@/lib/ai/provider-health-ledger"); + return { + ...actual, + postgresProviderHealthLedger: { + async getSkipHints() { + return new Map(); + }, + async recordSuccess() {}, + async recordFailure() {}, + }, + }; +}); +vi.mock("@/lib/insights/features", () => ({ + FeaturesPayloadTooLargeError: class extends Error { + sizeBytes = 0; + }, + extractFeatures: vi.fn(async () => FEATURES), + 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 { bindResponseTimeout } from "@/lib/ai/effective-timeout"; +import { + LocalOpenAICompatibleClient, + resetLocalJsonDialectCache, +} from "@/lib/ai/local-client"; +import { clearLastWorkingProviderCache } from "@/lib/ai/provider-runner"; +import { resetReasoningDialectCache } from "@/lib/ai/reasoning/dialect-cache"; +import { openInsightsCachedText } from "@/lib/insights/cached-text-cipher"; +import { _resetPinnedDispatcherForTests } from "@/lib/safe-fetch-dispatcher"; +import { NextRequest } from "next/server"; +import { withFeatureCacheScope } from "@/lib/insights/feature-cache"; +import { loadGenerationContext } from "@/lib/insights/briefing-request/context"; +import { generateBriefing } from "@/lib/insights/briefing-request/generate"; +import { forceWarmUser } from "../insight-pregenerate"; +import type { PrismaClient } from "@/generated/prisma/client"; +import type { ProviderChainResolved } from "@/lib/ai/provider-runner"; + +type CacheColumns = Parameters[0]; +const warmPrisma = prismaMock as unknown as PrismaClient; + +vi.stubEnv("ENCRYPTION_KEY", "c".repeat(64)); + +/** A signals block, so the grounding gate engages, and a medications block. */ +const FEATURES = { + signalsOfDay: [ + { + sourceMetric: "weight", + latest: 81.4, + avg7: 82, + avg30: 82.1, + deltaVs7: -0.6, + deltaVs30: -0.7, + spread30: 1.2, + recentAnomaly: null, + }, + ], + medications: [ + { + name: "Ramipril", + dose: "5 mg", + category: "cardio", + compliance7: 100, + compliance30: 86, + compliance90: 88, + streak: 12, + missedLast7: 0, + }, + ], +}; + +/** A local model's first answer that is prose around the JSON, not JSON. */ +const PROSE = "Here is your briefing: the week looked steady overall."; +/** Valid JSON quoting a figure the server never computed. */ +const UNGROUNDED = JSON.stringify({ + dailyBriefing: { paragraph: "Your weight is down 47.9 kg this week." }, +}); +/** Valid JSON citing a figure the server computed. */ +const GROUNDED = JSON.stringify({ + dailyBriefing: { paragraph: "Adherence sits at 86 % over the last 30 days." }, +}); + +/** What the model takes for every answer, and the person's setting. */ +const ANSWER_MS = 300_000; +const RESPONSE_TIMEOUT_SECONDS = 600; +/** Fake time advanced per step of the drive loop. */ +const STEP_MS = 1_000; + +const realSetTimeout = globalThis.setTimeout; +let server: http.Server; +let origin: string; +let replies: string[] = []; +let answered = 0; + +function envelope(content: string) { + return { + id: "chatcmpl-1274", + object: "chat.completion", + created: 1_760_000_000, + model: "slow-model", + choices: [ + { + index: 0, + message: { role: "assistant", content }, + finish_reason: "stop", + }, + ], + usage: { prompt_tokens: 5200, completion_tokens: 600, total_tokens: 5800 }, + }; +} + +beforeAll(async () => { + server = http.createServer((req, res) => { + req.resume(); + req.on("end", () => { + const content = replies.shift() ?? GROUNDED; + // A non-streamed answer: nothing, not even the headers, until it is done. + setTimeout(() => { + answered += 1; + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify(envelope(content))); + }, ANSWER_MS); + }); + }); + await new Promise((r) => server.listen(0, "127.0.0.1", r)); + origin = `http://127.0.0.1:${(server.address() as AddressInfo).port}`; +}); + +afterAll(async () => { + _resetPinnedDispatcherForTests(); + await new Promise((r) => server.close(() => r())); +}); + +const savedOrigins = process.env.AI_PRIVATE_ORIGINS; + +beforeEach(() => { + process.env.AI_PRIVATE_ORIGINS = origin; + replies = []; + answered = 0; + findUnique.mockReset(); + findUnique.mockResolvedValue({ + id: "u1", + locale: "de", + timezone: "Europe/Berlin", + insightsPrivacyMode: "aggregated", + insightsCachedAt: null, + insightsCachedTextEncrypted: null, + insightsCachedLocale: null, + insightsExcludeMetrics: [], + insightsSnapshotHash: null, + insightsBriefingRerollDate: null, + insightsWarmFailedAt: null, + aiResponseTimeoutSeconds: RESPONSE_TIMEOUT_SECONDS, + }); + updateMany.mockReset(); + updateMany.mockResolvedValue({ count: 1 }); + clearLastWorkingProviderCache(); + resetLocalJsonDialectCache(); + resetReasoningDialectCache(); + const client = new LocalOpenAICompatibleClient({ + baseUrl: `${origin}/v1`, + model: "slow-model", + }); + chain.current = [ + { + providerType: "local", + instance: bindResponseTimeout(client, RESPONSE_TIMEOUT_SECONDS), + }, + ]; + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); +}); + +afterEach(() => { + vi.useRealTimers(); + // A warm cut short leaves its request open on the server; drop it. + server.closeAllConnections(); + if (savedOrigins === undefined) delete process.env.AI_PRIVATE_ORIGINS; + else process.env.AI_PRIVATE_ORIGINS = savedOrigins; +}); + +/** + * Run `work` to completion, advancing fake time a step at a time and giving + * the sockets a real moment between steps. Returns the fake time it took. + */ +async function drive( + work: Promise, + limitMs: number, +): Promise<{ value: T; elapsedMs: number }> { + let settled = false; + let value: T | undefined; + let failure: unknown = null; + work.then( + (v) => { + settled = true; + value = v; + }, + (e: unknown) => { + settled = true; + failure = e; + }, + ); + let elapsedMs = 0; + while (!settled && elapsedMs <= limitMs) { + await new Promise((r) => realSetTimeout(r, 1)); + await vi.advanceTimersByTimeAsync(STEP_MS); + elapsedMs += STEP_MS; + } + if (failure) throw failure; + expect(settled, `still running after ${limitMs} ms`).toBe(true); + return { value: value as T, elapsedMs }; +} + +/** The briefing paragraph the warm committed, if it committed one. */ +function committedParagraph(): string | null { + const writes = updateMany.mock.calls + .map((c) => openInsightsCachedText((c[0] as { data: CacheColumns }).data)) + .filter((t): t is string => t !== null); + const last = writes.at(-1); + if (!last) return null; + const parsed = JSON.parse(last) as { + dailyBriefing?: { paragraph?: string }; + }; + return parsed.dailyBriefing?.paragraph ?? null; +} + +describe("briefing warm against a model that needs 300 s per answer (600 s setting)", () => { + it("commits a briefing that takes one answer", async () => { + replies = [GROUNDED]; + const { value, elapsedMs } = await drive( + forceWarmUser(warmPrisma, "u1", "de", { today: true }), + 30 * 60_000, + ); + expect(value.comprehensive).toBe("generated"); + expect(answered).toBe(1); + expect(elapsedMs).toBeGreaterThanOrEqual(ANSWER_MS); + expect(committedParagraph()).toMatch(/86 %/); + }); + + it("commits a briefing that needs the JSON retry and the grounding retry", async () => { + // Prose first (JSON retry), then an invented figure (grounding retry), + // then a grounded answer: three answers, 900 s, each well inside 600 s. + replies = [PROSE, UNGROUNDED, GROUNDED]; + const { value, elapsedMs } = await drive( + forceWarmUser(warmPrisma, "u1", "de", { today: true }), + 60 * 60_000, + ); + expect(value.comprehensive).toBe("generated"); + expect(answered).toBe(3); + expect(elapsedMs).toBeGreaterThanOrEqual(3 * ANSWER_MS); + expect(committedParagraph()).toMatch(/86 %/); + }); +}); + +describe("the regenerate's server half against the same model", () => { + it("answers the POST with the briefing after the generation and its grounding retry", async () => { + // The route generates inline with no deadline of its own; the browser + // waits `forceAdvisorTimeoutMs` (three calls at the setting, plus 30 s). + replies = [UNGROUNDED, GROUNDED]; + const work = withFeatureCacheScope(async () => { + const ctx = await loadGenerationContext({ + request: new NextRequest("http://localhost/api/insights/generate", { + method: "POST", + }), + userId: "u1", + userLocale: "de", + }); + return generateBriefing({ + ctx, + chain: chain.current, + ipAddress: null, + }); + }); + const { value, elapsedMs } = await drive(work, 30 * 60_000); + expect(value.ok).toBe(true); + expect(answered).toBe(2); + expect(elapsedMs).toBeGreaterThanOrEqual(2 * ANSWER_MS); + expect(committedParagraph()).toMatch(/86 %/); + }); +}); diff --git a/src/lib/jobs/__tests__/insight-pregenerate.test.ts b/src/lib/jobs/__tests__/insight-pregenerate.test.ts index f413497c1..bbdb13376 100644 --- a/src/lib/jobs/__tests__/insight-pregenerate.test.ts +++ b/src/lib/jobs/__tests__/insight-pregenerate.test.ts @@ -138,7 +138,7 @@ import { jobOf, loadedHandler, } from "../registry/__tests__/registry-helpers"; -import { AI_BUDGETS } from "@/lib/ai/ai-budgets"; +import { AI_BUDGETS, COMPREHENSIVE_MODEL_CALLS } from "@/lib/ai/ai-budgets"; function makePrisma( users: Array<{ @@ -482,10 +482,9 @@ describe("runInsightPregenerate — outcome tally", () => { observedSignal = opts.signal; setTimeout( () => resolve({ status: "generated", providerType: "x" }), - // v1.25.12 — above the default warm budget (~210 s = the 180 s - // comprehensive provider budget + headroom) so the bounded budget + // v1.25.12 — above the default warm budget so the bounded budget // fires before this "slow" generation settles. - 260_000, + comprehensiveWarmBudgetMs(null) + 50_000, ); }), ); @@ -501,7 +500,9 @@ describe("runInsightPregenerate — outcome tally", () => { statusGenerators, warmGenericMetrics, }); - await vi.advanceTimersByTimeAsync(230_000); + await vi.advanceTimersByTimeAsync( + comprehensiveWarmBudgetMs(null) + 20_000, + ); const result = await promise; // The timeout fired, aborted the still-running generation (so its @@ -610,7 +611,7 @@ describe("runInsightPregenerate — failure visibility + intra-day retry (v1.28. new Promise((resolve) => { setTimeout( () => resolve({ status: "generated", providerType: "x" }), - 260_000, + comprehensiveWarmBudgetMs(null) + 50_000, ); }), ); @@ -628,7 +629,9 @@ describe("runInsightPregenerate — failure visibility + intra-day retry (v1.28. warmGenericMetrics, enqueueRetry, }); - await vi.advanceTimersByTimeAsync(230_000); + await vi.advanceTimersByTimeAsync( + comprehensiveWarmBudgetMs(null) + 20_000, + ); const result = await promise; expect(result.failed).toBe(1); @@ -1230,10 +1233,9 @@ describe("forceWarmUser — on-demand single-user warm (v1.8.7.1)", () => { // advance past it so the warm continues. setTimeout( () => resolve({ status: "generated", providerType: "x" }), - // v1.25.12 — above the default warm budget (~210 s = the 180 s - // comprehensive provider budget + headroom) so the bounded budget + // v1.25.12 — above the default warm budget so the bounded budget // fires before this "slow" generation settles. - 260_000, + comprehensiveWarmBudgetMs(null) + 50_000, ); }), ); @@ -1250,7 +1252,9 @@ describe("forceWarmUser — on-demand single-user warm (v1.8.7.1)", () => { warmGenericMetrics, }); // Advance past the comprehensive budget so withTimeout fires. - await vi.advanceTimersByTimeAsync(230_000); + await vi.advanceTimersByTimeAsync( + comprehensiveWarmBudgetMs(null) + 20_000, + ); const result = await promise; for (const g of statusGenerators) { @@ -1372,10 +1376,9 @@ describe("forceWarmUser — on-demand single-user warm (v1.8.7.1)", () => { observedSignal = opts.signal; setTimeout( () => resolve({ status: "generated", providerType: "x" }), - // v1.25.12 — above the default warm budget (~210 s = the 180 s - // comprehensive provider budget + headroom) so the bounded budget + // v1.25.12 — above the default warm budget so the bounded budget // fires before this "slow" generation settles. - 260_000, + comprehensiveWarmBudgetMs(null) + 50_000, ); }), ); @@ -1391,7 +1394,9 @@ describe("forceWarmUser — on-demand single-user warm (v1.8.7.1)", () => { statusGenerators, warmGenericMetrics, }); - await vi.advanceTimersByTimeAsync(230_000); + await vi.advanceTimersByTimeAsync( + comprehensiveWarmBudgetMs(null) + 20_000, + ); const result = await promise; // The timeout fired and aborted the still-running generation. @@ -1436,32 +1441,37 @@ describe("queue registration", () => { }); }); -describe("comprehensiveWarmBudgetMs — warm budget scales with the response timeout (v1.25.3)", () => { +describe("comprehensiveWarmBudgetMs — the warm waits for every call of one generation", () => { const HEADROOM_MS = 30_000; const DEFAULT_MS = AI_BUDGETS.comprehensive.timeoutMs; + const CALLS = COMPREHENSIVE_MODEL_CALLS; - it("falls back to the comprehensive surface budget + headroom for an unset value", () => { - expect(comprehensiveWarmBudgetMs(null)).toBe(DEFAULT_MS + HEADROOM_MS); - expect(comprehensiveWarmBudgetMs(undefined)).toBe(DEFAULT_MS + HEADROOM_MS); + it("falls back to the comprehensive surface budget per call for an unset value", () => { + const unset = DEFAULT_MS * CALLS + HEADROOM_MS; + expect(comprehensiveWarmBudgetMs(null)).toBe(unset); + expect(comprehensiveWarmBudgetMs(undefined)).toBe(unset); // A non-positive stored value is treated as unset. - expect(comprehensiveWarmBudgetMs(0)).toBe(DEFAULT_MS + HEADROOM_MS); + expect(comprehensiveWarmBudgetMs(0)).toBe(unset); }); - it("scales the budget with a raised setting and stays above the old fixed 130 s cap", () => { - // The regression: any account whose response timeout was >= ~130 s had its - // comprehensive warm clipped by the old flat 130 s cap, never landing. - const budget = comprehensiveWarmBudgetMs(300); - expect(budget).toBe(300_000 + HEADROOM_MS); - expect(budget).toBeGreaterThan(130_000); + it("gives each sequential call the full setting (#1274)", () => { + // Three 300 s answers (the generation, the JSON retry, the grounding + // retry) under a 600 s setting used to meet a 630 s bound. + expect(CALLS).toBeGreaterThanOrEqual(4); + for (const seconds of [10, 60, 180, 300, 600]) { + expect(comprehensiveWarmBudgetMs(seconds)).toBe( + seconds * 1000 * CALLS + HEADROOM_MS, + ); + } + expect(comprehensiveWarmBudgetMs(600)).toBeGreaterThan(3 * 300_000); }); - it("clamps to an absolute ceiling that still sits above the 600 s write-time maximum", () => { - // The 600 s max a user can configure is honoured (not clipped below it); - // the ceiling adds the headroom so a legitimate 600 s setting fits. - expect(comprehensiveWarmBudgetMs(600)).toBe(600_000 + HEADROOM_MS); + it("clamps to a ceiling that still covers the 600 s write-time maximum", () => { + const ceiling = 600_000 * CALLS + HEADROOM_MS; + expect(comprehensiveWarmBudgetMs(600)).toBe(ceiling); // A value beyond the write-time bound can never reach here, but the clamp // is the structural backstop against a wedged call regardless. - expect(comprehensiveWarmBudgetMs(100_000)).toBe(600_000 + HEADROOM_MS); + expect(comprehensiveWarmBudgetMs(100_000)).toBe(ceiling); }); }); diff --git a/src/lib/jobs/insight-pregenerate.ts b/src/lib/jobs/insight-pregenerate.ts index b8a6ce6b8..8b0e36674 100644 --- a/src/lib/jobs/insight-pregenerate.ts +++ b/src/lib/jobs/insight-pregenerate.ts @@ -79,7 +79,7 @@ import { type MetricStatusMetricId, } from "@/lib/insights/metric-status-registry"; import { withTimeout } from "@/lib/insights/with-timeout"; -import { AI_BUDGETS } from "@/lib/ai/ai-budgets"; +import { AI_BUDGETS, COMPREHENSIVE_MODEL_CALLS } from "@/lib/ai/ai-budgets"; import { resolveEffectiveTimeoutMs } from "@/lib/ai/effective-timeout"; import { INSIGHT_PREGENERATE_QUEUE, @@ -153,71 +153,42 @@ const WARM_PASS_CONCURRENCY = 3; /** * Bounded budget for one comprehensive generation inside a warm pass — - * the forced single-user warm AND the nightly per-candidate loop. The - * comprehensive insight is the single heaviest generation (full feature - * extraction + a 1500-token completion) and a slow or stalled provider - * could pin the whole warm for a minute-plus, after which the entire job - * aborted with the per-status + generic-metric passes never run. Capping - * it lets both paths abandon a slow comprehensive and still warm the - * cheaper per-status (7) + generic-metric (~30) assessments, so the user's - * cards are warm even when the briefing generation stalls. In the nightly - * loop the cap also bounds head-of-line blocking: without it one stalled - * provider pinned the whole batch behind a single candidate. Set a hair - * below the queue's own retry horizon so a single tick stays bounded. + * the forced single-user warm AND the nightly per-candidate loop. A slow or + * stalled provider must not keep the per-status (7) + generic-metric (~30) + * passes from running, and in the nightly loop one stalled provider must not + * pin the whole batch behind a single candidate. The bound is a backstop + * against a wedged generation, never the deadline of a working one: every + * provider call carries its own abort, so this sits above the longest + * generation those calls allow. * - * v1.21.5 — raised from 45 s. The old cap sat BELOW the provider call's own - * timeout (`AI_BUDGETS.comprehensive.timeoutMs`, now 120 s), so on a large - * account the warm bound — not the provider — aborted the generation, and the - * briefing never landed. The cap must sit just above the provider timeout so - * the provider's own abort governs a genuine completion while this stays the - * backstop against a wedged call. The per-user budget gate + the WARM_PASS - * concurrency cap still bound total nightly cost. - * - * v1.25.3 — the cap is no longer a fixed constant. The per-user - * `User.aiResponseTimeoutSeconds` setting (bounded 10–600 s at write-time) is - * the operator's lever for slow self-hosted / local backends, and the AI - * settings UI tells users to RAISE it for a model that loads slowly. The old - * fixed 130 s cap sat below the upper end of that range, so any account whose - * response timeout was ≥ ~130 s had every comprehensive warm clipped by the - * warm bound — not the provider — before the generation could land, leaving the - * briefing on its hard empty state. The warm budget now derives from the - * resolved effective per-request timeout plus a fixed headroom for feature - * extraction / prompt assembly / IO, so raising the response timeout actually - * extends the warm/cron budget. A sane absolute ceiling (the 600 s write-time - * maximum plus the same headroom) keeps a single wedged call from pinning the - * worker forever, while staying ABOVE the largest value a user can configure. - */ - -/** - * Fixed slack added on top of the resolved upstream provider timeout when - * sizing the warm/cron outer abort: bounded-feature extraction, prompt - * assembly, and the cache write all run inside the warm wrapper but outside the - * single provider round-trip the response-timeout governs. + * That generation is several calls, not one. Each provider call gets the + * account's response timeout (`User.aiResponseTimeoutSeconds`, 10–600 s at + * write time, the operator's lever for a slow local model; the + * comprehensive surface default when unset), and one generation makes up to + * `COMPREHENSIVE_MODEL_CALLS` of them in a row. A bound of one call plus + * headroom cut a local model that needs 300 s per answer at 630 s, in the + * middle of its corrective retry, however high the setting stood (#1274). + * The headroom covers what runs inside the warm but outside the calls: + * feature extraction, prompt assembly, the cache write. */ const COMPREHENSIVE_WARM_HEADROOM_MS = 30_000; /** * Largest value `User.aiResponseTimeoutSeconds` accepts (see the 10–600 s - * write-time bound in `POST /api/user/ai-provider`), in ms. The warm ceiling is - * derived from it so the ceiling is always ABOVE the maximum a user can set. + * write-time bound in `POST /api/user/ai-provider`), in ms. The ceiling below + * derives from it so it always sits ABOVE the most a person can set. */ const AI_RESPONSE_TIMEOUT_MAX_MS = 600_000; -/** - * Absolute backstop on the warm outer abort. A misconfigured 600 s response - * timeout still cannot pin the worker beyond this; the ceiling sits above the - * write-time maximum (by exactly the headroom) so a legitimate 600 s setting is - * honoured rather than silently clipped. - */ +/** Absolute backstop: the longest generation the largest setting allows. */ const COMPREHENSIVE_WARM_CEILING_MS = - AI_RESPONSE_TIMEOUT_MAX_MS + COMPREHENSIVE_WARM_HEADROOM_MS; + AI_RESPONSE_TIMEOUT_MAX_MS * COMPREHENSIVE_MODEL_CALLS + + COMPREHENSIVE_WARM_HEADROOM_MS; /** - * Outer-abort budget (ms) for one comprehensive warm, derived from the user's - * resolved effective upstream timeout. `effectiveTimeoutMs + headroom`, clamped - * to the absolute ceiling. A null/unset setting falls back to the comprehensive - * surface budget (`AI_BUDGETS.comprehensive.timeoutMs`), so a default account - * gets `120 s + headroom` rather than the old flat 130 s. + * Outer-abort budget (ms) for one comprehensive warm: the account's per-call + * timeout times the calls one generation may make, plus headroom, clamped to + * the ceiling. */ export function comprehensiveWarmBudgetMs( aiResponseTimeoutSeconds: number | null | undefined, @@ -227,7 +198,8 @@ export function comprehensiveWarmBudgetMs( AI_BUDGETS.comprehensive.timeoutMs, ); return Math.min( - effectiveTimeoutMs + COMPREHENSIVE_WARM_HEADROOM_MS, + effectiveTimeoutMs * COMPREHENSIVE_MODEL_CALLS + + COMPREHENSIVE_WARM_HEADROOM_MS, COMPREHENSIVE_WARM_CEILING_MS, ); } diff --git a/src/lib/jobs/queue-runtime.ts b/src/lib/jobs/queue-runtime.ts index 071481bfe..9069b2b45 100644 --- a/src/lib/jobs/queue-runtime.ts +++ b/src/lib/jobs/queue-runtime.ts @@ -154,7 +154,7 @@ export const QUEUE_RUNTIME: Readonly> = { fn: "handleInsightPregenerateJob", }, exclusive: "lockedPass", - why: "Up to 200 accounts, a comprehensive generation each (a provider call bounded by the account's timeout, up to ten minutes) plus the status warm.", + why: "Up to 200 accounts, a comprehensive generation each (up to five provider calls in a row, each bounded by the account's timeout of at most ten minutes) plus the status warm.", }, "morning-digest-refresh": short( "One account, one comprehensive generation, bounded by the provider timeout.",