diff --git a/crates/tracedecay-daemon-service/src/invocation/types.rs b/crates/tracedecay-daemon-service/src/invocation/types.rs index 74c86cdbd4..f2b8812b5a 100644 --- a/crates/tracedecay-daemon-service/src/invocation/types.rs +++ b/crates/tracedecay-daemon-service/src/invocation/types.rs @@ -207,6 +207,7 @@ impl BoundedHookOrchestratorV1 { address: &HookOrchestrationAddressV1, operation: &Arc, emit_terminal: bool, + permit: Option, ) { let completions = { let mut in_flight = in_flight @@ -233,6 +234,12 @@ impl BoundedHookOrchestratorV1 { { in_flight.events.remove(&operation.event); } + // Admission checks the in-flight maps and takes a permit under + // this lock, so the cycle leaves both in one step: no admission + // sees it gone yet still holding its permit, or sees its permit + // free while it still looks in flight. Its terminals, the host's + // cue to send the session's next boundary, follow. + drop(permit); completions }; if emit_terminal { @@ -388,15 +395,14 @@ impl BoundedHookOrchestratorV1 { let superseded = operation .superseded .load(std::sync::atomic::Ordering::Acquire); - Self::settle_operation(&in_flight, &address, &operation, superseded); + Self::settle_operation(&in_flight, &address, &operation, superseded, None); return; }; if cancellation.is_cancelled() || work_cancellation.is_cancelled() { let superseded = operation .superseded .load(std::sync::atomic::Ordering::Acquire); - drop(permit); - Self::settle_operation(&in_flight, &address, &operation, superseded); + Self::settle_operation(&in_flight, &address, &operation, superseded, Some(permit)); return; } let resumable = request.clone(); @@ -456,12 +462,13 @@ impl BoundedHookOrchestratorV1 { .load(std::sync::atomic::Ordering::Acquire) } }; - // Free the permit before the operation leaves the in-flight maps. - // An admission that no longer finds this operation must not be - // backpressured by it, and a completion fires only once the next - // admission can run. - drop(permit); - Self::settle_operation(&in_flight, &address, &operation, emit_terminal); + Self::settle_operation( + &in_flight, + &address, + &operation, + emit_terminal, + Some(permit), + ); }); let Ok(task) = task else { Self::settle_operation( @@ -469,6 +476,7 @@ impl BoundedHookOrchestratorV1 { &spawned_address, &spawned_operation, false, + None, ); return Err(HookOrchestrationAdmissionV1::Unavailable); }; diff --git a/crates/tracedecay-daemon-service/src/invocation/types/context_scout_tests.rs b/crates/tracedecay-daemon-service/src/invocation/types/context_scout_tests.rs index 53c71e81d6..f1a891714c 100644 --- a/crates/tracedecay-daemon-service/src/invocation/types/context_scout_tests.rs +++ b/crates/tracedecay-daemon-service/src/invocation/types/context_scout_tests.rs @@ -257,6 +257,54 @@ async fn explicit_request_preserves_a_newer_native_event() { assert!(runtime.shutdown().await); } +/// A cycle's terminal is the host's cue to send the session's next boundary, +/// so the cycle must release its permit before it reports that terminal. +#[tokio::test] +async fn settled_cycle_admits_the_sessions_next_boundary_from_its_terminal() { + let mut first = native_request(221); + let mut envelope = first.hook.envelope().clone(); + envelope.event_id = [229; 16]; + let binding = HookScopeBindingV1 { + host: envelope.producer, + project_id: envelope.project_id, + repository_id: envelope.repository_id, + worktree_id: envelope.worktree_id, + worktree_epoch: envelope.worktree_epoch, + binding_token: envelope.binding_token, + capabilities: vec![tracedecay_hooks::HookCapabilityV1 { + family: tracedecay_hooks::HookEventFamily::SavedEdit, + support: tracedecay_hooks::stock_event_support( + envelope.producer, + tracedecay_hooks::HookEventFamily::SavedEdit, + ), + }], + }; + let next = + HookOrchestrationRequestV1::from_envelope(envelope, &binding, None, 1, false).unwrap(); + let runtime = BoundedHookOrchestratorV1::new(1, |_, _| async { + HookOrchestrationWorkOutcomeV1::Completed + }) + .unwrap(); + let (admitted, admission) = tokio::sync::oneshot::channel(); + let pending = StdMutex::new(Some((next, admitted))); + let owner = Arc::downgrade(&runtime); + first.completion = Some(Arc::new(move || { + if let Some((next, admitted)) = pending.lock().unwrap().take() + && let Some(runtime) = owner.upgrade() + { + let _ = admitted.send(runtime.admit(next)); + } + })); + + assert_eq!(runtime.admit(first), HookOrchestrationAdmissionV1::Enqueued); + assert_eq!( + admission.await.unwrap(), + HookOrchestrationAdmissionV1::Enqueued, + "a settled cycle must not backpressure the boundary its terminal invites" + ); + assert!(runtime.shutdown().await); +} + /// A boundary of a session with no Context Scout lifecycle runs once and /// settles. It reruns only when a later hook of that session carries a /// lifecycle, and a newer boundary of the session replaces it. @@ -354,6 +402,15 @@ async fn lifecycle_less_boundary_settles_and_resumes_on_its_sessions_lifecycle() Some(([210; 16], true)), "only the session's hook that carries a lifecycle resumes the parked boundary" ); + // The resumed cycle reports no terminal. Until it settles, the session's + // next boundary supersedes it and inherits its lifecycle. + tokio::time::timeout(std::time::Duration::from_secs(5), async { + while runtime.active_tasks() != 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("the resumed cycle settles"); session_hook(213, Some(lifecycle.clone())); let second = saved_edit(214); assert_eq!(