import assert from "node:assert/strict";
import { spawnSync } from "node:child_process";
import { fileURLToPath } from "node:url";
import test from "node:test";

import { closeLocalPostgresDataPool, createLocalPostgresDataClient } from "../src/lib/db/local-postgres-client-core.ts";
import { startPostgresFixture } from "./helpers/postgres-fixture.ts";

const runnerPath = fileURLToPath(
  new URL("../scripts/db-migrate.mjs", import.meta.url),
);

function dockerAvailable(): boolean {
  const probe = spawnSync("docker", ["version", "--format", "{{.Server.Version}}"], {
    encoding: "utf8",
    stdio: "ignore",
  });
  return probe.status === 0;
}

const skipWithoutDocker = dockerAvailable() ? false : "docker unavailable on this host";

const snapshot = {
  birth_date: "1997-08-08",
  latitude: 36.420487,
  longitude: 114.209936,
  timezone_offset: 8,
  birth_time_source: "family_exact",
  reported_birth_time: "05:00",
  active_birth_time: null,
  birth_time_period: null,
  uncertainty_before_minutes: 10,
  uncertainty_after_minutes: 10,
};
const fingerprint = "a".repeat(64);
const range = { start_time: "04:50", end_time: "05:10" };
const skillName = "jyotish-birth-time-rectification";
const skillVersion = "9.0.0";

function rpcError(error: unknown): string {
  if (!error || typeof error !== "object") return "";
  const value = error as { message?: unknown };
  return typeof value.message === "string" ? value.message : "";
}

test("v9 migration applies on a fresh database and re-applies idempotently", { skip: skipWithoutDocker }, async () => {
  const fixture = startPostgresFixture();
  const schemaUrl = fixture.connectionUrl("schema_owner", "schema-owner-test-password");
  const migrate = () =>
    spawnSync(process.execPath, [runnerPath], {
      encoding: "utf8",
      env: { ...process.env, SCHEMA_DATABASE_URL: schemaUrl },
    });
  try {
    const firstRun = migrate();
    assert.equal(firstRun.status, 0, firstRun.stderr);
    assert.match(firstRun.stdout, /applied 20260812010000_agentic_rectification_v9_runtime\.sql/);
    const secondRun = migrate();
    assert.equal(secondRun.status, 0, secondRun.stderr);
    assert.match(
      secondRun.stdout,
      /already applied 20260812010000_agentic_rectification_v9_runtime\.sql/,
    );
  } finally {
    fixture.stop();
  }
});

test("v9 open is atomic, idempotent and allows separate homepage cases", { skip: skipWithoutDocker }, async () => {
  const fixture = startPostgresFixture();
  const schemaUrl = fixture.connectionUrl("schema_owner", "schema-owner-test-password");
  try {
    const migration = spawnSync(process.execPath, [runnerPath], {
      encoding: "utf8",
      env: { ...process.env, SCHEMA_DATABASE_URL: schemaUrl },
    });
    assert.equal(migration.status, 0, migration.stderr);

    fixture.psqlAs(
      "identity_runtime",
      "identity-runtime-test-password",
      `
        insert into identity.users (name, email, email_verified, email_verified_at)
        values ('V9 User', 'v9-user@example.com', true, now())
      `,
    );
    const userId = fixture.psql(
      "select id from identity.users where email = 'v9-user@example.com'",
    );
    fixture.psql(`
      update public.profiles
      set birth_date = '1997-08-08',
          reported_birth_time = '05:00',
          birth_time_source = 'family_exact',
          uncertainty_before_minutes = 10,
          uncertainty_after_minutes = 10,
          latitude = 36.420487,
          longitude = 114.209936,
          timezone_offset = 8,
          birth_time_status = 'reported'
      where id = '${userId}';
    `);

    fixture.psql(`
      with provider as (
        insert into public.model_providers (code, name, provider_type, encrypted_api_key, enabled)
        values ('v9-test', 'V9 Test', 'openai', 'test-ciphertext', true)
        returning id
      ), config as (
        insert into public.model_configs (model_id)
        values ('v9-default-model')
        returning id
      )
      insert into public.model_config_versions (
        config_id, version, provider_id, label, provider_model, enabled, is_default, status, published_at
      )
      select config.id, 1, provider.id, 'V9 Default', 'gpt-test', true, true, 'published', now()
      from config cross join provider;
    `);

    const service = createLocalPostgresDataClient(
      fixture.connectionUrl("service_runtime", "service-runtime-test-password"),
      null,
      "service_role",
    );

    const first = await service.rpc("open_agentic_rectification_case", {
      p_user_id: userId,
      p_request_id: "11111111-1111-4111-8111-111111111111",
      p_intent: "homepage",
      p_session_id: null,
      p_skill_name: skillName,
      p_skill_version: skillVersion,
      p_baseline_profile_fingerprint: fingerprint,
      p_baseline_birth_snapshot: snapshot,
      p_candidate_range: range,
    });
    assert.equal(first.error, null, rpcError(first.error));
    const created = first.data as Record<string, unknown>;
    assert.equal(created.disposition, "created");
    assert.equal(created.should_start_opening, true);
    assert.equal(created.status, "draft");
    const caseId = String(created.case_id);
    const sessionId = String(created.session_id);

    assert.equal(
      fixture.psql(`select count(*) from public.agentic_rectification_cases where user_id = '${userId}'`),
      "1",
    );
    assert.equal(
      fixture.psql(`select count(*) from public.chat_sessions where user_id = '${userId}'`),
      "1",
    );
    assert.equal(
      fixture.psql(`select model_id || ':' || model_config_version from public.chat_sessions where id = '${sessionId}'`),
      "v9-default-model:1",
    );
    assert.equal(
      fixture.psql(
        `select count(*) from public.agentic_rectification_open_ledger where user_id = '${userId}'`,
      ),
      "1",
    );
    assert.equal(
      fixture.psql(
        `select agentic_rectification_case_id from public.chat_sessions where id = '${sessionId}'`,
      ),
      caseId,
    );

    // Replaying the same requestId must not create anything new.
    const replay = await service.rpc("open_agentic_rectification_case", {
      p_user_id: userId,
      p_request_id: "11111111-1111-4111-8111-111111111111",
      p_intent: "homepage",
      p_session_id: null,
      p_skill_name: skillName,
      p_skill_version: skillVersion,
      p_baseline_profile_fingerprint: fingerprint,
      p_baseline_birth_snapshot: snapshot,
      p_candidate_range: range,
    });
    assert.equal(replay.error, null, rpcError(replay.error));
    assert.equal((replay.data as Record<string, unknown>).case_id, caseId);
    assert.equal((replay.data as Record<string, unknown>).session_id, sessionId);
    assert.equal(
      fixture.psql(`select count(*) from public.chat_sessions where user_id = '${userId}'`),
      "1",
    );

    // A fresh homepage click creates a separate case and session while the
    // previous unfinished case remains available through exact-session history.
    const secondOpen = await service.rpc("open_agentic_rectification_case", {
      p_user_id: userId,
      p_request_id: "22222222-2222-4222-8222-222222222222",
      p_intent: "homepage",
      p_session_id: null,
      p_skill_name: skillName,
      p_skill_version: skillVersion,
      p_baseline_profile_fingerprint: fingerprint,
      p_baseline_birth_snapshot: snapshot,
      p_candidate_range: range,
    });
    assert.equal(secondOpen.error, null, rpcError(secondOpen.error));
    assert.equal((secondOpen.data as Record<string, unknown>).disposition, "created");
    assert.equal((secondOpen.data as Record<string, unknown>).should_start_opening, true);
    const secondCaseId = String((secondOpen.data as Record<string, unknown>).case_id);
    const secondSessionId = String((secondOpen.data as Record<string, unknown>).session_id);
    assert.notEqual(secondCaseId, caseId);
    assert.notEqual(secondSessionId, sessionId);
    assert.equal(
      fixture.psql(`select count(*) from public.agentic_rectification_cases where user_id = '${userId}' and status = any (public.agentic_rectification_resumable_statuses())`),
      "2",
    );
    assert.equal(
      fixture.psql(`select count(*) from public.chat_sessions where user_id = '${userId}'`),
      "2",
    );

    // The explicit new intent has the same create semantics.
    const thirdOpen = await service.rpc("open_agentic_rectification_case", {
      p_user_id: userId,
      p_request_id: "33333333-3333-4333-8333-333333333333",
      p_intent: "new",
      p_session_id: null,
      p_skill_name: skillName,
      p_skill_version: skillVersion,
      p_baseline_profile_fingerprint: fingerprint,
      p_baseline_birth_snapshot: snapshot,
      p_candidate_range: range,
    });
    assert.equal(thirdOpen.error, null, rpcError(thirdOpen.error));
    assert.equal((thirdOpen.data as Record<string, unknown>).disposition, "created");
    assert.equal(
      fixture.psql(`select count(*) from public.agentic_rectification_cases where user_id = '${userId}' and status = any (public.agentic_rectification_resumable_statuses())`),
      "3",
    );

    // Session intent opens the exact session.
    const sessionOpen = await service.rpc("open_agentic_rectification_case", {
      p_user_id: userId,
      p_request_id: "44444444-4444-4444-8444-444444444444",
      p_intent: "session",
      p_session_id: sessionId,
      p_skill_name: skillName,
      p_skill_version: skillVersion,
      p_baseline_profile_fingerprint: "session-view",
      p_baseline_birth_snapshot: {},
      p_candidate_range: { start_time: "00:00", end_time: "23:59" },
    });
    assert.equal(sessionOpen.error, null, rpcError(sessionOpen.error));
    assert.equal((sessionOpen.data as Record<string, unknown>).case_id, caseId);
    assert.equal((sessionOpen.data as Record<string, unknown>).session_id, sessionId);
  } finally {
    // The service client registered a per-URL pool in the global cache; close
    // only this test's own service URL before destroying PostgreSQL, otherwise
    // the async pool teardown surfaces 57P01 after the fixture is gone. The
    // fixture must still stop even if pool close fails.
    try {
      await closeLocalPostgresDataPool(
        fixture.connectionUrl("service_runtime", "service-runtime-test-password"),
      );
    } finally {
      fixture.stop();
    }
  }
});

