eidetic-engine 0.15.1

//! In-process single-flight coalescing for duplicate read-heavy operations.
//!
//! The coalescer is intentionally process-local and non-durable. It only shares
//! an in-flight result for callers with the same redaction-safe
//! [`SingleFlightKey`](crate::models::SingleFlightKey); it never mutates memory,
//! indexes, Beads, Agent Mail, or cache state.

use std::collections::{BTreeSet, HashMap};
use std::sync::{
    Arc, Barrier, Condvar, Mutex, OnceLock,
    atomic::{AtomicU64, AtomicUsize, Ordering},
};
use std::thread;
use std::time::{Duration, Instant};

use crate::models::{
    SingleFlightKey, SingleFlightKeyInput, SingleFlightLastKeyPosture, SingleFlightPostureReport,
    SingleFlightSurface, SingleFlightSurfaceCounters, SingleFlightSurfacePosture,
};

pub const SINGLEFLIGHT_FOLLOWER_TIMEOUT_CODE: &str = "singleflight_follower_timeout";
pub const SINGLEFLIGHT_LEADER_FAILED_CODE: &str = "singleflight_leader_failed";
pub const SINGLEFLIGHT_STATE_POISONED_CODE: &str = "singleflight_state_poisoned";
const SINGLEFLIGHT_TRACE_SURFACE: &str = "singleflight";
const SINGLEFLIGHT_TRACE_REQUEST_ID: &str = "singleflight_group_run";
const SINGLEFLIGHT_TRACE_WORKSPACE_ID: &str = "process_local";
const SINGLEFLIGHT_LEADER_START_EVENT: &str = "leader_start";
const SINGLEFLIGHT_FOLLOWER_JOIN_EVENT: &str = "follower_join";
const SINGLEFLIGHT_FOLLOWER_TIMEOUT_EVENT: &str = "follower_timeout";
const SINGLEFLIGHT_LEADER_COMPLETE_EVENT: &str = "leader_complete";
const SINGLEFLIGHT_COALESCED_RESULT_REUSED_EVENT: &str = "coalesced_result_reused";
const SINGLEFLIGHT_LEADER_FAILED_EVENT: &str = "leader_failed";
#[cfg(test)]
const SINGLEFLIGHT_REQUIRED_TELEMETRY_PHASES: [&str; 5] = [
    SINGLEFLIGHT_LEADER_START_EVENT,
    SINGLEFLIGHT_FOLLOWER_JOIN_EVENT,
    SINGLEFLIGHT_FOLLOWER_TIMEOUT_EVENT,
    SINGLEFLIGHT_LEADER_COMPLETE_EVENT,
    SINGLEFLIGHT_COALESCED_RESULT_REUSED_EVENT,
];
const GRAPH_FEATURE_ENRICHMENT_FOLLOWER_TIMEOUT: Duration = Duration::from_secs(30);
const GRAPH_FEATURE_ENRICHMENT_BURST_SCHEMA_V1: &str =
    "ee.graph.feature_enrichment.singleflight_burst.v1";
const GRAPH_FEATURE_ENRICHMENT_BURST_FOLLOWER_JOIN_TIMEOUT: Duration = Duration::from_secs(2);
const GRAPH_FEATURE_ENRICHMENT_BURST_MAX_IDENTICAL_REQUESTS: usize = 32;
const GRAPH_FEATURE_ENRICHMENT_BURST_MAX_DISTINCT_REQUESTS: usize = 8;
const GRAPH_FEATURE_ENRICHMENT_BURST_MAX_NODE_COUNT: usize = 256;

static GRAPH_FEATURE_ENRICHMENT_GROUP: OnceLock<
    SingleFlightGroup<crate::graph::GraphFeatureEnrichmentReport>,
> = OnceLock::new();

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum SingleFlightRole {
    Leader,
    Follower,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct SingleFlightRun<T> {
    pub value: T,
    pub role: SingleFlightRole,
    pub shared: bool,
    pub degraded_codes: Vec<&'static str>,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum SingleFlightError {
    FollowerTimeout {
        key_hash: String,
        timeout_ms: u64,
    },
    LeaderFailed {
        key_hash: String,
        role: SingleFlightRole,
        message: String,
    },
    StatePoisoned {
        key_hash: String,
    },
}

impl SingleFlightError {
    #[must_use]
    pub const fn code(&self) -> &'static str {
        match self {
            Self::FollowerTimeout { .. } => SINGLEFLIGHT_FOLLOWER_TIMEOUT_CODE,
            Self::LeaderFailed { .. } => SINGLEFLIGHT_LEADER_FAILED_CODE,
            Self::StatePoisoned { .. } => SINGLEFLIGHT_STATE_POISONED_CODE,
        }
    }

    #[must_use]
    pub const fn severity(&self) -> &'static str {
        match self {
            Self::FollowerTimeout { .. } => "medium",
            Self::LeaderFailed { .. } => "medium",
            Self::StatePoisoned { .. } => "high",
        }
    }

    #[must_use]
    pub const fn repair(&self) -> &'static str {
        match self {
            Self::FollowerTimeout { .. } => "Retry the read with a longer wait budget.",
            Self::LeaderFailed { .. } => {
                "Inspect the leader operation error; followers observed the same failure."
            }
            Self::StatePoisoned { .. } => {
                "Restart the process to clear poisoned in-memory single-flight state."
            }
        }
    }
}

impl std::fmt::Display for SingleFlightError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::FollowerTimeout {
                key_hash,
                timeout_ms,
            } => write!(
                formatter,
                "follower timed out waiting for key {key_hash} after {timeout_ms}ms"
            ),
            Self::LeaderFailed {
                key_hash,
                role,
                message,
            } => write!(
                formatter,
                "{role:?} observed leader failure for key {key_hash}: {message}"
            ),
            Self::StatePoisoned { key_hash } => {
                write!(
                    formatter,
                    "single-flight state was poisoned for key {key_hash}"
                )
            }
        }
    }
}

impl std::error::Error for SingleFlightError {}

#[derive(Debug)]
pub struct SingleFlightGroup<T> {
    entries: Mutex<HashMap<String, Arc<SingleFlightEntry<T>>>>,
    counters: SingleFlightCounters,
    last_key: Mutex<Option<SingleFlightLastKeyPosture>>,
}

impl<T> Default for SingleFlightGroup<T> {
    fn default() -> Self {
        Self {
            entries: Mutex::new(HashMap::new()),
            counters: SingleFlightCounters::default(),
            last_key: Mutex::new(None),
        }
    }
}

