fix(rectification): 请求内 Case 档案只读投影写即失效缓存
同一请求里 dossier/compute 按 (fn, userId, caseId) 合并重复读，其它 RPC 与 .from 立即失效。路由入口各包一处，零调用点改动。
This commit is contained in:
jesse-ux
@@ -0,0 +1,185 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { readFileSync } from "node:fs";
|
||||
import test from "node:test";
|
||||
|
||||
import { withRectificationRequestCache } from "../src/lib/rectification-agentic/v9/request-cache.ts";
|
||||
import { loadV9CaseDossier } from "../src/lib/rectification-agentic/v9/tool-service.ts";
|
||||
import { runV9AgentTurn } from "../src/lib/rectification-agentic/v9/agent-run.ts";
|
||||
import { RECTIFICATION_SKILL_NAME } from "../src/lib/rectification-agentic/v9/case-status.ts";
|
||||
import {
|
||||
CASE_COMPUTE_RPC,
|
||||
CASE_DOSSIER_RPC,
|
||||
CASE_ID,
|
||||
SESSION_ID,
|
||||
TURN_ID,
|
||||
USER_ID,
|
||||
countCachedProjectionRpcs,
|
||||
dossierFixture,
|
||||
fakeAccounting,
|
||||
receiptHandlers,
|
||||
} from "./rectification-v9-test-support.ts";
|
||||
|
||||
/** Measured after wrapping: route-style prefetch + one evidence turn. */
|
||||
const CACHED_AGENT_TURN_DOSSIER_BUDGET = 4;
|
||||
|
||||
const WRAPPED_ADMIN_CLIENT = /accounting = withRectificationRequestCache\(createAdminSupabaseClient\(\)\)/;
|
||||
const BARE_ADMIN_CLIENT = /accounting = createAdminSupabaseClient\(\)/;
|
||||
|
||||
const CACHED_ROUTES = [
|
||||
"../src/app/api/rectification/agent/route.ts",
|
||||
"../src/app/api/rectification/cases/[caseId]/turns/[turnId]/regenerate/route.ts",
|
||||
"../src/app/api/rectification/cases/[caseId]/route.ts",
|
||||
"../src/app/api/rectification/cases/[caseId]/repair-exit/route.ts",
|
||||
"../src/app/api/rectification/cases/[caseId]/candidates/accept/route.ts",
|
||||
] as const;
|
||||
|
||||
function projectionArgs(caseId = CASE_ID) {
|
||||
return { p_user_id: USER_ID, p_case_id: caseId };
|
||||
}
|
||||
|
||||
function delayedClient(innerCalls: string[], delayMs = 20) {
|
||||
return {
|
||||
rpc(fn: string, args: Record<string, unknown>) {
|
||||
void args;
|
||||
innerCalls.push(fn);
|
||||
return new Promise<{ data: unknown; error: null }>((resolve) => {
|
||||
setTimeout(() => resolve({ data: { ok: fn }, error: null }), delayMs);
|
||||
});
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
test("repeated case_dossier reads of the same user and case hit the in-flight cache once", async () => {
|
||||
const accounting = fakeAccounting({
|
||||
[CASE_DOSSIER_RPC]: () => dossierFixture(),
|
||||
}, { requestCache: true });
|
||||
const first = await loadV9CaseDossier(accounting.client, USER_ID, CASE_ID);
|
||||
const second = await loadV9CaseDossier(accounting.client, USER_ID, CASE_ID);
|
||||
assert.equal(first.case.caseId, CASE_ID);
|
||||
assert.equal(second.case.caseId, CASE_ID);
|
||||
assert.deepEqual(countCachedProjectionRpcs(accounting.calls), { dossier: 1, compute: 0 });
|
||||
});
|
||||
|
||||
test("a non-projection rpc invalidates so the next dossier read hits the backing client", async () => {
|
||||
const accounting = fakeAccounting({
|
||||
[CASE_DOSSIER_RPC]: () => dossierFixture(),
|
||||
append_agentic_rectification_turn: () => ({ turn_id: TURN_ID }),
|
||||
}, { requestCache: true });
|
||||
await accounting.client.rpc(CASE_DOSSIER_RPC, projectionArgs());
|
||||
await accounting.client.rpc(CASE_DOSSIER_RPC, projectionArgs());
|
||||
await accounting.client.rpc("append_agentic_rectification_turn", projectionArgs());
|
||||
await accounting.client.rpc(CASE_DOSSIER_RPC, projectionArgs());
|
||||
assert.equal(countCachedProjectionRpcs(accounting.calls).dossier, 2);
|
||||
});
|
||||
|
||||
test("two caseIds do not share a cached projection", async () => {
|
||||
const otherCaseId = "99999999-9999-4999-8999-999999999999";
|
||||
const accounting = fakeAccounting({
|
||||
[CASE_DOSSIER_RPC]: () => dossierFixture(),
|
||||
}, { requestCache: true });
|
||||
await accounting.client.rpc(CASE_DOSSIER_RPC, projectionArgs(CASE_ID));
|
||||
await accounting.client.rpc(CASE_DOSSIER_RPC, projectionArgs(otherCaseId));
|
||||
await accounting.client.rpc(CASE_DOSSIER_RPC, projectionArgs(CASE_ID));
|
||||
assert.equal(
|
||||
accounting.calls.filter((call) => call.fn === CASE_DOSSIER_RPC).length,
|
||||
2,
|
||||
);
|
||||
});
|
||||
|
||||
test("concurrent duplicate reads share one backing rpc", async () => {
|
||||
const innerCalls: string[] = [];
|
||||
const wrapped = withRectificationRequestCache(delayedClient(innerCalls));
|
||||
const args = projectionArgs();
|
||||
const [first, second] = await Promise.all([
|
||||
wrapped.rpc(CASE_DOSSIER_RPC, args),
|
||||
wrapped.rpc(CASE_DOSSIER_RPC, args),
|
||||
]);
|
||||
assert.deepEqual(first, { data: { ok: CASE_DOSSIER_RPC }, error: null });
|
||||
assert.deepEqual(second, { data: { ok: CASE_DOSSIER_RPC }, error: null });
|
||||
assert.deepEqual(innerCalls, [CASE_DOSSIER_RPC]);
|
||||
});
|
||||
|
||||
test("dossier and compute projections cache independently", async () => {
|
||||
const accounting = fakeAccounting({
|
||||
[CASE_DOSSIER_RPC]: () => dossierFixture(),
|
||||
[CASE_COMPUTE_RPC]: () => ({ case_id: CASE_ID }),
|
||||
}, { requestCache: true });
|
||||
await accounting.client.rpc(CASE_DOSSIER_RPC, projectionArgs());
|
||||
await accounting.client.rpc(CASE_COMPUTE_RPC, projectionArgs());
|
||||
await accounting.client.rpc(CASE_DOSSIER_RPC, projectionArgs());
|
||||
await accounting.client.rpc(CASE_COMPUTE_RPC, projectionArgs());
|
||||
assert.deepEqual(countCachedProjectionRpcs(accounting.calls), { dossier: 1, compute: 1 });
|
||||
});
|
||||
|
||||
test(".from passthrough invalidates the projection cache", async () => {
|
||||
const accounting = fakeAccounting({
|
||||
[CASE_DOSSIER_RPC]: () => dossierFixture(),
|
||||
}, { requestCache: true });
|
||||
await accounting.client.rpc(CASE_DOSSIER_RPC, projectionArgs());
|
||||
await accounting.client.from("profiles").update({ birth_time_source: "approximate" }).eq("id", USER_ID);
|
||||
await accounting.client.rpc(CASE_DOSSIER_RPC, projectionArgs());
|
||||
assert.equal(countCachedProjectionRpcs(accounting.calls).dossier, 2);
|
||||
assert.equal(accounting.profilePatches.length, 1);
|
||||
});
|
||||
|
||||
test("rectification routes wrap createAdminSupabaseClient once and never use the bare client", () => {
|
||||
for (const relative of CACHED_ROUTES) {
|
||||
const source = readFileSync(new URL(relative, import.meta.url), "utf8");
|
||||
const wraps = source.match(new RegExp(WRAPPED_ADMIN_CLIENT.source, "g")) ?? [];
|
||||
assert.equal(wraps.length, 1, relative);
|
||||
assert.equal(BARE_ADMIN_CLIENT.test(source), false, relative);
|
||||
}
|
||||
});
|
||||
|
||||
test("a cached evidence turn stays within the measured case_dossier budget", async () => {
|
||||
const handlers = {
|
||||
...receiptHandlers,
|
||||
[CASE_DOSSIER_RPC]: () => dossierFixture(),
|
||||
append_agentic_rectification_turn: () => ({ turn_id: TURN_ID }),
|
||||
finalize_agentic_rectification_turn: () => ({ turn_id: TURN_ID, status: "completed", idempotent: false }),
|
||||
};
|
||||
const unwrapped = fakeAccounting(handlers);
|
||||
const wrapped = fakeAccounting(handlers, { requestCache: true });
|
||||
const stream = {
|
||||
stream: async () => ({
|
||||
fullStream: (async function* () {
|
||||
yield { type: "start" };
|
||||
yield { type: "tool-call", payload: { toolName: "skill", args: { name: RECTIFICATION_SKILL_NAME } } };
|
||||
yield { type: "tool-result", payload: { toolName: "skill" } };
|
||||
yield { type: "tool-call", payload: { toolName: "rectification-read-case", args: { caseId: CASE_ID } } };
|
||||
yield { type: "tool-result", payload: { toolName: "rectification-read-case" } };
|
||||
yield { type: "text-delta", payload: { text: "记下了。" } };
|
||||
yield { type: "finish" };
|
||||
})(),
|
||||
totalUsage: Promise.resolve({ inputTokens: 10, outputTokens: 20 }),
|
||||
}),
|
||||
getSkill: async () => ({ name: RECTIFICATION_SKILL_NAME, instructions: "skill" }),
|
||||
};
|
||||
const base = {
|
||||
userId: USER_ID,
|
||||
caseId: CASE_ID,
|
||||
sessionId: SESSION_ID,
|
||||
requestId: "aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee",
|
||||
action: "evidence" as const,
|
||||
message: "2016年9月离开家去北京工作",
|
||||
modelName: "gpt-4o-mini",
|
||||
billing: {
|
||||
reserve: async () => ({ success: true, status: 200 }),
|
||||
complete: async () => true,
|
||||
release: async () => true,
|
||||
},
|
||||
emit: () => {},
|
||||
buildAgent: async () => stream as never,
|
||||
};
|
||||
await loadV9CaseDossier(unwrapped.client, USER_ID, CASE_ID);
|
||||
await runV9AgentTurn({ ...base, accounting: unwrapped.client });
|
||||
await loadV9CaseDossier(wrapped.client, USER_ID, CASE_ID);
|
||||
await runV9AgentTurn({ ...base, accounting: wrapped.client });
|
||||
const before = countCachedProjectionRpcs(unwrapped.calls);
|
||||
const after = countCachedProjectionRpcs(wrapped.calls);
|
||||
assert.ok(after.dossier < before.dossier);
|
||||
assert.ok(
|
||||
after.dossier <= CACHED_AGENT_TURN_DOSSIER_BUDGET,
|
||||
`cached dossier rpc ${after.dossier} exceeded budget ${CACHED_AGENT_TURN_DOSSIER_BUDGET} (uncached ${before.dossier})`,
|
||||
);
|
||||
});
|
||||
Reference in New Issue
Block a user