"""Issue 18 comprehensive settlement acceptance contracts (synthetic SQLite only).

Public APIs are resolved at call time so the pre-implementation suite collects and
fails with an actionable assertion rather than an import/attribute error.
"""
import hashlib
import json
import sqlite3
from concurrent.futures import ThreadPoolExecutor

import pytest

from jarca_ledger_mcp import service
from jarca_ledger_mcp.db import init_db


def api(name):
    fn = getattr(service, name, None)
    assert callable(fn), f"required comprehensive settlement API missing: service.{name}"
    return fn


def dbfile(tmp_path):
    path = str(tmp_path / "comprehensive-synthetic.sqlite3")
    init_db(path)
    return path


def ok(value):
    assert value.get("success") is True, value
    return value


def new_draft(path, month="2026-09", key="draft"):
    created = service.create_manual_asset("Synthetic cash", 100, "2026-09-30", "cash", db_path=path)
    assert created.get("success") is True, created
    return ok(api("create_settlement_draft")(month, "synthetic-operator", key, db_path=path))["draft"]


def close_draft(path, draft, idem="close"):
    review_result = ok(api("review_settlement_draft")(draft["id"], db_path=path))
    review = review_result
    approved = api("approve_settlement_draft")(draft["id"], draft["version"], review["review_fingerprint"], "synthetic-reviewer", "synthetic approval", "ref-synthetic", db_path=path)
    ok(approved)
    fresh = ok(api("get_settlement_draft")(draft["id"], db_path=path))["draft"]
    return api("finalize_settlement_draft")(draft["id"], fresh["version"], review["review_fingerprint"], idem, db_path=path)


def test_draft_previews_six_totals_without_creating_run(tmp_path):
    path = dbfile(tmp_path)
    draft = new_draft(path)
    result = ok(api("review_settlement_draft")(draft["id"], db_path=path))
    review = result
    required = {"income_total", "expense_total", "net_cashflow", "total_assets", "total_liabilities", "net_worth"}
    assert required <= review.keys() and required <= review["totals"].keys()
    assert review["aggregation_version"] == "comprehensive-v1"
    assert review["snapshot_schema_version"] == 1
    assert isinstance(review["cashflow_items"], list)
    with sqlite3.connect(path) as db:
        assert db.execute("SELECT COUNT(*) FROM asset_snapshot_runs").fetchone()[0] == 0
        assert db.execute("SELECT COUNT(*) FROM settlement_drafts WHERE id=?", (draft["id"],)).fetchone()[0] == 1


def test_closed_header_persists_six_totals_policy_and_versions(tmp_path):
    path = dbfile(tmp_path)
    draft = new_draft(path)
    result = ok(close_draft(path, draft))
    assert result.get("run_id")
    saved = result
    for key in ("income_total", "expense_total", "net_cashflow", "total_assets", "total_liabilities", "net_worth"):
        assert key in saved
    assert saved["aggregation_version"] == "comprehensive-v1"
    assert saved["snapshot_schema_version"] == 1
    with sqlite3.connect(path) as db:
        columns = {row[1] for row in db.execute("PRAGMA table_info(asset_snapshot_runs)")}
        assert {"income_total", "expense_total", "net_cashflow", "total_assets", "total_liabilities", "net_worth", "cashflow_policy_json", "aggregation_version", "snapshot_schema_version", "source_snapshot_json"} <= columns
        row = db.execute("SELECT income_total,expense_total,net_cashflow,total_assets,total_liabilities,net_worth FROM asset_snapshot_runs WHERE id=?", (saved["run_id"],)).fetchone()
        assert len(row) == 6


def test_asset_detail_preserves_raw_evidence_and_signed_liability_contributions(tmp_path):
    path = dbfile(tmp_path)
    a = service.create_ledger_account("Synthetic debt A", type="loan", liquidity_bucket="loan", db_path=path)["account"]["id"]
    b = service.create_ledger_account("Synthetic debt B", type="loan", liquidity_bucket="loan", db_path=path)["account"]["id"]
    for account, amount, key in ((a, -900, "loan-a"), (b, 200, "loan-b")):
        v = service.register_account_valuation(account, "2026-09-30", amount, valuation_kind="loan_balance", representative_liability_definition="principal_only", verification_status="verified", idempotency_key=key, db_path=path)
        assert v["success"], v
    draft = new_draft(path)
    review_result = ok(api("review_settlement_draft")(draft["id"], db_path=path))
    review = review_result
    entries = review.get("items", [])
    assert entries
    assert any(x.get("amount") in (-900, 200) for x in entries)
    result = ok(close_draft(path, draft))
    evidence = json.dumps(result, sort_keys=True)
    assert "applied_amount" in evidence
    assert "-900" in evidence and "200" in evidence


