aft/callgraph_store/

join.rs

//! Manifest-only callgraph assembly.
//!
//! This module deliberately accepts a manifest plus immutable blob payloads rather
//! than a checkout root.  Extraction is content-addressed and path-free; binding a
//! blob to a manifest path and resolving its cross-file references happens here.
//! The existing resolver uses String file identities. Non-UTF-8 source members
//! remain in byte-addressed facts but are reported as unbound rather than being
//! converted lossily; supporting them requires a separate resolver identity change.

use std::collections::{BTreeMap, BTreeSet};
use std::fmt;

use serde::{Deserialize, Serialize};
use tree_sitter::Parser;

use super::facts::{BlobKey, FactPaths, ManifestFacts, ProjectFacts};
use crate::callgraph::{self, FileCallData, SymbolMeta};
use crate::imports::{ImportBlock, ImportForm, ImportGroup, ImportKind, ImportStatement};
use crate::parser::{grammar_for, LangId};
use crate::symbols::SymbolKind;
use crate::views::{Manifest, ManifestEntry, RelPath};
use std::collections::HashMap;
use std::path::Path;
use std::rc::Rc;
use std::sync::Arc;

const TOP_LEVEL_SYMBOL: &str = "<top-level>";

/// An error raised while decoding or assembling manifest-addressed callgraph data.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum ManifestJoinError {
    UnsupportedLanguage(String),
    Parse(String),
    InvalidBlob(String),
    MissingBlob(String),
    InvalidConfig { path: Vec<u8>, reason: String },
}

impl fmt::Display for ManifestJoinError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::UnsupportedLanguage(language) => {
                write!(formatter, "unsupported callgraph blob language {language}")
            }
            Self::Parse(reason) => write!(formatter, "callgraph blob parse failed: {reason}"),
            Self::InvalidBlob(reason) => write!(formatter, "invalid callgraph blob: {reason}"),
            Self::MissingBlob(key) => write!(
                formatter,
                "manifest references missing callgraph blob {key}"
            ),
            Self::InvalidConfig { path, reason } => write!(
                formatter,
                "invalid manifest config {}: {reason}",
                String::from_utf8_lossy(path)
            ),
        }
    }
}

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

/// Reads immutable payloads by the full key recorded in a manifest entry.
///
/// Implementations may be backed by the family blob store, but this interface
/// intentionally exposes no checkout path or directory operation to the join.
pub trait ManifestBlobReader {
    fn read_callgraph_blob(&self, full_key: &str) -> Result<Option<Vec<u8>>, ManifestJoinError>;

    fn read_callgraph_blob_decoded(
        &self,
        full_key: &str,
    ) -> Result<Option<Arc<CallgraphBlob>>, ManifestJoinError> {
        self.read_callgraph_blob(full_key)?
            .map(|payload| CallgraphBlob::from_bytes(&payload).map(Arc::new))
            .transpose()
    }
}

/// Tree-sitter node position in canonical pre-order traversal order.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct AstPreorderNode {
    pub ordinal: u32,
    pub kind: String,
    pub byte_start: usize,
    pub byte_end: usize,
}

/// A symbol captured by extraction.  Its ordinal is the source AST node's
/// pre-order position, not a per-path or database-generated identifier.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct BlobSymbol {
    pub ordinal: u32,
    pub name: String,
    pub scoped_name: String,
    pub kind: String,
    pub exported: bool,
    pub is_default_export: bool,
    pub start_line: u32,
    pub start_col: u32,
    pub end_line: u32,
    pub end_col: u32,
    pub signature: Option<String>,
}

/// The parse-level class of a reference.  No target path is present in a blob.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum BlobRefKind {
    Call,
    ValueRef,
    Import,
    Module,
    Reexport,
    ExportAlias,
}

/// An unresolved reference extracted from one source blob.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct BlobRef {
    /// Canonical tree-sitter AST pre-order position of this reference.
    pub ordinal: u32,
    pub kind: BlobRefKind,
    pub caller_symbol: Option<String>,
    pub short_name: Option<String>,
    pub full_ref: Option<String>,
    pub module_path: Option<String>,
    pub line: u32,
    pub byte_start: usize,
    pub byte_end: usize,
    pub path_override: Option<String>,
    pub local_name: Option<String>,
    pub requested_name: Option<String>,
    pub namespace_alias: Option<String>,
    pub wildcard: bool,
    pub import_kind: Option<String>,
}

/// A parsed import retained in the blob so binding can resolve aliases without
/// re-reading source text.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct BlobImport {
    pub ordinal: u32,
    pub module_path: String,
    pub names: Vec<String>,
    pub default_import: Option<String>,
    pub namespace_import: Option<String>,
    pub byte_start: usize,
    pub byte_end: usize,
    pub raw_text: String,
    pub type_only: bool,
    pub side_effect: bool,
}

/// Path-free parse output stored for a regular source file.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct ParseBlob {
    pub extractor_version: String,
    pub language: String,
    pub ast_nodes: Vec<AstPreorderNode>,
    pub symbols: Vec<BlobSymbol>,
    pub default_export_symbol: Option<String>,
    pub exported_symbols: Vec<String>,
    pub callable_symbols: Vec<String>,
    pub imports: Vec<BlobImport>,
    pub refs: Vec<BlobRef>,
}

/// Raw source is retained only for configuration files and ignore-list members.
/// These blobs are parsed during joining because their interpretation depends on
/// the manifest view they configure.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct ConfigBlob {
    pub extractor_version: String,
    pub language: String,
    pub source: Vec<u8>,
}

/// The immutable callgraph blob payload.  A regular source blob has parse output
/// only; a configuration blob is intentionally raw so its manifest-scoped
/// resolver settings can be interpreted during assembly.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum CallgraphBlob {
    Parse(ParseBlob),
    Config(ConfigBlob),
}

impl CallgraphBlob {
    /// Extracts path-free callgraph parse output from source bytes and the
    /// extractor version that names the corresponding content key.
    pub fn extract(
        source: &str,
        language: &str,
        extractor_version: impl Into<String>,
    ) -> Result<Self, ManifestJoinError> {
        let lang = language_id(language)
            .ok_or_else(|| ManifestJoinError::UnsupportedLanguage(language.to_string()))?;
        let extractor_version = extractor_version.into();
        let ast_nodes = ast_preorder_nodes(source, lang)?;
        let mut data = callgraph::build_file_data_from_source_with_lang(
            std::path::Path::new("__callgraph_blob__"),
            source,
            lang,
        )
        .map_err(|error| ManifestJoinError::Parse(error.to_string()))?;
        if lang == LangId::Rust {
            super::extend_rust_imports_with_nested_uses(source, &mut data);
        }
        let symbols = blob_symbols(source, &data, &ast_nodes);
        let imports = blob_imports(&data, &ast_nodes);
        let mut refs = blob_refs(source, &data, &ast_nodes);
        refs.extend(rust_module_refs(source, lang, &ast_nodes));
        let empty =
            Manifest::new([]).map_err(|error| ManifestJoinError::Parse(error.to_string()))?;
        let reader = |_: &BlobKey| None;
        let empty_facts = ManifestFacts {
            manifest: &empty,
            blobs: &reader,
        };
        let paths = FactPaths {
            root: Path::new("/"),
            facts: &empty_facts,
        };
        let file = Path::new("/__callgraph_blob__");
        let mut structural =
            super::collect_reexport_refs(paths.root, file, "__callgraph_blob__", source, &paths)
                .raw_refs;
        structural.extend(
            super::collect_source_less_export_alias_refs("__callgraph_blob__", source).raw_refs,
        );
        if lang == LangId::Rust {
            structural.extend(
                super::collect_rust_pub_use_reexport_refs(
                    paths.root,
                    file,
                    "__callgraph_blob__",
                    &data.import_block.imports,
                    &super::LineIndex::new(source),
                    &paths,
                )
                .raw_refs,
            );
        }
        refs.extend(
            structural
                .into_iter()
                .map(|raw| structural_ref(raw, &ast_nodes)),
        );
        let mut exported_symbols = data.exported_symbols.clone();
        exported_symbols.sort();
        let mut callable_symbols = data.calls_by_symbol.keys().cloned().collect::<Vec<_>>();
        callable_symbols.sort();
        refs.sort_by(|left, right| {
            (
                left.ordinal,
                left.kind,
                left.byte_start,
                left.byte_end,
                left.full_ref.as_deref(),
            )
                .cmp(&(
                    right.ordinal,
                    right.kind,
                    right.byte_start,
                    right.byte_end,
                    right.full_ref.as_deref(),
                ))
        });
        refs.dedup_by(|left, right| {
            left.ordinal == right.ordinal
                && left.kind == right.kind
                && left.byte_start == right.byte_start
                && left.byte_end == right.byte_end
                && left.full_ref == right.full_ref
        });

        Ok(Self::Parse(ParseBlob {
            extractor_version,
            language: language.to_string(),
            ast_nodes,
            symbols,
            default_export_symbol: data.default_export_symbol,
            exported_symbols,
            callable_symbols,
            imports,
            refs,
        }))
    }

    /// Builds a manifest configuration input.  The caller must key it with
    /// `language = "config"` and the same extractor version stored here.
    pub fn config(source: impl Into<Vec<u8>>, extractor_version: impl Into<String>) -> Self {
        Self::Config(ConfigBlob {
            extractor_version: extractor_version.into(),
            language: "config".to_string(),
            source: source.into(),
        })
    }

    /// Uses one canonical JSON encoding for the immutable payload bytes.
    pub fn to_bytes(&self) -> Result<Vec<u8>, ManifestJoinError> {
        serde_json::to_vec(self).map_err(|error| ManifestJoinError::InvalidBlob(error.to_string()))
    }

    /// Decodes a payload after the blob store has verified its digest and schema.
    pub fn from_bytes(bytes: &[u8]) -> Result<Self, ManifestJoinError> {
        serde_json::from_slice(bytes)
            .map_err(|error| ManifestJoinError::InvalidBlob(error.to_string()))
    }

    pub fn parse(&self) -> Option<&ParseBlob> {
        match self {
            Self::Parse(blob) => Some(blob),
            Self::Config(_) => None,
        }
    }

    pub fn config_source(&self) -> Option<&ConfigBlob> {
        match self {
            Self::Parse(_) => None,
            Self::Config(blob) => Some(blob),
        }
    }
}

/// The stable identity of one bound blob reference.  The path breaks ties when
/// identical content is bound at more than one manifest path.
#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
pub struct CallerRefKey {
    pub caller_blob_key: String,
    pub ref_ordinal: u32,
    pub caller_path: Vec<u8>,
}

/// The manifest-derived resolution state for one reference.
#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
pub enum ResolutionStatus {
    Resolved,
    Unresolved,
}