impl<T> SingleFlightGroup<T>
where
    T: Clone,
{
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    pub fn run<F>(
        &self,
        key: &SingleFlightKey,
        follower_timeout: Duration,
        operation: F,
    ) -> Result<SingleFlightRun<T>, SingleFlightError>
    where
        F: FnOnce() -> Result<T, String>,
    {
        self.record_last_key(key);
        let key_hash = key.key_hash.clone();
        let (entry, is_leader) = self.entry_for(&key_hash)?;

        if is_leader {
            let leader_started = Instant::now();

            struct LeaderGuard<'a, T: Clone> {
                group: &'a SingleFlightGroup<T>,
                key_hash: String,
                entry: Arc<SingleFlightEntry<T>>,
                leader_started: Instant,
                completed: bool,
            }

            impl<T: Clone> Drop for LeaderGuard<'_, T> {
                fn drop(&mut self) {
                    if !self.completed {
                        let _ = self.group.remove_entry(&self.key_hash, &self.entry);
                        self.group
                            .counters
                            .leader_failures
                            .fetch_add(1, Ordering::SeqCst);
                        trace_singleflight_checkpoint(
                            SINGLEFLIGHT_LEADER_FAILED_EVENT,
                            &self.key_hash,
                            duration_ms(self.leader_started.elapsed()),
                            &[SINGLEFLIGHT_LEADER_FAILED_CODE],
                        );
                        if let Ok(mut state) = self.entry.state.lock() {
                            *state =
                                SingleFlightState::Completed(Err("leader panicked".to_owned()));
                            self.entry.ready.notify_all();
                        } else {
                            self.group
                                .counters
                                .state_poisoned
                                .fetch_add(1, Ordering::SeqCst);
                            self.entry.ready.notify_all();
                        }
                    }
                }
            }

            let mut guard = LeaderGuard {
                group: self,
                key_hash: key_hash.clone(),
                entry: Arc::clone(&entry),
                leader_started,
                completed: false,
            };

            let result = operation();
            let completion = self.complete_leader(&key_hash, &entry, leader_started, result);
            if !matches!(&completion, Err(SingleFlightError::StatePoisoned { .. })) {
                guard.completed = true;
            }
            completion
        } else {
            self.counters.follower_joins.fetch_add(1, Ordering::SeqCst);
            trace_singleflight_checkpoint(SINGLEFLIGHT_FOLLOWER_JOIN_EVENT, &key_hash, 0, &[]);
            match self.wait_for_leader(&key_hash, &entry, follower_timeout) {
                Err(SingleFlightError::FollowerTimeout {
                    key_hash,
                    timeout_ms,
                }) => Self::run_follower_timeout_fallback(&key_hash, timeout_ms, operation),
                other => other,
            }
        }
    }

    pub fn active_len(&self) -> Result<usize, SingleFlightError> {
        let entries = self
            .entries
            .lock()
            .map_err(|_| SingleFlightError::StatePoisoned {
                key_hash: "<group>".to_owned(),
            })?;
        Ok(entries.len())
    }

    pub fn posture(
        &self,
        surface: SingleFlightSurface,
        configured: bool,
        follower_timeout: Duration,
    ) -> SingleFlightSurfacePosture {
        let last_key = self.last_key_snapshot();
        let active_leader_count = match self.active_len() {
            Ok(count) => capped_u32(count),
            Err(_) => {
                self.counters.state_poisoned.fetch_add(1, Ordering::SeqCst);
                0
            }
        };
        SingleFlightSurfacePosture::new(
            surface,
            configured,
            active_leader_count,
            self.counters.snapshot(),
            duration_ms(follower_timeout),
            last_key,
        )
    }

    fn record_last_key(&self, key: &SingleFlightKey) {
        match self.last_key.lock() {
            Ok(mut last_key) => {
                *last_key = Some(SingleFlightLastKeyPosture::from_key(key));
            }
            Err(_) => {
                self.counters.state_poisoned.fetch_add(1, Ordering::SeqCst);
            }
        }
    }

    fn last_key_snapshot(&self) -> Option<SingleFlightLastKeyPosture> {
        match self.last_key.lock() {
            Ok(last_key) => last_key.clone(),
            Err(_) => {
                self.counters.state_poisoned.fetch_add(1, Ordering::SeqCst);
                None
            }
        }
    }

    fn entry_for(
        &self,
        key_hash: &str,
    ) -> Result<(Arc<SingleFlightEntry<T>>, bool), SingleFlightError> {
        let mut entries = self
            .entries
            .lock()
            .map_err(|_| SingleFlightError::StatePoisoned {
                key_hash: key_hash.to_owned(),
            })?;
        if let Some(entry) = entries.get(key_hash) {
            entry.followers.fetch_add(1, Ordering::SeqCst);
            return Ok((Arc::clone(entry), false));
        }

        let entry = Arc::new(SingleFlightEntry::default());
        entries.insert(key_hash.to_owned(), Arc::clone(&entry));
        self.counters.leader_starts.fetch_add(1, Ordering::SeqCst);
        trace_singleflight_checkpoint(SINGLEFLIGHT_LEADER_START_EVENT, key_hash, 0, &[]);
        Ok((entry, true))
    }

    fn complete_leader(
        &self,
        key_hash: &str,
        entry: &Arc<SingleFlightEntry<T>>,
        leader_started: Instant,
        result: Result<T, String>,
    ) -> Result<SingleFlightRun<T>, SingleFlightError> {
        {
            let mut state = entry
                .state
                .lock()
                .map_err(|_| SingleFlightError::StatePoisoned {
                    key_hash: key_hash.to_owned(),
                })?;
            *state = SingleFlightState::Completed(result.clone());
            entry.ready.notify_all();
        }
        self.remove_entry(key_hash, entry)?;

        match result {
            Ok(value) => {
                self.counters
                    .completed_leaders
                    .fetch_add(1, Ordering::SeqCst);
                trace_singleflight_checkpoint(
                    SINGLEFLIGHT_LEADER_COMPLETE_EVENT,
                    key_hash,
                    duration_ms(leader_started.elapsed()),
                    &[],
                );
                Ok(SingleFlightRun {
                    value,
                    role: SingleFlightRole::Leader,
                    shared: entry.followers.load(Ordering::SeqCst) > 0,
                    degraded_codes: Vec::new(),
                })
            }
            Err(message) => {
                self.counters.leader_failures.fetch_add(1, Ordering::SeqCst);
                trace_singleflight_checkpoint(
                    SINGLEFLIGHT_LEADER_FAILED_EVENT,
                    key_hash,
                    duration_ms(leader_started.elapsed()),
                    &[SINGLEFLIGHT_LEADER_FAILED_CODE],
                );
                Err(SingleFlightError::LeaderFailed {
                    key_hash: key_hash.to_owned(),
                    role: SingleFlightRole::Leader,
                    message,
                })
            }
        }
    }

    fn wait_for_leader(
        &self,
        key_hash: &str,
        entry: &SingleFlightEntry<T>,
        follower_timeout: Duration,
    ) -> Result<SingleFlightRun<T>, SingleFlightError> {
        let started = Instant::now();
        let deadline = match started.checked_add(follower_timeout) {
            Some(deadline) => deadline,
            None => {
                self.counters
                    .follower_timeouts
                    .fetch_add(1, Ordering::SeqCst);
                trace_singleflight_checkpoint(
                    SINGLEFLIGHT_FOLLOWER_TIMEOUT_EVENT,
                    key_hash,
                    0,
                    &[SINGLEFLIGHT_FOLLOWER_TIMEOUT_CODE],
                );
                return Err(SingleFlightError::FollowerTimeout {
                    key_hash: key_hash.to_owned(),
                    timeout_ms: duration_ms(follower_timeout),
                });
            }
        };
        let mut state = entry
            .state
            .lock()
            .map_err(|_| SingleFlightError::StatePoisoned {
                key_hash: key_hash.to_owned(),
            })?;

        loop {
            match &*state {
                SingleFlightState::Pending => {
                    let remaining = match deadline.checked_duration_since(Instant::now()) {
                        Some(remaining) if !remaining.is_zero() => remaining,
                        _ => {
                            self.counters
                                .follower_timeouts
                                .fetch_add(1, Ordering::SeqCst);
                            trace_singleflight_checkpoint(
                                SINGLEFLIGHT_FOLLOWER_TIMEOUT_EVENT,
                                key_hash,
                                duration_ms(started.elapsed()),
                                &[SINGLEFLIGHT_FOLLOWER_TIMEOUT_CODE],
                            );
                            return Err(SingleFlightError::FollowerTimeout {
                                key_hash: key_hash.to_owned(),
                                timeout_ms: duration_ms(follower_timeout),
                            });
                        }
                    };
                    let (next_state, wait) =
                        entry.ready.wait_timeout(state, remaining).map_err(|_| {
                            SingleFlightError::StatePoisoned {
                                key_hash: key_hash.to_owned(),
                            }
                        })?;
                    state = next_state;
                    if wait.timed_out() && matches!(*state, SingleFlightState::Pending) {
                        self.counters
                            .follower_timeouts
                            .fetch_add(1, Ordering::SeqCst);
                        trace_singleflight_checkpoint(
                            SINGLEFLIGHT_FOLLOWER_TIMEOUT_EVENT,
                            key_hash,
                            duration_ms(started.elapsed()),
                            &[SINGLEFLIGHT_FOLLOWER_TIMEOUT_CODE],
                        );
                        return Err(SingleFlightError::FollowerTimeout {
                            key_hash: key_hash.to_owned(),
                            timeout_ms: duration_ms(follower_timeout),
                        });
                    }
                }
                SingleFlightState::Completed(Ok(value)) => {
                    self.counters.reused_results.fetch_add(1, Ordering::SeqCst);
                    trace_singleflight_checkpoint(
                        SINGLEFLIGHT_COALESCED_RESULT_REUSED_EVENT,
                        key_hash,
                        duration_ms(started.elapsed()),
                        &[],
                    );
                    return Ok(SingleFlightRun {
                        value: value.clone(),
                        role: SingleFlightRole::Follower,
                        shared: true,
                        degraded_codes: Vec::new(),
                    });
                }
                SingleFlightState::Completed(Err(message)) => {
                    trace_singleflight_checkpoint(
                        SINGLEFLIGHT_LEADER_FAILED_EVENT,
                        key_hash,
                        duration_ms(started.elapsed()),
                        &[SINGLEFLIGHT_LEADER_FAILED_CODE],
                    );
                    return Err(SingleFlightError::LeaderFailed {
                        key_hash: key_hash.to_owned(),
                        role: SingleFlightRole::Follower,
                        message: message.clone(),
                    });
                }
            }
        }
    }

    fn run_follower_timeout_fallback<F>(
        key_hash: &str,
        timeout_ms: u64,
        operation: F,
    ) -> Result<SingleFlightRun<T>, SingleFlightError>
    where
        F: FnOnce() -> Result<T, String>,
    {
        match operation() {
            Ok(value) => Ok(SingleFlightRun {
                value,
                role: SingleFlightRole::Follower,
                shared: false,
                degraded_codes: vec![SINGLEFLIGHT_FOLLOWER_TIMEOUT_CODE],
            }),
            Err(message) => Err(SingleFlightError::LeaderFailed {
                key_hash: key_hash.to_owned(),
                role: SingleFlightRole::Follower,
                message: format!("follower fallback after {timeout_ms}ms failed: {message}"),
            }),
        }
    }

    fn remove_entry(
        &self,
        key_hash: &str,
        entry: &Arc<SingleFlightEntry<T>>,
    ) -> Result<(), SingleFlightError> {
        let mut entries = self
            .entries
            .lock()
            .map_err(|_| SingleFlightError::StatePoisoned {
                key_hash: key_hash.to_owned(),
            })?;
        if entries
            .get(key_hash)
            .is_some_and(|current| Arc::ptr_eq(current, entry))
        {
            entries.remove(key_hash);
        }
        Ok(())
    }
}