def test_source_snapshot_preserves_reproducible_values_without_raw_secrets(tmp_path):
    path = dbfile(tmp_path)
    account = service.create_ledger_account("Synthetic evidence", last_balance=42, liquidity_bucket="cash", db_path=path)["account"]["id"]
    # Evidence is synthetic; no credentials or user-provided freeform payloads.
    v = service.register_account_valuation(account, "2026-09-30", 42, valuation_kind="balance", verification_status="verified", idempotency_key="value", source_ref="synthetic-ref", db_path=path)
    assert v["success"], v
    result = ok(close_draft(path, new_draft(path)))
    saved = result
    assert saved.get("source_snapshot") or saved.get("source_snapshot_json")
    encoded = json.dumps(saved, sort_keys=True)
    assert "synthetic-ref" in encoded
    for secret in ("raw_payload", "password", "cookie", "token"):
        assert secret not in encoded.lower()


@pytest.mark.parametrize("mutation", ["ledger_amount", "category_classification"])
def test_source_policy_and_category_drift_invalidate_approval(tmp_path, mutation):
    path = dbfile(tmp_path)
    category = service.create_ledger_category("Food", "Meal", cashflow_kind="expense", db_path=path)
    assert category.get("success") is True, category
    transaction = service.create_ledger_transaction(transaction_date="2026-09-12", type="expense", amount=17, category="Food", subcategory="Meal", description="synthetic drift", dedupe=False, db_path=path)
    assert transaction.get("success") is True, transaction
    txid = transaction.get("transaction", transaction).get("id")
    draft = new_draft(path)
    review_result = ok(api("review_settlement_draft")(draft["id"], db_path=path))
    review = review_result
    ok(api("approve_settlement_draft")(draft["id"], draft["version"], review["review_fingerprint"], "synthetic-reviewer", "approved", "ref", db_path=path))
    with sqlite3.connect(path) as db:
        if mutation == "ledger_amount":
            db.execute("UPDATE transactions SET amount=amount+1 WHERE id=?", (txid,))
        else:
            tables = {r[0] for r in db.execute("SELECT name FROM sqlite_master WHERE type='table'")}
            assert "ledger_categories" in tables
            cols = {r[1] for r in db.execute("PRAGMA table_info(ledger_categories)")}
            assert {"category", "subcategory", "cashflow_kind"} <= cols
            db.execute("UPDATE ledger_categories SET cashflow_kind='income' WHERE category=? AND subcategory=?", ("Food", "Meal"))
        db.commit()
    closed = api("finalize_settlement_draft")(draft["id"], draft["version"], review["review_fingerprint"], "close", db_path=path)
    assert closed.get("success") is False, closed
    with sqlite3.connect(path) as db:
        assert db.execute("SELECT COUNT(*) FROM asset_snapshot_runs").fetchone()[0] == 0


@pytest.mark.parametrize("failure_point", ["header", "assets", "cashflow", "completion"])
def test_mid_close_failure_rolls_back_all_settlement_state(tmp_path, failure_point):
    path = dbfile(tmp_path)
    ok(service.create_ledger_category("Work", "Salary", cashflow_kind="income", db_path=path))
    ok(service.create_ledger_transaction("2026-09-10", "income", 50, "Work", "Salary", db_path=path))
    draft = new_draft(path)
    review_result = ok(api("review_settlement_draft")(draft["id"], db_path=path))
    review = review_result
    ok(api("approve_settlement_draft")(draft["id"], draft["version"], review["review_fingerprint"], "synthetic-reviewer", "approved", "ref", db_path=path))
    with sqlite3.connect(path) as db:
        target = {"header":"asset_snapshot_runs", "assets":"asset_snapshots", "cashflow":"settlement_cashflow_snapshots", "completion":"settlement_drafts"}[failure_point]
        # Missing future table is an explicit RED assertion, not a SQL error masquerading as rollback.
        assert db.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (target,)).fetchone(), f"required comprehensive table missing: {target}"
        event = "UPDATE" if failure_point == "completion" else "INSERT"
        condition = " WHEN NEW.status='completed'" if failure_point == "completion" else ""
        db.execute(f"CREATE TRIGGER fail_{failure_point} BEFORE {event} ON {target}{condition} BEGIN SELECT RAISE(ABORT,'synthetic injected failure'); END")
    failed = api("finalize_settlement_draft")(draft["id"], draft["version"], review["review_fingerprint"], "close", db_path=path)
    assert failed.get("success") is False
    with sqlite3.connect(path) as db:
        assert db.execute("SELECT COUNT(*) FROM asset_snapshot_runs").fetchone()[0] == 0
        assert db.execute("SELECT COUNT(*) FROM asset_snapshots").fetchone()[0] == 0
        assert db.execute("SELECT COUNT(*) FROM settlement_cashflow_snapshots").fetchone()[0] == 0
        assert db.execute("SELECT status FROM settlement_drafts WHERE id=?", (draft["id"],)).fetchone()[0] == "active"


