Merge remote-tracking branch 'origin/release/v1.46.0' into fix/end-medication-contributor-base

# Conflicts:
#	src/__tests__/baselines/source-file-lines.json
#	src/__tests__/baselines/source-function-lines.json
This commit is contained in:
Marc-André Bombeck committed 2026-10-24 22:39:30 +02:00
commit a35ff5f5ea
47 files changed
+2606 -447

No files matched your search

+1 -34
View File
@@ -62,6 +62,7 @@ import {
STATUS_MODEL_CALLS,
} from "@/lib/ai/ai-budgets";
import { AI_RESPONSE_TIMEOUT_MAX_SECONDS } from "@/lib/ai/effective-timeout";
import { MODEL_RUN_SITES } from "./helpers/model-run-sites";
const SRC = join(process.cwd(), "src");
const ENTRY_POINTS = [
@@ -74,40 +75,6 @@ const ENTRY_POINTS = [
"executeSupplementLabelExtraction",
];
/** Each importer of an entry point, and the queues whose jobs reach it. */
const MODEL_RUN_SITES: Readonly<Record<string, readonly string[]>> = {
"lib/insights/status-card-generation.ts": [
"insight-status-generate",
"insights-general-status",
"insight-pregenerate",
],
"lib/insights/derived/derived-assessment-ai.ts": ["insight-status-generate"],
"lib/insights/status-batch.ts": [
"insights-general-status",
"insight-pregenerate",
],
"lib/insights/narrative/period-narrative-generate.ts": [
"period-narrative-warm",
],
"lib/ai/coach/conversation-summary.ts": ["coach-memory-refresh"],
"lib/ai/coach/facts.ts": ["coach-memory-refresh"],
"lib/ai/coach/plans.ts": ["coach-memory-refresh"],
"lib/jobs/workout-insight-generate.ts": ["workout-insight-generate"],
"lib/jobs/insight-pregenerate.ts": ["insight-pregenerate"],
"lib/jobs/morning-digest-refresh.ts": ["morning-digest-refresh"],
"lib/documents/extract.ts": ["document-ai-run", "document-index"],
"lib/documents/index-document.ts": ["document-index", "document-ai-run"],
"lib/documents/ai-runs/index-run.ts": ["document-ai-run"],
"lib/documents/ai-runs/summary-run.ts": ["document-ai-run"],
"lib/documents/ai-runs/suggest-run.ts": ["document-ai-run"],
"lib/labs/ocr-extract.ts": ["document-ai-run"],
"lib/jobs/document-ai-run.ts": ["document-ai-run"],
"lib/jobs/document-content-index-backfill.ts": [
"document-content-index-backfill",
],
"lib/jobs/document-summary.ts": ["document-summary"],
};
/**
* Importers no queue reaches: a request waits on them, and the browser's
* abort follows the published setting (`clientAbortMs`).
@@ -0,0 +1,257 @@
/**
* Structural guard (#1279): every queue whose jobs reach a model retries a
* bounded number of times, with a backoff, never in a tight loop.
*
* pg-boss gives a send that names no retry options the queue's defaults:
* two retries, no delay, no backoff. A job that reaches a model and fails is
* then run again at once, twice, against a provider that just failed. So
* every place a job for such a queue is created says what it wants: each
* `send` names `retryLimit` (at most `MAX_RETRY_LIMIT`), and unless that is
* 0, `retryBackoff: true` and a `retryDelay` of at least `MIN_RETRY_DELAY_S`
* seconds; each cron carries the same in its schedule options.
*
* The queues are the ones `MODEL_RUN_SITES` attributes to a model-run entry
* point (complete by `ai-job-expiry-guard.test.ts`), plus the queues whose
* jobs call a provider instance directly (`DIRECT_CALL_QUEUES`), which this
* guard keeps complete: every file under `lib/jobs` that calls
* `generateCompletion` must be named there.
*
* Its limit: a `send` whose options come from a variable rather than an
* object literal at the call is reported, not resolved; write the literal.
* The bound on what one failure costs a record is elsewhere (the status
* note's growing window, the chain's bench for background work); this guard
* holds the queue layer.
*
* Mutation check: drop `retryBackoff` from a send in the set, raise a
* `retryLimit` above three, remove a cron's schedule options, or call
* `generateCompletion` from a new job file, and this goes red. The "plants"
* case runs the checker over sources shaped like the tree before #1279.
*/
import { join } from "node:path";
import { describe, expect, it } from "vitest";
import {
readSourceText,
stripComments,
walkSourceFiles,
} from "./helpers/source-files";
import { MODEL_RUN_SITES } from "./helpers/model-run-sites";
import { allJobDefinitions } from "@/lib/jobs/registry";
const SRC = join(process.cwd(), "src");
/** The most retries a model-bound job may ask for. */
const MAX_RETRY_LIMIT = 3;
/** The least first retry delay, in seconds, when a job retries at all. */
const MIN_RETRY_DELAY_S = 30;
/** Job files that call a provider instance directly, and their queue. */
const DIRECT_CALL_QUEUES: Readonly<Record<string, string>> = {
"lib/jobs/reaction-line.ts": "reaction-line-generate",
"lib/jobs/coach-nudge-ai.ts": "coach-nudge",
};
function modelQueues(): string[] {
return [
...new Set([
...Object.values(MODEL_RUN_SITES).flat(),
...Object.values(DIRECT_CALL_QUEUES),
]),
].sort();
}
interface SourceFile {
rel: string;
text: string;
}
function sources(): SourceFile[] {
return walkSourceFiles(SRC, { floor: 3000 })
.filter((p) => !p.startsWith("generated/"))
.filter((p) => !p.includes("__tests__"))
.filter((p) => !p.endsWith(".test.ts") && !p.endsWith(".test.tsx"))
.filter((p) => p.endsWith(".ts"))
.map((rel) => ({
rel,
text: stripComments(readSourceText(join(SRC, rel))),
}));
}
/** `export const NAME = "queue"` declarations: constant name → queue. */
function queueConstants(files: readonly SourceFile[]): Map<string, string> {
const out = new Map<string, string>();
for (const { text } of files) {
for (const m of text.matchAll(
/export const ([A-Z][A-Z0-9_]*_QUEUE)\s*=\s*"([^"]+)"/g,
)) {
out.set(m[1], m[2]);
}
}
return out;
}
/** The text of the call whose `(` sits at `open`, parentheses balanced. */
function callText(text: string, open: number): string {
let depth = 0;
for (let i = open; i < text.length; i += 1) {
if (text[i] === "(") depth += 1;
else if (text[i] === ")") {
depth -= 1;
if (depth === 0) return text.slice(open, i + 1);
}
}
return text.slice(open);
}
/** What is wrong with one set of retry options, or `null`. */
function retryProblem(options: {
retryLimit?: unknown;
retryDelay?: unknown;
retryBackoff?: unknown;
}): string | null {
const limit = options.retryLimit;
if (typeof limit !== "number") return "names no retryLimit";
if (limit > MAX_RETRY_LIMIT)
return `retryLimit ${limit} above ${MAX_RETRY_LIMIT}`;
if (limit === 0) return null;
if (options.retryBackoff !== true) return "retries without retryBackoff";
const delay = options.retryDelay;
if (typeof delay !== "number" || delay < MIN_RETRY_DELAY_S) {
return `retryDelay below ${MIN_RETRY_DELAY_S}s`;
}
return null;
}
/** The retry options an object literal in `call` spells out. */
function literalRetryOptions(call: string) {
const num = (key: string) => {
const m = new RegExp(`\\b${key}:\\s*(\\d+)`).exec(call);
return m ? Number(m[1]) : undefined;
};
const backoff = /\bretryBackoff:\s*(true|false)/.exec(call);
return {
retryLimit: num("retryLimit"),
retryDelay: num("retryDelay"),
retryBackoff: backoff ? backoff[1] === "true" : undefined,
};
}
/** Every send to a model-bound queue whose retry options fall short. */
function sendFindings(files: readonly SourceFile[], queues: readonly string[]) {
const constants = queueConstants(files);
const wanted = new Set(queues);
const problems: string[] = [];
let sends = 0;
for (const { rel, text } of files) {
for (const m of text.matchAll(/\.send(?:After)?\(\s*([A-Z][A-Z0-9_]*)/g)) {
const queue = constants.get(m[1]);
if (!queue || !wanted.has(queue)) continue;
sends += 1;
const open = text.indexOf("(", m.index);
const problem = retryProblem(literalRetryOptions(callText(text, open)));
if (problem) problems.push(`${rel}: send to ${queue} ${problem}`);
}
}
return { problems, sends };
}
/** Every cron of a model-bound queue whose schedule options fall short. */
function cronFindings(
jobs: ReadonlyArray<{
queue: string;
cron?: string;
scheduleOptions?: Readonly<Record<string, unknown>>;
}>,
queues: readonly string[],
) {
const wanted = new Set(queues);
const problems: string[] = [];
for (const job of jobs) {
if (!job.cron || !wanted.has(job.queue)) continue;
const problem = retryProblem(job.scheduleOptions ?? {});
if (problem) problems.push(`cron ${job.queue} ${problem}`);
}
return problems;
}
/** Job files that call `generateCompletion` and are not named above. */
function unnamedDirectCallers(files: readonly SourceFile[]): string[] {
return files
.filter((f) => f.rel.startsWith("lib/jobs/"))
.filter((f) => /\.generateCompletion(?:Stream)?\(/.test(f.text))
.map((f) => f.rel)
.filter((rel) => !(rel in DIRECT_CALL_QUEUES))
.sort();
}
describe("model-bound queues retry a bounded number of times, backed off", () => {
const files = sources();
const queues = modelQueues();
it("knows the model-bound queues (an empty set is a failure)", () => {
expect(queues.length).toBeGreaterThanOrEqual(12);
const declared = new Set(allJobDefinitions().map((j) => j.queue));
expect(queues.filter((q) => !declared.has(q))).toEqual([]);
});
it("names every job file that calls a provider directly", () => {
expect(unnamedDirectCallers(files)).toEqual([]);
});
it("every send to one of them names a bounded, backed-off retry", () => {
const { problems, sends } = sendFindings(files, queues);
expect(sends).toBeGreaterThanOrEqual(12);
expect(problems).toEqual([]);
});
it("every cron of one of them carries the same in its schedule options", () => {
expect(cronFindings(allJobDefinitions(), queues)).toEqual([]);
});
it("plants: sends and crons shaped like the tree before #1279 are caught", () => {
const planted: SourceFile[] = [
{
rel: "lib/jobs/planted-queue.ts",
text: 'export const PLANTED_MEMORY_QUEUE = "planted-memory";\nexport const PLANTED_CARD_QUEUE = "planted-card";',
},
{
rel: "lib/jobs/planted-send.ts",
text: [
"await boss.send(PLANTED_MEMORY_QUEUE, payload, {",
" singletonKey: `quiet:${id}`,",
" startAfter: 600,",
"});",
"await boss.send(PLANTED_CARD_QUEUE, payload, { retryLimit: 5, retryDelay: 0 });",
].join("\n"),
},
{
rel: "lib/jobs/planted-direct.ts",
text: "const r = await provider.generateCompletion(params);",
},
];
expect(
sendFindings(planted, ["planted-memory", "planted-card"]).problems,
).toEqual([
"lib/jobs/planted-send.ts: send to planted-memory names no retryLimit",
"lib/jobs/planted-send.ts: send to planted-card retryLimit 5 above 3",
]);
expect(
cronFindings(
[{ queue: "coach-nudge", cron: "15 5 * * *" }],
["coach-nudge"],
),
).toEqual(["cron coach-nudge names no retryLimit"]);
expect(unnamedDirectCallers(planted)).toEqual([
"lib/jobs/planted-direct.ts",
]);
expect(retryProblem({ retryLimit: 2, retryDelay: 60 })).toBe(
"retries without retryBackoff",
);
expect(
retryProblem({ retryLimit: 3, retryDelay: 5, retryBackoff: true }),
).toBe("retryDelay below 30s");
expect(retryProblem({ retryLimit: 0 })).toBeNull();
});
});
@@ -1,7 +1,7 @@
{
"$comment": "Production files over 800 lines and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.",
"count": 88,
"total": 98267,
"count": 87,
"total": 97455,
"entries": {
"app/medications/page-client.tsx": 839,
"components/charts/mood-chart.tsx": 1166,
@@ -64,7 +64,6 @@
"lib/insights/correlations.ts": 1026,
"lib/insights/features.ts": 1875,
"lib/insights/mood-aggregates.ts": 1053,
"lib/insights/status-cache.ts": 812,
"lib/integrations/status.ts": 1445,
"lib/jobs/coach-nudge.ts": 1084,
"lib/jobs/insight-pregenerate.ts": 1189,
@@ -1,7 +1,7 @@
{
"$comment": "Production functions over 200 lines, keyed \"file :: name\", and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.",
"count": 351,
"total": 113731,
"total": 113672,
"entries": {
"app/achievements/page.tsx :: AchievementsPage": 206,
"app/api/admin/backups/upload/route.ts :: apiHandler(arg)": 314,
@@ -254,8 +254,8 @@
"lib/ai/coach/turn/pipeline.ts :: runTurn": 274,
"lib/ai/coach/turn/reply-guards.ts :: guardReply": 331,
"lib/ai/codex-client.ts :: parseResponse": 302,
"lib/ai/local-client.ts :: generateCompletionStream": 298,
"lib/ai/openai-client.ts :: generateCompletion": 375,
"lib/ai/local-client.ts :: generateCompletionStream": 294,
"lib/ai/openai-client.ts :: generateCompletion": 335,
"lib/analytics/bp-in-target-fast-path.ts :: computeFromRollups": 203,
"lib/analytics/compliance/display.ts :: calculateCompliance": 212,
"lib/analytics/correlations-fast-path.ts :: computeCorrelationHypothesesFastPath": 256,
@@ -298,7 +298,7 @@
"lib/insights/comprehensive-generate.ts :: generateComprehensiveInsight": 876,
"lib/insights/features.ts :: extractFeatures": 796,
"lib/insights/mood-aggregates.ts :: fetchMoodAggregates": 246,
"lib/insights/status-provider.ts :: runStatusCompletion": 241,
"lib/insights/status-provider.ts :: runStatusCompletion": 226,
"lib/integrations/connection-health-load.ts :: loadConnectionFacts": 227,
"lib/integrations/status.ts :: recordSyncFailure": 292,
"lib/jobs/apple-health-import-worker.ts :: handleAppleHealthImport": 300,
@@ -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": 9273,
"total": 9272,
"entries": {
"__tests__/account-payload-consumer-guard.test.ts": 2,
"__tests__/acting-account-boundary-guard.test.ts": 7,
@@ -1751,7 +1751,7 @@
"lib/ai/local-client.ts": 18,
"lib/ai/medical-references.ts": 1,
"lib/ai/mock-client.ts": 5,
"lib/ai/openai-client.ts": 17,
"lib/ai/openai-client.ts": 16,
"lib/ai/prompts/__tests__/assessment-prompt-contract.test.ts": 5,
"lib/ai/prompts/__tests__/briefing-personalisation.test.ts": 1,
"lib/ai/prompts/__tests__/coach-prompt-locale.test.ts": 1,
+38
View File
@@ -0,0 +1,38 @@
/**
* Each file that imports a background model-run entry point, and the queues
* whose jobs reach it. Shared by the guards that hold those queues to their
* expiry (`ai-job-expiry-guard.test.ts`) and to bounded, backed-off retries
* (`ai-job-retry-bound-guard.test.ts`); the first proves the table complete.
*/
export const MODEL_RUN_SITES: Readonly<Record<string, readonly string[]>> = {
"lib/insights/status-card-generation.ts": [
"insight-status-generate",
"insights-general-status",
"insight-pregenerate",
],
"lib/insights/derived/derived-assessment-ai.ts": ["insight-status-generate"],
"lib/insights/status-batch.ts": [
"insights-general-status",
"insight-pregenerate",
],
"lib/insights/narrative/period-narrative-generate.ts": [
"period-narrative-warm",
],
"lib/ai/coach/conversation-summary.ts": ["coach-memory-refresh"],
"lib/ai/coach/facts.ts": ["coach-memory-refresh"],
"lib/ai/coach/plans.ts": ["coach-memory-refresh"],
"lib/jobs/workout-insight-generate.ts": ["workout-insight-generate"],
"lib/jobs/insight-pregenerate.ts": ["insight-pregenerate"],
"lib/jobs/morning-digest-refresh.ts": ["morning-digest-refresh"],
"lib/documents/extract.ts": ["document-ai-run", "document-index"],
"lib/documents/index-document.ts": ["document-index", "document-ai-run"],
"lib/documents/ai-runs/index-run.ts": ["document-ai-run"],
"lib/documents/ai-runs/summary-run.ts": ["document-ai-run"],
"lib/documents/ai-runs/suggest-run.ts": ["document-ai-run"],
"lib/labs/ocr-extract.ts": ["document-ai-run"],
"lib/jobs/document-ai-run.ts": ["document-ai-run"],
"lib/jobs/document-content-index-backfill.ts": [
"document-content-index-backfill",
],
"lib/jobs/document-summary.ts": ["document-summary"],
};
@@ -67,6 +67,14 @@ vi.mock("@/lib/auth/audit", () => ({
auditLog: vi.fn(),
}));
vi.mock("@/lib/ai/provider-health-ledger", () => ({
forgetProviderHealth: vi.fn(),
}));
vi.mock("@/lib/insights/status-negative-window", () => ({
clearStatusNegativeWindows: vi.fn(),
}));
import { GET, PATCH } from "../route";
import { prisma } from "@/lib/db";
import { resolveProviderAvailability } from "@/lib/ai/provider";
@@ -77,6 +85,8 @@ import { getAssistantFlags } from "@/lib/feature-flags";
import { providerCredentialPolicy } from "@/lib/sharing/provider-work-authority";
import { hasActiveConsentForSurface } from "@/lib/ai/consent-guard";
import { auditLog } from "@/lib/auth/audit";
import { forgetProviderHealth } from "@/lib/ai/provider-health-ledger";
import { clearStatusNegativeWindows } from "@/lib/insights/status-negative-window";
function patchRequest(body: unknown): Request {
return new Request("http://localhost/api/user/ai-provider", {
@@ -327,6 +337,26 @@ describe("PATCH /api/user/ai-provider — the gateway's own columns (#470)", ()
);
});
it("starts background work afresh after a change to what answers (#1279)", async () => {
// A provider benched for background work and notes backing off would
// otherwise sit out a pause the old setup earned.
vi.mocked(isPublicUrl).mockReturnValue(true);
const res = await patch(
patchRequest({
provider: "OPENAI_COMPATIBLE",
compatModel: "gpt-oss-120b",
}),
);
expect(res.status).toBe(200);
expect(vi.mocked(forgetProviderHealth).mock.calls.map((c) => c[0])).toEqual(
expect.arrayContaining([
{ userId: "u-1", providerType: "openai-compatible" },
{ userId: "u-1", providerType: "local" },
]),
);
expect(clearStatusNegativeWindows).toHaveBeenCalledWith("u-1");
});
it("refuses a private base URL the operator has not allowlisted", async () => {
vi.mocked(isPublicUrl).mockReturnValue(false);
vi.mocked(isLocalAiHostAllowed).mockReturnValue(false);
+31 -1
View File
@@ -23,9 +23,39 @@ import {
serializeProviderChain,
} from "@/lib/ai/provider-chain";
import { reasoningEffortFor } from "@/lib/ai/reasoning-effort";
import { forgetProviderHealth } from "@/lib/ai/provider-health-ledger";
import type { ProviderChainType } from "@/lib/ai/provider-chain";
import { clearStatusNegativeWindows } from "@/lib/insights/status-negative-window";
export const dynamic = "force-dynamic";
/** The chain entries this route configures (Codex signs in elsewhere). */
const USER_SETTINGS_PROVIDERS: readonly ProviderChainType[] = [
"openai",
"anthropic",
"local",
"openai-compatible",
];
/**
* Write the changed provider settings. #1279 — and start background work
* afresh: it skips a provider the health ledger benched, and a status note
* waits out its negative window, so without this a fixed setup would sit out
* a pause the old one earned.
*/
async function saveProviderSettings(
userId: string,
updates: Record<string, unknown>,
): Promise<void> {
await prisma.user.update({ where: { id: userId }, data: updates });
await Promise.all([
...USER_SETTINGS_PROVIDERS.map((providerType) =>
forgetProviderHealth({ userId, providerType }),
),
clearStatusNegativeWindows(userId),
]);
}
export const GET = apiHandler(async () => {
const { user } = await requireAuth();
annotate({ action: { name: "user.ai-provider.get" } });
@@ -389,7 +419,7 @@ export const PATCH = apiHandler(async (request: NextRequest) => {
: null;
if (Object.keys(updates).length > 0) {
await prisma.user.update({ where: { id: user.id }, data: updates });
await saveProviderSettings(user.id, updates);
}
for (const [slot, value, column] of keySlots) {
@@ -0,0 +1,212 @@
/**
* #1279 — a model behind an OpenAI-compatible gateway or a local server that
* reasons however it is asked: gpt-oss behind vLLM accepts only low / medium /
* high and ignores `reasoning_effort: "none"`, and a GLM chat template reads an
* effort it does not know as its maximum. Every reply is the thought alone,
* cut off at the token ceiling (`finish_reason: "length"`).
*
* The #1267 repeat with `none` is right for Gemma on Ollama and wasted on
* these. Driven through the real clients, the real `safeFetch` and the pinned
* dispatcher against a loopback server, these cases pin the rule: one repeat
* per call, never `none` again once a `none` reply was answerless too, the
* thinking room on every later call, and what each failed call cost carried
* on its error so the day's ledger can charge it.
*/
import http from "node:http";
import type { AddressInfo } from "node:net";
import {
afterAll,
afterEach,
beforeAll,
beforeEach,
describe,
expect,
it,
vi,
} from "vitest";
const emitSignal = vi.hoisted(() => vi.fn());
vi.mock("@/lib/logging/signal", () => ({
emitSignal: (...a: unknown[]) => emitSignal(...a),
}));
import { LocalOpenAICompatibleClient } from "../local-client";
import { OpenAIClient } from "../openai-client";
import { resetReasoningDialectCache } from "../reasoning/dialect-cache";
import { singleUserTurn, type CompletionParams } from "../types";
import { _resetPinnedDispatcherForTests } from "@/lib/safe-fetch-dispatcher";
interface SeenRequest {
reasoningEffort: unknown;
maxTokens: unknown;
}
/**
* `always`: the thought fills the ceiling whatever the effort (gpt-oss, GLM).
* `answers-unreasoned`: the same, except asked with `none` it answers
* (Gemma 4 on Ollama).
*/
let mode: "always" | "answers-unreasoned" = "always";
let server: http.Server;
let origin: string;
let seen: SeenRequest[] = [];
const ANSWER = JSON.stringify({ summary: "Steady." });
function reply(effort: unknown) {
const answers = mode === "answers-unreasoned" && effort === "none";
return {
id: "chatcmpl-1279",
object: "chat.completion",
model: "gpt-oss-120b",
choices: [
{
index: 0,
message: answers
? { role: "assistant", content: ANSWER }
: {
role: "assistant",
content: null,
reasoning_content: "We need to look at the readings. ".repeat(40),
},
finish_reason: answers ? "stop" : "length",
},
],
usage: { prompt_tokens: 2400, completion_tokens: 300, total_tokens: 2700 },
};
}
beforeAll(async () => {
server = http.createServer((req, res) => {
let raw = "";
req.on("data", (c) => (raw += c));
req.on("end", () => {
const body = JSON.parse(raw) as Record<string, unknown>;
seen.push({
reasoningEffort: body.reasoning_effort,
maxTokens: body.max_tokens ?? body.max_completion_tokens,
});
res.writeHead(200, { "Content-Type": "application/json" });
res.end(JSON.stringify(reply(body.reasoning_effort)));
});
});
await new Promise<void>((r) => server.listen(0, "127.0.0.1", r));
origin = `http://127.0.0.1:${(server.address() as AddressInfo).port}`;
});
afterAll(async () => {
_resetPinnedDispatcherForTests();
await new Promise<void>((r) => server.close(() => r()));
});
const savedOrigins = process.env.AI_PRIVATE_ORIGINS;
beforeEach(() => {
process.env.AI_PRIVATE_ORIGINS = origin;
mode = "always";
seen = [];
emitSignal.mockClear();
resetReasoningDialectCache();
});
afterEach(() => {
if (savedOrigins === undefined) delete process.env.AI_PRIVATE_ORIGINS;
else process.env.AI_PRIVATE_ORIGINS = savedOrigins;
});
function gateway() {
return new OpenAIClient({
apiKey: "",
model: "gpt-oss-120b",
baseUrl: `${origin}/v1`,
providerType: "openai-compatible",
});
}
function local() {
return new LocalOpenAICompatibleClient({
baseUrl: `${origin}/v1`,
model: "gpt-oss-120b",
});
}
/** A status card call as `runStatusCompletion` shapes it: no level. */
function statusParams(): CompletionParams {
return singleUserTurn({
system: "You write the status note.",
user: "Snapshot: …",
maxTokens: 300,
timeoutMs: 30_000,
responseFormat: "json",
});
}
function actions(): string[] {
return emitSignal.mock.calls.map((c) => (c[0] as { action: string }).action);
}
async function failures(
client: { generateCompletion(p: CompletionParams): Promise<unknown> },
calls: number,
) {
const errors: Array<Record<string, unknown>> = [];
for (let i = 0; i < calls; i += 1) {
errors.push(
(await client
.generateCompletion(statusParams())
.catch((e) => e)) as Record<string, unknown>,
);
}
return errors;
}
describe.each([
["the OpenAI-compatible gateway", gateway, "ai.gateway"],
["the Local client", local, "ai.local"],
] as const)(
"%s, a model that reasons whatever it is asked",
(_, make, prefix) => {
it("repeats one call once, then never sends `none` again", async () => {
const errors = await failures(make(), 5);
expect(seen.map((s) => s.reasoningEffort)).toEqual([
undefined,
"none",
undefined,
undefined,
undefined,
undefined,
]);
for (const err of errors) {
expect(err).toBeInstanceOf(Error);
expect(err.kind).toBe("reasoning_only");
expect(err.httpStatus).toBe(0);
}
// One line when the repeat is learned, one when it is learned useless,
// nothing per call after that.
expect(actions()).toEqual([
`${prefix}.reasoning_answerless`,
`${prefix}.reasoning_unswitchable`,
]);
});
it("leaves the thinking room on every call after the first", async () => {
await failures(make(), 3);
expect(seen.map((s) => s.maxTokens)).toEqual([300, 1324, 1324, 1324]);
});
it("carries what a failed call cost on its error, both requests included", async () => {
const [first, second] = await failures(make(), 2);
expect(first.tokensUsed).toBe(5400);
expect(second.tokensUsed).toBe(2700);
});
it("still repeats with `none` for a model that answers unreasoned, and bills both requests", async () => {
mode = "answers-unreasoned";
const result = await make().generateCompletion(statusParams());
expect(result.content).toBe(ANSWER);
expect(result.tokensUsed).toBe(5400);
expect(seen.map((s) => s.reasoningEffort)).toEqual([undefined, "none"]);
});
},
);
@@ -0,0 +1,207 @@
/**
* #1279 — background work and a provider the health ledger has benched.
*
* A chain whose only entry answered every call with reasoning and no answer
* was called by every background job: the ledger counted the failures and
* paused the provider, but a paused provider only moved to the tail of the
* chain, and the tail is still walked. A job now leaves a benched provider
* out (the local floor only once it is paused), says so once per window, and
* a person's own call still reaches it and reports its error.
*/
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
const emitSignal = vi.hoisted(() => vi.fn());
vi.mock("@/lib/logging/signal", () => ({
emitSignal: (...a: unknown[]) => emitSignal(...a),
}));
import {
singleUserTurn,
type AIProvider,
type CompletionParams,
type CompletionResult,
} from "../types";
import {
AllProvidersFailedError,
clearLastWorkingProviderCache,
runRawCompletionWithFallback,
} from "../provider-runner";
import {
HARD_FAILURE_SKIP_THRESHOLD,
PROVIDER_PAUSE_THRESHOLD,
createInMemoryProviderHealthLedger,
} from "../provider-health-ledger";
import type { ProviderChainType } from "../provider-chain";
/** Answers every call with reasoning and no answer, as the clients report it. */
class AnswerlessProvider implements AIProvider {
readonly type: AIProvider["type"];
calls = 0;
constructor(type: AIProvider["type"] = "openai-compatible") {
this.type = type;
}
async generateCompletion(_: CompletionParams): Promise<CompletionResult> {
this.calls += 1;
throw Object.assign(new Error("returned empty content"), {
httpStatus: 0,
kind: "reasoning_only",
tokensUsed: 2700,
});
}
}
function walk(
surface: "job" | "coach",
userId: string,
providerType: ProviderChainType,
instance: AIProvider,
ledger: ReturnType<typeof createInMemoryProviderHealthLedger>,
) {
return runRawCompletionWithFallback({
userId,
surface,
providers: [{ providerType, instance }],
params: singleUserTurn({ system: "s", user: "u" }),
ledger,
});
}
function signals(action: string) {
return emitSignal.mock.calls
.map((c) => c[0] as { action: string; meta: Record<string, unknown> })
.filter((s) => s.action === action);
}
beforeEach(() => {
clearLastWorkingProviderCache();
emitSignal.mockClear();
});
afterEach(() => {
vi.useRealTimers();
});
describe("#1279: background work and a benched provider", () => {
it("stops calling the only provider once the ledger benches it, for jobs", async () => {
const ledger = createInMemoryProviderHealthLedger();
const gateway = new AnswerlessProvider();
for (let i = 0; i < HARD_FAILURE_SKIP_THRESHOLD; i += 1) {
await expect(
walk("job", "u-bench", "openai-compatible", gateway, ledger),
).rejects.toBeInstanceOf(AllProvidersFailedError);
}
expect(gateway.calls).toBe(HARD_FAILURE_SKIP_THRESHOLD);
// Ten more jobs inside the window: none of them sends a request.
for (let i = 0; i < 10; i += 1) {
const err = await walk(
"job",
"u-bench",
"openai-compatible",
gateway,
ledger,
).catch((e) => e);
expect(err).toBeInstanceOf(AllProvidersFailedError);
expect((err as AllProvidersFailedError).attempts).toEqual([
expect.objectContaining({
providerType: "openai-compatible",
failureKind: "paused",
httpStatus: null,
}),
]);
}
expect(gateway.calls).toBe(HARD_FAILURE_SKIP_THRESHOLD);
// One line for the window, not one per job.
expect(signals("ai.chain.background_paused")).toHaveLength(1);
expect(signals("ai.chain.background_paused")[0].meta).toMatchObject({
provider: "openai-compatible",
reason: "backoff",
});
});
it("still sends a person's own call to the benched provider and reports its error", async () => {
const ledger = createInMemoryProviderHealthLedger();
const gateway = new AnswerlessProvider();
for (let i = 0; i < PROVIDER_PAUSE_THRESHOLD; i += 1) {
await walk("job", "u-coach", "openai-compatible", gateway, ledger).catch(
() => undefined,
);
}
const before = gateway.calls;
const err = await walk(
"coach",
"u-coach",
"openai-compatible",
gateway,
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();
const gateway = new AnswerlessProvider();
for (let i = 0; i < HARD_FAILURE_SKIP_THRESHOLD; i += 1) {
await walk("job", "u-window", "openai-compatible", gateway, ledger).catch(
() => undefined,
);
}
await walk("job", "u-window", "openai-compatible", gateway, ledger).catch(
() => undefined,
);
// The five-minute backoff lifts; the next job tries, fails, re-arms.
vi.advanceTimersByTime(6 * 60 * 1000);
await walk("job", "u-window", "openai-compatible", gateway, ledger).catch(
() => undefined,
);
await walk("job", "u-window", "openai-compatible", gateway, ledger).catch(
() => undefined,
);
expect(gateway.calls).toBe(HARD_FAILURE_SKIP_THRESHOLD + 1);
expect(signals("ai.chain.background_paused")).toHaveLength(2);
});
it("keeps the local floor through a short backoff and leaves it out once paused", async () => {
const ledger = createInMemoryProviderHealthLedger();
const local = new AnswerlessProvider("local");
for (let i = 0; i < HARD_FAILURE_SKIP_THRESHOLD; i += 1) {
await walk("job", "u-local", "local", local, ledger).catch(
() => undefined,
);
}
// Backoff: the floor is still called.
await walk("job", "u-local", "local", local, ledger).catch(() => undefined);
expect(local.calls).toBe(HARD_FAILURE_SKIP_THRESHOLD + 1);
while (local.calls < PROVIDER_PAUSE_THRESHOLD) {
await walk("job", "u-local", "local", local, ledger).catch(
() => undefined,
);
}
// Paused: background work leaves it alone.
await walk("job", "u-local", "local", local, ledger).catch(() => undefined);
await walk("job", "u-local", "local", local, ledger).catch(() => undefined);
expect(local.calls).toBe(PROVIDER_PAUSE_THRESHOLD);
});
it("sums what the failed hops reported spending", async () => {
const ledger = createInMemoryProviderHealthLedger();
const err = await runRawCompletionWithFallback({
userId: "u-spend",
surface: "job",
providers: [
{
providerType: "openai-compatible",
instance: new AnswerlessProvider(),
},
{ providerType: "local", instance: new AnswerlessProvider("local") },
],
params: singleUserTurn({ system: "s", user: "u" }),
ledger,
}).catch((e) => e);
expect((err as AllProvidersFailedError).tokensSpent).toBe(5400);
});
});
+84
View File
@@ -0,0 +1,84 @@
/**
* #1279 — background work and a provider the health ledger has benched.
*
* An interactive walk keeps a benched provider at the tail of the chain, so a
* person never loses their only provider to a cooldown. Background work has
* nobody waiting and no reason to call a provider that just failed several
* 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
* (`hardFailureCooldownMs` in `provider-health-ledger.ts`). The ledger is
* durable, so the bench holds across workers and restarts.
*
* The local floor is left out only once it is paused, past a brief backoff:
* it costs no money, but a model that never answers still occupies the
* machine it runs on.
*/
import { emitSignal } from "@/lib/logging/signal";
import type { ProviderChainType } from "./provider-chain";
import type { ProviderSkipHint } from "./provider-health-ledger";
export interface BenchedProvider<C> {
candidate: C;
hint: ProviderSkipHint;
}
/**
* Split a chain into what background work walks and what it leaves out.
* `isFloor` names the providers that are never benched for a person (the
* local model).
*/
export function partitionForBackgroundWork<
C extends { providerType: ProviderChainType },
>(
providers: readonly C[],
skips: ReadonlyMap<ProviderChainType, ProviderSkipHint>,
isFloor: (type: ProviderChainType) => boolean,
): { walkable: C[]; benched: Array<BenchedProvider<C>> } {
const walkable: C[] = [];
const benched: Array<BenchedProvider<C>> = [];
for (const candidate of providers) {
const hint = skips.get(candidate.providerType);
const leftOut =
hint !== undefined &&
(!isFloor(candidate.providerType) || hint.reason === "paused");
if (leftOut) benched.push({ candidate, hint });
else walkable.push(candidate);
}
return { walkable, benched };
}
/**
* One `ai.chain.background_paused` line per bench window, per process: the
* first background job that finds the provider benched says so, the rest
* mark it on their own event only. Keyed by user and provider, holding the
* window's end, so a new window logs once more.
*/
const LOGGED_SLOT = Symbol.for("healthlog.ai.backgroundBenchLogged");
const logged: Map<string, number> = ((globalThis as Record<symbol, unknown>)[
LOGGED_SLOT
] ??= new Map()) as Map<string, number>;
export function signalBackgroundBench(
userId: string,
providerType: ProviderChainType,
hint: ProviderSkipHint,
): void {
const key = `${userId}\u0000${providerType}`;
const until = hint.retryAt?.getTime() ?? 0;
if (logged.get(key) === until) return;
logged.set(key, until);
emitSignal({
action: "ai.chain.background_paused",
level: "warn",
meta: {
provider: providerType,
reason: hint.reason,
retry_in_minutes:
until > 0
? Math.max(0, Math.ceil((until - Date.now()) / 60_000))
: null,
},
});
}
@@ -34,6 +34,10 @@ describe("enqueueCoachMemoryRefresh", () => {
singletonSeconds: COACH_MEMORY_QUIET_MS / 1000,
startAfter: COACH_MEMORY_QUIET_MS / 1000,
expireInSeconds: COACH_MEMORY_REFRESH_EXPIRE_SECONDS,
// #1279 — bounded, backed-off retries instead of pg-boss's two at once.
retryLimit: 2,
retryDelay: 60,
retryBackoff: true,
});
expect(COACH_MEMORY_QUIET_MS).toBe(30 * 60_000);
});
@@ -251,7 +251,7 @@ describe("#1276: a gateway model that always reasons", () => {
});
it.each<Parser>(["fielded", "template-opened", "raw"])(
"keeps room to reason after one answerless reply taught it `none` (%s)",
"keeps room to reason after the `none` repeat came back answerless too (%s)",
async (mode) => {
parser = mode;
// A long round reasons past the whole ceiling once; the #1267 rule
@@ -281,8 +281,10 @@ describe("#1276: a gateway model that always reasons", () => {
expect(out.result.content).toBe(ANSWER);
expect(executeCoachTool).toHaveBeenCalledTimes(1);
for (const body of bodies()) {
expect(body.reasoning_effort).toBe("none");
// `none` is a request, not a promise: the thinking room stays.
// #1279 — the `none` repeat came back answerless too, so the model
// is not asked with `none` again: the turn goes out at its own level.
expect(body.reasoning_effort).toBe("medium");
// The thinking room stays.
expect(body.max_tokens).toBe(600 + 4096);
}
},
+5
View File
@@ -85,6 +85,11 @@ export async function enqueueCoachMemoryRefresh(
singletonSeconds: quietSeconds,
startAfter: Math.max(1, Math.round(delayMs / 1000)),
expireInSeconds: COACH_MEMORY_REFRESH_EXPIRE_SECONDS,
// #1279 — pg-boss's queue default retried a failed pass twice at
// once; the same two retries now wait a minute, then longer.
retryLimit: 2,
retryDelay: 60,
retryBackoff: true,
},
);
annotate({
+5
View File
@@ -40,6 +40,11 @@ export type ProviderFailureKind =
| "embedded_error"
/** A non-2xx HTTP status. */
| "http"
/**
* #1279 — not sent: background work waits out a provider the health
* ledger has benched after failures in a row (`provider-runner.ts`).
*/
| "paused"
/** A transport failure with no code this list knows. */
| "network"
| "unknown";
+38 -38
View File
@@ -25,7 +25,6 @@ import {
type JsonModeDialect,
} from "./json-dialect";
import type { ReasoningEffort } from "./reasoning-effort";
import { emitSignal } from "@/lib/logging/signal";
import {
ANSWERS_ONLY_UNREASONED,
annotateReasoningDowngrade,
@@ -34,6 +33,11 @@ import {
rememberReasoningDialect,
} from "./reasoning/dialect-cache";
import { uncontrolledThinkingRoom } from "./reasoning/levels";
import {
askAgainUnreasoned,
reasonsUnasked,
withEarlierSpend,
} from "./reasoning/answerless";
import {
ReasoningCollector,
separateMessage,
@@ -108,7 +112,7 @@ function withStrictJsonPrefix(messages: AiMessage[]): AiMessage[] {
* treats as a hard failure; `kind` says whether the reply held reasoning and
* no answer, or nothing at all, so the log names it instead of "network".
*/
function emptyReplyError(hadReasoning: boolean): Error {
function emptyReplyError(hadReasoning: boolean, spent: number | null): Error {
const err = new Error(
hadReasoning
? "Local AI returned reasoning but no answer"
@@ -118,6 +122,8 @@ function emptyReplyError(hadReasoning: boolean): Error {
httpStatus: 0,
kind: hadReasoning ? "reasoning_only" : "empty_response",
upstream: "local",
// #1279 — what the reply cost, for the day's ledger to charge.
...(spent !== null ? { tokensUsed: spent } : {}),
});
return err;
}
@@ -204,8 +210,7 @@ export class LocalOpenAICompatibleClient implements AIProvider {
// room once the model has shown it reasons unasked.
const thinkingTokens = params.reasoning
? uncontrolledThinkingRoom(params.reasoning.effort)
: learnedReasoningDialect(this.reasoningDialectKey()) ===
ANSWERS_ONLY_UNREASONED
: reasonsUnasked(this.reasoningDialectKey())
? uncontrolledThinkingRoom("off")
: 0;
const body = JSON.stringify({
@@ -298,34 +303,32 @@ export class LocalOpenAICompatibleClient implements AIProvider {
* "none"` answers in `content`. So when a request that did not already say
* `none` comes back empty with reasoning (or after asking to reason), learn
* that this endpoint and model answer only unreasoned, and tell the caller
* to ask once more. Returns false when the empty reply is final.
* to ask once more (#1279: at most once, and never again for a model whose
* `none` reply was answerless too). False when the empty reply is final.
*/
private learnAnswerlessReasoning(
params: CompletionParams,
hadReasoning: boolean,
finishReason: string | undefined,
): boolean {
const sent = this.sentReasoningEffort(params);
if (sent === "none") return false;
if (!hadReasoning && sent === null) return false;
rememberReasoningDialect(
this.reasoningDialectKey(),
ANSWERS_ONLY_UNREASONED,
);
annotateReasoningDowngrade("local", sent ?? "default", "none");
emitSignal({
action: "ai.local.reasoning_answerless",
level: "warn",
meta: {
effort: sent ?? "default",
finish:
finishReason === "length" || finishReason === "stop"
? finishReason
: "other",
reasoning: hadReasoning,
},
return askAgainUnreasoned({
key: this.reasoningDialectKey(),
provider: "local",
signals: "ai.local",
sent: this.sentReasoningEffort(params),
hadReasoning,
finishReason,
});
return true;
}
/** #1279 — the streamed call once more, carrying this reply's spend. */
private streamAgain(
params: CompletionParams,
onDelta: (delta: string) => void,
spent: number | null,
): Promise<CompletionResult> {
const again = this.generateCompletionStream(params, onDelta);
return withEarlierSpend(again, spent);
}
async generateCompletion(
@@ -470,8 +473,9 @@ export class LocalOpenAICompatibleClient implements AIProvider {
room,
);
if (!content) {
const spent = json.usage?.total_tokens ?? null;
if (this.learnAnswerlessReasoning(params, hadReasoning, finish)) {
return this.generateCompletion(params);
return withEarlierSpend(this.generateCompletion(params), spent);
}
// The reasoning that did arrive still reaches `onReasoning`, as before.
collector.result(
@@ -480,7 +484,7 @@ export class LocalOpenAICompatibleClient implements AIProvider {
// v1.20.1 — sentinel httpStatus + kind so the chain classifier can tell
// an empty 200-OK reply apart from a transport failure. Cascade unchanged
// (`status <= 0` is already a hard failure).
throw emptyReplyError(hadReasoning);
throw emptyReplyError(hadReasoning, spent);
}
const reasoning = collector.result(
json.usage?.completion_tokens_details?.reasoning_tokens ?? null,
@@ -597,17 +601,13 @@ export class LocalOpenAICompatibleClient implements AIProvider {
room,
);
if (!content) {
if (
this.learnAnswerlessReasoning(
params,
hadReasoning,
json?.choices?.[0]?.finish_reason,
)
) {
return this.generateCompletionStream(params, onDelta);
const spent = json?.usage?.total_tokens ?? null;
const finish = json?.choices?.[0]?.finish_reason;
if (this.learnAnswerlessReasoning(params, hadReasoning, finish)) {
return this.streamAgain(params, onDelta, spent);
}
collector.result(null);
throw emptyReplyError(hadReasoning);
throw emptyReplyError(hadReasoning, spent);
}
const reasoning = collector.result(null);
onDelta(content);
@@ -804,10 +804,10 @@ export class LocalOpenAICompatibleClient implements AIProvider {
const hadReasoning =
streamedReasoning || endState === "inside" || tail.took;
if (this.learnAnswerlessReasoning(params, hadReasoning, finishReason)) {
return this.generateCompletionStream(params, onDelta);
return this.streamAgain(params, onDelta, tokensUsed);
}
collector.result(reasoningTokens);
throw emptyReplyError(hadReasoning);
throw emptyReplyError(hadReasoning, tokensUsed);
}
const reasoning = collector.result(reasoningTokens);
+60 -48
View File
@@ -20,6 +20,11 @@ import {
learnedReasoningDialect,
rememberReasoningDialect,
} from "./reasoning/dialect-cache";
import {
askAgainUnreasoned,
reasonsUnasked,
withEarlierSpend,
} from "./reasoning/answerless";
import {
REASONING_THINKING_BUDGET,
uncontrolledThinkingRoom,
@@ -434,53 +439,13 @@ export class OpenAIClient implements AIProvider {
// A reply with tool calls and no prose is valid (F1 tool loop); only an
// empty reply with neither content NOR tool calls is an error.
if (!content && !toolCalls) {
// #1267 — a gateway in front of a thinking model (Ollama serving
// Gemma 4, say) can answer with the thought in `reasoning` and no
// content. Asked with `none`, the same model answers; learn that once
// and ask again. Same rule as the Local client.
const sentEffort =
reasoningWire.kind !== "none" ? reasoningWire.effort : entryEffort;
const hadReasoning = reply.hadReasoning;
if (
this.isGateway &&
sentEffort !== "none" &&
(hadReasoning || sentEffort !== null)
) {
rememberReasoningDialect(
this.reasoningDialectKey(),
ANSWERS_ONLY_UNREASONED,
);
annotateReasoningDowngrade(this.type, sentEffort ?? "default", "none");
emitSignal({
action: "ai.gateway.reasoning_answerless",
level: "warn",
meta: {
effort: sentEffort ?? "default",
finish:
choice?.finish_reason === "length" ||
choice?.finish_reason === "stop"
? choice.finish_reason
: "other",
reasoning: hadReasoning,
},
});
return this.generateCompletion(params);
}
// v1.20.1 — tag the empty-reply throw with a sentinel `httpStatus: 0` +
// `kind` so the provider-chain classifier can tell an empty 200-OK reply
// apart from a genuine transport failure (ECONNRESET, DNS, timeout) in
// observability. The cascade is unchanged: `isHardProviderFailure`
// already treats `status <= 0` as a hard failure and walks to the next
// provider.
const err = new Error(
`${this.isGateway ? "OpenAI-compatible gateway" : "OpenAI"} returned empty content`,
);
Object.assign(err, {
httpStatus: 0,
kind: hadReasoning ? "reasoning_only" : "empty_response",
upstream: this.isGateway ? "openai-compatible" : "openai",
return this.answerless(params, {
sent:
reasoningWire.kind !== "none" ? reasoningWire.effort : entryEffort,
hadReasoning: reply.hadReasoning,
finishReason: choice?.finish_reason,
spent: json.usage?.total_tokens ?? null,
});
throw err;
}
const summary = reply.reasoning ? [reply.reasoning.trim()] : [];
@@ -538,6 +503,54 @@ export class OpenAIClient implements AIProvider {
};
}
/**
* A reply with neither an answer nor tool calls. #1267 — a gateway in front
* of a thinking model (Ollama serving Gemma 4, say) can answer with the
* thought in `reasoning` and no content; asked with `none`, the same model
* answers, so that is learned once and the call asked again. #1279 — once,
* and never `none` again for a model the repeat showed ignores it. Same
* rule as the Local client (`reasoning/answerless.ts`).
*
* Otherwise the empty reply is thrown, tagged with a sentinel
* `httpStatus: 0` and its `kind` (v1.20.1) so the chain classifier can tell
* it apart from a transport failure (ECONNRESET, DNS, timeout). The cascade
* is unchanged: `isHardProviderFailure` treats `status <= 0` as hard. It
* carries what the reply cost, for the day's ledger to charge.
*/
private answerless(
params: CompletionParams,
reply: {
sent: string | null;
hadReasoning: boolean;
finishReason: string | null | undefined;
spent: number | null;
},
): Promise<CompletionResult> {
if (
this.isGateway &&
askAgainUnreasoned({
key: this.reasoningDialectKey(),
provider: this.type,
signals: "ai.gateway",
sent: reply.sent,
hadReasoning: reply.hadReasoning,
finishReason: reply.finishReason,
})
) {
return withEarlierSpend(this.generateCompletion(params), reply.spent);
}
const err = new Error(
`${this.isGateway ? "OpenAI-compatible gateway" : "OpenAI"} returned empty content`,
);
Object.assign(err, {
httpStatus: 0,
kind: reply.hadReasoning ? "reasoning_only" : "empty_response",
upstream: this.isGateway ? "openai-compatible" : "openai",
...(reply.spent !== null ? { tokensUsed: reply.spent } : {}),
});
throw err;
}
// ── v1.41 reasoning ───────────────────────────────────────────────────────
/**
@@ -561,8 +574,7 @@ export class OpenAIClient implements AIProvider {
if (params.reasoning) {
return uncontrolledThinkingRoom(params.reasoning.effort);
}
return learnedReasoningDialect(this.reasoningDialectKey()) ===
ANSWERS_ONLY_UNREASONED
return reasonsUnasked(this.reasoningDialectKey())
? uncontrolledThinkingRoom("off")
: 0;
}
+65 -6
View File
@@ -13,6 +13,12 @@ import {
type ProviderFailure,
type ProviderFailureKind,
} from "./failure-kind";
import { spentTokensOf } from "./reasoning/answerless";
import {
partitionForBackgroundWork,
signalBackgroundBench,
type BenchedProvider,
} from "./background-bench";
import {
isOperatorFundedProvider,
readDailySpend,
@@ -182,9 +188,15 @@ export interface FallbackHop {
httpStatus: number | null;
/**
* #1267 — what the call ran into (`failure-kind.ts`). Absent on a hop that
* never called the provider (the operator cost cap).
* never called the provider (the operator cost cap); `paused` on one that
* background work skipped (#1279).
*/
failureKind?: ProviderFailureKind;
/**
* #1279 — tokens the failed call spent upstream, when its reply said so
* (an answer that never came still billed its reasoning).
*/
tokensSpent?: number;
}
/**
@@ -205,6 +217,12 @@ export class AllProvidersFailedError extends Error {
* the thing that broke.
*/
readonly primaryCredentialExpired: boolean;
/**
* #1279 — tokens the failed hops spent upstream, summed. A caller that
* reserved budget charges this rather than refunding the whole reservation:
* a model that reasons without answering is paid for every time.
*/
readonly tokensSpent: number;
constructor(attempts: FallbackHop[]) {
super(
@@ -214,6 +232,10 @@ export class AllProvidersFailedError extends Error {
);
this.name = "AllProvidersFailedError";
this.attempts = attempts;
this.tokensSpent = attempts.reduce(
(sum, a) => sum + (a.tokensSpent ?? 0),
0,
);
const first = attempts[0];
this.primaryCredentialExpired =
first !== undefined &&
@@ -305,7 +327,11 @@ async function resolveChainOrder(
userId: string,
providers: ProviderChainResolved[],
ledger: ProviderHealthLedger,
): Promise<ProviderChainResolved[]> {
surface: BudgetSurface,
): Promise<{
ordered: ProviderChainResolved[];
benched: Array<BenchedProvider<ProviderChainResolved>>;
}> {
const skips = await ledger.getSkipHints(userId);
if (skips.size > 0) {
annotate({
@@ -317,8 +343,13 @@ async function resolveChainOrder(
},
});
}
const afterSkips = applyHealthLedgerSkips(providers, skips);
return applyLastWorkingCache(userId, afterSkips);
// #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 afterSkips = applyHealthLedgerSkips(walkable, skips);
return { ordered: applyLastWorkingCache(userId, afterSkips), benched };
}
function summariseError(e: unknown): {
@@ -514,10 +545,36 @@ async function runRawChain(
throw new AllProvidersFailedError([]);
}
const ordered = await resolveChainOrder(userId, providers, ledger);
const { ordered, benched } = await resolveChainOrder(
userId,
providers,
ledger,
surface,
);
const hops: FallbackHop[] = [];
const failedHops: HopSignal[] = [];
// #1279 — background work skips a provider the ledger benched; no request
// is sent. The skip is recorded after the walked hops, so a chain left
// with nothing fails as `AllProvidersFailedError` like any other, and it
// is logged once per window.
const benchedHops: FallbackHop[] = benched.map(({ candidate }, i) => ({
providerType: candidate.providerType,
attempt: ordered.length + i + 1,
failureReason: "benched-for-background-work",
httpStatus: null,
failureKind: "paused",
}));
for (const { candidate, hint } of benched) {
annotate({
meta: {
ai_chain_background_benched: candidate.providerType,
ai_chain_background_benched_reason: hint.reason,
},
});
signalBackgroundBench(userId, candidate.providerType, hint);
}
for (let i = 0; i < ordered.length; i += 1) {
const candidate = ordered[i];
// v1.37.19 — hop-time operator-cost guard. A request whose
@@ -621,12 +678,14 @@ async function runRawChain(
...failure,
});
void ledger.recordFailure(userId, candidate.providerType, summary.status);
const spent = spentTokensOf(error);
const hop: FallbackHop = {
providerType: candidate.providerType,
attempt: i + 1,
failureReason: summary.reason,
httpStatus: summary.status,
failureKind: failure.failure,
...(spent !== null ? { tokensSpent: spent } : {}),
};
hops.push(hop);
annotate({
@@ -648,7 +707,7 @@ async function runRawChain(
},
});
signalFailedHops(failedHops, false, surface);
throw new AllProvidersFailedError(hops);
throw new AllProvidersFailedError([...hops, ...benchedHops]);
}
/**
+168
View File
@@ -0,0 +1,168 @@
/**
* #1279 — what the Local and the OpenAI-compatible gateway clients do with a
* reply that reached the model and came back without an answer.
*
* A reply with reasoning and no answer is asked once more with
* `reasoning_effort: "none"` (#1267): Ollama serves Gemma 4 that way, and the
* repeat answers. The lesson holds for the hour (`ANSWERS_ONLY_UNREASONED`).
*
* When the repeat, or any later call sent as `none`, comes back the same way,
* `none` did nothing for this model: gpt-oss accepts only low / medium / high,
* and a GLM chat template reads an effort it does not know as its maximum
* (vllm-project/vllm#54744). That is learned too (`REASONS_REGARDLESS`): later
* calls go out as the call asks, with the thinking room #1276 keeps, and an
* answerless reply is final. So one call costs two requests at most, and
* after the first such call, one.
*
* What bounds the calls themselves is not here: the chain keeps background
* work off a provider that keeps failing (`provider-runner.ts`), and each job
* backs off the record it failed on.
*/
import { emitSignal } from "@/lib/logging/signal";
import {
ANSWERS_ONLY_UNREASONED,
REASONS_REGARDLESS,
annotateReasoningDowngrade,
learnedReasoningDialect,
rememberReasoningDialect,
type ReasoningDialectKey,
} from "./dialect-cache";
/**
* What the client does next with an answerless reply.
* - `ask_unreasoned`: learned that the model answers only unreasoned; send
* the call once more (it now goes out as `none`).
* - `final`: throw the empty reply. `learnedRegardless` is true when this
* reply taught that `none` does not help.
*/
export type AnswerlessStep =
{ kind: "ask_unreasoned" } | { kind: "final"; learnedRegardless: boolean };
/**
* The next step for a reply with no answer. `sent` is the effort the request
* carried (`null`: no field); `hadReasoning` whether the reply reasoned.
*/
export function nextAnswerlessStep(
key: ReasoningDialectKey,
sent: string | null,
hadReasoning: boolean,
): AnswerlessStep {
if (learnedReasoningDialect(key) === REASONS_REGARDLESS) {
return { kind: "final", learnedRegardless: false };
}
if (sent === "none") {
if (!hadReasoning) return { kind: "final", learnedRegardless: false };
rememberReasoningDialect(key, REASONS_REGARDLESS);
return { kind: "final", learnedRegardless: true };
}
if (!hadReasoning && sent === null) {
return { kind: "final", learnedRegardless: false };
}
rememberReasoningDialect(key, ANSWERS_ONLY_UNREASONED);
return { kind: "ask_unreasoned" };
}
/**
* The clients' whole answerless step: decide, mark the downgrade, and write
* the one line each lesson gets (`<signals>.reasoning_answerless` when the
* repeat is learned, `<signals>.reasoning_unswitchable` when it is learned
* useless), never one per call. True when the call goes out once more.
*/
export function askAgainUnreasoned(args: {
key: ReasoningDialectKey;
/** The provider tag for the downgrade mark (`local`, `openai-compatible`). */
provider: string;
signals: "ai.local" | "ai.gateway";
sent: string | null;
hadReasoning: boolean;
finishReason: string | null | undefined;
}): boolean {
const finish =
args.finishReason === "length" || args.finishReason === "stop"
? args.finishReason
: "other";
const step = nextAnswerlessStep(args.key, args.sent, args.hadReasoning);
if (step.kind === "final") {
if (step.learnedRegardless) {
emitSignal({
action: `${args.signals}.reasoning_unswitchable`,
level: "warn",
meta: { finish },
});
}
return false;
}
annotateReasoningDowngrade(args.provider, args.sent ?? "default", "none");
emitSignal({
action: `${args.signals}.reasoning_answerless`,
level: "warn",
meta: {
effort: args.sent ?? "default",
finish,
reasoning: args.hadReasoning,
},
});
return true;
}
/**
* True when the model has shown it reasons without being asked to, so a call
* without a level of its own gets the thinking room (#1276).
*/
export function reasonsUnasked(key: ReasoningDialectKey): boolean {
const learned = learnedReasoningDialect(key);
return learned === ANSWERS_ONLY_UNREASONED || learned === REASONS_REGARDLESS;
}
/**
* The repeat's outcome with the first request's tokens added: they were spent
* upstream whether the repeat answers or fails too, and the day's ledger
* charges what a call cost, not its last request.
*/
export async function withEarlierSpend<T extends { tokensUsed: number | null }>(
repeat: Promise<T>,
earlier: number | null,
): Promise<T> {
if (earlier === null || earlier <= 0) return repeat;
try {
const result = await repeat;
return {
...result,
tokensUsed: (result.tokensUsed ?? 0) + earlier,
};
} catch (err) {
addSpentTokens(err, earlier);
throw err;
}
}
/**
* Tokens a failed call spent upstream, as its error carries them
* (`tokensUsed`), or `null` when it carries none.
*/
export function spentTokensOf(err: unknown): number | null {
const value = (err as { tokensUsed?: unknown } | null)?.tokensUsed;
return typeof value === "number" && Number.isFinite(value) && value > 0
? value
: null;
}
/** Add `tokens` to what the error `err` says its call spent. */
export function addSpentTokens(err: unknown, tokens: number | null): void {
if (tokens === null || tokens <= 0) return;
if (err === null || typeof err !== "object") return;
Object.assign(err, { tokensUsed: (spentTokensOf(err) ?? 0) + tokens });
}
/**
* Tokens a failed provider chain reported spending upstream
* (`AllProvidersFailedError.tokensSpent`), or 0 when it reported none. What a
* caller that reserved budget charges instead of a full refund: a model that
* reasons without answering is billed for every call.
*/
export function spentOnFailure(err: unknown): number {
const value = (err as { tokensSpent?: unknown } | null)?.tokensSpent;
return typeof value === "number" && Number.isFinite(value) && value > 0
? value
: 0;
}
+24 -1
View File
@@ -47,10 +47,33 @@ export interface ReasoningDialectKey {
*/
export const ANSWERS_ONLY_UNREASONED = "off";
/**
* #1279 — the learned value for an endpoint and model that answered with
* reasoning alone even when asked with `none`: gpt-oss behind vLLM accepts
* only low / medium / high and ignores it, and a GLM chat template reads an
* effort it does not know as its maximum. Asking such a model with `none` is
* a wasted call, or a longer thought. Both clients send what the call asks
* for (with the thinking room) and never repeat an answerless reply while
* this holds.
*/
export const REASONS_REGARDLESS = "unswitchable";
/** How long a learned downgrade holds before the full request is tried again. */
export const DIALECT_TTL_MS = 60 * 60 * 1000;
const learned = new Map<string, { value: string; expiresAt: number }>();
/**
* Pinned on `globalThis`: a module imported from more than one bundler layer
* is instantiated once per layer (see `caches` in `server-cache.ts`), so the
* Coach's route handlers and the worker's jobs each kept a copy and paid for
* the same lesson separately. One map per process.
*/
const LEARNED_SLOT = Symbol.for("healthlog.ai.reasoningDialects");
const learned: Map<string, { value: string; expiresAt: number }> = ((
globalThis as Record<symbol, unknown>
)[LEARNED_SLOT] ??= new Map()) as Map<
string,
{ value: string; expiresAt: number }
>;
function cacheKey(key: ReasoningDialectKey): string {
return `${key.provider}\u0000${key.endpoint}\u0000${key.model}`;
+15 -5
View File
@@ -45,6 +45,7 @@ import {
resolveDailyCap,
type BudgetCostOwner,
} from "@/lib/ai/coach/budget";
import { spentTokensOf } from "@/lib/ai/reasoning/answerless";
import {
loadOwnedDocument,
prepareVisionInput,
@@ -289,13 +290,22 @@ async function tryProviderIndex(
outcome: { indexed: true, source: "vision", tokenCount },
note: null,
};
} catch {
} catch (err) {
// Refund the reservation and let the caller fall through to local — a
// transient provider miss must never leave a text-layer PDF unsearchable.
await reconcileSpend(userId, reservation.reserved, 0, dateKey, 0, {
servedBy: null,
reservedOwner: reservation.owner,
});
// #1279 — what the reply reported spending stays charged: a model that
// reasons without answering bills every call.
await reconcileSpend(
userId,
reservation.reserved,
spentTokensOf(err) ?? 0,
dateKey,
0,
{
servedBy: null,
reservedOwner: reservation.owner,
},
);
return { outcome: null, note: "provider-error" };
}
}
@@ -185,6 +185,17 @@ describe("runBriefingCompletion — ledger accounting", () => {
expect(ledgerTotal).toBe(0);
});
it("charges what a failed chain reported spending (#1279)", async () => {
runRawCompletionWithFallback.mockRejectedValue(
Object.assign(new Error("chain exhausted"), { tokensSpent: 5400 }),
);
await expect(runBriefingCompletion(completionArgs())).rejects.toThrow(
"chain exhausted",
);
expect(ledgerTotal).toBe(5400);
});
it("re-throws the provider error unchanged so callers keep their mapping", async () => {
class AllProvidersFailedError extends Error {}
runRawCompletionWithFallback.mockRejectedValue(
@@ -15,6 +15,7 @@
import { describe, it, expect, vi, beforeEach } from "vitest";
const deleteMany = vi.fn();
const noteUpdateMany = vi.fn();
const noteFindMany = vi.fn();
const aiCapabilityForRecord = vi.fn();
const enqueueStatusGeneration = vi.fn();
@@ -39,6 +40,7 @@ vi.mock("@/lib/db", () => ({
insightStatusCache: {
deleteMany: (...a: unknown[]) => deleteMany(...a),
findMany: (...a: unknown[]) => noteFindMany(...a),
updateMany: (...a: unknown[]) => noteUpdateMany(...a),
},
user: { findUnique: (...a: unknown[]) => userFindUnique(...a) },
appSettings: {
@@ -316,11 +318,11 @@ describe("invalidateStatusInsightsForTypes", () => {
expect(enqueueStatusGeneration).not.toHaveBeenCalled();
});
it("counts only a stored note as fresh, so a recently stalled scope still refreshes", async () => {
// A row that carries only a negative-cache window has no note, so a
// scope that recently stalled must retry rather than be debounced into
// staying cold. The probe asks the database for noted rows only; with
// none returned, every dirtied scope refreshes.
it("counts only a stored note as fresh, so a stalled scope refreshes once its window lapses", async () => {
// A row that carries only a lapsed negative-cache window has no note,
// so a scope that stalled must retry rather than be debounced into
// staying cold. The first probe asks for noted rows only; with none
// returned, every dirtied scope refreshes.
await invalidateStatusInsightsForTypes("u1", ["WEIGHT"], {
source: "manual",
});
@@ -329,6 +331,23 @@ describe("invalidateStatusInsightsForTypes", () => {
expect(enqueuedScopes()).toEqual(["bmi", "general", "weight"]);
});
it("leaves a scope whose negative window is still open to wait it out (#1279)", async () => {
// Every sync batch used to re-enqueue a scope whose generation had just
// failed, however often it failed. The second probe asks for open
// windows; a scope it returns is not enqueued.
noteFindMany.mockImplementation(
async (args: { where: { retryAt?: { gt: Date } } }) =>
args.where.retryAt ? [freshRow("weight"), freshRow("general")] : [],
);
const before = Date.now();
await invalidateStatusInsightsForTypes("u1", ["WEIGHT"], {
source: "manual",
});
const probe = noteFindMany.mock.calls[1][0].where;
expect(probe.retryAt.gt.getTime()).toBeGreaterThanOrEqual(before - 1000);
expect(enqueuedScopes()).toEqual(["bmi"]);
});
it("scopes the freshness probe to the user's resolved locale and a 1 h gap cutoff", async () => {
userFindUnique.mockResolvedValue({ locale: "en" });
const before = Date.now();
@@ -394,6 +413,20 @@ describe("enqueueStatusRefillForUser", () => {
expect([...locales]).toEqual(["de"]);
});
it("closes the user's negative windows first: the person asked (#1279)", async () => {
await enqueueStatusRefillForUser("u1", "en");
expect(noteUpdateMany).toHaveBeenCalledWith({
where: {
userId: "u1",
OR: [{ retryAt: { not: null } }, { failureStreak: { gt: 0 } }],
},
data: { retryAt: null, negativeReason: null, failureStreak: 0 },
});
expect(noteUpdateMany.mock.invocationCallOrder[0]).toBeLessThan(
enqueueStatusGeneration.mock.invocationCallOrder[0],
);
});
it("never deletes cache rows (the hash baselines survive the refill)", async () => {
await enqueueStatusRefillForUser("u1", "en");
expect(deleteMany).not.toHaveBeenCalled();
@@ -141,7 +141,9 @@ describe("generatePulseStatusForUser — provider timeout fallback", () => {
expect(persisted.create.textEncrypted).toBeUndefined();
expect(persisted.create.negativeReason).toBe("timeout");
// The update arm moves only the window, never an existing note.
// #1279 — and the streak that sizes the next window.
expect(Object.keys(persisted.update).sort()).toEqual([
"failureStreak",
"negativeReason",
"retryAt",
]);
@@ -53,7 +53,6 @@ import {
readStatusNegativeCache,
refreshUnchangedStatusInsight,
resolveReadOnlyStatusMiss,
writeStatusNegativeWindow,
writeStatusNote,
} from "../status-cache";
@@ -100,7 +99,7 @@ beforeEach(() => {
probeProviderPresence.mockResolvedValue(true);
});
describe("writeStatusNote / writeStatusNegativeWindow", () => {
describe("writeStatusNote", () => {
it("upserts one encrypted row per (user, metric, locale) and clears the negative window", async () => {
vi.mocked(prisma.insightStatusCache.upsert).mockResolvedValue({} as never);
await writeStatusNote({
@@ -128,21 +127,8 @@ describe("writeStatusNote / writeStatusNegativeWindow", () => {
expect(arg.update.snapshotHash).toBe("s".repeat(64));
expect(arg.update.retryAt).toBeNull();
expect(arg.update.negativeReason).toBeNull();
});
it("opens a negative window without touching the stored note", async () => {
const retryAt = new Date(Date.now() + 60_000);
await writeStatusNegativeWindow({
userId: "u1",
cacheAction: "insights.pulse-status.de",
todayKey: TODAY,
reason: "timeout",
retryAt,
});
const arg = vi.mocked(prisma.insightStatusCache.upsert).mock
.calls[0][0] as { update: Record<string, unknown> };
// Only the window moves: a stall must not hide yesterday's text.
expect(arg.update).toEqual({ retryAt, negativeReason: "timeout" });
// #1279 — a note ends the run of failures that sized its window.
expect(arg.update.failureStreak).toBe(0);
});
});
@@ -0,0 +1,160 @@
/**
* #1279 — the negative-cache window of a status note grows with the
* generations in a row that ended without a note, is not moved or counted
* by a failure inside it, resets when the person asks again, and is what the
* provider chokepoint reads before any generation.
*/
import { beforeEach, describe, expect, it, vi } from "vitest";
const findUnique = vi.fn();
const upsert = vi.fn();
const updateMany = vi.fn();
vi.mock("@/lib/db", () => ({
prisma: {
insightStatusCache: {
findUnique: (...a: unknown[]) => findUnique(...a),
upsert: (...a: unknown[]) => upsert(...a),
updateMany: (...a: unknown[]) => updateMany(...a),
},
},
}));
import {
STATUS_NEGATIVE_WINDOW_BASE_MS,
clearStatusNegativeWindows,
statusNegativeWindowMs,
statusNoteCoolingDown,
writeStatusNegativeWindow,
} from "../status-negative-window";
const TODAY = "2026-05-31";
const MINUTE = 60 * 1000;
interface UpsertArgs {
where: { userId_metric_locale: Record<string, string> };
create: Record<string, unknown>;
update: { retryAt: Date; negativeReason: string; failureStreak: number };
}
function written(): UpsertArgs {
return upsert.mock.calls[0][0] as UpsertArgs;
}
beforeEach(() => {
vi.resetAllMocks();
});
describe("writeStatusNegativeWindow", () => {
it("opens a five-minute window without touching the stored note", async () => {
findUnique.mockResolvedValue(null);
const before = Date.now();
await writeStatusNegativeWindow({
userId: "u1",
cacheAction: "insights.pulse-status.de",
todayKey: TODAY,
reason: "timeout",
});
expect(written().where.userId_metric_locale).toEqual({
userId: "u1",
metric: "pulse",
locale: "de",
});
// Only the window moves: a stall must not hide yesterday's text.
expect(Object.keys(written().update).sort()).toEqual([
"failureStreak",
"negativeReason",
"retryAt",
]);
expect(written().update.negativeReason).toBe("timeout");
expect(written().update.failureStreak).toBe(1);
const opened = written().update.retryAt.getTime() - before;
expect(opened).toBeGreaterThanOrEqual(STATUS_NEGATIVE_WINDOW_BASE_MS);
expect(opened).toBeLessThan(STATUS_NEGATIVE_WINDOW_BASE_MS + 5000);
});
it("grows the window with each failure in a row", async () => {
findUnique.mockResolvedValue({
retryAt: new Date(Date.now() - 1000),
failureStreak: 3,
});
const before = Date.now();
await writeStatusNegativeWindow({
userId: "u1",
cacheAction: "insights.metric:HEART_RATE-status.en",
todayKey: TODAY,
reason: "error",
});
expect(written().update.failureStreak).toBe(4);
expect(written().update.retryAt.getTime() - before).toBeGreaterThanOrEqual(
40 * MINUTE,
);
});
it("neither moves nor counts a window that is still open", async () => {
findUnique.mockResolvedValue({
retryAt: new Date(Date.now() + MINUTE),
failureStreak: 2,
});
await writeStatusNegativeWindow({
userId: "u1",
cacheAction: "insights.weight-status.en",
todayKey: TODAY,
reason: "error",
});
expect(upsert).not.toHaveBeenCalled();
});
it("ignores a name that is not a status note", async () => {
await writeStatusNegativeWindow({
userId: "u1",
cacheAction: "workouts.insight",
todayKey: TODAY,
reason: "error",
});
expect(findUnique).not.toHaveBeenCalled();
expect(upsert).not.toHaveBeenCalled();
});
});
describe("statusNegativeWindowMs", () => {
it("doubles from five minutes to a two-hour ceiling", () => {
expect([1, 2, 3, 4, 5, 6, 7, 30].map(statusNegativeWindowMs)).toEqual(
[5, 10, 20, 40, 80, 120, 120, 120].map((m) => m * MINUTE),
);
});
});
describe("statusNoteCoolingDown", () => {
it("is true only while the window is open, and reads nothing for other names", async () => {
findUnique.mockResolvedValueOnce({
retryAt: new Date(Date.now() + MINUTE),
});
expect(await statusNoteCoolingDown("u1", "insights.weight-status.en")).toBe(
true,
);
findUnique.mockResolvedValueOnce({ retryAt: new Date(Date.now() - 1) });
expect(await statusNoteCoolingDown("u1", "insights.weight-status.en")).toBe(
false,
);
expect(await statusNoteCoolingDown("u1", "workouts.insight")).toBe(false);
expect(findUnique).toHaveBeenCalledTimes(2);
});
});
describe("clearStatusNegativeWindows", () => {
it("closes every open window and streak of one record", async () => {
await clearStatusNegativeWindows("u1");
expect(updateMany).toHaveBeenCalledWith({
where: {
userId: "u1",
OR: [{ retryAt: { not: null } }, { failureStreak: { gt: 0 } }],
},
data: { retryAt: null, negativeReason: null, failureStreak: 0 },
});
});
it("never throws: the windows lapse on their own", async () => {
updateMany.mockRejectedValue(new Error("pool exhausted"));
await expect(clearStatusNegativeWindows("u1")).resolves.toBeUndefined();
});
});
@@ -20,6 +20,8 @@ import { describe, it, expect, vi, beforeEach } from "vitest";
let ledgerTotal = 0;
/** …and of `coach_usage.operator_tokens`. */
let ledgerOperator = 0;
/** The status note row the cooling-down gate reads. */
let noteWindow: { retryAt: Date | null } | null = null;
vi.mock("@/lib/db", () => ({
prisma: {
@@ -55,6 +57,10 @@ vi.mock("@/lib/db", () => ({
user: {
findUnique: vi.fn(async () => ({ aiResponseTimeoutSeconds: null })),
},
// #1279 — the note's negative window, read before any provider work.
insightStatusCache: {
findUnique: vi.fn(async () => noteWindow),
},
},
}));
@@ -126,6 +132,7 @@ beforeEach(() => {
vi.clearAllMocks();
ledgerTotal = 0;
ledgerOperator = 0;
noteWindow = null;
resolveProviderChain.mockResolvedValue(OPERATOR_CHAIN);
resolveProvider.mockResolvedValue({ type: "none" });
aiCapabilityForRecord.mockResolvedValue({
@@ -312,6 +319,39 @@ describe("runStatusCompletion — ledger accounting", () => {
expect(result.kind).toBe("error");
expect(ledgerTotal).toBe(0);
});
it("asks no provider about a note inside its negative window (#1279)", async () => {
noteWindow = { retryAt: new Date(Date.now() + 60_000) };
const result = await runStatusCompletion(
completionArgs({ cacheAction: "insights.weight-status.en" }),
);
expect(result.kind).toBe("error");
expect(runRawCompletionWithFallback).not.toHaveBeenCalled();
expect(ledgerTotal).toBe(0);
// The window lapsed: the note is generated again.
noteWindow = { retryAt: new Date(Date.now() - 1) };
mockProviderReply(100);
const again = await runStatusCompletion(
completionArgs({ cacheAction: "insights.weight-status.en" }),
);
expect(again.kind).toBe("ok");
});
it("charges what a failed chain reported spending (#1279)", async () => {
// A model that reasons and never answers is billed for every call; a
// full refund let background work spend without the ceiling seeing it.
runRawCompletionWithFallback.mockRejectedValue(
Object.assign(new Error("All 1 configured providers failed"), {
tokensSpent: 5400,
}),
);
const result = await runStatusCompletion(completionArgs());
expect(result.kind).toBe("error");
expect(ledgerTotal).toBe(5400);
});
});
describe("runStatusCompletion — cost owner decides the ceiling", () => {
@@ -41,14 +41,18 @@ vi.mock("@/lib/db", () => ({
async (args: {
where: {
OR: Array<{ metric: string; locale: string }>;
generatedAt: { gte: Date };
generatedAt?: { gte: Date };
};
}) =>
args.where.OR.filter(
}) => {
// The second probe asks for open negative windows; every
// generation in this simulation succeeds, so there are none.
const since = args.where.generatedAt?.gte;
if (!since) return [];
return args.where.OR.filter(
(key) =>
(sim.notes.get(key.metric) ?? -Infinity) >=
args.where.generatedAt.gte.getTime(),
).map((key) => ({ metric: key.metric, locale: key.locale })),
(sim.notes.get(key.metric) ?? -Infinity) >= since.getTime(),
).map((key) => ({ metric: key.metric, locale: key.locale }));
},
),
},
},
+18 -8
View File
@@ -35,6 +35,7 @@ import {
type ProviderChainResolved,
type RunRawWithFallbackResult,
} from "@/lib/ai/provider-runner";
import { spentOnFailure } from "@/lib/ai/reasoning/answerless";
import {
buildDateKey,
reconcileSpend,
@@ -195,14 +196,23 @@ export async function runBriefingCompletion(
},
});
} catch (err) {
// No usable reply: refund the whole reservation rather than bill an
// invented figure. A partially-burned upstream call is possible here, but
// we have no reported count to charge, and over-charging a failed
// generation would ration the retry the user is entitled to.
await reconcileSpend(args.userId, reservation.reserved, 0, dateKey, 0, {
servedBy: null,
reservedOwner: reservation.owner,
}).catch(() => {
// No usable reply: refund the reservation rather than bill an invented
// figure. A partially-burned upstream call is possible here, but without
// a reported count there is nothing to charge, and over-charging a
// failed generation would ration the retry the user is entitled to.
// #1279 — what the hops did report spending is charged: a model that
// reasons without answering bills every call.
await reconcileSpend(
args.userId,
reservation.reserved,
spentOnFailure(err),
dateKey,
0,
{
servedBy: null,
reservedOwner: reservation.owner,
},
).catch(() => {
// Best-effort: a failed refund leaves the conservative reservation in
// place (never an undercount) and must not mask the provider error.
});
@@ -18,6 +18,12 @@ vi.mock("@/lib/insights/status-cache", () => ({
gateUnchangedStatusInput: vi.fn().mockResolvedValue(null),
}));
const persistTimeoutNegativeStub = vi.fn();
vi.mock("@/lib/insights/timeout-fallback", () => ({
persistTimeoutNegativeStub: (...a: unknown[]) =>
persistTimeoutNegativeStub(...a),
}));
const findUnique = vi.fn();
vi.mock("@/lib/db", () => ({
prisma: { user: { findUnique: (a: unknown) => findUnique(a) } },
@@ -156,3 +162,25 @@ describe("generateDerivedScoreAssessment — per-user tz label (QA F5)", () => {
expect(capturedPrompt).not.toContain("(Europe/Berlin)");
});
});
describe("generateDerivedScoreAssessment — a failed generation (#1279)", () => {
it("opens the note's negative window, so the next sync does not ask again", async () => {
vi.mocked(runStatusCompletion).mockResolvedValue({ kind: "error" });
const result = await generateDerivedScoreAssessment({
metric: "RECOVERY_SCORE",
userId: "u-ny",
derived: okDerived(),
locale: "en",
now: new Date("2026-06-20T12:00:00.000Z"),
});
expect(result).toMatchObject({ kind: "error", retryable: true });
expect(persistTimeoutNegativeStub).toHaveBeenCalledWith({
userId: "u-ny",
cacheAction: "insights.derived-score:RECOVERY_SCORE-status.en",
todayKey: "2026-06-20",
reason: "error",
});
});
});
@@ -37,6 +37,7 @@ import {
type AssessmentStatus,
} from "@/lib/insights/status-shared";
import { runStatusCompletion } from "@/lib/insights/status-provider";
import { persistTimeoutNegativeStub } from "@/lib/insights/timeout-fallback";
import {
readFreshStatusText,
resolveReadOnlyStatusMiss,
@@ -223,6 +224,32 @@ export async function resolveDerivedAssessment(args: {
// ── generate (worker entry) ──────────────────────────────────────────────
/**
* A generation that timed out or failed. #1279 — it opens the same negative
* window every other note opens, so the score is not asked about again on
* the next sync while it keeps failing.
*/
async function failedAssessment(
userId: string,
cacheAction: string,
todayKey: string,
kind: "timeout" | "error",
): Promise<AssessmentStatus> {
await persistTimeoutNegativeStub({
userId,
cacheAction,
todayKey,
reason: kind,
});
return {
kind,
text: null,
hasProvider: true,
updatedAt: null,
retryable: true,
};
}
/**
* Worker entry: build the signal-led prompt for one derived score, run the
* provider chain, and persist the AI assessment under the shared status-cache
@@ -341,13 +368,7 @@ export async function generateDerivedScoreAssessment(args: {
retryable: false,
};
}
return {
kind: outcome.kind,
text: null,
hasProvider: true,
updatedAt: null,
retryable: true,
};
return failedAssessment(args.userId, cacheAction, todayKey, outcome.kind);
}
// Parse and screen through the same fail-closed chokepoint as every other
+24 -40
View File
@@ -84,7 +84,7 @@ export function statusCacheAction(scope: string, locale: string): string {
const CACHE_ACTION_SHAPE = /^insights\.(.+)-status\.([^.]+)$/;
/** The row key behind a note's name, or `null` for a name of another shape. */
function statusCacheKey(
export function statusCacheKey(
cacheAction: string,
): { metric: string; locale: string } | null {
const match = CACHE_ACTION_SHAPE.exec(cacheAction);
@@ -200,6 +200,7 @@ export async function writeStatusNote(args: {
generatedAt,
retryAt: null,
negativeReason: null,
failureStreak: 0,
};
await prisma.insightStatusCache.upsert({
where: { userId_metric_locale: { userId: args.userId, ...key } },
@@ -230,38 +231,13 @@ async function redateStatusNote(
generatedAt,
retryAt: null,
negativeReason: null,
failureStreak: 0,
},
});
}
return generatedAt;
}
/**
* Open a negative-cache window for a note: the generation stalled, erred, or
* was screened. Leaves any stored note exactly as it was.
*/
export async function writeStatusNegativeWindow(args: {
userId: string;
cacheAction: string;
todayKey: string;
reason: "timeout" | "error" | "screened";
retryAt: Date;
}): Promise<void> {
const key = statusCacheKey(args.cacheAction);
if (!key) return;
await prisma.insightStatusCache.upsert({
where: { userId_metric_locale: { userId: args.userId, ...key } },
create: {
userId: args.userId,
...key,
dateKey: args.todayKey,
retryAt: args.retryAt,
negativeReason: args.reason,
},
update: { retryAt: args.retryAt, negativeReason: args.reason },
});
}
export interface FreshStatusCacheHit {
kind: "generated";
text: string;
@@ -635,10 +611,12 @@ export async function readPreviousStatusNote(args: {
}
/**
* Which of `cacheActions` carry a note written at or after `since`. The
* ingest invalidator skips those, so a fresh note survives a sync drip.
* Which of `cacheActions` are not due for another generation: a note written
* at or after `since`, or (#1279) an open negative-cache window. The ingest
* invalidator skips those, so a fresh note survives a sync drip and a note
* whose generation keeps failing is not selected again by every sync.
*/
export async function statusNotesWrittenSince(args: {
export async function statusNotesNotDue(args: {
userId: string;
cacheActions: readonly string[];
since: Date;
@@ -653,17 +631,23 @@ export async function statusNotesWrittenSince(args: {
const [metric, locale] = k.split("\u0000");
return { metric, locale };
});
const rows = await prisma.insightStatusCache.findMany({
where: {
userId: args.userId,
OR: keys,
generatedAt: { gte: args.since },
textEncrypted: { not: null },
},
select: { metric: true, locale: true },
});
const [written, coolingDown] = await Promise.all([
prisma.insightStatusCache.findMany({
where: {
userId: args.userId,
OR: keys,
generatedAt: { gte: args.since },
textEncrypted: { not: null },
},
select: { metric: true, locale: true },
}),
prisma.insightStatusCache.findMany({
where: { userId: args.userId, OR: keys, retryAt: { gt: new Date() } },
select: { metric: true, locale: true },
}),
]);
const fresh = new Set<string>();
for (const row of rows) {
for (const row of [...written, ...coolingDown]) {
const action = byKey.get(`${row.metric}\u0000${row.locale}`);
if (action) fresh.add(action);
}
+9 -5
View File
@@ -22,8 +22,9 @@ import {
} from "@/lib/insights/status-shared";
import {
statusCacheAction,
statusNotesWrittenSince,
statusNotesNotDue,
} from "@/lib/insights/status-cache";
import { clearStatusNegativeWindows } from "@/lib/insights/status-negative-window";
import {
metricIdForMeasurementType,
metricStatusScope,
@@ -275,6 +276,9 @@ export async function enqueueStatusRefillForUser(
});
return 0;
}
// #1279 — the person asked for fresh notes: a note backing off after
// failed generations is asked again now, not when its window lapses.
await clearStatusNegativeWindows(userId);
const scopes = new Set<InsightStatusScope>(PER_STATUS_SCOPES);
try {
// v1.42 — a loose index scan over the live-rows index (`live-types.ts`)
@@ -301,9 +305,9 @@ export async function enqueueStatusRefillForUser(
* invalidator so a fresh assessment survives the sync drip.
*
* One indexed read per user answers it for every candidate scope at once.
* Only a stored note counts: a row that carries nothing but a negative-cache
* window is not fresh, so a scope that recently stalled still gets a retry
* enqueued.
* #1279 — a scope inside an open negative-cache window counts too: it used to
* be re-enqueued by every sync batch however often its generation had just
* failed. Once the window lapses the next sync enqueues it again.
*/
async function findRecentlyWarmedScopes(
userId: string,
@@ -321,7 +325,7 @@ async function findRecentlyWarmedScopes(
actionToScope.set(statusCacheAction(scope, locale), scope);
}
const written = await statusNotesWrittenSince({
const written = await statusNotesNotDue({
userId,
cacheActions: candidateActions,
since: cutoff,
+113
View File
@@ -0,0 +1,113 @@
/**
* The negative-cache window of a status note: how long a note whose
* generation stalled, erred or was screened waits before it is generated
* again.
*
* The window used to be a fixed five minutes that only the read path
* honoured. #1279 — a model that never answers then failed every five
* minutes per note, and a note the next data sync selected again ignored the
* window altogether. The window now grows with the failures in a row
* (`failureStreak`), and every generation waits it out at the provider
* chokepoint (`runStatusCompletion`), whichever path selected the note.
*/
import { prisma } from "@/lib/db";
import { statusCacheKey } from "@/lib/insights/status-cache";
/** The first negative-cache window after a generation that left no note. */
export const STATUS_NEGATIVE_WINDOW_BASE_MS = 5 * 60 * 1000;
/** The longest window: a note is tried again at least this often. */
export const STATUS_NEGATIVE_WINDOW_MAX_MS = 2 * 60 * 60 * 1000;
/**
* The window after `streak` generations in a row ended without a note:
* five minutes, doubling per further failure, up to two hours (#1279).
*/
export function statusNegativeWindowMs(streak: number): number {
const doublings = Math.min(Math.max(streak - 1, 0), 16);
return Math.min(
STATUS_NEGATIVE_WINDOW_BASE_MS * 2 ** doublings,
STATUS_NEGATIVE_WINDOW_MAX_MS,
);
}
/**
* Open a negative-cache window for a note: the generation stalled, erred, or
* was screened. Leaves any stored note exactly as it was.
*
* #1279 — the window grows with the failures in a row (`failureStreak`,
* `statusNegativeWindowMs`), so a model that never answers is asked about a
* note less and less often instead of every five minutes. A failure while a
* window is still open (a generation that was refused for it, or one that
* raced another) neither moves it nor counts.
*/
export async function writeStatusNegativeWindow(args: {
userId: string;
cacheAction: string;
todayKey: string;
reason: "timeout" | "error" | "screened";
}): Promise<void> {
const key = statusCacheKey(args.cacheAction);
if (!key) return;
const where = { userId_metric_locale: { userId: args.userId, ...key } };
const row = await prisma.insightStatusCache.findUnique({
where,
select: { retryAt: true, failureStreak: true },
});
const now = Date.now();
if (row?.retryAt && row.retryAt.getTime() > now) return;
const failureStreak = (row?.failureStreak ?? 0) + 1;
const retryAt = new Date(now + statusNegativeWindowMs(failureStreak));
await prisma.insightStatusCache.upsert({
where,
create: {
userId: args.userId,
...key,
dateKey: args.todayKey,
retryAt,
negativeReason: args.reason,
failureStreak,
},
update: { retryAt, negativeReason: args.reason, failureStreak },
});
}
/**
* True while the note's negative-cache window is open. #1279 — checked right
* before a status generation reaches a provider (`runStatusCompletion`), so
* no path that selects the note again (the next data sync, a warm pass, a
* queued job) can ask about it inside the window.
*/
export async function statusNoteCoolingDown(
userId: string,
cacheAction: string,
): Promise<boolean> {
const key = statusCacheKey(cacheAction);
if (!key) return false;
const row = await prisma.insightStatusCache.findUnique({
where: { userId_metric_locale: { userId, ...key } },
select: { retryAt: true },
});
return row?.retryAt != null && row.retryAt.getTime() > Date.now();
}
/**
* Close every open window on one record's notes, streaks included: the person
* asked for fresh notes (the regenerate action) or changed what answers them.
* Best-effort: a failed write leaves windows that lapse on their own.
*/
export async function clearStatusNegativeWindows(
userId: string,
): Promise<void> {
try {
await prisma.insightStatusCache.updateMany({
where: {
userId,
OR: [{ retryAt: { not: null } }, { failureStreak: { gt: 0 } }],
},
data: { retryAt: null, negativeReason: null, failureStreak: 0 },
});
} catch {
// The windows lapse by themselves.
}
}
+54 -35
View File
@@ -12,6 +12,7 @@ import {
runRawCompletionWithFallback,
type ProviderChainResolved,
} from "@/lib/ai/provider-runner";
import { spentOnFailure } from "@/lib/ai/reasoning/answerless";
import { aiEgressRefusal } from "@/lib/ai/capabilities/egress";
import { annotate } from "@/lib/logging/context";
import {
@@ -35,6 +36,7 @@ import {
type BackgroundReasoningJob,
} from "@/lib/ai/reasoning/levels";
import { resolveJobReasoning } from "@/lib/ai/reasoning/controls";
import { statusNoteCoolingDown } from "./status-negative-window";
/**
* v1.41 — the fallback ceiling for a status-path call that reasons. Thinking
@@ -191,6 +193,17 @@ export async function runStatusCompletion(
return { kind: "none", reason: "no_provider" };
}
// #1279 — a note whose last generations ended without one waits out its
// window here, whichever path selected it again (a data sync, a warm pass,
// a queued job). Reported as `error`, the transient miss, like a stall.
if (await statusNoteCoolingDown(userId, cacheAction)) {
annotate({
action: { name: "insights.status.cooling_down" },
meta: { cacheAction },
});
return { kind: "error" };
}
// The wire re-check for exactly this chain (`aiEgressRefusal`, the one
// helper every chokepoint uses): the capability again, and whether sending
// to these providers needs a consent receipt under the capability's rule
@@ -285,28 +298,7 @@ export async function runStatusCompletion(
"job",
);
if (!reservation.allowed) {
// Over the day's ceiling. Reported as `error` — a TRANSIENT miss the caller
// serves the fallback for without persisting it — deliberately NOT `none`,
// which callers cache as the settled "no provider configured" assessment.
// The distinct annotation keeps the refusal observable even though the
// result shape is shared.
// v1.38.19 — the shared `error` outcome cannot say
// WHY the generation stopped, so the annotation has to. `surface: "job"`
// plus `cap` separates "this job used up the background share" from "the
// whole day is spent", and `limit` says which counter tripped. The operator
// sees the same split on the admin provider-health card.
annotate({
action: { name: "insights.status.budget_exceeded" },
meta: {
cacheAction,
owner: reservation.owner,
surface: "job",
limit: reservation.limit,
cap: jobCap,
totalAfter: reservation.totalAfter,
operatorAfter: reservation.operatorAfter,
},
});
annotateBudgetRefusal(cacheAction, reservation, jobCap);
return { kind: "error" };
}
@@ -340,22 +332,17 @@ export async function runStatusCompletion(
null,
);
if (raced.timedOut) {
// A timed-out generation may still have burned upstream tokens, but we have
// no reported count to charge — refund the reservation rather than bill an
// invented figure.
await reconcileSpend(userId, reservation.reserved, 0, dateKey, 0, {
if (raced.timedOut || raced.errored || raced.value === null) {
// A timeout or failure: charge only what a reply reported spending, never
// an invented figure. #1279 — a failed chain's report counts: a model
// that reasons without answering bills every call, and a full refund let
// background work spend without the day's ceiling ever seeing it.
const spent = spentOnFailure(raced.error);
await reconcileSpend(userId, reservation.reserved, spent, dateKey, 0, {
servedBy: null,
reservedOwner: reservation.owner,
});
return { kind: "timeout" };
}
if (raced.errored || raced.value === null) {
await reconcileSpend(userId, reservation.reserved, 0, dateKey, 0, {
servedBy: null,
reservedOwner: reservation.owner,
});
return { kind: "error" };
return { kind: raced.timedOut ? "timeout" : "error" };
}
const { result, workingProvider } = raced.value;
@@ -409,3 +396,35 @@ export async function runStatusCompletion(
tokensUsed: result.tokensUsed ?? null,
};
}
/**
* Over the day's ceiling. Reported as `error` — a TRANSIENT miss the caller
* serves the fallback for without persisting it — deliberately NOT `none`,
* which callers cache as the settled "no provider configured" assessment.
* The distinct annotation keeps the refusal observable even though the
* result shape is shared.
*
* v1.38.19 — the shared `error` outcome cannot say WHY the generation
* stopped, so the annotation has to. `surface: "job"` plus `cap` separates
* "this job used up the background share" from "the whole day is spent", and
* `limit` says which counter tripped. The operator sees the same split on the
* admin provider-health card.
*/
function annotateBudgetRefusal(
cacheAction: string,
reservation: Awaited<ReturnType<typeof reserveBudget>>,
jobCap: number,
): void {
annotate({
action: { name: "insights.status.budget_exceeded" },
meta: {
cacheAction,
owner: reservation.owner,
surface: "job",
limit: reservation.limit,
cap: jobCap,
totalAfter: reservation.totalAfter,
operatorAfter: reservation.operatorAfter,
},
});
}
+9 -11
View File
@@ -1,4 +1,4 @@
import { writeStatusNegativeWindow } from "@/lib/insights/status-cache";
import { writeStatusNegativeWindow } from "@/lib/insights/status-negative-window";
import { annotate } from "@/lib/logging/context";
/**
@@ -31,19 +31,18 @@ import { annotate } from "@/lib/logging/context";
* but it tripped a safety contract (see `@/lib/ai/safety/outbound-screen`).
* It rides the same envelope as a stall on purpose — the surface policy for a
* background-generated cached card is WITHHOLD, and this path already means
* exactly "serve the deterministic line, persist no model text". The 5-minute
* negative stub gives the next generation a fresh attempt rather than blanking
* the card until midnight.
* exactly "serve the deterministic line, persist no model text". The negative
* stub gives the next generation a fresh attempt once its window lapses
* rather than blanking the card until midnight.
*/
export type StatusFallbackReason = "timeout" | "error" | "screened";
/** Short retry window after a provider stall before a re-enqueue is allowed. */
export const TIMEOUT_NEGATIVE_CACHE_MS = 5 * 60 * 1000;
/**
* Persist a short-TTL negative-cache stub for `(userId, cacheAction)`.
* Best-effort: a failure here just means the read-only resolver may
* re-enqueue sooner than the window intended.
* Persist a negative-cache stub for `(userId, cacheAction)`. Its window is five
* minutes after a first failure and grows with each failure in a row, up to
* two hours (#1279, `writeStatusNegativeWindow`). Best-effort: a failure here
* just means the read-only resolver may re-enqueue sooner than the window
* intended.
*/
export async function persistTimeoutNegativeStub(args: {
userId: string;
@@ -57,7 +56,6 @@ export async function persistTimeoutNegativeStub(args: {
cacheAction: args.cacheAction,
todayKey: args.todayKey,
reason: args.reason,
retryAt: new Date(Date.now() + TIMEOUT_NEGATIVE_CACHE_MS),
});
} catch {
// Negative cache is best-effort — never throw out of the timeout path.
+3 -1
View File
@@ -26,6 +26,8 @@ export interface TimeoutEnvelope<T> {
/** True when the upstream rejected before the budget expired. */
errored: boolean;
value: T;
/** What the upstream rejected with, when `errored`. */
error?: unknown;
}
export function withTimeout<T>(
@@ -83,7 +85,7 @@ export function withTimeout<T>(
action: { name: "insights.status.provider_error" },
meta: { reason: reason.slice(0, 240) },
});
resolve({ timedOut: false, errored: true, value: fallback });
resolve({ timedOut: false, errored: true, value: fallback, error });
});
});
}
@@ -0,0 +1,480 @@
/**
* #1279 — one simulated hour against an OpenAI-compatible gateway whose model
* answers every request with reasoning and no answer, cut off at the token
* ceiling (`finish_reason: "length"`), as gpt-oss behind vLLM did with the
* gateway's Reasoning left at Default and nobody using the app.
*
* What runs for real: the sync-driven re-selection of status cards
* (`invalidateStatusInsightsForTypes`), the production enqueue with its
* queue name, singleton key and send options (`enqueueStatusGeneration`),
* the queue's consumer (`runInsightStatusGenerate`), the card's single-call
* path and fallback (`runPreparedStatusCard`, `returnTimeoutFallback`), the
* status chokepoint (`runStatusCompletion`), the chain walker with the
* health ledger's in-memory twin, the gateway client and its reasoning
* lessons, and the note store's negative windows over an in-memory table.
* What is faked: the wire (`safeFetch` answers like the gateway and moves
* the clock by the model's latency), pg-boss (a queue that honours the
* singleton key, the exclusive policy and the retry options a send names),
* the clock, and the reads that would need Postgres for anything else.
*
* A data sync dirties five cards every minute. Before the fix, every sync
* re-selected every card whose last generation had failed, and each job
* reached the model: run against that code, this hour counted 151 requests,
* 150 of them sent with `none`, none of them charged to the day's ledger.
* The second case also loses the reasoning lesson between jobs, which is
* what the reported log shows (two requests behind every answerless line):
* 300 requests before the fix. The bound has to hold without the lesson.
*/
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import type { InsightStatusGeneratePayload } from "@/lib/jobs/insight-status-generate-shared";
// ── The clock and the wire ───────────────────────────────────────────────────
const wire = vi.hoisted(() => ({
requests: [] as Array<{ effort: unknown; maxTokens: unknown }>,
latencyMs: 2600,
}));
vi.mock("@/lib/safe-fetch", async (importActual) => ({
...(await importActual<typeof import("@/lib/safe-fetch")>()),
safeFetch: async (_url: string, init: { body: string }) => {
const body = JSON.parse(init.body) as Record<string, unknown>;
wire.requests.push({
effort: body.reasoning_effort,
maxTokens: body.max_tokens ?? body.max_completion_tokens,
});
vi.setSystemTime(Date.now() + wire.latencyMs);
return new Response(
JSON.stringify({
id: "chatcmpl-1279",
object: "chat.completion",
model: "gpt-oss-120b",
choices: [
{
index: 0,
message: {
role: "assistant",
content: null,
reasoning_content: "We need to look at the readings. ".repeat(30),
},
finish_reason: "length",
},
],
usage: {
prompt_tokens: 2470,
completion_tokens: 300,
total_tokens: 2770,
},
}),
{ status: 200, headers: { "content-type": "application/json" } },
);
},
}));
// ── pg-boss, as far as these sends use it ────────────────────────────────────
interface QueuedJob {
data: InsightStatusGeneratePayload;
key: string | null;
runAt: number;
retriesLeft: number;
retryDelayS: number;
backoff: boolean;
attempt: number;
}
const boss = vi.hoisted(() => ({
jobs: [] as QueuedJob[],
sends: 0,
/** The last time each singleton key was sent, for `singletonSeconds`. */
slots: new Map<string, number>(),
}));
vi.mock("@/lib/jobs/boss-instance", () => ({
getGlobalBoss: () => ({
async send(
_queue: string,
data: InsightStatusGeneratePayload,
opts: Record<string, unknown> = {},
) {
boss.sends += 1;
const key =
typeof opts.singletonKey === "string" ? opts.singletonKey : null;
const now = Date.now();
if (key) {
// The queue's exclusive policy: one queued or running job per key.
if (boss.jobs.some((j) => j.key === key)) return null;
const window = Number(opts.singletonSeconds ?? 0) * 1000;
const last = boss.slots.get(key);
if (last !== undefined && now - last < window) return null;
boss.slots.set(key, now);
}
boss.jobs.push({
data,
key,
runAt: now + Number(opts.startAfter ?? 0) * 1000,
retriesLeft: Number(opts.retryLimit ?? 2),
retryDelayS: Number(opts.retryDelay ?? 0),
backoff: opts.retryBackoff === true,
attempt: 0,
});
return "job-id";
},
}),
}));
// ── The note store and the reads around it ───────────────────────────────────
interface NoteRow {
userId: string;
metric: string;
locale: string;
dateKey: string;
textEncrypted: Uint8Array | null;
generatedAt: Date | null;
retryAt: Date | null;
negativeReason: string | null;
failureStreak: number;
}
const store = vi.hoisted(() => ({ notes: new Map<string, NoteRow>() }));
function noteKey(w: { userId: string; metric: string; locale: string }) {
return `${w.userId}|${w.metric}|${w.locale}`;
}
vi.mock("@/lib/db", () => ({
prisma: {
user: {
findUnique: async () => ({
locale: "en",
timezone: "UTC",
aiResponseTimeoutSeconds: null,
}),
},
insightStatusCache: {
findUnique: async (args: {
where: {
userId_metric_locale: {
userId: string;
metric: string;
locale: string;
};
};
}) => store.notes.get(noteKey(args.where.userId_metric_locale)) ?? null,
upsert: async (args: {
where: {
userId_metric_locale: {
userId: string;
metric: string;
locale: string;
};
};
create: Partial<NoteRow>;
update: Partial<NoteRow>;
}) => {
const k = noteKey(args.where.userId_metric_locale);
const prev = store.notes.get(k);
const next = prev
? { ...prev, ...args.update }
: ({
textEncrypted: null,
generatedAt: null,
retryAt: null,
negativeReason: null,
failureStreak: 0,
...args.create,
} as NoteRow);
store.notes.set(k, next);
return next;
},
findMany: async (args: {
where: {
userId: string;
OR: Array<{ metric: string; locale: string }>;
generatedAt?: { gte: Date };
textEncrypted?: { not: null };
retryAt?: { gt: Date };
};
}) =>
args.where.OR.flatMap((key) => {
const row = store.notes.get(
noteKey({ userId: args.where.userId, ...key }),
);
if (!row) return [];
const w = args.where;
if (
w.generatedAt &&
!(row.generatedAt && row.generatedAt >= w.generatedAt.gte)
) {
return [];
}
if (w.textEncrypted && row.textEncrypted === null) return [];
if (w.retryAt && !(row.retryAt && row.retryAt > w.retryAt.gt))
return [];
return [{ metric: row.metric, locale: row.locale }];
}),
updateMany: async () => ({ count: 0 }),
},
},
}));
// The day's token ledger: every reservation allowed, the charge recorded.
const ledger = vi.hoisted(() => ({ charged: 0 }));
vi.mock("@/lib/ai/coach/budget", async (importActual) => ({
...(await importActual<typeof import("@/lib/ai/coach/budget")>()),
buildDateKey: async () => "2026-10-24",
reserveBudget: async (_u: string, estimated: number) => ({
allowed: true,
reserved: estimated,
owner: "user",
limit: null,
totalAfter: 0,
operatorAfter: 0,
}),
reconcileSpend: async (_u: string, _reserved: number, actual: number) => {
ledger.charged += actual;
},
}));
// The chain's durable health ledger, as its in-memory twin, fresh per case.
const health = vi.hoisted(() => ({
current: null as null | {
getSkipHints(userId: string): Promise<Map<string, unknown>>;
recordSuccess(userId: string, type: string): Promise<void>;
recordFailure(
userId: string,
type: string,
status: number | null,
): Promise<void>;
},
}));
vi.mock("@/lib/ai/provider-health-ledger", async (importActual) => {
const actual =
await importActual<typeof import("@/lib/ai/provider-health-ledger")>();
return {
...actual,
postgresProviderHealthLedger: {
getSkipHints: (u: string) => health.current!.getSkipHints(u),
recordSuccess: (u: string, t: string) =>
health.current!.recordSuccess(u, t),
recordFailure: (u: string, t: string, s: number | null) =>
health.current!.recordFailure(u, t, s),
},
};
});
// The one chain entry: the gateway, as the settings build it.
vi.mock("@/lib/ai/provider", async () => {
const { OpenAIClient } = await import("@/lib/ai/openai-client");
const gateway = new OpenAIClient({
apiKey: "",
model: "gpt-oss-120b",
baseUrl: "https://gateway.example/v1",
providerType: "openai-compatible",
});
return {
resolveProviderChain: async () => [
{ providerType: "openai-compatible", instance: gateway },
],
resolveProvider: async () => ({ type: "none" }),
probeProviderPresence: async () => true,
};
});
const AVAILABLE = { available: true, reason: null, onDeviceAllowed: true };
vi.mock("@/lib/ai/capabilities/gate", () => ({
aiCapabilityForRecord: async () => AVAILABLE,
aiCapabilityForJob: async () => AVAILABLE,
aiCapabilityToServe: async () => AVAILABLE,
}));
vi.mock("@/lib/ai/capabilities/egress", () => ({
aiEgressRefusal: async () => null,
}));
vi.mock("@/lib/ai/capabilities/usage", () => ({
loadRecordAiUsage: async () => ({
level: "standard",
settings: { syncRegenGapHours: 6 },
}),
}));
vi.mock("@/lib/sharing/provider-work-authority", () => ({
providerWorkAuthorityForRecord: () => ({ origin: "owner" }),
mayEnqueueProviderWork: () => true,
mayDispatchProviderWork: async () => true,
withProviderWorkAuthority: async (_a: unknown, fn: () => Promise<unknown>) =>
fn(),
}));
const signals = vi.hoisted(() => [] as Array<{ action: string }>);
vi.mock("@/lib/logging/signal", () => ({
emitSignal: (s: { action: string }) => signals.push(s),
logCaught: () => undefined,
caughtAs: () => undefined,
}));
import { invalidateStatusInsightsForTypes } from "@/lib/insights/status-invalidation";
import {
runInsightStatusGenerate,
type InsightStatusMetric,
} from "@/lib/jobs/insight-status-generate";
import { runPreparedStatusCard } from "@/lib/insights/status-card-generation";
import { returnTimeoutFallback } from "@/lib/insights/timeout-fallback";
import { statusCacheAction } from "@/lib/insights/status-cache";
import { resetReasoningDialectCache } from "@/lib/ai/reasoning/dialect-cache";
import { clearLastWorkingProviderCache } from "@/lib/ai/provider-runner";
import { createInMemoryProviderHealthLedger } from "@/lib/ai/provider-health-ledger";
const USER = "self-hoster";
const HOUR_MS = 60 * 60 * 1000;
const SYNC_EVERY_MS = 60 * 1000;
/** A watch drip: weight, blood pressure and pulse dirty five cards. */
const SYNC_TYPES = ["WEIGHT", "BLOOD_PRESSURE_SYS", "PULSE"] as const;
/** A card generator as the worker runs it: one status call, its fallback. */
function cardGenerator(scope: InsightStatusMetric) {
return async (userId: string, options: { locale: string }) => {
const cacheAction = statusCacheAction(scope, options.locale);
return runPreparedStatusCard({
phase: "pending",
metric: scope,
userId,
cacheAction,
systemPrompt: "You write the status note.",
userPrompt: `Snapshot for ${scope}: …`,
temperature: 0.45,
noProvider: { hasProvider: false, cached: true, updatedAt: null },
finalize: async () => ({
hasProvider: true,
cached: false,
updatedAt: null,
}),
timeout: (reason) =>
returnTimeoutFallback({
cacheAction,
reason,
stubText: "Deterministic line.",
userId,
todayKey: "2026-10-24",
}),
});
};
}
const GENERATORS: Parameters<typeof runInsightStatusGenerate>[1] = {
general: cardGenerator("general"),
"blood-pressure": cardGenerator("blood-pressure"),
weight: cardGenerator("weight"),
pulse: cardGenerator("pulse"),
bmi: cardGenerator("bmi"),
mood: cardGenerator("mood"),
"medication-compliance": cardGenerator("medication-compliance"),
};
/** Let fire-and-forget writes (the negative stub) land. */
async function settle() {
for (let i = 0; i < 5; i += 1) await Promise.resolve();
await new Promise((r) => setImmediate(r));
}
/**
* Run the worker for one simulated hour: a sync every minute, the queue
* drained one job at a time as soon as a job is due.
*/
async function simulateHour(opts: { loseLessonBetweenJobs: boolean }) {
const end = Date.now() + HOUR_MS;
let nextSync = Date.now();
let jobsRun = 0;
let syncs = 0;
while (Date.now() < end) {
if (Date.now() >= nextSync) {
syncs += 1;
await invalidateStatusInsightsForTypes(USER, SYNC_TYPES, {
source: "sync",
});
await settle();
nextSync += SYNC_EVERY_MS;
}
const due = boss.jobs
.filter((j) => j.runAt <= Date.now())
.sort((a, b) => a.runAt - b.runAt)[0];
if (!due) {
const nextJob = Math.min(...boss.jobs.map((j) => j.runAt), Infinity);
vi.setSystemTime(Math.min(nextSync, nextJob, end));
continue;
}
jobsRun += 1;
try {
await runInsightStatusGenerate(due.data, GENERATORS);
boss.jobs.splice(boss.jobs.indexOf(due), 1);
} catch {
if (due.retriesLeft > 0) {
due.retriesLeft -= 1;
due.attempt += 1;
const delay =
due.retryDelayS * (due.backoff ? 2 ** (due.attempt - 1) : 1);
due.runAt = Date.now() + delay * 1000;
} else {
boss.jobs.splice(boss.jobs.indexOf(due), 1);
}
}
await settle();
if (opts.loseLessonBetweenJobs) resetReasoningDialectCache();
vi.setSystemTime(Date.now() + 100);
}
return { jobsRun, syncs };
}
beforeEach(() => {
vi.useFakeTimers({ toFake: ["Date"] });
vi.setSystemTime(new Date("2026-10-24T15:40:00.000Z"));
wire.requests = [];
boss.jobs = [];
boss.sends = 0;
boss.slots.clear();
store.notes.clear();
ledger.charged = 0;
signals.length = 0;
resetReasoningDialectCache();
clearLastWorkingProviderCache();
health.current = createInMemoryProviderHealthLedger();
});
afterEach(() => {
vi.useRealTimers();
});
describe("#1279: an hour of background status work against a model that never answers", () => {
it("sends a handful of requests, not hundreds", async () => {
const { jobsRun, syncs } = await simulateHour({
loseLessonBetweenJobs: false,
});
// A sync every minute kept dirtying the five cards …
expect(syncs).toBe(60);
expect(jobsRun).toBeGreaterThan(0);
// … but the model was asked a bounded number of times.
expect(wire.requests.length).toBeLessThanOrEqual(10);
// `none` went out once, as the one repeat, and never again.
expect(wire.requests.filter((r) => r.effort === "none")).toHaveLength(1);
// And what those requests cost reached the day's ledger.
expect(ledger.charged).toBe(wire.requests.length * 2770);
});
it("stays bounded when the lesson does not survive from one job to the next", async () => {
await simulateHour({ loseLessonBetweenJobs: true });
expect(wire.requests.length).toBeLessThanOrEqual(14);
});
it("says once that background work paused, not once per job", async () => {
await simulateHour({ loseLessonBetweenJobs: false });
const paused = signals.filter(
(s) => s.action === "ai.chain.background_paused",
);
expect(paused.length).toBeGreaterThanOrEqual(1);
expect(paused.length).toBeLessThanOrEqual(4);
expect(
signals.filter((s) => s.action === "ai.gateway.reasoning_answerless"),
).toHaveLength(1);
});
});
@@ -0,0 +1,144 @@
/**
* #1279 — the content-index backfill against a provider that fails every
* call (a model that reasons and never answers). Its cap counted indexed
* documents only, so a provider that never answered never reached it, and
* the walk transcribed every un-indexed document in turn until the job's
* time ran out. It now stops after a few failures in a row, counts calls
* against the cap, and charges what the failed replies reported spending.
*/
import { beforeEach, describe, expect, it, vi } from "vitest";
const findMany = vi.fn();
const transcribeDocument = vi.fn();
const reconcileSpend = vi.fn();
vi.mock("@/lib/db", () => ({
prisma: {
inboundDocument: {
findMany: (...a: unknown[]) => findMany(...a),
findFirst: async (args: { where: { id: string } }) => ({
id: args.where.id,
kind: "OTHER",
contentEncrypted: new Uint8Array([1, 2, 3]),
contentCodec: "binary2",
mimeType: "image/png",
status: "STORED",
}),
},
documentContentIndex: {
upsert: async () => ({}),
findMany: async () => [],
update: async () => ({}),
},
},
}));
vi.mock("@/lib/documents/store", () => ({
decryptDocumentContent: () => Buffer.from([1, 2, 3]),
}));
vi.mock("@/lib/labs/ocr-upload", () => ({
detectOcrMimeType: () => "image/png",
}));
vi.mock("@/lib/documents/rasterize-pdf", () => ({ rasterizePdf: vi.fn() }));
vi.mock("@/lib/documents/describe", () => ({
transcribeDocument: (...a: unknown[]) => transcribeDocument(...a),
DocumentDescribeError: class DocumentDescribeError extends Error {},
}));
vi.mock("@/lib/documents/content-index", () => ({
upsertContentIndex: async () => ({ tokenCount: 3 }),
CONTENT_TOKENIZER_VERSION: "1",
decryptIndexText: () => "decrypted index text",
tokeniseAndHash: () => ["hash-a"],
}));
vi.mock("@/lib/documents/provider-order", () => ({
resolveDocumentVisionProvider: async () => ({
chain: [{ providerType: "openai-compatible", instance: {} }],
withheld: null,
pick: {
entry: { providerType: "openai-compatible", instance: {} },
providerType: "openai-compatible",
pdfSupported: true,
},
}),
}));
vi.mock("@/lib/ai/capabilities/gate", () => ({
aiCapabilityForJob: async () => ({
available: true,
reason: null,
onDeviceAllowed: true,
}),
aiCapabilityForRecord: vi.fn(),
}));
vi.mock("@/lib/ai/coach/budget", () => ({
buildDateKey: () => "2026-10-24",
reserveBudget: async () => ({ allowed: true, reserved: 3000, owner: "user" }),
reconcileSpend: (...a: unknown[]) => reconcileSpend(...a),
resolveDailyCap: () => 1_000_000,
resolveDailyCapFor: () => 1_000_000,
resolveCostOwner: () => "user",
}));
vi.mock("@/lib/jobs/boss-instance", () => ({ getGlobalBoss: vi.fn() }));
import {
BACKFILL_FAILURES_IN_A_ROW,
runContentIndexBackfillForUser,
} from "../document-content-index-backfill";
/** 120 un-indexed images, served 50 to a page like the real walk asks. */
function seedDocuments(count: number) {
const ids = Array.from({ length: count }, (_, i) => ({
id: `d${String(i).padStart(4, "0")}`,
}));
findMany.mockImplementation(
async (args: { take: number; cursor?: { id: string } }) => {
const start = args.cursor
? ids.findIndex((d) => d.id === args.cursor!.id) + 1
: 0;
return ids.slice(start, start + args.take);
},
);
}
beforeEach(() => {
vi.resetAllMocks();
reconcileSpend.mockResolvedValue(undefined);
});
describe("#1279: the content-index backfill and a provider that never answers", () => {
it("stops after a few failures in a row instead of walking every document", async () => {
seedDocuments(120);
transcribeDocument.mockRejectedValue(
Object.assign(new Error("returned empty content"), {
httpStatus: 0,
kind: "reasoning_only",
tokensUsed: 2700,
}),
);
const result = await runContentIndexBackfillForUser("user-1");
expect(transcribeDocument).toHaveBeenCalledTimes(
BACKFILL_FAILURES_IN_A_ROW,
);
expect(result).toMatchObject({
indexed: 0,
failed: BACKFILL_FAILURES_IN_A_ROW,
reason: "provider-failing",
});
// What each failed reply reported spending stays on the day's ledger.
for (const call of reconcileSpend.mock.calls) {
expect(call[2]).toBe(2700);
}
});
it("keeps walking past one document the model cannot read", async () => {
seedDocuments(5);
transcribeDocument
.mockRejectedValueOnce(new Error("unreadable"))
.mockResolvedValue({ text: "glucose 90" });
const result = await runContentIndexBackfillForUser("user-1");
expect(transcribeDocument).toHaveBeenCalledTimes(5);
expect(result).toMatchObject({ indexed: 4, failed: 1, reason: "ok" });
});
});
+88 -36
View File
@@ -39,7 +39,9 @@ import {
reserveBudget,
resolveCostOwner,
resolveDailyCapFor,
type BudgetCostOwner,
} from "@/lib/ai/coach/budget";
import type { ProviderChainType } from "@/lib/ai/provider-chain";
import { prisma } from "@/lib/db";
import {
loadOwnedDocument,
@@ -55,6 +57,7 @@ import { transcribeDocument } from "@/lib/documents/describe";
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
import { resolveDocumentVisionProvider } from "@/lib/documents/provider-order";
import { annotate } from "@/lib/logging/context";
import { spentTokensOf } from "@/lib/ai/reasoning/answerless";
import { CONTENT_INDEX_BACKFILL_QUEUE } from "./document-content-index-backfill-queue";
/** Discovery page size (id-only; blobs are loaded one at a time). */
@@ -68,9 +71,26 @@ const PAGE_SIZE = 50;
*/
export const CONTENT_INDEX_BACKFILL_EXPIRE_SECONDS = 3 * 60 * 60;
/** Max documents one job indexes before yielding (bounds spend + runtime). */
/**
* Max provider calls one job makes before yielding (bounds spend + runtime).
* #1279 — calls, not successes: counted on indexed documents only, a
* provider that failed every call never reached it, and the walk went on
* until the job's time ran out.
*/
const MAX_DOCS_PER_RUN = 200;
/**
* #1279 — provider calls in a row that may fail before the walk stops. One
* document the model cannot read is that document; several in a row is the
* provider, and the rest of the walk would only repeat it.
*/
export const BACKFILL_FAILURES_IN_A_ROW = 3;
/** The provider the document pick chose. */
type VisionPick = NonNullable<
Awaited<ReturnType<typeof resolveDocumentVisionProvider>>["pick"]
>;
/** Image MIME types the vision path can always index. */
const IMAGE_MIMES = ["image/jpeg", "image/png", "image/webp"] as const;
@@ -100,7 +120,13 @@ export interface ContentIndexBackfillSummary {
*/
failed: number;
reason:
"ok" | "no-provider" | "no-consent" | "budget-reached" | "unavailable";
| "ok"
| "no-provider"
| "no-consent"
| "budget-reached"
| "unavailable"
/** #1279 — `BACKFILL_FAILURES_IN_A_ROW` provider calls failed in a row. */
| "provider-failing";
/**
* The job's time budget ran out before the walk did. The documents not
* reached stay un-indexed for the next run.
@@ -234,12 +260,14 @@ export async function runContentIndexBackfillForUser(
let indexed = 0;
let skipped = 0;
let failed = stale.failed;
let calls = 0;
let failedInARow = 0;
let reason: ContentIndexBackfillSummary["reason"] = "ok";
let cursor: string | null = null;
let stoppedEarly = false;
outer: for (;;) {
if (indexed >= MAX_DOCS_PER_RUN) break;
if (calls >= MAX_DOCS_PER_RUN) break;
const rows: { id: string }[] = await prisma.inboundDocument.findMany({
where: {
userId,
@@ -258,7 +286,11 @@ export async function runContentIndexBackfillForUser(
if (rows.length === 0) break;
for (const { id } of rows) {
if (indexed >= MAX_DOCS_PER_RUN) break outer;
if (calls >= MAX_DOCS_PER_RUN) break outer;
if (failedInARow >= BACKFILL_FAILURES_IN_A_ROW) {
reason = "provider-failing";
break outer;
}
if (shouldStop()) {
stoppedEarly = true;
break outer;
@@ -301,40 +333,18 @@ export async function runContentIndexBackfillForUser(
continue;
}
try {
const { text } = await transcribeDocument({
provider: pick.entry.instance,
providerType: pick.providerType,
images: vision.images,
documents: vision.documents,
});
await reconcileSpend(
userId,
reservation.reserved,
reservation.reserved,
dateKey,
0,
{
servedBy: pick.entry.providerType,
reservedOwner: reservation.owner,
},
);
await upsertContentIndex({
userId,
documentId: document.id,
text,
source: "vision",
providerType: pick.providerType,
});
calls += 1;
const read = await transcribeAndIndex(userId, document.id, pick, vision, {
reserved: reservation.reserved,
owner: reservation.owner,
dateKey,
});
if (read === "indexed") {
indexed += 1;
} catch {
// A provider miss on one document must not abort the batch — refund and
// leave it un-indexed (a later run retries it).
await reconcileSpend(userId, reservation.reserved, 0, dateKey, 0, {
servedBy: null,
reservedOwner: reservation.owner,
});
failedInARow = 0;
} else {
failed += 1;
failedInARow += 1;
}
}
@@ -363,6 +373,48 @@ export async function runContentIndexBackfillForUser(
};
}
/**
* One document's provider read and its index row, with the reservation
* settled either way. A provider miss on one document must not abort the
* batch: it is refunded and left un-indexed (a later run retries it).
* #1279 — what the failed reply reported spending stays charged, and the
* caller stops the walk after several misses in a row.
*/
async function transcribeAndIndex(
userId: string,
documentId: string,
pick: VisionPick,
vision: Extract<Awaited<ReturnType<typeof prepareVisionInput>>, { ok: true }>,
budget: { reserved: number; owner: BudgetCostOwner; dateKey: string },
): Promise<"indexed" | { error: unknown }> {
const settle = (spent: number, servedBy: ProviderChainType | null) =>
reconcileSpend(userId, budget.reserved, spent, budget.dateKey, 0, {
servedBy,
reservedOwner: budget.owner,
});
try {
const { text } = await transcribeDocument({
provider: pick.entry.instance,
providerType: pick.providerType,
images: vision.images,
documents: vision.documents,
});
await settle(budget.reserved, pick.entry.providerType);
await upsertContentIndex({
userId,
documentId,
text,
source: "vision",
providerType: pick.providerType,
});
return "indexed";
} catch (err) {
await settle(spentTokensOf(err) ?? 0, null);
// Counted into the run's `failed` tally by the caller.
return { error: err };
}
}
/**
* Enqueue a per-user content-index backfill. Coalesced by `singletonKey` so a
* user cannot pile up parallel runs; returns whether a job was created.
+3
View File
@@ -227,9 +227,12 @@ const jobs: readonly JobDefinition[] = [
work: { localConcurrency: 1 },
load: async () => (await handlers()).handleCoachMemoryRefresh,
}),
// #1279 — the nudge tick reaches a model; pg-boss's queue default retried
// a failed tick twice at once. The same two retries now back off.
defineJob({
queue: COACH_NUDGE_QUEUE,
cron: COACH_NUDGE_CRON,
scheduleOptions: { retryLimit: 2, retryDelay: 60, retryBackoff: true },
work: { localConcurrency: 1 },
load: async () => (await handlers()).handleCoachNudge,
}),