presolve_compiler/

service.rs

//! Local, durable compiler-service host (Phase L4).
//!
//! The host accepts complete request-owned workspace inputs and delegates every
//! compilation to `platform::CompilerSessionState`; it never reads workspace
//! files or persists source text.

#![allow(
    clippy::missing_errors_doc,
    clippy::missing_panics_doc,
    clippy::needless_pass_by_value,
    clippy::single_match_else,
    clippy::too_many_lines
)]

use std::collections::{BTreeMap, BTreeSet};
use std::fmt::Write as _;
use std::fs::{self, File, OpenOptions};
use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;

use sha2::{Digest as _, Sha256};

use crate::persistent_cache::{
    CacheInspectionReportV1, CacheKeyInputV1, CacheOutcomeV1, CacheReportSelector,
    CacheTelemetryV1, CachedCompileResultV1, PersistentArtifactCacheV1,
};
use crate::platform::{
    self, CacheLimits, CancellationToken, CanonicalReusableProductV1, CompilationOutcome,
    CompileWorkspaceRequest, CompilerSessionState, ContractVersion, IncrementalCompilationModeV1,
    IncrementalCompileWorkspaceRequestV1, IncrementalFallbackReasonV1, RequestedCompilationMode,
    SourceRevisionId, SourceUnitId, WorkspaceConfiguration, WorkspaceGraph, WorkspaceId,
    WorkspaceInput, WorkspaceSnapshot, WorkspaceSource,
};
use crate::watch::{self, WatchChangeBatchMetadataV1, WatchSessionV1};
use crate::workspace::{self, WorkspaceBuildPlanV1, WorkspaceManifestV1, WorkspacePackageGraphV1};

pub const COMPILER_SERVICE_PROTOCOL_VERSION: u32 = 1;
pub const COMPILER_SERVICE_DESCRIPTOR_SCHEMA_VERSION: u32 = 1;
pub const COMPILER_SERVICE_REQUEST_SCHEMA_VERSION: u32 = 1;
pub const COMPILER_SERVICE_RESPONSE_SCHEMA_VERSION: u32 = 1;
pub const DURABLE_SESSION_SCHEMA_VERSION: u32 = 1;
pub const DURABLE_COMMIT_SCHEMA_VERSION: u32 = 1;
pub const SESSION_JOURNAL_SCHEMA_VERSION: u32 = 1;
pub const PERSISTENCE_MANIFEST_SCHEMA_VERSION: u32 = 1;
pub const SERVICE_INSPECTION_SCHEMA_VERSION: u32 = 1;
pub const SESSION_INSPECTION_SCHEMA_VERSION: u32 = 1;
pub const MAXIMUM_FRAME_BYTES: usize = 16 * 1024 * 1024;