/// A logical derived row.  It is intentionally independent of SQLite rowids and
/// other physical database details.
#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
pub struct DerivedRow {
    pub caller_blob_key: String,
    pub ref_ordinal: u32,
    pub caller_path: Vec<u8>,
    pub kind: BlobRefKind,
    pub status: ResolutionStatus,
    pub target_path: Option<Vec<u8>>,
    pub target_symbol: Option<String>,
}

impl DerivedRow {
    pub fn ref_key(&self) -> CallerRefKey {
        CallerRefKey {
            caller_blob_key: self.caller_blob_key.clone(),
            ref_ordinal: self.ref_ordinal,
            caller_path: self.caller_path.clone(),
        }
    }
}

/// The result of resolving one manifest.  `resolution_order` exposes the exact
/// canonical order consumed by the resolver for deterministic test coverage.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct JoinResult {
    pub rows: BTreeSet<DerivedRow>,
    pub resolution_order: Vec<CallerRefKey>,
    pub unbound_non_utf8_paths: Vec<Vec<u8>>,
}

impl JoinResult {
    /// Serializes logical rows in canonical order.  This is the comparison form
    /// for equal manifests; callers must not compare SQLite file bytes.
    pub fn canonical_serialization(&self) -> Vec<u8> {
        let mut output = Vec::new();
        for row in &self.rows {
            append_field(&mut output, row.caller_blob_key.as_bytes());
            append_field(&mut output, &row.ref_ordinal.to_be_bytes());
            append_field(&mut output, &row.caller_path);
            append_field(&mut output, &[row.kind as u8]);
            append_field(
                &mut output,
                &[match row.status {
                    ResolutionStatus::Resolved => 1,
                    ResolutionStatus::Unresolved => 0,
                }],
            );
            append_optional_field(&mut output, row.target_path.as_deref());
            append_optional_field(&mut output, row.target_symbol.as_deref().map(str::as_bytes));
        }
        output
    }
}

/// Incremental assembly details used to verify precise invalidation behavior.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct IncrementalJoinResult {
    pub result: JoinResult,
    pub re_resolved: BTreeSet<CallerRefKey>,
    pub full_re_resolve: bool,
}

fn changed_manifest_paths(previous: &Manifest, current: &Manifest) -> BTreeSet<Vec<u8>> {
    let previous_entries = previous
        .entries()
        .map(|(path, entry)| (path.as_bytes().to_vec(), entry))
        .collect::<BTreeMap<_, _>>();
    let current_entries = current
        .entries()
        .map(|(path, entry)| (path.as_bytes().to_vec(), entry))
        .collect::<BTreeMap<_, _>>();
    previous_entries
        .keys()
        .chain(current_entries.keys())
        .collect::<BTreeSet<_>>()
        .into_iter()
        .filter(|path| previous_entries.get(*path) != current_entries.get(*path))
        .cloned()
        .collect()
}

fn manifest_resolution_input(manifest: &Manifest, path: &[u8]) -> bool {
    let lookup = if path.first() == Some(&0) {
        return manifest
            .entries()
            .find(|(candidate, _)| candidate.as_bytes() == path)
            .is_some_and(|(_, entry)| matches!(entry, ManifestEntry::Synthetic { .. }));
    } else {
        RelPath::new(path.to_vec()).ok()
    };
    lookup
        .as_ref()
        .and_then(|path| manifest.get(path))
        .is_some_and(|entry| {
            matches!(
                entry,
                ManifestEntry::Regular {
                    resolution_input: true,
                    ..
                }
            )
        })
}

fn append_field(output: &mut Vec<u8>, value: &[u8]) {
    output.extend_from_slice(&(value.len() as u64).to_be_bytes());
    output.extend_from_slice(value);
}

fn append_optional_field(output: &mut Vec<u8>, value: Option<&[u8]>) {
    match value {
        Some(value) => {
            output.push(1);
            append_field(output, value);
        }
        None => output.push(0),
    }
}

fn language_id(language: &str) -> Option<LangId> {
    Some(match language {
        "typescript" => LangId::TypeScript,
        "tsx" => LangId::Tsx,
        "javascript" => LangId::JavaScript,
        "python" => LangId::Python,
        "rust" => LangId::Rust,
        "go" => LangId::Go,
        "c" => LangId::C,
        "cpp" => LangId::Cpp,
        "cuda" => LangId::Cuda,
        "metal" => LangId::Metal,
        "zig" => LangId::Zig,
        "csharp" => LangId::CSharp,
        "bash" => LangId::Bash,
        "html" => LangId::Html,
        "markdown" => LangId::Markdown,
        "solidity" => LangId::Solidity,
        "scss" => LangId::Scss,
        "vue" => LangId::Vue,
        "json" => LangId::Json,
        "scala" => LangId::Scala,
        "java" => LangId::Java,
        "ruby" => LangId::Ruby,
        "kotlin" => LangId::Kotlin,
        "swift" => LangId::Swift,
        "php" => LangId::Php,
        "lua" => LangId::Lua,
        "perl" => LangId::Perl,
        "yaml" => LangId::Yaml,
        "pascal" => LangId::Pascal,
        "r" => LangId::R,
        "groovy" => LangId::Groovy,
        "objc" => LangId::ObjC,
        "toml" => LangId::Toml,
        _ => return None,
    })
}

fn ast_preorder_nodes(
    source: &str,
    lang: LangId,
) -> Result<Vec<AstPreorderNode>, ManifestJoinError> {
    let mut parser = Parser::new();
    parser
        .set_language(&grammar_for(lang))
        .map_err(|error| ManifestJoinError::Parse(error.to_string()))?;
    let tree = parser
        .parse(source, None)
        .ok_or_else(|| ManifestJoinError::Parse("tree-sitter returned no tree".to_string()))?;
    let mut nodes = Vec::new();
    let mut stack = vec![tree.root_node()];
    while let Some(node) = stack.pop() {
        nodes.push(AstPreorderNode {
            ordinal: nodes.len() as u32,
            kind: node.kind().to_string(),
            byte_start: node.start_byte(),
            byte_end: node.end_byte(),
        });
        let children = node.children(&mut node.walk()).collect::<Vec<_>>();
        stack.extend(children.into_iter().rev());
    }
    Ok(nodes)
}

fn blob_symbols(
    source: &str,
    data: &FileCallData,
    ast_nodes: &[AstPreorderNode],
) -> Vec<BlobSymbol> {
    let mut symbols = data
        .symbol_metadata
        .iter()
        .map(|(scoped_name, meta)| {
            blob_symbol(
                source,
                scoped_name,
                meta,
                &data.default_export_symbol,
                ast_nodes,
            )
        })
        .collect::<Vec<_>>();
    symbols.sort_by(|left, right| {
        (
            left.ordinal,
            left.start_line,
            left.start_col,
            left.scoped_name.as_str(),
        )
            .cmp(&(
                right.ordinal,
                right.start_line,
                right.start_col,
                right.scoped_name.as_str(),
            ))
    });
    symbols
}

fn blob_symbol(
    source: &str,
    scoped_name: &str,
    meta: &SymbolMeta,
    default_export: &Option<String>,
    ast_nodes: &[AstPreorderNode],
) -> BlobSymbol {
    let byte_start = byte_offset(source, meta.range.start_line, meta.range.start_col);
    let byte_end = byte_offset(source, meta.range.end_line, meta.range.end_col).max(byte_start);
    BlobSymbol {
        ordinal: ordinal_for_range(ast_nodes, byte_start, byte_end),
        name: unqualified_symbol_name(scoped_name).to_string(),
        scoped_name: scoped_name.to_string(),
        kind: symbol_kind_name(&meta.kind).to_string(),
        exported: meta.exported,
        is_default_export: default_export.as_deref() == Some(scoped_name),
        start_line: meta.range.start_line,
        start_col: meta.range.start_col,
        end_line: meta.range.end_line,
        end_col: meta.range.end_col,
        signature: meta.signature.clone(),
    }
}

fn blob_imports(data: &FileCallData, ast_nodes: &[AstPreorderNode]) -> Vec<BlobImport> {
    let mut imports = data
        .import_block
        .imports
        .iter()
        .map(|import| BlobImport {
            ordinal: ordinal_for_range(ast_nodes, import.byte_range.start, import.byte_range.end),
            module_path: import.module_path.clone(),
            names: import.names.clone(),
            default_import: import.default_import.clone(),
            namespace_import: import.namespace_import.clone(),
            byte_start: import.byte_range.start,
            byte_end: import.byte_range.end,
            raw_text: import.raw_text.clone(),
            type_only: import.kind == ImportKind::Type,
            side_effect: import.kind == ImportKind::SideEffect,
        })
        .collect::<Vec<_>>();
    imports.sort_by(|left, right| {
        (left.ordinal, left.module_path.as_str()).cmp(&(right.ordinal, right.module_path.as_str()))
    });
    imports
}

fn blob_refs(source: &str, data: &FileCallData, ast_nodes: &[AstPreorderNode]) -> Vec<BlobRef> {
    let mut refs = Vec::new();
    for (caller_symbol, calls) in &data.calls_by_symbol {
        for call in calls {
            refs.push(call_ref(caller_symbol, call, BlobRefKind::Call, ast_nodes));
        }
    }
    for (caller_symbol, calls) in &data.value_refs_by_symbol {
        for call in calls {
            refs.push(call_ref(
                caller_symbol,
                call,
                BlobRefKind::ValueRef,
                ast_nodes,
            ));
        }
    }
    for import in &data.import_block.imports {
        refs.push(BlobRef {
            ordinal: ordinal_for_range(ast_nodes, import.byte_range.start, import.byte_range.end),
            kind: BlobRefKind::Import,
            caller_symbol: None,
            short_name: None,
            full_ref: Some(import.module_path.clone()),
            module_path: Some(import.module_path.clone()),
            line: line_for_byte(source, import.byte_range.start),
            byte_start: import.byte_range.start,
            byte_end: import.byte_range.end,
            path_override: None,
            local_name: None,
            requested_name: None,
            namespace_alias: import.namespace_import.clone(),
            wildcard: super::import_is_wildcard(import),
            import_kind: None,
        });
    }
    refs
}

fn call_ref(
    caller_symbol: &str,
    call: &callgraph::CallSite,
    kind: BlobRefKind,
    ast_nodes: &[AstPreorderNode],
) -> BlobRef {
    BlobRef {
        ordinal: ordinal_for_range(ast_nodes, call.byte_start, call.byte_end),
        kind,
        caller_symbol: Some(caller_symbol.to_string()),
        short_name: Some(call.callee_name.clone()),
        full_ref: Some(call.full_callee.clone()),
        module_path: None,
        line: call.line,
        byte_start: call.byte_start,
        byte_end: call.byte_end,
        path_override: None,
        local_name: None,
        requested_name: None,
        namespace_alias: None,
        wildcard: false,
        import_kind: None,
    }
}

