From fc4b9fbda7ea82af2eeb1ac0039758bbb07d0c19 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 24 Oct 2026 23:36:48 -0700 Subject: [PATCH 1/9] simplify(diagnostics): publish compiler status through the project owner The TypeScript producer recorded each run into a process-global, path-keyed LAST_RUNS map that nothing ever removed, and availability reads joined it by pathname. A successor owner at the same path could therefore report its predecessor's run as its own. Each project owner now creates one TypeScriptProducerStatusHandleV1 at open. The producer publishes into it and the owner's primitive runtime reads from it; the global map, last_record, last_producer_run and the path-only status API are deleted. Teardown retires the status with the owner. Co-Authored-By: Claude Opus 5.5 --- .../src/diagnostics_producer.rs | 179 ++++++++++++------ .../src/primitives/production.rs | 6 + .../src/primitives/runtime.rs | 57 ++++-- .../src/daemon/project_open_owners.rs | 13 ++ .../compiler_diagnostics_producer.rs | 6 +- .../project_open_owners/primitive_runtime.rs | 7 + 6 files changed, 192 insertions(+), 76 deletions(-) diff --git a/crates/tracedecay-application/src/diagnostics_producer.rs b/crates/tracedecay-application/src/diagnostics_producer.rs index a04c0fff56..c42994382f 100644 --- a/crates/tracedecay-application/src/diagnostics_producer.rs +++ b/crates/tracedecay-application/src/diagnostics_producer.rs @@ -11,7 +11,7 @@ //! generation carries one snapshot of all their merged findings, but a project //! is rechecked only when the generation changed a file that can change what //! tsc reports for it; the rest republish their previous result without -//! running tsc. Each run's outcome is recorded per project root, and each +//! running tsc. Each run's outcome is published to the owning project's status handle, and each //! tsconfig's check within it, so a read can name the real state of the //! project that owns the file it asked about (still running, compiler failed, //! compiler missing, nothing resolved) instead of a bare "no producer" or a @@ -20,7 +20,7 @@ use std::collections::{BTreeMap, BTreeSet}; use std::panic::AssertUnwindSafe; use std::path::{Path, PathBuf}; -use std::sync::{Arc, Mutex, OnceLock}; +use std::sync::{Arc, RwLock}; use futures_util::FutureExt; use futures_util::future::join_all; @@ -82,54 +82,77 @@ pub enum TypeScriptProjectCheckV1 { }, } -#[derive(Clone)] -struct ProducerRecord { +/// The latest producer outcome of one project owner: the run and how each +/// tsconfig fared in it. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct TypeScriptProducerStatusV1 { run: CompilerProducerRunV1, checks: BTreeMap, } -fn last_runs() -> &'static Mutex> { - static LAST_RUNS: OnceLock>> = OnceLock::new(); - LAST_RUNS.get_or_init(|| Mutex::new(BTreeMap::new())) -} +impl TypeScriptProducerStatusV1 { + #[must_use] + pub fn new( + run: CompilerProducerRunV1, + checks: BTreeMap, + ) -> Self { + Self { run, checks } + } -fn last_record(project_root: &Path) -> Option { - last_runs() - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .get(project_root) - .cloned() + #[must_use] + pub fn run(&self) -> &CompilerProducerRunV1 { + &self.run + } } -/// The most recent producer run recorded for `project_root`, if any. -#[must_use] -pub fn last_producer_run(project_root: &Path) -> Option { - last_record(project_root).map(|record| record.run) +/// Read view of one project owner's producer status. The producer task and the +/// owner's diagnostics reads share one handle created with the owner, so a +/// successor owner at the same path starts with no status and a superseded +/// producer can only publish into the handle it was given. +#[derive(Clone, Debug, Default)] +pub struct TypeScriptProducerStatusHandleV1 { + latest: Arc>>>, } -fn record_producer_run( - project_root: &Path, - run: &CompilerProducerRunV1, - checks: BTreeMap, -) { - let record = ProducerRecord { - run: run.clone(), - checks, - }; - last_runs() - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .insert(project_root.to_path_buf(), record); +impl TypeScriptProducerStatusHandleV1 { + fn publish(&self, status: TypeScriptProducerStatusV1) { + *self + .latest + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(Arc::new(status)); + } + + /// The most recent producer status, if the producer has run. + #[must_use] + pub fn current(&self) -> Option> { + self.latest + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone() + } } /// What one project root's producer remembers between generations: the file /// digests of the generation it last checked and how each project fared. -#[derive(Default)] pub struct TypeScriptProducerStateV1 { + status: TypeScriptProducerStatusHandleV1, checked_files: Option>, projects: BTreeMap, } +impl TypeScriptProducerStateV1 { + /// State for one project owner's producer, publishing its status into + /// `status`. + #[must_use] + pub fn new(status: TypeScriptProducerStatusHandleV1) -> Self { + Self { + status, + checked_files: None, + projects: BTreeMap::new(), + } + } +} + #[derive(Clone)] enum ProjectResult { Checked { @@ -154,8 +177,8 @@ const CONCURRENT_CHECKS: usize = 4; /// `build_info_dir`, only when `identity` changed a file that can change what /// tsc reports for it since the generation `state` last checked; every other /// project republishes its previous result without running tsc. The outcome -/// is recorded for [`last_producer_run`], and each tsconfig's check for -/// [`typescript_diagnostics_availability`]. +/// and each tsconfig's check are published to the status handle `state` was +/// built with, which [`typescript_diagnostics_availability`] reads. /// /// A clean project publishes an empty snapshot, which is what lets a read /// answer "no diagnostics" as evidence instead of as a missing producer. @@ -185,7 +208,9 @@ pub async fn run_typescript_producer_v1( &mut checks, ) .await; - record_producer_run(project_root, &run, checks); + state + .status + .publish(TypeScriptProducerStatusV1::new(run.clone(), checks)); run } @@ -477,11 +502,12 @@ pub enum TypeScriptDiagnosticsAvailabilityV1 { /// Resolves what a read can truthfully say about the TypeScript producer: the /// project owning `file` (or, for a workspace read, the first checkable one), -/// its filesystem state, and the last recorded run. +/// its filesystem state, and the owner's last published run. #[must_use] pub fn typescript_diagnostics_availability( project_root: &Path, file: Option<&Path>, + status: &TypeScriptProducerStatusHandleV1, ) -> TypeScriptDiagnosticsAvailabilityV1 { let project = match file { Some(file) => match typescript_file_owner(project_root, file) { @@ -504,12 +530,12 @@ pub fn typescript_diagnostics_availability( let TypeScriptProject { tsconfig, compiler } = project; match compiler { Some(compiler) => { - let record = last_record(project_root); + let status = status.current(); TypeScriptDiagnosticsAvailabilityV1::Configured { - check: record + check: status .as_ref() - .and_then(|record| record.checks.get(&tsconfig).cloned()), - last_run: record.map(|record| record.run), + .and_then(|status| status.checks.get(&tsconfig).cloned()), + last_run: status.map(|status| status.run.clone()), tsconfig, compiler, } @@ -542,9 +568,9 @@ mod tests { use tracedecay_runtime_core::test_executable::write_executable_script; use super::{ - CompilerProducerRunV1, TypeScriptDiagnosticsAvailabilityV1, TypeScriptProjectCheckV1, - compiler_diagnostic, last_producer_run, record_producer_run, - typescript_diagnostics_availability, + CompilerProducerRunV1, TypeScriptDiagnosticsAvailabilityV1, + TypeScriptProducerStatusHandleV1, TypeScriptProducerStatusV1, TypeScriptProjectCheckV1, + compiler_diagnostic, typescript_diagnostics_availability, }; #[cfg(unix)] use super::{TypeScriptProducerStateV1, run_typescript_producer_v1}; @@ -585,16 +611,21 @@ mod tests { } #[test] - fn the_last_run_is_recorded_per_project_root() { - let root = tempfile::tempdir().expect("project root"); - assert_eq!(last_producer_run(root.path()), None); + fn a_status_handle_publishes_to_its_owner_only() { + let owner = TypeScriptProducerStatusHandleV1::default(); + assert_eq!(owner.current(), None); let run = CompilerProducerRunV1::CompilerFailed { reason: "tsc exited with 2".to_owned(), }; - record_producer_run(root.path(), &run, BTreeMap::new()); - assert_eq!(last_producer_run(root.path()), Some(run)); - let other = tempfile::tempdir().expect("other root"); - assert_eq!(last_producer_run(other.path()), None); + owner.publish(TypeScriptProducerStatusV1::new( + run.clone(), + BTreeMap::new(), + )); + assert_eq!( + owner.current().map(|status| status.run().clone()), + Some(run) + ); + assert_eq!(TypeScriptProducerStatusHandleV1::default().current(), None); } /// A read about a file reports the check of the tsconfig that owns it, @@ -615,9 +646,9 @@ mod tests { let failed = TypeScriptProjectCheckV1::Failed { reason: "TS5083: Cannot read file".to_owned(), }; - record_producer_run( - root, - &CompilerProducerRunV1::CodeIndexGenerationUnavailable, + let status = TypeScriptProducerStatusHandleV1::default(); + status.publish(TypeScriptProducerStatusV1::new( + CompilerProducerRunV1::CodeIndexGenerationUnavailable, BTreeMap::from([ ( root.join("packages/app/tsconfig.json"), @@ -625,11 +656,15 @@ mod tests { ), (lib.clone(), failed.clone()), ]), - ); + )); let TypeScriptDiagnosticsAvailabilityV1::Configured { tsconfig, check, .. - } = typescript_diagnostics_availability(root, Some(Path::new("packages/lib/src/a.ts"))) + } = typescript_diagnostics_availability( + root, + Some(Path::new("packages/lib/src/a.ts")), + &status, + ) else { panic!("lib owns its sources and the root compiler checks it"); }; @@ -797,7 +832,7 @@ mod tests { #[tokio::test] async fn only_projects_a_generation_changed_are_rechecked() { let fixture = ProducerFixture::new(); - let mut state = TypeScriptProducerStateV1::default(); + let mut state = TypeScriptProducerStateV1::new(TypeScriptProducerStatusHandleV1::default()); let both = vec![ ( "file.packages.app.src.index.ts".to_owned(), @@ -847,4 +882,36 @@ mod tests { ); assert_eq!(fixture.checked(), ["broken"]); } + + /// A producer run reports to the handle of the owner it was built for. A + /// successor owner opened at the same path reads no status until its own + /// producer runs, so the predecessor's run is never taken for its own. + #[cfg(unix)] + #[tokio::test] + async fn a_successor_owner_does_not_inherit_the_predecessors_status() { + let fixture = ProducerFixture::new(); + let predecessor = TypeScriptProducerStatusHandleV1::default(); + let mut state = TypeScriptProducerStateV1::new(predecessor.clone()); + let run = fixture.run("generation.producer.1", &mut state).await; + assert_eq!( + predecessor.current().map(|status| status.run().clone()), + Some(run) + ); + + let successor = TypeScriptProducerStatusHandleV1::default(); + let file = Path::new("packages/app/src/index.ts"); + let TypeScriptDiagnosticsAvailabilityV1::Configured { + check, last_run, .. + } = typescript_diagnostics_availability(&fixture.root, Some(file), &successor) + else { + panic!("app is checkable"); + }; + assert_eq!((check, last_run), (None, None)); + let TypeScriptDiagnosticsAvailabilityV1::Configured { check, .. } = + typescript_diagnostics_availability(&fixture.root, Some(file), &predecessor) + else { + panic!("app is checkable"); + }; + assert_eq!(check, Some(TypeScriptProjectCheckV1::Checked)); + } } diff --git a/crates/tracedecay-application/src/primitives/production.rs b/crates/tracedecay-application/src/primitives/production.rs index d97e35deea..4e74e9cda5 100644 --- a/crates/tracedecay-application/src/primitives/production.rs +++ b/crates/tracedecay-application/src/primitives/production.rs @@ -30,6 +30,7 @@ use super::runtime::{ }; use super::symbol_graph::SymbolGraphCursorPort; use crate::code_index::CodeIndexIgnoredDependencyAdmissionPortV1; +use crate::diagnostics_producer::TypeScriptProducerStatusHandleV1; use crate::diagnostics_publication::CodeIndexPublicationIdentityPortV1; use crate::diagnostics_query::DiagnosticQueryCursor; use crate::lsp_runtime::LspCodeIndexProjectionIdentityPort; @@ -868,6 +869,7 @@ pub struct ProductionPrimitiveOpenRequestV1 { convergence_park: Arc, access: ProjectSourceAccessSnapshot, admitted_root_uri: String, + compiler_status: TypeScriptProducerStatusHandleV1, } impl ProductionPrimitiveOpenRequestV1 { @@ -878,6 +880,7 @@ impl ProductionPrimitiveOpenRequestV1 { temporal: Arc, access: ProjectSourceAccessSnapshot, admitted_root_uri: String, + compiler_status: TypeScriptProducerStatusHandleV1, ) -> Self { Self { source_runtime, @@ -890,6 +893,7 @@ impl ProductionPrimitiveOpenRequestV1 { convergence_park: code.convergence_park, access, admitted_root_uri, + compiler_status, } } } @@ -910,6 +914,7 @@ pub async fn open_production_primitive_runtime( convergence_park, access, admitted_root_uri, + compiler_status, } = request; let database = source_runtime.db().clone(); let project_root = source_runtime.project_root().to_path_buf(); @@ -987,6 +992,7 @@ pub async fn open_production_primitive_runtime( test_runs, test_run_scope, convergence_park, + compiler_status, ) } diff --git a/crates/tracedecay-application/src/primitives/runtime.rs b/crates/tracedecay-application/src/primitives/runtime.rs index 1175a62fff..cee53d825a 100644 --- a/crates/tracedecay-application/src/primitives/runtime.rs +++ b/crates/tracedecay-application/src/primitives/runtime.rs @@ -50,8 +50,8 @@ use super::symbol_graph::{CanonicalSymbolGraphAdapter, SymbolGraphCursorPort}; use crate::ProjectSourceAccessSnapshot; use crate::code_index::CodeIndexIgnoredDependencyAdmissionPortV1; use crate::diagnostics_producer::{ - CompilerProducerRunV1, TypeScriptDiagnosticsAvailabilityV1, TypeScriptProjectCheckV1, - typescript_diagnostics_availability, + CompilerProducerRunV1, TypeScriptDiagnosticsAvailabilityV1, TypeScriptProducerStatusHandleV1, + TypeScriptProjectCheckV1, typescript_diagnostics_availability, }; use crate::managed_test_runs::{ ManagedTestRunCurrentScope, ManagedTestRunReadOutcome, ManagedTestRunStaleReason, @@ -344,6 +344,9 @@ pub struct OwnedPrimitiveRuntime { /// The filesystem root `admitted_root_uri` names; diagnostics reads probe /// the producer state here when no publication exists. admitted_project_root: PathBuf, + /// This owner's TypeScript producer status; diagnostics reads that find no + /// publication describe it, never another owner's run at the same path. + compiler_status: TypeScriptProducerStatusHandleV1, test_runs: RetainedManagedTestRuns, test_run_scope: Arc, convergence_park: Arc, @@ -538,6 +541,7 @@ pub fn open_primitive_project_runtime( test_runs: RetainedManagedTestRuns, test_run_scope: Arc, convergence_park: Arc, + compiler_status: TypeScriptProducerStatusHandleV1, ) -> Result { scope.validate()?; let admitted_project_root = validate_admitted_root_uri(&admitted_root_uri)?; @@ -575,6 +579,7 @@ pub fn open_primitive_project_runtime( access, admitted_root_uri, admitted_project_root, + compiler_status, test_runs, test_run_scope, convergence_park, @@ -928,6 +933,7 @@ async fn dispatch_admitted( evidence.omissions.first().map(|omission| omission.reason), &runtime.admitted_project_root, file, + &runtime.compiler_status, ); } // Another package's publication is no evidence about a file whose @@ -937,8 +943,11 @@ async fn dispatch_admitted( .payload .as_ref() .is_some_and(|result| result.diagnostics.is_empty()) - && let Some(unchecked) = - unchecked_owner_problem(&runtime.admitted_project_root, file)? + && let Some(unchecked) = unchecked_owner_problem( + &runtime.admitted_project_root, + file, + &runtime.compiler_status, + )? { return problem(&context, &operation, unchecked); } @@ -1849,11 +1858,12 @@ fn diagnostics_unavailable_problem( reason: Option, project_root: &Path, file: Option<&Path>, + status: &TypeScriptProducerStatusHandleV1, ) -> Result, ApplicationContractError> { problem( context, operation, - diagnostics_absence_problem(reason, project_root, file)?, + diagnostics_absence_problem(reason, project_root, file, status)?, ) } @@ -1867,6 +1877,7 @@ fn diagnostics_absence_problem( reason: Option, project_root: &Path, file: Option<&Path>, + status: &TypeScriptProducerStatusHandleV1, ) -> Result { let (code, message) = match reason { Some(OmissionReason::Stale) => ( @@ -1875,7 +1886,7 @@ fn diagnostics_absence_problem( .to_owned(), ), Some(OmissionReason::Unsupported) => { - return unpublished_diagnostics_problem(project_root, file); + return unpublished_diagnostics_problem(project_root, file, status); } Some(OmissionReason::Redacted) => ( "application.diagnostics.redacted", @@ -1901,8 +1912,9 @@ fn diagnostics_absence_problem( fn unpublished_diagnostics_problem( project_root: &Path, file: Option<&Path>, + status: &TypeScriptProducerStatusHandleV1, ) -> Result { - let problem = match typescript_diagnostics_availability(project_root, file) { + let problem = match typescript_diagnostics_availability(project_root, file, status) { TypeScriptDiagnosticsAvailabilityV1::NoTsconfig { searched } => { no_tsconfig_problem(file, &searched) } @@ -1982,8 +1994,9 @@ fn unpublished_diagnostics_problem( fn unchecked_owner_problem( project_root: &Path, file: &Path, + status: &TypeScriptProducerStatusHandleV1, ) -> Result, ApplicationContractError> { - let problem = match typescript_diagnostics_availability(project_root, Some(file)) { + let problem = match typescript_diagnostics_availability(project_root, Some(file), status) { TypeScriptDiagnosticsAvailabilityV1::CompilerMissing { tsconfig, install_command, @@ -2091,9 +2104,9 @@ mod tests { use super::{ ExtendedPrimitivePort, MAX_SAFE_DIAGNOSTIC_MESSAGE_BYTES, OmissionReason, PrimitiveCapacity, PrimitiveDispatch, PrimitiveRequest, StorageStatusPrimitiveRequest, - diagnostics_absence_problem, safe_problem_message, session_structural_refusal_problem, - symbol_temporal_state, unchecked_owner_problem, unpublished_diagnostics_problem, - valid_owned_primitive_request, validate_admitted_root_uri, + TypeScriptProducerStatusHandleV1, diagnostics_absence_problem, safe_problem_message, + session_structural_refusal_problem, symbol_temporal_state, unchecked_owner_problem, + unpublished_diagnostics_problem, valid_owned_primitive_request, validate_admitted_root_uri, }; use tracedecay_contracts::retrieval::{ CodeGraphReadFreshnessV1, GraphRelationRequest, ImplementationSelector, @@ -2173,8 +2186,13 @@ mod tests { "application.diagnostics.unsupported", ), ] { - let problem = diagnostics_absence_problem(reason, project.path(), None) - .expect("typed diagnostics absence"); + let problem = diagnostics_absence_problem( + reason, + project.path(), + None, + &TypeScriptProducerStatusHandleV1::default(), + ) + .expect("typed diagnostics absence"); assert_eq!(problem.kind(), kind, "reason {reason:?}"); assert_eq!( problem @@ -2196,6 +2214,7 @@ mod tests { fn unpublished_diagnostics_name_the_producer_route() { let project = tempfile::tempdir().expect("project root"); let root = project.path(); + let status = TypeScriptProducerStatusHandleV1::default(); let file = Path::new("packages/app/src/index.ts"); for (scope, detail) in [ @@ -2223,7 +2242,8 @@ mod tests { }, ), ] { - let no_tsconfig = unpublished_diagnostics_problem(root, scope).expect("no tsconfig"); + let no_tsconfig = + unpublished_diagnostics_problem(root, scope, &status).expect("no tsconfig"); assert_eq!(no_tsconfig.detail(), Some(&detail)); assert_eq!(no_tsconfig.kind(), ApplicationProblemKind::Unsupported); assert_eq!( @@ -2254,7 +2274,8 @@ mod tests { .expect("package tsconfig"); std::fs::write(root.join("pnpm-lock.yaml"), "lockfileVersion: '9.0'\n").expect("lock"); for scope in [None, Some(file)] { - let missing = unpublished_diagnostics_problem(root, scope).expect("missing compiler"); + let missing = + unpublished_diagnostics_problem(root, scope, &status).expect("missing compiler"); assert_eq!(missing.kind(), ApplicationProblemKind::Unsupported); assert_eq!(missing.legal_actions(), [LegalAction::Refresh]); assert_eq!( @@ -2269,7 +2290,7 @@ mod tests { ); } assert!( - unchecked_owner_problem(root, file) + unchecked_owner_problem(root, file, &status) .expect("owner state") .is_some(), "an empty page is no evidence for a file whose compiler is missing" @@ -2280,11 +2301,11 @@ mod tests { std::fs::write(bin.join(if cfg!(windows) { "tsc.cmd" } else { "tsc" }), "") .expect("workspace tsc"); assert!( - unchecked_owner_problem(root, file) + unchecked_owner_problem(root, file, &status) .expect("owner state") .is_none() ); - let pending = unpublished_diagnostics_problem(root, Some(file)).expect("pending"); + let pending = unpublished_diagnostics_problem(root, Some(file), &status).expect("pending"); assert_eq!( pending.detail(), Some(&ApplicationProblemDetailV1::DiagnosticsPending { diff --git a/crates/tracedecay/src/daemon/project_open_owners.rs b/crates/tracedecay/src/daemon/project_open_owners.rs index 25a14b809b..c01b17a082 100644 --- a/crates/tracedecay/src/daemon/project_open_owners.rs +++ b/crates/tracedecay/src/daemon/project_open_owners.rs @@ -23,6 +23,7 @@ use tracedecay_agent_hosts::native_integration::{ DaemonNativeIntegrationAnalysisV1, DaemonNativeIntegrationServiceRegistry, NativeIntegrationTargetV1, }; +use tracedecay_application::diagnostics_producer::TypeScriptProducerStatusHandleV1; use tracedecay_application::lsp_runtime::DaemonLspSessionFactory; use tracedecay_application::source_authorization::ProjectSourceAccessSnapshot; use tracedecay_code_index_runtime::git_transactions::DaemonGitIndexTransactionServiceRegistry; @@ -169,6 +170,10 @@ pub(crate) async fn install_project_open_source_edit_owners_for_test( /// callable-code authorization and the primitive runtime. Returns the /// admitted root URI they were bound to. #[tracing::instrument(name = "daemon.project.owners.code_reads", level = "trace", skip_all)] +#[expect( + clippy::too_many_arguments, + reason = "Project-open wiring hands each owner its admitted authorities explicitly." +)] async fn register_project_code_read_owners( invocation: &DaemonInvocationState, project_root: &Path, @@ -177,6 +182,7 @@ async fn register_project_code_read_owners( access: &ProjectSourceAccessSnapshot, session_db: tracedecay_global_db::RegisteredGlobalDbLeaseV1, temporal: Arc, + compiler_status: TypeScriptProducerStatusHandleV1, ) -> Result { // Primitive reads are part of the admitted core route. Publish their // runtime and callable-code authorization before the slower mutation, @@ -219,6 +225,7 @@ async fn register_project_code_read_owners( session_db, temporal, access.clone(), + compiler_status, ) .await } @@ -271,6 +278,7 @@ pub(super) async fn register_reset_required_route_code_read_owners( &access, session_store, Arc::new(primitive_runtime::ResetRequiredSessionLookupV1), + TypeScriptProducerStatusHandleV1::default(), )) .await .map(drop) @@ -380,6 +388,9 @@ pub(super) async fn register_project_open_production_owners( let temporal = Arc::new(DaemonSessionLookupPrimitiveV1::new( server.project_session_application_retrieval_service(&access.scope)?, )); + // One status handle per project owner: the producer publishes into it and + // this owner's diagnostics reads describe it. + let compiler_status = TypeScriptProducerStatusHandleV1::default(); let admitted_root_uri = Box::pin(register_project_code_read_owners( invocation, project_root, @@ -388,6 +399,7 @@ pub(super) async fn register_project_open_production_owners( &access, session_db.clone(), temporal, + compiler_status.clone(), )) .await?; owner_phase("code_read_owners_registered"); @@ -792,6 +804,7 @@ pub(super) async fn register_project_open_production_owners( project_root.to_path_buf(), &scope, Arc::clone(&graph), + compiler_status, ); tracing::info!( diff --git a/crates/tracedecay/src/daemon/project_open_owners/compiler_diagnostics_producer.rs b/crates/tracedecay/src/daemon/project_open_owners/compiler_diagnostics_producer.rs index ac17ea0262..1c76d13fbc 100644 --- a/crates/tracedecay/src/daemon/project_open_owners/compiler_diagnostics_producer.rs +++ b/crates/tracedecay/src/daemon/project_open_owners/compiler_diagnostics_producer.rs @@ -13,7 +13,8 @@ use std::path::{Path, PathBuf}; use std::sync::Arc; use tracedecay_application::diagnostics_producer::{ - CompilerProducerRunV1, TypeScriptProducerStateV1, run_typescript_producer_v1, + CompilerProducerRunV1, TypeScriptProducerStateV1, TypeScriptProducerStatusHandleV1, + run_typescript_producer_v1, }; use tracedecay_application::diagnostics_store::DiagnosticsStore; use tracedecay_code_index_runtime::code_index_scheduler::CodeIndexOwnerSignalsV1; @@ -39,6 +40,7 @@ pub(super) fn spawn_typescript_diagnostics_producer( project_root: PathBuf, scope: &tracedecay_contracts::ResolvedScope, graph: Arc, + status: TypeScriptProducerStatusHandleV1, ) -> bool { if tracedecay_code_index_runtime::project_reads::code_index_disabled_for_scope( &invocation.code_index_schedulers, @@ -72,7 +74,7 @@ pub(super) fn spawn_typescript_diagnostics_producer( let schedulers = invocation.code_index_schedulers.clone(); let mut signals = CodeIndexOwnerSignalsV1::subscribe(&schedulers, &project_root).await; let build_info_dir = graph.store_layout().data_root.join(BUILD_INFO_DIR); - let mut state = TypeScriptProducerStateV1::default(); + let mut state = TypeScriptProducerStateV1::new(status); // One producer run per generation, whatever its outcome: the root's // signals also wake on serving and pass changes, and a project // whose compiler keeps failing must not re-run it on each of them. diff --git a/crates/tracedecay/src/daemon/project_open_owners/primitive_runtime.rs b/crates/tracedecay/src/daemon/project_open_owners/primitive_runtime.rs index 0bcc6488fa..3464b0cda9 100644 --- a/crates/tracedecay/src/daemon/project_open_owners/primitive_runtime.rs +++ b/crates/tracedecay/src/daemon/project_open_owners/primitive_runtime.rs @@ -3,6 +3,7 @@ use std::path::Path; use std::sync::Arc; +use tracedecay_application::diagnostics_producer::TypeScriptProducerStatusHandleV1; use tracedecay_application::primitives::{ ProductionPrimitiveCodeAuthoritiesV1, ProductionPrimitiveOpenRequestV1, admitted_root_uri_for_project, @@ -34,6 +35,10 @@ impl TemporalRetrievalPort for ResetRequiredSessionLookupV1 { } #[tracing::instrument(name = "daemon.project.owners.primitive", level = "trace", skip_all)] +#[expect( + clippy::too_many_arguments, + reason = "Project-open wiring hands each owner its admitted authorities explicitly." +)] pub(super) async fn open_and_register_project_primitive_runtime( invocation: &DaemonInvocationState, project_root: &Path, @@ -42,6 +47,7 @@ pub(super) async fn open_and_register_project_primitive_runtime( session_db: tracedecay_global_db::RegisteredGlobalDbLeaseV1, temporal: Arc, access: ProjectSourceAccessSnapshot, + compiler_status: TypeScriptProducerStatusHandleV1, ) -> Result { let admitted_root_uri = admitted_root_uri_for_project(project_root).map_err(|error| TraceDecayError::Config { @@ -76,6 +82,7 @@ pub(super) async fn open_and_register_project_primitive_runtime( temporal, access, admitted_root_uri.clone(), + compiler_status, ), ) .await From 42931ccaf57535aba3e743ea7c56ec4662478657 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 25 Oct 2026 10:56:41 -0700 Subject: [PATCH 2/9] test(daemon): verify listing lease through retirement eligibility --- .../tracedecay/src/daemon/tests/ownership.rs | 67 +++++++++++++++++-- 1 file changed, 61 insertions(+), 6 deletions(-) diff --git a/crates/tracedecay/src/daemon/tests/ownership.rs b/crates/tracedecay/src/daemon/tests/ownership.rs index 98d75f79f6..4e8db89fea 100644 --- a/crates/tracedecay/src/daemon/tests/ownership.rs +++ b/crates/tracedecay/src/daemon/tests/ownership.rs @@ -445,11 +445,36 @@ async fn mounted_graph_listing_takes_no_server_lease_while_it_reads() { .await .expect("mount project server"); + engine.lifecycle.begin_draining(); + tokio::time::timeout( + std::time::Duration::from_secs(5), + engine.lifecycle.wait_for_idle(), + ) + .await + .expect("fixture startup owners must release their server handles"); + let startup_lifecycle = DaemonLifecycle::default(); + let startup_server = Arc::clone(&server); + let (startup_entered, startup_parked) = tokio::sync::oneshot::channel(); + super::super::project_open_orchestration::spawn_lifecycle_automation_scheduler_activation( + startup_lifecycle.clone(), + async move { + startup_entered.send(()).expect("startup owner entered"); + std::future::pending::<()>().await; + drop(startup_server); + }, + ); + startup_parked.await.expect("startup owner parked"); + + let graph = server.cg().await; + let key = ProjectServerKey::from_open_project(&graph, &handshake).expect("mounted owner key"); + let route = ProjectRouteKey::from_handshake(graph.project_root(), &handshake) + .expect("mounted route key"); + let server_witness = Arc::downgrade(&server); let swap = server.graph_slot().write_owned().await; + drop(server); let mut registry_gate = Some(engine.store_administration.project_servers().lock().await); let listing = engine.store_administration.mounted_project_graphs(); let mut listing = std::pin::pin!(listing); - let handles = Arc::strong_count(&server); let registry_lock = engine.store_administration.project_servers().lock(); let mut registry_lock = std::pin::pin!(registry_lock); // Force the first listing poll to wait for another registry user. Tokio's @@ -457,7 +482,7 @@ async fn mounted_graph_listing_takes_no_server_lease_while_it_reads() { // Drive both futures with real wakeups: the fair registry mutex admits // the listing before this later lock request, then must release it while // the graph write guard still prevents the listing from completing. - let registry = tokio::time::timeout( + let mut registry = tokio::time::timeout( std::time::Duration::from_secs(5), std::future::poll_fn(|context| { assert!( @@ -470,14 +495,44 @@ async fn mounted_graph_listing_takes_no_server_lease_while_it_reads() { ) .await .expect("a parked graph listing must release the registry lock for route binds"); + assert_eq!(registry.servers.len(), 1); + assert!( + registry + .retire_lru_ready_under_graph_pressure(|server| Arc::strong_count(server) > 1) + .is_err(), + "the controlled startup handle must deny retirement while the listing is parked" + ); + assert!(registry.get(&key).is_some()); + startup_lifecycle.begin_draining(); + tokio::time::timeout( + std::time::Duration::from_secs(5), + startup_lifecycle.wait_for_idle(), + ) + .await + .expect("startup owner must release its handle while the listing is parked"); + let (owner, retired) = registry + .retire_lru_ready_under_graph_pressure(|server| Arc::strong_count(server) > 1) + .expect("a parked graph listing must not lease the idle server") + .expect("the idle owner must be eligible for retirement"); + assert_eq!(owner, key.owner); + assert_eq!(retired.len(), 1); + let (retired_key, retired_server) = retired.into_iter().next().expect("retired server"); + assert_eq!(retired_key, key); assert_eq!( - Arc::strong_count(&server), - handles, - "a parked graph listing must hold no server handle" + Arc::as_ptr(&retired_server), + server_witness.as_ptr(), + "retirement must transfer the exact listed server" ); + assert!(registry.get(&key).is_none()); + assert!(registry.get_route(&route).is_none()); + registry.insert_route(route, retired_key, retired_server); drop(registry); drop(swap); - assert_eq!(listing.await.len(), 1); + let listed = listing.await; + assert_eq!(listed.len(), 1); + assert!(Arc::ptr_eq(&listed[0], &graph)); + drop((listed, graph)); + engine.shutdown_all().await; } /// A mounted route alias answers without re-running registry admission: the From 91024a64edf9d99d92cd879e5856f6e665d5eda2 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 25 Oct 2026 14:28:51 -0700 Subject: [PATCH 3/9] fix(ci): bound Windows compiler concurrency --- .github/workflows/ci.yml | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3e82ac6c34..12ef2e354b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -310,7 +310,7 @@ jobs: *) echo "Expected a Bazel target pattern: $target" >&2; exit 1 ;; esac done - bazel test --config=ci --test_output=all "${targets[@]}" + bazel test --config=ci --jobs=2 --test_output=all "${targets[@]}" - name: Upload failed test diagnostics if: failure() @@ -596,7 +596,9 @@ jobs: shell: bash # Compile and link every target, but defer unused test runfiles trees. # Bazel still materializes runfiles required by build-time tools. - run: bazel build --config=ci --nobuild_runfile_links //... + # Four concurrent Rust actions exhausted the hosted runner's commit + # limit while mapping asupersync metadata (Windows error 1455). + run: bazel build --config=ci --jobs=2 --nobuild_runfile_links //... - name: Build and test native targets if: ${{ github.event_name == 'workflow_dispatch' && inputs.run_os }} From 099ed215de20c7c752600b1767272836a07580d1 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 25 Oct 2026 14:45:41 -0700 Subject: [PATCH 4/9] fix(ci): scope native concurrency to Windows --- .github/workflows/ci.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 12ef2e354b..dc96d97cb4 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -310,7 +310,7 @@ jobs: *) echo "Expected a Bazel target pattern: $target" >&2; exit 1 ;; esac done - bazel test --config=ci --jobs=2 --test_output=all "${targets[@]}" + bazel test --config=ci --test_output=all "${targets[@]}" - name: Upload failed test diagnostics if: failure() @@ -621,7 +621,7 @@ jobs: *) echo "Expected a Bazel target pattern: $target" >&2; exit 1 ;; esac done - bazel test --config=ci --test_output=all "${targets[@]}" + bazel test --config=ci --jobs=2 --test_output=all "${targets[@]}" - name: Upload failed test diagnostics if: failure() From d6c5075f396b8a209d4d5ab8e42fb7365493292a Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 26 Oct 2026 16:51:06 -0700 Subject: [PATCH 5/9] fix(diagnostics): reuse retained compiler status on reopen --- .../src/primitives/production.rs | 31 +- .../src/primitives/runtime.rs | 41 ++- .../src/invocation.rs | 2 +- .../src/invocation/primitive.rs | 54 ++-- .../src/project_runtime.rs | 80 +++-- .../src/project_runtime/request_snapshot.rs | 12 +- .../src/project_runtime/tests.rs | 68 +++- crates/tracedecay/BUILD.bazel | 2 +- .../src/daemon/project_open_owners.rs | 17 +- .../project_open_owners/primitive_runtime.rs | 12 +- crates/tracedecay/src/daemon/tests.rs | 1 + .../src/daemon/tests/compiler_status.rs | 290 ++++++++++++++++++ dashboard/BUILD.bazel | 7 + scripts/bazel/gen_builds.py | 5 + 14 files changed, 504 insertions(+), 118 deletions(-) create mode 100644 crates/tracedecay/src/daemon/tests/compiler_status.rs diff --git a/crates/tracedecay-application/src/primitives/production.rs b/crates/tracedecay-application/src/primitives/production.rs index 4e74e9cda5..c4c8667cd4 100644 --- a/crates/tracedecay-application/src/primitives/production.rs +++ b/crates/tracedecay-application/src/primitives/production.rs @@ -858,6 +858,7 @@ pub struct ProductionPrimitiveCodeAuthoritiesV1 { pub convergence_park: Arc, } +#[derive(Clone)] pub struct ProductionPrimitiveOpenRequestV1 { source_runtime: Arc, code_graph: Arc, @@ -869,7 +870,6 @@ pub struct ProductionPrimitiveOpenRequestV1 { convergence_park: Arc, access: ProjectSourceAccessSnapshot, admitted_root_uri: String, - compiler_status: TypeScriptProducerStatusHandleV1, } impl ProductionPrimitiveOpenRequestV1 { @@ -880,7 +880,6 @@ impl ProductionPrimitiveOpenRequestV1 { temporal: Arc, access: ProjectSourceAccessSnapshot, admitted_root_uri: String, - compiler_status: TypeScriptProducerStatusHandleV1, ) -> Self { Self { source_runtime, @@ -893,9 +892,32 @@ impl ProductionPrimitiveOpenRequestV1 { convergence_park: code.convergence_park, access, admitted_root_uri, - compiler_status, } } + + pub fn validate_registration_root( + &self, + project_root: &Path, + ) -> Result<(), ApplicationContractError> { + if self.source_runtime.project_root() != project_root { + return Err(ApplicationContractError::Inconsistent { + field: "application primitive registration root", + }); + } + Ok(()) + } + + pub fn validate_retained( + &self, + runtime: &PrimitiveProjectRuntime, + ) -> Result<(), ApplicationContractError> { + runtime.validate_reopen( + self.source_runtime.db(), + &self.session_db, + &self.access, + &self.admitted_root_uri, + ) + } } /// Opens the complete owned application primitive runtime from production authorities. @@ -914,7 +936,6 @@ pub async fn open_production_primitive_runtime( convergence_park, access, admitted_root_uri, - compiler_status, } = request; let database = source_runtime.db().clone(); let project_root = source_runtime.project_root().to_path_buf(); @@ -992,7 +1013,7 @@ pub async fn open_production_primitive_runtime( test_runs, test_run_scope, convergence_park, - compiler_status, + TypeScriptProducerStatusHandleV1::default(), ) } diff --git a/crates/tracedecay-application/src/primitives/runtime.rs b/crates/tracedecay-application/src/primitives/runtime.rs index cee53d825a..7c6576af6d 100644 --- a/crates/tracedecay-application/src/primitives/runtime.rs +++ b/crates/tracedecay-application/src/primitives/runtime.rs @@ -377,7 +377,7 @@ impl PrimitiveCapacity { /// keep the exact project-open owners alive for the same lifetime as dispatch. pub struct PrimitiveProjectRuntime { database: Database, - dispatch: Arc, + dispatch: Arc, } /// Replace the mount-time authority carried by an owned primitive result with @@ -401,12 +401,47 @@ pub fn reauthorize_primitive_evidence( impl PrimitiveProjectRuntime { pub fn dispatch(&self) -> Arc { - Arc::clone(&self.dispatch) + self.dispatch.clone() } pub fn database(&self) -> &Database { &self.database } + + pub fn compiler_status(&self) -> TypeScriptProducerStatusHandleV1 { + self.dispatch.compiler_status.clone() + } + + pub(super) fn validate_reopen( + &self, + database: &Database, + session_db: &RegisteredGlobalDbLeaseV1, + access: &ProjectSourceAccessSnapshot, + admitted_root_uri: &str, + ) -> Result<(), ApplicationContractError> { + // Request admission checks each renewed grant; its expiry is not + // part of the retained source/configuration authority. + let retained = &self.dispatch.access; + let sessions = &self.dispatch.test_runs.store; + if self.database.canonical_database_path() != database.canonical_database_path() + || self.database.opened_file_identity() != database.opened_file_identity() + || sessions.binding() != session_db.binding() + || sessions.verified_locator() != session_db.verified_locator() + || self.dispatch.admitted_root_uri != admitted_root_uri + || retained.scope != access.scope + || retained.requester != access.requester + || retained.binding != access.binding + || retained.configuration_revision != access.configuration_revision + || retained.configuration_digest != access.configuration_digest + || retained.configuration_provenance_digest != access.configuration_provenance_digest + || retained.effective_capabilities != access.effective_capabilities + { + return Err(ApplicationContractError::Inconsistent { + field: "retained application primitive authority", + }); + } + Ok(()) + } } impl PrimitiveDispatch for OwnedPrimitiveRuntime { @@ -573,7 +608,7 @@ pub fn open_primitive_project_runtime( health, extended, }; - let dispatch: Arc = Arc::new(OwnedPrimitiveRuntime { + let dispatch = Arc::new(OwnedPrimitiveRuntime { project_runtime: services, scope, access, diff --git a/crates/tracedecay-daemon-service/src/invocation.rs b/crates/tracedecay-daemon-service/src/invocation.rs index eef8697c99..30f84ce214 100644 --- a/crates/tracedecay-daemon-service/src/invocation.rs +++ b/crates/tracedecay-daemon-service/src/invocation.rs @@ -134,7 +134,7 @@ use tracedecay_application::lsp_runtime::{ use tracedecay_application::operation_stream::{ OperationEmitter, OperationEventAuthority, OperationKind, operation_event_authority, }; -use tracedecay_application::primitives::{PrimitiveDispatch, PrimitiveProjectRuntime}; +use tracedecay_application::primitives::PrimitiveProjectRuntime; use tracedecay_contracts::feedback::observations::{ FeedbackAnchorOperationV1, FeedbackArgumentRejectionClassV1, FeedbackDeliveryRouteV1, FeedbackOperationV1, FeedbackOutcomeV1, FeedbackRejectedArgumentV1, FeedbackSourceEventV1, diff --git a/crates/tracedecay-daemon-service/src/invocation/primitive.rs b/crates/tracedecay-daemon-service/src/invocation/primitive.rs index 297c88b3f3..af3ec397bc 100644 --- a/crates/tracedecay-daemon-service/src/invocation/primitive.rs +++ b/crates/tracedecay-daemon-service/src/invocation/primitive.rs @@ -7,6 +7,7 @@ use tracedecay_agent_hosts::agents::context_scout::{ ContextScoutMutationResultV1, ContextScoutMutationSettlementOutcomeV1, ContextScoutPublicMutationV1, ContextScoutRecentStateV1, }; +use tracedecay_application::diagnostics_producer::TypeScriptProducerStatusHandleV1; use tracedecay_application::primitives::{ ProductionPrimitiveOpenRequestV1, open_production_primitive_runtime, }; @@ -1375,41 +1376,38 @@ impl DaemonPrimitiveRuntimeRegistrar { } } - /// Retains the already-opened project runtime as its teardown owner. - /// Scope/access were bound by the concrete project-open factory. - pub async fn register( + pub async fn open_and_register( &self, project_root: PathBuf, - project_runtime: PrimitiveProjectRuntime, - ) -> Result, DaemonPrimitiveRuntimeRegistrationError> { - let dispatch = project_runtime.dispatch(); + request: ProductionPrimitiveOpenRequestV1, + ) -> Result { + request.validate_registration_root(&project_root)?; self.service .project_runtimes - .register(project_root, project_runtime) + .register_or_reconcile_with( + project_root, + |runtime: &mut PrimitiveProjectRuntime| { + request.validate_retained(runtime).map_err(Into::into) + }, + || async { + open_production_primitive_runtime(request.clone()) + .await + .map_err(Into::into) + }, + PrimitiveProjectRuntime::compiler_status, + ) .await - .map_err(|error| match error { - ProjectRuntimeRegistryError::AlreadyRegistered => { - DaemonPrimitiveRuntimeRegistrationError::AlreadyRegistered - } - ProjectRuntimeRegistryError::Closed => { - DaemonPrimitiveRuntimeRegistrationError::RegistryClosed - } - ProjectRuntimeRegistryError::ConcurrentBuildFailed { detail } => { - DaemonPrimitiveRuntimeRegistrationError::ConcurrentBuildFailed { detail } - } - })?; - Ok(dispatch) } +} - pub async fn open_and_register( - &self, - project_root: PathBuf, - request: ProductionPrimitiveOpenRequestV1, - ) -> Result<(), DaemonPrimitiveRuntimeRegistrationError> { - let runtime = open_production_primitive_runtime(request).await?; - match self.register(project_root, runtime).await { - Ok(_) | Err(DaemonPrimitiveRuntimeRegistrationError::AlreadyRegistered) => Ok(()), - Err(error) => Err(error), +impl From for DaemonPrimitiveRuntimeRegistrationError { + fn from(error: ProjectRuntimeRegistryError) -> Self { + match error { + ProjectRuntimeRegistryError::AlreadyRegistered => Self::AlreadyRegistered, + ProjectRuntimeRegistryError::Closed => Self::RegistryClosed, + ProjectRuntimeRegistryError::ConcurrentBuildFailed { detail } => { + Self::ConcurrentBuildFailed { detail } + } } } } diff --git a/crates/tracedecay-daemon-service/src/project_runtime.rs b/crates/tracedecay-daemon-service/src/project_runtime.rs index 1842d0c7d2..f272aecd49 100644 --- a/crates/tracedecay-daemon-service/src/project_runtime.rs +++ b/crates/tracedecay-daemon-service/src/project_runtime.rs @@ -689,7 +689,11 @@ impl Drop for FeedbackRegistrationLease { } impl ProjectRuntimeBuildReservationLeaseV1 { - fn commit(mut self, component: C) -> Result<(), ProjectRuntimeRegistryError> + fn commit( + mut self, + component: C, + project: impl FnOnce(&C) -> T, + ) -> Result where C: ProjectRuntimeComponent, { @@ -707,8 +711,9 @@ impl ProjectRuntimeBuildReservationLeaseV1 { |reservation| Arc::ptr_eq(reservation, &self.reservation), ) && C::peek(runtime).is_none() => { + let result = project(&component); *C::slot(runtime) = Some(component); - Ok(()) + Ok(result) } _ => Err(ProjectRuntimeRegistryError::AlreadyRegistered), } @@ -723,7 +728,7 @@ impl ProjectRuntimeBuildReservationLeaseV1 { drop(root_fences); self.active = false; let outcome = match &result { - Ok(()) => ProjectRuntimeBuildOutcomeV1::Succeeded, + Ok(_) => ProjectRuntimeBuildOutcomeV1::Succeeded, Err(error) => ProjectRuntimeBuildOutcomeV1::Failed { detail: error.to_string(), }, @@ -847,44 +852,6 @@ impl ProjectRuntimeRegistryV1 { }) } - /// Publish a component, refusing if this project already has a live one. - #[tracing::instrument( - name = "daemon.service.project_runtime.register", - level = "trace", - skip_all - )] - pub(crate) async fn register( - &self, - project_root: PathBuf, - component: C, - ) -> Result<(), ProjectRuntimeRegistryError> - where - C: ProjectRuntimeComponent, - { - loop { - let mut reservation_changed = self.reservation_changed.subscribe(); - { - let root_fences = self.lock_root_fences(); - let mut runtimes = self.lock_runtimes(); - if self.closed.load(Ordering::Acquire) || root_fences.contains(&project_root) { - return Err(ProjectRuntimeRegistryError::Closed); - } - let runtime = runtimes.entry(project_root.clone()).or_default(); - if !runtime.reservations.contains(&TypeId::of::()) { - let slot = C::slot(runtime); - if slot.is_some() { - return Err(ProjectRuntimeRegistryError::AlreadyRegistered); - } - *slot = Some(component); - return Ok(()); - } - } - if reservation_changed.changed().await.is_err() { - return Err(ProjectRuntimeRegistryError::Closed); - } - } - } - /// Publish a component over whatever was there. /// /// Only for components whose caller has already established that the @@ -1233,23 +1200,45 @@ impl ProjectRuntimeRegistryV1 { /// is reserved under the registry locks, then `build` runs after both /// locks are released. Registrations for the same typed slot join that /// reservation and receive its exact terminal outcome. + pub(crate) async fn register_or_reconcile( + &self, + project_root: PathBuf, + reconcile: R, + build: B, + ) -> Result<(), E> + where + C: ProjectRuntimeComponent, + E: From + fmt::Display, + R: FnOnce(&mut C) -> Result<(), E>, + B: FnOnce() -> Fut, + Fut: Future>, + { + self.register_or_reconcile_with(project_root, reconcile, build, |_| ()) + .await + } + + /// Project the installed component under the same locks that admit its + /// registration or reconcile its incumbent. A rejected build never returns + /// a projection of the discarded candidate. #[tracing::instrument( name = "daemon.service.project_runtime.register_or_reconcile", level = "trace", skip_all )] - pub(crate) async fn register_or_reconcile( + pub(crate) async fn register_or_reconcile_with( &self, project_root: PathBuf, reconcile: R, build: B, - ) -> Result<(), E> + project: P, + ) -> Result where C: ProjectRuntimeComponent, E: From + fmt::Display, R: FnOnce(&mut C) -> Result<(), E>, B: FnOnce() -> Fut, Fut: Future>, + P: FnOnce(&C) -> T, { loop { let mut reservation_changed = self.reservation_changed.subscribe(); @@ -1271,7 +1260,8 @@ impl ProjectRuntimeRegistryV1 { } else { let slot = C::slot(runtime); if let Some(incumbent) = slot.as_mut() { - return reconcile(incumbent); + reconcile(incumbent)?; + return Ok(project(incumbent)); } let reservation = ProjectRuntimeBuildReservationV1::new(); runtime.reservations.push(type_id); @@ -1311,7 +1301,7 @@ impl ProjectRuntimeRegistryV1 { } if let Some(lease) = build_lease { return match build().await { - Ok(component) => lease.commit::(component).map_err(Into::into), + Ok(component) => lease.commit(component, project).map_err(Into::into), Err(error) => { lease.fail(error.to_string()); Err(error) diff --git a/crates/tracedecay-daemon-service/src/project_runtime/request_snapshot.rs b/crates/tracedecay-daemon-service/src/project_runtime/request_snapshot.rs index a1b5beb320..7526d9e0c0 100644 --- a/crates/tracedecay-daemon-service/src/project_runtime/request_snapshot.rs +++ b/crates/tracedecay-daemon-service/src/project_runtime/request_snapshot.rs @@ -282,7 +282,7 @@ mod tests { Arc::new(UnavailableAdvisoryCycle), ); registry - .register(canonical.clone(), owner) + .publish(canonical.clone(), owner) .await .expect("canonical owner registration"); @@ -356,7 +356,7 @@ mod tests { let admitted_project = ProjectId::new("project.admitted-owner").expect("project id"); let replacement_project = ProjectId::new("project.replacement-owner").expect("project id"); registry - .register( + .publish( project_root.clone(), DaemonAdvisoryCycleInvocationOwner::new( admitted_project.clone(), @@ -417,7 +417,7 @@ mod tests { let foreign = PathBuf::from("/projects/foreign-root"); let admitted_project = ProjectId::new("project.admitted-root").expect("project id"); registry - .register( + .publish( admitted.clone(), DaemonAdvisoryCycleInvocationOwner::new( admitted_project.clone(), @@ -427,7 +427,7 @@ mod tests { .await .expect("admitted owner registration"); registry - .register( + .publish( foreign.clone(), DaemonAdvisoryCycleInvocationOwner::new( ProjectId::new("project.foreign-root").expect("project id"), @@ -482,7 +482,7 @@ mod tests { let original_project = ProjectId::new("project.original-target").expect("project id"); let replacement_project = ProjectId::new("project.replacement-target").expect("project id"); registry - .register( + .publish( original.clone(), DaemonAdvisoryCycleInvocationOwner::new( original_project.clone(), @@ -492,7 +492,7 @@ mod tests { .await .expect("original owner registration"); registry - .register( + .publish( replacement.clone(), DaemonAdvisoryCycleInvocationOwner::new( replacement_project, diff --git a/crates/tracedecay-daemon-service/src/project_runtime/tests.rs b/crates/tracedecay-daemon-service/src/project_runtime/tests.rs index b80642f5e8..d5fc9f6072 100644 --- a/crates/tracedecay-daemon-service/src/project_runtime/tests.rs +++ b/crates/tracedecay-daemon-service/src/project_runtime/tests.rs @@ -70,10 +70,17 @@ async fn a_second_registration_is_refused_and_leaves_the_incumbent_live() { let project = root("alpha"); registry - .register(project.clone(), component(1)) + .publish(project.clone(), component(1)) .await .expect("the first registration owns the empty slot"); - let refused = registry.register(project.clone(), component(2)).await; + let refused = registry + .register_or_reconcile_with::( + project.clone(), + |_| Err(ProjectRuntimeRegistryError::AlreadyRegistered), + || async { panic!("a rejected incumbent must not build a replacement") }, + mark, + ) + .await; assert_eq!(refused, Err(ProjectRuntimeRegistryError::AlreadyRegistered)); assert_eq!( @@ -495,7 +502,7 @@ async fn concurrent_successful_registrations_build_only_once() { registrations.spawn(async move { barrier.wait().await; registry - .register_or_reconcile::( + .register_or_reconcile_with::( project, |incumbent| { assert_eq!(mark(incumbent), Some(11)); @@ -506,6 +513,7 @@ async fn concurrent_successful_registrations_build_only_once() { release_build.notified().await; Ok(component(11)) }, + Arc::clone, ) .await }); @@ -521,9 +529,11 @@ async fn concurrent_successful_registrations_build_only_once() { let mut successes = 0; while let Some(result) = registrations.join_next().await { - result + let returned = result .expect("registration task") .expect("joined registration succeeds"); + let installed = registry.get::(&project).await.unwrap(); + assert!(Arc::ptr_eq(&installed, &returned)); successes += 1; } assert_eq!(successes, REGISTRATIONS); @@ -938,7 +948,14 @@ async fn shutting_down_empties_the_registry() { assert!(registry.is_empty()); assert!(!registry.holds::(&root("alpha")).await); assert_eq!( - registry.register(root("late"), component(3)).await, + registry + .register_or_reconcile_with::( + root("late"), + |_| Ok(()), + || async { Ok(component(3)) }, + mark, + ) + .await, Err(ProjectRuntimeRegistryError::Closed), "registration must fail after shutdown closes the registry" ); @@ -974,7 +991,14 @@ async fn retiring_exact_roots_fences_republication_without_closing_other_project ); assert!(!registry.holds::(&retired).await); assert_eq!( - registry.register(retired.clone(), component(3)).await, + registry + .register_or_reconcile_with::( + retired.clone(), + |_| Ok(()), + || async { Ok(component(3)) }, + mark, + ) + .await, Err(ProjectRuntimeRegistryError::Closed) ); assert_eq!( @@ -1042,3 +1066,35 @@ async fn daemon_shutdown_does_not_wait_for_drained_runtime_deallocation() { .recv_timeout(std::time::Duration::from_secs(5)) .expect("the detached release must still free the drained runtime"); } + +#[tokio::test] +async fn registration_projection_refuses_shutdown_during_build() { + let registry = ProjectRuntimeRegistryV1::default(); + let project = root("projection-shutdown"); + let building_registry = registry.clone(); + let building_project = project.clone(); + let (started, building) = tokio::sync::oneshot::channel(); + let (release, resume) = tokio::sync::oneshot::channel(); + let registration = tokio::spawn(async move { + building_registry + .register_or_reconcile_with::( + building_project, + |_| Ok(()), + || async { + started.send(()).unwrap(); + resume.await.unwrap(); + Ok(component(42)) + }, + mark, + ) + .await + }); + building.await.unwrap(); + registry.begin_shutdown(); + release.send(()).unwrap(); + assert_eq!( + registration.await.unwrap(), + Err(ProjectRuntimeRegistryError::Closed) + ); + assert!(!registry.holds::(&project).await); +} diff --git a/crates/tracedecay/BUILD.bazel b/crates/tracedecay/BUILD.bazel index 1bbfc4a2ad..d0e1ea9992 100644 --- a/crates/tracedecay/BUILD.bazel +++ b/crates/tracedecay/BUILD.bazel @@ -475,7 +475,7 @@ rust_test( compile_data = glob(["tests/**","assets/**","fixtures/**","data/**","resources/**","vendor/**"], allow_empty = True) + ["//crates/tracedecay-agent-hosts:src_agents_hermes_templates", "//tests:fixtures_provider_normalization_claude", "//tests:fixtures_provider_normalization_codex", "//tests:fixtures_provider_normalization_cursor", "//tests:fixtures_remote_tls"], data = glob(["tests/**","assets/**","fixtures/**","data/**","resources/**","vendor/**"], allow_empty = True) + ["//tests:fixtures"], - runfiles_env = {"TRACEDECAY_TEST_BIN": "//crates/tracedecay-cli:tracedecay__test"}, + runfiles_env = {"TRACEDECAY_TEST_BIN": "//crates/tracedecay-cli:tracedecay__test", "TRACEDECAY_TEST_TYPESCRIPT_COMPILER": "//dashboard:typescript_compiler"}, ) filegroup( diff --git a/crates/tracedecay/src/daemon/project_open_owners.rs b/crates/tracedecay/src/daemon/project_open_owners.rs index c01b17a082..a5e669911f 100644 --- a/crates/tracedecay/src/daemon/project_open_owners.rs +++ b/crates/tracedecay/src/daemon/project_open_owners.rs @@ -168,12 +168,8 @@ pub(crate) async fn install_project_open_source_edit_owners_for_test( /// Registers the owners that serve code reads on an admitted route: the /// callable-code authorization and the primitive runtime. Returns the -/// admitted root URI they were bound to. +/// admitted root URI and the compiler status retained by their runtime. #[tracing::instrument(name = "daemon.project.owners.code_reads", level = "trace", skip_all)] -#[expect( - clippy::too_many_arguments, - reason = "Project-open wiring hands each owner its admitted authorities explicitly." -)] async fn register_project_code_read_owners( invocation: &DaemonInvocationState, project_root: &Path, @@ -182,8 +178,7 @@ async fn register_project_code_read_owners( access: &ProjectSourceAccessSnapshot, session_db: tracedecay_global_db::RegisteredGlobalDbLeaseV1, temporal: Arc, - compiler_status: TypeScriptProducerStatusHandleV1, -) -> Result { +) -> Result<(String, TypeScriptProducerStatusHandleV1)> { // Primitive reads are part of the admitted core route. Publish their // runtime and callable-code authorization before the slower mutation, // delivery, native-integration, and Work owners finish mounting. @@ -225,7 +220,6 @@ async fn register_project_code_read_owners( session_db, temporal, access.clone(), - compiler_status, ) .await } @@ -278,7 +272,6 @@ pub(super) async fn register_reset_required_route_code_read_owners( &access, session_store, Arc::new(primitive_runtime::ResetRequiredSessionLookupV1), - TypeScriptProducerStatusHandleV1::default(), )) .await .map(drop) @@ -388,10 +381,7 @@ pub(super) async fn register_project_open_production_owners( let temporal = Arc::new(DaemonSessionLookupPrimitiveV1::new( server.project_session_application_retrieval_service(&access.scope)?, )); - // One status handle per project owner: the producer publishes into it and - // this owner's diagnostics reads describe it. - let compiler_status = TypeScriptProducerStatusHandleV1::default(); - let admitted_root_uri = Box::pin(register_project_code_read_owners( + let (admitted_root_uri, compiler_status) = Box::pin(register_project_code_read_owners( invocation, project_root, server, @@ -399,7 +389,6 @@ pub(super) async fn register_project_open_production_owners( &access, session_db.clone(), temporal, - compiler_status.clone(), )) .await?; owner_phase("code_read_owners_registered"); diff --git a/crates/tracedecay/src/daemon/project_open_owners/primitive_runtime.rs b/crates/tracedecay/src/daemon/project_open_owners/primitive_runtime.rs index 3464b0cda9..c8c84e0b34 100644 --- a/crates/tracedecay/src/daemon/project_open_owners/primitive_runtime.rs +++ b/crates/tracedecay/src/daemon/project_open_owners/primitive_runtime.rs @@ -35,10 +35,6 @@ impl TemporalRetrievalPort for ResetRequiredSessionLookupV1 { } #[tracing::instrument(name = "daemon.project.owners.primitive", level = "trace", skip_all)] -#[expect( - clippy::too_many_arguments, - reason = "Project-open wiring hands each owner its admitted authorities explicitly." -)] pub(super) async fn open_and_register_project_primitive_runtime( invocation: &DaemonInvocationState, project_root: &Path, @@ -47,8 +43,7 @@ pub(super) async fn open_and_register_project_primitive_runtime( session_db: tracedecay_global_db::RegisteredGlobalDbLeaseV1, temporal: Arc, access: ProjectSourceAccessSnapshot, - compiler_status: TypeScriptProducerStatusHandleV1, -) -> Result { +) -> Result<(String, TypeScriptProducerStatusHandleV1)> { let admitted_root_uri = admitted_root_uri_for_project(project_root).map_err(|error| TraceDecayError::Config { message: format!("project-open admitted root URI denied: {error}"), @@ -65,7 +60,7 @@ pub(super) async fn open_and_register_project_primitive_runtime( message: "project-open primitive runtime requires the mounted ignored-dependency admission authority" .to_owned(), })?; - invocation + let compiler_status = invocation .primitive_runtime_registrar() .open_and_register( project_root.to_path_buf(), @@ -82,12 +77,11 @@ pub(super) async fn open_and_register_project_primitive_runtime( temporal, access, admitted_root_uri.clone(), - compiler_status, ), ) .await .map_err(primitive_runtime_registration_error)?; - Ok(admitted_root_uri) + Ok((admitted_root_uri, compiler_status)) } fn primitive_runtime_registration_error( diff --git a/crates/tracedecay/src/daemon/tests.rs b/crates/tracedecay/src/daemon/tests.rs index fdef12b96b..216028181b 100644 --- a/crates/tracedecay/src/daemon/tests.rs +++ b/crates/tracedecay/src/daemon/tests.rs @@ -33,6 +33,7 @@ use tracedecay_daemon_service::shutdown::DaemonLifecycle; mod bootstrap; mod code_index_hydration; +mod compiler_status; mod feedback_impact; mod handshake; mod invocation_ownership; diff --git a/crates/tracedecay/src/daemon/tests/compiler_status.rs b/crates/tracedecay/src/daemon/tests/compiler_status.rs new file mode 100644 index 0000000000..b6bffc362a --- /dev/null +++ b/crates/tracedecay/src/daemon/tests/compiler_status.rs @@ -0,0 +1,290 @@ +#![cfg(unix)] + +use std::path::{Path, PathBuf}; +use std::process::Command; +use std::sync::Arc; +use std::time::Duration; + +use tempfile::TempDir; +use tokio::sync::mpsc; +use tracedecay_application::primitives::PrimitiveProjectRuntime; +use tracedecay_contracts::retrieval::{ + DiagnosticsPrimitiveRequest, DiagnosticsPrimitiveScope, PrimitiveRequest, +}; +use tracedecay_contracts::{CancellationContext, Deadline}; +use tracedecay_domain::UtcMicros; +use tracedecay_tool_catalog::ApplicationSurfaceOperation; +use tracing_subscriber::layer::{Context, Layer, SubscriberExt}; +use tracing_subscriber::registry::LookupSpan; + +use super::{ + apply_project_setting_via_surface, enter_test_daemon_database_scope, git, + initialize_test_project, test_client_identity_for, test_daemon_engine_for_profile, + test_handshake_defaults, +}; +use crate::daemon::{ + DaemonEngine, DaemonHandshake, DaemonInvocationOutcome, DaemonInvocationRequest, + MAX_CACHED_PROJECT_SERVERS, ProjectRouteKey, execute_daemon_invocation, +}; + +struct CompilerRunCompleted(mpsc::UnboundedSender<()>); + +impl Layer for CompilerRunCompleted +where + S: tracing::Subscriber + for<'a> LookupSpan<'a>, +{ + fn on_close(&self, id: tracing::span::Id, context: Context<'_, S>) { + if context + .span(&id) + .is_some_and(|span| span.name() == "usecases.diagnostics.typescript_producer") + { + let _ = self.0.send(()); + } + } +} + +fn install_project_compiler(project: &Path, compiler: &Path, missing: &str) { + std::fs::create_dir_all(project.join("node_modules/.bin")).unwrap(); + std::os::unix::fs::symlink(compiler, project.join("node_modules/.bin/tsc")).unwrap(); + std::fs::write( + project.join("node_modules/check.json"), + format!(r#"{{"extends":"./{missing}.json"}}"#), + ) + .unwrap(); +} + +async fn compiler_completed(completed: &mut mpsc::UnboundedReceiver<()>) { + tokio::time::timeout(Duration::from_mins(1), completed.recv()) + .await + .expect("the production producer must finish its real tsc check") + .expect("compiler completion observer"); +} + +async fn diagnostics( + engine: &DaemonEngine, + handshake: &DaemonHandshake, +) -> DaemonInvocationOutcome { + let now = tracedecay_contracts::now_micros(); + execute_daemon_invocation( + engine, + handshake, + DaemonInvocationRequest::primitive( + "request.compiler-status-reopen", + ApplicationSurfaceOperation::DiagnosticsRead, + PrimitiveRequest::DiagnosticsRead(DiagnosticsPrimitiveRequest { + scope: DiagnosticsPrimitiveScope::Workspace, + maximum_diagnostics: 10, + cursor: None, + }), + now, + Deadline::new(UtcMicros(now.0 + 30_000_000)).unwrap(), + CancellationContext::active("cancel.compiler-status-reopen").unwrap(), + ) + .unwrap(), + ) + .await + .outcome +} + +fn compiler_failure(outcome: &DaemonInvocationOutcome) -> String { + let DaemonInvocationOutcome::ApplicationProblem { problem } = outcome else { + panic!("a failed compiler must answer a public typed problem: {outcome:?}"); + }; + let diagnostic = problem.diagnostic().expect("compiler failure diagnostic"); + assert_eq!(diagnostic.code, "application.diagnostics.producer-failed"); + diagnostic.message.clone() +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn route_eviction_reopen_reads_retained_compiler_status_with_leased_sibling() { + const CHILD: &str = "TRACEDECAY_TEST_COMPILER_STATUS_REOPEN_CHILD"; + if std::env::var_os(CHILD).is_none() { + let name = format!( + "{}::route_eviction_reopen_reads_retained_compiler_status_with_leased_sibling", + module_path!().split_once("::").unwrap().1 + ); + let home = TempDir::new().unwrap(); + let output = Command::new(std::env::current_exe().unwrap()) + .args(["--exact", &name, "--test-threads=1", "--nocapture"]) + .env(CHILD, "1") + .env("HOME", home.path()) + .env("XDG_CONFIG_HOME", home.path().join("config")) + .env("GIT_CONFIG_GLOBAL", "/dev/null") + .env("GIT_CONFIG_NOSYSTEM", "1") + .output() + .expect("isolated compiler-status daemon journey"); + assert!( + output.status.success() + && String::from_utf8_lossy(&output.stdout).contains("test result: ok. 1 passed"), + "compiler-status daemon journey failed:\n{}\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr), + ); + return; + } + + let compiler = PathBuf::from( + std::env::var_os("TRACEDECAY_TEST_TYPESCRIPT_COMPILER") + .expect("set TRACEDECAY_TEST_TYPESCRIPT_COMPILER to the installed pinned tsc"), + ) + .canonicalize() + .expect("installed pinned tsc"); + let version = Command::new(&compiler).arg("--version").output().unwrap(); + assert!( + version.status.success(), + "installed tsc must run: {version:?}" + ); + println!( + "installed project compiler: {}", + String::from_utf8_lossy(&version.stdout).trim() + ); + let (completed, mut completions) = mpsc::unbounded_channel(); + tracing::subscriber::set_global_default( + tracing_subscriber::registry().with(CompilerRunCompleted(completed)), + ) + .unwrap(); + + let temp = TempDir::new().unwrap(); + let primary = temp.path().join("primary"); + let profile = temp.path().join("profile"); + std::fs::create_dir_all(primary.join("src")).unwrap(); + std::fs::write(primary.join("src/index.ts"), "export const value = 1;\n").unwrap(); + std::fs::write(primary.join(".gitignore"), "node_modules/\n").unwrap(); + std::fs::write( + primary.join("tsconfig.json"), + r#"{"extends":"./node_modules/check.json","include":["src/**/*.ts"]}"#, + ) + .unwrap(); + git(&primary, &["init", "--quiet", "--initial-branch=main"]); + git(&primary, &["add", "--all"]); + git( + &primary, + &["commit", "--quiet", "-m", "test: compiler ownership"], + ); + let client_identity = test_client_identity_for(profile.clone()); + initialize_test_project(&primary, &client_identity).await; + let primary = primary.canonicalize().unwrap(); + let handshake = DaemonHandshake { + project_path: Some(primary.clone()), + client_identity, + ..test_handshake_defaults() + }; + let _database_scope = enter_test_daemon_database_scope(&profile, "compiler-status-reopen"); + let configuration_engine = test_daemon_engine_for_profile(&profile); + drop( + apply_project_setting_via_surface(&configuration_engine, &handshake, |_| { + ( + tracedecay_domain::configuration::SettingKey::new( + tracedecay_domain::configuration::SYNC_WATCH_LINKED_WORKTREES_SETTING_KEY, + ) + .unwrap(), + tracedecay_domain::configuration::ConfigurationValueV1::Boolean(true), + ) + }) + .await, + ); + let shutdown = configuration_engine.shutdown_all().await; + assert!(shutdown.project_servers.is_clean(), "{shutdown:?}"); + assert!(shutdown.background.unfinished().is_empty(), "{shutdown:?}"); + drop(configuration_engine); + install_project_compiler(&primary, &compiler, "predecessor-missing"); + let engine = test_daemon_engine_for_profile(&profile); + let (primary_key, _, primary_server, _) = engine.open_project_server(&handshake).await.unwrap(); + compiler_completed(&mut completions).await; + let before = diagnostics(&engine, &handshake).await; + assert!(compiler_failure(&before).contains("predecessor-missing.json")); + let retained_dispatch = engine + .invocation + .service + .project_runtimes + .read::(&primary, PrimitiveProjectRuntime::dispatch) + .await + .unwrap(); + drop(primary_server); + + let mut leased_siblings = Vec::new(); + let mut sibling_handshake = None; + for index in 0..MAX_CACHED_PROJECT_SERVERS { + let linked = temp.path().join(format!("linked-{index}")); + git( + &primary, + &[ + "worktree", + "add", + "--quiet", + "--detach", + linked.to_str().unwrap(), + ], + ); + if index == 0 { + install_project_compiler(&linked, &compiler, "sibling-missing"); + } + let linked_handshake = DaemonHandshake { + project_path: Some(linked), + ..handshake.clone() + }; + let (key, _, server, _) = engine.open_project_server(&linked_handshake).await.unwrap(); + assert_eq!( + key.owner, primary_key.owner, + "linked roots share the real store owner" + ); + if index == 0 { + compiler_completed(&mut completions).await; + sibling_handshake = Some(linked_handshake); + } + leased_siblings.push(server); + } + let route = ProjectRouteKey::from_handshake(&primary, &handshake).unwrap(); + { + let mut servers = engine.store_administration.project_servers().lock().await; + assert_eq!(servers.servers.len(), MAX_CACHED_PROJECT_SERVERS); + assert!( + servers.get_route(&route).is_none(), + "the primary route must actually be evicted" + ); + assert!( + servers + .retire_lru_ready_under_graph_pressure(|server| Arc::strong_count(server) > 1) + .is_err(), + "leased siblings must prevent whole-owner retirement" + ); + } + tokio::time::timeout( + Duration::from_mins(1), + engine + .store_administration + .wait_for_retirement_reaper_count_for_test(0), + ) + .await + .expect("route eviction cleanup must finish before reopen"); + let still_retained = engine + .invocation + .service + .project_runtimes + .read::(&primary, PrimitiveProjectRuntime::dispatch) + .await + .expect("sibling route eviction retains the invocation runtime"); + assert!(Arc::ptr_eq(&retained_dispatch, &still_retained)); + std::fs::write( + primary.join("node_modules/check.json"), + r#"{"extends":"./reopened-missing.json"}"#, + ) + .unwrap(); + assert_eq!(git(&primary, &["status", "--porcelain"]), ""); + drop(leased_siblings.pop()); + let (_, _, reopened, _) = engine.open_project_server(&handshake).await.unwrap(); + compiler_completed(&mut completions).await; + let after = diagnostics(&engine, &handshake).await; + let sibling = diagnostics(&engine, sibling_handshake.as_ref().unwrap()).await; + drop((retained_dispatch, still_retained, reopened, leased_siblings)); + let shutdown = engine.shutdown_all().await; + assert!(shutdown.project_servers.is_clean(), "{shutdown:?}"); + assert!(shutdown.background.unfinished().is_empty(), "{shutdown:?}"); + assert!(engine.invocation.service.project_runtimes.is_empty()); + assert!( + compiler_failure(&after).contains("reopened-missing.json"), + "reopen must expose the completed new check, not the retained predecessor: {after:?}" + ); + assert!(compiler_failure(&sibling).contains("sibling-missing.json")); + println!("public diagnostics observed predecessor, reopened, and isolated sibling failures"); +} diff --git a/dashboard/BUILD.bazel b/dashboard/BUILD.bazel index 4557220314..87b41ee06d 100644 --- a/dashboard/BUILD.bazel +++ b/dashboard/BUILD.bazel @@ -38,6 +38,13 @@ vitest_bin.vitest_test( tags = ["cpu:4"], ) +tsc_bin.tsc_binary( + name = "typescript_compiler", + testonly = True, + # The daemon invokes this compiler from each isolated project's directory. + env = {"BAZEL_BINDIR": "."}, +) + tsc_bin.tsc_test( name = "typecheck", include_types = True, diff --git a/scripts/bazel/gen_builds.py b/scripts/bazel/gen_builds.py index 366be494d0..8b3ee65a1f 100755 --- a/scripts/bazel/gen_builds.py +++ b/scripts/bazel/gen_builds.py @@ -172,6 +172,11 @@ # path, so no suite depends on Cargo's target// layout. Binaries name # their `__test` build, which carries the test run's features as Cargo's would. RUNTIME_BINARIES = ( + ( + "TRACEDECAY_TEST_TYPESCRIPT_COMPILER", + "//dashboard:typescript_compiler", + r'var_os\("TRACEDECAY_TEST_TYPESCRIPT_COMPILER"\)', + ), ( "TRACEDECAY_TYPESCRIPT_SDK_DIST", "//sdks/typescript:dist", From fa7c69bcff72cc52e0c0699be9f5bf6a87d9043c Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 26 Oct 2026 17:51:31 -0700 Subject: [PATCH 6/9] fix(diagnostics): fence superseded compiler publications --- crates/tracedecay-application/BUILD.bazel | 1 + .../src/diagnostics_producer.rs | 299 ++++++------------ .../src/diagnostics_producer/overlap_tests.rs | 169 ++++++++++ .../src/diagnostics_producer/status.rs | 257 +++++++++++++++ .../src/primitives/runtime.rs | 18 +- .../compiler_diagnostics_producer.rs | 6 +- 6 files changed, 535 insertions(+), 215 deletions(-) create mode 100644 crates/tracedecay-application/src/diagnostics_producer/overlap_tests.rs create mode 100644 crates/tracedecay-application/src/diagnostics_producer/status.rs diff --git a/crates/tracedecay-application/BUILD.bazel b/crates/tracedecay-application/BUILD.bazel index 6c83d7e155..1f79b59bdc 100644 --- a/crates/tracedecay-application/BUILD.bazel +++ b/crates/tracedecay-application/BUILD.bazel @@ -137,6 +137,7 @@ rust_test( rustc_flags = perf_rustc_flags(1), compile_data = glob(["tests/**","assets/**","fixtures/**","data/**","resources/**","vendor/**"], allow_empty = True), data = glob(["tests/**","assets/**","fixtures/**","data/**","resources/**","vendor/**"], allow_empty = True), + runfiles_env = {"TRACEDECAY_TEST_TYPESCRIPT_COMPILER": "//dashboard:typescript_compiler"}, ) filegroup( diff --git a/crates/tracedecay-application/src/diagnostics_producer.rs b/crates/tracedecay-application/src/diagnostics_producer.rs index c42994382f..267340b5a6 100644 --- a/crates/tracedecay-application/src/diagnostics_producer.rs +++ b/crates/tracedecay-application/src/diagnostics_producer.rs @@ -20,13 +20,13 @@ use std::collections::{BTreeMap, BTreeSet}; use std::panic::AssertUnwindSafe; use std::path::{Path, PathBuf}; -use std::sync::{Arc, RwLock}; +use std::sync::Arc; use futures_util::FutureExt; use futures_util::future::join_all; use sha2::{Digest, Sha256}; use tokio::sync::Semaphore; -use tracedecay_domain::{CodeGenerationId, ContentDigest, UtcMicros}; +use tracedecay_domain::{ContentDigest, UtcMicros}; use tracedecay_lsp::{ SearchedTsconfig, TypeScriptFileOwner, TypeScriptProject, TypeScriptProjectConfigInputs, run_typescript_compiler, typescript_build_info_program, typescript_file_owner, @@ -42,111 +42,33 @@ use crate::diagnostics_publication::{ }; use crate::diagnostics_store::DiagnosticsStore; -/// Outcome of one producer run over one project root. -#[derive(Clone, Debug, PartialEq, Eq)] -pub enum CompilerProducerRunV1 { - Published { - generation: CodeGenerationId, - inserted: u64, - unresolved: usize, - /// Projects tsc checked for this generation. - checked: usize, - /// Projects whose previous result this generation republished. - reused: usize, - }, - /// The compiler reported findings, but none resolved onto a file the - /// code-index generation knows, so nothing was published. - NoResolvableDiagnostics { - unresolved: Vec, - }, - /// The code index has no complete generation for this root yet. - CodeIndexGenerationUnavailable, - /// No project could be checked: every compiler failed (spawn failure, a - /// tsconfig naming no inputs, a crash), or none is installed. - CompilerFailed { - reason: String, - }, - PublicationFailed { - reason: String, - }, -} - -/// How one tsconfig fared in the most recent run. A project with no compiler -/// is not recorded: reads resolve that from the filesystem. -#[derive(Clone, Debug, PartialEq, Eq)] -pub enum TypeScriptProjectCheckV1 { - /// Its findings are part of the published snapshot. - Checked, - Failed { - reason: String, - }, -} - -/// The latest producer outcome of one project owner: the run and how each -/// tsconfig fared in it. -#[derive(Clone, Debug, PartialEq, Eq)] -pub struct TypeScriptProducerStatusV1 { - run: CompilerProducerRunV1, - checks: BTreeMap, -} - -impl TypeScriptProducerStatusV1 { - #[must_use] - pub fn new( - run: CompilerProducerRunV1, - checks: BTreeMap, - ) -> Self { - Self { run, checks } - } - - #[must_use] - pub fn run(&self) -> &CompilerProducerRunV1 { - &self.run - } -} - -/// Read view of one project owner's producer status. The producer task and the -/// owner's diagnostics reads share one handle created with the owner, so a -/// successor owner at the same path starts with no status and a superseded -/// producer can only publish into the handle it was given. -#[derive(Clone, Debug, Default)] -pub struct TypeScriptProducerStatusHandleV1 { - latest: Arc>>>, -} +mod status; -impl TypeScriptProducerStatusHandleV1 { - fn publish(&self, status: TypeScriptProducerStatusV1) { - *self - .latest - .write() - .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(Arc::new(status)); - } +#[cfg(all(test, unix))] +mod overlap_tests; - /// The most recent producer status, if the producer has run. - #[must_use] - pub fn current(&self) -> Option> { - self.latest - .read() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .clone() - } -} +use status::TypeScriptProducerStatusWriterV1; +pub use status::{ + CompilerProducerRunV1, TypeScriptProducerStatusHandleV1, TypeScriptProducerStatusV1, + TypeScriptProjectCheckV1, +}; /// What one project root's producer remembers between generations: the file /// digests of the generation it last checked and how each project fared. pub struct TypeScriptProducerStateV1 { - status: TypeScriptProducerStatusHandleV1, + status: TypeScriptProducerStatusWriterV1, checked_files: Option>, projects: BTreeMap, } impl TypeScriptProducerStateV1 { - /// State for one project owner's producer, publishing its status into - /// `status`. + /// Appoint this producer as the current writer of `status`. Construct its + /// state before spawning it: a retained runtime may still have an older + /// producer completing a check while the old route is being retired. #[must_use] pub fn new(status: TypeScriptProducerStatusHandleV1) -> Self { Self { - status, + status: status.start_producer(), checked_files: None, projects: BTreeMap::new(), } @@ -197,45 +119,6 @@ pub async fn run_typescript_producer_v1( observed_at: UtcMicros, ) -> CompilerProducerRunV1 { let mut checks = BTreeMap::new(); - let run = produce( - project_root, - projects, - identity, - store, - build_info_dir, - state, - observed_at, - &mut checks, - ) - .await; - state - .status - .publish(TypeScriptProducerStatusV1::new(run.clone(), checks)); - run -} - -#[expect( - clippy::too_many_arguments, - reason = "One producer run: its inputs, the state it carries, and the checks it records." -)] -async fn produce( - project_root: &Path, - projects: &[TypeScriptProject], - identity: &CodeIndexPublicationIdentityV1, - store: &DiagnosticsStore<'_>, - build_info_dir: &Path, - state: &mut TypeScriptProducerStateV1, - observed_at: UtcMicros, - checks: &mut BTreeMap, -) -> CompilerProducerRunV1 { - let (Ok(analyzer_revision), Ok(configuration_revision)) = ( - compiler_diagnostic_analyzer_revision_v1(), - compiler_diagnostic_configuration_revision_v1(), - ) else { - return CompilerProducerRunV1::PublicationFailed { - reason: "compiler pillar identity is unavailable".to_owned(), - }; - }; let files = identity .file_digests() .map(|(path, digest)| (path.to_owned(), digest.clone())) @@ -282,53 +165,70 @@ async fn produce( checks.insert(tsconfig.clone(), check); } state.projects = results; - if !checks - .values() - .any(|check| *check == TypeScriptProjectCheckV1::Checked) - { - return CompilerProducerRunV1::CompilerFailed { - reason: if failures.is_empty() { - "no TypeScript project under the project root has a compiler installed".to_owned() + let Some(publication) = state.status.begin_publication().await else { + return CompilerProducerRunV1::Superseded; + }; + let run = match ( + compiler_diagnostic_analyzer_revision_v1(), + compiler_diagnostic_configuration_revision_v1(), + ) { + (Ok(analyzer_revision), Ok(configuration_revision)) => { + if checks + .values() + .any(|check| *check == TypeScriptProjectCheckV1::Checked) + { + match publish_compiler_diagnostics_for_identity_v1( + project_root, + identity, + store, + &parsed, + analyzer_revision, + configuration_revision, + observed_at, + ) + .await + { + CompilerDiagnosticPublicationOutcomeV1::Published { + generation, + report, + unresolved, + } => CompilerProducerRunV1::Published { + generation, + inserted: report.inserted, + unresolved: unresolved.len(), + checked, + reused, + }, + CompilerDiagnosticPublicationOutcomeV1::NoResolvableDiagnostics { + unresolved, + } => CompilerProducerRunV1::NoResolvableDiagnostics { + unresolved: unresolved.iter().map(ToString::to_string).collect(), + }, + CompilerDiagnosticPublicationOutcomeV1::CodeIndexIdentityUnavailable + | CompilerDiagnosticPublicationOutcomeV1::CodeIndexGenerationUnavailable => { + CompilerProducerRunV1::CodeIndexGenerationUnavailable + } + CompilerDiagnosticPublicationOutcomeV1::Failed { reason } => { + CompilerProducerRunV1::PublicationFailed { reason } + } + } } else { - failures.join("; ") - }, - }; - } - match publish_compiler_diagnostics_for_identity_v1( - project_root, - identity, - store, - &parsed, - analyzer_revision, - configuration_revision, - observed_at, - ) - .await - { - CompilerDiagnosticPublicationOutcomeV1::Published { - generation, - report, - unresolved, - } => CompilerProducerRunV1::Published { - generation, - inserted: report.inserted, - unresolved: unresolved.len(), - checked, - reused, - }, - CompilerDiagnosticPublicationOutcomeV1::NoResolvableDiagnostics { unresolved } => { - CompilerProducerRunV1::NoResolvableDiagnostics { - unresolved: unresolved.iter().map(ToString::to_string).collect(), + CompilerProducerRunV1::CompilerFailed { + reason: if failures.is_empty() { + "no TypeScript project under the project root has a compiler installed" + .to_owned() + } else { + failures.join("; ") + }, + } } } - CompilerDiagnosticPublicationOutcomeV1::CodeIndexIdentityUnavailable - | CompilerDiagnosticPublicationOutcomeV1::CodeIndexGenerationUnavailable => { - CompilerProducerRunV1::CodeIndexGenerationUnavailable - } - CompilerDiagnosticPublicationOutcomeV1::Failed { reason } => { - CompilerProducerRunV1::PublicationFailed { reason } - } - } + _ => CompilerProducerRunV1::PublicationFailed { + reason: "compiler pillar identity is unavailable".to_owned(), + }, + }; + publication.publish(TypeScriptProducerStatusV1::new(run.clone(), checks)); + run } /// Logical paths whose content differs between two generations, including @@ -610,28 +510,10 @@ mod tests { assert_eq!(warning.code, None); } - #[test] - fn a_status_handle_publishes_to_its_owner_only() { - let owner = TypeScriptProducerStatusHandleV1::default(); - assert_eq!(owner.current(), None); - let run = CompilerProducerRunV1::CompilerFailed { - reason: "tsc exited with 2".to_owned(), - }; - owner.publish(TypeScriptProducerStatusV1::new( - run.clone(), - BTreeMap::new(), - )); - assert_eq!( - owner.current().map(|status| status.run().clone()), - Some(run) - ); - assert_eq!(TypeScriptProducerStatusHandleV1::default().current(), None); - } - /// A read about a file reports the check of the tsconfig that owns it, /// not whichever package happened to publish. - #[test] - fn availability_reports_the_owning_tsconfigs_own_check() { + #[tokio::test] + async fn availability_reports_the_owning_tsconfigs_own_check() { let root = tempfile::tempdir().expect("project root"); let root = root.path(); for package in ["app", "lib"] { @@ -647,16 +529,21 @@ mod tests { reason: "TS5083: Cannot read file".to_owned(), }; let status = TypeScriptProducerStatusHandleV1::default(); - status.publish(TypeScriptProducerStatusV1::new( - CompilerProducerRunV1::CodeIndexGenerationUnavailable, - BTreeMap::from([ - ( - root.join("packages/app/tsconfig.json"), - TypeScriptProjectCheckV1::Checked, - ), - (lib.clone(), failed.clone()), - ]), - )); + status + .start_producer() + .begin_publication() + .await + .unwrap() + .publish(TypeScriptProducerStatusV1::new( + CompilerProducerRunV1::CodeIndexGenerationUnavailable, + BTreeMap::from([ + ( + root.join("packages/app/tsconfig.json"), + TypeScriptProjectCheckV1::Checked, + ), + (lib.clone(), failed.clone()), + ]), + )); let TypeScriptDiagnosticsAvailabilityV1::Configured { tsconfig, check, .. diff --git a/crates/tracedecay-application/src/diagnostics_producer/overlap_tests.rs b/crates/tracedecay-application/src/diagnostics_producer/overlap_tests.rs new file mode 100644 index 0000000000..b1b76ae6c2 --- /dev/null +++ b/crates/tracedecay-application/src/diagnostics_producer/overlap_tests.rs @@ -0,0 +1,169 @@ +//! A real compiler run may finish after its replacement has published. + +use std::path::PathBuf; +use std::sync::Arc; +use std::time::Duration; + +use tracedecay_domain::UtcMicros; +use tracedecay_domain::test_fixtures::id; +use tracedecay_lsp::typescript_projects; +use tracedecay_runtime_core::db::engine::TestConnection; +use tracedecay_runtime_core::test_executable::write_executable_script; + +use super::{ + CompilerProducerRunV1, TypeScriptDiagnosticsAvailabilityV1, TypeScriptProducerStateV1, + TypeScriptProducerStatusHandleV1, TypeScriptProjectCheckV1, run_typescript_producer_v1, + typescript_diagnostics_availability, +}; +use crate::diagnostics_publication::CodeIndexPublicationIdentityV1; +use crate::diagnostics_store::DiagnosticsStore; + +#[tokio::test] +async fn an_overlapping_old_compiler_cannot_replace_new_records_or_status_for_the_same_generation() +{ + let compiler = PathBuf::from( + std::env::var_os("TRACEDECAY_TEST_TYPESCRIPT_COMPILER") + .expect("set TRACEDECAY_TEST_TYPESCRIPT_COMPILER to the installed pinned tsc"), + ) + .canonicalize() + .expect("installed pinned tsc"); + let temp = tempfile::tempdir().expect("isolated compiler project"); + let root = temp.path().join("project"); + std::fs::create_dir_all(root.join("src")).unwrap(); + let source = + b"export function echo(value) { return value; }\nexport const label: string = 1;\n"; + std::fs::write(root.join("src/index.ts"), source).unwrap(); + std::fs::write( + root.join("tsconfig.json"), + r#"{"extends":"./node_modules/check.json","include":["src/**/*.ts"]}"#, + ) + .unwrap(); + let bin = root.join("node_modules/.bin"); + std::fs::create_dir_all(&bin).unwrap(); + std::os::unix::fs::symlink(compiler, bin.join("real-tsc")).unwrap(); + std::fs::write( + root.join("node_modules/check.json"), + r#"{"compilerOptions":{"noImplicitAny":true}}"#, + ) + .unwrap(); + std::fs::write(root.join("node_modules/hold-compiler"), "").unwrap(); + // Delay delivery of the installed compiler's actual result. The wrapper + // synthesizes no diagnostics: tsc has already read the old config before + // the replacement is admitted, but its producer is still awaiting exit. + write_executable_script( + &bin.join("tsc"), + b"#!/bin/sh\n\ + if [ -f node_modules/hold-compiler ]; then\n\ + ./node_modules/.bin/real-tsc \"$@\" > node_modules/old-output 2>&1\n\ + result=$?\n\ + : > node_modules/old-compiled\n\ + while [ ! -f node_modules/release-old ]; do sleep 0.01; done\n\ + cat node_modules/old-output\n\ + exit \"$result\"\n\ + fi\n\ + exec ./node_modules/.bin/real-tsc \"$@\"\n", + ) + .unwrap(); + let connection = TestConnection::open(&temp.path().join("diagnostics.db")); + let store = DiagnosticsStore::new_runtime(&connection); + let identity = CodeIndexPublicationIdentityV1::new( + id("generation.compiler-overlap"), + id("repository.compiler-overlap"), + None, + None, + None, + [( + "src/index.ts".to_owned(), + id("file.compiler-overlap"), + tracedecay_code_index::intake::content_digest(source), + )], + ); + let projects = typescript_projects(&root); + assert_eq!(projects.len(), 1); + let build_info = temp.path().join("typescript-build-info"); + let status = TypeScriptProducerStatusHandleV1::default(); + let mut predecessor = TypeScriptProducerStateV1::new(status.clone()); + let old_run = run_typescript_producer_v1( + &root, + &projects, + &identity, + &store, + &build_info, + &mut predecessor, + UtcMicros(1_700_000_000_000_000), + ); + let replacement_run = async { + while !root.join("node_modules/old-compiled").try_exists().unwrap() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + let old_output = std::fs::read_to_string(root.join("node_modules/old-output")).unwrap(); + assert!(old_output.contains("TS7006"), "{old_output}"); + assert!(old_output.contains("TS2322"), "{old_output}"); + assert_eq!(status.current(), None, "the old producer is still running"); + let mut replacement = TypeScriptProducerStateV1::new(status.clone()); + std::fs::write( + root.join("node_modules/check.json"), + r#"{"compilerOptions":{"noImplicitAny":false}}"#, + ) + .unwrap(); + std::fs::remove_file(root.join("node_modules/hold-compiler")).unwrap(); + let run = run_typescript_producer_v1( + &root, + &projects, + &identity, + &store, + &build_info, + &mut replacement, + UtcMicros(1_700_000_000_000_001), + ) + .await; + assert_eq!( + run, + CompilerProducerRunV1::Published { + generation: identity.generation_id().clone(), + inserted: 1, + unresolved: 0, + checked: 1, + reused: 0, + } + ); + let records = store + .current_records(identity.generation_id()) + .await + .unwrap(); + assert_eq!(records.len(), 1); + assert_eq!(records[0].code, "TS2322"); + let published = status.current().expect("replacement status"); + assert_eq!(published.run(), &run); + std::fs::write(root.join("node_modules/release-old"), "").unwrap(); + (published, records) + }; + let (old_run, (replacement_status, replacement_records)) = + tokio::time::timeout(Duration::from_mins(1), async { + tokio::join!(old_run, replacement_run) + }) + .await + .expect("both real compiler processes must complete"); + assert_eq!(old_run, CompilerProducerRunV1::Superseded); + assert_eq!( + store + .current_records(identity.generation_id()) + .await + .unwrap(), + replacement_records, + "late old findings must not create a newer publication of the same generation" + ); + assert!( + Arc::ptr_eq(&replacement_status, &status.current().unwrap()), + "the late old result must leave the replacement's snapshot intact" + ); + let TypeScriptDiagnosticsAvailabilityV1::Configured { + check: Some(TypeScriptProjectCheckV1::Checked), + last_run: Some(last_run), + .. + } = typescript_diagnostics_availability(&root, Some("src/index.ts".as_ref()), &status) + else { + panic!("the file must report its completed replacement check"); + }; + assert_eq!(&last_run, replacement_status.run()); +} diff --git a/crates/tracedecay-application/src/diagnostics_producer/status.rs b/crates/tracedecay-application/src/diagnostics_producer/status.rs new file mode 100644 index 0000000000..854b4a912e --- /dev/null +++ b/crates/tracedecay-application/src/diagnostics_producer/status.rs @@ -0,0 +1,257 @@ +//! Compiler status retained by one primitive runtime, with one current producer. + +use std::collections::BTreeMap; +use std::path::PathBuf; +use std::sync::{Arc, RwLock}; + +use tokio::sync::{Mutex, MutexGuard}; +use tracedecay_domain::CodeGenerationId; + +/// Outcome of one producer run over one project root. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum CompilerProducerRunV1 { + Published { + generation: CodeGenerationId, + inserted: u64, + unresolved: usize, + /// Projects tsc checked for this generation. + checked: usize, + /// Projects whose previous result this generation republished. + reused: usize, + }, + /// The compiler reported findings, but none resolved onto a file the + /// code-index generation knows, so nothing was published. + NoResolvableDiagnostics { + unresolved: Vec, + }, + /// The code index has no complete generation for this root yet. + CodeIndexGenerationUnavailable, + /// No project could be checked: every compiler failed (spawn failure, a + /// tsconfig naming no inputs, a crash), or none is installed. + CompilerFailed { + reason: String, + }, + PublicationFailed { + reason: String, + }, + /// A newer producer was appointed before this run entered publication. + Superseded, +} + +/// How one tsconfig fared in the most recent run. A project with no compiler +/// is not recorded: reads resolve that from the filesystem. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum TypeScriptProjectCheckV1 { + /// Its findings are part of the published snapshot. + Checked, + Failed { + reason: String, + }, +} + +/// The latest producer outcome of one project owner: the run and how each +/// tsconfig fared in it. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct TypeScriptProducerStatusV1 { + pub(super) run: CompilerProducerRunV1, + pub(super) checks: BTreeMap, +} + +impl TypeScriptProducerStatusV1 { + #[must_use] + pub fn new( + run: CompilerProducerRunV1, + checks: BTreeMap, + ) -> Self { + Self { run, checks } + } + + #[must_use] + pub fn run(&self) -> &CompilerProducerRunV1 { + &self.run + } +} + +/// Read view of the status retained by one primitive runtime. Reopening its +/// route preserves the last completed check and appoints a new producer; an +/// older task cannot overwrite it while that task's server is being retired. +#[derive(Clone, Debug, Default)] +pub struct TypeScriptProducerStatusHandleV1 { + status: Arc>, + publication: Arc>, +} + +#[derive(Debug, Default)] +struct ProducerStatus { + producer: Option>, + latest: Option>, +} + +impl TypeScriptProducerStatusHandleV1 { + /// Claim publication before spawning the producer, so task polling order + /// cannot let an older producer claim after its replacement. + pub(super) fn start_producer(&self) -> TypeScriptProducerStatusWriterV1 { + let producer = Arc::new(()); + self.status + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .producer = Some(Arc::clone(&producer)); + TypeScriptProducerStatusWriterV1 { + status: self.clone(), + producer, + } + } + + /// The most recent completed status from an admitted producer, if any. + #[must_use] + pub fn current(&self) -> Option> { + self.status + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .latest + .clone() + } +} + +/// Publication authority for one producer instance, not for a pathname. +pub(super) struct TypeScriptProducerStatusWriterV1 { + status: TypeScriptProducerStatusHandleV1, + producer: Arc<()>, +} + +impl TypeScriptProducerStatusWriterV1 { + /// Compiler work may overlap. Publication is ordered for this retained + /// runtime: an already-admitted old publication finishes before the new + /// one, and an old producer entering after replacement cannot publish. + pub(super) async fn begin_publication(&self) -> Option> { + let gate = self.status.publication.lock().await; + let current = self + .status + .status + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .producer + .as_ref() + .is_some_and(|producer| Arc::ptr_eq(producer, &self.producer)); + current.then_some(TypeScriptProducerPublicationV1 { + writer: self, + _gate: gate, + }) + } + + fn publish(&self, snapshot: TypeScriptProducerStatusV1) { + let mut status = self + .status + .status + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner); + // The authority check and replacement share this lock. A new claim + // cannot interleave between accepting this producer and its write. + if status + .producer + .as_ref() + .is_some_and(|producer| Arc::ptr_eq(producer, &self.producer)) + { + status.latest = Some(Arc::new(snapshot)); + } + } +} + +/// Held across the record transaction and the matching status publication. +pub(super) struct TypeScriptProducerPublicationV1<'a> { + writer: &'a TypeScriptProducerStatusWriterV1, + _gate: MutexGuard<'a, ()>, +} + +impl TypeScriptProducerPublicationV1<'_> { + pub(super) fn publish(self, snapshot: TypeScriptProducerStatusV1) { + self.writer.publish(snapshot); + } +} + +#[cfg(test)] +mod tests { + use super::{ + CompilerProducerRunV1, TypeScriptProducerStatusHandleV1, TypeScriptProducerStatusV1, + }; + use std::collections::BTreeMap; + + fn failed(reason: &str) -> TypeScriptProducerStatusV1 { + TypeScriptProducerStatusV1::new( + CompilerProducerRunV1::CompilerFailed { + reason: reason.to_owned(), + }, + BTreeMap::new(), + ) + } + + #[test] + fn a_status_handle_publishes_to_its_owner_only() { + let owner = TypeScriptProducerStatusHandleV1::default(); + assert_eq!(owner.current(), None); + owner.start_producer().publish(failed("owner")); + assert_eq!(owner.current().as_deref(), Some(&failed("owner"))); + assert_eq!(TypeScriptProducerStatusHandleV1::default().current(), None); + } + + #[test] + fn a_replacement_claim_fences_old_runs_before_the_replacement_is_polled() { + let owner = TypeScriptProducerStatusHandleV1::default(); + let predecessor = owner.start_producer(); + predecessor.publish(failed("last completed check")); + let sibling = TypeScriptProducerStatusHandleV1::default(); + let sibling_producer = sibling.start_producer(); + sibling_producer.publish(failed("sibling")); + + let replacement = owner.start_producer(); + predecessor.publish(failed("late old check")); + assert_eq!( + owner.current().as_deref(), + Some(&failed("last completed check")), + "a superseded task is fenced even before the new task runs" + ); + replacement.publish(failed("replacement")); + predecessor.publish(failed("another late old check")); + assert_eq!(owner.current().as_deref(), Some(&failed("replacement"))); + sibling_producer.publish(failed("sibling next check")); + assert_eq!( + sibling.current().as_deref(), + Some(&failed("sibling next check")), + "claims in one retained runtime do not fence a linked sibling" + ); + } + + #[tokio::test] + async fn an_entered_publication_finishes_before_its_replacement_and_does_not_gate_a_sibling() { + let owner = TypeScriptProducerStatusHandleV1::default(); + let predecessor = owner.start_producer(); + let old_publication = predecessor.begin_publication().await.unwrap(); + let replacement = owner.start_producer(); + let mut next_publication = Box::pin(replacement.begin_publication()); + std::future::poll_fn(|cx| { + assert!(next_publication.as_mut().poll(cx).is_pending()); + std::task::Poll::Ready(()) + }) + .await; + let sibling = TypeScriptProducerStatusHandleV1::default(); + sibling + .start_producer() + .begin_publication() + .await + .unwrap() + .publish(failed("sibling")); + assert_eq!(sibling.current().as_deref(), Some(&failed("sibling"))); + + old_publication.publish(failed("old record transaction completed")); + assert_eq!(owner.current(), None, "a new claim fences the old status"); + next_publication + .await + .unwrap() + .publish(failed("replacement")); + assert_eq!(owner.current().as_deref(), Some(&failed("replacement"))); + assert!( + predecessor.begin_publication().await.is_none(), + "the superseded producer must not enter another record transaction" + ); + } +} diff --git a/crates/tracedecay-application/src/primitives/runtime.rs b/crates/tracedecay-application/src/primitives/runtime.rs index 7c6576af6d..5cfe5944f4 100644 --- a/crates/tracedecay-application/src/primitives/runtime.rs +++ b/crates/tracedecay-application/src/primitives/runtime.rs @@ -1971,14 +1971,16 @@ fn unpublished_diagnostics_problem( // is the same pending state: the producer publishes exactly once // per generation, so the current generation's run is what the // caller is waiting on. - None | Some(CompilerProducerRunV1::CodeIndexGenerationUnavailable) => { - ApplicationProblem::from_detail( - ApplicationProblemDetailV1::DiagnosticsPending { - producer: compiler, - generation: None, - }, - ) - } + None + | Some( + CompilerProducerRunV1::CodeIndexGenerationUnavailable + | CompilerProducerRunV1::Superseded, + ) => ApplicationProblem::from_detail( + ApplicationProblemDetailV1::DiagnosticsPending { + producer: compiler, + generation: None, + }, + ), Some(CompilerProducerRunV1::Published { generation, .. }) => { ApplicationProblem::from_detail( ApplicationProblemDetailV1::DiagnosticsPending { diff --git a/crates/tracedecay/src/daemon/project_open_owners/compiler_diagnostics_producer.rs b/crates/tracedecay/src/daemon/project_open_owners/compiler_diagnostics_producer.rs index 1c76d13fbc..f835c91169 100644 --- a/crates/tracedecay/src/daemon/project_open_owners/compiler_diagnostics_producer.rs +++ b/crates/tracedecay/src/daemon/project_open_owners/compiler_diagnostics_producer.rs @@ -49,6 +49,10 @@ pub(super) fn spawn_typescript_diagnostics_producer( log_producer_event(&project_root, "code_index_disabled"); return false; } + // Reopen can retain the primitive runtime while the prior server's + // compiler task is still retiring. Fence it before this task can yield + // in discovery; scheduling order must not choose the current producer. + let mut state = TypeScriptProducerStateV1::new(status); owner.spawn_background_task(tracing::Instrument::instrument( async move { let Some(projects) = discover_projects(&project_root).await else { @@ -74,7 +78,6 @@ pub(super) fn spawn_typescript_diagnostics_producer( let schedulers = invocation.code_index_schedulers.clone(); let mut signals = CodeIndexOwnerSignalsV1::subscribe(&schedulers, &project_root).await; let build_info_dir = graph.store_layout().data_root.join(BUILD_INFO_DIR); - let mut state = TypeScriptProducerStateV1::new(status); // One producer run per generation, whatever its outcome: the root's // signals also wake on serving and pass changes, and a project // whose compiler keeps failing must not re-run it on each of them. @@ -191,6 +194,7 @@ fn log_producer_run(project_root: &Path, run: &CompilerProducerRunV1) { CompilerProducerRunV1::PublicationFailed { reason } => { ("publication_failed", reason.clone()) } + CompilerProducerRunV1::Superseded => ("superseded", String::new()), }; log_daemon_event( "typescript_diagnostics_producer", From 7f17d81ad1a2e580b5567a728a971b43172e21e3 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 26 Oct 2026 19:12:27 -0700 Subject: [PATCH 7/9] fix(primitives): refresh retained authority on configuration changes Rebuild configuration-bound primitive ports when a retained route reopens against a changed configuration, while preserving its producer status and operation capacity. Compare the installed dispatch with the captured one before replacement so late preparations cannot overwrite a successor. Exercise unchanged and changed configuration reopen through real daemon route eviction, a leased sibling, the configuration surface and pinned tsc. --- .../src/primitives/production.rs | 4 +- .../src/primitives/runtime.rs | 44 +++- .../src/invocation/primitive.rs | 31 ++- .../src/daemon/tests/compiler_status.rs | 226 ++++++++++++++++-- 4 files changed, 276 insertions(+), 29 deletions(-) diff --git a/crates/tracedecay-application/src/primitives/production.rs b/crates/tracedecay-application/src/primitives/production.rs index c4c8667cd4..280a1b1b04 100644 --- a/crates/tracedecay-application/src/primitives/production.rs +++ b/crates/tracedecay-application/src/primitives/production.rs @@ -907,10 +907,12 @@ impl ProductionPrimitiveOpenRequestV1 { Ok(()) } + /// Validate the stable store/source authority and report whether the + /// retained facade already has this configuration and capability set. pub fn validate_retained( &self, runtime: &PrimitiveProjectRuntime, - ) -> Result<(), ApplicationContractError> { + ) -> Result { runtime.validate_reopen( self.source_runtime.db(), &self.session_db, diff --git a/crates/tracedecay-application/src/primitives/runtime.rs b/crates/tracedecay-application/src/primitives/runtime.rs index 5cfe5944f4..748e3f6236 100644 --- a/crates/tracedecay-application/src/primitives/runtime.rs +++ b/crates/tracedecay-application/src/primitives/runtime.rs @@ -375,6 +375,7 @@ impl PrimitiveCapacity { /// Dropping this value releases the dispatch facade and every Arc-backed /// project primitive authority together. The database and Arc-backed ports /// keep the exact project-open owners alive for the same lifetime as dispatch. +#[derive(Clone)] pub struct PrimitiveProjectRuntime { database: Database, dispatch: Arc, @@ -418,7 +419,7 @@ impl PrimitiveProjectRuntime { session_db: &RegisteredGlobalDbLeaseV1, access: &ProjectSourceAccessSnapshot, admitted_root_uri: &str, - ) -> Result<(), ApplicationContractError> { + ) -> Result { // Request admission checks each renewed grant; its expiry is not // part of the retained source/configuration authority. let retained = &self.dispatch.access; @@ -431,15 +432,48 @@ impl PrimitiveProjectRuntime { || retained.scope != access.scope || retained.requester != access.requester || retained.binding != access.binding - || retained.configuration_revision != access.configuration_revision - || retained.configuration_digest != access.configuration_digest - || retained.configuration_provenance_digest != access.configuration_provenance_digest - || retained.effective_capabilities != access.effective_capabilities { return Err(ApplicationContractError::Inconsistent { field: "retained application primitive authority", }); } + Ok( + retained.configuration_revision == access.configuration_revision + && retained.configuration_digest == access.configuration_digest + && retained.configuration_provenance_digest + == access.configuration_provenance_digest + && retained.effective_capabilities == access.effective_capabilities, + ) + } + + /// Install one complete configuration-bound facade after preparing it + /// outside the registry lock. The captured dispatch fences a concurrent + /// reopen, so a delayed preparation cannot restore older authority. + pub fn reconcile_reopen( + &mut self, + captured: &Self, + replacement: &Self, + ) -> Result<(), ApplicationContractError> { + if self.validate_reopen( + &replacement.database, + &replacement.dispatch.test_runs.store, + &replacement.dispatch.access, + &replacement.dispatch.admitted_root_uri, + )? { + return Ok(()); + } + if !Arc::ptr_eq(&self.dispatch, &captured.dispatch) { + return Err(ApplicationContractError::Inconsistent { + field: "application primitive authority changed during reopen", + }); + } + let mut dispatch = (*replacement.dispatch).clone(); + // The store owner survives route eviction. Its producer publication + // fence and in-flight operation capacity survive this facade refresh + // too; only the configuration-bound ports and access are replaced. + dispatch.compiler_status = self.dispatch.compiler_status.clone(); + dispatch.capacity = self.dispatch.capacity.clone(); + self.dispatch = Arc::new(dispatch); Ok(()) } } diff --git a/crates/tracedecay-daemon-service/src/invocation/primitive.rs b/crates/tracedecay-daemon-service/src/invocation/primitive.rs index af3ec397bc..0063180995 100644 --- a/crates/tracedecay-daemon-service/src/invocation/primitive.rs +++ b/crates/tracedecay-daemon-service/src/invocation/primitive.rs @@ -1382,18 +1382,43 @@ impl DaemonPrimitiveRuntimeRegistrar { request: ProductionPrimitiveOpenRequestV1, ) -> Result { request.validate_registration_root(&project_root)?; - self.service + let retained = self + .service .project_runtimes .register_or_reconcile_with( - project_root, + project_root.clone(), |runtime: &mut PrimitiveProjectRuntime| { - request.validate_retained(runtime).map_err(Into::into) + request + .validate_retained(runtime) + .map(|_| ()) + .map_err(DaemonPrimitiveRuntimeRegistrationError::from) }, || async { open_production_primitive_runtime(request.clone()) .await + .map_err(DaemonPrimitiveRuntimeRegistrationError::from) + }, + PrimitiveProjectRuntime::clone, + ) + .await?; + if request.validate_retained(&retained)? { + return Ok(retained.compiler_status()); + } + + // A sibling can retain this root's store owner after route eviction. + // Rebuild every configuration-bound port together rather than merely + // updating the access snapshot while leaving old cursor authorities. + let replacement = open_production_primitive_runtime(request).await?; + self.service + .project_runtimes + .register_or_reconcile_with( + project_root, + |runtime: &mut PrimitiveProjectRuntime| { + runtime + .reconcile_reopen(&retained, &replacement) .map_err(Into::into) }, + || async { Ok(replacement.clone()) }, PrimitiveProjectRuntime::compiler_status, ) .await diff --git a/crates/tracedecay/src/daemon/tests/compiler_status.rs b/crates/tracedecay/src/daemon/tests/compiler_status.rs index b6bffc362a..d5572aca29 100644 --- a/crates/tracedecay/src/daemon/tests/compiler_status.rs +++ b/crates/tracedecay/src/daemon/tests/compiler_status.rs @@ -53,13 +53,6 @@ fn install_project_compiler(project: &Path, compiler: &Path, missing: &str) { .unwrap(); } -async fn compiler_completed(completed: &mut mpsc::UnboundedReceiver<()>) { - tokio::time::timeout(Duration::from_mins(1), completed.recv()) - .await - .expect("the production producer must finish its real tsc check") - .expect("compiler completion observer"); -} - async fn diagnostics( engine: &DaemonEngine, handshake: &DaemonHandshake, @@ -86,6 +79,35 @@ async fn diagnostics( .outcome } +async fn completed_compiler_failure( + engine: &DaemonEngine, + handshake: &DaemonHandshake, + completed: &mut mpsc::UnboundedReceiver<()>, + expected: &str, +) -> DaemonInvocationOutcome { + tokio::time::timeout(Duration::from_mins(1), async { + loop { + let outcome = diagnostics(engine, handshake).await; + if let DaemonInvocationOutcome::ApplicationProblem { problem } = &outcome + && problem.diagnostic().is_some_and(|diagnostic| { + diagnostic.code == "application.diagnostics.producer-failed" + && diagnostic.message.contains(expected) + }) + { + return outcome; + } + // Siblings can finish checks too. A completed run wakes the read; + // only this route's expected public outcome completes the wait. + completed + .recv() + .await + .expect("compiler completion observer"); + } + }) + .await + .expect("the production producer must expose its real tsc outcome") +} + fn compiler_failure(outcome: &DaemonInvocationOutcome) -> String { let DaemonInvocationOutcome::ApplicationProblem { problem } = outcome else { panic!("a failed compiler must answer a public typed problem: {outcome:?}"); @@ -190,20 +212,27 @@ async fn route_eviction_reopen_reads_retained_compiler_status_with_leased_siblin install_project_compiler(&primary, &compiler, "predecessor-missing"); let engine = test_daemon_engine_for_profile(&profile); let (primary_key, _, primary_server, _) = engine.open_project_server(&handshake).await.unwrap(); - compiler_completed(&mut completions).await; - let before = diagnostics(&engine, &handshake).await; + let before = completed_compiler_failure( + &engine, + &handshake, + &mut completions, + "predecessor-missing.json", + ) + .await; assert!(compiler_failure(&before).contains("predecessor-missing.json")); - let retained_dispatch = engine + let retained_runtime = engine .invocation .service .project_runtimes - .read::(&primary, PrimitiveProjectRuntime::dispatch) + .read::(&primary, PrimitiveProjectRuntime::clone) .await .unwrap(); + let retained_dispatch = retained_runtime.dispatch(); drop(primary_server); let mut leased_siblings = Vec::new(); let mut sibling_handshake = None; + let mut last_sibling_handshake = None; for index in 0..MAX_CACHED_PROJECT_SERVERS { let linked = temp.path().join(format!("linked-{index}")); git( @@ -228,8 +257,15 @@ async fn route_eviction_reopen_reads_retained_compiler_status_with_leased_siblin key.owner, primary_key.owner, "linked roots share the real store owner" ); + last_sibling_handshake = Some(linked_handshake.clone()); if index == 0 { - compiler_completed(&mut completions).await; + completed_compiler_failure( + &engine, + &linked_handshake, + &mut completions, + "sibling-missing.json", + ) + .await; sibling_handshake = Some(linked_handshake); } leased_siblings.push(server); @@ -273,18 +309,168 @@ async fn route_eviction_reopen_reads_retained_compiler_status_with_leased_siblin assert_eq!(git(&primary, &["status", "--porcelain"]), ""); drop(leased_siblings.pop()); let (_, _, reopened, _) = engine.open_project_server(&handshake).await.unwrap(); - compiler_completed(&mut completions).await; - let after = diagnostics(&engine, &handshake).await; + let after = completed_compiler_failure( + &engine, + &handshake, + &mut completions, + "reopened-missing.json", + ) + .await; let sibling = diagnostics(&engine, sibling_handshake.as_ref().unwrap()).await; - drop((retained_dispatch, still_retained, reopened, leased_siblings)); - let shutdown = engine.shutdown_all().await; - assert!(shutdown.project_servers.is_clean(), "{shutdown:?}"); - assert!(shutdown.background.unfinished().is_empty(), "{shutdown:?}"); - assert!(engine.invocation.service.project_runtimes.is_empty()); assert!( compiler_failure(&after).contains("reopened-missing.json"), "reopen must expose the completed new check, not the retained predecessor: {after:?}" ); assert!(compiler_failure(&sibling).contains("sibling-missing.json")); - println!("public diagnostics observed predecessor, reopened, and isolated sibling failures"); + let unchanged = engine + .invocation + .service + .project_runtimes + .read::(&primary, PrimitiveProjectRuntime::clone) + .await + .unwrap(); + assert!( + Arc::ptr_eq(&retained_dispatch, &unchanged.dispatch()), + "same-configuration reopen keeps the installed facade" + ); + let old_configuration = reopened + .cg() + .await + .configuration_runtime() + .client() + .current() + .await + .unwrap(); + drop(reopened); + + // Reopen the sibling displaced by the primary. Every other route is + // leased, so the now-idle primary is evicted while its runtime survives. + let (_, _, last_sibling, _) = engine + .open_project_server(last_sibling_handshake.as_ref().unwrap()) + .await + .unwrap(); + assert!( + engine + .store_administration + .project_servers() + .lock() + .await + .get_route(&route) + .is_none(), + "the primary must be evicted before its configuration changes" + ); + tokio::time::timeout( + Duration::from_mins(1), + engine + .store_administration + .wait_for_retirement_reaper_count_for_test(0), + ) + .await + .expect("second route eviction cleanup"); + drop( + apply_project_setting_via_surface(&engine, sibling_handshake.as_ref().unwrap(), |_| { + ( + tracedecay_domain::configuration::SettingKey::new( + tracedecay_domain::configuration::TELEMETRY_TIMINGS_SETTING_KEY, + ) + .unwrap(), + tracedecay_domain::configuration::ConfigurationValueV1::Boolean(false), + ) + }) + .await, + ); + std::fs::write( + primary.join("node_modules/check.json"), + r#"{"extends":"./configured-missing.json"}"#, + ) + .unwrap(); + drop(last_sibling); + let (_, _, configured, _) = engine.open_project_server(&handshake).await.unwrap(); + let current_configuration = configured + .cg() + .await + .configuration_runtime() + .client() + .current() + .await + .unwrap(); + assert_ne!( + old_configuration.revision_id(), + current_configuration.revision_id(), + "reopen must encounter the configuration committed through its sibling" + ); + let refreshed = engine + .invocation + .service + .project_runtimes + .read::(&primary, PrimitiveProjectRuntime::clone) + .await + .unwrap(); + assert!( + !Arc::ptr_eq(&retained_dispatch, &refreshed.dispatch()), + "changed configuration replaces the complete primitive facade" + ); + let after_configuration = completed_compiler_failure( + &engine, + &handshake, + &mut completions, + "configured-missing.json", + ) + .await; + assert!(compiler_failure(&after_configuration).contains("configured-missing.json")); + assert!(Arc::ptr_eq( + &retained_runtime.compiler_status().current().unwrap(), + &refreshed.compiler_status().current().unwrap(), + )); + + let mut delayed_reopen = refreshed.clone(); + assert!( + delayed_reopen + .reconcile_reopen(&unchanged, &retained_runtime) + .is_err(), + "a delayed old-configuration candidate cannot replace the refreshed facade" + ); + assert!(Arc::ptr_eq( + &delayed_reopen.dispatch(), + &refreshed.dispatch() + )); + let sibling_runtime = engine + .invocation + .service + .project_runtimes + .read::( + sibling_handshake + .as_ref() + .unwrap() + .project_path + .as_ref() + .unwrap(), + PrimitiveProjectRuntime::clone, + ) + .await + .unwrap(); + assert!( + delayed_reopen + .reconcile_reopen(&refreshed, &sibling_runtime) + .is_err(), + "a linked sibling's source authority cannot replace the primary's" + ); + let sibling_after = diagnostics(&engine, sibling_handshake.as_ref().unwrap()).await; + assert!(compiler_failure(&sibling_after).contains("sibling-missing.json")); + drop(( + retained_runtime, + retained_dispatch, + still_retained, + unchanged, + refreshed, + delayed_reopen, + sibling_runtime, + configured, + leased_siblings, + )); + let shutdown = engine.shutdown_all().await; + assert!(shutdown.project_servers.is_clean(), "{shutdown:?}"); + assert!(shutdown.background.unfinished().is_empty(), "{shutdown:?}"); + assert!(engine.invocation.service.project_runtimes.is_empty()); + println!("public diagnostics observed retained, reconfigured, and isolated sibling owners"); } From c1ca94e5157425d9fdf77e70f71543acd66234df Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 26 Oct 2026 21:53:51 -0700 Subject: [PATCH 8/9] fix(work): refresh retained configuration on route reopen Route eviction can retain a project's Work runtime while a sibling commits new configuration. Reopening the route previously refused that same store authority before the compiler producer could register. Refresh the policy, topology, routing and evidence admission together under the captured runtime and canonical current-configuration fence. Keep exact store, actor, scope, grant and Work authority checks. Refresh the retained fan-out recovery view without replacing recovery owners or their signal. Extend the real compiler route journey to verify stable grant identity, current Work authority, stale and sibling rejection, and retained authority after refusals. Preserve all existing assertions and budgets. Validation: pinned rustfmt and diff checks passed; independently reviewed. Native tests were not run locally because of disk and memory constraints. Hosted CI must verify the repaired route journey. --- .../src/invocation/registrars.rs | 98 ++++++++++--- .../src/invocation/types.rs | 15 ++ .../work/workflow_fan_out/recovery.rs | 17 ++- .../src/invocation/work_routing.rs | 10 ++ .../src/daemon/tests/compiler_status.rs | 134 +++++++++++++++++- 5 files changed, 253 insertions(+), 21 deletions(-) diff --git a/crates/tracedecay-daemon-service/src/invocation/registrars.rs b/crates/tracedecay-daemon-service/src/invocation/registrars.rs index 261cf30ec6..58a1de2114 100644 --- a/crates/tracedecay-daemon-service/src/invocation/registrars.rs +++ b/crates/tracedecay-daemon-service/src/invocation/registrars.rs @@ -1061,6 +1061,46 @@ impl DaemonWorkRuntimeRegistrar { canonical_sha256(&authority).map_err(|error| TraceDecayError::Config { message: format!("Workflow authority digest failed: {error}"), })?; + let retained = self + .service + .project_runtimes + .read::(&project_root, Clone::clone) + .await; + let refresh = if let Some(retained) = retained.filter(|retained| { + !retained.configuration_matches( + &policy_digest, + &configuration_digest, + &work_topology_policy, + &proposal_routing, + ) + }) { + let configuration = self + .service + .project_runtimes + .read::(&project_root, Clone::clone) + .await + .filter(|configuration| { + configuration.scope == grant.scope && configuration.actor == actor + }) + .ok_or_else(|| TraceDecayError::Config { + message: + "Work configuration refresh requires its registered configuration authority" + .to_owned(), + })?; + let current = configuration + .runtime + .client() + .current_state() + .await + .map_err(|error| TraceDecayError::Config { + message: format!( + "Work configuration refresh could not read current state: {error}" + ), + })?; + Some((retained, current)) + } else { + None + }; let observability_producer = self .service .observability_producer(Some(&project_root)) @@ -1081,27 +1121,53 @@ impl DaemonWorkRuntimeRegistrar { // whose server was replaced, and left it permanently // degraded. Its own project scope is already proven by // `grant.scope` and the authority digest. - if registered.actor == actor + if registered.database.binding() == database.binding() + && registered.database.verified_locator() == database.verified_locator() + && registered.actor == actor && registered.grant.digest == grant.digest && registered.grant.scope == grant.scope && registered.authority_digest == authority_digest - && registered.policy_digest == policy_digest - && registered.configuration_digest == configuration_digest - && registered - .proposal_routing - .same_configuration_as(&proposal_routing) { - // The same authority re-registering only renews its grant. - if registered.grant != grant { - let recovery = registered.workflow_fan_out_recovery.as_ref().ok_or_else( - || TraceDecayError::Config { - message: "Workflow fan-out recovery owner is not mounted" - .to_owned(), - }, - )?; - recovery.refresh_grant(grant.clone()); - registered.grant = grant.clone(); + if !registered.configuration_matches( + &policy_digest, + &configuration_digest, + &work_topology_policy, + &proposal_routing, + ) { + // The durable current-state read belongs to this + // captured incumbent. Check both inside the registry + // transition so a delayed reopen cannot overwrite a + // newer installed configuration. + let Some((captured, current)) = &refresh else { + return Err(TraceDecayError::Config { + message: "Work configuration changed during reopen".to_owned(), + }); + }; + if !registered.configuration_matches( + &captured.policy_digest, + &captured.configuration_digest, + &captured.work_topology_policy, + &captured.proposal_routing, + ) || !proposal_routing.matches_current_configuration(current) + { + return Err(TraceDecayError::Config { + message: "Work configuration changed during reopen".to_owned(), + }); + } } + let recovery = registered.workflow_fan_out_recovery.clone().ok_or_else( + || TraceDecayError::Config { + message: "Workflow fan-out recovery owner is not mounted" + .to_owned(), + }, + )?; + registered.grant = grant.clone(); + registered.policy_digest = policy_digest.clone(); + registered.configuration_digest = configuration_digest.clone(); + registered.work_topology_policy = work_topology_policy.clone(); + registered.proposal_routing = proposal_routing.clone(); + registered.evidence_retrieval = evidence_retrieval.clone(); + recovery.refresh_admission(registered); return Ok(()); } Err(TraceDecayError::Config { diff --git a/crates/tracedecay-daemon-service/src/invocation/types.rs b/crates/tracedecay-daemon-service/src/invocation/types.rs index c230e0b6da..ea910da401 100644 --- a/crates/tracedecay-daemon-service/src/invocation/types.rs +++ b/crates/tracedecay-daemon-service/src/invocation/types.rs @@ -984,6 +984,21 @@ pub struct RegisteredWorkRuntime { } impl RegisteredWorkRuntime { + pub(super) fn configuration_matches( + &self, + policy_digest: &ManifestDigest, + configuration_digest: &ManifestDigest, + work_topology_policy: &tracedecay_domain::configuration::WorkTopologyPolicyV1, + proposal_routing: &super::work_routing::DaemonWorkProposalRoutingAuthorityV1, + ) -> bool { + &self.policy_digest == policy_digest + && &self.configuration_digest == configuration_digest + && &self.work_topology_policy == work_topology_policy + && self + .proposal_routing + .same_configuration_as(proposal_routing) + } + /// The registered grant, bounded to one request observed at `observed_at`. pub(crate) fn grant_at(&self, observed_at: UtcMicros) -> CapabilityGrantSnapshot { mounted_grant_at(&self.grant, observed_at) diff --git a/crates/tracedecay-daemon-service/src/invocation/work/workflow_fan_out/recovery.rs b/crates/tracedecay-daemon-service/src/invocation/work/workflow_fan_out/recovery.rs index 457347505f..8d2010cf47 100644 --- a/crates/tracedecay-daemon-service/src/invocation/work/workflow_fan_out/recovery.rs +++ b/crates/tracedecay-daemon-service/src/invocation/work/workflow_fan_out/recovery.rs @@ -348,12 +348,21 @@ impl WorkflowFanOutRecoveryOwnerV1 { }) } - pub(crate) fn refresh_grant(&self, grant: tracedecay_contracts::CapabilityGrantSnapshot) { - self.inner + pub(crate) fn refresh_admission(&self, registered: &RegisteredWorkRuntime) { + let mut retained = self + .inner .registered .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .grant = grant; + .unwrap_or_else(std::sync::PoisonError::into_inner); + // Future recovery pages use the same complete admission as requests. + // Keep the existing recovery owners and signal; copying the published + // fan-out owner into its own retained state would also create a cycle. + retained.grant = registered.grant.clone(); + retained.policy_digest = registered.policy_digest.clone(); + retained.configuration_digest = registered.configuration_digest.clone(); + retained.work_topology_policy = registered.work_topology_policy.clone(); + retained.proposal_routing = registered.proposal_routing.clone(); + retained.evidence_retrieval = registered.evidence_retrieval.clone(); } /// Stop this owner from starting another fan-out reconciliation pass, diff --git a/crates/tracedecay-daemon-service/src/invocation/work_routing.rs b/crates/tracedecay-daemon-service/src/invocation/work_routing.rs index 41000b17a6..29f17d48ff 100644 --- a/crates/tracedecay-daemon-service/src/invocation/work_routing.rs +++ b/crates/tracedecay-daemon-service/src/invocation/work_routing.rs @@ -125,6 +125,16 @@ impl DaemonWorkProposalRoutingAuthorityV1 { && self.executable_bindings == other.executable_bindings } + pub(super) fn matches_current_configuration( + &self, + current: &tracedecay_global_db::configuration::contracts::ports::ConfigurationCurrentStateV1, + ) -> bool { + self.configuration_revision == current.revision_id + && self.configuration_snapshot == current.snapshot.snapshot_id + && self.configuration_digest == current.snapshot.effective_behavior_digest + && self.resolution_provenance_digest == current.snapshot.resolution_provenance_digest + } + pub(super) fn matches_scope(&self, scope: &ResolvedScope) -> bool { &self.scope == scope } diff --git a/crates/tracedecay/src/daemon/tests/compiler_status.rs b/crates/tracedecay/src/daemon/tests/compiler_status.rs index d5572aca29..180141b569 100644 --- a/crates/tracedecay/src/daemon/tests/compiler_status.rs +++ b/crates/tracedecay/src/daemon/tests/compiler_status.rs @@ -8,11 +8,17 @@ use std::time::Duration; use tempfile::TempDir; use tokio::sync::mpsc; use tracedecay_application::primitives::PrimitiveProjectRuntime; +use tracedecay_application::project_open_authorization::project_open_work_grant; +use tracedecay_configuration::config::PinnedRuntimeConfiguration; use tracedecay_contracts::retrieval::{ DiagnosticsPrimitiveRequest, DiagnosticsPrimitiveScope, PrimitiveRequest, }; use tracedecay_contracts::{CancellationContext, Deadline}; -use tracedecay_domain::UtcMicros; +use tracedecay_daemon_service::{ + DaemonWorkProposalRoutingAuthorityV1, daemon_owned_project_source_access_at, +}; +use tracedecay_domain::errors::TraceDecayError; +use tracedecay_domain::{ManifestDigest, UtcMicros, WorkAuthority, canonical_sha256}; use tracedecay_tool_catalog::ApplicationSurfaceOperation; use tracing_subscriber::layer::{Context, Layer, SubscriberExt}; use tracing_subscriber::registry::LookupSpan; @@ -117,6 +123,95 @@ fn compiler_failure(outcome: &DaemonInvocationOutcome) -> String { diagnostic.message.clone() } +async fn register_work_for_configuration( + engine: &DaemonEngine, + target_root: &Path, + server: &crate::mcp::McpServer, + configuration: &PinnedRuntimeConfiguration, +) -> (ManifestDigest, bool, Result<(), TraceDecayError>) { + let graph = server.cg().await; + let scope = tracedecay_code_index_runtime::resolved_scope_for_project( + graph.project_root(), + &graph + .configuration_runtime() + .configuration_target() + .project_id, + ) + .unwrap(); + let access = daemon_owned_project_source_access_at( + &scope, + graph.project_root(), + configuration, + tracedecay_contracts::now_micros(), + ) + .unwrap(); + let database = graph + .store_runtime_registry() + .mount_registered_project_sessions(scope.project_id.clone()) + .await + .unwrap(); + let grant = project_open_work_grant(&access, tracedecay_contracts::now_micros()) + .unwrap() + .unwrap(); + let authority = WorkAuthority::new( + scope.project_id.clone(), + scope.repository_id.clone(), + scope.worktree_id.clone(), + access.requester.clone(), + grant.digest.clone(), + ) + .unwrap(); + let policy_digest = canonical_sha256(&( + "tracedecay.daemon.configuration-policy.v1", + &scope.scope_digest, + &access.configuration_digest, + &access.configuration_provenance_digest, + )) + .unwrap(); + let routing = DaemonWorkProposalRoutingAuthorityV1::mount( + scope.clone(), + configuration, + &access.configuration_digest, + &grant, + ) + .unwrap(); + let topology = tracedecay_configuration::config::topology::resolved_work_topology_policy( + configuration.snapshot(), + ) + .unwrap() + .clone(); + drop(graph); + let registrar = engine.invocation.work_runtime_registrar(); + let matches = registrar + .authority_matches( + target_root, + &authority, + &access.requester, + &grant, + &policy_digest, + &access.configuration_digest, + ) + .await; + let grant_digest = grant.digest.clone(); + let result = registrar + .register( + target_root.to_path_buf(), + database, + authority, + access.requester, + grant, + policy_digest, + access.configuration_digest, + topology, + routing, + server + .work_evidence_retrieval(&scope, engine.invocation.work_federated_query_authority()) + .unwrap(), + ) + .await; + (grant_digest, matches, result) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn route_eviction_reopen_reads_retained_compiler_status_with_leased_sibling() { const CHILD: &str = "TRACEDECAY_TEST_COMPILER_STATUS_REOPEN_CHILD"; @@ -341,6 +436,10 @@ async fn route_eviction_reopen_reads_retained_compiler_status_with_leased_siblin .current() .await .unwrap(); + let (old_work_grant, old_work_matches, old_work_registration) = + register_work_for_configuration(&engine, &primary, &reopened, &old_configuration).await; + assert!(old_work_matches); + old_work_registration.expect("same-configuration Work registration remains admitted"); drop(reopened); // Reopen the sibling displaced by the primary. Every other route is @@ -418,6 +517,39 @@ async fn route_eviction_reopen_reads_retained_compiler_status_with_leased_siblin ) .await; assert!(compiler_failure(&after_configuration).contains("configured-missing.json")); + let (current_work_grant, current_work_matches, current_work_registration) = + register_work_for_configuration(&engine, &primary, &configured, ¤t_configuration) + .await; + assert_eq!( + old_work_grant, current_work_grant, + "telemetry changes preserve durable Work grant identity" + ); + assert!( + current_work_matches, + "reopen must already have refreshed the Work configuration authority" + ); + current_work_registration.expect("current Work configuration remains admitted"); + let (_, stale_work_matches, stale_work_registration) = + register_work_for_configuration(&engine, &primary, &configured, &old_configuration).await; + assert!(!stale_work_matches); + assert!( + stale_work_registration.is_err(), + "a delayed old configuration cannot replace current Work authority" + ); + let sibling_server = leased_siblings.first().unwrap(); + let (_, foreign_work_matches, foreign_work_registration) = + register_work_for_configuration(&engine, &primary, sibling_server, ¤t_configuration) + .await; + assert!(!foreign_work_matches); + assert!( + foreign_work_registration.is_err(), + "a linked sibling cannot replace the primary's Work authority" + ); + let (_, preserved_work_matches, preserved_work_registration) = + register_work_for_configuration(&engine, &primary, &configured, ¤t_configuration) + .await; + assert!(preserved_work_matches); + preserved_work_registration.expect("rejected candidates leave current Work authority intact"); assert!(Arc::ptr_eq( &retained_runtime.compiler_status().current().unwrap(), &refreshed.compiler_status().current().unwrap(), From 22182a83d808093552a40d58d883c98fdab1e331 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 26 Oct 2026 22:32:52 -0700 Subject: [PATCH 9/9] fix(work): fence retained configuration refreshes Hold the canonical configuration writer lane while validating and publishing a changed retained Work runtime. Refuse stale candidates without replacing the incumbent or waiting for a new owner. Exercise the real route journey with a newer durable configuration committed while refresh is paused, then admit the current revision. --- .../src/invocation.rs | 45 +++++ .../src/invocation/registrars.rs | 175 ++++++++++-------- .../src/project_runtime.rs | 33 ++++ .../src/configuration/store.rs | 25 +++ .../src/daemon/tests/compiler_status.rs | 90 +++++++++ 5 files changed, 293 insertions(+), 75 deletions(-) diff --git a/crates/tracedecay-daemon-service/src/invocation.rs b/crates/tracedecay-daemon-service/src/invocation.rs index 30f84ce214..6859dd4e6d 100644 --- a/crates/tracedecay-daemon-service/src/invocation.rs +++ b/crates/tracedecay-daemon-service/src/invocation.rs @@ -321,6 +321,13 @@ pub struct DaemonConfigurationRuntimeRegistrationPauseV1 { pub allow_return: tokio::sync::oneshot::Sender<()>, } +#[cfg(any(test, feature = "test-helpers"))] +struct WorkRuntimeRefreshPauseV1 { + project_root: PathBuf, + candidate_captured: tokio::sync::oneshot::Sender<()>, + allow_refresh: tokio::sync::oneshot::Receiver<()>, +} + #[derive(Clone)] pub struct DaemonInvocationService { code_index_schedulers: @@ -384,6 +391,8 @@ pub struct DaemonInvocationService { #[cfg(any(test, feature = "test-helpers"))] configuration_runtime_registration_pause: Arc>>, + #[cfg(any(test, feature = "test-helpers"))] + work_runtime_refresh_pause: Arc>>, /// Home of the daemon owner's user, where daemon-run automation deploys /// managed skills. `None` when the owning profile has no home. owner_home: Option, @@ -434,6 +443,8 @@ impl DaemonInvocationService { native_integration_status_broadcasts: Arc::new(Mutex::new(BTreeMap::new())), #[cfg(any(test, feature = "test-helpers"))] configuration_runtime_registration_pause: Arc::new(Mutex::new(None)), + #[cfg(any(test, feature = "test-helpers"))] + work_runtime_refresh_pause: Arc::new(Mutex::new(None)), owner_home: None, } } @@ -514,6 +525,40 @@ impl DaemonInvocationService { } } + #[cfg(any(test, feature = "test-helpers"))] + pub async fn pause_work_runtime_refresh( + &self, + project_root: PathBuf, + ) -> ( + tokio::sync::oneshot::Receiver<()>, + tokio::sync::oneshot::Sender<()>, + ) { + let (candidate_captured, candidate_observed) = tokio::sync::oneshot::channel(); + let (allow_refresh, refresh_allowed) = tokio::sync::oneshot::channel(); + *self.work_runtime_refresh_pause.lock().await = Some(WorkRuntimeRefreshPauseV1 { + project_root, + candidate_captured, + allow_refresh: refresh_allowed, + }); + (candidate_observed, allow_refresh) + } + + #[cfg(any(test, feature = "test-helpers"))] + async fn take_work_runtime_refresh_pause( + &self, + project_root: &Path, + ) -> Option { + let mut pause = self.work_runtime_refresh_pause.lock().await; + if pause + .as_ref() + .is_some_and(|pause| pause.project_root == project_root) + { + pause.take() + } else { + None + } + } + /// The one status broadcast shared by the native-integration invocation /// handler and every LSP session factory registered for `project_root`. pub async fn native_integration_status_broadcast( diff --git a/crates/tracedecay-daemon-service/src/invocation/registrars.rs b/crates/tracedecay-daemon-service/src/invocation/registrars.rs index 58a1de2114..62002a7ee2 100644 --- a/crates/tracedecay-daemon-service/src/invocation/registrars.rs +++ b/crates/tracedecay-daemon-service/src/invocation/registrars.rs @@ -4,7 +4,9 @@ use super::*; use tracedecay_application::feedback::ProductionFeedbackCyclePartsV1; use tracedecay_contracts::AnalyzerAdmittedDiagnosticProviderV1; use tracedecay_contracts::diagnostics::FeedbackDiagnosticProviderAdmissionV1; -use tracedecay_global_db::configuration::contracts::ports::ConfigurationOperationFuture; +use tracedecay_global_db::configuration::contracts::ports::{ + ConfigurationCurrentStateV1, ConfigurationOperationFuture, +}; use tracedecay_global_db::configuration::contracts::types::ConfigurationError; mod lsp; @@ -1087,17 +1089,7 @@ impl DaemonWorkRuntimeRegistrar { "Work configuration refresh requires its registered configuration authority" .to_owned(), })?; - let current = configuration - .runtime - .client() - .current_state() - .await - .map_err(|error| TraceDecayError::Config { - message: format!( - "Work configuration refresh could not read current state: {error}" - ), - })?; - Some((retained, current)) + Some((retained, configuration)) } else { None }; @@ -1109,73 +1101,106 @@ impl DaemonWorkRuntimeRegistrar { message: "registered Work runtime requires the mounted observability producer" .to_owned(), })?; + let reconcile = |registered: &mut RegisteredWorkRuntime, + current: Option<&ConfigurationCurrentStateV1>| { + // Store authority only. The evidence-retrieval adapter is + // built fresh by every route that mounts it, so comparing + // its object identity refused the second route of one + // project, a linked worktree, or a reopen of a route + // whose server was replaced, and left it permanently + // degraded. Its own project scope is already proven by + // `grant.scope` and the authority digest. + if registered.database.binding() == database.binding() + && registered.database.verified_locator() == database.verified_locator() + && registered.actor == actor + && registered.grant.digest == grant.digest + && registered.grant.scope == grant.scope + && registered.authority_digest == authority_digest + { + if !registered.configuration_matches( + &policy_digest, + &configuration_digest, + &work_topology_policy, + &proposal_routing, + ) { + // Currentness is checked under the configuration + // writer lane. The captured incumbent also prevents + // overwriting a newer registration. + let Some((captured, _)) = &refresh else { + return Err(TraceDecayError::Config { + message: "Work configuration changed during reopen".to_owned(), + }); + }; + if !registered.configuration_matches( + &captured.policy_digest, + &captured.configuration_digest, + &captured.work_topology_policy, + &captured.proposal_routing, + ) || !current.is_some_and(|current| { + proposal_routing.matches_current_configuration(current) + }) { + return Err(TraceDecayError::Config { + message: "Work configuration changed during reopen".to_owned(), + }); + } + } + let recovery = registered + .workflow_fan_out_recovery + .clone() + .ok_or_else(|| TraceDecayError::Config { + message: "Workflow fan-out recovery owner is not mounted".to_owned(), + })?; + registered.grant = grant.clone(); + registered.policy_digest = policy_digest.clone(); + registered.configuration_digest = configuration_digest.clone(); + registered.work_topology_policy = work_topology_policy.clone(); + registered.proposal_routing = proposal_routing.clone(); + registered.evidence_retrieval = evidence_retrieval.clone(); + recovery.refresh_admission(registered); + return Ok(()); + } + Err(TraceDecayError::Config { + message: "a different Work authority is already registered for this project" + .to_owned(), + }) + }; + if let Some((_, configuration)) = &refresh { + #[cfg(any(test, feature = "test-helpers"))] + if let Some(pause) = self + .service + .take_work_runtime_refresh_pause(&project_root) + .await + { + let _ = pause.candidate_captured.send(()); + let _ = pause.allow_refresh.await; + } + return configuration + .runtime + .configuration_store() + .with_current_state(|current| { + self.service + .project_runtimes + .reconcile_registered(&project_root, |registered| { + reconcile(registered, Some(current)) + })? + .ok_or_else(|| TraceDecayError::Config { + message: "Work configuration changed during reopen".to_owned(), + }) + }) + .await + .map_err(|error| TraceDecayError::Config { + message: format!( + "Work configuration refresh could not fence current state: {error}" + ), + })?; + } + // Initial admission and an unchanged retained pin keep their existing + // lifetime. Only a replacement must prove currentness at publication. self.service .project_runtimes .register_or_reconcile( project_root.clone(), - |registered: &mut RegisteredWorkRuntime| { - // Store authority only. The evidence-retrieval adapter is - // built fresh by every route that mounts it, so comparing - // its object identity refused the second route of one - // project, a linked worktree, or a reopen of a route - // whose server was replaced, and left it permanently - // degraded. Its own project scope is already proven by - // `grant.scope` and the authority digest. - if registered.database.binding() == database.binding() - && registered.database.verified_locator() == database.verified_locator() - && registered.actor == actor - && registered.grant.digest == grant.digest - && registered.grant.scope == grant.scope - && registered.authority_digest == authority_digest - { - if !registered.configuration_matches( - &policy_digest, - &configuration_digest, - &work_topology_policy, - &proposal_routing, - ) { - // The durable current-state read belongs to this - // captured incumbent. Check both inside the registry - // transition so a delayed reopen cannot overwrite a - // newer installed configuration. - let Some((captured, current)) = &refresh else { - return Err(TraceDecayError::Config { - message: "Work configuration changed during reopen".to_owned(), - }); - }; - if !registered.configuration_matches( - &captured.policy_digest, - &captured.configuration_digest, - &captured.work_topology_policy, - &captured.proposal_routing, - ) || !proposal_routing.matches_current_configuration(current) - { - return Err(TraceDecayError::Config { - message: "Work configuration changed during reopen".to_owned(), - }); - } - } - let recovery = registered.workflow_fan_out_recovery.clone().ok_or_else( - || TraceDecayError::Config { - message: "Workflow fan-out recovery owner is not mounted" - .to_owned(), - }, - )?; - registered.grant = grant.clone(); - registered.policy_digest = policy_digest.clone(); - registered.configuration_digest = configuration_digest.clone(); - registered.work_topology_policy = work_topology_policy.clone(); - registered.proposal_routing = proposal_routing.clone(); - registered.evidence_retrieval = evidence_retrieval.clone(); - recovery.refresh_admission(registered); - return Ok(()); - } - Err(TraceDecayError::Config { - message: - "a different Work authority is already registered for this project" - .to_owned(), - }) - }, + |registered| reconcile(registered, None), || async { let durable_write_signal = super::types::WorkDurableWriteSignalV1::default(); diff --git a/crates/tracedecay-daemon-service/src/project_runtime.rs b/crates/tracedecay-daemon-service/src/project_runtime.rs index f272aecd49..5e5900d8ae 100644 --- a/crates/tracedecay-daemon-service/src/project_runtime.rs +++ b/crates/tracedecay-daemon-service/src/project_runtime.rs @@ -1192,6 +1192,39 @@ impl ProjectRuntimeRegistryV1 { .then_some(result) } + /// Reconcile an existing component without waiting for a reservation or + /// constructing a new owner. Callers can hold an external authority fence + /// through this synchronous transition without awaiting another writer. + pub(crate) fn reconcile_registered( + &self, + project_root: &Path, + reconcile: R, + ) -> Result, E> + where + C: ProjectRuntimeComponent, + E: From, + R: FnOnce(&mut C) -> Result, + { + let root_fences = self.lock_root_fences(); + let mut runtimes = self.lock_runtimes(); + if self.closed.load(Ordering::Acquire) || root_fences.contains(project_root) { + return Err(ProjectRuntimeRegistryError::Closed.into()); + } + let Some(runtime) = runtimes.get_mut(project_root) else { + return Ok(None); + }; + let type_id = TypeId::of::(); + if runtime.reservations.contains(&type_id) + || runtime.registration_builds.contains_key(&type_id) + { + return Ok(None); + } + let Some(incumbent) = C::slot(runtime).as_mut() else { + return Ok(None); + }; + reconcile(incumbent).map(Some) + } + /// Register a component, or accept an incumbent the caller recognizes as /// the same authority. /// diff --git a/crates/tracedecay-global-db/src/configuration/store.rs b/crates/tracedecay-global-db/src/configuration/store.rs index 392aa6de88..bf07c3fb66 100644 --- a/crates/tracedecay-global-db/src/configuration/store.rs +++ b/crates/tracedecay-global-db/src/configuration/store.rs @@ -908,6 +908,31 @@ impl OwnedGlobalDbConfigurationControlStore { self.db.clone() } + /// Publish a synchronous runtime projection while configuration commits + /// are excluded by this store's canonical writer lane. The callback must + /// not wait for another database operation or build a new runtime owner. + /// + /// This does not write configuration. Dropping the immediate transaction + /// rolls back before releasing its lane, with no await after publication. + pub fn with_current_state<'a, T, F>(&'a self, publish: F) -> ConfigurationOperationFuture<'a, T> + where + T: Send + 'a, + F: FnOnce(&ConfigurationCurrentStateV1) -> T + Send + 'a, + { + let db = self.database(); + Box::pin(async move { + let transaction = db + .begin_write_transaction() + .await + .map_err(|_| ConfigurationError::Unavailable)?; + let current = current_state_from_transaction(&transaction).await?; + Self::require_active_mutation_scope(db.as_ref())?; + let result = publish(¤t); + drop(transaction); + Ok(result) + }) + } + /// Revalidate mutation access against the capability's client-bound write /// authority, so no path-derived authority can be substituted here. fn require_active_mutation_scope(db: &RegisteredGlobalDb) -> Result<(), ConfigurationError> { diff --git a/crates/tracedecay/src/daemon/tests/compiler_status.rs b/crates/tracedecay/src/daemon/tests/compiler_status.rs index 180141b569..694c1b84a9 100644 --- a/crates/tracedecay/src/daemon/tests/compiler_status.rs +++ b/crates/tracedecay/src/daemon/tests/compiler_status.rs @@ -550,6 +550,96 @@ async fn route_eviction_reopen_reads_retained_compiler_status_with_leased_siblin .await; assert!(preserved_work_matches); preserved_work_registration.expect("rejected candidates leave current Work authority intact"); + + // Retain A while a route captures B, then commit C before the route can + // publish. The configuration writer fence must refuse B without changing A. + drop( + apply_project_setting_via_surface(&engine, sibling_handshake.as_ref().unwrap(), |_| { + ( + tracedecay_domain::configuration::SettingKey::new( + tracedecay_domain::configuration::TELEMETRY_TIMINGS_SETTING_KEY, + ) + .unwrap(), + tracedecay_domain::configuration::ConfigurationValueV1::Boolean(true), + ) + }) + .await, + ); + let candidate_configuration = configured + .cg() + .await + .configuration_runtime() + .client() + .current() + .await + .unwrap(); + assert_ne!( + current_configuration.revision_id(), + candidate_configuration.revision_id() + ); + let (candidate_captured, allow_refresh) = engine + .invocation + .service + .pause_work_runtime_refresh(primary.clone()) + .await; + let ((_, raced_matches, raced_registration), latest_configuration) = tokio::join!( + register_work_for_configuration(&engine, &primary, &configured, &candidate_configuration), + async { + candidate_captured.await.expect("Work candidate captured"); + drop( + apply_project_setting_via_surface( + &engine, + sibling_handshake.as_ref().unwrap(), + |_| { + ( + tracedecay_domain::configuration::SettingKey::new( + tracedecay_domain::configuration::TELEMETRY_TIMINGS_SETTING_KEY, + ) + .unwrap(), + tracedecay_domain::configuration::ConfigurationValueV1::Boolean(false), + ) + }, + ) + .await, + ); + let latest = configured + .cg() + .await + .configuration_runtime() + .client() + .current() + .await + .unwrap(); + assert_ne!(candidate_configuration.revision_id(), latest.revision_id()); + allow_refresh.send(()).expect("resume Work refresh"); + latest + }, + ); + assert!(!raced_matches); + assert!( + matches!(raced_registration, Err(TraceDecayError::Config { ref message }) + if message == "Work configuration changed during reopen"), + "a durable commit before publication must refuse the superseded candidate: {raced_registration:?}" + ); + let (_, unchanged_work_matches, unchanged_work_registration) = + register_work_for_configuration(&engine, &primary, &configured, ¤t_configuration) + .await; + assert!( + unchanged_work_matches, + "refusing B must leave the complete previously installed A authority intact" + ); + unchanged_work_registration.expect("same-pin admission remains valid after a later commit"); + let (_, latest_work_matches, latest_work_registration) = + register_work_for_configuration(&engine, &primary, &configured, &latest_configuration) + .await; + assert!(!latest_work_matches); + latest_work_registration.expect("the durable current C authority can be installed"); + let (_, installed_work_matches, installed_work_registration) = + register_work_for_configuration(&engine, &primary, &configured, &latest_configuration) + .await; + assert!(installed_work_matches); + installed_work_registration.expect("C remains admitted after publication"); + assert!(Arc::ptr_eq( &retained_runtime.compiler_status().current().unwrap(), &refreshed.compiler_status().current().unwrap(),