test("v9 enforces profile gating, ownership and terminal read-only", { skip: skipWithoutDocker }, async () => {
  const fixture = startPostgresFixture();
  const schemaUrl = fixture.connectionUrl("schema_owner", "schema-owner-test-password");
  try {
    const migration = spawnSync(process.execPath, [runnerPath], {
      encoding: "utf8",
      env: { ...process.env, SCHEMA_DATABASE_URL: schemaUrl },
    });
    assert.equal(migration.status, 0, migration.stderr);

    fixture.psqlAs(
      "identity_runtime",
      "identity-runtime-test-password",
      `
        insert into identity.users (name, email, email_verified, email_verified_at) values
          ('Incomplete', 'incomplete@example.com', true, now()),
          ('Owner', 'owner@example.com', true, now());
      `,
    );
    const incompleteId = fixture.psql(
      "select id from identity.users where email = 'incomplete@example.com'",
    );
    const ownerId = fixture.psql(
      "select id from identity.users where email = 'owner@example.com'",
    );
    fixture.psql(`
      update public.profiles set birth_date = '1997-08-08' where id = '${incompleteId}';
      update public.profiles
      set birth_date = '1997-08-08',
          reported_birth_time = '05:00',
          birth_time_source = 'family_exact',
          latitude = 36.420487, longitude = 114.209936, timezone_offset = 8,
          birth_time_status = 'reported'
      where id = '${ownerId}';
    `);

    const service = createLocalPostgresDataClient(
      fixture.connectionUrl("service_runtime", "service-runtime-test-password"),
      null,
      "service_role",
    );

    // Incomplete profile: no case is created.
    const incomplete = await service.rpc("open_agentic_rectification_case", {
      p_user_id: incompleteId,
      p_request_id: "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa",
      p_intent: "homepage",
      p_session_id: null,
      p_skill_name: skillName,
      p_skill_version: skillVersion,
      p_baseline_profile_fingerprint: fingerprint,
      p_baseline_birth_snapshot: { birth_date: "1997-08-08" },
      p_candidate_range: range,
    });
    assert.match(rpcError(incomplete.error), /agentic_rectification_profile_incomplete/);
    assert.equal(
      fixture.psql(`select count(*) from public.agentic_rectification_cases where user_id = '${incompleteId}'`),
      "0",
    );

    // Owner opens a case, closes it, then reopens it read-only.
    const opened = await service.rpc("open_agentic_rectification_case", {
      p_user_id: ownerId,
      p_request_id: "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb",
      p_intent: "homepage",
      p_session_id: null,
      p_skill_name: skillName,
      p_skill_version: skillVersion,
      p_baseline_profile_fingerprint: fingerprint,
      p_baseline_birth_snapshot: snapshot,
      p_candidate_range: range,
    });
    assert.equal(opened.error, null, rpcError(opened.error));
    const ownerCaseId = String((opened.data as Record<string, unknown>).case_id);
    const ownerSessionId = String((opened.data as Record<string, unknown>).session_id);

    const closed = await service.rpc("close_agentic_rectification_case", {
      p_user_id: ownerId,
      p_case_id: ownerCaseId,
      p_reason: "completed_by_user",
    });
    assert.equal(closed.error, null, rpcError(closed.error));
    assert.equal((closed.data as Record<string, unknown>).status, "closed");
    assert.equal(
      fixture.psql(`select status from public.agentic_rectification_cases where id = '${ownerCaseId}'`),
      "closed",
    );
    assert.equal(
      fixture.psql(`select completed_at is not null from public.agentic_rectification_cases where id = '${ownerCaseId}'`),
      "t",
    );

    const readonly = await service.rpc("open_agentic_rectification_case", {
      p_user_id: ownerId,
      p_request_id: "cccccccc-cccc-4ccc-8ccc-cccccccccccc",
      p_intent: "session",
      p_session_id: ownerSessionId,
      p_skill_name: skillName,
      p_skill_version: skillVersion,
      p_baseline_profile_fingerprint: "session-view",
      p_baseline_birth_snapshot: {},
      p_candidate_range: { start_time: "00:00", end_time: "23:59" },
    });
    assert.equal(readonly.error, null, rpcError(readonly.error));
    assert.equal((readonly.data as Record<string, unknown>).disposition, "readonly");
    assert.equal((readonly.data as Record<string, unknown>).should_start_opening, false);

    // A different user cannot open the owner's session.
    const foreign = await service.rpc("open_agentic_rectification_case", {
      p_user_id: incompleteId,
      p_request_id: "dddddddd-dddd-4ddd-8ddd-dddddddddddd",
      p_intent: "session",
      p_session_id: ownerSessionId,
      p_skill_name: skillName,
      p_skill_version: skillVersion,
      p_baseline_profile_fingerprint: "session-view",
      p_baseline_birth_snapshot: {},
      p_candidate_range: { start_time: "00:00", end_time: "23:59" },
    });
    assert.match(rpcError(foreign.error), /agentic_rectification_session_not_found/);

    // Terminal cases reject evidence proposals and turn appends.
    const turn = await service.rpc("append_agentic_rectification_turn", {
      p_user_id: ownerId,
      p_case_id: ownerCaseId,
      p_user_message: "我2016年9月离开家去北京工作",
      p_assistant_message: "已记录这段经历。",
      p_model_name: "test-model",
      p_model_version: "1",
      p_status: "completed",
      p_request_id: "eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee",
    });
    assert.match(rpcError(turn.error), /agentic_rectification_case_terminal/);

    const proposal = await service.rpc("propose_agentic_rectification_evidence", {
      p_user_id: ownerId,
      p_case_id: ownerCaseId,
      p_source_turn_id: "eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee",
      p_user_quote: "离开家去北京",
      p_subject: "self",
      p_event_kind: "relocation",
      p_domain: "relocation",
      p_occurred_from: "2016-09-01",
      p_occurred_to: null,
      p_date_precision: "month",
      p_summary: "2016年9月离开家去北京",
    });
    assert.match(rpcError(proposal.error), /agentic_rectification_case_terminal/);
  } finally {
    try {
      await closeLocalPostgresDataPool(
        fixture.connectionUrl("service_runtime", "service-runtime-test-password"),
      );
    } finally {
      fixture.stop();
    }
  }
});

