projectatlas_service/

analysis.rs

//! Closed, bounded architecture, impact, and static-trace projections.

mod impact;

#[cfg(test)]
mod analysis_test_observer {
    use std::cell::RefCell;

    /// Named production phase reached by one synchronous analysis request.
    #[derive(Clone, Debug, Eq, PartialEq)]
    pub(super) enum AnalysisPhaseEvent {
        /// Exact byte ledger passed to the architecture community projection.
        CompositionBudget {
            /// Remaining bytes handed to architecture findings.
            symbol_byte_budget: u64,
            /// Existing architecture findings' append charge.
            existing_finding_append_bytes: u64,
            /// Remaining bytes handed to community findings.
            community_budget: u64,
        },
        /// Induced relationship closure is about to traverse its first bounded frontier.
        Traversal,
        /// Admitted symbol hydration is about to enter bounded storage work.
        SymbolHydration,
        /// Topology and finding composition is about to retain aggregate output state.
        Composition,
        /// Optional dead-code candidate discovery has entered complete-scope work.
        DeadCodeDiscovery,
        /// One bounded candidate-classification batch is about to run.
        ClassificationHydration,
        /// One bounded candidate relation traversal is about to run.
        CandidateTraversal,
        /// One candidate relation report has returned its typed truncation state.
        CandidateReport {
            /// Whether the candidate report has a resumable continuation.
            has_continuation: bool,
            /// Whether the candidate report reached the edge limit.
            has_edges_limit: bool,
        },
        /// Terminal adjacency was probed before its generation was rechecked.
        TerminalProbe,
        /// Occurrence evidence was probed before its generation was rechecked.
        OccurrenceProbe,
        /// The occurrence evidence query was about to validate its generation.
        OccurrenceProbeBeforeRead,
        /// Terminal candidate coverage was loaded before its generation was rechecked.
        TerminalCandidateCoverageProbe {
            /// Aggregate relation work retained after the coverage batch.
            intermediate_bytes: u64,
        },
        /// The repository-wide candidate entity page is about to run.
        CandidateEntityHydration {
            /// Intermediate bytes left after relation traversal.
            remaining_intermediate_bytes: u64,
        },
        /// The repository-wide candidate page has been reduced to exact candidates.
        CandidateEnumeration,
        /// Adapter-specific output fitting has begun under the retained request control.
        OutputRendering,
    }

    /// Thread-local callback used only while one unit test owns observation.
    type Observer = Box<dyn FnMut(AnalysisPhaseEvent)>;

    thread_local! {
        /// Observer scoped to the synchronous analysis test thread.
        static OBSERVER: RefCell<Option<Observer>> = RefCell::new(None);
    }

    /// Restores a prior nested observer on every return or unwind path.
    struct ObserverGuard {
        /// Observer replaced for the current scope.
        previous: Option<Observer>,
    }

    impl Drop for ObserverGuard {
        fn drop(&mut self) {
            OBSERVER.with(|slot| {
                drop(slot.replace(self.previous.take()));
            });
        }
    }

    /// Run one operation while observing named production analysis phases.
    pub(super) fn observe_analysis_phase<T>(
        observer: impl FnMut(AnalysisPhaseEvent) + 'static,
        operation: impl FnOnce() -> T,
    ) -> T {
        let previous = OBSERVER.with(|slot| slot.replace(Some(Box::new(observer))));
        let _guard = ObserverGuard { previous };
        operation()
    }

    /// Notify the current thread-local observer when one is installed.
    pub(super) fn notify(event: AnalysisPhaseEvent) {
        OBSERVER.with(|slot| {
            if let Some(observer) = slot.borrow_mut().as_mut() {
                observer(event);
            }
        });
    }
}

#[cfg(test)]
use super::relations::classification_path;
use super::relations::{
    ExternalRelationIdentity, external_relation_identities, hydrate_single_detailed_node,
    load_detailed_relations, resolve_relation_anchor_for_analysis,
};
use super::{
    CoverageTrustState, DetailedRelationBudget, DetailedRelationNode, DetailedRelationQuery,
    DetailedRelationReport, DetailedRelationRow, DetailedRelationWork, RelationAnchor,
    RelationDirection, RelationNextCall, RelationPurpose, RelationResolutionFilter,
    RelationTotalState, ServiceError, ServiceResult, coverage_trust, selected_project_binding,
};
use impact::{LoadedVcs, digest_vcs_paths, impact_findings, load_vcs_paths};
use projectatlas_core::graph::{
    Completeness, ConfidenceClass, CoverageRecord, EntitySelector, ExtendedRelationKind,
    GraphEntity, GraphEntityKey, GraphIdentityText, GraphLimitKind, GraphLimits, GraphRelationKind,
    ProjectInstanceId, RelationResolution, RepositoryNodePath,
};
#[cfg(test)]
use projectatlas_core::language::ContentClassification;
use projectatlas_core::language::ContentSelection;
use projectatlas_core::symbols::{CodeSymbol, RelationKind};
use projectatlas_core::{
    CanonicalProjectRoot, IndexCancellation, IndexGeneration, IndexWorkControl, IndexWorkStage,
};
#[cfg(test)]
use projectatlas_db::MAX_FILE_CONTENT_CLASSIFICATION_PATHS;
use projectatlas_db::{
    AtlasStore, DbError, MAX_REPOSITORY_GRAPH_FRONTIER, MAX_SYMBOL_BATCH_DECODED_BYTES,
    MAX_SYMBOL_BATCH_PATHS, MAX_SYMBOL_BATCH_ROWS, RepositoryGraphAdjacencyContinuation,
    RepositoryGraphDirection, RepositoryGraphReadBudget, RepositoryGraphReadWork,
    SymbolBatchReadBudget, SymbolBatchReadLimit,
};
use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, BTreeSet, VecDeque};
use std::io::{self, Write};
use std::time::{Duration, Instant};

/// Current closed relation-analysis cursor schema.
const ANALYSIS_CURSOR_VERSION: u16 = 1;
/// Domain separator for the native project-root continuation binding.
const ANALYSIS_ROOT_DOMAIN: &str = "projectatlas:analysis-root:v1";
/// Maximum decoded cursor bytes accepted from an adapter.
const ANALYSIS_CURSOR_MAX_BYTES: usize = 256 * 1024;
/// Maximum nodes retained by one analysis projection.
const MAX_ANALYSIS_NODES: u32 = 512;
/// Maximum edges retained by one analysis projection.
const MAX_ANALYSIS_EDGES: u32 = 2_048;
/// Version of the deterministic architecture-community projection.
const COMMUNITY_ALGORITHM_VERSION: u16 = 1;
/// Version of the stable community ordering and identifier contract.
const COMMUNITY_ORDERING_VERSION: u16 = 1;
/// Maximum asynchronous label-propagation rounds per request.
const COMMUNITY_MAX_ITERATIONS: u32 = 24;
/// Self-vote keeps isolated and weakly connected nodes deterministic.
const COMMUNITY_SELF_WEIGHT: u32 = 1;
/// Conservative duplicate serialized-byte charge for community working state.
const COMMUNITY_WORKING_SET_MULTIPLIER: u64 = 8;
/// Conservative per-node/edge container overhead for community working state.
const COMMUNITY_WORKING_SET_ENTRY_BYTES: u64 = 256;
/// Conservative fixed map, vector, and propagation-state overhead.
const COMMUNITY_WORKING_SET_FIXED_BYTES: u64 = 32 * 1024;
/// Fixed JSON array framing already retained by the analysis composition.
const JSON_ARRAY_FRAMING_BYTES: u64 = 2;
/// Fixed JSON array separator inserted before an appended finding.
const JSON_ARRAY_SEPARATOR_BYTES: u64 = 1;

/// Closed analysis projection selected on the existing relation route.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum RelationAnalysisMode {
    /// Components, communities, cycles, purpose, complexity, and bottleneck candidates.
    Architecture,
    /// Version-control-aware affected nodes and conservative dead-code candidates.
    Impact,
    /// One node-simple static relationship path to an exact target label.
    Trace,
    /// Bounded reachability from an explicit, non-persistent entrypoint profile.
    Entrypoint,
}

/// Request-owned entrypoint reachability profile.
///
/// Profiles are deliberately not discovered or persisted. The surrounding
/// relation budget supplies the input, traversal, and output ceilings.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct EntrypointProfile {
    /// Stable human-readable profile name.
    pub name: String,
    /// Exact file or fully disambiguated symbol anchors.
    pub anchors: Vec<RelationAnchor>,
    /// Closed relation families admitted by this profile.
    pub relations: Vec<GraphRelationKind>,
}

/// Typed profile-level reachability disposition.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum EntrypointProfileCoverage {
    /// Every admitted entity and relation was inspected under the profile bounds.
    Complete,
    /// A declared relation, entity, or resource boundary prevented a safe negative.
    Partial,
}

/// Non-persistent profile metadata returned with entrypoint findings.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
pub struct EntrypointProfileResult {
    /// Request-owned stable profile name.
    pub name: String,
    /// Exact anchors used for traversal.
    pub anchors: Vec<RelationAnchor>,
    /// Exact relation families used for traversal.
    pub relations: Vec<GraphRelationKind>,
    /// Coverage disposition for the complete local entity scope.
    pub coverage: EntrypointProfileCoverage,
    /// Number of reachable local entities retained in the result.
    pub reachable: u32,
    /// Number of evidence-backed unreachable candidates retained in the result.
    pub unreachable_candidates: u32,
}

/// Version-control scope used by the impact projection.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum GitImpactSelection {
    /// Uncommitted tracked and untracked working-tree changes.
    WorkingTree,
    /// Changes staged in the index.
    Index,
    /// Files changed between two exact revision expressions.
    RevisionRange {
        /// Older revision expression.
        base: String,
        /// Newer revision expression.
        head: String,
    },
}

/// Complete request for one bounded closed analysis projection.
#[derive(Clone, Debug)]
pub struct RelationAnalysisQuery {
    /// Existing normalized relation traversal, filters, budgets, and cursor.
    pub relations: DetailedRelationQuery,
    /// Closed projection to compute over the bounded traversal.
    pub mode: RelationAnalysisMode,
    /// Exact file or symbol selector required by static trace mode.
    pub trace_target: Option<RelationAnchor>,
    /// Optional VCS scope; impact defaults to the working tree.
    pub vcs: Option<GitImpactSelection>,
    /// Include weak communities that exclude containment edges.
    pub include_communities: bool,
    /// Include iterative strongly-connected-component findings.
    pub include_cycles: bool,
    /// Include conservative non-exported dead-code candidates.
    pub include_dead_code: bool,
    /// Explicit non-persistent profile for entrypoint reachability mode.
    pub entrypoint_profile: Option<EntrypointProfile>,
}

/// Confidence disposition of one analysis finding.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum AnalysisStatus {
    /// The bounded evidence establishes the stated structural fact.
    Confirmed,
    /// The evidence is useful for review but is not a semantic proof.
    Candidate,
    /// Complete bounded evidence establishes a negative result.
    Absent,
    /// A declared coverage or traversal boundary prevents a safe conclusion.
    Inconclusive,
}

/// Closed finding families emitted by relation analysis.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum AnalysisFindingKind {
    /// Weak component over every admitted local relation.
    Component,
    /// Weak community with containment edges excluded.
    Community,
    /// Strongly connected dependency cycle.
    DependencyCycle,
    /// Owners share one approved purpose.
    PurposeAlignment,
    /// Connected owners retain conflicting approved purposes.
    PurposeDrift,
    /// Declaration span and graph-degree structural candidate.
    StructuralComplexity,
    /// High-degree or cross-owner dependency junction.
    Bottleneck,
    /// Node affected by the selected VCS path set.
    Impact,
    /// Conservative non-exported declaration with no trusted inbound relation.
    DeadCode,
    /// Node-simple static relationship path.
    StaticTrace,
    /// Ambiguous or unresolved static relation that prevents a closed conclusion.
    ResolutionGap,
    /// Reachability disposition for an explicit entrypoint profile.
    EntrypointReachability,
}

/// One deterministic bounded analysis finding.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
pub struct AnalysisFinding {
    /// Closed finding family.
    pub kind: AnalysisFindingKind,
    /// Evidence disposition.
    pub status: AnalysisStatus,
    /// Concise interpretation that does not overclaim semantic proof.
    pub summary: String,
    /// Typed reusable nodes with purpose, coverage, and next-call routing.
    pub nodes: Vec<AnalysisNode>,
    /// Optional exact structural metric, such as degree or declaration span.
    pub metric: Option<u64>,
    /// Exact relation evidence when this finding is caused by a resolution gap.
    pub evidence: Option<AnalysisRelationEvidence>,
    /// Deterministic community metadata when this is a community finding.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub community: Option<CommunityAnalysis>,
}

/// Convergence state of one bounded label-propagation run.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum CommunityConvergence {
    /// Every node retained its selected label during a complete round.
    Converged,
    /// The fixed iteration ceiling was reached before convergence.
    IterationLimit,
    /// The input coverage or resource envelope did not permit a partition.
    Inconclusive,
}

/// Coverage state attached to one community projection.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum CommunityCoverage {
    /// All admitted nodes and resolved local edges were inspected.
    Complete,
    /// A traversal or relation boundary prevented a complete partition.
    Partial,
}

/// One fixed relation-family weight in the v1 community contract.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
pub struct CommunityRelationWeight {
    /// Relation family receiving this vote weight.
    pub relation: GraphRelationKind,
    /// Positive vote weight used by label propagation.
    pub weight: u32,
}

/// Versioned limits and input selectors used by one community run.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
pub struct CommunityParameters {
    /// Label-propagation algorithm version.
    pub algorithm_version: u16,
    /// Stable node, label, and identifier ordering version.
    pub ordering_version: u16,
    /// Maximum label-propagation rounds.
    pub max_iterations: u32,
    /// Maximum nodes admitted to one projection.
    pub node_limit: u32,
    /// Maximum edges admitted to one projection.
    pub edge_limit: u32,
    /// Maximum encoded output bytes inherited from the bounded analysis route.
    pub output_bytes: u32,
    /// Optional detailed relation-family filter.
    pub relation: Option<GraphRelationKind>,
}

/// One resolved relation retained as community evidence.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
pub struct CommunityEdgeEvidence {
    /// Canonical source entity identity.
    pub source: String,
    /// Canonical target entity identity.
    pub target: String,
    /// Resolved local relation family.
    pub relation: GraphRelationKind,
    /// Vote weight applied to this edge.
    pub weight: u32,
}

/// Complete non-persistent metadata for one deterministic community.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
pub struct CommunityAnalysis {
    /// Stable ID derived from the versioned parameters and sorted members.
    pub id: String,
    /// Exact members with their existing source, purpose, and coverage evidence.
    pub members: Vec<AnalysisNode>,
    /// Resolved local edges induced by these members.
    pub evidence: Vec<CommunityEdgeEvidence>,
    /// Fixed relation weights admitted by this run.
    pub weights: Vec<CommunityRelationWeight>,
    /// Versioned algorithm and resource parameters.
    pub parameters: CommunityParameters,
    /// Number of completed propagation rounds.
    pub iteration: u32,
    /// Typed convergence disposition.
    pub convergence: CommunityConvergence,
    /// Typed graph coverage disposition.
    pub coverage: CommunityCoverage,
    /// Whether a fixed node, edge, or output bound prevented a complete result.
    pub truncated: bool,
}

/// One reusable analysis node retaining its authoritative navigation evidence.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
pub struct AnalysisNode {
    /// Exact entity, current approved purpose, and path coverage.
    pub node: DetailedRelationNode,
    /// Existing public call that can consume this selector directly.
    pub next_call: Option<RelationNextCall>,
}

/// Exact ambiguous or unresolved relation retained inside its fitted finding.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
pub struct AnalysisRelationEvidence {
    /// Full normalized relation, including resolution reference and confidence.
    pub relation: projectatlas_core::graph::LogicalRelation,
    /// Exact detailed-relation request that can inspect this gap directly.
    pub next_call: Option<RelationAnalysisNextCall>,
}

/// Exact existing relation call that can inspect one resolution gap directly.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
pub struct RelationAnalysisNextCall {
    /// Exact local source anchor.
    pub anchor: RelationAnchor,
    /// Direction that retains the source-side unresolved or ambiguous relation.
    pub direction: RelationDirection,
    /// Exact legacy or extended relation family.
    pub relation: GraphRelationKind,
    /// Exact resolution class retained by the finding.
    pub resolution: RelationResolutionFilter,
    /// Minimum confidence that retains the exact relation trust class.
    pub minimum_confidence: ConfidenceClass,
    /// Explicit classified-content restriction retained across the next call.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub content_selection: Option<ContentSelection>,
}

/// Typed VCS availability retained in impact responses.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(tag = "state", rename_all = "snake_case")]
pub enum VcsImpact {
    /// Architecture and trace modes did not request VCS evidence.
    NotRequested,
    /// Git returned a bounded normalized path set.
    Available {
        /// Exact selector used by the request.
        selection: GitImpactSelection,
        /// Exact number of normalized repository-relative changed paths.
        changed_path_count: u64,
    },
    /// Git is unavailable, the root is not a worktree, or the command failed.
    Unavailable {
        /// Exact selector that could not be evaluated.
        selection: GitImpactSelection,
        /// Bounded actionable reason.
        reason: String,
    },
}

/// Exact work performed in addition to the existing detailed traversal.
#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize)]
pub struct RelationAnalysisWork {
    /// Existing detailed traversal and hydration work.
    pub relations: DetailedRelationWork,
    /// Adjacency rows inspected while closing edges among admitted nodes.
    pub closure_inspected_edges: u32,
    /// Raw `SQLite` bytes decoded by induced-edge closure reads.
    pub closure_decoded_bytes: u64,
    /// Git stdout bytes retained while normalizing impact evidence.
    pub vcs_retained_bytes: u64,
    /// Number of admitted entities analyzed.
    pub analyzed_nodes: u32,
    /// Number of unique admitted local edges analyzed.
    pub analyzed_edges: u32,
    /// Symbols retained by bounded per-file exact-identity hydration.
    pub hydrated_symbols: u32,
    /// Decoded symbol and range-index bytes inspected during finding computation.
    pub hydrated_symbol_bytes: u64,
    /// Whether symbol path, row, or byte limits omitted candidates.
    pub symbol_hydration_truncated: bool,
    /// Exact serialized-equivalent analysis-owned bytes retained at return.
    pub retained_composition_bytes: u64,
    /// Peak aggregate relation, closure, VCS, symbol, topology,
    /// community-working-set, and finding bytes.
    pub peak_intermediate_bytes: u64,
    /// Whether composition limits omitted supported analysis projections.
    pub composition_truncated: bool,
    /// Exact bytes emitted by the selected adapter envelope.
    pub rendered_output_bytes: u64,
}

/// Bounded analysis report returned through the existing CLI and MCP relation route.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
pub struct RelationAnalysisReport {
    /// Closed projection selected by the caller.
    pub mode: RelationAnalysisMode,
    /// Existing exact relation anchor with authoritative purpose and coverage.
    pub anchor: DetailedRelationNode,
    /// Complete graph generation used by every relation row.
    pub generation: projectatlas_core::IndexGeneration,
    /// Accepted purpose revision used by every projected node.
    pub authored_purpose_revision: u64,
    /// Existing generation- and query-bound traversal continuation.
    pub continuation: Option<String>,
    /// Findings retained in this fitted response.
    pub returned: u32,
    /// Exact or lower-bound finding cardinality before output-prefix fitting.
    pub total: RelationTotalState,
    /// Whether traversal, closure, or output fitting omitted supported evidence.
    pub truncated: bool,
    /// Stable unique hard limits reached by traversal, closure, or rendering.
    pub reached_limits: Vec<GraphLimitKind>,
    /// Typed VCS evidence for impact mode.
    pub vcs: VcsImpact,
    /// Request-owned entrypoint profile metadata, when that mode is selected.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub entrypoint_profile: Option<EntrypointProfileResult>,
    /// Exact bounded work retained by this analysis.
    pub work: RelationAnalysisWork,
    /// Deterministically ordered structural findings.
    pub findings: Vec<AnalysisFinding>,
}

/// Unrendered analysis report that can fit an adapter's exact final envelope.
pub struct RelationAnalysisDraft {
    /// Fully hydrated maximum candidate report.
    report: RelationAnalysisReport,
    /// Exact encoded-output ceiling requested by the caller.
    output_bytes: u32,
    /// Result-defining traversal budget bound into continuations.
    budget: DetailedRelationBudget,
    /// Normalized query identity bound into continuations.
    cursor_binding: AnalysisCursorBinding,
    /// Repository and authored-purpose generation bound into continuations.
    cursor_snapshot: AnalysisCursorSnapshot,
    /// Relation cursor replayed when output fitting omits finding rows.
    replay_relation_cursor: Option<String>,
    /// Findings emitted before this draft began.
    finding_offset: u32,
    /// Optional normalized VCS evidence identity.
    vcs_digest: Option<[u8; 32]>,
    /// Exact external identities reached by the bounded detailed traversal.
    external_relation_identities: BTreeSet<ExternalRelationIdentity>,
    /// Shared request cancellation and deadline retained through rendering.
    control: IndexWorkControl,
}

impl RelationAnalysisDraft {
    /// Fully hydrated report before adapter-specific output fitting.
    #[must_use]
    pub const fn candidate_report(&self) -> &RelationAnalysisReport {
        &self.report
    }

    /// Borrow the exact request control retained through output rendering.
    pub(super) const fn control(&self) -> &IndexWorkControl {
        &self.control
    }

    /// Move the call-scoped rendezvous identities out before output fitting.
    pub(super) fn take_external_relation_identities(
        &mut self,
    ) -> BTreeSet<ExternalRelationIdentity> {
        std::mem::take(&mut self.external_relation_identities)
    }