fn rust_module_refs(source: &str, lang: LangId, ast_nodes: &[AstPreorderNode]) -> Vec<BlobRef> {
    if lang != LangId::Rust {
        return Vec::new();
    }
    let mut parser = Parser::new();
    if parser.set_language(&grammar_for(lang)).is_err() {
        return Vec::new();
    }
    let Some(tree) = parser.parse(source, None) else {
        return Vec::new();
    };
    let mut refs = Vec::new();
    let mut stack = vec![tree.root_node()];
    while let Some(node) = stack.pop() {
        if node.kind() == "mod_item"
            && node
                .named_children(&mut node.walk())
                .all(|child| child.kind() != "declaration_list")
        {
            if let Some(name) = node.child_by_field_name("name") {
                let module_name = source[name.byte_range()].to_string();
                refs.push(BlobRef {
                    ordinal: ordinal_for_range(ast_nodes, node.start_byte(), node.end_byte()),
                    kind: BlobRefKind::Module,
                    caller_symbol: None,
                    short_name: Some(module_name.clone()),
                    full_ref: Some(module_name.clone()),
                    module_path: Some(module_name),
                    line: node.start_position().row as u32 + 1,
                    byte_start: node.start_byte(),
                    byte_end: node.end_byte(),
                    path_override: super::rust_module_path_override(source, node)
                        .map(str::to_string),
                    local_name: None,
                    requested_name: None,
                    namespace_alias: None,
                    wildcard: false,
                    import_kind: None,
                });
            }
        }
        let children = node.children(&mut node.walk()).collect::<Vec<_>>();
        stack.extend(children.into_iter().rev());
    }
    refs
}

fn ordinal_for_range(ast_nodes: &[AstPreorderNode], byte_start: usize, byte_end: usize) -> u32 {
    ast_nodes
        .iter()
        .filter(|node| node.byte_start <= byte_start && node.byte_end >= byte_end)
        .min_by_key(|node| (node.byte_end.saturating_sub(node.byte_start), node.ordinal))
        .map(|node| node.ordinal)
        .unwrap_or(0)
}

fn byte_offset(source: &str, line: u32, column: u32) -> usize {
    let mut offset = 0usize;
    for (index, segment) in source.split_inclusive('\n').enumerate() {
        if index as u32 == line {
            return offset + (column as usize).min(segment.len());
        }
        offset += segment.len();
    }
    source.len()
}

fn line_for_byte(source: &str, byte_start: usize) -> u32 {
    source[..byte_start.min(source.len())]
        .bytes()
        .filter(|byte| *byte == b'\n')
        .count() as u32
        + 1
}

fn symbol_kind_name(kind: &SymbolKind) -> &'static str {
    match kind {
        SymbolKind::Function => "function",
        SymbolKind::Kernel => "kernel",
        SymbolKind::Class => "class",
        SymbolKind::Method => "method",
        SymbolKind::Struct => "struct",
        SymbolKind::Interface => "interface",
        SymbolKind::Enum => "enum",
        SymbolKind::TypeAlias => "type_alias",
        SymbolKind::Variable => "variable",
        SymbolKind::Heading => "heading",
        SymbolKind::FileSummary => "file_summary",
    }
}

fn unqualified_symbol_name(scoped_name: &str) -> &str {
    if scoped_name == TOP_LEVEL_SYMBOL {
        return scoped_name;
    }
    scoped_name.rsplit("::").next().unwrap_or(scoped_name)
}

#[cfg(test)]
mod tests {
    use super::*;
    #[test]
    fn blob_ordinals_are_tree_sitter_preorder_positions() {
        let source = "export function run() { return helper(); }\nfunction helper() {}\n";
        let blob = CallgraphBlob::extract(source, "typescript", "join-test-v1").unwrap();
        let repeated = CallgraphBlob::extract(source, "typescript", "join-test-v1").unwrap();
        let different_version =
            CallgraphBlob::extract(source, "typescript", "join-test-v2").unwrap();
        assert_eq!(blob.to_bytes().unwrap(), repeated.to_bytes().unwrap());
        assert_ne!(
            blob.to_bytes().unwrap(),
            different_version.to_bytes().unwrap()
        );
        let parse = blob.parse().unwrap();
        assert_eq!(parse.ast_nodes[0].ordinal, 0);
        assert!(parse.refs.iter().all(|reference| parse
            .ast_nodes
            .iter()
            .any(|node| node.ordinal == reference.ordinal)));
        let helper = parse
            .refs
            .iter()
            .find(|reference| reference.short_name.as_deref() == Some("helper"))
            .unwrap();
        let node = parse
            .ast_nodes
            .iter()
            .find(|node| node.ordinal == helper.ordinal)
            .unwrap();
        assert!(node.byte_start <= helper.byte_start && node.byte_end >= helper.byte_end);
        assert_eq!(node.kind, "call_expression");
        let mut parser = Parser::new();
        parser
            .set_language(&grammar_for(LangId::TypeScript))
            .unwrap();
        let tree = parser.parse(source, None).unwrap();
        let mut cursor = tree.walk();
        let mut expected = Vec::new();
        'preorder: loop {
            let node = cursor.node();
            expected.push(AstPreorderNode {
                ordinal: expected.len() as u32,
                kind: node.kind().to_string(),
                byte_start: node.start_byte(),
                byte_end: node.end_byte(),
            });
            if cursor.goto_first_child() {
                continue;
            }
            while !cursor.goto_next_sibling() {
                if !cursor.goto_parent() {
                    break 'preorder;
                }
            }
        }
        assert_eq!(parse.ast_nodes, expected);
    }
}

fn structural_ref(raw: super::RawRef, nodes: &[AstPreorderNode]) -> BlobRef {
    BlobRef {
        ordinal: ordinal_for_range(nodes, raw.byte_start, raw.byte_end),
        kind: if raw.kind == "export_alias" {
            BlobRefKind::ExportAlias
        } else {
            BlobRefKind::Reexport
        },
        caller_symbol: raw.caller_symbol,
        short_name: raw.short_name,
        full_ref: raw.full_ref,
        module_path: raw.module_path,
        line: raw.line,
        byte_start: raw.byte_start,
        byte_end: raw.byte_end,
        path_override: None,
        local_name: raw.local_name,
        requested_name: raw.requested_name,
        namespace_alias: raw.namespace_alias,
        wildcard: raw.wildcard,
        import_kind: raw.import_kind,
    }
}

fn bound_name(name: &str, path: &str) -> String {
    name.replace(
        "<default:__callgraph_blob__>",
        &format!(
            "<default:{}>",
            Path::new(path)
                .file_name()
                .unwrap_or_default()
                .to_string_lossy()
        ),
    )
}

impl ParseBlob {
    fn file_data(&self, path: &str) -> Result<FileCallData, ManifestJoinError> {
        let lang = language_id(&self.language)
            .ok_or_else(|| ManifestJoinError::UnsupportedLanguage(self.language.clone()))?;
        let mut calls_by_symbol: HashMap<String, Vec<callgraph::CallSite>> = HashMap::new();
        let mut value_refs_by_symbol: HashMap<String, Vec<callgraph::CallSite>> = HashMap::new();
        for raw in &self.refs {
            let map = match raw.kind {
                BlobRefKind::Call => &mut calls_by_symbol,
                BlobRefKind::ValueRef => &mut value_refs_by_symbol,
                _ => continue,
            };
            let Some(caller) = &raw.caller_symbol else {
                continue;
            };
            map.entry(bound_name(caller, path))
                .or_default()
                .push(callgraph::CallSite {
                    callee_name: raw.short_name.clone().unwrap_or_default(),
                    full_callee: raw.full_ref.clone().unwrap_or_default(),
                    line: raw.line,
                    byte_start: raw.byte_start,
                    byte_end: raw.byte_end,
                });
        }
        for symbol in &self.callable_symbols {
            calls_by_symbol.entry(bound_name(symbol, path)).or_default();
        }
        let mut symbol_metadata = HashMap::new();
        for symbol in &self.symbols {
            let kind = match symbol.kind.as_str() {
                "function" => SymbolKind::Function,
                "method" => SymbolKind::Method,
                "class" => SymbolKind::Class,
                "struct" => SymbolKind::Struct,
                "interface" => SymbolKind::Interface,
                "enum" => SymbolKind::Enum,
                "type_alias" => SymbolKind::TypeAlias,
                "heading" => SymbolKind::Heading,
                "file_summary" => SymbolKind::FileSummary,
                _ => SymbolKind::Variable,
            };
            symbol_metadata.insert(
                bound_name(&symbol.scoped_name, path),
                SymbolMeta {
                    kind,
                    exported: symbol.exported,
                    signature: symbol.signature.clone(),
                    line: symbol.start_line + 1,
                    range: crate::symbols::Range {
                        start_line: symbol.start_line,
                        start_col: symbol.start_col,
                        end_line: symbol.end_line,
                        end_col: symbol.end_col,
                    },
                    entry_point_attribute: None,
                },
            );
        }
        let imports = self
            .imports
            .iter()
            .map(|import| ImportStatement {
                module_path: import.module_path.clone(),
                names: import.names.clone(),
                default_import: import.default_import.clone(),
                namespace_import: import.namespace_import.clone(),
                kind: if import.type_only {
                    ImportKind::Type
                } else if import.side_effect {
                    ImportKind::SideEffect
                } else {
                    ImportKind::Value
                },
                group: ImportGroup::Internal,
                byte_range: import.byte_start..import.byte_end,
                raw_text: import.raw_text.clone(),
                form: if lang == LangId::Rust {
                    ImportForm::RustUse {
                        visibility: import.default_import.clone(),
                        named: import.names.clone(),
                    }
                } else {
                    ImportForm::Es {
                        default_import: import.default_import.clone(),
                        namespace_import: import.namespace_import.clone(),
                        named: import.names.clone(),
                        type_only: import.type_only,
                        side_effect: import.side_effect,
                        attribute_clause: None,
                        attribute_type: None,
                    }
                },
            })
            .collect::<Vec<_>>();
        Ok(FileCallData {
            calls_by_symbol,
            value_refs_by_symbol,
            symbol_metadata,
            exported_symbols: self
                .exported_symbols
                .iter()
                .map(|s| bound_name(s, path))
                .collect(),
            default_export_symbol: self
                .default_export_symbol
                .as_ref()
                .map(|s| bound_name(s, path)),
            import_block: ImportBlock {
                imports,
                byte_range: None,
            },
            lang,
        })
    }

