mir_analyzer/batch/

run.rs

use super::*;

impl AnalysisSession {
    /// Run the full batch analysis pipeline on a set of file paths.
    pub fn analyze_paths(&self, paths: &[PathBuf], opts: &BatchOptions) -> AnalysisResult {
        let php_version = self.batch_php_version(opts);
        let mut all_issues = Vec::new();
        let _t0 = std::time::Instant::now();

        // ---- Load PHP built-in stubs (before definition collection so user code can override)
        self.load_batch_stubs(php_version);
        // Index vendor autoload.files (global function/constant helpers such as
        // Laravel's `confirm()`, `select()`, etc.) before body analysis so
        // calls to these functions resolve rather than emitting UndefinedFunction.
        self.ensure_vendor_eager_functions();
        let _t_stubs = _t0.elapsed();

        // ---- Read files in parallel ----------------------------------
        let parsed_files: Vec<ParsedProjectFile> = paths
            .par_iter()
            .filter_map(|path| match std::fs::read_to_string(path) {
                Ok(src) => {
                    let file = Arc::from(path.to_string_lossy().as_ref());
                    Some(ParsedProjectFile::new(file, Arc::from(src)))
                }
                Err(e) => {
                    eprintln!("Cannot read {}: {}", path.display(), e);
                    None
                }
            })
            .collect();
        let _t_read = _t0.elapsed();

        let file_data: Vec<(Arc<str>, Arc<str>)> = parsed_files
            .iter()
            .map(|parsed| (parsed.file.clone(), parsed.source.clone()))
            .collect();

        // ---- Detect files deleted since the last run ------------------------
        // A file analyzed previously but now gone from disk leaves dependents
        // holding results that assume its definitions still exist. (A file
        // merely absent from this run's path set but still on disk is NOT a
        // deletion — checking disk existence avoids evicting during partial-path
        // analysis.) Drop their own entries here; dependent eviction happens
        // below, once per-file surface fingerprints are known.
        //
        // `topology_changed` tracks whether this run touched the cache (any file
        // changed, added, or removed). When nothing changed, the reverse-dep
        // graph loaded from disk is still accurate, so the rebuild + full
        // cache.bin rewrite at the end of this pass is skipped.
        let mut topology_changed = false;
        let mut removed_files: Vec<String> = Vec::new();
        if let Some(cache) = &self.cache {
            let current: rustc_hash::FxHashSet<&str> =
                file_data.iter().map(|(f, _)| f.as_ref()).collect();
            removed_files = cache
                .cached_files()
                .into_iter()
                .filter(|f| !current.contains(f.as_str()) && !std::path::Path::new(f).exists())
                .collect();
            for f in &removed_files {
                cache.evict(f);
            }
            topology_changed = !removed_files.is_empty();
        }

        // ---- Register Salsa source inputs for incremental follow-up calls ----
        {
            let mut guard = self.db.salsa.write();
            for parsed in &parsed_files {
                guard.upsert_source_file(parsed.file.clone(), parsed.source.clone());
            }
        }
        let _t_salsa_reg = _t0.elapsed();

        // ---- Definition collection from the already-parsed AST -------
        // Returns (FileDefinitions, content_hash, has_hard_parse_errors) so we
        // can prime the parse cache before the pre-warm loop below.
        type Pass1Entry = (FileDefinitions, [u8; 32], bool, String);
        let file_defs: Vec<Pass1Entry> = parsed_files
            .par_iter()
            .map(|parsed| {
                let content_hash = hash_source(parsed.source());
                let has_hard_parse_errors = parsed
                    .errors()
                    .iter()
                    .any(crate::parser::is_hard_parse_error);
                let mut all_issues: Vec<Issue> = parsed
                    .errors()
                    .iter()
                    .filter(|err| !crate::parser::is_spurious_reserved_class_error(err))
                    .map(|err| {
                        crate::parser::parse_error_to_issue(
                            err,
                            &parsed.file,
                            parsed.source(),
                            parsed.source_map(),
                        )
                    })
                    .collect();
                let collector = crate::collector::DefinitionCollector::new_for_slice(
                    parsed.file.clone(),
                    parsed.source(),
                    parsed.source_map(),
                );
                let (mut slice, collector_issues) = collector.collect_slice(parsed.owned());
                all_issues.extend(collector_issues);
                mir_codebase::definitions::deduplicate_params_in_slice(&mut slice);
                let defs = FileDefinitions {
                    slice: Arc::new(slice),
                    issues: Arc::new(all_issues),
                };
                let surface_hash = surface_fingerprint(parsed.source(), parsed.owned());
                (defs, content_hash, has_hard_parse_errors, surface_hash)
            })
            .collect();
        let _t_collect_defs = _t0.elapsed();

        // Pair each file with its cross-file surface fingerprint (par_iter().map
        // preserves order, so file_defs aligns with parsed_files).
        let surface_hashes: HashMap<Arc<str>, String> = parsed_files
            .iter()
            .zip(file_defs.iter())
            .map(|(parsed, (_defs, _h, _e, surface))| (parsed.file.clone(), surface.clone()))
            .collect();

        // Hex content hashes derived from the pass-1 digests, so the
        // content-changed pass and the body pass below don't re-hash every
        // source file (BLAKE3 over all bytes, twice).
        let content_hexes: HashMap<Arc<str>, String> = parsed_files
            .iter()
            .zip(file_defs.iter())
            .map(|(parsed, (_defs, hash, _e, _surface))| {
                (
                    parsed.file.clone(),
                    blake3::Hash::from(*hash).to_hex().to_string(),
                )
            })
            .collect();
        let transient_replay = self
            .cache
            .is_none()
            .then(|| {
                self.transient_batch_replay(
                    php_version.cache_byte(),
                    opts.skip_symbols,
                    &content_hexes,
                )
            })
            .flatten();

        // ---- Cross-file invalidation: evict dependents whose surface changed --
        // A file whose content changed re-analyzes itself regardless (its body
        // pass misses the cache below). It cascades to dependents only when its
        // declaration-level surface changed: a body-only edit to a declared-
        // return callable leaves every dependent's result intact. A missing or
        // pre-firewall (empty) stored surface is treated as unknown and cascades.
        if let Some(cache) = &self.cache {
            let content_changed: Vec<String> = file_data
                .iter()
                .filter_map(|(f, _src)| {
                    let valid = content_hexes.get(f).is_some_and(|h| cache.is_valid(f, h));
                    if valid {
                        None
                    } else {
                        Some(f.to_string())
                    }
                })
                .collect();
            if !content_changed.is_empty() {
                topology_changed = true;
            }
            let mut seeds: Vec<String> = content_changed
                .into_iter()
                .filter(|f| {
                    let new_surface = surface_hashes
                        .get(f.as_str())
                        .map(String::as_str)
                        .unwrap_or("");
                    // Keep (cascade) unless the stored surface is known and equal.
                    !matches!(
                        cache.surface_hash(f),
                        Some(stored) if !stored.is_empty() && stored == new_surface
                    )
                })
                .collect();
            seeds.extend(removed_files.iter().cloned());
            if !seeds.is_empty() {
                cache.evict_with_dependents(&seeds);
            }
        }

        // Prime the in-process parse cache so the pre-warm loop below avoids
        // re-parsing every project file through collect_file_definitions.
        {
            let guard = self.db.salsa.read();
            let php_v = php_version.cache_byte();
            for (defs, hash, has_hard_parse_errors, _surface) in &file_defs {
                if !*has_hard_parse_errors {
                    guard.prime_parse_cache(
                        *hash,
                        php_v,
                        Arc::clone(&defs.slice),
                        Arc::clone(&defs.issues),
                    );
                }
            }
        }

        let mut files_with_parse_errors: HashSet<Arc<str>> = HashSet::default();
        {
            // Commit subtype-index class edges alongside issue collection —
            // parity with `ingest_file`'s single-file path, so goto-implementation
            // sees implementors from a batch/vendor run without waiting for
            // each file to be individually touched by an on-demand commit path.
            let guard = self.db.salsa.read();
            for (parsed, (defs, _hash, _hard_err, _surface)) in parsed_files.iter().zip(file_defs) {
                for issue in defs.issues.iter() {
                    if matches!(issue.kind, mir_issues::IssueKind::ParseError { .. })
                        && issue.severity == mir_issues::Severity::Error
                    {
                        files_with_parse_errors.insert(issue.location.file.clone());
                    }
                }
                let entries = crate::db::subtype_index::entries_from_slice(&defs.slice);
                guard.set_file_class_edges(&parsed.file, entries);
                all_issues.extend(Arc::unwrap_or_clone(defs.issues));
            }
        }
        let _t_ingest = _t0.elapsed();

        // ---- Pre-warm collect_file_definitions for project files -------------
        {
            let db_prewarm = {
                let guard = self.db.salsa.read();
                (**guard).clone()
            };
            let project_source_files: Vec<SourceFile> = {
                let guard = self.db.salsa.read();
                parsed_files
                    .iter()
                    .filter_map(|p| (**guard).lookup_source_file(&p.file))
                    .collect()
            };
            project_source_files
                .into_par_iter()
                .for_each_with(db_prewarm, |db, sf| {
                    let _ = collect_file_definitions(db as &dyn MirDatabase, sf);
                });
        }
        let _t_prewarm_ms = (_t0.elapsed() - _t_ingest).as_secs_f64() * 1000.0;

        // Fold the freshly-registered project files into the workspace symbol
        // index singleton. The singleton may have been built from vendor before
        // this run (CLI indexes vendor before analyze_paths); since adding files
        // no longer nulls it, project classes would otherwise be invisible to
        // find_class_like and reported as false UndefinedClass.
        self.refresh_workspace_index();

        // ---- Lazy-load unknown classes via PSR-4 ----------------------------
        let _t_before_lazy = _t0.elapsed();
        if let Some(psr4) = self.psr4.clone() {
            self.lazy_load_missing_classes(psr4, php_version, &mut all_issues);
        }
        let _t_lazyload_ms = (_t0.elapsed() - _t_before_lazy).as_secs_f64() * 1000.0;

        // ---- Class-level checks ---------------------------------------------
        let analyzed_file_set: HashSet<Arc<str>> =
            file_data.iter().map(|(f, _)| f.clone()).collect();

        // Definitions are all collected and indexed — fire the plugin hook
        // (Psalm's AfterCodebasePopulated) before any body/class analysis.
        if let Some(plugins) = mir_plugin::snapshot() {
            if plugins.hooks().after_codebase_populated {
                let files: Vec<Arc<str>> = analyzed_file_set.iter().cloned().collect();
                plugins.after_codebase_populated(&mut mir_plugin::AfterCodebasePopulatedEvent {
                    files: &files,
                });
            }
        }
        let _t_class_analyzer = std::time::Instant::now();
        {
            let class_db = {
                let guard = self.db.salsa.read();
                (**guard).clone()
            };
            let class_issues = crate::class::ClassAnalyzer::with_files(
                &class_db,
                analyzed_file_set.clone(),
                &file_data,
            )
            .analyze_all();
            all_issues.extend(class_issues);
        }
        let _t_class_analyzer_ms = _t_class_analyzer.elapsed().as_secs_f64() * 1000.0;

        let _t_class_checks = _t0.elapsed();

        let mut db_main = {
            let guard = self.db.salsa.read();
            (**guard).clone()
        };
        // All index mutation for the body pass is done (lazy_load_missing_classes
        // + refresh ran above; lazy_load_from_body_issues runs *after* this pass
        // on a separate db). Freeze the index on this ephemeral clone so each
        // find_class_like borrows it instead of cloning the singleton's three
        // Arcs per call — the per-worker `map_with` clone bumps the refcount once.
        db_main.freeze_workspace_index();

        // ---- Body analysis: function/method bodies in parallel --------------
        type BodyResult = (
            Arc<str>,
            Vec<Issue>,
            Vec<crate::symbol::ResolvedSymbol>,
            Vec<RefLoc>,
        );
        let body_results: Vec<BodyResult> = parsed_files
            .par_iter()
            .filter(|parsed| !files_with_parse_errors.contains(&parsed.file))
            .map_with(db_main, |db, parsed| {
                let mut driver = BodyAnalyzer::new(&*db as &dyn MirDatabase, php_version);
                // Diagnostics-only consumers never read the symbol vecs —
                // don't build them (a Type clone per reference) at all.
                driver.collect_symbols = !opts.skip_symbols;
                if let Some(replay) = transient_replay.as_ref() {
                    if let Some(hit) = replay.files.get(&parsed.file) {
                        let locs: Vec<RefLoc> = hit
                            .ref_locs
                            .iter()
                            .map(|(symbol, line, col_start, col_end)| RefLoc {
                                symbol_key: Arc::clone(symbol),
                                file: parsed.file.clone(),
                                line: *line,
                                col_start: *col_start,
                                col_end: *col_end,
                            })
                            .collect();
                        return (
                            parsed.file.clone(),
                            hit.issues.to_vec(),
                            hit.symbols.to_vec(),
                            locs,
                        );
                    }
                }
                let (issues, symbols) = if let Some(cache) = &self.cache {
                    let h = content_hexes
                        .get(parsed.file.as_ref())
                        .cloned()
                        .unwrap_or_else(|| hash_content(parsed.source()));
                    if let Some((cached_issues, ref_locs)) = cache.get(&parsed.file, &h) {
                        // Cache replay: rebuild the file's complete reference
                        // set straight from the cached tuples — no pending-
                        // buffer detour. Symbol keys are Arc-shared with the
                        // cache entry, so this allocates no strings.
                        let locs: Vec<RefLoc> = ref_locs
                            .iter()
                            .map(|(symbol, line, col_start, col_end)| RefLoc {
                                symbol_key: Arc::clone(symbol),
                                file: parsed.file.clone(),
                                line: *line,
                                col_start: *col_start,
                                col_end: *col_end,
                            })
                            .collect();
                        return (
                            parsed.file.clone(),
                            cached_issues.to_vec(),
                            Vec::new(),
                            locs,
                        );
                    }
                    let (issues, symbols) = driver.analyze_bodies(
                        parsed.owned(),
                        parsed.file.clone(),
                        parsed.source(),
                        parsed.source_map(),
                    );
                    let pending = db.take_pending_ref_locs();
                    let cache_locs: Arc<[crate::cache::CachedRefLoc]> = pending
                        .iter()
                        .map(|r| (Arc::clone(&r.symbol_key), r.line, r.col_start, r.col_end))
                        .collect();
                    let surface = surface_hashes
                        .get(parsed.file.as_ref())
                        .cloned()
                        .unwrap_or_default();
                    cache.put(
                        &parsed.file,
                        h,
                        surface,
                        issues.as_slice().into(),
                        cache_locs,
                    );
                    if let Some(cb) = &opts.on_file_done {
                        cb();
                    }
                    let symbols = if opts.skip_symbols {
                        Vec::new()
                    } else {
                        symbols
                    };
                    return (parsed.file.clone(), issues, symbols, pending);
                } else {
                    driver.analyze_bodies(
                        parsed.owned(),
                        parsed.file.clone(),
                        parsed.source(),
                        parsed.source_map(),
                    )
                };
                let pending = db.take_pending_ref_locs();
                if let Some(cb) = &opts.on_file_done {
                    cb();
                }
                // Drop the per-file symbol vec inside the worker when the
                // consumer opted out — the orchestrator never accumulates.
                let symbols = if opts.skip_symbols {
                    Vec::new()
                } else {
                    symbols
                };
                (parsed.file.clone(), issues, symbols, pending)
            })
            .collect();

        let _t_body_analysis = _t0.elapsed();

        if self.cache.is_none() {
            let replay_files: HashMap<Arc<str>, crate::session::BatchReplayFile> = body_results
                .iter()
                .map(|(file, issues, symbols, ref_locs)| {
                    let cached_ref_locs: Arc<[crate::cache::CachedRefLoc]> = ref_locs
                        .iter()
                        .map(|r| (Arc::clone(&r.symbol_key), r.line, r.col_start, r.col_end))
                        .collect();
                    (
                        file.clone(),
                        crate::session::BatchReplayFile {
                            issues: issues.as_slice().into(),
                            ref_locs: cached_ref_locs,
                            symbols: symbols.as_slice().into(),
                        },
                    )
                })
                .collect();
            self.store_transient_batch_replay(
                php_version.cache_byte(),
                opts.skip_symbols,
                &content_hexes,
                replay_files,
            );
        }

        // Serial commit with replace semantics: each file's output (or cache
        // replay) is its complete reference set, so stale entries from a
        // prior run cannot survive an append.
        let mut all_symbols = Vec::new();
        {
            let guard = self.db.salsa.read();
            for (file, issues, symbols, ref_locs) in body_results {
                all_issues.extend(issues);
                all_symbols.extend(symbols);
                guard.set_file_reference_locations(file.as_ref(), ref_locs);
            }
        }

        // ---- Post-analysis lazy loading: FQCNs used without `use` imports ------
        if let Some(psr4) = self.psr4.clone() {
            self.lazy_load_from_body_issues(
                psr4,
                php_version,
                &file_data,
                &files_with_parse_errors,
                &mut all_issues,
                &mut all_symbols,
                opts.skip_symbols,
            );
        }

        // ---- Build reverse dep graph and persist it for the next run ---------
        // Must run AFTER `commit_reference_locations_batch` (above): the graph's
        // call-site / instantiation / inferred-return edges are derived from the
        // committed reference-location map. Built any earlier (the salsa db is
        // fresh each session) that map is empty, so only structural edges
        // (parent/interface/trait/declared types) survive — and any dependent
        // reachable only through a call site or inferred type goes stale.
        if topology_changed {
            if let Some(cache) = &self.cache {
                let db_snapshot = {
                    let guard = self.db.salsa.read();
                    (**guard).clone()
                };
                let rev = build_reverse_deps(&db_snapshot);
                cache.set_reverse_deps(rev);
            }
        }

        // Persist cache hits/misses to disk
        if let Some(cache) = &self.cache {
            cache.flush();
        }

        // ---- Dead-code detection -------------------------------------------
        if opts.should_run_dead_code() {
            let salsa = self.snapshot_db();
            let _t_dead_code = std::time::Instant::now();
            let dead_code_issues =
                crate::dead_code::DeadCodeAnalyzer::with_files(&salsa, analyzed_file_set.clone())
                    .analyze();
            all_issues.extend(dead_code_issues);
            if std::env::var("MIR_TIMING").is_ok() {
                eprintln!(
                    "[timing] dead_code_analyzer={:.0}ms",
                    _t_dead_code.elapsed().as_secs_f64() * 1000.0
                );
            }
        }

        let _t_total = _t0.elapsed();
        if std::env::var("MIR_TIMING").is_ok() {
            eprintln!(
                "[timing] stubs={:.0}ms read={:.0}ms salsa_reg={:.0}ms collect_defs={:.0}ms ingest={:.0}ms class_checks={:.0}ms (prewarm={:.0}ms lazy_load={:.0}ms class_analyzer={:.0}ms) body_analysis={:.0}ms total={:.0}ms",
                _t_stubs.as_secs_f64() * 1000.0,
                (_t_read - _t_stubs).as_secs_f64() * 1000.0,
                (_t_salsa_reg - _t_read).as_secs_f64() * 1000.0,
                (_t_collect_defs - _t_salsa_reg).as_secs_f64() * 1000.0,
                (_t_ingest - _t_collect_defs).as_secs_f64() * 1000.0,
                (_t_class_checks - _t_ingest).as_secs_f64() * 1000.0,
                _t_prewarm_ms,
                _t_lazyload_ms,
                _t_class_analyzer_ms,
                (_t_body_analysis - _t_class_checks).as_secs_f64() * 1000.0,
                _t_total.as_secs_f64() * 1000.0,
            );
        }

        // Plugin issue veto (Psalm's BeforeAddIssue) — applied to the final
        // set before config-level suppression so a veto can't be shadowed.
        if let Some(plugins) = mir_plugin::snapshot() {
            if plugins.hooks().before_add_issue {
                all_issues.retain(|i| plugins.before_add_issue(i));
            }
        }

        opts.apply(&mut all_issues);
        let analyzed_files_vec: Vec<Arc<str>> = analyzed_file_set.iter().cloned().collect();
        self.apply_suppressions_and_emit_unused(&mut all_issues, &analyzed_files_vec);
        if let Some(dump) = crate::metrics::dump() {
            eprintln!("{dump}");
        }

        // ---- Build workspace symbol index singleton -------------------------
        {
            let mut guard = self.db.salsa.write();
            guard.rebuild_workspace_symbol_index();
        }

        AnalysisResult::build(all_issues, rustc_hash::FxHashMap::default(), all_symbols)
    }
    /// Re-analyze a single file (definition collection + body analysis) within the batch context.
    ///
    /// Mirrors the old `ProjectAnalyzer::re_analyze_file` cache-aware path.
    /// Use [`Self::reanalyze_dependents`] for LSP-style per-file flows that
    /// don't need batch options.
    pub fn re_analyze_file(
        &self,
        file_path: &str,
        new_content: &str,
        opts: &BatchOptions,
    ) -> AnalysisResult {
        self.clear_transient_batch_replay();
        let php_version = self.batch_php_version(opts);

        // Mirror-only workspace files may be waiting in the pending index
        // set. Reconcile them before the content-hash fast path; otherwise a
        // cached result computed against a partial workspace could be replayed
        // before `settle_workspace_index()` gets a chance to invalidate it.
        self.settle_workspace_index();

        // Fast path: content unchanged and cache has a valid entry.
        if let Some(cache) = &self.cache {
            let h = hash_content(new_content);
            if let Some((cached_issues, ref_locs)) = cache.get(file_path, &h) {
                let mut issues = cached_issues.to_vec();
                let file: Arc<str> = Arc::from(file_path);
                // Replace semantics: the cached set is the file's complete
                // reference set, so stale entries from a prior version are
                // cleared rather than appended over.
                let locs: Vec<RefLoc> = ref_locs
                    .iter()
                    .map(|(symbol, line, col_start, col_end)| RefLoc {
                        symbol_key: Arc::clone(symbol),
                        file: file.clone(),
                        line: *line,
                        col_start: *col_start,
                        col_end: *col_end,
                    })
                    .collect();
                let guard = self.db.salsa.read();
                guard.set_file_reference_locations(file_path, locs);
                drop(guard);
                opts.apply(&mut issues);
                self.apply_suppressions_and_emit_unused(&mut issues, std::slice::from_ref(&file));
                return AnalysisResult::build(issues, HashMap::default(), Vec::new());
            }
        }

        let file: Arc<str> = Arc::from(file_path);

        {
            let mut guard = self.db.salsa.write();
            guard.remove_file_definitions(file_path);
        }

        let collected =
            self.db
                .collect_and_ingest_file_with_parsed(file.clone(), new_content, php_version);
        let file_defs = collected.file_defs;

        let mut all_issues: Vec<Issue> = Arc::unwrap_or_clone(file_defs.issues.clone());

        {
            let mut guard = self.db.salsa.write();
            if guard.workspace_symbol_index_singleton().is_some() {
                if let Some(sf) = guard.lookup_source_file(file.as_ref()) {
                    if guard.file_declarations_changed(sf) {
                        guard.rebuild_workspace_symbol_index();
                    }
                }
            }
        }

        let (symbols, surface_hash) = {
            let guard = self.db.salsa.write();
            let parsed = collected
                .parsed
                .unwrap_or_else(|| php_rs_parser::parse(new_content));
            let surface_hash = surface_fingerprint(new_content, &parsed.program);

            let has_hard_errors = parsed.errors.iter().any(crate::parser::is_hard_parse_error);
            let symbols = if !has_hard_errors {
                let db_ref: &dyn MirDatabase = &**guard;
                let mut driver = BodyAnalyzer::new(db_ref, php_version);
                driver.collect_symbols = !opts.skip_symbols;
                let (body_issues, symbols) = driver.analyze_bodies(
                    &parsed.program,
                    file.clone(),
                    new_content,
                    &parsed.source_map,
                );
                all_issues.extend(body_issues);
                let pending = guard.take_pending_ref_locs();
                guard.set_file_reference_locations(file.as_ref(), pending);
                if opts.skip_symbols {
                    Vec::new()
                } else {
                    symbols
                }
            } else {
                Vec::new()
            };
            (symbols, surface_hash)
        };

        // Bake inline-suppression marks in *before* caching: suppression is a
        // pure function of file content (and the cache key hashes content), so
        // the cached issues should already carry their marks. The cache-hit
        // branch above replays this file's source without re-registering the
        // `SourceFile` input, so the db-backed post-filter cannot recompute
        // marks there — caching the canonical result is what keeps a fresh
        // process honoring `@mir-ignore` on an unchanged file.
        mark_suppressed(
            &mut all_issues,
            &crate::suppression::SuppressionMap::from_source(new_content),
        );

        if let Some(cache) = &self.cache {
            let h = hash_content(new_content);
            // Cascade to dependents only when this file's cross-file surface
            // changed; a body-only edit to a declared-return callable leaves
            // their results valid. Unknown (absent/empty) stored surface cascades.
            let surface_changed = match cache.surface_hash(file_path) {
                Some(stored) if !stored.is_empty() => stored != surface_hash,
                _ => true,
            };
            if surface_changed {
                cache.evict_with_dependents(&[file_path.to_string()]);
            }
            let db = self.snapshot_db();
            let ref_locs = extract_reference_locations(&db, &file);
            cache.put(
                file_path,
                h,
                surface_hash,
                all_issues.as_slice().into(),
                ref_locs,
            );
        }

        opts.apply(&mut all_issues);
        // Emit `UnusedSuppress` for named suppressions that matched nothing —
        // present on the cache-hit branch above and on `analyze_paths`, but
        // missing here, so every non-cached re-analysis (the actual
        // incremental/LSP-edit pipeline) silently dropped this diagnostic.
        // `mark_suppressed` above already baked suppression marks into what
        // gets cached; this only adds the (uncached, always-fresh) unused-
        // suppression pass on top, mirroring the cache-hit branch's order.
        self.apply_suppressions_and_emit_unused(&mut all_issues, std::slice::from_ref(&file));
        AnalysisResult::build(all_issues, HashMap::default(), symbols)
    }

