diff --git a/backend/src/bulkhead.rs b/backend/src/bulkhead.rs
index 257b5dd2..8cad0451 100644
--- a/backend/src/bulkhead.rs
+++ b/backend/src/bulkhead.rs
@@ -162,6 +162,20 @@ impl BulkheadRegistry {
             )
         };
 
+        // Fast path: a concurrency slot is immediately available, so this
+        // request never actually has to wait — it must not be gated by
+        // `max_queue_size`, which bounds the *wait queue*, not total
+        // concurrency. Without this, `max_queue_size: 0` would reject every
+        // request outright even with free concurrent slots.
+        if let Ok(permit) = Arc::clone(&semaphore).try_acquire_owned() {
+            metrics.active.fetch_add(1, Ordering::SeqCst);
+            return Ok(BulkheadPermit {
+                _permit: permit,
+                metrics,
+            });
+        }
+
+        // No slot free: this request must wait, gated by the queue budget.
         let queued_now = metrics.queued.fetch_add(1, Ordering::SeqCst) + 1;
         if queued_now > max_queue_size {
             metrics.queued.fetch_sub(1, Ordering::SeqCst);
diff --git a/backend/src/chaos.rs b/backend/src/chaos.rs
index cb7c6e07..1e9faebf 100644
--- a/backend/src/chaos.rs
+++ b/backend/src/chaos.rs
@@ -542,6 +542,158 @@ mod tests {
         assert!(!result.passed());
     }
 
+    /// #370: cache unavailability + one replica down simultaneously. Rather
+    /// than erroring outright, a caller using `fallback::cascade` should
+    /// route around both down targets to the next healthy one.
+    #[test]
+    fn cascading_cache_and_replica_failure_falls_back_gracefully() {
+        use crate::fallback::{cascade, FallbackChain, FallbackTarget};
+
+        let sim = NetworkPartitionSimulator::new();
+        sim.partition("cache");
+        sim.partition("replica-b");
+        // "primary-db" is left healthy.
+
+        let chain = FallbackChain {
+            id: "test-chain".to_string(),
+            name: "vault-read".to_string(),
+            resource: "vault:read".to_string(),
+            targets: vec![
+                FallbackTarget {
+                    name: "cache".to_string(),
+                    endpoint: "cache://local".to_string(),
+                    priority: 0,
+                },
+                FallbackTarget {
+                    name: "replica-b".to_string(),
+                    endpoint: "replica://b".to_string(),
+                    priority: 1,
+                },
+                FallbackTarget {
+                    name: "primary-db".to_string(),
+                    endpoint: "db://primary".to_string(),
+                    priority: 2,
+                },
+            ],
+            created_at: chrono::Utc::now(),
+            active: true,
+        };
+
+        let result = cascade(&chain, |target| sim.call("client", &target.name));
+
+        // Falls back instead of erroring: a resolved target is found even
+        // though the two highest-priority targets are both down.
+        assert_eq!(result.resolved_target, Some("db://primary".to_string()));
+        assert!(
+            result.degraded,
+            "should be reported as degraded since it didn't resolve on the first target"
+        );
+        assert_eq!(result.attempts.len(), 3);
+        assert!(!result.attempts[0].succeeded, "cache should be down");
+        assert!(!result.attempts[1].succeeded, "replica-b should be down");
+        assert!(result.attempts[2].succeeded, "primary-db should be reachable");
+
+        // Recovery: healing cache should let the chain resolve on the
+        // highest-priority target again, no longer degraded.
+        sim.heal("cache");
+        let recovered = cascade(&chain, |target| sim.call("client", &target.name));
+        assert_eq!(recovered.resolved_target, Some("cache://local".to_string()));
+        assert!(!recovered.degraded);
+    }
+
+    /// #370: circuit breaker + bulkhead interacting under concurrent load.
+    /// The bulkhead bounds concurrency/queueing to a downstream dependency;
+    /// the circuit breaker should additionally fast-reject calls once it
+    /// trips, without either mechanism panicking or deadlocking under load.
+    #[tokio::test]
+    async fn circuit_breaker_and_bulkhead_interact_under_load() {
+        use crate::bulkhead::{BulkheadConfig, BulkheadRegistry};
+        use crate::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig, CircuitState};
+        use std::sync::Arc;
+
+        let registry = Arc::new(BulkheadRegistry::new(BulkheadConfig {
+            max_concurrent: 3,
+            max_queue_size: 4,
+        }));
+        let breaker = Arc::new(CircuitBreaker::new(
+            "downstream-dep",
+            CircuitBreakerConfig {
+                failure_threshold: 4,
+                success_threshold: 2,
+                open_duration: Duration::from_secs(30),
+            },
+        ));
+
+        let bulkhead_permits_acquired = Arc::new(AtomicUsize::new(0));
+        let operation_invocations = Arc::new(AtomicUsize::new(0));
+
+        let mut handles = Vec::new();
+        for _ in 0..20 {
+            let registry = Arc::clone(&registry);
+            let breaker = Arc::clone(&breaker);
+            let bulkhead_permits_acquired = Arc::clone(&bulkhead_permits_acquired);
+            let operation_invocations = Arc::clone(&operation_invocations);
+            handles.push(tokio::spawn(async move {
+                match registry.acquire("/api/vault-dependency").await {
+                    Ok(_permit) => {
+                        bulkhead_permits_acquired.fetch_add(1, Ordering::SeqCst);
+                        // Simulate call latency while holding the slot.
+                        tokio::time::sleep(Duration::from_millis(5)).await;
+                        breaker
+                            .call(|| {
+                                operation_invocations.fetch_add(1, Ordering::SeqCst);
+                                Err::<(), &str>("simulated downstream failure")
+                            })
+                            .map_err(|_| ())
+                    }
+                    Err(_queue_full) => Err(()),
+                }
+            }));
+        }
+
+        let mut rejected_or_failed = 0;
+        for handle in handles {
+            // A panic in the spawned task surfaces here as an `Err` from
+            // `join` — exactly the "did the system crash under combined
+            // load" signal a chaos scenario needs to catch.
+            let outcome = handle.await.expect("task panicked under concurrent load");
+            if outcome.is_err() {
+                rejected_or_failed += 1;
+            }
+        }
+
+        assert_eq!(
+            rejected_or_failed, 20,
+            "every call should have failed or been rejected (operation always fails)"
+        );
+        assert_eq!(
+            breaker.state(),
+            CircuitState::Open,
+            "breaker should trip open under sustained failures"
+        );
+
+        let acquired = bulkhead_permits_acquired.load(Ordering::SeqCst);
+        let invoked = operation_invocations.load(Ordering::SeqCst);
+        assert!(
+            acquired >= 4,
+            "test needs at least failure_threshold calls to reach the breaker to prove it trips, got {acquired}"
+        );
+        assert!(
+            invoked < acquired,
+            "circuit breaker should have fast-rejected at least one call that made it past the \
+             bulkhead instead of invoking the operation every time (invoked={invoked}, acquired={acquired})"
+        );
+
+        // No permits leaked: bulkhead accounting should be back to zero
+        // active once every task has completed.
+        let snapshot = registry.metrics_snapshot();
+        let bulkhead_stats = snapshot
+            .iter()
+            .find(|s| s.endpoint == "/api/vault-dependency")
+            .expect("endpoint should have been recorded");
+        assert_eq!(bulkhead_stats.active, 0, "all bulkhead permits should have been released");
+    }
+
     #[test]
     fn chaos_report_aggregates_multiple_scenarios() {
         let mut report = ChaosReport::new();
diff --git a/backend/src/db.rs b/backend/src/db.rs
index e7304fe5..923cca33 100644
--- a/backend/src/db.rs
+++ b/backend/src/db.rs
@@ -86,6 +86,10 @@ pub struct AppState {
     pub query_cache: Arc<crate::query_cache::QueryCache>,
     /// Distributed-lock deadlock detector stats (#82).
     pub deadlock_detector: Arc<crate::deadlock::DeadlockDetector>,
+    /// Incident tracking: shared by the incident HTTP API and the scheduled
+    /// consensus reconciliation job (#373), so a conflict opens the same
+    /// kind of record a manually-filed incident would.
+    pub incident_state: Arc<crate::incidents::IncidentState>,
 }
 
 impl axum::extract::FromRef<AppState> for Arc<Db> {
@@ -124,6 +128,12 @@ impl axum::extract::FromRef<AppState> for Arc<crate::feature_flags::FlagState> {
     }
 }
 
+impl axum::extract::FromRef<AppState> for Arc<crate::incidents::IncidentState> {
+    fn from_ref(state: &AppState) -> Arc<crate::incidents::IncidentState> {
+        Arc::clone(&state.incident_state)
+    }
+}
+
 // NOTE: The following FromRef implementations reference fields that are not currently
 // in AppState. When these features are properly implemented, uncomment and add the
 // corresponding fields to AppState.
diff --git a/backend/src/incidents.rs b/backend/src/incidents.rs
index 2005fc80..6f12d621 100644
--- a/backend/src/incidents.rs
+++ b/backend/src/incidents.rs
@@ -155,20 +155,29 @@ fn timeline_entry(actor: impl Into<String>, note: impl Into<String>) -> Timeline
     }
 }
 