def test_close_rejects_stored_detail_sum_tampering(tmp_path):
    path = dbfile(tmp_path)
    ok(service.create_ledger_category("Work", "Salary", cashflow_kind="income", db_path=path))
    ok(service.create_ledger_transaction("2026-09-10", "income", 50, "Work", "Salary", db_path=path))
    draft = new_draft(path)
    with sqlite3.connect(path) as db:
        assert db.execute("SELECT 1 FROM sqlite_master WHERE name='settlement_cashflow_snapshots'").fetchone(), "cashflow detail storage is required"
        db.execute("CREATE TRIGGER corrupt_saved_detail AFTER INSERT ON settlement_cashflow_snapshots BEGIN UPDATE settlement_cashflow_snapshots SET amount=amount+1 WHERE id=NEW.id; END")
    result = close_draft(path, draft)
    assert result.get("success") is False, result
    with sqlite3.connect(path) as db:
        for table in ("asset_snapshot_runs", "asset_snapshots", "settlement_cashflow_snapshots"):
            assert db.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] == 0
        assert db.execute("SELECT status FROM settlement_drafts WHERE id=?", (draft["id"],)).fetchone()[0] == "active"


def test_replay_and_correction_keep_each_run_immutable(tmp_path):
    path = dbfile(tmp_path)
    first = new_draft(path)
    initial = ok(close_draft(path, first, "first-close"))
    replay = ok(api("finalize_settlement_draft")(first["id"], first["version"], initial["review_fingerprint"], "first-close", db_path=path))
    assert replay == initial
    correction = ok(api("create_settlement_draft")("2026-09", "synthetic-operator", "correction", correction_of_run_id=initial["run_id"], correction_reason="synthetic correction", db_path=path))["draft"]
    corrected = ok(close_draft(path, correction, "correction-close"))
    assert corrected["run_id"] != initial["run_id"]
    old = ok(api("get_monthly_settlement_snapshot")(run_id=initial["run_id"], db_path=path))
    new = ok(api("get_monthly_settlement_snapshot")(run_id=corrected["run_id"], db_path=path))
    assert old["run_id"] == initial["run_id"] and new["run_id"] == corrected["run_id"]


def test_concurrent_close_retries_return_one_saved_result(tmp_path):
    path = dbfile(tmp_path)
    draft = new_draft(path)
    review_result = ok(api("review_settlement_draft")(draft["id"], db_path=path))
    review = review_result
    ok(api("approve_settlement_draft")(draft["id"], draft["version"], review["review_fingerprint"], "synthetic-reviewer", "approved", "ref", db_path=path))
    def close():
        return api("finalize_settlement_draft")(draft["id"], draft["version"], review["review_fingerprint"], "same-key", db_path=path)
    with ThreadPoolExecutor(max_workers=4) as pool:
        results = list(pool.map(lambda _: close(), range(4)))
    successes = [r for r in results if r.get("success")]
    assert len(successes) == 4
    assert len({r["run_id"] for r in successes}) == 1


def test_readonly_snapshot_missing_database_does_not_create_file(tmp_path):
    path = tmp_path / "missing.sqlite3"
    result = api("get_monthly_settlement_snapshot")(month="2026-09", db_path=str(path))
    assert result.get("success") is False
    assert not path.exists()


def test_saved_reader_has_six_totals_and_legacy_null_is_not_live_zero(tmp_path):
    path = dbfile(tmp_path)
    # Empty database has no run; a saved reader must identify no snapshot, never fabricate current totals.
    result = api("get_monthly_settlement_snapshot")(month="2026-09", db_path=path)
    assert result.get("success") is False or result.get("source") == "settlement_snapshot"
    for key in ("income_total", "expense_total", "net_cashflow", "total_assets", "total_liabilities", "net_worth"):
        assert key not in result or result[key] is None


def test_cashflow_and_legacy_api_shapes_remain_distinct(tmp_path):
    path = dbfile(tmp_path)
    before = service.get_monthly_cashflow_report("2026-09", db_path=path)
    assert before.get("success") is True
    live_keys = set(before)
    draft = new_draft(path)
    review_result = ok(api("review_settlement_draft")(draft["id"], db_path=path))
    review = review_result
    assert {"income_total", "expense_total", "net_cashflow"} <= review.keys()
    after = service.get_monthly_cashflow_report("2026-09", db_path=path)
    assert set(after) == live_keys


def test_policy_setter_persists_explicit_lists_and_changes_review_fingerprint(tmp_path):
    path = dbfile(tmp_path)
    draft = new_draft(path)
    before = ok(api("review_settlement_draft")(draft["id"], db_path=path))["review_fingerprint"]
    policy = {"savings_categories":[{"category":"Savings","subcategory":"Reserve"}], "category_decisions":[]}
    changed = ok(api("set_settlement_cashflow_policy")(draft["id"], draft["version"], "synthetic-operator", policy, db_path=path))
    updated = changed["draft"]
    after = ok(api("review_settlement_draft")(draft["id"], db_path=path))["review_fingerprint"]
    assert updated["version"] == draft["version"] + 1
    assert before != after