    fn bind(
        &self,
        path: &str,
        facts: &FactPaths<'_>,
    ) -> Result<super::FileExtract, ManifestJoinError> {
        self.bind_with_dependencies(path, facts, None)
    }

    fn bind_with_dependencies(
        &self,
        path: &str,
        facts: &FactPaths<'_>,
        cached: Option<&BTreeMap<u32, BTreeSet<String>>>,
    ) -> Result<super::FileExtract, ManifestJoinError> {
        let data = self.file_data(path)?;
        let nodes = self
            .symbols
            .iter()
            .map(|symbol| {
                let scoped_name = bound_name(&symbol.scoped_name, path);
                super::NodeRecord {
                    id: format!("{path}:{}:{scoped_name}", symbol.ordinal),
                    file_path: path.to_string(),
                    name: bound_name(&symbol.name, path),
                    scoped_name,
                    kind: symbol.kind.clone(),
                    range: crate::symbols::Range {
                        start_line: symbol.start_line,
                        start_col: symbol.start_col,
                        end_line: symbol.end_line,
                        end_col: symbol.end_col,
                    },
                    range_ordinal: symbol.ordinal,
                    signature: symbol.signature.clone(),
                    exported: symbol.exported,
                    is_default_export: symbol.is_default_export,
                    is_type_like: false,
                    is_callgraph_entry_point: false,
                }
            })
            .collect::<Vec<_>>();
        // Overloads can share a scoped name. Preserve the first node chosen by
        // the source-order scan without rescanning all symbols for every call.
        let mut caller_nodes = HashMap::with_capacity(nodes.len());
        for node in &nodes {
            #[cfg(test)]
            CALLER_NODE_LOOKUP_WORK.with(|work| work.set(work.get() + 1));
            caller_nodes.entry(&node.scoped_name).or_insert(&node.id);
        }
        let abs = facts.root.join(path);
        let mut raw_refs = Vec::new();
        for (position, raw) in self.refs.iter().enumerate() {
            let position = u32::try_from(position).expect("reference vector fits u32");
            let dependencies =
                if let Some(dependencies) = cached.and_then(|cache| cache.get(&position)) {
                    dependencies.clone()
                } else if raw.kind == BlobRefKind::Module {
                    super::rust_external_module_target(
                        &abs,
                        raw.path_override.as_deref(),
                        raw.module_path.as_deref().unwrap_or_default(),
                        facts,
                    )
                    .and_then(|p| facts.canonical(&p))
                    .map(|p| super::relative_path(facts.root, &p))
                    .into_iter()
                    .collect()
                } else if let Some(module) = &raw.module_path {
                    super::module_dependencies(facts.root, &abs, module, facts)
                } else {
                    BTreeSet::new()
                };
            let caller_symbol = raw
                .caller_symbol
                .as_ref()
                .map(|name| bound_name(name, path));
            let caller_node = caller_symbol.as_ref().and_then(|name| {
                #[cfg(test)]
                CALLER_NODE_LOOKUP_WORK.with(|work| work.set(work.get() + 1));
                #[cfg(test)]
                if SCAN_CALLER_NODES.with(std::cell::Cell::get) {
                    return nodes
                        .iter()
                        .find(|node| {
                            CALLER_NODE_LOOKUP_WORK.with(|work| work.set(work.get() + 1));
                            &node.scoped_name == name
                        })
                        .map(|node| node.id.clone());
                }
                caller_nodes.get(name).map(|id| (*id).clone())
            });
            raw_refs.push(super::RawRef {
                ref_id: format!("{path}:{}:{:?}", raw.ordinal, raw.kind),
                caller_node,
                caller_symbol,
                caller_file: path.to_string(),
                kind: match raw.kind {
                    BlobRefKind::Call => "call",
                    BlobRefKind::ValueRef => "value_ref",
                    BlobRefKind::Import => "import",
                    BlobRefKind::Module => "module",
                    BlobRefKind::Reexport => "reexport",
                    BlobRefKind::ExportAlias => "export_alias",
                }
                .to_string(),
                short_name: raw.short_name.clone(),
                full_ref: raw.full_ref.clone(),
                module_path: raw.module_path.clone(),
                import_kind: raw.import_kind.clone(),
                local_name: raw.local_name.clone(),
                requested_name: raw.requested_name.clone(),
                namespace_alias: raw.namespace_alias.clone(),
                wildcard: raw.wildcard,
                line: raw.line,
                byte_start: raw.byte_start,
                byte_end: raw.byte_end,
                dependencies,
            });
        }
        Ok(super::FileExtract {
            rel_path: path.to_string(),
            freshness: crate::cache_freshness::FileFreshness {
                mtime: std::time::UNIX_EPOCH,
                size: 0,
                content_hash: crate::cache_freshness::zero_hash(),
            },
            lang: data.lang,
            data,
            nodes,
            raw_refs,
            dispatch_hints: Vec::new(),
            surface_fingerprint: String::new(),
        })
    }
}

/// A manifest-backed index is the ordinary resolver index with different facts.
type ManifestProjectIndex<'a> = super::ProjectIndex<'a>;

impl JoinResult {
    pub fn from_manifest(
        manifest: &Manifest,
        blobs: &impl ManifestBlobReader,
    ) -> Result<Self, ManifestJoinError> {
        let loaded = manifest_payloads(manifest, blobs)?;
        let reader = |key: &BlobKey| loaded.get(key).cloned();
        let facts = Rc::new(ManifestFacts {
            manifest,
            blobs: &reader,
        });
        Self::from_facts(manifest, &loaded, Path::new("/"), facts, None)
    }

    fn from_facts<'a>(
        manifest: &'a Manifest,
        loaded: &BTreeMap<String, Arc<[u8]>>,
        root: &Path,
        facts: Rc<dyn ProjectFacts + 'a>,
        selected: Option<&BTreeSet<CallerRefKey>>,
    ) -> Result<Self, ManifestJoinError> {
        let paths = FactPaths {
            root,
            facts: facts.as_ref(),
        };
        let mut extracts = HashMap::new();
        let mut work = Vec::new();
        let mut unbound_non_utf8_paths = Vec::new();
        for (path, entry) in manifest.entries() {
            let ManifestEntry::Regular { planes, .. } = entry else {
                continue;
            };
            let Some(key) = &planes.callgraph else {
                continue;
            };
            let bytes = loaded
                .get(key)
                .ok_or_else(|| ManifestJoinError::MissingBlob(key.clone()))?;
            let CallgraphBlob::Parse(blob) = CallgraphBlob::from_bytes(bytes)? else {
                continue;
            };
            let Ok(rel) = std::str::from_utf8(path.as_bytes()) else {
                unbound_non_utf8_paths.push(path.as_bytes().to_vec());
                continue;
            };
            let extract = blob.bind(rel, &paths)?;
            for (raw, bound) in blob.refs.iter().zip(&extract.raw_refs) {
                let ref_key = CallerRefKey {
                    caller_blob_key: key.clone(),
                    ref_ordinal: raw.ordinal,
                    caller_path: path.as_bytes().to_vec(),
                };
                if selected.is_none_or(|set| set.contains(&ref_key)) {
                    work.push((ref_key, (raw.kind, bound.clone())));
                }
            }
            extracts.insert(rel.to_string(), extract);
        }
        let files = extracts
            .iter()
            .map(|(path, extract)| {
                (
                    path.clone(),
                    super::DbFileIndex::from_extract(root, extract, &paths),
                )
            })
            .collect();
        let caller_data = extracts
            .iter()
            .map(|(path, extract)| (path.clone(), &extract.data))
            .collect();
        let mut index = ManifestProjectIndex::from_parts(
            root,
            files,
            caller_data,
            super::WorkspaceCratePrefixCache::default(),
            facts,
        );
        index.unbound_non_utf8_paths = unbound_non_utf8_paths;
        if !index.unbound_non_utf8_paths.is_empty() {
            log::warn!(
                "callgraph index left {} non-UTF-8 source paths unbound",
                index.unbound_non_utf8_paths.len()
            );
        }
        let mut result = Self {
            rows: BTreeSet::new(),
            resolution_order: Vec::new(),
            unbound_non_utf8_paths: index.unbound_non_utf8_paths.clone(),
        };
        work.sort_by(|a, b| (&a.0, a.1 .0).cmp(&(&b.0, b.1 .0)));
        for (key, (kind, raw)) in work {
            let resolved = super::resolve_ref(raw, &index)
                .map_err(|error| ManifestJoinError::Parse(error.to_string()))?;
            result.rows.insert(DerivedRow {
                caller_blob_key: key.caller_blob_key.clone(),
                ref_ordinal: key.ref_ordinal,
                caller_path: key.caller_path.clone(),
                kind,
                status: if resolved.target_file.is_some() {
                    ResolutionStatus::Resolved
                } else {
                    ResolutionStatus::Unresolved
                },
                target_path: resolved.target_file.map(String::into_bytes),
                target_symbol: resolved.target_symbol,
            });
            result.resolution_order.push(key);
        }
        Ok(result)
    }

    pub fn update(
        &self,
        previous_manifest: &Manifest,
        manifest: &Manifest,
        blobs: &impl ManifestBlobReader,
    ) -> Result<IncrementalJoinResult, ManifestJoinError> {
        let changed = changed_manifest_paths(previous_manifest, manifest);
        let full_re_resolve = changed.iter().any(|path| {
            manifest_resolution_input(previous_manifest, path)
                || manifest_resolution_input(manifest, path)
        });
        let loaded = manifest_payloads(manifest, blobs)?;
        let reader = |key: &BlobKey| loaded.get(key).cloned();
        let mut current_keys = BTreeSet::new();
        for (path, entry) in manifest.entries() {
            if std::str::from_utf8(path.as_bytes()).is_err() {
                continue;
            }
            let ManifestEntry::Regular { planes, .. } = entry else {
                continue;
            };
            let Some(key) = &planes.callgraph else {
                continue;
            };
            if let CallgraphBlob::Parse(blob) = CallgraphBlob::from_bytes(&loaded[key])? {
                current_keys.extend(blob.refs.iter().map(|raw| CallerRefKey {
                    caller_blob_key: key.clone(),
                    ref_ordinal: raw.ordinal,
                    caller_path: path.as_bytes().to_vec(),
                }));
            }
        }
        let previous_rows = self
            .rows
            .iter()
            .map(|row| (row.ref_key(), row))
            .collect::<BTreeMap<_, _>>();
        let selected = current_keys
            .iter()
            .filter(|key| {
                full_re_resolve
                    || changed.contains(&key.caller_path)
                    || previous_rows.get(*key).is_none_or(|row| {
                        row.target_path
                            .as_ref()
                            .is_some_and(|path| changed.contains(path))
                    })
            })
            .cloned()
            .collect::<BTreeSet<_>>();
        let facts = Rc::new(ManifestFacts {
            manifest,
            blobs: &reader,
        });
        let mut result =
            Self::from_facts(manifest, &loaded, Path::new("/"), facts, Some(&selected))?;
        result.rows.extend(
            self.rows
                .iter()
                .filter(|row| {
                    current_keys.contains(&row.ref_key()) && !selected.contains(&row.ref_key())
                })
                .cloned(),
        );
        result.resolution_order = result.rows.iter().map(DerivedRow::ref_key).collect();
        result.resolution_order.sort();
        Ok(IncrementalJoinResult {
            result,
            re_resolved: selected,
            full_re_resolve,
        })
    }
}

