from __future__ import annotations import hashlib import json import os import sqlite3 import sys from pathlib import Path from typing import Any WORKTREE = Path("/home/agent/.hermes/worktrees/FinanceManager-main-deploy") sys.path.insert(0, str(WORKTREE / "src")) from jarvis_finance.services.household_import import ( authorize_owner_neutral_cluster, preview_household_import, ) DB = Path("/home/agent/jarvis_runtime/finance-system/data/finance.sqlite3") BASE_PAYLOAD = Path("/home/agent/jarvis_runtime/finance-system/sprint16.1-uat/real-preview-payload-private.json") ROOT = Path("/home/agent/jarvis_runtime/finance-system/sprint16.3-analysis") REQUEST_OUT = ROOT / "sprint16.3-final-request-private.json" PREVIEW_OUT = ROOT / "sprint16.3-final-preview-private.json" EVIDENCE_OUT = ROOT / "sprint16.3-final-preview-evidence-private.json" SOURCE_FILES = [ Path("/home/agent/.hermes/private/finance/sprint16/raiffeisen.csv"), Path("/home/agent/.hermes/private/finance/sprint16/akb_transactions.csv"), Path("/home/agent/.hermes/private/finance/sprint16/visa.csv"), Path("/home/agent/.hermes/private/finance/sprint16/migros.csv"), ] EXPECTED = [ (91, "expense", ("galaxus", "digitec")), (25, "expense", ("ikea",)), (19, "income", ("einwohnergemeinde",)), (14, "income", ()), (13, "expense", ()), (13, "expense", ("sonneland",)), (11, "income", ("gasser",)), (10, "expense", ("jumbo",)), (10, "expense", ("saira",)), (9, "income", ("gutschrift",)), (8, "expense", ("beck",)), (7, "expense", ()), (7, "expense", ("geld gesendet",)), (7, "income", ("erne",)), (6, "expense", ("dauerauftrag",)), (6, "expense", ("dauerauftrag",)), (6, "expense", ("ubertrag", "übertrag")), (5, "expense", ("tanken", "parkieren", "fahrzeug")), (5, "expense", ("ubertrag", "übertrag")), (4, "expense", ("sport", "freizeit", "zentrum")), ] CATEGORY_BY_UNIT = { 1: "Einmalige Anschaffungen", 2: "Einmalige Anschaffungen", 3: "Lohn Melanie", 5: "Essen & Haushalt", 6: "Auto Melanie", 7: "Sonstige Einnahmen", 8: "Liegenschaftsunterhalt", 9: "Essen & Haushalt", 10: "Rückerstattungen", 11: "Essen & Haushalt", 12: "Essen & Haushalt", 14: "Lohn Marcel", 18: "Auto Melanie", 20: "Hobbys & Selfcare", } TRANSFER_UNITS = {4, 15, 16, 17, 19} SKIP_UNITS = {13} def private_write(path: Path, value: Any) -> None: tmp = path.with_suffix(path.suffix + ".tmp") tmp.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8") os.chmod(tmp, 0o600) os.replace(tmp, path) os.chmod(path, 0o600) def sha_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def snapshot() -> dict[str, Any]: conn = sqlite3.connect(f"file:{DB}?mode=ro", uri=True) tables = [r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name")] result: dict[str, Any] = {"tables": {}, "integrity": conn.execute("PRAGMA integrity_check").fetchone()[0], "foreign_keys": list(conn.execute("PRAGMA foreign_key_check"))} for table in tables: digest = hashlib.sha256(table.encode()) count = 0 for row in conn.execute(f'SELECT * FROM "{table}" ORDER BY rowid'): count += 1 digest.update(json.dumps(list(row), ensure_ascii=False, default=str, separators=(",", ":")).encode()) result["tables"][table] = {"count": count, "digest": digest.hexdigest()} conn.close() return result def name_matches(label: str, accepted: tuple[str, ...]) -> bool: normalized = " ".join(label.casefold().replace("ü", "u").split()) return not accepted or any(token.casefold().replace("ü", "u") in normalized for token in accepted) def run() -> None: before = snapshot() source_before = {str(path): sha_file(path) for path in SOURCE_FILES} payload = json.loads(BASE_PAYLOAD.read_text(encoding="utf-8")) conn = sqlite3.connect(f"file:{DB}?mode=ro", uri=True) conn.row_factory = sqlite3.Row base = preview_household_import(conn, payload) clusters = sorted(base["merchant_clusters"], key=lambda item: (-item["count"], item["merchant_family"]))[:20] if len(clusters) != 20: raise RuntimeError(f"expected 20 decision units, got {len(clusters)}") identity_checks: list[dict[str, Any]] = [] for unit, (cluster, expected) in enumerate(zip(clusters, EXPECTED, strict=True), 1): count, semantics, accepted = expected passed = ( cluster["count"] == count and cluster["transaction_semantics"] == semantics and name_matches(cluster["merchant_family"], accepted) ) identity_checks.append({ "unit": unit, "count": cluster["count"], "transaction_semantics": cluster["transaction_semantics"], "identity_hash": hashlib.sha256(cluster["merchant_family"].encode()).hexdigest(), "passed": passed, }) if not passed: raise RuntimeError(f"decision unit {unit} identity/count/semantic mismatch") category_rows = conn.execute( "SELECT category_id,name,category_type FROM budget_categories WHERE is_active=1 ORDER BY rowid" ).fetchall() by_name: dict[str, list[sqlite3.Row]] = {} for row in category_rows: by_name.setdefault(str(row["name"]), []).append(row) decisions: list[dict[str, Any]] = [] for unit, cluster in enumerate(clusters, 1): if unit in SKIP_UNITS: continue if unit in TRANSFER_UNITS: decisions.append(authorize_owner_neutral_cluster( conn, { "confirm_owner_attestation": True, "preview_request": payload, "approval_evidence": "owner_attested_known_household_counterparty_v1", "cluster_token": cluster["cluster_token"], "approved_row_tokens": cluster["row_tokens"], }, os.environ.get("JARVIS_FINANCE_OPERATOR_APPROVAL_KEY"), )) continue category_name = CATEGORY_BY_UNIT[unit] matches = by_name.get(category_name, []) if len(matches) != 1: raise RuntimeError(f"category {category_name!r} must resolve uniquely, got {len(matches)}") category = matches[0] if str(category["category_type"]) != cluster["transaction_semantics"]: raise RuntimeError(f"decision unit {unit} category semantic mismatch") decisions.append({ "cluster_token": cluster["cluster_token"], "category_id": str(category["category_id"]), "decision_type": "category", "excluded_row_tokens": [], }) final_request = payload | {"cluster_decisions": decisions} final = preview_household_import(conn, final_request) conn.close() after = snapshot() source_after = {str(path): sha_file(path) for path in SOURCE_FILES} unit13_tokens = set(clusters[12]["row_tokens"]) unit13_items = [item for item in final["items"] if item["row_token"] in unit13_tokens] neutral_tokens = {token for unit in TRANSFER_UNITS for token in clusters[unit - 1]["row_tokens"]} neutral_items = [item for item in final["items"] if item["row_token"] in neutral_tokens] applied_category_tokens = {token for unit in CATEGORY_BY_UNIT for token in clusters[unit - 1]["row_tokens"]} applied_category_items = [item for item in final["items"] if item["row_token"] in applied_category_tokens] marketplace_items = [ item for item in final["items"] if any(token in item["merchant"].casefold() for token in ("galaxus", "digitec")) ] marketplace_refunds = [item for item in marketplace_items if item["transaction_semantics"] == "refund"] assertions = { "identity_checks_passed": all(item["passed"] for item in identity_checks), "decision_unit_count": len(decisions) == 19, "unit13_remains_seven_manual_rows": len(unit13_items) == 7 and all(item["user_state"] == "decision_needed" for item in unit13_items), "unit13_not_guessed_as_transfer": all(item["transaction_semantics"] != "user_confirmed_unmatched_transfer" for item in unit13_items), "all_owner_confirmed_transfer_rows_are_neutral": len(neutral_items) == 37 and all(item["transaction_semantics"] == "user_confirmed_unmatched_transfer" for item in neutral_items), "all_category_rows_resolved": len(applied_category_items) == 232 and all(item["category_id"] for item in applied_category_items), "marketplace_refunds_remain_refunds": len(marketplace_refunds) == 21 and all( item.get("category_name") != "Einmalige Anschaffungen" for item in marketplace_refunds ), "concept_v2_gate_met": final["concept_v2_readiness"]["gate_met"] is True, "technically_confirmable": final["technically_confirmable"] is True, "business_ready_for_confirm": final["business_ready_for_confirm"] is True, "strict_indicator_separate": final["strict_quality_indicator"]["indicator_met"] is False, "errors_empty": final["errors"] == [], "database_unchanged": before == after, "source_hashes_unchanged": source_before == source_after, "integrity_ok": before["integrity"] == after["integrity"] == "ok", "foreign_keys_zero": before["foreign_keys"] == after["foreign_keys"] == [], "under_hard_uat_seconds": final["performance"]["api_total_seconds"] <= 120, } if not all(assertions.values()): raise RuntimeError(f"final preview assertions failed: {assertions}") private_write(REQUEST_OUT, final_request) private_write(PREVIEW_OUT, final) private_write(EVIDENCE_OUT, { "assertions": assertions, "identity_checks": identity_checks, "source_hashes": source_after, "db_snapshot": after, "preview_fingerprint": final["preview_fingerprint"], "baseline_fingerprint": final["baseline_fingerprint"], "counts": final["counts"], "concept_v2_readiness": final["concept_v2_readiness"], "strict_quality_indicator": final["strict_quality_indicator"], "review_threshold": final["review_threshold"], "expected_budget_effect": final["expected_budget_effect"], "expected_writes": final["expected_writes"], "performance": final["performance"], }) print(json.dumps({ "assertions": assertions, "counts": final["counts"], "concept_v2_readiness": final["concept_v2_readiness"], "strict_quality_indicator": final["strict_quality_indicator"], "review_threshold": final["review_threshold"], "expected_writes": final["expected_writes"], "performance": final["performance"], }, ensure_ascii=False)) if __name__ == "__main__": run()