def test_completed_get_and_saved_snapshot_ignore_later_live_ledger_changes(tmp_path):
    path = dbfile(tmp_path)
    draft = new_draft(path)
    saved = ok(close_draft(path, draft))
    ok(service.create_ledger_category("Later", "Income", cashflow_kind="income", db_path=path))
    ok(service.create_ledger_transaction("2026-09-20", "income", 99, "Later", "Income", db_path=path))
    history = ok(api("get_settlement_draft")(draft["id"], db_path=path))
    fetched = ok(api("get_monthly_settlement_snapshot")(run_id=saved["run_id"], db_path=path))
    assert history["final_response"] == saved
    assert fetched["income_total"] == saved["income_total"]


def _historical_asset_review_fingerprint(path, draft, review):
    """Exact f67 asset-only serial, with no new cashflow/policy/version inputs."""
    from jarca_ledger_mcp.application import settlement_draft_service as old
    with sqlite3.connect(path) as conn:
        conn.row_factory = sqlite3.Row
        stored = old._items(conn, draft["id"])
        states = []
        for item in stored:
            candidates = [dict(v) for v in conn.execute("SELECT * FROM account_valuations WHERE account_id=? ORDER BY id", (item["source_id"],))] if item["source_type"] == "account" else []
            states.append({"stored": item, "current": old._current_semantics(conn, item), "candidates": candidates})
        serial = {"draft_id": draft["id"], "version": draft["version"], "basis_date": draft["basis_date"], "states": states,
                  "warnings": [], "latest_run": old._latest_close(conn, draft["basis_date"]), "correction_parent": draft["correction_of_run_id"],
                  "totals": {k: review[k] for k in ("total_assets", "total_liabilities", "net_worth")}}
    return hashlib.sha256(json.dumps(serial, sort_keys=True, ensure_ascii=False).encode()).hexdigest()


def test_legacy_asset_approval_requires_comprehensive_review_before_close(tmp_path):
    path = dbfile(tmp_path)
    draft = new_draft(path)
    review = ok(service.review_settlement_draft(draft["id"], db_path=path))
    assert not review["warnings"], "this historical serial fixture is deliberately warning-free"
    historic = _historical_asset_review_fingerprint(path, draft, review)
    if "aggregation_version" not in review:
        assert historic == review["review_fingerprint"], "historical fixture must exactly match the original baseline approval"
    with sqlite3.connect(path) as conn:
        conn.execute("UPDATE settlement_drafts SET approval_fingerprint=?,approval_version=?,approved_by='synthetic-reviewer',approved_at='2026-09-30',approval_reason='historic asset review',approval_reference='historic-ref',approval_flags=? WHERE id=?",
                     (historic, draft["version"], json.dumps({"allow_missing_exclusions": False, "allow_carried_before": False, "allow_manual_substitute": False}), draft["id"]))
    refused = service.finalize_settlement_draft(draft["id"], draft["version"], historic, "historic-close", db_path=path)
    assert refused["success"] is False, "an asset-only historical approval cannot authorize comprehensive cashflow interpretation"
    with sqlite3.connect(path) as conn:
        assert conn.execute("SELECT COUNT(*) FROM asset_snapshot_runs").fetchone()[0] == 0
        assert conn.execute("SELECT status,final_response FROM settlement_drafts WHERE id=?", (draft["id"],)).fetchone() == ("active", "")
    fresh = ok(service.review_settlement_draft(draft["id"], db_path=path))
    assert fresh["review_fingerprint"] != historic
    ok(service.approve_settlement_draft(draft["id"], draft["version"], fresh["review_fingerprint"], "synthetic-reviewer", "comprehensive reapproval", "new-ref", db_path=path))
    saved = ok(service.finalize_settlement_draft(draft["id"], draft["version"], fresh["review_fingerprint"], "fresh-close", db_path=path))
    ok(service.create_ledger_category("Later", "Income", cashflow_kind="income", db_path=path))
    ok(service.create_ledger_transaction("2026-09-20", "income", 99, "Later", "Income", db_path=path))
    assert ok(service.get_settlement_draft(draft["id"], db_path=path))["final_response"] == saved
    assert ok(service.finalize_settlement_draft(draft["id"], draft["version"], fresh["review_fingerprint"], "fresh-close", db_path=path)) == saved


def _seed_reader_run(path, comprehensive=False, source=None):
    init_db(path)
    with sqlite3.connect(path) as conn:
        conn.execute("INSERT INTO asset_snapshot_runs(id,basis_date,close_status,created_at) VALUES ('synthetic-run','2026-09-30','complete','2026-09-30')")
        if comprehensive:
            conn.execute("UPDATE asset_snapshot_runs SET settlement_month='2026-09',currency='KRW',income_total=20,expense_total=0,net_cashflow=20,total_assets=0,total_liabilities=0,net_worth=0,cashflow_policy_json='{}',aggregation_version='comprehensive-v1',snapshot_schema_version=1,source_snapshot_json=? WHERE id='synthetic-run'", (json.dumps(source or {}),))
            conn.execute("INSERT INTO settlement_cashflow_snapshots(id,settlement_run_id,cashflow_kind,category,subcategory,amount,transaction_count,currency) VALUES ('synthetic-detail','synthetic-run','income','Salary','Monthly',20,1,'KRW')")


