diff --git a/docs/data-model.md b/docs/data-model.md index 70f26ce..7234106 100644 --- a/docs/data-model.md +++ b/docs/data-model.md @@ -60,6 +60,13 @@ | `return_rate` | REAL | 선택 | NULL | finite number | ratio | 아니오 | | `representative_liability_definition` | TEXT | loan 필수 | NULL | principal_only/principal_interest_fees | enum | 아니오 | +#### `select_account_valuation` 선택 규칙과 기존 인덱스 매핑 + +- `select_account_valuation`은 `account_id is mandatory`인 단일 평가 선택 query다. `currency and valuation_kind/purpose are optional filters`이며, `basis_date exact match or <= upper bound` 조건을 지원한다. +- 선택 결과는 `verified leaf valuations only`다. 즉 superseded parent가 아닌 leaf만 권위로 사용하며, `future-dated and unverified valuations are excluded`한다. +- `idx_account_valuations_account_date supports account/date selection`; `idx_account_valuations_status_date supports status/date selection`. `neither index covers account + currency + purpose`이므로 해당 조합의 전체 선택 조건을 기존 인덱스가 대신 보장한다고 해석하지 않는다. +- `idempotency and single_correction indexes enforce write integrity, not selection`이다. 이 인덱스들은 재기록 방지와 correction 단일 child 불변조건을 위한 것이며 `select_account_valuation`의 후보 검색 인덱스가 아니다. + ### manual-assets — 수동 자산/부채 - 책임: transaction이 아닌 평가 source diff --git a/jarca_ledger_mcp/application/accounts_write_services.py b/jarca_ledger_mcp/application/accounts_write_services.py index 4cb38a5..fd8399f 100644 --- a/jarca_ledger_mcp/application/accounts_write_services.py +++ b/jarca_ledger_mcp/application/accounts_write_services.py @@ -42,7 +42,7 @@ def update_ledger_account(account_id: str, db_path: str | None = None, **fields: if current is None: return fail("not_found", "account not found") balance_update = "last_balance" in fields and fields.get("last_balance") is not None basis_update = "last_balance_basis_date" in fields and fields.get("last_balance_basis_date") is not None - if basis_update and not balance_update: + if basis_update and (not balance_update or fields.get("last_balance") == ""): return fail("validation_error", "last_balance_basis_date requires last_balance") normalized_basis = None if basis_update: diff --git a/jarca_ledger_mcp/application/assets_service.py b/jarca_ledger_mcp/application/assets_service.py index f2ecb7b..f3bb109 100644 --- a/jarca_ledger_mcp/application/assets_service.py +++ b/jarca_ledger_mcp/application/assets_service.py @@ -259,6 +259,16 @@ def _manual_asset_valuation_items(conn: sqlite3.Connection, active_only: bool = return included, excluded +def _persisted_evidence_array(value: Any) -> list[dict[str, Any]] | None: + try: + decoded = json.loads(value or "[]") + except (TypeError, json.JSONDecodeError): + return None + if not isinstance(decoded, list) or any(not isinstance(item, dict) for item in decoded): + return None + return decoded + + def _snapshot_rows_for_run(conn: sqlite3.Connection, run_id: str) -> tuple[list[sqlite3.Row], dict[str, Any]]: """Load one immutable close run and only the rows persisted by that run.""" run = conn.execute( @@ -287,6 +297,11 @@ def _snapshot_rows_for_run(conn: sqlite3.Connection, run_id: str) -> tuple[list[ )""", (basis_date, run["_run_rowid"]), ).fetchone()[0] + evidence = { + field: _persisted_evidence_array(run[field]) + for field in ("applied_evidence", "excluded_evidence", "carried_evidence") + } + integrity_error = any(value is None for value in evidence.values()) metadata = { "snapshot_run_id": run["id"], "is_correction": bool(run["correction_of_run_id"]), @@ -298,9 +313,10 @@ def _snapshot_rows_for_run(conn: sqlite3.Connection, run_id: str) -> tuple[list[ "approved_at": run["approved_at"], "approval_reason": run["approval_reason"], "approval_reference": run["approval_reference"], - "applied_evidence": json.loads(run["applied_evidence"] or "[]"), - "excluded_evidence": json.loads(run["excluded_evidence"] or "[]"), - "carried_evidence": json.loads(run["carried_evidence"] or "[]"), + "applied_evidence": evidence["applied_evidence"] or [], + "excluded_evidence": evidence["excluded_evidence"] or [], + "carried_evidence": evidence["carried_evidence"] or [], + "evidence_integrity_error": integrity_error, "closed_at": run["created_at"] or "", "previous_snapshot_row_count": previous_rows, "total_snapshot_rows_for_basis_date": total_rows, @@ -372,7 +388,10 @@ def _closed_run_metadata(metadata: dict[str, Any]) -> dict[str, Any]: for item in exclusions: if item.get("candidate"): key = (item.get("source_type", ""), item.get("source_id", "")) - warning = candidate_warnings.setdefault(key, {"code": "unverified_candidate", "source_type": key[0], "source_id": key[1], "candidates": []}) + warning_shape = {"code": "unverified_candidate", "source_type": key[0], "source_id": key[1], "candidates": []} + if item.get("blocking") is False: + warning_shape["blocking"] = False + warning = candidate_warnings.setdefault(key, warning_shape) warning["candidates"].append(item["candidate"]) else: warnings.append({"code": item.get("reason", "excluded_item"), "source_type": item.get("source_type", ""), "source_id": item.get("source_id", ""), "currency": item.get("currency", "KRW")}) @@ -667,6 +686,8 @@ def close_asset_snapshot(basis_date: str = "", source_note: str = "", memo: str if preview_fingerprint and preview_fingerprint != _preview_fingerprint(preview, basis, selection_policy): return fail("authority_drift", "asset authority changed since preview", basis_date=basis) latest_rows, latest_meta = _snapshot_rows_for_latest_run(conn, basis) + if latest_meta.get("evidence_integrity_error"): + return fail("data_integrity_error", "snapshot evidence is malformed", snapshot_run_id=latest_meta.get("snapshot_run_id", "")) 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( @@ -734,6 +755,8 @@ def close_asset_snapshot(basis_date: str = "", source_note: str = "", memo: str or immutable_replay_run ): replay_rows, replay_meta = _snapshot_rows_for_run(conn, existing_run["id"]) + if replay_meta.get("evidence_integrity_error"): + return fail("data_integrity_error", "snapshot evidence is malformed", snapshot_run_id=existing_run["id"]) replay_items = [_public_evidence(item) for item in replay_meta.get("applied_evidence", [])] replay_warnings = _closed_warnings(replay_meta, basis, replay_rows) replay_public = _closed_run_metadata(replay_meta) @@ -816,7 +839,7 @@ def close_asset_snapshot(basis_date: str = "", source_note: str = "", memo: str ), ) snapshot_ids.append(snapshot_id) - applied_evidence = [{k: item.get(k) for k in ("id", "source_type", "source_id", "authority_source", "name", "type", "liquidity_bucket", "active", "amount", "valuation_amount", "last_balance", "last_synced_at", "currency", "valuation_id", "valuation_kind", "valuation_date", "age_days", "carried_forward", "verification_status", "evidence_kind") if item.get(k) is not None} for item in included] + applied_evidence = [{k: item.get(k) for k in ("id", "source_type", "source_id", "authority_source", "name", "type", "liquidity_bucket", "active", "amount", "valuation_amount", "last_balance", "last_synced_at", "currency", "valuation_id", "valuation_kind", "valuation_date", "age_days", "carried_forward", "verification_status", "evidence_kind", "unverified_candidates") if item.get(k) is not None} 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( { @@ -824,6 +847,7 @@ def close_asset_snapshot(basis_date: str = "", source_note: str = "", memo: str "source_id": warning.get("source_id", ""), "reason": "unverified_candidate", "candidate": candidate, + "blocking": warning.get("blocking", True), "state_hash": next( state["state_hash"] for state in preview["_candidate_state_hashes"] @@ -922,6 +946,8 @@ def get_asset_snapshot_report(basis_date: str = "", db_path: str | None = None, return fail("validation_error", f"invalid date: {basis_date}") with read_db(db_path) as conn: rows, close_metadata = _snapshot_rows_for_latest_run(conn, basis) + if close_metadata.get("evidence_integrity_error"): + return fail("data_integrity_error", "snapshot evidence is malformed", snapshot_run_id=close_metadata.get("snapshot_run_id", "")) has_close_run = bool(close_metadata.get("snapshot_run_id")) snapshots = _snapshot_items(conn, basis) if rows else [] legacy_snapshots = snapshots if rows and not has_close_run else [] @@ -1032,8 +1058,10 @@ def get_net_worth_timeseries(start_date: str = "", end_date: str = "", months: i points = [] for row in basis_rows: basis = row["basis_date"] - items = _snapshot_items(conn, basis) _, metadata = _snapshot_rows_for_latest_run(conn, basis) + if metadata.get("evidence_integrity_error"): + return fail("data_integrity_error", "snapshot evidence is malformed", snapshot_run_id=metadata.get("snapshot_run_id", "")) + items = _snapshot_items(conn, basis) run_public = _closed_run_metadata(metadata) run_public["warnings"] = _closed_warnings(metadata, basis, conn.execute("SELECT rowid AS _rowid, * FROM asset_snapshots WHERE basis_date=? ORDER BY created_at ASC, name ASC", (basis,)).fetchall()) run_provenance = _evidence_provenance(metadata.get("applied_evidence", []), basis) diff --git a/tests/test_asset_snapshot_idempotency.py b/tests/test_asset_snapshot_idempotency.py index 17be3e1..e9958a7 100644 --- a/tests/test_asset_snapshot_idempotency.py +++ b/tests/test_asset_snapshot_idempotency.py @@ -1,10 +1,133 @@ from __future__ import annotations +import json import sqlite3 from jarca_ledger_mcp import service +def test_replay_preserves_redacted_unverified_candidates_after_correction(tmp_path): + db_path = str(tmp_path / "ledger.sqlite3") + account = service.create_ledger_account("투자 계좌", type="investment", db_path=db_path) + account_id = account["account"]["id"] + verified = service.register_account_valuation( + account_id, "2026-06-30", 100, verification_status="verified", + idempotency_key="valuation-verified", db_path=db_path, + ) + candidate = service.register_account_valuation( + account_id, "2026-06-30", 110, verification_status="unverified", + idempotency_key="valuation-candidate", db_path=db_path, + ) + + first = service.close_asset_snapshot( + "2026-06-30", idempotency_key="candidate-close-a", db_path=db_path, + ) + assert first["success"] is True + assert first["included_items"][0]["unverified_candidates"] == [{ + "id": candidate["valuation"]["id"], + "basis_date": "2026-06-30", + "currency": "KRW", + "valuation_kind": "balance", + "verification_status": "unverified", + }] + assert {warning["blocking"] for warning in first["warnings"] if warning["code"] == "unverified_candidate"} == {False} + assert first["included_items"][0]["valuation_id"] == verified["valuation"]["id"] + with sqlite3.connect(db_path) as conn: + persisted = conn.execute("SELECT applied_evidence, excluded_evidence FROM asset_snapshot_runs WHERE id=?", (first["snapshot_run_id"],)).fetchone() + applied_evidence, excluded_evidence = map(json.loads, persisted) + assert applied_evidence[0]["unverified_candidates"] == first["included_items"][0]["unverified_candidates"] + candidate_evidence = [item for item in excluded_evidence if item.get("reason") == "unverified_candidate"] + assert candidate_evidence[0]["candidate"] == first["included_items"][0]["unverified_candidates"][0] + assert candidate_evidence[0]["blocking"] is False + assert candidate_evidence[0]["state_hash"] + + corrected = service.close_asset_snapshot( + "2026-06-30", correction_reason="preserve evidence", idempotency_key="candidate-close-b", + db_path=db_path, + ) + assert corrected["success"] is True + + with sqlite3.connect(db_path) as conn: + before_replay = conn.execute("SELECT COUNT(*) FROM asset_snapshots").fetchone()[0] + before_runs = conn.execute("SELECT COUNT(*) FROM asset_snapshot_runs").fetchone()[0] + replay = service.close_asset_snapshot( + "2026-06-30", idempotency_key="candidate-close-a", db_path=db_path, + ) + with sqlite3.connect(db_path) as conn: + after_replay = conn.execute("SELECT COUNT(*) FROM asset_snapshots").fetchone()[0] + after_runs = conn.execute("SELECT COUNT(*) FROM asset_snapshot_runs").fetchone()[0] + + assert replay == first + assert replay["included_items"] == first["included_items"] + assert replay["warnings"] == first["warnings"] + assert after_replay == before_replay + assert after_runs == before_runs + + +def test_replay_preserves_blocking_candidate_warning_shape(tmp_path): + db_path = str(tmp_path / "ledger.sqlite3") + account_id = service.create_ledger_account( + "미확인 계좌", type="investment", db_path=db_path, + )["account"]["id"] + service.register_account_valuation( + account_id, "2026-06-30", 110, verification_status="unverified", + idempotency_key="only-candidate", db_path=db_path, + ) + + first = service.close_asset_snapshot( + "2026-06-30", idempotency_key="blocking-candidate-close", + allow_missing_exclusions=True, approved_by="reviewer", + approval_reason="exclude unverified evidence", + approval_reference="synthetic-approval", db_path=db_path, + ) + assert first["success"] is True + warning = next(item for item in first["warnings"] if item["code"] == "unverified_candidate") + assert "blocking" not in warning + + replay = service.close_asset_snapshot( + "2026-06-30", idempotency_key="blocking-candidate-close", + allow_missing_exclusions=True, approved_by="reviewer", + approval_reason="exclude unverified evidence", + approval_reference="synthetic-approval", db_path=db_path, + ) + + assert replay == first + replay_warning = next(item for item in replay["warnings"] if item["code"] == "unverified_candidate") + assert "blocking" not in replay_warning + + +def test_malformed_persisted_evidence_fails_closed_without_snapshot_writes(tmp_path): + db_path = str(tmp_path / "ledger.sqlite3") + service.create_manual_asset("현금", 100, "2026-06-30", "cash", db_path=db_path) + first = service.close_asset_snapshot( + "2026-06-30", idempotency_key="malformed-evidence-close", db_path=db_path, + ) + assert first["success"] is True + + with sqlite3.connect(db_path) as conn: + conn.execute( + "UPDATE asset_snapshot_runs SET excluded_evidence=? WHERE id=?", + ("{malformed", first["snapshot_run_id"]), + ) + before_rows = conn.execute("SELECT COUNT(*) FROM asset_snapshots").fetchone()[0] + before_runs = conn.execute("SELECT COUNT(*) FROM asset_snapshot_runs").fetchone()[0] + + report = service.get_asset_snapshot_report("2026-06-30", db_path=db_path) + replay = service.close_asset_snapshot( + "2026-06-30", idempotency_key="malformed-evidence-close", db_path=db_path, + ) + timeseries = service.get_net_worth_timeseries( + "2026-06-01", "2026-06-30", db_path=db_path, + ) + + for result in (report, replay, timeseries): + assert result["success"] is False + assert result["error_code"] == "data_integrity_error" + with sqlite3.connect(db_path) as conn: + assert conn.execute("SELECT COUNT(*) FROM asset_snapshots").fetchone()[0] == before_rows + assert conn.execute("SELECT COUNT(*) FROM asset_snapshot_runs").fetchone()[0] == before_runs + + def test_exact_run_replay_after_correction_reconstructs_original_persisted_run(tmp_path): db_path = str(tmp_path / "ledger.sqlite3") service.create_manual_asset("현금", 100, "2026-06-30", "cash", db_path=db_path) diff --git a/tests/test_import_current_balance_authority.py b/tests/test_import_current_balance_authority.py index 97672cb..fa60d16 100644 --- a/tests/test_import_current_balance_authority.py +++ b/tests/test_import_current_balance_authority.py @@ -65,6 +65,42 @@ def test_late_import_current_balance_is_authoritative_by_evidence_basis_date(mon } +def test_empty_balance_with_basis_date_is_rejected_without_mutation_before_import(tmp_path, monkeypatch): + path = str(tmp_path / "ledger.sqlite3") + account = service.create_ledger_account("KB", institution="KB국민은행", db_path=path)["account"]["id"] + before = service.list_ledger_accounts(active_only=False, db_path=path)["accounts"][0] + before_valuations = service.list_account_valuations(account, db_path=path)["valuations"] + + rejected = service.update_ledger_account( + account, + last_balance="", + last_balance_basis_date="2026-07-01", + db_path=path, + ) + + after = service.list_ledger_accounts(active_only=False, db_path=path)["accounts"][0] + after_valuations = service.list_account_valuations(account, db_path=path)["valuations"] + assert rejected["success"] is False + assert rejected["error_code"] == "validation_error" + assert after == before + assert after_valuations == before_valuations + + monkeypatch.setattr( + imports_service, + "parse_kb_bank_pdf", + lambda *_: _parsed(_transaction("2026-06-30", 900, 1), "hash-valid"), + ) + monkeypatch.setattr(imports_service, "now_iso", lambda: "2026-07-02T09:00:00") + imported = imports_service.import_kb_bank_pdf("synthetic", "pw", account_id=account, db_path=path) + + current = service.list_ledger_accounts(active_only=False, db_path=path)["accounts"][0] + valuations = service.list_account_valuations(account, db_path=path)["valuations"] + assert imported["success"] is True + assert current["last_balance"] == 900 + assert current["last_balance_basis_date"] == "2026-06-30" + assert {(item["basis_date"], item["amount"]) for item in valuations} == {("2026-06-30", 900)} + + def test_account_write_sync_timestamp_does_not_infer_balance_basis_or_candidate(tmp_path): path = str(tmp_path / "ledger.sqlite3") unknown = service.create_ledger_account( diff --git a/tests/test_valuation_contract_docs.py b/tests/test_valuation_contract_docs.py index c473ee1..f3ca20f 100644 --- a/tests/test_valuation_contract_docs.py +++ b/tests/test_valuation_contract_docs.py @@ -43,6 +43,25 @@ def test_interface_contract_documents_public_valuation_enums_validation_and_mapp assert marker in interface, marker +def test_data_model_documents_valuation_selection_contract_and_index_mapping() -> None: + data_model = _text("docs/data-model.md") + + required_contract = ( + "select_account_valuation", + "account_id is mandatory", + "currency and valuation_kind/purpose are optional filters", + "basis_date exact match or <= upper bound", + "verified leaf valuations only", + "future-dated and unverified valuations are excluded", + "idx_account_valuations_account_date supports account/date selection", + "idx_account_valuations_status_date supports status/date selection", + "neither index covers account + currency + purpose", + "idempotency and single_correction indexes enforce write integrity, not selection", + ) + for marker in required_contract: + assert marker in data_model, marker + + def test_loan_cache_writers_document_no_inferred_valuation_history() -> None: data_model = _text("docs/data-model.md") interface = _text("docs/interface-contract.md")