omena-lsp-server 0.3.0

//! M3: the off-loop workspace-republish executor (rfcs#111 §8.5, §12 M3).
//!
//! `prepare` flushes the republish lane through its settle gate on the loop
//! and captures a copy-on-write query snapshot; `collect` runs the
//! abort-capable parallel wave against that snapshot on a worker thread —
//! the loop stays free; `apply` publishes loop-side in canonical order under
//! the loop's per-tick chunk budget, writing behind through the LOOP state's
//! disk-cache session (the snapshot carries a default session on purpose).
//! A settle-window reopen bumps the lane generation: the wave aborts at the
//! next item boundary and pending applies are dropped — their keys are
//! republished by the reopened window's tide, and the publication order key
//! forbids any stale overwrite.

use crate::LspShellState;
use crate::diagnostics_follow_up::{
    TIDE_REPUBLISH_LANE_CONFIG, workspace_republish_frontier_passed,
};
use crate::disk_cache::DiskDiagnosticsCacheSlotV0;
use crate::lsp_output::ScheduledLspOutput;
use crate::state::LspQuerySnapshotV0;
use crate::tide::TideGateInputsV0;
use serde_json::Value;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

#[derive(Debug)]
pub struct TideWorkspaceRepublishJobV0 {
    snapshot: LspQuerySnapshotV0,
    uris: Vec<String>,
    pub generation: u64,
    gen_watch: Arc<AtomicU64>,
}

impl TideWorkspaceRepublishJobV0 {
    #[cfg(test)]
    pub(crate) fn target_uris_for_test(&self) -> &[String] {
        self.uris.as_slice()
    }
}

#[derive(Debug)]
pub struct TideWorkspaceRepublishItemV0 {
    pub(crate) uri: String,
    pub(crate) diagnostics: Value,
    pub(crate) disk_cache_slot: Option<DiskDiagnosticsCacheSlotV0>,
}

#[derive(Debug)]
pub struct TideWorkspaceRepublishResultV0 {
    pub generation: u64,
    pub items: Vec<TideWorkspaceRepublishItemV0>,
    /// Wave-ineligible targets of THIS chunk. When the tide is still current
    /// at completion these fall back to the per-file deferred arm; a
    /// disowned tide drops them (the reopened window covers the corpus).
    pub uncovered_uris: Vec<String>,
    /// The last chunk of the tide: the loop completes the tide once this
    /// arrives and the apply queue drains.
    pub final_chunk: bool,
}

/// Gate evaluation + snapshot capture, on the loop. `idle` is the courtesy
/// input (no recent client message); aging overrides it after
/// [`TIDE_REPUBLISH_LANE_CONFIG`]'s bound, the frontier never.
pub fn prepare_tide_workspace_republish_job(
    state: &mut LspShellState,
    idle: bool,
) -> Option<TideWorkspaceRepublishJobV0> {
    if !state.external_sif_refresh_deferred {
        return None;
    }
    let inputs = TideGateInputsV0 {
        frontier_passed: workspace_republish_frontier_passed(state),
        idle,
    };
    let flush = state.tide_republish_lane.try_flush(
        inputs,
        state.tide_tick,
        &TIDE_REPUBLISH_LANE_CONFIG,
    )?;
    state
        .tide_republish_gen_watch
        .store(flush.generation, Ordering::Relaxed);
    // Demand-shaped targeting (rfcs#111 demand lattice): `All` covers the
    // corpus, `Cone` covers the seeds' reverse-dependency closure at flush
    // time. Open documents come first — the user is looking at them — and
    // chunked streaming below turns that ordering into convergence latency.
    let uris = crate::diagnostics_follow_up::tide_republish_target_uris(state, &flush.demand);
    if uris.is_empty() {
        state.tide_republish_lane.tide_completed(flush.generation);
        return None;
    }
    crate::loop_trace!(
        "republish-tide prepared gen={} targets={} demand={}",
        flush.generation,
        uris.len(),
        match &flush.demand {
            crate::tide::TideRepublishDemandV0::All => "all".to_string(),
            crate::tide::TideRepublishDemandV0::Cone(seeds) => format!("cone({})", seeds.len()),
            crate::tide::TideRepublishDemandV0::None => "none".to_string(),
        }
    );
    Some(TideWorkspaceRepublishJobV0 {
        snapshot: state.query_snapshot(),
        uris,
        generation: flush.generation,
        gen_watch: Arc::clone(&state.tide_republish_gen_watch),
    })
}