    /// Fit one complete adapter envelope by retaining the largest finding prefix.
    ///
    /// # Errors
    ///
    /// Returns an adapter error when rendering fails or even the empty finding
    /// envelope exceeds the declared output or aggregate intermediate ceiling.
    pub fn fit_output<F, E, O>(self, mut encode: F) -> Result<(RelationAnalysisReport, O), E>
    where
        F: FnMut(&RelationAnalysisReport, &IndexWorkControl) -> Result<O, E>,
        E: From<ServiceError>,
        O: AsRef<[u8]>,
    {
        if self.report.mode == RelationAnalysisMode::Entrypoint {
            return self.fit_entrypoint_output(encode);
        }
        check_control(Some(&self.control)).map_err(E::from)?;
        #[cfg(test)]
        analysis_test_observer::notify(analysis_test_observer::AnalysisPhaseEvent::OutputRendering);
        let original_report_bytes =
            serialized_bytes_controlled(&self.report, Some(&self.control)).map_err(E::from)?;
        let construction_peak = self.report.work.peak_intermediate_bytes;
        let mut low = 0;
        let mut high = self.report.findings.len();
        let mut best = None;
        let mut output_limited = false;
        let mut intermediate_limited = false;
        let mut empty_output_oversized = false;
        let mut empty_intermediate_oversized = false;
        while low <= high {
            check_control(Some(&self.control)).map_err(E::from)?;
            let middle = low + (high - low) / 2;
            let fit_limits = [
                output_limited.then_some(GraphLimitKind::OutputBytes),
                intermediate_limited.then_some(GraphLimitKind::IntermediateBytes),
            ];
            let mut candidate =
                analysis_prefix(&self.report, middle, fit_limits.into_iter().flatten());
            if middle < self.report.findings.len() {
                let middle = u32::try_from(middle).map_err(|_overflow| {
                    E::from(ServiceError::InvalidInput(
                        "analysis finding offset overflowed".to_string(),
                    ))
                })?;
                let finding_offset = self.finding_offset.checked_add(middle).ok_or_else(|| {
                    E::from(ServiceError::InvalidInput(
                        "analysis finding offset overflowed".to_string(),
                    ))
                })?;
                candidate.continuation = Some(
                    encode_analysis_cursor(
                        self.replay_relation_cursor.as_deref(),
                        finding_offset,
                        &self.cursor_binding,
                        self.cursor_snapshot,
                        self.vcs_digest,
                        self.budget,
                    )
                    .map_err(E::from)?,
                );
            }
            check_control(Some(&self.control)).map_err(E::from)?;
            let mut encoded = encode(&candidate, &self.control)?;
            check_control(Some(&self.control)).map_err(E::from)?;
            let mut stable = false;
            for _ in 0..8 {
                check_control(Some(&self.control)).map_err(E::from)?;
                let rendered = u64::try_from(encoded.as_ref().len()).map_err(|source| {
                    E::from(ServiceError::InvalidInput(format!(
                        "analysis rendered byte count overflowed: {source}"
                    )))
                })?;
                let candidate_report_bytes =
                    serialized_bytes_controlled(&candidate, Some(&self.control))
                        .map_err(E::from)?;
                let fitting_peak = original_report_bytes
                    .checked_add(candidate_report_bytes)
                    .and_then(|bytes| bytes.checked_add(rendered))
                    .ok_or_else(|| {
                        E::from(ServiceError::InvalidInput(
                            "analysis output fitting byte count overflowed".to_string(),
                        ))
                    })?;
                let peak = construction_peak.max(fitting_peak);
                if candidate.work.rendered_output_bytes == rendered
                    && candidate.work.peak_intermediate_bytes == peak
                {
                    stable = true;
                    break;
                }
                candidate.work.rendered_output_bytes = rendered;
                candidate.work.peak_intermediate_bytes = peak;
                drop(encoded);
                encoded = encode(&candidate, &self.control)?;
                check_control(Some(&self.control)).map_err(E::from)?;
            }
            if !stable {
                return Err(E::from(ServiceError::InvalidInput(
                    "analysis output accounting did not stabilize".to_string(),
                )));
            }
            let output_fits = encoded.as_ref().len() <= self.output_bytes as usize;
            let intermediate_fits =
                candidate.work.peak_intermediate_bytes <= self.budget.intermediate_bytes();
            if output_fits && intermediate_fits {
                best = Some((candidate, encoded));
                low = middle.saturating_add(1);
            } else {
                if middle == 0 {
                    empty_output_oversized = !output_fits;
                    empty_intermediate_oversized = !intermediate_fits;
                    break;
                }
                let newly_output_limited = !output_fits && !output_limited;
                let newly_intermediate_limited = !intermediate_fits && !intermediate_limited;
                output_limited |= !output_fits;
                intermediate_limited |= !intermediate_fits;
                if newly_output_limited || newly_intermediate_limited {
                    best = None;
                    low = 0;
                }
                high = middle - 1;
            }
        }
        check_control(Some(&self.control)).map_err(E::from)?;
        best.ok_or_else(|| {
            let message = if empty_intermediate_oversized {
                "empty analysis envelope exceeds the aggregate intermediate-byte budget"
            } else if empty_output_oversized {
                "graph output byte limit is too small for the empty analysis envelope"
            } else if intermediate_limited {
                "analysis output fitting exceeds the aggregate intermediate-byte budget"
            } else {
                "graph output byte limit is too small for the empty analysis envelope"
            };
            E::from(ServiceError::InvalidInput(message.to_string()))
        })
    }

    /// Fit an entrypoint report atomically; partial finding prefixes are not replayable.
    fn fit_entrypoint_output<F, E, O>(self, mut encode: F) -> Result<(RelationAnalysisReport, O), E>
    where
        F: FnMut(&RelationAnalysisReport, &IndexWorkControl) -> Result<O, E>,
        E: From<ServiceError>,
        O: AsRef<[u8]>,
    {
        check_control(Some(&self.control)).map_err(E::from)?;
        #[cfg(test)]
        analysis_test_observer::notify(analysis_test_observer::AnalysisPhaseEvent::OutputRendering);
        let mut candidate = self.report;
        let mut encoded = encode(&candidate, &self.control)?;
        for _ in 0..8 {
            check_control(Some(&self.control)).map_err(E::from)?;
            let rendered = u64::try_from(encoded.as_ref().len()).map_err(|source| {
                E::from(ServiceError::InvalidInput(format!(
                    "entrypoint rendered byte count overflowed: {source}"
                )))
            })?;
            let candidate_report_bytes =
                serialized_bytes_controlled(&candidate, Some(&self.control)).map_err(E::from)?;
            let fitting_peak = candidate_report_bytes
                .checked_add(rendered)
                .ok_or_else(|| {
                    E::from(ServiceError::InvalidInput(
                        "entrypoint output fitting byte count overflowed".to_string(),
                    ))
                })?;
            let peak = candidate.work.peak_intermediate_bytes.max(fitting_peak);
            if candidate.work.rendered_output_bytes == rendered
                && candidate.work.peak_intermediate_bytes == peak
            {
                if encoded.as_ref().len() > self.output_bytes as usize {
                    return Err(E::from(ServiceError::InvalidInput(
                        "entrypoint profile output exceeds its declared byte budget".to_string(),
                    )));
                }
                if peak > self.budget.intermediate_bytes() {
                    return Err(E::from(ServiceError::InvalidInput(
                        "entrypoint profile output exceeds its aggregate intermediate-byte budget"
                            .to_string(),
                    )));
                }
                return Ok((candidate, encoded));
            }
            candidate.work.rendered_output_bytes = rendered;
            candidate.work.peak_intermediate_bytes = peak;
            encoded = encode(&candidate, &self.control)?;
        }
        Err(E::from(ServiceError::InvalidInput(
            "entrypoint output accounting did not stabilize".to_string(),
        )))
    }
}

#[derive(Clone, Serialize)]
/// One local normalized edge admitted to topology algorithms.
struct LocalEdge {
    /// Canonical source entity identity.
    source: String,
    /// Canonical target entity identity.
    target: String,
    /// Closed relation family.
    kind: GraphRelationKind,
    /// Whether the owning relation evidence is complete.
    complete: bool,
}

#[derive(Default)]
/// Analysis-owned work accumulated beyond the detailed relation traversal.
struct SupplementalWork {
    /// Persisted symbols retained for declaration-aware findings.
    hydrated_symbols: u32,
    /// Decoded rows plus serialized-equivalent range-index bytes retained.
    hydrated_symbol_bytes: u64,
    /// Peak symbol hydration bytes including request-path auxiliaries.
    hydrated_symbol_peak_bytes: u64,
    /// Whether a declared hydration bound omitted candidates.
    symbol_hydration_truncated: bool,
    /// Stable limits reached by supplemental work.
    reached_limits: Vec<GraphLimitKind>,
    /// Analysis-owned bytes retained in the composed report.
    retained_composition_bytes: u64,
    /// Whether composition fitting omitted supported evidence.
    composition_truncated: bool,
    /// Conservative transient bytes charged while constructing community state.
    community_working_set_bytes: u64,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
/// Normalized result-defining request identity encoded in every cursor.
struct AnalysisCursorBinding {
    /// Digest of the selected canonical project root.
    root_digest: [u8; 32],
    /// Exact normalized traversal anchor.
    anchor: RelationAnchor,
    /// Traversal direction.
    direction: RelationDirection,
    /// Optional relation-family filter.
    relation: Option<GraphRelationKind>,
    /// Minimum admitted trust class.
    minimum_confidence: ConfidenceClass,
    /// Resolution-class filter.
    resolution: RelationResolutionFilter,
    /// Classified-content restriction that changes admitted topology.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    content_selection: Option<ContentSelection>,
    /// Closed feature selections that affect results.
    options: AnalysisCursorOptions,
    /// Exact traversal resource envelope.
    budget: DetailedRelationBudget,
    /// Detailed relation algorithm contract version.
    algorithm_version: u16,
    /// Detailed relation ordering contract version.
    ordering_version: u16,
    /// Closed analysis projection.
    mode: RelationAnalysisMode,
    /// Optional exact trace target.
    trace_target: Option<RelationAnchor>,
    /// Optional VCS selector.
    vcs: Option<GitImpactSelection>,
    /// Explicit entrypoint profile, when selected.
    entrypoint_profile: Option<EntrypointProfile>,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
/// Typed inclusion state for cursor-bound optional analysis behavior.
enum AnalysisFeatureSelection {
    /// Feature output is excluded.
    Excluded,
    /// Feature output is included.
    Included,
}

impl From<bool> for AnalysisFeatureSelection {
    fn from(included: bool) -> Self {
        if included {
            Self::Included
        } else {
            Self::Excluded
        }
    }
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
/// Closed feature selections grouped into one cursor option contract.
struct AnalysisCursorOptions {
    /// Retain exact relation occurrence spans.
    relation_occurrences: AnalysisFeatureSelection,
    /// Emit weak community candidates.
    communities: AnalysisFeatureSelection,
    /// Emit dependency-cycle candidates.
    cycles: AnalysisFeatureSelection,
    /// Emit conservative dead-code candidates.
    dead_code: AnalysisFeatureSelection,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
/// Repository state that must remain stable while a cursor is replayed.
struct AnalysisCursorSnapshot {
    /// Stable selected project instance.
    project: ProjectInstanceId,
    /// Complete repository-graph generation.
    generation: projectatlas_core::IndexGeneration,
    /// Accepted authored-purpose revision.
    authored_purpose_revision: u64,
}

#[derive(Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
/// Bounded continuation for relation traversal and finding-prefix replay.
struct AnalysisCursor {
    /// Cursor schema version.
    version: u16,
    /// Result-defining normalized request identity.
    binding: AnalysisCursorBinding,
    /// Repository state captured by the first page.
    snapshot: AnalysisCursorSnapshot,
    /// Optional underlying relation traversal continuation.
    relation_cursor: Option<String>,
    /// Findings emitted before the next fitted page.
    finding_offset: u32,
    /// Optional normalized VCS evidence digest.
    vcs_digest: Option<[u8; 32]>,
}

/// Load one closed analysis view over the existing bounded relation service.
///
/// # Errors
///
/// Returns the existing detailed-relation errors plus bounded VCS, closure, and
/// output validation failures.
pub fn load_relation_analysis(
    store: &AtlasStore,
    query: &RelationAnalysisQuery,
    control: Option<&IndexWorkControl>,
) -> ServiceResult<RelationAnalysisDraft> {
    if query.mode == RelationAnalysisMode::Entrypoint {
        return load_entrypoint_profile_draft(store, query, control);
    }
    load_relation_analysis_with_closure_deadline(store, query, control, None, false)
}

/// Load analysis while retaining its bounded external traversal identities.
pub(super) fn load_relation_analysis_for_federation(
    store: &AtlasStore,
    query: &RelationAnalysisQuery,
    control: Option<&IndexWorkControl>,
) -> ServiceResult<RelationAnalysisDraft> {
    if query.mode == RelationAnalysisMode::Entrypoint {
        return Err(ServiceError::InvalidInput(
            "entrypoint profiles require one project root".to_string(),
        ));
    }
    load_relation_analysis_with_closure_deadline(store, query, control, None, true)
}

/// Load analysis with an optional earlier closure-stage deadline ceiling.
fn load_relation_analysis_with_closure_deadline(
    store: &AtlasStore,
    query: &RelationAnalysisQuery,
    control: Option<&IndexWorkControl>,
    closure_deadline_ceiling: Option<Instant>,
    retain_external_relation_identities: bool,
) -> ServiceResult<RelationAnalysisDraft> {
    validate_analysis_query(query)?;
    let started = Instant::now();
    let deadline = started
        .checked_add(Duration::from_millis(query.relations.budget.deadline_ms()))
        .unwrap_or(started);
    let deadline = control
        .and_then(IndexWorkControl::deadline)
        .map_or(deadline, |caller_deadline| caller_deadline.min(deadline));
    let analysis_control = control.map_or_else(
        || IndexWorkControl::with_deadline(IndexCancellation::new(), deadline),
        |caller| {
            caller.with_timeout_ceiling(deadline.saturating_duration_since(caller.started_at()))
        },
    );
    let control = Some(&analysis_control);
    check_control(control)?;
    let selected_binding = selected_project_binding(store)?;
    let cursor_binding = analysis_cursor_binding(query, &selected_binding.project_root_identity)?;
    let decoded_cursor = query
        .relations
        .cursor
        .as_deref()
        .map(|cursor| decode_analysis_cursor(cursor, &cursor_binding))
        .transpose()?;
    let mut relation_query = query.relations.clone();
    relation_query.budget = bounded_analysis_budget(query.relations.budget)?;
    relation_query.cursor = decoded_cursor
        .as_ref()
        .and_then(|cursor| cursor.relation_cursor.clone());
    let replay_relation_cursor = relation_query.cursor.clone();
    let finding_offset = decoded_cursor
        .as_ref()
        .map_or(0, |cursor| cursor.finding_offset);
    let relations = load_detailed_relations(store, &relation_query, control)?;
    let external_relation_identities = if retain_external_relation_identities {
        external_relation_identities(&relations)
    } else {
        BTreeSet::new()
    };
    let external_relation_identity_bytes =
        serialized_bytes_controlled(&external_relation_identities, control)?;
    let cursor_snapshot = AnalysisCursorSnapshot {
        project: relations.anchor.entity.key().project(),
        generation: relations.generation,
        authored_purpose_revision: relations.authored_purpose_revision,
    };
    if decoded_cursor
        .as_ref()
        .is_some_and(|cursor| cursor.snapshot != cursor_snapshot)
    {
        return Err(ServiceError::RelationCursorStale {
            field: "analysis snapshot",
        });
    }
    check_control(control)?;
    let mut nodes = collect_nodes(&relations, control)?;
    let mut edges = collect_report_edges(&relations, control)?;
    let mut closure_query = query.clone();
    closure_query.relations = relation_query;
    let closure = close_induced_edges(
        store,
        &closure_query,
        &relations.work,
        closure_deadline_ceiling.map_or(deadline, |ceiling| ceiling.min(deadline)),
        &nodes,
        &mut edges,
        control,
    )?;
    check_control(control)?;
    let evidence_complete =
        relation_evidence_complete(&relations, &nodes, &edges, query, &closure, false);
    let community_evidence_complete =
        relation_evidence_complete(&relations, &nodes, &edges, query, &closure, true);
    let dead_code_scope_complete = dead_code_scope_complete(&relations, query);
    check_control(control)?;
    let vcs_load = if query.mode == RelationAnalysisMode::Impact {
        let selection = query.vcs.clone().unwrap_or(GitImpactSelection::WorkingTree);
        load_vcs_paths(
            selected_binding.project_root_identity.as_path(),
            selection,
            query.relations.budget.intermediate_bytes().saturating_sub(
                relations
                    .work
                    .intermediate_bytes
                    .saturating_add(closure.decoded_bytes)
                    .saturating_add(external_relation_identity_bytes),
            ),
            deadline,
            control,
        )
    } else {
        LoadedVcs {
            report: VcsImpact::NotRequested,
            changed_paths: Vec::new(),
            retained_bytes: 0,
        }
    };
    let vcs = vcs_load.report;
    let vcs_digest = (query.mode == RelationAnalysisMode::Impact)
        .then(|| digest_vcs_paths(&vcs_load.changed_paths, control))
        .transpose()?;
    if decoded_cursor
        .as_ref()
        .is_some_and(|cursor| cursor.vcs_digest != vcs_digest)
    {
        return Err(ServiceError::RelationCursorStale {
            field: "VCS evidence",
        });
    }
    let mut supplemental_work = SupplementalWork::default();
    let analysis_allowance = query.relations.budget.intermediate_bytes().saturating_sub(
        relations
            .work
            .intermediate_bytes
            .saturating_add(closure.decoded_bytes)
            .saturating_add(external_relation_identity_bytes),
    );
    let projection_allowance = analysis_allowance.saturating_sub(vcs_load.retained_bytes);
    #[cfg(test)]
    analysis_test_observer::notify(analysis_test_observer::AnalysisPhaseEvent::Composition);
    let mut topology_bytes =
        serialized_bytes_controlled(&(nodes.values().collect::<Vec<_>>(), &edges), control)?;
    let mut gaps = resolution_gap_findings(&relations, control)?;
    gaps.extend(closure.resolution_gaps.iter().cloned());
    check_control(control)?;
    gaps.sort_by(|left, right| resolution_gap_identity(left).cmp(resolution_gap_identity(right)));
    gaps.dedup_by(|left, right| resolution_gap_identity(left) == resolution_gap_identity(right));
    check_control(control)?;
    let gap_bytes = serialized_bytes_controlled(&gaps, control)?;
    let projection_safe = topology_bytes.saturating_add(gap_bytes) <= projection_allowance;
    let mut findings = if projection_safe {
        gaps
    } else {
        supplemental_work.composition_truncated = true;
        push_limit(
            &mut supplemental_work.reached_limits,
            GraphLimitKind::IntermediateBytes,
        );
        edges.clear();
        nodes.retain(|_, node| node.entity.key() == relations.anchor.entity.key());
        topology_bytes =
            serialized_bytes_controlled(&(nodes.values().collect::<Vec<_>>(), &edges), control)?;
        vec![AnalysisFinding {
            kind: AnalysisFindingKind::Component,
            status: AnalysisStatus::Inconclusive,
            summary: "analysis composition crossed the shared intermediate-byte budget".to_string(),
            nodes: vec![analysis_node(&relations.anchor)],
            metric: None,
            evidence: None,
            community: None,
        }]
    };
    let initial_finding_bytes = serialized_bytes_controlled(&findings, control)?;
    let initial_finding_count = findings.len();
    let symbol_byte_budget = projection_allowance
        .saturating_sub(topology_bytes)
        .saturating_sub(initial_finding_bytes);
    if projection_safe {
        findings.extend(match query.mode {
            RelationAnalysisMode::Architecture => architecture_findings(
                store,
                &nodes,
                &edges,
                evidence_complete,
                community_evidence_complete,
                query,
                symbol_byte_budget,
                initial_finding_count,
                &mut supplemental_work,
                control,
            )?,
            RelationAnalysisMode::Impact => impact_findings(
                store,
                &nodes,
                &edges,
                evidence_complete,
                dead_code_scope_complete,
                &vcs,
                &vcs_load.changed_paths,
                query,
                symbol_byte_budget,
                &mut supplemental_work,
                control,
            )?,
            RelationAnalysisMode::Trace => {
                trace_findings(&relations, query.trace_target.as_ref(), evidence_complete)?
            }
            RelationAnalysisMode::Entrypoint => Vec::new(),
        });
    }
    let generated_finding_count = u32::try_from(findings.len()).map_err(|_overflow| {
        ServiceError::InvalidInput("analysis finding count overflowed".to_string())
    })?;
    let mut finding_bytes = serialized_bytes_controlled(&findings, control)?;
    let generated_composition_bytes = vcs_load
        .retained_bytes
        .saturating_add(topology_bytes)
        .saturating_add(finding_bytes);
    supplemental_work.retained_composition_bytes = vcs_load
        .retained_bytes
        .saturating_add(topology_bytes)
        .saturating_add(finding_bytes);
    while supplemental_work.retained_composition_bytes > analysis_allowance && findings.len() > 1 {
        check_control(control)?;
        findings.pop();
        supplemental_work.composition_truncated = true;
        push_limit(
            &mut supplemental_work.reached_limits,
            GraphLimitKind::IntermediateBytes,
        );
        finding_bytes = serialized_bytes_controlled(&findings, control)?;
        supplemental_work.retained_composition_bytes = vcs_load
            .retained_bytes
            .saturating_add(topology_bytes)
            .saturating_add(finding_bytes);
    }
    if supplemental_work.retained_composition_bytes > analysis_allowance {
        findings.clear();
        supplemental_work.composition_truncated = true;
        push_limit(
            &mut supplemental_work.reached_limits,
            GraphLimitKind::IntermediateBytes,
        );
        finding_bytes = serialized_bytes_controlled(&findings, control)?;
        supplemental_work.retained_composition_bytes = vcs_load
            .retained_bytes
            .saturating_add(topology_bytes)
            .saturating_add(finding_bytes);
    }
    let hydration_peak = relations
        .work
        .intermediate_bytes
        .saturating_add(closure.decoded_bytes)
        .saturating_add(vcs_load.retained_bytes)
        .saturating_add(topology_bytes)
        .saturating_add(initial_finding_bytes)
        .saturating_add(external_relation_identity_bytes)
        .saturating_add(supplemental_work.hydrated_symbol_peak_bytes);
    let generation_peak = relations
        .work
        .intermediate_bytes
        .saturating_add(closure.decoded_bytes)
        .saturating_add(external_relation_identity_bytes)
        .saturating_add(generated_composition_bytes)
        .saturating_add(supplemental_work.community_working_set_bytes);
    let final_peak = relations
        .work
        .intermediate_bytes
        .saturating_add(closure.decoded_bytes)
        .saturating_add(supplemental_work.retained_composition_bytes)
        .saturating_add(external_relation_identity_bytes);
    let peak_intermediate_bytes = hydration_peak.max(generation_peak).max(final_peak);
    let community_projection_truncated = findings.iter().any(|finding| {
        finding
            .community
            .as_ref()
            .is_some_and(|community| community.truncated)
    });
    let full_finding_count = u32::try_from(findings.len()).map_err(|_overflow| {
        ServiceError::InvalidInput("analysis finding count overflowed".to_string())
    })?;
    if finding_offset > full_finding_count {
        return Err(ServiceError::RelationCursorInvalid {
            reason: "analysis finding offset exceeds the recomputed page",
        });
    }
    findings.drain(..finding_offset as usize);
    let mut reached_limits = relations.reached_limits.clone();
    if !closure.complete {
        push_limit(
            &mut reached_limits,
            if closure.deadline_reached {
                GraphLimitKind::Deadline
            } else {
                GraphLimitKind::Edges
            },
        );
    }
    for limit in &supplemental_work.reached_limits {
        push_limit(&mut reached_limits, *limit);
    }
    let work = RelationAnalysisWork {
        relations: relations.work.clone(),
        closure_inspected_edges: closure.inspected_edges,
        closure_decoded_bytes: closure.decoded_bytes,
        vcs_retained_bytes: vcs_load.retained_bytes,
        analyzed_nodes: u32::try_from(nodes.len()).unwrap_or(u32::MAX),
        analyzed_edges: u32::try_from(edges.len()).unwrap_or(u32::MAX),
        hydrated_symbols: supplemental_work.hydrated_symbols,
        hydrated_symbol_bytes: supplemental_work.hydrated_symbol_bytes,
        symbol_hydration_truncated: supplemental_work.symbol_hydration_truncated,
        retained_composition_bytes: supplemental_work.retained_composition_bytes,
        peak_intermediate_bytes,
        composition_truncated: supplemental_work.composition_truncated,
        rendered_output_bytes: 0,
    };
    let analysis_truncated = (!evidence_complete && relations.truncated)
        || !closure.complete
        || supplemental_work.symbol_hydration_truncated
        || supplemental_work.composition_truncated
        || community_projection_truncated;
    let returned = u32::try_from(findings.len()).unwrap_or(u32::MAX);
    let total = if analysis_truncated {
        RelationTotalState::AtLeast(u64::from(generated_finding_count))
    } else {
        RelationTotalState::Exact(u64::from(full_finding_count))
    };
    let continuation = if evidence_complete {
        None
    } else {
        relations
            .continuation
            .as_deref()
            .map(|cursor| {
                encode_analysis_cursor(
                    Some(cursor),
                    0,
                    &cursor_binding,
                    cursor_snapshot,
                    vcs_digest,
                    query.relations.budget,
                )
            })
            .transpose()?
    };
    let report = RelationAnalysisReport {
        mode: query.mode,
        anchor: relations.anchor,
        generation: relations.generation,
        authored_purpose_revision: relations.authored_purpose_revision,
        continuation,
        returned,
        total,
        truncated: analysis_truncated,
        reached_limits,
        vcs,
        entrypoint_profile: None,
        work,
        findings,
    };
    nodes.clear();
    check_control(control)?;
    Ok(RelationAnalysisDraft {
        report,
        output_bytes: query.relations.budget.output_bytes(),
        budget: query.relations.budget,
        cursor_binding,
        cursor_snapshot,
        replay_relation_cursor,
        finding_offset,
        vcs_digest,
        external_relation_identities,
        control: analysis_control,
    })
}

/// Validate closed mode, selector, option, and budget combinations.
fn validate_analysis_query(query: &RelationAnalysisQuery) -> ServiceResult<()> {
    if query.mode == RelationAnalysisMode::Entrypoint {
        let profile = query.entrypoint_profile.as_ref().ok_or_else(|| {
            ServiceError::InvalidInput(
                "entrypoint analysis requires an explicit profile".to_string(),
            )
        })?;
        validate_entrypoint_profile(profile)?;
        if query.relations.direction != RelationDirection::Outbound
            || query.relations.resolution != RelationResolutionFilter::Any
        {
            return Err(ServiceError::InvalidInput(
                "entrypoint profiles require outbound traversal with any resolution".to_string(),
            ));
        }
        if query.relations.cursor.is_some() {
            return Err(ServiceError::RelationCursorInvalid {
                reason: "entrypoint profiles do not support relation cursors; restart the profile",
            });
        }
        if query.relations.relation.is_some() {
            return Err(ServiceError::InvalidInput(
                "entrypoint profiles carry relation families in the profile itself".to_string(),
            ));
        }
        if query.trace_target.is_some()
            || query.vcs.is_some()
            || query.include_communities
            || query.include_cycles
            || query.include_dead_code
        {
            return Err(ServiceError::InvalidInput(
                "entrypoint profiles do not accept unrelated analysis controls".to_string(),
            ));
        }
        return Ok(());
    }
    if query.entrypoint_profile.is_some() {
        return Err(ServiceError::InvalidInput(
            "entrypoint profile is valid only for entrypoint analysis".to_string(),
        ));
    }
    if query.mode == RelationAnalysisMode::Trace {
        let Some(target) = query.trace_target.as_ref() else {
            return Err(ServiceError::InvalidInput(
                "analysis trace requires an exact file or symbol target".to_string(),
            ));
        };
        if matches!(
            target,
            RelationAnchor::Symbol {
                symbol_kind: None,
                ..
            }
        ) || matches!(
            target,
            RelationAnchor::Symbol {
                signature: None,
                ..
            }
        ) {
            return Err(ServiceError::InvalidInput(
                "analysis trace symbol targets require exact kind and signature".to_string(),
            ));
        }
    } else if query.trace_target.is_some() {
        return Err(ServiceError::InvalidInput(
            "trace_target is valid only for static trace analysis".to_string(),
        ));
    }
    if query.mode != RelationAnalysisMode::Impact && query.vcs.is_some() {
        return Err(ServiceError::InvalidInput(
            "VCS selection is valid only for impact analysis".to_string(),
        ));
    }
    if query.mode != RelationAnalysisMode::Architecture
        && (query.include_communities || query.include_cycles)
    {
        return Err(ServiceError::InvalidInput(
            "community and cycle controls are valid only for architecture analysis".to_string(),
        ));
    }
    if query.mode != RelationAnalysisMode::Impact && query.include_dead_code {
        return Err(ServiceError::InvalidInput(
            "dead-code controls are valid only for impact analysis".to_string(),
        ));
    }
    Ok(())
}

/// Validate the request-owned profile before any graph read occurs.
fn validate_entrypoint_profile(profile: &EntrypointProfile) -> ServiceResult<()> {
    let name = profile.name.trim();
    if name.is_empty() || name.len() > 128 || name != profile.name {
        return Err(ServiceError::InvalidInput(
            "entrypoint profile name must be nonempty, trimmed, and at most 128 bytes".to_string(),
        ));
    }
    if profile.anchors.is_empty() || profile.anchors.len() > 32 {
        return Err(ServiceError::InvalidInput(
            "entrypoint profile requires 1 to 32 anchors".to_string(),
        ));
    }
    let mut anchors = BTreeSet::new();
    for anchor in &profile.anchors {
        let encoded = serde_json::to_string(anchor)?;
        if !anchors.insert(encoded) {
            return Err(ServiceError::InvalidInput(
                "entrypoint profile anchors must be unique".to_string(),
            ));
        }
        if matches!(
            anchor,
            RelationAnchor::Symbol {
                symbol_kind: None,
                ..
            }
        ) || matches!(
            anchor,
            RelationAnchor::Symbol {
                signature: None,
                ..
            }
        ) {
            return Err(ServiceError::InvalidInput(
                "entrypoint symbol anchors require exact kind and signature".to_string(),
            ));
        }
        if let RelationAnchor::Symbol {
            name,
            parent,
            signature,
            ..
        } = anchor
            && (name.trim().is_empty()
                || name != name.trim()
                || signature
                    .as_deref()
                    .is_some_and(|value| value.trim().is_empty())
                || parent
                    .as_deref()
                    .is_some_and(|value| value.trim().is_empty()))
        {
            return Err(ServiceError::InvalidInput(
                "entrypoint symbol anchors require nonempty exact identity fields".to_string(),
            ));
        }
    }
    if profile.relations.is_empty() || profile.relations.len() > GraphRelationKind::ALL.len() {
        return Err(ServiceError::InvalidInput(
            "entrypoint profile requires 1 to 12 relation families".to_string(),
        ));
    }
    let mut relations = BTreeSet::new();
    for relation in &profile.relations {
        if !GraphRelationKind::ALL.contains(relation) || !relations.insert(relation.as_str()) {
            return Err(ServiceError::InvalidInput(
                "entrypoint profile relation families must be unique supported values".to_string(),
            ));
        }
    }
    Ok(())
}

/// Build a bounded, non-persistent entrypoint profile over the existing graph.
fn load_entrypoint_profile_draft(
    store: &AtlasStore,
    query: &RelationAnalysisQuery,
    control: Option<&IndexWorkControl>,
) -> ServiceResult<RelationAnalysisDraft> {
    validate_analysis_query(query)?;
    let profile = query
        .entrypoint_profile
        .as_ref()
        .ok_or_else(|| ServiceError::InvalidInput("entrypoint profile is missing".to_string()))?;
    let started = Instant::now();
    let budget = bounded_analysis_budget(query.relations.budget)?;
    validate_entrypoint_profile_budget(profile, budget)?;
    let deadline = started
        .checked_add(Duration::from_millis(budget.deadline_ms()))
        .unwrap_or(started);
    let analysis_control = control.map_or_else(
        || IndexWorkControl::with_deadline(IndexCancellation::new(), deadline),
        |caller| {
            caller.with_timeout_ceiling(deadline.saturating_duration_since(caller.started_at()))
        },
    );
    let control = Some(&analysis_control);
    check_control(control)?;
    let selected_binding = selected_project_binding(store)?;
    let generation = store.repository_graph_generation()?.ok_or_else(|| {
        ServiceError::InvalidInput(
            "repository graph has no complete generation for entrypoint analysis".to_string(),
        )
    })?;
    let mut reachable = BTreeMap::<String, DetailedRelationNode>::new();
    let mut edges = Vec::new();
    let mut relation_work = DetailedRelationWork::default();
    let mut canonical_anchors = Vec::with_capacity(profile.anchors.len());
    let mut canonical_anchor_keys = BTreeSet::new();
    let mut resolved_anchor_keys = BTreeMap::<String, String>::new();
    let mut resolved_anchor_entities = BTreeMap::<String, GraphEntity>::new();
    let mut first_resolved_anchor = None;
    let mut complete = true;
    let mut reached_limits = Vec::new();
    'anchors: for anchor in &profile.anchors {
        let anchor_budget = match entrypoint_anchor_budget(budget, &relation_work)? {
            Ok(anchor_budget) => anchor_budget,
            Err(limit) => {
                complete = false;
                push_limit(&mut reached_limits, limit);
                break 'anchors;
            }
        };
        let (entity, anchor_work) = match resolve_relation_anchor_for_analysis(
            store,
            selected_binding.project_instance_id,
            generation,
            anchor,
            anchor_budget,
            control,
        ) {
            Ok(result) => result,
            Err(ServiceError::Db(DbError::GraphContract(
                projectatlas_core::graph::GraphContractError::InvalidLimits {
                    reason: "graph read decoded bytes exceed the batch budget",
                },
            ))) => {
                complete = false;
                push_limit(&mut reached_limits, GraphLimitKind::IntermediateBytes);
                break 'anchors;
            }
            Err(error) => return Err(error),
        };
        add_relation_work(&mut relation_work, &anchor_work)?;
        first_resolved_anchor.get_or_insert_with(|| entity.clone());
        let entity_key = entity.key().canonical_identity().to_string();
        if !canonical_anchor_keys.insert(entity_key.clone()) {
            return Err(ServiceError::InvalidInput(
                "entrypoint profile anchors must resolve to unique entities".to_string(),
            ));
        }
        let canonical_anchor = relation_anchor_for_entity(&entity).ok_or_else(|| {
            ServiceError::InvalidInput(
                "entrypoint profile anchor is not addressable by an exact anchor".to_string(),
            )
        })?;
        let anchor_identity = serde_json::to_string(&canonical_anchor)
            .map_err(|error| ServiceError::InvalidInput(error.to_string()))?;
        resolved_anchor_keys.insert(anchor_identity.clone(), entity_key);
        resolved_anchor_entities.insert(anchor_identity, entity);
        canonical_anchors.push(canonical_anchor);
    }
    let mut first_anchor = None;
    let mut authored_purpose_revision = 0;
    let mut purpose_revision_initialized = false;
    let entity_selected = |node: &DetailedRelationNode| {
        !matches!(node.entity.selector(), EntitySelector::External { .. })
            && (query.relations.content_selection == ContentSelection::UnspecifiedLegacy
                || node.classification.is_some_and(|classification| {
                    query.relations.content_selection.includes(classification)
                }))
    };