test("v9 evidence lifecycle: quote grounding, idempotency, confirm and revision lineage", { skip: skipWithoutDocker }, async () => {
  const fixture = startPostgresFixture();
  const schemaUrl = fixture.connectionUrl("schema_owner", "schema-owner-test-password");
  try {
    const migration = spawnSync(process.execPath, [runnerPath], {
      encoding: "utf8",
      env: { ...process.env, SCHEMA_DATABASE_URL: schemaUrl },
    });
    assert.equal(migration.status, 0, migration.stderr);

    fixture.psqlAs(
      "identity_runtime",
      "identity-runtime-test-password",
      `
        insert into identity.users (name, email, email_verified, email_verified_at)
        values ('Evidence', 'evidence@example.com', true, now())
      `,
    );
    const userId = fixture.psql(
      "select id from identity.users where email = 'evidence@example.com'",
    );
    fixture.psql(`
      update public.profiles
      set birth_date = '1997-08-08',
          reported_birth_time = '05:00',
          birth_time_source = 'family_exact',
          latitude = 36.420487, longitude = 114.209936, timezone_offset = 8,
          birth_time_status = 'reported'
      where id = '${userId}';
    `);

    const service = createLocalPostgresDataClient(
      fixture.connectionUrl("service_runtime", "service-runtime-test-password"),
      null,
      "service_role",
    );

    const opened = await service.rpc("open_agentic_rectification_case", {
      p_user_id: userId,
      p_request_id: "ffffffff-ffff-4fff-8fff-ffffffffffff",
      p_intent: "homepage",
      p_session_id: null,
      p_skill_name: skillName,
      p_skill_version: skillVersion,
      p_baseline_profile_fingerprint: fingerprint,
      p_baseline_birth_snapshot: snapshot,
      p_candidate_range: range,
    });
    assert.equal(opened.error, null, rpcError(opened.error));
    const caseId = String((opened.data as Record<string, unknown>).case_id);

    const turn = await service.rpc("append_agentic_rectification_turn", {
      p_user_id: userId,
      p_case_id: caseId,
      p_user_message: "2016年9月我离开家去北京开始工作",
      p_assistant_message: "好的，已记录。",
      p_model_name: "test-model",
      p_model_version: "1",
      p_status: "completed",
      p_request_id: "12121212-1212-4212-8212-121212121212",
    });
    assert.equal(turn.error, null, rpcError(turn.error));
    const turnId = String((turn.data as Record<string, unknown>).turn_id);

    const proposed = await service.rpc("propose_agentic_rectification_evidence", {
      p_user_id: userId,
      p_case_id: caseId,
      p_source_turn_id: turnId,
      p_user_quote: "离开家去北京",
      p_subject: "self",
      p_event_kind: "relocation",
      p_domain: "relocation",
      p_occurred_from: "2016-09-01",
      p_occurred_to: null,
      p_date_precision: "month",
      p_summary: "2016年9月离开家去北京",
    });
    assert.equal(proposed.error, null, rpcError(proposed.error));
    const evidenceId = String((proposed.data as Record<string, unknown>).evidence_id);
    assert.equal((proposed.data as Record<string, unknown>).idempotent, false);
    assert.equal(
      fixture.psql(`select status from public.agentic_rectification_evidence where id = '${evidenceId}'`),
      "draft",
    );

    // Replay must not create a second evidence row.
    const replay = await service.rpc("propose_agentic_rectification_evidence", {
      p_user_id: userId,
      p_case_id: caseId,
      p_source_turn_id: turnId,
      p_user_quote: "离开家去北京",
      p_subject: "self",
      p_event_kind: "relocation",
      p_domain: "relocation",
      p_occurred_from: "2016-09-01",
      p_occurred_to: null,
      p_date_precision: "month",
      p_summary: "2016年9月离开家去北京",
    });
    assert.equal(replay.error, null, rpcError(replay.error));
    assert.equal((replay.data as Record<string, unknown>).evidence_id, evidenceId);
    assert.equal((replay.data as Record<string, unknown>).idempotent, true);
    assert.equal(
      fixture.psql(`select count(*) from public.agentic_rectification_evidence where case_id = '${caseId}'`),
      "1",
    );

    // Quotes not present in the source turn are rejected.
    const ungrounded = await service.rpc("propose_agentic_rectification_evidence", {
      p_user_id: userId,
      p_case_id: caseId,
      p_source_turn_id: turnId,
      p_user_quote: "去了上海",
      p_subject: "self",
      p_event_kind: "relocation",
      p_domain: "relocation",
      p_occurred_from: "2020-01-01",
      p_occurred_to: null,
      p_date_precision: "year",
      p_summary: "2020年去了上海",
    });
    assert.match(rpcError(ungrounded.error), /agentic_rectification_quote_not_grounded/);

    // Confirm transitions draft -> confirmed with a timestamp.
    const confirmed = await service.rpc("confirm_agentic_rectification_evidence", {
      p_user_id: userId,
      p_case_id: caseId,
      p_evidence_id: evidenceId,
    });
    assert.equal(confirmed.error, null, rpcError(confirmed.error));
    assert.equal((confirmed.data as Record<string, unknown>).status, "confirmed");
    assert.equal(
      fixture.psql(`select confirmed_at is not null from public.agentic_rectification_evidence where id = '${evidenceId}'`),
      "t",
    );

    // Revision creates a superseding row and never overwrites history.
    const revised = await service.rpc("revise_agentic_rectification_evidence", {
      p_user_id: userId,
      p_case_id: caseId,
      p_evidence_id: evidenceId,
      p_user_quote: "离开家去北京",
      p_occurred_from: "2016-10-01",
      p_occurred_to: null,
      p_date_precision: "month",
      p_summary: "2016年10月离开家去北京",
    });
    assert.equal(revised.error, null, rpcError(revised.error));
    const revisionId = String((revised.data as Record<string, unknown>).evidence_id);
    assert.equal((revised.data as Record<string, unknown>).supersedes_evidence_id, evidenceId);
    assert.equal(
      fixture.psql(`select status from public.agentic_rectification_evidence where id = '${evidenceId}'`),
      "superseded",
    );
    assert.equal(
      fixture.psql(`select status from public.agentic_rectification_evidence where id = '${revisionId}'`),
      "pending_confirmation",
    );
    assert.equal(
      fixture.psql(
        `select count(*) from public.agentic_rectification_evidence where case_id = '${caseId}'`,
      ),
      "2",
    );
  } finally {
    try {
      await closeLocalPostgresDataPool(
        fixture.connectionUrl("service_runtime", "service-runtime-test-password"),
      );
    } finally {
      fixture.stop();
    }
  }
});