/// Worker-side compute, streaming (rfcs#111 §8.5): ONE shared-graph parallel
/// wave — one memo-host sync, one substrate, one condensation — with a
/// per-item sink that emits each target the moment its pool task finishes.
/// Open documents were ordered first by prepare, so they converge first.
/// The generation watch aborts disowned tides at item boundaries; the final
/// event carries the uncovered remainder for the fallback arm.
pub fn collect_tide_workspace_republish_streaming(
    job: TideWorkspaceRepublishJobV0,
    emit: &(dyn Fn(TideWorkspaceRepublishResultV0) -> bool + Sync),
) {
    // A panic in the shared setup (host sync, substrate, condensation)
    // would otherwise unwind past the final-chunk emit, leaving the loop's
    // in-flight gate stuck at 1 and workspace republish silently dead for
    // the session. Guarantee a final chunk on EVERY exit: on panic, report
    // every target uncovered — the fallback arm re-covers them per-file,
    // and items already streamed are superseded harmlessly.
    let generation = job.generation;
    let uncovered_fallback = job.uris.clone();
    if std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        collect_tide_workspace_republish_streaming_inner(job, emit)
    }))
    .is_err()
    {
        let _ = emit(TideWorkspaceRepublishResultV0 {
            generation,
            items: Vec::new(),
            uncovered_uris: uncovered_fallback,
            final_chunk: true,
        });
    }
}

fn collect_tide_workspace_republish_streaming_inner(
    job: TideWorkspaceRepublishJobV0,
    emit: &(dyn Fn(TideWorkspaceRepublishResultV0) -> bool + Sync),
) {
    let covered = std::sync::Mutex::new(std::collections::BTreeSet::<usize>::new());
    let sink =
        |index: usize,
         diagnostics: serde_json::Value,
         disk_cache_slot: Option<crate::disk_cache::DiskDiagnosticsCacheSlotV0>| {
            if let Ok(mut covered) = covered.lock() {
                covered.insert(index);
            }
            let Some(uri) = job.uris.get(index) else {
                return;
            };
            let _ = emit(TideWorkspaceRepublishResultV0 {
                generation: job.generation,
                items: vec![TideWorkspaceRepublishItemV0 {
                    uri: uri.clone(),
                    diagnostics,
                    disk_cache_slot,
                }],
                uncovered_uris: Vec::new(),
                final_chunk: false,
            });
        };
    let _ = crate::parallel_style_wave::resolved_parallel_style_wave_targets_from_read_view_with_abort_and_sink(
        &job.snapshot,
        job.uris.as_slice(),
        crate::parallel_style_wave::PARALLEL_STYLE_WAVE_MIN_PARALLEL_TARGETS,
        Some((job.gen_watch.as_ref(), job.generation)),
        Some(&sink),
    );
    let covered = covered.into_inner().unwrap_or_default();
    let uncovered_uris = job
        .uris
        .iter()
        .enumerate()
        .filter(|(index, _)| !covered.contains(index))
        .map(|(_, uri)| uri.clone())
        .collect::<Vec<_>>();
    crate::loop_trace!(
        "republish-tide collected gen={} covered={} uncovered={}",
        job.generation,
        covered.len(),
        uncovered_uris.len()
    );
    let _ = emit(TideWorkspaceRepublishResultV0 {
        generation: job.generation,
        items: Vec::new(),
        uncovered_uris,
        final_chunk: true,
    });
}