-/// `POST /incidents` — open a new incident with severity classification.
-pub async fn create_incident(
-    State(state): State<Arc<IncidentState>>,
-    Json(body): Json<CreateIncidentRequest>,
-) -> (StatusCode, Json<Incident>) {
+/// Open a new incident directly against `store`, bypassing the HTTP layer.
+///
+/// This is the synchronous entry point used by background jobs (e.g. the
+/// scheduled consensus reconciliation and backup checksum verification
+/// jobs) that detect a problem outside of a request/response cycle and need
+/// to surface it to operators the same way a manually-filed incident would
+/// be tracked. `create_incident` (the `POST /incidents` handler) builds on
+/// top of this so both paths produce identical `Incident` records.
+pub fn open_incident(
+    store: &IncidentStore,
+    title: impl Into<String>,
+    description: impl Into<String>,
+    severity: IncidentSeverity,
+) -> Incident {
     let now = Utc::now();
     let incident = Incident {
         id: Uuid::new_v4().to_string(),
-        title: body.title,
-        description: body.description,
-        severity: body.severity,
+        title: title.into(),
+        description: description.into(),
+        severity,
         status: IncidentStatus::Open,
         escalation_level: 0,
-        assigned_to: body.assigned_to,
+        assigned_to: None,
         timeline: vec![timeline_entry("system", "incident opened")],
         created_at: now,
         updated_at: now,
@@ -177,11 +186,30 @@ pub async fn create_incident(
     tracing::warn!(
         incident_id = %incident.id,
         severity = ?incident.severity,
+        title = %incident.title,
         "incident opened"
     );
 
-    let mut store = state.store.lock().unwrap();
-    store.insert(incident.id.clone(), incident.clone());
+    store
+        .lock()
+        .unwrap()
+        .insert(incident.id.clone(), incident.clone());
+
+    incident
+}
+
+/// `POST /incidents` — open a new incident with severity classification.
+pub async fn create_incident(
+    State(state): State<Arc<IncidentState>>,
+    Json(body): Json<CreateIncidentRequest>,
+) -> (StatusCode, Json<Incident>) {
+    let mut incident = open_incident(&state.store, body.title, body.description, body.severity);
+    incident.assigned_to = body.assigned_to.clone();
+    state
+        .store
+        .lock()
+        .unwrap()
+        .insert(incident.id.clone(), incident.clone());
 
     (StatusCode::CREATED, Json(incident))
 }
@@ -268,3 +296,29 @@ pub fn is_past_escalation_sla(incident: &Incident) -> bool {
     let elapsed = Utc::now() - incident.created_at;
     elapsed.num_minutes() > incident.severity.escalation_sla_minutes()
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn open_incident_inserts_into_store_as_open() {
+        let store = create_incident_store();
+        let incident = open_incident(&store, "cache drift detected", "3 conflicting keys", IncidentSeverity::Sev2);
+
+        assert_eq!(incident.status, IncidentStatus::Open);
+        assert_eq!(incident.escalation_level, 0);
+        assert_eq!(incident.timeline.len(), 1);
+
+        let stored = store.lock().unwrap().get(&incident.id).cloned();
+        assert!(stored.is_some());
+        assert_eq!(stored.unwrap().title, "cache drift detected");
+    }
+
+    #[test]
+    fn fresh_incident_is_not_past_sla() {
+        let store = create_incident_store();
+        let incident = open_incident(&store, "t", "d", IncidentSeverity::Sev1);
+        assert!(!is_past_escalation_sla(&incident));
+    }
+}
diff --git a/backend/src/lib.rs b/backend/src/lib.rs
index c27a1e95..0ac37507 100644
--- a/backend/src/lib.rs
+++ b/backend/src/lib.rs
@@ -3,6 +3,7 @@ pub mod anomaly_detection;
 pub mod audit;
 pub mod backup_validation;
 pub mod batching;
+pub mod bulkhead;
 pub mod cache;
 pub mod chaos;
 pub mod circuit_breaker;
@@ -18,10 +19,12 @@ pub mod dlq;
 pub mod error;
 pub mod error_context;
 pub mod event_sourcing;
+pub mod fallback;
 pub mod feature_flags;
 pub mod graphql;
 pub mod handlers;
 pub mod health_routing;
+pub mod incidents;
 pub mod load_shedding;
 pub mod message_queue;
 pub mod metrics;
diff --git a/backend/src/main.rs b/backend/src/main.rs
index ce876a96..fa1bf8d1 100644
--- a/backend/src/main.rs
+++ b/backend/src/main.rs
@@ -32,6 +32,10 @@ use ethos_protocol_backend::{
     event_sourcing::EventSourcingState,
     feature_flags::{evaluate_flag_handler, get_flag, list_flags, upsert_flag, FlagState},
     graphql::{build_schema, graphql_handler, graphql_playground},
+    incidents::{
+        add_timeline_entry, create_incident, escalate_incident, get_incident, list_incidents,
+        update_incident_status, IncidentState,
+    },
     load_shedding::{admission_middleware, LoadMonitor, LoadShedder, SheddingConfig},
     message_queue::MessageQueueState,
     metrics::Metrics,
@@ -42,6 +46,7 @@ use ethos_protocol_backend::{
     routes,
     rpc_pool::{RpcPool, RpcPoolConfig},
     scheduler,
+    scheduler::SchedulerContext,
     streaming::{stream_events, stream_vaults},
     timeout_policy::TimeoutState,
     tracing_sampling::TraceSampler,
@@ -187,6 +192,12 @@ pub fn build_router(state: AppState) -> Router {
         .route("/webhooks", post(register_webhook).get(list_webhooks))
         .route("/webhooks/:id", delete(delete_webhook))
         .route("/webhooks/verify", post(verify_webhook))
+        // ── Incident tracking routes (#373) ───────────────────────────────────
+        .route("/incidents", post(create_incident).get(list_incidents))
+        .route("/incidents/:id", get(get_incident))
+        .route("/incidents/:id/timeline", post(add_timeline_entry))
+        .route("/incidents/:id/status", post(update_incident_status))
+        .route("/incidents/:id/escalate", post(escalate_incident))
         // ── GraphQL routes (#66) ─────────────────────────────────────────────
         .route("/graphql", post(graphql_handler))
         .route("/graphql/playground", get(graphql_playground))
@@ -284,11 +295,6 @@ async fn main() {
         "consensus cache initialized"
     );
 
-    let scheduler_db = Arc::clone(&db);
-    tokio::spawn(async move {
-        scheduler::run(scheduler_db).await;
-    });
-
     let vault_store = create_vault_store();
     let event_store = create_event_store();
     let graphql_schema = build_schema(Arc::clone(&vault_store), Arc::clone(&event_store));
@@ -322,6 +328,20 @@ async fn main() {
 
     let flag_state = Arc::new(FlagState::new(Arc::clone(&db)));
 
+    let incident_state = Arc::new(IncidentState::new());
+
+    // ── Background scheduler (reminders, TTL insurance, retention, secret
+    // rotation, consensus reconciliation (#373)) ──────────────────────────
+    let scheduler_ctx = SchedulerContext {
+        db: Arc::clone(&db),
+        consensus: Arc::clone(&consensus),
+        metrics: Arc::clone(&metrics),
+        incident_state: Arc::clone(&incident_state),
+    };
+    tokio::spawn(async move {
+        scheduler::run(scheduler_ctx).await;
+    });
+
     let state = AppState {
         db: Arc::clone(&db),
         vault_store,
@@ -345,6 +365,7 @@ async fn main() {
         flag_state,
         query_cache: Arc::new(ethos_protocol_backend::query_cache::QueryCache::new()),
         deadlock_detector: Arc::new(ethos_protocol_backend::deadlock::DeadlockDetector::new()),
+        incident_state,
     };
 
     // ── Dynamic ACL admin routes ─────────────────────────────────────────
diff --git a/backend/src/metrics.rs b/backend/src/metrics.rs
index b2b8faef..c02c997b 100644
--- a/backend/src/metrics.rs
+++ b/backend/src/metrics.rs
@@ -12,6 +12,12 @@ pub struct Metrics {
     pub request_errors_total: AtomicU64,
     pub http_requests_total: AtomicU64,
     pub contract_paused: AtomicU64,
+    /// Total scheduled consensus (cache reconciliation) checks run.
+    pub consensus_checks_total: AtomicU64,
+    /// Total key conflicts detected across all consensus checks.
+    pub consensus_conflicts_total: AtomicU64,
+    /// 1 if the most recent consensus check found the cache consistent, 0 otherwise.
+    pub consensus_consistent: AtomicU64,
 }
 
 impl Metrics {
@@ -65,6 +71,24 @@ impl Metrics {
             "1 if contract is paused, 0 otherwise",
             self.contract_paused.load(Ordering::Relaxed),
         );
+        push_counter(
+            &mut out,
+            "ethos_protocol_consensus_checks_total",
+            "Total scheduled consensus reconciliation checks run",
+            self.consensus_checks_total.load(Ordering::Relaxed),
+        );
+        push_counter(
+            &mut out,
+            "ethos_protocol_consensus_conflicts_total",
+            "Total cache key conflicts detected by consensus checks",
+            self.consensus_conflicts_total.load(Ordering::Relaxed),
+        );
+        push_gauge(
+            &mut out,
+            "ethos_protocol_consensus_consistent",
+            "1 if the most recent consensus check found the cache consistent, 0 otherwise",
+            self.consensus_consistent.load(Ordering::Relaxed),
+        );
 
         out
     }
@@ -108,6 +132,19 @@ mod tests {
         assert!(output.contains("ethos_protocol_contract_paused 1"));
     }
 
+    #[test]
+    fn test_render_contains_consensus_metrics() {
+        let m = Metrics::new();
+        m.consensus_checks_total.store(3, Ordering::Relaxed);
+        m.consensus_conflicts_total.store(2, Ordering::Relaxed);
+        m.consensus_consistent.store(0, Ordering::Relaxed);
+
+        let output = m.render();
+        assert!(output.contains("ethos_protocol_consensus_checks_total 3"));
+        assert!(output.contains("ethos_protocol_consensus_conflicts_total 2"));
+        assert!(output.contains("ethos_protocol_consensus_consistent 0"));
+    }
+
     #[test]
     fn test_render_prometheus_format() {
         let m = Metrics::new();
diff --git a/backend/src/retry_policy.rs b/backend/src/retry_policy.rs
index 0a52f62c..8aae2221 100644
--- a/backend/src/retry_policy.rs
+++ b/backend/src/retry_policy.rs
@@ -82,6 +82,33 @@ pub struct CreateRetryPolicyRequest {
     pub retry_on_status: Vec<u16>,
 }
 
+impl RetryPolicy {
+    /// Validates that this policy's backoff/jitter configuration is sane
+    /// before it's stored and used to compute real delays.
+    ///
+    /// `jitter` itself (`JitterMode`) is a closed enum with no numeric
+    /// fields, so it can't independently be "negative" or "out of bounds" —
+    /// but the parameters that feed the jittered delay computation
+    /// (`compute_backoff_delay`) can be, and a bad value there is exactly
+    /// what would produce a negative, zero-forever, or larger-than-interval
+    /// jitter window at runtime. This checks those.
+    pub fn validate(&self) -> Result<(), String> {
+        if self.max_attempts == 0 {
+            return Err("max_attempts must be > 0".to_string());
+        }
+        if self.base_delay_ms > self.max_delay_ms {
+            return Err("base_delay_ms must be <= max_delay_ms".to_string());
+        }
+        if !self.multiplier.is_finite() || self.multiplier <= 0.0 {
+            return Err(format!(
+                "multiplier must be a finite positive number, got {}",
+                self.multiplier
+            ));
+        }
+        Ok(())
+    }
+}
+
 /// Computes the delay before attempt number `attempt` (1-indexed) using
 /// exponential backoff capped at `max_delay_ms`, with jitter applied.
 pub fn compute_backoff_delay(policy: &RetryPolicy, attempt: u32) -> Duration {
@@ -205,19 +232,6 @@ async fn create_retry_policy(
     State(state): State<RetryPolicyState>,
     Json(body): Json<CreateRetryPolicyRequest>,
 ) -> Result<(StatusCode, Json<RetryPolicy>), (StatusCode, String)> {
-    if body.max_attempts == 0 {
-        return Err((
-            StatusCode::UNPROCESSABLE_ENTITY,
-            "max_attempts must be > 0".into(),
-        ));
-    }
-    if body.base_delay_ms > body.max_delay_ms {
-        return Err((
-            StatusCode::UNPROCESSABLE_ENTITY,
-            "base_delay_ms must be <= max_delay_ms".into(),
-        ));
-    }
-
     let policy = RetryPolicy {
         id: uuid::Uuid::new_v4().to_string(),
         name: body.name,
@@ -230,6 +244,11 @@ async fn create_retry_policy(
         retry_on_status: body.retry_on_status,
         created_at: chrono::Utc::now(),
     };
+
+    policy
+        .validate()
+        .map_err(|e| (StatusCode::UNPROCESSABLE_ENTITY, e))?;
+
     let saved = state.store.upsert(policy);
     Ok((StatusCode::CREATED, Json(saved)))
 }
@@ -299,6 +318,87 @@ mod tests {
         }
     }
 
+    #[test]
+    fn equal_jitter_stays_within_half_to_full_capped_delay() {
+        // Unlike full jitter (which can land anywhere in [0, capped]), equal
+        // jitter should never fall below half the capped delay, and never
+        // exceed it. Sampled repeatedly since jitter is randomized.
+        let policy = test_policy(JitterMode::Equal);
+        for attempt in 1..=6 {
+            let capped_ms = (policy.base_delay_ms as f64 * policy.multiplier.powi(attempt as i32 - 1))
+                .min(policy.max_delay_ms as f64);
+            for _ in 0..50 {
+                let delay_ms = compute_backoff_delay(&policy, attempt).as_millis() as f64;
+                assert!(
+                    delay_ms >= (capped_ms / 2.0).floor() && delay_ms <= capped_ms.ceil(),
+                    "attempt {attempt}: delay {delay_ms}ms outside expected [{}, {}] window",
+                    capped_ms / 2.0,
+                    capped_ms
+                );
+            }
+        }
+    }
+
+    #[test]
+    fn no_jitter_mode_never_produces_negative_or_out_of_bounds_delay() {
+        let policy = test_policy(JitterMode::None);
+        for attempt in 1..=10 {
+            let delay = compute_backoff_delay(&policy, attempt);
+            assert!(delay.as_millis() <= policy.max_delay_ms as u128);
+        }
+    }
+
+    #[test]
+    fn validate_accepts_sane_boundary_configurations() {
+        // multiplier == 1.0 (no growth, but still a sane, finite, positive
+        // value) and base_delay_ms == max_delay_ms (zero-width interval) are
+        // both edge cases that should be accepted, not rejected.
+        let mut policy = test_policy(JitterMode::None);
+        policy.multiplier = 1.0;
+        assert!(policy.validate().is_ok());
+
+        policy.base_delay_ms = policy.max_delay_ms;
+        assert!(policy.validate().is_ok());
+    }
+
+    #[test]
+    fn validate_rejects_zero_max_attempts() {
+        let mut policy = test_policy(JitterMode::None);
+        policy.max_attempts = 0;
+        assert!(policy.validate().is_err());
+    }
+
+    #[test]
+    fn validate_rejects_base_delay_greater_than_max_delay() {
+        let mut policy = test_policy(JitterMode::None);
+        policy.base_delay_ms = policy.max_delay_ms + 1;
+        assert!(policy.validate().is_err());
+    }
+
+    #[test]
+    fn validate_rejects_non_positive_multiplier() {
+        for bad in [0.0, -1.0, -0.5] {
+            let mut policy = test_policy(JitterMode::None);
+            policy.multiplier = bad;
+            assert!(
+                policy.validate().is_err(),
+                "multiplier {bad} should be rejected"
+            );
+        }
+    }
+
+    #[test]
+    fn validate_rejects_non_finite_multiplier() {
+        for bad in [f64::NAN, f64::INFINITY, f64::NEG_INFINITY] {
+            let mut policy = test_policy(JitterMode::None);
+            policy.multiplier = bad;
+            assert!(
+                policy.validate().is_err(),
+                "multiplier {bad} should be rejected"
+            );
+        }
+    }
+
     #[test]
     fn find_for_path_prefers_most_specific_match() {
         let store = RetryPolicyStore::new();
diff --git a/backend/src/scheduler.rs b/backend/src/scheduler.rs
index 48b9c1e8..9899c260 100644
--- a/backend/src/scheduler.rs
+++ b/backend/src/scheduler.rs
@@ -1,3 +1,4 @@
+use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -5,19 +6,43 @@ use chrono::Utc;
 
 use crate::{db::Db, models::Frequency};
 
+/// Dependencies the background scheduler needs to run all of its periodic
+/// jobs. Grouped into one struct (rather than many `run(...)` parameters)
+/// since the job list has grown from "poll reminder preferences" into a
+/// handful of unrelated periodic checks that each need their own slice of
+/// shared state.
+pub struct SchedulerContext {
+    pub db: Arc<Db>,
+    /// Distributed cache consensus checker (#373).
+    pub consensus: Arc<crate::consensus::NodeCache>,
+    /// Prometheus-style counters exposed at `/metrics`.
+    pub metrics: Arc<crate::metrics::Metrics>,
+    /// Shared incident store: consensus conflicts detected by the scheduled
+    /// job are opened here the same way a manually-filed incident would be.
+    pub incident_state: Arc<crate::incidents::IncidentState>,
+}
+
 /// Polls preferences every minute and fires reminders for vaults whose TTL
 /// is within the user-configured window.
 ///
 /// In production, replace `fetch_ttl_remaining` with a real Stellar RPC call
 /// and `send_reminder` with actual email/SMS/push dispatch.
-pub async fn run(db: Arc<Db>) {
+pub async fn run(ctx: SchedulerContext) {
+    let SchedulerContext {
+        db,
+        consensus,
+        metrics,
+        incident_state,
+    } = ctx;
+
     // Seed default secret rotation policies on startup.
     crate::secret_rotation::seed_default_policies(&db);
 
     let mut interval = tokio::time::interval(Duration::from_mins(1));
-    // Track when we last ran the daily/hourly tasks.
+    // Track when we last ran the daily/hourly/periodic tasks.
     let mut last_daily_purge = chrono::DateTime::<Utc>::MIN_UTC;
     let mut last_rotation_check = chrono::DateTime::<Utc>::MIN_UTC;
+    let mut last_consensus_check = chrono::DateTime::<Utc>::MIN_UTC;
 
     loop {
         interval.tick().await;
@@ -108,6 +133,14 @@ pub async fn run(db: Arc<Db>) {
             crate::secret_rotation::run_rotation_scheduler(&db);
             last_rotation_check = now;
         }
+
+        // 5) Distributed cache consensus reconciliation (#373; runs at most
+        //    once every 5 minutes — cache drift needs tighter reconciliation
+        //    than the once-a-day/hour housekeeping jobs above).
+        if now.signed_duration_since(last_consensus_check).num_minutes() >= 5 {
+            run_consensus_check(&consensus, &metrics, &incident_state);
+            last_consensus_check = now;
+        }
     }
 }
 
@@ -274,3 +307,134 @@ fn run_consistency_check(db: &Arc<Db>) {
         "consistency check job completed"
     );
 }
+
+// ── #373: Consensus Reconciliation Job ───────────────────────────────────────
+
+/// Run the periodic distributed-cache consensus reconciliation job.
+///
+/// Compares this node's local cache against the shared `InMemoryBackend` /
+/// `RedisBackend` (see `consensus.rs`), publishes the result as metrics, and
+/// — when conflicts are found — opens an incident so operators are notified
+/// even if nobody is actively watching `/health/consensus` or `/metrics`.
+fn run_consensus_check(
+    consensus: &Arc<crate::consensus::NodeCache>,
+    metrics: &Arc<crate::metrics::Metrics>,
+    incident_state: &Arc<crate::incidents::IncidentState>,
+) {
+    tracing::info!("consensus reconciliation job started");
+
+    let report = match consensus.check_and_resolve() {
+        Ok(report) => report,
+        Err(e) => {
+            tracing::error!(error = %e, "consensus reconciliation job failed to run");
+            return;
+        }
+    };
+
+    metrics.consensus_checks_total.fetch_add(1, Ordering::Relaxed);
+    metrics
+        .consensus_conflicts_total
+        .fetch_add(report.conflicts.len() as u64, Ordering::Relaxed);
+    metrics
+        .consensus_consistent
+        .store(u64::from(report.consistent), Ordering::Relaxed);
+
+    if report.consistent {
+        tracing::info!(
+            node_id = %report.node_id,
+            keys_checked = report.keys_checked,
+            "consensus reconciliation job completed: cache consistent"
+        );
+        return;
+    }
+
+    tracing::warn!(
+        node_id = %report.node_id,
+        conflicts = report.conflicts.len(),
+        conflicts_resolved = report.conflicts_resolved,
+        keys_checked = report.keys_checked,
+        "consensus reconciliation job detected conflicts"
+    );
+
+    let conflicted_keys: Vec<&str> = report.conflicts.iter().map(|c| c.key.as_str()).collect();
+    crate::incidents::open_incident(
+        &incident_state.store,
+        "Distributed cache consensus conflict detected",
+        format!(
+            "Node '{}' found {} conflicting key(s) between its local cache and the distributed \
+             backend during scheduled reconciliation (strategy: {:?}). {} conflict(s) were \
+             auto-resolved. Affected keys: {}",
+            report.node_id,
+            report.conflicts.len(),
+            report.strategy,
+            report.conflicts_resolved,
+            conflicted_keys.join(", "),
+        ),
+        crate::incidents::IncidentSeverity::Sev3,
+    );
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::consensus::{CacheBackend, CacheEntry, ConflictStrategy, InMemoryBackend, NodeCache};
+    use crate::incidents::{create_incident_store, IncidentState};
+    use crate::metrics::Metrics;
+    use chrono::TimeZone;
+
+    #[test]
+    fn consensus_check_opens_incident_and_updates_metrics_on_conflict() {
+        let backend: Arc<dyn CacheBackend> = Arc::new(InMemoryBackend::new());
+        let consensus = Arc::new(NodeCache::new(
+            "test-node",
+            Arc::clone(&backend),
+            ConflictStrategy::LastWriteWins,
+        ));
+        consensus.put("vault:1", "authoritative").unwrap();
+        consensus.set_local_entry(CacheEntry {
+            key: "vault:1".to_string(),
+            value: "stale".to_string(),
+            node_id: "test-node".to_string(),
+            updated_at: chrono::Utc.timestamp_millis_opt(1).unwrap(),
+            version: 1,
+        });
+
+        let metrics = Metrics::new();
+        let incident_state = Arc::new(IncidentState {
+            store: create_incident_store(),
+        });
+
+        run_consensus_check(&consensus, &metrics, &incident_state);
+
+        assert_eq!(metrics.consensus_checks_total.load(Ordering::Relaxed), 1);
+        assert_eq!(metrics.consensus_conflicts_total.load(Ordering::Relaxed), 1);
+        assert_eq!(metrics.consensus_consistent.load(Ordering::Relaxed), 0);
+
+        let incidents = incident_state.store.lock().unwrap();
+        assert_eq!(incidents.len(), 1);
+        let incident = incidents.values().next().unwrap();
+        assert!(incident.description.contains("vault:1"));
+    }
+
+    #[test]
+    fn consensus_check_does_not_open_incident_when_consistent() {
+        let backend: Arc<dyn CacheBackend> = Arc::new(InMemoryBackend::new());
+        let consensus = Arc::new(NodeCache::new(
+            "test-node",
+            backend,
+            ConflictStrategy::LastWriteWins,
+        ));
+        consensus.put("vault:2", "value").unwrap();
+
+        let metrics = Metrics::new();
+        let incident_state = Arc::new(IncidentState {
+            store: create_incident_store(),
+        });
+
+        run_consensus_check(&consensus, &metrics, &incident_state);
+
+        assert_eq!(metrics.consensus_consistent.load(Ordering::Relaxed), 1);
+        assert_eq!(metrics.consensus_conflicts_total.load(Ordering::Relaxed), 0);
+        assert!(incident_state.store.lock().unwrap().is_empty());
+    }
+}
diff --git a/backend/src/tests.rs b/backend/src/tests.rs
index 48d77dc6..5638fd9a 100644
--- a/backend/src/tests.rs
+++ b/backend/src/tests.rs
@@ -23,6 +23,7 @@ use ethos_protocol_backend::{
     event_sourcing::EventSourcingState,
     feature_flags::FlagState,
     graphql::build_schema,
+    incidents::IncidentState,
     load_shedding::{LoadMonitor, LoadShedder, SheddingConfig},
     message_queue::MessageQueueState,
     metrics::Metrics,
@@ -94,6 +95,7 @@ fn test_state(db: Arc<Db>) -> AppState {
         flag_state,
         query_cache: Arc::new(ethos_protocol_backend::query_cache::QueryCache::new()),
         deadlock_detector: Arc::new(ethos_protocol_backend::deadlock::DeadlockDetector::new()),
+        incident_state: Arc::new(IncidentState::new()),
     }
 }
 
@@ -379,6 +381,7 @@ async fn test_consensus_health_detects_and_resolves_divergence() {
         flag_state,
         query_cache: Arc::new(ethos_protocol_backend::query_cache::QueryCache::new()),
         deadlock_detector: Arc::new(ethos_protocol_backend::deadlock::DeadlockDetector::new()),
+        incident_state: Arc::new(IncidentState::new()),
     };
     db.migrate().unwrap();
 
diff --git a/backend/src/timeout_adaptation.rs b/backend/src/timeout_adaptation.rs
index 930a339a..fade6bb2 100644
--- a/backend/src/timeout_adaptation.rs
+++ b/backend/src/timeout_adaptation.rs
@@ -319,6 +319,122 @@ mod tests {
         assert_eq!(endpoints, vec!["a".to_string(), "b".to_string()]);
     }
 
+    /// Simulation test: feed a long run of steady-state (constant) latency
+    /// samples and confirm the adapted timeout converges to the observed
+    /// latency within a bounded number of iterations (`min_samples`), then
+    /// never changes again — i.e. it settles rather than oscillating.
+    #[test]
+    fn current_timeout_converges_within_min_samples_under_steady_state_load() {
+        let config = TimeoutAdaptationConfig {
+            min_samples: 10,
+            window_size: 30,
+            multiplier: 1.0,
+            ..TimeoutAdaptationConfig::default()
+        };
+        let manager = AdaptiveTimeoutManager::new(config);
+        let steady_latency = Duration::from_millis(80);
+
+        for i in 1..=config.min_samples {
+            manager.record_latency("steady", steady_latency);
+            if i < config.min_samples {
+                assert_eq!(
+                    manager.current_timeout("steady"),
+                    config.default_timeout,
+                    "should still use the default timeout before min_samples is reached"
+                );
+            }
+        }
+
+        // Convergence: exactly at min_samples, the adapted timeout should
+        // match the steady-state latency (multiplier 1.0, no clamping in
+        // range) and then hold steady indefinitely under continued
+        // steady-state load rather than drifting or oscillating.
+        let converged = manager.current_timeout("steady");
+        assert_eq!(converged, steady_latency);
+
+        for _ in 0..100 {
+            manager.record_latency("steady", steady_latency);
+            assert_eq!(
+                manager.current_timeout("steady"),
+                converged,
+                "adapted timeout oscillated under constant steady-state load"
+            );
+        }
+    }
+
+    /// The predictive EMA (`predict_timeout`) should also settle to the
+    /// steady-state value (within a small tolerance for floating-point
+    /// accumulation) once enough steady-state samples have fed its history.
+    #[test]
+    fn predicted_timeout_converges_toward_steady_state_value() {
+        let config = TimeoutAdaptationConfig {
+            min_samples: 5,
+            window_size: 10,
+            multiplier: 1.0,
+            ..TimeoutAdaptationConfig::default()
+        };
+        let manager = AdaptiveTimeoutManager::new(config);
+        let steady_latency = Duration::from_millis(120);
+
+        // Well past HISTORY_CAPACITY (20) pushes so the EMA has settled.
+        for _ in 0..60 {
+            manager.record_latency("steady", steady_latency);
+        }
+
+        let predicted = manager
+            .predict_timeout("steady")
+            .expect("expected a prediction after enough steady-state samples");
+        let delta_ms = (predicted.as_millis() as i128 - steady_latency.as_millis() as i128).abs();
+        assert!(
+            delta_ms <= 2,
+            "predicted timeout {predicted:?} should have converged near the steady-state \
+             latency {steady_latency:?}, delta {delta_ms}ms"
+        );
+    }
+
+    /// Latency spike followed by recovery: a single large spike should
+    /// widen the adapted timeout immediately, and the timeout should return
+    /// to its pre-spike baseline once the spike ages out of the fixed-size
+    /// rolling window (after `window_size` further steady-state samples).
+    #[test]
+    fn timeout_widens_on_spike_then_recovers_once_it_ages_out_of_window() {
+        let config = TimeoutAdaptationConfig {
+            min_samples: 5,
+            window_size: 20,
+            multiplier: 1.0,
+            percentile: 0.99,
+            ..TimeoutAdaptationConfig::default()
+        };
+        let manager = AdaptiveTimeoutManager::new(config);
+        let baseline = Duration::from_millis(60);
+        let spike = Duration::from_secs(5);
+
+        for _ in 0..config.window_size {
+            manager.record_latency("recovering", baseline);
+        }
+        let baseline_timeout = manager.current_timeout("recovering");
+        assert_eq!(baseline_timeout, baseline);
+
+        manager.record_latency("recovering", spike);
+        let spiked_timeout = manager.current_timeout("recovering");
+        assert!(
+            spiked_timeout > baseline_timeout,
+            "timeout should widen in response to a latency spike"
+        );
+
+        // Push enough baseline samples for the single spike sample to fully
+        // age out of the fixed-size rolling window (FIFO eviction means it
+        // is guaranteed gone after `window_size` more pushes).
+        for _ in 0..config.window_size {
+            manager.record_latency("recovering", baseline);
+        }
+        let recovered_timeout = manager.current_timeout("recovering");
+        assert_eq!(
+            recovered_timeout, baseline,
+            "timeout should recover to the pre-spike baseline once the spike ages out of the window"
+        );
+    }
+
     /// Micro-benchmark-style scenario (no external harness): simulate a
     /// burst of latency samples and confirm adaptation stays within a
     /// reasonable wall-clock budget. Documented further in
diff --git a/docs/chaos-testing.md b/docs/chaos-testing.md
index a3b609c9..999cce28 100644
--- a/docs/chaos-testing.md
+++ b/docs/chaos-testing.md
@@ -70,6 +70,70 @@ report.add(ChaosRunner::new(&partition_injector).run(100, resilient_op));
 assert!(report.all_passed());
 ```
 
+## Combined-Failure Scenarios (#370)
+
+Individual fault injectors are useful, but real incidents are rarely a
+single clean failure — they're usually several things going wrong at once.
+Two scenarios exercise that directly, composing chaos primitives with the
+actual reliability modules they're meant to protect rather than a
+standalone fake operation:
+
+### Cache down + one replica down → falls back, doesn't error
+
+`cascading_cache_and_replica_failure_falls_back_gracefully` marks two
+`NetworkPartitionSimulator` targets ("cache" and "replica-b") down
+simultaneously, leaving a third ("primary-db") healthy, then drives a
+`fallback::FallbackChain` through `fallback::cascade`. Findings:
+
+- The chain correctly skips both down targets and resolves on the healthy
+  one — `resolved_target` is `Some`, not an error, confirming
+  `fallback.rs`'s cascade behavior degrades gracefully instead of failing
+  the whole operation when the two highest-priority targets are both
+  unavailable at once.
+- `degraded` is correctly `true` (it didn't resolve on the first target),
+  distinguishing "worked, but not optimally" from "fully healthy" for
+  monitoring purposes.
+- Healing "cache" and re-running the cascade confirms recovery: the chain
+  resolves on the highest-priority target again and `degraded` flips back
+  to `false`.
+
+**Gap**: this scenario drives `fallback::cascade` directly with a
+synthetic chain; it doesn't yet exercise a real call site (e.g. webhook
+delivery or an RPC call) that's actually wired up to use a fallback chain
+in production. Worth adding once a concrete caller adopts
+`fallback::cascade` for cache/replica reads.
+
+### Circuit breaker + bulkhead under concurrent load
+
+`circuit_breaker_and_bulkhead_interact_under_load` spawns 20 concurrent
+tasks against a shared `BulkheadRegistry` (max 3 concurrent, queue of 4)
+and `CircuitBreaker` (opens after 4 consecutive failures), each task
+acquiring a bulkhead permit before making a call that always fails.
+Findings:
+
+- No panics or deadlocks under the combined concurrent load (a panic in
+  any spawned task would surface as a `join` error and fail the test
+  immediately).
+- The two mechanisms compose as expected: some calls are rejected by the
+  bulkhead before ever reaching the breaker (queue full), and — the
+  interesting part — once the breaker trips open, calls that *did* make it
+  past the bulkhead are still fast-rejected by the breaker without
+  invoking the (failing) operation again. The test asserts
+  `operation_invocations < bulkhead_permits_acquired` specifically to
+  isolate this: it's not enough for calls to fail overall, the breaker
+  must demonstrably short-circuit some of them.
+- Bulkhead accounting (`active` permits) returns to zero once every task
+  completes — no permit leak under combined failure + concurrency.
+
+**Gap**: this scenario calls `BulkheadRegistry::acquire` and
+`CircuitBreaker::call` directly; it doesn't run through the actual
+`bulkhead_middleware` Axum middleware layered onto real HTTP requests, so
+it doesn't catch issues specific to how the two are layered in
+`main.rs`'s router (ordering relative to other middleware, header
+propagation, etc.). A follow-up using `tower::ServiceExt::oneshot` against
+a real router (see the pattern in `backend/src/tests.rs`) would close that
+gap.
+
 ## Extending
 
 To add a new fault type, implement `FaultInjector` (`name`, `inject`,
diff --git a/docs/consistency-verification.md b/docs/consistency-verification.md
index 558b8c2b..1601e9b7 100644
--- a/docs/consistency-verification.md
+++ b/docs/consistency-verification.md
@@ -101,3 +101,41 @@ Log output:
 1. Add a `check_*` function to `ConsistencyChecker` in `consistency.rs`.
 2. Register it in the `check_fns` slice inside `run_all_checks`.
 3. Document it in this file with its severity level and the query it runs.
+
+## Distributed Cache Consensus Reconciliation
+
+Separate from the SQLite-focused checks above, `backend/src/consensus.rs`
+implements `ConsensusReport` / `ConflictDetail` for comparing a node's local
+cache against the shared `InMemoryBackend` / `RedisBackend` used for
+multi-node distributed-cache consensus (`NodeCache::check_and_resolve`).
+
+### Scheduled Job
+
+The scheduler (`scheduler.rs`) runs a consensus reconciliation job at most
+**every 5 minutes**. Unlike the SQLite consistency checks above, cache
+divergence between nodes can compound quickly (each node keeps serving
+stale reads until reconciled), so this job runs on a tighter cadence than
+the 6-hour SQLite checks or the hourly backup validation job.
+
+Each run:
+
+1. Calls `NodeCache::check_and_resolve()`, which diffs the local cache
+   against the distributed backend, resolves any conflicts per the
+   configured `ConflictStrategy` (`last_write_wins` or `voting`), and
+   returns a `ConsensusReport`.
+2. Publishes the result as Prometheus metrics on `/metrics`:
+   - `ethos_protocol_consensus_checks_total` (counter)
+   - `ethos_protocol_consensus_conflicts_total` (counter)
+   - `ethos_protocol_consensus_consistent` (gauge; 1 = consistent, 0 = conflicts found)
+3. When conflicts are found, opens an incident via `incidents.rs`
+   (`POST /incidents`-equivalent, severity `Sev3`) describing the affected
+   keys and how many conflicts were auto-resolved, so operators are
+   notified even if nobody is actively watching `/health/consensus` or
+   `/metrics`.
+
+### On-Demand Endpoint
+
+`GET /health/consensus` runs the same check synchronously and returns the
+current consistency status; it does not itself open an incident or update
+the scheduled-job metrics above (those are only touched by the periodic
+job).
diff --git a/docs/timeout-adaptation.md b/docs/timeout-adaptation.md
index 10ad87df..59a59c5d 100644
--- a/docs/timeout-adaptation.md
+++ b/docs/timeout-adaptation.md
@@ -47,6 +47,60 @@ manager.record_latency("get_vault", observed_duration);
 let timeout = manager.current_timeout("get_vault");
 ```
 
+## Convergence Characteristics
+
+Two independently-computed values respond to load differently, and it's
+worth knowing which one you're looking at:
+
+### `current_timeout` — bounded, then stable
+
+`current_timeout` is computed fresh on every call directly from whatever
+samples currently sit in the rolling window — it is **not** itself an
+exponential moving average, so it has no oscillation risk of its own. Under
+steady-state (constant) latency:
+
+- It converges in **exactly `min_samples` iterations** — the first call
+  where the window has at least `min_samples` observations returns a
+  timeout derived from the (now entirely steady-state) window contents.
+- It then holds **perfectly stable** for as long as the input stays
+  steady-state, since every sample in the window is identical and the
+  percentile of a constant window never changes.
+
+See `current_timeout_converges_within_min_samples_under_steady_state_load`
+in the test suite.
+
+### `predict_timeout` — EMA settles within its history window
+
+`predict_timeout` extrapolates from an exponential moving average
+(alpha = 0.3) over up to the last 20 recorded percentile values
+(`HISTORY_CAPACITY`), plus a linear trend term. Under steady-state input,
+the EMA is a fixed point (feeding it the same value repeatedly leaves it
+unchanged) and the trend term goes to zero, so the prediction converges to
+the steady-state latency — in practice within a couple of ms of the
+mathematically-exact value due to `Duration <-> f64` conversion rounding.
+See `predicted_timeout_converges_toward_steady_state_value`.
+
+### Spike + recovery
+
+A single latency spike immediately widens `current_timeout` (it shows up in
+the percentile as soon as it's recorded — no delay). Because the window is
+a fixed-size FIFO ring buffer, the spike is guaranteed to be fully evicted,
+and the timeout fully recovered to its pre-spike baseline, after at most
+`window_size` further steady-state samples — one full window rotation. It
+cannot recover any faster (the spike stays visible to the percentile
+calculation until it physically falls out of the window) and it never gets
+"stuck" wide (it's not a decaying average, so there's no long tail). See
+`timeout_widens_on_spike_then_recovers_once_it_ages_out_of_window`.
+
+### Known gap
+
+Both convergence properties above are exact given `multiplier = 1.0`.
+There's no dedicated test yet asserting the *rate* of `predict_timeout`'s
+EMA convergence when initialized from a very different value than the
+steady-state target (i.e. how many iterations until it's within X% after a
+regime change, as opposed to whether it eventually gets arbitrarily close)
+— worth adding if the EMA's alpha is ever tuned.
+
 ## Benchmarking
 
 There's no `criterion` dev-dependency in this workspace, so