""" AUTH.4C3 — save_reconciliation.py ================================= Read-only OUTCOME_UNKNOWN-RECONCILIATION-CONTRACT für den SAVE-Executor. Schließt den Contract Gap: OUTCOME_UNKNOWN + Read-Back beweist exakt, dass das gewünschte Ziel bereits erreicht wurde -> terminal bestätigter Erfolg, OHNE zweiten SAVE, OHNE Retry, OHNE manuellen SUCCEEDED-State, OHNE SQL, OHNE Überschreiben historischer Unsicherheit. CORE SAFETY INVARIANT: OUTCOME_UNKNOWN bedeutet: der Executor darf NICHT wissen, ob die Mutation stattgefunden hat. Daher darf OUTCOME_UNKNOWN NIEMALS direkt automatisch erneut mutieren. Einzige sichere nächste Aktion: READ-ONLY RECONCILIATION. Erst wenn die autoritative Zielseite exakt beweist TARGET_ALREADY_REACHED, darf der Job ohne neue Mutation terminal bestätigt werden. KRITISCH: Dieses Modul besitzt KEIN write-Callable. Es gibt KEINE tolaria_save / write()-Dependency Injection. Reconciliation ist technisch unfähig zu schreiben. Dependencies: source_loader, read_back, job_store, audit. Isoliert implementiert (KEIN produktiver Container). Nutzt job_store + job_state_machine + job_schema. """ from __future__ import annotations import hashlib from typing import Any, Callable, Dict, Optional from job_schema import JOB_TYPE_SAVE from job_store import JobStore from job_state_machine import ( ST_OUTCOME_UNKNOWN, ST_RECONCILED, ST_SUCCEEDED, ) # --------------------------------------------------------------------------- # Reconciliation-Ergebnis-Klassifikation # --------------------------------------------------------------------------- # Nur TARGET_EXACT darf terminalen Erfolg bestätigen. RC_TARGET_EXACT = "TARGET_EXACT" RC_TARGET_ABSENT = "TARGET_ABSENT" RC_TARGET_MISMATCH = "TARGET_MISMATCH" RC_READ_UNAVAILABLE = "READ_UNAVAILABLE" RC_INVALID_JOB = "INVALID_JOB" RC_PROVENANCE_MISMATCH = "PROVENANCE_MISMATCH" RC_ALREADY_RECONCILED = "ALREADY_RECONCILED" # Audit-Events (AUTH.4C3) EV_RECONCILIATION_STARTED = "RECONCILIATION_STARTED" EV_RECONCILIATION_TARGET_EXACT = "RECONCILIATION_TARGET_EXACT" EV_CONFIRMED_EXECUTED = "CONFIRMED_EXECUTED" # Immutable Job-Felder (müssen nach OUTCOME_UNKNOWN unveränderlich sein) IMMUTABLE_RECONCILIATION_FIELDS = frozenset({ "object_id", "vault_path", "source_commit", "provenance_hash", "idempotency_key", "mission_id", }) class ReconciliationError(Exception): """Basis-Fehler für Reconciliation.""" class ReconciliationWriteError(ReconciliationError): """Wird geworfen, wenn Reconciliation versucht zu schreiben (darf nie passieren).""" class SaveReconciliationCore: """ Read-only Reconciliation-Core für OUTCOME_UNKNOWN-SAVE-Jobs. source_loader: Callable[[str, str, str], str] — lädt Content aus autoritativer Source (source_commit, object_id, vault_path). Identisch zu SaveExecutorCore. read_back: Callable[[str], Optional[str]] — liest den aktuellen Tolaria-Target (vault_path) -> content oder None. Credential-frei (Least Privilege). store: JobStore — für State-Transition + Audit. KEIN tolaria_save / write-Callable. Dieses Modul kann technisch NICHT schreiben. """ def __init__( self, store: JobStore, source_loader: Callable[[str, str, str], str], read_back: Callable[[str], Optional[str]], ): if store.worker_scope != "SAVE": raise ReconciliationError( "SaveReconciliationCore requires worker_scope='SAVE'") self.store = store self.source_loader = source_loader self.read_back = read_back # -- Hauptverarbeitung -------------------------------------------------- def reconcile(self, job_id: str, worker_id: str) -> Dict[str, Any]: """ Führt die read-only Reconciliation eines OUTCOME_UNKNOWN-SAVE-Jobs aus. Ablauf: 1. Job laden (muss existieren) 2. State prüfen (muss OUTCOME_UNKNOWN sein) 3. Job-Type prüfen (nur SAVE) 4. Immutable Felder validieren 5. Content aus autoritativer Source rekonstruieren (exact commit) 6. Provenance/Hash recompute + validieren 7. Tolaria-Target lesen (credential-frei) 8. Exakt vergleichen -> klassifizieren 9. Nur TARGET_EXACT -> RECONCILED -> SUCCEEDED (KEIN SAVE) Returns: finaler Job-State. """ job = self.store.get_job(job_id) if job is None: raise ReconciliationError(f"job not found: {job_id}") # State muss OUTCOME_UNKNOWN sein (nur dieser Zustand ist reconciliierbar) if job["state"] != ST_OUTCOME_UNKNOWN: # Idempotenz: bereits RECONCILED -> finalisiere zu SUCCEEDED if job["state"] == ST_RECONCILED: return self._finalize_reconciled(job_id, worker_id) # SUCCEEDED bleibt terminal (kein State-Rückschritt) if job["state"] == ST_SUCCEEDED: return job # Andere States (READY/CLAIMED/EXECUTING/FAILED/REJECTED/CREATED) # dürfen den Recovery-Pfad NICHT nutzen. raise ReconciliationError( f"job {job_id} state={job['state']} not reconciliable (must be OUTCOME_UNKNOWN)") # Job-Type: nur SAVE if job["job_type"] != JOB_TYPE_SAVE: raise ReconciliationError( f"job {job_id} job_type={job['job_type']} not reconciliable (must be SAVE)") # Immutable Felder validieren (müssen vorhanden + konsistent sein) self._validate_immutable_fields(job) # Audit: Reconciliation beginnt (read-only) self.store.record_audit_event( job_id, EV_RECONCILIATION_STARTED, worker_id=worker_id) # 1. Content aus autoritativer Source rekonstruieren (exact commit) try: content = self.source_loader( job["source_commit"], job["object_id"], job["vault_path"]) except Exception: # Source nicht verfügbar -> READ_UNAVAILABLE (kein Erfolg) return self._classify(job_id, RC_READ_UNAVAILABLE, worker_id) # 2. Provenance/Hash recompute + validieren if not self._validate_provenance(job, content): return self._classify(job_id, RC_PROVENANCE_MISMATCH, worker_id) # 3. Tolaria-Target lesen (credential-frei) try: target_content = self.read_back(job["vault_path"]) except Exception: # Read-Back nicht verfügbar -> READ_UNAVAILABLE (kein Erfolg) return self._classify(job_id, RC_READ_UNAVAILABLE, worker_id) # 4. Exakt vergleichen -> klassifizieren if target_content is None: # Target ABSENT -> NICHT automatisch SAVE wiederholen return self._classify(job_id, RC_TARGET_ABSENT, worker_id) if target_content != content: # Target MISMATCH -> FAIL CLOSED (kein Überschreiben, kein Repair) return self._classify(job_id, RC_TARGET_MISMATCH, worker_id) # 5. TARGET_EXACT: alle Felder exakt bestätigt return self._confirm_target_exact(job_id, worker_id) # -- TARGET_EXACT-Bestätigung ------------------------------------------- def _confirm_target_exact(self, job_id: str, worker_id: str) -> Dict[str, Any]: """ TARGET_EXACT bestätigt: KEIN SAVE. Job wird regulär finalisiert. OUTCOME_UNKNOWN -> RECONCILED -> SUCCEEDED. """ # Audit: TARGET_EXACT bestätigt (read-only) self.store.record_audit_event( job_id, EV_RECONCILIATION_TARGET_EXACT, worker_id=worker_id) # OUTCOME_UNKNOWN -> RECONCILED (persistenter Zwischenzustand) self.store.mark_reconciled(job_id, worker_id) # RECONCILED -> SUCCEEDED (terminal, KEINE neue Mutation) self.store.mark_succeeded(job_id, worker_id) # Audit: Mutation als BEREITS ausgeführt bestätigt (NICHT neu ausgeführt) self.store.record_audit_event( job_id, EV_CONFIRMED_EXECUTED, worker_id=worker_id) return self.store.get_job(job_id) # -- Idempotenz: bereits RECONCILED ------------------------------------- def _finalize_reconciled(self, job_id: str, worker_id: str) -> Dict[str, Any]: """ Job ist bereits RECONCILED (Crash nach RECONCILED vor SUCCEEDED, oder wiederholte Reconciliation). Finalisiere zu SUCCEEDED. Kein State-Rückschritt, kein neues EXECUTED, kein SAVE. """ # Audit: ALREADY_RECONCILED (idempotent) self.store.record_audit_event( job_id, EV_RECONCILIATION_TARGET_EXACT, worker_id=worker_id) self.store.mark_succeeded(job_id, worker_id) self.store.record_audit_event( job_id, EV_CONFIRMED_EXECUTED, worker_id=worker_id) return self.store.get_job(job_id) # -- Klassifikation (kein Erfolg) --------------------------------------- def _classify(self, job_id: str, result: str, worker_id: str) -> Dict[str, Any]: """ Nicht-TARGET_EXACT-Ergebnis: Job bleibt OUTCOME_UNKNOWN (kein State-Change). Kein automatischer Retry. Human Review. Der result_code wird im Job gesetzt (Diagnose-Information), der State bleibt unverändert. """ # Audit: Reconciliation-Ergebnis (kein Erfolg, kein State-Change) self.store.record_audit_event( job_id, f"RECONCILIATION_{result}", worker_id=worker_id, result_code=result) # result_code im Job setzen (kein State-Change) self.store.set_result_code(job_id, result) return self.store.get_job(job_id) # -- Validierung -------------------------------------------------------- def _validate_immutable_fields(self, job: Dict[str, Any]) -> None: """Prüft, dass alle immutable Reconciliation-Felder vorhanden sind.""" missing = IMMUTABLE_RECONCILIATION_FIELDS - set(job.keys()) if missing: raise ReconciliationError( f"job {job['job_id']} missing immutable fields: {sorted(missing)}") def _validate_provenance(self, job: Dict[str, Any], content: str) -> bool: """ RECOMPUTE: berechnet den Provenance-Hash aus dem rekonstruierten Content und vergleicht mit dem Job-Feld. RQ darf den Hash nicht blind bestimmen. """ computed = hashlib.sha256(content.encode("utf-8")).hexdigest() return computed == job["provenance_hash"]