    let mut frontier = if complete {
        canonical_anchors
    } else {
        Vec::new()
    };
    let mut scheduled_anchors = frontier
        .iter()
        .map(serde_json::to_string)
        .collect::<Result<BTreeSet<_>, _>>()
        .map_err(|error| ServiceError::InvalidInput(error.to_string()))?;
    let mut visited = BTreeSet::new();
    let mut depth = 0_u32;
    'profile: while !frontier.is_empty() && depth < budget.depth() {
        let mut next_frontier = Vec::new();
        for anchor in frontier.drain(..) {
            for relation in &profile.relations {
                check_control(control)?;
                let anchor_identity = serde_json::to_string(&anchor)
                    .map_err(|error| ServiceError::InvalidInput(error.to_string()))?;
                let anchor_is_retained = resolved_anchor_keys
                    .get(&anchor_identity)
                    .is_some_and(|key| reachable.contains_key(key));
                let retained_keys =
                    anchor_is_retained.then(|| reachable.keys().cloned().collect::<BTreeSet<_>>());
                let collect_occurrences = query.relations.include_occurrences
                    && relation_work.retained_occurrences < budget.occurrences_total();
                let step_budget = match entrypoint_step_budget(
                    budget,
                    &relation_work,
                    reachable.len(),
                    0,
                    anchor_is_retained,
                    false,
                    collect_occurrences,
                    None,
                )? {
                    Ok(step_budget) => step_budget,
                    Err(limit) => {
                        if limit == GraphLimitKind::Edges
                            && let Some(anchor_key) = resolved_anchor_keys.get(&anchor_identity)
                            && let Some(entity) = resolved_anchor_entities
                                .get(&anchor_identity)
                                .or_else(|| reachable.get(anchor_key).map(|node| &node.entity))
                            && entrypoint_terminal_adjacency_is_empty(
                                store,
                                generation,
                                entity.key(),
                                *relation,
                                query.relations.minimum_confidence,
                                query.relations.content_selection,
                                control,
                            )?
                        {
                            if reachable.contains_key(anchor_key) {
                                continue;
                            }
                            if purpose_revision_initialized
                                && store.authored_purpose_revision()? != authored_purpose_revision
                            {
                                return Err(ServiceError::RelationCursorStale {
                                    field: "entrypoint authored purpose revision",
                                });
                            }
                            let anchor_node = match load_entrypoint_terminal_candidate_coverage(
                                store,
                                entity,
                                generation,
                                budget,
                                &mut relation_work,
                                query.relations.content_selection,
                                control,
                            )? {
                                Ok(node) => node,
                                Err(limit) => {
                                    complete = false;
                                    push_limit(&mut reached_limits, limit);
                                    break 'profile;
                                }
                            };
                            if !entity_selected(&anchor_node) {
                                return Err(ServiceError::InvalidInput(
                                    "entrypoint anchor is outside the selected content".to_string(),
                                ));
                            }
                            if purpose_revision_initialized
                                && store.authored_purpose_revision()? != authored_purpose_revision
                            {
                                return Err(ServiceError::RelationCursorStale {
                                    field: "entrypoint authored purpose revision",
                                });
                            }
                            if !trusted_node_coverage(&anchor_node, &profile.relations) {
                                complete = false;
                                break 'profile;
                            }
                            if reachable.len() >= budget.nodes() as usize {
                                complete = false;
                                push_limit(&mut reached_limits, GraphLimitKind::Nodes);
                            }
                            if visited.len() >= budget.visited() as usize {
                                complete = false;
                                push_limit(&mut reached_limits, GraphLimitKind::Visited);
                            }
                            if !complete {
                                break 'profile;
                            }
                            let entity_key = entity.key().canonical_identity().to_string();
                            visited.insert(entity_key);
                            insert_node(&mut reachable, &anchor_node);
                            first_anchor.get_or_insert(anchor_node);
                            continue;
                        }
                        complete = false;
                        push_limit(&mut reached_limits, limit);
                        break 'profile;
                    }
                };
                let mut relation_query = query.relations.clone();
                relation_query.anchor = anchor.clone();
                relation_query.relation = Some(*relation);
                relation_query.direction = RelationDirection::Outbound;
                relation_query.resolution = RelationResolutionFilter::Any;
                relation_query.cursor = None;
                relation_query.budget = step_budget;
                relation_query.include_occurrences = collect_occurrences;
                #[cfg(test)]
                analysis_test_observer::notify(
                    analysis_test_observer::AnalysisPhaseEvent::Traversal,
                );
                let report = match load_detailed_relations(store, &relation_query, control) {
                    Ok(report) => report,
                    Err(ServiceError::Db(DbError::GraphContract(
                        projectatlas_core::graph::GraphContractError::InvalidLimits {
                            reason: "graph read decoded bytes exceed the batch budget",
                        },
                    ))) => {
                        complete = false;
                        push_limit(&mut reached_limits, GraphLimitKind::IntermediateBytes);
                        break 'profile;
                    }
                    Err(error) => return Err(error),
                };
                if report.generation != generation {
                    return Err(ServiceError::RelationCursorStale {
                        field: "entrypoint graph generation",
                    });
                }
                first_anchor.get_or_insert_with(|| report.anchor.clone());
                resolved_anchor_keys.insert(
                    anchor_identity,
                    report.anchor.entity.key().canonical_identity().to_string(),
                );
                if purpose_revision_initialized
                    && report.authored_purpose_revision != authored_purpose_revision
                {
                    return Err(ServiceError::RelationCursorStale {
                        field: "entrypoint authored purpose revision",
                    });
                }
                authored_purpose_revision = report.authored_purpose_revision;
                purpose_revision_initialized = true;
                add_relation_work(&mut relation_work, &report.work)?;
                if report.pruned_incomplete_paths > 0 {
                    complete = false;
                }
                if report.pruned_evidence_truncated {
                    complete = false;
                    push_limit(&mut reached_limits, GraphLimitKind::IntermediateBytes);
                }
                let filtered_edge_limit_is_terminal = entrypoint_filtered_edge_limit_is_terminal(
                    store,
                    generation,
                    &report,
                    *relation,
                    query.relations.minimum_confidence,
                    query.relations.content_selection,
                    control,
                )?;
                if query.relations.include_occurrences
                    && (!collect_occurrences || !report.pruned_relations.is_empty())
                {
                    match entrypoint_occurrence_evidence_is_incomplete(
                        store,
                        &report,
                        generation,
                        budget,
                        &mut relation_work,
                        control,
                        !collect_occurrences,
                    )? {
                        Ok(true) => {
                            complete = false;
                            push_limit(&mut reached_limits, GraphLimitKind::Occurrences);
                        }
                        Err(limit) => {
                            complete = false;
                            push_limit(&mut reached_limits, limit);
                        }
                        Ok(false) => {}
                    }
                }
                for limit in &report.reached_limits {
                    if !filtered_edge_limit_is_terminal
                        || (*limit != GraphLimitKind::Edges && *limit != GraphLimitKind::Rows)
                    {
                        push_limit(&mut reached_limits, *limit);
                    }
                }
                if relation_work.inspected_edges > budget.edges() {
                    complete = false;
                    push_limit(&mut reached_limits, GraphLimitKind::Edges);
                    break 'profile;
                }
                if relation_work.intermediate_bytes > budget.intermediate_bytes() {
                    complete = false;
                    push_limit(&mut reached_limits, GraphLimitKind::IntermediateBytes);
                    break 'profile;
                }
                complete &= trusted_node_coverage(&report.anchor, &profile.relations);
                let anchor_key = report.anchor.entity.key().canonical_identity().to_string();
                visited.insert(anchor_key);
                insert_node(&mut reachable, &report.anchor);
                for row in &report.rows {
                    check_control(control)?;
                    let resolved = matches!(
                        row.relation.resolution(),
                        RelationResolution::Resolved { .. } | RelationResolution::External { .. }
                    ) && row.relation.completeness() == Completeness::Complete;
                    complete &= resolved && trusted_relation_row(row, &profile.relations);
                    insert_node(&mut reachable, &row.source);
                    if let Some(target) = &row.target
                        && entity_selected(target)
                    {
                        insert_node(&mut reachable, target);
                        if resolved {
                            let target_key = target.entity.key().canonical_identity().to_string();
                            if visited.insert(target_key.clone()) {
                                match relation_anchor_for_entity(&target.entity) {
                                    Some(target_anchor) => {
                                        if let Ok(anchor_key) =
                                            serde_json::to_string(&target_anchor)
                                        {
                                            let is_new =
                                                scheduled_anchors.insert(anchor_key.clone());
                                            resolved_anchor_keys
                                                .entry(anchor_key)
                                                .or_insert_with(|| target_key.clone());
                                            if is_new {
                                                next_frontier.push(target_anchor);
                                            }
                                        }
                                    }
                                    None if !matches!(
                                        target.entity.selector(),
                                        EntitySelector::External { .. }
                                    ) =>
                                    {
                                        complete = false;
                                    }
                                    None => {}
                                }
                            }
                        }
                    }
                    for node in &row.path {
                        if entity_selected(node) {
                            insert_node(&mut reachable, node);
                        }
                    }
                    if let Some(edge) =
                        local_edge(&row.relation, &row.source.entity, row.target.as_ref())
                    {
                        let edge_bytes = u64::try_from(edge.source.len())
                            .unwrap_or(u64::MAX)
                            .checked_add(u64::try_from(edge.target.len()).unwrap_or(u64::MAX))
                            .and_then(|bytes| {
                                bytes.checked_add(std::mem::size_of::<LocalEdge>() as u64)
                            })
                            .ok_or_else(entrypoint_work_overflow)?;
                        let retained_edge_bytes = relation_work
                            .intermediate_bytes
                            .checked_add(edge_bytes)
                            .ok_or_else(entrypoint_work_overflow)?;
                        if retained_edge_bytes > budget.intermediate_bytes() {
                            complete = false;
                            push_limit(&mut reached_limits, GraphLimitKind::IntermediateBytes);
                            break 'profile;
                        }
                        relation_work.intermediate_bytes = retained_edge_bytes;
                        edges.push(edge);
                    }
                }
                if reachable.len() > budget.nodes() as usize {
                    if let Some(retained_keys) = retained_keys {
                        reachable.retain(|key, _| retained_keys.contains(key));
                    }
                    complete = false;
                    push_limit(&mut reached_limits, GraphLimitKind::Nodes);
                    break 'profile;
                }
                if report.continuation.is_some() {
                    if !filtered_edge_limit_is_terminal {
                        complete = false;
                        push_limit(&mut reached_limits, GraphLimitKind::Rows);
                    }
                } else {
                    complete &= filtered_edge_limit_is_terminal
                        || (!report.truncated
                            && report.reached_limits.is_empty()
                            && matches!(report.total, RelationTotalState::Exact(_)));
                }
            }
        }
        frontier = next_frontier;
        depth = depth.saturating_add(1);
    }
    if !frontier.is_empty() {
        complete = false;
        push_limit(&mut reached_limits, GraphLimitKind::Depth);
    }
    edges.sort_by(|left, right| {
        (&left.source, &left.target, left.kind.as_str()).cmp(&(
            &right.source,
            &right.target,
            right.kind.as_str(),
        ))
    });
    edges.dedup_by(|left, right| {
        left.source == right.source && left.target == right.target && left.kind == right.kind
    });
    if first_anchor.is_none()
        && first_resolved_anchor.is_none()
        && reached_limits.contains(&GraphLimitKind::IntermediateBytes)
    {
        return Err(ServiceError::ResourceLimit {
            limit: GraphLimitKind::IntermediateBytes,
        });
    }
    let anchor = first_anchor
        .or_else(|| {
            first_resolved_anchor.map(|entity| {
                entrypoint_unavailable_node(&entity, query.relations.content_selection)
            })
        })
        .ok_or_else(|| {
            ServiceError::InvalidInput("entrypoint profile resolved no anchors".to_string())
        })?;
    let entity_limit = budget.nodes();
    let reachable_keys = reachable.keys().cloned().collect::<BTreeSet<_>>();
    let mut protected_reachable_keys = reachable_keys.clone();
    for node in reachable.values() {
        if let EntitySelector::Symbol { symbol } = node.entity.selector() {
            let file_selector = EntitySelector::File {
                path: symbol.file.clone(),
            };
            protected_reachable_keys.insert(
                GraphEntityKey::new(node.entity.key().project(), &file_selector)
                    .canonical_identity()
                    .to_string(),
            );
        }
    }
    let mut protected_enclosure_entities = Vec::new();
    if complete {
        let owner_paths = reachable
            .values()
            .filter_map(|node| match node.entity.selector() {
                EntitySelector::Symbol { symbol } => Some(symbol.file.as_str().to_string()),
                _ => None,
            })
            .collect::<BTreeSet<_>>();
        for path in owner_paths {
            check_control(control)?;
            let path = RepositoryNodePath::new(std::path::Path::new(&path))
                .map_err(|error| ServiceError::InvalidInput(error.to_string()))?;
            let remaining_intermediate = budget
                .intermediate_bytes()
                .saturating_sub(relation_work.intermediate_bytes);
            if remaining_intermediate == 0 {
                complete = false;
                push_limit(&mut reached_limits, GraphLimitKind::IntermediateBytes);
                break;
            }
            let read_budget = RepositoryGraphReadBudget::new(
                1,
                RepositoryGraphReadBudget::MAX_RETURNED_ROWS,
                remaining_intermediate.min(RepositoryGraphReadBudget::MAX_DECODED_BYTES),
                RepositoryGraphReadBudget::MAX_HYDRATED_ENTITIES,
                RepositoryGraphReadBudget::MAX_HYDRATED_PATHS,
            )
            .map_err(|error| ServiceError::InvalidInput(error.to_string()))?;
            let owner_page = match store.repository_graph_entities_by_path_bounded(
                generation_project(&anchor.entity),
                generation,
                &path,
                RepositoryGraphReadBudget::MAX_RETURNED_ROWS,
                read_budget,
                control,
            ) {
                Ok(owner_page) => owner_page,
                Err(DbError::GraphContract(
                    projectatlas_core::graph::GraphContractError::InvalidLimits {
                        reason: "graph read decoded bytes exceed the batch budget",
                    },
                )) => {
                    complete = false;
                    push_limit(&mut reached_limits, GraphLimitKind::IntermediateBytes);
                    break;
                }
                Err(error) => return Err(error.into()),
            };
            add_repository_read_work(&mut relation_work, &owner_page.work)?;
            if owner_page.page.truncated {
                complete = false;
                push_limit(&mut reached_limits, GraphLimitKind::Nodes);
                break;
            }
            protected_enclosure_entities.extend(owner_page.page.rows);
        }
    }
    protect_reachable_symbol_enclosures(
        &reachable,
        &protected_enclosure_entities,
        &mut protected_reachable_keys,
    );
    let mut retained_candidate_keys = reachable_keys;
    let mut unreachable = BTreeMap::new();
    let remaining_intermediate = budget
        .intermediate_bytes()
        .saturating_sub(relation_work.intermediate_bytes);
    if remaining_intermediate == 0 {
        complete = false;
        push_limit(&mut reached_limits, GraphLimitKind::IntermediateBytes);
    }
    if complete {
        let protected_entity_overhead = match u32::try_from(protected_reachable_keys.len()) {
            Ok(overhead) => overhead,
            Err(_overflow) => {
                complete = false;
                push_limit(&mut reached_limits, GraphLimitKind::Nodes);
                0
            }
        };
        let candidate_page_limit = match entity_limit.checked_add(protected_entity_overhead) {
            Some(limit) if limit <= GraphLimits::MAX_ROWS => limit,
            _ => {
                complete = false;
                push_limit(&mut reached_limits, GraphLimitKind::Nodes);
                0
            }
        };
        let all_entities = if candidate_page_limit == 0 {
            None
        } else {
            #[cfg(test)]
            analysis_test_observer::notify(
                analysis_test_observer::AnalysisPhaseEvent::CandidateEntityHydration {
                    remaining_intermediate_bytes: remaining_intermediate,
                },
            );
            let read_budget = RepositoryGraphReadBudget::new(
                1,
                candidate_page_limit,
                remaining_intermediate.min(RepositoryGraphReadBudget::MAX_DECODED_BYTES),
                candidate_page_limit.saturating_add(1).saturating_mul(2),
                candidate_page_limit.saturating_add(1).saturating_mul(2),
            )
            .map_err(|error| ServiceError::InvalidInput(error.to_string()))?;
            match store.repository_graph_entrypoint_candidates_page_bounded(
                generation_project(&anchor.entity),
                generation,
                candidate_page_limit,
                query.relations.content_selection,
                read_budget,
                control,
            ) {
                Ok(all_entities) => {
                    add_repository_read_work(&mut relation_work, &all_entities.work)?;
                    Some(all_entities.page)
                }
                Err(DbError::GraphContract(
                    projectatlas_core::graph::GraphContractError::InvalidLimits {
                        reason: "graph read decoded bytes exceed the batch budget",
                    },
                )) => {
                    complete = false;
                    push_limit(&mut reached_limits, GraphLimitKind::IntermediateBytes);
                    None
                }
                Err(error) => return Err(error.into()),
            }
        };
        if let Some(all_entities) = all_entities.as_ref() {
            protect_reachable_symbol_enclosures(
                &reachable,
                &all_entities.rows,
                &mut protected_reachable_keys,
            );
        }
        let candidate_entities = all_entities
            .as_ref()
            .map(|all_entities| {
                if all_entities.truncated {
                    complete = false;
                    push_limit(&mut reached_limits, GraphLimitKind::Nodes);
                }
                all_entities
                    .rows
                    .iter()
                    .filter(|entity| {
                        matches!(
                            entity.selector(),
                            EntitySelector::File { .. } | EntitySelector::Symbol { .. }
                        ) && !protected_reachable_keys.contains(entity.key().canonical_identity())
                    })
                    .collect::<Vec<_>>()
            })
            .unwrap_or_default();
        #[cfg(test)]
        analysis_test_observer::notify(
            analysis_test_observer::AnalysisPhaseEvent::CandidateEnumeration,
        );
        if complete {
            let current_generation = store.repository_graph_generation()?.ok_or_else(|| {
                ServiceError::InvalidInput(
                    "repository graph has no complete generation for entrypoint analysis"
                        .to_string(),
                )
            })?;
            if current_generation != generation {
                return Err(ServiceError::RelationCursorStale {
                    field: "entrypoint graph generation",
                });
            }
            for entity in candidate_entities {
                check_control(control)?;
                if !complete {
                    break;
                }
                let candidate_key = entity.key().canonical_identity().to_string();
                let candidate_anchor = relation_anchor_for_entity(entity).ok_or_else(|| {
                    ServiceError::InvalidInput(
                        "entrypoint candidate is not addressable by an exact anchor".to_string(),
                    )
                })?;
                let mut candidate_report_anchor = None;
                let mut candidate_unretained_keys = BTreeSet::new();
                for relation in &profile.relations {
                    let accounted_candidates = retained_candidate_keys
                        .len()
                        .saturating_sub(reachable.len());
                    let anchor_is_retained = retained_candidate_keys.contains(&candidate_key);
                    let collect_occurrences = query.relations.include_occurrences
                        && relation_work.retained_occurrences < budget.occurrences_total();
                    let step_budget = match entrypoint_step_budget(
                        budget,
                        &relation_work,
                        reachable.len(),
                        accounted_candidates,
                        anchor_is_retained,
                        true,
                        collect_occurrences,
                        None,
                    )? {
                        Ok(step_budget) => step_budget,
                        Err(GraphLimitKind::Edges)
                            if entrypoint_terminal_adjacency_is_empty(
                                store,
                                generation,
                                entity.key(),
                                *relation,
                                query.relations.minimum_confidence,
                                query.relations.content_selection,
                                control,
                            )? =>
                        {
                            match entrypoint_step_budget(
                                budget,
                                &relation_work,
                                reachable.len(),
                                accounted_candidates,
                                anchor_is_retained,
                                true,
                                collect_occurrences,
                                Some(1),
                            )? {
                                Ok(_) => {}
                                Err(limit) => {
                                    complete = false;
                                    push_limit(&mut reached_limits, limit);
                                    break;
                                }
                            }
                            let candidate_node = if let Some(node) = candidate_report_anchor.clone()
                            {
                                node
                            } else {
                                let node = match load_entrypoint_terminal_candidate_coverage(
                                    store,
                                    entity,
                                    generation,
                                    budget,
                                    &mut relation_work,
                                    query.relations.content_selection,
                                    control,
                                )? {
                                    Ok(node) => node,
                                    Err(limit) => {
                                        complete = false;
                                        push_limit(&mut reached_limits, limit);
                                        break;
                                    }
                                };
                                candidate_report_anchor = Some(node.clone());
                                node
                            };
                            if !trusted_node_coverage(&candidate_node, &profile.relations) {
                                complete = false;
                                break;
                            }
                            if !anchor_is_retained {
                                candidate_unretained_keys.insert(candidate_key.clone());
                            }
                            let remaining_candidate_capacity =
                                u32::try_from(retained_candidate_keys.len()).unwrap_or(u32::MAX);
                            let remaining_nodes = usize::try_from(
                                budget.nodes().saturating_sub(remaining_candidate_capacity),
                            )
                            .unwrap_or(usize::MAX);
                            let remaining_visited = usize::try_from(
                                budget
                                    .visited()
                                    .saturating_sub(remaining_candidate_capacity),
                            )
                            .unwrap_or(usize::MAX);
                            if candidate_unretained_keys.len() > remaining_nodes {
                                complete = false;
                                push_limit(&mut reached_limits, GraphLimitKind::Nodes);
                            }
                            if candidate_unretained_keys.len() > remaining_visited {
                                complete = false;
                                push_limit(&mut reached_limits, GraphLimitKind::Visited);
                            }
                            if !complete {
                                break;
                            }
                            candidate_report_anchor.get_or_insert(candidate_node);
                            continue;
                        }
                        Err(limit) => {
                            complete = false;
                            push_limit(&mut reached_limits, limit);
                            break;
                        }
                    };
                    let mut candidate_query = query.relations.clone();
                    candidate_query.anchor = candidate_anchor.clone();
                    candidate_query.relation = Some(*relation);
                    candidate_query.direction = RelationDirection::Outbound;
                    candidate_query.resolution = RelationResolutionFilter::Any;
                    candidate_query.cursor = None;
                    candidate_query.budget = step_budget;
                    candidate_query.include_occurrences = collect_occurrences;
                    #[cfg(test)]
                    analysis_test_observer::notify(
                        analysis_test_observer::AnalysisPhaseEvent::CandidateTraversal,
                    );
                    let candidate_report =
                        match load_detailed_relations(store, &candidate_query, control) {
                            Ok(report) => report,
                            Err(ServiceError::Db(DbError::GraphContract(
                                projectatlas_core::graph::GraphContractError::InvalidLimits {
                                    reason: "graph read decoded bytes exceed the batch budget",
                                },
                            ))) => {
                                complete = false;
                                push_limit(&mut reached_limits, GraphLimitKind::IntermediateBytes);
                                break;
                            }
                            Err(error) => return Err(error),
                        };
                    #[cfg(test)]
                    analysis_test_observer::notify(
                        analysis_test_observer::AnalysisPhaseEvent::CandidateReport {
                            has_continuation: candidate_report.continuation.is_some(),
                            has_edges_limit: candidate_report
                                .reached_limits
                                .contains(&GraphLimitKind::Edges),
                        },
                    );
                    if candidate_report.generation != generation {
                        return Err(ServiceError::RelationCursorStale {
                            field: "entrypoint graph generation",
                        });
                    }
                    if candidate_report.authored_purpose_revision != authored_purpose_revision {
                        return Err(ServiceError::RelationCursorStale {
                            field: "entrypoint authored purpose revision",
                        });
                    }
                    add_relation_work(&mut relation_work, &candidate_report.work)?;
                    if candidate_report.pruned_evidence_truncated {
                        complete = false;
                        push_limit(&mut reached_limits, GraphLimitKind::IntermediateBytes);
                    }
                    let filtered_edge_limit_is_terminal =
                        entrypoint_filtered_edge_limit_is_terminal(
                            store,
                            generation,
                            &candidate_report,
                            *relation,
                            query.relations.minimum_confidence,
                            query.relations.content_selection,
                            control,
                        )?;
                    if query.relations.include_occurrences
                        && (!collect_occurrences || !candidate_report.pruned_relations.is_empty())
                    {
                        match entrypoint_occurrence_evidence_is_incomplete(
                            store,
                            &candidate_report,
                            generation,
                            budget,
                            &mut relation_work,
                            control,
                            !collect_occurrences,
                        )? {
                            Ok(true) => {
                                complete = false;
                                push_limit(&mut reached_limits, GraphLimitKind::Occurrences);
                            }
                            Err(limit) => {
                                complete = false;
                                push_limit(&mut reached_limits, limit);
                            }
                            Ok(false) => {}
                        }
                    }
                    if relation_work.inspected_edges > budget.edges() {
                        complete = false;
                        push_limit(&mut reached_limits, GraphLimitKind::Edges);
                        break;
                    }
                    if relation_work.intermediate_bytes > budget.intermediate_bytes() {
                        complete = false;
                        push_limit(&mut reached_limits, GraphLimitKind::IntermediateBytes);
                        break;
                    }
                    candidate_unretained_keys.extend(candidate_report_unretained_local_keys(
                        &candidate_report,
                        &retained_candidate_keys,
                        query.relations.content_selection,
                    ));
                    let remaining_candidate_capacity =
                        u32::try_from(retained_candidate_keys.len()).unwrap_or(u32::MAX);
                    let remaining_nodes = usize::try_from(
                        budget.nodes().saturating_sub(remaining_candidate_capacity),
                    )
                    .unwrap_or(usize::MAX);
                    let remaining_visited = usize::try_from(
                        budget
                            .visited()
                            .saturating_sub(remaining_candidate_capacity),
                    )
                    .unwrap_or(usize::MAX);
                    if candidate_unretained_keys.len() > remaining_nodes {
                        complete = false;
                        push_limit(&mut reached_limits, GraphLimitKind::Nodes);
                    }
                    if candidate_unretained_keys.len() > remaining_visited {
                        complete = false;
                        push_limit(&mut reached_limits, GraphLimitKind::Visited);
                    }
                    if !complete {
                        break;
                    }
                    for limit in &candidate_report.reached_limits {
                        if !filtered_edge_limit_is_terminal
                            || (*limit != GraphLimitKind::Edges && *limit != GraphLimitKind::Rows)
                        {
                            push_limit(&mut reached_limits, *limit);
                        }
                    }
                    if candidate_report.continuation.is_some() && !filtered_edge_limit_is_terminal {
                        push_limit(&mut reached_limits, GraphLimitKind::Rows);
                    }
                    if !entrypoint_report_complete(
                        &candidate_report,
                        &profile.relations,
                        filtered_edge_limit_is_terminal,
                    ) {
                        complete = false;
                        break;
                    }
                    candidate_report_anchor.get_or_insert(candidate_report.anchor);
                }
                if complete && let Some(candidate_report_anchor) = candidate_report_anchor {
                    unreachable.insert(candidate_key.clone(), candidate_report_anchor);
                    for key in candidate_unretained_keys {
                        retained_candidate_keys.insert(key);
                    }
                }
            }
        }
    }
    if complete {
        validate_entrypoint_generation(generation, store.repository_graph_generation())?;
        if purpose_revision_initialized
            && store.authored_purpose_revision()? != authored_purpose_revision
        {
            return Err(ServiceError::RelationCursorStale {
                field: "entrypoint authored purpose revision",
            });
        }
    }
    let coverage = if complete {
        EntrypointProfileCoverage::Complete
    } else {
        EntrypointProfileCoverage::Partial
    };
    let mut findings = Vec::new();
    if complete {
        findings.push(AnalysisFinding {
            kind: AnalysisFindingKind::EntrypointReachability,
            status: AnalysisStatus::Confirmed,
            summary: "explicit entrypoints reach these local entities through complete admitted relations"
                .to_string(),
            nodes: Vec::new(),
            metric: Some(reachable.len() as u64),
            evidence: None,
            community: None,
        });
        if !unreachable.is_empty() {
            findings.push(AnalysisFinding {
                kind: AnalysisFindingKind::EntrypointReachability,
                status: AnalysisStatus::Candidate,
                summary: "local entities are unreachable from the explicit entrypoints; review exact source evidence before any deletion decision"
                    .to_string(),
                nodes: Vec::new(),
                metric: Some(unreachable.len() as u64),
                evidence: None,
                community: None,
            });
        }
    } else {
        findings.push(AnalysisFinding {
            kind: AnalysisFindingKind::EntrypointReachability,
            status: AnalysisStatus::Inconclusive,
            summary: "entrypoint reachability is inconclusive because relation coverage or a declared bound is partial"
                .to_string(),
            nodes: Vec::new(),
            metric: None,
            evidence: None,
            community: None,
        });
    }
    let mut profile_result = EntrypointProfileResult {
        name: profile.name.clone(),
        anchors: profile.anchors.clone(),
        relations: profile.relations.clone(),
        coverage,
        reachable: u32::try_from(reachable.len()).unwrap_or(u32::MAX),
        unreachable_candidates: if complete {
            u32::try_from(unreachable.len()).unwrap_or(u32::MAX)
        } else {
            0
        },
    };
    let mut work = RelationAnalysisWork {
        relations: relation_work,
        analyzed_nodes: u32::try_from(reachable.len().saturating_add(unreachable.len()))
            .unwrap_or(u32::MAX),
        analyzed_edges: u32::try_from(edges.len()).unwrap_or(u32::MAX),
        ..RelationAnalysisWork::default()
    };
    let metadata_composition_bytes =
        serialized_bytes_controlled(&(&findings, &profile_result), control)?;
    let mut node_composition_bytes = serialized_analysis_nodes_bytes(reachable.values(), control)?;
    if complete && !unreachable.is_empty() {
        node_composition_bytes = node_composition_bytes
            .checked_add(serialized_analysis_nodes_bytes(
                unreachable.values(),
                control,
            )?)
            .ok_or_else(entrypoint_work_overflow)?;
    }
    let composition_fits = work
        .relations
        .intermediate_bytes
        .checked_add(metadata_composition_bytes)
        .and_then(|bytes| bytes.checked_add(node_composition_bytes))
        .is_some_and(|bytes| bytes <= budget.intermediate_bytes());
    if composition_fits {
        if complete {
            findings[0].nodes = reachable.values().map(analysis_node).collect();
            if !unreachable.is_empty() {
                findings[1].nodes = unreachable.values().map(analysis_node).collect();
            }
            profile_result.unreachable_candidates =
                u32::try_from(unreachable.len()).unwrap_or(u32::MAX);
        } else {
            findings[0].nodes = reachable.values().map(analysis_node).collect();
        }
    } else {
        complete = false;
        work.composition_truncated = true;
        push_limit(&mut reached_limits, GraphLimitKind::IntermediateBytes);
        profile_result.coverage = EntrypointProfileCoverage::Partial;
        profile_result.reachable = 0;
        profile_result.unreachable_candidates = 0;
        findings.truncate(1);
        findings[0].status = AnalysisStatus::Inconclusive;
        findings[0].summary =
            "entrypoint evidence was omitted because the declared composition bound was partial"
                .to_string();
        findings[0].nodes.clear();
        findings[0].metric = None;
    }
    work.retained_composition_bytes =
        serialized_bytes_controlled(&(&findings, &profile_result), control)?;
    work.peak_intermediate_bytes = work
        .relations
        .intermediate_bytes
        .checked_add(work.retained_composition_bytes)
        .ok_or_else(entrypoint_work_overflow)?;
    if work.peak_intermediate_bytes > budget.intermediate_bytes() {
        return Err(ServiceError::InvalidInput(
            "entrypoint profile construction exceeds its aggregate intermediate-byte budget"
                .to_string(),
        ));
    }
    if !complete {
        profile_result.coverage = EntrypointProfileCoverage::Partial;
        for finding in &mut findings {
            finding.status = AnalysisStatus::Inconclusive;
        }
    }
    let report = RelationAnalysisReport {
        mode: RelationAnalysisMode::Entrypoint,
        anchor,
        generation,
        authored_purpose_revision,
        continuation: None,
        returned: u32::try_from(findings.len()).unwrap_or(u32::MAX),
        total: if complete {
            RelationTotalState::Exact(findings.len() as u64)
        } else {
            RelationTotalState::AtLeast(findings.len() as u64)
        },
        truncated: !complete,
        reached_limits,
        vcs: VcsImpact::NotRequested,
        entrypoint_profile: Some(profile_result),
        work,
        findings,
    };
    let cursor_snapshot = AnalysisCursorSnapshot {
        project: generation_project(&report.anchor.entity),
        generation,
        authored_purpose_revision,
    };
    let cursor_binding = analysis_cursor_binding(query, &selected_binding.project_root_identity)?;
    Ok(RelationAnalysisDraft {
        report,
        output_bytes: budget.output_bytes(),
        budget,
        cursor_binding,
        cursor_snapshot,
        replay_relation_cursor: None,
        finding_offset: 0,
        vcs_digest: None,
        external_relation_identities: BTreeSet::new(),
        control: analysis_control,
    })
}

/// Reject a profile whose required initial anchors cannot fit its shared state.
fn validate_entrypoint_profile_budget(
    profile: &EntrypointProfile,
    budget: DetailedRelationBudget,
) -> ServiceResult<()> {
    let anchor_count = u32::try_from(profile.anchors.len()).unwrap_or(u32::MAX);
    if anchor_count > budget.nodes() {
        return Err(ServiceError::InvalidInput(
            "entrypoint profile anchor count exceeds the node budget".to_string(),
        ));
    }
    if anchor_count > budget.visited() {
        return Err(ServiceError::InvalidInput(
            "entrypoint profile anchor count exceeds the visited budget".to_string(),
        ));
    }
    Ok(())
}

/// Add one detailed traversal work ledger without widening its types.
fn add_relation_work(
    total: &mut DetailedRelationWork,
    next: &DetailedRelationWork,
) -> ServiceResult<()> {
    total.returned_rows = total
        .returned_rows
        .checked_add(next.returned_rows)
        .ok_or_else(entrypoint_work_overflow)?;
    total.inspected_edges = total
        .inspected_edges
        .checked_add(next.inspected_edges)
        .ok_or_else(entrypoint_work_overflow)?;
    total.active_nodes = total.active_nodes.max(next.active_nodes);
    total.visited_nodes = total.visited_nodes.max(next.visited_nodes);
    total.retained_occurrences = total
        .retained_occurrences
        .checked_add(next.retained_occurrences)
        .ok_or_else(entrypoint_work_overflow)?;
    total.database_requested_rows = total
        .database_requested_rows
        .checked_add(next.database_requested_rows)
        .ok_or_else(entrypoint_work_overflow)?;
    total.database_returned_rows = total
        .database_returned_rows
        .checked_add(next.database_returned_rows)
        .ok_or_else(entrypoint_work_overflow)?;
    total.database_decoded_bytes = total
        .database_decoded_bytes
        .checked_add(next.database_decoded_bytes)
        .ok_or_else(entrypoint_work_overflow)?;
    total.hydrated_entities = total
        .hydrated_entities
        .checked_add(next.hydrated_entities)
        .ok_or_else(entrypoint_work_overflow)?;
    total.hydrated_purpose_paths = total
        .hydrated_purpose_paths
        .checked_add(next.hydrated_purpose_paths)
        .ok_or_else(entrypoint_work_overflow)?;
    total.hydrated_classification_paths = total
        .hydrated_classification_paths
        .checked_add(next.hydrated_classification_paths)
        .ok_or_else(entrypoint_work_overflow)?;
    total.retained_composition_bytes = total
        .retained_composition_bytes
        .saturating_add(next.retained_composition_bytes);
    total.intermediate_bytes = total
        .intermediate_bytes
        .saturating_add(next.intermediate_bytes);
    total.rendered_output_bytes = total
        .rendered_output_bytes
        .saturating_add(next.rendered_output_bytes);
    Ok(())
}

/// Add one bounded repository-graph read to the profile work ledger.
fn add_repository_read_work(
    total: &mut DetailedRelationWork,
    next: &RepositoryGraphReadWork,
) -> ServiceResult<()> {
    total.database_requested_rows = total
        .database_requested_rows
        .checked_add(next.requested_rows)
        .ok_or_else(entrypoint_work_overflow)?;
    total.database_returned_rows = total
        .database_returned_rows
        .checked_add(next.returned_rows)
        .ok_or_else(entrypoint_work_overflow)?;
    total.database_decoded_bytes = total
        .database_decoded_bytes
        .checked_add(next.decoded_bytes)
        .ok_or_else(entrypoint_work_overflow)?;
    total.hydrated_entities = total
        .hydrated_entities
        .checked_add(next.hydrated_entities)
        .ok_or_else(entrypoint_work_overflow)?;
    total.hydrated_purpose_paths = total
        .hydrated_purpose_paths
        .checked_add(next.hydrated_paths)
        .ok_or_else(entrypoint_work_overflow)?;
    total.intermediate_bytes = total.intermediate_bytes.saturating_add(next.decoded_bytes);
    Ok(())
}

/// Load candidate classifications in bounded, cancellable batches.
#[cfg(test)]
fn load_entrypoint_candidate_classifications<'entity>(
    store: &AtlasStore,
    entities: impl IntoIterator<Item = &'entity GraphEntity>,
    budget: DetailedRelationBudget,
    relation_work: &mut DetailedRelationWork,
    control: Option<&IndexWorkControl>,
) -> ServiceResult<Option<BTreeMap<String, projectatlas_core::language::ContentClassification>>> {
    let mut path_set = BTreeSet::new();
    for entity in entities {
        check_control(control)?;
        let Some(path) = classification_path(entity) else {
            continue;
        };
        if path_set.contains(&path) {
            continue;
        }
        let path_bytes = classification_path_bytes(&path)?;
        if relation_work
            .intermediate_bytes
            .checked_add(path_bytes)
            .is_none_or(|bytes| bytes > budget.intermediate_bytes())
        {
            return Ok(None);
        }
        path_set.insert(path);
        relation_work.intermediate_bytes = relation_work
            .intermediate_bytes
            .checked_add(path_bytes)
            .ok_or_else(entrypoint_work_overflow)?;
    }
    let paths = path_set.into_iter().collect::<Vec<_>>();
    let mut classifications = BTreeMap::new();
    for chunk in paths.chunks(MAX_FILE_CONTENT_CLASSIFICATION_PATHS) {
        check_control(control)?;
        #[cfg(test)]
        analysis_test_observer::notify(
            analysis_test_observer::AnalysisPhaseEvent::ClassificationHydration,
        );
        let row_upper_bound = classification_rows_upper_bound(chunk, control)?;
        if relation_work
            .intermediate_bytes
            .checked_add(row_upper_bound)
            .is_none_or(|bytes| bytes > budget.intermediate_bytes())
        {
            return Ok(None);
        }
        let rows = store.file_content_classifications_for_paths(chunk)?;
        check_control(control)?;
        let decoded_bytes = classification_rows_bytes(&rows, control)?;
        let retained_bytes = classification_rows_bytes(&rows, control)?;
        let requested_rows =
            u32::try_from(chunk.len()).map_err(|_overflow| entrypoint_work_overflow())?;
        let returned_rows =
            u32::try_from(rows.len()).map_err(|_overflow| entrypoint_work_overflow())?;
        relation_work.database_requested_rows = relation_work
            .database_requested_rows
            .checked_add(requested_rows)
            .ok_or_else(entrypoint_work_overflow)?;
        relation_work.database_returned_rows = relation_work
            .database_returned_rows
            .checked_add(returned_rows)
            .ok_or_else(entrypoint_work_overflow)?;
        relation_work.database_decoded_bytes = relation_work
            .database_decoded_bytes
            .checked_add(decoded_bytes)
            .ok_or_else(entrypoint_work_overflow)?;
        relation_work.hydrated_classification_paths = relation_work
            .hydrated_classification_paths
            .checked_add(returned_rows)
            .ok_or_else(entrypoint_work_overflow)?;
        let retained_intermediate = relation_work
            .intermediate_bytes
            .checked_add(decoded_bytes)
            .and_then(|bytes| bytes.checked_add(retained_bytes))
            .ok_or_else(entrypoint_work_overflow)?;
        if retained_intermediate > budget.intermediate_bytes() {
            return Ok(None);
        }
        relation_work.intermediate_bytes = retained_intermediate;
        classifications.extend(rows.into_iter().map(|row| (row.path, row.classification)));
        check_control(control)?;
    }
    Ok(Some(classifications))
}

/// Count one bounded path retained while classification batches run.
#[cfg(test)]
fn classification_path_bytes(path: &str) -> ServiceResult<u64> {
    u64::try_from(path.len())
        .ok()
        .and_then(|bytes| bytes.checked_add(8))
        .ok_or_else(entrypoint_work_overflow)
}

/// Bound one classification batch before materializing its database rows.
#[cfg(test)]
fn classification_rows_upper_bound(
    paths: &[String],
    control: Option<&IndexWorkControl>,
) -> ServiceResult<u64> {
    let maximum_classification_bytes = ContentClassification::ALL
        .iter()
        .map(|classification| classification.as_str().len())
        .max()
        .map_or(0, |bytes| u64::try_from(bytes).unwrap_or(u64::MAX));
    paths.iter().try_fold(0_u64, |bytes, path| {
        check_control(control)?;
        let row_bytes = u64::try_from(path.len())
            .ok()
            .and_then(|path_bytes| path_bytes.checked_add(maximum_classification_bytes))
            .and_then(|row_bytes| row_bytes.checked_add(8))
            .and_then(|row_bytes| row_bytes.checked_mul(2))
            .ok_or_else(entrypoint_work_overflow)?;
        bytes
            .checked_add(row_bytes)
            .ok_or_else(entrypoint_work_overflow)
    })
}

/// Count conservative decoded and retained bytes for one classification batch.
#[cfg(test)]
fn classification_rows_bytes(
    rows: &[projectatlas_db::FileContentClassification],
    control: Option<&IndexWorkControl>,
) -> ServiceResult<u64> {
    rows.iter().try_fold(0_u64, |bytes, row| {
        check_control(control)?;
        let row_bytes = u64::try_from(row.path.len())
            .ok()
            .and_then(|path| {
                u64::try_from(row.classification.as_str().len())
                    .ok()
                    .and_then(|classification| path.checked_add(classification))
            })
            .and_then(|bytes| bytes.checked_add(8))
            .ok_or_else(entrypoint_work_overflow)?;
        bytes
            .checked_add(row_bytes)
            .ok_or_else(entrypoint_work_overflow)
    })
}

/// Derive one initial-anchor budget from the remaining profile-wide capacity.
fn entrypoint_anchor_budget(
    budget: DetailedRelationBudget,
    work: &DetailedRelationWork,
) -> ServiceResult<Result<DetailedRelationBudget, GraphLimitKind>> {
    let remaining_intermediate = budget
        .intermediate_bytes()
        .saturating_sub(work.intermediate_bytes);
    if remaining_intermediate < 64 * 1_024 {
        return Ok(Err(GraphLimitKind::IntermediateBytes));
    }
    Ok(Ok(budget.with_aggregate_limits(
        None,
        None,
        None,
        None,
        Some(remaining_intermediate),
        None,
    )?))
}

/// Check whether occurrence hydration omitted evidence after the profile total was exhausted.
fn entrypoint_occurrence_evidence_is_incomplete(
    store: &AtlasStore,
    report: &DetailedRelationReport,
    generation: projectatlas_core::IndexGeneration,
    budget: DetailedRelationBudget,
    relation_work: &mut DetailedRelationWork,
    control: Option<&IndexWorkControl>,
    include_report_rows: bool,
) -> ServiceResult<Result<bool, GraphLimitKind>> {
    let mut admitted_relations = if include_report_rows {
        report
            .rows
            .iter()
            .map(|row| row.relation.clone())
            .collect::<Vec<_>>()
    } else {
        Vec::new()
    };
    admitted_relations.extend(report.pruned_relations.iter().cloned());
    if admitted_relations.is_empty() {
        return Ok(Ok(false));
    }
    for chunk in admitted_relations.chunks(MAX_REPOSITORY_GRAPH_FRONTIER) {
        check_control(control)?;
        let remaining_intermediate = budget
            .intermediate_bytes()
            .saturating_sub(relation_work.intermediate_bytes);
        if remaining_intermediate < 64 * 1_024 {
            return Ok(Err(GraphLimitKind::IntermediateBytes));
        }
        let relations = chunk.to_vec();
        let batch_rows = u32::try_from(relations.len()).map_err(|_overflow| {
            ServiceError::InvalidInput("occurrence evidence batch size overflowed".to_string())
        })?;
        let read_budget = RepositoryGraphReadBudget::new(
            batch_rows,
            batch_rows,
            remaining_intermediate.min(RepositoryGraphReadBudget::MAX_DECODED_BYTES),
            1,
            batch_rows.saturating_mul(2).max(1),
        )
        .map_err(|error| ServiceError::InvalidInput(error.to_string()))?;
        #[cfg(test)]
        analysis_test_observer::notify(
            analysis_test_observer::AnalysisPhaseEvent::OccurrenceProbeBeforeRead,
        );
        let batch = match store.repository_graph_occurrence_pages_bounded(
            &relations,
            1,
            read_budget,
            control,
        ) {
            Ok(batch) => batch,
            Err(DbError::GraphContract(
                projectatlas_core::graph::GraphContractError::InvalidLimits {
                    reason: "graph read decoded bytes exceed the batch budget",
                },
            )) => return Ok(Err(GraphLimitKind::IntermediateBytes)),
            Err(
                DbError::GraphContract(
                    projectatlas_core::graph::GraphContractError::GenerationMismatch { .. },
                )
                | DbError::GraphRowShape {
                    table: "project_identity",
                    reason: "typed graph generation does not match complete publication",
                },
            ) => {
                return Err(ServiceError::RelationCursorStale {
                    field: "entrypoint graph generation",
                });
            }
            Err(error) => return Err(error.into()),
        };
        add_repository_read_work(relation_work, &batch.work)?;
        #[cfg(test)]
        analysis_test_observer::notify(analysis_test_observer::AnalysisPhaseEvent::OccurrenceProbe);
        let current_generation = match store.repository_graph_generation() {
            Ok(generation) => generation.ok_or_else(|| {
                ServiceError::InvalidInput(
                    "repository graph has no complete generation for entrypoint analysis"
                        .to_string(),
                )
            })?,
            Err(DbError::GraphRowShape {
                table: "project_identity",
                reason: "typed graph generation does not match complete publication",
            }) => {
                return Err(ServiceError::RelationCursorStale {
                    field: "entrypoint graph generation",
                });
            }
            Err(error) => return Err(error.into()),
        };
        if current_generation != generation {
            return Err(ServiceError::RelationCursorStale {
                field: "entrypoint graph generation",
            });
        }
        if batch
            .pages
            .iter()
            .any(|page| page.truncated || !page.rows.is_empty())
        {
            return Ok(Ok(true));
        }
    }
    Ok(Ok(false))
}

/// Retain one resolved entity when no relation page was needed for it.
fn entrypoint_unavailable_node(
    entity: &GraphEntity,
    content_selection: ContentSelection,
) -> DetailedRelationNode {
    let path = entity_path(entity).map(str::to_string);
    DetailedRelationNode {
        entity: entity.clone(),
        classification: None,
        content_selection: content_selection
            .explicit_value()
            .map(|_| content_selection),
        purpose: RelationPurpose::Unavailable { path },
        coverage: Vec::new(),
    }
}

/// Protect every indexed symbol that encloses a reachable symbol.
fn protect_reachable_symbol_enclosures(
    reachable: &BTreeMap<String, DetailedRelationNode>,
    entities: &[GraphEntity],
    protected_keys: &mut BTreeSet<String>,
) {
    let mut pending = reachable
        .values()
        .filter_map(|node| match node.entity.selector() {
            EntitySelector::Symbol { symbol } => symbol.parent.as_ref().map(|parent| {
                (
                    symbol.file.as_str().to_string(),
                    parent.as_str().to_string(),
                )
            }),
            _ => None,
        })
        .collect::<VecDeque<_>>();
    while let Some((file, parent)) = pending.pop_front() {
        for entity in entities {
            let EntitySelector::Symbol { symbol } = entity.selector() else {
                continue;
            };
            let qualified_name_matches = symbol.name.as_str() == parent
                || symbol.parent.as_ref().is_some_and(|candidate_parent| {
                    format!("{candidate_parent}::{name}", name = symbol.name.as_str()) == parent
                        || format!("{candidate_parent}.{name}", name = symbol.name.as_str())
                            == parent
                });
            if symbol.file.as_str() != file || !qualified_name_matches {
                continue;
            }
            let key = entity.key().canonical_identity().to_string();
            if protected_keys.insert(key)
                && let Some(next_parent) = symbol.parent.as_ref()
            {
                pending.push_back((file.clone(), next_parent.as_str().to_string()));
            }
        }
    }
}

/// Derive one one-step detailed budget from the remaining profile-wide capacity.
fn entrypoint_step_budget(
    budget: DetailedRelationBudget,
    work: &DetailedRelationWork,
    retained_nodes: usize,
    validated_candidates: usize,
    anchor_is_retained: bool,
    candidate_anchor_overhead: bool,
    include_occurrences: bool,
    remaining_edges_override: Option<u32>,
) -> ServiceResult<Result<DetailedRelationBudget, GraphLimitKind>> {
    let accounted_nodes = retained_nodes.saturating_add(validated_candidates);
    let accounted_nodes = u32::try_from(accounted_nodes).unwrap_or(u32::MAX);
    let validated_candidates = u32::try_from(validated_candidates).unwrap_or(u32::MAX);
    let remaining_edges = remaining_edges_override
        .unwrap_or_else(|| budget.edges().saturating_sub(work.inspected_edges));
    let remaining_nodes = budget.nodes().saturating_sub(accounted_nodes);
    let remaining_visited = budget.visited().saturating_sub(accounted_nodes);
    let step_nodes = if candidate_anchor_overhead {
        budget.nodes()
    } else if anchor_is_retained {
        budget.nodes().saturating_sub(validated_candidates)
    } else {
        remaining_nodes
    };
    let step_visited = if candidate_anchor_overhead {
        budget.visited()
    } else if anchor_is_retained {
        budget.visited().saturating_sub(validated_candidates)
    } else {
        remaining_visited
    };
    let remaining_occurrences = if include_occurrences {
        budget
            .occurrences_total()
            .saturating_sub(work.retained_occurrences)
    } else {
        budget.occurrences_total()
    };
    let remaining_intermediate = budget
        .intermediate_bytes()
        .saturating_sub(work.intermediate_bytes);
    let rows = budget.page_rows().min(remaining_edges);
    if remaining_edges == 0 {
        return Ok(Err(GraphLimitKind::Edges));
    }
    if remaining_nodes == 0 && !anchor_is_retained {
        return Ok(Err(GraphLimitKind::Nodes));
    }
    if remaining_visited == 0 && !anchor_is_retained {
        return Ok(Err(GraphLimitKind::Visited));
    }
    if remaining_intermediate < 64 * 1_024 {
        return Ok(Err(GraphLimitKind::IntermediateBytes));
    }
    if rows == 0 {
        return Ok(Err(GraphLimitKind::Rows));
    }
    let limits = GraphLimits::new(
        rows,
        budget.occurrences_per_relation(),
        1,
        budget.output_bytes(),
    )
    .map_err(|error| ServiceError::InvalidInput(error.to_string()))?;
    let step = DetailedRelationBudget::from_graph_limits(limits).with_aggregate_limits(
        Some(remaining_edges),
        Some(step_nodes),
        Some(step_visited),
        Some(remaining_occurrences),
        Some(remaining_intermediate),
        Some(budget.deadline_ms()),
    )?;
    Ok(Ok(step))
}

/// Return the typed error for aggregate entrypoint work overflow.
fn entrypoint_work_overflow() -> ServiceError {
    ServiceError::InvalidInput("entrypoint relation work overflowed".to_string())
}

/// Reconcile a raw edge limit when no admitted row remains after the page.
fn entrypoint_filtered_edge_limit_is_terminal(
    store: &AtlasStore,
    generation: projectatlas_core::IndexGeneration,
    report: &DetailedRelationReport,
    relation: GraphRelationKind,
    minimum_confidence: ConfidenceClass,
    content_selection: ContentSelection,
    control: Option<&IndexWorkControl>,
) -> ServiceResult<bool> {
    if !report.reached_limits.contains(&GraphLimitKind::Edges)
        || report.reached_limits.iter().any(|limit| {
            *limit != GraphLimitKind::Edges
                && (report.rows.is_empty() || *limit != GraphLimitKind::Rows)
        })
        || report.pruned_incomplete_paths > 0
        || report.pruned_evidence_truncated
    {
        return Ok(false);
    }
    if let Some(continuation) = report.adjacency_continuation.as_ref() {
        let has_rows = store.repository_graph_adjacency_continuation_has_filtered_rows(
            continuation,
            minimum_confidence,
            content_selection,
            control,
        )?;
        validate_entrypoint_generation(generation, store.repository_graph_generation())?;
        #[cfg(test)]
        analysis_test_observer::notify(analysis_test_observer::AnalysisPhaseEvent::TerminalProbe);
        return Ok(!has_rows);
    }
    if !report.rows.is_empty() {
        return Ok(false);
    }
    entrypoint_terminal_adjacency_is_empty(
        store,
        generation,
        report.anchor.entity.key(),
        relation,
        minimum_confidence,
        content_selection,
        control,
    )
}

/// Return the project identity for an already generation-bound graph entity.
fn generation_project(entity: &GraphEntity) -> ProjectInstanceId {
    entity.key().project()
}

/// Probe one terminal adjacency and reject a publication change immediately afterward.
fn entrypoint_terminal_adjacency_is_empty(
    store: &AtlasStore,
    generation: projectatlas_core::IndexGeneration,
    key: &GraphEntityKey,
    relation: GraphRelationKind,
    minimum_confidence: ConfidenceClass,
    content_selection: ContentSelection,
    control: Option<&IndexWorkControl>,
) -> ServiceResult<bool> {
    let empty = store.repository_graph_adjacency_is_empty_filtered(
        key,
        RepositoryGraphDirection::Outbound,
        relation,
        minimum_confidence,
        content_selection,
        control,
    )?;
    #[cfg(test)]
    analysis_test_observer::notify(analysis_test_observer::AnalysisPhaseEvent::TerminalProbe);
    validate_entrypoint_generation(generation, store.repository_graph_generation())?;
    Ok(empty)
}

/// Load trusted coverage for a candidate whose admitted terminal adjacency is empty.
fn load_entrypoint_terminal_candidate_coverage(
    store: &AtlasStore,
    entity: &GraphEntity,
    generation: projectatlas_core::IndexGeneration,
    budget: DetailedRelationBudget,
    relation_work: &mut DetailedRelationWork,
    content_selection: ContentSelection,
    control: Option<&IndexWorkControl>,
) -> ServiceResult<Result<DetailedRelationNode, GraphLimitKind>> {
    check_control(control)?;
    let remaining_intermediate = budget
        .intermediate_bytes()
        .saturating_sub(relation_work.intermediate_bytes);
    if remaining_intermediate < 64 * 1_024 {
        return Ok(Err(GraphLimitKind::IntermediateBytes));
    }
    let candidate_budget =
        budget.with_aggregate_limits(None, None, None, None, Some(remaining_intermediate), None)?;
    let (node, metadata_work) = match hydrate_single_detailed_node(
        store,
        entity,
        generation,
        content_selection,
        candidate_budget,
        control,
    ) {
        Ok(value) => value,
        Err(ServiceError::Db(DbError::GraphContract(
            projectatlas_core::graph::GraphContractError::InvalidLimits {
                reason: "graph read decoded bytes exceed the batch budget",
            },
        ))) => return Ok(Err(GraphLimitKind::IntermediateBytes)),
        Err(ServiceError::InvalidInput(reason))
            if reason == "detailed relation intermediate-byte budget is exhausted"
                || reason == "terminal node metadata exceeded the intermediate-byte budget" =>
        {
            return Ok(Err(GraphLimitKind::IntermediateBytes));
        }
        Err(ServiceError::Db(
            DbError::GraphContract(
                projectatlas_core::graph::GraphContractError::GenerationMismatch { .. },
            )
            | DbError::GraphRowShape {
                table: "project_identity",
                reason: "typed graph generation does not match complete publication",
            },
        )) => {
            return Err(ServiceError::RelationCursorStale {
                field: "entrypoint graph generation",
            });
        }
        Err(error) => return Err(error),
    };
    add_relation_work(relation_work, &metadata_work)?;
    #[cfg(test)]
    analysis_test_observer::notify(
        analysis_test_observer::AnalysisPhaseEvent::TerminalCandidateCoverageProbe {
            intermediate_bytes: relation_work.intermediate_bytes,
        },
    );
    validate_entrypoint_generation(generation, store.repository_graph_generation())?;
    check_control(control)?;
    Ok(Ok(node))
}

/// Return whether a detailed node carries trusted coverage for admitted relations.
fn trusted_node_coverage(
    node: &DetailedRelationNode,
    admitted_relations: &[GraphRelationKind],
) -> bool {
    let applies = |coverage: &CoverageRecord| {
        coverage
            .relation()
            .is_none_or(|relation| admitted_relations.contains(&relation))
    };
    let applicable = node
        .coverage
        .iter()
        .filter(|coverage| applies(coverage))
        .collect::<Vec<_>>();
    !applicable.is_empty()
        && applicable
            .iter()
            .all(|coverage| coverage_trust(coverage.state()) == CoverageTrustState::Trusted)
        && (applicable
            .iter()
            .any(|coverage| coverage.relation().is_none())
            || admitted_relations.iter().all(|relation| {
                applicable
                    .iter()
                    .any(|coverage| coverage.relation() == Some(*relation))
            }))
}

/// Return whether every local endpoint in one relation row has trusted coverage.
fn trusted_relation_row(
    row: &DetailedRelationRow,
    admitted_relations: &[GraphRelationKind],
) -> bool {
    trusted_node_coverage(&row.source, admitted_relations)
        && row.target.as_ref().is_none_or(|target| {
            matches!(target.entity.selector(), EntitySelector::External { .. })
                || trusted_node_coverage(target, admitted_relations)
        })
        && row
            .path
            .iter()
            .all(|node| trusted_node_coverage(node, admitted_relations))
}

/// Return whether one terminal candidate traversal supports a safe negative.
fn entrypoint_report_complete(
    report: &DetailedRelationReport,
    admitted_relations: &[GraphRelationKind],
    filtered_edge_limit_is_terminal: bool,
) -> bool {
    (filtered_edge_limit_is_terminal
        || (matches!(
            report.total,
            RelationTotalState::Exact(total) if total == u64::from(report.returned)
        ) && !report.truncated
            && report.continuation.is_none()
            && report.reached_limits.is_empty()))
        && report.pruned_incomplete_paths == 0
        && !report.pruned_evidence_truncated
        && trusted_node_coverage(&report.anchor, admitted_relations)
        && report.rows.iter().all(|row| {
            matches!(
                row.relation.resolution(),
                RelationResolution::Resolved { .. } | RelationResolution::External { .. }
            ) && row.relation.completeness() == Completeness::Complete
                && trusted_relation_row(row, admitted_relations)
        })
}

/// Return local identities exposed by candidate traversal that were not retained already.
fn candidate_report_unretained_local_keys(
    report: &DetailedRelationReport,
    retained_keys: &BTreeSet<String>,
    content_selection: ContentSelection,
) -> BTreeSet<String> {
    let mut keys = BTreeSet::new();
    let mut retain = |node: &DetailedRelationNode| {
        if !matches!(node.entity.selector(), EntitySelector::External { .. })
            && (content_selection == ContentSelection::UnspecifiedLegacy
                || node
                    .classification
                    .is_some_and(|classification| content_selection.includes(classification)))
            && !retained_keys.contains(node.entity.key().canonical_identity())
        {
            keys.insert(node.entity.key().canonical_identity().to_string());
        }
    };
    retain(&report.anchor);
    for row in &report.rows {
        retain(&row.source);
        if let Some(target) = row.target.as_ref() {
            retain(target);
        }
        row.path.iter().for_each(&mut retain);
    }
    keys
}

/// Require one unchanged complete generation at a freshness boundary.
fn validate_entrypoint_generation(
    expected: IndexGeneration,
    current: Result<Option<IndexGeneration>, DbError>,
) -> ServiceResult<()> {
    match current {
        Ok(Some(current)) if current == expected => Ok(()),
        Ok(Some(_) | None)
        | Err(
            DbError::GraphPublicationUnavailable
            | DbError::GraphRowShape {
                table: "project_identity",
                reason: "typed graph generation does not match complete publication",
            }
            | DbError::GraphContract(
                projectatlas_core::graph::GraphContractError::GenerationMismatch { .. },
            ),
        ) => Err(ServiceError::RelationCursorStale {
            field: "entrypoint graph generation",
        }),
        Err(error) => Err(error.into()),
    }
}

/// Clamp the existing traversal budget to analysis product ceilings.
fn bounded_analysis_budget(
    budget: DetailedRelationBudget,
) -> ServiceResult<DetailedRelationBudget> {
    let limits = GraphLimits::new(
        budget.page_rows().min(MAX_ANALYSIS_NODES),
        budget.occurrences_per_relation(),
        budget.depth(),
        budget.output_bytes(),
    )
    .map_err(|error| ServiceError::InvalidInput(error.to_string()))?;
    DetailedRelationBudget::from_graph_limits(limits).with_aggregate_limits(
        Some(budget.edges().min(MAX_ANALYSIS_EDGES)),
        Some(budget.nodes().min(MAX_ANALYSIS_NODES)),
        Some(budget.visited().min(MAX_ANALYSIS_NODES)),
        Some(budget.occurrences_total()),
        Some(budget.intermediate_bytes()),
        Some(budget.deadline_ms()),
    )
}

/// Normalize every result-defining request field for cursor identity.
fn analysis_cursor_binding(
    query: &RelationAnalysisQuery,
    project_root: &CanonicalProjectRoot,
) -> ServiceResult<AnalysisCursorBinding> {
    Ok(AnalysisCursorBinding {
        root_digest: super::canonical_root_digest(ANALYSIS_ROOT_DOMAIN, project_root)?,
        anchor: query.relations.anchor.clone(),
        direction: query.relations.direction,
        relation: query.relations.relation,
        minimum_confidence: query.relations.minimum_confidence,
        resolution: query.relations.resolution,
        content_selection: query
            .relations
            .content_selection
            .explicit_value()
            .map(|_| query.relations.content_selection),
        options: AnalysisCursorOptions {
            relation_occurrences: query.relations.include_occurrences.into(),
            communities: query.include_communities.into(),
            cycles: query.include_cycles.into(),
            dead_code: query.include_dead_code.into(),
        },
        budget: query.relations.budget,
        algorithm_version: ANALYSIS_CURSOR_VERSION,
        ordering_version: 1,
        mode: query.mode,
        trace_target: query.trace_target.clone(),
        vcs: (query.mode == RelationAnalysisMode::Impact)
            .then(|| query.vcs.clone().unwrap_or(GitImpactSelection::WorkingTree)),
        entrypoint_profile: query.entrypoint_profile.clone(),
    })
}

/// Decode and validate one analysis continuation against its request.
fn decode_analysis_cursor(
    encoded: &str,
    expected: &AnalysisCursorBinding,
) -> ServiceResult<AnalysisCursor> {
    if encoded.is_empty() || encoded.len() > ANALYSIS_CURSOR_MAX_BYTES {
        return Err(ServiceError::RelationCursorInvalid {
            reason: "analysis cursor length is empty or above the product ceiling",
        });
    }
    let cursor: AnalysisCursor = serde_json::from_str(encoded).map_err(|_malformed| {
        ServiceError::RelationCursorInvalid {
            reason: "analysis cursor JSON is malformed or contains unknown fields",
        }
    })?;
    if cursor.version != ANALYSIS_CURSOR_VERSION {
        return Err(ServiceError::RelationCursorStale {
            field: "analysis algorithm version",
        });
    }
    if cursor.binding != *expected {
        return Err(ServiceError::RelationCursorMismatched {
            field: "analysis query",
        });
    }
    Ok(cursor)
}

/// Encode one bounded analysis continuation.
fn encode_analysis_cursor(
    relation_cursor: Option<&str>,
    finding_offset: u32,
    binding: &AnalysisCursorBinding,
    snapshot: AnalysisCursorSnapshot,
    vcs_digest: Option<[u8; 32]>,
    budget: DetailedRelationBudget,
) -> ServiceResult<String> {
    let encoded = serde_json::to_string(&AnalysisCursor {
        version: ANALYSIS_CURSOR_VERSION,
        binding: binding.clone(),
        snapshot,
        relation_cursor: relation_cursor.map(str::to_string),
        finding_offset,
        vcs_digest,
    })?;
    if encoded.len() > ANALYSIS_CURSOR_MAX_BYTES
        || encoded.len() > budget.intermediate_bytes() as usize
    {
        return Err(ServiceError::RelationCursorInvalid {
            reason: "encoded analysis cursor exceeds the intermediate-state ceiling",
        });
    }
    Ok(encoded)
}

/// Collect unique non-external nodes from a detailed relation page.
fn collect_nodes(
    report: &DetailedRelationReport,
    control: Option<&IndexWorkControl>,
) -> ServiceResult<BTreeMap<String, DetailedRelationNode>> {
    let mut nodes = BTreeMap::new();
    insert_node(&mut nodes, &report.anchor);
    for row in &report.rows {
        check_control(control)?;
        insert_node(&mut nodes, &row.source);
        if let Some(target) = &row.target {
            insert_node(&mut nodes, target);
        }
        for node in &row.path {
            insert_node(&mut nodes, node);
        }
    }
    Ok(nodes)
}

/// Project unresolved and ambiguous relation rows into typed findings.
fn resolution_gap_findings(
    report: &DetailedRelationReport,
    control: Option<&IndexWorkControl>,
) -> ServiceResult<Vec<AnalysisFinding>> {
    let mut findings = Vec::new();
    for row in &report.rows {
        check_control(control)?;
        if matches!(
            row.relation.resolution(),
            RelationResolution::Ambiguous { .. } | RelationResolution::Unresolved { .. }
        ) {
            findings.push(resolution_gap_finding(&row.relation, &row.source));
        }
    }
    Ok(findings)
}

/// Build one exact resolution-gap finding from a detailed relation row.
fn resolution_gap_finding(
    relation: &projectatlas_core::graph::LogicalRelation,
    source: &DetailedRelationNode,
) -> AnalysisFinding {
    AnalysisFinding {
        kind: AnalysisFindingKind::ResolutionGap,
        status: AnalysisStatus::Inconclusive,
        summary: "ambiguous or unresolved relation blocks a closed structural conclusion"
            .to_string(),
        nodes: vec![analysis_node(source)],
        metric: None,
        evidence: Some(AnalysisRelationEvidence {
            relation: relation.clone(),
            next_call: relation_gap_next_call(relation, source),
        }),
        community: None,
    }
}

/// Return the canonical relation identity used to sort and deduplicate gaps.
fn resolution_gap_identity(finding: &AnalysisFinding) -> &str {
    finding
        .evidence
        .as_ref()
        .map_or("", |evidence| evidence.relation.key().canonical_identity())
}

/// Build an exact existing relation request for one source-side gap.
fn relation_gap_next_call(
    relation: &projectatlas_core::graph::LogicalRelation,
    source: &DetailedRelationNode,
) -> Option<RelationAnalysisNextCall> {
    let anchor = relation_anchor_for_entity(&source.entity)?;
    let resolution = match relation.resolution() {
        RelationResolution::Ambiguous { .. } => RelationResolutionFilter::Ambiguous,
        RelationResolution::Unresolved { .. } => RelationResolutionFilter::Unresolved,
        RelationResolution::Resolved { .. } => RelationResolutionFilter::Resolved,
        RelationResolution::External { .. } => RelationResolutionFilter::External,
    };
    Some(RelationAnalysisNextCall {
        anchor,
        direction: RelationDirection::Outbound,
        relation: relation.kind(),
        resolution,
        minimum_confidence: relation.confidence(),
        content_selection: source.content_selection,
    })
}

/// Convert a locally addressable graph entity into a relation anchor.
fn relation_anchor_for_entity(entity: &GraphEntity) -> Option<RelationAnchor> {
    match entity.selector() {
        EntitySelector::File { path } => Some(RelationAnchor::File { file: path.clone() }),
        EntitySelector::Symbol { symbol } => Some(RelationAnchor::Symbol {
            file: symbol.file.clone(),
            name: symbol.name.as_str().to_string(),
            symbol_kind: Some(symbol.kind),
            parent: symbol
                .parent
                .as_ref()
                .map(|parent| parent.as_str().to_string()),
            signature: Some(symbol.signature.as_str().to_string()),
        }),
        EntitySelector::Project
        | EntitySelector::Folder { .. }
        | EntitySelector::Package { .. }
        | EntitySelector::External { .. } => None,
    }
}

/// Insert one local node once by canonical identity.
fn insert_node(nodes: &mut BTreeMap<String, DetailedRelationNode>, node: &DetailedRelationNode) {
    if !matches!(node.entity.selector(), EntitySelector::External { .. }) {
        nodes
            .entry(node.entity.key().canonical_identity().to_string())
            .or_insert_with(|| node.clone());
    }
}

/// Collect local edges from one detailed relation page.
fn collect_report_edges(
    report: &DetailedRelationReport,
    control: Option<&IndexWorkControl>,
) -> ServiceResult<Vec<LocalEdge>> {
    let mut edges = Vec::new();
    for row in &report.rows {
        check_control(control)?;
        if let Some(edge) = local_edge(&row.relation, &row.source.entity, row.target.as_ref()) {
            edges.push(edge);
        }
    }
    Ok(edges)
}

/// Convert a resolved local relation row into an algorithm edge.
fn local_edge(
    relation: &projectatlas_core::graph::LogicalRelation,
    source: &GraphEntity,
    target: Option<&DetailedRelationNode>,
) -> Option<LocalEdge> {
    let target = target?;
    if matches!(target.entity.selector(), EntitySelector::External { .. }) {
        return None;
    }
    Some(LocalEdge {
        source: source.key().canonical_identity().to_string(),
        target: target.entity.key().canonical_identity().to_string(),
        kind: relation.kind(),
        complete: relation.completeness() == Completeness::Complete,
    })
}

/// Completeness state for admitted community relations during closure.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
enum CommunityClosureScope {
    /// Every admitted community relation stayed within the induced nodes.
    #[default]
    Closed,
    /// At least one admitted community relation escaped the induced nodes.
    Open,
}

/// Work and evidence retained while closing edges among admitted nodes.
#[derive(Default)]
struct ClosureWork {
    /// Whether every admitted frontier completed under all bounds.
    complete: bool,
    /// Whether the closure-stage deadline stopped otherwise valid work.
    deadline_reached: bool,
    /// Whether the induced node scope was closed in the requested direction.
    induced_scope_closed: bool,
    /// Scope status for admitted community relations.
    community_scope: CommunityClosureScope,
    /// Database adjacency rows inspected by closure.
    inspected_edges: u32,
    /// Database-decoded bytes retained during closure.
    decoded_bytes: u64,
    /// Resolution gaps discovered after the first relation page.
    resolution_gaps: Vec<AnalysisFinding>,
}

/// Close bounded local edges among the nodes admitted by detailed traversal.
fn close_induced_edges(
    store: &AtlasStore,
    query: &RelationAnalysisQuery,
    relation_work: &DetailedRelationWork,
    deadline: Instant,
    nodes: &BTreeMap<String, DetailedRelationNode>,
    edges: &mut Vec<LocalEdge>,
    control: Option<&IndexWorkControl>,
) -> ServiceResult<ClosureWork> {
    let mut work = ClosureWork {
        complete: true,
        induced_scope_closed: query.relations.direction == RelationDirection::Outbound,
        community_scope: if query.relations.direction == RelationDirection::Outbound {
            CommunityClosureScope::Closed
        } else {
            CommunityClosureScope::Open
        },
        ..ClosureWork::default()
    };
    let mut keys = Vec::with_capacity(nodes.len());
    let mut known = BTreeSet::new();
    for (key, node) in nodes {
        check_control(control)?;
        keys.push(node.entity.key().clone());
        known.insert(key.clone());
    }
    #[cfg(test)]
    analysis_test_observer::notify(analysis_test_observer::AnalysisPhaseEvent::Traversal);
    check_control(control)?;
    for chunk in keys.chunks(MAX_REPOSITORY_GRAPH_FRONTIER) {
        let mut continuation: Option<RepositoryGraphAdjacencyContinuation> = None;
        loop {
            if Instant::now() >= deadline {
                work.complete = false;
                work.deadline_reached = true;
                break;
            }
            check_control(control)?;
            let remaining = query.relations.budget.edges().saturating_sub(
                relation_work
                    .inspected_edges
                    .saturating_add(work.inspected_edges),
            );
            if remaining == 0 {
                work.complete = false;
                break;
            }
            let page_limit = remaining.min(query.relations.budget.page_rows()).max(1);
            let decoded_remaining = query
                .relations
                .budget
                .intermediate_bytes()
                .saturating_sub(
                    relation_work
                        .intermediate_bytes
                        .saturating_add(work.decoded_bytes),
                )
                .min(RepositoryGraphReadBudget::MAX_DECODED_BYTES);
            if decoded_remaining == 0 {
                work.complete = false;
                break;
            }
            let endpoints = page_limit.saturating_add(1).saturating_mul(2);
            let budget = RepositoryGraphReadBudget::new(
                u32::try_from(chunk.len()).map_err(|_overflow| {
                    ServiceError::InvalidInput("analysis frontier overflowed".to_string())
                })?,
                page_limit,
                decoded_remaining,
                endpoints,
                endpoints,
            )
            .map_err(|error| ServiceError::InvalidInput(error.to_string()))?;
            let read = store.repository_graph_adjacency_page_filtered_bounded_with_documents(
                chunk,
                RepositoryGraphDirection::Outbound,
                query.relations.relation,
                super::relations::include_document_relations(&query.relations),
                continuation.as_ref(),
                page_limit,
                budget,
                control,
            )?;
            work.inspected_edges = work
                .inspected_edges
                .checked_add(u32::try_from(read.page.rows.len()).map_err(|_overflow| {
                    ServiceError::InvalidInput("analysis edge work overflowed".to_string())
                })?)
                .ok_or_else(|| {
                    ServiceError::InvalidInput("analysis edge work overflowed".to_string())
                })?;
            work.decoded_bytes = work.decoded_bytes.saturating_add(read.work.decoded_bytes);
            for row in read.page.rows {
                check_control(control)?;
                if !analysis_relation_matches(&row.detail.relation, &query.relations) {
                    continue;
                }
                let community_relation =
                    community_relation_weight(row.detail.relation.kind()).is_some();
                let source_key = row.detail.source.key().canonical_identity().to_string();
                if matches!(
                    row.detail.relation.resolution(),
                    RelationResolution::Ambiguous { .. } | RelationResolution::Unresolved { .. }
                ) {
                    work.induced_scope_closed = false;
                    if community_relation {
                        work.community_scope = CommunityClosureScope::Open;
                    }
                    if let Some(source) = nodes.get(&source_key) {
                        work.resolution_gaps
                            .push(resolution_gap_finding(&row.detail.relation, source));
                    }
                    continue;
                }
                let Some(target) = row.detail.target else {
                    work.induced_scope_closed = false;
                    if community_relation {
                        work.community_scope = CommunityClosureScope::Open;
                    }
                    continue;
                };
                let target_key = target.key().canonical_identity().to_string();
                if known.contains(&source_key) && known.contains(&target_key) {
                    edges.push(LocalEdge {
                        source: source_key,
                        target: target_key,
                        kind: row.detail.relation.kind(),
                        complete: row.detail.relation.completeness() == Completeness::Complete,
                    });
                } else {
                    work.induced_scope_closed = false;
                    if community_relation {
                        work.community_scope = CommunityClosureScope::Open;
                    }
                }
            }
            if read.page.truncated {
                continuation = read.page.continuation;
                if continuation.is_none() {
                    return Err(ServiceError::InvalidInput(
                        "truncated analysis closure omitted its continuation".to_string(),
                    ));
                }
            } else {
                break;
            }
        }
        if !work.complete {
            break;
        }
    }
    edges.sort_by(|left, right| {
        (&left.source, &left.target, format!("{:?}", left.kind)).cmp(&(
            &right.source,
            &right.target,
            format!("{:?}", right.kind),
        ))
    });
    edges.dedup_by(|left, right| {
        left.source == right.source && left.target == right.target && left.kind == right.kind
    });
    work.resolution_gaps
        .sort_by(|left, right| resolution_gap_identity(left).cmp(resolution_gap_identity(right)));
    work.resolution_gaps
        .dedup_by(|left, right| resolution_gap_identity(left) == resolution_gap_identity(right));
    Ok(work)
}

/// Determine whether topology findings have complete admitted evidence.
fn relation_evidence_complete(
    report: &DetailedRelationReport,
    nodes: &BTreeMap<String, DetailedRelationNode>,
    edges: &[LocalEdge],
    query: &RelationAnalysisQuery,
    closure: &ClosureWork,
    community_only: bool,
) -> bool {
    closure.complete
        && if community_only {
            closure.community_scope == CommunityClosureScope::Closed
        } else {
            closure.induced_scope_closed
        }
        && report.reached_limits.is_empty()
        && query.relations.direction == RelationDirection::Outbound
        && edges
            .iter()
            .filter(|edge| !community_only || community_relation_weight(edge.kind).is_some())
            .all(|edge| edge.complete)
        && nodes.values().all(|node| match node.entity.selector() {
            EntitySelector::Project => true,
            EntitySelector::Folder { .. }
            | EntitySelector::File { .. }
            | EntitySelector::Package { .. }
            | EntitySelector::Symbol { .. } => {
                !node.coverage.is_empty()
                    && node.coverage.iter().all(|coverage| {
                        coverage_trust(coverage.state()) == CoverageTrustState::Trusted
                    })
            }
            EntitySelector::External { .. } => false,
        })
}

/// Determine whether the exact inbound scope can support dead-code candidates.
fn dead_code_scope_complete(
    report: &DetailedRelationReport,
    query: &RelationAnalysisQuery,
) -> bool {
    let exact_symbol_anchor = matches!(
        &query.relations.anchor,
        RelationAnchor::Symbol {
            symbol_kind: Some(_),
            signature: Some(_),
            ..
        }
    );
    let exact_total = matches!(
        report.total,
        RelationTotalState::Exact(total) if total == u64::from(report.returned)
    );
    query.relations.direction == RelationDirection::Inbound
        && query.relations.relation.is_none()
        && query.relations.resolution == RelationResolutionFilter::Resolved
        && query.relations.minimum_confidence == ConfidenceClass::Low
        && exact_symbol_anchor
        && exact_total
        && !report.truncated
        && report.continuation.is_none()
        && report.reached_limits.is_empty()
        && !report.anchor.coverage.is_empty()
        && report
            .anchor
            .coverage
            .iter()
            .all(|coverage| coverage_trust(coverage.state()) == CoverageTrustState::Trusted)
        && report.rows.iter().all(|row| {
            matches!(
                row.relation.resolution(),
                RelationResolution::Resolved { .. }
            ) && row.relation.completeness() == Completeness::Complete
        })
}

/// Apply detailed relation family, trust, and resolution filters.
fn analysis_relation_matches(
    relation: &projectatlas_core::graph::LogicalRelation,
    query: &DetailedRelationQuery,
) -> bool {
    super::relations::relation_matches(relation, query)
}

/// Compute bounded architecture findings over admitted topology and symbols.
fn architecture_findings(
    store: &AtlasStore,
    nodes: &BTreeMap<String, DetailedRelationNode>,
    edges: &[LocalEdge],
    complete: bool,
    community_complete: bool,
    query: &RelationAnalysisQuery,
    symbol_byte_budget: u64,
    preceding_finding_count: usize,
    supplemental_work: &mut SupplementalWork,
    control: Option<&IndexWorkControl>,
) -> ServiceResult<Vec<AnalysisFinding>> {
    let mut findings = structural_findings(
        store,
        nodes,
        edges,
        complete,
        symbol_byte_budget,
        supplemental_work,
        control,
    )?;
    if !complete {
        findings.push(AnalysisFinding {
            kind: AnalysisFindingKind::Component,
            status: AnalysisStatus::Inconclusive,
            summary: "architecture candidates are incomplete because traversal or local coverage is partial"
                .to_string(),
            nodes: analysis_nodes_for(nodes, &nodes.keys().cloned().collect::<Vec<_>>()),
            metric: Some(nodes.len() as u64),
            evidence: None,
            community: None,
        });
    }
    let components = weak_components(nodes, edges, false);
    for component in &components {
        findings.push(AnalysisFinding {
            kind: AnalysisFindingKind::Component,
            status: AnalysisStatus::Candidate,
            summary: "component candidate from a weakly connected admitted relation set"
                .to_string(),
            nodes: analysis_nodes_for(nodes, component),
            metric: Some(component.len() as u64),
            evidence: None,
            community: None,
        });
        findings.push(purpose_finding(nodes, component, complete));
    }
    if query.include_communities {
        let mut existing_finding_bytes = serialized_bytes_controlled(&findings, control)?;
        let mut existing_finding_append_bytes = serialized_findings_append_bytes(
            existing_finding_bytes,
            findings.len(),
            preceding_finding_count,
        );
        if existing_finding_append_bytes > symbol_byte_budget {
            supplemental_work.composition_truncated = true;
            push_limit(
                &mut supplemental_work.reached_limits,
                GraphLimitKind::IntermediateBytes,
            );
            while existing_finding_append_bytes > symbol_byte_budget && findings.len() > 1 {
                check_control(control)?;
                findings.pop();
                existing_finding_bytes = serialized_bytes_controlled(&findings, control)?;
                existing_finding_append_bytes = serialized_findings_append_bytes(
                    existing_finding_bytes,
                    findings.len(),
                    preceding_finding_count,
                );
            }
            if existing_finding_append_bytes > symbol_byte_budget {
                findings.clear();
                existing_finding_bytes = serialized_bytes_controlled(&findings, control)?;
                existing_finding_append_bytes = serialized_findings_append_bytes(
                    existing_finding_bytes,
                    findings.len(),
                    preceding_finding_count,
                );
            }
        }
        let community_budget = symbol_byte_budget.saturating_sub(existing_finding_append_bytes);
        #[cfg(test)]
        analysis_test_observer::notify(
            analysis_test_observer::AnalysisPhaseEvent::CompositionBudget {
                symbol_byte_budget,
                existing_finding_append_bytes,
                community_budget,
            },
        );
        let existing_community_finding_count =
            preceding_finding_count.saturating_add(findings.len());
        let (community_findings, community_working_set_bytes, community_limits) =
            community_findings_with_budget(
                nodes,
                edges,
                community_complete,
                query,
                community_budget,
                existing_community_finding_count,
                control,
            )?;
        supplemental_work.community_working_set_bytes = supplemental_work
            .community_working_set_bytes
            .max(community_working_set_bytes);
        let community_truncated = !community_limits.is_empty();
        for limit in community_limits {
            push_limit(&mut supplemental_work.reached_limits, limit);
        }
        if community_truncated
            || community_findings.iter().any(|finding| {
                finding
                    .community
                    .as_ref()
                    .is_some_and(|community| community.truncated)
            })
        {
            supplemental_work.composition_truncated = true;
        }
        findings.extend(community_findings);
    }
    if query.include_cycles {
        let dependency_edges = edges
            .iter()
            .filter(|edge| dependency_relation(edge.kind))
            .cloned()
            .collect::<Vec<_>>();
        let cycles = strongly_connected_components(nodes, &dependency_edges)
            .into_iter()
            .filter(|component| {
                component.len() > 1
                    || dependency_edges.iter().any(|edge| {
                        component.first() == Some(&edge.source) && edge.source == edge.target
                    })
            })
            .collect::<Vec<_>>();
        if cycles.is_empty() {
            findings.push(AnalysisFinding {
                kind: AnalysisFindingKind::DependencyCycle,
                status: if complete {
                    AnalysisStatus::Absent
                } else {
                    AnalysisStatus::Inconclusive
                },
                summary: if complete {
                    "no dependency cycle exists in the complete admitted bounded scope"
                } else {
                    "no cycle was observed, but traversal or coverage is incomplete"
                }
                .to_string(),
                nodes: Vec::new(),
                metric: Some(0),
                evidence: None,
                community: None,
            });
        } else {
            for cycle in cycles {
                findings.push(AnalysisFinding {
                    kind: AnalysisFindingKind::DependencyCycle,
                    status: AnalysisStatus::Candidate,
                    summary: "iterative dependency-family SCC found a static cycle candidate"
                        .to_string(),
                    nodes: analysis_nodes_for(nodes, &cycle),
                    metric: Some(cycle.len() as u64),
                    evidence: None,
                    community: None,
                });
            }
        }
    }
    Ok(findings)
}

/// Return whether a relation participates in dependency SCC and impact flow.
fn dependency_relation(kind: GraphRelationKind) -> bool {
    matches!(
        kind,
        GraphRelationKind::Legacy(
            RelationKind::Imports | RelationKind::Calls | RelationKind::DependsOn
        ) | GraphRelationKind::Extended(
            ExtendedRelationKind::Tests
                | ExtendedRelationKind::RoutesTo
                | ExtendedRelationKind::Configures
        )
    )
}

/// Classify purpose alignment for one connected component.
fn purpose_finding(
    nodes: &BTreeMap<String, DetailedRelationNode>,
    component: &[String],
    complete: bool,
) -> AnalysisFinding {
    let mut purposes = BTreeSet::new();
    let mut unavailable = false;
    for key in component {
        match nodes.get(key).map(|node| &node.purpose) {
            Some(RelationPurpose::Approved { purpose, .. }) => {
                purposes.insert(purpose.clone());
            }
            Some(RelationPurpose::Unavailable { .. } | RelationPurpose::NotApplicable) | None => {
                unavailable = true;
            }
        }
    }
    let (kind, status, summary) = if purposes.len() > 1 {
        (
            AnalysisFindingKind::PurposeDrift,
            AnalysisStatus::Candidate,
            "connected owners retain multiple approved purpose responsibilities",
        )
    } else if purposes.len() == 1 && !unavailable && complete {
        (
            AnalysisFindingKind::PurposeAlignment,
            AnalysisStatus::Confirmed,
            "connected owners share one approved purpose responsibility",
        )
    } else {
        (
            AnalysisFindingKind::PurposeAlignment,
            AnalysisStatus::Inconclusive,
            if complete {
                "purpose alignment is unavailable for at least one admitted owner"
            } else {
                "purpose alignment is inconclusive under partial traversal or coverage"
            },
        )
    };
    AnalysisFinding {
        kind,
        status,
        summary: summary.to_string(),
        nodes: analysis_nodes_for(nodes, component),
        metric: Some(purposes.len() as u64),
        evidence: None,
        community: None,
    }
}

/// Compute declaration-span and graph-degree structural candidates.
fn structural_findings(
    store: &AtlasStore,
    nodes: &BTreeMap<String, DetailedRelationNode>,
    edges: &[LocalEdge],
    topology_complete: bool,
    symbol_byte_budget: u64,
    supplemental_work: &mut SupplementalWork,
    control: Option<&IndexWorkControl>,
) -> ServiceResult<Vec<AnalysisFinding>> {
    let degrees = degrees(nodes, edges);
    let symbols_by_file = load_admitted_symbols(store, nodes, symbol_byte_budget, control)?;
    supplemental_work.hydrated_symbols = supplemental_work
        .hydrated_symbols
        .saturating_add(symbols_by_file.rows_retained);
    supplemental_work.hydrated_symbol_bytes = supplemental_work
        .hydrated_symbol_bytes
        .saturating_add(symbols_by_file.retained_bytes);
    supplemental_work.hydrated_symbol_peak_bytes = supplemental_work
        .hydrated_symbol_peak_bytes
        .max(symbols_by_file.peak_bytes);
    supplemental_work.symbol_hydration_truncated |= !symbols_by_file.complete;
    supplemental_work
        .reached_limits
        .extend(symbols_by_file.reached_limits.iter().copied());
    let mut candidates = Vec::new();
    for (key, node) in nodes {
        check_control(control)?;
        let Some((path, name, kind, parent, signature)) = symbol_identity(&node.entity) else {
            continue;
        };
        if let Some(symbol) = symbols_by_file.rows_for_path(path).and_then(|symbols| {
            symbols.iter().find(|candidate| {
                candidate.name == name
                    && candidate.kind == kind
                    && candidate.parent.as_deref() == parent
                    && candidate.signature == signature
            })
        }) {
            let span = symbol
                .line_end
                .saturating_sub(symbol.line_start)
                .saturating_add(1) as u64;
            let degree = degrees.get(key).copied().unwrap_or_default() as u64;
            candidates.push((span, degree, key.clone()));
        }
    }
    let symbols_complete = symbols_by_file.complete;
    drop(symbols_by_file);
    let mut findings = Vec::new();
    if let Some((span, _degree, key)) = candidates.iter().max_by(std::cmp::Ord::cmp) {
        findings.push(AnalysisFinding {
            kind: AnalysisFindingKind::StructuralComplexity,
            status: if symbols_complete && topology_complete {
                AnalysisStatus::Candidate
            } else {
                AnalysisStatus::Inconclusive
            },
            summary: if symbols_complete && topology_complete {
                "largest language-valid declaration span in the admitted scope; not cyclomatic complexity"
            } else {
                "structural candidate observed, but bounded symbol hydration omitted admitted declarations"
            }
            .to_string(),
            nodes: analysis_nodes_for(nodes, std::slice::from_ref(key)),
            metric: Some(*span),
            evidence: None,
            community: None,
        });
    }
    if let Some((_span, degree, key)) = candidates
        .iter()
        .max_by(|left, right| (left.1, left.0, &left.2).cmp(&(right.1, right.0, &right.2)))
    {
        findings.push(AnalysisFinding {
            kind: AnalysisFindingKind::Bottleneck,
            status: if symbols_complete && topology_complete {
                AnalysisStatus::Candidate
            } else {
                AnalysisStatus::Inconclusive
            },
            summary: "highest admitted static fan-in plus fan-out junction".to_string(),
            nodes: analysis_nodes_for(nodes, std::slice::from_ref(key)),
            metric: Some(*degree),
            evidence: None,
            community: None,
        });
    }
    Ok(findings)
}

/// Bounded persisted symbols plus a compact exact-path range index.
struct AdmittedSymbols {
    /// Deterministically sorted symbols returned by the database.
    rows: Vec<CodeSymbol>,
    /// Sorted exact-path ranges into `rows`.
    ranges: Vec<SymbolPathRange>,
    /// Whether every admitted path, row, and byte fit the envelope.
    complete: bool,
    /// Number of persisted symbols retained.
    rows_retained: u32,
    /// Database-decoded rows plus serialized-equivalent range-index bytes.
    retained_bytes: u64,
    /// Retained bytes plus serialized-equivalent request-path auxiliaries.
    peak_bytes: u64,
    /// Stable public limits that omitted symbol candidates.
    reached_limits: Vec<GraphLimitKind>,
}

#[derive(Serialize)]
/// One sorted exact-path slice into the retained symbol vector.
struct SymbolPathRange {
    /// Exact repository-relative owning path.
    path: String,
    /// Inclusive symbol-vector offset.
    start: usize,
    /// Exclusive symbol-vector offset.
    end: usize,
}

impl AdmittedSymbols {
    /// Return the retained persisted symbols for one exact path.
    fn rows_for_path(&self, path: &str) -> Option<&[CodeSymbol]> {
        let index = self
            .ranges
            .binary_search_by(|range| range.path.as_str().cmp(path))
            .ok()?;
        let range = &self.ranges[index];
        self.rows.get(range.start..range.end)
    }
}

/// Load one bounded indexed symbol batch and build its compact path ranges.
fn load_admitted_symbols(
    store: &AtlasStore,
    nodes: &BTreeMap<String, DetailedRelationNode>,
    byte_budget: u64,
    control: Option<&IndexWorkControl>,
) -> ServiceResult<AdmittedSymbols> {
    let path_limit = usize::try_from(MAX_SYMBOL_BATCH_PATHS).map_err(|source| {
        ServiceError::InvalidInput(format!("symbol path ceiling overflowed: {source}"))
    })?;
    let mut paths = Vec::new();
    let mut path_truncated = false;
    for node in nodes.values() {
        check_control(control)?;
        let EntitySelector::Symbol { symbol } = node.entity.selector() else {
            continue;
        };
        let path = symbol.file.as_str();
        let Err(index) = paths.binary_search_by(|candidate: &String| candidate.as_str().cmp(path))
        else {
            continue;
        };
        paths.insert(index, path.to_string());
        if paths.len() > path_limit {
            paths.pop();
            path_truncated = true;
        }
    }
    let paths = paths.into_boxed_slice().into_vec();
    let byte_limit = byte_budget.min(MAX_SYMBOL_BATCH_DECODED_BYTES);
    let mut reached_limits = Vec::new();
    if path_truncated {
        push_limit(&mut reached_limits, GraphLimitKind::Rows);
    }
    if paths.is_empty() {
        return Ok(AdmittedSymbols {
            rows: Vec::new(),
            ranges: Vec::new(),
            complete: !path_truncated,
            rows_retained: 0,
            retained_bytes: 0,
            peak_bytes: symbol_path_request_bytes(&paths, control)?,
            reached_limits,
        });
    }
    #[cfg(test)]
    analysis_test_observer::notify(analysis_test_observer::AnalysisPhaseEvent::SymbolHydration);
    check_control(control)?;
    let grouping_reserve = symbol_hydration_reserve_bytes(&paths, control)?;
    let decoded_byte_limit = byte_limit.saturating_sub(grouping_reserve);
    if decoded_byte_limit == 0 {
        push_limit(&mut reached_limits, GraphLimitKind::IntermediateBytes);
        return Ok(AdmittedSymbols {
            rows: Vec::new(),
            ranges: Vec::new(),
            complete: false,
            rows_retained: 0,
            retained_bytes: 0,
            peak_bytes: symbol_path_request_bytes(&paths, control)?,
            reached_limits,
        });
    }
    let read = store.load_symbols_for_paths_bounded(
        &paths,
        SymbolBatchReadBudget::new(
            MAX_SYMBOL_BATCH_PATHS,
            MAX_SYMBOL_BATCH_ROWS,
            decoded_byte_limit.min(MAX_SYMBOL_BATCH_DECODED_BYTES),
        )?,
        control,
    )?;
    match read.reached_limit {
        Some(SymbolBatchReadLimit::Paths | SymbolBatchReadLimit::Rows) => {
            push_limit(&mut reached_limits, GraphLimitKind::Rows);
        }
        Some(SymbolBatchReadLimit::DecodedBytes) => {
            push_limit(&mut reached_limits, GraphLimitKind::IntermediateBytes);
        }
        None => {}
    }
    let ranges = symbol_path_ranges(&read.rows, control)?;
    let retained_bytes = read
        .work
        .decoded_bytes
        .saturating_add(symbol_range_index_bytes(&ranges, control)?);
    let peak_bytes = retained_bytes.saturating_add(symbol_path_request_bytes(&paths, control)?);
    Ok(AdmittedSymbols {
        rows: read.rows,
        ranges,
        complete: !path_truncated && !read.truncated,
        rows_retained: read.work.returned_rows,
        retained_bytes,
        peak_bytes,
        reached_limits,
    })
}

/// Build sorted non-overlapping exact-path ranges over sorted symbol rows.
fn symbol_path_ranges(
    rows: &[CodeSymbol],
    control: Option<&IndexWorkControl>,
) -> ServiceResult<Vec<SymbolPathRange>> {
    let mut ranges = Vec::new();
    let mut start = 0;
    while start < rows.len() {
        check_control(control)?;
        let mut end = start.saturating_add(1);
        while end < rows.len() && rows[end].path == rows[start].path {
            check_control(control)?;
            end = end.saturating_add(1);
        }
        ranges.push(SymbolPathRange {
            path: rows[start].path.clone(),
            start,
            end,
        });
        start = end;
    }
    Ok(ranges.into_boxed_slice().into_vec())
}

/// Count serialized request-path and worst-case path-range auxiliaries.
fn symbol_hydration_reserve_bytes(
    paths: &[String],
    control: Option<&IndexWorkControl>,
) -> ServiceResult<u64> {
    let mut ranges = Vec::with_capacity(paths.len());
    for path in paths {
        check_control(control)?;
        ranges.push(SymbolPathRange {
            path: path.clone(),
            start: 0,
            end: 0,
        });
    }
    Ok(symbol_path_request_bytes(paths, control)?
        .saturating_add(serialized_bytes_controlled(&ranges, control)?))
}

/// Count serialized-equivalent request-path bytes.
fn symbol_path_request_bytes(
    paths: &[String],
    control: Option<&IndexWorkControl>,
) -> ServiceResult<u64> {
    serialized_bytes_controlled(paths, control)
}

/// Count serialized-equivalent retained path-range bytes.
fn symbol_range_index_bytes(
    ranges: &[SymbolPathRange],
    control: Option<&IndexWorkControl>,
) -> ServiceResult<u64> {
    serialized_bytes_controlled(ranges, control)
}

/// Allocation-free serialized-equivalent byte counter.
struct SerializedByteCounter<'a> {
    /// Bytes that the serializer would write.
    bytes: u64,
    /// Shared request control checked between serializer writes.
    control: Option<&'a IndexWorkControl>,
    /// Whether serialization stopped on cancellation or deadline.
    interrupted: bool,
}

