fix: harden valuation evidence selection

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)

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))

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
+                     ORDER BY v.basis_date DESC, v.recorded_at DESC, v.id DESC""",
-                                  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""",
                 (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({

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