def test_legacy_reader_preserves_unrecorded_currency_and_marks_legacy(tmp_path):
    from jarca_ledger_mcp.application.monthly_settlement_service import get_monthly_settlement_snapshot
    path = str(tmp_path / "reader-legacy-null.sqlite3")
    _seed_reader_run(path)
    result = ok(get_monthly_settlement_snapshot(run_id="synthetic-run", db_path=path))
    assert result["legacy"] is True and result["is_comprehensive"] is False
    assert result["currency"] is None, "legacy currency must not be invented as KRW"
    assert all(value is None for value in result["totals"].values())


def test_legacy_reader_accepts_original_header_without_migration(tmp_path):
    from jarca_ledger_mcp.application.monthly_settlement_service import get_monthly_settlement_snapshot
    path = str(tmp_path / "reader-old-columns.sqlite3")
    _seed_reader_run(path)
    with sqlite3.connect(path) as conn:
        for column in ("settlement_month", "currency", "income_total", "expense_total", "net_cashflow", "total_assets", "total_liabilities", "net_worth", "cashflow_policy_json", "aggregation_version", "snapshot_schema_version", "source_snapshot_json"):
            conn.execute(f"ALTER TABLE asset_snapshot_runs DROP COLUMN {column}")
    for kwargs in ({"run_id": "synthetic-run"}, {"month": "2026-09"}):
        result = get_monthly_settlement_snapshot(db_path=path, **kwargs)
        assert result["success"], "pre-comprehensive headers must be read without new-column SQL"
        assert all(value is None for value in result["totals"].values())
    with sqlite3.connect(path) as conn:
        assert "settlement_month" not in {row[1] for row in conn.execute("PRAGMA table_info(asset_snapshot_runs)")}


def test_stored_reader_keeps_required_transaction_date(tmp_path):
    from jarca_ledger_mcp.application.monthly_settlement_service import get_monthly_settlement_snapshot
    path = str(tmp_path / "reader-source-date.sqlite3")
    tx = {"id": "synthetic-tx", "transaction_date": "2026-09-01", "type": "income", "amount": 20, "currency": "KRW", "category": "Salary", "subcategory": "Monthly", "cashflow_kind": "income", "applied_amount": 20}
    _seed_reader_run(path, True, {"transactions": [tx], "assets": []})
    result = ok(get_monthly_settlement_snapshot(run_id="synthetic-run", db_path=path))
    assert result["source_snapshot"]["transactions"] == [tx]


def test_stored_reader_allowlists_source_root_and_nested_decision(tmp_path):
    from jarca_ledger_mcp.application.monthly_settlement_service import get_monthly_settlement_snapshot
    path = str(tmp_path / "reader-source-private.sqlite3")
    marker = "SYNTHETIC_PRIVATE_SNAPSHOT"
    tx = {"id": "synthetic-tx", "transaction_date": "2026-09-01", "type": "income", "amount": 20, "currency": "KRW", "category": "Salary", "subcategory": "Monthly", "cashflow_kind": "income", "applied_amount": 20, "decision": {"category": "Salary", "subcategory": "Monthly", "cashflow_kind": "income", "reason": "synthetic decision", "private_note": marker}}
    _seed_reader_run(path, True, {"transactions": [tx], "assets": [], "private_note": marker})
    result = ok(get_monthly_settlement_snapshot(run_id="synthetic-run", db_path=path))
    assert marker not in json.dumps(result), "unknown root and nested source fields are not public financial evidence"


@pytest.mark.parametrize("target", ["header_total", "asset_amount", "asset_currency", "header_source", "header_policy", "header_version"])
def test_close_rejects_persisted_header_and_asset_tampering(tmp_path, target):
    path = dbfile(tmp_path)
    ok(service.create_ledger_category("Salary", "Monthly", cashflow_kind="income", db_path=path))
    ok(service.create_ledger_transaction("2026-09-01", "income", 20, "Salary", "Monthly", db_path=path))
    draft = new_draft(path)
    review = ok(service.review_settlement_draft(draft["id"], db_path=path))
    ok(service.approve_settlement_draft(draft["id"], draft["version"], review["review_fingerprint"], "reviewer", "synthetic", "ref", db_path=path))
    if target.startswith("asset_"):
        column, expression = ("amount", "amount+1") if target == "asset_amount" else ("currency", "'USD'")
        trigger = f"CREATE TRIGGER synthetic_tamper AFTER INSERT ON asset_snapshots BEGIN UPDATE asset_snapshots SET {column}={expression} WHERE id=NEW.id; END"
    else:
        column, expression = {"header_total": ("income_total", "income_total+1"), "header_source": ("source_snapshot_json", "'{}'"), "header_policy": ("cashflow_policy_json", "'{\"savings_categories\":[{\"category\":\"Other\",\"subcategory\":\"Pair\"}],\"category_decisions\":[]}'"), "header_version": ("aggregation_version", "'unknown-version'")}[target]
        trigger = f"CREATE TRIGGER synthetic_tamper AFTER INSERT ON settlement_cashflow_snapshots BEGIN UPDATE asset_snapshot_runs SET {column}={expression} WHERE id=NEW.settlement_run_id; END"
    with sqlite3.connect(path) as conn:
        conn.execute(trigger)
    result = service.finalize_settlement_draft(draft["id"], draft["version"], review["review_fingerprint"], "tamper-close", db_path=path)
    assert result["success"] is False, "persisted header/detail must match the approved result before commit"
    with sqlite3.connect(path) as conn:
        for table in ("asset_snapshot_runs", "asset_snapshots", "settlement_cashflow_snapshots"):
            assert conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] == 0
        assert conn.execute("SELECT status,final_run_id,final_response FROM settlement_drafts WHERE id=?", (draft["id"],)).fetchone() == ("active", "", "")


