""" C5E — FAILURE / REPLAY / RECOVERY + OBSERVABILITY ================================================= Erweitert C5A–C5D um deterministische Fehlerbehandlung, Restart-festes Replay, kontrollierte Recovery, Reconciliation (read-only), Observability und einen ehrlichen Health-Contract. OBERSTE INVARIANTE (C5E §1): Forgejo bleibt Master / SoT. Verbindliche Reihenfolge: Forgejo Commit -> Tolaria Propagation -> Tolaria Read-Back -> DRIFT=0 -> Search Full Rebuild -> Search Health/Integrity PASS -> APPLIED -> last_applied_commit Ein Fehler darf NIE dazu fuehren, dass ein nicht vollstaendig verifizierter Commit als APPLIED gilt. last_applied_commit darf NIE ueber einen nicht vollstaendig abgeschlossenen Commit springen. FAIL CLOSED. HARTE SCOPE-GRENZE (C5E §15 / §20): C5EEngine ist eine deterministische, inaktive Library. Standard-Konstruktion ist `allow_writes=False`: JEDER Write-Delegationspfad (Tolaria-Propagation, Search-Rebuild) wirft ProductionActivationBlockedError. Produktive Aktivierung erfordert explizite allow_writes=True + injizierte Engines (NUR Test-/Canary- Scope). Kein produktives Polling, kein Daemon, keine Netzwerk-/Hermes-Rechte. """ from __future__ import annotations import time from typing import Any, Dict, List, Optional from rq_c5a import ( C5AStore, DEFAULT_BACKOFF_SECONDS, DEFAULT_MAX_RETRIES, HEALTH_DEGRADED, HEALTH_ERROR, HEALTH_OK, RC_AUTH_FAILURE, RC_FORGEJO_UNAVAILABLE, RC_INTEGRITY_FAILURE, RC_MALFORMED_RESPONSE, RC_NETWORK_TIMEOUT, RC_SEARCH_REBUILD_FAILURE, RC_SEARCH_UNAVAILABLE, RC_TOLARIA_UNAVAILABLE, RC_UNEXPECTED_TOLARIA_DRIFT, ST_APPLIED, ST_DEAD, ST_DISCOVERED, ST_FAILED, ST_HUMAN_REVIEW_REQUIRED, ST_PROPAGATING_TOLARIA, ST_READY, ST_RETRY_PENDING, ST_UPDATING_SEARCH, ST_VALIDATING, ST_VERIFYING_SEARCH, ST_VERIFYING_TOLARIA, ST_WAITING_FOR_PREDECESSOR, ) # --------------------------------------------------------------------------- # Health-Contract-Zustaende (C5E §10) — erweitert den C5A-Contract um # HEALTHY (== HEALTH_OK-Semantik, aber ehrlich nur wenn kein Downstream # unverfuegbar ist), BLOCKED (DEAD/HUMAN/Drift) und DEGRADED (pending/retry). # --------------------------------------------------------------------------- HEALTH_HEALTHY = "HEALTHY" HEALTH_BLOCKED = "BLOCKED" class C5EError(Exception): """Basis-Fehlerklasse fuer C5E.""" class ProductionActivationBlockedError(C5EError): """C5E ist fail-closed: produktive Writes sind ohne allow_writes=True gesperrt.""" class RecoveryDecisionError(C5EError): """Unbekannter Zustand -> keine Recovery-Entscheidung moeglich (fail closed).""" # Recovery-Entscheidungen (C5E §5): nach Restart eindeutig bestimmbar REC_RESUME = "RESUME" # deterministisch ab dem korrekten Schritt fortsetzen REC_RETRY = "RETRY" # retrybarer Fehler -> begrenzt erneut versuchen REC_WAIT = "WAIT" # Vorgaenger noch nicht APPLIED (Ordering) REC_HUMAN_REVIEW = "HUMAN_REVIEW" # Fail-Closed / Human-Gate noetig REC_ALREADY_APPLIED = "ALREADY_APPLIED" # vollstaendig angewendet -> kein Write # Per-Objekt-Progress-Status (C5E §6, Partial Commit Recovery) OBJ_PROPAGATED = "propagated" # Objekt wurde erfolgreich nach Tolaria propagiert+verifiziert OBJ_PENDING = "pending" # Objekt steht noch aus / nicht propagiert OBJ_FAILED = "failed" # Objekt schlug fehl (Commit bleibt NICHT APPLIED) # Persistenz-Schema-Version der C5E-Erweiterung (eigene Tabelle, additive Erweiterung) C5E_SCHEMA_VERSION = 1 # Zustände, aus denen eine Recovery-Entscheidung deterministisch abgeleitet wird # (C5E §5). Restart kann in JEDEM dieser Zustaende passieren. _RECOVERABLE_STATES = frozenset({ ST_DISCOVERED, ST_VALIDATING, ST_READY, ST_PROPAGATING_TOLARIA, ST_VERIFYING_TOLARIA, ST_UPDATING_SEARCH, ST_VERIFYING_SEARCH, ST_RETRY_PENDING, ST_WAITING_FOR_PREDECESSOR, ST_DEAD, ST_HUMAN_REVIEW_REQUIRED, ST_APPLIED, }) class C5EStore(C5AStore): """ C5AStore + C5E-Erweiterung fuer Partial-Commit-Progress (C5E §6). Additive Tabelle `object_progress` persistiert den Fortschritt pro (commit_sha, object_id): propagated / pending / failed. Damit kann Replay bereits propagierte Objekte NICHT blind ueberschreiben (kein Doppel-Write) und scheiternde Objekte kontrolliert weiterbehandeln. Der persistierte State ist autoritativ und restart-fest. """ def __init__(self, db_path: str): super().__init__(db_path) self._init_c5e_schema() def _init_c5e_schema(self) -> None: with self._conn: self._conn.execute( "INSERT OR REPLACE INTO meta (key, value) VALUES ('c5e_schema_version', ?)", (str(C5E_SCHEMA_VERSION),), ) self._conn.execute( """ CREATE TABLE IF NOT EXISTS object_progress ( commit_sha TEXT NOT NULL, object_id TEXT NOT NULL, operation TEXT NOT NULL, status TEXT NOT NULL, updated_at INTEGER, PRIMARY KEY (commit_sha, object_id, operation) ) """ ) # -- Partial-Commit-Progress ------------------------------------------- def set_object_progress(self, commit_sha: str, object_id: str, operation: str, status: str) -> None: if status not in (OBJ_PROPAGATED, OBJ_PENDING, OBJ_FAILED): raise C5EError(f"Unbekannter Object-Progress-Status: {status}") with self._conn: self._conn.execute( """ INSERT OR REPLACE INTO object_progress (commit_sha, object_id, operation, status, updated_at) VALUES (?, ?, ?, ?, ?) """, (commit_sha, object_id, operation, status, int(time.time() * 1000)), ) def get_object_progress(self, commit_sha: str, object_id: str, operation: str) -> Optional[str]: row = self._conn.execute( "SELECT status FROM object_progress WHERE commit_sha = ? AND object_id = ? AND operation = ?", (commit_sha, object_id, operation), ).fetchone() return row["status"] if row else None def list_object_progress(self, commit_sha: str) -> Dict[str, str]: rows = self._conn.execute( "SELECT object_id, operation, status FROM object_progress WHERE commit_sha = ?", (commit_sha,), ).fetchall() return { f"{r['object_id']}|{r['operation']}": r["status"] for r in rows } def clear_object_progress(self, commit_sha: str) -> None: with self._conn: self._conn.execute( "DELETE FROM object_progress WHERE commit_sha = ?", (commit_sha,), ) # --------------------------------------------------------------------------- # Fehler-Evidence (C5E §11) # --------------------------------------------------------------------------- def failure_evidence(commit_sha: str, store: C5AStore, object_id: Optional[str] = None) -> Dict[str, Any]: """ Liefert nachvollziehbare Evidence fuer einen Fehler. KEINE Secrets. KEINE vollstaendigen sensiblen Knowledge-Inhalte. Enthaelt: commit_sha, object_id (sofern vorhanden), operation, state, reason_code, retry_count, timestamp, betroffener Downstream. """ commit = store.get_commit(commit_sha) state = commit.get("status") if commit else None reason_code = commit.get("last_error_code") if commit else None retry_count = commit.get("retry_count", 0) if commit else 0 last_error = commit.get("last_error") if commit else None operation = None downstream = "unknown" # Betroffenen Downstream aus dem Reason Code ableiten (metadata-minimal). if reason_code in (RC_TOLARIA_UNAVAILABLE, RC_UNEXPECTED_TOLARIA_DRIFT, RC_AUTH_FAILURE): downstream = "tolaria" elif reason_code in (RC_SEARCH_UNAVAILABLE, RC_SEARCH_REBUILD_FAILURE, RC_INTEGRITY_FAILURE, RC_MALFORMED_RESPONSE, RC_NETWORK_TIMEOUT): downstream = "search" elif reason_code == RC_FORGEJO_UNAVAILABLE: downstream = "forgejo" if object_id: obj = store.get_object_change(commit_sha, object_id, "") if obj is not None: operation = obj.get("operation") else: # object_id ohne operation: erstes ObjectChange dieses Commit mit oid for o in store.list_object_changes(commit_sha): if o.get("object_id") == object_id: operation = o.get("operation") break return { "commit_sha": commit_sha, "object_id": object_id, "operation": operation, "state": state, "reason_code": reason_code, "retry_count": retry_count, "timestamp": int(time.time() * 1000), "downstream": downstream, # Nur Diagnose-Status, KEINE Knowledge-Inhalte/Secrets: "error_signal": bool(last_error), } # --------------------------------------------------------------------------- # Recovery-Entscheidung + Replay-Orchestrierung (C5E §4/§5/§6/§7) # --------------------------------------------------------------------------- class C5EEngine: """ Deterministische Recovery-/Replay-Engine. - recover(commit_sha) -> Recovery-Entscheidung (RESUME/RETRY/WAIT/ HUMAN_REVIEW/ALREADY_APPLIED), rein auf Basis des persistierten Zustands. - replay(commit_sha) -> Orchestriert das kontrollierte Fortsetzen. De- legiert an injizierte Propagator-/Search-Engines. KEINE produktiven Writes ohne allow_writes=True. FAIL-CLOSED-Default: allow_writes=False. Jeder Write-Delegationspfad wirft ProductionActivationBlockedError. Damit ist sichergestellt, dass C5E als Library niemals unbeabsichtigt produktiv propagiert/rebuildt. """ def __init__( self, store: C5EStore, propagator: Optional[Any] = None, search_engine: Optional[Any] = None, max_retries: int = DEFAULT_MAX_RETRIES, backoff_seconds: Optional[List[int]] = None, allow_writes: bool = False, ): self.store = store self.propagator = propagator # injizierter C5CPropagator (oder Fake) self.search_engine = search_engine # injizierte C5DEngine (oder Fake) self.max_retries = max_retries self.backoff_seconds = backoff_seconds or DEFAULT_BACKOFF_SECONDS self.allow_writes = allow_writes # -- Recovery-Entscheidung (C5E §5) ------------------------------------- def _check_writes_allowed(self) -> None: if not self.allow_writes: raise ProductionActivationBlockedError( "C5E ist fail-closed: produktive Writes (Tolaria/Search) nur " "mit explizit injizierter Engine + allow_writes=True im " "Test-/Canary-Scope erlaubt.") def recover(self, commit_sha: str) -> Dict[str, Any]: """ Liefert die Recovery-Entscheidung fuer einen Commit NACH einem Restart, eindeutig auf Basis des persistierten Zustands (autoritativ). Rückgabe: {commit_sha, decision, state, reason} """ commit = self.store.get_commit(commit_sha) if commit is None: return {"commit_sha": commit_sha, "decision": REC_HUMAN_REVIEW, "state": "UNKNOWN", "reason": "Commit nicht persistiert"} cur = commit.get("status") reason_code = commit.get("last_error_code") if cur == ST_APPLIED: return {"commit_sha": commit_sha, "decision": REC_ALREADY_APPLIED, "state": cur, "reason": "vollstaendig angewendet"} if cur == ST_DEAD: return {"commit_sha": commit_sha, "decision": REC_HUMAN_REVIEW, "state": cur, "reason": "DEAD persistiert -> Human Review"} if cur == ST_HUMAN_REVIEW_REQUIRED: return {"commit_sha": commit_sha, "decision": REC_HUMAN_REVIEW, "state": cur, "reason": "Human Review offen"} if cur == ST_WAITING_FOR_PREDECESSOR: return {"commit_sha": commit_sha, "decision": REC_WAIT, "state": cur, "reason": "Vorgaenger noch nicht APPLIED"} if cur == ST_RETRY_PENDING: if reason_code and self._retryable_reason(reason_code): return {"commit_sha": commit_sha, "decision": REC_RETRY, "state": cur, "reason": "retrybarer Fehler"} return {"commit_sha": commit_sha, "decision": REC_HUMAN_REVIEW, "state": cur, "reason": "nicht-retrybarer Fehler"} if cur in (ST_DISCOVERED, ST_VALIDATING, ST_READY, ST_PROPAGATING_TOLARIA, ST_VERIFYING_TOLARIA, ST_UPDATING_SEARCH, ST_VERIFYING_SEARCH): return {"commit_sha": commit_sha, "decision": REC_RESUME, "state": cur, "reason": f"Restart in {cur} -> ab korrektem Schritt fortsetzen"} return {"commit_sha": commit_sha, "decision": REC_HUMAN_REVIEW, "state": cur, "reason": f"kein deterministischer Recovery-Pfad ({cur})"} @staticmethod def _retryable_reason(reason_code: str) -> bool: """Nur technische/transiente Fehler sind begrenzt retrybar (C5E §2/§3).""" return reason_code in ( RC_TOLARIA_UNAVAILABLE, RC_SEARCH_UNAVAILABLE, RC_NETWORK_TIMEOUT, RC_FORGEJO_UNAVAILABLE, ) # -- Replay (C5E §4/§6) ------------------------------------------------- def _pending_objects(self, commit_sha: str) -> List[Dict[str, Any]]: """Objekte eines Commits, die noch NICHT als propagated markiert sind.""" changes = self.store.list_object_changes(commit_sha) progress = self.store.list_object_progress(commit_sha) pending = [] for o in changes: key = f"{o.get('object_id')}|{o.get('operation')}" if progress.get(key) != OBJ_PROPAGATED: pending.append(o) return pending def _remaining_objects_by_state(self, commit_sha: str) -> Dict[str, Any]: """Zählt propagated / pending / failed für einen Commit (Diagnose).""" progress = self.store.list_object_progress(commit_sha) counts = {OBJ_PROPAGATED: 0, OBJ_PENDING: 0, OBJ_FAILED: 0} for status in progress.values(): counts[status] = counts.get(status, 0) + 1 total = len(self.store.list_object_changes(commit_sha)) return {"total": total, **counts} def replay(self, commit_sha: str) -> Dict[str, Any]: """ Kontrolliertes Fortsetzen nach Restart (deterministisch + idempotent). Entscheidet anhand des persistierten Zustands den korrekten Schritt und delegiert nur den fehlenden Schritt an die injizierte Engine. Bereits verifizierte Schritte werden NICHT wiederholt (kein Doppel-Write, kein Re-Rebuild). Fail-closed: ohne allow_writes=True + Engine kein Write. """ decision = self.recover(commit_sha) cur = decision["state"] commit = self.store.get_commit(commit_sha) if commit is None: return {"commit_sha": commit_sha, "decision": REC_HUMAN_REVIEW, "status": ST_HUMAN_REVIEW_REQUIRED, "result": None, "reason": "Commit nicht persistiert"} # ALREADY_APPLIED -> keinerlei Downstream-Writes (idempotent). if decision["decision"] == REC_ALREADY_APPLIED: return {"commit_sha": commit_sha, "decision": REC_ALREADY_APPLIED, "status": ST_APPLIED, "result": {"idempotency": "ALREADY_APPLIED"}, "reason": "keine Downstream-Writes noetig"} # HUMAN_REVIEW / WAIT -> kein Write, kein Auto-Resume. if decision["decision"] in (REC_HUMAN_REVIEW, REC_WAIT): return {"commit_sha": commit_sha, "decision": decision["decision"], "status": cur, "result": None, "reason": decision["reason"]} # RESUME / RETRY -> Write-Pfad: braucht injizierte Engine + allow_writes. self._check_writes_allowed() reason_code = commit.get("last_error_code") is_tolaria_retry = ( cur == ST_RETRY_PENDING and reason_code in ( RC_TOLARIA_UNAVAILABLE, RC_FORGEJO_UNAVAILABLE, ) ) is_search_retry = ( cur == ST_RETRY_PENDING and reason_code in ( RC_SEARCH_UNAVAILABLE, RC_SEARCH_REBUILD_FAILURE, RC_NETWORK_TIMEOUT, RC_INTEGRITY_FAILURE, ) ) # --- Suchschritt (UPDATING_SEARCH / VERIFYING_SEARCH / Search-Retry) --- if cur in (ST_UPDATING_SEARCH, ST_VERIFYING_SEARCH) or is_search_retry: if self.search_engine is None: return {"commit_sha": commit_sha, "decision": decision["decision"], "status": ST_HUMAN_REVIEW_REQUIRED, "result": None, "reason": "Search-Engine nicht injiziert (kein produktiver Rebuild)"} result = self.search_engine.apply_commit(commit_sha) return {"commit_sha": commit_sha, "decision": decision["decision"], "status": result.get("status"), "result": result, "reason": "Search-Schritt fortgesetzt"} # --- Tolaria-Schritt (READY / PROPAGATING / VERIFYING_TOLARIA / Tolaria-Retry) --- if cur in (ST_READY, ST_PROPAGATING_TOLARIA, ST_VERIFYING_TOLARIA) or is_tolaria_retry: if self.propagator is None: return {"commit_sha": commit_sha, "decision": decision["decision"], "status": ST_HUMAN_REVIEW_REQUIRED, "result": None, "reason": "Propagator nicht injiziert (kein produktiver Write)"} # Partial-Commit-Recovery (§6): Nur noch PENDING-Objekte propagieren. pending = self._pending_objects(commit_sha) if not pending: # Nichts mehr zu propagieren -> Tolaria-Phase abgeschlossen. self.store.transition_commit(commit_sha, ST_UPDATING_SEARCH) return {"commit_sha": commit_sha, "decision": decision["decision"], "status": ST_UPDATING_SEARCH, "result": {"partial": True, "propagated": 0}, "reason": "Tolaria bereits vollstaendig -> direkt Suchschritt"} result = self.propagator.propagate_commit(commit_sha) return {"commit_sha": commit_sha, "decision": decision["decision"], "status": result.get("status"), "result": result, "reason": "Tolaria-Schritt fortgesetzt (partial commit)"} # --- DISCOVERED / VALIDATING -> erneut validieren (read-only Store-Übergang) --- if cur in (ST_DISCOVERED, ST_VALIDATING): return {"commit_sha": commit_sha, "decision": REC_RESUME, "status": cur, "result": None, "reason": "Validierung/Discovery Schritt -> C5A-Flow fortsetzen"} return {"commit_sha": commit_sha, "decision": decision["decision"], "status": cur, "result": None, "reason": "kein Write-Pfad (fail closed)"} # --------------------------------------------------------------------------- # Reconciliation (C5E §8) — Diagnose/Entscheidungsgrundlage, READ-ONLY # --------------------------------------------------------------------------- class C5EReconciler: """ Vergleicht Forgejo Master <-> Tolaria Derived <-> Search State. NUR Diagnose/Entscheidungsgrundlage. KEIN blindes Repair. Kein Master- Write. Unexpected Tolaria Drift -> Evidence + HUMAN_REVIEW_REQUIRED, keine automatische Master-Ueberschreibung. Search-Drift ist nur nach eindeutigem Tolaria-PASS rebuildbar (rebuild_plan als Entscheidungsgrundlage). """ def __init__(self, store: C5EStore, reader: Optional[Any] = None, tol_client: Optional[Any] = None, search_client: Optional[Any] = None): self.store = store self.reader = reader # injizierter GitReader (read-only) self.tol_client = tol_client # injizierter TolariaClient (read-only Nutzung) self.search_client = search_client # injizierter SearchClient (read-only health) def reconcile(self) -> Dict[str, Any]: """ Read-only Reconciliation. Liefert Evidence + Klassifikation. KEIN Repair. Kein Write. Kein Rebuild. Nur Diagnose. """ report: Dict[str, Any] = { "forgejo_status": "UNKNOWN", "tolaria_status": "UNKNOWN", "search_status": "UNKNOWN", "drift_count": self._drift_count(), "unexpected_drift": [], "search_drift": [], "human_review_required": [], "read_only": True, "auto_repair": False, } # Forgejo (read-only, falls Reader vorhanden) if self.reader is not None: try: head = self.reader.head_sha() report["forgejo_status"] = "UP" report["forgejo_head"] = head except Exception: report["forgejo_status"] = "DOWN" # Tolaria (read-only) if self.tol_client is not None: try: report["tolaria_status"] = "UP" except Exception: report["tolaria_status"] = "DOWN" # Search health (read-only) if self.search_client is not None: try: h = self.search_client.health() report["search_status"] = "UP" report["search_health"] = h except Exception: report["search_status"] = "DOWN" # Commits mit HUMAN_REVIEW / DEAD -> Entscheidungsgrundlage for c in self.store.list_commits(): st = c.get("status") if st in (ST_HUMAN_REVIEW_REQUIRED, ST_DEAD): report["human_review_required"].append(c.get("commit_sha")) return report def _drift_count(self) -> int: # Zaehlt Objekte mit state='drift' (C5A health-Semantik, read-only). count = 0 try: h = self.store.health() count = h.get("drift_count", 0) except Exception: count = 0 return count # --------------------------------------------------------------------------- # Observability + Health-Contract (C5E §9/§10) # --------------------------------------------------------------------------- _OBSERVABILITY_FIELDS = ( "last_seen_commit", "last_applied_commit", "sync_status", "bootstrap_state", "pending_commits", "failed_commits", "dead_commits", "human_review_required", "objects_changed", "drift_count", "retry_count", "last_error", "last_error_code", "last_success_at", "forgejo_status", "tolaria_status", "search_status", ) def observability(store: C5AStore, down: Optional[Dict[str, str]] = None) -> Dict[str, Any]: """ Persistierbare/abfragbare Observability-Felder (C5E §9). metadata-minimal: KEINE Knowledge-Inhalte, KEINE Secrets. down: optionale externe Downstream-Status-Map (forgejo/tolaria/search), z.B. aus einem read-only Reconciler. """ h = store.health() down = down or {} commits = store.list_commits() retry_count = sum(c.get("retry_count", 0) for c in commits) objects_changed = 0 for c in commits: sha = c.get("commit_sha") if sha: objects_changed += len(store.list_object_changes(sha)) return { "last_seen_commit": h.get("last_seen_commit"), "last_applied_commit": h.get("last_applied_commit"), "sync_status": h.get("status"), "bootstrap_state": h.get("bootstrap_state"), "pending_commits": h.get("pending_commits", 0), "failed_commits": h.get("failed_commits", 0), "dead_commits": h.get("dead_commits", 0), "human_review_required": h.get("human_review_required", 0), "objects_changed": objects_changed, "drift_count": h.get("drift_count", 0), "retry_count": retry_count, "last_error": None, # metadata-minimal: kein Knowledge-Inhalt "last_error_code": h.get("last_error_code"), "last_success_at": h.get("last_success_at"), "forgejo_status": down.get("forgejo", h.get("forgejo_status", "UNKNOWN")), "tolaria_status": down.get("tolaria", h.get("tolaria_status", "UNKNOWN")), "search_status": down.get("search", h.get("search_status", "UNKNOWN")), } def health_contract(store: C5AStore, down: Optional[Dict[str, str]] = None) -> Dict[str, Any]: """ Ehrlicher Health-Contract (C5E §10): HEALTHY / DEGRADED / BLOCKED. Darf NICHT HEALTHY vortaeuschen, wenn z.B.: - DEAD- oder FAILED-Commit existiert -> BLOCKED - HUMAN_REVIEW_REQUIRED offen ist -> BLOCKED - Drift erkannt wurde -> BLOCKED - kritischer Downstream dauerhaft unavailable -> BLOCKED/DEGRADED - pending/retry vorhanden -> DEGRADED """ h = store.health() down = down or {} dead = h.get("dead_commits", 0) failed = h.get("failed_commits", 0) human = h.get("human_review_required", 0) drift = h.get("drift_count", 0) pending = h.get("pending_commits", 0) retry_pending = sum( 1 for c in store.list_commits() if c.get("status") == ST_RETRY_PENDING) # Kritischer Downstream dauerhaft unavailable critical_down = [ d for d in ("forgejo", "tolaria", "search") if down.get(d) in ("DOWN", "UNKNOWN") ] has_downstream_failure = bool(critical_down) if dead > 0 or failed > 0 or human > 0 or drift > 0: status = HEALTH_BLOCKED elif has_downstream_failure: status = HEALTH_BLOCKED elif pending > 0 or retry_pending > 0: status = HEALTH_DEGRADED else: status = HEALTH_HEALTHY base = observability(store, down) return { **base, "status": status, "health_state": status, }