fn manifest_payloads(
    manifest: &Manifest,
    blobs: &impl ManifestBlobReader,
) -> Result<BTreeMap<String, Arc<[u8]>>, ManifestJoinError> {
    let mut loaded = BTreeMap::new();
    for (_, entry) in manifest.entries() {
        let ManifestEntry::Regular { planes, .. } = entry else {
            continue;
        };
        let Some(key) = &planes.callgraph else {
            continue;
        };
        if !loaded.contains_key(key) {
            let bytes = blobs
                .read_callgraph_blob(key)?
                .ok_or_else(|| ManifestJoinError::MissingBlob(key.clone()))?;
            loaded.insert(key.clone(), Arc::from(bytes));
        }
    }
    Ok(loaded)
}

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

/// Bound reference dependencies and resolver probes are generation-specific, unlike
/// parse blobs. The owner must invalidate these whenever a probed path changes.
#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
pub(crate) struct ViewBindingDependencies {
    // AST ordinals can collide for structural references. Vector positions are
    // unique and stable within the immutable caller blob used to validate reuse.
    pub references: BTreeMap<u32, BTreeSet<String>>,
    pub dependencies: BTreeSet<String>,
    #[serde(default)]
    pub consulted_facts: BTreeSet<(String, String)>,
    #[serde(default)]
    pub unattributed: bool,
    #[serde(default)]
    binding_facts: ConfigConsultations,
    #[serde(default)]
    resolution_facts: ConfigConsultations,
    binding_probes: BTreeSet<String>,
    resolved_dependencies: BTreeSet<String>,
    surface_queries: Vec<(ViewSurfaceQuery, String)>,
    #[serde(skip)]
    surface: Option<ViewFileSurface>,
}

impl ViewBindingDependencies {
    pub(crate) fn from_surface_json(payload: &str) -> Result<Self, serde_json::Error> {
        Ok(Self {
            surface: Some(serde_json::from_str(payload)?),
            ..Self::default()
        })
    }

    pub(crate) fn copy_surface_from(&mut self, source: &Self) {
        self.surface.clone_from(&source.surface);
    }

    pub(crate) fn surface_json(&self) -> Result<Option<String>, serde_json::Error> {
        self.surface.as_ref().map(serde_json::to_string).transpose()
    }
}

/// Inputs that determine a reference's target, excluding call-site identity.
/// Rust qualified imports are visible only after their declaration; the count
/// of visible imports identifies that monotone prefix even across nested uses.
#[derive(Hash, PartialEq, Eq)]
struct ViewResolutionBinding {
    kind: String,
    full_ref: Option<String>,
    short_name: Option<String>,
    visible_rust_imports: usize,
}

impl ViewResolutionBinding {
    fn new(raw: &super::RawRef, caller: &FileCallData) -> Self {
        Self {
            kind: raw.kind.clone(),
            full_ref: raw.full_ref.clone(),
            short_name: raw.short_name.clone(),
            visible_rust_imports: if caller.lang == LangId::Rust {
                caller
                    .import_block
                    .imports
                    .partition_point(|import| import.byte_range.start <= raw.byte_start)
            } else {
                0
            },
        }
    }
}

pub(crate) struct SelectedManifestJoin {
    pub result: JoinResult,
    pub bindings: BTreeMap<String, ViewBindingDependencies>,
    pub resolved_callers: BTreeSet<String>,
    pub rebuilt_surface_entries: usize,
    pub rebuilt_surface_paths: BTreeSet<String>,
    pub decoded_caller_blobs: usize,
    pub resolved_bindings: usize,
}

/// Compact, deterministic snapshot of one file's resolver index. It deliberately
/// excludes source, AST nodes, and call sites: only callers actually resolved need
/// those payloads. Membership probes invalidate module/reexport targets along with
/// bindings; source changes invalidate the snapshot through the manifest diff.
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
struct ViewFileSurface {
    language: String,
    exports: BTreeSet<String>,
    default_export: Option<String>,
    export_aliases: BTreeMap<String, String>,
    node_by_scoped: BTreeMap<String, String>,
    node_by_bare: BTreeMap<String, String>,
    node_kind_by_id: BTreeMap<String, String>,
    module_targets: BTreeMap<String, Option<String>>,
    declared_module_targets: BTreeMap<String, Option<String>>,
    reexports: Vec<(Option<String>, BTreeMap<String, String>, bool)>,
}

impl ViewFileSurface {
    fn capture(language: &str, index: &super::DbFileIndex) -> Self {
        Self {
            language: language.into(),
            exports: index.exports.iter().cloned().collect(),
            default_export: index.default_export.clone(),
            export_aliases: index.export_aliases.clone().into_iter().collect(),
            node_by_scoped: index.node_by_scoped.clone().into_iter().collect(),
            node_by_bare: index.node_by_bare.clone().into_iter().collect(),
            node_kind_by_id: index.node_kind_by_id.clone().into_iter().collect(),
            module_targets: index.module_targets.clone().into_iter().collect(),
            declared_module_targets: index.declared_module_targets.clone().into_iter().collect(),
            reexports: index
                .reexports
                .iter()
                .map(|r| {
                    (
                        r.target_file.clone(),
                        r.named.clone().into_iter().collect(),
                        r.wildcard,
                    )
                })
                .collect(),
        }
    }

    fn restore(&self) -> super::DbFileIndex {
        super::DbFileIndex {
            lang: language_id(&self.language),
            exports: self.exports.iter().cloned().collect(),
            default_export: self.default_export.clone(),
            export_aliases: self.export_aliases.clone().into_iter().collect(),
            node_by_scoped: self.node_by_scoped.clone().into_iter().collect(),
            node_by_bare: self.node_by_bare.clone().into_iter().collect(),
            node_kind_by_id: self.node_kind_by_id.clone().into_iter().collect(),
            module_targets: self.module_targets.clone().into_iter().collect(),
            declared_module_targets: self.declared_module_targets.clone().into_iter().collect(),
            reexports: self
                .reexports
                .iter()
                .map(|(target_file, named, wildcard)| super::ReexportIndex {
                    target_file: target_file.clone(),
                    named: named.clone().into_iter().collect(),
                    wildcard: *wildcard,
                })
                .collect(),
        }
    }
}

/// Configuration files read by the manifest resolver's workspace/package and
/// tsconfig lookup (callgraph.rs), and Rust crate lookup (callgraph_store).
/// Directory discovery uses a compact membership domain for these names. Their
/// content changes invalidate only callers that consulted changed fields.
pub(crate) fn view_resolution_config(path: &[u8]) -> bool {
    matches!(
        path.rsplit(|byte| *byte == b'/').next(),
        Some(b"package.json" | b"tsconfig.json" | b"pnpm-workspace.yaml" | b"Cargo.toml")
    )
}

/// Field identities are generation-owned. Raw-read paths are transient validation
/// evidence, cleared after classifying a caller so memo-hit timing is not persisted.
#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
struct ConfigConsultations {
    facts: BTreeSet<(String, String)>,
    reads: BTreeSet<String>,
    unattributed: bool,
}
impl ConfigConsultations {
    fn extend(&mut self, other: &Self) {
        self.facts.extend(other.facts.iter().cloned());
        self.reads.extend(other.reads.iter().cloned());
        self.unattributed |= other.unattributed;
    }
    fn unattributed(&self) -> bool {
        self.unattributed
            || self
                .reads
                .iter()
                .any(|path| !self.facts.iter().any(|(input, _)| input == path))
    }
}

#[derive(Clone, Default)]
struct ConsultationTrace {
    config: ConfigConsultations,
    probes: BTreeSet<String>,
}
type ConsultationMemoKey = (std::path::PathBuf, String, String);

/// Memo answers and their provenance belong to one immutable manifest join.
/// Replaying provenance on hits keeps missing candidates and facts attributable.
struct ViewBindingFacts<'a> {
    inner: Rc<dyn ProjectFacts + 'a>,
    probes: std::cell::RefCell<BTreeSet<String>>,
    consultations: std::cell::RefCell<ConfigConsultations>,
    memo_stack: std::cell::RefCell<Vec<(ConsultationMemoKey, ConsultationTrace)>>,
    memo_traces: std::cell::RefCell<BTreeMap<ConsultationMemoKey, ConsultationTrace>>,
    workspace_packages:
        std::cell::RefCell<BTreeMap<(std::path::PathBuf, String), Option<std::path::PathBuf>>>,
    workspace_members:
        std::cell::RefCell<BTreeMap<std::path::PathBuf, Arc<Vec<std::path::PathBuf>>>>,
    canonical_cache: std::cell::RefCell<HashMap<Vec<u8>, Option<Vec<u8>>>>,
    file_cache: std::cell::RefCell<HashMap<Vec<u8>, bool>>,
    config_cache: std::cell::RefCell<HashMap<Vec<u8>, Option<Arc<[u8]>>>>,
    directory_cache: std::cell::RefCell<HashMap<Vec<u8>, Vec<super::facts::DirEntry>>>,
}

impl<'a> ViewBindingFacts<'a> {
    fn new(inner: Rc<dyn ProjectFacts + 'a>) -> Self {
        Self {
            inner,
            probes: Default::default(),
            consultations: Default::default(),
            memo_stack: Default::default(),
            memo_traces: Default::default(),
            workspace_packages: Default::default(),
            workspace_members: Default::default(),
            canonical_cache: Default::default(),
            file_cache: Default::default(),
            config_cache: Default::default(),
            directory_cache: Default::default(),
        }
    }
    fn file_fact(&self, rel: &[u8]) -> bool {
        if let Some(value) = self.file_cache.borrow().get(rel) {
            return *value;
        }
        let value = self.inner.is_file(rel);
        self.file_cache.borrow_mut().insert(rel.to_vec(), value);
        value
    }