def test_stored_aggregation_version_drift_invalidates_approval(tmp_path):
    path = dbfile(tmp_path)
    draft = new_draft(path)
    review = ok(service.review_settlement_draft(draft["id"], db_path=path))
    ok(service.approve_settlement_draft(draft["id"], draft["version"], review["review_fingerprint"], "reviewer", "synthetic", "ref", db_path=path))
    with sqlite3.connect(path) as conn:
        conn.execute("UPDATE settlement_drafts SET aggregation_version='unknown-version' WHERE id=?", (draft["id"],))
    result = service.finalize_settlement_draft(draft["id"], draft["version"], review["review_fingerprint"], "version-drift", db_path=path)
    assert result["success"] is False, "unknown stored logic version cannot silently use current aggregation"


def test_completed_review_returns_saved_totals_and_fingerprint(tmp_path):
    path = dbfile(tmp_path)
    draft = new_draft(path)
    saved = ok(close_draft(path, draft))
    ok(service.create_ledger_category("Later", "Income", cashflow_kind="income", db_path=path))
    ok(service.create_ledger_transaction("2026-09-20", "income", 99, "Later", "Income", db_path=path))
    review = ok(service.review_settlement_draft(draft["id"], db_path=path))
    assert review["totals"] == saved["totals"]
    assert review["review_fingerprint"] == saved["review_fingerprint"]


def test_excluded_loan_source_has_no_applied_contribution(tmp_path):
    path = dbfile(tmp_path)
    values = []
    for name, amount in (("Synthetic A", -900), ("Synthetic B", 200)):
        account = ok(service.create_ledger_account(name, type="loan", liquidity_bucket="loan", db_path=path))["account"]["id"]
        value = ok(service.register_account_valuation(account, "2026-09-30", amount, valuation_kind="loan_balance", representative_liability_definition="principal_only", verification_status="verified", idempotency_key=name, db_path=path))["valuation"]
        values.append((account, value["id"]))
    draft = new_draft(path)
    edited = ok(service.set_settlement_draft_item(draft["id"], "account", values[1][0], draft["version"], "operator", valuation_id=values[1][1], included=False, exclusion_reason="synthetic exclusion", db_path=path))
    review = edited["review"]
    assert review["total_liabilities"] == 900
    excluded = next(x for x in review["source_snapshot"]["assets"] if x["source_id"] == values[1][0])
    assert excluded["amount"] == 200 and excluded["applied_amount"] is None


def test_cashflow_detail_currency_matches_header_without_losing_original(tmp_path):
    path = dbfile(tmp_path)
    ok(service.create_ledger_category("Salary", "Monthly", cashflow_kind="income", db_path=path))
    tx = ok(service.create_ledger_transaction("2026-09-01", "income", 20, "Salary", "Monthly", db_path=path))["transaction"]
    with sqlite3.connect(path) as conn:
        conn.execute("UPDATE transactions SET currency='krw' WHERE id=?", (tx["id"],))
    saved = ok(close_draft(path, new_draft(path)))
    assert saved["cashflow_items"][0]["currency"] == "KRW"
    assert saved["source_snapshot"]["transactions"][0]["currency"] == "krw"


def test_saved_reader_retains_asset_authority_and_liability_flags(tmp_path):
    path = dbfile(tmp_path)
    saved = ok(close_draft(path, new_draft(path)))
    snapshot = ok(service.get_monthly_settlement_snapshot(run_id=saved["run_id"], db_path=path))
    asset = snapshot["source_snapshot"]["assets"][0]
    assert asset.get("authority_source") == "manual_assets"
    assert asset.get("is_liability") is False
    assert asset.get("status") == "ready"


