Fix Review 63 snapshot and balance regressions

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()

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_item = _public_evidence({
-            _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"])
+        authority_fingerprint = _preview_fingerprint(preview, basis, selection_policy)
-        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"]}
+        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}
-        fingerprint = hashlib.sha256(json.dumps(fingerprint_payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()).hexdigest()
+        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"],
+                            valuation_date=(replay_items[0].get("valuation_date", "") if replay_items else ""),
-                            age_days=replay_provenance["age_days"],
+                            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:

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"

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"