    /// Collect type definitions only from `paths` into the codebase
    /// without analyzing method bodies or emitting issues. Used to load
    /// vendor types.
    ///
    /// When a disk-backed cache is attached, per-file `StubSlice` results
    /// from previous runs are reused on a content-hash match, eliminating
    /// the parse + definition-collection step. Cache misses run the normal
    /// pipeline and write back so subsequent runs hit.
    pub fn collect_definitions(&self, paths: &[PathBuf]) {
        self.clear_transient_batch_replay();
        let _timing = std::env::var("MIR_TIMING").is_ok();
        let _t0 = std::time::Instant::now();

        let php_v = self.php_version.cache_byte();

        struct FileEntry {
            file: Arc<str>,
            src: Arc<str>,
            hash: [u8; 32],
            cached: Option<mir_codebase::definitions::StubSlice>,
        }
        let entries: Vec<FileEntry> = paths
            .par_iter()
            .filter_map(|path| {
                let src = std::fs::read_to_string(path).ok()?;
                let file: Arc<str> = Arc::from(path.to_string_lossy().as_ref());
                let src: Arc<str> = Arc::from(src);
                let hash = hash_source(&src);
                let cached = self.db.stub_cache.as_ref().and_then(|c| {
                    let (mut slice, _issues) = c.get(&file, &hash, php_v)?;
                    prepare_for_ingest(&mut slice);
                    Some(slice)
                });
                Some(FileEntry {
                    file,
                    src,
                    hash,
                    cached,
                })
            })
            .collect();
        let _t_read = _t0.elapsed();

        let source_files: Vec<SourceFile> = {
            let mut guard = self.db.salsa.write();
            entries
                .iter()
                .map(|e| {
                    guard.upsert_source_file_with_durability(
                        e.file.clone(),
                        e.src.clone(),
                        salsa::Durability::HIGH,
                    )
                })
                .collect()
        };
        let _t_reg = _t0.elapsed();

        let db_pass1 = {
            let guard = self.db.salsa.read();
            (**guard).clone()
        };
        let stub_cache = self.db.stub_cache.clone();
        let prepared: Vec<(Arc<str>, mir_codebase::definitions::StubSlice)> = entries
            .into_par_iter()
            .zip(source_files.into_par_iter())
            .map_with(db_pass1, |db, (mut entry, salsa_file)| {
                if let Some(slice) = entry.cached.take() {
                    let slice_arc = Arc::new(slice);
                    db.parse_cache().insert(
                        entry.hash,
                        php_v,
                        Arc::clone(&slice_arc),
                        Arc::new(Vec::new()),
                    );
                    return (entry.file.clone(), (*slice_arc).clone());
                }
                let defs = collect_file_definitions(&*db, salsa_file);
                if let Some(cache) = stub_cache.as_ref() {
                    cache.put(&entry.file, &entry.hash, php_v, &defs.slice, &defs.issues);
                }
                (entry.file.clone(), (*defs.slice).clone())
            })
            .collect();
        let _t_collect = _t0.elapsed();
        // Commit subtype-index class edges for the vendor tree: without this,
        // goto-implementation can't surface implementors that live only in
        // vendor/ (index_batch's project-file warm sweep has the same gap for
        // project files, but no StubSlice is cheaply available there — see
        // the persistence work tracked separately).
        {
            let guard = self.db.salsa.read();
            for (file, slice) in &prepared {
                let entries = crate::db::subtype_index::entries_from_slice(slice);
                guard.set_file_class_edges(file, entries);
            }
        }
        drop(prepared);
        let _t_ingest = _t0.elapsed();

        if _timing {
            let (hits, misses) = self.stub_cache_stats();
            eprintln!(
                "[vendor] read={:.0}ms reg={:.0}ms collect={:.0}ms ingest={:.0}ms total={:.0}ms (cache hits={hits} misses={misses})",
                _t_read.as_secs_f64() * 1000.0,
                (_t_reg - _t_read).as_secs_f64() * 1000.0,
                (_t_collect - _t_reg).as_secs_f64() * 1000.0,
                (_t_ingest - _t_collect).as_secs_f64() * 1000.0,
                _t_ingest.as_secs_f64() * 1000.0,
            );
        }

        {
            let mut guard = self.db.salsa.write();
            guard.rebuild_workspace_symbol_index();
        }

        crate::collector::print_collector_stats();
    }
}

