diff --git a/jarca_ledger_mcp/application/accounts_write_services.py b/jarca_ledger_mcp/application/accounts_write_services.py index 48a1dd5..4cb38a5 100644 --- a/jarca_ledger_mcp/application/accounts_write_services.py +++ b/jarca_ledger_mcp/application/accounts_write_services.py @@ -25,6 +25,8 @@ def create_ledger_account(name: str, type: str = "bank", institution: str = "", try: basis_date = normalize_date(last_balance_basis_date, default_today=False) if last_balance_basis_date else "" except Exception as e: return fail("validation_error", str(e)) + if basis_date and (last_balance is None or last_balance == ""): + return fail("validation_error", "last_balance_basis_date requires last_balance") with db(db_path) as conn: conn.execute("INSERT INTO accounts(id,name,type,institution,currency,active,memo,account_number_masked,holder_name,last_balance,last_balance_basis_date,last_synced_at,statement_status,liquidity_bucket,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", (account_id,name.strip(),type,institution,currency.upper(),int(active),memo,account_number_masked,holder_name,last_balance,basis_date,last_synced_at,statement_status_norm,liquidity_bucket_norm,ts,ts)) row = conn.execute("SELECT * FROM accounts WHERE id=?", (account_id,)).fetchone() diff --git a/jarca_ledger_mcp/application/assets_service.py b/jarca_ledger_mcp/application/assets_service.py index eaabf1e..4728f0d 100644 --- a/jarca_ledger_mcp/application/assets_service.py +++ b/jarca_ledger_mcp/application/assets_service.py @@ -261,7 +261,9 @@ def _manual_asset_valuation_items(conn: sqlite3.Connection, active_only: bool = 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("SELECT * FROM asset_snapshot_runs WHERE id=?", (run_id,)).fetchone() + run = conn.execute( + "SELECT rowid AS _run_rowid, * FROM asset_snapshot_runs WHERE id=?", (run_id,) + ).fetchone() if not run: return [], {} basis_date = run["basis_date"] @@ -272,6 +274,19 @@ def _snapshot_rows_for_run(conn: sqlite3.Connection, run_id: str) -> tuple[list[ total_rows = conn.execute( "SELECT COUNT(*) FROM asset_snapshots WHERE basis_date=?", (basis_date,) ).fetchone()[0] + previous_rows = conn.execute( + """SELECT COUNT(*) + FROM asset_snapshots AS snapshot + LEFT JOIN asset_snapshot_runs AS snapshot_run + ON snapshot_run.id=snapshot.snapshot_run_id + WHERE snapshot.basis_date=? + AND ( + snapshot.snapshot_run_id IS NULL + OR snapshot.snapshot_run_id='' + OR snapshot_run.rowid < ? + )""", + (basis_date, run["_run_rowid"]), + ).fetchone()[0] metadata = { "snapshot_run_id": run["id"], "is_correction": bool(run["correction_of_run_id"]), @@ -287,6 +302,7 @@ def _snapshot_rows_for_run(conn: sqlite3.Connection, run_id: str) -> tuple[list[ "excluded_evidence": json.loads(run["excluded_evidence"] or "[]"), "carried_evidence": json.loads(run["carried_evidence"] or "[]"), "closed_at": run["created_at"] or "", + "previous_snapshot_row_count": previous_rows, "total_snapshot_rows_for_basis_date": total_rows, } return rows, metadata @@ -319,14 +335,15 @@ def _snapshot_items(conn: sqlite3.Connection, basis_date: str) -> list[dict[str, ) valuation_date = evidence.get("valuation_date") or snap.get("basis_date", "") carried_forward = bool(evidence.get("carried_forward", False)) - items.append( - _public_evidence({ + public_item = _public_evidence({ "id": snap.get("id", ""), "source_id": snap.get("source_id", ""), "source_type": snap.get("source_type", ""), "authority_source": "asset_snapshots", "valuation_source": snap.get("valuation_source", ""), "name": snap.get("name", ""), + "currency": snap.get("currency", ""), + "valuation_id": snap.get("valuation_id", ""), "liquidity_bucket": bucket, "amount": int(snap["amount"]) if snap.get("amount") is not None else None, "valuation_date": valuation_date, @@ -341,7 +358,9 @@ def _snapshot_items(conn: sqlite3.Connection, basis_date: str) -> list[dict[str, "correction_of_run_id": snap.get("correction_of_run_id", ""), "correction_reason": snap.get("correction_reason", ""), }) - ) + if not snap.get("valuation_id"): + public_item.pop("valuation_id", None) + items.append(public_item) return items @@ -408,6 +427,14 @@ def _closed_warnings(metadata: dict[str, Any], basis_date: str, rows: list[sqlit valuation_date = str(evidence.get("valuation_date") or "")[:10] if valuation_date and valuation_date < basis_date: warnings.append({"code": "stale_valuation", "source_type": evidence.get("source_type", ""), "source_id": evidence.get("source_id", ""), "name": row.get("name", ""), "valuation_date": valuation_date, "basis_date": basis_date}) + previous_rows = int(metadata.get("previous_snapshot_row_count") or 0) + if previous_rows: + warnings.append({ + "code": "existing_snapshot_rows", + "basis_date": basis_date, + "snapshot_row_count": previous_rows, + "policy": "append_only_correction_required", + }) return warnings @@ -607,7 +634,6 @@ def preview_asset_snapshot_close(basis_date: str = "", db_path: str | None = Non return fail("validation_error", str(e)) with read_db(db_path) as conn: preview = _snapshot_preview_from_conn(conn, basis, selection_policy) - preview.pop("_included_items", None) preview["selection_policy"] = selection_policy approval_eligible = [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")] @@ -618,7 +644,10 @@ def preview_asset_snapshot_close(basis_date: str = "", db_path: str | None = Non preview["close_status"] = "incomplete_draft" else: preview["close_status"] = "complete_exact" + # Fingerprint the private authority collection before redacting it from + # the public preview. Close performs the same calculation. preview["preview_fingerprint"] = _preview_fingerprint(preview, basis, selection_policy) + preview.pop("_included_items", None) preview.pop("_candidate_state_hashes", None) return ok(**preview) @@ -649,9 +678,10 @@ def close_asset_snapshot(basis_date: str = "", source_note: str = "", memo: str 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_items = preview.get("_included_items", preview["included_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 fingerprint_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() + authority_fingerprint = _preview_fingerprint(preview, basis, selection_policy) + request_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} + request_fingerprint = hashlib.sha256(json.dumps(request_payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()).hexdigest() + fingerprint = json.dumps({"authority": authority_fingerprint, "request": request_fingerprint}, sort_keys=True, separators=(",", ":")) 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") if carried_items and (not allow_as_of_carry_forward or not approved_by.strip() or not approval_reason.strip() or not approval_reference.strip()): @@ -659,7 +689,22 @@ def close_asset_snapshot(basis_date: str = "", source_note: str = "", memo: str if idempotency_key: existing_run = conn.execute("SELECT * FROM asset_snapshot_runs WHERE idempotency_key=?", (idempotency_key,)).fetchone() if existing_run: - if existing_run["request_fingerprint"] == fingerprint: + try: + stored_fingerprint = json.loads(existing_run["request_fingerprint"]) + except (TypeError, json.JSONDecodeError): + stored_fingerprint = {"authority": existing_run["request_fingerprint"], "request": ""} + latest_basis_run = conn.execute( + "SELECT correction_of_run_id FROM asset_snapshot_runs WHERE basis_date=? ORDER BY rowid DESC LIMIT 1", + (basis,), + ).fetchone() + authority_changed_after_correction = bool( + latest_basis_run + and latest_basis_run["correction_of_run_id"] == existing_run["id"] + ) + if stored_fingerprint.get("request") == request_fingerprint and ( + stored_fingerprint.get("authority") == authority_fingerprint + or authority_changed_after_correction + ): replay_rows, replay_meta = _snapshot_rows_for_run(conn, 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) @@ -679,13 +724,16 @@ def close_asset_snapshot(basis_date: str = "", source_note: str = "", memo: str close_status=existing_run["close_status"], selection_policy=existing_run["selection_policy"], inserted_count=len(replay_rows), - inserted_snapshot_ids=[row["id"] for row in replay_rows], + inserted_snapshot_ids=[ + next((row["id"] for row in replay_rows if row["source_type"] == item.get("source_type") and row["source_id"] == item.get("source_id")), "") + for item in replay_meta.get("applied_evidence", []) + ], warnings=replay_warnings, carried_items=replay_public["carried_items"], approval=approval, correction=correction, provenance=_canonical_provenance( - valuation_date=replay_provenance["valuation_date"], - age_days=replay_provenance["age_days"], + valuation_date=(replay_items[0].get("valuation_date", "") if replay_items else ""), + age_days=(replay_items[0].get("age_days") if replay_items else None), carried=replay_provenance["carried_forward"], verification_status=replay_provenance["verification_status"], evidence_kind=replay_provenance["evidence_kind"], @@ -740,7 +788,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", "currency", "valuation_id", "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") 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( { @@ -945,10 +993,13 @@ def get_net_worth_timeseries(start_date: str = "", end_date: str = "", months: i return fail("validation_error", str(e)) with read_db(db_path) as conn: basis_rows = conn.execute( - """SELECT DISTINCT basis_date FROM asset_snapshots + """SELECT basis_date FROM asset_snapshots + WHERE basis_date>=? AND basis_date<=? + UNION + SELECT basis_date FROM asset_snapshot_runs WHERE basis_date>=? AND basis_date<=? ORDER BY basis_date ASC""", - (start_norm, end_norm), + (start_norm, end_norm, start_norm, end_norm), ).fetchall() points = [] for row in basis_rows: diff --git a/tests/test_issue_12_review63_accounts.py b/tests/test_issue_12_review63_accounts.py new file mode 100644 index 0000000..690b2fb --- /dev/null +++ b/tests/test_issue_12_review63_accounts.py @@ -0,0 +1,76 @@ +from types import SimpleNamespace + +from jarca_ledger_mcp import service +from jarca_ledger_mcp.application import imports_service + + +def _transaction(date, balance): + return SimpleNamespace( + transaction_date=date, + transaction_datetime=f"{date} 10:00:00", + amount=100, + counterparty="상대", + raw_summary="입금", + branch="", + remittance_memo="", + balance_after=balance, + type="income", + page=1, + action="입금", + external_id=lambda source_hash: f"review63:{source_hash}", + raw_payload=lambda: "{}", + ) + + +def test_create_rejects_basis_only_without_mutation(tmp_path): + path = str(tmp_path / "ledger.sqlite3") + + rejected = service.create_ledger_account( + "기준일만 있는 계좌", + last_balance_basis_date="2026-07-31", + db_path=path, + ) + + assert rejected == { + "success": False, + "error_code": "validation_error", + "error": "last_balance_basis_date requires last_balance", + } + baseline = service.create_ledger_account("정상 계좌", db_path=path) + assert baseline["success"] is True + listed = service.list_ledger_accounts(active_only=False, db_path=path) + assert listed["count"] == 1 + assert listed["accounts"][0]["name"] == "정상 계좌" + + +def test_basis_only_create_cannot_block_later_dated_import(tmp_path, monkeypatch): + path = str(tmp_path / "ledger.sqlite3") + rejected = service.create_ledger_account( + "기준일만 있는 계좌", + last_balance_basis_date="2026-07-31", + db_path=path, + ) + assert rejected["success"] is False + + account = service.create_ledger_account("정상 계좌", db_path=path)["account"]["id"] + monkeypatch.setattr( + imports_service, + "parse_kb_bank_pdf", + lambda *_: { + "transactions": [_transaction("2026-06-30", 50)], + "validation": {"ok": True}, + "source_hash": "review63", + "metadata": {"final_balance": 50, "account_number_masked": ""}, + "file_path": "/tmp/synthetic.pdf", + "page_count": 1, + }, + ) + + imported = imports_service.import_kb_bank_pdf( + "ignored", "pw", account_id=account, db_path=path, + ) + + current = service.list_ledger_accounts(active_only=False, db_path=path)["accounts"][0] + assert imported["success"] is True + assert current["last_balance"] == 50 + assert current["last_balance_basis_date"] == "2026-06-30" diff --git a/tests/test_issue_12_review63_assets.py b/tests/test_issue_12_review63_assets.py new file mode 100644 index 0000000..ffd531a --- /dev/null +++ b/tests/test_issue_12_review63_assets.py @@ -0,0 +1,162 @@ +from __future__ import annotations + +import sqlite3 + +from jarca_ledger_mcp import service + + +def _approval_kwargs(): + return { + "allow_missing_exclusions": True, + "approved_by": "reviewer", + "approval_reason": "reviewed", + "approval_reference": "review-63", + } + + +def test_f001_preview_fingerprint_uses_same_unredacted_authority_and_rejects_drift(tmp_path): + db_path = str(tmp_path / "ledger.sqlite3") + service.create_manual_asset("현금", 100, "2026-06-30", "cash", db_path=db_path) + + preview = service.preview_asset_snapshot_close("2026-06-30", db_path=db_path) + closed = service.close_asset_snapshot( + "2026-06-30", preview_fingerprint=preview["preview_fingerprint"], db_path=db_path + ) + assert closed["success"] is True + + service.update_manual_asset( + next(iter(service.list_manual_assets(db_path=db_path)["manual_assets"]))["id"], + valuation_amount=101, + db_path=db_path, + ) + with sqlite3.connect(db_path) as conn: + rows_before = conn.execute("SELECT COUNT(*) FROM asset_snapshot_runs").fetchone()[0] + drifted = service.close_asset_snapshot( + "2026-06-30", preview_fingerprint=preview["preview_fingerprint"], db_path=db_path + ) + with sqlite3.connect(db_path) as conn: + rows_after = conn.execute("SELECT COUNT(*) FROM asset_snapshot_runs").fetchone()[0] + assert drifted["success"] is False + assert drifted["error_code"] == "authority_drift" + assert rows_after == rows_before + + +def test_f002_close_replay_is_byte_for_byte_equal_for_mixed_account_manual_as_of(tmp_path): + db_path = str(tmp_path / "ledger.sqlite3") + account = service.create_ledger_account( + "투자계좌", type="investment", currency="KRW", db_path=db_path + ) + service.register_account_valuation( + account["account"]["id"], "2026-06-29", 200, valuation_kind="total_value", + verification_status="verified", idempotency_key="v-29", db_path=db_path, + ) + service.create_manual_asset("보증금", 300, "2026-06-30", "real_estate", db_path=db_path) + + first = service.close_asset_snapshot( + "2026-06-30", selection_policy="as_of", idempotency_key="mixed-replay", + allow_as_of_carry_forward=True, **_approval_kwargs(), db_path=db_path, + ) + assert first["success"] is True + replay = service.close_asset_snapshot( + "2026-06-30", selection_policy="as_of", idempotency_key="mixed-replay", + allow_as_of_carry_forward=True, **_approval_kwargs(), db_path=db_path, + ) + assert replay == first + + +def test_f002_correction_replay_preserves_full_initial_response(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="correction-initial", db_path=db_path + ) + assert first["success"] is True + service.update_manual_asset( + next(iter(service.list_manual_assets(db_path=db_path)["manual_assets"]))["id"], + valuation_amount=120, db_path=db_path, + ) + corrected = service.close_asset_snapshot( + "2026-06-30", correction_reason="updated valuation", + idempotency_key="correction-run", db_path=db_path, + ) + assert corrected["success"] is True + assert any(warning["code"] == "existing_snapshot_rows" for warning in corrected["warnings"]) + + correction_replay = service.close_asset_snapshot( + "2026-06-30", correction_reason="updated valuation", + idempotency_key="correction-run", db_path=db_path, + ) + assert correction_replay == corrected + + initial_replay = service.close_asset_snapshot( + "2026-06-30", idempotency_key="correction-initial", db_path=db_path + ) + assert initial_replay == first + + +def test_f003_zero_row_conditional_runs_are_timeseries_authority_with_aliases(tmp_path): + db_path = str(tmp_path / "ledger.sqlite3") + service.create_ledger_account("미평가", db_path=db_path) + first = service.close_asset_snapshot( + "2026-06-30", idempotency_key="zero-initial", **_approval_kwargs(), db_path=db_path + ) + assert first["success"] is True + assert first["inserted_count"] == 0 + assert first["total_assets"] == first["total_liabilities"] == first["net_worth"] == 0 + + corrected = service.close_asset_snapshot( + "2026-06-30", correction_reason="reconfirm zero", idempotency_key="zero-correction", + **_approval_kwargs(), db_path=db_path, + ) + assert corrected["success"] is True + assert corrected["inserted_count"] == 0 + + series = service.get_net_worth_timeseries( + start_date="2026-06-01", end_date="2026-06-30", db_path=db_path + ) + assert series["success"] is True + assert len(series["points"]) == 1 + point = series["points"][0] + assert point["basis_date"] == "2026-06-30" + assert point["net_worth"] == point["total_assets"] == point["total_liabilities"] == 0 + assert point["close_status"] == "conditional" + assert point["approval"]["approval_reference"] == "review-63" + assert point["excluded_items"] + assert point["snapshot_run_id"] == corrected["snapshot_run_id"] + assert point["provenance"]["monthly_source_snapshot_id"] == corrected["snapshot_run_id"] + + +def test_f004_closed_report_restores_frozen_valuation_identity_without_legacy_invention(tmp_path): + db_path = str(tmp_path / "ledger.sqlite3") + account = service.create_ledger_account("계좌", db_path=db_path) + valuation = service.register_account_valuation( + account["account"]["id"], "2026-06-29", 500, verification_status="verified", + idempotency_key="frozen-v1", db_path=db_path, + ) + first = service.close_asset_snapshot( + "2026-06-30", selection_policy="as_of", idempotency_key="frozen-close", + allow_as_of_carry_forward=True, **_approval_kwargs(), db_path=db_path, + ) + assert first["success"] is True + report_before = service.get_asset_snapshot_report("2026-06-30", db_path=db_path) + item_before = report_before["included_items"][0] + assert item_before["valuation_id"] == valuation["valuation"]["id"] + assert item_before["currency"] == "KRW" + + service.register_account_valuation( + account["account"]["id"], "2026-06-29", 900, verification_status="verified", + idempotency_key="frozen-v2", db_path=db_path, + ) + report_after = service.get_asset_snapshot_report("2026-06-30", db_path=db_path) + assert report_after["included_items"][0]["valuation_id"] == item_before["valuation_id"] + assert report_after["included_items"][0]["currency"] == item_before["currency"] + assert report_after["selection_policy"] == "as_of" + + legacy_db = str(tmp_path / "legacy.sqlite3") + service.create_asset_snapshot( + "2026-06-30", "account", "legacy-account", "legacy", 10, "cash", db_path=legacy_db + ) + legacy = service.get_asset_snapshot_report("2026-06-30", db_path=legacy_db) + legacy_item = legacy["included_items"][0] + assert "valuation_id" not in legacy_item + assert legacy_item["currency"] == "KRW"