static NEXT_SERVICE: AtomicU64 = AtomicU64::new(1);

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ServiceError {
    pub code: &'static str,
    pub message: String,
    pub retryable: bool,
}
impl ServiceError {
    fn new(code: &'static str, message: impl Into<String>) -> Self {
        Self {
            code,
            message: message.into(),
            retryable: false,
        }
    }
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CompilerServiceDescriptor {
    pub schema_version: u32,
    pub protocol_version: u32,
    pub service_instance_id: String,
    pub compiler_contract: ContractVersion,
    pub platform_contract: String,
    pub persistence_version: u32,
    pub capabilities: Vec<String>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ServiceOperation {
    Handshake,
    ServiceStatus,
    OpenSession,
    ResumeSession,
    Compile,
    Cancel,
    SessionStatus,
    CloseSession,
    Shutdown,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ServiceResponseStatus {
    Ok,
    Rejected,
    Cancelled,
    Failed,
}
#[derive(Debug, Clone)]
pub struct CompleteSource {
    pub path: String,
    pub source: String,
    pub language: Option<platform::SourceLanguage>,
}
#[derive(Debug, Clone)]
pub struct CompileRequest {
    pub configuration: WorkspaceConfiguration,
    pub candidate_snapshot: WorkspaceSnapshot,
    pub sources: Vec<CompleteSource>,
    pub mode: RequestedCompilationMode,
    /// Optional L5 inspection data. `None` preserves the L4 response surface.
    pub incremental_report: IncrementalReportSelector,
    /// Explicit test-only proof mode. It never enables production semantics.
    pub verify_exact_equivalence: bool,
    /// Optional L6 cache telemetry. `None` preserves the L4/L5 response surface.
    pub cache_report: CacheReportSelector,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum IncrementalReportSelector {
    None,
    Summary,
    Full,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct IncrementalExecutionReportV1 {
    pub schema: &'static str,
    pub plan_fingerprint: platform::Digest,
    pub mode: IncrementalCompilationModeV1,
    pub changed_inputs: Vec<platform::IncrementalInputChangeV1>,
    pub invalidated_identities: Vec<SourceUnitId>,
    pub reused_product_identities: Vec<platform::ProductKey>,
    pub recomputed_work_unit_identities: Vec<SourceUnitId>,
    pub fallback_reasons: Vec<IncrementalFallbackReasonV1>,
    pub publication_outcome: &'static str,
    pub exact_equivalence_verified: Option<bool>,
}
impl IncrementalExecutionReportV1 {
    /// # Panics
    ///
    /// Panics only if serializing an owned Rust string fails, which `serde_json`
    /// guarantees for strings.
    #[must_use]
    pub fn to_canonical_json(&self) -> Vec<u8> {
        let quote = |value: &str| serde_json::to_string(value).expect("strings serialize");
        let changes = self
            .changed_inputs
            .iter()
            .map(|change| {
                format!(
                    "{{\"kind\":{},\"identity\":{}}}",
                    quote(change.kind.as_str()),
                    quote(&change.identity)
                )
            })
            .collect::<Vec<_>>()
            .join(",");
        let ids = |values: &[SourceUnitId]| {
            values
                .iter()
                .map(|value| quote(value.as_str()))
                .collect::<Vec<_>>()
                .join(",")
        };
        let reused = self
            .reused_product_identities
            .iter()
            .map(|value| quote(value.as_str()))
            .collect::<Vec<_>>()
            .join(",");
        let fallback = self
            .fallback_reasons
            .iter()
            .map(|reason| quote(reason.code()))
            .collect::<Vec<_>>()
            .join(",");
        format!("{{\"schema\":{},\"plan_fingerprint\":{},\"mode\":{},\"changed_inputs\":[{}],\"invalidated_identities\":[{}],\"reused_product_identities\":[{}],\"recomputed_work_unit_identities\":[{}],\"fallback_reasons\":[{}],\"publication_outcome\":{},\"exact_equivalence_verified\":{}}}\n",quote(self.schema),quote(self.plan_fingerprint.as_str()),quote(self.mode.as_str()),changes,ids(&self.invalidated_identities),reused,ids(&self.recomputed_work_unit_identities),fallback,quote(self.publication_outcome),self.exact_equivalence_verified.map_or_else(|| "null".into(), |value| value.to_string())).into_bytes()
    }
}
#[derive(Debug, Clone)]
pub struct CompileResponse {
    pub commit_sequence: u64,
    pub snapshot: Arc<WorkspaceSnapshot>,
    pub graph: Arc<WorkspaceGraph>,
    pub mode: String,
    pub incremental_report: Option<IncrementalExecutionReportV1>,
    pub cache_report: Option<CacheTelemetryV1>,
}
#[derive(Debug, Clone)]
pub struct WorkspacePackageCompileRequestV1 {
    pub package_id: String,
    pub expected_commit_sequence: u64,
    pub request: CompileRequest,
}
#[derive(Debug, Clone)]
pub struct WorkspaceCompileRequestV1 {
    pub manifest: WorkspaceManifestV1,
    pub packages: Vec<WorkspacePackageCompileRequestV1>,
    pub operation_id: String,
}
#[derive(Debug, Clone)]
pub struct WorkspacePackageResultV1 {
    pub package_id: String,
    pub status: String,
    pub snapshot_id: Option<String>,
}
#[derive(Debug, Clone)]
pub struct WorkspaceBuildResultV1 {
    pub workspace_id: String,
    pub status: String,
    pub manifest_identity: String,
    pub graph_identity: String,
    pub plan_identity: String,
    pub package_results: Vec<WorkspacePackageResultV1>,
}
/// Transient L8 input.  The complete workspace request is never placed in a
/// snapshot, event journal, or durable service state.
#[derive(Debug, Clone)]
pub struct WatchCandidateV1 {
    pub candidate_id: String,
    pub fingerprint_hint: Option<String>,
    pub verify_exact_equivalence: bool,
    pub workspace_request: WorkspaceCompileRequestV1,
}
#[derive(Debug, Clone)]
pub struct WatchChangeBatchV1 {
    pub schema: String,
    pub version: u32,
    pub watch_session_id: String,
    pub sequence: u64,
    pub observed_changes: Vec<watch::ObservedChangeV1>,
    pub candidate: WatchCandidateV1,
}
struct WatchHostSession {
    state: WatchSessionV1,
    /// At most one unexecuted complete replacement request.  This is strictly
    /// process-local and is released on coalesce, completion, or stop.
    pending: Option<WatchCandidateV1>,
}
#[derive(Debug, Clone)]
struct DurableWorkspaceStateV1 {
    manifest: WorkspaceManifestV1,
    graph: WorkspacePackageGraphV1,
    plan: WorkspaceBuildPlanV1,
    result: WorkspaceBuildResultV1,
}

pub fn encode_frame(payload: &[u8]) -> Result<Vec<u8>, ServiceError> {
    if payload.is_empty() {
        return Err(ServiceError::new("invalid_frame", "empty payload"));
    }
    if payload.len() > MAXIMUM_FRAME_BYTES {
        return Err(ServiceError::new(
            "frame_too_large",
            "payload exceeds service limit",
        ));
    }
    let mut frame = format!("{:08x}:", payload.len()).into_bytes();
    frame.extend_from_slice(payload);
    frame.push(b'\n');
    Ok(frame)
}
pub fn decode_frame(frame: &[u8]) -> Result<Vec<u8>, ServiceError> {
    if frame.len() < 10 || frame[8] != b':' || *frame.last().unwrap_or(&0) != b'\n' {
        return Err(ServiceError::new("invalid_frame", "malformed frame"));
    }
    let header = std::str::from_utf8(&frame[..8])
        .map_err(|_| ServiceError::new("invalid_frame", "non-ASCII length"))?;
    if !header
        .bytes()
        .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
    {
        return Err(ServiceError::new(
            "invalid_frame",
            "length is not lowercase hexadecimal",
        ));
    }
    let length = usize::from_str_radix(header, 16)
        .map_err(|_| ServiceError::new("invalid_frame", "invalid length"))?;
    if length == 0 || length > MAXIMUM_FRAME_BYTES || frame.len() != length + 10 {
        return Err(ServiceError::new(
            "invalid_frame",
            "truncated or trailing frame",
        ));
    }
    Ok(frame[9..frame.len() - 1].to_vec())
}

pub struct CompilerServiceHost {
    root: PathBuf,
    descriptor: CompilerServiceDescriptor,
    sessions: BTreeMap<String, DurableSession>,
    cache: PersistentArtifactCacheV1,
    workspaces: BTreeMap<String, DurableWorkspaceStateV1>,
    watch_sessions: BTreeMap<String, WatchHostSession>,
}
struct DurableSession {
    workspace_id: WorkspaceId,
    configuration: WorkspaceConfiguration,
    compiler_contract: ContractVersion,
    commit_sequence: u64,
    closed: bool,
    l3: CompilerSessionState,
    incremental_baseline: Option<IncrementalBaseline>,
}
#[derive(Clone)]
struct IncrementalBaseline {
    publication_identity: String,
    configuration: WorkspaceConfiguration,
    source_fingerprints: BTreeMap<SourceUnitId, SourceRevisionId>,
    snapshot: Arc<WorkspaceSnapshot>,
    graph: Arc<WorkspaceGraph>,
    reusable_products: Vec<CanonicalReusableProductV1>,
    compiler_contract: ContractVersion,
}
impl IncrementalBaseline {
    fn is_consistent(&self) -> bool {
        self.publication_identity == self.snapshot.snapshot_id.as_str()
            && self.compiler_contract == self.snapshot.compiler_contract
            && platform::workspace_configuration_fingerprint_v1(&self.configuration)
                .is_ok_and(|fingerprint| fingerprint == self.snapshot.configuration_fingerprint)
            && self.source_fingerprints
                == self
                    .snapshot
                    .units
                    .iter()
                    .map(|unit| (unit.source_unit_id.clone(), unit.source_revision_id.clone()))
                    .collect()
    }
}
impl CompilerServiceHost {
    pub fn start(
        root: impl AsRef<Path>,
        compiler_contract: ContractVersion,
    ) -> Result<Self, ServiceError> {
        Self::start_with_cache(root, None, compiler_contract)
    }
    pub fn start_with_cache(
        root: impl AsRef<Path>,
        cache_root: Option<&Path>,
        compiler_contract: ContractVersion,
    ) -> Result<Self, ServiceError> {
        let root = root.as_ref().join("service");
        fs::create_dir_all(root.join("sessions")).map_err(io_error)?;
        let service_id = format!(
            "service-instance:{:032x}",
            NEXT_SERVICE.fetch_add(1, Ordering::Relaxed)
        );
        let descriptor = CompilerServiceDescriptor {
            schema_version: 1,
            protocol_version: 1,
            service_instance_id: service_id,
            compiler_contract: compiler_contract.clone(),
            platform_contract: "presolve-platform-l3:1".into(),
            persistence_version: 1,
            capabilities: vec![
                "canonical_inspection".into(),
                "durable_sessions".into(),
                "incremental_compile".into(),
                "request_cancellation".into(),
                "session_recovery".into(),
                "explicit_watch_sessions".into(),
            ],
        };
        let host = Self {
            root,
            descriptor,
            sessions: BTreeMap::new(),
            cache: PersistentArtifactCacheV1::open(cache_root, &compiler_contract),
            workspaces: BTreeMap::new(),
            watch_sessions: BTreeMap::new(),
        };
        host.write_manifest()?;
        Ok(host)
    }
    #[must_use]
    pub fn descriptor(&self) -> &CompilerServiceDescriptor {
        &self.descriptor
    }
    pub fn open_session(
        &mut self,
        configuration: WorkspaceConfiguration,
        claimed: &WorkspaceId,
    ) -> Result<String, ServiceError> {
        let workspace = platform::derive_workspace_id_v1(&configuration).map_err(platform_error)?;
        if &workspace != claimed {
            return Err(ServiceError::new(
                "workspace_mismatch",
                "workspace identity does not match configuration",
            ));
        }
        let id = format!(
            "session:sha256:{:x}",
            Sha256::digest(
                format!(
                    "presolve-durable-session-v1\0{}\0{}",
                    workspace.as_str(),
                    NEXT_SERVICE.fetch_add(1, Ordering::Relaxed)
                )
                .as_bytes()
            )
        );
        let l3 = CompilerSessionState::new(
            workspace.clone(),
            self.descriptor.compiler_contract.clone(),
            CacheLimits::default(),
        );
        let session = DurableSession {
            workspace_id: workspace,
            configuration,
            compiler_contract: self.descriptor.compiler_contract.clone(),
            commit_sequence: 0,
            closed: false,
            l3,
            incremental_baseline: None,
        };
        self.write_session(&id, &session)?;
        append_journal(&self.root, &id, 1, "session_created", 0, None)?;
        self.sessions.insert(id.clone(), session);
        Ok(id)
    }
    /// Closes a live session and releases its non-durable L5 baseline before
    /// recording the existing L4 durable closed-state marker.
    pub fn close_session(&mut self, session_id: &str) -> Result<(), ServiceError> {
        let root = self.root.clone();
        let session = self
            .sessions
            .get_mut(session_id)
            .ok_or_else(|| ServiceError::new("session_not_found", "session not loaded"))?;
        if session.closed {
            return Ok(());
        }
        session.incremental_baseline = None;
        session.l3.close();
        session.closed = true;
        write_session_at(&root, session_id, session)?;
        append_journal(
            &root,
            session_id,
            session.commit_sequence.saturating_mul(2).saturating_add(2),
            "session_closed",
            session.commit_sequence,
            None,
        )
    }
    pub fn verify_cache(
        &self,
        _explicit_root: &Path,
    ) -> Result<CacheInspectionReportV1, ServiceError> {
        self.cache.verify().map_err(cache_operation_error)
    }
    pub fn inspect_cache(
        &self,
        _explicit_root: &Path,
    ) -> Result<CacheInspectionReportV1, ServiceError> {
        self.cache.inspect().map_err(cache_operation_error)
    }
    pub fn clean_cache(&self, _explicit_root: &Path) -> Result<Vec<String>, ServiceError> {
        self.cache.clean().map_err(cache_operation_error)
    }
    /// Creates a process-local L8 watch session.  This has no persistence and
    /// intentionally performs no compilation or input discovery.
    pub fn create_watch_session(
        &mut self,
        configuration: watch::WatchSessionConfigurationV1,
    ) -> Result<watch::WatchSessionSnapshotV1, ServiceError> {
        let id = configuration.watch_session_id.clone();
        if self.watch_sessions.contains_key(&id) {
            return Err(ServiceError::new(
                "L8W001_WATCH_SESSION_ALREADY_EXISTS",
                "watch session already exists",
            ));
        }
        if self.watch_sessions.len() >= watch::MAX_WATCH_SESSIONS {
            return Err(ServiceError::new(
                "L8W010_WATCH_RESOURCE_LIMIT_EXCEEDED",
                "maximum process-local watch sessions reached",
            ));
        }
        let state = WatchSessionV1::new(configuration).map_err(watch_error)?;
        let snapshot = state.snapshot();
        self.watch_sessions.insert(
            id,
            WatchHostSession {
                state,
                pending: None,
            },
        );
        Ok(snapshot)
    }
    /// Accepts an exact complete replacement candidate. The caller, not the
    /// compiler service, owns all observation and source acquisition.
    pub fn submit_watch_change_batch(
        &mut self,
        batch: WatchChangeBatchV1,
        monotonic_milliseconds: u64,
    ) -> Result<watch::WatchAcceptanceV1, ServiceError> {
        let workspace_id = batch
            .candidate
            .workspace_request
            .manifest
            .normalize_validate()
            .map_err(workspace_error)?
            .workspace_id;
        let session = self
            .watch_sessions
            .get_mut(&batch.watch_session_id)
            .ok_or_else(|| {
                ServiceError::new("L8W002_WATCH_SESSION_NOT_FOUND", "watch session not found")
            })?;
        if session.state.configuration().workspace_id != workspace_id {
            return Err(ServiceError::new(
                "L8W003_WATCH_WORKSPACE_MISMATCH",
                "candidate workspace does not match watch session",
            ));
        }
        let fingerprint = watch_candidate_fingerprint(&batch.candidate.workspace_request)?;
        if batch
            .candidate
            .fingerprint_hint
            .as_ref()
            .is_some_and(|hint| hint != &fingerprint)
        {
            return Err(ServiceError::new(
                "L8W008_CANDIDATE_IDENTITY_HINT_MISMATCH",
                "candidate fingerprint hint does not match the derived identity",
            ));
        }
        let metadata = WatchChangeBatchMetadataV1 {
            schema: batch.schema,
            version: batch.version,
            watch_session_id: batch.watch_session_id,
            sequence: batch.sequence,
            observed_changes: batch.observed_changes,
            candidate: watch::WatchCandidateMetadataV1 {
                candidate_id: batch.candidate.candidate_id.clone(),
                fingerprint,
            },
        };
        let acceptance = session
            .state
            .accept(metadata, monotonic_milliseconds)
            .map_err(watch_error)?;
        if acceptance.retained_pending {
            // Replacing this drops the prior complete candidate and its authored
            // source text immediately; the state machine itself is source-free.
            session.pending = Some(batch.candidate);
        }
        Ok(acceptance)
    }
    pub fn flush_watch_session(&mut self, watch_session_id: &str) -> Result<(), ServiceError> {
        self.watch_sessions
            .get_mut(watch_session_id)
            .ok_or_else(|| {
                ServiceError::new("L8W002_WATCH_SESSION_NOT_FOUND", "watch session not found")
            })?
            .state
            .flush()
            .map_err(watch_error)
    }
    /// Runs one caller-controlled deterministic scheduler turn. This is an
    /// internal service executor hook, not a wall-clock timer or public API.
    pub fn run_watch_scheduler_turn(
        &mut self,
        watch_session_id: &str,
        monotonic_milliseconds: u64,
    ) -> Result<Option<watch::WatchExecutionReportV1>, ServiceError> {
        let (plan, candidate) = {
            let session = self
                .watch_sessions
                .get_mut(watch_session_id)
                .ok_or_else(|| {
                    ServiceError::new("L8W002_WATCH_SESSION_NOT_FOUND", "watch session not found")
                })?;
            let Some(plan) = session.state.scheduler_turn(monotonic_milliseconds) else {
                return Ok(None);
            };
            let candidate = session.pending.take().ok_or_else(|| {
                ServiceError::new(
                    "L8W007_INVALID_CHANGE_BATCH",
                    "missing transient pending candidate",
                )
            })?;
            (plan, candidate)
        };
        // The sole active request is owned on this stack only; it is released as
        // soon as L7 returns. L7 remains the only compiler/workspace executor.
        let outcome = self.compile_workspace_v1(candidate.workspace_request);
        let (success, identity) = match &outcome {
            Ok(result) if result.status == "succeeded" => {
                (true, Some(workspace_result_identity(result)))
            }
            _ => (false, None),
        };
        let report = self
            .watch_sessions
            .get_mut(watch_session_id)
            .expect("session cannot be removed during serialized turn")
            .state
            .finish(success, identity);
        debug_assert!(report.as_ref().is_some_and(|report| report.plan == plan));
        match outcome {
            Ok(_) => Ok(report),
            // L7 failure is represented in the watch event/report while prior
            // watch success remains untouched; callers can inspect the session.
            Err(error) => {
                if report.is_some() {
                    Ok(report)
                } else {
                    Err(error)
                }
            }
        }
    }
    pub fn poll_watch_events(
        &self,
        watch_session_id: &str,
        after_event_sequence: u64,
    ) -> Result<Vec<watch::WatchEventV1>, ServiceError> {
        self.watch_sessions
            .get(watch_session_id)
            .ok_or_else(|| {
                ServiceError::new("L8W002_WATCH_SESSION_NOT_FOUND", "watch session not found")
            })?
            .state
            .poll(after_event_sequence)
            .map_err(watch_error)
    }
    pub fn inspect_watch_session(
        &self,
        watch_session_id: &str,
    ) -> Result<watch::WatchSessionSnapshotV1, ServiceError> {
        self.watch_sessions
            .get(watch_session_id)
            .map(|session| session.state.snapshot())
            .ok_or_else(|| {
                ServiceError::new("L8W002_WATCH_SESSION_NOT_FOUND", "watch session not found")
            })
    }
    pub fn stop_watch_session(
        &mut self,
        watch_session_id: &str,
    ) -> Result<watch::WatchSessionSnapshotV1, ServiceError> {
        let session = self
            .watch_sessions
            .get_mut(watch_session_id)
            .ok_or_else(|| {
                ServiceError::new("L8W002_WATCH_SESSION_NOT_FOUND", "watch session not found")
            })?;
        session.pending = None;
        session.state.stop();
        Ok(session.state.snapshot())
    }
    pub fn compile_workspace_v1(
        &mut self,
        request: WorkspaceCompileRequestV1,
    ) -> Result<WorkspaceBuildResultV1, ServiceError> {
        let manifest = request
            .manifest
            .normalize_validate()
            .map_err(workspace_error)?;
        let graph = workspace::graph(&manifest).map_err(workspace_error)?;
        let expected = manifest
            .packages
            .iter()
            .map(|p| p.package_id.clone())
            .collect::<BTreeSet<_>>();
        let supplied = request
            .packages
            .iter()
            .map(|p| p.package_id.clone())
            .collect::<BTreeSet<_>>();
        if expected != supplied || supplied.len() != request.packages.len() {
            return Err(ServiceError::new(
                "L7W007_PACKAGE_REQUEST_SET_MISMATCH",
                "workspace request keys do not match manifest",
            ));
        }
        let mut requests = request
            .packages
            .into_iter()
            .map(|p| (p.package_id.clone(), p))
            .collect::<BTreeMap<_, _>>();
        for package in &manifest.packages {
            let item = requests
                .get(&package.package_id)
                .expect("validated request");
            if package.session_id != item.request.candidate_snapshot.workspace_id.as_str()
                && !self.sessions.contains_key(&package.session_id)
            {
                return Err(ServiceError::new(
                    "L7W011_WORKSPACE_SESSION_OWNERSHIP_CONFLICT",
                    "package session does not exist",
                ));
            }
            if let Some(hint) = &package.configuration_identity_hint {
                let actual =
                    platform::workspace_configuration_fingerprint_v1(&item.request.configuration)
                        .map_err(platform_error)?;
                if hint != actual.as_str() {
                    return Err(ServiceError::new(
                        "L7W008_CONFIGURATION_IDENTITY_HINT_MISMATCH",
                        "configuration identity hint mismatch",
                    ));
                }
            }
        }
        let fingerprints = requests
            .iter()
            .map(|(id, item)| {
                (
                    id.clone(),
                    item.request
                        .candidate_snapshot
                        .snapshot_id
                        .as_str()
                        .to_owned(),
                )
            })
            .collect();
        let plan = workspace::plan(&graph, fingerprints);
        let mut results = Vec::new();
        let mut failed = false;
        for stage in &plan.stages {
            for package_id in &stage.packages {
                let item = requests.remove(package_id).expect("plan package");
                let session_id = manifest
                    .packages
                    .iter()
                    .find(|p| p.package_id == *package_id)
                    .expect("descriptor")
                    .session_id
                    .clone();
                if failed {
                    results.push(WorkspacePackageResultV1 {
                        package_id: package_id.clone(),
                        status: "skipped_fail_fast".into(),
                        snapshot_id: None,
                    });
                    continue;
                }
                match self.compile(&session_id, item.expected_commit_sequence, item.request) {
                    Ok(response) => results.push(WorkspacePackageResultV1 {
                        package_id: package_id.clone(),
                        status: "succeeded".into(),
                        snapshot_id: Some(response.snapshot.snapshot_id.to_string()),
                    }),
                    Err(_) => {
                        failed = true;
                        results.push(WorkspacePackageResultV1 {
                            package_id: package_id.clone(),
                            status: "failed".into(),
                            snapshot_id: None,
                        });
                    }
                }
            }
        }
        let status = if failed { "failed" } else { "succeeded" }.to_owned();
        let result = WorkspaceBuildResultV1 {
            workspace_id: manifest.workspace_id.clone(),
            status,
            manifest_identity: graph.manifest_identity.clone(),
            graph_identity: graph.graph_identity.clone(),
            plan_identity: plan.plan_identity.clone(),
            package_results: results,
        };
        if !failed {
            let state = DurableWorkspaceStateV1 {
                manifest: manifest.clone(),
                graph: graph.clone(),
                plan: plan.clone(),
                result: result.clone(),
            };
            self.publish_workspace_state(&state)?;
            self.workspaces.insert(manifest.workspace_id.clone(), state);
        }
        Ok(result)
    }
    pub fn inspect_workspace(
        &self,
        workspace_id: &str,
    ) -> Result<WorkspaceBuildResultV1, ServiceError> {
        self.workspaces
            .get(workspace_id)
            .map(|state| state.result.clone())
            .ok_or_else(|| ServiceError::new("workspace_not_found", "workspace state not found"))
    }
    pub fn verify_workspace(&self, workspace_id: &str) -> Result<(), ServiceError> {
        let state = self
            .workspaces
            .get(workspace_id)
            .ok_or_else(|| ServiceError::new("workspace_not_found", "workspace state not found"))?;
        if state.graph.manifest_identity != state.manifest.identity()
            || state.plan.manifest_identity != state.graph.manifest_identity
        {
            return Err(ServiceError::new(
                "workspace_state_invalid",
                "workspace state integrity mismatch",
            ));
        }
        Ok(())
    }
    pub fn remove_workspace_state(&mut self, workspace_id: &str) -> Result<(), ServiceError> {
        self.workspaces.remove(workspace_id);
        let path = self
            .root
            .join("workspaces")
            .join(format!("{workspace_id}.json"));
        if path.exists() {
            fs::remove_file(path).map_err(io_error)?;
        }
        Ok(())
    }
    pub fn compile(
        &mut self,
        session_id: &str,
        expected: u64,
        request: CompileRequest,
    ) -> Result<CompileResponse, ServiceError> {
        let root = self.root.clone();
        let session = self
            .sessions
            .get_mut(session_id)
            .ok_or_else(|| ServiceError::new("session_not_found", "session not loaded"))?;
        if session.closed {
            return Err(ServiceError::new("session_closed", "session is closed"));
        }
        if expected != session.commit_sequence {
            return Err(ServiceError::new(
                "stale_session_state",
                "commit sequence is stale",
            ));
        }
        platform::canonical_workspace_configuration_json_v1(&request.configuration)
            .map_err(|error| ServiceError::new("invalid_request", error.message))?;
        let workspace = WorkspaceInput {
            configuration: request.configuration.clone(),
            sources: request
                .sources
                .iter()
                .map(|source| WorkspaceSource {
                    path: source.path.clone(),
                    source: source.source.clone(),
                    language: source.language,
                })
                .collect(),
            compiler_contract: session.compiler_contract.clone(),
        };
        let derived = WorkspaceSnapshot::from_input(&workspace).map_err(platform_failure)?;
        if derived
            .to_canonical_json()
            .map_err(|error| ServiceError::new("invalid_request", error.message))?
            != request
                .candidate_snapshot
                .to_canonical_json()
                .map_err(|error| ServiceError::new("invalid_request", error.message))?
        {
            return Err(ServiceError::new(
                "source_revision_mismatch",
                "complete source input does not match candidate snapshot",
            ));
        }
        let raw_baseline = session.incremental_baseline.as_ref();
        let baseline_is_malformed = raw_baseline.is_some_and(|baseline| !baseline.is_consistent());
        let baseline = raw_baseline.filter(|_| !baseline_is_malformed);
        let mut plan = platform::plan_incremental_compilation_v1(
            baseline.map(|value| (value.snapshot.as_ref(), value.graph.as_ref())),
            &derived,
        );
        if baseline_is_malformed {
            platform::force_clean_fallback_v1(
                &mut plan,
                &derived,
                IncrementalFallbackReasonV1::MalformedBaselineGraph,
            );
        }
        let cache_input = CacheKeyInputV1 {
            compiler_contract: session.compiler_contract.clone(),
            configuration_fingerprint: derived.configuration_fingerprint.as_str().into(),
            source_universe_fingerprint: platform::source_universe_fingerprint_v1(&derived)
                .as_str()
                .into(),
            compile_mode: match request.mode {
                RequestedCompilationMode::Automatic => "automatic",
                RequestedCompilationMode::Full => "full",
            },
        };
        if plan.mode == IncrementalCompilationModeV1::NoChange {
            let Some(baseline) = baseline else {
                return Err(ServiceError::new(
                    "internal_invariant_failed",
                    "L5 no-change plan has no baseline",
                ));
            };
            baseline.snapshot.validate().map_err(platform_failure)?;
            baseline.graph.validate().map_err(platform_failure)?;
            let report = selected_report(
                request.incremental_report,
                &plan,
                Vec::new(),
                Vec::new(),
                "published",
                Some(true),
            );
            return Ok(CompileResponse {
                commit_sequence: session.commit_sequence,
                snapshot: Arc::clone(&baseline.snapshot),
                graph: Arc::clone(&baseline.graph),
                mode: plan.mode.as_str().into(),
                incremental_report: report,
                cache_report: selected_cache_report(
                    request.cache_report,
                    CacheTelemetryV1 {
                        enabled: self.cache.enabled(),
                        outcome: CacheOutcomeV1::NotChecked,
                        reasons: Vec::new(),
                        cache_key: cache_input.key(),
                        payload_length: None,
                        result_fingerprint: None,
                        entry_published: false,
                        entry_replaced: false,
                    },
                ),
            });
        }
        // A restored L6 baseline deliberately has no parser products. Do not
        // pretend it can support L5 durable partial reuse.
        if plan.mode == IncrementalCompilationModeV1::Incremental
            && baseline.is_some_and(|value| value.reusable_products.is_empty())
        {
            platform::force_clean_fallback_v1(
                &mut plan,
                &derived,
                IncrementalFallbackReasonV1::ReuseProductRejected,
            );
        }
        let (cache_hit, mut cache_telemetry) = self.cache.lookup(&cache_input);
        if let Some(hit) = cache_hit {
            let previous_configuration = session.configuration.clone();
            let previous_workspace_id = session.workspace_id.clone();
            let previous_sequence = session.commit_sequence;
            session.configuration = request.configuration.clone();
            session.workspace_id = derived.workspace_id.clone();
            session.commit_sequence += 1;
            if publish_commit(
                &root,
                session_id,
                session,
                session.commit_sequence,
                &hit.snapshot,
                &hit.graph,
            )
            .is_ok()
            {
                session.incremental_baseline = Some(baseline_from_result(
                    &session.configuration,
                    Arc::new(hit.snapshot.clone()),
                    Arc::new(hit.graph.clone()),
                    Vec::new(),
                    &session.compiler_contract,
                ));
                return Ok(CompileResponse {
                    commit_sequence: session.commit_sequence,
                    snapshot: Arc::new(hit.snapshot),
                    graph: Arc::new(hit.graph),
                    mode: hit.response_mode,
                    incremental_report: selected_report(
                        request.incremental_report,
                        &plan,
                        Vec::new(),
                        Vec::new(),
                        "published",
                        None,
                    ),
                    cache_report: selected_cache_report(request.cache_report, cache_telemetry),
                });
            }
            session.configuration = previous_configuration;
            session.workspace_id = previous_workspace_id;
            session.commit_sequence = previous_sequence;
            cache_telemetry.outcome = CacheOutcomeV1::Miss;
            cache_telemetry.reasons.push(
                crate::persistent_cache::CacheReasonCodeV1::CanonicalProductValidationFailure,
            );
        }
        let verification_workspace = request.verify_exact_equivalence.then(|| workspace.clone());
        let (outcome, reused, recomputed) =
            if plan.mode == IncrementalCompilationModeV1::Incremental {
                let reusable_products = baseline
                    .map(|value| value.reusable_products.clone())
                    .unwrap_or_default();
                let execution = session.l3.compile_workspace_incremental_v1(
                    IncrementalCompileWorkspaceRequestV1 {
                        workspace,
                        cancellation: CancellationToken::new(),
                        plan: plan.clone(),
                        reusable_products,
                    },
                );
                (
                    execution.outcome,
                    execution.reused_product_identities,
                    execution.recomputed_work_units,
                )
            } else {
                let outcome = session.l3.compile_workspace(CompileWorkspaceRequest {
                    workspace,
                    // A fallback is intentionally a clean canonical L3 compile.
                    mode: RequestedCompilationMode::Full,
                    cancellation: CancellationToken::new(),
                });
                (outcome, Vec::new(), plan.recompute_work_units.clone())
            };
        let CompilationOutcome::Committed(committed) = outcome else {
            return Err(ServiceError::new(
                "compiler_platform_failed",
                "compiler did not commit",
            ));
        };
        let equivalence = if let Some(clean_workspace) = verification_workspace {
            let mut clean = CompilerSessionState::new(
                derived.workspace_id.clone(),
                session.compiler_contract.clone(),
                CacheLimits::default(),
            );
            let clean = clean.compile_workspace(CompileWorkspaceRequest {
                workspace: clean_workspace,
                mode: RequestedCompilationMode::Full,
                cancellation: CancellationToken::new(),
            });
            let CompilationOutcome::Committed(clean) = clean else {
                return Err(ServiceError::new(
                    "incremental_equivalence_failed",
                    "isolated clean L3 compilation failed",
                ));
            };
            if clean
                .snapshot
                .to_canonical_json()
                .map_err(platform_serialization)?
                != committed
                    .snapshot
                    .to_canonical_json()
                    .map_err(platform_serialization)?
                || clean
                    .graph
                    .to_canonical_json()
                    .map_err(platform_serialization)?
                    != committed
                        .graph
                        .to_canonical_json()
                        .map_err(platform_serialization)?
            {
                return Err(ServiceError::new(
                    "incremental_equivalence_failed",
                    "first canonical mismatch is workspace snapshot or graph",
                ));
            }
            Some(true)
        } else {
            None
        };
        let previous_configuration = session.configuration.clone();
        let previous_workspace_id = session.workspace_id.clone();
        let previous_sequence = session.commit_sequence;
        session.configuration = request.configuration;
        session.workspace_id = derived.workspace_id.clone();
        session.commit_sequence += 1;
        if let Err(error) = publish_commit(
            &root,
            session_id,
            session,
            session.commit_sequence,
            &committed.snapshot,
            &committed.graph,
        ) {
            session.configuration = previous_configuration;
            session.workspace_id = previous_workspace_id;
            session.commit_sequence = previous_sequence;
            return Err(error);
        }
        session.incremental_baseline = Some(baseline_from_result(
            &session.configuration,
            Arc::clone(&committed.snapshot),
            Arc::clone(&committed.graph),
            committed.reusable_products,
            &session.compiler_contract,
        ));
        let cache_telemetry = self.cache.publish(
            &cache_input,
            &CachedCompileResultV1 {
                snapshot: (*committed.snapshot).clone(),
                graph: (*committed.graph).clone(),
                response_mode: plan.mode.as_str().into(),
            },
        );
        Ok(CompileResponse {
            commit_sequence: session.commit_sequence,
            snapshot: committed.snapshot,
            graph: committed.graph,
            mode: plan.mode.as_str().into(),
            incremental_report: selected_report(
                request.incremental_report,
                &plan,
                reused,
                recomputed,
                "published",
                equivalence,
            ),
            cache_report: selected_cache_report(request.cache_report, cache_telemetry),
        })
    }
    fn write_manifest(&self) -> Result<(), ServiceError> {
        atomic_write(&self.root.join("manifest.json"),format!("{{\"schema_version\":1,\"protocol_version\":1,\"persistence_version\":1,\"compiler_contract\":{},\"platform_contract\":\"presolve-platform-l3:1\"}}\n",json(&self.descriptor.compiler_contract.to_string())).as_bytes())
    }
    fn write_session(&self, id: &str, session: &DurableSession) -> Result<(), ServiceError> {
        let config = platform::canonical_workspace_configuration_json_v1(&session.configuration)
            .map_err(|error| ServiceError::new("persistence_io_failed", error.message))?;
        let config = String::from_utf8(config).map_err(|_| {
            ServiceError::new(
                "internal_invariant_failed",
                "non-UTF8 canonical configuration",
            )
        })?;
        let config = config.trim_end();
        let dir = self.session_dir(id);
        fs::create_dir_all(dir.join("commits")).map_err(io_error)?;
        atomic_write(&dir.join("session.json"),format!("{{\"schema_version\":1,\"session_id\":{},\"workspace_id\":{},\"workspace_configuration\":{},\"compiler_contract\":{},\"state\":{},\"current_commit_sequence\":{}}}\n",json(id),json(session.workspace_id.as_str()),config,json(session.compiler_contract.as_str()),json(if session.closed{"closed"}else{"open"}),session.commit_sequence).as_bytes())
    }
    fn publish_workspace_state(&self, state: &DurableWorkspaceStateV1) -> Result<(), ServiceError> {
        let directory = self.root.join("workspaces");
        fs::create_dir_all(&directory).map_err(io_error)?;
        let packages = state
            .result
            .package_results
            .iter()
            .map(|p| {
                format!(
                    "{{\"package_id\":{},\"status\":{},\"snapshot_id\":{}}}",
                    json(&p.package_id),
                    json(&p.status),
                    p.snapshot_id
                        .as_ref()
                        .map_or_else(|| "null".into(), |v| json(v))
                )
            })
            .collect::<Vec<_>>()
            .join(",");
        let bytes=format!("{{\"schema\":\"presolve.durable-workspace-state\",\"version\":1,\"workspace_id\":{},\"manifest_identity\":{},\"graph_identity\":{},\"plan_identity\":{},\"package_results\":[{}]}}\n",json(&state.result.workspace_id),json(&state.result.manifest_identity),json(&state.result.graph_identity),json(&state.result.plan_identity),packages);
        atomic_write(
            &directory.join(format!("{}.json", state.result.workspace_id)),
            bytes.as_bytes(),
        )
    }
}
fn selected_report(
    selector: IncrementalReportSelector,
    plan: &platform::IncrementalCompilationPlanV1,
    reused_product_identities: Vec<platform::ProductKey>,
    recomputed_work_unit_identities: Vec<SourceUnitId>,
    publication_outcome: &'static str,
    exact_equivalence_verified: Option<bool>,
) -> Option<IncrementalExecutionReportV1> {
    if selector == IncrementalReportSelector::None {
        return None;
    }
    let mut report = IncrementalExecutionReportV1 {
        schema: platform::INCREMENTAL_EXECUTION_REPORT_V1_SCHEMA,
        plan_fingerprint: plan.plan_fingerprint.clone(),
        mode: plan.mode,
        changed_inputs: plan.input_changes.clone(),
        invalidated_identities: plan.invalidation_closure.clone(),
        reused_product_identities,
        recomputed_work_unit_identities,
        fallback_reasons: plan.fallback_reasons.clone(),
        publication_outcome,
        exact_equivalence_verified,
    };
    if selector == IncrementalReportSelector::Summary {
        report.changed_inputs.clear();
        report.invalidated_identities.clear();
        report.reused_product_identities.clear();
        report.recomputed_work_unit_identities.clear();
        report.fallback_reasons.clear();
    }
    Some(report)
}
fn selected_cache_report(
    selector: CacheReportSelector,
    mut telemetry: CacheTelemetryV1,
) -> Option<CacheTelemetryV1> {
    if selector == CacheReportSelector::None {
        return None;
    }
    if selector == CacheReportSelector::Summary {
        telemetry.payload_length = None;
        telemetry.result_fingerprint = None;
        telemetry.entry_published = false;
        telemetry.entry_replaced = false;
    }
    Some(telemetry)
}
fn baseline_from_result(
    configuration: &WorkspaceConfiguration,
    snapshot: Arc<WorkspaceSnapshot>,
    graph: Arc<WorkspaceGraph>,
    reusable_products: Vec<CanonicalReusableProductV1>,
    compiler_contract: &ContractVersion,
) -> IncrementalBaseline {
    IncrementalBaseline {
        publication_identity: snapshot.snapshot_id.to_string(),
        configuration: configuration.clone(),
        source_fingerprints: snapshot
            .units
            .iter()
            .map(|unit| (unit.source_unit_id.clone(), unit.source_revision_id.clone()))
            .collect(),
        snapshot,
        graph,
        reusable_products,
        compiler_contract: compiler_contract.clone(),
    }
}
fn publish_commit(
    root: &Path,
    id: &str,
    session: &DurableSession,
    sequence: u64,
    snapshot: &WorkspaceSnapshot,
    graph: &WorkspaceGraph,
) -> Result<(), ServiceError> {
    let dir = session_directory(root, id);
    let tmp = dir.join(format!("commits/{sequence:020}.tmp"));
    let final_dir = dir.join(format!("commits/{sequence:020}"));
    fs::create_dir_all(&tmp).map_err(io_error)?;
    let snapshot = snapshot
        .to_canonical_json()
        .map_err(|error| ServiceError::new("commit_publication_failed", error.message))?;
    let graph = graph
        .to_canonical_json()
        .map_err(|error| ServiceError::new("commit_publication_failed", error.message))?;
    atomic_write(&tmp.join("workspace-snapshot.json"), &snapshot)?;
    atomic_write(&tmp.join("workspace-graph.json"), &graph)?;
    atomic_write(
        &tmp.join("products.json"),
        b"{\"schema_version\":1,\"products\":[]}\n",
    )?;
    let snapshot_id = snapshot_id(&snapshot)?;
    atomic_write(&tmp.join("commit.json"),format!("{{\"schema_version\":1,\"session_id\":{},\"commit_sequence\":{},\"workspace_id\":{},\"snapshot_id\":{},\"workspace_graph_schema_version\":1,\"compiler_contract\":{}}}\n",json(id),sequence,json(session.workspace_id.as_str()),json(&snapshot_id),json(session.compiler_contract.as_str())).as_bytes())?;
    fs::rename(&tmp, &final_dir).map_err(io_error)?;
    append_journal(
        root,
        id,
        sequence * 2,
        "commit_prepared",
        sequence,
        Some(&snapshot_id),
    )?;
    atomic_write(&dir.join("current"), format!("{sequence:020}\n").as_bytes())?;
    write_session_at(root, id, session)?;
    append_journal(
        root,
        id,
        sequence * 2 + 1,
        "commit_published",
        sequence,
        Some(&snapshot_id),
    )
}
fn append_journal(
    root: &Path,
    id: &str,
    sequence: u64,
    kind: &str,
    commit: u64,
    snapshot: Option<&str>,
) -> Result<(), ServiceError> {
    let mut file = OpenOptions::new()
        .create(true)
        .append(true)
        .open(session_directory(root, id).join("journal.ndjson"))
        .map_err(io_error)?;
    writeln!(file,"{{\"schema_version\":1,\"journal_sequence\":{},\"operation_id\":\"operation:service\",\"kind\":{},\"commit_sequence\":{},\"snapshot_id\":{}}}",sequence,json(kind),commit,snapshot.map_or("null".into(),json)).map_err(io_error)?;
    file.sync_all().map_err(io_error)
}
impl CompilerServiceHost {
    fn session_dir(&self, id: &str) -> PathBuf {
        self.root
            .join("sessions")
            .join(id.strip_prefix("session:sha256:").unwrap_or(id))
    }
}
fn session_directory(root: &Path, id: &str) -> PathBuf {
    root.join("sessions")
        .join(id.strip_prefix("session:sha256:").unwrap_or(id))
}
fn write_session_at(root: &Path, id: &str, session: &DurableSession) -> Result<(), ServiceError> {
    let config = platform::canonical_workspace_configuration_json_v1(&session.configuration)
        .map_err(|error| ServiceError::new("persistence_io_failed", error.message))?;
    let config = String::from_utf8(config).map_err(|_| {
        ServiceError::new(
            "internal_invariant_failed",
            "non-UTF8 canonical configuration",
        )
    })?;
    let dir = session_directory(root, id);
    fs::create_dir_all(dir.join("commits")).map_err(io_error)?;
    atomic_write(&dir.join("session.json"),format!("{{\"schema_version\":1,\"session_id\":{},\"workspace_id\":{},\"workspace_configuration\":{},\"compiler_contract\":{},\"state\":{},\"current_commit_sequence\":{}}}\n",json(id),json(session.workspace_id.as_str()),config.trim_end(),json(session.compiler_contract.as_str()),json(if session.closed{"closed"}else{"open"}),session.commit_sequence).as_bytes())
}
fn atomic_write(path: &Path, bytes: &[u8]) -> Result<(), ServiceError> {
    let tmp = path.with_extension("tmp");
    let mut file = File::create(&tmp).map_err(io_error)?;
    file.write_all(bytes).map_err(io_error)?;
    file.sync_all().map_err(io_error)?;
    fs::rename(tmp, path).map_err(io_error)
}
fn json(value: &str) -> String {
    serde_json::to_string(value).expect("strings serialize")
}
fn snapshot_id(bytes: &[u8]) -> Result<String, ServiceError> {
    let value: serde_json::Value = serde_json::from_slice(bytes)
        .map_err(|_| ServiceError::new("commit_publication_failed", "invalid snapshot bytes"))?;
    value
        .get("snapshot_id")
        .and_then(serde_json::Value::as_str)
        .map(str::to_owned)
        .ok_or_else(|| ServiceError::new("commit_publication_failed", "missing snapshot identity"))
}
fn io_error(error: std::io::Error) -> ServiceError {
    ServiceError::new("persistence_io_failed", error.to_string())
}
fn platform_error(error: platform::PlatformValidationError) -> ServiceError {
    ServiceError::new(error.code, error.message)
}
fn platform_failure(error: platform::PlatformFailure) -> ServiceError {
    ServiceError::new("compiler_platform_failed", error.message)
}
fn platform_serialization(error: platform::PlatformSerializationError) -> ServiceError {
    ServiceError::new("compiler_platform_failed", error.message)
}
fn cache_operation_error(reason: crate::persistent_cache::CacheReasonCodeV1) -> ServiceError {
    ServiceError::new("cache_operation_failed", reason.code())
}
fn workspace_error(error: workspace::WorkspaceErrorV1) -> ServiceError {
    ServiceError::new(error.code(), "workspace manifest validation failed")
}
fn watch_error(error: watch::WatchErrorV1) -> ServiceError {
    ServiceError::new(error.code(), "watch-session validation failed")
}
/// Candidate identity is derived from the complete normalized L7 request. It
/// binds all package snapshot identities (which bind exact source universes),
/// configuration identities, request modes, and expected publication state.
fn watch_candidate_fingerprint(
    request: &WorkspaceCompileRequestV1,
) -> Result<String, ServiceError> {
    let manifest = request
        .manifest
        .normalize_validate()
        .map_err(workspace_error)?;
    let mut packages = request.packages.iter().collect::<Vec<_>>();
    packages.sort_by(|a, b| a.package_id.cmp(&b.package_id));
    let mut encoded = format!("{}|{}|", manifest.identity(), request.operation_id);
    for package in packages {
        let config =
            platform::canonical_workspace_configuration_json_v1(&package.request.configuration)
                .map_err(platform_serialization)?;
        write!(
            encoded,
            "{}|{}|{}|{}|{:?}|{}|{:?}|{:?}|",
            package.package_id,
            package.expected_commit_sequence,
            package.request.candidate_snapshot.snapshot_id.as_str(),
            String::from_utf8_lossy(&config),
            package.request.mode,
            package.request.verify_exact_equivalence,
            package.request.incremental_report,
            package.request.cache_report,
        )
        .expect("write to string");
        let mut sources = package.request.sources.iter().collect::<Vec<_>>();
        sources.sort_by(|a, b| a.path.cmp(&b.path));
        for source in sources {
            write!(
                encoded,
                "{}\0{}\0{:?}|",
                source.path, source.source, source.language
            )
            .expect("write to string");
        }
    }
    Ok(format!("sha256:{:x}", Sha256::digest(encoded.as_bytes())))
}
fn workspace_result_identity(result: &WorkspaceBuildResultV1) -> String {
    let packages = result
        .package_results
        .iter()
        .map(|package| {
            format!(
                "{}:{}:{:?}",
                package.package_id, package.status, package.snapshot_id
            )
        })
        .collect::<Vec<_>>()
        .join("|");
    format!(
        "sha256:{:x}",
        Sha256::digest(
            format!(
                "{}|{}|{}|{}|{}|{}",
                result.workspace_id,
                result.status,
                result.manifest_identity,
                result.graph_identity,
                result.plan_identity,
                packages
            )
            .as_bytes()
        )
    )
}

pub mod protocol {
    pub use super::{
        decode_frame, encode_frame, CompilerServiceDescriptor, ServiceError, ServiceOperation,
        ServiceResponseStatus,
    };
}
pub mod host {
    pub use super::{
        CompileRequest, CompileResponse, CompilerServiceHost, CompleteSource,
        IncrementalExecutionReportV1, IncrementalReportSelector,
    };
}
pub mod session_store {
    pub use super::CompilerServiceHost;
}
pub mod journal {
    pub use super::CompilerServiceHost;
}
pub mod transport {
    pub use super::{decode_frame, encode_frame};
}
pub mod inspection {
    pub use super::CompilerServiceDescriptor;
}

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

    fn contract() -> ContractVersion {
        ContractVersion::new("presolve-compiler:0.1.0-alpha.8")
    }
    fn request(
        configuration: WorkspaceConfiguration,
        sources: Vec<(&str, &str)>,
        report: IncrementalReportSelector,
        verify: bool,
    ) -> CompileRequest {
        let input = WorkspaceInput {
            configuration: configuration.clone(),
            sources: sources
                .iter()
                .map(|(path, source)| WorkspaceSource {
                    path: (*path).into(),
                    source: (*source).into(),
                    language: None,
                })
                .collect(),
            compiler_contract: contract(),
        };
        CompileRequest {
            configuration,
            candidate_snapshot: WorkspaceSnapshot::from_input(&input).unwrap(),
            sources: sources
                .into_iter()
                .map(|(path, source)| CompleteSource {
                    path: path.into(),
                    source: source.into(),
                    language: None,
                })
                .collect(),
            mode: RequestedCompilationMode::Automatic,
            incremental_report: report,
            verify_exact_equivalence: verify,
            cache_report: CacheReportSelector::None,
        }
    }
    fn root(label: &str) -> std::path::PathBuf {
        let root = std::env::temp_dir().join(format!(
            "presolve-service-l5-{label}-{}",
            NEXT_SERVICE.fetch_add(1, Ordering::Relaxed)
        ));
        let _ = std::fs::remove_dir_all(&root);
        root
    }
    fn opened_host(
        root: &std::path::Path,
        configuration: &WorkspaceConfiguration,
    ) -> (CompilerServiceHost, String) {
        let workspace = platform::derive_workspace_id_v1(configuration).unwrap();
        let mut host = CompilerServiceHost::start(root, contract()).unwrap();
        let session = host
            .open_session(configuration.clone(), &workspace)
            .unwrap();
        (host, session)
    }
    #[test]
    fn l8_explicit_watch_session_delegates_one_complete_l7_candidate() {
        let root = root("l8-watch");
        let configuration = WorkspaceConfiguration::default();
        let (mut host, session) = opened_host(&root, &configuration);
        host.create_watch_session(watch::WatchSessionConfigurationV1 {
            schema: watch::WATCH_SESSION_CONFIGURATION_V1_SCHEMA.into(),
            version: 1,
            watch_session_id: "watch-test".into(),
            workspace_id: "watch-space".into(),
            debounce: watch::WatchDebounceV1 {
                quiet_period_milliseconds: 0,
                maximum_delay_milliseconds: 0,
            },
            supersession_policy: "cancel_obsolete".into(),
            event_detail: "summary".into(),
            event_journal_capacity: 16,
        })
        .unwrap();
        let package_request = request(
            configuration,
            vec![("src/App.ts", "export class App {}")],
            IncrementalReportSelector::Full,
            true,
        );
        let workspace_request = WorkspaceCompileRequestV1 {
            manifest: WorkspaceManifestV1 {
                schema: workspace::WORKSPACE_MANIFEST_V1_SCHEMA.into(),
                version: 1,
                workspace_id: "watch-space".into(),
                packages: vec![workspace::WorkspacePackageDescriptorV1 {
                    package_id: "app".into(),
                    session_id: session,
                    display_name: None,
                    configuration_identity_hint: None,
                    metadata: BTreeMap::new(),
                }],
                dependencies: vec![],
                policy: workspace::WorkspacePolicyV1 {
                    failure_mode: "fail_fast".into(),
                    execution_mode: "deterministic_serial".into(),
                    result_detail: "summary".into(),
                },
            },
            packages: vec![WorkspacePackageCompileRequestV1 {
                package_id: "app".into(),
                expected_commit_sequence: 0,
                request: package_request,
            }],
            operation_id: "watch-operation".into(),
        };
        host.submit_watch_change_batch(
            WatchChangeBatchV1 {
                schema: watch::WATCH_CHANGE_BATCH_V1_SCHEMA.into(),
                version: 1,
                watch_session_id: "watch-test".into(),
                sequence: 1,
                observed_changes: vec![watch::ObservedChangeV1 {
                    kind: "modified".into(),
                    logical_path: "src/App.ts".into(),
                    previous_logical_path: None,
                }],
                candidate: WatchCandidateV1 {
                    candidate_id: "candidate".into(),
                    fingerprint_hint: None,
                    verify_exact_equivalence: true,
                    workspace_request,
                },
            },
            0,
        )
        .unwrap();
        let report = host
            .run_watch_scheduler_turn("watch-test", 0)
            .unwrap()
            .unwrap();
        assert_eq!(report.outcome, "succeeded");
        let snapshot = host.inspect_watch_session("watch-test").unwrap();
        assert_eq!(snapshot.pending_count, 0);
        assert!(snapshot.last_successful_workspace_result_identity.is_some());
        assert!(!String::from_utf8_lossy(&read_tree(&root)).contains("export class App"));
        std::fs::remove_dir_all(root).unwrap();
    }
    fn read_tree(path: &Path) -> Vec<u8> {
        if path.is_file() {
            return std::fs::read(path).unwrap();
        }
        std::fs::read_dir(path)
            .unwrap()
            .flat_map(|entry| read_tree(&entry.unwrap().path()))
            .collect()
    }

    #[test]
    fn frames_are_exactly_length_delimited() {
        let frame = encode_frame(br#"{"protocol_version":1}"#).unwrap();
        assert_eq!(decode_frame(&frame).unwrap(), br#"{"protocol_version":1}"#);
        assert!(decode_frame(b"00000000:\n").is_err());
        assert!(decode_frame(b"00000002:{}x\n").is_err());
    }

    #[test]
    fn complete_candidate_commit_is_durable_without_source_persistence() {
        let root = std::env::temp_dir().join(format!(
            "presolve-service-test-{}",
            NEXT_SERVICE.fetch_add(1, Ordering::Relaxed)
        ));
        let configuration = WorkspaceConfiguration::default();
        let workspace = platform::derive_workspace_id_v1(&configuration).unwrap();
        let contract = ContractVersion::new("presolve-compiler:0.1.0-alpha.8");
        let mut host = CompilerServiceHost::start(&root, contract.clone()).unwrap();
        let session = host
            .open_session(configuration.clone(), &workspace)
            .unwrap();
        let input = WorkspaceInput {
            configuration: configuration.clone(),
            sources: vec![WorkspaceSource {
                path: "src/App.tsx".into(),
                source: "export class App {}".into(),
                language: None,
            }],
            compiler_contract: contract,
        };
        let snapshot = WorkspaceSnapshot::from_input(&input).unwrap();
        let result = host
            .compile(
                &session,
                0,
                CompileRequest {
                    configuration,
                    candidate_snapshot: snapshot,
                    sources: vec![CompleteSource {
                        path: "src/App.tsx".into(),
                        source: "export class App {}".into(),
                        language: None,
                    }],
                    mode: RequestedCompilationMode::Full,
                    incremental_report: IncrementalReportSelector::None,
                    verify_exact_equivalence: false,
                    cache_report: CacheReportSelector::None,
                },
            )
            .unwrap();
        assert_eq!(result.commit_sequence, 1);
        let persisted = root
            .join("service/sessions")
            .join(session.strip_prefix("session:sha256:").unwrap())
            .join("commits/00000000000000000001");
        assert!(persisted.join("workspace-snapshot.json").is_file());
        assert!(persisted.join("workspace-graph.json").is_file());
        assert!(
            std::fs::metadata(persisted.join("commit.json"))
                .unwrap()
                .len()
                > 0
        );
        assert!(!String::from_utf8_lossy(&read_tree(&persisted)).contains("export class App {}"));
        std::fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn l5_content_edit_reuses_validated_parse_products_and_equals_clean() {
        let root = root("content");
        let configuration = WorkspaceConfiguration::default();
        let (mut host, session) = opened_host(&root, &configuration);
        let baseline = vec![
            ("src/Dependency.ts", "export class Dependency {}"),
            (
                "src/App.ts",
                "import { Dependency } from './Dependency'; export class App {}",
            ),
            ("src/Stable.ts", "export class Stable {}"),
        ];
        let first = host
            .compile(
                &session,
                0,
                request(
                    configuration.clone(),
                    baseline.clone(),
                    IncrementalReportSelector::Full,
                    true,
                ),
            )
            .unwrap();
        assert_eq!(first.mode, "cold");
        let report = first.incremental_report.unwrap();
        assert_eq!(report.mode, IncrementalCompilationModeV1::Cold);
        assert_eq!(report.exact_equivalence_verified, Some(true));
        let candidate = vec![
            (
                "src/Dependency.ts",
                "export class Dependency { value = 1; }",
            ),
            (
                "src/App.ts",
                "import { Dependency } from './Dependency'; export class App {}",
            ),
            ("src/Stable.ts", "export class Stable {}"),
        ];
        let second = host
            .compile(
                &session,
                1,
                request(
                    configuration.clone(),
                    candidate.clone(),
                    IncrementalReportSelector::Full,
                    true,
                ),
            )
            .unwrap();
        let report = second.incremental_report.unwrap();
        assert_eq!(second.mode, "incremental");
        assert!(!report.reused_product_identities.is_empty());
        assert_eq!(report.exact_equivalence_verified, Some(true));
        let no_change = host
            .compile(
                &session,
                2,
                request(
                    configuration,
                    candidate,
                    IncrementalReportSelector::Full,
                    true,
                ),
            )
            .unwrap();
        assert_eq!(no_change.commit_sequence, 2);
        assert_eq!(no_change.mode, "no_change");
        assert_eq!(
            no_change.snapshot.to_canonical_json().unwrap(),
            second.snapshot.to_canonical_json().unwrap()
        );
        host.close_session(&session).unwrap();
        let closed = host.sessions.get(&session).unwrap();
        assert!(closed.closed);
        assert!(closed.incremental_baseline.is_none());
        std::fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn l5_service_restart_has_no_durable_baseline() {
        let root = root("restart");
        let configuration = WorkspaceConfiguration::default();
        let baseline = vec![("src/App.ts", "export class App {}")];
        let (mut first, first_session) = opened_host(&root, &configuration);
        first
            .compile(
                &first_session,
                0,
                request(
                    configuration.clone(),
                    baseline.clone(),
                    IncrementalReportSelector::None,
                    false,
                ),
            )
            .unwrap();
        drop(first);
        let (mut restarted, session) = opened_host(&root, &configuration);
        let response = restarted
            .compile(
                &session,
                0,
                request(
                    configuration,
                    baseline,
                    IncrementalReportSelector::Full,
                    true,
                ),
            )
            .unwrap();
        assert_eq!(response.mode, "cold");
        assert!(response
            .incremental_report
            .unwrap()
            .reused_product_identities
            .is_empty());
        std::fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn l6_persistent_complete_result_cache_hits_after_restart_without_source_text() {
        let root = root("l6-service");
        let cache_root = root.join("explicit-cache");
        let configuration = WorkspaceConfiguration::default();
        let workspace = platform::derive_workspace_id_v1(&configuration).unwrap();
        let sentinel = "L6_SOURCE_SENTINEL_IDENTIFIER_COMMENT_STRING";
        let source = format!("// {sentinel}\nexport class App {{ value = '{sentinel}'; }}");
        let mut first =
            CompilerServiceHost::start_with_cache(&root, Some(&cache_root), contract()).unwrap();
        let session = first
            .open_session(configuration.clone(), &workspace)
            .unwrap();
        let mut clean_request = request(
            configuration.clone(),
            vec![("src/App.ts", source.as_str())],
            IncrementalReportSelector::None,
            true,
        );
        clean_request.cache_report = CacheReportSelector::Full;
        let clean = first.compile(&session, 0, clean_request).unwrap();
        assert!(clean.cache_report.as_ref().unwrap().entry_published);
        assert!(cache_root.join("manifest.json").is_file());
        drop(first);

        let mut restarted =
            CompilerServiceHost::start_with_cache(&root, Some(&cache_root), contract()).unwrap();
        let session = restarted
            .open_session(configuration.clone(), &workspace)
            .unwrap();
        let mut hit_request = request(
            configuration,
            vec![("src/App.ts", source.as_str())],
            IncrementalReportSelector::None,
            false,
        );
        hit_request.cache_report = CacheReportSelector::Full;
        let hit = restarted.compile(&session, 0, hit_request).unwrap();
        assert_eq!(
            hit.cache_report.as_ref().unwrap().outcome,
            CacheOutcomeV1::Hit
        );
        assert_eq!(
            clean.snapshot.to_canonical_json().unwrap(),
            hit.snapshot.to_canonical_json().unwrap()
        );
        assert_eq!(
            clean.graph.to_canonical_json().unwrap(),
            hit.graph.to_canonical_json().unwrap()
        );
        let cache_bytes = read_tree(&cache_root);
        assert!(!String::from_utf8_lossy(&cache_bytes).contains(sentinel));
        let report = restarted.inspect_cache(&cache_root).unwrap();
        assert_eq!(report.valid_keys.len(), 1);
        assert_eq!(restarted.clean_cache(&cache_root).unwrap().len(), 1);
        std::fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn l6_corruption_and_disabled_cache_fall_back_to_l5() {
        let root = root("l6-corrupt");
        let cache_root = root.join("explicit-cache");
        let configuration = WorkspaceConfiguration::default();
        let workspace = platform::derive_workspace_id_v1(&configuration).unwrap();
        let sources = vec![("src/App.ts", "export class App {}")];
        let mut host =
            CompilerServiceHost::start_with_cache(&root, Some(&cache_root), contract()).unwrap();
        let session = host
            .open_session(configuration.clone(), &workspace)
            .unwrap();
        host.compile(
            &session,
            0,
            request(
                configuration.clone(),
                sources.clone(),
                IncrementalReportSelector::None,
                false,
            ),
        )
        .unwrap();
        drop(host);
        let payload = cache_root.join("entries");
        let prefix = std::fs::read_dir(&payload)
            .unwrap()
            .next()
            .unwrap()
            .unwrap()
            .path();
        let entry = std::fs::read_dir(prefix)
            .unwrap()
            .next()
            .unwrap()
            .unwrap()
            .path()
            .join("payload.bin");
        std::fs::write(&entry, b"corrupt").unwrap();
        let mut restarted =
            CompilerServiceHost::start_with_cache(&root, Some(&cache_root), contract()).unwrap();
        let session = restarted
            .open_session(configuration.clone(), &workspace)
            .unwrap();
        let mut corrupt_request = request(
            configuration.clone(),
            sources.clone(),
            IncrementalReportSelector::None,
            true,
        );
        corrupt_request.cache_report = CacheReportSelector::Full;
        let result = restarted.compile(&session, 0, corrupt_request).unwrap();
        assert_ne!(result.cache_report.unwrap().outcome, CacheOutcomeV1::Hit);
        drop(restarted);
        let mut disabled = CompilerServiceHost::start(&root, contract()).unwrap();
        let session = disabled.open_session(configuration, &workspace).unwrap();
        let mut request = request(
            WorkspaceConfiguration::default(),
            sources,
            IncrementalReportSelector::None,
            true,
        );
        request.cache_report = CacheReportSelector::Full;
        assert_eq!(
            disabled
                .compile(&session, 0, request)
                .unwrap()
                .cache_report
                .unwrap()
                .outcome,
            CacheOutcomeV1::Miss
        );
        std::fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn l5_add_delete_configuration_and_malformed_baselines_clean_fallback() {
        let root = root("fallback");
        let configuration = WorkspaceConfiguration::default();
        let (mut host, session) = opened_host(&root, &configuration);
        let baseline = vec![("src/App.ts", "export class App {}")];
        host.compile(
            &session,
            0,
            request(
                configuration.clone(),
                baseline.clone(),
                IncrementalReportSelector::None,
                false,
            ),
        )
        .unwrap();
        let added = host
            .compile(
                &session,
                1,
                request(
                    configuration.clone(),
                    vec![
                        ("src/App.ts", "export class App {}"),
                        ("src/Added.ts", "export class Added {}"),
                    ],
                    IncrementalReportSelector::Full,
                    true,
                ),
            )
            .unwrap();
        let report = added.incremental_report.unwrap();
        assert_eq!(added.mode, "clean_fallback");
        assert!(report
            .fallback_reasons
            .contains(&IncrementalFallbackReasonV1::SourceUniverseMembershipUnmodeled));
        let mut changed_configuration = configuration.clone();
        changed_configuration.feature_flags.push("strict".into());
        let config_changed = host
            .compile(
                &session,
                2,
                request(
                    changed_configuration,
                    vec![
                        ("src/App.ts", "export class App {}"),
                        ("src/Added.ts", "export class Added {}"),
                    ],
                    IncrementalReportSelector::Full,
                    true,
                ),
            )
            .unwrap();
        assert!(config_changed
            .incremental_report
            .unwrap()
            .fallback_reasons
            .contains(&IncrementalFallbackReasonV1::ConfigurationChanged));
        host.sessions
            .get_mut(&session)
            .unwrap()
            .incremental_baseline
            .as_mut()
            .unwrap()
            .publication_identity = "malformed".into();
        let malformed = host
            .compile(
                &session,
                3,
                request(
                    configuration,
                    baseline,
                    IncrementalReportSelector::Full,
                    true,
                ),
            )
            .unwrap();
        assert!(malformed
            .incremental_report
            .unwrap()
            .fallback_reasons
            .contains(&IncrementalFallbackReasonV1::MalformedBaselineGraph));
        std::fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn l5_failure_isolation_no_persistence_and_twenty_run_determinism() {
        let configuration = WorkspaceConfiguration::default();
        let baseline = vec![
            ("src/Dependency.ts", "export class Dependency {}"),
            (
                "src/App.ts",
                "import { Dependency } from './Dependency'; export class App {}",
            ),
        ];
        let candidate = vec![
            (
                "src/Dependency.ts",
                "export class Dependency { value = 1; }",
            ),
            (
                "src/App.ts",
                "import { Dependency } from './Dependency'; export class App {}",
            ),
        ];
        let mut expected_report = None;
        for run in 0..20 {
            let root = root("determinism");
            let (mut host, session) = opened_host(&root, &configuration);
            host.compile(
                &session,
                0,
                request(
                    configuration.clone(),
                    baseline.clone(),
                    IncrementalReportSelector::None,
                    false,
                ),
            )
            .unwrap();
            let mut invalid = request(
                configuration.clone(),
                baseline.clone(),
                IncrementalReportSelector::None,
                false,
            );
            invalid.sources = vec![
                CompleteSource {
                    path: "src/App.ts".into(),
                    source: "export class App {}".into(),
                    language: None,
                },
                CompleteSource {
                    path: "src/App.ts".into(),
                    source: "export class Duplicate {}".into(),
                    language: None,
                },
            ];
            assert!(host.compile(&session, 1, invalid).is_err());
            let response = host
                .compile(
                    &session,
                    1,
                    request(
                        configuration.clone(),
                        candidate.clone(),
                        IncrementalReportSelector::Full,
                        true,
                    ),
                )
                .unwrap();
            let report = response.incremental_report.unwrap().to_canonical_json();
            if let Some(expected) = &expected_report {
                assert_eq!(&report, expected, "determinism run {run}");
            } else {
                expected_report = Some(report);
            }
            let persisted = read_tree(&root.join("service/sessions"));
            assert!(!String::from_utf8_lossy(&persisted).contains("value = 1"));
            std::fs::remove_dir_all(root).unwrap();
        }
    }
}