#[derive(Debug, Default)]
struct SingleFlightCounters {
    leader_starts: AtomicU64,
    completed_leaders: AtomicU64,
    follower_joins: AtomicU64,
    follower_timeouts: AtomicU64,
    leader_failures: AtomicU64,
    reused_results: AtomicU64,
    state_poisoned: AtomicU64,
}

impl SingleFlightCounters {
    fn snapshot(&self) -> SingleFlightSurfaceCounters {
        SingleFlightSurfaceCounters {
            leader_start_count: self.leader_starts.load(Ordering::SeqCst),
            completed_leader_count: self.completed_leaders.load(Ordering::SeqCst),
            follower_join_count: self.follower_joins.load(Ordering::SeqCst),
            follower_timeout_count: self.follower_timeouts.load(Ordering::SeqCst),
            leader_failure_count: self.leader_failures.load(Ordering::SeqCst),
            reused_result_count: self.reused_results.load(Ordering::SeqCst),
            state_poisoned_count: self.state_poisoned.load(Ordering::SeqCst),
        }
    }
}

#[must_use]
pub fn singleflight_posture_report() -> SingleFlightPostureReport {
    let graph_feature_enrichment = GRAPH_FEATURE_ENRICHMENT_GROUP.get().map_or_else(
        || {
            SingleFlightSurfacePosture::new(
                SingleFlightSurface::GraphFeatureEnrichment,
                true,
                0,
                SingleFlightSurfaceCounters::default(),
                duration_ms(GRAPH_FEATURE_ENRICHMENT_FOLLOWER_TIMEOUT),
                None,
            )
        },
        |group| {
            group.posture(
                SingleFlightSurface::GraphFeatureEnrichment,
                true,
                GRAPH_FEATURE_ENRICHMENT_FOLLOWER_TIMEOUT,
            )
        },
    );

    SingleFlightPostureReport::from_surfaces(vec![graph_feature_enrichment])
}

