From bdfcea5e7ce1ea3f14a1445065e95d039ba3e6eb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 25 Oct 2026 00:24:51 +0200 Subject: [PATCH] fix(insights): let "Re-run analysis" reach a benched provider The on-demand briefing (POST /api/insights/generate) runs through the briefing chokepoint, which named its chain walk "job" for the background share of the day's ceiling. Since background work leaves out a provider the health ledger benched, a person pressing "Re-run analysis" got no request at all: the call failed with a connection refusal that never happened, and a refused credential was never reported back. The chain runners now take the walk apart from the budget surface: a call can be rationed as a job and still have a person waiting on it (interactive). runBriefingCompletion takes the flag and the on-demand route's calls (the first pass and the grounding retry) pass it, so they keep the job share and walk a benched provider at the tail, as background-bench promises for interactive walks. Every other job-surface caller is a worker or is enqueued; the status, narrative and warm routes only read and queue, and document reads and lab scans resolve a single provider without a chain. Refs #1279 --- .../baselines/version-stamp-comments.json | 4 +- .../insights/generate/__tests__/route.test.ts | 32 ++++++++ .../provider-runner-background-bench.test.ts | 23 ++++++ src/lib/ai/background-bench.ts | 4 +- src/lib/ai/provider-runner.ts | 75 ++++++++++--------- src/lib/insights/briefing-provider.ts | 12 ++- .../insights/briefing-request/completion.ts | 2 + 7 files changed, 110 insertions(+), 42 deletions(-) diff --git a/src/__tests__/baselines/version-stamp-comments.json b/src/__tests__/baselines/version-stamp-comments.json index 064d48b73..c2c255493 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": 9263, + "total": 9262, "entries": { "__tests__/account-payload-consumer-guard.test.ts": 2, "__tests__/acting-account-boundary-guard.test.ts": 7, @@ -1783,7 +1783,7 @@ "lib/ai/prompts/weight.ts": 2, "lib/ai/provider-chain.ts": 1, "lib/ai/provider-health-ledger.ts": 3, - "lib/ai/provider-runner.ts": 15, + "lib/ai/provider-runner.ts": 14, "lib/ai/provider.ts": 10, "lib/ai/reasoning/__tests__/controls.test.ts": 1, "lib/ai/reasoning/controls.ts": 2, diff --git a/src/app/api/insights/generate/__tests__/route.test.ts b/src/app/api/insights/generate/__tests__/route.test.ts index bab35438f..59fd5db0a 100644 --- a/src/app/api/insights/generate/__tests__/route.test.ts +++ b/src/app/api/insights/generate/__tests__/route.test.ts @@ -71,6 +71,9 @@ vi.mock("@/lib/db", () => ({ consentReceipt: { findFirst: vi.fn(async () => ({ id: "receipt-1" })), }, + // The provider health ledger the chain walk reads. Nothing benched by + // default; the bench case below seeds a row. + providerHealth: { findMany: vi.fn(async () => []) }, }, })); @@ -284,6 +287,35 @@ describe("POST /api/insights/generate — daily token ceiling", () => { }); }); +describe("POST /api/insights/generate — a provider the health ledger benched", () => { + // #1279 — "Re-run analysis" is a person waiting on the answer. Background + // work leaves a benched provider out; this call walks it at the tail of + // the chain, so it reaches the provider and reports what the provider + // said, not a connection failure that never happened. + it("still asks the benched provider and reports its own refusal", async () => { + vi.mocked(prisma.providerHealth.findMany).mockResolvedValueOnce([ + { + providerType: "admin-openai", + lastResult: "auth_failed", + consecutiveFailures: 1, + nextRetryAt: new Date(Date.now() + 60 * 60 * 1000), + }, + ] as never); + const err = Object.assign(new Error("OpenAI request failed (401)"), { + httpStatus: 401, + }); + makeProviderThatThrows(err); + const provider = await vi.mocked(resolveProvider)("u-1"); + + const res = await POST(jsonRequest({ force: true }) as never); + + expect(provider.generateCompletion).toHaveBeenCalledTimes(1); + expect(res.status).toBe(422); + const body = (await res.json()) as ApiErrorEnvelope; + expect(body.error).toMatch(/AI provider rejected/i); + }); +}); + describe("POST /api/insights/generate — provider error mapping", () => { it("rejects a body over the 16 KB cap with 413 before parsing", async () => { const res = await POST( diff --git a/src/lib/ai/__tests__/provider-runner-background-bench.test.ts b/src/lib/ai/__tests__/provider-runner-background-bench.test.ts index 18b29e814..708c8bbac 100644 --- a/src/lib/ai/__tests__/provider-runner-background-bench.test.ts +++ b/src/lib/ai/__tests__/provider-runner-background-bench.test.ts @@ -141,6 +141,29 @@ describe("#1279: background work and a benched provider", () => { ); }); + it("walks the benched provider for a job-rationed call a person waits on", async () => { + const ledger = createInMemoryProviderHealthLedger(); + const gateway = new AnswerlessProvider(); + for (let i = 0; i < PROVIDER_PAUSE_THRESHOLD; i += 1) { + await walk("job", "u-rerun", "openai-compatible", gateway, ledger).catch( + () => undefined, + ); + } + const before = gateway.calls; + const err = await runRawCompletionWithFallback({ + userId: "u-rerun", + surface: "job", + interactive: true, + providers: [{ providerType: "openai-compatible", instance: gateway }], + params: singleUserTurn({ system: "s", user: "u" }), + ledger, + }).catch((e) => e); + expect(gateway.calls).toBe(before + 1); + expect((err as AllProvidersFailedError).attempts[0].failureKind).toBe( + "reasoning_only", + ); + }); + it("logs once more when a new window opens", async () => { vi.useFakeTimers(); const ledger = createInMemoryProviderHealthLedger(); diff --git a/src/lib/ai/background-bench.ts b/src/lib/ai/background-bench.ts index 0a5fb5614..a2b8d594f 100644 --- a/src/lib/ai/background-bench.ts +++ b/src/lib/ai/background-bench.ts @@ -7,7 +7,9 @@ * times in a row: a chain whose only entry answered every call with reasoning * and no answer was called by every background job, all day, because the tail * is still walked. So a job leaves every benched provider out until its - * window lifts, and the window grows while the failures continue + * window lifts (a job-rationed call a person waits on, the on-demand + * briefing, passes `interactive` and walks it at the tail like any + * interactive call), and the window grows while the failures continue * (`hardFailureCooldownMs` in `provider-health-ledger.ts`). The ledger is * durable, so the bench holds across workers and restarts. * diff --git a/src/lib/ai/provider-runner.ts b/src/lib/ai/provider-runner.ts index c4bd364c8..11de8b6fc 100644 --- a/src/lib/ai/provider-runner.ts +++ b/src/lib/ai/provider-runner.ts @@ -327,7 +327,7 @@ async function resolveChainOrder( userId: string, providers: ProviderChainResolved[], ledger: ProviderHealthLedger, - surface: BudgetSurface, + background: boolean, ): Promise<{ ordered: ProviderChainResolved[]; benched: Array>; @@ -344,10 +344,9 @@ async function resolveChainOrder( }); } // #1279 — background work leaves a benched provider out entirely. - const { walkable, benched } = - surface === "job" - ? partitionForBackgroundWork(providers, skips, (t) => NEVER_SKIP.has(t)) - : { walkable: providers, benched: [] }; + const { walkable, benched } = background + ? partitionForBackgroundWork(providers, skips, (t) => NEVER_SKIP.has(t)) + : { walkable: providers, benched: [] }; const afterSkips = applyHealthLedgerSkips(walkable, skips); return { ordered: applyLastWorkingCache(userId, afterSkips), benched }; } @@ -511,6 +510,29 @@ function signalFailedHops( } } +/** What both raw runners take. */ +interface ChainWalkArgs { + userId: string; + providers: ProviderChainResolved[]; + params: CompletionParams; + /** + * Which ceiling an operator-funded FALLBACK hop is rationed against. + * `"job"` gets the background share, exactly as the reservation did; + * `"coach"` gets the whole ceiling. + */ + surface: BudgetSurface; + /** + * #1279 — a person is waiting on this call although it is rationed as + * background work (the on-demand briefing). A `"job"` walk otherwise leaves + * every provider the health ledger benched out (`background-bench.ts`); + * this one walks them at the tail of the chain like any interactive call, + * so the person gets the provider's own answer or refusal. + */ + interactive?: boolean; + /** The health ledger; the durable Postgres one unless a test passes its own. */ + ledger?: ProviderHealthLedger; +} + /** * Shared chain walker for the two RAW runners (streaming + buffered). Owns the * one set of chain / fallback / health-ledger semantics — empty-chain guard, @@ -522,13 +544,7 @@ function signalFailedHops( * Schema-class failures still bubble (they are not hard) exactly as before. */ async function runRawChain( - args: { - userId: string; - providers: ProviderChainResolved[]; - params: CompletionParams; - surface: BudgetSurface; - ledger?: ProviderHealthLedger; - }, + args: ChainWalkArgs, // v1.41 — `params` reaches every candidate unchanged, `reasoning` and // `onReasoning` included. A fallback hop may see `providerState` another // provider made on an earlier round; each client re-inserts only its own @@ -549,7 +565,7 @@ async function runRawChain( userId, providers, ledger, - surface, + surface === "job" && !args.interactive, ); const hops: FallbackHop[] = []; const failedHops: HopSignal[] = []; @@ -720,21 +736,12 @@ async function runRawChain( * full reply — to confirm real tokens flowed. The returned shape matches the * buffered runner so the route is path-agnostic. */ -export async function runStreamingRawCompletionWithFallback(args: { - userId: string; - providers: ProviderChainResolved[]; - params: CompletionParams; - /** - * v1.38.19 — which ceiling an operator-funded FALLBACK - * hop is rationed against. `"job"` gets the background share, exactly as the - * reservation did; `"coach"` gets the whole ceiling. - */ - surface: BudgetSurface; - /** Called once per streamed token/chunk for providers that stream. */ - onDelta: (delta: string) => void; - /** See `RunWithFallbackParams.ledger`. */ - ledger?: ProviderHealthLedger; -}): Promise { +export async function runStreamingRawCompletionWithFallback( + args: ChainWalkArgs & { + /** Called once per streamed token/chunk for providers that stream. */ + onDelta: (delta: string) => void; + }, +): Promise { return runRawChain( args, // Prefer the provider's streaming wire; fall back to the buffered call for @@ -772,15 +779,9 @@ export interface RunRawWithFallbackResult { * last-working cache + structured logging are shared with the streaming * sibling via `runRawChain`. */ -export async function runRawCompletionWithFallback(args: { - userId: string; - providers: ProviderChainResolved[]; - params: CompletionParams; - /** See {@link runStreamingRawCompletionWithFallback}'s `surface`. */ - surface: BudgetSurface; - /** See `RunWithFallbackParams.ledger`. */ - ledger?: ProviderHealthLedger; -}): Promise { +export async function runRawCompletionWithFallback( + args: ChainWalkArgs, +): Promise { return runRawChain(args, (candidate) => candidate.instance.generateCompletion(args.params), ); diff --git a/src/lib/insights/briefing-provider.ts b/src/lib/insights/briefing-provider.ts index d8ad8aa2d..ad28340c5 100644 --- a/src/lib/insights/briefing-provider.ts +++ b/src/lib/insights/briefing-provider.ts @@ -104,6 +104,14 @@ export interface RunBriefingCompletionArgs { * budget; the client adds the thinking budget itself. */ reasoningJob?: BackgroundReasoningJob; + /** + * #1279 — a person is waiting on this call: the on-demand "Re-run + * analysis". It keeps the background share of the day's ceiling, and walks + * a provider the health ledger benched at the tail of the chain instead of + * leaving it out, so the person gets that provider's own answer or refusal. + * Absent for the scheduled generator. + */ + interactive?: boolean; } /** @@ -175,9 +183,9 @@ export async function runBriefingCompletion( try { outcome = await runRawCompletionWithFallback({ userId: args.userId, - // A background generator: an operator-funded fallback hop is rationed at - // the job share, exactly as the reservation above was. + // Rationed at the job share, exactly as the reservation above was. surface: "job", + ...(args.interactive ? { interactive: true } : {}), providers: args.chain, params: { ...singleUserTurn({ diff --git a/src/lib/insights/briefing-request/completion.ts b/src/lib/insights/briefing-request/completion.ts index 923c29720..de7bb137a 100644 --- a/src/lib/insights/briefing-request/completion.ts +++ b/src/lib/insights/briefing-request/completion.ts @@ -62,6 +62,8 @@ export function runBriefingCall( // the client's 60 s default on large accounts. timeoutMs: ctx.effectiveTimeoutMs, stage: options.stage, + // The person who pressed "Re-run analysis" is waiting on this call. + interactive: true, // A briefing asked for by hand reasons exactly like the scheduled one; // the retries repair a reply and do not. ...(options.reasoning ? { reasoningJob: "daily_briefing" as const } : {}),