def test_saved_reader_projects_safe_header_exclusion_carry_and_approval_evidence(tmp_path):
    path = dbfile(tmp_path)
    saved = ok(close_draft(path, new_draft(path)))
    excluded = [{"source_type":"account", "source_id":"missing-account-41", "reason":"missing_verified_exact_valuation",
                 "valuation_date":"2026-09-27", "arbitrary_metadata":{"password":"DO-NOT-LEAK"}}]
    carried = [{"source_type":"account", "source_id":"carried-account-42", "reason":"as_of_carry_forward",
                "valuation_date":"2026-09-27", "arbitrary_metadata":{"token":"DO-NOT-LEAK"}}]
    approval = {"approved_by":"synthetic-reviewer", "approval_reason":"synthetic conditional approval",
                "approval_reference":"synthetic-ref", "allow_missing_exclusions":True,
                "allow_as_of_carry_forward":True, "private_note":"DO-NOT-LEAK"}
    with sqlite3.connect(path) as conn:
        conn.execute("UPDATE asset_snapshot_runs SET close_status='conditional',excluded_evidence=?,carried_evidence=?,approved_by=?,approval_reason=?,approval_reference=? WHERE id=?",
                     (json.dumps(excluded), json.dumps(carried), approval["approved_by"], approval["approval_reason"], approval["approval_reference"], saved["run_id"]))
        conn.commit()
        before = list(conn.iterdump())
    exact = ok(service.get_monthly_settlement_snapshot(run_id=saved["run_id"], db_path=path))
    latest = ok(service.get_monthly_settlement_snapshot(month="2026-09", db_path=path))
    assert exact == latest
    assert exact["run_id"] == saved["run_id"] and exact["close_status"] == "conditional"
    assert exact["excluded_items"][0]["source_id"] == "missing-account-41"
    assert exact["excluded_items"][0]["reason"] == "missing_verified_exact_valuation"
    assert exact["carried_items"][0]["source_id"] == "carried-account-42"
    assert exact["carried_items"][0]["valuation_date"] == "2026-09-27"
    assert exact["applied_evidence"]
    assert exact["approval"]["approval_reason"] == "synthetic conditional approval"
    encoded = json.dumps(exact, sort_keys=True)
    assert "DO-NOT-LEAK" not in encoded and "private_note" not in encoded and "arbitrary_metadata" not in encoded
    with sqlite3.connect(path) as conn:
        assert list(conn.iterdump()) == before


@pytest.mark.parametrize("schema_version,aggregation_version", [(1, None), (1, ""), (1, "unknown-version"), (2, "comprehensive-v1")])
def test_saved_reader_rejects_unsupported_comprehensive_versions_without_writes(tmp_path, schema_version, aggregation_version):
    path = dbfile(tmp_path)
    saved = ok(close_draft(path, new_draft(path)))
    with sqlite3.connect(path) as conn:
        conn.execute("UPDATE asset_snapshot_runs SET snapshot_schema_version=?,aggregation_version=? WHERE id=?", (schema_version, aggregation_version, saved["run_id"]))
        conn.commit()
        before = list(conn.iterdump())
    for query in ({"run_id": saved["run_id"]}, {"month": "2026-09"}):
        result = service.get_monthly_settlement_snapshot(db_path=path, **query)
        assert result["success"] is False, "unsupported comprehensive logic/schema is not a recognized authoritative result"
        assert result["error_code"] == "db_error"
        assert "totals" not in result and "cashflow_items" not in result
    with sqlite3.connect(path) as conn:
        assert list(conn.iterdump()) == before


@pytest.mark.parametrize("aggregation_version", ["unknown-version", "comprehensive-v1", ""])
def test_saved_reader_does_not_mislabel_partial_version_metadata_as_legacy(tmp_path, aggregation_version):
    path = dbfile(tmp_path)
    saved = ok(close_draft(path, new_draft(path)))
    with sqlite3.connect(path) as conn:
        conn.execute("UPDATE asset_snapshot_runs SET snapshot_schema_version=NULL,aggregation_version=? WHERE id=?", (aggregation_version, saved["run_id"]))
        conn.commit()
        before = list(conn.iterdump())
    for query in ({"run_id": saved["run_id"]}, {"month": "2026-09"}):
        result = service.get_monthly_settlement_snapshot(db_path=path, **query)
        assert result["success"] is False, "partial version metadata is not a genuine NULL-version legacy run"
        assert result["error_code"] == "db_error"
        assert "totals" not in result
    with sqlite3.connect(path) as conn:
        assert list(conn.iterdump()) == before
        conn.execute("UPDATE asset_snapshot_runs SET aggregation_version=NULL WHERE id=?", (saved["run_id"],))
    legacy = ok(service.get_monthly_settlement_snapshot(run_id=saved["run_id"], db_path=path))
    assert legacy["legacy"] and not legacy["is_comprehensive"]
    assert all(value is None for value in legacy["totals"].values())


