diff --git a/src/lib/jobs/__tests__/ai-job-capability-inventory.test.ts b/src/lib/jobs/__tests__/ai-job-capability-inventory.test.ts index 00f6886bc..28df5d5eb 100644 --- a/src/lib/jobs/__tests__/ai-job-capability-inventory.test.ts +++ b/src/lib/jobs/__tests__/ai-job-capability-inventory.test.ts @@ -41,10 +41,6 @@ const AI_JOBS: Record = { gate: "aiCapabilityForJob", keys: ["statusText"], }, - "lib/jobs/reminder/insights-handlers.ts": { - gate: "aiCapabilityForJob", - keys: ["statusText"], - }, "lib/jobs/insight-status-generate.ts": { gate: "aiCapabilityForJob", keys: ["statusText"], diff --git a/src/lib/jobs/__tests__/status-cron-candidates.test.ts b/src/lib/jobs/__tests__/status-cron-candidates.test.ts index 075c531ab..331de5d62 100644 --- a/src/lib/jobs/__tests__/status-cron-candidates.test.ts +++ b/src/lib/jobs/__tests__/status-cron-candidates.test.ts @@ -175,44 +175,53 @@ describe("findStatusCronCandidates — gates", () => { }); }); -describe("mood-status queue registration (dead-queue guard)", () => { - // v1.18.1 — the nightly status-ladder wiring moved out of the 2143-LOC - // reminder-worker boot file into the status registrar. The dead-queue guard - // follows the wiring there. +describe("the nightly status batch is the one status cron", () => { const workerSrc = fs.readFileSync( path.resolve(__dirname, "../reminder/register-status.ts"), "utf8", ); + const RETIRED = [ + "insights-blood-pressure-status", + "insights-weight-status", + "insights-pulse-status", + "insights-bmi-status", + "insights-mood-status", + "insights-medication-compliance-status", + ]; - it("registers the queue in the allQueues createQueue loop", () => { + it("registers and schedules the 02:00 batch with the nightly pass options", () => { const match = workerSrc.match(/const allQueues\s*=\s*\[([\s\S]*?)\];/); expect(match).not.toBeNull(); - expect(match![1]).toMatch(/\bMOOD_STATUS_QUEUE\b/); - }); - - it("schedules the 02:30 cron with the nightly pass options (retries and expiry)", () => { + expect(match![1]).toMatch(/\bGENERAL_STATUS_QUEUE\b/); expect(workerSrc).toMatch( - /\[\s*MOOD_STATUS_QUEUE\s*,\s*MOOD_STATUS_CRON\s*,\s*nightlyInsightPassOptions\s*\]/, + /\[\s*GENERAL_STATUS_QUEUE\s*,\s*GENERAL_STATUS_CRON\s*,\s*nightlyInsightPassOptions\s*\]/, ); - expect(workerSrc).toMatch(/MOOD_STATUS_CRON\s*=\s*"30 2 \* \* \*"/); + expect(workerSrc).toMatch(/GENERAL_STATUS_CRON\s*=\s*"0 2 \* \* \*"/); }); - it("registers a createAndWork handler for the queue", () => { - expect(workerSrc).toMatch(/createAndWork[\s\S]{0,120}MOOD_STATUS_QUEUE/); + it("retires the six per-metric crons on an instance that still has them", () => { + const retired = workerSrc.match( + /const RETIRED_STATUS_QUEUES\s*=\s*\[([\s\S]*?)\]/, + ); + expect(retired).not.toBeNull(); + for (const name of RETIRED) expect(retired![1]).toContain(`"${name}"`); + expect(workerSrc).toMatch( + /await retireQueues\(boss,\s*RETIRED_STATUS_QUEUES\)/, + ); + const live = workerSrc.replace(retired![0], ""); + for (const name of RETIRED) expect(live).not.toContain(`"${name}"`); }); - it("drives all seven status crons through the shared discovery", () => { + it("drives the batch through the shared discovery", () => { const statusSrc = fs.readFileSync( path.resolve(__dirname, "../reminder/insights-handlers.ts"), "utf8", ); - expect(statusSrc).toMatch(/findStatusCronCandidates\(prisma\)/); - // The old iterate-every-user discovery must be gone from the status - // handlers (the WHOOP/data-backup cohort reads keep their own scans). const statusBlock = statusSrc.slice( - statusSrc.indexOf("async function runStatusCronGenerate"), + statusSrc.indexOf("async function runStatusBatchCron"), statusSrc.indexOf("async function handleInsightPregenerateJob"), ); + expect(statusBlock).toMatch(/findStatusCronCandidates\(prisma\)/); expect(statusBlock).not.toMatch(/prisma\.user\.findMany/); }); }); diff --git a/src/lib/jobs/insight-pregenerate.ts b/src/lib/jobs/insight-pregenerate.ts index 927a64fc9..b2666e40b 100644 --- a/src/lib/jobs/insight-pregenerate.ts +++ b/src/lib/jobs/insight-pregenerate.ts @@ -749,7 +749,7 @@ export async function runInsightPregenerate( // (`insights:${userId}`), so this never starves a user's manual // regenerate quota. The gate bounds the comprehensive cost only: a // blocked user still gets the refill-only status warm below, because - // the 02:xx status crons already skipped every pregenerate candidate + // the 02:00 status batch already skipped every pregenerate candidate // on the assumption that THIS pass warms their cards — exiting early // here left those cards cold until the first on-visit generation. const budget = briefing.available @@ -906,7 +906,7 @@ export async function runInsightPregenerate( // generator's content-hash gate turns a same-data regeneration into a // timestamp refresh with no provider call. The warm runs on every // comprehensive outcome (budget-blocked, failed, timed-out, - // consent-skipped included) because the 02:xx status crons skip every + // consent-skipped included) because the 02:00 status batch skips every // pregenerate candidate on the assumption that THIS pass covers them. // // Nothing to warm when the status notes themselves are unavailable. diff --git a/src/lib/jobs/queue-runtime.ts b/src/lib/jobs/queue-runtime.ts index 9861861e0..12d817ce7 100644 --- a/src/lib/jobs/queue-runtime.ts +++ b/src/lib/jobs/queue-runtime.ts @@ -72,15 +72,6 @@ export type QueueRuntime = const short = (why: string): QueueRuntime => ({ runtime: "short", why }); -const nightlyStatusPass = (fn: string): QueueRuntime => ({ - runtime: "long", - expireInSeconds: NIGHTLY_INSIGHT_PASS_EXPIRE_SECONDS, - expiryVia: ["nightlyInsightPassOptions"], - stop: { file: "lib/jobs/reminder/insights-handlers.ts", fn }, - exclusive: "lockedPass", - why: "Nightly status pass: one provider call per candidate account, serially, over the whole instance.", -}); - const cohortSync = (provider: string): QueueRuntime => short( `${provider} sync: a recent window per connection (incremental since the last sync); a webhook job is one connection.`, @@ -140,17 +131,17 @@ export const QUEUE_RUNTIME: Readonly> = { "strava-backfill": forwarder, // ── Status and insights ──────────────────────────────────────────── - "insights-general-status": nightlyStatusPass("handleGeneralStatusGenerate"), - "insights-blood-pressure-status": nightlyStatusPass( - "handleBloodPressureStatusGenerate", - ), - "insights-weight-status": nightlyStatusPass("handleWeightStatusGenerate"), - "insights-pulse-status": nightlyStatusPass("handlePulseStatusGenerate"), - "insights-bmi-status": nightlyStatusPass("handleBmiStatusGenerate"), - "insights-mood-status": nightlyStatusPass("handleMoodStatusGenerate"), - "insights-medication-compliance-status": nightlyStatusPass( - "handleMedicationComplianceStatusGenerate", - ), + "insights-general-status": { + runtime: "long", + expireInSeconds: NIGHTLY_INSIGHT_PASS_EXPIRE_SECONDS, + expiryVia: ["nightlyInsightPassOptions"], + stop: { + file: "lib/jobs/reminder/insights-handlers.ts", + fn: "handleGeneralStatusGenerate", + }, + exclusive: "lockedPass", + why: "Nightly status batch: one provider call per candidate account, serially, over the whole instance.", + }, "insight-pregenerate": { runtime: "long", expireInSeconds: INSIGHT_PREGENERATE_EXPIRE_SECONDS, diff --git a/src/lib/jobs/reminder/__tests__/registrar-shared.test.ts b/src/lib/jobs/reminder/__tests__/registrar-shared.test.ts new file mode 100644 index 000000000..00d3998ad --- /dev/null +++ b/src/lib/jobs/reminder/__tests__/registrar-shared.test.ts @@ -0,0 +1,45 @@ +/** + * The registrar's boot-time decisions that reach pg-boss on every instance. + */ +import { describe, expect, it, vi } from "vitest"; + +vi.mock("@/lib/db", () => ({ + prisma: { $executeRaw: vi.fn().mockResolvedValue(0) }, +})); +vi.mock("@/lib/logging/transports", () => ({ emitIfSampled: vi.fn() })); + +import { retireQueues } from "../registrar-shared"; + +function fakeBoss() { + return { + createQueue: vi.fn().mockResolvedValue(undefined), + updateQueue: vi.fn().mockResolvedValue(undefined), + schedule: vi.fn().mockResolvedValue(undefined), + work: vi.fn().mockResolvedValue("worker-id"), + unschedule: vi.fn().mockResolvedValue(undefined), + deleteQueue: vi.fn().mockResolvedValue(undefined), + }; +} + +describe("retireQueues", () => { + it("removes the schedule before the queue, and one failure does not stop the rest", async () => { + const boss = fakeBoss(); + const order: string[] = []; + boss.unschedule.mockImplementation(async (name: string) => { + order.push(`unschedule:${name}`); + }); + boss.deleteQueue.mockImplementation(async (name: string) => { + order.push(`delete:${name}`); + if (name === "old-a") throw new Error("locked"); + }); + await expect( + retireQueues(boss as never, ["old-a", "old-b"]), + ).resolves.toBeUndefined(); + expect(order).toEqual([ + "unschedule:old-a", + "delete:old-a", + "unschedule:old-b", + "delete:old-b", + ]); + }); +}); diff --git a/src/lib/jobs/reminder/insights-handlers.ts b/src/lib/jobs/reminder/insights-handlers.ts index dff543080..54fec9bdb 100644 --- a/src/lib/jobs/reminder/insights-handlers.ts +++ b/src/lib/jobs/reminder/insights-handlers.ts @@ -1,5 +1,6 @@ /** - * Nightly insight status crons (general / blood pressure / weight / pulse / BMI / mood / medication compliance) plus the insight pregenerate and per-user status-generate handlers. + * The nightly status batch plus the insight pregenerate and per-user + * status-generate handlers. * * Extracted from reminder-worker.ts, which owns the queue names, cron * schedules, and boss.work registrations. @@ -23,155 +24,32 @@ import { import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome"; import { jobBudget } from "@/lib/jobs/job-budget"; import { withBackgroundEvent } from "@/lib/logging/background"; -import { generateBloodPressureStatusForUser } from "@/lib/insights/blood-pressure-status"; -import { generateWeightStatusForUser } from "@/lib/insights/weight-status"; -import { generatePulseStatusForUser } from "@/lib/insights/pulse-status"; -import { generateBmiStatusForUser } from "@/lib/insights/bmi-status"; -import { generateMoodStatusForUser } from "@/lib/insights/mood-status"; -import { generateMedicationComplianceStatusForUser } from "@/lib/insights/medication-compliance-status"; import { generateStatusBatchForUser } from "@/lib/insights/status-batch"; import { findStatusCronCandidates } from "@/lib/jobs/status-cron-candidates"; import { annotate } from "@/lib/logging/context"; -import { aiCapabilityForJob } from "@/lib/ai/capabilities/gate"; import { getWorkerPrisma } from "./shared"; export interface GeneralStatusPayload { triggeredAt: string; } -export interface BloodPressureStatusPayload { - triggeredAt: string; -} - -export interface WeightStatusPayload { - triggeredAt: string; -} - -export interface PulseStatusPayload { - triggeredAt: string; -} - -export interface BmiStatusPayload { - triggeredAt: string; -} - -export interface MoodStatusPayload { - triggeredAt?: string; -} - -export interface MedicationComplianceStatusPayload { - triggeredAt: string; -} - /** - * Shared driver for the nightly 02:xx per-metric status crons. User - * discovery is centralised in `findStatusCronCandidates`, which applies - * the operator's `insightStatus` switch, the person's AI analysis switch, - * and the pregenerate-candidate skip (users with a stale comprehensive - * cache belong to the 04:30 pre-generate pass, which re-warms every - * per-status note anyway — see `status-cron-candidates.ts` for the full - * division of nightly labour). Each user's `statusText` capability is then - * resolved before a generator runs. - * The generators normalise `locale` themselves (de stays de, everything - * else gets English prose). + * The nightly status batch. The 02:00 cron runs `generateStatusBatchForUser` + * for every candidate (`findStatusCronCandidates`): it builds every + * per-metric snapshot once, issues ONE provider call for the metrics still + * needing the model, and fans the answer into the per-metric cache rows. A + * metric the batch omits, or every metric when the call fails, falls back to + * that card's own path inside the batch, so the pass covers every card on its + * own. Each user's `statusText` capability is resolved inside the batch before + * a snapshot is built. * - * The pass is what succeeds or fails. One user whose generation threw is that - * user's cold card, reported as the `failed` count; failing the job would - * re-run the whole cohort over it. - * - * The pass walks every candidate with a provider call each, so on a large - * instance it can outlast its job. `shouldStop` is the job's time budget: the - * pass stops between users once it is spent and reports `stopped_early`. The - * users it did not reach keep yesterday's note until the next night or their - * next visit; nothing is retried beside a pass that is still running. + * The pass is what succeeds or fails. One user whose batch threw is that + * user's cold cards, reported as the `failed` count; failing the job would + * re-run the whole cohort over it. `shouldStop` is the job's time budget: the + * pass stops between users once it is spent and reports `stopped_early`, and + * the users it did not reach are warmed by their next visit or night. */ -export async function runStatusCronGenerate( - taskName: string, - generate: ( - userId: string, - options: { locale: string | null; force: boolean }, - ) => Promise, - shouldStop: () => boolean = () => false, -): Promise { - return withBackgroundEvent(taskName, async (evt) => { - const prisma = getWorkerPrisma(); - try { - const users = await findStatusCronCandidates(prisma); - - if (users.length === 0) { - return jobDone({ total: 0, generated: 0, failed: 0 }); - } - - let generated = 0; - let failed = 0; - let stoppedEarly = false; - - for (const user of users) { - if (shouldStop()) { - stoppedEarly = true; - break; - } - // The capability before the generator builds a snapshot. The batch - // entry below checks for itself. - const capability = await aiCapabilityForJob(user.id, "statusText"); - if (!capability.available) { - annotate({ - action: { name: "insights.status.cron.skipped" }, - meta: { task: taskName, reason: capability.reason }, - }); - continue; - } - try { - await generate(user.id, { - locale: await resolveJobLocale(user.locale), - force: false, - }); - generated++; - } catch (error) { - failed++; - evt.addWarning( - `${taskName} generation failed for user ${user.id}: ${error}`, - ); - } - } - - evt.setBackground({ - task_name: taskName, - result: { generated, failed, total: users.length, stoppedEarly }, - }); - return jobDone({ - total: users.length, - generated, - failed, - stopped_early: stoppedEarly, - }); - } catch (err) { - evt.setError(err); - recordError(); - await reportWorkerError(taskName, err); - throw err; - } - }); -} - -/** - * v1.18.11 (P2) — nightly status batch. The 02:00 anchor cron runs - * `generateStatusBatchForUser` for every candidate: it builds all seven - * per-metric snapshots once and issues ONE provider call for the metrics - * still needing the LLM (seed-pinned + grounded inside `runStatusCompletion`, - * unchanged), fanning the response into the SAME per-metric cache rows the - * standalone generators wrote. - * - * The six later per-metric crons (02:05–02:30) stay registered and keep their - * own drivers: a card the batch wrote today resolves inside their `prepare` - * step as a calendar-day cache hit (`served`, no provider call, no snapshot - * rebuild), so they cost a cache read and cover the edge cases the batch - * can't — a user discovered after 02:00, a metric the batch omitted, or a - * batch-call failure (which falls each metric back to its single-card path - * inside the batch itself). Net effect: the nightly ladder pays one call per - * user instead of seven, with no queue/registry churn. - */ -async function runStatusBatchCron( +export async function runStatusBatchCron( taskName: string, shouldStop: () => boolean, ): Promise { @@ -189,8 +67,8 @@ async function runStatusBatchCron( let failed = 0; let stoppedEarly = false; for (const user of users) { - // Same budget as the per-metric passes: stop between users, and - // leave the rest to the later crons and the next night. + // Stop between users; the rest are warmed by a visit or the next + // night. if (shouldStop()) { stoppedEarly = true; break; @@ -233,8 +111,7 @@ async function runStatusBatchCron( }, }); // A user whose batch call threw already fell back to the single-card - // path inside the batch, and the six later per-metric crons cover what - // is still cold. The batch pass ran, so it reports the count. + // path inside the batch. The pass ran, so it reports the count. return jobDone({ total: users.length, generated, @@ -257,66 +134,6 @@ export function handleGeneralStatusGenerate( return runStatusBatchCron("job.insights.batch", jobBudget(jobs)); } -export function handleBloodPressureStatusGenerate( - jobs: Job[], -): Promise { - return runStatusCronGenerate( - "job.insights.blood_pressure", - generateBloodPressureStatusForUser, - jobBudget(jobs), - ); -} - -export function handleWeightStatusGenerate( - jobs: Job[], -): Promise { - return runStatusCronGenerate( - "job.insights.weight", - generateWeightStatusForUser, - jobBudget(jobs), - ); -} - -export function handlePulseStatusGenerate( - jobs: Job[], -): Promise { - return runStatusCronGenerate( - "job.insights.pulse", - generatePulseStatusForUser, - jobBudget(jobs), - ); -} - -export function handleBmiStatusGenerate( - jobs: Job[], -): Promise { - return runStatusCronGenerate( - "job.insights.bmi", - generateBmiStatusForUser, - jobBudget(jobs), - ); -} - -export function handleMoodStatusGenerate( - jobs: Job[], -): Promise { - return runStatusCronGenerate( - "job.insights.mood", - generateMoodStatusForUser, - jobBudget(jobs), - ); -} - -export function handleMedicationComplianceStatusGenerate( - jobs: Job[], -): Promise { - return runStatusCronGenerate( - "job.insights.medication_compliance", - generateMedicationComplianceStatusForUser, - jobBudget(jobs), - ); -} - export async function handleInsightPregenerateJob( jobs: Job[], ): Promise { diff --git a/src/lib/jobs/reminder/register-status.ts b/src/lib/jobs/reminder/register-status.ts index bd8d3c09a..1f3d961c6 100644 --- a/src/lib/jobs/reminder/register-status.ts +++ b/src/lib/jobs/reminder/register-status.ts @@ -1,8 +1,8 @@ /** * Status / insight / computed-score queue registrar. * - * Owns the nightly per-metric insight-status ladder (general / BP / weight / - * pulse / BMI / mood / medication-compliance), the comprehensive-insight + * Owns the nightly status batch (every per-metric status card, one provider + * call per account), the comprehensive-insight * pre-generation, the on-demand per-metric status generation, the nightly * computed scores (recovery / stress / strain), the proactive Coach nudge, the * Coach memory refresh, the period-narrative warm, and the medication @@ -112,24 +112,13 @@ import { createAndWork, insightRetryOptions, nightlyInsightPassOptions, + retireQueues, type QueuePolicyTable, type ScheduleEntry, } from "./registrar-shared"; import { GeneralStatusPayload, - BloodPressureStatusPayload, - WeightStatusPayload, - PulseStatusPayload, - BmiStatusPayload, - MoodStatusPayload, - MedicationComplianceStatusPayload, handleGeneralStatusGenerate, - handleBloodPressureStatusGenerate, - handleWeightStatusGenerate, - handlePulseStatusGenerate, - handleBmiStatusGenerate, - handleMoodStatusGenerate, - handleMedicationComplianceStatusGenerate, handleInsightPregenerateJob, handleInsightStatusGenerate, } from "./insights-handlers"; @@ -138,45 +127,25 @@ const GENERAL_STATUS_QUEUE = "insights-general-status"; const GENERAL_STATUS_CRON = "0 2 * * *"; // daily at 02:00 -const BLOOD_PRESSURE_STATUS_QUEUE = "insights-blood-pressure-status"; - -const BLOOD_PRESSURE_STATUS_CRON = "5 2 * * *"; // daily at 02:05 - -const WEIGHT_STATUS_QUEUE = "insights-weight-status"; - -const WEIGHT_STATUS_CRON = "10 2 * * *"; // daily at 02:10 - -const PULSE_STATUS_QUEUE = "insights-pulse-status"; - -const PULSE_STATUS_CRON = "15 2 * * *"; // daily at 02:15 - -const BMI_STATUS_QUEUE = "insights-bmi-status"; - -const BMI_STATUS_CRON = "20 2 * * *"; // daily at 02:20 -// v1.15.20 — mood joins the nightly per-metric status ladder. Same gate + -// discovery as the six older crons (see status-cron-candidates.ts); 02:30 -// continues the 5-minute stagger after BMI (02:20) and compliance (02:25). - -const MOOD_STATUS_QUEUE = "insights-mood-status"; - -const MOOD_STATUS_CRON = "30 2 * * *"; // daily at 02:30 - -const MEDICATION_COMPLIANCE_STATUS_QUEUE = - "insights-medication-compliance-status"; - -const MEDICATION_COMPLIANCE_STATUS_CRON = "25 2 * * *"; // daily at 02:25 +/** + * The six per-metric status crons that followed the 02:00 batch (02:05 to + * 02:30). The batch builds every card and falls each metric it could not + * cover back to that card's own path, so they only re-read cards the batch + * had just written; a user who joins later is warmed by the visit or the next + * night. Their schedules and queues are removed on an instance that still has + * them. + */ +const RETIRED_STATUS_QUEUES = [ + "insights-blood-pressure-status", + "insights-weight-status", + "insights-pulse-status", + "insights-bmi-status", + "insights-mood-status", + "insights-medication-compliance-status", +] as const; const allQueues = [ GENERAL_STATUS_QUEUE, - BLOOD_PRESSURE_STATUS_QUEUE, - WEIGHT_STATUS_QUEUE, - PULSE_STATUS_QUEUE, - BMI_STATUS_QUEUE, - // v1.15.20 — mood joins the nightly status ladder. The queue MUST be - // registered here or pg-boss never provisions it and the 02:30 schedule - // silently no-ops (the v1.4.37 dead-queue class). - MOOD_STATUS_QUEUE, - MEDICATION_COMPLIANCE_STATUS_QUEUE, // v1.7.0 — nightly comprehensive-insight pre-generation so the daily // briefing is warm before the user opens /insights or the dashboard // snapshot. Without this entry the 04:30 schedule silently no-ops. @@ -238,21 +207,6 @@ const allQueues = [ const schedules: ScheduleEntry[] = [ [GENERAL_STATUS_QUEUE, GENERAL_STATUS_CRON, nightlyInsightPassOptions], - [ - BLOOD_PRESSURE_STATUS_QUEUE, - BLOOD_PRESSURE_STATUS_CRON, - nightlyInsightPassOptions, - ], - [WEIGHT_STATUS_QUEUE, WEIGHT_STATUS_CRON, nightlyInsightPassOptions], - [PULSE_STATUS_QUEUE, PULSE_STATUS_CRON, nightlyInsightPassOptions], - [BMI_STATUS_QUEUE, BMI_STATUS_CRON, nightlyInsightPassOptions], - // v1.15.20 — mood status nightly, continuing the 02:xx ladder. - [MOOD_STATUS_QUEUE, MOOD_STATUS_CRON, nightlyInsightPassOptions], - [ - MEDICATION_COMPLIANCE_STATUS_QUEUE, - MEDICATION_COMPLIANCE_STATUS_CRON, - nightlyInsightPassOptions, - ], // v1.7.0 — nightly 04:30 Europe/Berlin comprehensive-insight // pre-generation. Budget-gated per user inside the handler. [ @@ -383,6 +337,7 @@ export async function registerStatusQueues( boss: PgBoss, ): Promise { await createAndSchedule(boss, allQueues, schedules, queuePolicies); + await retireQueues(boss, RETIRED_STATUS_QUEUES); await createAndWork( boss, @@ -394,54 +349,6 @@ export async function registerStatusQueues( handleGeneralStatusGenerate, ), ); - await createAndWork( - boss, - BLOOD_PRESSURE_STATUS_QUEUE, - { localConcurrency: 1 }, - lockedPass( - BLOOD_PRESSURE_STATUS_QUEUE, - () => WHOLE_PASS, - handleBloodPressureStatusGenerate, - ), - ); - await createAndWork( - boss, - WEIGHT_STATUS_QUEUE, - { localConcurrency: 1 }, - lockedPass( - WEIGHT_STATUS_QUEUE, - () => WHOLE_PASS, - handleWeightStatusGenerate, - ), - ); - await createAndWork( - boss, - PULSE_STATUS_QUEUE, - { localConcurrency: 1 }, - lockedPass(PULSE_STATUS_QUEUE, () => WHOLE_PASS, handlePulseStatusGenerate), - ); - await createAndWork( - boss, - BMI_STATUS_QUEUE, - { localConcurrency: 1 }, - lockedPass(BMI_STATUS_QUEUE, () => WHOLE_PASS, handleBmiStatusGenerate), - ); - await createAndWork( - boss, - MOOD_STATUS_QUEUE, - { localConcurrency: 1 }, - lockedPass(MOOD_STATUS_QUEUE, () => WHOLE_PASS, handleMoodStatusGenerate), - ); - await createAndWork( - boss, - MEDICATION_COMPLIANCE_STATUS_QUEUE, - { localConcurrency: 1 }, - lockedPass( - MEDICATION_COMPLIANCE_STATUS_QUEUE, - () => WHOLE_PASS, - handleMedicationComplianceStatusGenerate, - ), - ); // v1.7.0 — nightly comprehensive-insight pre-generation. // // v1.16.8 — localConcurrency 2 (was 1). The queue carries BOTH the diff --git a/src/lib/jobs/reminder/registrar-shared.ts b/src/lib/jobs/reminder/registrar-shared.ts index 454943d81..b76474815 100644 --- a/src/lib/jobs/reminder/registrar-shared.ts +++ b/src/lib/jobs/reminder/registrar-shared.ts @@ -169,7 +169,7 @@ export const insightRetryOptions = { } as const; /** - * Send options for a nightly AI pass over every account (the status crons). + * Send options for a nightly AI pass over every account (the status batch). * * The retries of `insightRetryOptions`, plus a two-hour expiry. Each pass * makes a provider call per account, and on a large instance that outlasted @@ -333,6 +333,26 @@ export async function createAndSchedule( } } +/** + * Remove queues the tree no longer declares: their schedule first, so the + * timekeeper stops minting jobs nobody works, then the queue with whatever it + * still holds. A queue that is already gone is a no-op, so every boot after + * the first does nothing. Never fails a boot. + */ +export async function retireQueues( + boss: PgBoss, + queues: readonly string[], +): Promise { + for (const name of queues) { + try { + await boss.unschedule(name); + await boss.deleteQueue(name); + } catch (err) { + workerLog("error", `[queue-retire] failed to retire ${name}`, err); + } + } +} + /** * Bind a handler to a queue. The one legal way to reach `boss.work` in this * tree — the `healthlog/job-handler-outcome` lint rule refuses a bare diff --git a/src/lib/jobs/status-cron-candidates.ts b/src/lib/jobs/status-cron-candidates.ts index 9b62b23a6..25a545ddf 100644 --- a/src/lib/jobs/status-cron-candidates.ts +++ b/src/lib/jobs/status-cron-candidates.ts @@ -1,7 +1,7 @@ /** - * Shared user discovery for the nightly 02:xx per-metric status crons - * (general / blood-pressure / weight / pulse / bmi / medication- - * compliance / mood). + * User discovery for the nightly 02:00 status batch, which writes every + * per-metric status card (general / blood-pressure / weight / pulse / bmi / + * medication-compliance / mood). * * Before this module each cron iterated EVERY user row: accounts that * disabled the coach surface (`disableCoach`) and operator-disabled @@ -12,18 +12,18 @@ * Division of nightly labour (documented here once, referenced by the * handlers): * - * - 02:xx status crons → users the 04:30 `insight-pregenerate` pass will + * - 02:00 status batch → users the 04:30 `insight-pregenerate` pass will * not reach: a comprehensive cache fresher than `PREGENERATE_STALE_MS`, * so that pass skips them, while their per-status notes still age out * daily and only these crons re-fill them. * - 04:30 insight-pregenerate → the rest: users with a comprehensive cache * older than `PREGENERATE_STALE_MS`. That pass warms every per-status * note whenever `statusText` is available for the user, whatever became - * of the briefing, so a 02:xx generation for them would be redone two + * of the briefing, so a 02:00 generation for them would be redone two * hours later. * - * Cheap gates, in SQL and in order (the per-user `statusText` capability in - * each handler, and the chokepoint behind it, are what decide): + * Cheap gates, in SQL and in order (the per-user `statusText` capability + * inside the batch, and the chokepoint behind it, are what decide): * 1. The operator's `insightStatus` switch (master applied). Off, the whole * pass is empty. * 2. The person's AI analysis switch (the `insights` module). Someone who @@ -32,7 +32,7 @@ * else. * 3. The pregenerate-candidate skip described above, only while the * `briefing` switch is on; when it is off the 04:30 pass no-ops, so the - * 02:xx crons keep covering everyone. + * 02:00 batch keeps covering everyone. */ import type { PrismaClient } from "@/generated/prisma/client"; import { getAssistantFlags } from "@/lib/feature-flags"; @@ -59,7 +59,7 @@ export async function findStatusCronCandidates( if (users.length === 0) return []; // When the briefing switch is off the 04:30 pregenerate pass no-ops, so no - // user has a comprehensive claim and the 02:xx crons keep all. + // user has a comprehensive claim and the 02:00 batch keeps all. if (!flags.briefing) { return users.map((u) => ({ id: u.id, locale: u.locale })); } diff --git a/src/lib/jobs/subsystem-surface.ts b/src/lib/jobs/subsystem-surface.ts index ea38c7ec3..aa15142e6 100644 --- a/src/lib/jobs/subsystem-surface.ts +++ b/src/lib/jobs/subsystem-surface.ts @@ -50,12 +50,6 @@ const subsystemSurface = { // register-status.ts "insights-general-status": { audience: "system" }, - "insights-blood-pressure-status": { audience: "system" }, - "insights-weight-status": { audience: "system" }, - "insights-pulse-status": { audience: "system" }, - "insights-bmi-status": { audience: "system" }, - "insights-mood-status": { audience: "system" }, - "insights-medication-compliance-status": { audience: "system" }, "insight-pregenerate": { audience: "system" }, "insight-status-generate": { audience: "system" }, "morning-digest-refresh": { audience: "system" }, diff --git a/tests/integration/background-switch-freshness.test.ts b/tests/integration/background-switch-freshness.test.ts index 98140c248..99f3ac81f 100644 --- a/tests/integration/background-switch-freshness.test.ts +++ b/tests/integration/background-switch-freshness.test.ts @@ -4,16 +4,17 @@ * Reads inside one request are memoised on the request's wide event, so a * page that resolves the AI capabilities five times reads the settings row * once. A background job opens one wide event for its whole run, and the - * status crons loop over every candidate inside it: memoising there meant the + * status batch loops over every candidate inside it: memoising there meant the * switches, the modules and the consent kinds were read once at the start of * the run and served to every later user, and the wire re-check read the same * cached answer. Turning the Assistant off at 02:05 did not stop the 02:00 * pass. * - * Driven here over the real loop, the real capability loader and Postgres: - * the first user's generation turns the switch off, and the remaining users - * must not reach a generator, while the wire re-check inside the first user's - * own run must see the change too. + * Driven here over the real loop, the real capability loader and Postgres. + * The batch is stubbed past its first step, the `statusText` capability it + * resolves before building anything: the first user's run turns the switch + * off, the remaining users must not get past that step, and the wire re-check + * inside the first user's own run must see the change too. */ import { beforeEach, describe, expect, it, vi } from "vitest"; @@ -27,7 +28,30 @@ vi.mock("@/lib/jobs/status-cron-candidates", () => ({ ), })); -import { runStatusCronGenerate } from "@/lib/jobs/reminder/insights-handlers"; +const batch = vi.hoisted(() => ({ + run: async (_userId: string): Promise => {}, +})); + +vi.mock("@/lib/insights/status-batch", async () => { + const { aiCapabilityForJob } = await import("@/lib/ai/capabilities/gate"); + return { + generateStatusBatchForUser: vi.fn(async (userId: string) => { + const result = { + served: 0, + batched: 0, + fellBack: 0, + batchCallMade: false, + }; + if (!(await aiCapabilityForJob(userId, "statusText")).available) { + return result; + } + await batch.run(userId); + return { ...result, batched: 1, batchCallMade: true }; + }), + }; +}); + +import { runStatusBatchCron } from "@/lib/jobs/reminder/insights-handlers"; import { aiEgressRefusal } from "@/lib/ai/capabilities/egress"; let counter = 0; @@ -74,7 +98,7 @@ describe("a switch turned off mid-run", () => { const reached: string[] = []; const wire: Array = []; - const generate = vi.fn(async (userId: string) => { + batch.run = async (userId: string) => { reached.push(userId); // The wire re-check before the switch moves: open. wire.push( @@ -87,9 +111,9 @@ describe("a switch turned off mid-run", () => { (await aiEgressRefusal("statusText", userId, ["anthropic"]))?.reason ?? null, ); - }); + }; - await runStatusCronGenerate("job.test.fresh_switch", generate); + await runStatusBatchCron("job.test.fresh_switch", () => false); expect(reached).toEqual([candidates.ids[0]]); expect(wire[0]).toBeNull();