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,