@pytest.fixture
def schema_thirteen_conditional_close(tmp_path):
    """Execute immutable pre-comprehensive public APIs, without another checkout."""
    import importlib
    import importlib.abc
    import importlib.util
    from pathlib import Path
    import subprocess
    import sys

    revision = "f67c07adf01d1f17415bf7cf836f46997ca6f21c"
    root = Path(__file__).resolve().parents[1]
    namespace = "_legacy_schema_thirteen"
    names = subprocess.check_output(
        ["git", "ls-tree", "-r", "--name-only", revision, "jarca_ledger_mcp"],
        cwd=root, text=True,
    ).splitlines()
    blobs = {
        name: subprocess.check_output(["git", "show", f"{revision}:{name}"], cwd=root)
        for name in names if name.endswith(".py")
    }

    class BaselineModules(importlib.abc.MetaPathFinder, importlib.abc.Loader):
        def find_spec(self, fullname, path=None, target=None):
            if fullname != namespace and not fullname.startswith(namespace + "."):
                return None
            relative = fullname[len(namespace):].replace(".", "/")
            candidate = "jarca_ledger_mcp" + relative
            package = candidate + "/__init__.py"
            module = candidate + ".py"
            if package in blobs:
                return importlib.util.spec_from_loader(fullname, self, is_package=True)
            if module in blobs:
                return importlib.util.spec_from_loader(fullname, self)
            return None

        def create_module(self, spec):
            return None

        def exec_module(self, module):
            relative = module.__name__[len(namespace):].replace(".", "/")
            candidate = "jarca_ledger_mcp" + relative
            filename = candidate + ("/__init__.py" if hasattr(module, "__path__") else ".py")
            module.__file__ = f"git:{revision}:{filename}"
            exec(compile(blobs[filename], module.__file__, "exec"), module.__dict__)

    finder = BaselineModules()
    sys.meta_path.insert(0, finder)
    try:
        old = importlib.import_module(namespace + ".service")
        old_db = importlib.import_module(namespace + ".db")
        path = str(tmp_path / "baseline-schema-thirteen.sqlite3")
        old_db.init_db(path)
        missing = ok(old.create_ledger_account("Synthetic missing", "bank", liquidity_bucket="cash", db_path=path))["account"]["id"]
        carried = ok(old.create_ledger_account("Synthetic carry", "bank", liquidity_bucket="cash", db_path=path))["account"]["id"]
        ok(old.register_account_valuation(carried, "2026-09-27", 100, verification_status="verified", idempotency_key="baseline-carry", db_path=path))
        closed = ok(old.close_asset_snapshot(
            "2026-09-30", selection_policy="as_of", idempotency_key="baseline-close",
            allow_missing_exclusions=True, allow_as_of_carry_forward=True,
            approved_by="synthetic-reviewer", approval_reason="synthetic conditional approval",
            approval_reference="synthetic-ref", db_path=path,
        ))
        original = ok(old.get_asset_snapshot_report("2026-09-30", db_path=path))
        yield path, missing, carried, closed, original
    finally:
        sys.meta_path.remove(finder)
        for name in list(sys.modules):
            if name == namespace or name.startswith(namespace + "."):
                del sys.modules[name]


def test_schema_thirteen_saved_reader_retains_real_conditional_header_without_writes(schema_thirteen_conditional_close):
    path, missing, carried, closed, original = schema_thirteen_conditional_close
    init_db(path)
    assert ok(service.get_asset_snapshot_report("2026-09-30", db_path=path)) == original
    with sqlite3.connect(path) as conn:
        before = list(conn.iterdump())
    exact = ok(service.get_monthly_settlement_snapshot(run_id=closed["snapshot_run_id"], db_path=path))
    latest = ok(service.get_monthly_settlement_snapshot(month="2026-09", db_path=path))
    assert exact == latest
    assert exact["legacy"] and not exact["is_comprehensive"]
    assert all(value is None for value in exact["totals"].values())
    assert exact["close_status"] == "conditional"
    excluded = next(item for item in exact["excluded_items"] if item["source_id"] == missing)
    assert excluded["reason"]
    carry = next(item for item in exact["carried_items"] if item["source_id"] == carried)
    assert carry["valuation_date"] == "2026-09-27"
    assert exact["applied_evidence"]
    assert exact["approval"]["approved_by"] == "synthetic-reviewer"
    with sqlite3.connect(path) as conn:
        assert list(conn.iterdump()) == before


def test_saved_evidence_privacy_contract_filters_keys_not_allowlisted_scalar_values():
    from pathlib import Path
    from jarca_ledger_mcp.application.assets_service import _public_evidence
    value = {"candidate": {"name": "synthetic public scalar", "memo": "private", "source_attributes": {"reason": "synthetic reason", "private_key": "private"}}, "unrecognized": "private"}
    assert _public_evidence(value, strict=True) == {"candidate": {"name": "synthetic public scalar", "source_attributes": {"reason": "synthetic reason"}}}
    contract = (Path(__file__).resolve().parents[1] / "docs/interface-contract.md").read_text()
    assert "allowlisted scalar values are preserved" in contract
    assert "not a content-based secret redactor" in contract