pub fn run_graph_feature_enrichment<F>(
    workspace_identity: &str,
    workspace_generation: u64,
    graph_generation: Option<u64>,
    source_mode: &str,
    options: &crate::graph::GraphFeatureEnrichmentOptions,
    operation: F,
) -> Result<SingleFlightRun<crate::graph::GraphFeatureEnrichmentReport>, SingleFlightError>
where
    F: FnOnce() -> crate::graph::GraphFeatureEnrichmentReport,
{
    run_graph_feature_enrichment_with_group(
        GraphFeatureEnrichmentSingleFlightInput {
            group: GRAPH_FEATURE_ENRICHMENT_GROUP.get_or_init(SingleFlightGroup::new),
            follower_timeout: GRAPH_FEATURE_ENRICHMENT_FOLLOWER_TIMEOUT,
            workspace_identity,
            workspace_generation,
            graph_generation,
            source_mode,
            options,
        },
        operation,
    )
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct GraphFeatureEnrichmentBurstOptions {
    pub identical_requests: usize,
    pub distinct_requests: usize,
    pub node_count: usize,
}

impl Default for GraphFeatureEnrichmentBurstOptions {
    fn default() -> Self {
        Self {
            identical_requests: 6,
            distinct_requests: 2,
            node_count: 64,
        }
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct GraphFeatureEnrichmentBurstReport {
    pub schema: &'static str,
    pub workspace_identity_hash: String,
    pub requested_identical: usize,
    pub requested_distinct: usize,
    pub requested_node_count: usize,
    pub effective_identical: usize,
    pub effective_distinct: usize,
    pub effective_node_count: usize,
    pub execution_count: usize,
    pub identical_leader_count: usize,
    pub identical_follower_count: usize,
    pub distinct_leader_count: usize,
    pub distinct_follower_count: usize,
    pub timeout_count: usize,
    pub leader_failure_count: usize,
    pub state_poisoned_count: usize,
    pub latency_ms: GraphFeatureEnrichmentBurstLatency,
    pub identical_result_hashes: Vec<String>,
    pub distinct_result_hashes: Vec<String>,
    pub passed: bool,
    pub diagnoses: Vec<String>,
}

impl GraphFeatureEnrichmentBurstReport {
    #[must_use]
    pub fn data_json(&self) -> serde_json::Value {
        serde_json::json!({
            "schema": self.schema,
            "command": "graph feature-enrichment --singleflight-burst",
            "workspaceIdentityHash": &self.workspace_identity_hash,
            "requested": {
                "identical": self.requested_identical,
                "distinct": self.requested_distinct,
                "nodeCount": self.requested_node_count,
            },
            "effective": {
                "identical": self.effective_identical,
                "distinct": self.effective_distinct,
                "nodeCount": self.effective_node_count,
            },
            "limits": {
                "maxIdentical": GRAPH_FEATURE_ENRICHMENT_BURST_MAX_IDENTICAL_REQUESTS,
                "maxDistinct": GRAPH_FEATURE_ENRICHMENT_BURST_MAX_DISTINCT_REQUESTS,
                "maxNodeCount": GRAPH_FEATURE_ENRICHMENT_BURST_MAX_NODE_COUNT,
            },
            "summary": {
                "passed": self.passed,
                "executionCount": self.execution_count,
                "identicalLeaderCount": self.identical_leader_count,
                "identicalFollowerCount": self.identical_follower_count,
                "distinctLeaderCount": self.distinct_leader_count,
                "distinctFollowerCount": self.distinct_follower_count,
                "timeoutCount": self.timeout_count,
                "leaderFailureCount": self.leader_failure_count,
                "statePoisonedCount": self.state_poisoned_count,
            },
            "latencyMs": {
                "p50": self.latency_ms.p50,
                "p95": self.latency_ms.p95,
                "p99": self.latency_ms.p99,
                "max": self.latency_ms.max,
            },
            "resultHashes": {
                "identical": &self.identical_result_hashes,
                "identicalUniqueCount": self.identical_result_hashes.len(),
                "distinct": &self.distinct_result_hashes,
                "distinctUniqueCount": self.distinct_result_hashes.len(),
            },
            "diagnoses": &self.diagnoses,
        })
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct GraphFeatureEnrichmentBurstLatency {
    pub p50: u64,
    pub p95: u64,
    pub p99: u64,
    pub max: u64,
}

#[must_use]
pub fn run_graph_feature_enrichment_burst_smoke(
    workspace_identity: &str,
    enrichment_options: &crate::graph::GraphFeatureEnrichmentOptions,
    burst_options: &GraphFeatureEnrichmentBurstOptions,
) -> GraphFeatureEnrichmentBurstReport {
    let requested_identical = burst_options.identical_requests;
    let requested_distinct = burst_options.distinct_requests;
    let requested_node_count = burst_options.node_count;
    let identical_count =
        requested_identical.clamp(2, GRAPH_FEATURE_ENRICHMENT_BURST_MAX_IDENTICAL_REQUESTS);
    let distinct_count =
        requested_distinct.min(GRAPH_FEATURE_ENRICHMENT_BURST_MAX_DISTINCT_REQUESTS);
    let node_count = requested_node_count.clamp(1, GRAPH_FEATURE_ENRICHMENT_BURST_MAX_NODE_COUNT);
    let mut input_diagnoses = burst_input_diagnoses(
        requested_identical,
        requested_distinct,
        requested_node_count,
        identical_count,
        distinct_count,
        node_count,
    );
    let group = Arc::new(SingleFlightGroup::new());
    let executions = Arc::new(AtomicUsize::new(0));
    let identical_barrier = Arc::new(Barrier::new(identical_count));
    let identical_key = graph_feature_enrichment_singleflight_key(
        workspace_identity,
        77,
        Some(13),
        "burst_smoke",
        enrichment_options,
    );
    let mut handles = Vec::with_capacity(identical_count.saturating_add(distinct_count));

    for _ in 0..identical_count {
        let group = Arc::clone(&group);
        let executions = Arc::clone(&executions);
        let options = enrichment_options.clone();
        let identical_barrier = Arc::clone(&identical_barrier);
        let identical_key_hash = identical_key.key_hash.clone();
        let workspace_identity = workspace_identity.to_owned();
        handles.push(thread::spawn(move || {
            identical_barrier.wait();
            let started = Instant::now();
            let outcome = run_graph_feature_enrichment_with_group(
                GraphFeatureEnrichmentSingleFlightInput {
                    group: &group,
                    follower_timeout: Duration::from_secs(5),
                    workspace_identity: &workspace_identity,
                    workspace_generation: 77,
                    graph_generation: Some(13),
                    source_mode: "burst_smoke",
                    options: &options,
                },
                || {
                    wait_for_followers(
                        &group,
                        &identical_key_hash,
                        identical_count.saturating_sub(1),
                        GRAPH_FEATURE_ENRICHMENT_BURST_FOLLOWER_JOIN_TIMEOUT,
                    );
                    executions.fetch_add(1, Ordering::SeqCst);
                    crate::graph::enrich_graph_features(
                        &burst_centrality_report(node_count, 0),
                        &options,
                    )
                },
            );
            BurstOutcome::from_run("identical", started.elapsed(), outcome)
        }));
    }

    if distinct_count > 0 {
        let distinct_barrier = Arc::new(Barrier::new(distinct_count));
        for index in 0..distinct_count {
            let group = Arc::clone(&group);
            let executions = Arc::clone(&executions);
            let options = enrichment_options.clone();
            let distinct_barrier = Arc::clone(&distinct_barrier);
            let workspace_identity = workspace_identity.to_owned();
            handles.push(thread::spawn(move || {
                distinct_barrier.wait();
                let started = Instant::now();
                let outcome = run_graph_feature_enrichment_with_group(
                    GraphFeatureEnrichmentSingleFlightInput {
                        group: &group,
                        follower_timeout: Duration::from_secs(5),
                        workspace_identity: &workspace_identity,
                        workspace_generation: 100_u64
                            .saturating_add(u64::try_from(index).unwrap_or(u64::MAX)),
                        graph_generation: Some(
                            200_u64.saturating_add(u64::try_from(index).unwrap_or(u64::MAX)),
                        ),
                        source_mode: "burst_smoke",
                        options: &options,
                    },
                    || {
                        executions.fetch_add(1, Ordering::SeqCst);
                        crate::graph::enrich_graph_features(
                            &burst_centrality_report(node_count, index.saturating_add(1)),
                            &options,
                        )
                    },
                );
                BurstOutcome::from_run("distinct", started.elapsed(), outcome)
            }));
        }
    }

    let mut outcomes = Vec::with_capacity(handles.len());
    for handle in handles {
        match handle.join() {
            Ok(outcome) => outcomes.push(outcome),
            Err(_) => outcomes.push(BurstOutcome::thread_panic()),
        }
    }

    summarize_burst(
        workspace_identity,
        requested_identical,
        requested_distinct,
        requested_node_count,
        identical_count,
        distinct_count,
        node_count,
        executions.load(Ordering::SeqCst),
        &outcomes,
        &mut input_diagnoses,
    )
}

fn burst_input_diagnoses(
    requested_identical: usize,
    requested_distinct: usize,
    requested_node_count: usize,
    effective_identical: usize,
    effective_distinct: usize,
    effective_node_count: usize,
) -> Vec<String> {
    let mut diagnoses = Vec::new();
    if requested_identical > effective_identical {
        diagnoses.push(format!(
            "capped identical requests from {requested_identical} to {effective_identical}"
        ));
    } else if requested_identical < effective_identical {
        diagnoses.push(format!(
            "raised identical requests from {requested_identical} to {effective_identical}"
        ));
    }
    if requested_distinct > effective_distinct {
        diagnoses.push(format!(
            "capped distinct requests from {requested_distinct} to {effective_distinct}"
        ));
    }
    if requested_node_count > effective_node_count {
        diagnoses.push(format!(
            "capped node count from {requested_node_count} to {effective_node_count}"
        ));
    } else if requested_node_count < effective_node_count {
        diagnoses.push(format!(
            "raised node count from {requested_node_count} to {effective_node_count}"
        ));
    }
    diagnoses
}

struct GraphFeatureEnrichmentSingleFlightInput<'a> {
    group: &'a SingleFlightGroup<crate::graph::GraphFeatureEnrichmentReport>,
    follower_timeout: Duration,
    workspace_identity: &'a str,
    workspace_generation: u64,
    graph_generation: Option<u64>,
    source_mode: &'a str,
    options: &'a crate::graph::GraphFeatureEnrichmentOptions,
}

fn run_graph_feature_enrichment_with_group<F>(
    input: GraphFeatureEnrichmentSingleFlightInput<'_>,
    operation: F,
) -> Result<SingleFlightRun<crate::graph::GraphFeatureEnrichmentReport>, SingleFlightError>
where
    F: FnOnce() -> crate::graph::GraphFeatureEnrichmentReport,
{
    let key = graph_feature_enrichment_singleflight_key(
        input.workspace_identity,
        input.workspace_generation,
        input.graph_generation,
        input.source_mode,
        input.options,
    );

    let mut run = input
        .group
        .run(&key, input.follower_timeout, || Ok(operation()))?;
    if run
        .degraded_codes
        .contains(&SINGLEFLIGHT_FOLLOWER_TIMEOUT_CODE)
        && !run
            .value
            .degraded
            .iter()
            .any(|entry| entry.code == SINGLEFLIGHT_FOLLOWER_TIMEOUT_CODE)
    {
        run.value
            .degraded
            .push(graph_feature_enrichment_timeout_degradation());
    }
    Ok(run)
}

fn graph_feature_enrichment_singleflight_key(
    workspace_identity: &str,
    workspace_generation: u64,
    graph_generation: Option<u64>,
    source_mode: &str,
    options: &crate::graph::GraphFeatureEnrichmentOptions,
) -> SingleFlightKey {
    let max_features = options.max_features.to_string();
    let min_combined_score = stable_f64_option(options.min_combined_score);
    let max_selection_boost = stable_f64_option(options.max_selection_boost);
    let option_pairs = [
        ("source_mode", source_mode),
        ("max_features", max_features.as_str()),
        ("min_combined_score_bits", min_combined_score.as_str()),
        ("max_selection_boost_bits", max_selection_boost.as_str()),
    ];
    let mut key_input = SingleFlightKeyInput::new(
        SingleFlightSurface::GraphFeatureEnrichment,
        workspace_identity,
        workspace_generation,
        crate::graph::GRAPH_FEATURE_ENRICHMENT_SCHEMA_V1,
    );
    key_input.graph_generation = graph_generation;
    key_input.option_pairs = &option_pairs;
    SingleFlightKey::from_input(&key_input)
}

fn wait_for_followers<T>(
    group: &SingleFlightGroup<T>,
    key_hash: &str,
    expected_followers: usize,
    timeout: Duration,
) {
    if expected_followers == 0 {
        return;
    }
    let started = Instant::now();
    loop {
        let follower_count = group.entries.lock().map_or(0, |entries| {
            entries
                .get(key_hash)
                .map_or(0, |entry| entry.followers.load(Ordering::SeqCst))
        });
        if follower_count >= expected_followers || started.elapsed() >= timeout {
            return;
        }
        thread::sleep(Duration::from_millis(1));
    }
}

fn stable_f64_option(value: f64) -> String {
    format!("{:016x}", value.to_bits())
}

fn graph_feature_enrichment_timeout_degradation() -> crate::graph::GraphFeatureEnrichmentDegradation
{
    crate::graph::GraphFeatureEnrichmentDegradation {
        code: SINGLEFLIGHT_FOLLOWER_TIMEOUT_CODE,
        severity: "medium",
        message:
            "Single-flight follower timed out; computed graph feature enrichment independently."
                .to_owned(),
        repair: "Retry the read with a longer wait budget.".to_owned(),
    }
}

#[derive(Debug)]
struct BurstOutcome {
    request_kind: &'static str,
    role: Option<SingleFlightRole>,
    shared: bool,
    result_hash: Option<String>,
    elapsed_ms: u64,
    error_code: Option<&'static str>,
}

impl BurstOutcome {
    fn from_run(
        request_kind: &'static str,
        elapsed: Duration,
        outcome: Result<
            SingleFlightRun<crate::graph::GraphFeatureEnrichmentReport>,
            SingleFlightError,
        >,
    ) -> Self {
        match outcome {
            Ok(run) => {
                let error_code = run.degraded_codes.first().copied();
                Self {
                    request_kind,
                    role: Some(run.role),
                    shared: run.shared,
                    result_hash: Some(graph_feature_report_hash(&run.value)),
                    elapsed_ms: duration_ms(elapsed),
                    error_code,
                }
            }
            Err(error) => Self {
                request_kind,
                role: None,
                shared: false,
                result_hash: None,
                elapsed_ms: duration_ms(elapsed),
                error_code: Some(error.code()),
            },
        }
    }

    fn thread_panic() -> Self {
        Self {
            request_kind: "panic",
            role: None,
            shared: false,
            result_hash: None,
            elapsed_ms: 0,
            error_code: Some(SINGLEFLIGHT_LEADER_FAILED_CODE),
        }
    }
}

fn summarize_burst(
    workspace_identity: &str,
    requested_identical: usize,
    requested_distinct: usize,
    requested_node_count: usize,
    identical_count: usize,
    distinct_count: usize,
    node_count: usize,
    execution_count: usize,
    outcomes: &[BurstOutcome],
    input_diagnoses: &mut Vec<String>,
) -> GraphFeatureEnrichmentBurstReport {
    let mut identical_leader_count = 0;
    let mut identical_follower_count = 0;
    let mut distinct_leader_count = 0;
    let mut distinct_follower_count = 0;
    let mut timeout_count = 0;
    let mut leader_failure_count = 0;
    let mut state_poisoned_count = 0;
    let mut identical_hashes = BTreeSet::new();
    let mut distinct_hashes = BTreeSet::new();
    let mut latency_values = Vec::with_capacity(outcomes.len());

    for outcome in outcomes {
        latency_values.push(outcome.elapsed_ms);
        match (outcome.request_kind, outcome.role) {
            ("identical", Some(SingleFlightRole::Leader)) => identical_leader_count += 1,
            ("identical", Some(SingleFlightRole::Follower)) => identical_follower_count += 1,
            ("distinct", Some(SingleFlightRole::Leader)) => distinct_leader_count += 1,
            ("distinct", Some(SingleFlightRole::Follower)) => distinct_follower_count += 1,
            _ => {}
        }
        match outcome.error_code {
            Some(SINGLEFLIGHT_FOLLOWER_TIMEOUT_CODE) => timeout_count += 1,
            Some(SINGLEFLIGHT_LEADER_FAILED_CODE) => leader_failure_count += 1,
            Some(SINGLEFLIGHT_STATE_POISONED_CODE) => state_poisoned_count += 1,
            _ => {}
        }
        if let Some(hash) = &outcome.result_hash {
            match outcome.request_kind {
                "identical" => {
                    identical_hashes.insert(hash.clone());
                }
                "distinct" => {
                    distinct_hashes.insert(hash.clone());
                }
                _ => {}
            }
        }
    }

    let identical_shared_count = outcomes
        .iter()
        .filter(|outcome| outcome.request_kind == "identical" && outcome.shared)
        .count();
    let mut diagnoses = std::mem::take(input_diagnoses);
    if identical_leader_count != 1 {
        diagnoses.push(format!(
            "expected one identical leader, got {identical_leader_count}"
        ));
    }
    if identical_follower_count != identical_count.saturating_sub(1) {
        diagnoses.push(format!(
            "expected {} identical followers, got {identical_follower_count}",
            identical_count.saturating_sub(1)
        ));
    }
    if identical_shared_count != identical_count {
        diagnoses.push(format!(
            "expected every identical response to be marked shared, got {identical_shared_count}"
        ));
    }
    if distinct_leader_count != distinct_count {
        diagnoses.push(format!(
            "expected {distinct_count} distinct leaders, got {distinct_leader_count}"
        ));
    }
    if distinct_follower_count != 0 {
        diagnoses.push(format!(
            "expected zero distinct followers, got {distinct_follower_count}"
        ));
    }
    if identical_hashes.len() != 1 {
        diagnoses.push(format!(
            "expected one identical result hash, got {}",
            identical_hashes.len()
        ));
    }
    if distinct_hashes.len() != distinct_count {
        diagnoses.push(format!(
            "expected {distinct_count} distinct result hashes, got {}",
            distinct_hashes.len()
        ));
    }
    if timeout_count > 0 || leader_failure_count > 0 || state_poisoned_count > 0 {
        diagnoses.push(format!(
            "observed errors: timeout={timeout_count}, leader_failure={leader_failure_count}, state_poisoned={state_poisoned_count}"
        ));
    }
    let expected_executions = 1_usize.saturating_add(distinct_count);
    if execution_count != expected_executions {
        diagnoses.push(format!(
            "expected {expected_executions} expensive executions, got {execution_count}"
        ));
    }

    GraphFeatureEnrichmentBurstReport {
        schema: GRAPH_FEATURE_ENRICHMENT_BURST_SCHEMA_V1,
        workspace_identity_hash: format!(
            "blake3:{}",
            blake3::hash(workspace_identity.as_bytes()).to_hex()
        ),
        requested_identical,
        requested_distinct,
        requested_node_count,
        effective_identical: identical_count,
        effective_distinct: distinct_count,
        effective_node_count: node_count,
        execution_count,
        identical_leader_count,
        identical_follower_count,
        distinct_leader_count,
        distinct_follower_count,
        timeout_count,
        leader_failure_count,
        state_poisoned_count,
        latency_ms: burst_latency(&latency_values),
        identical_result_hashes: identical_hashes.into_iter().collect(),
        distinct_result_hashes: distinct_hashes.into_iter().collect(),
        passed: diagnoses.is_empty(),
        diagnoses,
    }
}

fn graph_feature_report_hash(report: &crate::graph::GraphFeatureEnrichmentReport) -> String {
    let canonical = serde_json::to_string(&report.data_json())
        .unwrap_or_else(|error| format!("graph_feature_report_serialization_error:{error}"));
    format!("blake3:{}", blake3::hash(canonical.as_bytes()).to_hex())
}

fn burst_latency(values: &[u64]) -> GraphFeatureEnrichmentBurstLatency {
    GraphFeatureEnrichmentBurstLatency {
        p50: latency_percentile(values, 50),
        p95: latency_percentile(values, 95),
        p99: latency_percentile(values, 99),
        max: values.iter().copied().max().unwrap_or(0),
    }
}

fn latency_percentile(values: &[u64], percentile: usize) -> u64 {
    if values.is_empty() {
        return 0;
    }
    let mut sorted = values.to_vec();
    sorted.sort_unstable();
    let last_index = sorted.len().saturating_sub(1);
    let index = last_index.saturating_mul(percentile).saturating_add(50) / 100;
    sorted[index.min(last_index)]
}

fn burst_centrality_report(
    node_count: usize,
    salt: usize,
) -> crate::graph::CentralityRefreshReport {
    let denominator = node_count.max(1) as f64;
    let salt_offset = salt as f64 * 0.000_001;
    let scores: Vec<_> = (0..node_count)
        .map(|index| {
            let forward = (node_count.saturating_sub(index)) as f64 / denominator;
            let reverse = index.saturating_add(1) as f64 / denominator;
            crate::graph::MemoryCentralityScore {
                memory_id: format!("mem_{salt:04}_{index:05}"),
                pagerank: forward + salt_offset,
                betweenness: reverse + salt_offset,
                hub: 0.0,
                authority: 0.0,
            }
        })
        .collect();

    crate::graph::CentralityRefreshReport {
        version: env!("CARGO_PKG_VERSION"),
        status: crate::graph::CentralityRefreshStatus::Refreshed,
        pagerank_status: crate::graph::CentralityAlgorithmStatus::Computed,
        betweenness_status: crate::graph::CentralityAlgorithmStatus::Computed,
        hits_status: crate::graph::CentralityAlgorithmStatus::Computed,
        dry_run: false,
        node_count,
        edge_count: node_count.saturating_mul(2),
        projection_ms: 0.0,
        pagerank_ms: 0.0,
        betweenness_ms: 0.0,
        hits_ms: 0.0,
        total_ms: 0.0,
        scores,
        top_pagerank: Vec::new(),
        top_betweenness: Vec::new(),
        top_hubs: Vec::new(),
        top_authorities: Vec::new(),
    }
}

#[derive(Debug)]
struct SingleFlightEntry<T> {
    state: Mutex<SingleFlightState<T>>,
    ready: Condvar,
    followers: AtomicUsize,
}

impl<T> Default for SingleFlightEntry<T> {
    fn default() -> Self {
        Self {
            state: Mutex::new(SingleFlightState::Pending),
            ready: Condvar::new(),
            followers: AtomicUsize::new(0),
        }
    }
}

#[derive(Clone, Debug)]
enum SingleFlightState<T> {
    Pending,
    Completed(Result<T, String>),
}

fn duration_ms(duration: Duration) -> u64 {
    u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
}

fn capped_u32(value: usize) -> u32 {
    u32::try_from(value).unwrap_or(u32::MAX)
}

fn trace_singleflight_checkpoint(
    phase: &'static str,
    key_hash: &str,
    elapsed_ms: u64,
    degraded_codes: &[&str],
) {
    tracing::info!(
        target: "ee::singleflight",
        workspace_id = SINGLEFLIGHT_TRACE_WORKSPACE_ID,
        request_id = SINGLEFLIGHT_TRACE_REQUEST_ID,
        bead_id = option_env!("EE_TRACE_BEAD_ID").unwrap_or("bd-gni47.3"),
        surface = SINGLEFLIGHT_TRACE_SURFACE,
        phase,
        elapsed_ms,
        degraded_codes = ?degraded_codes,
        key_hash = %key_hash,
        "single-flight checkpoint"
    );
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::graph::{
        CentralityRefreshReport, CentralityRefreshStatus, GraphFeatureEnrichmentOptions,
        MemoryCentralityScore, enrich_graph_features,
    };
    use std::sync::{
        Arc, Barrier,
        atomic::{AtomicUsize, Ordering},
        mpsc,
    };
    use std::thread;

    type TestResult = Result<(), String>;

    fn key(label: &str) -> SingleFlightKey {
        let mut input = SingleFlightKeyInput::new(
            SingleFlightSurface::Context,
            "workspace-a",
            7,
            "ee.context.v1",
        );
        input.query_text = Some(label);
        SingleFlightKey::from_input(&input)
    }

    #[test]
    fn identical_concurrent_requests_share_one_leader() -> TestResult {
        let group = Arc::new(SingleFlightGroup::new());
        let key = key("same read");
        let calls = Arc::new(AtomicUsize::new(0));
        let barrier = Arc::new(Barrier::new(6));
        let mut handles = Vec::new();

        for _ in 0..6 {
            let group = Arc::clone(&group);
            let key = key.clone();
            let calls = Arc::clone(&calls);
            let barrier = Arc::clone(&barrier);
            handles.push(thread::spawn(move || {
                barrier.wait();
                group.run(&key, Duration::from_secs(5), || {
                    calls.fetch_add(1, Ordering::SeqCst);
                    thread::sleep(Duration::from_millis(50));
                    Ok("shared-result".to_owned())
                })
            }));
        }

        let mut leader_count = 0;
        let mut follower_count = 0;
        for handle in handles {
            let run = handle
                .join()
                .map_err(|_| "thread panicked".to_owned())?
                .map_err(|error| format!("single-flight run failed: {error:?}"))?;
            assert_eq!(run.value, "shared-result");
            match run.role {
                SingleFlightRole::Leader => leader_count += 1,
                SingleFlightRole::Follower => follower_count += 1,
            }
        }

        assert_eq!(calls.load(Ordering::SeqCst), 1);
        assert_eq!(leader_count, 1);
        assert_eq!(follower_count, 5);
        assert_eq!(group.active_len().map_err(|error| format!("{error:?}"))?, 0);
        let posture = group.posture(
            SingleFlightSurface::Context,
            true,
            Duration::from_millis(250),
        );
        assert_eq!(posture.status, "idle");
        assert_eq!(posture.active_leader_count, 0);
        assert_eq!(posture.leader_start_count, 1);
        assert_eq!(posture.completed_leader_count, 1);
        assert_eq!(posture.follower_join_count, 5);
        assert_eq!(posture.reused_result_count, 5);
        Ok(())
    }

    #[test]
    fn leader_reports_shared_when_follower_claimed_entry_before_completion() -> TestResult {
        let group = SingleFlightGroup::<String>::new();
        let key = key("fast shared read");
        let (leader_entry, is_leader) = group
            .entry_for(&key.key_hash)
            .map_err(|error| format!("leader entry failed: {error:?}"))?;
        assert!(is_leader);

        let (_follower_entry, follower_is_leader) = group
            .entry_for(&key.key_hash)
            .map_err(|error| format!("follower entry failed: {error:?}"))?;
        assert!(!follower_is_leader);

        let run = group
            .complete_leader(
                &key.key_hash,
                &leader_entry,
                Instant::now(),
                Ok("shared-result".to_owned()),
            )
            .map_err(|error| format!("leader completion failed: {error:?}"))?;
        assert_eq!(run.role, SingleFlightRole::Leader);
        assert!(
            run.shared,
            "leader must report a shared result once a follower has claimed the entry"
        );
        assert_eq!(group.active_len().map_err(|error| format!("{error:?}"))?, 0);
        Ok(())
    }

    #[test]
    fn posture_records_only_redaction_safe_last_key_generations() -> TestResult {
        let group = SingleFlightGroup::new();
        let mut input = SingleFlightKeyInput::new(
            SingleFlightSurface::Context,
            "/workspace/secret-path",
            11,
            "ee.context.v2",
        );
        input.index_generation = Some(17);
        input.graph_generation = Some(23);
        input.query_text = Some("raw query with token secret should never surface");
        let key = SingleFlightKey::from_input(&input);

        group
            .run(&key, Duration::from_secs(1), || Ok("ok".to_owned()))
            .map_err(|error| format!("single-flight run failed: {error}"))?;

        let posture = group.posture(SingleFlightSurface::Context, true, Duration::from_secs(1));
        let last_key = posture
            .last_key
            .as_ref()
            .ok_or_else(|| "posture should include last key".to_owned())?;
        assert_eq!(last_key.key_hash, key.key_hash);
        assert_eq!(last_key.workspace_generation, 11);
        assert_eq!(last_key.index_generation, Some(17));
        assert_eq!(last_key.graph_generation, Some(23));

        let serialized = serde_json::to_string(&posture)
            .map_err(|error| format!("serialize single-flight posture: {error}"))?;
        assert!(serialized.contains(&key.key_hash));
        assert!(!serialized.contains("raw query"));
        assert!(!serialized.contains("token secret"));
        assert!(!serialized.contains("/workspace/secret-path"));
        Ok(())
    }

    #[test]
    fn distinct_keys_execute_independently() -> TestResult {
        let group = Arc::new(SingleFlightGroup::new());
        let calls = Arc::new(AtomicUsize::new(0));
        let barrier = Arc::new(Barrier::new(4));
        let mut handles = Vec::new();

        for index in 0..4 {
            let group = Arc::clone(&group);
            let key = key(&format!("query-{index}"));
            let calls = Arc::clone(&calls);
            let barrier = Arc::clone(&barrier);
            handles.push(thread::spawn(move || {
                barrier.wait();
                group.run(&key, Duration::from_secs(5), || {
                    calls.fetch_add(1, Ordering::SeqCst);
                    Ok(index)
                })
            }));
        }

        for handle in handles {
            let run = handle
                .join()
                .map_err(|_| "thread panicked".to_owned())?
                .map_err(|error| format!("single-flight run failed: {error:?}"))?;
            assert_eq!(run.role, SingleFlightRole::Leader);
            assert!(!run.shared);
        }
        assert_eq!(calls.load(Ordering::SeqCst), 4);
        Ok(())
    }

    #[test]
    fn follower_timeout_fails_open_and_does_not_cancel_leader() -> TestResult {
        let group = Arc::new(SingleFlightGroup::new());
        let key = key("slow read");
        let calls = Arc::new(AtomicUsize::new(0));
        let (leader_started_tx, leader_started_rx) = mpsc::channel();

        let leader_group = Arc::clone(&group);
        let leader_key = key.clone();
        let leader_calls = Arc::clone(&calls);
        let leader = thread::spawn(move || {
            leader_group.run(&leader_key, Duration::from_secs(5), || {
                leader_calls.fetch_add(1, Ordering::SeqCst);
                leader_started_tx
                    .send(())
                    .map_err(|error| format!("failed to signal leader start: {error}"))?;
                thread::sleep(Duration::from_millis(150));
                Ok("leader-finished".to_owned())
            })
        });

        leader_started_rx
            .recv_timeout(Duration::from_secs(1))
            .map_err(|error| format!("leader did not start: {error}"))?;
        let follower_run = group
            .run(&key, Duration::from_millis(10), || {
                calls.fetch_add(1, Ordering::SeqCst);
                Ok("follower-fallback".to_owned())
            })
            .map_err(|error| format!("follower fallback should succeed: {error:?}"))?;
        assert_eq!(follower_run.value, "follower-fallback");
        assert_eq!(follower_run.role, SingleFlightRole::Follower);
        assert!(!follower_run.shared);
        assert_eq!(
            follower_run.degraded_codes,
            vec![SINGLEFLIGHT_FOLLOWER_TIMEOUT_CODE]
        );

        let leader_run = leader
            .join()
            .map_err(|_| "thread panicked".to_owned())?
            .map_err(|error| format!("leader failed unexpectedly: {error:?}"))?;
        assert_eq!(leader_run.value, "leader-finished");
        assert_eq!(leader_run.role, SingleFlightRole::Leader);
        assert_eq!(calls.load(Ordering::SeqCst), 2);
        let posture = group.posture(SingleFlightSurface::Search, true, Duration::from_millis(10));
        assert_eq!(posture.status, "observed_failures");
        assert_eq!(posture.follower_timeout_count, 1);
        assert_eq!(posture.completed_leader_count, 1);
        assert_eq!(posture.reused_result_count, 0);
        Ok(())
    }

    #[test]
    fn telemetry_contract_names_required_events() {
        assert_eq!(
            SINGLEFLIGHT_REQUIRED_TELEMETRY_PHASES,
            [
                "leader_start",
                "follower_join",
                "follower_timeout",
                "leader_complete",
                "coalesced_result_reused",
            ]
        );
    }

    #[test]
    fn leader_failure_is_visible_to_waiting_followers() -> TestResult {
        let group = Arc::new(SingleFlightGroup::<String>::new());
        let key = key("failed read");
        let (leader_started_tx, leader_started_rx) = mpsc::channel();

        let leader_group = Arc::clone(&group);
        let leader_key = key.clone();
        let leader = thread::spawn(move || {
            leader_group.run(&leader_key, Duration::from_secs(5), || {
                leader_started_tx
                    .send(())
                    .map_err(|error| format!("failed to signal leader start: {error}"))?;
                thread::sleep(Duration::from_millis(50));
                Err("leader cancelled".to_owned())
            })
        });

        leader_started_rx
            .recv_timeout(Duration::from_secs(1))
            .map_err(|error| format!("leader did not start: {error}"))?;
        let follower_error = match group.run(&key, Duration::from_secs(5), || {
            Ok("should-not-run".to_owned())
        }) {
            Ok(run) => {
                return Err(format!(
                    "follower should observe leader failure, got {run:?}"
                ));
            }
            Err(error) => error,
        };
        assert_eq!(follower_error.code(), SINGLEFLIGHT_LEADER_FAILED_CODE);
        match follower_error {
            SingleFlightError::LeaderFailed { role, message, .. } => {
                assert_eq!(role, SingleFlightRole::Follower);
                assert_eq!(message, "leader cancelled");
            }
            other => return Err(format!("unexpected follower error: {other:?}")),
        }

        let leader_error = match leader.join().map_err(|_| "thread panicked".to_owned())? {
            Ok(run) => {
                return Err(format!(
                    "leader should return operation failure, got {run:?}"
                ));
            }
            Err(error) => error,
        };
        match leader_error {
            SingleFlightError::LeaderFailed { role, message, .. } => {
                assert_eq!(role, SingleFlightRole::Leader);
                assert_eq!(message, "leader cancelled");
            }
            other => return Err(format!("unexpected leader error: {other:?}")),
        }
        let posture = group.posture(
            SingleFlightSurface::GraphFeatureEnrichment,
            true,
            Duration::from_secs(5),
        );
        assert_eq!(posture.status, "observed_failures");
        assert_eq!(posture.leader_failure_count, 1);
        Ok(())
    }

    #[test]
    fn leader_panic_is_visible_to_followers_and_posture() -> TestResult {
        let group = Arc::new(SingleFlightGroup::<String>::new());
        let key = key("panicked read");
        let (leader_started_tx, leader_started_rx) = mpsc::channel();
        let (release_leader_tx, release_leader_rx) = mpsc::channel();

        let leader_group = Arc::clone(&group);
        let leader_key = key.clone();
        let leader = thread::spawn(move || {
            std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                leader_group.run(&leader_key, Duration::from_secs(5), || {
                    leader_started_tx
                        .send(())
                        .map_err(|error| format!("failed to signal leader start: {error}"))?;
                    release_leader_rx
                        .recv_timeout(Duration::from_secs(1))
                        .map_err(|error| format!("leader release not received: {error}"))?;
                    panic!("synthetic leader panic");
                })
            }))
        });

        leader_started_rx
            .recv_timeout(Duration::from_secs(1))
            .map_err(|error| format!("leader did not start: {error}"))?;

        let follower_group = Arc::clone(&group);
        let follower_key = key.clone();
        let follower = thread::spawn(move || {
            follower_group.run(&follower_key, Duration::from_secs(5), || {
                Ok("should-not-run".to_owned())
            })
        });

        wait_for_registered_follower(&group, &key.key_hash)?;
        release_leader_tx
            .send(())
            .map_err(|error| format!("failed to release leader: {error}"))?;

        let leader_panic = leader.join().map_err(|_| "thread panicked".to_owned())?;
        assert!(
            leader_panic.is_err(),
            "leader panic should be caught by the test"
        );

        let follower_error = match follower.join().map_err(|_| "thread panicked".to_owned())? {
            Ok(run) => {
                return Err(format!("follower should observe leader panic, got {run:?}"));
            }
            Err(error) => error,
        };
        assert_eq!(follower_error.code(), SINGLEFLIGHT_LEADER_FAILED_CODE);
        match follower_error {
            SingleFlightError::LeaderFailed { role, message, .. } => {
                assert_eq!(role, SingleFlightRole::Follower);
                assert_eq!(message, "leader panicked");
            }
            other => return Err(format!("unexpected follower error: {other:?}")),
        }

        let posture = group.posture(SingleFlightSurface::Context, true, Duration::from_secs(5));
        assert_eq!(posture.status, "observed_failures");
        assert_eq!(posture.leader_failure_count, 1);
        assert_eq!(posture.active_leader_count, 0);
        Ok(())
    }

    #[derive(Debug)]
    struct PanicOnClone;

    impl Clone for PanicOnClone {
        fn clone(&self) -> Self {
            panic!("synthetic clone panic")
        }
    }

    #[test]
    fn leader_cleanup_runs_when_result_clone_panics_during_completion() -> TestResult {
        let group = SingleFlightGroup::<PanicOnClone>::new();
        let key = key("clone panics during completion");

        let panic = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            let _ = group.run(&key, Duration::from_secs(1), || Ok(PanicOnClone));
        }));
        assert!(
            panic.is_err(),
            "test fixture should panic while cloning the leader result"
        );

        assert_eq!(
            group.active_len().map_err(|error| format!("{error:?}"))?,
            0,
            "leader guard must remove the active entry when completion panics after the operation returns"
        );
        let posture = group.posture(SingleFlightSurface::Context, true, Duration::from_secs(1));
        assert_eq!(posture.active_leader_count, 0);
        assert_eq!(posture.leader_failure_count, 1);
        assert_eq!(posture.state_poisoned_count, 1);
        Ok(())
    }

    fn wait_for_registered_follower(
        group: &SingleFlightGroup<String>,
        key_hash: &str,
    ) -> TestResult {
        let started = Instant::now();
        loop {
            let follower_count = {
                let entries = group
                    .entries
                    .lock()
                    .map_err(|_| "single-flight entries lock poisoned".to_owned())?;
                entries
                    .get(key_hash)
                    .map_or(0, |entry| entry.followers.load(Ordering::SeqCst))
            };
            if follower_count > 0 {
                return Ok(());
            }
            if started.elapsed() >= Duration::from_secs(1) {
                return Err("follower did not join the single-flight entry".to_owned());
            }
            thread::sleep(Duration::from_millis(1));
        }
    }

    #[test]
    fn graph_feature_enrichment_wrapper_shares_identical_report() -> TestResult {
        let group = Arc::new(SingleFlightGroup::new());
        let leader_started = Arc::new(Barrier::new(2));
        let release_leader = Arc::new(Barrier::new(2));
        let executions = Arc::new(AtomicUsize::new(0));
        let options = GraphFeatureEnrichmentOptions::default();
        let mut handles = Vec::new();

        for _ in 0..4 {
            let group = Arc::clone(&group);
            let leader_started = Arc::clone(&leader_started);
            let release_leader = Arc::clone(&release_leader);
            let executions = Arc::clone(&executions);
            let options = options.clone();
            handles.push(thread::spawn(move || {
                run_graph_feature_enrichment_with_group(
                    GraphFeatureEnrichmentSingleFlightInput {
                        group: &group,
                        follower_timeout: Duration::from_secs(2),
                        workspace_identity: "/workspace/eidetic_engine_cli",
                        workspace_generation: 12,
                        graph_generation: Some(7),
                        source_mode: "graph_snapshot",
                        options: &options,
                    },
                    || {
                        executions.fetch_add(1, Ordering::SeqCst);
                        leader_started.wait();
                        release_leader.wait();
                        enrich_graph_features(&centrality_report(), &options)
                    },
                )
                .map_err(|error| format!("single-flight run failed: {error}"))
            }));
        }

        leader_started.wait();
        thread::sleep(Duration::from_millis(50));
        release_leader.wait();

        let mut leader_count = 0;
        let mut follower_count = 0;
        let mut reports = Vec::new();
        for handle in handles {
            let run = handle.join().map_err(|_| "thread panicked".to_owned())??;
            match run.role {
                SingleFlightRole::Leader => leader_count += 1,
                SingleFlightRole::Follower => follower_count += 1,
            }
            reports.push(run.value.data_json());
        }

        assert_eq!(executions.load(Ordering::SeqCst), 1);
        assert_eq!(leader_count, 1);
        assert_eq!(follower_count, 3);
        for report in reports.iter().skip(1) {
            assert_eq!(report, &reports[0]);
        }

        Ok(())
    }

    #[test]
    fn graph_feature_enrichment_timeout_fallback_returns_degraded_report() -> TestResult {
        let group = Arc::new(SingleFlightGroup::new());
        let options = GraphFeatureEnrichmentOptions::default();
        let executions = Arc::new(AtomicUsize::new(0));
        let (leader_started_tx, leader_started_rx) = mpsc::channel();
        let (release_leader_tx, release_leader_rx) = mpsc::channel();

        let leader_group = Arc::clone(&group);
        let leader_options = options.clone();
        let leader_executions = Arc::clone(&executions);
        let leader = thread::spawn(move || {
            run_graph_feature_enrichment_with_group(
                GraphFeatureEnrichmentSingleFlightInput {
                    group: &leader_group,
                    follower_timeout: Duration::from_secs(2),
                    workspace_identity: "/workspace/eidetic_engine_cli",
                    workspace_generation: 12,
                    graph_generation: Some(7),
                    source_mode: "graph_snapshot",
                    options: &leader_options,
                },
                || {
                    leader_executions.fetch_add(1, Ordering::SeqCst);
                    leader_started_tx.send(()).expect("signal leader start");
                    release_leader_rx
                        .recv_timeout(Duration::from_secs(1))
                        .expect("leader release");
                    enrich_graph_features(&centrality_report(), &leader_options)
                },
            )
        });

        leader_started_rx
            .recv_timeout(Duration::from_secs(1))
            .map_err(|error| format!("leader did not start: {error}"))?;
        let follower_run = run_graph_feature_enrichment_with_group(
            GraphFeatureEnrichmentSingleFlightInput {
                group: &group,
                follower_timeout: Duration::from_millis(10),
                workspace_identity: "/workspace/eidetic_engine_cli",
                workspace_generation: 12,
                graph_generation: Some(7),
                source_mode: "graph_snapshot",
                options: &options,
            },
            || {
                executions.fetch_add(1, Ordering::SeqCst);
                enrich_graph_features(&centrality_report(), &options)
            },
        )
        .map_err(|error| format!("follower fallback should succeed: {error}"))?;

        assert_eq!(follower_run.role, SingleFlightRole::Follower);
        assert!(!follower_run.shared);
        assert_eq!(
            follower_run.degraded_codes,
            vec![SINGLEFLIGHT_FOLLOWER_TIMEOUT_CODE]
        );
        assert!(
            follower_run
                .value
                .degraded
                .iter()
                .any(|entry| entry.code == SINGLEFLIGHT_FOLLOWER_TIMEOUT_CODE),
            "fallback report should surface the timeout degradation"
        );
        assert!(
            !follower_run.value.features.is_empty(),
            "fallback report should contain independently computed features"
        );

        release_leader_tx
            .send(())
            .map_err(|error| format!("failed to release leader: {error}"))?;
        let leader_run = leader
            .join()
            .map_err(|_| "leader thread panicked".to_owned())?
            .map_err(|error| format!("leader failed unexpectedly: {error}"))?;
        assert_eq!(leader_run.role, SingleFlightRole::Leader);
        assert_eq!(executions.load(Ordering::SeqCst), 2);

        Ok(())
    }

    #[test]
    fn graph_feature_enrichment_burst_caps_oversized_inputs() -> TestResult {
        let report = run_graph_feature_enrichment_burst_smoke(
            "/workspace/eidetic_engine_cli",
            &GraphFeatureEnrichmentOptions::default(),
            &GraphFeatureEnrichmentBurstOptions {
                identical_requests: GRAPH_FEATURE_ENRICHMENT_BURST_MAX_IDENTICAL_REQUESTS
                    .saturating_add(1),
                distinct_requests: GRAPH_FEATURE_ENRICHMENT_BURST_MAX_DISTINCT_REQUESTS
                    .saturating_add(1),
                node_count: GRAPH_FEATURE_ENRICHMENT_BURST_MAX_NODE_COUNT.saturating_add(1),
            },
        );

        assert_eq!(
            report.requested_identical,
            GRAPH_FEATURE_ENRICHMENT_BURST_MAX_IDENTICAL_REQUESTS.saturating_add(1)
        );
        assert_eq!(
            report.requested_distinct,
            GRAPH_FEATURE_ENRICHMENT_BURST_MAX_DISTINCT_REQUESTS.saturating_add(1)
        );
        assert_eq!(
            report.requested_node_count,
            GRAPH_FEATURE_ENRICHMENT_BURST_MAX_NODE_COUNT.saturating_add(1)
        );
        assert_eq!(
            report.effective_identical,
            GRAPH_FEATURE_ENRICHMENT_BURST_MAX_IDENTICAL_REQUESTS
        );
        assert_eq!(
            report.effective_distinct,
            GRAPH_FEATURE_ENRICHMENT_BURST_MAX_DISTINCT_REQUESTS
        );
        assert_eq!(
            report.effective_node_count,
            GRAPH_FEATURE_ENRICHMENT_BURST_MAX_NODE_COUNT
        );
        assert!(
            !report.passed,
            "capped hidden harness inputs must not report a clean pass"
        );
        assert!(
            report
                .diagnoses
                .iter()
                .any(|diagnosis| diagnosis.starts_with("capped identical requests")),
            "diagnoses should mention capped identical requests: {:?}",
            report.diagnoses
        );
        assert!(
            report
                .diagnoses
                .iter()
                .any(|diagnosis| diagnosis.starts_with("capped distinct requests")),
            "diagnoses should mention capped distinct requests: {:?}",
            report.diagnoses
        );
        assert!(
            report
                .diagnoses
                .iter()
                .any(|diagnosis| diagnosis.starts_with("capped node count")),
            "diagnoses should mention capped node count: {:?}",
            report.diagnoses
        );
        assert!(
            report.execution_count
                <= GRAPH_FEATURE_ENRICHMENT_BURST_MAX_DISTINCT_REQUESTS.saturating_add(1),
            "execution count should stay bounded by one identical leader plus capped distinct leaders"
        );

        let json = report.data_json();
        assert_eq!(
            json["requested"]["identical"],
            serde_json::json!(
                GRAPH_FEATURE_ENRICHMENT_BURST_MAX_IDENTICAL_REQUESTS.saturating_add(1)
            )
        );
        assert_eq!(
            json["effective"]["identical"],
            serde_json::json!(GRAPH_FEATURE_ENRICHMENT_BURST_MAX_IDENTICAL_REQUESTS)
        );
        assert_eq!(
            json["limits"]["maxNodeCount"],
            serde_json::json!(GRAPH_FEATURE_ENRICHMENT_BURST_MAX_NODE_COUNT)
        );

        Ok(())
    }

    #[test]
    fn graph_feature_enrichment_burst_reports_floor_adjusted_inputs() -> TestResult {
        let report = run_graph_feature_enrichment_burst_smoke(
            "/workspace/eidetic_engine_cli",
            &GraphFeatureEnrichmentOptions::default(),
            &GraphFeatureEnrichmentBurstOptions {
                identical_requests: 0,
                distinct_requests: 0,
                node_count: 0,
            },
        );

        assert_eq!(report.requested_identical, 0);
        assert_eq!(report.effective_identical, 2);
        assert_eq!(report.requested_node_count, 0);
        assert_eq!(report.effective_node_count, 1);
        assert!(
            !report.passed,
            "floor-adjusted hidden harness inputs must not report a clean pass"
        );
        assert!(
            report
                .diagnoses
                .iter()
                .any(|diagnosis| diagnosis == "raised identical requests from 0 to 2"),
            "diagnoses should mention raised identical requests: {:?}",
            report.diagnoses
        );
        assert!(
            report
                .diagnoses
                .iter()
                .any(|diagnosis| diagnosis == "raised node count from 0 to 1"),
            "diagnoses should mention raised node count: {:?}",
            report.diagnoses
        );

        Ok(())
    }

    fn centrality_report() -> CentralityRefreshReport {
        let scores = vec![
            MemoryCentralityScore {
                memory_id: "mem_a".to_owned(),
                pagerank: 0.9,
                betweenness: 0.2,
                hub: 0.0,
                authority: 0.0,
            },
            MemoryCentralityScore {
                memory_id: "mem_b".to_owned(),
                pagerank: 0.3,
                betweenness: 0.8,
                hub: 0.0,
                authority: 0.0,
            },
        ];
        CentralityRefreshReport {
            version: env!("CARGO_PKG_VERSION"),
            status: CentralityRefreshStatus::Refreshed,
            pagerank_status: crate::graph::CentralityAlgorithmStatus::Computed,
            betweenness_status: crate::graph::CentralityAlgorithmStatus::Computed,
            hits_status: crate::graph::CentralityAlgorithmStatus::Computed,
            dry_run: false,
            node_count: scores.len(),
            edge_count: 1,
            projection_ms: 0.0,
            pagerank_ms: 0.0,
            betweenness_ms: 0.0,
            hits_ms: 0.0,
            total_ms: 0.0,
            top_pagerank: scores.clone(),
            top_betweenness: scores.clone(),
            top_hubs: Vec::new(),
            top_authorities: Vec::new(),
            scores,
        }
    }
}