Jyotisha/frontend/tests/database-local-business.test.ts

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

import {
  closeLocalPostgresDataPools,
  createLocalPostgresDataClient,
} from "../src/lib/db/local-postgres-client-core.ts";
import { loadLatestAgenticRectificationResult } from "../src/lib/rectification-agentic/session.ts";
import { startPostgresFixture } from "./helpers/postgres-fixture.ts";

const runnerPath = fileURLToPath(
  new URL("../scripts/db-migrate.mjs", import.meta.url),
);
const acceptedExactFamilyMigration = readFileSync(
  new URL("../supabase/migrations/20260816010000_accept_exact_family_birth_times.sql", import.meta.url),
  "utf8",
);
const pinArchiveMigration = readFileSync(
  new URL("../supabase/migrations/20260901020000_chat_session_pin_archive.sql", import.meta.url),
  "utf8",
);

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

function candidateIds(value: unknown): string[] {
  if (!value || typeof value !== "object") return [];
  const candidates = (value as { candidates?: unknown }).candidates;
  if (!Array.isArray(candidates)) return [];
  return candidates.flatMap((candidate) => {
    if (!candidate || typeof candidate !== "object") return [];
    const candidateId = (candidate as { candidate_id?: unknown }).candidate_id;
    return typeof candidateId === "string" ? [candidateId] : [];
  });
}

