From 2a6076e86be3fc039305f788cdbca0b38b0e37fe Mon Sep 17 00:00:00 2001 From: gitea-impl-agent Date: Tue, 6 Oct 2026 09:47:58 +0000 Subject: [PATCH] fix: preserve snapshot compatibility and loan fallback --- .../application/account_valuation_service.py | 10 ++-- .../application/assets_service.py | 23 ++++---- tests/test_issue_12_regressions.py | 52 +++++++++++++++++++ 3 files changed, 73 insertions(+), 12 deletions(-) diff --git a/jarca_ledger_mcp/application/account_valuation_service.py b/jarca_ledger_mcp/application/account_valuation_service.py index e8ac581..8d50eba 100644 --- a/jarca_ledger_mcp/application/account_valuation_service.py +++ b/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: diff --git a/jarca_ledger_mcp/application/assets_service.py b/jarca_ledger_mcp/application/assets_service.py index bbb4d8f..b5309eb 100644 --- a/jarca_ledger_mcp/application/assets_service.py +++ b/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() diff --git a/tests/test_issue_12_regressions.py b/tests/test_issue_12_regressions.py index df7861a..33d54ad 100644 --- a/tests/test_issue_12_regressions.py +++ b/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