test("v9 legacy backfill maps statuses, keeps one resumable per user and is idempotent", { skip: skipWithoutDocker }, async () => {
  const fixture = startPostgresFixture();
  const schemaUrl = fixture.connectionUrl("schema_owner", "schema-owner-test-password");
  try {
    const migration = spawnSync(process.execPath, [runnerPath], {
      encoding: "utf8",
      env: { ...process.env, SCHEMA_DATABASE_URL: schemaUrl },
    });
    assert.equal(migration.status, 0, migration.stderr);

    fixture.psqlAs(
      "identity_runtime",
      "identity-runtime-test-password",
      `
        insert into identity.users (name, email, email_verified, email_verified_at) values
          ('Legacy', 'legacy@example.com', true, now()),
          ('LegacyConfirmed', 'legacy-confirmed@example.com', true, now()),
          ('LegacyCollect', 'legacy-collect@example.com', true, now()),
          ('LegacyDraft', 'legacy-draft@example.com', true, now()),
          ('LegacyDup', 'legacy-dup@example.com', true, now());
      `,
    );
    const legacyId = fixture.psql("select id from identity.users where email = 'legacy@example.com'");
    const confirmedId = fixture.psql("select id from identity.users where email = 'legacy-confirmed@example.com'");
    const collectId = fixture.psql("select id from identity.users where email = 'legacy-collect@example.com'");
    const draftId = fixture.psql("select id from identity.users where email = 'legacy-draft@example.com'");
    const dupId = fixture.psql("select id from identity.users where email = 'legacy-dup@example.com'");
    fixture.psql(`
      update public.profiles
      set birth_date = '1997-08-08', reported_birth_time = '05:00',
          birth_time_source = 'family_exact',
          latitude = 36.420487, longitude = 114.209936, timezone_offset = 8,
          birth_time_status = 'reported'
      where id in ('${legacyId}', '${confirmedId}', '${collectId}', '${draftId}', '${dupId}');

      -- Legacy session with activity + a user-accepted result (newest -> active).
      insert into public.chat_sessions (id, user_id, title, theme, session_type, messages, updated_at)
      values (
        'aaaa1111-1111-4111-8111-111111111111', '${legacyId}', '旧校正A', 'general',
        'birth_time_rectification',
        '[{"role":"user","text":"2016年9月我离开家去北京工作"},{"role":"assistant","text":"已记录"}]',
        now() - interval '1 day'
      );
      -- Legacy session with activity but no selection (older -> superseded).
      insert into public.chat_sessions (id, user_id, title, theme, session_type, messages, updated_at)
      values (
        'bbbb1111-1111-4111-8111-111111111111', '${legacyId}', '旧校正B', 'general',
        'birth_time_rectification',
        '[{"role":"user","text":"2020年我开始担任管理职责"}]',
        now() - interval '3 days'
      );
      -- Repeated empty legacy session (old -> abandoned).
      insert into public.chat_sessions (id, user_id, title, theme, session_type, messages, updated_at)
      values (
        'cccc1111-1111-4111-8111-111111111111', '${legacyId}', '空校正', 'general',
        'birth_time_rectification', '[]', now() - interval '10 days'
      );
      -- Engine-confirmed legacy session for the second user.
      insert into public.chat_sessions (id, user_id, title, theme, session_type, messages, updated_at)
      values (
        'dddd1111-1111-4111-8111-111111111111', '${confirmedId}', '已确认校正', 'general',
        'birth_time_rectification',
        '[{"role":"user","text":"2015年我进入大学"}]',
        now() - interval '2 days'
      );
      -- Messages but no results -> collecting_evidence (active).
      insert into public.chat_sessions (id, user_id, title, theme, session_type, messages, updated_at)
      values (
        'eeee1111-1111-4111-8111-111111111111', '${collectId}', '收集校正', 'general',
        'birth_time_rectification',
        '[{"role":"user","text":"2018年我换了城市"}]',
        now() - interval '1 day'
      );
      -- Single empty session -> draft.
      insert into public.chat_sessions (id, user_id, title, theme, session_type, messages, updated_at)
      values (
        'ffff1111-1111-4111-8111-111111111111', '${draftId}', '新空校正', 'general',
        'birth_time_rectification', '[]', now() - interval '1 day'
      );
      -- Duplicate empty sessions: newest -> draft, older -> abandoned.
      insert into public.chat_sessions (id, user_id, title, theme, session_type, messages, updated_at)
      values
        ('10111111-1111-4111-8111-111111111111', '${dupId}', '重复空1', 'general',
         'birth_time_rectification', '[]', now() - interval '1 day'),
        ('20222222-2222-4222-8222-222222222222', '${dupId}', '重复空2', 'general',
         'birth_time_rectification', '[]', now() - interval '2 days');

      insert into public.agentic_rectification_results (
        user_id, session_id, engine_result_id, canonical_input_hash, algorithm_version,
        candidate_range, candidates, overall_confidence, selection_allowed, confirmation_allowed,
        representative_time, selected_time, selection_kind, selected_at,
        baseline_birth_date, baseline_reported_birth_time, baseline_birth_time_source,
        baseline_uncertainty_before_minutes, baseline_uncertainty_after_minutes,
        baseline_latitude, baseline_longitude, baseline_timezone_offset
      ) values (
        '${legacyId}', 'aaaa1111-1111-4111-8111-111111111111',
        'legacy-engine-1', 'legacy-hash-1', 'legacy-v1',
        '{"start_time":"04:00","end_time":"06:00"}',
        '[{"rank":1,"time":"05:00","relative_support":70,"tied_minute_count":1}]',
        'medium', true, false, '05:00', '05:00', 'user_accepted', now() - interval '1 day',
        '1997-08-08', '05:00', 'family_exact', 10, 10, 36.420487, 114.209936, 8
      );
      insert into public.agentic_rectification_results (
        user_id, session_id, engine_result_id, canonical_input_hash, algorithm_version,
        candidate_range, candidates, overall_confidence, selection_allowed, confirmation_allowed,
        representative_time, selected_time, selection_kind, selected_at,
        baseline_birth_date, baseline_reported_birth_time, baseline_birth_time_source,
        baseline_uncertainty_before_minutes, baseline_uncertainty_after_minutes,
        baseline_latitude, baseline_longitude, baseline_timezone_offset
      ) values (
        '${confirmedId}', 'dddd1111-1111-4111-8111-111111111111',
        'legacy-engine-2', 'legacy-hash-2', 'legacy-v1',
        '{"start_time":"04:00","end_time":"06:00"}',
        '[{"rank":1,"time":"05:00","relative_support":90,"tied_minute_count":1}]',
        'high', true, true, '05:00', '05:00', 'engine_confirmed', now() - interval '2 days',
        '1997-08-08', '05:00', 'family_exact', 10, 10, 36.420487, 114.209936, 8
      );
    `);

    const service = createLocalPostgresDataClient(
      fixture.connectionUrl("service_runtime", "service-runtime-test-password"),
      null,
      "service_role",
    );

    // Preflight count: 8 legacy sessions without a V9 case.
    assert.equal(
      fixture.psql(`
        select count(*) from public.chat_sessions
        where session_type = 'birth_time_rectification'
          and not exists (
            select 1 from public.agentic_rectification_cases c where c.session_id = public.chat_sessions.id
          )
      `),
      "8",
    );

    const backfill = await service.rpc("backfill_agentic_rectification_legacy_cases", {});
    assert.equal(backfill.error, null, rpcError(backfill.error));
    const stats = backfill.data as Record<string, unknown>;
    assert.equal(stats.cases_created, 8);
    assert.equal(stats.results_mapped, 2);
    assert.equal(stats.confirmed, 1);
    assert.equal(stats.candidate_accepted, 1);
    assert.equal(stats.collecting_evidence, 1);
    assert.equal(stats.draft, 2);
    assert.equal(stats.abandoned, 2);
    assert.equal(stats.superseded, 1);

    // Statuses landed correctly.
    assert.equal(
      fixture.psql(`select status from public.agentic_rectification_cases where session_id = 'dddd1111-1111-4111-8111-111111111111'`),
      "confirmed",
    );
    assert.equal(
      fixture.psql(`select status from public.agentic_rectification_cases where session_id = 'ffff1111-1111-4111-8111-111111111111'`),
      "draft",
    );
    assert.equal(
      fixture.psql(`select status from public.agentic_rectification_cases where session_id = 'eeee1111-1111-4111-8111-111111111111'`),
      "collecting_evidence",
    );
    assert.equal(
      fixture.psql(`select status from public.agentic_rectification_cases where session_id = 'aaaa1111-1111-4111-8111-111111111111'`),
      "candidate_accepted",
    );
    assert.equal(
      fixture.psql(`select status from public.agentic_rectification_cases where session_id = 'bbbb1111-1111-4111-8111-111111111111'`),
      "superseded",
    );
    assert.equal(
      fixture.psql(`select status from public.agentic_rectification_cases where session_id = '10111111-1111-4111-8111-111111111111'`),
      "draft",
    );
    assert.equal(
      fixture.psql(`select status from public.agentic_rectification_cases where session_id = '20222222-2222-4222-8222-222222222222'`),
      "abandoned",
    );

    // Exactly one resumable case per user.
    assert.equal(
      fixture.psql(`
        select count(*) from (
          select user_id from public.agentic_rectification_cases
          where status in ('draft','collecting_evidence','candidate_ready','candidate_accepted','needs_rebaseline','paused')
          group by user_id having count(*) > 1
        ) conflicts
      `),
      "0",
    );

    // Results are mapped to their cases.
    assert.equal(
      fixture.psql(`select count(*) from public.agentic_rectification_results where case_id is null`),
      "0",
    );

    // Verification report is clean.
    const verify = await service.rpc("verify_agentic_rectification_backfill", {});
    assert.equal(verify.error, null, rpcError(verify.error));
    const report = verify.data as Record<string, unknown>;
    assert.equal(report.orphan_cases, 0);
    assert.equal(report.resumable_conflicts, 0);
    assert.equal(report.legacy_sessions_without_case, 0);

    // Re-running the backfill creates nothing new (idempotent).
    const again = await service.rpc("backfill_agentic_rectification_legacy_cases", {});
    assert.equal(again.error, null, rpcError(again.error));
    assert.equal((again.data as Record<string, unknown>).cases_created, 0);
  } finally {
    try {
      await closeLocalPostgresDataPool(
        fixture.connectionUrl("service_runtime", "service-runtime-test-password"),
      );
    } finally {
      fixture.stop();
    }
  }
});