impl Write for SerializedByteCounter<'_> {
    fn write(&mut self, buffer: &[u8]) -> io::Result<usize> {
        if self
            .control
            .is_some_and(|control| control.check(IndexWorkStage::RepositoryTraversal).is_err())
        {
            self.interrupted = true;
            // `Write::write_all` retries `Interrupted` indefinitely. Use a
            // non-retriable transport error and translate it back to the
            // control's typed terminal failure after serialization returns.
            return Err(io::Error::other("analysis serialization interrupted"));
        }
        self.bytes = self
            .bytes
            .checked_add(u64::try_from(buffer.len()).unwrap_or(u64::MAX))
            .ok_or_else(|| io::Error::other("analysis serialized byte count overflowed"))?;
        Ok(buffer.len())
    }

    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}

/// Measure one bounded value while retaining the request deadline and cancellation.
fn serialized_bytes_controlled<T: Serialize + ?Sized>(
    value: &T,
    control: Option<&IndexWorkControl>,
) -> ServiceResult<u64> {
    let mut counter = SerializedByteCounter {
        bytes: 0,
        control,
        interrupted: false,
    };
    let result = serde_json::to_writer(&mut counter, value);
    if counter.interrupted {
        check_control(control)?;
    }
    result?;
    check_control(control)?;
    Ok(counter.bytes)
}

