From 6ebd5e8711bc1401a7e4f8d968bffef53bbb7ccc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Fri, 14 Aug 2026 05:53:17 +0200 Subject: [PATCH] Guard the operator's cap on fallback hops and reap dead summary promises MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - The provider-chain walker now checks the day's recorded spend against OPERATOR_COST_CAP before invoking an operator-funded admin-* entry. A request whose primary provider runs on the user's own key reserves under the generous user-plan ceiling; a fallback hop (or health-ledger reorder) onto the operator's key could then spend past the operator's cap unchecked. An exhausted cap skips the candidate like a failed hop. - summaryState PENDING gets a giver-upper: the detail serialiser degrades a PENDING older than an hour to UNAVAILABLE (which offers the manual generate action), and an hourly reaper persists the same heal following the orphan-ImportJob reconcile precedent — a job dead past its retry budget no longer leaves the sheet saying "generating" forever. --- src/lib/ai/__tests__/provider-runner.test.ts | 87 +++++++++++++++++++ src/lib/ai/coach/budget.ts | 36 +++++++- src/lib/ai/provider-runner.ts | 35 ++++++++ .../document-summary-serialisation.test.ts | 26 +++++- src/lib/documents/store.ts | 21 +++++ .../__tests__/document-summary-reaper.test.ts | 32 +++++++ src/lib/jobs/document-summary-reaper.ts | 77 ++++++++++++++++ src/lib/jobs/reminder/register-maintenance.ts | 23 +++++ 8 files changed, 333 insertions(+), 4 deletions(-) create mode 100644 src/lib/jobs/__tests__/document-summary-reaper.test.ts create mode 100644 src/lib/jobs/document-summary-reaper.ts diff --git a/src/lib/ai/__tests__/provider-runner.test.ts b/src/lib/ai/__tests__/provider-runner.test.ts index b34a5fd1e..51affa883 100644 --- a/src/lib/ai/__tests__/provider-runner.test.ts +++ b/src/lib/ai/__tests__/provider-runner.test.ts @@ -10,6 +10,19 @@ import { // provider-health ledger. These pure chain tests do not stand up a DB, // so swap the default for a no-op; the dedicated ledger-aware cases // below inject an in-memory ledger explicitly. +// v1.37.19 (A7-2) — the runner consults the day's recorded spend before an +// operator-funded hop. These pure chain tests do not stand up a DB, so the +// ledger read is swapped for a controllable in-memory figure. +const budgetState = vi.hoisted(() => ({ spent: 0 })); +vi.mock("../coach/budget", async () => { + const actual = + await vi.importActual("../coach/budget"); + return { + ...actual, + readDailySpend: vi.fn(async () => budgetState.spent), + }; +}); + vi.mock("../provider-health-ledger", async () => { const actual = await vi.importActual< typeof import("../provider-health-ledger") @@ -97,6 +110,7 @@ function err(status: number, msg = "boom"): Error & { httpStatus: number } { beforeEach(() => { clearLastWorkingProviderCache(); + budgetState.spent = 0; }); afterEach(() => { @@ -671,3 +685,76 @@ describe("runStreamingRawCompletionWithFallback", () => { expect(result.fallbackHops.map((h) => h.providerType)).toEqual(["openai"]); }); }); + +// v1.37.19 (A7-2) — hop-time operator-cost guard. +// +// Watched red: with the `isOperatorFundedProvider` check removed from +// `runRawChain` (the pre-fix walker), the exhausted-cap case below fails — +// the admin-openai fallback was invoked and spent operator money although +// the request had only been reserved under the user-plan ceiling. +describe("runRawCompletionWithFallback — operator-cost cap at hop time", () => { + const params: CompletionParams = singleUserTurn({ system: "s", user: "u" }); + + it("refuses the admin-* fallback hop once the operator cap is exhausted", async () => { + budgetState.spent = 200_000; // OPERATOR_COST_CAP + const codex = new ScriptedProvider({ + type: "codex", + script: [{ ok: false, error: err(500) }], + }); + const adminOpenai = new ScriptedProvider({ script: [{ ok: true }] }); + + await expect( + runRawCompletionWithFallback({ + userId: "u-cap", + providers: [ + { providerType: "codex", instance: codex }, + { providerType: "admin-openai", instance: adminOpenai }, + ], + params, + }), + ).rejects.toMatchObject({ + attempts: [ + expect.objectContaining({ providerType: "codex" }), + expect.objectContaining({ + providerType: "admin-openai", + failureReason: "operator-cost-cap-exhausted", + }), + ], + }); + // The operator-funded provider was never invoked. + expect(adminOpenai.callCount).toBe(0); + }); + + it("lets the admin-* fallback run while the operator cap has headroom", async () => { + budgetState.spent = 100; + const codex = new ScriptedProvider({ + type: "codex", + script: [{ ok: false, error: err(500) }], + }); + const adminOpenai = new ScriptedProvider({ script: [{ ok: true }] }); + + const outcome = await runRawCompletionWithFallback({ + userId: "u-headroom", + providers: [ + { providerType: "codex", instance: codex }, + { providerType: "admin-openai", instance: adminOpenai }, + ], + params, + }); + expect(outcome.workingProvider.providerType).toBe("admin-openai"); + expect(adminOpenai.callCount).toBe(1); + }); + + it("never consults the ledger for a user-funded chain", async () => { + budgetState.spent = 999_999_999; + const openai = new ScriptedProvider({ script: [{ ok: true }] }); + const outcome = await runRawCompletionWithFallback({ + userId: "u-own-key", + providers: [{ providerType: "openai", instance: openai }], + params, + }); + // The user's own key is not the operator's money — the cap does not + // apply and the call proceeds regardless of the recorded spend. + expect(outcome.workingProvider.providerType).toBe("openai"); + }); +}); diff --git a/src/lib/ai/coach/budget.ts b/src/lib/ai/coach/budget.ts index 35c026222..d559a8e82 100644 --- a/src/lib/ai/coach/budget.ts +++ b/src/lib/ai/coach/budget.ts @@ -64,13 +64,43 @@ export function resolveDailyCap( chain: ReadonlyArray<{ providerType: ProviderChainType }>, ): number { const primary = chain[0]?.providerType; - return primary === "admin-openai" || - primary === "admin-codex" || - primary === undefined + return primary === undefined || isOperatorFundedProvider(primary) ? OPERATOR_COST_CAP : USER_PLAN_CAP; } +/** + * v1.37.19 (A7-2) — does this provider spend the OPERATOR's money? + * `admin-openai` (the server's own API key) and `admin-codex` (the + * server's shared ChatGPT account) do; everything else is the user's own + * egress. One predicate so `resolveDailyCap` and the chain walker's + * hop-time guard cannot drift on the classification. + */ +export function isOperatorFundedProvider(type: ProviderChainType): boolean { + return type === "admin-openai" || type === "admin-codex"; +} + +/** + * v1.37.19 (A7-2) — the day's recorded spend for one user (0 when no row). + * + * Read by the chain walker's hop-time operator-cap guard: a request whose + * PRIMARY provider runs on the user's own key reserves under the generous + * `USER_PLAN_CAP`, but the chain may still FALL BACK onto an operator-funded + * `admin-*` entry (or the health-ledger reorder may promote one) — and that + * hop must not spend past `OPERATOR_COST_CAP` just because the reservation + * was checked against the wrong owner's ceiling. + */ +export async function readDailySpend( + userId: string, + dateKey: string = buildDateKey(), +): Promise { + const row = await prisma.coachUsage.findUnique({ + where: { userId_dateKey: { userId, dateKey } }, + select: { totalTokens: true }, + }); + return row?.totalTokens ?? 0; +} + /** * Build the UTC day-key for a given clock. Defaults to "now". * diff --git a/src/lib/ai/provider-runner.ts b/src/lib/ai/provider-runner.ts index cc62c83da..b54109a74 100644 --- a/src/lib/ai/provider-runner.ts +++ b/src/lib/ai/provider-runner.ts @@ -6,6 +6,11 @@ import { type ProviderSkipHint, } from "./provider-health-ledger"; import { annotate } from "@/lib/logging/context"; +import { + isOperatorFundedProvider, + OPERATOR_COST_CAP, + readDailySpend, +} from "@/lib/ai/coach/budget"; /** * The local model is the guaranteed floor (Epic B Pillar 4): a @@ -394,6 +399,36 @@ async function runRawChain( for (let i = 0; i < ordered.length; i += 1) { const candidate = ordered[i]; + // v1.37.19 (A7-2) — hop-time operator-cost guard. A request whose + // primary provider runs on the USER's own key reserves its budget + // under the generous user-plan ceiling; when the chain then falls + // back onto an operator-funded `admin-*` entry (or the health-ledger + // reorder promotes one), that hop would spend the OPERATOR's money + // under the wrong owner's cap. Check the day's recorded spend + // against OPERATOR_COST_CAP before letting an operator-funded + // candidate run; an exhausted cap skips the candidate exactly like a + // failed hop (next entry, or AllProvidersFailedError at the end). + if (isOperatorFundedProvider(candidate.providerType)) { + const spent = await readDailySpend(userId); + if (spent >= OPERATOR_COST_CAP) { + const hop: FallbackHop = { + providerType: candidate.providerType, + attempt: i + 1, + failureReason: "operator-cost-cap-exhausted", + httpStatus: null, + }; + hops.push(hop); + annotate({ + action: { name: "ai.chain.operator_cap_refused" }, + meta: { + [`ai_chain_hop_${i + 1}_provider`]: candidate.providerType, + [`ai_chain_hop_${i + 1}_reason`]: "operator-cost-cap-exhausted", + operator_cap_spent: spent, + }, + }); + continue; + } + } try { const result = await invoke(candidate); rememberWorkingProvider(userId, candidate.providerType); diff --git a/src/lib/documents/__tests__/document-summary-serialisation.test.ts b/src/lib/documents/__tests__/document-summary-serialisation.test.ts index a4be9e6b1..dc5c0e88a 100644 --- a/src/lib/documents/__tests__/document-summary-serialisation.test.ts +++ b/src/lib/documents/__tests__/document-summary-serialisation.test.ts @@ -68,7 +68,7 @@ describe("serialiseDocumentDetail — summary state", () => { expect(dto.summaryGeneratedAt).toBe("2026-02-15T09:00:00.000Z"); }); - it.each(["NONE", "PENDING", "WITHHELD", "UNAVAILABLE"] as const)( + it.each(["NONE", "WITHHELD", "UNAVAILABLE"] as const)( "passes %s through with no summary", (summaryState) => { const dto = serialiseDocumentDetail(row({ summaryState }), []); @@ -78,6 +78,30 @@ describe("serialiseDocumentDetail — summary state", () => { }, ); + it("passes a FRESH PENDING through — the job is credibly in flight", () => { + const dto = serialiseDocumentDetail( + row({ summaryState: "PENDING", updatedAt: new Date() }), + [], + ); + expect(dto.summaryState).toBe("PENDING"); + }); + + // Watched red: with the stale-PENDING branch removed from + // `serialiseDocumentDetail` (the pre-v1.37.19 serialiser), this case fails — + // a summary job dead past its retry budget left the sheet saying + // "generating" forever, with no manual action offered. + it("degrades a PENDING older than the TTL to UNAVAILABLE", () => { + const dto = serialiseDocumentDetail( + row({ + summaryState: "PENDING", + updatedAt: new Date(Date.now() - 2 * 60 * 60 * 1000), + }), + [], + ); + expect(dto.summary).toBeNull(); + expect(dto.summaryState).toBe("UNAVAILABLE"); + }); + it("degrades an undecryptable READY row to UNAVAILABLE", () => { // A rotated-away key. There is no summary to show, so the DTO must not // keep promising one — and the ciphertext is never returned either. diff --git a/src/lib/documents/store.ts b/src/lib/documents/store.ts index 15022db31..7f322a15b 100644 --- a/src/lib/documents/store.ts +++ b/src/lib/documents/store.ts @@ -46,6 +46,16 @@ export type DocumentContentCodec = (typeof DOCUMENT_CONTENT_CODECS)[number]; /** The codec every NEW upload is written with. */ export const ACTIVE_DOCUMENT_CODEC: DocumentContentCodec = "binary2"; +/** + * How long a `summaryState: "PENDING"` claim stays credible. The summary job + * retries twice within minutes; a PENDING whose row has not been touched for + * an hour is a dead promise (worker crash past the retry budget, or a + * delete-during-queue then restore). Shared by the read-time heal in + * `serialiseDocumentDetail` and the hourly persistence sweep + * (`jobs/document-summary-reaper.ts`) so the two ends cannot drift. + */ +export const SUMMARY_PENDING_TTL_MS = 60 * 60 * 1000; + /** Encrypt the raw document bytes into the `Bytes` payload the schema stores. */ export function encryptDocumentToBytes(bytes: Buffer): Uint8Array { const ciphertext = encrypt(bytes.toString("base64")); @@ -291,6 +301,17 @@ export function serialiseDocumentDetail( // READY with no ciphertext should be unreachable; trust the bytes, not the // flag, rather than promising a summary that is not there. summaryState = "UNAVAILABLE"; + } else if ( + summaryState === "PENDING" && + Date.now() - doc.updatedAt.getTime() > SUMMARY_PENDING_TTL_MS + ) { + // A PENDING older than the TTL is a dead promise — the job died past its + // retry budget and nothing will resolve it. Degrade to UNAVAILABLE so the + // sheet stops saying "generating" forever and offers the manual generate + // action instead. The hourly reaper persists the same heal + // (`jobs/document-summary-reaper.ts`); this read-time arm just makes the + // very next open honest without waiting for the tick. + summaryState = "UNAVAILABLE"; } return { ...serialiseDocument( diff --git a/src/lib/jobs/__tests__/document-summary-reaper.test.ts b/src/lib/jobs/__tests__/document-summary-reaper.test.ts new file mode 100644 index 000000000..48ee31e9a --- /dev/null +++ b/src/lib/jobs/__tests__/document-summary-reaper.test.ts @@ -0,0 +1,32 @@ +/** + * The hourly stale-PENDING summary reaper (the persistence arm of the heal + * `serialiseDocumentDetail` applies at read time). + * + * Watched red: with the TTL predicate dropped from the updateMany `where` + * (healing EVERY pending row), the cutoff assertion fails — a summary job + * legitimately in flight must never be claimed. + */ +import { describe, expect, it, vi } from "vitest"; + +import { SUMMARY_PENDING_TTL_MS } from "@/lib/documents/store"; + +import { reapStalePendingSummaries } from "../document-summary-reaper"; + +describe("reapStalePendingSummaries", () => { + it("heals only PENDING rows older than the TTL, to UNAVAILABLE", async () => { + const updateMany = vi.fn(async () => ({ count: 3 })); + const prisma = { inboundDocument: { updateMany } } as never; + const now = new Date("2026-08-14T12:00:00.000Z"); + + const healed = await reapStalePendingSummaries(prisma, now); + + expect(healed).toBe(3); + expect(updateMany).toHaveBeenCalledWith({ + where: { + summaryState: "PENDING", + updatedAt: { lt: new Date(now.getTime() - SUMMARY_PENDING_TTL_MS) }, + }, + data: { summaryState: "UNAVAILABLE" }, + }); + }); +}); diff --git a/src/lib/jobs/document-summary-reaper.ts b/src/lib/jobs/document-summary-reaper.ts new file mode 100644 index 000000000..f3e7b9362 --- /dev/null +++ b/src/lib/jobs/document-summary-reaper.ts @@ -0,0 +1,77 @@ +/** + * Hourly reaper for document-summary rows stuck on PENDING. + * + * PENDING promises the reader "a job will resolve this" — and the enqueue + * side only claims it once a pg-boss job genuinely exists. But a job can + * still die past its retry budget (worker crash, provider outage outliving + * retryLimit:2), and a soft-deleted-then-restored document can outlive the + * queue row entirely. Nothing then ever writes a terminal state, so the + * detail sheet says "wird generiert" forever and the OpenAPI-exported state + * keeps exporting the lie. + * + * Two ends close it (both against the same TTL): + * - read-time: `serialiseDocumentDetail` degrades a stale PENDING to + * UNAVAILABLE, so the very next detail open is honest; + * - this sweep: persists the same heal, so list chips, exports and any + * other reader converge too. Mirrors the orphan-ImportJob reconcile + * precedent (a stuck in-flight marker must converge to a visible + * terminal state within a couple of ticks). + * + * UNAVAILABLE is the honest terminal here — "could not produce" — and the + * detail view offers the manual generate button in that state, so the user + * regains the action a dead PENDING was hiding. A job that does eventually + * run after the heal still wins: the summary writer flips any non-READY + * state to READY when it lands. + */ +import { type Job } from "pg-boss"; +import { withBackgroundEvent } from "@/lib/logging/background"; +import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; +import { getWorkerPrisma } from "@/lib/jobs/reminder/shared"; +import { SUMMARY_PENDING_TTL_MS } from "@/lib/documents/store"; +import type { PrismaClient } from "@/generated/prisma/client"; + +export const DOCUMENT_SUMMARY_REAPER_QUEUE = "document-summary-reaper"; +/** Hourly at :50 — off the maintenance herd's minute marks. */ +export const DOCUMENT_SUMMARY_REAPER_CRON = "50 * * * *"; + +export interface DocumentSummaryReaperPayload { + triggeredAt?: string; +} + +/** + * Flip PENDING rows whose last write is older than the TTL to UNAVAILABLE. + * Returns the number of rows healed. `updatedAt` is the staleness clock — + * the PENDING claim bumps it, and any later legitimate progress does too, + * so a row still moving is never claimed. + */ +export async function reapStalePendingSummaries( + prisma: PrismaClient, + now: Date = new Date(), +): Promise { + const cutoff = new Date(now.getTime() - SUMMARY_PENDING_TTL_MS); + const { count } = await prisma.inboundDocument.updateMany({ + where: { summaryState: "PENDING", updatedAt: { lt: cutoff } }, + data: { summaryState: "UNAVAILABLE" }, + }); + return count; +} + +export async function handleDocumentSummaryReaper( + jobs: Job[], +): Promise { + void jobs; + return withBackgroundEvent("job.document_summary_reaper", async (evt) => { + const prisma = getWorkerPrisma(); + try { + const healed = await reapStalePendingSummaries(prisma); + evt.setAction({ name: "documents.summary.reap_stale_pending" }); + evt.addMeta("summary_pending_healed", healed); + // A tick with nothing stuck heals zero rows and is still a run that + // did its work — the zero is the fact, not an absence of one. + return jobDone({ summary_pending_healed: healed }); + } catch (err) { + evt.addWarning(`document-summary-reaper failed: ${err}`); + return jobFailed("document summary reaper failed", err); + } + }); +} diff --git a/src/lib/jobs/reminder/register-maintenance.ts b/src/lib/jobs/reminder/register-maintenance.ts index b12a8207f..2f29ccc4c 100644 --- a/src/lib/jobs/reminder/register-maintenance.ts +++ b/src/lib/jobs/reminder/register-maintenance.ts @@ -196,6 +196,12 @@ import { handleDocumentPurge, type DocumentPurgePayload, } from "@/lib/jobs/document-purge"; +import { + DOCUMENT_SUMMARY_REAPER_QUEUE, + DOCUMENT_SUMMARY_REAPER_CRON, + handleDocumentSummaryReaper, + type DocumentSummaryReaperPayload, +} from "@/lib/jobs/document-summary-reaper"; import { CYCLE_PREDICTION_REFRESH_QUEUE, CYCLE_PREDICTION_REFRESH_CRON, @@ -423,6 +429,11 @@ const allQueues = [ // the daily schedule silently no-ops and "deleted" documents hold backup // weight forever. DOCUMENT_PURGE_QUEUE, + // Document vault — hourly reaper for summary rows stuck on PENDING (a + // job dead past its retry budget). Without this entry the schedule + // silently no-ops and a dead "wird generiert" claim never converges to + // the honest UNAVAILABLE state. + DOCUMENT_SUMMARY_REAPER_QUEUE, // Document vault P2 — on-demand content-search index backfill. Fired by the // "index all documents" action (no cron): indexes a user's not-yet-indexed // documents via one provider transcription each, consent + budget gated. @@ -569,6 +580,10 @@ const schedules: ScheduleEntry[] = [ // Document vault — daily 04:10 Europe/Berlin purge for tombstoned // documents past the 30-day undo grace. [DOCUMENT_PURGE_QUEUE, DOCUMENT_PURGE_CRON, cronIsTheRetry], + // Document vault — hourly heal of summary rows stuck on PENDING. The next + // tick is the retry: the updateMany is idempotent and the TTL predicate + // re-selects anything a failed tick left behind. + [DOCUMENT_SUMMARY_REAPER_QUEUE, DOCUMENT_SUMMARY_REAPER_CRON, cronIsTheRetry], // v1.32.1 (issue #588) — every-15-minute orphan-ImportJob sweep. Re-runs // the same reconcile the boot path uses, so a stuck "unpacking" row // whose worker crashed/restarted without the boot-time pass catching it @@ -834,6 +849,14 @@ export async function registerMaintenanceQueues( { localConcurrency: 1 }, handleDocumentPurge, ); + // Document vault — hourly stale-PENDING summary reaper. Single-flight; + // the underlying updateMany is idempotent so a duplicate tick is a no-op. + await createAndWork( + boss, + DOCUMENT_SUMMARY_REAPER_QUEUE, + { localConcurrency: 1 }, + handleDocumentSummaryReaper, + ); await createAndWork( boss, PR_DETECTION_QUEUE,