    fn record(&self, path: &[u8]) {
        // Directory discovery may probe thousands of missing package manifests.
        // One membership domain rechecks those bindings on config add/remove;
        // individual config content changes use the field consultations instead.
        if view_resolution_config(path) {
            self.record(VIEW_CONFIG_MEMBERSHIP_DOMAIN.as_bytes());
            return;
        }
        if let Ok(path) = std::str::from_utf8(path) {
            let mut parts = Vec::new();
            for part in path.split('/') {
                match part {
                    "" | "." => {}
                    ".." => {
                        parts.pop();
                    }
                    _ => parts.push(part),
                }
            }
            let path = parts.join("/");
            self.probes.borrow_mut().insert(path.clone());
            for (_, trace) in self.memo_stack.borrow_mut().iter_mut() {
                trace.probes.insert(path.clone());
            }
        }
    }

    fn take_config(&self) -> ConfigConsultations {
        let mut result = std::mem::take(&mut *self.consultations.borrow_mut());
        result.unattributed = result.unattributed();
        result.reads.clear();
        result
    }

    fn config_event(&self, path: &[u8], name: Option<&str>) {
        let Ok(path) = std::str::from_utf8(path) else {
            self.consultations.borrow_mut().unattributed = true;
            return;
        };
        let mut event = ConfigConsultations::default();
        if let Some(name) = name {
            event.facts.insert((path.into(), name.into()));
        } else {
            event.reads.insert(path.into());
        }
        self.consultations.borrow_mut().extend(&event);
        for (_, trace) in self.memo_stack.borrow_mut().iter_mut() {
            trace.config.extend(&event);
        }
    }

    fn take(&self) -> BTreeSet<String> {
        std::mem::take(&mut *self.probes.borrow_mut())
    }
}

impl ProjectFacts for ViewBindingFacts<'_> {
    fn records_config_facts(&self) -> bool {
        true
    }
    fn config_fact(&self, rel: &[u8], name: &str) {
        self.record(rel);
        if matches!(name, "workspaces" | "packages") {
            self.record(VIEW_CONFIG_MEMBERSHIP_DOMAIN.as_bytes());
        }
        self.config_event(rel, Some(name));
    }
    fn memo_start(&self, path: &Path, kind: &str, name: &str) {
        self.memo_stack.borrow_mut().push((
            (path.into(), kind.into(), name.into()),
            ConsultationTrace::default(),
        ));
    }
    fn memo_finish(&self, path: &Path, kind: &str, name: &str) {
        let (key, trace) = self
            .memo_stack
            .borrow_mut()
            .pop()
            .expect("balanced resolver memo recording");
        debug_assert_eq!(key, (path.into(), kind.into(), name.into()));
        self.memo_traces.borrow_mut().insert(key, trace);
    }
    fn memo_replay(&self, path: &Path, kind: &str, name: &str) {
        let traces = self.memo_traces.borrow();
        let Some(trace) = traces.get(&(path.into(), kind.into(), name.into())) else {
            self.consultations.borrow_mut().unattributed = true;
            return;
        };
        self.consultations.borrow_mut().extend(&trace.config);
        self.probes
            .borrow_mut()
            .extend(trace.probes.iter().cloned());
        for (_, parent) in self.memo_stack.borrow_mut().iter_mut() {
            parent.config.extend(&trace.config);
            parent.probes.extend(trace.probes.iter().cloned());
        }
    }
    fn workspace_package(&self, root: &Path, name: &str) -> Option<Option<std::path::PathBuf>> {
        self.workspace_packages
            .borrow()
            .get(&(root.into(), name.into()))
            .cloned()
    }
    fn remember_workspace_package(
        &self,
        root: &Path,
        name: &str,
        value: Option<std::path::PathBuf>,
    ) {
        self.workspace_packages
            .borrow_mut()
            .insert((root.into(), name.into()), value);
    }
    fn workspace_members(&self, root: &Path) -> Option<Arc<Vec<std::path::PathBuf>>> {
        self.workspace_members.borrow().get(root).cloned()
    }
    fn remember_workspace_members(&self, root: &Path, value: Arc<Vec<std::path::PathBuf>>) {
        self.workspace_members
            .borrow_mut()
            .insert(root.into(), value);
    }
    fn is_file(&self, rel: &[u8]) -> bool {
        let is_file = self.file_fact(rel);
        if is_file {
            self.record(rel);
        }
        is_file
    }
    fn is_dir(&self, rel: &[u8]) -> bool {
        self.inner.is_dir(rel)
    }
    fn config_bytes(&self, rel: &[u8]) -> Option<Arc<[u8]>> {
        self.consultations.borrow_mut().unattributed = true;
        for (_, trace) in self.memo_stack.borrow_mut().iter_mut() {
            trace.config.unattributed = true;
        }
        self.attributed_config_bytes(rel)
    }
    fn attributed_config_bytes(&self, rel: &[u8]) -> Option<Arc<[u8]>> {
        self.config_event(rel, None);
        self.record(rel);
        if let Some(value) = self.config_cache.borrow().get(rel) {
            return value.clone();
        }
        let value = self.inner.config_bytes(rel);
        self.config_cache
            .borrow_mut()
            .insert(rel.to_vec(), value.clone());
        value
    }
    fn symlink_target(&self, rel: &[u8]) -> Option<&[u8]> {
        self.inner.symlink_target(rel)
    }
    fn canonical(&self, rel: &[u8]) -> Option<Vec<u8>> {
        // FactPaths canonicalizes before testing existence, so misses must be
        // recorded here as well as in is_file (not only after canonicalization).
        let cached = self.canonical_cache.borrow().get(rel).cloned();
        let canonical = cached.unwrap_or_else(|| {
            let value = self.inner.canonical(rel);
            self.canonical_cache
                .borrow_mut()
                .insert(rel.to_vec(), value.clone());
            value
        });
        // Existing directory probes are workspace-discovery implementation detail.
        // Config add/remove rechecks that discovery through its membership domain;
        // source-file probes and misses remain caller-specific dependencies.
        if canonical.as_ref().is_none_or(|path| self.file_fact(path)) {
            self.record(rel);
        }
        canonical
    }
    fn list_dir(&self, rel: &[u8]) -> Vec<super::facts::DirEntry> {
        if let Some(value) = self.directory_cache.borrow().get(rel) {
            return value.clone();
        }
        let value = self.inner.list_dir(rel);
        self.directory_cache
            .borrow_mut()
            .insert(rel.to_vec(), value.clone());
        value
    }
}

/// Resolve selected callers against the complete symbol index. Cached binding
/// dependencies avoid resolving unchanged imports merely to rebuild that index.
/// `selected=None` is the cold path; otherwise the owner supplies the transitive
/// reverse-dependency closure and caches from the same base generation.
#[cfg(test)]
pub(crate) fn join_selected_manifest(
    manifest: &Manifest,
    blobs: &impl ManifestBlobReader,
    selected: Option<&BTreeSet<String>>,
    cached: &BTreeMap<String, ViewBindingDependencies>,
) -> Result<SelectedManifestJoin, ManifestJoinError> {
    join_manifest_with_surfaces(manifest, blobs, selected, cached, None)
}

/// Rebind only changed callers or callers that probed changed membership. Other
/// candidates replay their prior consumer-specific surface queries and retain
/// their reference rows when every answer is unchanged.
pub(crate) fn join_selected_manifest_reusing_surfaces(
    manifest: &Manifest,
    blobs: &impl ManifestBlobReader,
    selected: Option<&BTreeSet<String>>,
    cached: &BTreeMap<String, ViewBindingDependencies>,
    changed: &BTreeSet<String>,
    membership_changed: &BTreeSet<String>,
    fact_invalidated: &BTreeSet<String>,
) -> Result<SelectedManifestJoin, ManifestJoinError> {
    join_manifest_with_surfaces(
        manifest,
        blobs,
        selected,
        cached,
        Some((changed, membership_changed, fact_invalidated)),
    )
}