test("local PostgreSQL applies the reviewed business schema and serves authenticated business calls", 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 20260715000000_account_credits\.sql/);
    assert.match(migration.stdout, /applied 20260721150000_align_conversational_finance_domain\.sql/);
    assert.match(migration.stdout, /applied 20260723010000_restore_conversational_message_history\.sql/);
    assert.match(migration.stdout, /applied 20260723020000_mark_captured_conversational_messages\.sql/);
    assert.match(migration.stdout, /applied 20260728010000_conversational_event_semantics\.sql/);
    assert.match(migration.stdout, /applied 20260728020000_rectification_agent_v5\.sql/);
    assert.match(migration.stdout, /applied 20260804010000_agentic_rectification_candidate_acceptance\.sql/);
    assert.match(migration.stdout, /applied 20260804020000_preserve_reported_birth_time_on_candidate_acceptance\.sql/);
    assert.match(migration.stdout, /applied 20260805010000_reconcile_admin_redemption_audit\.sql/);
    assert.match(migration.stdout, /applied 20260805020000_reconcile_payment_admin_schema\.sql/);
    assert.match(migration.stdout, /applied 20260805030000_reconcile_rectification_v4_conversational_turns\.sql/);
    assert.match(migration.stdout, /applied 20260806000000_personal_reports\.sql/);
    assert.match(migration.stdout, /applied 20260806010000_admin_rbac\.sql/);
    assert.match(migration.stdout, /applied 20260806020000_billing_products_subscriptions\.sql/);
    assert.match(migration.stdout, /applied 20260806030000_settle_order_usage_authorization\.sql/);
    assert.match(migration.stdout, /applied 20260806040000_model_configuration\.sql/);
    assert.match(migration.stdout, /applied 20260806050000_operations_feature_flags\.sql/);
    assert.match(migration.stdout, /applied 20260811010000_consultation_status_service_role_read\.sql/);
    assert.match(migration.stdout, /applied 20260812010000_agentic_rectification_v9_runtime\.sql/);
    assert.match(migration.stdout, /applied 20260813010000_agentic_rectification_v9_agent_api\.sql/);
    assert.match(migration.stdout, /applied 20260813060000_rectification_turn_regeneration\.sql/);
    assert.match(migration.stdout, /applied 20260814020000_rectification_v10_runtime\.sql/);
    assert.match(migration.stdout, /applied 20260814025000_personal_report_document_v2\.sql/);
    assert.match(migration.stdout, /applied 20260814040000_personal_report_jobs_v2\.sql/);
    assert.match(migration.stdout, /applied 20260816010000_accept_exact_family_birth_times\.sql/);
    assert.match(migration.stdout, /applied 20260818010000_admin_product_catalog_mutations\.sql/);
    assert.match(migration.stdout, /applied 20260819010000_rectification_ingest_precision_plateau\.sql/);
    assert.match(migration.stdout, /applied 20260822010000_declared_birth_window_clocks\.sql/);
    assert.match(migration.stdout, /applied 20260823010000_rectification_occupation_dateless\.sql/);
    assert.match(migration.stdout, /applied 20260823020000_rectification_inference_choice_write\.sql/);
    assert.match(migration.stdout, /applied 20260824010000_rectification_inference_transition_ledger\.sql/);
    assert.match(migration.stdout, /applied 20260824020000_rectification_choice_action\.sql/);
    assert.match(migration.stdout, /applied 20260824030000_rectification_turn_origin\.sql/);
    assert.match(migration.stdout, /applied 20260831020000_feature_pricing_admin_runtime_read_policy\.sql/);
    assert.match(migration.stdout, /applied 20260901010000_append_consultation_question\.sql/);
    assert.match(migration.stdout, /applied 20260901020000_chat_session_pin_archive\.sql/);
    assert.match(migration.stdout, /applied 20260905010000_personal_report_longform_appendices\.sql/);
    assert.equal(
      existsSync(fileURLToPath(new URL("../db/migrations/20260901020000_chat_session_pin_archive.sql", import.meta.url))),
      false,
      "business migration must not be copied into frontend/db/migrations (BUG-127/BUG-144)",
    );
    assert.equal(
      existsSync(fileURLToPath(new URL("../db/migrations/20260905010000_personal_report_longform_appendices.sql", import.meta.url))),
      false,
      "business migration must not be copied into frontend/db/migrations (BUG-127/BUG-144)",
    );
    fixture.psql(pinArchiveMigration);
    assert.equal(
      fixture.psql(`
        select is_nullable || ':' || data_type
        from information_schema.columns
        where table_schema = 'public'
          and table_name = 'chat_sessions'
          and column_name = 'pinned'
      `),
      "NO:boolean",
    );
    assert.equal(
      fixture.psql(`
        select is_nullable || ':' || data_type
        from information_schema.columns
        where table_schema = 'public'
          and table_name = 'chat_sessions'
          and column_name = 'archived_at'
      `),
      "YES:timestamp with time zone",
    );

    assert.equal(
      fixture.psql(`
        select is_nullable || ':' || data_type
        from information_schema.columns
        where table_schema = 'public'
          and table_name = 'birth_time_rectification_turns'
          and column_name = 'user_message'
      `),
      "YES:text",
    );
    assert.equal(
      fixture.psql(`
        select is_nullable || ':' || data_type || ':' || column_default
        from information_schema.columns
        where table_schema = 'public'
          and table_name = 'birth_time_rectification_turns'
          and column_name = 'user_message_captured'
      `),
      "NO:boolean:f",
    );
    assert.equal(
      fixture.psql(`
        select
          has_function_privilege(
            'service_role',
            'public.load_conversational_rectification_case_with_history(uuid, uuid)',
            'execute'
          ) || ':' ||
          has_function_privilege(
            'authenticated',
            'public.load_conversational_rectification_case_with_history(uuid, uuid)',
            'execute'
          )
      `),
      "true:f",
    );
    assert.equal(
      fixture.psql(`
        select
          has_table_privilege('service_role', 'public.consultation_requests', 'select') || ':' ||
          has_table_privilege('anon', 'public.consultation_requests', 'select') || ':' ||
          has_table_privilege('authenticated', 'public.consultation_requests', 'select')
      `),
      "true:f:f",
    );

    fixture.psql(`
      insert into public.feature_pricing(feature_key, model_tier, credit_cost, version)
      values ('rectification', 'standard', 3, 1)
    `);
    assert.equal(
      fixture.psqlAs(
        "admin_runtime",
        "admin-runtime-test-password",
        "select feature_key || ':' || model_tier || ':' || credit_cost || ':' || status from public.feature_pricing",
      ),
      "rectification:standard:3:draft",
    );

    // Public table allowlist. Last change (BUG-548):
    // original: omitted personal_report_longform_appendices
    // new: insert it between personal_report_jobs and personal_report_sections
    // reason: 20260905010000 added the appendix cache table; this list must match live pg_tables
    assert.equal(
      fixture.psql(`
        select string_agg(tablename, ',' order by tablename)
        from pg_tables
        where schemaname = 'public'
      `),
      [
        "admin_permissions",
        "admin_role_permissions",
        "admin_roles",
        "admin_session_revocations",
        "admin_user_roles",
        "admin_users",
        "agentic_rectification_candidate_decisions",
        "agentic_rectification_candidates",
        "agentic_rectification_case_conversation_summaries",
        "agentic_rectification_cases",
        "agentic_rectification_choice_actions",
        "agentic_rectification_conversation_focuses",
        "agentic_rectification_evidence",
        "agentic_rectification_inference_transitions",
        "agentic_rectification_open_ledger",
        "agentic_rectification_results",
        "agentic_rectification_run_attempts",
        "agentic_rectification_run_phases",
        "agentic_rectification_skill_run_receipts",
        "agentic_rectification_skill_upgrade_receipts",
        "agentic_rectification_tool_receipts",
        "agentic_rectification_turn_regenerations",
        "agentic_rectification_turns",
        "billing_products",
        "birth_time_rectification_action_receipts",
        "birth_time_rectification_agent_runs",
        "birth_time_rectification_billing",
        "birth_time_rectification_candidate_feature_snapshots",
        "birth_time_rectification_cases",
        "birth_time_rectification_diagnostics",
        "birth_time_rectification_dynamic_state",
        "birth_time_rectification_event_evidence",
        "birth_time_rectification_handoff_attach_receipts",
        "birth_time_rectification_handoff_settlements",
        "birth_time_rectification_pending_evidence",
        "birth_time_rectification_public_messages",
        "birth_time_rectification_question_handoffs",
        "birth_time_rectification_scoring_jobs",
        "birth_time_rectification_turns",
        "birth_time_rectification_v4_actions",
        "birth_time_rectification_v4_candidate_snapshots",
        "birth_time_rectification_v4_cases",
        "birth_time_rectification_v4_event_revisions",
        "birth_time_rectification_v4_events",
        "birth_time_rectification_v4_handoff_attach_receipts",
        "birth_time_rectification_v4_handoff_settlements",
        "birth_time_rectification_v4_handoffs",
        "birth_time_rectification_v4_jobs",
        "birth_time_rectification_v4_turns",
        "chart_profiles",
        "chat_sessions",
        "consultation_requests",
        "credit_request_cancellations",
        "credit_transactions",
        "epay_settings",
        "feature_flags",
        // This migration intentionally adds the server-owned feature pricing table.
        "feature_pricing",
        "model_config_versions",
        "model_configs",
        "model_connection_test_evidence",
        "model_providers",
        "model_publish_events",
        "notification_templates",
        "payment_orders",
        "payment_packages",
        "personal_report_jobs",
        "personal_report_longform_appendices",
        "personal_report_sections",
        "personal_reports",
        "pricing_experiment_events",
        "product_entitlements",
        "profiles",
        "redemption_attempts",
        "redemption_codes",
        "synastry_reports",
        "usage_events",
        "usage_ledger",
        "usage_reservations",
        "user_product_redemptions",
        "user_subscriptions",
      ].join(","),
    );
    assert.equal(
      fixture.psql(`
        select string_agg(column_name, ',' order by column_name)
        from information_schema.columns
        where table_schema = 'public'
          and table_name = 'agentic_rectification_turns'
          and column_name in ('client_action_id', 'content_hash', 'message_origin')
      `),
      "client_action_id,content_hash,message_origin",
    );
    assert.equal(
      fixture.psql(`
        select
          has_function_privilege(
            'service_role',
            'public.record_agentic_rectification_turn_origin(uuid, uuid, uuid, text, uuid, text)',
            'execute'
          ) || ':' ||
          has_function_privilege(
            'authenticated',
            'public.record_agentic_rectification_turn_origin(uuid, uuid, uuid, text, uuid, text)',
            'execute'
          )
      `),
      "true:f",
    );

    fixture.psqlAs(
      "identity_runtime",
      "identity-runtime-test-password",
      `
        insert into identity.users (name, email, email_verified, email_verified_at)
        values ('Local User', 'local-user@example.com', true, now())
      `,
    );
    const userId = fixture.psql(
      "select id from identity.users where email = 'local-user@example.com'",
    );
    assert.equal(
      fixture.psql(`select email from auth.users where id = '${userId}'`),
      "local-user@example.com",
    );
    assert.equal(
      fixture.psql(`select email || ':' || credits from public.profiles where id = '${userId}'`),
      "local-user@example.com:0",
    );
    assert.equal(
      fixture.psqlAs(
        "app_runtime",
        "app-runtime-test-password",
        `set role authenticated;
         select set_config('request.jwt.claim.sub', '${userId}', true);
         select email from public.profiles where id = '${userId}'`,
      ),
      `SET\n${userId}\nlocal-user@example.com`,
    );
    fixture.psql(`update public.profiles set birth_date = '1997-08-08' where id = '${userId}'`);
    fixture.psql(`
      update public.profiles
      set reported_birth_time = '05:00',
          active_birth_time = null,
          birth_time_source = 'family_exact',
          uncertainty_before_minutes = 0,
          uncertainty_after_minutes = 0,
          birth_time_status = 'reported',
          rectification_case_id = null
      where id = '${userId}'
    `);
    fixture.psql(acceptedExactFamilyMigration);
    assert.equal(
      fixture.psql(`
        select to_char(reported_birth_time, 'HH24:MI') || ':' ||
          to_char(active_birth_time, 'HH24:MI') || ':' || birth_time_status
        from public.profiles
        where id = '${userId}'
      `),
      "05:00:05:00:accepted",
    );
    fixture.psql(`
      update public.profiles
      set reported_birth_time = '05:00',
          active_birth_time = null,
          birth_time_source = 'family_exact',
          uncertainty_before_minutes = 10,
          uncertainty_after_minutes = 10,
          birth_time_status = 'reported',
          rectification_case_id = null
      where id = '${userId}'
    `);
    fixture.psql(acceptedExactFamilyMigration);
    assert.equal(
      fixture.psql(`
        select active_birth_time is null || ':' || birth_time_status
        from public.profiles
        where id = '${userId}'
      `),
      "true:reported",
    );
    fixture.psql(`
      update public.profiles
      set reported_birth_time = null,
          active_birth_time = null,
          birth_time_source = null,
          uncertainty_before_minutes = null,
          uncertainty_after_minutes = null,
          birth_time_status = null,
          rectification_case_id = null
      where id = '${userId}'
    `);

    const local = createLocalPostgresDataClient(
      fixture.connectionUrl("app_runtime", "app-runtime-test-password"),
      { id: userId, email: "local-user@example.com" },
    );
    const profile = await local.from("profiles")
      .select("id,email,credits,birth_date,created_at")
      .eq("id", userId)
      .single();
    assert.equal(profile.error, null);
    const { created_at: createdAt, ...profileData } = profile.data as Record<string, unknown>;
    assert.equal(typeof createdAt, "string");
    assert.match(String(createdAt), /^\d{4}-\d{2}-\d{2}T/);
    assert.deepEqual(profileData, {
      id: userId,
      email: "local-user@example.com",
      credits: 0,
      birth_date: "1997-08-08",
    });
    const nonAbandonedProfiles = await local.from("profiles")
      .select("id")
      .neq("email", "not-local-user@example.com")
      .single();
    assert.equal(nonAbandonedProfiles.error, null);
    assert.deepEqual(nonAbandonedProfiles.data, { id: userId });
    const admin = createLocalPostgresDataClient(
      fixture.connectionUrl("service_runtime", "service-runtime-test-password"),
      null,
      "service_role",
    );
    const adminProfile = await admin.from("profiles")
      .select("id")
      .eq("id", userId)
      .single();
    assert.equal(adminProfile.error, null);
    assert.deepEqual(adminProfile.data, { id: userId });

    const serviceScalar = (sql: string) => {
      const output = fixture.psqlAs(
        "service_runtime",
        "service-runtime-test-password",
        `set role service_role; ${sql}`,
      );
      return output.split("\n").at(-1) ?? "";
    };

    const terminalCaseId = "44444444-4444-4444-8444-444444444444";
    const terminalSessionId = "55555555-5555-4555-8555-555555555555";
    const terminalRequestId = "77777777-7777-4777-8777-777777777777";
    const skillSha256 = "a".repeat(64);
    const skillSourceCommit = "b".repeat(40);
    fixture.psql(`
      insert into public.chat_sessions (
        id, user_id, title, theme, model_id, messages, session_type, updated_at
      ) values (
        '${terminalSessionId}', '${userId}', 'V10 terminal finalize', 'general',
        'test-model', '[]', 'birth_time_rectification', now()
      );
      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 (
        '${terminalCaseId}', '${userId}', '${terminalSessionId}', 'collecting_evidence',
        'jyotish-birth-time-rectification', '10.0.0', '${skillSha256}',
        '${skillSourceCommit}', 'v10-terminal-fixture', '{}', '{}'
      );
    `);

    const terminalTurnId = serviceScalar(`
      select result ->> 'turn_id'
      from (
        select public.append_agentic_rectification_turn(
          '${userId}', '${terminalCaseId}', '确认这个时间', null,
          'test-model', null, 'pending', '${terminalRequestId}'
        ) as result
      ) appended
    `);
    assert.match(terminalTurnId, /^[0-9a-f-]{36}$/);
    assert.equal(
      serviceScalar(`
        select (result ->> 'turn_id') || ':' ||
          (result ->> 'should_execute') || ':' ||
          (result ->> 'idempotent')
        from (
          select public.append_agentic_rectification_turn(
            '${userId}', '${terminalCaseId}', '确认这个时间', null,
            'test-model', null, 'pending', '${terminalRequestId}'
          ) as result
        ) replayed
      `),
      `${terminalTurnId}:false:true`,
    );
    assert.equal(
      fixture.psql(`
        select count(*)
        from public.agentic_rectification_turns
        where case_id = '${terminalCaseId}' and request_id = '${terminalRequestId}'
      `),
      "1",
    );
    assert.throws(
      () => serviceScalar(`
        select public.append_agentic_rectification_turn(
          '${userId}', '${terminalCaseId}', '不同的请求正文', null,
          'test-model', null, 'pending', '${terminalRequestId}'
        )
      `),
      /agentic_rectification_request_mismatch/,
    );
    for (const legacySignature of [
      "public.append_agentic_rectification_turn(uuid,uuid,text,text,text,text,text)",
      "public.finalize_agentic_rectification_turn(uuid,uuid,uuid,text,text)",
      "public.finalize_agentic_rectification_turn(uuid,uuid,uuid,text,text,uuid)",
    ]) {
      assert.equal(
        fixture.psql(`select pg_catalog.has_function_privilege('service_role', '${legacySignature}', 'EXECUTE')`),
        "f",
      );
    }

    const terminalAttemptId = serviceScalar(`
      select result ->> 'attempt_id'
      from (
        select public.create_agentic_rectification_run_attempt(
          '${userId}', '${terminalCaseId}', '${terminalTurnId}', 1
        ) as result
      ) claimed
    `);
    fixture.psql(`
      update public.agentic_rectification_cases
      set status = 'closed', completed_at = now(), closed_reason = 'other'
      where id = '${terminalCaseId}';
    `);
    assert.equal(
      serviceScalar(`
        select (result ->> 'turn_id') || ':' ||
          (result ->> 'should_execute') || ':' ||
          (result ->> 'idempotent')
        from (
          select public.append_agentic_rectification_turn(
            '${userId}', '${terminalCaseId}', '确认这个时间', null,
            'test-model', null, 'pending', '${terminalRequestId}'
          ) as result
        ) replayed
      `),
      `${terminalTurnId}:false:true`,
    );
    assert.throws(
      () => serviceScalar(`
        select public.append_agentic_rectification_turn(
          '${userId}', '${terminalCaseId}', '关闭后新请求', null,
          'test-model', null, 'pending', 'dddddddd-dddd-4ddd-8ddd-dddddddddddd'
        )
      `),
      /agentic_rectification_case_terminal/,
    );
    assert.equal(
      serviceScalar(`
        select (result ->> 'attempt_id') || ':' ||
          (result ->> 'should_execute') || ':' ||
          (result ->> 'already_in_progress') || ':' ||
          (result ->> 'idempotent')
        from (
          select public.create_agentic_rectification_run_attempt(
            '${userId}', '${terminalCaseId}', '${terminalTurnId}', 1
          ) as result
        ) replayed
      `),
      `${terminalAttemptId}:false:true:true`,
    );
    assert.throws(
      () => serviceScalar(`
        select public.create_agentic_rectification_run_attempt(
          '${userId}', '${terminalCaseId}', '${terminalTurnId}', 2
        )
      `),
      /agentic_rectification_case_terminal/,
    );
    serviceScalar(`
      select public.insert_agentic_rectification_skill_run_receipt(
        '${userId}', '${terminalCaseId}', '${terminalTurnId}', '${terminalAttemptId}',
        'turn', 'jyotish-birth-time-rectification', '10.0.0',
        '${skillSha256}', '${skillSourceCommit}'
      )
    `);
    serviceScalar(`
      select public.insert_agentic_rectification_run_phase(
        '${userId}', '${terminalCaseId}', '${terminalTurnId}',
        'billing.settled', null, 1, '${terminalAttemptId}'
      )
    `);
    serviceScalar(`
      select public.insert_agentic_rectification_run_phase(
        '${userId}', '${terminalCaseId}', '${terminalTurnId}',
        'run.completed', null, 2, '${terminalAttemptId}'
      )
    `);
    assert.equal(
      serviceScalar(`
        select (result ->> 'status') || ':' || (result ->> 'idempotent')
        from (
          select public.finalize_agentic_rectification_run_attempt(
            '${userId}', '${terminalCaseId}', '${terminalTurnId}', '${terminalAttemptId}',
            'completed', null, '{"inputTokens":11,"outputTokens":7}'::jsonb
          ) as result
        ) finalized
      `),
      "completed:false",
    );
    assert.equal(
      serviceScalar(`
        select (result ->> 'status') || ':' || (result ->> 'idempotent')
        from (
          select public.finalize_agentic_rectification_turn(
            '${userId}', '${terminalCaseId}', '${terminalTurnId}', '${terminalAttemptId}',
            'completed', '已确认并完成', '${terminalAttemptId}'
          ) as result
        ) finalized
      `),
      "completed:false",
    );
    assert.throws(
      () => serviceScalar(`
        select public.finalize_agentic_rectification_turn(
          '${userId}', '${terminalCaseId}', '${terminalTurnId}', '${terminalAttemptId}',
          'failed', null, null
        )
      `),
      /agentic_rectification_turn_already_completed/,
    );
    assert.equal(
      fixture.psql(`
        select status || ':' || assistant_message || ':' || successful_attempt_id
        from public.agentic_rectification_turns
        where id = '${terminalTurnId}'
      `),
      `completed:已确认并完成:${terminalAttemptId}`,
    );

    const supersededCaseId = "88888888-8888-4888-8888-888888888888";
    const supersededSessionId = "99999999-9999-4999-8999-999999999999";
    const supersededRequestId = "cccccccc-cccc-4ccc-8ccc-cccccccccccc";
    fixture.psql(`
      insert into public.chat_sessions (
        id, user_id, title, theme, model_id, messages, session_type, updated_at
      ) values (
        '${supersededSessionId}', '${userId}', 'V10 superseded attempt', 'general',
        'test-model', '[]', 'birth_time_rectification', now()
      );
      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 (
        '${supersededCaseId}', '${userId}', '${supersededSessionId}', 'collecting_evidence',
        'jyotish-birth-time-rectification', '10.0.0', '${skillSha256}',
        '${skillSourceCommit}', 'v10-superseded-fixture', '{}', '{}'
      );
    `);
    const supersededTurnId = serviceScalar(`
      select result ->> 'turn_id'
      from (
        select public.append_agentic_rectification_turn(
          '${userId}', '${supersededCaseId}', '继续核对', null,
          'test-model', null, 'pending', '${supersededRequestId}'
        ) as result
      ) appended
    `);
    const oldAttemptId = serviceScalar(`
      select result ->> 'attempt_id'
      from (
        select public.create_agentic_rectification_run_attempt(
          '${userId}', '${supersededCaseId}', '${supersededTurnId}', 1
        ) as result
      ) claimed
    `);
    serviceScalar(`
      select public.finalize_agentic_rectification_run_attempt(
        '${userId}', '${supersededCaseId}', '${supersededTurnId}', '${oldAttemptId}',
        'retryable', 'stream_unfinished', '{}'::jsonb
      )
    `);
    const latestAttemptId = serviceScalar(`
      select result ->> 'attempt_id'
      from (
        select public.create_agentic_rectification_run_attempt(
          '${userId}', '${supersededCaseId}', '${supersededTurnId}', 2
        ) as result
      ) claimed
    `);
    assert.notEqual(latestAttemptId, oldAttemptId);
    assert.throws(
      () => serviceScalar(`
        select public.finalize_agentic_rectification_turn(
          '${userId}', '${supersededCaseId}', '${supersededTurnId}', '${oldAttemptId}',
          'retryable', null, null
        )
      `),
      /agentic_rectification_attempt_superseded/,
    );

    const sessionId = "11111111-1111-4111-8111-111111111111";
    const inserted = await local.from("chat_sessions").insert({
      id: sessionId,
      user_id: userId,
      title: "Local conversation",
      theme: "general",
      model_id: "test-model",
      messages: [],
      session_type: "consultation",
      rectification_case_id: null,
      updated_at: new Date().toISOString(),
    }).select("id").single();
    assert.equal(inserted.error, null);
    assert.deepEqual(inserted.data, { id: sessionId });

    const pinDefaults = await local
      .from("chat_sessions")
      .select("pinned,archived_at")
      .eq("id", sessionId)
      .single();
    assert.equal(pinDefaults.error, null);
    assert.deepEqual(pinDefaults.data, { pinned: false, archived_at: null });

    const ownerPin = await local
      .from("chat_sessions")
      .update({ pinned: true, archived_at: "2026-09-01T00:00:00.000Z" })
      .eq("id", sessionId)
      .select("pinned,archived_at")
      .single();
    assert.equal(ownerPin.error, null);
    assert.equal((ownerPin.data as { pinned: boolean }).pinned, true);
    assert.ok((ownerPin.data as { archived_at: string | null }).archived_at);

    fixture.psqlAs(
      "identity_runtime",
      "identity-runtime-test-password",
      `
        insert into identity.users (name, email, email_verified, email_verified_at)
        values ('Pin Archive Other', 'pin-archive-other@example.com', true, now())
      `,
    );
    const otherUserId = fixture.psql(
      "select id from identity.users where email = 'pin-archive-other@example.com'",
    );
    const other = createLocalPostgresDataClient(
      fixture.connectionUrl("app_runtime", "app-runtime-test-password"),
      { id: otherUserId, email: "pin-archive-other@example.com" },
    );
    const stolen = await other
      .from("chat_sessions")
      .update({ pinned: false, archived_at: null })
      .eq("id", sessionId)
      .select("id");
    assert.equal((Array.isArray(stolen.data) ? stolen.data : []).length, 0);
    assert.equal(
      fixture.psql(`select pinned from public.chat_sessions where id = '${sessionId}'`),
      "t",
    );
    assert.equal(
      fixture.psql(`select archived_at is not null from public.chat_sessions where id = '${sessionId}'`),
      "t",
    );

    const ownerRestore = await local
      .from("chat_sessions")
      .update({ pinned: false, archived_at: null })
      .eq("id", sessionId)
      .select("pinned,archived_at")
      .single();
    assert.equal(ownerRestore.error, null);
    assert.deepEqual(ownerRestore.data, { pinned: false, archived_at: null });

    const beforeTomorrow = await local
      .from("chat_sessions")
      .select("id")
      .eq("user_id", userId)
      .eq("session_type", "consultation")
      .lte("updated_at", new Date(Date.now() + 86_400_000).toISOString())
      .single();
    assert.equal(beforeTomorrow.error, null);
    assert.deepEqual(beforeTomorrow.data, { id: sessionId });

    const matchingTitle = await local
      .from("chat_sessions")
      .select("id")
      .eq("user_id", userId)
      .like("title", "Local%")
      .single();
    assert.equal(matchingTitle.error, null);
    assert.deepEqual(matchingTitle.data, { id: sessionId });

    const injectedLike = await local
      .from("chat_sessions")
      .select("id")
      .eq("user_id", userId)
      .like("title", "Local%' OR true --");
    assert.equal(injectedLike.error, null);
    assert.deepEqual(injectedLike.data, []);

    const rectificationSessionId = "22222222-2222-4222-8222-222222222222";
    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}';
      insert into public.chat_sessions (id, user_id, title, theme, model_id, messages, session_type, updated_at)
      values ('${rectificationSessionId}', '${userId}', 'Rectification', 'general', 'test-model', '[]', 'birth_time_rectification', now());
      insert into public.agentic_rectification_results (
        id, user_id, session_id, engine_result_id, canonical_input_hash, algorithm_version,
        candidate_range, candidates, overall_confidence, selection_allowed, confirmation_allowed,
        representative_time, 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 (
        '33333333-3333-4333-8333-333333333333', '${userId}', '${rectificationSessionId}',
        'engine-result-1', 'canonical-hash-1', 'test-v1', '{}',
        '[{"rank":1,"time":"04:55","relative_support":60,"tied_minute_count":1},{"rank":2,"time":"05:07","relative_support":40,"tied_minute_count":1}]',
        'medium', true, false, '04:55', '1997-08-08', '05:00', 'family_exact', 10, 10,
        36.420487, 114.209936, 8
      );
    `);
    const latestCandidate = await loadLatestAgenticRectificationResult(
      admin as never,
      userId,
      rectificationSessionId,
    );
    assert.equal(latestCandidate?.resultId, "33333333-3333-4333-8333-333333333333");
    assert.equal(latestCandidate?.selectionAllowed, true);
    assert.deepEqual(latestCandidate?.candidates.map(({ time, relative_support }) => ({ time, relative_support })), [
      { time: "04:55", relative_support: 60 },
      { time: "05:07", relative_support: 40 },
    ]);
    assert.equal(
      fixture.psql(`
        select
          not has_function_privilege('service_role', 'public.accept_agentic_rectification_candidate(uuid,uuid,uuid,time without time zone)', 'EXECUTE')
          and has_function_privilege('service_role', 'public.accept_agentic_rectification_candidate_for_case_v2(uuid,uuid,uuid,uuid,uuid,jsonb)', 'EXECUTE')
      `),
      "t",
    );
    assert.throws(
      () => fixture.psqlAs(
        "service_runtime",
        "service-runtime-test-password",
        `set role service_role;
         select public.accept_agentic_rectification_candidate(
           '${userId}', '${rectificationSessionId}', '33333333-3333-4333-8333-333333333333', '04:55'
         )`,
      ),
      /permission denied for function accept_agentic_rectification_candidate/,
    );
    assert.equal(
      fixture.psql(`select coalesce(to_char(active_birth_time, 'HH24:MI'), 'null') || ':' || birth_time_status || ':' || to_char(reported_birth_time, 'HH24:MI') from public.profiles where id = '${userId}'`),
      "null:reported:05:00",
    );

    const rectificationCaseId = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa";
    const baselineBirthSnapshot = {
      birth_date: "1997-08-08",
      reported_birth_time: "05:00",
      active_birth_time: null,
      birth_time_source: "family_exact",
      birth_time_period: null,
      uncertainty_before_minutes: 10,
      uncertainty_after_minutes: 10,
      latitude: 36.420487,
      longitude: 114.209936,
      timezone_offset: 8,
    };
    const candidateRange = { start_time: "04:50", end_time: "05:10" };
    const candidateInferenceState = {
      algorithm_version: "database-local-business-v2",
      candidate_set_id: "04:50-05:10:04:55,05:07",
      revision: 0,
      phase: "candidate_ready",
      result_status: "completed_with_range",
      range_start: "04:50",
      range_end: "05:10",
      candidates: [
        { id: "04:55", time: "04:55", cluster_range: ["04:55", "04:55"], prior_score: 60, posterior_score: 60, probability: 0.6, status: "active", rank: 1, strong_conflict_count: 0 },
        { id: "05:07", time: "05:07", cluster_range: ["05:07", "05:07"], prior_score: 40, posterior_score: 40, probability: 0.4, status: "active", rank: 2, strong_conflict_count: 0 },
      ],
      events: [],
      probes: [],
      answered_probes: [],
      rounds: [],
      entropy: 1,
      representative_time: "04:55",
      credible_range: ["04:55", "05:07"],
    };
    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 (
        '${rectificationCaseId}', '${userId}', '${rectificationSessionId}', 'candidate_ready',
        'jyotish-birth-time-rectification', '10.0.0', '${"e".repeat(64)}',
        '${JSON.stringify(baselineBirthSnapshot)}'::jsonb,
        '${JSON.stringify(candidateRange)}'::jsonb
      )
    `);

    const persisted = await admin.rpc("persist_agentic_rectification_candidate_v2", {
      p_user_id: userId,
      p_case_id: rectificationCaseId,
      p_engine_result_id: "database-local-business-v2",
      p_evidence_ledger_fingerprint: "f".repeat(64),
      p_candidate_range_fingerprint: "1".repeat(64),
      p_skill_version: "10.0.0",
      p_algorithm_version: "database-local-business-v2",
      p_event_contract_version: "rectification-event-contract-v2",
      p_decision_policy_version: "decision-policy-v2",
      p_candidate_range: candidateRange,
      p_candidates: [
        { rank: 1, time: "04:55", relative_support: 60, tied_minute_count: 1 },
        { rank: 2, time: "05:07", relative_support: 40, tied_minute_count: 1 },
      ],
      p_decision_receipt: {
        display_allowed: true,
        accept_allowed: true,
        confirm_allowed: false,
        representative_time: "04:55",
        overall_confidence: "medium",
        margin_percent: 20,
        inference_state: candidateInferenceState,
      },
      p_execution_ledger: [
        { phase: "candidate.score", status: "completed", engine: "fixture-engine" },
        { phase: "decision.evaluate", status: "completed", policy: "decision-policy-v2" },
      ],
    });
    assert.equal(persisted.error, null, rpcError(persisted.error));
    const persistedRow = persisted.data as Record<string, unknown>;
    const v2ResultId = String(persistedRow.result_id);
    const [firstCandidateId, secondCandidateId] = candidateIds(persistedRow);
    assert.ok(firstCandidateId);
    assert.ok(secondCandidateId);

    const firstAcceptRequestId = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb";
    const firstAccepted = await admin.rpc("accept_agentic_rectification_candidate_for_case_v2", {
      p_user_id: userId,
      p_case_id: rectificationCaseId,
      p_result_id: v2ResultId,
      p_candidate_id: firstCandidateId,
      p_request_id: firstAcceptRequestId,
    });
    assert.equal(firstAccepted.error, null, rpcError(firstAccepted.error));
    assert.deepEqual(
      {
        saved_time: (firstAccepted.data as Record<string, unknown>).saved_time,
        status: (firstAccepted.data as Record<string, unknown>).status,
        case_status: (firstAccepted.data as Record<string, unknown>).case_status,
        idempotent: (firstAccepted.data as Record<string, unknown>).idempotent,
      },
      { saved_time: "04:55", status: "accepted", case_status: "candidate_accepted", idempotent: false },
    );
    assert.equal(
      fixture.psql(`
        select to_char(active_birth_time, 'HH24:MI') || ':' ||
          to_char(birth_time, 'HH24:MI') || ':' || birth_time_status || ':' ||
          to_char(reported_birth_time, 'HH24:MI')
        from public.profiles where id = '${userId}'
      `),
      "04:55:04:55:accepted:05:00",
    );

    const firstReplay = await admin.rpc("accept_agentic_rectification_candidate_for_case_v2", {
      p_user_id: userId,
      p_case_id: rectificationCaseId,
      p_result_id: v2ResultId,
      p_candidate_id: firstCandidateId,
      p_request_id: firstAcceptRequestId,
    });
    assert.equal(firstReplay.error, null, rpcError(firstReplay.error));
    assert.equal((firstReplay.data as Record<string, unknown>).idempotent, true);

    const secondAccepted = await admin.rpc("accept_agentic_rectification_candidate_for_case_v2", {
      p_user_id: userId,
      p_case_id: rectificationCaseId,
      p_result_id: v2ResultId,
      p_candidate_id: secondCandidateId,
      p_request_id: "cccccccc-cccc-4ccc-8ccc-cccccccccccc",
    });
    assert.equal(secondAccepted.error, null, rpcError(secondAccepted.error));
    assert.deepEqual(
      {
        saved_time: (secondAccepted.data as Record<string, unknown>).saved_time,
        status: (secondAccepted.data as Record<string, unknown>).status,
        case_status: (secondAccepted.data as Record<string, unknown>).case_status,
        idempotent: (secondAccepted.data as Record<string, unknown>).idempotent,
      },
      { saved_time: "05:07", status: "accepted", case_status: "candidate_accepted", idempotent: false },
    );
    assert.equal(
      fixture.psql(`select to_char(active_birth_time, 'HH24:MI') || ':' || to_char(birth_time, 'HH24:MI') || ':' || to_char(reported_birth_time, 'HH24:MI') from public.profiles where id = '${userId}'`),
      "05:07:05:07:05:00",
    );

    assert.equal(
      fixture.psql(`select invalidated_at is null from public.agentic_rectification_results where id = '33333333-3333-4333-8333-333333333333'`),
      "t",
    );
    fixture.psql(`update public.profiles set reported_birth_time = '05:01' where id = '${userId}'`);
    assert.equal(
      fixture.psql(`select invalidated_at is not null from public.agentic_rectification_results where id = '33333333-3333-4333-8333-333333333333'`),
      "t",
    );
    assert.equal(
      fixture.psql(`select invalidated_at is not null from public.agentic_rectification_results where id = '${v2ResultId}'`),
      "t",
    );
    const expiredAccept = await admin.rpc("accept_agentic_rectification_candidate_for_case_v2", {
      p_user_id: userId,
      p_case_id: rectificationCaseId,
      p_result_id: v2ResultId,
      p_candidate_id: firstCandidateId,
      p_request_id: "dddddddd-dddd-4ddd-8ddd-dddddddddddd",
    });
    assert.match(rpcError(expiredAccept.error), /agentic_rectification_candidate_expired/);
    assert.equal(
      await loadLatestAgenticRectificationResult(admin as never, userId, rectificationSessionId),
      null,
    );

    assert.equal(
      fixture.psql(`
        select
          has_function_privilege(
            'service_role',
            'public.append_consultation_question(uuid, text, uuid, jsonb)',
            'execute'
          ) || ':' ||
          has_function_privilege(
            'authenticated',
            'public.append_consultation_question(uuid, text, uuid, jsonb)',
            'execute'
          ) || ':' ||
          has_function_privilege(
            'anon',
            'public.append_consultation_question(uuid, text, uuid, jsonb)',
            'execute'
          )
      `),
      "true:f:f",
    );

    const consultationSessionId = "88888888-8888-4888-8888-888888888888";
    fixture.psql(`
      insert into public.chat_sessions (
        id, user_id, title, theme, model_id, messages, session_type, updated_at
      ) values (
        '${consultationSessionId}', '${userId}', '新对话', 'general',
        'test-model', '[]', 'consultation', now()
      );
    `);
    const firstQuestion = await admin.rpc("append_consultation_question", {
      p_user_id: userId,
      p_request_id: "append-request-1",
      p_session_id: consultationSessionId,
      p_question_message: { role: "user", text: "第一问会不会丢" },
    });
    assert.equal(firstQuestion.error, null, rpcError(firstQuestion.error));
    const firstQuestionRow = Array.isArray(firstQuestion.data) ? firstQuestion.data[0] : firstQuestion.data;
    assert.equal((firstQuestionRow as { success?: unknown }).success, true);
    assert.equal(
      fixture.psql(`
        select title || ':' || jsonb_array_length(messages)::text || ':' || (messages->0->>'text')
        from public.chat_sessions where id = '${consultationSessionId}'
      `),
      "第一问会不会丢:1:第一问会不会丢",
    );

    const firstQuestionReplay = await admin.rpc("append_consultation_question", {
      p_user_id: userId,
      p_request_id: "append-request-1",
      p_session_id: consultationSessionId,
      p_question_message: { role: "user", text: "不该写入的重复提问" },
    });
    assert.equal(firstQuestionReplay.error, null, rpcError(firstQuestionReplay.error));
    const firstQuestionReplayRow = Array.isArray(firstQuestionReplay.data) ? firstQuestionReplay.data[0] : firstQuestionReplay.data;
    assert.equal((firstQuestionReplayRow as { success?: unknown }).success, true);
    assert.equal(
      fixture.psql(`select jsonb_array_length(messages) from public.chat_sessions where id = '${consultationSessionId}'`),
      "1",
    );

    const secondQuestion = await admin.rpc("append_consultation_question", {
      p_user_id: userId,
      p_request_id: "append-request-2",
      p_session_id: consultationSessionId,
      p_question_message: { role: "user", text: "第二问也会留下" },
    });
    assert.equal(secondQuestion.error, null, rpcError(secondQuestion.error));
    assert.equal(
      fixture.psql(`
        select jsonb_array_length(messages)::text || ':' || (messages->0->>'text') || ':' || (messages->1->>'text')
        from public.chat_sessions where id = '${consultationSessionId}'
      `),
      "2:第一问会不会丢:第二问也会留下",
    );

    fixture.psql(`
      update public.chat_sessions
      set messages = (
        select coalesce(jsonb_agg(jsonb_build_object('role','user','text','x','requestId', n::text)), '[]'::jsonb)
        from generate_series(1, 200) as n
      )
      where id = '${consultationSessionId}'
    `);
    const sessionFull = await admin.rpc("append_consultation_question", {
      p_user_id: userId,
      p_request_id: "append-request-full",
      p_session_id: consultationSessionId,
      p_question_message: { role: "user", text: "满了就不能再写" },
    });
    assert.equal(sessionFull.error, null, rpcError(sessionFull.error));
    const sessionFullRow = Array.isArray(sessionFull.data) ? sessionFull.data[0] : sessionFull.data;
    assert.deepEqual(sessionFullRow, { success: false, error_code: "session_full" });
    assert.equal(
      fixture.psql(`
        select jsonb_array_length(messages)
        from public.chat_sessions where id = '${consultationSessionId}'
      `),
      "200",
    );

    fixture.psql(`
      insert into public.redemption_codes (code_hash, code_mask, credits)
      values ('${"a".repeat(64)}', 'JYOTISH-****-TEST', 3)
    `);
    const redeemed = await local.rpc("redeem_code", {
      p_code_hash: "a".repeat(64),
    });
    assert.equal(redeemed.error, null);
    assert.deepEqual(redeemed.data, [{ success: true, credits: 3, awarded_credits: 3, error_code: null }]);
  } finally {
    await closeLocalPostgresDataPools();
    fixture.stop();
  }
});