/// Charge one finding as an append to an already retained JSON findings array.
///
/// The enclosing array's `[]` framing is charged by the composition owner. An
/// append therefore adds only the finding bytes and, after the first finding,
/// the fixed comma separator.
const fn serialized_finding_append_bytes(
    finding_bytes: u64,
    preceding_finding_count: usize,
) -> u64 {
    finding_bytes.saturating_add(if preceding_finding_count == 0 {
        0
    } else {
        JSON_ARRAY_SEPARATOR_BYTES
    })
}

/// Charge a serialized findings vector appended to an existing JSON array.
const fn serialized_findings_append_bytes(
    findings_bytes: u64,
    finding_count: usize,
    preceding_finding_count: usize,
) -> u64 {
    if finding_count == 0 {
        return 0;
    }
    findings_bytes.saturating_sub(if preceding_finding_count == 0 {
        JSON_ARRAY_FRAMING_BYTES
    } else {
        JSON_ARRAY_SEPARATOR_BYTES
    })
}

/// Find one node-simple static relation path to an exact target.
fn trace_findings(
    report: &DetailedRelationReport,
    target: Option<&RelationAnchor>,
    evidence_complete: bool,
) -> ServiceResult<Vec<AnalysisFinding>> {
    let target = target.ok_or_else(|| {
        ServiceError::InvalidInput("analysis trace requires an exact target".to_string())
    })?;
    let mut matching = BTreeSet::new();
    if entity_matches_anchor(&report.anchor.entity, target) {
        matching.insert(report.anchor.entity.key().canonical_identity().to_string());
    }
    for row in &report.rows {
        for node in &row.path {
            if entity_matches_anchor(&node.entity, target) {
                matching.insert(node.entity.key().canonical_identity().to_string());
            }
        }
    }
    if matching.len() > 1 {
        return Err(ServiceError::InvalidInput(
            "analysis trace target is ambiguous in the admitted graph scope".to_string(),
        ));
    }
    let target_key = matching.iter().next();
    if target_key.is_some_and(|key| key == report.anchor.entity.key().canonical_identity()) {
        return Ok(vec![AnalysisFinding {
            kind: AnalysisFindingKind::StaticTrace,
            status: AnalysisStatus::Confirmed,
            summary: "static trace target is the selected anchor".to_string(),
            nodes: vec![analysis_node(&report.anchor)],
            metric: Some(0),
            evidence: None,
            community: None,
        }]);
    }
    if let Some(row) = target_key.and_then(|target_key| {
        report.rows.iter().find(|row| {
            row.path
                .last()
                .is_some_and(|node| node.entity.key().canonical_identity() == target_key.as_str())
        })
    }) {
        return Ok(vec![AnalysisFinding {
            kind: AnalysisFindingKind::StaticTrace,
            status: AnalysisStatus::Confirmed,
            summary: "node-simple static relation path; not a runtime execution trace".to_string(),
            nodes: row.path.iter().map(analysis_node).collect(),
            metric: Some(u64::from(row.depth)),
            evidence: None,
            community: None,
        }]);
    }
    Ok(vec![AnalysisFinding {
        kind: AnalysisFindingKind::StaticTrace,
        status: if evidence_complete {
            AnalysisStatus::Absent
        } else {
            AnalysisStatus::Inconclusive
        },
        summary: if evidence_complete {
            "target is not reachable in the complete admitted static scope"
        } else {
            "target was not observed before the bounded traversal or coverage stopped"
        }
        .to_string(),
        nodes: Vec::new(),
        metric: None,
        evidence: None,
        community: None,
    }])
}

