fix: preserve snapshot compatibility and loan fallback

jarca_ledger_mcp/application/account_valuation_service.py

@@ -79,10 +79,12 @@ def _validate_common(account_id: str, basis_date: str, amount: int, currency: st
     if source not in _VALID_SOURCE_TYPES:
         return fail("validation_error", f"source_type must be one of {sorted(_VALID_SOURCE_TYPES)}")
     definition = representative_liability_definition.strip() if representative_liability_definition else None
-    if kind in {"loan", "loan_balance"} and definition not in _VALID_LOAN_DEFINITIONS:
+    if definition is not None and definition not in _VALID_LOAN_DEFINITIONS:
+        return fail("validation_error", "representative liability definition is invalid")
+    if kind in {"loan", "loan_balance"} and definition is None:
         return fail("validation_error", "loan valuation requires representative_liability_definition")
-    if kind not in {"loan", "loan_balance"} and definition is not None:
-        return fail("validation_error", "non-loan valuation cannot define representative liability")
+    if kind not in {"loan", "loan_balance", "balance"} and definition is not None:
+        return fail("validation_error", "valuation kind cannot define representative liability")
     if amount_int is None:
         return fail("validation_error", "amount must be an integer")
     return basis, amount_int, currency_norm, status, source, definition
@@ -182,6 +184,8 @@ def register_account_valuation(
             return fail("validation_error", "valuation currency must match account currency")
         if (valuation_kind or "balance").strip().lower() in {"loan", "loan_balance"} and str(account["type"]).lower() != "loan":
             return fail("validation_error", "loan valuation kind requires a loan account")
+        if definition is not None and str(account["type"]).lower() != "loan":
+            return fail("validation_error", "representative liability definition requires a loan account")
         if idempotency_key:
             existing = conn.execute("SELECT * FROM account_valuations WHERE account_id=? AND idempotency_key=?", (account_id, idempotency_key)).fetchone()
             if existing:

jarca_ledger_mcp/application/assets_service.py

@@ -134,10 +134,7 @@ def _account_valuation_items(conn: sqlite3.Connection, basis_date: str = "", sel
                 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"]
-            ):
+            if account_type == "loan" and not valuation["representative_liability_definition"]:
                 excluded.append({
                     **base,
                     "reason": "missing_representative_liability_definition",
@@ -420,7 +417,7 @@ def _snapshot_preview_from_conn(conn: sqlite3.Connection, basis: str, selection_
     }
 
 
-def preview_asset_snapshot_close(basis_date: str = "", selection_policy: str = "exact", db_path: str | None = None) -> dict[str, Any]:
+def preview_asset_snapshot_close(basis_date: str = "", db_path: str | None = None, selection_policy: str = "exact") -> dict[str, Any]:
     if selection_policy not in {"exact", "as_of"}:
         return fail("validation_error", "selection_policy must be exact or as_of")
     try:
@@ -436,7 +433,7 @@ def preview_asset_snapshot_close(basis_date: str = "", selection_policy: str = "
     return ok(**preview)
 
 
-def close_asset_snapshot(basis_date: str = "", source_note: str = "", memo: str = "", correction_reason: str = "", selection_policy: str = "exact", idempotency_key: str = "", allow_missing_exclusions: bool = False, allow_as_of_carry_forward: bool = False, approved_by: str = "", approval_reason: str = "", approval_reference: str = "", db_path: str | None = None) -> dict[str, Any]:
+def close_asset_snapshot(basis_date: str = "", source_note: str = "", memo: str = "", correction_reason: str = "", db_path: str | None = None, selection_policy: str = "exact", idempotency_key: str = "", allow_missing_exclusions: bool = False, allow_as_of_carry_forward: bool = False, approved_by: str = "", approval_reason: str = "", approval_reference: str = "") -> dict[str, Any]:
     if selection_policy not in {"exact", "as_of"}:
         return fail("validation_error", "selection_policy must be exact or as_of")
     try:
@@ -448,6 +445,15 @@ 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)
+        latest_rows, latest_meta = _snapshot_rows_for_latest_run(conn, basis)
+        legacy_snapshot_rows = bool(latest_rows and not latest_meta.get("snapshot_run_id"))
+        if legacy_snapshot_rows and not correction_reason.strip():
+            return fail(
+                "legacy_snapshot_correction_required",
+                "legacy snapshot rows require an explicit correction reason before close",
+                basis_date=basis,
+                existing_snapshot_row_count=preview["existing_snapshot_row_count"],
+            )
         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)
@@ -464,10 +470,9 @@ def close_asset_snapshot(basis_date: str = "", source_note: str = "", memo: str
                     return ok(reused=True, snapshot_run_id=existing_run["id"], close_status=existing_run["close_status"], basis_date=basis)
                 return fail("idempotency_conflict", "snapshot close idempotency payload differs")
         included = preview["included_items"]
-        latest_rows, latest_meta = _snapshot_rows_for_latest_run(conn, basis)
         # A conditional close may legitimately contain zero snapshot rows.
         # The run record, not row count, is the authoritative close marker.
-        is_correction = bool(latest_meta.get("snapshot_run_id"))
+        is_correction = bool(latest_meta.get("snapshot_run_id") or legacy_snapshot_rows)
         if is_correction and not correction_reason.strip():
             return fail("validation_error", "correction_reason is required when closing a basis_date that already has a snapshot run", basis_date=basis, existing_snapshot_row_count=preview["existing_snapshot_row_count"])
         correction_of_run_id = latest_meta.get("snapshot_run_id", "")
@@ -571,7 +576,7 @@ def create_asset_snapshot(basis_date: str, source_type: str, source_id: str, nam
     return ok(asset_snapshot=row_to_dict(row))
 
 
-def get_asset_snapshot_report(basis_date: str = "", selection_policy: str = "exact", db_path: str | None = None) -> dict[str, Any]:
+def get_asset_snapshot_report(basis_date: str = "", db_path: str | None = None, selection_policy: str = "exact") -> dict[str, Any]:
     if selection_policy not in {"exact", "as_of"}:
         return fail("validation_error", "selection_policy must be exact or as_of")
     basis = normalize_date(basis_date, default_today=True) if basis_date else current_kst_date().isoformat()

tests/test_issue_12_regressions.py

@@ -453,6 +453,58 @@ def test_legacy_snapshot_rows_do_not_override_live_authority_or_claim_closed(tmp
     assert report["legacy_snapshot_row_count"] == 1
 
 
+def test_snapshot_public_api_preserves_legacy_db_path_position(tmp_path):
+    path = str(tmp_path / "ledger.sqlite3")
+    preview = service.preview_asset_snapshot_close("2026-06-30", path)
+    closed = service.close_asset_snapshot("2026-06-30", "", "", "", path)
+    report = service.get_asset_snapshot_report("2026-06-30", path)
+    assert preview["success"] is True
+    assert closed["success"] is True
+    assert report["success"] is True
+    assert report["is_closed_snapshot"] is True
+
+
+def test_legacy_snapshot_requires_explicit_correction_before_new_close(tmp_path):
+    path = str(tmp_path / "ledger.sqlite3")
+    service.create_asset_snapshot(
+        "2026-06-30", "account", "legacy", "legacy", 999, "cash", db_path=path,
+    )
+    close = service.close_asset_snapshot("2026-06-30", db_path=path)
+    report = service.get_asset_snapshot_report("2026-06-30", db_path=path)
+    assert close["success"] is False
+    assert close["error_code"] == "legacy_snapshot_correction_required"
+    assert report["source"] == "legacy_asset_snapshots"
+    assert report["net_worth"] == 999
+    assert report["is_closed_snapshot"] is False
+
+
+def test_loan_balance_fallback_with_representative_definition_is_liability(tmp_path):
+    path = str(tmp_path / "ledger.sqlite3")
+    account = service.create_ledger_account("대출", type="loan", db_path=path)["account"]["id"]
+    recorded = service.register_account_valuation(
+        account, "2026-06-30", 100, valuation_kind="balance",
+        representative_liability_definition="principal_only",
+        verification_status="verified", db_path=path,
+    )
+    report = service.get_asset_snapshot_report("2026-06-30", db_path=path)
+    assert recorded["success"] is True
+    assert report["total_assets"] == 0
+    assert report["total_liabilities"] == 100
+    assert report["net_worth"] == -100
+
+
+def test_loan_balance_fallback_rejects_unknown_representative_definition(tmp_path):
+    path = str(tmp_path / "ledger.sqlite3")
+    account = service.create_ledger_account("대출", type="loan", db_path=path)["account"]["id"]
+    recorded = service.register_account_valuation(
+        account, "2026-06-30", 100, valuation_kind="balance",
+        representative_liability_definition="outstanding_principal",
+        verification_status="verified", db_path=path,
+    )
+    assert recorded["success"] is False
+    assert recorded["error_code"] == "validation_error"
+
+
 def test_account_write_rolls_back_when_conn_level_candidate_write_fails(tmp_path, monkeypatch):
     path = str(tmp_path / "ledger.sqlite3")
     import jarca_ledger_mcp.application.accounts_write_services as writes