/// Loop-side apply for ONE item — the caller pumps a bounded chunk per tick
/// (I4) and must have verified the tide generation is still current.
/// Write-behind runs through the loop state's real disk-cache session, then
/// the tiered publish emits in the same shape as every other arm.
pub fn apply_tide_workspace_republish_item(
    state: &mut LspShellState,
    item: TideWorkspaceRepublishItemV0,
) -> Vec<ScheduledLspOutput> {
    if let Some(slot) = item.disk_cache_slot.as_ref() {
        slot.store_write_behind(state, &item.diagnostics);
    }
    crate::diagnostics_scheduler::publish_tiered_diagnostics_notifications(
        state,
        item.uri.as_str(),
        item.diagnostics,
    )
}

/// Completion: re-arm the lane; when the tide is still current, uncovered
/// targets re-enter the per-file deferred arm so no key is silently skipped.
pub fn complete_tide_workspace_republish(
    state: &mut LspShellState,
    generation: u64,
    uncovered_uris: Vec<String>,
) -> crate::LspDiagnosticsFollowUpEffectsV0 {
    let current = state.tide_republish_lane.generation() == generation;
    // Read the flushed demand BEFORE discharging it: the source refresh
    // below is shaped by the same demand that shaped the tide's own
    // targets (review finding: an unshaped all-open-sources broadcast at
    // every completion contradicted the cone discipline and put
    // corpus-scale gather work on the loop).
    let source_scope_seeds = match state.tide_republish_lane.in_flight_demand() {
        Some(crate::tide::TideRepublishDemandV0::Cone(seeds)) => Some(seeds.clone()),
        _ => None,
    };
    state.tide_republish_lane.tide_completed(generation);
    if !current {
        return crate::LspDiagnosticsFollowUpEffectsV0::default();
    }
    let mut effects = crate::diagnostics_scheduler::DiagnosticsScheduleEffectsV0::default();
    if !uncovered_uris.is_empty() {
        crate::loop_trace!(
            "republish-tide leftovers gen={} n={}",
            generation,
            uncovered_uris.len()
        );
        effects.extend(
            crate::diagnostics_scheduler::run_diagnostics_schedule_effects(
                state,
                crate::diagnostics_scheduler::DiagnosticsScheduleEvent::WatchedFiles {
                    uris: uncovered_uris,
                },
            ),
        );
    }
    // The wave re-rendered every STYLE target against the settled corpus;
    // open SOURCE documents were rendered against whatever corpus existed
    // when their tab opened — possibly BEFORE this settle admitted the
    // stylesheets they reference, and nothing else ever re-evaluates them
    // (the fan-outs fire on style EVENTS, not on corpus growth). Completion
    // of a current-generation tide is the moment the corpus became
    // authoritative, so open sources re-enter the same per-file deferred
    // arm as the uncovered style leftovers.
    let open_source_uris = crate::diagnostics_scheduler::open_document_uris_for_diagnostics(state)
        .into_iter()
        .filter(|uri| !crate::protocol::is_style_document_uri(uri.as_str()))
        .collect::<Vec<_>>();
    // Cone tides refresh only the sources that DEPEND on the cone's seeds
    // (the same filter the didChange fan-out uses); the unscoped arm is
    // reserved for All tides — the cold-settle case whose whole point is
    // that every earlier source render predates the corpus.
    let open_source_uris = match source_scope_seeds {
        Some(seeds) => {
            let mut scoped = std::collections::BTreeSet::new();
            for seed in seeds {
                scoped.extend(
                    crate::diagnostics_scheduler::scoped_source_republish_uris_for_style_change(
                        state,
                        seed.as_str(),
                        open_source_uris.clone(),
                    ),
                );
            }
            scoped.into_iter().collect::<Vec<_>>()
        }
        None => open_source_uris,
    };
    if !open_source_uris.is_empty() {
        crate::loop_trace!(
            "republish-tide source refresh gen={} n={}",
            generation,
            open_source_uris.len()
        );
        effects.extend(
            crate::diagnostics_scheduler::diagnostics_effects_for_document_uris(
                state,
                open_source_uris,
                true,
            ),
        );
    }
    crate::LspDiagnosticsFollowUpEffectsV0 {
        outputs: effects.outputs,
        deferred_diagnostics: effects.deferred_diagnostics,
    }
}