fix: harden valuation concurrency and currency scope

jarca_ledger_mcp/application/account_valuation_service.py

@@ -174,6 +174,7 @@ def register_account_valuation(
     ts = now_iso()
     values = (amount_int, currency_norm, (valuation_kind or "balance").strip().lower(), source_norm, source_ref or "", status, memo or "", as_of_at, timezone, optional_ints["cost_basis"], optional_ints["net_contributed"], reported_principal_kind, reported_principal_label, optional_ints["profit"], return_rate, return_basis, optional_ints["principal"], optional_ints["interest"], optional_ints["fee"], definition)
     with db(db_path) as conn:
+        conn.execute("BEGIN IMMEDIATE")
         account = conn.execute("SELECT * FROM accounts WHERE id=?", (account_id,)).fetchone()
         if account is None:
             return fail("not_found", "account not found")
@@ -196,10 +197,10 @@ def register_account_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,supersedes_id,correction_reason,as_of_at,timezone,cost_basis,net_contributed,reported_principal_kind,reported_principal_label,profit,return_rate,return_basis,principal,interest,fee,representative_liability_definition)
             VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", (valuation_id, account_id, basis, *values[:6], ts, values[6], idempotency_key or "", supersedes_id, correction_reason, *values[7:]))
-        except sqlite3.IntegrityError as exc:
+        except sqlite3.IntegrityError:
             if supersedes_id:
                 return fail("correction_conflict", "valuation already has a correction")
-            return fail("idempotency_conflict", str(exc))
+            return fail("idempotency_conflict", "idempotency_key already has a different payload")
         except Exception as exc:
             return fail("validation_error", str(exc))
         row = conn.execute("SELECT * FROM account_valuations WHERE id=?", (valuation_id,)).fetchone()
