jsfamily/jarca-ledger-mcp
5
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: preserve migrated legacy snapshot closes

Browse Source
This commit is contained in:
gitea-impl-agent
2026-10-08 03:30:59 +00:00
parent 432e6f851a
commit 4b8b74ba2a
34 changed files with 178 additions and 2128 deletions
Download Patch File Download Diff File Expand all files Collapse all files
jarca_ledger_mcp/application/assets_service.py
+32 -3
View File
@@ -328,6 +328,32 @@ def _snapshot_rows_for_latest_run(conn: sqlite3.Connection, basis_date: str) ->
rows = conn.execute("SELECT rowid AS _rowid, * FROM asset_snapshots WHERE basis_date=? ORDER BY created_at ASC, name ASC", (basis_date,)).fetchall()
run = conn.execute("SELECT * FROM asset_snapshot_runs WHERE basis_date=? ORDER BY rowid DESC LIMIT 1", (basis_date,)).fetchone()
if not run:
legacy_run_rows = [row for row in rows if row["closed_at"] and row["snapshot_run_id"]]
if legacy_run_rows:
# Schema 11 persisted run identity only on the snapshot rows. Keep
# its rowid-based latest-run rule rather than falling back to live
# authority merely because schema 12 has no matching run record.
latest = max(
legacy_run_rows,
key=lambda row: (
int(row["_rowid"]),
row["closed_at"] or row["created_at"] or "",
row["snapshot_run_id"],
),
)
latest_run_id = latest["snapshot_run_id"]
latest_rows = [row for row in legacy_run_rows if row["snapshot_run_id"] == latest_run_id]
return latest_rows, {
"snapshot_run_id": latest_run_id,
"legacy_closed": True,
"is_correction": bool(latest["is_correction"]),
"correction_of_run_id": latest["correction_of_run_id"] or "",
"correction_reason": latest["correction_reason"] or "",
"close_status": "legacy_read_only",
"selection_policy": latest["selection_policy"] or "exact",
"closed_at": latest["closed_at"] or latest["created_at"] or "",
"total_snapshot_rows_for_basis_date": len(rows),
}
return rows, {"snapshot_run_id": "", "is_correction": False, "correction_of_run_id": "", "correction_reason": "", "total_snapshot_rows_for_basis_date": len(rows)} if rows else {}
return _snapshot_rows_for_run(conn, run["id"])
@@ -403,7 +429,7 @@ def _closed_run_metadata(metadata: dict[str, Any]) -> dict[str, Any]:
"snapshot_run_id": metadata.get("snapshot_run_id", ""),
"close_status": metadata.get("close_status", ""),
"selection_policy": metadata.get("selection_policy", ""),
"approval": {k: metadata.get(k, "") for k in ("approved_by", "approved_at", "approval_reason", "approval_reference")},
"approval": {} if metadata.get("legacy_closed") else {k: metadata.get(k, "") for k in ("approved_by", "approved_at", "approval_reason", "approval_reference")},
"correction": {k: metadata.get(k, "") for k in ("is_correction", "correction_of_run_id", "correction_reason")},
}
@@ -949,16 +975,19 @@ def get_asset_snapshot_report(basis_date: str = "", db_path: str | None = None,
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"))
legacy_closed = bool(close_metadata.get("legacy_closed"))
snapshots = _snapshot_items(conn, basis) if rows else []
legacy_snapshots = snapshots if rows and not has_close_run else []
if has_close_run:
if has_close_run or legacy_closed:
run_public = _closed_run_metadata(close_metadata)
run_public["warnings"] = _closed_warnings(close_metadata, basis, rows)
if legacy_closed:
run_public["approval"] = {}
summary = _summarize_valuation_items(snapshots, run_public["excluded_items"])
evidence_provenance = _evidence_provenance(close_metadata.get("applied_evidence", []), basis)
return _ok_with_provenance_aliases(
basis_date=basis,
source="asset_snapshots",
source="legacy_asset_snapshots" if legacy_closed else "asset_snapshots",
is_closed_snapshot=True,
snapshot_run_id=close_metadata.get("snapshot_run_id", ""),
snapshot_count=len(snapshots),