/// Return the fixed relation-family weights admitted by community v1.
fn community_relation_weights() -> Vec<CommunityRelationWeight> {
    GraphRelationKind::ALL
        .into_iter()
        .filter_map(|relation| {
            community_relation_weight(relation)
                .map(|weight| CommunityRelationWeight { relation, weight })
        })
        .collect()
}

/// Return the fixed vote weight for an admitted relation family.
fn community_relation_weight(kind: GraphRelationKind) -> Option<u32> {
    match kind {
        GraphRelationKind::Legacy(RelationKind::Calls) => Some(8),
        GraphRelationKind::Legacy(RelationKind::Imports) => Some(5),
        GraphRelationKind::Legacy(RelationKind::DependsOn) => Some(3),
        GraphRelationKind::Extended(
            ExtendedRelationKind::Tests | ExtendedRelationKind::RoutesTo,
        ) => Some(4),
        GraphRelationKind::Extended(
            ExtendedRelationKind::References
            | ExtendedRelationKind::Configures
            | ExtendedRelationKind::Reads
            | ExtendedRelationKind::Writes,
        ) => Some(2),
        GraphRelationKind::Extended(
            ExtendedRelationKind::Documents | ExtendedRelationKind::Deploys,
        ) => Some(1),
        GraphRelationKind::Legacy(RelationKind::Contains) => None,
    }
}

