diff --git a/crates/tracedecay-application/src/lsp_runtime.rs b/crates/tracedecay-application/src/lsp_runtime.rs index ee018fbc824..9f7f0dffca3 100644 --- a/crates/tracedecay-application/src/lsp_runtime.rs +++ b/crates/tracedecay-application/src/lsp_runtime.rs @@ -349,6 +349,7 @@ pub(crate) struct StoredLspTestRunExpansionV1 { pub fn lsp_session_factory( feedback_runtime: Arc, test_run_store: RegisteredGlobalDbLeaseV1, + operation_events: crate::operation_stream::OperationEventAuthority, code_index: Arc, workspace_index: Arc, diagnostic_records: Arc, @@ -368,7 +369,7 @@ where code_index, workspace_index, )?); - let test_runs = lsp_test_result_port(project.clone(), test_run_store); + let test_runs = lsp_test_result_port(project.clone(), test_run_store, operation_events); let diagnostic_projection = Arc::new(DiagnosticsStoreLspFeedbackProjection::new( diagnostic_records, project.clone(), diff --git a/crates/tracedecay-application/src/lsp_runtime/managed_test_runs.rs b/crates/tracedecay-application/src/lsp_runtime/managed_test_runs.rs index d9dd1da6d55..c89c4f456dd 100644 --- a/crates/tracedecay-application/src/lsp_runtime/managed_test_runs.rs +++ b/crates/tracedecay-application/src/lsp_runtime/managed_test_runs.rs @@ -25,7 +25,7 @@ use crate::managed_test_runs::{ CanonicalManagedTestRunReader, ManagedTestRunCurrentScope, ManagedTestRunReadOutcome, ManagedTestRunStaleReason, ManagedTestRunUnavailableReason, }; -use crate::operation_stream::{ManagedTestRunActivity, operation_event_authority}; +use crate::operation_stream::{ManagedTestRunActivity, OperationEventAuthority}; /// Real test execution result projection owner. Feedback impact owns affected /// test identities; execution results remain in their separate canonical @@ -171,9 +171,10 @@ impl ManagedTestRunProjection { pub(crate) fn lsp_test_result_port( project: Arc, store: RegisteredGlobalDbLeaseV1, + operation_events: OperationEventAuthority, ) -> Arc { Arc::new(ManagedTestRunProjection::new( - CanonicalManagedTestRunReader::new(store, operation_event_authority()), + CanonicalManagedTestRunReader::new(store, operation_events), project, )) } diff --git a/crates/tracedecay-application/src/operation_stream.rs b/crates/tracedecay-application/src/operation_stream.rs index 9e77ca4cf82..4d22384926f 100644 --- a/crates/tracedecay-application/src/operation_stream.rs +++ b/crates/tracedecay-application/src/operation_stream.rs @@ -7,8 +7,8 @@ use std::collections::{BTreeMap, BTreeSet, VecDeque}; use std::fmt; use std::pin::Pin; +use std::sync::Arc; use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; -use std::sync::{Arc, OnceLock}; use std::task::{Context, Poll}; use schemars::JsonSchema; @@ -128,14 +128,6 @@ pub enum OperationEventItem { pub type OperationEvent = StreamEvent; -static OPERATION_EVENTS: OnceLock = OnceLock::new(); - -pub fn operation_event_authority() -> OperationEventAuthority { - OPERATION_EVENTS - .get_or_init(OperationEventAuthority::default) - .clone() -} - #[derive(Clone, Debug, PartialEq, Eq)] pub struct OperationBinding { operation_id: OperationId, @@ -842,12 +834,9 @@ impl OperationEventAuthority { /// `FrontierExpired` rather than a fabricated snapshot. /// /// A record whose `OperationEmitter` is still alive is kept. The - /// process-global authority is shared by every daemon composition in the - /// process and by the project workflow handlers that begin managed test - /// runs outside any composition, so one composition's shutdown must not - /// truncate a stream another producer is between admission and its first - /// result on. The emitter is the only holder of the cancellation - /// receiver, so its receiver count is the producer liveness signal. + /// owning daemon must leave accepted, unsettled producers able to publish + /// their terminal receipt. The emitter's cancellation receiver remains + /// the local producer-liveness signal. #[tracing::instrument(name = "usecases.operation.expire_idle", level = "trace", skip_all)] pub async fn expire_idle(&self) { let mut state = self.inner.state.lock().await; diff --git a/crates/tracedecay-daemon-service/src/invocation.rs b/crates/tracedecay-daemon-service/src/invocation.rs index 46f9e9f82ae..ff5daa9300f 100644 --- a/crates/tracedecay-daemon-service/src/invocation.rs +++ b/crates/tracedecay-daemon-service/src/invocation.rs @@ -132,7 +132,7 @@ use tracedecay_application::lsp_runtime::{ production_semantic_authorities, }; use tracedecay_application::operation_stream::{ - OperationEmitter, OperationEventAuthority, OperationKind, operation_event_authority, + OperationEmitter, OperationEventAuthority, OperationKind, }; use tracedecay_application::primitives::PrimitiveProjectRuntime; use tracedecay_contracts::feedback::observations::{ @@ -238,8 +238,7 @@ pub use feedback::{ DaemonAdvisoryCycleMountChangedFuture, DaemonAdvisoryCycleMountFuture, DaemonAdvisoryCycleMountV1, DaemonFeedbackInvocationOwner, DaemonFeedbackProximityInvocationFuture, DaemonFeedbackProximityInvocationRequest, - advisory_cycle_invocation_result, daemon_operation_event_authority, - feedback_proximity_invocation_result, + advisory_cycle_invocation_result, feedback_proximity_invocation_result, }; pub use primitive::{ DaemonContextScoutRuntimeRegistrar, DaemonContextScoutRuntimeRegistrationError, @@ -431,7 +430,7 @@ impl DaemonInvocationService { request_cancellations: crate::request_cancellation::RequestCancellationRegistryV1::default(), store_observability: StoreObservabilityRegistryV1::default(), - operation_events: daemon_operation_event_authority(), + operation_events: OperationEventAuthority::default(), github_stack_coordinator: Arc::new( tracedecay_application::stack_coordinator::DaemonGitHubStackCoordinatorV1::default( ), diff --git a/crates/tracedecay-daemon-service/src/invocation/feedback.rs b/crates/tracedecay-daemon-service/src/invocation/feedback.rs index 4d445b0c373..8a425f5e20a 100644 --- a/crates/tracedecay-daemon-service/src/invocation/feedback.rs +++ b/crates/tracedecay-daemon-service/src/invocation/feedback.rs @@ -8,10 +8,6 @@ use tracedecay_contracts::RequestContext; use tracedecay_contracts::feedback::FeedbackPortFuture; use tracedecay_domain::CommitId; -pub fn daemon_operation_event_authority() -> OperationEventAuthority { - operation_event_authority() -} - pub(crate) struct DaemonFeedbackInvocationRequest { pub(crate) operation: DaemonInvocationOperation, pub(crate) request_handle: String, diff --git a/crates/tracedecay-daemon-service/src/invocation/lsp.rs b/crates/tracedecay-daemon-service/src/invocation/lsp.rs index 7c19fb938f7..d85a56c5529 100644 --- a/crates/tracedecay-daemon-service/src/invocation/lsp.rs +++ b/crates/tracedecay-daemon-service/src/invocation/lsp.rs @@ -435,8 +435,8 @@ impl DaemonInvocationService { self.store_observability.join_retirement_drains().await; step("observability_retirements_joined"); self.session_holder_databases.lock().await.clear(); - // The operation-event authority is process-global, not owned by this - // composition: only frontiers without a live producer expire here. + // This daemon owns its frontiers; an accepted producer may retain + // its terminal-publication window after other owners have drained. self.operation_events.expire_idle().await; step("idle_operation_events_expired"); let lease_shutdown_clean = lease_shutdown.is_ok(); diff --git a/crates/tracedecay-daemon-service/src/invocation/registrars/lsp.rs b/crates/tracedecay-daemon-service/src/invocation/registrars/lsp.rs index 27b47d49f2e..5d79b5c5f84 100644 --- a/crates/tracedecay-daemon-service/src/invocation/registrars/lsp.rs +++ b/crates/tracedecay-daemon-service/src/invocation/registrars/lsp.rs @@ -123,6 +123,7 @@ impl DaemonLspOwnerRegistrar { lsp_session_factory( feedback_runtime, registered_database.clone(), + self.service.operation_events(), code_index.clone() as Arc, workspace_index, diagnostic_records, diff --git a/crates/tracedecay-daemon-service/src/invocation/tests/mod.rs b/crates/tracedecay-daemon-service/src/invocation/tests/mod.rs index 0ff2002dd2c..16e29df1bb2 100644 --- a/crates/tracedecay-daemon-service/src/invocation/tests/mod.rs +++ b/crates/tracedecay-daemon-service/src/invocation/tests/mod.rs @@ -135,3 +135,83 @@ mod project_lifecycle_tests; mod source_edit_tests; mod work_routing_fixtures; mod work_tests; + +#[tokio::test] +async fn independent_daemon_compositions_have_separate_managed_test_frontiers() { + let first = DaemonInvocationService::default(); + let second = DaemonInvocationService::default(); + let request = tracedecay_contracts::RequestId::new("request.test.independent-daemons") + .expect("request id"); + let deadline = tracedecay_contracts::Deadline::new(tracedecay_domain::UtcMicros(10_000)) + .expect("deadline"); + let _first_run = first + .operation_events() + .begin_managed_test_run( + "file:///composition-first/".to_owned(), + request.clone(), + deadline.clone(), + ) + .await + .expect("first composition admits its operation"); + let second_run = second + .operation_events() + .begin_managed_test_run("file:///composition-second/".to_owned(), request, deadline) + .await + .expect("second composition admits the same request id"); + first.operation_events().expire_idle().await; + second_run + .test_result("suite::after-first-shutdown".to_owned(), true) + .await + .expect("the second owner's accepted producer can still publish"); +} + +#[tokio::test] +async fn cancelled_managed_test_in_one_daemon_cannot_cancel_another() { + let first = DaemonInvocationService::default(); + let second = DaemonInvocationService::default(); + // Same project and request identity on purpose: only the daemon-owned + // authority separates these concurrent generations. A process-global + // event table would share the first cancellation with the second run. + let request = tracedecay_contracts::RequestId::new("request.test.isolated-cancellation") + .expect("request id"); + let deadline = tracedecay_contracts::Deadline::new(tracedecay_domain::UtcMicros(10_000)) + .expect("deadline"); + let root = "file:///shared-composition/"; + let first_run = first + .operation_events() + .begin_managed_test_run(root.to_owned(), request.clone(), deadline.clone()) + .await + .expect("first daemon admits the operation"); + let second_run = second + .operation_events() + .begin_managed_test_run(root.to_owned(), request, deadline) + .await + .expect("second daemon independently admits the same operation"); + assert_eq!( + first_run + .request_managed_test_cancellation() + .await + .expect("first daemon accepts cancellation"), + tracedecay_application::operation_stream::OperationCancelOutcome::Requested + ); + assert!(first_run.is_cancelled()); + assert!( + !second_run.is_cancelled(), + "cancelling the first daemon's operation must not cancel the second" + ); + + // Retire the first daemon's expired frontier. The other daemon's active + // stream must still publish and accept its own independent cancellation. + first.operation_events().expire_idle().await; + second_run + .test_result("suite::independent".to_owned(), true) + .await + .expect("first daemon retirement must not affect the second stream"); + assert_eq!( + second_run + .request_managed_test_cancellation() + .await + .expect("second daemon cancellation is independent"), + tracedecay_application::operation_stream::OperationCancelOutcome::Requested + ); +} diff --git a/crates/tracedecay-daemon-service/src/lib.rs b/crates/tracedecay-daemon-service/src/lib.rs index 98593b3f712..8cd01302330 100644 --- a/crates/tracedecay-daemon-service/src/lib.rs +++ b/crates/tracedecay-daemon-service/src/lib.rs @@ -113,7 +113,7 @@ pub use invocation::{ RegisteredRetainedRequestContextError, RegisteredRetainedRuntime, RegisteredWorkRuntime, SwitchableFeedbackCycleRuntimeV1, UnavailableFeedbackCycleRuntimeV1, admit_registered_hook_orchestration, advisory_cycle_invocation_result, - callable_code_request_context, daemon_operation_event_authority, execute_profile_configuration, + callable_code_request_context, execute_profile_configuration, feedback_proximity_invocation_result, register_hook_orchestration_runtime, settle_daemon_work_delivery, unregister_hook_orchestration_runtime, }; diff --git a/crates/tracedecay-mcp/src/handlers/workflow.rs b/crates/tracedecay-mcp/src/handlers/workflow.rs index 0f2a7a4dc5a..58e4abf3316 100644 --- a/crates/tracedecay-mcp/src/handlers/workflow.rs +++ b/crates/tracedecay-mcp/src/handlers/workflow.rs @@ -35,7 +35,7 @@ use tracedecay_application::diagnose::{Severity, parse_cargo_output}; use tracedecay_application::diagnostics_publication::CodeIndexPublicationIdentityPortV1; use tracedecay_application::diagnostics_store::DiagnosticsStore; use tracedecay_application::operation_stream::{ - OperationEmitter, OperationEventError, operation_event_authority, + OperationEmitter, OperationEventAuthority, OperationEventError, }; use tracedecay_code_index::is_test_file; use tracedecay_contracts::feedback::TestResultProjectionV1; @@ -461,6 +461,7 @@ fn severity_wire(s: Severity) -> DiagnoseSeverityV1 { /// the requesting session the run is attributed to when the request named one. pub struct ManagedTestRunRecording { pub store: RegisteredGlobalDbLeaseV1, + pub events: OperationEventAuthority, pub session_id: Option, } @@ -719,7 +720,9 @@ impl ManagedTestRun { else { return Ok(()); }; - publish_test_run_terminal( + // Once recording is refused, preserve the store failure even when a + // concurrent lifecycle shutdown also prevents terminal publication. + if let Err(publish_error) = publish_test_run_terminal( &self.emitter, test_run_receipt( self.started_at, @@ -728,15 +731,21 @@ impl ManagedTestRun { 0, ), ) - .await?; + .await + { + tracing::warn!( + error = %publish_error, + "could not publish failed managed test-run start receipt" + ); + } Err(test_run_record_error(error)) } /// Settles the durable record with the run's exit status and outcome /// counts, then publishes the terminal receipt. Readers rely on that /// order: a run whose live stream terminated has its outcome recorded - /// unless recording it failed. The receipt is published either way, so - /// no subscriber waits on a run that already ended. + /// unless recording it failed. A refused durable write publishes a failed + /// live terminal, never an unverifiable success, so subscribers stop waiting. #[tracing::instrument(name = "mcp.workflow.affected_tests.finish", level = "trace", skip_all)] pub(super) async fn finish( &self, @@ -766,8 +775,32 @@ impl ManagedTestRun { .record_managed_test_run_outcome(&self.operation_id(), &outcome) .await .map_err(test_run_record_error); - publish_test_run_terminal(&self.emitter, receipt.clone()).await?; + // A completed test process is not a completed durable run if its + // outcome could not be recorded. Close the live stream with failure, + // rather than publishing success that no reader can verify. + let stream_receipt = if recorded.is_ok() { + receipt.clone() + } else { + test_run_receipt( + self.started_at, + &self.deadline, + OperationTermination::Failed, + bytes_consumed, + ) + }; + // A refused stream publication cannot obscure a durable-store failure. + // When the durable write succeeds, publication errors still propagate. + let published = publish_test_run_terminal(&self.emitter, stream_receipt).await; + if let (Err(record_error), Err(publish_error)) = (&recorded, &published) { + tracing::warn!( + operation_id = %self.operation_id(), + storage_error = %record_error, + publication_error = %publish_error, + "managed test-run outcome and failed terminal publication both refused" + ); + } recorded?; + published?; Ok(receipt) } @@ -807,7 +840,9 @@ async fn begin_test_run( })?; let document_content_digests = managed_test_document_content_digests(&root, changed_paths).await?; - let emitter = operation_event_authority() + let emitter = admission + .recording + .events .begin_managed_test_run(root_uri.clone(), request_id, admission.deadline.clone()) .await .map_err(|error| test_run_event_error(&error))?; diff --git a/crates/tracedecay-mcp/src/handlers/workflow/affected_tests_tests.rs b/crates/tracedecay-mcp/src/handlers/workflow/affected_tests_tests.rs index b3f6c1e554e..aedaa9f606e 100644 --- a/crates/tracedecay-mcp/src/handlers/workflow/affected_tests_tests.rs +++ b/crates/tracedecay-mcp/src/handlers/workflow/affected_tests_tests.rs @@ -10,11 +10,14 @@ use std::future::ready; use std::sync::Arc; use sha2::{Digest, Sha256}; +use tracedecay_application::operation_stream::{ + OperationEventAuthority, OperationId, OperationKind, +}; use tracedecay_code_index::graph_projection::HermeticCodeGraphProjectionStore; use tracedecay_code_index::lineage::{GenerationSymbolIndexV1, LineageSymbolRecordV1}; use tracedecay_contracts::{ CancellationSignal, CapabilityGrantId, CapabilityGrantSnapshot, DisclosureClass, - RequestContext, RequestId, ResolvedScope, + RequestContext, RequestId, ResolvedScope, StreamEventKind, }; use tracedecay_domain::{ ActorId, BoundedSanitizedText, CanonicalRelationEdgeV1, ChunkerRevision, CodeGenerationId, @@ -48,6 +51,7 @@ where .await; let recording = ManagedTestRunRecording { store: sessions.registered.clone(), + events: OperationEventAuthority::default(), session_id: None, }; let completion = @@ -1151,7 +1155,7 @@ fn scoped_file_symbol_page_needs_only_a_file_scale_budget() { #[tokio::test] async fn request_cancellation_cancels_the_waiting_managed_run() { let root = tempfile::tempdir().unwrap(); - let emitter = operation_event_authority() + let emitter = OperationEventAuthority::default() .begin_managed_test_run( Url::from_directory_path(root.path()).unwrap().to_string(), mint_global_request_id(GlobalRequestSurface::ManagedTestRun).unwrap(), @@ -1208,3 +1212,163 @@ fn assert_failed_terminal(output: &Value) { assert_eq!(output["terminal"]["receipt"]["termination"], "failed"); assert!(output["note"].is_null()); } + +#[tokio::test] +async fn refused_durable_test_result_cannot_publish_a_completed_live_terminal() { + let cancellation = CancellationSignal::active("cancellation.affected-tests.settlement-refusal") + .expect("fixture cancellation"); + let context = verified_graph_context(&cancellation); + let authority = OperationEventAuthority::default(); + let emitter = authority + .begin(&context, OperationKind::TestRun, UtcMicros(1)) + .await + .expect("admit an authenticated operation"); + let subscription = authority + .subscribe( + &OperationId::from_request(context.request_id().clone()), + &context, + UtcMicros(1), + 0, + None, + ) + .await + .expect("subscribe before settlement"); + let (_, _, mut stream) = subscription.into_sse_parts(); + let accepted = tokio::time::timeout(std::time::Duration::from_secs(2), stream.next()) + .await + .expect("initial event must arrive") + .expect("accepted event"); + assert!(matches!(accepted.kind, StreamEventKind::Item(_))); + + let store = tracedecay_global_db::tests::harness::RegisteredGlobalDbHarness::open( + "affected-tests-unrecorded-terminal", + ) + .await; + // No durable start row exists. An outcome UPDATE must affect zero rows; + // the live stream must not tell an observer the run completed successfully. + let managed = ManagedTestRun { + emitter, + recording: ManagedTestRunRecording { + store: store.registered.clone(), + events: authority.clone(), + session_id: None, + }, + started_at: UtcMicros(10), + deadline: Deadline::new(UtcMicros(i64::MAX)).expect("fixture deadline"), + }; + let error = managed + .finish( + OperationTermination::Completed, + 7, + Some(0), + &LibtestReport::default(), + ) + .await + .expect_err("unrecorded outcome cannot settle durably"); + assert!( + matches!(&error, TraceDecayError::Database { .. }), + "durable-store refusal must remain typed: {error}" + ); + let terminal = tokio::time::timeout(std::time::Duration::from_secs(2), stream.next()) + .await + .expect("a refused record must still settle the live stream") + .expect("terminal event"); + assert!(matches!( + terminal.kind, + StreamEventKind::Terminal(termination) + if termination.termination == OperationTermination::Failed + && termination.receipt.termination == OperationTermination::Failed + && termination.receipt.budget.bytes_consumed == 7 + )); + assert!( + stream.next().await.is_none(), + "live stream ends after refusal" + ); +} + +#[tokio::test] +async fn durable_store_refusal_remains_primary_when_terminal_publication_also_fails() { + let cancellation = + CancellationSignal::active("cancellation.affected-tests.concurrent-terminal") + .expect("fixture cancellation"); + let context = verified_graph_context(&cancellation); + let authority = OperationEventAuthority::default(); + let emitter = authority + .begin(&context, OperationKind::TestRun, UtcMicros(1)) + .await + .expect("admit an authenticated operation"); + let subscription = authority + .subscribe( + &OperationId::from_request(context.request_id().clone()), + &context, + UtcMicros(1), + 0, + None, + ) + .await + .expect("observe the original operation terminal"); + let (_, _, mut stream) = subscription.into_sse_parts(); + let deadline = Deadline::new(UtcMicros(i64::MAX)).expect("fixture deadline"); + // Deliberately pre-settle this event owner so the subsequent Failed + // publication is refused with TerminalAlreadyPublished. + emitter + .terminal(test_run_receipt( + UtcMicros(10), + &deadline, + OperationTermination::Completed, + 0, + )) + .await + .expect("pre-settle lifecycle"); + + let store = tracedecay_global_db::tests::harness::RegisteredGlobalDbHarness::open( + "affected-tests-dual-settlement-refusal", + ) + .await; + let run = ManagedTestRun { + emitter, + recording: ManagedTestRunRecording { + store: store.registered.clone(), + events: authority.clone(), + session_id: None, + }, + started_at: UtcMicros(10), + deadline, + }; + // No start record exists, so the durable write is also refused. + // Keep that primary failure even when event publication independently + // rejects a second, nonidentical terminal. + let error = run + .finish( + OperationTermination::Completed, + 7, + Some(0), + &LibtestReport::default(), + ) + .await + .expect_err("no successful durability should be reported"); + assert!( + matches!(&error, TraceDecayError::Database { .. }), + "the storage refusal must take precedence: {error}" + ); + let accepted = tokio::time::timeout(std::time::Duration::from_secs(2), stream.next()) + .await + .expect("original accepted event must be observable") + .expect("accepted event"); + assert!(matches!(accepted.kind, StreamEventKind::Item(_))); + let terminal = tokio::time::timeout(std::time::Duration::from_secs(2), stream.next()) + .await + .expect("original terminal must remain observable") + .expect("original terminal event"); + assert!(matches!( + terminal.kind, + StreamEventKind::Terminal(termination) + if termination.termination == OperationTermination::Completed + && termination.receipt.termination == OperationTermination::Completed + && termination.receipt.budget.bytes_consumed == 0 + )); + assert!( + stream.next().await.is_none(), + "refused later publication must not insert a second terminal" + ); +} diff --git a/crates/tracedecay/src/daemon/http_application_router.rs b/crates/tracedecay/src/daemon/http_application_router.rs index a78536d6331..08bb14e2d99 100644 --- a/crates/tracedecay/src/daemon/http_application_router.rs +++ b/crates/tracedecay/src/daemon/http_application_router.rs @@ -23,6 +23,7 @@ async fn build_http_application_router( owner: &tracedecay_runtime_core::config::ProfileRoot, project_id: &str, project_path: &Path, + operation_events: tracedecay_application::operation_stream::OperationEventAuthority, ) -> Result { let project_id = tracedecay_domain::ProjectId::new(project_id.to_owned()).map_err(|error| { TraceDecayError::Config { @@ -45,7 +46,7 @@ async fn build_http_application_router( let client = crate::daemon::core_client::invocation_client_for_current(&owner, handshake)?; tracedecay_daemon_service::application_surface::http_application_router( client, - tracedecay_daemon_service::daemon_operation_event_authority(), + operation_events, project_id, ) .map_err(|error| TraceDecayError::Config { @@ -64,6 +65,7 @@ pub(super) fn install_http_application_cold_resolver( invocation: DaemonInvocationState, project_open_gates: Arc>, ) -> Result<()> { + let operation_events = invocation.service.operation_events(); registry.install_remote_deletion_runtime_owners( super::remote_deletion::RemoteDeletionRuntimeOwners { administration: store_administration.clone(), @@ -73,6 +75,7 @@ pub(super) fn install_http_application_cold_resolver( )?; registry.install_resolver(move |project_id| { let store_administration = store_administration.clone(); + let operation_events = operation_events.clone(); tracing::Instrument::instrument( async move { let database = store_administration.registered_profile_database().await?; @@ -121,6 +124,7 @@ pub(super) fn install_http_application_cold_resolver( store_administration.owner_profile()?, project_id.as_str(), &canonical_root, + operation_events, ) .await .map(Some) @@ -177,11 +181,13 @@ pub(super) async fn mount_http_application_router( owner: &tracedecay_runtime_core::config::ProfileRoot, project_id: &str, project_path: &Path, + operation_events: tracedecay_application::operation_stream::OperationEventAuthority, ) -> Result<()> { if !registry.is_active() { return Ok(()); } - let router = build_http_application_router(owner, project_id, project_path).await?; + let router = + build_http_application_router(owner, project_id, project_path, operation_events).await?; registry.mount(project_id, router).await } @@ -231,6 +237,7 @@ mod tests { &owner, project_id, project.path(), + tracedecay_application::operation_stream::OperationEventAuthority::default(), )); let mut yielded = false; let peer = std::future::poll_fn(|task| { diff --git a/crates/tracedecay/src/daemon/production_harness.rs b/crates/tracedecay/src/daemon/production_harness.rs index bc20b8dd552..f3b10a0555a 100644 --- a/crates/tracedecay/src/daemon/production_harness.rs +++ b/crates/tracedecay/src/daemon/production_harness.rs @@ -868,6 +868,18 @@ impl ProductionProjectCompositionHarnessV1 { }) } + /// The real stream owner of this isolated daemon composition, for lifecycle tests. + pub fn operation_events( + &self, + ) -> Result { + self.resources + .as_ref() + .map(|resources| resources.invocation.service.operation_events()) + .ok_or_else(|| TraceDecayError::Config { + message: "production composition has already shut down".to_owned(), + }) + } + pub fn server(&self, project_root: impl AsRef) -> Result> { let canonical_project_path = tracedecay_runtime_core::path_safety::canonical_existing_identity( diff --git a/crates/tracedecay/src/daemon/production_harness/configuration_idempotency_journey_test.rs b/crates/tracedecay/src/daemon/production_harness/configuration_idempotency_journey_test.rs index ddb7dfde6d2..45172f1d176 100644 --- a/crates/tracedecay/src/daemon/production_harness/configuration_idempotency_journey_test.rs +++ b/crates/tracedecay/src/daemon/production_harness/configuration_idempotency_journey_test.rs @@ -20,7 +20,6 @@ use tracedecay_sdk::operations::{ApplicationConfigurationSet, TypedOperation}; use tracedecay_tool_catalog::ApplicationSurfaceOperation; use super::journey_test_support::{initialize_project, tool_payload}; -use tracedecay_daemon_service::daemon_operation_event_authority; const HTTP_AUTH_TOKEN: &str = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; @@ -294,7 +293,7 @@ async fn configuration_http_sdk( let router = tracedecay_daemon_service::application_surface::http_application_router_with_executor( executor, - daemon_operation_event_authority(), + resources.invocation.service.operation_events(), target.project_id.clone(), ) .expect("canonical HTTP application router"); diff --git a/crates/tracedecay/src/daemon/project_composition.rs b/crates/tracedecay/src/daemon/project_composition.rs index ec7cc332d7e..ca8160c2574 100644 --- a/crates/tracedecay/src/daemon/project_composition.rs +++ b/crates/tracedecay/src/daemon/project_composition.rs @@ -1678,6 +1678,7 @@ impl ProjectOpenInputs<'_> { http_owner, &core.project_id, self.canonical_project_path, + self.invocation.service.operation_events(), ) .await?; self.log_phase("http_application_mounted", None, full_setup_started); diff --git a/crates/tracedecay/src/daemon/project_open_owners/advisory_runtime.rs b/crates/tracedecay/src/daemon/project_open_owners/advisory_runtime.rs index bd756c6db29..60cd5a21542 100644 --- a/crates/tracedecay/src/daemon/project_open_owners/advisory_runtime.rs +++ b/crates/tracedecay/src/daemon/project_open_owners/advisory_runtime.rs @@ -113,10 +113,9 @@ use tracedecay_daemon_service::{ DaemonAdvisoryCycleMountFuture, DaemonAdvisoryCycleMountV1, DaemonFeedbackProximityInvocationFuture, DaemonFeedbackProximityInvocationRequest, HookOrchestrationRequestV1, HookOrchestrationTriggerV1, HookOrchestrationWorkOutcomeV1, - advisory_cycle_invocation_result, daemon_operation_event_authority, - daemon_owned_project_source_access_at, feedback_proximity_invocation_result, - project_open_source_access_authority, register_hook_orchestration_runtime, - unregister_hook_orchestration_runtime, + advisory_cycle_invocation_result, daemon_owned_project_source_access_at, + feedback_proximity_invocation_result, project_open_source_access_authority, + register_hook_orchestration_runtime, unregister_hook_orchestration_runtime, }; use tracedecay_domain::errors::{Result, TraceDecayError}; use tracedecay_mcp::handlers::hook_runtime::daemon_mint_hook_v2_file_id; @@ -135,6 +134,7 @@ mod tests; #[derive(Clone)] struct ProjectOpenAdvisoryFeedbackCycleV1 { producer: Arc, + operation_events: tracedecay_application::operation_stream::OperationEventAuthority, root_uri: String, feedback_scope: FeedbackScopeV1, github_review_read: Option<(GitHubPullRequestIdV1, GitHubReviewReadOperationV1)>, @@ -277,7 +277,8 @@ impl ProjectOpenAdvisoryFeedbackCycleV1 { None => ProductionCiFailureDiscoveryOutcomeV1::NotConfigured, }; let expires_at = UtcMicros(observed_at.0.saturating_add(5 * 60 * 1_000_000)); - let operation = daemon_operation_event_authority() + let operation = self + .operation_events .begin( &invocation.context, OperationKind::FeedbackDiagnostics, @@ -2327,6 +2328,7 @@ async fn register_production_advisory_owner( }); let advisory_cycle = Arc::new(ProjectOpenAdvisoryFeedbackCycleV1 { producer, + operation_events: invocation.service.operation_events(), root_uri: state.admitted_root_uri.clone(), feedback_scope: feedback_scope.clone(), github_review_read, diff --git a/crates/tracedecay/src/mcp/server.rs b/crates/tracedecay/src/mcp/server.rs index e662233c61a..a12d68da8b8 100644 --- a/crates/tracedecay/src/mcp/server.rs +++ b/crates/tracedecay/src/mcp/server.rs @@ -473,6 +473,7 @@ pub struct McpServer { application_invocation_executor: Option>, daemon_invocation_service: Option, + operation_events: tracedecay_application::operation_stream::OperationEventAuthority, delivery_settlement_authority: Option>, pub(crate) delivery_settlement_recorder: @@ -804,6 +805,10 @@ impl McpServer { #[cfg(any(test, feature = "test-transport"))] host_admission_test_runtime, } = context; + let operation_events = daemon_invocation_service + .as_ref() + .map(tracedecay_daemon_service::DaemonInvocationService::operation_events) + .unwrap_or_default(); let persisted_tokens_saved = match cg.get_tokens_saved().await { Ok(persisted) => Some(persisted), Err(error) => { @@ -1034,6 +1039,7 @@ impl McpServer { application_surface_client: tokio::sync::OnceCell::new(), application_invocation_executor, daemon_invocation_service, + operation_events, delivery_settlement_authority, delivery_settlement_recorder, project_server_live, diff --git a/crates/tracedecay/src/mcp/server/graph_tool_owner.rs b/crates/tracedecay/src/mcp/server/graph_tool_owner.rs index db5f5392b9b..363f2e0897c 100644 --- a/crates/tracedecay/src/mcp/server/graph_tool_owner.rs +++ b/crates/tracedecay/src/mcp/server/graph_tool_owner.rs @@ -106,6 +106,7 @@ impl McpServer { application_invocation_target: tracedecay_contracts::InvocationTarget::CurrentProject, dashboard_application_invocation_executor: self.application_invocation_executor.clone(), daemon_invocation_service: self.daemon_invocation_service.as_ref(), + operation_events: Some(self.operation_events.clone()), dashboard_delivery_settlement_authority: self.delivery_settlement_authority.clone(), application_request_id: None, application_deadline: None, diff --git a/crates/tracedecay/src/mcp/tools/handlers/dispatch_groups.rs b/crates/tracedecay/src/mcp/tools/handlers/dispatch_groups.rs index f188b1b6061..058192ebb2f 100644 --- a/crates/tracedecay/src/mcp/tools/handlers/dispatch_groups.rs +++ b/crates/tracedecay/src/mcp/tools/handlers/dispatch_groups.rs @@ -787,15 +787,17 @@ async fn compute_run_affected_tests( let recording = options .registered_project_session_db .clone() - .map(|store| workflow::ManagedTestRunRecording { + .zip(options.operation_events.clone()) + .map(|(store, events)| workflow::ManagedTestRunRecording { store, + events, session_id: mcp_analytics_session_id(&args), }) .ok_or_else(|| { TraceDecayError::project_route( "runtime_mounting", true, - "managed test runs are recorded in the project session store, which is still mounting", + "managed test runs require a mounted project session store and owning operation-event authority", ) }); workflow::compute_run_affected_tests( diff --git a/crates/tracedecay/src/mcp/tools/handlers/mod.rs b/crates/tracedecay/src/mcp/tools/handlers/mod.rs index 5de8117601d..17ae397489c 100644 --- a/crates/tracedecay/src/mcp/tools/handlers/mod.rs +++ b/crates/tracedecay/src/mcp/tools/handlers/mod.rs @@ -309,6 +309,8 @@ pub struct ToolCallRegistryOptions<'a> { Option>, pub(crate) daemon_invocation_service: Option<&'a tracedecay_daemon_service::DaemonInvocationService>, + pub(crate) operation_events: + Option, pub(crate) dashboard_delivery_settlement_authority: Option>, pub application_request_id: Option, @@ -388,6 +390,7 @@ impl Default for ToolCallRegistryOptions<'_> { application_invocation_executor: None, dashboard_application_invocation_executor: None, daemon_invocation_service: None, + operation_events: None, dashboard_delivery_settlement_authority: None, application_request_id: None, application_deadline: None, diff --git a/crates/tracedecay/tests/hooks_lsp_suite/hook_spool_drain_test.rs b/crates/tracedecay/tests/hooks_lsp_suite/hook_spool_drain_test.rs index 914e15c444b..854e540790a 100644 --- a/crates/tracedecay/tests/hooks_lsp_suite/hook_spool_drain_test.rs +++ b/crates/tracedecay/tests/hooks_lsp_suite/hook_spool_drain_test.rs @@ -10,8 +10,12 @@ use std::process::{Command, Stdio}; use std::time::{Duration, Instant}; use serde_json::json; -use tracedecay_domain::NativeHostIdentityV1; use tracedecay_domain::canonical_text::sha256_hex; +use tracedecay_domain::{ + DeliveryChannelIdentityV1, DeliveryEventClassV1, DeliverySettlementAttemptV1, + DeliverySettlementOutcomeV1, DeliverySettlementV1, DeliverySurfaceFamilyV1, + NativeHostIdentityV1, +}; use tracedecay_hooks::{ HOOK_SYNCHRONOUS_BUDGET, HookDeliveryReceiptWriterV1, HookDeliveryRetentionV1, HookDeliverySourceReceiptV1, hook_delivery_receipt_spool_root, @@ -378,7 +382,9 @@ fn a_receipt_spool_root_that_is_not_a_directory_is_reported_on_status() { } /// A receipt backlog larger than one drain batch settles in back-to-back -/// sweeps, and status reads `drained` only once none is left. +/// sweeps, and status reads `drained` only once none is left. Stage valid +/// receipt-only deliveries so unrelated end-of-turn producer backpressure +/// cannot turn a receipt-batch regression into a producer-work retry test. #[test] fn a_receipt_backlog_beyond_one_batch_drains_without_waiting_for_the_sweep() { const BACKLOG: usize = 65; @@ -387,44 +393,51 @@ fn a_receipt_backlog_beyond_one_batch_drains_without_waiting_for_the_sweep() { let project = init_git_project(&home); let daemon = spawn_tracedecay_daemon(&home); tracedecay(&home, &project, &["init"]); + drop(daemon); + let layout = default_profile_sharded_layout(&project, &home.join(".tracedecay")).unwrap(); - let records = layout - .data_root - .join("hook-v2-spool/cursor-desktop/records.v1.bin"); let receipts = hook_delivery_receipt_spool_root(&layout.data_root, NativeHostIdentityV1::CursorDesktop); - - drop(daemon); - for index in 0..BACKLOG { - capture_stop(&home, &project, &format!("backlog-{index}")); - } - let held = published_receipts(&receipts) - .into_iter() - .map(|path| { - let receipt = serde_json::from_slice::( - &std::fs::read(&path).unwrap(), - ) - .unwrap(); - std::fs::remove_file(&path).unwrap(); - receipt + let now = tracedecay_runtime_core::tracedecay::utc_now_or_one(); + let held = (0..BACKLOG) + .map(|index| { + HookDeliverySourceReceiptV1::new(DeliverySettlementV1 { + attempt: DeliverySettlementAttemptV1 { + owner_event_id: format!("hook:native:receipt-backlog-{index}"), + event_class: DeliveryEventClassV1::Activity, + channel: DeliveryChannelIdentityV1 { + surface: DeliverySurfaceFamilyV1::Hook, + channel_ref: "hook:cursor_desktop:receipt-backlog".to_owned(), + }, + work_attempt: None, + eligible: 1, + valid_at: now, + attempted_at: now, + }, + outcome: DeliverySettlementOutcomeV1::Delivered, + settled_at: now, + drop_reason: None, + }) + .expect("valid hook delivery receipt") }) .collect::>(); + let unique = held + .iter() + .map(|receipt| receipt.receipt_id) + .collect::>(); assert_eq!( - held.len(), + unique.len(), BACKLOG, - "each stop publishes one delivery receipt" - ); - - // Drain the stops' events first so only the receipt backlog remains. - let daemon = spawn_tracedecay_daemon(&home); - assert!( - drain_latency(&records, PROMPT_DRAIN).is_some(), - "the stops' events must drain within {PROMPT_DRAIN:?}" + "every receipt has a distinct identity" ); - drop(daemon); for receipt in &held { publish(&receipts, receipt); } + assert_eq!( + published_receipts(&receipts).len(), + BACKLOG, + "all receipts must be durable before the daemon restarts" + ); let _daemon = spawn_tracedecay_daemon(&home); assert!( diff --git a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/affected_tests_behavior_test.rs b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/affected_tests_behavior_test.rs index 67292ac6d9a..0b206a748fd 100644 --- a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/affected_tests_behavior_test.rs +++ b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/affected_tests_behavior_test.rs @@ -11,7 +11,6 @@ use std::time::Duration; use serde_json::{Value, json}; use tracedecay::mcp::McpServer; -use tracedecay_application::operation_stream::operation_event_authority; use tracedecay_contracts::request_identity::{GlobalRequestSurface, mint_global_request_id}; use tracedecay_contracts::{Deadline, OperationBudgetUsage, OperationReceipt}; use tracedecay_domain::UtcMicros; @@ -661,12 +660,9 @@ async fn concurrent_read_refusals_keep_each_callers_request_identity() { fixture.harness.shutdown().await; } -/// The operation-event authority is process-global, so every daemon -/// composition in this test process shares it. A peer composition shutting -/// down (the production `DaemonInvocationState::shutdown` path) between a -/// managed run's admission and its first result must leave the accepted -/// record in place: the first result lands at sequence 1 behind it instead -/// of failing with an expired history frontier. +/// Independently composed daemons own distinct operation authorities. Both +/// must admit the same request ID, and shutting down a peer must not truncate +/// an already admitted run in the surviving composition. #[tokio::test] async fn operation_history_keeps_an_admitted_test_run_across_a_peer_composition_shutdown() { let owner = production_composition_fixture_with_sources(write_affected_fixture).await; @@ -678,14 +674,21 @@ async fn operation_history_keeps_an_admitted_test_run_across_a_peer_composition_ .expect("affected fixture root URI") .to_string(); let deadline = Deadline::new(UtcMicros(i64::MAX)).expect("deadline"); - let emitter = operation_event_authority() - .begin_managed_test_run( - root_uri, - mint_global_request_id(GlobalRequestSurface::ManagedTestRun).expect("request id"), - deadline.clone(), - ) + let request_id = + mint_global_request_id(GlobalRequestSurface::ManagedTestRun).expect("request id"); + let owner_events = owner + .harness + .operation_events() + .expect("first composition authority"); + let peer_events = peer.operation_events().expect("peer composition authority"); + let emitter = owner_events + .begin_managed_test_run(root_uri.clone(), request_id.clone(), deadline.clone()) .await .expect("managed test run admitted before the peer shuts down"); + let _peer_emitter = peer_events + .begin_managed_test_run(root_uri, request_id, deadline.clone()) + .await + .expect("independent composition must admit the same request ID"); peer.shutdown().await;