from __future__ import annotations import hashlib import json import re import unicodedata from datetime import date, datetime, timedelta from decimal import Decimal, InvalidOperation from sqlite3 import Connection from typing import Any from zoneinfo import ZoneInfo from jarvis_finance.audit.log import record_audit_event from jarvis_finance.imports.common import stable_id, utc_now from jarvis_finance.services.performance_hardening import _truewealth_account CONTRACT = "truewealth_raiffeisen_recipient_v1" SOURCE = "truewealth_raiffeisen_cashflow_v1" ZERO = Decimal(0) def _canonical(value: Any) -> str: return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) def _fingerprint(value: Any) -> str: return hashlib.sha256(_canonical(value).encode("utf-8")).hexdigest() def _normalise(value: object) -> str: text = unicodedata.normalize("NFKD", str(value or "")) text = "".join(char for char in text if not unicodedata.combining(char)) return " ".join(re.sub(r"[^a-z0-9]+", " ", text.lower()).split()) def _exact_normalised_phrase(needle: str | None, haystack: str) -> bool: if not needle: return False return re.search(rf"(?:^| ){re.escape(needle)}(?: |$)", haystack) is not None def _decimal(value: object) -> Decimal: try: result = Decimal(str(value)) except (InvalidOperation, TypeError, ValueError) as exc: raise ValueError("Bank cashflow amount is invalid") from exc if not result.is_finite(): raise ValueError("Bank cashflow amount is invalid") return result def _period(period_from: str, period_to: str) -> tuple[str, str]: try: start = date.fromisoformat(period_from) end = date.fromisoformat(period_to) except ValueError as exc: raise ValueError("True Wealth bank preview requires ISO dates") from exc if start > end: raise ValueError("True Wealth bank preview period is invalid") return start.isoformat(), end.isoformat() def _rule_payload( *, recipient_name: str, recipient_account_identity: str | None, counterparty_hashes: list[str], reference_tokens: list[str], attestation: str, ) -> dict[str, Any]: name = _normalise(recipient_name) account_identity = _normalise(recipient_account_identity) hashes = sorted({str(value).strip().lower() for value in counterparty_hashes if str(value).strip()}) tokens = sorted({_normalise(value) for value in reference_tokens if _normalise(value)}) if any(len(value) < 16 for value in hashes): raise ValueError("Counterparty hashes must contain at least 16 characters") if account_identity and len(account_identity.replace(" ", "")) < 8: raise ValueError("Recipient account identity is too short for exact matching") if any(len(value.replace(" ", "")) < 6 for value in tokens): raise ValueError("Payment reference tokens must contain at least 6 characters") note = attestation.strip() if not name and not account_identity and not hashes and not tokens: raise ValueError("Recipient rule requires a stable recipient identity") if len(note) < 12: raise ValueError("Recipient rule requires an explicit attestation") return { "contract": CONTRACT, "recipient_name": name, "recipient_account_identity": account_identity or None, "counterparty_hashes": hashes, "reference_tokens": tokens, "attestation": note, } def _candidate_rows(conn: Connection, start: str, end: str) -> list[dict[str, Any]]: rows = conn.execute( """SELECT b.budget_transaction_id,b.account_id,b.transaction_date,b.booking_date, b.description,b.payee,b.amount_original,b.currency_original,b.amount_chf,b.fx_rate_to_chf, b.fx_status,b.transaction_type, b.status,b.source_candidate_id,a.name AS account_name,a.account_type AS budget_account_type, a.currency AS budget_account_currency,a.is_active AS budget_account_active, c.raw_fingerprint,c.notes AS candidate_notes,c.merchant, c.source_type AS candidate_source_type,c.account_source AS candidate_account_source FROM budget_transactions b JOIN budget_accounts a ON a.budget_account_id=b.account_id LEFT JOIN budget_transaction_candidates c ON c.transaction_candidate_id=b.source_candidate_id WHERE c.source_type='raiffeisen_bank' AND b.transaction_date BETWEEN ? AND ? AND b.status='confirmed' AND CAST(b.amount_original AS NUMERIC)<0 ORDER BY b.transaction_date,b.budget_transaction_id""", (start, end), ).fetchall() return [dict(row) for row in rows] def _counterparty_hash(row: dict[str, Any]) -> str | None: try: notes = json.loads(str(row.get("candidate_notes") or "{}")) except (TypeError, json.JSONDecodeError): return None value = str(notes.get("counterparty_hash") or "").strip().lower() return value or None def _existing_truewealth_flows(conn: Connection, account_id: str, start: str, end: str) -> list[dict[str, Any]]: return [ dict(row) for row in conn.execute( """SELECT transaction_id,trade_date,ABS(CAST(net_amount_original AS NUMERIC)) amount, currency_original,source_reference,external_transaction_id,row_hash,source_type FROM transactions WHERE account_id=? AND trade_date BETWEEN ? AND ? AND transaction_type='external_deposit' AND is_confirmed=1 AND COALESCE(is_voided,0)=0 ORDER BY trade_date,transaction_id""", (account_id, start, end), ).fetchall() ] def _confirmed_collision_links(conn: Connection) -> dict[str, str]: links: dict[str, str] = {} rows = conn.execute( """SELECT new_values_json FROM audit_log WHERE source=? AND action='recipient_rule_and_history_confirmed' AND confirmed=1 ORDER BY created_at,audit_id""", (SOURCE,), ).fetchall() for row in rows: try: payload = json.loads(str(row[0] or "{}")) except (TypeError, json.JSONDecodeError): continue for link in payload.get("collision_links") or []: bank_id = str(link.get("bank_transaction_id") or "") cashflow_id = str(link.get("existing_cashflow_id") or "") if bank_id and cashflow_id: links[bank_id] = cashflow_id return links def preview_truewealth_bank_payments( conn: Connection, *, period_from: str, period_to: str, recipient_name: str, recipient_account_identity: str | None, counterparty_hashes: list[str], reference_tokens: list[str], attestation: str, collision_decisions: dict[str, str] | None = None, ) -> dict[str, Any]: """Read-only deterministic candidate preview over already imported Raiffeisen rows.""" start, end = _period(period_from, period_to) rule = _rule_payload( recipient_name=recipient_name, recipient_account_identity=recipient_account_identity, counterparty_hashes=counterparty_hashes, reference_tokens=reference_tokens, attestation=attestation, ) decisions = {str(key): str(value) for key, value in (collision_decisions or {}).items()} if any(value not in {"link_existing", "record_distinct"} for value in decisions.values()): raise ValueError("Unsupported economic-collision decision") account_id = _truewealth_account(conn) bank_rows = _candidate_rows(conn, start, end) existing = _existing_truewealth_flows(conn, account_id, start, end) confirmed_collision_links = _confirmed_collision_links(conn) existing_by_economic: dict[tuple[str, str, str], list[dict[str, Any]]] = {} existing_by_lineage: dict[str, list[dict[str, Any]]] = {} for existing_row in existing: economic_key = ( str(existing_row["trade_date"]), format(_decimal(existing_row["amount"]).normalize(), "f"), str(existing_row["currency_original"]).upper(), ) existing_by_economic.setdefault(economic_key, []).append(existing_row) for lineage in ( existing_row.get("source_reference"), existing_row.get("external_transaction_id"), existing_row.get("row_hash"), ): if lineage: existing_by_lineage.setdefault(str(lineage), []).append(existing_row) used_existing_ids: set[str] = set() applied_decisions: set[str] = set() seen_bank_keys: set[str] = set() candidates: list[dict[str, Any]] = [] excluded: list[dict[str, Any]] = [] for row in bank_rows: text = _normalise(" ".join(str(row.get(key) or "") for key in ("payee", "merchant", "description"))) payee = _normalise(row.get("payee") or row.get("merchant")) counterparty = _counterparty_hash(row) reasons: list[str] = [] strong = False if counterparty and counterparty in rule["counterparty_hashes"]: reasons.append("counterparty_identity_exact") strong = True if _exact_normalised_phrase(rule["recipient_account_identity"], text): reasons.append("recipient_account_identity_exact") strong = True name_exact = False if rule["recipient_name"] and payee and rule["recipient_name"] == payee: reasons.append("recipient_name_exact_supporting") name_exact = True matched_tokens = [token for token in rule["reference_tokens"] if _exact_normalised_phrase(token, text)] if matched_tokens: reasons.append("payment_reference_exact") strong = strong or str(row["transaction_type"]) == "transfer" name_tokens = [part for part in rule["recipient_name"].split() if len(part) >= 5] text_tokens = set(text.split()) overlap = [part for part in name_tokens if part in text_tokens] weak = bool( str(row["transaction_type"]) == "transfer" and (name_exact or len(overlap) >= 2 or bool(name_tokens and name_tokens[0] in text_tokens)) ) amount = abs(_decimal(row["amount_original"])) bank_key = str(row.get("raw_fingerprint") or row["budget_transaction_id"]) duplicate_bank_row = bank_key in seen_bank_keys seen_bank_keys.add(bank_key) base = { "bank_transaction_id": str(row["budget_transaction_id"]), "booking_date": str(row["booking_date"] or row["transaction_date"]), "value_date": str(row["transaction_date"]), "date_basis": "bank_value_date_proxy", "amount": format(amount.normalize(), "f"), "currency": str(row["currency_original"]).upper(), "overall_wealth_treatment": "internal_wealth_transfer", "truewealth_scope_treatment": "external_deposit", "household_treatment": "not_expense_not_income", } source_reasons: list[str] = [] if str(row.get("budget_account_type") or "") != "checking": source_reasons.append("raiffeisen_cash_account_role_required") if int(row.get("budget_account_active") or 0) != 1: source_reasons.append("active_raiffeisen_account_required") if str(row.get("budget_account_currency") or "").upper() != "CHF": source_reasons.append("raiffeisen_account_currency_not_chf") if str(row.get("candidate_source_type") or "") != "raiffeisen_bank": source_reasons.append("raiffeisen_source_provenance_required") if str(row.get("candidate_account_source") or "") != str(row["account_id"]): source_reasons.append("raiffeisen_account_source_mismatch") original_amount = _decimal(row["amount_original"]) chf_amount = _decimal(row["amount_chf"]) if row.get("amount_chf") is not None else None fx_rate = _decimal(row["fx_rate_to_chf"]) if row.get("fx_rate_to_chf") is not None else None if str(row["currency_original"]).upper() != "CHF" or chf_amount is None: source_reasons.append("confirmed_chf_amount_required") if str(row.get("fx_status") or "") != "not_needed": source_reasons.append("confirmed_fx_status_required") if chf_amount is not None and chf_amount != original_amount: source_reasons.append("chf_amount_inconsistent_with_original") if fx_rate != Decimal("1"): source_reasons.append("chf_identity_fx_rate_required") if source_reasons: recipient_matched = strong or weak target = candidates if recipient_matched else excluded target.append( { **base, "overall_wealth_treatment": ( "not_applied_pending_review" if recipient_matched else "not_applicable_excluded" ), "truewealth_scope_treatment": ( "not_applied_pending_review" if recipient_matched else "not_applicable_excluded" ), "household_treatment": ( "unchanged_pending_review" if recipient_matched else "unchanged_excluded" ), "match_quality": "uncertain" if recipient_matched else "excluded", "recognition_reasons": sorted( [*source_reasons, "recipient_match_with_unresolved_source_quality"] if recipient_matched else source_reasons ), **( {"disposition": "unclear", "existing_cashflow_id": None} if recipient_matched else {} ), } ) continue if not strong and not weak: excluded.append( { **base, "overall_wealth_treatment": "not_applicable_excluded", "truewealth_scope_treatment": "not_applicable_excluded", "household_treatment": "unchanged_excluded", "match_quality": "excluded", "recognition_reasons": ["stable_recipient_identity_not_matched"], } ) continue quality = "secure" if strong else "uncertain" if quality == "uncertain": base["overall_wealth_treatment"] = "not_applied_pending_review" base["truewealth_scope_treatment"] = "not_applied_pending_review" base["household_treatment"] = "unchanged_pending_review" economic = (base["value_date"], base["amount"], base["currency"]) exact_matches: list[dict[str, Any]] = [] for lineage in (base["bank_transaction_id"], bank_key): exact_matches.extend(existing_by_lineage.get(str(lineage), [])) existing_row = next( ( item for item in exact_matches if str(item["transaction_id"]) not in used_existing_ids ), None, ) economic_collision = bool(existing_by_economic.get(economic)) and existing_row is None decision = decisions.get(base["bank_transaction_id"]) confirmed_link_id = confirmed_collision_links.get(base["bank_transaction_id"]) if confirmed_link_id: linked_matches = [ item for item in existing if str(item["transaction_id"]) == confirmed_link_id and str(item["transaction_id"]) not in used_existing_ids and ( str(item["trade_date"]), format(_decimal(item["amount"]).normalize(), "f"), str(item["currency_original"]).upper(), ) == economic ] if len(linked_matches) == 1: existing_row = linked_matches[0] reasons.append("economic_collision_linked_by_confirmed_lineage") economic_collision = False else: quality = "uncertain" economic_collision = False base["overall_wealth_treatment"] = "not_applied_pending_review" base["truewealth_scope_treatment"] = "not_applied_pending_review" base["household_treatment"] = "unchanged_pending_review" reasons.append("confirmed_collision_link_invalid") if economic_collision and decision and quality != "secure": raise ValueError("Collision decision cannot substitute for stable recipient identity") if economic_collision and decision: applied_decisions.add(base["bank_transaction_id"]) economic_matches = [ item for item in existing_by_economic[economic] if str(item["transaction_id"]) not in used_existing_ids ] if decision == "link_existing": if len(economic_matches) != 1: raise ValueError("link_existing requires one unambiguous existing cashflow") existing_row = economic_matches[0] reasons.append("economic_collision_linked_by_explicit_decision") else: reasons.append("economic_collision_recorded_distinct_by_explicit_decision") quality = "secure" base["overall_wealth_treatment"] = "internal_wealth_transfer" base["truewealth_scope_treatment"] = "external_deposit" base["household_treatment"] = "not_expense_not_income" economic_collision = False if existing_row is not None: used_existing_ids.add(str(existing_row["transaction_id"])) if economic_collision and not duplicate_bank_row: quality = "uncertain" base["overall_wealth_treatment"] = "not_applied_pending_review" base["truewealth_scope_treatment"] = "not_applied_pending_review" base["household_treatment"] = "unchanged_pending_review" reasons.append("economic_collision_without_exact_lineage") disposition = ( "duplicate" if duplicate_bank_row else "existing" if existing_row else "new" if quality == "secure" else "unclear" ) candidates.append( { **base, "match_quality": quality, "recognition_reasons": sorted(reasons or ["recipient_text_partial_only"]), "disposition": disposition, "existing_cashflow_id": str(existing_row["transaction_id"]) if existing_row else None, } ) unused_decisions = set(decisions) - applied_decisions if unused_decisions: raise ValueError("Collision decision does not match a current unresolved economic collision") secure = [row for row in candidates if row["match_quality"] == "secure"] uncertain = [row for row in candidates if row["match_quality"] == "uncertain"] write_rows = [row for row in secure if row["disposition"] == "new"] amounts = sorted({row["amount"] for row in secure}, key=Decimal) amount_change = { "detected": len(amounts) > 1, "amounts": amounts, "rule_uses_amount": False, } fingerprint_inputs = { "rule": rule, "collision_decisions": decisions, "confirmed_collision_links": confirmed_collision_links, "period_from": start, "period_to": end, "bank_rows": bank_rows, "existing": existing, } fingerprint = _fingerprint(fingerprint_inputs) return { "preview_id": f"tw-bank-preview-{fingerprint[:24]}", "input_fingerprint": fingerprint, "account_id": account_id, "period_from": start, "period_to": end, "rule_fingerprint": _fingerprint(rule), "rule_status": "preview_only_not_activated", "candidates": candidates, "secure_candidates": secure, "uncertain_candidates": uncertain, "excluded_candidates": excluded, "cashflows_to_write": write_rows, "counts": { "secure": len(secure), "uncertain": len(uncertain), "excluded": len(excluded), "new": len(write_rows), "existing": sum(row["disposition"] == "existing" for row in secure), "duplicates": sum(row["disposition"] == "duplicate" for row in secure), }, "detected_amount_change": amount_change, "covered_period": {"from": start, "to": end}, "can_confirm": bool(secure) and not uncertain, "planned_changes": { "recipient_rule_audit": 1, "cashflow_transactions": len(write_rows), "coverage_records": 0, "rule_activations": 0, "valuation_snapshots": 0, }, "reason_codes": (["uncertain_recipient_matches_require_review"] if uncertain else []) + (["no_secure_recipient_matches"] if not secure else []), } def _insert_confirmed_cashflows(conn: Connection, preview: dict[str, Any], confirmation_id: str) -> int: now = utc_now() written = 0 for row in preview["cashflows_to_write"]: lineage = conn.execute( """SELECT c.raw_fingerprint FROM budget_transactions b JOIN budget_transaction_candidates c ON c.transaction_candidate_id=b.source_candidate_id WHERE b.budget_transaction_id=?""", (row["bank_transaction_id"],), ).fetchone() if not lineage or not str(lineage["raw_fingerprint"] or ""): raise ValueError("Confirmed bank cashflow requires stable source lineage") raw_fingerprint = str(lineage["raw_fingerprint"]) economic = _fingerprint( { "account_id": preview["account_id"], "bank_transaction_id": row["bank_transaction_id"], "raw_fingerprint": raw_fingerprint, "date": row["value_date"], "amount": row["amount"], "currency": row["currency"], } ) conn.execute( """INSERT INTO transactions( transaction_id,transaction_type,activity_kind,account_id,trade_date,booking_date,event_timestamp, gross_amount_original,net_amount_original,currency_original,fx_rate_to_chf,fx_source,fx_status, gross_amount_chf,net_amount_chf,source_type,source_id,external_transaction_id,row_hash, is_confirmed,quality_status,notes,source_reference,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( stable_id("tw-bank-flow", economic), "external_deposit", "external_deposit", preview["account_id"], row["value_date"], row["booking_date"], row["value_date"], row["amount"], row["amount"], row["currency"], "1" if row["currency"] == "CHF" else None, "identity" if row["currency"] == "CHF" else None, "ok" if row["currency"] == "CHF" else "missing", row["amount"] if row["currency"] == "CHF" else None, row["amount"] if row["currency"] == "CHF" else None, SOURCE, confirmation_id, raw_fingerprint, economic, 1, "complete" if row["currency"] == "CHF" else "partial", "Owner-confirmed Raiffeisen to True Wealth transfer; bank value date proxy", row["bank_transaction_id"], now, ), ) written += int(conn.execute("SELECT changes()").fetchone()[0]) return written def _preview_from_request(conn: Connection, request: dict[str, Any]) -> dict[str, Any]: return preview_truewealth_bank_payments( conn, period_from=str(request.get("period_from") or ""), period_to=str(request.get("period_to") or ""), recipient_name=str(request.get("recipient_name") or ""), recipient_account_identity=( str(request["recipient_account_identity"]) if request.get("recipient_account_identity") else None ), counterparty_hashes=[str(value) for value in request.get("counterparty_hashes") or []], reference_tokens=[str(value) for value in request.get("reference_tokens") or []], attestation=str(request.get("attestation") or ""), collision_decisions={ str(key): str(value) for key, value in (request.get("collision_decisions") or {}).items() }, ) def confirm_truewealth_bank_payments(conn: Connection, request: dict[str, Any]) -> dict[str, Any]: if request.get("confirm") is not True: raise ValueError("True Wealth bank cashflow confirmation requires confirm=true") confirmation_id = str(request.get("confirmation_id") or "").strip() if not confirmation_id: raise ValueError("True Wealth bank cashflow confirmation requires confirmation_id") request_fingerprint = _fingerprint( { key: request.get(key) for key in ( "period_from", "period_to", "recipient_name", "recipient_account_identity", "counterparty_hashes", "reference_tokens", "collision_decisions", "attestation", "preview_id", "input_fingerprint", ) } ) previous = conn.execute( """SELECT audit_id,new_values_json FROM audit_log WHERE source=? AND action='recipient_rule_and_history_confirmed' AND entity_id=? ORDER BY created_at DESC LIMIT 1""", (SOURCE, confirmation_id), ).fetchone() if previous: stored = json.loads(str(previous["new_values_json"] or "{}")) if ( stored.get("input_fingerprint") != request.get("input_fingerprint") or stored.get("confirmation_request_fingerprint") != request_fingerprint ): raise ValueError("confirmation_id was already used with a different payload") values = { key: stored[key] for key in ( "confirmation_id", "input_fingerprint", "rule_fingerprint", "account_id", "period_from", "period_to", "written_cashflows", "coverage_activated", "rule_activated", ) } return {**values, "audit_id": str(previous["audit_id"]), "idempotent": True} preview = _preview_from_request(conn, request) if preview["preview_id"] != request.get("preview_id") or preview["input_fingerprint"] != request.get("input_fingerprint"): raise ValueError("True Wealth bank cashflow preview is stale") if not preview["can_confirm"]: raise ValueError("True Wealth bank cashflow preview contains unresolved matches") try: conn.execute("BEGIN IMMEDIATE") locked_previous = conn.execute( """SELECT audit_id,new_values_json FROM audit_log WHERE source=? AND action='recipient_rule_and_history_confirmed' AND entity_id=? ORDER BY created_at DESC LIMIT 1""", (SOURCE, confirmation_id), ).fetchone() if locked_previous: stored = json.loads(str(locked_previous["new_values_json"] or "{}")) if ( stored.get("input_fingerprint") != request.get("input_fingerprint") or stored.get("confirmation_request_fingerprint") != request_fingerprint ): raise ValueError("confirmation_id was already used with a different payload") conn.commit() values = { key: stored[key] for key in ( "confirmation_id", "input_fingerprint", "rule_fingerprint", "account_id", "period_from", "period_to", "written_cashflows", "coverage_activated", "rule_activated", ) } return {**values, "audit_id": str(locked_previous["audit_id"]), "idempotent": True} locked = _preview_from_request(conn, request) if locked["input_fingerprint"] != request.get("input_fingerprint"): raise ValueError("True Wealth bank cashflow preview became stale before confirmation") written = _insert_confirmed_cashflows(conn, locked, confirmation_id) operational_rule = { "recipient_name": "", "recipient_account_identity": _normalise(request.get("recipient_account_identity")), "counterparty_hashes": sorted({str(value).strip().lower() for value in request.get("counterparty_hashes") or []}), "reference_tokens": sorted({_normalise(value) for value in request.get("reference_tokens") or []}), } values = { "confirmation_id": confirmation_id, "input_fingerprint": locked["input_fingerprint"], "rule_fingerprint": locked["rule_fingerprint"], "account_id": locked["account_id"], "period_from": locked["period_from"], "period_to": locked["period_to"], "written_cashflows": written, "coverage_activated": False, "rule_activated": False, } audit_id = record_audit_event( conn, source=SOURCE, action="recipient_rule_and_history_confirmed", entity_type="truewealth_recipient_rule", entity_id=confirmation_id, old_values={}, new_values={ **values, "confirmation_request_fingerprint": request_fingerprint, "operational_rule": operational_rule, "confirmed_candidate_ids": [row["bank_transaction_id"] for row in locked["secure_candidates"]], "collision_links": [ { "bank_transaction_id": row["bank_transaction_id"], "existing_cashflow_id": row["existing_cashflow_id"], } for row in locked["secure_candidates"] if row.get("existing_cashflow_id") and "economic_collision_linked_by_explicit_decision" in row["recognition_reasons"] ], "confirmed_summary": { "candidate_count": len(locked["secure_candidates"]), "written_count": written, "amount_total_chf": format( sum((Decimal(row["amount"]) for row in locked["cashflows_to_write"]), Decimal("0")), "f", ), }, }, user_text_note="Explicit owner confirmation recorded; raw attestation redacted", confirmed=True, created_by="user", ) conn.commit() except Exception: conn.rollback() raise return {**values, "audit_id": audit_id, "idempotent": False} def activate_truewealth_recipient_rule(conn: Connection, request: dict[str, Any]) -> dict[str, Any]: """Separate explicit activation; confirmation of historical rows never activates automation.""" if request.get("confirm") is not True: raise ValueError("True Wealth recipient-rule activation requires confirm=true") rule_fingerprint = str(request.get("rule_fingerprint") or "") activation_id = str(request.get("activation_id") or "").strip() confirmed = conn.execute( """SELECT new_values_json FROM audit_log WHERE source=? AND action='recipient_rule_and_history_confirmed' ORDER BY created_at DESC""", (SOURCE,), ).fetchall() matching = [json.loads(str(row[0] or "{}")) for row in confirmed] confirmed_rule = next( (item for item in matching if item.get("rule_fingerprint") == rule_fingerprint), None, ) if confirmed_rule is None: raise ValueError("Recipient rule must be confirmed before activation") existing = conn.execute( "SELECT audit_id,new_values_json FROM audit_log WHERE source=? AND action='recipient_rule_activated' AND entity_id=?", (SOURCE, activation_id), ).fetchone() if existing: stored = json.loads(str(existing["new_values_json"] or "{}")) if stored.get("rule_fingerprint") != rule_fingerprint: raise ValueError("activation_id was already used with a different rule") values = { "activation_id": stored["activation_id"], "rule_fingerprint": stored["rule_fingerprint"], "active": True, } return {**values, "audit_id": str(existing["audit_id"]), "idempotent": True} try: conn.execute("BEGIN IMMEDIATE") locked_existing = conn.execute( "SELECT audit_id,new_values_json FROM audit_log WHERE source=? AND action='recipient_rule_activated' AND entity_id=?", (SOURCE, activation_id), ).fetchone() if locked_existing: stored = json.loads(str(locked_existing["new_values_json"] or "{}")) if stored.get("rule_fingerprint") != rule_fingerprint: raise ValueError("activation_id was already used with a different rule") conn.commit() values = {"activation_id": stored["activation_id"], "rule_fingerprint": stored["rule_fingerprint"], "active": True} return {**values, "audit_id": str(locked_existing["audit_id"]), "idempotent": True} values = {"activation_id": activation_id, "rule_fingerprint": rule_fingerprint, "active": True} audit_values = { **values, "account_id": confirmed_rule["account_id"], "active_after": confirmed_rule["period_to"], "rule": confirmed_rule["operational_rule"], } audit_id = record_audit_event( conn, source=SOURCE, action="recipient_rule_activated", entity_type="truewealth_recipient_rule", entity_id=activation_id, old_values={}, new_values=audit_values, confirmed=True, created_by="user", ) conn.commit() except Exception: conn.rollback() raise return {**values, "audit_id": audit_id, "idempotent": False} def process_activated_truewealth_bank_payments( conn: Connection, *, as_of: str | None, ) -> dict[str, Any]: """Apply only unambiguous future rows for the explicitly activated exact rule.""" day = ( date.fromisoformat(as_of) if as_of else datetime.now(ZoneInfo("Europe/Zurich")).date() ).isoformat() active_row = conn.execute( """SELECT entity_id,new_values_json FROM audit_log WHERE source=? AND action='recipient_rule_activated' AND confirmed=1 ORDER BY created_at DESC,audit_id DESC LIMIT 1""", (SOURCE,), ).fetchone() if not active_row: return { "status": "activation_required", "written_cashflows": 0, "reason_codes": ["truewealth_recipient_rule_activation_required"], } active = json.loads(str(active_row["new_values_json"] or "{}")) if not active.get("active") or not active.get("rule") or not active.get("active_after"): return { "status": "blocked", "written_cashflows": 0, "reason_codes": ["activated_truewealth_rule_evidence_incomplete"], } start = (date.fromisoformat(str(active["active_after"])) + timedelta(days=1)).isoformat() if start > day: return {"status": "no_candidates", "written_cashflows": 0, "reason_codes": []} rule = dict(active["rule"]) preview = preview_truewealth_bank_payments( conn, period_from=start, period_to=day, recipient_name=str(rule.get("recipient_name") or ""), recipient_account_identity=rule.get("recipient_account_identity"), counterparty_hashes=list(rule.get("counterparty_hashes") or []), reference_tokens=list(rule.get("reference_tokens") or []), attestation="Diese Empfängeridentität gehört zum persönlichen True-Wealth-Portfolio; Aktivierung zuvor bestätigt.", ) if preview["uncertain_candidates"]: return { "status": "blocked", "written_cashflows": 0, "reason_codes": ["future_truewealth_matches_require_review"], "input_fingerprint": preview["input_fingerprint"], } if not preview["cashflows_to_write"]: return { "status": "no_candidates", "written_cashflows": 0, "reason_codes": [], "input_fingerprint": preview["input_fingerprint"], } automation_id = stable_id( "tw-active-rule-run", str(active_row["entity_id"]), preview["input_fingerprint"], ) existing = conn.execute( """SELECT audit_id,new_values_json FROM audit_log WHERE source=? AND action='recipient_rule_future_rows_applied' AND entity_id=?""", (SOURCE, automation_id), ).fetchone() if existing: stored = json.loads(str(existing["new_values_json"] or "{}")) return {**stored, "audit_id": str(existing["audit_id"]), "idempotent": True} try: conn.execute("BEGIN IMMEDIATE") locked = preview_truewealth_bank_payments( conn, period_from=start, period_to=day, recipient_name=str(rule.get("recipient_name") or ""), recipient_account_identity=rule.get("recipient_account_identity"), counterparty_hashes=list(rule.get("counterparty_hashes") or []), reference_tokens=list(rule.get("reference_tokens") or []), attestation="Diese Empfängeridentität gehört zum persönlichen True-Wealth-Portfolio; Aktivierung zuvor bestätigt.", ) if locked["input_fingerprint"] != preview["input_fingerprint"] or locked["uncertain_candidates"]: raise ValueError("Activated True Wealth rule inputs became stale or ambiguous") written = _insert_confirmed_cashflows(conn, locked, automation_id) values = { "status": "complete", "written_cashflows": written, "input_fingerprint": locked["input_fingerprint"], "rule_fingerprint": active["rule_fingerprint"], "coverage_activated": False, } audit_id = record_audit_event( conn, source=SOURCE, action="recipient_rule_future_rows_applied", entity_type="truewealth_recipient_rule", entity_id=automation_id, old_values={}, new_values=values, confirmed=True, created_by="system", ) conn.commit() except Exception: conn.rollback() raise return {**values, "audit_id": audit_id, "idempotent": False}