fix(streams): refuse completed terminal for unrecorded tests

crates/tracedecay-application/src/lsp_runtime.rs

@@ -349,6 +349,7 @@ pub(crate) struct StoredLspTestRunExpansionV1 {
 pub fn lsp_session_factory<F>(
     feedback_runtime: Arc<FeedbackRuntime>,
     test_run_store: RegisteredGlobalDbLeaseV1,
+    operation_events: crate::operation_stream::OperationEventAuthority,
     code_index: Arc<dyn LspCodeIndexProjectionIdentityPort>,
     workspace_index: Arc<dyn crate::lsp_support::LspWorkspaceDocumentIndexPort>,
     diagnostic_records: Arc<dyn LspFeedbackDiagnosticRecordPort>,
@@ -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(),

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<RegisteredProjectLspAuthority>,
     store: RegisteredGlobalDbLeaseV1,
+    operation_events: OperationEventAuthority,
 ) -> Arc<dyn LspTestRunProjectionPort> {
     Arc::new(ManagedTestRunProjection::new(
-        CanonicalManagedTestRunReader::new(store, operation_event_authority()),
+        CanonicalManagedTestRunReader::new(store, operation_events),
         project,
     ))
 }

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<OperationEventItem>;
 
-static OPERATION_EVENTS: OnceLock<OperationEventAuthority> = 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;

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(
                 ),

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,

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();

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<dyn LspCodeIndexProjectionIdentityPort>,
                 workspace_index,
                 diagnostic_records,

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
+    );
+}

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,
 };

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<String>,
 }
 
@@ -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))?;