mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-08 12:32:43 +00:00
fix(ai): read AI switches, modules and consent afresh inside background jobs
A background job opens one wide event for its whole run, and the per-request memo cached the operator switches, module availability, module preferences, provider presence and consent kinds on it. The nightly status passes loop over every candidate inside that one event, so a switch turned off mid-run did not stop the remaining users, and the wire re-check read the same cached answer. memoizePerRequest takes a freshInBackground option; every input that decides whether AI work may run sets it, so a background caller reads them on every call while a request still reads them once.
This commit is contained in:
1 parent
4ae231d757
commit
06251d7219
10 files changed
+236
-67
No files matched your search
@@ -0,0 +1,27 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
|
||||
import { eventStorage } from "@/lib/logging/context";
|
||||
import { WideEventBuilder } from "@/lib/logging/event-builder";
|
||||
import { memoizePerRequest } from "@/lib/request-cache";
|
||||
|
||||
function twice(kind: "http" | "background", fresh: boolean) {
|
||||
const factory = vi.fn(async () => Math.random());
|
||||
return eventStorage.run(new WideEventBuilder(kind), async () => {
|
||||
const options = fresh ? { freshInBackground: true } : {};
|
||||
await memoizePerRequest("k", factory, options);
|
||||
await memoizePerRequest("k", factory, options);
|
||||
return factory.mock.calls.length;
|
||||
});
|
||||
}
|
||||
|
||||
describe("memoizePerRequest", () => {
|
||||
it("reads once per request", async () => {
|
||||
expect(await twice("http", false)).toBe(1);
|
||||
expect(await twice("http", true)).toBe(1);
|
||||
});
|
||||
|
||||
it("reads once per background event unless the input must stay fresh", async () => {
|
||||
expect(await twice("background", false)).toBe(1);
|
||||
expect(await twice("background", true)).toBe(2);
|
||||
});
|
||||
});
|
||||
@@ -56,13 +56,17 @@ export interface AiCapabilityScope {
|
||||
|
||||
/** The kinds of the record's active consent receipts. */
|
||||
function loadActiveConsentKinds(recordId: string): Promise<Set<string>> {
|
||||
return memoizePerRequest(`ai-consent-kinds:${recordId}`, async () => {
|
||||
const rows = await prisma.consentReceipt.findMany({
|
||||
where: { userId: recordId, revokedAt: null },
|
||||
select: { kind: true },
|
||||
});
|
||||
return new Set(rows.map((row) => row.kind));
|
||||
});
|
||||
return memoizePerRequest(
|
||||
`ai-consent-kinds:${recordId}`,
|
||||
async () => {
|
||||
const rows = await prisma.consentReceipt.findMany({
|
||||
where: { userId: recordId, revokedAt: null },
|
||||
select: { kind: true },
|
||||
});
|
||||
return new Set(rows.map((row) => row.kind));
|
||||
},
|
||||
{ freshInBackground: true },
|
||||
);
|
||||
}
|
||||
|
||||
function scopeKey(scope: AiCapabilityScope): string {
|
||||
@@ -114,6 +118,7 @@ export function loadAiCapabilityInputs(
|
||||
return null;
|
||||
}
|
||||
},
|
||||
{ freshInBackground: true },
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@@ -985,6 +985,7 @@ export function probeProviderChain(
|
||||
managedBy: entries.length > 0 ? managedBy : null,
|
||||
};
|
||||
},
|
||||
{ freshInBackground: true },
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@@ -16,7 +16,7 @@ vi.mock("@/lib/db", () => ({
|
||||
// pre-v1.4.33 assertion semantics.
|
||||
let mockEvent: object | null = null;
|
||||
vi.mock("@/lib/logging/context", () => ({
|
||||
getEvent: () => mockEvent ?? { addWarning: vi.fn() },
|
||||
getEvent: () => mockEvent ?? { getKind: () => "http", addWarning: vi.fn() },
|
||||
}));
|
||||
|
||||
import { prisma } from "@/lib/db";
|
||||
@@ -154,7 +154,7 @@ describe("per-request memoisation", () => {
|
||||
// Pin a stable stub object so every `getEvent()` inside this test
|
||||
// returns the same reference — the memo keys on identity, so a
|
||||
// shared reference == shared cache.
|
||||
mockEvent = { addWarning: vi.fn() };
|
||||
mockEvent = { getKind: () => "http", addWarning: vi.fn() };
|
||||
|
||||
const a = await getAssistantFlags();
|
||||
const b = await getAssistantFlags();
|
||||
@@ -174,9 +174,9 @@ describe("per-request memoisation", () => {
|
||||
assistantDocumentAiEnabled: true,
|
||||
});
|
||||
|
||||
mockEvent = { addWarning: vi.fn(), id: "req-1" };
|
||||
mockEvent = { getKind: () => "http", addWarning: vi.fn(), id: "req-1" };
|
||||
await getAssistantFlags();
|
||||
mockEvent = { addWarning: vi.fn(), id: "req-2" };
|
||||
mockEvent = { getKind: () => "http", addWarning: vi.fn(), id: "req-2" };
|
||||
await getAssistantFlags();
|
||||
|
||||
expect(FIND).toHaveBeenCalledTimes(2);
|
||||
|
||||
@@ -49,35 +49,40 @@ const ASSISTANT_FLAGS_OFF: AssistantFlagSet = Object.freeze({
|
||||
* switch; failing open meant a database blip switched every AI egress on.
|
||||
*
|
||||
* Memoised per request, so a page that fires several gated reads at once reads
|
||||
* the singleton row once.
|
||||
* the singleton row once. Read afresh on every call inside a background job,
|
||||
* whose one event spans a whole multi-user pass.
|
||||
*/
|
||||
export function loadAssistantSwitches(): Promise<AssistantFlagSet | null> {
|
||||
return memoizePerRequest("assistant-flags", async () => {
|
||||
try {
|
||||
const settings = await prisma.appSettings.findUnique({
|
||||
where: { id: "singleton" },
|
||||
select: {
|
||||
assistantEnabled: true,
|
||||
assistantCoachEnabled: true,
|
||||
assistantBriefingEnabled: true,
|
||||
assistantInsightStatusEnabled: true,
|
||||
assistantDocumentAiEnabled: true,
|
||||
},
|
||||
});
|
||||
return resolveAssistantFlags({
|
||||
enabled: settings?.assistantEnabled ?? true,
|
||||
coach: settings?.assistantCoachEnabled ?? true,
|
||||
briefing: settings?.assistantBriefingEnabled ?? true,
|
||||
insightStatus: settings?.assistantInsightStatusEnabled ?? true,
|
||||
documentAi: settings?.assistantDocumentAiEnabled ?? true,
|
||||
});
|
||||
} catch {
|
||||
getEvent()?.addWarning(
|
||||
"Failed to load assistant switches; every AI capability is off for this request",
|
||||
);
|
||||
return null;
|
||||
}
|
||||
});
|
||||
return memoizePerRequest(
|
||||
"assistant-flags",
|
||||
async () => {
|
||||
try {
|
||||
const settings = await prisma.appSettings.findUnique({
|
||||
where: { id: "singleton" },
|
||||
select: {
|
||||
assistantEnabled: true,
|
||||
assistantCoachEnabled: true,
|
||||
assistantBriefingEnabled: true,
|
||||
assistantInsightStatusEnabled: true,
|
||||
assistantDocumentAiEnabled: true,
|
||||
},
|
||||
});
|
||||
return resolveAssistantFlags({
|
||||
enabled: settings?.assistantEnabled ?? true,
|
||||
coach: settings?.assistantCoachEnabled ?? true,
|
||||
briefing: settings?.assistantBriefingEnabled ?? true,
|
||||
insightStatus: settings?.assistantInsightStatusEnabled ?? true,
|
||||
documentAi: settings?.assistantDocumentAiEnabled ?? true,
|
||||
});
|
||||
} catch {
|
||||
getEvent()?.addWarning(
|
||||
"Failed to load assistant switches; every AI capability is off for this request",
|
||||
);
|
||||
return null;
|
||||
}
|
||||
},
|
||||
{ freshInBackground: true },
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -288,6 +288,11 @@ export class WideEventBuilder {
|
||||
return structuredClone(this.event) as WideEvent;
|
||||
}
|
||||
|
||||
/** `http` for a request, `background` for a job or a hand-rolled route. */
|
||||
getKind(): EventKind {
|
||||
return this.event.kind ?? "http";
|
||||
}
|
||||
|
||||
getDurationMs(): number {
|
||||
return Math.round(performance.now() - this.startTime);
|
||||
}
|
||||
|
||||
+26
-22
@@ -169,28 +169,32 @@ export function resolveModuleEnabled(
|
||||
|
||||
/** Internal: load the gate inputs once per request. */
|
||||
function loadInputs(userId: string): Promise<ModuleGateInputs> {
|
||||
return memoizePerRequest(`module-gate-inputs:${userId}`, async () => {
|
||||
const [user, cycleProfile] = await Promise.all([
|
||||
prisma.user.findUnique({
|
||||
where: { id: userId },
|
||||
select: {
|
||||
gender: true,
|
||||
disableCoach: true,
|
||||
modulePreferencesJson: true,
|
||||
},
|
||||
}),
|
||||
prisma.cycleProfile.findUnique({
|
||||
where: { userId },
|
||||
select: { cycleTrackingEnabled: true },
|
||||
}),
|
||||
]);
|
||||
return {
|
||||
gender: user?.gender ?? null,
|
||||
disableCoach: user?.disableCoach ?? false,
|
||||
modulePreferences: normalisePrefs(user?.modulePreferencesJson),
|
||||
cycleTrackingEnabled: cycleProfile?.cycleTrackingEnabled ?? null,
|
||||
};
|
||||
});
|
||||
return memoizePerRequest(
|
||||
`module-gate-inputs:${userId}`,
|
||||
async () => {
|
||||
const [user, cycleProfile] = await Promise.all([
|
||||
prisma.user.findUnique({
|
||||
where: { id: userId },
|
||||
select: {
|
||||
gender: true,
|
||||
disableCoach: true,
|
||||
modulePreferencesJson: true,
|
||||
},
|
||||
}),
|
||||
prisma.cycleProfile.findUnique({
|
||||
where: { userId },
|
||||
select: { cycleTrackingEnabled: true },
|
||||
}),
|
||||
]);
|
||||
return {
|
||||
gender: user?.gender ?? null,
|
||||
disableCoach: user?.disableCoach ?? false,
|
||||
modulePreferences: normalisePrefs(user?.modulePreferencesJson),
|
||||
cycleTrackingEnabled: cycleProfile?.cycleTrackingEnabled ?? null,
|
||||
};
|
||||
},
|
||||
{ freshInBackground: true },
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -104,13 +104,17 @@ async function loadAvailabilityBlob(): Promise<unknown> {
|
||||
* all-available so the modules stay visible on first boot.
|
||||
*/
|
||||
export async function getOperatorModuleAvailability(): Promise<OperatorModuleAvailability> {
|
||||
return memoizePerRequest("operator-module-availability", async () => {
|
||||
const [blob, switches] = await Promise.all([
|
||||
loadAvailabilityBlob(),
|
||||
getAssistantFlags(),
|
||||
]);
|
||||
return resolveOperatorAvailability(blob, switches.coach);
|
||||
});
|
||||
return memoizePerRequest(
|
||||
"operator-module-availability",
|
||||
async () => {
|
||||
const [blob, switches] = await Promise.all([
|
||||
loadAvailabilityBlob(),
|
||||
getAssistantFlags(),
|
||||
]);
|
||||
return resolveOperatorAvailability(blob, switches.coach);
|
||||
},
|
||||
{ freshInBackground: true },
|
||||
);
|
||||
}
|
||||
|
||||
/** A module key whose operator layer the blob owns (every key but the Coach). */
|
||||
|
||||
@@ -35,6 +35,22 @@ function getCache(builder: WideEventBuilder): Map<string, unknown> {
|
||||
return cache;
|
||||
}
|
||||
|
||||
export interface MemoizeOptions {
|
||||
/**
|
||||
* Read afresh on every call inside a background event.
|
||||
*
|
||||
* A request is short and acts for one record, so one read of the operator's
|
||||
* switches serves all of it. A background job opens one wide event for its
|
||||
* whole run, and a nightly pass loops over every user inside it: a cached
|
||||
* switch, module map, provider presence or consent set would be the answer
|
||||
* from the start of the run, served to every later user and to the wire
|
||||
* re-check, so turning the Assistant off at 02:05 would not stop the pass
|
||||
* that began at 02:00. Every input that decides whether AI work may run
|
||||
* sets this.
|
||||
*/
|
||||
freshInBackground?: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* Memoise `factory()` for the lifetime of the current request. Returns
|
||||
* the factory's result directly when no event context is active so
|
||||
@@ -49,11 +65,15 @@ function getCache(builder: WideEventBuilder): Map<string, unknown> {
|
||||
export function memoizePerRequest<T>(
|
||||
key: string,
|
||||
factory: () => Promise<T>,
|
||||
options: MemoizeOptions = {},
|
||||
): Promise<T> {
|
||||
const event = getEvent();
|
||||
if (!event) {
|
||||
return factory();
|
||||
}
|
||||
if (options.freshInBackground && event.getKind() === "background") {
|
||||
return factory();
|
||||
}
|
||||
const cache = getCache(event);
|
||||
const existing = cache.get(key);
|
||||
if (existing !== undefined) {
|
||||
|
||||
@@ -0,0 +1,98 @@
|
||||
/**
|
||||
* A switch turned off while a nightly job is running stops the rest of it.
|
||||
*
|
||||
* Reads inside one request are memoised on the request's wide event, so a
|
||||
* page that resolves the AI capabilities five times reads the settings row
|
||||
* once. A background job opens one wide event for its whole run, and the
|
||||
* status crons loop over every candidate inside it: memoising there meant the
|
||||
* switches, the modules and the consent kinds were read once at the start of
|
||||
* the run and served to every later user, and the wire re-check read the same
|
||||
* cached answer. Turning the Assistant off at 02:05 did not stop the 02:00
|
||||
* pass.
|
||||
*
|
||||
* Driven here over the real loop, the real capability loader and Postgres:
|
||||
* the first user's generation turns the switch off, and the remaining users
|
||||
* must not reach a generator, while the wire re-check inside the first user's
|
||||
* own run must see the change too.
|
||||
*/
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
import { getPrismaClient, truncateAllTables } from "./setup";
|
||||
|
||||
const candidates = vi.hoisted(() => ({ ids: [] as string[] }));
|
||||
|
||||
vi.mock("@/lib/jobs/status-cron-candidates", () => ({
|
||||
findStatusCronCandidates: vi.fn(async () =>
|
||||
candidates.ids.map((id) => ({ id, locale: "en" })),
|
||||
),
|
||||
}));
|
||||
|
||||
import { runStatusCronGenerate } from "@/lib/jobs/reminder/insights-handlers";
|
||||
import { aiEgressRefusal } from "@/lib/ai/capabilities/egress";
|
||||
|
||||
let counter = 0;
|
||||
|
||||
async function seedUserWithProvider(): Promise<string> {
|
||||
const n = counter++;
|
||||
const user = await getPrismaClient().user.create({
|
||||
data: {
|
||||
username: `fresh-switch-${n}`,
|
||||
email: `fresh-switch-${n}@example.test`,
|
||||
role: "USER",
|
||||
timezone: "UTC",
|
||||
locale: "en",
|
||||
aiProvider: "ANTHROPIC",
|
||||
aiAnthropicKeyEncrypted: "v1:presence-only",
|
||||
consentReceipts: {
|
||||
create: { kind: "ai_full", artefact: "test", signedAt: new Date() },
|
||||
},
|
||||
},
|
||||
});
|
||||
return user.id;
|
||||
}
|
||||
|
||||
async function setInsightStatusSwitch(enabled: boolean): Promise<void> {
|
||||
await getPrismaClient().appSettings.upsert({
|
||||
where: { id: "singleton" },
|
||||
create: { id: "singleton", assistantInsightStatusEnabled: enabled },
|
||||
update: { assistantInsightStatusEnabled: enabled },
|
||||
});
|
||||
}
|
||||
|
||||
beforeEach(async () => {
|
||||
await truncateAllTables(getPrismaClient());
|
||||
await setInsightStatusSwitch(true);
|
||||
});
|
||||
|
||||
describe("a switch turned off mid-run", () => {
|
||||
it("stops the remaining users of the same background pass", async () => {
|
||||
candidates.ids = [
|
||||
await seedUserWithProvider(),
|
||||
await seedUserWithProvider(),
|
||||
await seedUserWithProvider(),
|
||||
];
|
||||
const reached: string[] = [];
|
||||
const wire: Array<string | null> = [];
|
||||
|
||||
const generate = vi.fn(async (userId: string) => {
|
||||
reached.push(userId);
|
||||
// The wire re-check before the switch moves: open.
|
||||
wire.push(
|
||||
(await aiEgressRefusal("statusText", userId, ["anthropic"]))?.reason ??
|
||||
null,
|
||||
);
|
||||
await setInsightStatusSwitch(false);
|
||||
// The same user's wire re-check after it moved: closed.
|
||||
wire.push(
|
||||
(await aiEgressRefusal("statusText", userId, ["anthropic"]))?.reason ??
|
||||
null,
|
||||
);
|
||||
});
|
||||
|
||||
await runStatusCronGenerate("job.test.fresh_switch", generate);
|
||||
|
||||
expect(reached).toEqual([candidates.ids[0]]);
|
||||
expect(wire[0]).toBeNull();
|
||||
expect(wire[1]).not.toBeNull();
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user