Files
Jyotisha/frontend/tests/research/rectification-segment-persisted-replay.ts
T
2026-10-17 21:16:32 +08:00

352 lines
23 KiB
TypeScript

import assert from "node:assert/strict";
import { createHash, randomUUID } from "node:crypto";
import { spawnSync } from "node:child_process";
import { fileURLToPath } from "node:url";
import { closeLocalPostgresDataPool, createLocalPostgresDataClient } from "../../src/lib/db/local-postgres-client-core.ts";
import { resolveActiveSkillPackage } from "../../src/lib/skill-package-registry.ts";
import { stillValidCandidates } from "../../src/lib/rectification-agentic/core/credible-range.ts";
import { publicNextAction } from "../../src/lib/rectification-agentic/core/rectification-decision.ts";
import { DEFAULT_MAX_DISCRIMINATION_ROUNDS, type AnswerClass, type InferenceState } from "../../src/lib/rectification-agentic/core/types.ts";
import { rawMinuteWeights, segmentsForMinutes, summarizeSegmentWeights, targetChartsForDomain } from "../../src/lib/rectification-agentic/core/segment-summary.ts";
import { applyRectificationChoice, inferenceForPersistedAnswer, persistNextInterviewAfterChoice } from "../../src/lib/rectification-agentic/v9/answer-choice.ts";
import { CHOICE_ACTION, outcomeIdForOption } from "../../src/lib/rectification-agentic/v9/choice-action.ts";
import type { ChoiceKey } from "../../src/lib/rectification-agentic/v9/choice-card.ts";
import { decideAfterInferenceChange, rectificationFollowupCatalog } from "../../src/lib/rectification-agentic/v9/decision-from-dossier.ts";
import type { EvidenceKind } from "../../src/lib/rectification-agentic/v9/evidence-model.ts";
import { resolveRectificationProductDomain } from "../../src/lib/rectification-agentic/v9/product-domain.ts";
import { scoreAndPersistCurrentEvidence } from "../../src/lib/rectification-agentic/v9/score-persist.ts";
import { confirmV9Evidence, loadV9CaseDossier, proposeV9Evidence } from "../../src/lib/rectification-agentic/v9/tool-service.ts";
import { startPostgresFixture } from "../helpers/postgres-fixture.ts";
import { segmentReplayOracleAnswer } from "../../scripts/rectification-segment-oracle.ts";
// Fictional owned Case seeds only. Production turn/evidence/choice calls are
// never replaced by fixture inserts, including when the original ACL refuses.
type ReplayRow = {
case_key: string; radius: number; oracle_truth_minute: string;
snapshot: Record<string, unknown> & { birth_date: string; reported_birth_time: string };
range: { start_time: string; end_time: string };
events: { event_kind: EvidenceKind; domain: string; date_start: string; date_end: string; precision: string; summary: string }[];
};
type AnswerMode = "all_correct" | "flip_1" | "flip_3" | "flip_5";
type StudyInput = {
modes?: AnswerMode[];
orderings?: Array<"off" | "on">;
code_sha?: string;
};
const chunks: Buffer[] = [];
for await (const chunk of process.stdin) chunks.push(Buffer.from(chunk));
const input = JSON.parse(Buffer.concat(chunks).toString("utf8")) as { rows: ReplayRow[]; study?: StudyInput };
const study = input.study ?? null;
const orderingFlags = study ? (study.orderings ?? ["off"]).map((item) => item === "on") : [false, true];
const modes: AnswerMode[] = study?.modes?.length ? study.modes : ["all_correct"];
const maxRounds = study ? DEFAULT_MAX_DISCRIMINATION_ROUNDS : 6;
const passes = orderingFlags.flatMap((enabled) => modes.map((mode) => ({ enabled, mode })));
function caseLabel(key: string): string {
return createHash("sha256").update(key).digest("hex").slice(0, 16);
}
function shiftDate(day: string, delta: number): string {
const date = new Date(`${day}T00:00:00Z`);
date.setUTCDate(date.getUTCDate() + delta);
return date.toISOString().slice(0, 10);
}
function inclusiveMinutes(rows: readonly { start_at: string; end_at: string }[]): number {
return rows.reduce((sum, row) => {
const width = (Date.parse(`${row.end_at}:00Z`) - Date.parse(`${row.start_at}:00Z`)) / 60000 + 1;
return Number.isFinite(width) && width > 0 ? sum + width : sum;
}, 0);
}
function envelopeIntervals(state: InferenceState, birthDate: string): { start_at: string; end_at: string }[] {
if (state.credible_intervals?.length) {
return state.credible_intervals.map((row) => ({ start_at: row.start_at, end_at: row.end_at }));
}
const range = state.credible_range;
if (!range) return [];
if (range[0] <= range[1]) return [{ start_at: `${birthDate}T${range[0]}`, end_at: `${birthDate}T${range[1]}` }];
const previous = shiftDate(birthDate, -1);
return [
{ start_at: `${previous}T${range[0]}`, end_at: `${previous}T23:59` },
{ start_at: `${birthDate}T00:00`, end_at: `${birthDate}T${range[1]}` },
];
}
function containsTruth(rows: readonly { start_at: string; end_at: string }[], birthDate: string, minute: string): boolean {
const stamps = [-1, 0, 1].map((delta) => `${shiftDate(birthDate, delta)}T${minute}`);
return rows.some((row) => stamps.some((stamp) => stamp >= row.start_at && stamp <= row.end_at));
}
function median(values: number[]): number | null {
if (!values.length) return null;
const sorted = [...values].sort((left, right) => left - right);
const mid = Math.floor(sorted.length / 2);
return sorted.length % 2 ? sorted[mid]! : (sorted[mid - 1]! + sorted[mid]!) / 2;
}
function flippedClass(value: AnswerClass): AnswerClass {
return value === "no" ? "yes" : "no";
}
function dayOffset(birthDate: string, stamp: string): number {
const birth = Date.parse(`${birthDate}T00:00:00Z`);
const day = Date.parse(`${stamp.slice(0, 10)}T00:00:00Z`);
return Math.round((day - birth) / 86400000);
}
let displayFixture: Record<string, unknown> | null = null;
const records: Record<string, unknown>[] = [];
const previousOrder = process.env.RECTIFICATION_SEGMENT_ORDER;
const quote = (value: unknown) => `'${String(value).replaceAll("'", "''")}'`;
const fixture = startPostgresFixture();
const url = fixture.connectionUrl("service_runtime", "service-runtime-test-password");
const authenticatedUrl = fixture.connectionUrl("app_runtime", "app-runtime-test-password");
let fatal: string | null = null;
try {
const migrated = spawnSync(process.execPath, [fileURLToPath(new URL("../../scripts/db-migrate.mjs", import.meta.url))], {
encoding: "utf8", env: { ...process.env, SCHEMA_DATABASE_URL: fixture.connectionUrl("schema_owner", "schema-owner-test-password") },
});
assert.equal(migrated.status, 0, migrated.stderr);
const service = createLocalPostgresDataClient(url, null, "service_role");
const skill = resolveActiveSkillPackage("jyotish-birth-time-rectification");
for (const row of input.rows) {
for (const pass of passes) {
const enabled = pass.enabled;
const mode = pass.mode;
const flipAt = mode === "flip_1" ? 1 : mode === "flip_3" ? 3 : mode === "flip_5" ? 5 : null;
process.env.RECTIFICATION_SEGMENT_ORDER = enabled ? "on" : "off";
const userId = randomUUID(), caseId = randomUUID(), sessionId = randomUUID(), sourceSessionId = randomUUID();
const trace: Record<string, unknown> = {
case_key: study ? caseLabel(row.case_key) : row.case_key,
radius: row.radius,
...(study ? { mode } : {}),
ordering: enabled ? "on-experiment" : "off-default", opening: "explicit-fictional-case-seed-not-opening-success",
asked: [], gates: [], stop: null, stage: "seed", metrics: null };
const ledger = () => JSON.parse(fixture.psql(`select json_build_object(
'turns',(select count(*) from public.agentic_rectification_turns where case_id=${quote(caseId)}),
'evidence',(select count(*) from public.agentic_rectification_evidence where case_id=${quote(caseId)}),
'results',(select count(*) from public.agentic_rectification_results where case_id=${quote(caseId)}),
'actions',(select count(*) from public.agentic_rectification_choice_actions where case_id=${quote(caseId)}),
'transitions',(select count(*) from public.agentic_rectification_inference_transitions where case_id=${quote(caseId)}))`));
try {
fixture.psqlAs("identity_runtime", "identity-runtime-test-password", `insert into identity.users(id,name,email,email_verified)
values(${quote(userId)},'Fictional Public Replay Owner',${quote(`${userId}@example.invalid`)},true)`);
const snapshot = row.snapshot;
fixture.psql(`update public.profiles set birth_date=${quote(snapshot.birth_date)},reported_birth_time=${quote(snapshot.reported_birth_time)},
birth_time_source='approximate',birth_time_status='reported',uncertainty_before_minutes=${row.radius},uncertainty_after_minutes=${row.radius},
latitude=${Number(snapshot.latitude)},longitude=${Number(snapshot.longitude)},timezone_offset=${Number(snapshot.timezone_offset)},
timezone_id=${snapshot.timezone_id ? quote(snapshot.timezone_id) : "null"} where id=${quote(userId)};
insert into public.chat_sessions(id,user_id,title,theme,session_type,messages) values
(${quote(sourceSessionId)},${quote(userId)},'Fictional replay source','general','consultation','[]'),
(${quote(sessionId)},${quote(userId)},'Fictional persisted replay','general','birth_time_rectification','[]')`);
// Match POST /cases/open: owned source reads use the authenticated
// session client; the accounting client is reserved for Case RPCs.
const authenticated = createLocalPostgresDataClient(authenticatedUrl, { id: userId, email: null }, "authenticated");
const domain = await resolveRectificationProductDomain(
authenticated as unknown as Parameters<typeof resolveRectificationProductDomain>[0], userId,
{ intent: "homepage", sourceSessionId, requestId: randomUUID() },
);
fixture.psql(`insert into public.agentic_rectification_cases(id,user_id,session_id,status,skill_name,skill_version,skill_sha256,skill_source_commit,
baseline_profile_fingerprint,baseline_birth_snapshot,candidate_range,rectification_domain) values
(${quote(caseId)},${quote(userId)},${quote(sessionId)},'collecting_evidence',${quote(skill.name)},${quote(skill.version)},${quote(skill.sha256)},
${skill.sourceCommit ? quote(skill.sourceCommit) : "null"},${quote("a".repeat(64))},${quote(JSON.stringify(snapshot))},${quote(JSON.stringify(row.range))},${quote(domain)})`);
const seeded = await loadV9CaseDossier(service, userId, caseId);
assert.equal(seeded.case.rectificationDomain, domain);
trace.domain = seeded.case.rectificationDomain;
trace.targets = targetChartsForDomain(seeded.case.rectificationDomain);
trace.stage = "append_turn_v10";
trace.ledger_before = ledger();
// Same V10 request-idempotent overload as agent-run-prepare.ts; the V9 overload is revoked from service_role.
const appended = await service.rpc("append_agentic_rectification_turn", { p_user_id: userId, p_case_id: caseId,
p_user_message: row.events.map(event => event.summary).join("; "), p_assistant_message: null,
p_model_name: "public-benchmark-oracle-not-provider", p_model_version: null, p_status: "pending", p_request_id: randomUUID() });
assert.equal(appended.error, null, JSON.stringify(appended.error));
const turn = { turnId: (appended.data as { turn_id: string }).turn_id };
trace.stage = "evidence";
for (const event of row.events) {
const proposed = await proposeV9Evidence(service, userId, caseId, {
sourceTurnId: turn.turnId, quote: event.summary, subject: event.domain === "family" ? "family" : "self",
eventKind: event.event_kind, domain: event.domain, occurredFrom: event.date_start, occurredTo: event.date_end,
datePrecision: event.precision, summary: event.summary,
});
assert.equal(proposed.outcome, "accepted", proposed.errorCode ?? "evidence rejected");
assert.ok(proposed.evidenceId);
await confirmV9Evidence(service, userId, caseId, proposed.evidenceId);
}
trace.stage = "scoreAndPersistCurrentEvidence";
await scoreAndPersistCurrentEvidence({ accounting: service, userId, caseId });
let scoredOrdinal = 0;
let flipApplied = false;
for (let round = 0; round < maxRounds; round++) {
let dossier = await loadV9CaseDossier(service, userId, caseId);
const state = inferenceForPersistedAnswer(dossier.latestResult?.decisionReceipt);
const decision = decideAfterInferenceChange({ dossier, state, userStopped: false, birthDate: snapshot.birth_date });
const catalog = rectificationFollowupCatalog(dossier.latestResult, dossier.evidence);
(trace.gates as unknown[]).push({ round, session_outcome: decision.sessionOutcome,
stop_reason: decision.stopReason ?? null, next_action: publicNextAction(decision), revision: state?.revision ?? null,
eligible_event_keys: catalog.eventProbes.map(probe => probe.semantic_key).sort(),
contrast_keys: catalog.contrastPacket?.probes.map(probe => probe.semanticKey).sort() ?? [] });
trace.stage = "persistNextInterviewAfterChoice";
const interview = await persistNextInterviewAfterChoice({ accounting: service, userId, caseId, dossier,
decisionState: state, decision, nextAction: publicNextAction(decision), birthDate: snapshot.birth_date,
askedTurnId: turn.turnId });
if (!interview.choiceReady || interview.terminalNote) {
trace.stop = interview.terminalNote ? "production-terminal" : "production-no-renderable-choice";
break;
}
dossier = await loadV9CaseDossier(service, userId, caseId);
const focus = dossier.conversationSummary.activeFocus;
assert.ok(focus);
const ownedState = inferenceForPersistedAnswer(dossier.latestResult?.decisionReceipt);
const schema = focus.expectedAnswerSchema;
const probe = ownedState?.probes.find(probe => probe.id === schema.probe_id);
if (!probe) {
trace.stop = "oracle-missing-for-production-focus";
(trace.gates as unknown[]).push({ intent: focus.intent, source: interview.followup?.source ?? null,
choice_kind: schema.choice_kind ?? null, scoring: schema.scoring !== false });
break;
}
const oracleClass = segmentReplayOracleAnswer(probe, row.oracle_truth_minute, ownedState?.transitions);
const choice = schema.choice as { options?: { key: ChoiceKey }[] } | undefined;
const scoring = schema.scoring !== false;
if (scoring) scoredOrdinal += 1;
const wantFlip = flipAt !== null && scoring && scoredOrdinal === flipAt;
const answerClass = wantFlip ? flippedClass(oracleClass) : oracleClass;
const option = choice?.options?.find(option => outcomeIdForOption(option.key, schema) === answerClass);
if (!option) {
trace.stop = wantFlip ? "flip-option-missing" : "oracle-no-renderable-class-option";
break;
}
if (wantFlip) flipApplied = true;
const before = ledger();
trace.stage = "applyRectificationChoice";
const applied = await applyRectificationChoice(service, { userId, caseId, sessionId, actionId: randomUUID(),
action: CHOICE_ACTION, focusId: focus.id, optionId: option.key, expectedRevision: ownedState!.revision, deferFollowup: true });
assert.equal(applied.applied, true);
const reloaded = await loadV9CaseDossier(service, userId, caseId);
const afterState = inferenceForPersistedAnswer(reloaded.latestResult?.decisionReceipt);
(trace.asked as unknown[]).push({ round, semantic_key: probe.semantic_key, optionId: option.key,
answer_class: outcomeIdForOption(option.key, schema), source: probe.source, choice_kind: schema.choice_kind,
scoring: schema.scoring !== false, ledger_before: before, ledger_after: ledger(),
revision_before: ownedState!.revision, revision_after: afterState?.revision ?? null });
assert.equal(afterState?.revision, ownedState!.revision + 1);
if (round === maxRounds - 1) trace.stop = maxRounds === 6 ? "six-question-budget" : "round-cap";
}
const final = await loadV9CaseDossier(service, userId, caseId);
const state = inferenceForPersistedAnswer(final.latestResult?.decisionReceipt);
if (study && state) {
const envelope = envelopeIntervals(state, snapshot.birth_date);
const stored = (state.living_display_intervals ?? []).filter((row) => row?.start_at && row?.end_at);
const shown = stored.length ? stored : envelope;
const living = stillValidCandidates(state.candidates);
const livingRows = living.flatMap((row) => row.cluster_intervals ?? []);
const remaining = living.length > 0 && living.every((row) => (row.cluster_intervals?.length ?? 0) > 0)
? inclusiveMinutes(livingRows) : inclusiveMinutes(envelope);
const clocks = (rows: readonly { start_at: string; end_at: string }[]) => rows.map((row) => ({
start: row.start_at.slice(11, 16), end: row.end_at.slice(11, 16),
}));
trace.question_count = (trace.asked as unknown[]).length;
trace.flip_applied = flipApplied;
trace.truth_in_envelope = containsTruth(envelope, snapshot.birth_date, row.oracle_truth_minute);
trace.truth_in_display = containsTruth(shown, snapshot.birth_date, row.oracle_truth_minute);
trace.remaining_minutes = remaining;
trace.display_width_minutes = inclusiveMinutes(shown);
trace.display_source = stored.length ? "stored" : "envelope";
trace.credible_range = state.credible_range;
trace.credible_intervals = clocks(envelope);
trace.living_display_intervals = stored.length ? clocks(stored) : null;
if (!displayFixture && row.radius === 60 && mode === "all_correct" && stored.length >= 2) {
displayFixture = {
radius: 60,
mode: "all_correct",
display_width_minutes: inclusiveMinutes(shown),
segment_count: stored.length,
intervals: stored.map((part) => ({
day_offset: dayOffset(snapshot.birth_date, part.start_at),
start: part.start_at.slice(11, 16),
end: part.end_at.slice(11, 16),
})),
members: living.map((candidate) => ({
posterior_score: candidate.posterior_score,
intervals: (candidate.cluster_intervals ?? []).map((part) => ({
day_offset: dayOffset(snapshot.birth_date, part.start_at),
start: part.start_at.slice(11, 16),
end: part.end_at.slice(11, 16),
})),
})),
};
}
}
if (!study && state?.segment_minutes?.length) {
const offsets = Object.fromEntries(state.segment_minutes.map(minute => [minute.time, minute.offset]));
const truthOffset = offsets[row.oracle_truth_minute];
assert.notEqual(truthOffset, undefined);
const weights = rawMinuteWeights(state.candidates.map(candidate => ({ time: candidate.time,
score: candidate.raw_posterior_score ?? 0, cluster_times: candidate.cluster_times, eliminated: candidate.raw_eliminated })));
trace.metrics = Object.fromEntries((["D9", "D10"] as const).map(chart => [chart,
summarizeSegmentWeights(weights, segmentsForMinutes(state.segment_minutes!, [chart]), offsets, truthOffset)]));
}
trace.stage = "completed";
trace.ledger_after = ledger();
} catch (error) {
trace.stop = "blocked";
trace.error = study ? (error instanceof Error ? error.name : "unknown_error")
: (error instanceof Error ? error.message : "unknown_error");
trace.ledger_after = ledger();
}
records.push(trace);
}
}
} catch (error) {
fatal = error instanceof Error ? error.message : "runner_failure";
} finally {
if (previousOrder === undefined) delete process.env.RECTIFICATION_SEGMENT_ORDER;
else process.env.RECTIFICATION_SEGMENT_ORDER = previousOrder;
await closeLocalPostgresDataPool(url);
await closeLocalPostgresDataPool(authenticatedUrl);
fixture.stop();
}
const aggregate = Object.fromEntries([...new Set(input.rows.map(row => row.radius))].flatMap(radius =>
["off-default", "on-experiment"].map(ordering => {
const rows = records.filter(row => row.radius === radius && row.ordering === ordering);
const measured = rows.filter(row => row.metrics !== null);
const metrics = Object.fromEntries((["D9", "D10"] as const).map(chart => {
const values = measured.map(row => (row.metrics as Record<string, { truth_retained: boolean; top_is_truth: boolean }>)[chart]!);
return [chart, { measured: values.length, truth_kept: values.filter(value => value.truth_retained).length,
top_hit: values.filter(value => value.top_is_truth).length }];
}));
return [`${radius}:${ordering}`, { attempted: rows.length, measured: measured.length,
blocked: rows.filter(row => row.stop === "blocked").length,
asked: rows.reduce((total, row) => total + (row.asked as unknown[]).length, 0),
completed_six: rows.filter(row => (row.asked as unknown[]).length === 6).length, metrics,
redline_status: measured.length === 77 ? "requires-independent-cell-comparison" : "not-evaluable" }];
})));
const studySummary = study ? Object.fromEntries(modes.flatMap((mode) => (
[...new Set(input.rows.map((row) => row.radius))].map((radius) => {
const rows = records.filter((row) => row.radius === radius && row.mode === mode && row.stop !== "blocked"
&& typeof row.truth_in_envelope === "boolean");
const envelope = rows.filter((row) => row.truth_in_envelope === true).length;
const display = rows.filter((row) => row.truth_in_display === true).length;
const numbers = (key: string) => rows.map((row) => row[key]).filter((value): value is number => typeof value === "number");
return [`${radius}:${mode}`, {
attempted: rows.length,
blocked: records.filter((row) => row.radius === radius && row.mode === mode && row.stop === "blocked").length,
envelope_contains: envelope,
display_contains: display,
envelope_rate: rows.length ? envelope / rows.length : null,
display_rate: rows.length ? display / rows.length : null,
remaining_minutes_median: median(numbers("remaining_minutes")),
display_width_median: median(numbers("display_width_minutes")),
question_count_median: median(numbers("question_count")),
flip_applied: rows.filter((row) => row.flip_applied === true).length,
}];
})
))) : null;
const blocked = fatal !== null || records.some(row => row.stop === "blocked" || (!study && row.metrics === null));
process.stdout.write(JSON.stringify({
scope: study
? "narrower-delivery replay; containment booleans only; no birth clock, date, place or name"
: "explicit-seeded persisted replay; off-default vs on-experiment; no opening success or provider claim",
fatal: study && fatal ? "runner_failure" : fatal,
records, aggregate, study_summary: studySummary, blocked,
max_rounds: maxRounds,
display_fixture: displayFixture,
code_sha: study?.code_sha ?? null,
distinct_cases: new Set(input.rows.map(row => study ? caseLabel(row.case_key) : row.case_key)).size,
requested_traces: input.rows.length * passes.length,
completed_six: records.filter(row => (row.asked as unknown[]).length === 6).length }) + "\n");
process.exitCode = blocked ? 1 : 0;