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
This commit is contained in:
Marc-André Bombeck committed 2026-10-25 00:24:51 +02:00
1 parent b90980f2a2
commit bdfcea5e7c
7 files changed
+110 -42

No files matched your search

@@ -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,
@@ -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(
@@ -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();
+3 -1
View File
@@ -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.
*
+38 -37
View File
@@ -327,7 +327,7 @@ async function resolveChainOrder(
userId: string,
providers: ProviderChainResolved[],
ledger: ProviderHealthLedger,
surface: BudgetSurface,
background: boolean,
): Promise<{
ordered: ProviderChainResolved[];
benched: Array<BenchedProvider<ProviderChainResolved>>;
@@ -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<RunRawWithFallbackResult> {
export async function runStreamingRawCompletionWithFallback(
args: ChainWalkArgs & {
/** Called once per streamed token/chunk for providers that stream. */
onDelta: (delta: string) => void;
},
): Promise<RunRawWithFallbackResult> {
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<RunRawWithFallbackResult> {
export async function runRawCompletionWithFallback(
args: ChainWalkArgs,
): Promise<RunRawWithFallbackResult> {
return runRawChain(args, (candidate) =>
candidate.instance.generateCompletion(args.params),
);
+10 -2
View File
@@ -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({
@@ -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 } : {}),