fn join_manifest_with_surfaces(
    manifest: &Manifest,
    blobs: &impl ManifestBlobReader,
    selected: Option<&BTreeSet<String>>,
    cached: &BTreeMap<String, ViewBindingDependencies>,
    reuse: Option<(&BTreeSet<String>, &BTreeSet<String>, &BTreeSet<String>)>,
) -> Result<SelectedManifestJoin, ManifestJoinError> {
    let mut profile = crate::views::materialization::profile::PhaseTimer::new("join");
    let loaded = std::cell::RefCell::new(BTreeMap::<BlobKey, Arc<[u8]>>::new());
    let load = |key: &BlobKey| -> Result<Arc<[u8]>, ManifestJoinError> {
        if let Some(bytes) = loaded.borrow().get(key) {
            return Ok(bytes.clone());
        }
        let bytes: Arc<[u8]> = blobs
            .read_callgraph_blob(key)?
            .ok_or_else(|| ManifestJoinError::MissingBlob(key.clone()))?
            .into();
        loaded.borrow_mut().insert(key.clone(), bytes.clone());
        Ok(bytes)
    };
    let decoded = std::cell::RefCell::new(HashMap::<BlobKey, Arc<CallgraphBlob>>::new());
    let decode = |key: &BlobKey| -> Result<Arc<CallgraphBlob>, ManifestJoinError> {
        if let Some(blob) = decoded.borrow().get(key) {
            return Ok(blob.clone());
        }
        let blob = blobs
            .read_callgraph_blob_decoded(key)?
            .ok_or_else(|| ManifestJoinError::MissingBlob(key.clone()))?;
        decoded.borrow_mut().insert(key.clone(), blob.clone());
        Ok(blob)
    };
    let read_error = std::cell::RefCell::new(None);
    let reader = |key: &BlobKey| match load(key) {
        Ok(bytes) => Some(bytes),
        Err(error) => {
            *read_error.borrow_mut() = Some(error);
            None
        }
    };
    profile.finish("load_payloads");
    let facts = Rc::new(ViewBindingFacts::new(Rc::new(ManifestFacts {
        manifest,
        blobs: &reader,
    })));
    let root = Path::new("/");
    let paths = FactPaths {
        root,
        facts: facts.as_ref(),
    };
    let mut extracts = HashMap::new();
    let mut files = HashMap::new();
    let mut work: BTreeMap<String, Vec<_>> = BTreeMap::new();
    let mut bindings = BTreeMap::new();
    let mut unbound_non_utf8_paths = Vec::new();
    let mut rebuilt_surface_paths = BTreeSet::new();
    let mut decoded_caller_blobs = 0;
    let mut restored_surfaces = 0;
    let mut rebuilt_changed = 0;
    let mut rebuilt_facts = 0;
    let mut rebuilt_membership = 0;
    let mut rebuilt_missing = 0;
    for (path, entry) in manifest.entries() {
        let ManifestEntry::Regular { planes, .. } = entry else {
            continue;
        };
        let Some(key) = &planes.callgraph else {
            continue;
        };
        let Ok(rel) = std::str::from_utf8(path.as_bytes()) else {
            if matches!(decode(key)?.as_ref(), CallgraphBlob::Parse(_)) {
                unbound_non_utf8_paths.push(path.as_bytes().to_vec());
            }
            continue;
        };
        let resolve = selected.is_none_or(|set| set.contains(rel));
        let cache = cached.get(rel).filter(|cache| {
            if let Some((changed, membership, fact_invalidated)) = reuse {
                selected.is_some()
                    && !changed.contains(rel)
                    && !fact_invalidated.contains(rel)
                    && cache.dependencies.is_disjoint(membership)
            } else {
                !resolve
            }
        });
        if let Some((cache, surface)) =
            cache.and_then(|cache| cache.surface.as_ref().map(|surface| (cache, surface)))
        {
            restored_surfaces += 1;
            files.insert(rel.to_string(), surface.restore());
            if resolve {
                bindings.insert(rel.to_string(), cache.clone());
            }
            continue;
        }
        let decoded_blob = decode(key)?;
        let CallgraphBlob::Parse(blob) = decoded_blob.as_ref() else {
            continue;
        };
        decoded_caller_blobs += 1;
        facts.take();
        facts.take_config();
        let extract =
            blob.bind_with_dependencies(rel, &paths, cache.map(|cache| &cache.references))?;
        let file_index = super::DbFileIndex::from_extract(root, &extract, &paths);
        rebuilt_surface_paths.insert(rel.to_string());
        match reuse {
            Some((changed, _, _)) if changed.contains(rel) => rebuilt_changed += 1,
            Some((_, _, invalidated)) if invalidated.contains(rel) => rebuilt_facts += 1,
            Some((_, membership, _))
                if cached
                    .get(rel)
                    .is_some_and(|old| !old.dependencies.is_disjoint(membership)) =>
            {
                rebuilt_membership += 1;
            }
            _ => rebuilt_missing += 1,
        }
        let mut binding = cache.cloned().unwrap_or_default();
        binding.surface = Some(ViewFileSurface::capture(&blob.language, &file_index));
        files.insert(rel.to_string(), file_index);
        if cache.is_none() {
            binding.references = blob
                .refs
                .iter()
                .zip(&extract.raw_refs)
                .enumerate()
                .map(|(position, (_, bound))| {
                    (
                        u32::try_from(position).expect("reference vector fits u32"),
                        bound.dependencies.clone(),
                    )
                })
                .collect();
            binding.binding_probes = facts.take();
            binding.binding_facts = facts.take_config();
            binding.dependencies = binding.references.values().flatten().cloned().collect();
            binding
                .dependencies
                .extend(binding.binding_probes.iter().cloned());
        }
        bindings.insert(rel.to_string(), binding);
        if resolve {
            let caller_work = work.entry(rel.to_string()).or_default();
            for (raw, bound) in blob.refs.iter().zip(&extract.raw_refs) {
                caller_work.push((
                    CallerRefKey {
                        caller_blob_key: key.clone(),
                        ref_ordinal: raw.ordinal,
                        caller_path: path.as_bytes().to_vec(),
                    },
                    (raw.kind, bound.clone()),
                ));
            }
        }
        extracts.insert(rel.to_string(), extract);
    }
    profile.finish("decode_bind_index_entries");
    let index = ManifestProjectIndex::from_parts(
        root,
        files,
        HashMap::new(),
        super::WorkspaceCratePrefixCache::default(),
        facts.clone(),
    );
    let mut result = JoinResult {
        rows: BTreeSet::new(),
        resolution_order: Vec::new(),
        unbound_non_utf8_paths,
    };
    let resolved_callers = bindings
        .keys()
        .filter(|path| {
            if selected.is_some_and(|set| !set.contains(*path)) {
                return false;
            }
            let Some((changed, _, _)) = reuse else {
                return true;
            };
            if selected.is_none() || changed.contains(*path) {
                return true;
            }
            cached.get(*path).is_none_or(|old| {
                old.references != bindings[*path].references
                    || old
                        .surface_queries
                        .iter()
                        .any(|(query, expected)| query.answer(&index) != *expected)
            })
        })
        .cloned()
        .collect::<BTreeSet<_>>();
    for (path, binding) in &mut bindings {
        if resolved_callers.contains(path) {
            binding.resolved_dependencies.clear();
            binding.surface_queries.clear();
            binding.resolution_facts = ConfigConsultations::default();
        } else if let Some(old) = cached.get(path) {
            binding.resolved_dependencies = old.resolved_dependencies.clone();
            binding.surface_queries = old.surface_queries.clone();
        }
    }
    profile.finish("index_and_surface_replay");
    // Surface replay needs no call sites. Decode an unchanged caller only after
    // replay proves that its reference results may change.
    for caller in &resolved_callers {
        if extracts.contains_key(caller) {
            continue;
        }
        let path = RelPath::new(caller.as_bytes().to_vec()).expect("bound manifest path");
        let Some(ManifestEntry::Regular { planes, .. }) = manifest.get(&path) else {
            continue;
        };
        let key = planes.callgraph.as_ref().expect("bound caller key");
        let decoded_blob = decode(key)?;
        let CallgraphBlob::Parse(blob) = decoded_blob.as_ref() else {
            continue;
        };
        decoded_caller_blobs += 1;
        facts.take();
        facts.take_config();
        let extract =
            blob.bind_with_dependencies(caller, &paths, Some(&bindings[caller].references))?;
        let caller_work = work.entry(caller.clone()).or_default();
        for (raw, bound) in blob.refs.iter().zip(&extract.raw_refs) {
            caller_work.push((
                CallerRefKey {
                    caller_blob_key: key.clone(),
                    ref_ordinal: raw.ordinal,
                    caller_path: caller.as_bytes().to_vec(),
                },
                (raw.kind, bound.clone()),
            ));
        }
        extracts.insert(caller.clone(), extract);
    }
    let index = ManifestProjectIndex::from_parts(
        root,
        index.files,
        extracts
            .iter()
            .map(|(path, extract)| (path.clone(), &extract.data))
            .collect(),
        super::WorkspaceCratePrefixCache::default(),
        facts.clone(),
    );
    profile.finish("decode_resolved_callers");
    let surface_index = ViewSurfaceIndex {
        inner: &index,
        queries: Default::default(),
    };
    let bases: BTreeMap<_, BTreeSet<_>> = resolved_callers
        .iter()
        .map(|caller| {
            let binding = &bindings[caller];
            (
                caller.clone(),
                binding
                    .references
                    .values()
                    .flatten()
                    .chain(binding.binding_probes.iter())
                    .cloned()
                    .collect(),
            )
        })
        .collect();
    let mut resolved_bindings = 0;
    for (caller, mut caller_work) in work {
        if !resolved_callers.contains(&caller) {
            continue;
        }
        caller_work.sort_by(|a, b| (&a.0, a.1 .0).cmp(&(&b.0, b.1 .0)));
        let caller_data = &extracts[&caller].data;
        let basis = &bases[&caller];
        let binding = bindings
            .get_mut(&caller)
            .expect("bound caller dependencies");
        let mut resolutions =
            HashMap::<ViewResolutionBinding, (Option<String>, Option<String>)>::new();
        let mut caller_queries = BTreeMap::new();
        for (key, (kind, raw)) in caller_work {
            let memo_key = ViewResolutionBinding::new(&raw, caller_data);
            // Dependencies belonging to a call site are not part of the memoized
            // target. Preserve them even when another reference resolved its binding.
            binding.resolved_dependencies.extend(
                raw.dependencies
                    .iter()
                    .filter(|dependency| !basis.contains(*dependency))
                    .cloned(),
            );
            let (target_file, target_symbol) = match resolutions.entry(memo_key) {
                std::collections::hash_map::Entry::Occupied(entry) => entry.get().clone(),
                std::collections::hash_map::Entry::Vacant(entry) => {
                    facts.take();
                    facts.take_config();
                    let resolved = super::resolve_ref(raw, &surface_index)
                        .map_err(|error| ManifestJoinError::Parse(error.to_string()))?;
                    // The memo table is caller-local, so one consultation union per
                    // binding preserves caller ownership on cache hits.
                    binding.resolution_facts.extend(&facts.take_config());
                    caller_queries.extend(surface_index.take());
                    binding.resolved_dependencies.extend(
                        resolved
                            .dependencies
                            .into_iter()
                            .chain(facts.take())
                            .filter(|dependency| !basis.contains(dependency)),
                    );
                    entry
                        .insert((resolved.target_file, resolved.target_symbol))
                        .clone()
                }
            };
            result.rows.insert(DerivedRow {
                caller_blob_key: key.caller_blob_key.clone(),
                ref_ordinal: key.ref_ordinal,
                caller_path: key.caller_path.clone(),
                kind,
                status: if target_file.is_some() {
                    ResolutionStatus::Resolved
                } else {
                    ResolutionStatus::Unresolved
                },
                target_path: target_file.map(String::into_bytes),
                target_symbol,
            });
            result.resolution_order.push(key);
        }
        resolved_bindings += resolutions.len();
        binding.surface_queries = caller_queries.into_iter().collect();
    }
    result.resolution_order.sort();
    profile.finish("resolve_and_record");
    for binding in bindings.values_mut() {
        binding.consulted_facts = binding
            .binding_facts
            .facts
            .union(&binding.resolution_facts.facts)
            .cloned()
            .collect();
        binding.unattributed =
            binding.binding_facts.unattributed() || binding.resolution_facts.unattributed();
        binding.dependencies = binding
            .references
            .values()
            .flatten()
            .cloned()
            .chain(binding.binding_probes.iter().cloned())
            .chain(binding.resolved_dependencies.iter().cloned())
            .chain(
                binding
                    .surface_queries
                    .iter()
                    .flat_map(|(query, _)| query.dependencies()),
            )
            .collect();
    }
    profile.finish("dependency_union");
    if std::env::var_os("AFT_VIEW_PROFILE").is_some() {
        eprintln!("view_profile join.surfaces restored={restored_surfaces} rebuilt_changed={rebuilt_changed} rebuilt_facts={rebuilt_facts} rebuilt_membership={rebuilt_membership} rebuilt_missing={rebuilt_missing} selected={} resolved={} skipped={}", selected.map_or(bindings.len(), BTreeSet::len), resolved_callers.len(), bindings.keys().filter(|path| selected.is_none_or(|set| set.contains(*path)) && !resolved_callers.contains(*path)).count());
    }
    if let Some(error) = read_error.borrow_mut().take() {
        return Err(error);
    }
    Ok(SelectedManifestJoin {
        result,
        bindings,
        resolved_callers,
        rebuilt_surface_entries: rebuilt_surface_paths.len(),
        rebuilt_surface_paths,
        decoded_caller_blobs,
        resolved_bindings,
    })
}

