From 02d3d8b7bbe49d83401db6a6664c24a28cebfc00 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sat, 24 Oct 2026 11:00:32 +0200 Subject: [PATCH] fix(insights): let slow-model status notes and AI jobs outlast their calls A status run (a metric card, the batched assessment, a derived score, a workout line, a Coach memory pass) is one model call plus the one repeat the Local and gateway clients send when a thinking model answered with reasoning alone, each with the person's response timeout. The run's outer bound was a single call, so the repeat was cut off, and the open card stopped polling after 48 s, so a note from a model that needs minutes never reached the page. The bound is now the per-call timeout times STATUS_MODEL_CALLS, and the card's poll ceiling follows the published response timeout (fast for the first 48 s, every 15 s after), only while the route reports the note in flight; the fallback text stays meanwhile. The status queue is exclusive, so the longer poll cannot queue a second generation beside a running one. The morning refresh is one comprehensive generation of up to five calls in a row, with no outer bound and pg-boss's fifteen-minute expiry, so a large setting had pg-boss fail and retry it while it was still running. It now runs behind the same bound as the warms, and the job carries an expiry derived from that worst case. The per-metric note, the workout line and the Coach memory refresh get expiries derived the same way. queue-runtime.ts gains a bounded class for these jobs: model time read off model-time.ts, an expiry above it, carried on every send. A new guard names every caller of runStatusCompletion and generateComprehensiveInsight with the queues that reach it, and holds each such queue to bounded above its runs or long under a run lock. Refs #1274 --- src/__tests__/ai-call-timeout-guard.test.ts | 14 +- src/__tests__/ai-client-abort-guard.test.ts | 44 +++- src/__tests__/ai-job-expiry-guard.test.ts | 223 ++++++++++++++++++ .../baselines/source-file-lines.json | 4 +- src/__tests__/baselines/test-as-never.json | 4 +- .../baselines/version-stamp-comments.json | 4 +- .../client-poll-inventory-guard.test.ts | 5 +- src/__tests__/job-queue-runtime-guard.test.ts | 19 +- .../chat/__tests__/golden-transcripts.test.ts | 3 +- .../__tests__/route-cancelled-turn.test.ts | 3 +- .../chat/__tests__/route-fence-guard.test.ts | 3 +- .../__tests__/route-guard-activation.test.ts | 3 +- .../chat/__tests__/route-guard-ledger.test.ts | 3 +- .../chat/__tests__/route-guard.test.ts | 3 +- .../__tests__/route-snapshot-once.test.ts | 3 +- .../chat/__tests__/route-tool-mode.test.ts | 3 +- .../__tests__/route-workout-scope.test.ts | 3 +- src/app/insights/medications/page.tsx | 13 +- ...ealthkit-metric-page-hrv-measures.test.tsx | 1 + .../__tests__/lab-biomarker-detail.test.tsx | 1 + src/hooks/use-insight-status.ts | 91 +++++-- src/lib/ai/ai-budgets.ts | 11 + .../__tests__/coach-memory-shared.test.ts | 2 + src/lib/ai/coach/coach-memory-shared.ts | 21 ++ src/lib/ai/effective-timeout.ts | 8 + .../status-provider-slow-model.test.ts | 173 ++++++++++++++ src/lib/insights/status-provider.ts | 36 ++- .../__tests__/morning-digest-refresh.test.ts | 166 +++++++++++-- src/lib/jobs/insight-pregenerate.ts | 54 +---- src/lib/jobs/insight-status-generate-queue.ts | 15 ++ .../jobs/insight-status-generate-shared.ts | 6 +- src/lib/jobs/model-time.ts | 77 ++++++ src/lib/jobs/morning-digest-refresh-shared.ts | 18 ++ src/lib/jobs/morning-digest-refresh.ts | 43 +++- src/lib/jobs/queue-runtime.ts | 66 +++++- .../__tests__/registry-contract.test.ts | 4 + src/lib/jobs/registry/status.ts | 15 +- .../jobs/workout-insight-generate-shared.ts | 16 ++ src/lib/validations/ai-provider.ts | 14 +- .../__golden__/queue-registration.json | 2 +- 40 files changed, 1053 insertions(+), 144 deletions(-) create mode 100644 src/__tests__/ai-job-expiry-guard.test.ts create mode 100644 src/lib/insights/__tests__/status-provider-slow-model.test.ts create mode 100644 src/lib/jobs/model-time.ts diff --git a/src/__tests__/ai-call-timeout-guard.test.ts b/src/__tests__/ai-call-timeout-guard.test.ts index 7704967fe..452926bcd 100644 --- a/src/__tests__/ai-call-timeout-guard.test.ts +++ b/src/__tests__/ai-call-timeout-guard.test.ts @@ -26,7 +26,8 @@ * 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. + * waits multiply by. A status run waits for its call and the one + * reasoning repeat (`STATUS_MODEL_CALLS`). * * 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, @@ -214,8 +215,15 @@ const OUTER_BOUND_SITES: Record< }, "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.", + bound: /\bstatusRunBoundMs\(\s*effectiveTimeoutMs\s*\)/, + sizedBy: /\breturn\s+effectiveTimeoutMs\s*\*\s*STATUS_MODEL_CALLS\b/g, + why: "One status run in a background job: the per-call timeout times STATUS_MODEL_CALLS, the call and the one repeat a thinking model gets (#1274). The card keeps its fallback copy meanwhile.", + }, + "lib/jobs/morning-digest-refresh.ts": { + sites: 1, + bound: + /\bcomprehensiveWarmBudgetMs\(\s*user\.aiResponseTimeoutSeconds\s*\)/, + why: "The morning refresh's one comprehensive generation: the same per-call timeout times COMPREHENSIVE_MODEL_CALLS the warms use, below the job's expiry (#1274).", }, }; diff --git a/src/__tests__/ai-client-abort-guard.test.ts b/src/__tests__/ai-client-abort-guard.test.ts index 715ed673c..02b489030 100644 --- a/src/__tests__/ai-client-abort-guard.test.ts +++ b/src/__tests__/ai-client-abort-guard.test.ts @@ -23,16 +23,28 @@ * `insight-pregenerate.test.ts` proves it outlasts the warm's own bound for * every setting. * + * The status cards poll the same way while their note is generated, and + * stopped after 48 s whatever the setting, so a slow model's note never + * reached the open card. Every status poll passes + * `statusPollMaxAttempts(responseTimeoutMs)`, to the schedule and to the + * exhaustion that ends it; `status-provider-slow-model.test.ts` proves the + * window outlasts the run's own bound for every setting. + * * Mutation check: put `FORCE_ADVISOR_TIMEOUT_MS = 45_000` back into the * advisor, drop the `signal` from one about-me save, or drop the ceiling - * argument from the advisor read's poll, and this goes red. + * argument from the advisor read's poll or from one status poll, and this + * goes red. */ import { readFileSync } from "node:fs"; import { join } from "node:path"; import { describe, expect, it } from "vitest"; -import { stripComments, walkSourceFiles } from "./helpers/source-files"; +import { + readSourceText, + stripComments, + walkSourceFiles, +} from "./helpers/source-files"; const SRC = join(process.cwd(), "src"); @@ -83,6 +95,34 @@ describe("browser aborts for model-backed requests", () => { ); }); + it("every status card polls for its note as long as the setting allows", () => { + const calls: Array<{ rel: string; text: string }> = []; + for (const rel of walkSourceFiles(SRC, { floor: 3000 })) { + if (rel.includes("__tests__/") || rel.startsWith("generated/")) continue; + const source = stripComments(readSourceText(join(SRC, rel))); + for (const m of source.matchAll( + /\b(?:nextStatusPollInterval|applyStatusPollExhaustion)\s*\(/g, + )) { + const text = callText(source, (m.index ?? 0) + m[0].length - 1); + // The definitions, not calls. + if (/:\s*(?:boolean|InsightStatusData)\s*\|\s*undefined/.test(text)) { + continue; + } + calls.push({ rel, text }); + } + } + // Three hooks poll and end the poll, the compliance card polls. + expect(calls.length).toBeGreaterThanOrEqual(7); + expect( + calls + .filter( + ({ text }) => + !/statusPollMaxAttempts\(\s*responseTimeoutMs\s*\)/.test(text), + ) + .map(({ rel }) => rel), + ).toEqual([]); + }); + it("every about-me save carries the setting-derived signal", () => { const sites: string[] = []; const missing: string[] = []; diff --git a/src/__tests__/ai-job-expiry-guard.test.ts b/src/__tests__/ai-job-expiry-guard.test.ts new file mode 100644 index 000000000..f48d4739a --- /dev/null +++ b/src/__tests__/ai-job-expiry-guard.test.ts @@ -0,0 +1,223 @@ +/** + * Structural guard: no job that waits on the person's response timeout can + * outlive its pg-boss expiry. + * + * Every model call has the record owner's response timeout to itself, up to + * ten minutes (`AI_RESPONSE_TIMEOUT_MAX_SECONDS`). pg-boss fails a job that + * runs past its expiry (fifteen minutes unless the send says otherwise) and + * retries it while the handler keeps going. The morning refresh is one + * comprehensive generation of up to five calls in a row, so under a large + * setting it was retried beside itself (#1274); the per-metric note, the + * workout line and the Coach memory pass could each do the same once a status + * run waited for its reasoning repeat. + * + * Two entry points carry every background model run at the person's + * timeout: `runStatusCompletion` (a status run, `STATUS_MODEL_CALLS` calls) + * and `generateComprehensiveInsight` (`COMPREHENSIVE_MODEL_CALLS` calls). The + * guard holds: + * + * 1. every file that imports one of them is named below with the queues + * whose jobs reach it, so a new caller has to be attributed; + * 2. each of those queues is `bounded` in `queue-runtime.ts`, with model + * time at least the runs it makes at the largest setting and an expiry + * above that, or `long`, where the run's lock makes the retry after an + * overrun report `already_running` instead of working beside it; + * 3. a queue declared `short` cannot be one of them. + * + * Its limit: the run counts per queue are declared here, not derived; the + * Coach memory refresh's three runs, for instance, are three calls in + * `runCoachMemoryRefresh` that a reader has to count. Model calls outside the + * two entry points (the document and lab runs, the reaction line, the nudge) + * keep their own deadlines and are not covered here. + * + * Mutation check: declare `morning-digest-refresh` short again, drop a run + * from a queue's model time, or import `runStatusCompletion` into a new file, + * and this goes red. The "plants" case runs the check over the table as it + * stood before #1274. + */ +import { join } from "node:path"; + +import { describe, expect, it } from "vitest"; + +import { + readSourceText, + stripComments, + walkSourceFiles, +} from "./helpers/source-files"; +import { QUEUE_RUNTIME, type QueueRuntime } from "@/lib/jobs/queue-runtime"; +import { + COMPREHENSIVE_WARM_CEILING_MS, + STATUS_RUN_CEILING_MS, +} from "@/lib/jobs/model-time"; +import { + COMPREHENSIVE_MODEL_CALLS, + STATUS_MODEL_CALLS, +} from "@/lib/ai/ai-budgets"; +import { AI_RESPONSE_TIMEOUT_MAX_SECONDS } from "@/lib/ai/effective-timeout"; + +const SRC = join(process.cwd(), "src"); +const ENTRY_POINTS = ["runStatusCompletion", "generateComprehensiveInsight"]; + +/** Each importer of an entry point, and the queues whose jobs reach it. */ +const MODEL_RUN_SITES: Readonly> = { + "lib/insights/status-card-generation.ts": [ + "insight-status-generate", + "insights-general-status", + "insight-pregenerate", + ], + "lib/insights/derived/derived-assessment-ai.ts": ["insight-status-generate"], + "lib/insights/status-batch.ts": [ + "insights-general-status", + "insight-pregenerate", + ], + "lib/insights/narrative/period-narrative-generate.ts": [ + "period-narrative-warm", + ], + "lib/ai/coach/conversation-summary.ts": ["coach-memory-refresh"], + "lib/ai/coach/facts.ts": ["coach-memory-refresh"], + "lib/ai/coach/plans.ts": ["coach-memory-refresh"], + "lib/jobs/workout-insight-generate.ts": ["workout-insight-generate"], + "lib/jobs/insight-pregenerate.ts": ["insight-pregenerate"], + "lib/jobs/morning-digest-refresh.ts": ["morning-digest-refresh"], +}; + +/** What one job of a bounded queue runs, at most, one after another. */ +const RUNS: Readonly< + Record +> = { + "insight-status-generate": { + status: 1, + why: "Each scope's generator makes one status run (runPreparedStatusCard or the derived assessment).", + }, + "workout-insight-generate": { + status: 1, + why: "One status run behind the provider claim.", + }, + "coach-memory-refresh": { + status: 3, + why: "The summary, the facts and the plan proposals, one run each, in a row.", + }, + "morning-digest-refresh": { + comprehensive: 1, + why: "One forced comprehensive generation.", + }, +}; + +/** Value specifiers of every `import { … } from` in `text`. */ +function importedNames(text: string): string[] { + return [...text.matchAll(/import\s*\{([^}]*)\}\s*from\s*["'][^"']+["']/g)] + .flatMap((m) => m[1].split(",")) + .map((name) => name.trim()) + .filter((name) => name.length > 0 && !name.startsWith("type ")) + .map((name) => name.split(/\s+as\s+/)[0]); +} + +function importers(): string[] { + return 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((rel) => + importedNames(stripComments(readSourceText(join(SRC, rel)))).some( + (name) => ENTRY_POINTS.includes(name), + ), + ) + .sort(); +} + +/** Seconds of model work `runs` makes at the largest setting. */ +function modelSecondsOf(runs: { status?: number; comprehensive?: number }) { + return ( + ((runs.status ?? 0) * STATUS_RUN_CEILING_MS + + (runs.comprehensive ?? 0) * COMPREHENSIVE_WARM_CEILING_MS) / + 1000 + ); +} + +/** Every finding for the queues that reach a model run. */ +function findings(table: Readonly>): string[] { + const queues = [...new Set(Object.values(MODEL_RUN_SITES).flat())].sort(); + const problems: string[] = []; + for (const queue of queues) { + const entry = table[queue]; + if (!entry) { + problems.push(`${queue}: not in QUEUE_RUNTIME`); + continue; + } + if (entry.runtime === "long") { + if (entry.exclusive !== "lockedPass") { + problems.push(`${queue}: long without the run lock`); + } + continue; + } + if (entry.runtime !== "bounded") { + problems.push( + `${queue}: declared ${entry.runtime}, but it waits on the person's response timeout`, + ); + continue; + } + const runs = RUNS[queue]; + if (!runs) { + problems.push(`${queue}: bounded without its runs named here`); + continue; + } + const needed = modelSecondsOf(runs); + if (entry.modelSeconds < needed) { + problems.push( + `${queue}: model time ${entry.modelSeconds}s below the ${needed}s its runs take`, + ); + } + if (!(entry.expireInSeconds > entry.modelSeconds)) { + problems.push(`${queue}: expiry not above its model time`); + } + } + return problems; +} + +describe("jobs that wait on the response timeout outlast it", () => { + it("names every importer of a model run (an empty set is a failure)", () => { + const found = importers(); + expect(found.length).toBeGreaterThanOrEqual(10); + expect(found).toEqual(Object.keys(MODEL_RUN_SITES).sort()); + }); + + it("sizes the run ceilings from the largest setting and the call counts", () => { + expect(STATUS_RUN_CEILING_MS).toBe( + AI_RESPONSE_TIMEOUT_MAX_SECONDS * 1000 * STATUS_MODEL_CALLS, + ); + expect(COMPREHENSIVE_WARM_CEILING_MS).toBeGreaterThan( + AI_RESPONSE_TIMEOUT_MAX_SECONDS * 1000 * COMPREHENSIVE_MODEL_CALLS, + ); + }); + + it("every queue that reaches a model run is bounded above it or locked", () => { + expect(findings(QUEUE_RUNTIME)).toEqual([]); + }); + + it("names its runs only for queues that are bounded", () => { + expect( + Object.keys(RUNS).filter((q) => QUEUE_RUNTIME[q]?.runtime !== "bounded"), + ).toEqual([]); + }); + + it("plants: the table before #1274 is caught", () => { + const before: Record = { + ...QUEUE_RUNTIME, + "morning-digest-refresh": { + runtime: "short", + why: "One account, one comprehensive generation, bounded by the provider timeout.", + }, + "coach-memory-refresh": { + runtime: "bounded", + modelSeconds: STATUS_RUN_CEILING_MS / 1000, + expireInSeconds: 1400, + expiryVia: [], + why: "planted: one run counted instead of three", + }, + }; + expect(findings(before)).toEqual([ + `coach-memory-refresh: model time ${STATUS_RUN_CEILING_MS / 1000}s below the ${modelSecondsOf({ status: 3 })}s its runs take`, + "morning-digest-refresh: declared short, but it waits on the person's response timeout", + ]); + }); +}); diff --git a/src/__tests__/baselines/source-file-lines.json b/src/__tests__/baselines/source-file-lines.json index bf4ee77d3..15947d20d 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": 99521, + "total": 99479, "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": 1231, + "lib/jobs/insight-pregenerate.ts": 1189, "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/test-as-never.json b/src/__tests__/baselines/test-as-never.json index e852565a5..a13953429 100644 --- a/src/__tests__/baselines/test-as-never.json +++ b/src/__tests__/baselines/test-as-never.json @@ -1,7 +1,7 @@ { "$comment": "as never casts in tests, per test file under src/. Maintained by src/__tests__/test-realism-ratchet.test.ts.", "count": 524, - "total": 5495, + "total": 5490, "entries": { "__tests__/api/auth/login/mfa-branch.test.ts": 10, "__tests__/api/auth/mfa/verify.test.ts": 9, @@ -445,7 +445,7 @@ "lib/jobs/__tests__/medication-intake-import.test.ts": 1, "lib/jobs/__tests__/medication-reminder-check.test.ts": 27, "lib/jobs/__tests__/mood-reminder.test.ts": 15, - "lib/jobs/__tests__/morning-digest-refresh.test.ts": 6, + "lib/jobs/__tests__/morning-digest-refresh.test.ts": 1, "lib/jobs/__tests__/oauth-callback-claim-cleanup.test.ts": 1, "lib/jobs/__tests__/offhost-backup.test.ts": 5, "lib/jobs/__tests__/offhost-purge.test.ts": 14, diff --git a/src/__tests__/baselines/version-stamp-comments.json b/src/__tests__/baselines/version-stamp-comments.json index 41f17cc58..e3589a68f 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": 9283, + "total": 9282, "entries": { "__tests__/account-payload-consumer-guard.test.ts": 2, "__tests__/acting-account-boundary-guard.test.ts": 7, @@ -611,7 +611,7 @@ "app/insights/layout.tsx": 4, "app/insights/lean-body-mass/page.tsx": 1, "app/insights/loading.tsx": 1, - "app/insights/medications/page.tsx": 8, + "app/insights/medications/page.tsx": 7, "app/insights/mood/page-client.tsx": 4, "app/insights/muscle-mass/page.tsx": 1, "app/insights/oxygen/page.tsx": 1, diff --git a/src/__tests__/client-poll-inventory-guard.test.ts b/src/__tests__/client-poll-inventory-guard.test.ts index 8e905819b..48d16b993 100644 --- a/src/__tests__/client-poll-inventory-guard.test.ts +++ b/src/__tests__/client-poll-inventory-guard.test.ts @@ -122,12 +122,13 @@ const INVENTORY: Readonly> = { sites: 3, shape: "bounded", cadence: - "4 s, nextStatusPollInterval, 12 attempts (metric, biomarker, general cards)", + "4 s, then 15 s, nextStatusPollInterval, attempts from the response timeout so the note can land (metric, biomarker, general cards)", }, "app/insights/medications/page.tsx": { sites: 1, shape: "bounded", - cadence: "4 s, nextStatusPollInterval, 12 attempts (compliance card)", + cadence: + "4 s, then 15 s, nextStatusPollInterval, attempts from the response timeout so the note can land (compliance card)", }, "components/insights/use-insights-advisor.ts": { sites: 1, diff --git a/src/__tests__/job-queue-runtime-guard.test.ts b/src/__tests__/job-queue-runtime-guard.test.ts index 14c86c4e5..27a15294c 100644 --- a/src/__tests__/job-queue-runtime-guard.test.ts +++ b/src/__tests__/job-queue-runtime-guard.test.ts @@ -20,6 +20,8 @@ * stops on the job's budget where the table says (`jobBudget` / * `jobDeadline`, or a declared one-shot), and it is exclusive per identity * where the table says (the definition's `lock`, or the named marker); + * - a `bounded` queue (one unit of model work behind outer bounds) has an + * expiry above its declared model time and every send carries it; * - a `budgeted` queue stops on the budget and relies on the default expiry; * - a `short` queue's sends set no expiry of their own, which would mean the * job is longer than the table claims. @@ -161,7 +163,7 @@ function check( const definition = all.definitions.find((d) => d.queue === queue); const sends = all.sends.filter((s) => s.queue === queue); const scheduledExpiry = definition?.scheduleOptions?.expireInSeconds; - if (entry.runtime === "long") { + if (entry.runtime === "long" || entry.runtime === "bounded") { if ( !(entry.expireInSeconds > DEFAULT_JOB_EXPIRE_SECONDS) || !(entry.expireInSeconds < MAX_EXPIRE_SECONDS) @@ -179,6 +181,17 @@ function check( if (definition?.cron && typeof scheduledExpiry !== "number") { problems.push(`${queue}: its schedule carries no expiry`); } + } + if (entry.runtime === "bounded") { + if (!(entry.expireInSeconds > entry.modelSeconds)) { + problems.push( + `${queue}: expiry ${entry.expireInSeconds}s is not above its ${entry.modelSeconds}s of model work`, + ); + } + if (sends.length === 0 && !definition?.cron) { + problems.push(`${queue}: no send or schedule carries its expiry`); + } + } else if (entry.runtime === "long") { if (typeof entry.stop === "object" && "oneShot" in entry.stop) { if (entry.stop.oneShot.length < 20) { problems.push(`${queue}: a one-shot needs its reason`); @@ -313,7 +326,9 @@ describe("every queue declares its runtime", () => { { definitions: [planted({})], ...scan(texts(budgeted)) }, ), ).toEqual(["planted-long: expiry 900s out of range"]); - }); + // Each finding rescans every send in the tree; under a full parallel run + // that took six to nine seconds against the five-second default. + }, 30_000); }); /** diff --git a/src/app/api/insights/chat/__tests__/golden-transcripts.test.ts b/src/app/api/insights/chat/__tests__/golden-transcripts.test.ts index 5bdc6899b..563defbfa 100644 --- a/src/app/api/insights/chat/__tests__/golden-transcripts.test.ts +++ b/src/app/api/insights/chat/__tests__/golden-transcripts.test.ts @@ -89,7 +89,8 @@ vi.mock("@/lib/ai/consent-guard", () => ({ vi.mock("@/lib/ai/prompts/insight-generator", () => ({ PROMPT_VERSION: "golden", })); -vi.mock("@/lib/ai/ai-budgets", () => ({ +vi.mock("@/lib/ai/ai-budgets", async (importOriginal) => ({ + ...(await importOriginal()), AI_BUDGETS: { coach: { maxTokens: 600, temperature: 0.4 } }, })); vi.mock("@/lib/ai/coach/persistence", () => ({ diff --git a/src/app/api/insights/chat/__tests__/route-cancelled-turn.test.ts b/src/app/api/insights/chat/__tests__/route-cancelled-turn.test.ts index a945487be..f620f50f1 100644 --- a/src/app/api/insights/chat/__tests__/route-cancelled-turn.test.ts +++ b/src/app/api/insights/chat/__tests__/route-cancelled-turn.test.ts @@ -79,7 +79,8 @@ vi.mock("@/lib/ai/consent-guard", () => ({ assertConsentForChain: vi.fn(async () => undefined), })); vi.mock("@/lib/ai/prompts/insight-generator", () => ({ PROMPT_VERSION: "x" })); -vi.mock("@/lib/ai/ai-budgets", () => ({ +vi.mock("@/lib/ai/ai-budgets", async (importOriginal) => ({ + ...(await importOriginal()), AI_BUDGETS: { coach: { maxTokens: 1500, temperature: 0.4 } }, })); diff --git a/src/app/api/insights/chat/__tests__/route-fence-guard.test.ts b/src/app/api/insights/chat/__tests__/route-fence-guard.test.ts index f966c86c3..975429f3c 100644 --- a/src/app/api/insights/chat/__tests__/route-fence-guard.test.ts +++ b/src/app/api/insights/chat/__tests__/route-fence-guard.test.ts @@ -68,7 +68,8 @@ vi.mock("@/lib/ai/consent-guard", () => ({ assertConsentForChain: vi.fn(async () => undefined), })); vi.mock("@/lib/ai/prompts/insight-generator", () => ({ PROMPT_VERSION: "x" })); -vi.mock("@/lib/ai/ai-budgets", () => ({ +vi.mock("@/lib/ai/ai-budgets", async (importOriginal) => ({ + ...(await importOriginal()), AI_BUDGETS: { coach: { maxTokens: 1500, temperature: 0.4 } }, })); const { fetchConversationWithMessages } = vi.hoisted(() => ({ diff --git a/src/app/api/insights/chat/__tests__/route-guard-activation.test.ts b/src/app/api/insights/chat/__tests__/route-guard-activation.test.ts index 18b4e8175..9f189d00b 100644 --- a/src/app/api/insights/chat/__tests__/route-guard-activation.test.ts +++ b/src/app/api/insights/chat/__tests__/route-guard-activation.test.ts @@ -98,7 +98,8 @@ const { assertConsentForChain } = vi.hoisted(() => ({ })); vi.mock("@/lib/ai/consent-guard", () => ({ assertConsentForChain })); vi.mock("@/lib/ai/prompts/insight-generator", () => ({ PROMPT_VERSION: "x" })); -vi.mock("@/lib/ai/ai-budgets", () => ({ +vi.mock("@/lib/ai/ai-budgets", async (importOriginal) => ({ + ...(await importOriginal()), AI_BUDGETS: { coach: { maxTokens: 1500, temperature: 0.4 } }, })); diff --git a/src/app/api/insights/chat/__tests__/route-guard-ledger.test.ts b/src/app/api/insights/chat/__tests__/route-guard-ledger.test.ts index 7b2f1c9c1..ba5aec9df 100644 --- a/src/app/api/insights/chat/__tests__/route-guard-ledger.test.ts +++ b/src/app/api/insights/chat/__tests__/route-guard-ledger.test.ts @@ -91,7 +91,8 @@ const { assertConsentForChain } = vi.hoisted(() => ({ })); vi.mock("@/lib/ai/consent-guard", () => ({ assertConsentForChain })); vi.mock("@/lib/ai/prompts/insight-generator", () => ({ PROMPT_VERSION: "x" })); -vi.mock("@/lib/ai/ai-budgets", () => ({ +vi.mock("@/lib/ai/ai-budgets", async (importOriginal) => ({ + ...(await importOriginal()), AI_BUDGETS: { coach: { maxTokens: 1500, temperature: 0.4 } }, })); diff --git a/src/app/api/insights/chat/__tests__/route-guard.test.ts b/src/app/api/insights/chat/__tests__/route-guard.test.ts index 97d5c793f..6499eb771 100644 --- a/src/app/api/insights/chat/__tests__/route-guard.test.ts +++ b/src/app/api/insights/chat/__tests__/route-guard.test.ts @@ -82,7 +82,8 @@ const { assertConsentForChain } = vi.hoisted(() => ({ })); vi.mock("@/lib/ai/consent-guard", () => ({ assertConsentForChain })); vi.mock("@/lib/ai/prompts/insight-generator", () => ({ PROMPT_VERSION: "x" })); -vi.mock("@/lib/ai/ai-budgets", () => ({ +vi.mock("@/lib/ai/ai-budgets", async (importOriginal) => ({ + ...(await importOriginal()), AI_BUDGETS: { coach: { maxTokens: 1500, temperature: 0.4 } }, })); diff --git a/src/app/api/insights/chat/__tests__/route-snapshot-once.test.ts b/src/app/api/insights/chat/__tests__/route-snapshot-once.test.ts index 7166b9981..401598337 100644 --- a/src/app/api/insights/chat/__tests__/route-snapshot-once.test.ts +++ b/src/app/api/insights/chat/__tests__/route-snapshot-once.test.ts @@ -77,7 +77,8 @@ vi.mock("@/lib/ai/consent-guard", () => ({ assertConsentForChain: vi.fn(async () => undefined), })); vi.mock("@/lib/ai/prompts/insight-generator", () => ({ PROMPT_VERSION: "x" })); -vi.mock("@/lib/ai/ai-budgets", () => ({ +vi.mock("@/lib/ai/ai-budgets", async (importOriginal) => ({ + ...(await importOriginal()), AI_BUDGETS: { coach: { maxTokens: 1500, temperature: 0.4 } }, })); diff --git a/src/app/api/insights/chat/__tests__/route-tool-mode.test.ts b/src/app/api/insights/chat/__tests__/route-tool-mode.test.ts index c830b5e65..7f25ae425 100644 --- a/src/app/api/insights/chat/__tests__/route-tool-mode.test.ts +++ b/src/app/api/insights/chat/__tests__/route-tool-mode.test.ts @@ -99,7 +99,8 @@ const { assertConsentForChain } = vi.hoisted(() => ({ })); vi.mock("@/lib/ai/consent-guard", () => ({ assertConsentForChain })); vi.mock("@/lib/ai/prompts/insight-generator", () => ({ PROMPT_VERSION: "x" })); -vi.mock("@/lib/ai/ai-budgets", () => ({ +vi.mock("@/lib/ai/ai-budgets", async (importOriginal) => ({ + ...(await importOriginal()), AI_BUDGETS: { coach: { maxTokens: 1500, temperature: 0.4 } }, })); diff --git a/src/app/api/insights/chat/__tests__/route-workout-scope.test.ts b/src/app/api/insights/chat/__tests__/route-workout-scope.test.ts index 28a98783e..64fad5815 100644 --- a/src/app/api/insights/chat/__tests__/route-workout-scope.test.ts +++ b/src/app/api/insights/chat/__tests__/route-workout-scope.test.ts @@ -157,7 +157,8 @@ vi.mock("@/lib/ai/consent-guard", () => ({ assertConsentForChain: vi.fn(async () => undefined), })); vi.mock("@/lib/ai/prompts/insight-generator", () => ({ PROMPT_VERSION: "x" })); -vi.mock("@/lib/ai/ai-budgets", () => ({ +vi.mock("@/lib/ai/ai-budgets", async (importOriginal) => ({ + ...(await importOriginal()), AI_BUDGETS: { coach: { maxTokens: 1500, temperature: 0.4 } }, })); diff --git a/src/app/insights/medications/page.tsx b/src/app/insights/medications/page.tsx index d017986c9..bd17dc53f 100644 --- a/src/app/insights/medications/page.tsx +++ b/src/app/insights/medications/page.tsx @@ -4,9 +4,12 @@ import { useQuery } from "@tanstack/react-query"; import Link from "next/link"; import { Loader2, Pill } from "lucide-react"; -import { useAiCapability } from "@/hooks/use-ai-capability"; +import { useAiCapability, useAiProviderState } from "@/hooks/use-ai-capability"; import { useAuth } from "@/hooks/use-auth"; -import { nextStatusPollInterval } from "@/hooks/use-insight-status"; +import { + nextStatusPollInterval, + statusPollMaxAttempts, +} from "@/hooks/use-insight-status"; import { pollAttemptCount } from "@/lib/queries/refetch-interval"; import { queryKeys } from "@/lib/query-keys"; import { useTranslations } from "@/lib/i18n/context"; @@ -102,6 +105,7 @@ export default function InsightsMedikamentePage() { // The compliance summary and the per-medication sentences are status notes: // read only while the `statusText` capability is available. const statusText = useAiCapability("statusText"); + const { responseTimeoutMs } = useAiProviderState(); const { data: comprehensive, @@ -137,14 +141,13 @@ export default function InsightsMedikamentePage() { enabled: isAuthenticated && statusText.available, staleTime: 60 * 1000, retry: 0, - // v1.8.4 — bounded preparing poll: stop after the shared attempt - // ceiling so a persistently failing generation can't poll an open - // page forever. + // Bounded poll whose ceiling follows the response timeout (#1274). refetchInterval: (query) => nextStatusPollInterval( query.state.data?.preparing, pollAttemptCount(query.state), query.state.data?.revalidating, + statusPollMaxAttempts(responseTimeoutMs), ), }); diff --git a/src/components/insights/__tests__/healthkit-metric-page-hrv-measures.test.tsx b/src/components/insights/__tests__/healthkit-metric-page-hrv-measures.test.tsx index 455a59224..37a3c96f0 100644 --- a/src/components/insights/__tests__/healthkit-metric-page-hrv-measures.test.tsx +++ b/src/components/insights/__tests__/healthkit-metric-page-hrv-measures.test.tsx @@ -19,6 +19,7 @@ vi.mock("@/hooks/use-ai-capability", () => ({ reason: null, onDeviceAllowed: true, }), + useAiProviderState: () => ({ responseTimeoutMs: 60_000 }), })); vi.mock("next/dynamic", () => ({ default: () => { diff --git a/src/components/labs/__tests__/lab-biomarker-detail.test.tsx b/src/components/labs/__tests__/lab-biomarker-detail.test.tsx index 86092e3cb..b673dba53 100644 --- a/src/components/labs/__tests__/lab-biomarker-detail.test.tsx +++ b/src/components/labs/__tests__/lab-biomarker-detail.test.tsx @@ -25,6 +25,7 @@ vi.mock("@/hooks/use-ai-capability", () => ({ reason: null, onDeviceAllowed: true, }), + useAiProviderState: () => ({ responseTimeoutMs: 60_000 }), })); vi.mock("@/lib/api/api-fetch", async (importOriginal) => ({ ApiError: (await importOriginal()) diff --git a/src/hooks/use-insight-status.ts b/src/hooks/use-insight-status.ts index 40c5b2096..9c0dea614 100644 --- a/src/hooks/use-insight-status.ts +++ b/src/hooks/use-insight-status.ts @@ -2,7 +2,7 @@ import { useQuery, useQueryClient } from "@tanstack/react-query"; -import { useAiCapability } from "@/hooks/use-ai-capability"; +import { useAiCapability, useAiProviderState } from "@/hooks/use-ai-capability"; import { useAuth } from "@/hooks/use-auth"; import { pollAttemptCount } from "@/lib/queries/refetch-interval"; import { useTranslations } from "@/lib/i18n/context"; @@ -10,6 +10,8 @@ import type { MetricStatusMetricId } from "@/lib/insights/metric-status-registry import type { AssessmentStatus } from "@/lib/insights/status-shared"; import { queryKeys } from "@/lib/query-keys"; import { apiGet } from "@/lib/api/api-fetch"; +import { STATUS_MODEL_CALLS } from "@/lib/ai/ai-budgets"; +import { clientAbortMs } from "@/lib/ai/effective-timeout"; /** * Shape returned by every `/api/insights/-status` endpoint. @@ -68,24 +70,51 @@ const STATUS_TIMEOUT_MS = 8_000; const STATUS_POLL_MS = 4_000; /** - * v1.8.4 — hard ceiling on the preparing poll. The route is read-only and - * a worker warms the cache out of band; a healthy generation lands within - * a couple of intervals. If a provider is configured but generation - * persistently fails, `preparing` stays true forever and the open page - * would otherwise poll indefinitely (battery / network waste). Cap the - * poll at ~12 attempts (≈ 48 s at the 4 s cadence) and then fall back to - * the static empty / "no analysis yet" state. `dataUpdateCount` counts - * every settled fetch — initial load plus each poll round — so once it - * crosses the cap we stop scheduling the next interval. + * v1.8.4 — the fast part of the preparing poll: ~12 attempts (≈ 48 s at the + * 4 s cadence), the window a healthy generation lands in. It is also the + * ceiling for a caller that does not know the response timeout. + * `dataUpdateCount` counts every settled fetch — initial load plus each poll + * round — so once it crosses the ceiling we stop scheduling the next one. */ export const STATUS_POLL_MAX_ATTEMPTS = 12; +/** + * The cadence once the fast part has passed: a slow model is still within + * its time, and a read every 15 s picks its note up without hammering the + * route for twenty minutes. + */ +export const STATUS_SLOW_POLL_MS = 15_000; + +/** + * #1274 — the attempt ceiling for a status card while its generation runs. + * + * The note is written by a background job whose run may wait the person's + * response timeout on each of `STATUS_MODEL_CALLS` calls (the call, and the + * one repeat a thinking model gets when it answered with reasoning alone). + * A fixed 12 attempts stopped watching after 48 s, so a slow local model + * delivered its note to a page that had stopped looking, and the card kept + * its fallback until a reload. The window is that run read off the + * published setting (`clientAbortMs`), covered by the fast polls and then + * the slow ones, plus the first read and one interval for the queue to pick + * the job up. It never falls below the fast ceiling, and it only matters + * while the server reports the generation in flight: a terminal payload + * stops the poll whatever the ceiling. + */ +export function statusPollMaxAttempts( + responseTimeoutMs: number | null | undefined, +): number { + const runMs = clientAbortMs(responseTimeoutMs, STATUS_MODEL_CALLS); + const fastMs = STATUS_POLL_MAX_ATTEMPTS * STATUS_POLL_MS; + const slow = Math.max(0, Math.ceil((runMs - fastMs) / STATUS_SLOW_POLL_MS)); + return STATUS_POLL_MAX_ATTEMPTS + slow + 2; +} + /** * Decide whether a card should schedule its next poll. Pure and shared - * between the `useInsightStatus` hook and the inline medication-compliance - * query so both sites enforce the identical ceiling. Returns the interval - * in ms while polling is warranted, or `false` once the payload is terminal - * OR the attempt cap is reached. + * between the `useInsightStatus` hooks and the inline medication-compliance + * query so every site enforces the same schedule. Returns the interval in ms + * while polling is warranted, or `false` once the payload is terminal OR the + * attempt ceiling (`statusPollMaxAttempts`) is reached. * * v1.9.0 — `revalidating` joins `preparing` as a "keep polling" signal: when * the route serves last-good (stale) text it would otherwise be terminal and @@ -97,10 +126,13 @@ export function nextStatusPollInterval( preparing: boolean | undefined, dataUpdateCount: number, revalidating?: boolean | undefined, + maxAttempts: number = STATUS_POLL_MAX_ATTEMPTS, ): number | false { if (!preparing && !revalidating) return false; - if (dataUpdateCount >= STATUS_POLL_MAX_ATTEMPTS) return false; - return STATUS_POLL_MS; + if (dataUpdateCount >= maxAttempts) return false; + return dataUpdateCount < STATUS_POLL_MAX_ATTEMPTS + ? STATUS_POLL_MS + : STATUS_SLOW_POLL_MS; } /** @@ -233,11 +265,12 @@ function useStatusPollAttemptCount(queryKey: readonly unknown[]): number { export function applyStatusPollExhaustion( data: InsightStatusData | undefined, dataUpdateCount: number, + maxAttempts: number = STATUS_POLL_MAX_ATTEMPTS, ): InsightStatusData | undefined { if (!data) return data; const status = data.assessment ?? assessmentStatusFromData(data); const pending = status.kind === "preparing" || status.kind === "revalidating"; - const pollExhausted = pending && dataUpdateCount >= STATUS_POLL_MAX_ATTEMPTS; + const pollExhausted = pending && dataUpdateCount >= maxAttempts; if (!pollExhausted) return normalizeStatusData(data); return { ...data, @@ -300,6 +333,7 @@ export function useInsightStatus(metric: InsightStatusMetric) { // capability is unavailable (the card is hidden then anyway). Unavailable // while `/me` loads, so the first paint fires nothing. const statusText = useAiCapability("statusText"); + const { responseTimeoutMs } = useAiProviderState(); const { locale } = useTranslations(); const queryKey = QUERY_KEY_FACTORY[metric](locale); @@ -326,12 +360,17 @@ export function useInsightStatus(metric: InsightStatusMetric) { query.state.data?.preparing, pollAttemptCount(query.state), query.state.data?.revalidating, + statusPollMaxAttempts(responseTimeoutMs), ), }); const pollAttempts = useStatusPollAttemptCount(queryKey); return { ...query, - data: applyStatusPollExhaustion(query.data, pollAttempts), + data: applyStatusPollExhaustion( + query.data, + pollAttempts, + statusPollMaxAttempts(responseTimeoutMs), + ), }; } @@ -365,6 +404,7 @@ export function useInsightMetricStatus( ) { const { isAuthenticated } = useAuth(); const statusText = useAiCapability("statusText"); + const { responseTimeoutMs } = useAiProviderState(); const { locale } = useTranslations(); const queryKey = queryKeys.insightsMetricStatus(metric, locale); @@ -385,12 +425,17 @@ export function useInsightMetricStatus( query.state.data?.preparing, pollAttemptCount(query.state), query.state.data?.revalidating, + statusPollMaxAttempts(responseTimeoutMs), ), }); const pollAttempts = useStatusPollAttemptCount(queryKey); return { ...query, - data: applyStatusPollExhaustion(query.data, pollAttempts), + data: applyStatusPollExhaustion( + query.data, + pollAttempts, + statusPollMaxAttempts(responseTimeoutMs), + ), }; } @@ -416,6 +461,7 @@ export function useInsightBiomarkerAssessment( ) { const { isAuthenticated } = useAuth(); const statusText = useAiCapability("statusText"); + const { responseTimeoutMs } = useAiProviderState(); const { locale } = useTranslations(); const queryKey = queryKeys.insightsBiomarkerAssessment(biomarkerId, locale); @@ -440,11 +486,16 @@ export function useInsightBiomarkerAssessment( query.state.data?.preparing, pollAttemptCount(query.state), query.state.data?.revalidating, + statusPollMaxAttempts(responseTimeoutMs), ), }); const pollAttempts = useStatusPollAttemptCount(queryKey); return { ...query, - data: applyStatusPollExhaustion(query.data, pollAttempts), + data: applyStatusPollExhaustion( + query.data, + pollAttempts, + statusPollMaxAttempts(responseTimeoutMs), + ), }; } diff --git a/src/lib/ai/ai-budgets.ts b/src/lib/ai/ai-budgets.ts index fbf586b68..d006f8394 100644 --- a/src/lib/ai/ai-budgets.ts +++ b/src/lib/ai/ai-budgets.ts @@ -265,6 +265,17 @@ export const AI_BUDGETS = { */ export const COMPREHENSIVE_MODEL_CALLS = 5; +/** + * The most model calls one status run (`runStatusCompletion`: a status card, + * the batched assessment, a derived score, a period narrative, a workout + * line, a Coach memory pass) makes one after another: the call itself, plus + * the one repeat the Local and gateway clients send when a thinking model + * answered with reasoning alone. Each has the person's response timeout to + * itself, so the run's outer bound, the jobs around it and the page that + * polls for its text all wait for this many (#1274). + */ +export const STATUS_MODEL_CALLS = 2; + /** 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/ai/coach/__tests__/coach-memory-shared.test.ts b/src/lib/ai/coach/__tests__/coach-memory-shared.test.ts index 09522ae95..3c251954c 100644 --- a/src/lib/ai/coach/__tests__/coach-memory-shared.test.ts +++ b/src/lib/ai/coach/__tests__/coach-memory-shared.test.ts @@ -14,6 +14,7 @@ vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() })); import { COACH_MEMORY_QUIET_MS, + COACH_MEMORY_REFRESH_EXPIRE_SECONDS, COACH_MEMORY_REFRESH_QUEUE, enqueueCoachMemoryRefresh, } from "../coach-memory-shared"; @@ -32,6 +33,7 @@ describe("enqueueCoachMemoryRefresh", () => { singletonKey: "quiet:c1", singletonSeconds: COACH_MEMORY_QUIET_MS / 1000, startAfter: COACH_MEMORY_QUIET_MS / 1000, + expireInSeconds: COACH_MEMORY_REFRESH_EXPIRE_SECONDS, }); expect(COACH_MEMORY_QUIET_MS).toBe(30 * 60_000); }); diff --git a/src/lib/ai/coach/coach-memory-shared.ts b/src/lib/ai/coach/coach-memory-shared.ts index 4dbcedfa6..bfbb80393 100644 --- a/src/lib/ai/coach/coach-memory-shared.ts +++ b/src/lib/ai/coach/coach-memory-shared.ts @@ -21,9 +21,29 @@ */ import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { annotate } from "@/lib/logging/context"; +import { + STATUS_RUN_CEILING_MS, + expireAboveModelTime, +} from "@/lib/jobs/model-time"; export const COACH_MEMORY_REFRESH_QUEUE = "coach-memory-refresh"; +/** + * The status runs one refresh makes in a row: the conversation summary, the + * fact extraction and the plan proposals (`runCoachMemoryRefresh`). + */ +export const COACH_MEMORY_MODEL_RUNS = 3; + +/** + * The job's expiry: every run of one refresh at the largest response timeout, + * plus headroom. Three runs of up to two ten-minute calls each outlived + * pg-boss's fifteen-minute default, which then failed the job and retried it + * beside the refresh still writing (#1274). + */ +export const COACH_MEMORY_REFRESH_EXPIRE_SECONDS = expireAboveModelTime( + COACH_MEMORY_MODEL_RUNS * STATUS_RUN_CEILING_MS, +); + /** How long a conversation must be quiet before its memory is refreshed. */ export const COACH_MEMORY_QUIET_MS = 30 * 60_000; @@ -64,6 +84,7 @@ export async function enqueueCoachMemoryRefresh( singletonKey: `quiet:${payload.conversationId}`, singletonSeconds: quietSeconds, startAfter: Math.max(1, Math.round(delayMs / 1000)), + expireInSeconds: COACH_MEMORY_REFRESH_EXPIRE_SECONDS, }, ); annotate({ diff --git a/src/lib/ai/effective-timeout.ts b/src/lib/ai/effective-timeout.ts index 634f0c0ff..93c0b491b 100644 --- a/src/lib/ai/effective-timeout.ts +++ b/src/lib/ai/effective-timeout.ts @@ -6,6 +6,14 @@ import type { AIProvider, CompletionParams } from "./types"; */ export const PROVIDER_DEFAULT_TIMEOUT_MS = 60_000; +/** + * The range `User.aiResponseTimeoutSeconds` accepts at write time (Settings → + * AI). The ceiling is what every outer bound and job expiry around model work + * sizes its worst case from: a person can make each call wait this long. + */ +export const AI_RESPONSE_TIMEOUT_MIN_SECONDS = 10; +export const AI_RESPONSE_TIMEOUT_MAX_SECONDS = 600; + /** * Resolve the upstream provider timeout (ms) for a non-chat generation surface. * diff --git a/src/lib/insights/__tests__/status-provider-slow-model.test.ts b/src/lib/insights/__tests__/status-provider-slow-model.test.ts new file mode 100644 index 000000000..e5bce7962 --- /dev/null +++ b/src/lib/insights/__tests__/status-provider-slow-model.test.ts @@ -0,0 +1,173 @@ +/** + * #1274 — a status run waits for every call it may make. + * + * A status run is one model call plus, on the Local and gateway clients, one + * repeat when a thinking model answered with reasoning alone. Each has the + * person's response timeout to itself. The run's outer bound used to be one + * call, so a model that needs 300 s per answer under a 300 s setting, whose + * first answer was reasoning only, was cut off in the middle of the repeat + * that would have answered, and the card kept its fallback text. + * + * The provider runner is replaced by a clock-driven stand-in that answers + * after the time the two calls take; everything else in the chokepoint runs. + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +const { settings } = vi.hoisted(() => ({ + settings: { aiResponseTimeoutSeconds: 300 as number | null }, +})); + +vi.mock("@/lib/db", () => ({ + prisma: { + $queryRaw: vi.fn(async () => [{ total_tokens: 0, operator_tokens: 0 }]), + $executeRaw: vi.fn(async () => 1), + user: { + findUnique: vi.fn(async () => ({ + aiResponseTimeoutSeconds: settings.aiResponseTimeoutSeconds, + })), + }, + }, +})); + +vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() })); + +vi.mock("@/lib/ai/provider", () => ({ + resolveProviderChain: vi.fn(async () => [ + { providerType: "local", instance: {} }, + ]), + resolveProvider: vi.fn(async () => ({ type: "none" })), + probeProviderPresence: vi.fn(async () => true), +})); + +vi.mock("@/lib/ai/capabilities/gate", () => ({ + aiCapabilityForRecord: vi.fn(async () => ({ + available: true, + reason: null, + onDeviceAllowed: true, + })), +})); + +vi.mock("@/lib/ai/capabilities/egress", () => ({ + aiEgressRefusal: vi.fn(async () => null), +})); + +const { runRawCompletionWithFallback } = vi.hoisted(() => ({ + runRawCompletionWithFallback: vi.fn(), +})); +vi.mock("@/lib/ai/provider-runner", () => ({ + AllProvidersFailedError: class extends Error {}, + runRawCompletionWithFallback, +})); + +import { runStatusCompletion, statusRunBoundMs } from "../status-provider"; +import { STATUS_PROVIDER_TIMEOUT_MS } from "../with-timeout"; +import { STATUS_MODEL_CALLS } from "@/lib/ai/ai-budgets"; +import { resolveEffectiveTimeoutMs } from "@/lib/ai/effective-timeout"; +import { + nextStatusPollInterval, + statusPollMaxAttempts, +} from "@/hooks/use-insight-status"; + +/** A model that answers `afterMs` after the run starts. */ +function answersAfter(afterMs: number) { + runRawCompletionWithFallback.mockImplementation( + () => + new Promise((resolve) => + setTimeout( + () => + resolve({ + result: { + content: '{"summary":"Steady."}', + model: "slow-local", + tokensUsed: 80, + }, + workingProvider: { providerType: "local" }, + }), + afterMs, + ), + ), + ); +} + +function run() { + return runStatusCompletion({ + userId: "u1", + cacheAction: "status:test", + systemPrompt: "system", + userPrompt: "user", + capability: "statusText", + }); +} + +beforeEach(() => { + vi.useFakeTimers(); + runRawCompletionWithFallback.mockReset(); + settings.aiResponseTimeoutSeconds = 300; +}); + +afterEach(() => { + vi.useRealTimers(); +}); + +describe("runStatusCompletion — the outer bound allows every call of a run", () => { + it("lands a note whose reasoning-only first answer was repeated", async () => { + // First call: the full 300 s, reasoning only. The repeat answers at 150 s. + answersAfter(450_000); + const pending = run(); + await vi.advanceTimersByTimeAsync(450_000); + await expect(pending).resolves.toMatchObject({ + kind: "ok", + content: '{"summary":"Steady."}', + }); + }); + + it("still gives up on a run that outlasts both calls", async () => { + answersAfter(700_000); + const pending = run(); + await vi.advanceTimersByTimeAsync(600_000); + await expect(pending).resolves.toEqual({ kind: "timeout" }); + }); + + it("sizes the bound from the default when no setting is stored", () => { + expect(statusRunBoundMs(60_000)).toBe(60_000 * STATUS_MODEL_CALLS); + expect(STATUS_MODEL_CALLS).toBe(2); + }); +}); + +describe("the open card polls as long as the run may take", () => { + it("outlasts the run's own bound for every setting", () => { + // The page reads the published `responseTimeoutMs`: the setting in ms, + // or the 60 s provider default when nothing is set. + const cases: Array<[number | null, number]> = [ + [null, 60_000], + [10, 10_000], + [60, 60_000], + [180, 180_000], + [300, 300_000], + [600, 600_000], + ]; + for (const [seconds, published] of cases) { + const attempts = statusPollMaxAttempts(published); + let count = 1; // the read that enqueued the generation + let watchedMs = 0; + let interval = nextStatusPollInterval(true, count, false, attempts); + while (interval !== false) { + watchedMs += interval; + count += 1; + interval = nextStatusPollInterval(true, count, false, attempts); + } + const runBound = statusRunBoundMs( + resolveEffectiveTimeoutMs(seconds, STATUS_PROVIDER_TIMEOUT_MS), + ); + expect(watchedMs, `setting ${seconds ?? "unset"}`).toBeGreaterThan( + runBound, + ); + } + }); + + it("stays bounded at the largest setting", () => { + // Twenty minutes of two model calls at 600 s, read every 15 s after the + // first 48 s: about a hundred reads, not three hundred. + expect(statusPollMaxAttempts(600_000)).toBeLessThanOrEqual(100); + }); +}); diff --git a/src/lib/insights/status-provider.ts b/src/lib/insights/status-provider.ts index 954688f65..5dd56a26c 100644 --- a/src/lib/insights/status-provider.ts +++ b/src/lib/insights/status-provider.ts @@ -21,7 +21,11 @@ import { resolveCostOwner, resolveDailyCapFor, } from "@/lib/ai/coach/budget"; -import { AI_BUDGETS, REFERENCE_AI_SEED } from "@/lib/ai/ai-budgets"; +import { + AI_BUDGETS, + REFERENCE_AI_SEED, + STATUS_MODEL_CALLS, +} from "@/lib/ai/ai-budgets"; import { singleUserTurn } from "@/lib/ai/types"; import { STATUS_PROVIDER_TIMEOUT_MS, withTimeout } from "./with-timeout"; import { prisma } from "@/lib/db"; @@ -40,6 +44,19 @@ import { resolveJobReasoning } from "@/lib/ai/reasoning/controls"; */ export const REASONING_JOB_TIMEOUT_MS = 90_000; +/** + * The outer bound of one status run: the per-call timeout for each of the + * `STATUS_MODEL_CALLS` calls the run may make in a row. Bounded at one call, + * a thinking model whose first answer was reasoning alone timed out in the + * middle of the repeat that would have answered, and a slow local model the + * person had given 300 s or more per call never landed its note (#1274). + * Every caller is a background job; the job expiries in `queue-runtime.ts` + * sit above this at the largest setting. + */ +export function statusRunBoundMs(effectiveTimeoutMs: number): number { + return effectiveTimeoutMs * STATUS_MODEL_CALLS; +} + /** * Shared provider plumbing for the seven `*-status.ts` generators. * @@ -53,10 +70,9 @@ export const REASONING_JOB_TIMEOUT_MS = 90_000; * provider chain, fall back to the legacy single provider when the chain * is empty, and run `runRawCompletionWithFallback` so a degraded provider * cascades to the next. The whole thing is still wrapped in - * `withTimeout` at the aligned 60 s budget so a total stall can't pin the - * card — but a timeout / error is reported as a transient miss (the - * caller serves the fallback for this render without persisting it), - * never as the day's cached assessment. + * `withTimeout` (`statusRunBoundMs`) so a total stall cannot hold the job — + * but a timeout / error is reported as a transient miss (the caller serves + * the fallback without persisting it), never as the day's cached assessment. */ export type StatusProviderResult = @@ -147,7 +163,7 @@ export async function hasUsableStatusProvider( /** * Run a status generation across the user's provider chain, bounded by the * per-user response-timeout setting (falling back to `STATUS_PROVIDER_TIMEOUT_MS` - * when unset). The result discriminates between + * when unset) for each call the run may make (`statusRunBoundMs`). The result discriminates between * no-provider / timeout / provider-error / success so the caller can * decide what to persist — only `ok` is ever cached as the day's * assessment. @@ -205,9 +221,9 @@ export async function runStatusCompletion( // cards, the batched assessment, the derived assessments, and the period // narratives — so resolving it here threads the setting onto all of them at // once. A positive stored value wins (seconds → ms); unset falls back to the - // status-path budget. Applied to BOTH the upstream call's own `timeoutMs` - // and the outer `withTimeout` cap so a raised value is not silently clipped - // by the 60 s wall-clock that previously bounded the path. + // status-path budget. Applied to the upstream call's own `timeoutMs` and, + // times the calls a run may make, to the outer `withTimeout` cap, so a + // raised value is not silently clipped by a wall-clock sized for less. const settingsRow = await prisma.user.findUnique({ where: { id: userId }, select: { aiResponseTimeoutSeconds: true }, @@ -320,7 +336,7 @@ export async function runStatusCompletion( ...(reasoning ? { reasoning } : {}), }, }), - effectiveTimeoutMs, + statusRunBoundMs(effectiveTimeoutMs), null, ); diff --git a/src/lib/jobs/__tests__/morning-digest-refresh.test.ts b/src/lib/jobs/__tests__/morning-digest-refresh.test.ts index 050b97d6c..64dd37bf6 100644 --- a/src/lib/jobs/__tests__/morning-digest-refresh.test.ts +++ b/src/lib/jobs/__tests__/morning-digest-refresh.test.ts @@ -1,6 +1,14 @@ -import { describe, it, expect, vi, beforeEach } from "vitest"; +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; import { runMorningDigestRefresh } from "../morning-digest-refresh"; +import { MORNING_DIGEST_REFRESH_EXPIRE_SECONDS } from "../morning-digest-refresh-shared"; +import { + COMPREHENSIVE_WARM_CEILING_MS, + comprehensiveWarmBudgetMs, +} from "../model-time"; +import { QUEUE_RUNTIME } from "../queue-runtime"; +import { COMPREHENSIVE_MODEL_CALLS } from "@/lib/ai/ai-budgets"; +import { AI_RESPONSE_TIMEOUT_MAX_SECONDS } from "@/lib/ai/effective-timeout"; import type { GenerateOutcome } from "@/lib/insights/comprehensive-generate"; const LOCAL_DATE = "2026-07-16"; @@ -9,6 +17,7 @@ function makePrisma( user: { locale: string | null; morningDigestRefreshedOn: string | null; + aiResponseTimeoutSeconds?: number | null; } | null, ) { const updateMock = vi.fn(async () => ({})); @@ -21,6 +30,19 @@ function makePrisma( return { prisma, updateMock }; } +/** The handler over the stand-in client. */ +function refresh( + prisma: ReturnType["prisma"], + ...rest: Parameters extends [ + unknown, + ...infer R, + ] + ? R + : never +) { + return runMorningDigestRefresh(prisma as never, ...rest); +} + describe("runMorningDigestRefresh", () => { const retryMock = vi.fn(async () => {}); @@ -38,13 +60,17 @@ describe("runMorningDigestRefresh", () => { providerType: "mock", })); - const result = await runMorningDigestRefresh( - prisma as never, + const result = await refresh( + prisma, { userId: "u1", localDate: LOCAL_DATE }, { generate, enqueueRetry: retryMock }, ); - expect(generate).toHaveBeenCalledWith("u1", { locale: "en", force: true }); + expect(generate).toHaveBeenCalledWith("u1", { + locale: "en", + force: true, + signal: expect.any(AbortSignal), + }); expect(updateMock).toHaveBeenCalledWith({ where: { id: "u1" }, data: { morningDigestRefreshedOn: LOCAL_DATE }, @@ -62,13 +88,17 @@ describe("runMorningDigestRefresh", () => { status: "unchanged", })); - const result = await runMorningDigestRefresh( - prisma as never, + const result = await refresh( + prisma, { userId: "u1", localDate: LOCAL_DATE }, { generate, enqueueRetry: retryMock }, ); - expect(generate).toHaveBeenCalledWith("u1", { locale: "de", force: true }); + expect(generate).toHaveBeenCalledWith("u1", { + locale: "de", + force: true, + signal: expect.any(AbortSignal), + }); expect(updateMock).toHaveBeenCalled(); expect(result).toEqual({ status: "finalised", comprehensive: "unchanged" }); }); @@ -83,8 +113,8 @@ describe("runMorningDigestRefresh", () => { reason: "no-provider", })); - const result = await runMorningDigestRefresh( - prisma as never, + const result = await refresh( + prisma, { userId: "u1", localDate: LOCAL_DATE }, { generate, enqueueRetry: retryMock }, ); @@ -103,8 +133,8 @@ describe("runMorningDigestRefresh", () => { providerType: "mock", })); - const result = await runMorningDigestRefresh( - prisma as never, + const result = await refresh( + prisma, { userId: "u1", localDate: LOCAL_DATE }, { generate, enqueueRetry: retryMock }, ); @@ -124,8 +154,8 @@ describe("runMorningDigestRefresh", () => { reason: "provider-timeout", })); - const result = await runMorningDigestRefresh( - prisma as never, + const result = await refresh( + prisma, { userId: "u1", localDate: LOCAL_DATE }, { generate, enqueueRetry: retryMock }, ); @@ -141,8 +171,8 @@ describe("runMorningDigestRefresh", () => { status: "unchanged", })); - const result = await runMorningDigestRefresh( - prisma as never, + const result = await refresh( + prisma, { userId: "ghost", localDate: LOCAL_DATE }, { generate, enqueueRetry: retryMock }, ); @@ -152,3 +182,109 @@ describe("runMorningDigestRefresh", () => { expect(result.status).toBe("missing-user"); }); }); + +describe("runMorningDigestRefresh — a slow model under a large setting (#1274)", () => { + const retryMock = vi.fn(async () => {}); + + beforeEach(() => { + vi.useFakeTimers(); + retryMock.mockReset(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("waits for every call of the generation, not one", async () => { + // Five calls in a row at 290 s each under a 300 s setting. + const { prisma, updateMock } = makePrisma({ + locale: "en", + morningDigestRefreshedOn: null, + aiResponseTimeoutSeconds: 300, + }); + const generate = vi.fn( + () => + new Promise((resolve) => + setTimeout( + () => resolve({ status: "generated", providerType: "local" }), + COMPREHENSIVE_MODEL_CALLS * 290_000, + ), + ), + ); + const pending = refresh( + prisma, + { userId: "u1", localDate: LOCAL_DATE }, + { generate, enqueueRetry: retryMock }, + ); + await vi.advanceTimersByTimeAsync(COMPREHENSIVE_MODEL_CALLS * 290_000); + await expect(pending).resolves.toEqual({ + status: "finalised", + comprehensive: "generated", + }); + expect(updateMock).toHaveBeenCalled(); + }); + + it("aborts a wedged generation at the bound and leaves the day provisional", async () => { + const { prisma, updateMock } = makePrisma({ + locale: "en", + morningDigestRefreshedOn: null, + aiResponseTimeoutSeconds: 600, + }); + let signal: AbortSignal | undefined; + const generate = vi.fn( + (_userId: string, options: { signal?: AbortSignal }) => { + signal = options.signal; + return new Promise(() => {}); + }, + ); + const pending = refresh( + prisma, + { userId: "u1", localDate: LOCAL_DATE }, + { generate, enqueueRetry: retryMock }, + ); + await vi.advanceTimersByTimeAsync(comprehensiveWarmBudgetMs(600)); + await expect(pending).resolves.toEqual({ + status: "failed", + comprehensive: "failed", + }); + expect(signal?.aborted).toBe(true); + expect(updateMock).not.toHaveBeenCalled(); + expect(retryMock).toHaveBeenCalledWith({ userId: "u1", locale: "en" }); + }); + + it("rethrows a generator fault so the queue retries it", async () => { + const { prisma } = makePrisma({ + locale: "en", + morningDigestRefreshedOn: null, + }); + const generate = vi.fn(async (): Promise => { + throw new Error("pool exhausted"); + }); + await expect( + refresh( + prisma, + { userId: "u1", localDate: LOCAL_DATE }, + { generate, enqueueRetry: retryMock }, + ), + ).rejects.toThrow("pool exhausted"); + expect(retryMock).not.toHaveBeenCalled(); + }); + + it("the job's expiry sits above the longest generation the largest setting allows", () => { + // pg-boss fails and retries a job past its expiry while the handler keeps + // going; the default fifteen minutes is below five ten-minute calls. + expect(COMPREHENSIVE_WARM_CEILING_MS).toBeGreaterThanOrEqual( + AI_RESPONSE_TIMEOUT_MAX_SECONDS * 1000 * COMPREHENSIVE_MODEL_CALLS, + ); + expect(comprehensiveWarmBudgetMs(AI_RESPONSE_TIMEOUT_MAX_SECONDS)).toBe( + COMPREHENSIVE_WARM_CEILING_MS, + ); + expect(MORNING_DIGEST_REFRESH_EXPIRE_SECONDS * 1000).toBeGreaterThan( + COMPREHENSIVE_WARM_CEILING_MS, + ); + expect(QUEUE_RUNTIME["morning-digest-refresh"]).toMatchObject({ + runtime: "bounded", + expireInSeconds: MORNING_DIGEST_REFRESH_EXPIRE_SECONDS, + }); + }); +}); diff --git a/src/lib/jobs/insight-pregenerate.ts b/src/lib/jobs/insight-pregenerate.ts index 8b0e36674..f67c63230 100644 --- a/src/lib/jobs/insight-pregenerate.ts +++ b/src/lib/jobs/insight-pregenerate.ts @@ -79,8 +79,7 @@ import { type MetricStatusMetricId, } from "@/lib/insights/metric-status-registry"; import { withTimeout } from "@/lib/insights/with-timeout"; -import { AI_BUDGETS, COMPREHENSIVE_MODEL_CALLS } from "@/lib/ai/ai-budgets"; -import { resolveEffectiveTimeoutMs } from "@/lib/ai/effective-timeout"; +import { comprehensiveWarmBudgetMs } from "@/lib/jobs/model-time"; import { INSIGHT_PREGENERATE_QUEUE, enqueueForceWarm, @@ -152,57 +151,16 @@ const FORCE_WARM_DAILY_WINDOW_MS = 24 * 60 * 60 * 1000; 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. A slow or + * Bounded budget for one comprehensive generation inside a warm pass — 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. - * - * 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. + * provider call carries its own abort, so `comprehensiveWarmBudgetMs` + * (`model-time.ts`) sits above the longest generation those calls allow. */ -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 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: the longest generation the largest setting allows. */ -const COMPREHENSIVE_WARM_CEILING_MS = - AI_RESPONSE_TIMEOUT_MAX_MS * COMPREHENSIVE_MODEL_CALLS + - COMPREHENSIVE_WARM_HEADROOM_MS; - -/** - * 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, -): number { - const effectiveTimeoutMs = resolveEffectiveTimeoutMs( - aiResponseTimeoutSeconds, - AI_BUDGETS.comprehensive.timeoutMs, - ); - return Math.min( - effectiveTimeoutMs * COMPREHENSIVE_MODEL_CALLS + - COMPREHENSIVE_WARM_HEADROOM_MS, - COMPREHENSIVE_WARM_CEILING_MS, - ); -} +export { comprehensiveWarmBudgetMs }; /** * Short, redaction-safe error description for the failure annotations diff --git a/src/lib/jobs/insight-status-generate-queue.ts b/src/lib/jobs/insight-status-generate-queue.ts index bee53259d..1b1aa5455 100644 --- a/src/lib/jobs/insight-status-generate-queue.ts +++ b/src/lib/jobs/insight-status-generate-queue.ts @@ -3,8 +3,23 @@ * schedule, concurrency and send options. They live apart from the job so the * worker registry and every sender can read them without loading it. */ +import { + STATUS_RUN_CEILING_MS, + expireAboveModelTime, +} from "@/lib/jobs/model-time"; + export const INSIGHT_STATUS_GENERATE_QUEUE = "insight-status-generate"; +/** + * The job's expiry: one status run at the largest response timeout (the call + * and its one reasoning repeat), plus headroom. Two ten-minute calls outlive + * pg-boss's fifteen-minute default, which then failed the job and queued a + * retry beside the generation still running (#1274). + */ +export const INSIGHT_STATUS_GENERATE_EXPIRE_SECONDS = expireAboveModelTime( + STATUS_RUN_CEILING_MS, +); + /** * Low concurrency keeps the on-demand warm pass from saturating the * Prisma pool or fanning out an unbounded number of concurrent provider diff --git a/src/lib/jobs/insight-status-generate-shared.ts b/src/lib/jobs/insight-status-generate-shared.ts index e448872d9..771f44649 100644 --- a/src/lib/jobs/insight-status-generate-shared.ts +++ b/src/lib/jobs/insight-status-generate-shared.ts @@ -21,7 +21,10 @@ import { type ProviderWorkAuthority, } from "@/lib/sharing/provider-work-authority"; import { logCaught } from "@/lib/logging/signal"; -import { INSIGHT_STATUS_GENERATE_QUEUE } from "./insight-status-generate-queue"; +import { + INSIGHT_STATUS_GENERATE_EXPIRE_SECONDS, + INSIGHT_STATUS_GENERATE_QUEUE, +} from "./insight-status-generate-queue"; /** * The seven per-metric status scopes the queue can generate. Mirrors the @@ -152,6 +155,7 @@ export async function enqueueStatusGeneration( retryLimit: 3, retryDelay: 60, retryBackoff: true, + expireInSeconds: INSIGHT_STATUS_GENERATE_EXPIRE_SECONDS, }, ); annotate({ diff --git a/src/lib/jobs/model-time.ts b/src/lib/jobs/model-time.ts new file mode 100644 index 000000000..b71a8fa43 --- /dev/null +++ b/src/lib/jobs/model-time.ts @@ -0,0 +1,77 @@ +/** + * How long a job's model work can take at most. + * + * Every model call has the record owner's response timeout to itself + * (`User.aiResponseTimeoutSeconds`, up to `AI_RESPONSE_TIMEOUT_MAX_SECONDS`), + * and the work a job does is a known number of calls in a row behind an outer + * bound. pg-boss fails and retries a job that runs past its expiry, and it + * does not stop the handler, so an expiry below that work lets a retry run + * beside the first attempt (#1274). The bounds here are what the outer aborts + * use and what the expiries in `queue-runtime.ts` are derived from, so both + * move together when a call count or the setting's ceiling does. + * + * No server import: the queue modules that carry an expiry into a send read + * these without loading a generator. + */ +import { + AI_BUDGETS, + COMPREHENSIVE_MODEL_CALLS, + STATUS_MODEL_CALLS, +} from "@/lib/ai/ai-budgets"; +import { + AI_RESPONSE_TIMEOUT_MAX_SECONDS, + resolveEffectiveTimeoutMs, +} from "@/lib/ai/effective-timeout"; + +const AI_RESPONSE_TIMEOUT_MAX_MS = AI_RESPONSE_TIMEOUT_MAX_SECONDS * 1000; + +/** + * Room inside a comprehensive warm for what runs around its calls: feature + * extraction, prompt assembly, the cache write. + */ +const COMPREHENSIVE_WARM_HEADROOM_MS = 30_000; + +/** The longest comprehensive generation the largest setting allows. */ +export const COMPREHENSIVE_WARM_CEILING_MS = + AI_RESPONSE_TIMEOUT_MAX_MS * COMPREHENSIVE_MODEL_CALLS + + COMPREHENSIVE_WARM_HEADROOM_MS; + +/** + * Outer-abort budget (ms) for one comprehensive generation run out of band + * (the nightly warm, the forced warm, the morning refresh): the account's + * per-call timeout times the calls one generation may make, plus headroom, + * clamped to the ceiling. 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). + */ +export function comprehensiveWarmBudgetMs( + aiResponseTimeoutSeconds: number | null | undefined, +): number { + const effectiveTimeoutMs = resolveEffectiveTimeoutMs( + aiResponseTimeoutSeconds, + AI_BUDGETS.comprehensive.timeoutMs, + ); + return Math.min( + effectiveTimeoutMs * COMPREHENSIVE_MODEL_CALLS + + COMPREHENSIVE_WARM_HEADROOM_MS, + COMPREHENSIVE_WARM_CEILING_MS, + ); +} + +/** + * The longest one status run (`runStatusCompletion`) waits at the largest + * setting: each of its `STATUS_MODEL_CALLS` calls at the full timeout. + */ +export const STATUS_RUN_CEILING_MS = + AI_RESPONSE_TIMEOUT_MAX_MS * STATUS_MODEL_CALLS; + +/** + * Room a job's expiry leaves on top of its model work for what the handler + * does around it: reading the record, building the prompt, persisting. + */ +export const JOB_MODEL_HEADROOM_SECONDS = 120; + +/** A job expiry (seconds) that sits above `modelMs` of model work. */ +export function expireAboveModelTime(modelMs: number): number { + return Math.ceil(modelMs / 1000) + JOB_MODEL_HEADROOM_SECONDS; +} diff --git a/src/lib/jobs/morning-digest-refresh-shared.ts b/src/lib/jobs/morning-digest-refresh-shared.ts index 53a79c878..8e0f763b3 100644 --- a/src/lib/jobs/morning-digest-refresh-shared.ts +++ b/src/lib/jobs/morning-digest-refresh-shared.ts @@ -16,9 +16,26 @@ import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { annotate } from "@/lib/logging/context"; import { logCaught } from "@/lib/logging/signal"; +import { + COMPREHENSIVE_WARM_CEILING_MS, + expireAboveModelTime, +} from "@/lib/jobs/model-time"; export const MORNING_DIGEST_REFRESH_QUEUE = "morning-digest-refresh"; +/** + * The job's expiry: the longest comprehensive generation the largest response + * timeout allows, plus headroom. The refresh is one forced generation of up + * to `COMPREHENSIVE_MODEL_CALLS` calls in a row, each with the person's + * timeout (up to ten minutes), which outlived pg-boss's fifteen-minute + * default: pg-boss failed the job and queued a retry while the handler was + * still generating (#1274). The handler bounds the generation by the same + * worst case (`comprehensiveWarmBudgetMs`), so the job cannot outlive this. + */ +export const MORNING_DIGEST_REFRESH_EXPIRE_SECONDS = expireAboveModelTime( + COMPREHENSIVE_WARM_CEILING_MS, +); + export interface MorningDigestRefreshPayload { userId: string; /** @@ -72,6 +89,7 @@ export async function enqueueMorningDigestRefresh( retryLimit: 2, retryDelay: 60, retryBackoff: true, + expireInSeconds: MORNING_DIGEST_REFRESH_EXPIRE_SECONDS, }); annotate({ action: { name: "daily.morning_refresh.enqueued" }, diff --git a/src/lib/jobs/morning-digest-refresh.ts b/src/lib/jobs/morning-digest-refresh.ts index 9f2d60165..1baffccb5 100644 --- a/src/lib/jobs/morning-digest-refresh.ts +++ b/src/lib/jobs/morning-digest-refresh.ts @@ -35,6 +35,8 @@ import { type GenerateOutcome, } from "@/lib/insights/comprehensive-generate"; import { enqueuePregenerateFailureRetry } from "@/lib/jobs/insight-pregenerate-shared"; +import { comprehensiveWarmBudgetMs } from "@/lib/jobs/model-time"; +import { withTimeout } from "@/lib/insights/with-timeout"; import type { SupportedLocale } from "@/lib/insights/status-shared"; import { resolveJobLocale } from "@/lib/i18n/job-locale"; import { @@ -53,7 +55,7 @@ export interface MorningDigestRefreshResult { type GenerateFn = ( userId: string, - options: { locale: SupportedLocale; force?: boolean }, + options: { locale: SupportedLocale; force?: boolean; signal?: AbortSignal }, ) => Promise; /** @@ -81,7 +83,11 @@ export async function runMorningDigestRefresh( const user = await prisma.user.findUnique({ where: { id: userId }, - select: { locale: true, morningDigestRefreshedOn: true }, + select: { + locale: true, + morningDigestRefreshedOn: true, + aiResponseTimeoutSeconds: true, + }, }); if (!user) { annotate({ @@ -101,20 +107,45 @@ export async function runMorningDigestRefresh( } const locale = await resolveJobLocale(user.locale); - const outcome = await generate(userId, { locale, force: true }); + // The same bound the nightly and forced warms use: every call of one + // generation at the person's timeout, plus headroom. It is the worst case + // the job's expiry (`MORNING_DIGEST_REFRESH_EXPIRE_SECONDS`) sits above, so + // pg-boss never declares a working refresh dead and retries it beside + // itself (#1274); a wedged generation is aborted here instead. + const controller = new AbortController(); + let thrown: unknown = null; + const bounded = await withTimeout( + () => + generate(userId, { + locale, + force: true, + signal: controller.signal, + }).catch((err: unknown) => { + thrown = err; + throw err; + }), + comprehensiveWarmBudgetMs(user.aiResponseTimeoutSeconds), + null, + () => controller.abort(), + ); + // A throw is transient (a pool or provider fault): the queue's own retry + // re-runs the refresh, as it did before the bound. + if (bounded.errored) throw thrown; + const outcome = bounded.value; // A hard provider failure leaves the day provisional and hands recovery to // the existing 45-minute briefing-retry machinery. Every other outcome — // including `skipped` (no provider / no consent: there is no LLM briefing to // wait for, so the deterministic score + digest are as final as they get // once last night's sleep is in the record) — finalises the day. - if (outcome.status === "failed") { + if (outcome === null || outcome.status === "failed") { + const cause = outcome !== null ? `generator:${outcome.reason}` : "timeout"; annotate({ action: { name: "daily.morning_refresh.deferred" }, - meta: { local_date: localDate, cause: `generator:${outcome.reason}` }, + meta: { local_date: localDate, cause }, }); await enqueueRetry({ userId, locale }); - return { status: "failed", comprehensive: outcome.status }; + return { status: "failed", comprehensive: "failed" }; } await prisma.user.update({ diff --git a/src/lib/jobs/queue-runtime.ts b/src/lib/jobs/queue-runtime.ts index 9069b2b45..230cfbc87 100644 --- a/src/lib/jobs/queue-runtime.ts +++ b/src/lib/jobs/queue-runtime.ts @@ -14,6 +14,13 @@ * - `budgeted`: a pass over a long history that stops itself at three * quarters of the default expiry (`jobBudget`) and hands the rest to its * next run, so it ends before fifteen minutes by construction. + * - `bounded`: one unit of model work (one account's note, refresh or + * memory pass) whose calls wait the person's response timeout each, up + * to ten minutes, which can outlast fifteen minutes. Every call sits + * behind an outer bound, so the job's longest run is known + * (`modelSeconds`, from `model-time.ts`); it carries an explicit + * `expireInSeconds` above that on every send, and no retry can start + * beside a run that is still going (#1274). * - `long`: a job that may run for hours. It carries an explicit * `expireInSeconds` on every send and schedule, stops itself before that * expiry (`jobBudget` / `jobDeadline`), and holds something per identity @@ -43,6 +50,17 @@ import { HEALTH_CONNECT_IMPORT_SEND_OPTIONS } from "@/lib/jobs/health-connect-im import { MEASUREMENT_MAINTENANCE_SEND_OPTIONS } from "@/lib/jobs/measurement-maintenance"; import { PULSE_DAY_MEAN_SEND_OPTIONS } from "@/lib/jobs/pulse-day-mean"; import { ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS } from "@/lib/rollups/measurement-rollups"; +import { MORNING_DIGEST_REFRESH_EXPIRE_SECONDS } from "@/lib/jobs/morning-digest-refresh-shared"; +import { INSIGHT_STATUS_GENERATE_EXPIRE_SECONDS } from "@/lib/jobs/insight-status-generate-queue"; +import { WORKOUT_INSIGHT_GENERATE_EXPIRE_SECONDS } from "@/lib/jobs/workout-insight-generate-shared"; +import { + COACH_MEMORY_MODEL_RUNS, + COACH_MEMORY_REFRESH_EXPIRE_SECONDS, +} from "@/lib/ai/coach/coach-memory-shared"; +import { + COMPREHENSIVE_WARM_CEILING_MS, + STATUS_RUN_CEILING_MS, +} from "@/lib/jobs/model-time"; /** pg-boss's default job expiry, in seconds. */ export const DEFAULT_JOB_EXPIRE_SECONDS = 15 * 60; @@ -53,6 +71,18 @@ export type SourceSite = { file: string; fn: string }; export type QueueRuntime = | { runtime: "short"; why: string } | { runtime: "budgeted"; budget: SourceSite; why: string } + | { + runtime: "bounded"; + /** + * The longest the job's model work runs at the largest response-timeout + * setting, in seconds: the outer bounds in the code, not an estimate. + */ + modelSeconds: number; + expireInSeconds: number; + /** As for `long`: identifiers that carry the expiry into a send. */ + expiryVia: readonly string[]; + why: string; + } | { runtime: "long"; expireInSeconds: number; @@ -156,15 +186,29 @@ export const QUEUE_RUNTIME: Readonly> = { exclusive: "lockedPass", 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.", - ), - "insight-status-generate": short( - "One account, one metric, one provider call.", - ), + "morning-digest-refresh": { + runtime: "bounded", + modelSeconds: COMPREHENSIVE_WARM_CEILING_MS / 1000, + expireInSeconds: MORNING_DIGEST_REFRESH_EXPIRE_SECONDS, + expiryVia: ["MORNING_DIGEST_REFRESH_EXPIRE_SECONDS"], + why: "One account, one forced comprehensive generation: up to five calls in a row at the account's timeout, behind the warm's own bound.", + }, + "insight-status-generate": { + runtime: "bounded", + modelSeconds: STATUS_RUN_CEILING_MS / 1000, + expireInSeconds: INSIGHT_STATUS_GENERATE_EXPIRE_SECONDS, + expiryVia: ["INSIGHT_STATUS_GENERATE_EXPIRE_SECONDS"], + why: "One account, one note: one status run, the call and its one reasoning repeat at the account's timeout.", + }, "data-arrival": short("One arrival's reactions."), "reaction-line-generate": short("One provider call for one reaction line."), - "workout-insight-generate": short("One workout's insight."), + "workout-insight-generate": { + runtime: "bounded", + modelSeconds: STATUS_RUN_CEILING_MS / 1000, + expireInSeconds: WORKOUT_INSIGHT_GENERATE_EXPIRE_SECONDS, + expiryVia: ["WORKOUT_INSIGHT_GENERATE_EXPIRE_SECONDS"], + why: "One workout's line: one status run at the account's timeout.", + }, "recovery-score-compute": short( "Arithmetic over a recent window, capped at 500 accounts; no provider call.", ), @@ -185,7 +229,13 @@ export const QUEUE_RUNTIME: Readonly> = { exclusive: "lockedPass", why: "On a boundary night up to 200 accounts with a provider call per period each; a single-account warm is short.", }, - "coach-memory-refresh": short("One conversation's summary and facts."), + "coach-memory-refresh": { + runtime: "bounded", + modelSeconds: (COACH_MEMORY_MODEL_RUNS * STATUS_RUN_CEILING_MS) / 1000, + expireInSeconds: COACH_MEMORY_REFRESH_EXPIRE_SECONDS, + expiryVia: ["COACH_MEMORY_REFRESH_EXPIRE_SECONDS"], + why: "One conversation's summary, facts and plan proposals: three status runs in a row at the account's timeout.", + }, "coach-nudge": short( "Light deterministic checks per account; the provider part is capped at 25 calls and 90 seconds a tick.", ), diff --git a/src/lib/jobs/registry/__tests__/registry-contract.test.ts b/src/lib/jobs/registry/__tests__/registry-contract.test.ts index afcbdc995..ed563e6a0 100644 --- a/src/lib/jobs/registry/__tests__/registry-contract.test.ts +++ b/src/lib/jobs/registry/__tests__/registry-contract.test.ts @@ -38,6 +38,10 @@ const ALLOWED: Readonly> = { "lib/ai/coach/coach-memory-shared.ts": "the Coach memory queue's name and payload, with its byte codec; no model code", "lib/ai/coach/bytes-codec.ts": "pulled by coach-memory-shared; a codec", + "lib/ai/ai-budgets.ts": + "pulled by model-time for the call counts the bounded queues' expiries derive from; constants only", + "lib/ai/effective-timeout.ts": + "pulled by model-time for the response-timeout ceiling; arithmetic only, no model code", "lib/export/restore-lock.ts": "the per-account restore lock the consolidation passes take; a lock primitive", "lib/export/restore-failure-cause.ts": diff --git a/src/lib/jobs/registry/status.ts b/src/lib/jobs/registry/status.ts index 104007bcd..87df05d45 100644 --- a/src/lib/jobs/registry/status.ts +++ b/src/lib/jobs/registry/status.ts @@ -7,10 +7,12 @@ * pass and the daily-briefing fallback. * * Most queues here are keyless cron ticks, so they carry no policy. The - * LLM-bound warm queues (pre-generate, per-metric status, period narrative, - * Coach memory) pass `singletonSeconds` on every send, which pg-boss - * constrains whatever the policy, so they are left alone. The three that carry - * one key on the user's local date, which no wall-clock window can express. + * LLM-bound warm queues (pre-generate, period narrative, Coach memory) pass + * `singletonSeconds` on every send, which pg-boss constrains whatever the + * policy, so they are left alone. The per-metric status queue is exclusive as + * well: its run outlasts its slot while the open card keeps sending. The three + * that carry one key on the user's local date, which no wall-clock window can + * express. */ import type { Job } from "pg-boss"; @@ -133,6 +135,11 @@ const jobs: readonly JobDefinition[] = [ // a first visit cannot fan out unbounded provider calls. defineJob({ queue: INSIGHT_STATUS_GENERATE_QUEUE, + policy: { + policy: "exclusive", + reason: + "The open card re-reads the route while its note is generated, and every miss sends again under the user+metric key. A run may take two calls at a ten-minute timeout, far past the 120 s singletonSeconds slot, so only exclusive keeps a second generation from queueing beside the one still running.", + }, work: { localConcurrency: INSIGHT_STATUS_GENERATE_CONCURRENCY }, load: async () => (await insights()).handleInsightStatusGenerate, }), diff --git a/src/lib/jobs/workout-insight-generate-shared.ts b/src/lib/jobs/workout-insight-generate-shared.ts index da8351199..614060950 100644 --- a/src/lib/jobs/workout-insight-generate-shared.ts +++ b/src/lib/jobs/workout-insight-generate-shared.ts @@ -15,9 +15,24 @@ import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { annotate } from "@/lib/logging/context"; import { logCaught } from "@/lib/logging/signal"; +import { + STATUS_RUN_CEILING_MS, + expireAboveModelTime, +} from "@/lib/jobs/model-time"; export const WORKOUT_INSIGHT_GENERATE_QUEUE = "workout-insight-generate"; +/** + * The job's expiry: one status run at the largest response timeout (the call + * and its one reasoning repeat), plus headroom. Past pg-boss's fifteen-minute + * default the job was failed and retried while the paragraph was still being + * written (#1274); the claim kept the retry from calling the model twice, but + * the attempt was recorded as a failure. + */ +export const WORKOUT_INSIGHT_GENERATE_EXPIRE_SECONDS = expireAboveModelTime( + STATUS_RUN_CEILING_MS, +); + /** Serial — this is a provider path, kept off the request pool entirely. */ export const WORKOUT_INSIGHT_GENERATE_CONCURRENCY = 1; @@ -56,6 +71,7 @@ export async function enqueueWorkoutInsight( retryLimit: 3, retryDelay: 60, retryBackoff: true, + expireInSeconds: WORKOUT_INSIGHT_GENERATE_EXPIRE_SECONDS, }); return { enqueued: true }; } catch (err) { diff --git a/src/lib/validations/ai-provider.ts b/src/lib/validations/ai-provider.ts index dc8f92802..b3c80dbe4 100644 --- a/src/lib/validations/ai-provider.ts +++ b/src/lib/validations/ai-provider.ts @@ -14,6 +14,10 @@ * to return. A `.refine()` with a message that quotes its input would break * that, and would put a key in an error body. */ +import { + AI_RESPONSE_TIMEOUT_MAX_SECONDS, + AI_RESPONSE_TIMEOUT_MIN_SECONDS, +} from "@/lib/ai/effective-timeout"; import { z } from "zod/v4"; import { PROVIDER_CHAIN_TYPES } from "@/lib/ai/provider-chain"; @@ -106,8 +110,14 @@ export const aiProviderPatchSchema = z.object({ responseTimeoutSeconds: z .number() .int() - .min(10, "Response timeout must be between 10 and 600 seconds") - .max(600, "Response timeout must be between 10 and 600 seconds") + .min( + AI_RESPONSE_TIMEOUT_MIN_SECONDS, + "Response timeout must be between 10 and 600 seconds", + ) + .max( + AI_RESPONSE_TIMEOUT_MAX_SECONDS, + "Response timeout must be between 10 and 600 seconds", + ) .nullable() .optional(), /** diff --git a/tests/integration/__golden__/queue-registration.json b/tests/integration/__golden__/queue-registration.json index 47c00bd1b..47d73a4e2 100644 --- a/tests/integration/__golden__/queue-registration.json +++ b/tests/integration/__golden__/queue-registration.json @@ -958,7 +958,7 @@ "name": "insight-status-generate", "notify": false, "partition": false, - "policy": "standard", + "policy": "exclusive", "retention_seconds": 1209600, "retry_backoff": false, "retry_delay": 0,