test("v9 agent api migration applies, seeds the runtime flag and guards consent", { skip: skipWithoutDocker }, async () => {
  const fixture = startPostgresFixture();
  const schemaUrl = fixture.connectionUrl("schema_owner", "schema-owner-test-password");
  const fixtureSkillSha256 = "d".repeat(64);
  const fixtureSkillSourceCommit = "e".repeat(40);
  try {
    const migration = spawnSync(process.execPath, [runnerPath], {
      encoding: "utf8",
      env: { ...process.env, SCHEMA_DATABASE_URL: schemaUrl },
    });
    assert.equal(migration.status, 0, migration.stderr);
    assert.match(migration.stdout, /applied 20260813010000_agentic_rectification_v9_agent_api\.sql/);
    assert.match(migration.stdout, /applied 20260813020000_rectification_birth_context_activity\.sql/);
    assert.match(migration.stdout, /applied 20260811030000_feature_flags_admin_runtime_read_policy\.sql/);

    // The runtime selector flag is published and enabled.
    assert.equal(
      fixture.psql(`select enabled || ':' || rollout_percentage || ':' || status
        from public.feature_flags where flag_key = 'rectification_runtime_version'`),
      "true:100:published",
    );
    assert.equal(
      fixture.psqlAs(
        "admin_runtime",
        "admin-runtime-test-password",
        `select enabled || ':' || rollout_percentage || ':' || status
          from public.feature_flags
          where flag_key = 'rectification_runtime_version'`,
      ),
      "true:100:published",
    );

    // Run phases table exists with RLS.
    assert.equal(
      fixture.psql(`select count(*) from pg_tables where schemaname='public' and tablename='agentic_rectification_run_phases'`),
      "1",
    );

    // Set up a profile + case + pending turn, then exercise the turn
    // finalize and run-phase receipt RPCs. The identity row is seeded through
    // the legitimate identity_runtime role (syncs to auth.users, satisfying
    // the profiles FK); the public.* fixture rows are seeded through the
    // postgres admin connection because identity_runtime has no grants on
    // these tables (production least privilege). Tests must never widen
    // runtime grants or change migrations.
    fixture.psqlAs("identity_runtime", "identity-runtime-test-password", `
      insert into identity.users (id, name, email, email_verified, email_verified_at)
      values ('66666666-6666-4666-8666-666666666666', 'V9 Agent API Fixture', 'v9-agent-api-fixture@example.com', true, now())
    `);
    fixture.psql(`
      update public.profiles
      set birth_date = '1997-08-08', reported_birth_time = '05:00', active_birth_time = null,
          birth_time_source = 'family_exact', uncertainty_before_minutes = 10,
          uncertainty_after_minutes = 10, birth_place_label = '河北省邯郸市武安市',
          latitude = 36.420487, longitude = 114.209936,
          timezone_id = 'Asia/Shanghai', timezone_offset = 8
      where id = '66666666-6666-4666-8666-666666666666';
    `);
    fixture.psql(`
      insert into public.chat_sessions (id, user_id, title, theme, session_type, messages)
      values ('22222222-2222-4222-8222-222222222222', '66666666-6666-4666-8666-666666666666', '生时校正', 'general', 'birth_time_rectification', '[]'::jsonb);
    `);
    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
      ) values (
        '11111111-1111-4111-8111-111111111111', '66666666-6666-4666-8666-666666666666',
        '22222222-2222-4222-8222-222222222222', 'draft', 'jyotish-birth-time-rectification', '9.0.0',
        '${fixtureSkillSha256}', '${fixtureSkillSourceCommit}',
        '${fingerprint}',
        '{"birth_date":"1997-08-08","birth_place_label":"河北省邯郸市武安市","latitude":36.420487,"longitude":114.209936,"timezone_id":"Asia/Shanghai","timezone_offset":8,"birth_time_source":"family_exact"}'::jsonb,
        '{"start_time":"04:50","end_time":"05:10"}'::jsonb
      );
    `);
    fixture.psql(`
      insert into public.agentic_rectification_turns (id, case_id, user_message, status, model_name)
      values ('33333333-3333-4333-8333-333333333333', '11111111-1111-4111-8111-111111111111',
        '2016年9月离开家去北京开始工作', 'pending', 'gpt-4o-mini');
    `);

    const service = createLocalPostgresDataClient(
      fixture.connectionUrl("service_runtime", "service-runtime-test-password"),
      null,
      "service_role",
    );
    const attempt = await service.rpc("create_agentic_rectification_run_attempt", {
      p_user_id: "66666666-6666-4666-8666-666666666666",
      p_case_id: "11111111-1111-4111-8111-111111111111",
      p_turn_id: "33333333-3333-4333-8333-333333333333",
      p_attempt_number: 1,
    });
    assert.equal(attempt.error, null, rpcError(attempt.error));
    const attemptId = String((attempt.data as Record<string, unknown>).attempt_id);

    const phase = await service.rpc("insert_agentic_rectification_run_phase", {
      p_user_id: "66666666-6666-4666-8666-666666666666",
      p_case_id: "11111111-1111-4111-8111-111111111111",
      p_turn_id: "33333333-3333-4333-8333-333333333333",
      p_phase: "skill.loaded",
      p_tool_name: null,
      p_sequence: 1,
      p_attempt_id: attemptId,
    });
    assert.equal(phase.error, null, rpcError(phase.error));

    const insertReceipt = (tool: string, status: string, methods: string[]) =>
      service.rpc("insert_agentic_rectification_tool_receipt", {
        p_user_id: "66666666-6666-4666-8666-666666666666",
        p_case_id: "11111111-1111-4111-8111-111111111111",
        p_turn_id: "33333333-3333-4333-8333-333333333333",
        p_tool_name: tool,
        p_public_phase: tool === "rectification-read-case" ? "case.loaded" : "candidates.comparing",
        p_status: status,
        p_input_fingerprint: null,
        p_result_fingerprint: null,
        p_engine_version: tool === "rectification-compare-candidates" ? "fixture-engine" : null,
        p_safe_error_code: status === "failed" ? "fixture_failure" : null,
        p_executed_methods: methods,
        p_attempt_id: attemptId,
      });
    for (const result of [
      await insertReceipt("rectification-read-case", "completed", []),
      await insertReceipt("rectification-compare-candidates", "completed", [
        "d1-rashi", "vimshottari-dasha", "d10-dashamsa",
      ]),
      await insertReceipt("rectification-read-diagnostics", "failed", ["d9-navamsa"]),
    ]) {
      assert.equal(result.error, null, rpcError(result.error));
    }
    const invalidMethod = await insertReceipt(
      "rectification-compare-candidates",
      "completed",
      ["private-technique"],
    );
    assert.match(rpcError(invalidMethod.error), /agentic_rectification_invalid_input/);

    const skillReceipt = await service.rpc("insert_agentic_rectification_skill_run_receipt", {
      p_user_id: "66666666-6666-4666-8666-666666666666",
      p_case_id: "11111111-1111-4111-8111-111111111111",
      p_turn_id: "33333333-3333-4333-8333-333333333333",
      p_request_id: attemptId,
      p_run_kind: "turn",
      p_skill_name: "jyotish-birth-time-rectification",
      p_skill_version: "9.0.0",
      p_skill_sha256: fixtureSkillSha256,
      p_source_commit: fixtureSkillSourceCommit,
    });
    assert.equal(skillReceipt.error, null, rpcError(skillReceipt.error));

    for (const [publicPhase, sequence] of [
      ["skill.bound", 2],
      ["case.loaded", 3],
      ["intent.classified", 4],
      ["billing.settled", 5],
      ["run.completed", 6],
    ] as const) {
      const lifecyclePhase = await service.rpc("insert_agentic_rectification_run_phase", {
        p_user_id: "66666666-6666-4666-8666-666666666666",
        p_case_id: "11111111-1111-4111-8111-111111111111",
        p_turn_id: "33333333-3333-4333-8333-333333333333",
        p_phase: publicPhase,
        p_tool_name: null,
        p_sequence: sequence,
        p_attempt_id: attemptId,
      });
      assert.equal(lifecyclePhase.error, null, rpcError(lifecyclePhase.error));
    }

    const finalizedAttempt = await service.rpc("finalize_agentic_rectification_run_attempt", {
      p_user_id: "66666666-6666-4666-8666-666666666666",
      p_case_id: "11111111-1111-4111-8111-111111111111",
      p_turn_id: "33333333-3333-4333-8333-333333333333",
      p_attempt_id: attemptId,
      p_status: "completed",
      p_error_code: null,
      p_usage: {},
    });
    assert.equal(finalizedAttempt.error, null, rpcError(finalizedAttempt.error));

    const finalizedTurn = await service.rpc("finalize_agentic_rectification_turn", {
      p_user_id: "66666666-6666-4666-8666-666666666666",
      p_case_id: "11111111-1111-4111-8111-111111111111",
      p_turn_id: "33333333-3333-4333-8333-333333333333",
      p_attempt_id: attemptId,
      p_status: "completed",
      p_assistant_message: "好的，我会先核对出生时间范围。",
      p_successful_attempt_id: attemptId,
    });
    assert.equal(finalizedTurn.error, null, rpcError(finalizedTurn.error));

    const receipt = await service.rpc("get_agentic_rectification_turn_receipt", {
      p_user_id: "66666666-6666-4666-8666-666666666666",
      p_case_id: "11111111-1111-4111-8111-111111111111",
      p_turn_id: "33333333-3333-4333-8333-333333333333",
    });
    assert.equal(receipt.error, null, rpcError(receipt.error));
    const receiptRow = receipt.data as Record<string, unknown>;
    assert.equal(receiptRow.status, "completed");
    assert.equal(receiptRow.skill_version, "9.0.0");
    assert.equal(receiptRow.engine_version, "fixture-engine");
    assert.deepEqual(receiptRow.tools, [
      "rectification-read-case",
      "rectification-compare-candidates",
    ]);
    assert.deepEqual(receiptRow.methods, [
      "d1-rashi",
      "d10-dashamsa",
      "vimshottari-dasha",
    ]);

    const consentCandidate = await service.rpc("persist_agentic_rectification_candidate_v2", {
      p_user_id: "66666666-6666-4666-8666-666666666666",
      p_case_id: "11111111-1111-4111-8111-111111111111",
      p_engine_result_id: "agent-api-consent-engine",
      p_evidence_ledger_fingerprint: "f".repeat(64),
      p_candidate_range_fingerprint: "a".repeat(64),
      p_skill_version: "9.0.0",
      p_algorithm_version: "agent-api-consent-v2",
      p_event_contract_version: "rectification-event-contract-v2",
      p_decision_policy_version: "rectification-candidate-policy-v2",
      p_candidate_range: { start_time: "04:50", end_time: "05:10" },
      p_candidates: [{ rank: 1, time: "05:02", relative_support: 100, tied_minute_count: 1 }],
      p_decision_receipt: {
        display_allowed: true,
        accept_allowed: true,
        confirm_allowed: true,
        representative_time: "05:02",
        overall_confidence: "high",
        margin_percent: 100,
      },
      p_execution_ledger: [{ phase: "decision.evaluate", status: "completed" }],
    });
    assert.equal(consentCandidate.error, null, rpcError(consentCandidate.error));
    const consentCandidateRow = consentCandidate.data as Record<string, unknown>;
    const consentCandidateId = String(
      (consentCandidateRow.candidates as Array<Record<string, unknown>>)[0]?.candidate_id ?? "",
    );

    // Consent must be grounded in the current source turn before confirmation.
    const consent = await service.rpc("confirm_agentic_rectification_candidate_for_case_v2", {
      p_user_id: "66666666-6666-4666-8666-666666666666",
      p_case_id: "11111111-1111-4111-8111-111111111111",
      p_result_id: String(consentCandidateRow.result_id),
      p_candidate_id: consentCandidateId,
      p_request_id: "99999999-aaaa-4aaa-8aaa-aaaaaaaaaaaa",
      p_consent_quote: "就用05:02",
      p_source_turn_id: "33333333-3333-4333-8333-333333333333",
    });
    assert.match(rpcError(consent.error), /agentic_rectification_consent_not_grounded/);
  } finally {
    try {
      await closeLocalPostgresDataPool(
        fixture.connectionUrl("service_runtime", "service-runtime-test-password"),
      );
    } finally {
      fixture.stop();
    }
  }
});

test("v9 allows changing an accepted minute within the same live candidate result", { skip: skipWithoutDocker }, async () => {
  const fixture = startPostgresFixture();
  const schemaUrl = fixture.connectionUrl("schema_owner", "schema-owner-test-password");
  try {
    const migration = spawnSync(process.execPath, [runnerPath], {
      encoding: "utf8",
      env: { ...process.env, SCHEMA_DATABASE_URL: schemaUrl },
    });
    assert.equal(migration.status, 0, migration.stderr);

    const userId = "77777777-7777-4777-8777-777777777777";
    const sessionId = "88888888-8888-4888-8888-888888888888";
    const caseId = "99999999-9999-4999-8999-999999999999";

    fixture.psqlAs("identity_runtime", "identity-runtime-test-password", `
      insert into identity.users (id, name, email, email_verified, email_verified_at)
      values ('${userId}', 'Candidate Reselection Fixture', 'candidate-reselection@example.com', true, now())
    `);
    fixture.psql(`
      update public.profiles
      set birth_date = '1997-08-08',
          reported_birth_time = '05:00',
          active_birth_time = null,
          birth_time = '05:00',
          birth_time_status = 'reported',
          birth_time_source = 'family_exact',
          birth_time_period = null,
          uncertainty_before_minutes = 10,
          uncertainty_after_minutes = 10,
          birth_place_label = '河北省邯郸市武安市',
          latitude = 36.420487,
          longitude = 114.209936,
          timezone_id = 'Asia/Shanghai',
          timezone_offset = 8
      where id = '${userId}';

      insert into public.chat_sessions (id, user_id, title, theme, session_type, messages)
      values ('${sessionId}', '${userId}', '候选改选', 'general', 'birth_time_rectification', '[]'::jsonb);

      insert into public.agentic_rectification_cases (
        id, user_id, session_id, status, skill_name, skill_version,
        baseline_profile_fingerprint, baseline_birth_snapshot, candidate_range
      ) values (
        '${caseId}', '${userId}', '${sessionId}', 'candidate_ready',
        '${skillName}', '${skillVersion}', '${fingerprint}',
        '${JSON.stringify(snapshot)}'::jsonb,
        '${JSON.stringify(range)}'::jsonb
      );
    `);

    const service = createLocalPostgresDataClient(
      fixture.connectionUrl("service_runtime", "service-runtime-test-password"),
      null,
      "service_role",
    );
    const persisted = await service.rpc("persist_agentic_rectification_candidate_v2", {
      p_user_id: userId,
      p_case_id: caseId,
      p_engine_result_id: "candidate-reselection-engine",
      p_evidence_ledger_fingerprint: "b".repeat(64),
      p_candidate_range_fingerprint: "c".repeat(64),
      p_skill_version: skillVersion,
      p_algorithm_version: "candidate-reselection-v2",
      p_event_contract_version: "rectification-event-contract-v2",
      p_decision_policy_version: "rectification-candidate-policy-v2",
      p_candidate_range: range,
      p_candidates: [
        { rank: 1, time: "05:07", relative_support: 34, tied_minute_count: 1 },
        { rank: 2, time: "05:08", relative_support: 33, tied_minute_count: 2 },
        { rank: 3, time: "05:09", relative_support: 33, tied_minute_count: 2 },
      ],
      p_decision_receipt: {
        display_allowed: true,
        accept_allowed: true,
        confirm_allowed: true,
        representative_time: "05:08",
        overall_confidence: "low",
        margin_percent: 1,
      },
      p_execution_ledger: [
        { phase: "candidate.score", status: "completed", engine: "fixture-engine" },
        { phase: "decision.evaluate", status: "completed", policy: "rectification-candidate-policy-v2" },
      ],
    });
    assert.equal(persisted.error, null, rpcError(persisted.error));
    const persistedRow = persisted.data as Record<string, unknown>;
    const resultId = String(persistedRow.result_id);
    const persistedCandidates = persistedRow.candidates as Array<Record<string, unknown>>;
    const candidateIdFor = (time: string) => String(
      persistedCandidates.find((candidate) => candidate.time === time)?.candidate_id ?? "",
    );
    const firstCandidateId = candidateIdFor("05:07");
    const switchedCandidateId = candidateIdFor("05:08");
    const thirdCandidateId = candidateIdFor("05:09");
    assert.match(firstCandidateId, /^[0-9a-f-]{36}$/);
    assert.match(switchedCandidateId, /^[0-9a-f-]{36}$/);
    assert.match(thirdCandidateId, /^[0-9a-f-]{36}$/);

    const firstAccept = await service.rpc("accept_agentic_rectification_candidate_for_case_v2", {
      p_user_id: userId,
      p_case_id: caseId,
      p_result_id: resultId,
      p_candidate_id: firstCandidateId,
      p_request_id: "11111111-aaaa-4aaa-8aaa-aaaaaaaaaaaa",
    });
    assert.equal(firstAccept.error, null, rpcError(firstAccept.error));

    const switchRequestId = "22222222-aaaa-4aaa-8aaa-aaaaaaaaaaaa";
    const switched = await service.rpc("accept_agentic_rectification_candidate_for_case_v2", {
      p_user_id: userId,
      p_case_id: caseId,
      p_result_id: resultId,
      p_candidate_id: switchedCandidateId,
      p_request_id: switchRequestId,
    });
    assert.equal(switched.error, null, rpcError(switched.error));
    assert.deepEqual(
      {
        saved_time: (switched.data as Record<string, unknown>).saved_time,
        status: (switched.data as Record<string, unknown>).status,
        case_status: (switched.data as Record<string, unknown>).case_status,
        idempotent: (switched.data as Record<string, unknown>).idempotent,
      },
      { saved_time: "05:08", status: "accepted", case_status: "candidate_accepted", idempotent: false },
    );
    assert.equal(
      fixture.psql(`select active_birth_time || ':' || birth_time || ':' || birth_time_status from public.profiles where id = '${userId}'`),
      "05:08:00:05:08:00:accepted",
    );
    assert.equal(
      fixture.psql(`select selected_time || ':' || selection_kind || ':' || selected_candidate_id from public.agentic_rectification_results where id = '${resultId}'`),
      `05:08:00:user_accepted:${switchedCandidateId}`,
    );
    assert.equal(
      fixture.psql(`select accepted_time || ':' || status from public.agentic_rectification_cases where id = '${caseId}'`),
      "05:08:00:candidate_accepted",
    );

    const replay = await service.rpc("accept_agentic_rectification_candidate_for_case_v2", {
      p_user_id: userId,
      p_case_id: caseId,
      p_result_id: resultId,
      p_candidate_id: switchedCandidateId,
      p_request_id: switchRequestId,
    });
    assert.equal(replay.error, null, rpcError(replay.error));
    assert.equal((replay.data as Record<string, unknown>).idempotent, true);

    const consentTurn = await service.rpc("append_agentic_rectification_turn", {
      p_user_id: userId,
      p_case_id: caseId,
      p_user_message: "我确认使用05:08作为出生时间",
      p_assistant_message: "已记录你的确认。",
      p_model_name: "test-model",
      p_model_version: "1",
      p_status: "completed",
      p_request_id: "33333333-aaaa-4aaa-8aaa-aaaaaaaaaaaa",
    });
    assert.equal(consentTurn.error, null, rpcError(consentTurn.error));
    const consentTurnId = String((consentTurn.data as Record<string, unknown>).turn_id);

    const confirmed = await service.rpc("confirm_agentic_rectification_candidate_for_case_v2", {
      p_user_id: userId,
      p_case_id: caseId,
      p_result_id: resultId,
      p_candidate_id: switchedCandidateId,
      p_request_id: "44444444-aaaa-4aaa-8aaa-aaaaaaaaaaaa",
      p_consent_quote: "确认使用05:08",
      p_source_turn_id: consentTurnId,
    });
    assert.equal(confirmed.error, null, rpcError(confirmed.error));
    assert.equal((confirmed.data as Record<string, unknown>).status, "confirmed");

    const confirmedSwitch = await service.rpc("accept_agentic_rectification_candidate_for_case_v2", {
      p_user_id: userId,
      p_case_id: caseId,
      p_result_id: resultId,
      p_candidate_id: thirdCandidateId,
      p_request_id: "55555555-aaaa-4aaa-8aaa-aaaaaaaaaaaa",
    });
    assert.match(rpcError(confirmedSwitch.error), /agentic_rectification_case_terminal/);
  } finally {
    try {
      await closeLocalPostgresDataPool(
        fixture.connectionUrl("service_runtime", "service-runtime-test-password"),
      );
    } finally {
      fixture.stop();
    }
  }
});