/// Consumer-specific export surface: record the answers the resolver actually
/// used, rather than invalidating every caller when an unrelated export changes.
/// Querying these answers against a new index is cheaper than walking references
/// and is sound only while the caller's immutable parse blob remains unchanged.
#[derive(Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
enum ViewSurfaceQuery {
    Language(String),
    Module(String, String),
    Parent(String),
    Reexports(String),
    Node(String, String),
    Callable(String, String),
    Alias(String, String),
    Export(String, String),
    Default(String),
    Contains(String),
    Crate(String),
    CrateRoot(String),
    Inline(String, Vec<String>, String),
}

fn surface_value(value: &impl Serialize) -> String {
    blake3::hash(&serde_json::to_vec(value).expect("resolver surface serializes"))
        .to_hex()
        .to_string()
}

fn reexport_surface(value: &[super::ReexportIndex]) -> String {
    surface_value(
        &value
            .iter()
            .map(|entry| {
                (
                    &entry.target_file,
                    entry.named.iter().collect::<BTreeMap<_, _>>(),
                    entry.wildcard,
                )
            })
            .collect::<Vec<_>>(),
    )
}

pub(crate) const VIEW_CONFIG_MEMBERSHIP_DOMAIN: &str = "\0view:config-membership";
pub(crate) const VIEW_RUST_MODULE_DOMAIN: &str = "\0view:rust-module-index";

impl ViewSurfaceQuery {
    fn dependencies(&self) -> Vec<String> {
        match self {
            Self::Crate(_) => vec![VIEW_CONFIG_MEMBERSHIP_DOMAIN.into()],
            Self::CrateRoot(file) => {
                vec![file.clone(), VIEW_CONFIG_MEMBERSHIP_DOMAIN.into()]
            }
            Self::Parent(file) | Self::Inline(file, ..) => {
                vec![file.clone(), VIEW_RUST_MODULE_DOMAIN.into()]
            }
            Self::Language(file)
            | Self::Module(file, _)
            | Self::Reexports(file)
            | Self::Node(file, _)
            | Self::Callable(file, _)
            | Self::Alias(file, _)
            | Self::Export(file, _)
            | Self::Default(file)
            | Self::Contains(file) => vec![file.clone()],
        }
    }

    fn answer(&self, index: &impl super::ResolverIndex) -> String {
        match self {
            Self::Language(file) => {
                surface_value(&index.lang_for(file).map(|lang| format!("{lang:?}")))
            }
            Self::Module(file, module) => surface_value(&index.module_target(file, module)),
            Self::Parent(file) => surface_value(&index.module_parent(file)),
            Self::Reexports(file) => reexport_surface(&index.reexports_for(file)),
            Self::Node(file, symbol) => surface_value(&index.node_for_symbol(file, symbol)),
            Self::Callable(file, node) => surface_value(&index.node_is_callable(file, node)),
            Self::Alias(file, symbol) => surface_value(&index.export_alias(file, symbol)),
            Self::Export(file, symbol) => surface_value(&index.has_export(file, symbol)),
            Self::Default(file) => surface_value(&index.default_export(file)),
            Self::Contains(file) => surface_value(&index.contains_file(file)),
            Self::Crate(name) => surface_value(&index.crate_src_prefix(name)),
            Self::CrateRoot(file) => surface_value(&index.rust_crate_root_file(file)),
            Self::Inline(file, segments, symbol) => {
                surface_value(&index.inline_scoped_target(file, segments, symbol))
            }
        }
    }
}

struct ViewSurfaceIndex<'a, I> {
    inner: &'a I,
    queries: std::cell::RefCell<BTreeMap<ViewSurfaceQuery, String>>,
}

impl<I> ViewSurfaceIndex<'_, I> {
    fn record(&self, query: ViewSurfaceQuery, value: &impl Serialize) {
        self.queries
            .borrow_mut()
            .insert(query, surface_value(value));
    }
    fn take(&self) -> BTreeMap<ViewSurfaceQuery, String> {
        std::mem::take(&mut *self.queries.borrow_mut())
    }
}

impl<I: super::ResolverIndex> super::ResolverIndex for ViewSurfaceIndex<'_, I> {
    fn caller_data(&self, file: &str) -> Option<&FileCallData> {
        // resolve_ref reads only its own caller data; changed callers never reuse
        // surface answers, so the immutable blob itself guards this input.
        self.inner.caller_data(file)
    }
    fn lang_for(&self, file: &str) -> Option<LangId> {
        let value = self.inner.lang_for(file);
        self.record(
            ViewSurfaceQuery::Language(file.into()),
            &value.map(|lang| format!("{lang:?}")),
        );
        value
    }
    fn module_target(&self, file: &str, module: &str) -> Option<String> {
        let value = self.inner.module_target(file, module);
        self.record(ViewSurfaceQuery::Module(file.into(), module.into()), &value);
        value
    }
    fn module_parent(&self, file: &str) -> Option<(String, String)> {
        let value = self.inner.module_parent(file);
        self.record(ViewSurfaceQuery::Parent(file.into()), &value);
        value
    }
    fn reexports_for(&self, file: &str) -> Vec<super::ReexportIndex> {
        let value = self.inner.reexports_for(file);
        self.queries.borrow_mut().insert(
            ViewSurfaceQuery::Reexports(file.into()),
            reexport_surface(&value),
        );
        value
    }
    fn node_for_symbol(&self, file: &str, symbol: &str) -> Option<String> {
        let value = self.inner.node_for_symbol(file, symbol);
        self.record(ViewSurfaceQuery::Node(file.into(), symbol.into()), &value);
        value
    }
    fn node_is_callable(&self, file: &str, node: &str) -> bool {
        let value = self.inner.node_is_callable(file, node);
        self.record(ViewSurfaceQuery::Callable(file.into(), node.into()), &value);
        value
    }
    fn export_alias(&self, file: &str, symbol: &str) -> Option<String> {
        let value = self.inner.export_alias(file, symbol);
        self.record(ViewSurfaceQuery::Alias(file.into(), symbol.into()), &value);
        value
    }
    fn has_export(&self, file: &str, symbol: &str) -> bool {
        let value = self.inner.has_export(file, symbol);
        self.record(ViewSurfaceQuery::Export(file.into(), symbol.into()), &value);
        value
    }
    fn default_export(&self, file: &str) -> Option<String> {
        let value = self.inner.default_export(file);
        self.record(ViewSurfaceQuery::Default(file.into()), &value);
        value
    }
    fn contains_file(&self, file: &str) -> bool {
        let value = self.inner.contains_file(file);
        self.record(ViewSurfaceQuery::Contains(file.into()), &value);
        value
    }
    fn crate_src_prefix(&self, name: &str) -> Option<String> {
        let value = self.inner.crate_src_prefix(name);
        self.record(ViewSurfaceQuery::Crate(name.into()), &value);
        value
    }
    fn rust_crate_root_file(&self, file: &str) -> Option<String> {
        let value = self.inner.rust_crate_root_file(file);
        self.record(ViewSurfaceQuery::CrateRoot(file.into()), &value);
        value
    }
    fn inline_scoped_target(
        &self,
        file: &str,
        segments: &[String],
        symbol: &str,
    ) -> Option<(String, String)> {
        let value = self.inner.inline_scoped_target(file, segments, symbol);
        self.record(
            ViewSurfaceQuery::Inline(file.into(), segments.to_vec(), symbol.into()),
            &value,
        );
        value
    }
}

#[cfg(test)]
mod consultation_tests {
    use super::*;

    #[test]
    fn opaque_config_read_is_unattributed_even_after_a_known_field_read() {
        let manifest = Manifest::new([(
            RelPath::new(b"package.json".to_vec()).unwrap(),
            ManifestEntry::Regular {
                mode: 0o100644,
                planes: crate::views::RegularPlanes {
                    callgraph: Some("config".into()),
                    semantic: None,
                },
                resolution_input: true,
            },
        )])
        .unwrap();
        let bytes: Arc<[u8]> = CallgraphBlob::config(b"{}".to_vec(), "fixture")
            .to_bytes()
            .unwrap()
            .into();
        let reader = |_: &BlobKey| Some(bytes.clone());
        let facts = ViewBindingFacts::new(Rc::new(ManifestFacts {
            manifest: &manifest,
            blobs: &reader,
        }));
        facts.config_fact(b"package.json", "name");
        assert!(facts.attributed_config_bytes(b"package.json").is_some());
        assert!(!facts.consultations.borrow().unattributed());
        facts.memo_start(Path::new("/"), "test", "");
        assert!(facts.config_bytes(b"package.json").is_some());
        facts.memo_finish(Path::new("/"), "test", "");
        assert!(facts.take_config().unattributed());
        facts.memo_replay(Path::new("/"), "test", "");
        assert!(facts.take_config().unattributed());
    }
}

#[cfg(test)]
thread_local! {
    static CALLER_NODE_LOOKUP_WORK: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
    // Select the original caller-node scan in tests so the benchmark can
    // compare scan and indexed lookup in the same process and host-load window.
    pub(crate) static SCAN_CALLER_NODES: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
}

#[test]
fn binding_caller_node_lookup_is_linear_and_preserves_first_duplicate() {
    let source = (0..200)
        .map(|i| format!("export function f{i}() {{ f0(); }}\n"))
        .collect::<String>();
    let blob = CallgraphBlob::extract(&source, "typescript", "lookup-test").unwrap();
    let mut parse = blob.parse().unwrap().clone();
    let mut duplicate = parse.symbols[0].clone();
    duplicate.ordinal += 10000;
    parse.symbols.push(duplicate);
    let manifest = Manifest::new([]).unwrap();
    let reader = |_: &BlobKey| None;
    let facts = ManifestFacts {
        manifest: &manifest,
        blobs: &reader,
    };
    let paths = FactPaths {
        root: Path::new("/"),
        facts: &facts,
    };
    CALLER_NODE_LOOKUP_WORK.with(|work| work.set(0));
    let extract = parse
        .bind_with_dependencies("caller.ts", &paths, None)
        .unwrap();
    let work = CALLER_NODE_LOOKUP_WORK.with(std::cell::Cell::get);
    assert!(work > 0, "lookup work must actually be observed");
    assert!(
        work <= parse.symbols.len() + parse.refs.len(),
        "caller node lookup must be linear: {work} operations for {} symbols and {} refs",
        parse.symbols.len(),
        parse.refs.len()
    );
    for reference in &extract.raw_refs {
        let expected = reference.caller_symbol.as_ref().and_then(|caller| {
            extract
                .nodes
                .iter()
                .find(|node| &node.scoped_name == caller)
                .map(|node| node.id.clone())
        });
        assert_eq!(reference.caller_node, expected);
    }
}