From 2bc57afb46da107a536580cfa23f602544fd5769 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 06:30:12 -0800 Subject: [PATCH 01/16] fix(streams): refuse success when managed test settlement fails --- .../tracedecay-mcp/src/handlers/workflow.rs | 15 +++- .../handlers/workflow/affected_tests_tests.rs | 76 +++++++++++++++++++ 2 files changed, 90 insertions(+), 1 deletion(-) diff --git a/crates/tracedecay-mcp/src/handlers/workflow.rs b/crates/tracedecay-mcp/src/handlers/workflow.rs index 0f2a7a4dc5a..250213f247f 100644 --- a/crates/tracedecay-mcp/src/handlers/workflow.rs +++ b/crates/tracedecay-mcp/src/handlers/workflow.rs @@ -766,7 +766,20 @@ 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, + ) + }; + publish_test_run_terminal(&self.emitter, stream_receipt).await?; recorded?; Ok(receipt) } 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..9a3e043db2b 100644 --- a/crates/tracedecay-mcp/src/handlers/workflow/affected_tests_tests.rs +++ b/crates/tracedecay-mcp/src/handlers/workflow/affected_tests_tests.rs @@ -1208,3 +1208,79 @@ 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() { + use futures_util::StreamExt as _; + use tracedecay_application::operation_stream::{ + OperationEventAuthority, OperationId, OperationKind, + }; + use tracedecay_contracts::StreamEventKind; + + 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(), + 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"); +} From 3df8d6301a67fcc312f4a66527f4ab2085e4e1b0 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 06:32:29 -0800 Subject: [PATCH 02/16] test(streams): retain typed refusal on failed settlement --- crates/tracedecay-mcp/src/handlers/workflow.rs | 4 ++-- .../src/handlers/workflow/affected_tests_tests.rs | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/crates/tracedecay-mcp/src/handlers/workflow.rs b/crates/tracedecay-mcp/src/handlers/workflow.rs index 250213f247f..35296beb10b 100644 --- a/crates/tracedecay-mcp/src/handlers/workflow.rs +++ b/crates/tracedecay-mcp/src/handlers/workflow.rs @@ -735,8 +735,8 @@ impl ManagedTestRun { /// 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, 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 9a3e043db2b..3a71f8b4f00 100644 --- a/crates/tracedecay-mcp/src/handlers/workflow/affected_tests_tests.rs +++ b/crates/tracedecay-mcp/src/handlers/workflow/affected_tests_tests.rs @@ -1268,7 +1268,7 @@ async fn refused_durable_test_result_cannot_publish_a_completed_live_terminal() .await .expect_err("unrecorded outcome cannot settle durably"); assert!( - matches!(error, TraceDecayError::Database { .. }), + matches!(&error, TraceDecayError::Database { .. }), "durable-store refusal must remain typed: {error}" ); let terminal = tokio::time::timeout(std::time::Duration::from_secs(2), stream.next()) From 51c71ffc89b9f75f90b3b589fa278cb638bbd567 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 07:25:17 -0800 Subject: [PATCH 03/16] fix(streams): preserve durable refusal over terminal errors --- .../tracedecay-mcp/src/handlers/workflow.rs | 17 ++++- .../handlers/workflow/affected_tests_tests.rs | 67 +++++++++++++++++-- 2 files changed, 74 insertions(+), 10 deletions(-) diff --git a/crates/tracedecay-mcp/src/handlers/workflow.rs b/crates/tracedecay-mcp/src/handlers/workflow.rs index 35296beb10b..2c61969816c 100644 --- a/crates/tracedecay-mcp/src/handlers/workflow.rs +++ b/crates/tracedecay-mcp/src/handlers/workflow.rs @@ -719,7 +719,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,7 +730,13 @@ 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)) } @@ -779,8 +787,11 @@ impl ManagedTestRun { bytes_consumed, ) }; - publish_test_run_terminal(&self.emitter, stream_receipt).await?; + // 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; recorded?; + published?; Ok(receipt) } 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 3a71f8b4f00..63a143828aa 100644 --- a/crates/tracedecay-mcp/src/handlers/workflow/affected_tests_tests.rs +++ b/crates/tracedecay-mcp/src/handlers/workflow/affected_tests_tests.rs @@ -1,5 +1,6 @@ use super::*; +use futures_util::StreamExt as _; use serde_json::json; use crate::ToolResult; @@ -10,11 +11,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, @@ -1211,12 +1215,6 @@ fn assert_failed_terminal(output: &Value) { #[tokio::test] async fn refused_durable_test_result_cannot_publish_a_completed_live_terminal() { - use futures_util::StreamExt as _; - use tracedecay_application::operation_stream::{ - OperationEventAuthority, OperationId, OperationKind, - }; - use tracedecay_contracts::StreamEventKind; - let cancellation = CancellationSignal::active("cancellation.affected-tests.settlement-refusal") .expect("fixture cancellation"); @@ -1284,3 +1282,58 @@ async fn refused_durable_test_result_cannot_publish_a_completed_live_terminal() )); 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 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(), + 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}" + ); +} From 864e982cf6189af95586f39305bcc8b5a2f73d42 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 07:26:29 -0800 Subject: [PATCH 04/16] style(streams): match hosted Rust formatter for run regressions --- .../src/handlers/workflow/affected_tests_tests.rs | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) 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 63a143828aa..5d3f4603651 100644 --- a/crates/tracedecay-mcp/src/handlers/workflow/affected_tests_tests.rs +++ b/crates/tracedecay-mcp/src/handlers/workflow/affected_tests_tests.rs @@ -1215,9 +1215,8 @@ fn assert_failed_terminal(output: &Value) { #[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 cancellation = CancellationSignal::active("cancellation.affected-tests.settlement-refusal") + .expect("fixture cancellation"); let context = verified_graph_context(&cancellation); let authority = OperationEventAuthority::default(); let emitter = authority @@ -1280,14 +1279,16 @@ async fn refused_durable_test_result_cannot_publish_a_completed_live_terminal() && termination.receipt.termination == OperationTermination::Failed && termination.receipt.budget.bytes_consumed == 7 )); - assert!(stream.next().await.is_none(), "live stream ends after refusal"); + 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 cancellation = CancellationSignal::active("cancellation.affected-tests.concurrent-terminal") + .expect("fixture cancellation"); let context = verified_graph_context(&cancellation); let authority = OperationEventAuthority::default(); let emitter = authority From 21e5e01dc31adad0e4e95377cf10a57a82c8dc7d Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 08:07:52 -0800 Subject: [PATCH 05/16] fix(streams): satisfy Bazel format and Clippy checks --- .../src/handlers/workflow/affected_tests_tests.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) 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 5d3f4603651..42e8d30f7cc 100644 --- a/crates/tracedecay-mcp/src/handlers/workflow/affected_tests_tests.rs +++ b/crates/tracedecay-mcp/src/handlers/workflow/affected_tests_tests.rs @@ -1,6 +1,5 @@ use super::*; -use futures_util::StreamExt as _; use serde_json::json; use crate::ToolResult; @@ -1287,8 +1286,9 @@ async fn refused_durable_test_result_cannot_publish_a_completed_live_terminal() #[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 cancellation = + CancellationSignal::active("cancellation.affected-tests.concurrent-terminal") + .expect("fixture cancellation"); let context = verified_graph_context(&cancellation); let authority = OperationEventAuthority::default(); let emitter = authority From 4c209dc49adfd110d512ca006e1f5fbdd15526ae Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 09:14:45 -0800 Subject: [PATCH 06/16] fix(streams): report double-fault terminal publication refusal --- .../tracedecay-mcp/src/handlers/workflow.rs | 8 +++++ .../handlers/workflow/affected_tests_tests.rs | 31 +++++++++++++++++++ 2 files changed, 39 insertions(+) diff --git a/crates/tracedecay-mcp/src/handlers/workflow.rs b/crates/tracedecay-mcp/src/handlers/workflow.rs index 2c61969816c..3f1d38e19dc 100644 --- a/crates/tracedecay-mcp/src/handlers/workflow.rs +++ b/crates/tracedecay-mcp/src/handlers/workflow.rs @@ -790,6 +790,14 @@ impl ManagedTestRun { // 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) 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 42e8d30f7cc..081d53c5b24 100644 --- a/crates/tracedecay-mcp/src/handlers/workflow/affected_tests_tests.rs +++ b/crates/tracedecay-mcp/src/handlers/workflow/affected_tests_tests.rs @@ -1295,6 +1295,17 @@ async fn durable_store_refusal_remains_primary_when_terminal_publication_also_fa .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. @@ -1337,4 +1348,24 @@ async fn durable_store_refusal_remains_primary_when_terminal_publication_also_fa 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" + ); } From f71fc271d505875480f142873065c4b8ac2ffc75 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 10:29:02 -0800 Subject: [PATCH 07/16] refactor(streams): bind operation frontiers to daemon owners --- .../tracedecay-application/src/lsp_runtime.rs | 3 +- .../src/lsp_runtime/managed_test_runs.rs | 5 ++-- .../src/operation_stream.rs | 19 +++--------- .../src/invocation.rs | 6 ++-- .../src/invocation/feedback.rs | 4 --- .../src/invocation/lsp.rs | 4 +-- .../src/invocation/registrars/lsp.rs | 1 + .../src/invocation/tests/mod.rs | 29 +++++++++++++++++++ crates/tracedecay-daemon-service/src/lib.rs | 2 +- .../tracedecay-mcp/src/handlers/workflow.rs | 7 +++-- .../handlers/workflow/affected_tests_tests.rs | 3 +- .../src/daemon/http_application_router.rs | 6 +++- .../configuration_idempotency_journey_test.rs | 3 +- .../project_open_owners/advisory_runtime.rs | 9 ++++-- crates/tracedecay/src/mcp/server.rs | 6 ++++ .../src/mcp/server/graph_tool_owner.rs | 1 + .../src/mcp/tools/handlers/dispatch_groups.rs | 6 ++-- .../tracedecay/src/mcp/tools/handlers/mod.rs | 3 ++ 18 files changed, 78 insertions(+), 39 deletions(-) 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..b1979aec776 100644 --- a/crates/tracedecay-application/src/operation_stream.rs +++ b/crates/tracedecay-application/src/operation_stream.rs @@ -8,7 +8,7 @@ use std::collections::{BTreeMap, BTreeSet, VecDeque}; use std::fmt; use std::pin::Pin; use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; -use std::sync::{Arc, OnceLock}; +use std::sync::Arc; 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..c4d3703298c 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,7 +238,7 @@ pub use feedback::{ DaemonAdvisoryCycleMountChangedFuture, DaemonAdvisoryCycleMountFuture, DaemonAdvisoryCycleMountV1, DaemonFeedbackInvocationOwner, DaemonFeedbackProximityInvocationFuture, DaemonFeedbackProximityInvocationRequest, - advisory_cycle_invocation_result, daemon_operation_event_authority, + advisory_cycle_invocation_result, feedback_proximity_invocation_result, }; pub use primitive::{ @@ -431,7 +431,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..d8b05dee062 100644 --- a/crates/tracedecay-daemon-service/src/invocation/tests/mod.rs +++ b/crates/tracedecay-daemon-service/src/invocation/tests/mod.rs @@ -135,3 +135,32 @@ 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"); +} 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 3f1d38e19dc..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, } @@ -839,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 081d53c5b24..e271a82c7ce 100644 --- a/crates/tracedecay-mcp/src/handlers/workflow/affected_tests_tests.rs +++ b/crates/tracedecay-mcp/src/handlers/workflow/affected_tests_tests.rs @@ -51,6 +51,7 @@ where .await; let recording = ManagedTestRunRecording { store: sessions.registered.clone(), + events: OperationEventAuthority::default(), session_id: None, }; let completion = @@ -1154,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(), diff --git a/crates/tracedecay/src/daemon/http_application_router.rs b/crates/tracedecay/src/daemon/http_application_router.rs index a78536d6331..6cc81f208fe 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) 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_open_owners/advisory_runtime.rs b/crates/tracedecay/src/daemon/project_open_owners/advisory_runtime.rs index bd756c6db29..d08b9a7c088 100644 --- a/crates/tracedecay/src/daemon/project_open_owners/advisory_runtime.rs +++ b/crates/tracedecay/src/daemon/project_open_owners/advisory_runtime.rs @@ -113,8 +113,8 @@ 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, + 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, }; @@ -135,6 +135,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 +278,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 +2329,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..b106af35931 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(|service| service.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, From 92d25a01cf42ca5cbfffe54278eeca1f8a8a669a Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 10:42:07 -0800 Subject: [PATCH 08/16] fix(streams): bind settlement fixtures to their event owner --- .../src/handlers/workflow/affected_tests_tests.rs | 2 ++ 1 file changed, 2 insertions(+) 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 e271a82c7ce..aedaa9f606e 100644 --- a/crates/tracedecay-mcp/src/handlers/workflow/affected_tests_tests.rs +++ b/crates/tracedecay-mcp/src/handlers/workflow/affected_tests_tests.rs @@ -1250,6 +1250,7 @@ async fn refused_durable_test_result_cannot_publish_a_completed_live_terminal() emitter, recording: ManagedTestRunRecording { store: store.registered.clone(), + events: authority.clone(), session_id: None, }, started_at: UtcMicros(10), @@ -1328,6 +1329,7 @@ async fn durable_store_refusal_remains_primary_when_terminal_publication_also_fa emitter, recording: ManagedTestRunRecording { store: store.registered.clone(), + events: authority.clone(), session_id: None, }, started_at: UtcMicros(10), From 347b1b7556705d9514f96fe0f466a5b02ae6cf3e Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 10:44:09 -0800 Subject: [PATCH 09/16] test(streams): prove independent real daemon event owners --- .../src/daemon/production_harness.rs | 12 +++++++++ .../affected_tests_behavior_test.rs | 27 ++++++++++--------- 2 files changed, 26 insertions(+), 13 deletions(-) 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/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..fbe7a025ec9 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,7 @@ use std::time::Duration; use serde_json::{Value, json}; use tracedecay::mcp::McpServer; -use tracedecay_application::operation_stream::operation_event_authority; +use tracedecay_application::operation_stream::OperationEventAuthority; use tracedecay_contracts::request_identity::{GlobalRequestSurface, mint_global_request_id}; use tracedecay_contracts::{Deadline, OperationBudgetUsage, OperationReceipt}; use tracedecay_domain::UtcMicros; @@ -661,12 +661,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 +675,18 @@ 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; From 2897268e7f60c041d60b9c13d56c8a3c5504b4ad Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 10:44:44 -0800 Subject: [PATCH 10/16] style(streams): remove stale singleton-test import --- .../mcp_suite/mcp_handler_test/affected_tests_behavior_test.rs | 1 - 1 file changed, 1 deletion(-) 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 fbe7a025ec9..cd4dc28563b 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::OperationEventAuthority; use tracedecay_contracts::request_identity::{GlobalRequestSurface, mint_global_request_id}; use tracedecay_contracts::{Deadline, OperationBudgetUsage, OperationReceipt}; use tracedecay_domain::UtcMicros; From 7bb060c668e1f51fdd7b55e03eb8ac02c643b2aa Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 11:06:57 -0800 Subject: [PATCH 11/16] style(streams): apply Bazel formatter to owner cutover --- crates/tracedecay-application/src/operation_stream.rs | 2 +- crates/tracedecay-daemon-service/src/invocation.rs | 3 +-- .../src/daemon/project_open_owners/advisory_runtime.rs | 5 ++--- .../mcp_handler_test/affected_tests_behavior_test.rs | 9 ++++++--- 4 files changed, 10 insertions(+), 9 deletions(-) diff --git a/crates/tracedecay-application/src/operation_stream.rs b/crates/tracedecay-application/src/operation_stream.rs index b1979aec776..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::atomic::{AtomicU64, AtomicUsize, Ordering}; use std::sync::Arc; +use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; use std::task::{Context, Poll}; use schemars::JsonSchema; diff --git a/crates/tracedecay-daemon-service/src/invocation.rs b/crates/tracedecay-daemon-service/src/invocation.rs index c4d3703298c..ff5daa9300f 100644 --- a/crates/tracedecay-daemon-service/src/invocation.rs +++ b/crates/tracedecay-daemon-service/src/invocation.rs @@ -238,8 +238,7 @@ pub use feedback::{ DaemonAdvisoryCycleMountChangedFuture, DaemonAdvisoryCycleMountFuture, DaemonAdvisoryCycleMountV1, DaemonFeedbackInvocationOwner, DaemonFeedbackProximityInvocationFuture, DaemonFeedbackProximityInvocationRequest, - advisory_cycle_invocation_result, - feedback_proximity_invocation_result, + advisory_cycle_invocation_result, feedback_proximity_invocation_result, }; pub use primitive::{ DaemonContextScoutRuntimeRegistrar, DaemonContextScoutRuntimeRegistrationError, 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 d08b9a7c088..60cd5a21542 100644 --- a/crates/tracedecay/src/daemon/project_open_owners/advisory_runtime.rs +++ b/crates/tracedecay/src/daemon/project_open_owners/advisory_runtime.rs @@ -114,9 +114,8 @@ use tracedecay_daemon_service::{ DaemonFeedbackProximityInvocationFuture, DaemonFeedbackProximityInvocationRequest, HookOrchestrationRequestV1, HookOrchestrationTriggerV1, HookOrchestrationWorkOutcomeV1, 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, + 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; 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 cd4dc28563b..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 @@ -674,9 +674,12 @@ 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 request_id = mint_global_request_id(GlobalRequestSurface::ManagedTestRun) - .expect("request id"); - let owner_events = owner.harness.operation_events().expect("first composition authority"); + 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()) From 24fabcc32bc721357fcaad1d4165f4b2c2068105 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 12:14:04 -0800 Subject: [PATCH 12/16] fix(streams): thread daemon event owner through HTTP mounts --- crates/tracedecay/src/daemon/http_application_router.rs | 5 ++++- crates/tracedecay/src/daemon/project_composition.rs | 1 + 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/crates/tracedecay/src/daemon/http_application_router.rs b/crates/tracedecay/src/daemon/http_application_router.rs index 6cc81f208fe..08bb14e2d99 100644 --- a/crates/tracedecay/src/daemon/http_application_router.rs +++ b/crates/tracedecay/src/daemon/http_application_router.rs @@ -181,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 } @@ -235,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/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); From cbf262cd14b21b5c9f1012ba1e93ce6e1b003dd5 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 13:12:13 -0800 Subject: [PATCH 13/16] fix(mcp): use daemon method reference for event owner --- crates/tracedecay/src/mcp/server.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/tracedecay/src/mcp/server.rs b/crates/tracedecay/src/mcp/server.rs index b106af35931..a12d68da8b8 100644 --- a/crates/tracedecay/src/mcp/server.rs +++ b/crates/tracedecay/src/mcp/server.rs @@ -807,7 +807,7 @@ impl McpServer { } = context; let operation_events = daemon_invocation_service .as_ref() - .map(|service| service.operation_events()) + .map(tracedecay_daemon_service::DaemonInvocationService::operation_events) .unwrap_or_default(); let persisted_tokens_saved = match cg.get_tokens_saved().await { Ok(persisted) => Some(persisted), From 92a77d06e3d6d301a4dbf104633e140de8d1f244 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 15:12:29 -0800 Subject: [PATCH 14/16] test(streams): isolate cancellation across daemon owners Refs #4450. Add a real service-level regression opening the same root and request in two independent daemon compositions. The first owner's cancellation and expired-frontier cleanup must not cancel or remove the second owner's active run. Keep the existing distinct-root test and all lifecycle assertions. --- .../src/invocation/tests/mod.rs | 51 +++++++++++++++++++ 1 file changed, 51 insertions(+) diff --git a/crates/tracedecay-daemon-service/src/invocation/tests/mod.rs b/crates/tracedecay-daemon-service/src/invocation/tests/mod.rs index d8b05dee062..16e29df1bb2 100644 --- a/crates/tracedecay-daemon-service/src/invocation/tests/mod.rs +++ b/crates/tracedecay-daemon-service/src/invocation/tests/mod.rs @@ -164,3 +164,54 @@ async fn independent_daemon_compositions_have_separate_managed_test_frontiers() .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 + ); +} From e18430ebd3987247d8354c108d844d4e70ffaadb Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 17:14:49 -0800 Subject: [PATCH 15/16] test(hooks): isolate receipt batching from producer backpressure Refs #4450. The 65-stop backlog fixture introduced independent producer-work obligations and could report a truthful backpressured status after all receipts settled. Exercise the real receipt writer and daemon settlement with 65 validated, distinct hook delivery receipts but no unrelated producer events. Preserve the 10-second bound, exact 65-receipt assertion, full drained status and independent capture/replay tests. No production code or timing/backoff limits changed. --- .../hooks_lsp_suite/hook_spool_drain_test.rs | 73 +++++++++++-------- 1 file changed, 41 insertions(+), 32 deletions(-) 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..0563bc21daf 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,7 +10,11 @@ use std::process::{Command, Stdio}; use std::time::{Duration, Instant}; use serde_json::json; -use tracedecay_domain::NativeHostIdentityV1; +use tracedecay_domain::{ + DeliveryChannelIdentityV1, DeliveryEventClassV1, DeliverySettlementAttemptV1, + DeliverySettlementOutcomeV1, DeliverySettlementV1, DeliverySurfaceFamilyV1, + NativeHostIdentityV1, +}; use tracedecay_domain::canonical_text::sha256_hex; use tracedecay_hooks::{ HOOK_SYNCHRONOUS_BUDGET, HookDeliveryReceiptWriterV1, HookDeliveryRetentionV1, @@ -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,47 @@ 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::>(); - assert_eq!( - held.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:?}" - ); - drop(daemon); + let unique = held + .iter() + .map(|receipt| receipt.receipt_id) + .collect::>(); + assert_eq!(unique.len(), BACKLOG, "every receipt has a distinct identity"); 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!( From e473225a908f68d0a53fdea41a21aaf34eeb7ab0 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 22:06:32 -0800 Subject: [PATCH 16/16] style(hooks): format spool drain test with pinned rustfmt Refs #4450 The pinned rustfmt check reports import ordering and an over-long assert_eq! call in the spool-drain regression. Apply exactly those hunks without changing any assertion. Co-Authored-By: Claude Sonnet 5.5 --- .../tests/hooks_lsp_suite/hook_spool_drain_test.rs | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) 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 0563bc21daf..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,12 +10,12 @@ use std::process::{Command, Stdio}; use std::time::{Duration, Instant}; use serde_json::json; +use tracedecay_domain::canonical_text::sha256_hex; use tracedecay_domain::{ DeliveryChannelIdentityV1, DeliveryEventClassV1, DeliverySettlementAttemptV1, DeliverySettlementOutcomeV1, DeliverySettlementV1, DeliverySurfaceFamilyV1, NativeHostIdentityV1, }; -use tracedecay_domain::canonical_text::sha256_hex; use tracedecay_hooks::{ HOOK_SYNCHRONOUS_BUDGET, HookDeliveryReceiptWriterV1, HookDeliveryRetentionV1, HookDeliverySourceReceiptV1, hook_delivery_receipt_spool_root, @@ -425,7 +425,11 @@ fn a_receipt_backlog_beyond_one_batch_drains_without_waiting_for_the_sweep() { .iter() .map(|receipt| receipt.receipt_id) .collect::>(); - assert_eq!(unique.len(), BACKLOG, "every receipt has a distinct identity"); + assert_eq!( + unique.len(), + BACKLOG, + "every receipt has a distinct identity" + ); for receipt in &held { publish(&receipts, receipt); }