@@ -237,7 +238,7 @@ def select_account_valuation(account_id: str, basis_date: str, as_of: bool = Fal
         latest_verified_date = max((x["basis_date"] for x in verified), default="")
         candidate = [
             r for r in leaves
-            if r["verification_status"] != "verified"
+            if r["verification_status"] == "unverified"
             and r["basis_date"] <= basis
             and (r["basis_date"] == basis if not as_of else (not latest_verified_date or r["basis_date"] > latest_verified_date))
         ]

jarca_ledger_mcp/application/assets_service.py

@@ -78,12 +78,16 @@ def _account_valuation_items(conn: sqlite3.Connection, basis_date: str = "", sel
             "authority_source": "accounts.last_balance",
             "name": account.get("name", ""),
             "type": account.get("type", ""),
+            "currency": (account.get("currency") or "KRW").upper(),
             "liquidity_bucket": bucket,
             "valuation_date": account.get("last_synced_at", ""),
             "last_synced_at": account.get("last_synced_at", ""),
             "source_note": account.get("memo", ""),
             "memo": account.get("memo", ""),
         }
+        if base["currency"] != "KRW":
+            excluded.append({**base, "reason": "non_krw_without_fx"})
+            continue
         balance = account.get("last_balance")
         if basis_date:
             valuation_rows = conn.execute(
@@ -118,7 +122,7 @@ def _account_valuation_items(conn: sqlite3.Connection, basis_date: str = "", sel
                 if verified:
                     selected, selected_purpose = verified[0], purpose
                     break
-                candidate_rows.extend(v for v in scoped if v["verification_status"] != "verified" and v["basis_date"] <= basis_date and (v["basis_date"] == basis_date if selection_policy == "exact" else True))
+                candidate_rows.extend(v for v in scoped if v["verification_status"] == "unverified" and v["basis_date"] <= basis_date and (v["basis_date"] == basis_date if selection_policy == "exact" else True))
             if selected is None:
                 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
@@ -126,7 +130,7 @@ def _account_valuation_items(conn: sqlite3.Connection, basis_date: str = "", sel
             candidate_rows.extend(
                 v for v in leaves
                 if v["valuation_kind"] == selected_purpose
-                and v["verification_status"] != "verified"
+                and v["verification_status"] == "unverified"
                 and v["basis_date"] <= basis_date
                 and (v["basis_date"] == basis_date if selection_policy == "exact" else True)
             )
@@ -523,9 +527,9 @@ def close_asset_snapshot(basis_date: str = "", source_note: str = "", memo: str
         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))
-        except sqlite3.IntegrityError as exc:
+        except sqlite3.IntegrityError:
             conn.rollback()
-            return fail("snapshot_conflict", "another snapshot close already owns this basis date", basis_date=basis, detail=str(exc))
+            return fail("snapshot_conflict", "another snapshot close already owns this basis date", basis_date=basis)
     return ok(
         basis_date=basis,
         source="snapshot",

tests/test_issue_12_regressions.py

@@ -4,6 +4,7 @@ import json
 from concurrent.futures import ThreadPoolExecutor
 
 from jarca_ledger_mcp import service
+from jarca_ledger_mcp.application import assets_service
 from jarca_ledger_mcp.application.account_valuation_service import (
     _public_candidate,
     record_account_valuation_candidate,
@@ -147,10 +148,31 @@ def test_snapshot_close_idempotency_fingerprint_and_concurrent_first_close(tmp_p
     with ThreadPoolExecutor(max_workers=2) as pool:
         results = list(pool.map(lambda _: service.close_asset_snapshot("2026-06-30", db_path=other_path), range(2)))
     assert sum(bool(result.get("success")) for result in results) == 1
+    conflict = next(result for result in results if not result.get("success"))
+    assert "detail" not in conflict
     with sqlite3.connect(other_path) as conn:
         assert conn.execute("SELECT COUNT(*) FROM asset_snapshot_runs WHERE basis_date='2026-06-30'").fetchone()[0] == 1
 
 
+def test_snapshot_integrity_conflict_does_not_expose_sqlite_detail(tmp_path, monkeypatch):
+    path = str(tmp_path / "ledger.sqlite3")
+    service.create_manual_asset("현금", 100, "2026-06-30", "cash", currency="KRW", db_path=path)
+    first = service.close_asset_snapshot("2026-06-30", db_path=path)
+    real_new_id = assets_service.new_id
+
+    def colliding_run_id(prefix):
+        return first["snapshot_run_id"] if prefix == "snaprun" else real_new_id(prefix)
+
+    monkeypatch.setattr(assets_service, "new_id", colliding_run_id)
+    conflict = service.close_asset_snapshot(
+        "2026-06-30", correction_reason="retry collision", db_path=path,
+    )
+    assert conflict["success"] is False
+    assert conflict["error_code"] == "snapshot_conflict"
+    assert conflict["error"] == "another snapshot close already owns this basis date"
+    assert "detail" not in conflict
+
+
 def test_import_candidate_helper_reuses_deterministic_duplicate_on_same_connection(tmp_path):
     path = str(tmp_path / "ledger.sqlite3")
     account = service.create_ledger_account("계좌", db_path=path)["account"]["id"]
@@ -200,6 +222,42 @@ def test_correction_idempotency_compares_effective_preserved_payload(tmp_path):
     assert retry_with_omitted_memo["error_code"] == "idempotency_conflict"
 
 
+def test_concurrent_valuation_idempotency_reuses_same_payload(tmp_path):
+    path = str(tmp_path / "ledger.sqlite3")
+    account = service.create_ledger_account("계좌", db_path=path)["account"]["id"]
+
+    def register(_):
+        return service.register_account_valuation(
+            account, "2026-06-30", 100, verification_status="verified",
+            idempotency_key="same", db_path=path,
+        )
+
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        results = list(pool.map(register, range(2)))
+    assert all(result["success"] for result in results)
+    assert len({result["valuation"]["id"] for result in results}) == 1
+    assert sorted(result["reused"] for result in results) == [False, True]
+
+
+def test_concurrent_correction_idempotency_reuses_same_payload(tmp_path):
+    path = str(tmp_path / "ledger.sqlite3")
+    account = service.create_ledger_account("계좌", db_path=path)["account"]["id"]
+    original = service.register_account_valuation(
+        account, "2026-06-30", 100, verification_status="verified", db_path=path,
+    )["valuation"]
+
+    def correct(_):
+        return service.correct_account_valuation(
+            original["id"], 200, "정정", idempotency_key="same-correction", db_path=path,
+        )
+
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        results = list(pool.map(correct, range(2)))
+    assert all(result["success"] for result in results)
+    assert len({result["valuation"]["id"] for result in results}) == 1
+    assert sorted(result["reused"] for result in results) == [False, True]
+
+
 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"]
@@ -351,6 +409,34 @@ def test_snapshot_rejects_bogus_selection_policy_and_manual_usd_is_excluded(tmp_
     assert any(item.get("reason") == "non_krw_without_fx" for item in report["excluded_items"])
 
 
+def test_non_krw_account_valuation_is_excluded_without_fx_conversion(tmp_path):
+    path = str(tmp_path / "ledger.sqlite3")
+    account = service.create_ledger_account("달러 계좌", currency="USD", db_path=path)["account"]["id"]
+    service.register_account_valuation(
+        account, "2026-06-30", 100, currency="USD",
+        verification_status="verified", db_path=path,
+    )
+    report = service.get_asset_snapshot_report("2026-06-30", db_path=path)
+    assert report["total_assets"] == 0
+    assert any(item.get("reason") == "non_krw_without_fx" for item in report["excluded_items"])
+
+
+def test_rejected_valuation_is_not_an_unverified_candidate(tmp_path):
+    path = str(tmp_path / "ledger.sqlite3")
+    account = service.create_ledger_account("계좌", db_path=path)["account"]["id"]
+    service.register_account_valuation(
+        account, "2026-06-30", 100, verification_status="verified", db_path=path,
+    )
+    service.register_account_valuation(
+        account, "2026-06-30", 999, verification_status="rejected", db_path=path,
+    )
+    selected = service.select_account_valuation(account, "2026-06-30", db_path=path)
+    preview = service.preview_asset_snapshot_close("2026-06-30", db_path=path)
+    assert selected.get("warning") is None
+    assert selected.get("unverified_candidates", []) == []
+    assert not any(warning.get("code") == "unverified_candidate" for warning in preview["warnings"])
+
+
 def test_legacy_snapshot_rows_do_not_override_live_authority_or_claim_closed(tmp_path):
     path = str(tmp_path / "ledger.sqlite3")
     service.create_asset_snapshot(