From 921f130fb0c4ba5f0ad40dba712336a916ec1441 Mon Sep 17 00:00:00 2001 From: gitea-impl-agent Date: Tue, 6 Oct 2026 08:54:12 +0000 Subject: [PATCH] fix: harden valuation evidence selection --- .../application/account_valuation_service.py | 18 +++- .../application/assets_service.py | 57 +++++++++- .../application/reconciliation_service.py | 36 +++++-- tests/test_issue_12_regressions.py | 100 +++++++++++++++++- 4 files changed, 195 insertions(+), 16 deletions(-) diff --git a/jarca_ledger_mcp/application/account_valuation_service.py b/jarca_ledger_mcp/application/account_valuation_service.py index 88753b9..2f575fb 100644 --- a/jarca_ledger_mcp/application/account_valuation_service.py +++ b/jarca_ledger_mcp/application/account_valuation_service.py @@ -98,7 +98,23 @@ def record_account_valuation_candidate(conn: sqlite3.Connection, account_id: str basis = normalize_date(basis_date) existing = conn.execute("SELECT * FROM account_valuations WHERE account_id=? AND idempotency_key=?", (account_id, idempotency_key)).fetchone() if existing: - return ok(valuation=_valuation(existing), reused=True) + incoming = { + "basis_date": basis, + "amount": int(amount), + "currency": (currency or "KRW").upper(), + "valuation_kind": "balance", + "source_type": source_type, + "source_ref": source_ref or "", + "verification_status": "unverified", + "memo": memo, + } + if all(existing[key] == value for key, value in incoming.items()): + return ok(valuation=_valuation(existing), reused=True) + return fail( + "idempotency_conflict", + "idempotency_key already has a different candidate payload", + valuation=_valuation(existing), + ) valuation_id = new_id("valuation") conn.execute("""INSERT INTO account_valuations (id,account_id,basis_date,amount,currency,valuation_kind,source_type,source_ref,verification_status,recorded_at,memo,idempotency_key) diff --git a/jarca_ledger_mcp/application/assets_service.py b/jarca_ledger_mcp/application/assets_service.py index 39dd3b7..9ccccce 100644 --- a/jarca_ledger_mcp/application/assets_service.py +++ b/jarca_ledger_mcp/application/assets_service.py @@ -48,6 +48,13 @@ def _evidence_state_hash(item: dict[str, Any]) -> str: ).hexdigest() +def _valuation_row_state_hash(row: sqlite3.Row) -> str: + """Bind redacted candidate evidence to its complete immutable row.""" + return hashlib.sha256( + json.dumps(row_to_dict(row), sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() + ).hexdigest() + + def _account_valuation_items(conn: sqlite3.Connection, basis_date: str = "", selection_policy: str = "exact") -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]: rows = conn.execute("SELECT * FROM accounts WHERE active=1 ORDER BY name").fetchall() included: list[dict[str, Any]] = [] @@ -116,6 +123,13 @@ def _account_valuation_items(conn: sqlite3.Connection, basis_date: str = "", sel excluded.append({**base, "reason": "missing_verified_exact_valuation" if selection_policy == "exact" else "missing_verified_valuation", "unverified_candidates": [_public_candidate(v) for v in candidate_rows]}) continue valuation = selected + candidate_rows.extend( + v for v in leaves + if v["valuation_kind"] == selected_purpose + and v["verification_status"] != "verified" + and v["basis_date"] <= basis_date + and (v["basis_date"] == basis_date if selection_policy == "exact" else True) + ) if account_type == "loan" and ( selected_purpose not in {"loan", "loan_balance"} or not valuation["representative_liability_definition"] @@ -363,6 +377,8 @@ def _snapshot_preview_from_conn(conn: sqlite3.Connection, basis: str, selection_ if item.get("unverified_candidates"): warnings.append({"code": "unverified_candidate", "source_type": item.get("source_type", ""), "source_id": item.get("source_id", ""), "candidates": [{k: candidate.get(k) for k in ("id", "basis_date", "currency", "valuation_kind", "verification_status")} for candidate in item["unverified_candidates"]]}) for item in included: + if item.get("unverified_candidates"): + warnings.append({"code": "unverified_candidate", "blocking": False, "source_type": item.get("source_type", ""), "source_id": item.get("source_id", ""), "candidates": [{k: candidate.get(k) for k in ("id", "basis_date", "currency", "valuation_kind", "verification_status")} for candidate in item["unverified_candidates"]]}) item_date = str(item.get("valuation_date") or "")[:10] if item_date and item_date < basis: warnings.append({"code": "stale_valuation", "source_type": item.get("source_type", ""), "source_id": item.get("source_id", ""), "name": item.get("name", ""), "valuation_date": item_date, "basis_date": basis}) @@ -370,6 +386,23 @@ def _snapshot_preview_from_conn(conn: sqlite3.Connection, basis: str, selection_ warnings.append({"code": "invalid_liquidity_bucket", **item}) if existing_rows: warnings.append({"code": "existing_snapshot_rows", "basis_date": basis, "snapshot_row_count": int(existing_rows), "policy": "append_only_correction_required"}) + candidate_ids = sorted({ + candidate.get("id", "") + for warning in warnings + if warning.get("code") == "unverified_candidate" + for candidate in warning.get("candidates", []) + if candidate.get("id") + }) + candidate_state_hashes = [] + for candidate_id in candidate_ids: + candidate_row = conn.execute( + "SELECT * FROM account_valuations WHERE id=?", (candidate_id,), + ).fetchone() + if candidate_row is not None: + candidate_state_hashes.append({ + "id": candidate_id, + "state_hash": _valuation_row_state_hash(candidate_row), + }) return { "basis_date": basis, "source": "live_authority", @@ -378,6 +411,7 @@ def _snapshot_preview_from_conn(conn: sqlite3.Connection, basis: str, selection_ "existing_snapshot_row_count": int(existing_rows), "would_be_correction": bool(existing_rows), "warnings": warnings, + "_candidate_state_hashes": candidate_state_hashes, **summary, } @@ -392,8 +426,9 @@ def preview_asset_snapshot_close(basis_date: str = "", selection_policy: str = " with db(db_path) as conn: preview = _snapshot_preview_from_conn(conn, basis, selection_policy) preview["selection_policy"] = selection_policy - blocking = [w for w in preview["warnings"] if w.get("code", "").startswith(("missing_", "future_", "unknown_", "unverified", "ambiguous_", "stale_"))] + blocking = [w for w in preview["warnings"] if w.get("blocking", True) and w.get("code", "").startswith(("missing_", "future_", "unknown_", "unverified", "ambiguous_", "stale_"))] preview["close_status"] = "complete_exact" if selection_policy == "exact" and not blocking else "incomplete_draft" + preview.pop("_candidate_state_hashes", None) return ok(**preview) @@ -409,10 +444,10 @@ def close_asset_snapshot(basis_date: str = "", source_note: str = "", memo: str with db(db_path) as conn: conn.execute("BEGIN IMMEDIATE") preview = _snapshot_preview_from_conn(conn, basis, selection_policy) - missing_exclusions = [w for w in preview["warnings"] if w.get("code", "").startswith(("missing_", "future_", "unknown_", "unverified", "ambiguous_"))] + missing_exclusions = [w for w in preview["warnings"] if w.get("blocking", True) and w.get("code", "").startswith(("missing_", "future_", "unknown_", "unverified", "ambiguous_"))] carried_items = [item for item in preview["included_items"] if item.get("carried_forward")] conditional = bool(missing_exclusions or carried_items) - fingerprint_payload = {"basis_date": basis, "source_note": source_note, "memo": memo, "correction_reason": correction_reason, "selection_policy": selection_policy, "allow_missing_exclusions": allow_missing_exclusions, "allow_as_of_carry_forward": allow_as_of_carry_forward, "approved_by": approved_by, "approval_reason": approval_reason, "approval_reference": approval_reference, "included": [{**{k: i.get(k) for k in ("source_type", "source_id", "amount", "currency", "valuation_id", "valuation_date", "verification_status")}, "state_hash": _evidence_state_hash(i)} for i in preview["included_items"]], "excluded": [{**{k: i.get(k) for k in ("source_type", "source_id", "reason", "valuation_date", "verification_status")}, "state_hash": _evidence_state_hash(i)} for i in preview["excluded_items"]]} + fingerprint_payload = {"basis_date": basis, "source_note": source_note, "memo": memo, "correction_reason": correction_reason, "selection_policy": selection_policy, "allow_missing_exclusions": allow_missing_exclusions, "allow_as_of_carry_forward": allow_as_of_carry_forward, "approved_by": approved_by, "approval_reason": approval_reason, "approval_reference": approval_reference, "included": [{**{k: i.get(k) for k in ("source_type", "source_id", "amount", "currency", "valuation_id", "valuation_date", "verification_status")}, "state_hash": _evidence_state_hash(i)} for i in preview["included_items"]], "excluded": [{**{k: i.get(k) for k in ("source_type", "source_id", "reason", "valuation_date", "verification_status")}, "state_hash": _evidence_state_hash(i)} for i in preview["excluded_items"]], "warnings": [warning for warning in preview["warnings"] if warning.get("code") != "existing_snapshot_rows"], "candidate_state_hashes": preview["_candidate_state_hashes"]} fingerprint = hashlib.sha256(json.dumps(fingerprint_payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()).hexdigest() if missing_exclusions and (not allow_missing_exclusions or not approved_by.strip() or not approval_reason.strip() or not approval_reference.strip()): return fail("conditional_approval_required", "missing exclusions require explicit approval metadata and allow_missing_exclusions") @@ -469,6 +504,22 @@ def close_asset_snapshot(basis_date: str = "", source_note: str = "", memo: str summary = _summarize_valuation_items(included, []) applied_evidence = [{k: item.get(k) for k in ("source_type", "source_id", "amount", "currency", "valuation_id", "valuation_date", "verification_status")} for item in included] excluded_evidence = [{k: item.get(k) for k in ("source_type", "source_id", "reason", "currency")} for item in preview.get("excluded_items", [])] + excluded_evidence.extend( + { + "source_type": warning.get("source_type", ""), + "source_id": warning.get("source_id", ""), + "reason": "unverified_candidate", + "candidate": candidate, + "state_hash": next( + state["state_hash"] + for state in preview["_candidate_state_hashes"] + if state["id"] == candidate.get("id") + ), + } + for warning in preview["warnings"] + if warning.get("code") == "unverified_candidate" + for candidate in warning.get("candidates", []) + ) carried_evidence = [{k: item.get(k) for k in ("source_type", "source_id", "amount", "currency", "valuation_id", "valuation_date", "verification_status")} for item in included if item.get("carried_forward")] try: conn.execute("INSERT INTO asset_snapshot_runs(id,basis_date,close_status,selection_policy,idempotency_key,approved_by,approved_at,approval_reason,approval_reference,correction_of_run_id,correction_reason,applied_evidence,excluded_evidence,carried_evidence,request_fingerprint,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", (run_id, basis, "conditional" if conditional else "complete_exact", selection_policy, idempotency_key, approved_by, ts if approved_by else "", approval_reason, approval_reference, correction_of_run_id, correction_reason.strip(), json.dumps(applied_evidence, ensure_ascii=False, sort_keys=True), json.dumps(excluded_evidence, ensure_ascii=False, sort_keys=True), json.dumps(carried_evidence, ensure_ascii=False, sort_keys=True), fingerprint, ts)) diff --git a/jarca_ledger_mcp/application/reconciliation_service.py b/jarca_ledger_mcp/application/reconciliation_service.py index ed0b232..9d8c8f0 100644 --- a/jarca_ledger_mcp/application/reconciliation_service.py +++ b/jarca_ledger_mcp/application/reconciliation_service.py @@ -86,7 +86,9 @@ def get_balance_reconciliation_report(account_id: str = "", baseline_balance: in effective_baseline = baseline_balance_int if baseline_balance_int is not None else account["last_balance"] effective_at = baseline_at_norm or account["last_synced_at"] or now_iso() calc = _account_balance_row(conn, account["id"], effective_at) - valuation = conn.execute( + valuation = None + ambiguous_valuation = False + valuation_rows = conn.execute( """SELECT * FROM account_valuations v WHERE v.account_id=? AND v.verification_status='verified' AND v.basis_date<=? @@ -94,16 +96,25 @@ def get_balance_reconciliation_report(account_id: str = "", baseline_balance: in SELECT 1 FROM account_valuations child WHERE child.supersedes_id=v.id ) - ORDER BY CASE v.valuation_kind - WHEN 'total_value' THEN 0 - WHEN 'balance' THEN 1 - WHEN 'cash_balance' THEN 2 - ELSE 3 - END, - v.basis_date DESC, v.recorded_at DESC, v.id DESC - LIMIT 1""", + ORDER BY v.basis_date DESC, v.recorded_at DESC, v.id DESC""", (account["id"], effective_at[:10]), - ).fetchone() + ).fetchall() + purposes = ( + ("total_value", "balance", "cash_balance") + if account["type"] == "investment" + else ("cash_balance", "balance") + ) + for purpose in purposes: + scoped = [row for row in valuation_rows if row["valuation_kind"] == purpose] + if not scoped: + continue + latest_basis = max(row["basis_date"] for row in scoped) + latest = [row for row in scoped if row["basis_date"] == latest_basis] + if len(latest) != 1: + ambiguous_valuation = True + else: + valuation = latest[0] + break if effective_baseline is None: difference = None ratio = None @@ -112,7 +123,10 @@ def get_balance_reconciliation_report(account_id: str = "", baseline_balance: in difference = int(effective_baseline) - int(calc["calculated_balance"]) ratio = abs(difference) / abs(int(effective_baseline)) if int(effective_baseline) else None status = "OK" if abs(difference) <= tolerance_int else "REVIEW_NEEDED" - if account["type"] == "investment" and valuation is not None: + if account["type"] == "investment" and ambiguous_valuation: + status = "AMBIGUOUS_VALUATION" + difference = None + elif account["type"] == "investment" and valuation is not None: status = "VALUATION_MANAGED" difference = None items.append({ diff --git a/tests/test_issue_12_regressions.py b/tests/test_issue_12_regressions.py index 336ef15..b82d7b0 100644 --- a/tests/test_issue_12_regressions.py +++ b/tests/test_issue_12_regressions.py @@ -1,9 +1,13 @@ import sqlite3 import inspect +import json from concurrent.futures import ThreadPoolExecutor from jarca_ledger_mcp import service -from jarca_ledger_mcp.application.account_valuation_service import _public_candidate +from jarca_ledger_mcp.application.account_valuation_service import ( + _public_candidate, + record_account_valuation_candidate, +) from jarca_ledger_mcp.application.imports_service import _record_import_valuation_candidate from jarca_ledger_mcp.db import init_db @@ -158,6 +162,24 @@ def test_import_candidate_helper_reuses_deterministic_duplicate_on_same_connecti assert first is None and second is None and count == 1 +def test_candidate_idempotency_rejects_changed_payload(tmp_path): + path = str(tmp_path / "ledger.sqlite3") + account = service.create_ledger_account("계좌", db_path=path)["account"]["id"] + with sqlite3.connect(path) as conn: + conn.row_factory = sqlite3.Row + first = record_account_valuation_candidate( + conn, account, "2026-06-30", 100, "KRW", "import", + "source-a", "candidate-key", "memo-a", + ) + conflict = record_account_valuation_candidate( + conn, account, "2026-07-01", 200, "KRW", "import", + "source-b", "candidate-key", "memo-b", + ) + assert first["success"] is True + assert conflict["success"] is False + assert conflict["error_code"] == "idempotency_conflict" + + def test_valuation_currency_and_loan_kind_are_rejected(tmp_path): path = str(tmp_path / "ledger.sqlite3") bank = service.create_ledger_account("은행", currency="KRW", db_path=path)["account"]["id"] @@ -224,6 +246,82 @@ def test_snapshot_report_redacts_unverified_candidate_sensitive_fields(tmp_path) assert set(candidates[0]) == {"id", "basis_date", "currency", "valuation_kind", "verification_status"} +def test_preview_warns_about_unverified_candidate_when_verified_value_is_selected(tmp_path): + path = str(tmp_path / "ledger.sqlite3") + account = service.create_ledger_account( + "투자", type="investment", db_path=path, + )["account"]["id"] + service.register_account_valuation( + account, "2026-06-30", 100, valuation_kind="total_value", + verification_status="verified", idempotency_key="verified", db_path=path, + ) + service.register_account_valuation( + account, "2026-06-30", 999, valuation_kind="total_value", + verification_status="unverified", idempotency_key="candidate", db_path=path, + ) + preview = service.preview_asset_snapshot_close("2026-06-30", db_path=path) + warning = next(item for item in preview["warnings"] if item["code"] == "unverified_candidate") + assert warning["source_id"] == account + assert warning["candidates"][0]["verification_status"] == "unverified" + closed = service.close_asset_snapshot("2026-06-30", db_path=path) + assert closed["success"] is True + with sqlite3.connect(path) as conn: + evidence = json.loads(conn.execute( + "SELECT excluded_evidence FROM asset_snapshot_runs WHERE id=?", + (closed["snapshot_run_id"],), + ).fetchone()[0]) + assert evidence[0]["reason"] == "unverified_candidate" + assert evidence[0]["candidate"]["verification_status"] == "unverified" + + +def test_snapshot_idempotency_binds_redacted_candidate_state(tmp_path): + path = str(tmp_path / "ledger.sqlite3") + account = service.create_ledger_account( + "투자", type="investment", db_path=path, + )["account"]["id"] + service.register_account_valuation( + account, "2026-06-30", 100, valuation_kind="total_value", + verification_status="verified", idempotency_key="verified", db_path=path, + ) + candidate = service.register_account_valuation( + account, "2026-06-30", 999, valuation_kind="total_value", + verification_status="unverified", idempotency_key="candidate", db_path=path, + )["valuation"] + first = service.close_asset_snapshot( + "2026-06-30", idempotency_key="close-candidate", db_path=path, + ) + with sqlite3.connect(path) as conn: + conn.execute( + "UPDATE account_valuations SET amount=?, source_ref=?, memo=? WHERE id=?", + (1000, "changed-source", "changed-memo", candidate["id"]), + ) + retry = service.close_asset_snapshot( + "2026-06-30", idempotency_key="close-candidate", db_path=path, + ) + assert first["success"] is True + assert retry["success"] is False + assert retry["error_code"] == "idempotency_conflict" + + +def test_reconciliation_fails_closed_on_ambiguous_verified_leaf(tmp_path): + path = str(tmp_path / "ledger.sqlite3") + account = service.create_ledger_account( + "투자", type="investment", last_balance=100, + last_synced_at="2026-06-30", db_path=path, + )["account"]["id"] + for key, amount in (("v1", 100), ("v2", 200)): + service.register_account_valuation( + account, "2026-06-30", amount, valuation_kind="total_value", + verification_status="verified", idempotency_key=key, db_path=path, + ) + item = service.get_balance_reconciliation_report( + account_id=account, baseline_at="2026-06-30T00:00:00+09:00", db_path=path, + )["items"][0] + assert item["status"] == "AMBIGUOUS_VALUATION" + assert item["external_confirmed_value"] is None + assert item["valuation_id"] == "" + + def test_snapshot_rejects_bogus_selection_policy_and_manual_usd_is_excluded(tmp_path): path = str(tmp_path / "ledger.sqlite3") assert service.preview_asset_snapshot_close("2026-06-30", selection_policy="bogus", db_path=path)["success"] is False