#[cfg(test)]
mod tests {
    use rustc_hash::FxHashMap as HashMap;
    use std::fs;
    use std::sync::{Arc, Mutex, OnceLock};

    use crate::{AnalysisSession, BatchOptions, PhpVersion};

    fn metrics_test_lock() -> &'static Mutex<()> {
        static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
        LOCK.get_or_init(|| Mutex::new(()))
    }

    fn write_php(dir: &tempfile::TempDir, name: &str, contents: &str) -> std::path::PathBuf {
        let path = dir.path().join(name);
        fs::write(&path, contents).expect("write fixture");
        path
    }

    #[test]
    fn identical_batch_rerun_replays_without_symbols() {
        let _guard = metrics_test_lock().lock().expect("metrics test lock");
        let dir = tempfile::tempdir().expect("tempdir");
        let file = write_php(
            &dir,
            "Example.php",
            "<?php\nclass Example {\n    public function answer(): int { return 42; }\n}\n(new Example())->answer();\n",
        );
        let session = AnalysisSession::new(PhpVersion::LATEST);
        let opts = BatchOptions::new().without_symbols();
        let hash = crate::cache::hash_content(
            "<?php\nclass Example {\n    public function answer(): int { return 42; }\n}\n(new Example())->answer();\n",
        );
        let hashes = HashMap::from_iter([(Arc::from(file.to_string_lossy().as_ref()), hash)]);

        let first = session.analyze_paths(std::slice::from_ref(&file), &opts);
        assert!(first.issues.is_empty());
        assert!(
            session
                .transient_batch_replay(PhpVersion::LATEST.cache_byte(), true, &hashes)
                .is_some(),
            "expected first run to populate transient replay"
        );

        crate::metrics::test_reset();
        let second = session.analyze_paths(std::slice::from_ref(&file), &opts);
        assert!(second.issues.is_empty());
        assert!(second.symbols.is_empty());

        let dump = crate::metrics::dump().expect("metrics enabled in tests");
        assert!(
            dump.contains("whole-file walks     : 0"),
            "expected replay to skip body walks, got:\n{dump}"
        );
        assert!(
            dump.contains("symbols allocated    : 0"),
            "expected diagnostics-only replay to avoid symbol allocation, got:\n{dump}"
        );
    }

    #[test]
    fn identical_batch_rerun_replays_with_symbols() {
        let _guard = metrics_test_lock().lock().expect("metrics test lock");
        let dir = tempfile::tempdir().expect("tempdir");
        let file = write_php(
            &dir,
            "Example.php",
            "<?php\nclass Example {\n    public function answer(): int { return 42; }\n}\n(new Example())->answer();\n",
        );
        let session = AnalysisSession::new(PhpVersion::LATEST);
        let opts = BatchOptions::new();
        let hash = crate::cache::hash_content(
            "<?php\nclass Example {\n    public function answer(): int { return 42; }\n}\n(new Example())->answer();\n",
        );
        let hashes = HashMap::from_iter([(Arc::from(file.to_string_lossy().as_ref()), hash)]);

        let first = session.analyze_paths(std::slice::from_ref(&file), &opts);
        assert!(
            !first.symbols.is_empty(),
            "expected first run to collect symbols"
        );
        assert!(
            session
                .transient_batch_replay(PhpVersion::LATEST.cache_byte(), false, &hashes)
                .is_some(),
            "expected first run to populate transient replay"
        );

        crate::metrics::test_reset();
        let second = session.analyze_paths(std::slice::from_ref(&file), &opts);
        assert_eq!(second.symbols.len(), first.symbols.len());

        let dump = crate::metrics::dump().expect("metrics enabled in tests");
        assert!(
            dump.contains("whole-file walks     : 0"),
            "expected replay to skip body walks, got:\n{dump}"
        );
    }

    #[test]
    fn session_mutation_clears_batch_replay() {
        let _guard = metrics_test_lock().lock().expect("metrics test lock");
        let dir = tempfile::tempdir().expect("tempdir");
        let file = write_php(
            &dir,
            "Example.php",
            "<?php\nclass Example {\n    public function answer(): int { return 42; }\n}\n(new Example())->answer();\n",
        );
        let file_arc: Arc<str> = Arc::from(file.to_string_lossy().as_ref());
        let session = AnalysisSession::new(PhpVersion::LATEST);
        let opts = BatchOptions::new().without_symbols();

        let _ = session.analyze_paths(std::slice::from_ref(&file), &opts);
        session.ingest_file(
            Arc::clone(&file_arc),
            Arc::from(
                "<?php\nclass Example {\n    public function answer(): string { return 'forty two'; }\n}\n(new Example())->answer();\n",
            ),
        );

        crate::metrics::test_reset();
        let result = session.analyze_paths(std::slice::from_ref(&file), &opts);
        assert!(result.issues.is_empty());

        let dump = crate::metrics::dump().expect("metrics enabled in tests");
        assert!(
            dump.contains("whole-file walks     : 1"),
            "expected session mutation to invalidate replay, got:\n{dump}"
        );
    }
}