fix: expose dynamic scoring persistence

2026-07-19 07:50:23 +08:00
parent a8629043d8
commit 45b63d5cef
6 changed files with 430 additions and 7 deletions
@@ -5,6 +5,8 @@ import {
 } from "./birth-time-journey-turn-protocol.ts";
 import type { DynamicJourneyTurnState } from "./birth-time-journey-turn-protocol.ts";
 import type {
+  DynamicScoringJobCommand,
+  DynamicScoringJobFailureCommand,
  DynamicStoredRectificationCase,
  LegacyStoredRectificationCase,
  StoredRectificationCase,
@@ -68,6 +70,7 @@ function privateState(value: DynamicStoredRectificationCase): DynamicPrivateJour

 function isStaleRpc(error: RpcError): boolean {
  return error.message.includes("stale_birth_time_dynamic_turn")
+    || error.message.includes("stale_birth_time_dynamic_scoring_job")
    || error.message.includes("stale_birth_time_legacy_upgrade");
 }

@@ -84,6 +87,43 @@ export function createDynamicTurnPersistence(
    return loaded;
  }

+  async function savedDynamicScoring(
+    value: DynamicStoredRectificationCase,
+    expectedVersion: number,
+    result: RpcResult,
+  ): Promise<DynamicStoredRectificationCase> {
+    if (result.error) {
+      if (isStaleRpc(result.error)) {
+        const current = await loadCase(value.userId, value.id);
+        throw new StaleJourneyTurnError(
+          value.id,
+          expectedVersion,
+          current?.turnVersion ?? 0,
+        );
+      }
+      throw new BirthTimeJourneyStoreError("update_case");
+    }
+    const version = rpcVersionSchema.safeParse(result.data);
+    if (!version.success || version.data !== expectedVersion + 1) {
+      throw new BirthTimeJourneyStoreError("update_case");
+    }
+    const loaded = await loadedDynamic(value.userId, value.id);
+    if (
+      loaded.turnVersion !== version.data
+      || loaded.dynamicTurnState.turnVersion !== version.data
+    ) throw new BirthTimeJourneyStoreError("load_case");
+    return loaded;
+  }
+
+  function scoringIdentity(command: DynamicScoringJobCommand) {
+    return {
+      p_job_id: command.jobId,
+      p_expected_version: command.expectedVersion,
+      p_evidence_fingerprint: command.evidenceFingerprint,
+      p_algorithm_version: command.algorithmVersion,
+    };
+  }
+
  return {
    async saveDynamicTurn(
      value: DynamicStoredRectificationCase,
@@ -119,6 +159,38 @@ export function createDynamicTurnPersistence(
      return loadedDynamic(value.userId, value.id);
    },

+    async completeDynamicScoringJob(
+      value: DynamicStoredRectificationCase,
+      command: DynamicScoringJobCommand,
+    ): Promise<DynamicStoredRectificationCase> {
+      if (!value.candidateResult) throw new BirthTimeJourneyStoreError("update_case");
+      const result = await client.rpc("complete_birth_time_dynamic_scoring_job", {
+        p_user_id: value.userId,
+        p_case_id: value.id,
+        ...scoringIdentity(command),
+        p_public_turn_state: publicTurn(value, command.expectedVersion + 1),
+        p_snapshot: value.snapshot,
+        p_candidate_result: value.candidateResult,
+        p_private_state: privateState(value),
+      });
+      return savedDynamicScoring(value, command.expectedVersion, result);
+    },
+
+    async failDynamicScoringJob(
+      value: DynamicStoredRectificationCase,
+      command: DynamicScoringJobFailureCommand,
+    ): Promise<DynamicStoredRectificationCase> {
+      const result = await client.rpc("fail_birth_time_dynamic_scoring_job", {
+        p_user_id: value.userId,
+        p_case_id: value.id,
+        ...scoringIdentity(command),
+        p_failure_code: command.failureCode,
+        p_public_turn_state: publicTurn(value, command.expectedVersion + 1),
+        p_private_state: privateState(value),
+      });
+      return savedDynamicScoring(value, command.expectedVersion, result);
+    },
+
    async upgradeLegacyActiveCase(
      value: LegacyStoredRectificationCase,
    ): Promise<StoredRectificationCase> {
@@ -133,16 +133,27 @@ export type LegacyStoredRectificationCase = StoredRectificationCaseBase
 export type DynamicStoredRectificationCase = StoredRectificationCaseBase
  & DynamicStoredFields;

-export type StoredRectificationCase =
-  | LegacyStoredRectificationCase
+export type StoredRectificationCase = LegacyStoredRectificationCase
  | DynamicStoredRectificationCase;

+export type DynamicScoringJobCommand = {
+  readonly expectedVersion: number;
+  readonly jobId: string;
+  readonly evidenceFingerprint: string;
+  readonly algorithmVersion: string;
+};
+
+export type DynamicScoringJobFailureCommand = DynamicScoringJobCommand
+  & { readonly failureCode: string };
+
 export interface BirthTimeJourneyStore {
  saveAssessment(value: PersistedJourneyAssessment): Promise<string>;
  loadCase(userId: string, caseId: string): Promise<StoredRectificationCase | null>;
  saveScoring(value: LegacyStoredRectificationCase): Promise<void>;
  saveTurn(value: LegacyStoredRectificationCase, expectedVersion: number, actionId: string): Promise<StoredRectificationCase>;
  saveDynamicTurn(value: DynamicStoredRectificationCase, expectedVersion: number, actionId: string): Promise<DynamicStoredRectificationCase>;
+  completeDynamicScoringJob(value: DynamicStoredRectificationCase, command: DynamicScoringJobCommand): Promise<DynamicStoredRectificationCase>;
+  failDynamicScoringJob(value: DynamicStoredRectificationCase, command: DynamicScoringJobFailureCommand): Promise<DynamicStoredRectificationCase>;
  upgradeLegacyActiveCase(value: LegacyStoredRectificationCase): Promise<StoredRectificationCase>;
  createScoringJob(value: LegacyStoredRectificationCase, expectedVersion: number, actionId: string, job: ScoringJobSpec): Promise<StoredRectificationCase>;
  claimScoringJob(identity: ScoringJobIdentity): Promise<ScoringJobClaim>;
@@ -0,0 +1,65 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import { StaleJourneyTurnError } from "../src/lib/birth-time-journey-turn-persistence.ts";
+import { dynamicCase } from "./birth-time-dynamic-persistence-fixture.ts";
+import { memoryStore } from "./birth-time-journey-memory-store.ts";
+import { lowCandidate } from "./birth-time-journey-test-support.ts";
+
+const jobA = "8c9d09e8-91b6-4335-b891-122f205a050c";
+const jobB = "dc6f3fdc-b679-4878-a3f4-1037fd1ababb";
+const baseCommand = {
+  expectedVersion: 7,
+  jobId: jobA,
+  evidenceFingerprint: "evidence-fingerprint",
+  algorithmVersion: "birth-time-event-scoring-v1",
+};
+
+function completingCase() {
+  const stored = dynamicCase();
+  return { ...stored, candidateResult: lowCandidate };
+}
+
+test("shared memory store replays only the identical scoring completion", async () => {
+  const value = completingCase();
+  const memory = memoryStore(dynamicCase());
+  const first = await memory.store.completeDynamicScoringJob(value, baseCommand);
+
+  assert.equal(await memory.store.completeDynamicScoringJob(value, baseCommand), first);
+  for (const command of [
+    { ...baseCommand, evidenceFingerprint: "changed" },
+    { ...baseCommand, algorithmVersion: "birth-time-event-scoring-v2" },
+  ]) {
+    await assert.rejects(
+      memory.store.completeDynamicScoringJob(value, command),
+      StaleJourneyTurnError,
+    );
+  }
+  await assert.rejects(
+    memory.store.completeDynamicScoringJob({
+      ...value,
+      candidateResult: { ...lowCandidate, topScore: 9 },
+    }, baseCommand),
+    StaleJourneyTurnError,
+  );
+  await assert.rejects(
+    memory.store.failDynamicScoringJob(dynamicCase(), {
+      ...baseCommand,
+      jobId: jobB,
+      failureCode: "engine_unavailable",
+    }),
+    StaleJourneyTurnError,
+  );
+});
+
+test("shared memory store replays only the identical scoring failure", async () => {
+  const value = dynamicCase();
+  const memory = memoryStore(dynamicCase());
+  const command = { ...baseCommand, failureCode: "engine_unavailable" };
+  const first = await memory.store.failDynamicScoringJob(value, command);
+
+  assert.equal(await memory.store.failDynamicScoringJob(value, command), first);
+  await assert.rejects(
+    memory.store.failDynamicScoringJob(value, { ...command, failureCode: "timeout" }),
+    StaleJourneyTurnError,
+  );
+});
@@ -0,0 +1,205 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import { createDynamicTurnPersistence } from "../src/lib/birth-time-journey-dynamic-persistence.ts";
+import {
+  BirthTimeJourneyStoreError,
+  StaleJourneyTurnError,
+} from "../src/lib/birth-time-journey-turn-persistence.ts";
+import type { DynamicStoredRectificationCase } from "../src/lib/birth-time-journey-service.ts";
+import {
+  caseId,
+  dynamicCase,
+  ownerId,
+} from "./birth-time-dynamic-persistence-fixture.ts";
+import { lowCandidate } from "./birth-time-journey-test-support.ts";
+
+const jobId = "8c9d09e8-91b6-4335-b891-122f205a050c";
+const fingerprint = "evidence-fingerprint";
+const algorithmVersion = "birth-time-event-scoring-v1";
+
+function scoringCase(
+  action: "complete" | "fail",
+): DynamicStoredRectificationCase {
+  const stored = dynamicCase();
+  return {
+    ...stored,
+    candidateResult: action === "complete" ? lowCandidate : null,
+    dynamicTurnState: {
+      ...stored.dynamicTurnState,
+      nextAction: action === "complete"
+        ? { kind: "present_low_result", resultId: lowCandidate.resultId }
+        : { kind: "retry_scoring", jobId },
+      progress: { ...stored.dynamicTurnState.progress, phase: action === "complete" ? "result" : "scoring" },
+    },
+  };
+}
+
+function successfulRpc(target: DynamicStoredRectificationCase) {
+  let stored = dynamicCase();
+  let committed = false;
+  const calls: { readonly name: string; readonly args: Readonly<Record<string, unknown>> }[] = [];
+  const persistence = createDynamicTurnPersistence({
+    async rpc(name, args) {
+      calls.push({ name, args });
+      if (!committed) {
+        stored = {
+          ...target,
+          turnVersion: 8,
+          dynamicTurnState: { ...target.dynamicTurnState, turnVersion: 8 },
+        };
+        committed = true;
+      }
+      return { data: 8, error: null };
+    },
+  }, async () => stored, () => "2026-07-18");
+  return { calls, persistence };
+}
+
+function assertNoPublicLeak(args: Readonly<Record<string, unknown>>) {
+  const publicPayload = JSON.stringify({
+    turn: args.p_public_turn_state,
+    snapshot: args.p_snapshot,
+    candidate: args.p_candidate_result,
+  });
+  for (const forbidden of [
+    "partitionId",
+    "candidateScores",
+    "agentContext",
+    "active_birth_time",
+    "birth_time",
+    "activeBirthTime",
+    "birthTime",
+  ]) assert.equal(publicPayload.includes(forbidden), false);
+}
+
+test("dynamic scoring completion calls the exact RPC and replays the stored turn", async () => {
+  const value = scoringCase("complete");
+  const fake = successfulRpc(value);
+  const command = { expectedVersion: 7, jobId, evidenceFingerprint: fingerprint, algorithmVersion };
+
+  const first = await fake.persistence.completeDynamicScoringJob(value, command);
+  const replay = await fake.persistence.completeDynamicScoringJob(value, command);
+
+  assert.equal(first.turnVersion, 8);
+  assert.equal(replay, first);
+  assert.deepEqual(fake.calls.map((call) => call.name), [
+    "complete_birth_time_dynamic_scoring_job",
+    "complete_birth_time_dynamic_scoring_job",
+  ]);
+  assert.deepEqual(fake.calls[0]?.args, {
+    p_user_id: ownerId,
+    p_case_id: caseId,
+    p_job_id: jobId,
+    p_expected_version: 7,
+    p_evidence_fingerprint: fingerprint,
+    p_algorithm_version: algorithmVersion,
+    p_public_turn_state: { ...value.dynamicTurnState, turnVersion: 8 },
+    p_snapshot: value.snapshot,
+    p_candidate_result: lowCandidate,
+    p_private_state: {
+      candidateModel: value.candidateModel,
+      currentChoiceQuestion: value.currentChoiceQuestion,
+      choiceAnswers: value.choiceAnswers,
+      choiceEvidence: value.choiceEvidence,
+      dynamicControl: value.dynamicControl,
+      agentContext: value.agentContext,
+    },
+  });
+  assertNoPublicLeak(fake.calls[0]?.args ?? {});
+});
+
+test("dynamic scoring failure calls the exact RPC and replays the stored turn", async () => {
+  const value = scoringCase("fail");
+  const fake = successfulRpc(value);
+  const command = {
+    expectedVersion: 7,
+    jobId,
+    evidenceFingerprint: fingerprint,
+    algorithmVersion,
+    failureCode: "engine_unavailable",
+  };
+
+  const first = await fake.persistence.failDynamicScoringJob(value, command);
+  const replay = await fake.persistence.failDynamicScoringJob(value, command);
+
+  assert.equal(first.turnVersion, 8);
+  assert.equal(replay, first);
+  assert.deepEqual(fake.calls.map((call) => call.name), [
+    "fail_birth_time_dynamic_scoring_job",
+    "fail_birth_time_dynamic_scoring_job",
+  ]);
+  assert.deepEqual(fake.calls[0], {
+    name: "fail_birth_time_dynamic_scoring_job",
+    args: {
+      p_user_id: ownerId,
+      p_case_id: caseId,
+      p_job_id: jobId,
+      p_expected_version: 7,
+      p_evidence_fingerprint: fingerprint,
+      p_algorithm_version: algorithmVersion,
+      p_failure_code: "engine_unavailable",
+      p_public_turn_state: { ...value.dynamicTurnState, turnVersion: 8 },
+      p_private_state: {
+        candidateModel: value.candidateModel,
+        currentChoiceQuestion: value.currentChoiceQuestion,
+        choiceAnswers: value.choiceAnswers,
+        choiceEvidence: value.choiceEvidence,
+        dynamicControl: value.dynamicControl,
+        agentContext: value.agentContext,
+      },
+    },
+  });
+  assertNoPublicLeak(fake.calls[0]?.args ?? {});
+});
+
+test("dynamic scoring persistence rejects malformed versions and maps RPC errors", async () => {
+  const value = scoringCase("fail");
+  let currentVersion = 9;
+  let completionCalls = 0;
+  const persistence = createDynamicTurnPersistence({
+    async rpc(name) {
+      if (name === "complete_birth_time_dynamic_scoring_job") {
+        completionCalls += 1;
+        return { data: completionCalls === 1 ? "8" : 9, error: null };
+      }
+      if (name === "fail_birth_time_dynamic_scoring_job" && currentVersion === 9) {
+        return { data: null, error: { message: "stale_birth_time_dynamic_scoring_job" } };
+      }
+      return { data: null, error: { message: "database unavailable" } };
+    },
+  }, async () => ({
+    ...dynamicCase(),
+    turnVersion: currentVersion,
+    dynamicTurnState: { ...dynamicCase().dynamicTurnState, turnVersion: currentVersion },
+  }), () => "2026-07-18");
+
+  await assert.rejects(
+    persistence.completeDynamicScoringJob({ ...value, candidateResult: lowCandidate }, {
+      expectedVersion: 7, jobId, evidenceFingerprint: fingerprint, algorithmVersion,
+    }),
+    BirthTimeJourneyStoreError,
+  );
+  await assert.rejects(
+    persistence.completeDynamicScoringJob({ ...value, candidateResult: lowCandidate }, {
+      expectedVersion: 7, jobId, evidenceFingerprint: fingerprint, algorithmVersion,
+    }),
+    BirthTimeJourneyStoreError,
+  );
+  await assert.rejects(
+    persistence.failDynamicScoringJob(value, {
+      expectedVersion: 7, jobId, evidenceFingerprint: fingerprint,
+      algorithmVersion, failureCode: "engine_unavailable",
+    }),
+    (error) => error instanceof StaleJourneyTurnError
+      && error.expectedVersion === 7
+      && error.currentVersion === 9,
+  );
+  currentVersion = 10;
+  await assert.rejects(
+    persistence.failDynamicScoringJob(value, {
+      expectedVersion: 7, jobId, evidenceFingerprint: fingerprint,
+      algorithmVersion, failureCode: "engine_unavailable",
+    }),
+    BirthTimeJourneyStoreError,
+  );
+});
@@ -1,5 +1,8 @@
+import { isDeepStrictEqual } from "node:util";
 import type {
  BirthTimeJourneyStore,
+  DynamicScoringJobCommand,
+  DynamicScoringJobFailureCommand,
  DynamicStoredRectificationCase,
  PersistedJourneyAssessment,
  StoredRectificationCase,
@@ -17,6 +20,30 @@ class MissingTestCaseError extends Error {

 export const journeyCaseId = "7299894c-10a8-4b45-91d1-339007282c50";

+type DynamicScoringReceipt = {
+  readonly kind: "complete" | "fail";
+  readonly command: DynamicScoringJobCommand | DynamicScoringJobFailureCommand;
+  readonly result: DynamicStoredRectificationCase["candidateResult"];
+};
+
+function sameScoringOperation(
+  receipt: DynamicScoringReceipt,
+  next: DynamicScoringReceipt,
+): boolean {
+  const prior = receipt.command;
+  const command = next.command;
+  return receipt.kind === next.kind
+    && prior.expectedVersion === command.expectedVersion
+    && prior.jobId.toLowerCase() === command.jobId.toLowerCase()
+    && prior.evidenceFingerprint === command.evidenceFingerprint
+    && prior.algorithmVersion === command.algorithmVersion
+    && (receipt.kind !== "fail" || (
+      "failureCode" in prior && "failureCode" in command
+      && prior.failureCode === command.failureCode
+    ))
+    && (receipt.kind !== "complete" || isDeepStrictEqual(receipt.result, next.result));
+}
+
 export function memoryStore(
  initialCase?: StoredRectificationCase,
  asOfDate = "2026-07-18",
@@ -28,6 +55,7 @@ export function memoryStore(
  let committedTurnWrites = 0;
  let legacyWrites = 0;
  let guidedCandidateWrites = 0;
+  let dynamicScoringReceipt: DynamicScoringReceipt | null = null;
  const scoringJobs = createMemoryScoringJobs({
    read: () => savedCase,
    write: (value) => {
@@ -37,6 +65,31 @@ export function memoryStore(
      committedTurnWrites += 1;
    },
  });
+  function persistDynamicScoring(
+    value: DynamicStoredRectificationCase,
+    receipt: DynamicScoringReceipt,
+  ): DynamicStoredRectificationCase {
+    if (!savedDynamicCase) throw new MissingTestCaseError();
+    const expectedVersion = receipt.command.expectedVersion;
+    if (savedDynamicCase.turnVersion === expectedVersion + 1) {
+      if (dynamicScoringReceipt && sameScoringOperation(dynamicScoringReceipt, receipt)) {
+        return savedDynamicCase;
+      }
+      throw new StaleJourneyTurnError(value.id, expectedVersion, savedDynamicCase.turnVersion);
+    }
+    if (savedDynamicCase.turnVersion !== expectedVersion) {
+      throw new StaleJourneyTurnError(value.id, expectedVersion, savedDynamicCase.turnVersion);
+    }
+    const saved = {
+      ...value,
+      turnVersion: expectedVersion + 1,
+      dynamicTurnState: { ...value.dynamicTurnState, turnVersion: expectedVersion + 1 },
+    };
+    savedCase = saved;
+    savedDynamicCase = saved;
+    dynamicScoringReceipt = receipt;
+    return saved;
+  }
  const store: BirthTimeJourneyStore = {
    async saveAssessment(value) {
      savedAssessment = value;
@@ -93,6 +146,12 @@ export function memoryStore(
      committedTurnWrites += 1;
      return savedDynamic;
    },
+    async completeDynamicScoringJob(value, command) {
+      return persistDynamicScoring(value, { kind: "complete", command, result: value.candidateResult });
+    },
+    async failDynamicScoringJob(value, command) {
+      return persistDynamicScoring(value, { kind: "fail", command, result: null });
+    },
    async upgradeLegacyActiveCase(value) {
      if (isTerminalLegacyCase(value)) {
        return value;
@@ -161,6 +220,7 @@ export function memoryStore(
    replaceCase: (value: StoredRectificationCase) => {
      savedCase = value;
      savedDynamicCase = value.journeyProtocol === "dynamic-choice-v2" ? value : null;
+      dynamicScoringReceipt = null;
    },
  };
 }