/// Return the weight for one complete relation admitted by the community projection.
fn admitted_community_edge_weight(edge: &LocalEdge) -> Option<u32> {
    if !edge.complete {
        return None;
    }
    community_relation_weight(edge.kind)
}

/// Compute deterministic bounded weighted label-propagation communities.
#[cfg(test)]
fn community_findings(
    nodes: &BTreeMap<String, DetailedRelationNode>,
    edges: &[LocalEdge],
    complete: bool,
    query: &RelationAnalysisQuery,
    control: Option<&IndexWorkControl>,
) -> ServiceResult<Vec<AnalysisFinding>> {
    Ok(community_findings_with_budget(
        nodes,
        edges,
        complete,
        query,
        query.relations.budget.intermediate_bytes(),
        0,
        control,
    )?
    .0)
}

/// Compute communities with an explicit remaining intermediate-byte allowance.
fn community_findings_with_budget(
    nodes: &BTreeMap<String, DetailedRelationNode>,
    edges: &[LocalEdge],
    complete: bool,
    query: &RelationAnalysisQuery,
    intermediate_bytes: u64,
    preceding_finding_count: usize,
    control: Option<&IndexWorkControl>,
) -> ServiceResult<(Vec<AnalysisFinding>, u64, Vec<GraphLimitKind>)> {
    check_control(control)?;
    let weights = community_relation_weights();
    let parameters = CommunityParameters {
        algorithm_version: COMMUNITY_ALGORITHM_VERSION,
        ordering_version: COMMUNITY_ORDERING_VERSION,
        max_iterations: COMMUNITY_MAX_ITERATIONS,
        node_limit: query.relations.budget.nodes().min(MAX_ANALYSIS_NODES),
        edge_limit: query.relations.budget.edges().min(MAX_ANALYSIS_EDGES),
        output_bytes: query.relations.budget.output_bytes(),
        relation: query.relations.relation,
    };
    let complete = complete
        && edges
            .iter()
            .filter(|edge| community_relation_weight(edge.kind).is_some())
            .all(|edge| edge.complete);
    let marker_coverage = if complete {
        CommunityCoverage::Complete
    } else {
        CommunityCoverage::Partial
    };
    let truncation_marker = community_truncation_finding(&weights, parameters, 0, marker_coverage);
    let truncation_marker_bytes = serialized_bytes_controlled(&truncation_marker, control)?;
    let truncation_marker_append_bytes =
        serialized_finding_append_bytes(truncation_marker_bytes, preceding_finding_count);
    let working_set_bytes = community_working_set_upper_bound(nodes, edges, control)?;
    if working_set_bytes.saturating_add(truncation_marker_append_bytes) > intermediate_bytes {
        let findings = if truncation_marker_append_bytes <= intermediate_bytes {
            vec![truncation_marker]
        } else {
            Vec::new()
        };
        return Ok((findings, 0, vec![GraphLimitKind::IntermediateBytes]));
    }
    let construction_bytes = intermediate_bytes.saturating_sub(working_set_bytes);
    let (keys, admitted_edges, resource_limits) =
        admitted_community_scope(nodes, edges, parameters);
    let resource_truncated = !resource_limits.is_empty();
    if !complete || resource_truncated {
        let mut evidence = admitted_edges;
        if resource_truncated {
            evidence.truncate(parameters.edge_limit as usize);
        }
        let metadata_fits = serialized_finding_append_bytes(
            community_finding_upper_bound(nodes, &keys, &evidence, &weights, parameters, control)?,
            preceding_finding_count,
        ) <= construction_bytes;
        let metadata_keys = if metadata_fits { keys.as_slice() } else { &[] };
        let metadata_evidence = if metadata_fits {
            evidence.as_slice()
        } else {
            &[]
        };
        let metadata_truncated = resource_truncated || !metadata_fits;
        let mut truncation_limits = resource_limits;
        if !metadata_fits {
            push_limit(&mut truncation_limits, GraphLimitKind::IntermediateBytes);
        }
        let metadata = community_metadata(
            metadata_keys,
            metadata_evidence,
            weights,
            parameters,
            0,
            CommunityConvergence::Inconclusive,
            if complete {
                CommunityCoverage::Complete
            } else {
                CommunityCoverage::Partial
            },
            metadata_truncated,
            nodes,
        );
        return Ok((
            vec![AnalysisFinding {
                kind: AnalysisFindingKind::Community,
                status: AnalysisStatus::Inconclusive,
                summary: if resource_truncated {
                    "community projection crossed its fixed node or edge resource ceiling"
                } else {
                    "community projection is inconclusive because relation coverage is partial"
                }
                .to_string(),
                nodes: analysis_nodes_for(nodes, metadata_keys),
                metric: Some(metadata_keys.len() as u64),
                evidence: None,
                community: Some(metadata),
            }],
            working_set_bytes,
            truncation_limits,
        ));
    }

    let (labels, iteration, convergence) =
        propagate_community_labels(&keys, &admitted_edges, parameters.max_iterations, control)?;
    if convergence != CommunityConvergence::Converged {
        let metadata_fits = community_finding_upper_bound(
            nodes,
            &keys,
            &admitted_edges,
            &weights,
            parameters,
            control,
        )?;
        let metadata_fits = serialized_finding_append_bytes(metadata_fits, preceding_finding_count)
            <= construction_bytes;
        let metadata_keys = if metadata_fits { keys.as_slice() } else { &[] };
        let metadata_evidence = if metadata_fits {
            admitted_edges.as_slice()
        } else {
            &[]
        };
        let metadata_truncated = !metadata_fits;
        let mut truncation_limits = Vec::new();
        if !metadata_fits {
            truncation_limits.push(GraphLimitKind::IntermediateBytes);
        }
        let metadata = community_metadata(
            metadata_keys,
            metadata_evidence,
            weights,
            parameters,
            iteration,
            convergence,
            CommunityCoverage::Complete,
            metadata_truncated,
            nodes,
        );
        return Ok((
            vec![AnalysisFinding {
                kind: AnalysisFindingKind::Community,
                status: AnalysisStatus::Inconclusive,
                summary: "community label propagation reached its fixed iteration ceiling"
                    .to_string(),
                nodes: analysis_nodes_for(nodes, metadata_keys),
                metric: Some(metadata_keys.len() as u64),
                evidence: None,
                community: Some(metadata),
            }],
            working_set_bytes,
            truncation_limits,
        ));
    }

    let (findings, truncation_limits) = community_candidate_findings(
        nodes,
        &keys,
        &admitted_edges,
        &labels,
        &weights,
        parameters,
        iteration,
        convergence,
        control,
        preceding_finding_count,
        construction_bytes,
    )?;
    Ok((findings, working_set_bytes, truncation_limits))
}

