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 resumes instead of duplicating", { 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 second request with a fresh requestId resumes the same case.
    const resume = 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(resume.error, null, rpcError(resume.error));
    assert.equal((resume.data as Record<string, unknown>).disposition, "resumed");
    assert.equal((resume.data as Record<string, unknown>).should_start_opening, false);
    assert.equal((resume.data as Record<string, unknown>).case_id, caseId);
    assert.equal(
      fixture.psql(`select count(*) from public.chat_sessions where user_id = '${userId}'`),
      "1",
    );

    // intent new while an active case exists must fail safely.
    const conflict = 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.match(rpcError(conflict.error), /agentic_rectification_active_case_conflict/);
    assert.equal(
      fixture.psql(`select count(*) from public.chat_sessions where user_id = '${userId}'`),
      "1",
    );

    // 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",
    });
    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",
    });
    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");
  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 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, latitude = 36.420487,
          longitude = 114.209936, 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,
        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',
        '${fingerprint}',
        '{"birth_date":"1997-08-08","latitude":36.420487,"longitude":114.209936,"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 finalize = 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_status: "completed",
      p_assistant_message: "好的，我会先核对出生时间范围。",
    });
    assert.equal(finalize.error, null, rpcError(finalize.error));

    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,
    });
    assert.equal(phase.error, null, rpcError(phase.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");

    // Consent must be grounded in the source turn before candidate lookup.
    const consent = await service.rpc("confirm_agentic_rectification_birth_time", {
      p_user_id: "66666666-6666-4666-8666-666666666666",
      p_case_id: "11111111-1111-4111-8111-111111111111",
      p_result_id: "55555555-5555-4555-8555-555555555555",
      p_time: "05:02",
      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();
    }
  }
});