Guard the operator's cap on fallback hops and reap dead summary promises

- 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.
This commit is contained in:
Marc-André Bombeck committed 2026-08-14 05:53:17 +02:00
1 parent f67fcec1c9
commit 6ebd5e8711
8 files changed
+333 -4

No files matched your search

@@ -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<typeof import("../coach/budget")>("../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");
});
});
+33 -3
View File
@@ -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<number> {
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".
*
+35
View File
@@ -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);
@@ -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.
+21
View File
@@ -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<ArrayBuffer> {
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(
@@ -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" },
});
});
});
+77
View File
@@ -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<number> {
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<DocumentSummaryReaperPayload>[],
): Promise<JobOutcome> {
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);
}
});
}
@@ -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<DocumentSummaryReaperPayload>(
boss,
DOCUMENT_SUMMARY_REAPER_QUEUE,
{ localConcurrency: 1 },
handleDocumentSummaryReaper,
);
await createAndWork<PrDetectionPayload>(
boss,
PR_DETECTION_QUEUE,