/// Admit the stable node and edge prefix under the community resource ceiling.
fn admitted_community_scope(
    nodes: &BTreeMap<String, DetailedRelationNode>,
    edges: &[LocalEdge],
    parameters: CommunityParameters,
) -> (Vec<String>, Vec<CommunityEdgeEvidence>, Vec<GraphLimitKind>) {
    let all_keys = nodes.keys().cloned().collect::<Vec<_>>();
    let keys = all_keys
        .iter()
        .take(parameters.node_limit as usize)
        .cloned()
        .collect::<Vec<_>>();
    let admitted_nodes = keys.iter().collect::<BTreeSet<_>>();
    let mut admitted_edges = Vec::new();
    let mut edge_count = 0_usize;
    for edge in edges {
        let Some(weight) = admitted_community_edge_weight(edge) else {
            continue;
        };
        if !admitted_nodes.contains(&edge.source) || !admitted_nodes.contains(&edge.target) {
            continue;
        }
        edge_count = edge_count.saturating_add(1);
        if edge_count <= parameters.edge_limit as usize {
            admitted_edges.push(CommunityEdgeEvidence {
                source: edge.source.clone(),
                target: edge.target.clone(),
                relation: edge.kind,
                weight,
            });
        }
    }
    admitted_edges.sort_by(|left, right| {
        (&left.source, &left.target, left.relation.as_str()).cmp(&(
            &right.source,
            &right.target,
            right.relation.as_str(),
        ))
    });
    admitted_edges.dedup_by(|left, right| {
        left.source == right.source
            && left.target == right.target
            && left.relation == right.relation
    });
    let mut reached_limits = Vec::new();
    if all_keys.len() > parameters.node_limit as usize {
        reached_limits.push(GraphLimitKind::Nodes);
    }
    if edge_count > parameters.edge_limit as usize {
        reached_limits.push(GraphLimitKind::Edges);
    }
    (keys, admitted_edges, reached_limits)
}

/// Propagate labels in stable node order with a fixed sequential work budget.
fn propagate_community_labels(
    keys: &[String],
    edges: &[CommunityEdgeEvidence],
    max_iterations: u32,
    control: Option<&IndexWorkControl>,
) -> ServiceResult<(BTreeMap<String, String>, u32, CommunityConvergence)> {
    check_control(control)?;
    let mut adjacency = keys
        .iter()
        .map(|key| (key.clone(), Vec::<(String, u32)>::new()))
        .collect::<BTreeMap<_, _>>();
    for edge in edges {
        check_control(control)?;
        adjacency
            .entry(edge.source.clone())
            .or_default()
            .push((edge.target.clone(), edge.weight));
        adjacency
            .entry(edge.target.clone())
            .or_default()
            .push((edge.source.clone(), edge.weight));
    }
    for neighbors in adjacency.values_mut() {
        neighbors.sort();
    }

    // ponytail: one bounded sequential pass is deterministic and sufficient at
    // the 512-node ceiling; replace only after a representative profile proves
    // that parallel label updates repay their extra synchronization.
    let mut labels = keys
        .iter()
        .map(|key| (key.clone(), key.clone()))
        .collect::<BTreeMap<_, _>>();
    let mut converged = keys.is_empty();
    let mut iteration = 0;
    while !converged && iteration < max_iterations {
        check_control(control)?;
        iteration += 1;
        let mut changed = false;
        for key in keys {
            check_control(control)?;
            let mut scores = BTreeMap::<String, u32>::new();
            scores.insert(key.clone(), COMMUNITY_SELF_WEIGHT);
            if let Some(neighbors) = adjacency.get(key) {
                for (neighbor, weight) in neighbors {
                    check_control(control)?;
                    if let Some(label) = labels.get(neighbor) {
                        let score = scores.entry(label.clone()).or_default();
                        *score = score.saturating_add(*weight);
                    }
                }
            }
            let selected = select_community_label(key, scores);
            if labels.get(key) != Some(&selected) {
                labels.insert(key.clone(), selected);
                changed = true;
            }
        }
        converged = !changed;
    }

    Ok((
        labels,
        iteration,
        if converged {
            CommunityConvergence::Converged
        } else {
            CommunityConvergence::IterationLimit
        },
    ))
}

/// Select the highest-scoring label, breaking equal scores by stable key order.
fn select_community_label(key: &str, scores: BTreeMap<String, u32>) -> String {
    let mut selected = key.to_string();
    let mut selected_score = 0;
    for (label, score) in scores {
        if score > selected_score || (score == selected_score && label < selected) {
            selected = label;
            selected_score = score;
        }
    }
    selected
}

/// Convert stable labels into candidate findings with induced evidence.
fn community_candidate_findings(
    nodes: &BTreeMap<String, DetailedRelationNode>,
    keys: &[String],
    admitted_edges: &[CommunityEdgeEvidence],
    labels: &BTreeMap<String, String>,
    weights: &[CommunityRelationWeight],
    parameters: CommunityParameters,
    iteration: u32,
    convergence: CommunityConvergence,
    control: Option<&IndexWorkControl>,
    preceding_finding_count: usize,
    intermediate_bytes: u64,
) -> ServiceResult<(Vec<AnalysisFinding>, Vec<GraphLimitKind>)> {
    let mut groups = BTreeMap::<String, Vec<String>>::new();
    for key in keys {
        check_control(control)?;
        let label = labels.get(key).cloned().unwrap_or_else(|| key.clone());
        groups.entry(label).or_default().push(key.clone());
    }
    let mut findings = Vec::new();
    let mut retained_append_bytes = 0_u64;
    let mut truncated = false;
    for members in groups.values() {
        check_control(control)?;
        let candidate_bound = community_candidate_upper_bound(
            nodes,
            members,
            admitted_edges,
            weights,
            parameters,
            control,
        )?;
        let candidate_append_bytes = serialized_finding_append_bytes(
            candidate_bound,
            preceding_finding_count.saturating_add(findings.len()),
        );
        if retained_append_bytes.saturating_add(candidate_append_bytes) > intermediate_bytes {
            truncated = true;
            break;
        }
        let evidence = admitted_edges
            .iter()
            .filter(|edge| {
                members.binary_search(&edge.source).is_ok()
                    && members.binary_search(&edge.target).is_ok()
            })
            .cloned()
            .collect::<Vec<_>>();
        let metadata = community_metadata(
            members,
            &evidence,
            weights.to_vec(),
            parameters,
            iteration,
            convergence,
            CommunityCoverage::Complete,
            false,
            nodes,
        );
        let finding = AnalysisFinding {
            kind: AnalysisFindingKind::Community,
            status: AnalysisStatus::Candidate,
            summary: if members.len() == 1 {
                "singleton community from the complete admitted relation scope"
            } else {
                "deterministic weighted relationship community candidate"
            }
            .to_string(),
            nodes: analysis_nodes_for(nodes, members),
            metric: Some(members.len() as u64),
            evidence: None,
            community: Some(metadata),
        };
        let finding_bytes = serialized_bytes_controlled(&finding, control)?;
        let finding_append_bytes = serialized_finding_append_bytes(
            finding_bytes,
            preceding_finding_count.saturating_add(findings.len()),
        );
        if retained_append_bytes.saturating_add(finding_append_bytes) > intermediate_bytes {
            truncated = true;
            break;
        }
        retained_append_bytes = retained_append_bytes.saturating_add(finding_append_bytes);
        findings.push(finding);
    }
    if truncated {
        check_control(control)?;
        let marker = community_truncation_finding(
            weights,
            parameters,
            iteration,
            CommunityCoverage::Complete,
        );
        let marker_bytes = serialized_bytes_controlled(&marker, control)?;
        let marker_append_bytes = serialized_finding_append_bytes(
            marker_bytes,
            preceding_finding_count.saturating_add(findings.len()),
        );
        if retained_append_bytes.saturating_add(marker_append_bytes) > intermediate_bytes {
            findings.clear();
        } else {
            findings.push(marker);
        }
    }
    Ok((
        findings,
        truncated
            .then_some(vec![GraphLimitKind::IntermediateBytes])
            .unwrap_or_default(),
    ))
}

/// Charge every graph-sized community temporary before allocating it.
///
/// The edge charge mirrors the complete, weighted relation admission used by
/// `admitted_community_scope`; excluded relation families never own community
/// clones and therefore do not consume this working-set budget.
fn community_working_set_upper_bound(
    nodes: &BTreeMap<String, DetailedRelationNode>,
    edges: &[LocalEdge],
    control: Option<&IndexWorkControl>,
) -> ServiceResult<u64> {
    let mut endpoint_bytes = 0_u64;
    for key in nodes.keys() {
        check_control(control)?;
        endpoint_bytes = endpoint_bytes
            .saturating_add(u64::try_from(key.len()).unwrap_or(u64::MAX))
            .saturating_add(std::mem::size_of::<String>() as u64);
    }
    let mut admitted_edge_count = 0_u64;
    for edge in edges {
        check_control(control)?;
        if admitted_community_edge_weight(edge).is_none() {
            continue;
        }
        admitted_edge_count = admitted_edge_count.saturating_add(1);
        endpoint_bytes = endpoint_bytes
            .saturating_add(u64::try_from(edge.source.len()).unwrap_or(u64::MAX))
            .saturating_add(u64::try_from(edge.target.len()).unwrap_or(u64::MAX))
            .saturating_add(std::mem::size_of::<String>() as u64 * 2)
            .saturating_add(std::mem::size_of::<CommunityEdgeEvidence>() as u64);
    }
    let entry_count = u64::try_from(nodes.len())
        .unwrap_or(u64::MAX)
        .saturating_add(admitted_edge_count);
    Ok(endpoint_bytes
        .saturating_mul(COMMUNITY_WORKING_SET_MULTIPLIER)
        .saturating_add(entry_count.saturating_mul(COMMUNITY_WORKING_SET_ENTRY_BYTES))
        .saturating_add(COMMUNITY_WORKING_SET_FIXED_BYTES))
}

/// Build the typed marker returned when community construction cannot fit.
fn community_truncation_finding(
    weights: &[CommunityRelationWeight],
    parameters: CommunityParameters,
    iteration: u32,
    coverage: CommunityCoverage,
) -> AnalysisFinding {
    AnalysisFinding {
        kind: AnalysisFindingKind::Community,
        status: AnalysisStatus::Inconclusive,
        summary: "community projection crossed the intermediate-byte budget before all communities were constructed"
            .to_string(),
        nodes: Vec::new(),
        metric: None,
        evidence: None,
        community: Some(CommunityAnalysis {
            id: community_id(&[], weights, parameters),
            members: Vec::new(),
            evidence: Vec::new(),
            weights: weights.to_vec(),
            parameters,
            iteration,
            convergence: CommunityConvergence::Inconclusive,
            coverage,
            truncated: true,
        }),
    }
}

/// Conservatively charge one candidate before allocating its owned vectors.
fn community_finding_upper_bound(
    nodes: &BTreeMap<String, DetailedRelationNode>,
    members: &[String],
    evidence: &[CommunityEdgeEvidence],
    weights: &[CommunityRelationWeight],
    parameters: CommunityParameters,
    control: Option<&IndexWorkControl>,
) -> ServiceResult<u64> {
    let member_bytes = community_member_bytes(nodes, members, control)?;
    let evidence_bytes = serialized_bytes_controlled(evidence, control)?;
    let weights_bytes = serialized_bytes_controlled(weights, control)?;
    let parameter_bytes = serialized_bytes_controlled(&parameters, control)?;
    Ok(member_bytes
        .saturating_mul(2)
        .saturating_add(evidence_bytes)
        .saturating_add(weights_bytes)
        .saturating_add(parameter_bytes)
        .saturating_add((members.len() as u64).saturating_mul(2))
        .saturating_add(2_048))
}

/// Conservatively charge admitted evidence without allocating a group copy.
fn community_candidate_upper_bound(
    nodes: &BTreeMap<String, DetailedRelationNode>,
    members: &[String],
    admitted_edges: &[CommunityEdgeEvidence],
    weights: &[CommunityRelationWeight],
    parameters: CommunityParameters,
    control: Option<&IndexWorkControl>,
) -> ServiceResult<u64> {
    let member_bytes = community_member_bytes(nodes, members, control)?;
    let mut evidence_bytes = 0_u64;
    for edge in admitted_edges {
        check_control(control)?;
        if members.binary_search(&edge.source).is_ok()
            && members.binary_search(&edge.target).is_ok()
        {
            evidence_bytes = evidence_bytes
                .saturating_add(serialized_bytes_controlled(edge, control)?)
                .saturating_add(1);
        }
    }
    let weights_bytes = serialized_bytes_controlled(weights, control)?;
    let parameter_bytes = serialized_bytes_controlled(&parameters, control)?;
    Ok(member_bytes
        .saturating_mul(2)
        .saturating_add(evidence_bytes)
        .saturating_add(weights_bytes)
        .saturating_add(parameter_bytes)
        .saturating_add((members.len() as u64).saturating_mul(2))
        .saturating_add(2_048))
}

/// Measure projected members one at a time before retaining a finding.
fn community_member_bytes(
    nodes: &BTreeMap<String, DetailedRelationNode>,
    members: &[String],
    control: Option<&IndexWorkControl>,
) -> ServiceResult<u64> {
    let mut bytes = 0_u64;
    for key in members {
        check_control(control)?;
        if let Some(node) = nodes.get(key) {
            bytes =
                bytes.saturating_add(serialized_bytes_controlled(&analysis_node(node), control)?);
        }
    }
    Ok(bytes)
}

/// Build stable metadata for one community or typed inconclusive result.
fn community_metadata(
    members: &[String],
    evidence: &[CommunityEdgeEvidence],
    weights: Vec<CommunityRelationWeight>,
    parameters: CommunityParameters,
    iteration: u32,
    convergence: CommunityConvergence,
    coverage: CommunityCoverage,
    truncated: bool,
    nodes: &BTreeMap<String, DetailedRelationNode>,
) -> CommunityAnalysis {
    let mut members = members.to_vec();
    members.sort();
    let id = community_id(&members, &weights, parameters);
    let mut evidence = evidence.to_vec();
    evidence.sort_by(|left, right| {
        (&left.source, &left.target, left.relation.as_str()).cmp(&(
            &right.source,
            &right.target,
            right.relation.as_str(),
        ))
    });
    CommunityAnalysis {
        id,
        members: analysis_nodes_for(nodes, &members),
        evidence,
        weights,
        parameters,
        iteration,
        convergence,
        coverage,
        truncated,
    }
}

/// Hash one stable community identifier from its complete versioned inputs.
fn community_id(
    members: &[String],
    weights: &[CommunityRelationWeight],
    parameters: CommunityParameters,
) -> String {
    let mut hasher = blake3::Hasher::new();
    hasher.update(b"projectatlas:architecture-community");
    hasher.update(&[0]);
    hasher.update(&parameters.algorithm_version.to_le_bytes());
    hasher.update(&parameters.ordering_version.to_le_bytes());
    hasher.update(&parameters.max_iterations.to_le_bytes());
    hasher.update(&parameters.node_limit.to_le_bytes());
    hasher.update(&parameters.edge_limit.to_le_bytes());
    hasher.update(&parameters.output_bytes.to_le_bytes());
    hasher.update(&COMMUNITY_SELF_WEIGHT.to_le_bytes());
    hasher.update(
        parameters
            .relation
            .map_or("none", GraphRelationKind::as_str)
            .as_bytes(),
    );
    hasher.update(&[0]);
    for entry in weights {
        hasher.update(entry.relation.as_str().as_bytes());
        hasher.update(&[0]);
        hasher.update(&entry.weight.to_le_bytes());
    }
    for member in members {
        hasher.update(member.as_bytes());
        hasher.update(&[0]);
    }
    format!(
        "community-v{}-{}",
        parameters.algorithm_version,
        hasher.finalize().to_hex()
    )
}

/// Compute deterministic weak components with optional containment exclusion.
fn weak_components(
    nodes: &BTreeMap<String, DetailedRelationNode>,
    edges: &[LocalEdge],
    exclude_contains: bool,
) -> Vec<Vec<String>> {
    let mut adjacency = nodes
        .keys()
        .map(|key| (key.clone(), Vec::new()))
        .collect::<BTreeMap<_, _>>();
    for edge in edges {
        if exclude_contains && edge.kind == GraphRelationKind::Legacy(RelationKind::Contains) {
            continue;
        }
        adjacency
            .entry(edge.source.clone())
            .or_default()
            .push(edge.target.clone());
        adjacency
            .entry(edge.target.clone())
            .or_default()
            .push(edge.source.clone());
    }
    let mut visited = BTreeSet::new();
    let mut components = Vec::new();
    for start in nodes.keys() {
        if !visited.insert(start.clone()) {
            continue;
        }
        let mut queue = VecDeque::from([start.clone()]);
        let mut component = Vec::new();
        while let Some(node) = queue.pop_front() {
            component.push(node.clone());
            if let Some(neighbors) = adjacency.get(&node) {
                for neighbor in neighbors {
                    if visited.insert(neighbor.clone()) {
                        queue.push_back(neighbor.clone());
                    }
                }
            }
        }
        component.sort();
        components.push(component);
    }
    components
}

/// Compute iterative deterministic SCCs over dependency relations.
fn strongly_connected_components(
    nodes: &BTreeMap<String, DetailedRelationNode>,
    edges: &[LocalEdge],
) -> Vec<Vec<String>> {
    let keys = nodes.keys().cloned().collect::<Vec<_>>();
    let indices = keys
        .iter()
        .enumerate()
        .map(|(index, key)| (key.clone(), index))
        .collect::<BTreeMap<_, _>>();
    let mut forward = vec![Vec::new(); keys.len()];
    let mut reverse = vec![Vec::new(); keys.len()];
    for edge in edges {
        let (Some(&source), Some(&target)) = (indices.get(&edge.source), indices.get(&edge.target))
        else {
            continue;
        };
        forward[source].push(target);
        reverse[target].push(source);
    }
    let mut seen = vec![false; keys.len()];
    let mut order = Vec::new();
    for start in 0..keys.len() {
        if seen[start] {
            continue;
        }
        let mut stack = vec![(start, false)];
        while let Some((node, expanded)) = stack.pop() {
            if expanded {
                order.push(node);
                continue;
            }
            if seen[node] {
                continue;
            }
            seen[node] = true;
            stack.push((node, true));
            for &next in forward[node].iter().rev() {
                if !seen[next] {
                    stack.push((next, false));
                }
            }
        }
    }
    seen.fill(false);
    let mut components = Vec::new();
    for &start in order.iter().rev() {
        if seen[start] {
            continue;
        }
        seen[start] = true;
        let mut stack = vec![start];
        let mut component = Vec::new();
        while let Some(node) = stack.pop() {
            component.push(keys[node].clone());
            for &next in &reverse[node] {
                if !seen[next] {
                    seen[next] = true;
                    stack.push(next);
                }
            }
        }
        component.sort();
        components.push(component);
    }
    components
}

/// Count admitted local fan-in plus fan-out for every node.
fn degrees(
    nodes: &BTreeMap<String, DetailedRelationNode>,
    edges: &[LocalEdge],
) -> BTreeMap<String, usize> {
    let mut values = nodes
        .keys()
        .map(|key| (key.clone(), 0))
        .collect::<BTreeMap<_, _>>();
    for edge in edges {
        *values.entry(edge.source.clone()).or_default() += 1;
        *values.entry(edge.target.clone()).or_default() += 1;
    }
    values
}

/// Count trusted inbound usage relations for every node.
///
/// Structural containment establishes declaration ownership, not runtime or
/// compile-time use, so it cannot by itself suppress a dead-code candidate.
fn usage_indegrees(
    nodes: &BTreeMap<String, DetailedRelationNode>,
    edges: &[LocalEdge],
    control: Option<&IndexWorkControl>,
) -> ServiceResult<BTreeMap<String, usize>> {
    let mut values = BTreeMap::new();
    for key in nodes.keys() {
        check_control(control)?;
        values.insert(key.clone(), 0);
    }
    for edge in edges
        .iter()
        .filter(|edge| edge.kind != GraphRelationKind::Legacy(RelationKind::Contains))
    {
        check_control(control)?;
        *values.entry(edge.target.clone()).or_default() += 1;
    }
    Ok(values)
}

/// Project selected canonical identities into reusable analysis nodes.
fn analysis_nodes_for(
    nodes: &BTreeMap<String, DetailedRelationNode>,
    keys: &[String],
) -> Vec<AnalysisNode> {
    keys.iter()
        .filter_map(|key| nodes.get(key))
        .map(analysis_node)
        .collect()
}

/// Measure node payload added to an already serialized empty node array.
fn serialized_analysis_nodes_bytes<'node>(
    nodes: impl IntoIterator<Item = &'node DetailedRelationNode>,
    control: Option<&IndexWorkControl>,
) -> ServiceResult<u64> {
    let mut bytes = 0_u64;
    for (index, node) in nodes.into_iter().enumerate() {
        if index > 0 {
            bytes = bytes
                .checked_add(JSON_ARRAY_SEPARATOR_BYTES)
                .ok_or_else(entrypoint_work_overflow)?;
        }
        bytes = bytes
            .checked_add(serialized_bytes_controlled(&analysis_node(node), control)?)
            .ok_or_else(entrypoint_work_overflow)?;
    }
    Ok(bytes)
}

/// Preserve one detailed node and its exact reusable next call.
fn analysis_node(node: &DetailedRelationNode) -> AnalysisNode {
    let next_call = super::relations::next_call_for_entity(
        &node.entity,
        node.content_selection
            .unwrap_or(ContentSelection::UnspecifiedLegacy),
        node.classification,
    );
    AnalysisNode {
        node: node.clone(),
        next_call,
    }
}

/// Compare a normalized graph entity with an exact public relation anchor.
fn entity_matches_anchor(entity: &GraphEntity, target: &RelationAnchor) -> bool {
    match (entity.selector(), target) {
        (EntitySelector::File { path }, RelationAnchor::File { file }) => path == file,
        (
            EntitySelector::Symbol { symbol },
            RelationAnchor::Symbol {
                file,
                name,
                symbol_kind,
                parent,
                signature,
            },
        ) => {
            symbol.file == *file
                && symbol.name.as_str() == name
                && symbol_kind.is_none_or(|kind| symbol.kind == kind)
                && symbol.parent.as_ref().map(GraphIdentityText::as_str) == parent.as_deref()
                && signature
                    .as_deref()
                    .is_none_or(|signature| symbol.signature.as_str() == signature)
        }
        _ => false,
    }
}

/// Return the owning repository path for a file or symbol entity.
fn entity_path(entity: &GraphEntity) -> Option<&str> {
    match entity.selector() {
        EntitySelector::Folder { path } => Some(path.as_str()),
        EntitySelector::File { path } => Some(path.as_str()),
        EntitySelector::Package { package } => Some(package.manifest.as_str()),
        EntitySelector::Symbol { symbol } => Some(symbol.file.as_str()),
        EntitySelector::Project | EntitySelector::External { .. } => None,
    }
}

/// Borrow the exact declaration identity from one symbol entity.
fn symbol_identity(
    entity: &GraphEntity,
) -> Option<(
    &str,
    &str,
    projectatlas_core::symbols::SymbolKind,
    Option<&str>,
    &str,
)> {
    let EntitySelector::Symbol { symbol } = entity.selector() else {
        return None;
    };
    Some((
        symbol.file.as_str(),
        symbol.name.as_str(),
        symbol.kind,
        symbol.parent.as_ref().map(GraphIdentityText::as_str),
        symbol.signature.as_str(),
    ))
}

/// Clone one fitted finding prefix and update its returned work.
fn analysis_prefix(
    report: &RelationAnalysisReport,
    rows: usize,
    reached_limits: impl IntoIterator<Item = GraphLimitKind>,
) -> RelationAnalysisReport {
    let mut candidate = report.clone();
    if rows < candidate.findings.len() {
        candidate.findings.truncate(rows);
        candidate.truncated = true;
        for limit in reached_limits {
            push_limit(&mut candidate.reached_limits, limit);
        }
    }
    candidate.returned = u32::try_from(candidate.findings.len()).unwrap_or(u32::MAX);
    candidate.work.rendered_output_bytes = 0;
    candidate
}

/// Insert one stable limit once while preserving encounter order.
fn push_limit(limits: &mut Vec<GraphLimitKind>, limit: GraphLimitKind) {
    if !limits.contains(&limit) {
        limits.push(limit);
    }
}

/// Check shared cancellation and deadline state at traversal work boundaries.
fn check_control(control: Option<&IndexWorkControl>) -> ServiceResult<()> {
    if let Some(control) = control {
        control
            .check(IndexWorkStage::RepositoryTraversal)
            .map_err(DbError::from)?;
    }
    Ok(())
}

#[cfg(test)]
#[path = "analysis/tests.rs"]
mod tests;