diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 77c8ed4556..e13ec8b452 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -619,8 +619,14 @@ jobs:
 
       - name: Install the Windows C cross-compiler
         run: |
-          sudo apt-get update
-          sudo apt-get install -y --no-install-recommends gcc-mingw-w64-x86-64
+          # This compiler comes from Ubuntu, independently of third-party feeds
+          # preconfigured on the hosted image.
+          ubuntu_sources=(
+            -o Dir::Etc::sourcelist=sources.list.d/ubuntu.sources
+            -o Dir::Etc::sourceparts=-
+          )
+          sudo apt-get "${ubuntu_sources[@]}" update
+          sudo apt-get "${ubuntu_sources[@]}" install -y --no-install-recommends gcc-mingw-w64-x86-64
 
       - name: Check the Windows target
         run: >-
diff --git a/MODULE.bazel b/MODULE.bazel
index 275174e15e..bc3e333a16 100644
--- a/MODULE.bazel
+++ b/MODULE.bazel
@@ -2,6 +2,7 @@ module(name = "tracedecay", version = "0.0.0")
 
 bazel_dep(name = "platforms", version = "1.1.0")
 bazel_dep(name = "bazel_skylib", version = "1.9.2")
+bazel_dep(name = "bazel_lib", version = "3.7.0")
 bazel_dep(name = "rules_python", version = "2.4.2")
 bazel_dep(name = "rules_rust", version = "0.74.0")
 # Rustc adds every dependency search directory to PATH on Windows; large
diff --git a/crates/tracedecay-application/src/code_index.rs b/crates/tracedecay-application/src/code_index.rs
index d660d5dd30..d540914272 100644
--- a/crates/tracedecay-application/src/code_index.rs
+++ b/crates/tracedecay-application/src/code_index.rs
@@ -268,7 +268,7 @@ mod tests {
                     resident_bytes: bytes,
                     unreclaimable_bytes: bytes,
                     swapped_bytes: 0,
-                    cgroup_committed_bytes: None,
+                    cgroup_headroom_bytes: None,
                 })
             }),
         ));
diff --git a/crates/tracedecay-cli/tests/c_library_allocator.rs b/crates/tracedecay-cli/tests/c_library_allocator.rs
index c434a4fd49..df4f9ef0ad 100644
--- a/crates/tracedecay-cli/tests/c_library_allocator.rs
+++ b/crates/tracedecay-cli/tests/c_library_allocator.rs
@@ -19,10 +19,15 @@ static MIMALLOC: mimalloc::MiMalloc = mimalloc::MiMalloc;
 const MIB: usize = 1024 * 1024;
 
 /// Bytes glibc's arenas and its mmapped chunks hold for live allocations.
+///
+/// The hermetic toolchain links against glibc 2.31, which predates
+/// `mallinfo2`. `mallinfo` reports the same counters as `int`, which this
+/// test's few-MiB working set stays far below.
 fn glibc_in_use() -> usize {
-    // SAFETY: `mallinfo2` only reads allocator statistics.
-    let info = unsafe { libc::mallinfo2() };
-    info.uordblks + info.hblkhd
+    // SAFETY: `mallinfo` only reads allocator statistics.
+    let info = unsafe { libc::mallinfo() };
+    usize::try_from(info.uordblks).expect("in-use arena bytes fit in int")
+        + usize::try_from(info.hblkhd).expect("mmapped bytes fit in int")
 }
 
 #[test]
diff --git a/crates/tracedecay-cli/tests/core_cli_suite/source_provenance_test.rs b/crates/tracedecay-cli/tests/core_cli_suite/source_provenance_test.rs
index f59b31095b..b8b5db1d79 100644
--- a/crates/tracedecay-cli/tests/core_cli_suite/source_provenance_test.rs
+++ b/crates/tracedecay-cli/tests/core_cli_suite/source_provenance_test.rs
@@ -410,23 +410,8 @@ impl CargoFixture {
         let watched_input = directory.path().join("release-sha.txt");
         std::fs::create_dir_all(&cargo_home).expect("fixture cargo home");
         std::fs::write(&watched_input, ENV_SHA).expect("write watched input");
-        std::fs::create_dir_all(root.join("src")).expect("fixture source directory");
-        std::fs::create_dir_all(root.join("provenance/src"))
-            .expect("fixture build dependency source directory");
-
-        for relative in [
-            "Cargo.toml",
-            "Cargo.lock",
-            "build.rs",
-            "README.md",
-            "provenance/Cargo.toml",
-            "provenance/src/lib.rs",
-        ] {
-            std::fs::copy(fixture.join(relative), root.join(relative))
-                .unwrap_or_else(|error| panic!("copy checked-in fixture {relative}: {error}"));
-        }
-        std::fs::copy(fixture.join("src/main.rs"), root.join("src/main.rs"))
-            .expect("copy checked-in fixture src/main.rs");
+        crate::common::fixture::copy_fixture_tree(&fixture, &root)
+            .expect("copy writable provenance fixture");
 
         Self {
             _directory: directory,
diff --git a/crates/tracedecay-cli/tests/core_cli_suite/tool_daemon_test.rs b/crates/tracedecay-cli/tests/core_cli_suite/tool_daemon_test.rs
index fdfd7aa638..276401137f 100644
--- a/crates/tracedecay-cli/tests/core_cli_suite/tool_daemon_test.rs
+++ b/crates/tracedecay-cli/tests/core_cli_suite/tool_daemon_test.rs
@@ -434,8 +434,8 @@ fn assert_scope_set_probe_answered(output: &Output, context: &str) {
 /// project server, then reads
 /// that server's tool-call counter from its typed status. Until the full
 /// project server takes over from the core, the call and the status can land
-/// on different servers, so the pair is re-sent until the status server has
-/// counted a call.
+/// on different servers with independent counters. Wait for the status
+/// server's project session authority and a counted call after that handoff.
 fn wait_for_tool_status_server_tool_calls(home: &Path, project: &Path) -> u64 {
     let project_arg = project.to_string_lossy().to_string();
     let deadline = Instant::now() + CLI_ROUNDTRIP_TIMEOUT;
@@ -478,12 +478,18 @@ fn wait_for_tool_status_server_tool_calls(home: &Path, project: &Path) -> u64 {
         let tool_calls = payload["server"]["tool_calls"]
             .as_u64()
             .unwrap_or_else(|| panic!("typed status must carry its server counters: {payload}"));
-        if tool_calls >= 1 {
+        let project_sessions_mounted =
+            payload["server"]["retained_state_proxy"]["database_authorities"]["project_sessions"]
+                .as_bool()
+                .unwrap_or_else(|| {
+                    panic!("typed status must carry its mounted authorities: {payload}")
+                });
+        if project_sessions_mounted && tool_calls >= 1 {
             return tool_calls;
         }
         assert!(
             Instant::now() < deadline,
-            "the status server never counted a tool call before the CLI roundtrip deadline: {payload}"
+            "the full status server never mounted project sessions and counted a tool call before the CLI roundtrip deadline: {payload}"
         );
         std::thread::sleep(Duration::from_millis(100));
     }
@@ -2148,7 +2154,7 @@ fn daemon_reuses_project_engine_across_tool_clients() {
     let project = TempDir::new().unwrap();
     let home_path = canonical_existing_path(home.path());
     let project_path = canonical_existing_path(project.path());
-    init_project_with_cli(&home_path, &project_path);
+    init_committed_git_project_with_cli(&home_path, &project_path);
     let _daemon = spawn_tracedecay_daemon(&home_path);
 
     let first_tool_calls = wait_for_tool_status_server_tool_calls(&home_path, &project_path);
diff --git a/crates/tracedecay-cli/tests/host_journeys_suite/host_lifecycle_cli_acceptance/exact_restore.rs b/crates/tracedecay-cli/tests/host_journeys_suite/host_lifecycle_cli_acceptance/exact_restore.rs
index 4e3e1a64fb..35297ad53b 100644
--- a/crates/tracedecay-cli/tests/host_journeys_suite/host_lifecycle_cli_acceptance/exact_restore.rs
+++ b/crates/tracedecay-cli/tests/host_journeys_suite/host_lifecycle_cli_acceptance/exact_restore.rs
@@ -470,6 +470,15 @@ fn uninstall_after_an_upgrade_removes_tracedecay_residue_and_keeps_unprovable_st
         run_phase(&cli, id, &["install", "--agent", id], false);
     }
     strip_creation_records(&cli);
+    // Unproven directories must retain operator modes across update/uninstall,
+    // regardless of the umask that created them during the old installation.
+    for (relative, mode) in [(".cursor/plugins", 0o750), (".cursor/plugins/local", 0o710)] {
+        fs::set_permissions(
+            cli.home.path().join(relative),
+            fs::Permissions::from_mode(mode),
+        )
+        .unwrap();
+    }
     deposit_legacy_cursor_extension(&cli);
     let upgraded = home_snapshot(&cli);
     for residue in [
@@ -503,8 +512,8 @@ fn uninstall_after_an_upgrade_removes_tracedecay_residue_and_keeps_unprovable_st
                 "  + .cline/mcp.json: file 600 {}",
                 hex::encode(Sha256::digest(skeleton))
             ),
-            "  + .cursor/plugins: dir 755".to_string(),
-            "  + .cursor/plugins/local: dir 755".to_string(),
+            "  + .cursor/plugins: dir 750".to_string(),
+            "  + .cursor/plugins/local: dir 710".to_string(),
         ]
     );
 }
diff --git a/crates/tracedecay-cli/tests/work_loop_journey.rs b/crates/tracedecay-cli/tests/work_loop_journey.rs
index 8a138c445e..906fb5b155 100644
--- a/crates/tracedecay-cli/tests/work_loop_journey.rs
+++ b/crates/tracedecay-cli/tests/work_loop_journey.rs
@@ -376,13 +376,13 @@ impl ProductionDaemon {
         (status, parsed)
     }
 
-    /// Follows only the daemon's explicit same-request warming contract. A
-    /// terminal problem or malformed retry response is returned immediately,
+    /// Follows the daemon's explicit same-request warming and capacity retry
+    /// contract. A terminal problem or malformed retry response returns immediately,
     /// so callers still prove their exact success or refusal below.
     fn post_settled(&self, label: &str, route_path: &str, body: &Value) -> (u16, Value) {
         poll_until(label, || {
             let (status, answer) = self.post(route_path, body);
-            if status == 503
+            if matches!(status, 429 | 503)
                 && answer["kind"] == "problem"
                 && answer["value"]["problem"]["retryable"] == true
                 && answer["value"]["problem"]["retry_scope"] == "same_request"
diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/memory_tests.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/memory_tests.rs
index bca70e7f82..0ebe34ada8 100644
--- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/memory_tests.rs
+++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/memory_tests.rs
@@ -490,7 +490,7 @@ fn measured_rss_pressure_refuses_worker_admission_and_readmits_as_it_falls() {
                     resident_bytes: unreclaimable_bytes,
                     unreclaimable_bytes,
                     swapped_bytes: 0,
-                    cgroup_committed_bytes: None,
+                    cgroup_headroom_bytes: None,
                 })
         }),
     ));
@@ -577,7 +577,7 @@ fn a_refresh_after_cold_index_sizes_its_worker_slab_to_measured_headroom() {
                 resident_bytes: unreclaimable_bytes,
                 unreclaimable_bytes,
                 swapped_bytes: 0,
-                cgroup_committed_bytes: None,
+                cgroup_headroom_bytes: None,
             })
         }),
     ));
@@ -655,7 +655,7 @@ fn a_refresh_decodes_its_parent_before_planning_its_worker_slab() {
                 resident_bytes: 0,
                 unreclaimable_bytes: 0,
                 swapped_bytes: 0,
-                cgroup_committed_bytes: None,
+                cgroup_headroom_bytes: None,
             })
         }),
     ));
@@ -745,7 +745,7 @@ fn a_released_generation_decodes_again_only_once_its_bytes_fit_the_budget() {
                 resident_bytes: 0,
                 unreclaimable_bytes: 0,
                 swapped_bytes: 0,
-                cgroup_committed_bytes: None,
+                cgroup_headroom_bytes: None,
             })
         }),
     ));
@@ -829,7 +829,7 @@ fn a_decode_is_admitted_against_unreclaimable_bytes_not_clean_file_pages() {
         resident_bytes: GIB,
         unreclaimable_bytes: GIB,
         swapped_bytes: 0,
-        cgroup_committed_bytes: None,
+        cgroup_headroom_bytes: None,
     }));
     let sampled = Arc::clone(&view);
     let pressure = Arc::new(ResidentMemoryPressureV1::with_sampler(
@@ -853,7 +853,7 @@ fn a_decode_is_admitted_against_unreclaimable_bytes_not_clean_file_pages() {
         resident_bytes: high_watermark - 1,
         unreclaimable_bytes: 2 * GIB,
         swapped_bytes: 0,
-        cgroup_committed_bytes: None,
+        cgroup_headroom_bytes: None,
     };
     let decoded = scheduler
         .latest_complete()
@@ -879,7 +879,7 @@ fn a_decode_is_admitted_against_unreclaimable_bytes_not_clean_file_pages() {
         resident_bytes: high_watermark - 1,
         unreclaimable_bytes: high_watermark - 1,
         swapped_bytes: 0,
-        cgroup_committed_bytes: None,
+        cgroup_headroom_bytes: None,
     };
     assert!(matches!(
         scheduler.publication.load_active_shared(),
@@ -920,7 +920,7 @@ fn reconcile_stops_when_resident_memory_crosses_the_watermark() {
                 resident_bytes: unreclaimable_bytes,
                 unreclaimable_bytes,
                 swapped_bytes: 0,
-                cgroup_committed_bytes: None,
+                cgroup_headroom_bytes: None,
             })
         }),
     ));
diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/publication_store.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/publication_store.rs
index 7dd83962b2..cb62507fcc 100644
--- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/publication_store.rs
+++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/publication_store.rs
@@ -98,7 +98,7 @@ fn selected_graph_build_bound_reserves_transient_memory_and_retries_after_releas
                 resident_bytes: 0,
                 unreclaimable_bytes: 0,
                 swapped_bytes: 0,
-                cgroup_committed_bytes: None,
+                cgroup_headroom_bytes: None,
             })
         }),
     ));
diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/residency.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/residency.rs
index 114b5106f1..4cd725b7cb 100644
--- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/residency.rs
+++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/residency.rs
@@ -1492,7 +1492,7 @@ fn fixed_reading_authority() -> Arc<ProcessResidentMemoryV1> {
         resident_bytes: 256 * 1024 * 1024,
         unreclaimable_bytes: 256 * 1024 * 1024,
         swapped_bytes: 0,
-        cgroup_committed_bytes: None,
+        cgroup_headroom_bytes: None,
     };
     Arc::new(ProcessResidentMemoryV1::with_pressure(
         limit,
diff --git a/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs b/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs
index 9d057c1692..05bf0e812e 100644
--- a/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs
+++ b/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs
@@ -2207,7 +2207,7 @@ fn text_artifact_ceilings_reserve_through_process_resident_memory() {
                     resident_bytes: live_bytes,
                     unreclaimable_bytes: live_bytes,
                     swapped_bytes: 0,
-                    cgroup_committed_bytes: None,
+                    cgroup_headroom_bytes: None,
                 })
             }),
         ));
diff --git a/crates/tracedecay-daemon-service/src/invocation/dispatch.rs b/crates/tracedecay-daemon-service/src/invocation/dispatch.rs
index 02d486f67a..b6bc595321 100644
--- a/crates/tracedecay-daemon-service/src/invocation/dispatch.rs
+++ b/crates/tracedecay-daemon-service/src/invocation/dispatch.rs
@@ -1,6 +1,7 @@
 //! The daemon invocation dispatcher: `DaemonInvocationService::invoke`.
 
 use super::*;
+use tracedecay_contracts::CancellationSignal;
 use tracedecay_runtime_core::cancellation::CancellationToken;
 use tracedecay_runtime_core::path_safety::canonical_existing_identity;
 use tracedecay_tool_catalog::ApplicationSurfaceOperation;
@@ -785,13 +786,26 @@ impl DaemonInvocationService {
                 let Some(owner) = graph_tool_owner else {
                     return missing_registered_owner_problem(publication, request_id);
                 };
+                let signal = match self.request_cancellations.signal(&request_id, &request_cancellation) {
+                    Some(signal) => signal,
+                    None => match CancellationSignal::active(cancellation.token_id.as_str()) {
+                        Ok(signal) => signal,
+                        Err(_) => return DaemonInvocationResponse::problem(
+                            request_id,
+                            DaemonInvocationProblem::InvalidRequest,
+                        ),
+                    },
+                };
+                if let CancellationState::Cancelled { requested_at } = cancellation.state {
+                    signal.cancel(requested_at);
+                }
                 Box::pin(execute_graph_tool(
                     request_id,
                     owner,
                     surface_operation,
                     arguments,
                     deadline,
-                    cancellation,
+                    signal,
                     request_cancellation,
                 ))
                 .await
diff --git a/crates/tracedecay-daemon-service/src/invocation/graph_tool.rs b/crates/tracedecay-daemon-service/src/invocation/graph_tool.rs
index c215591889..ab7880061a 100644
--- a/crates/tracedecay-daemon-service/src/invocation/graph_tool.rs
+++ b/crates/tracedecay-daemon-service/src/invocation/graph_tool.rs
@@ -9,8 +9,7 @@ use std::sync::Arc;
 use serde_json::{Map, Value};
 use tracedecay_contracts::graph_tool::GraphToolCompletionV1;
 use tracedecay_contracts::{
-    ApplicationProblem, CancellationContext, CancellationSignal, CancellationState, Deadline,
-    RequestId, ResolvedScope, now_micros,
+    ApplicationProblem, CancellationSignal, Deadline, RequestId, ResolvedScope, now_micros,
 };
 use tracedecay_daemon_protocol::{
     DaemonInvocationOutcome, DaemonInvocationProblem, DaemonInvocationResponse,
@@ -92,21 +91,15 @@ pub(super) async fn execute_graph_tool(
     operation: ApplicationSurfaceOperation,
     arguments: Map<String, Value>,
     deadline: Deadline,
-    cancellation: CancellationContext,
+    signal: CancellationSignal,
     request_cancellation: tracedecay_runtime_core::cancellation::CancellationToken,
 ) -> DaemonInvocationResponse {
-    let (Ok(typed_request_id), Ok(signal)) = (
-        RequestId::new(request_id.clone()),
-        CancellationSignal::active(cancellation.token_id.as_str()),
-    ) else {
+    let Ok(typed_request_id) = RequestId::new(request_id.clone()) else {
         return DaemonInvocationResponse::problem(
             request_id,
             DaemonInvocationProblem::InvalidRequest,
         );
     };
-    if let CancellationState::Cancelled { requested_at } = &cancellation.state {
-        signal.cancel(*requested_at);
-    }
     let execution = owner.port.execute(GraphToolInvocationV1 {
         operation,
         arguments,
diff --git a/crates/tracedecay-daemon-service/src/request_cancellation.rs b/crates/tracedecay-daemon-service/src/request_cancellation.rs
index eee4ff1dc2..90e27ad813 100644
--- a/crates/tracedecay-daemon-service/src/request_cancellation.rs
+++ b/crates/tracedecay-daemon-service/src/request_cancellation.rs
@@ -12,6 +12,7 @@ use std::collections::BTreeMap;
 use std::sync::{Arc, Mutex, MutexGuard};
 use std::time::{Duration, Instant};
 
+use tracedecay_contracts::{CancellationSignal, clock::now_micros};
 use tracedecay_runtime_core::cancellation::CancellationToken;
 
 const PENDING_CAPACITY: usize = 1_024;
@@ -21,11 +22,17 @@ const COMPLETED_TTL: Duration = Duration::from_mins(1);
 
 #[derive(Default)]
 struct State {
-    active: BTreeMap<String, CancellationToken>,
+    active: BTreeMap<String, ActiveRequest>,
     pending: BTreeMap<String, Instant>,
     completed: BTreeMap<String, Instant>,
 }
 
+#[derive(Clone)]
+struct ActiveRequest {
+    token: CancellationToken,
+    signal: Option<CancellationSignal>,
+}
+
 /// The cancellation table of one invocation service generation.
 ///
 /// Clones share the same table. A [`Lease`] retains the table it was
@@ -55,6 +62,37 @@ impl RequestCancellationRegistryV1 {
     }
 
     pub fn register(&self, request_id: &str) -> Option<Lease> {
+        self.register_control(request_id, None)
+    }
+
+    /// Retains an in-process caller's live signal alongside its request lease.
+    /// Reconstructing it from its wire context would require an asynchronous
+    /// relay before a blocking worker could observe caller cancellation.
+    pub fn register_with_signal(
+        &self,
+        request_id: &str,
+        signal: CancellationSignal,
+    ) -> Option<Lease> {
+        self.register_control(request_id, Some(signal))
+    }
+
+    pub(crate) fn signal(
+        &self,
+        request_id: &str,
+        token: &CancellationToken,
+    ) -> Option<CancellationSignal> {
+        self.lock()
+            .active
+            .get(request_id)
+            .filter(|request| request.token.is_same_token(token))
+            .and_then(|request| request.signal.clone())
+    }
+
+    fn register_control(
+        &self,
+        request_id: &str,
+        signal: Option<CancellationSignal>,
+    ) -> Option<Lease> {
         let token = CancellationToken::for_application_request(request_id);
         let mut state = self.lock();
         expire_ephemeral(&mut state, Instant::now());
@@ -67,8 +105,17 @@ impl RequestCancellationRegistryV1 {
         state.completed.remove(request_id);
         if state.pending.remove(request_id).is_some() {
             token.cancel();
+            if let Some(signal) = &signal {
+                signal.cancel(now_micros());
+            }
         }
-        state.active.insert(request_id.to_owned(), token.clone());
+        state.active.insert(
+            request_id.to_owned(),
+            ActiveRequest {
+                token: token.clone(),
+                signal,
+            },
+        );
         Some(Lease {
             state: Arc::clone(&self.state),
             request_id: request_id.to_owned(),
@@ -81,9 +128,12 @@ impl RequestCancellationRegistryV1 {
         let mut state = self.lock();
         let now = Instant::now();
         expire_ephemeral(&mut state, now);
-        if let Some(token) = state.active.get(request_id).cloned() {
+        if let Some(request) = state.active.get(request_id).cloned() {
             drop(state);
-            token.cancel();
+            request.token.cancel();
+            if let Some(signal) = request.signal {
+                signal.cancel(now_micros());
+            }
             true
         } else if state.completed.contains_key(request_id) {
             true
@@ -128,7 +178,7 @@ impl Drop for Lease {
         if state
             .active
             .get(&self.request_id)
-            .is_some_and(|token| token.is_same_token(&self.token))
+            .is_some_and(|request| request.token.is_same_token(&self.token))
         {
             state.active.remove(&self.request_id);
             record_completed(&mut state, self.request_id.clone(), Instant::now());
@@ -141,6 +191,70 @@ mod tests {
     use std::time::{Duration, Instant};
 
     use super::{COMPLETED_CAPACITY, RequestCancellationRegistryV1, State, record_completed};
+    use tracedecay_contracts::{CancellationSignal, clock::now_micros};
+
+    #[test]
+    fn controlled_request_shares_live_cancellation_without_polling_a_relay() {
+        let registry = RequestCancellationRegistryV1::default();
+        let signal = CancellationSignal::active("cancel.controlled").unwrap();
+        let lease = registry
+            .register_with_signal("request.controlled", signal.clone())
+            .unwrap();
+        let worker = registry
+            .signal("request.controlled", &lease.token())
+            .unwrap();
+
+        assert!(signal.cancel(now_micros()));
+        assert!(worker.is_cancelled());
+
+        let foreign = RequestCancellationRegistryV1::default()
+            .register("request.controlled")
+            .unwrap();
+        assert!(
+            registry
+                .signal("request.controlled", &foreign.token())
+                .is_none()
+        );
+        drop(lease);
+        let retry = registry.register("request.controlled").unwrap();
+        assert!(
+            registry
+                .signal("request.controlled", &retry.token())
+                .is_none()
+        );
+        assert!(!retry.token().is_cancelled());
+    }
+
+    #[test]
+    fn daemon_cancel_reaches_the_controlled_signal_and_preserves_request_isolation() {
+        let registry = RequestCancellationRegistryV1::default();
+        let signal = CancellationSignal::active("cancel.controlled").unwrap();
+        let lease = registry
+            .register_with_signal("request.controlled", signal.clone())
+            .unwrap();
+        let other_signal = CancellationSignal::active("cancel.other").unwrap();
+        let other = registry
+            .register_with_signal("request.other", other_signal.clone())
+            .unwrap();
+
+        assert!(registry.cancel("request.controlled"));
+        assert!(signal.is_cancelled());
+        assert!(lease.token().is_cancelled());
+        assert!(!other_signal.is_cancelled());
+        assert!(!other.token().is_cancelled());
+    }
+
+    #[test]
+    fn pending_cancel_reaches_a_later_controlled_registration() {
+        let registry = RequestCancellationRegistryV1::default();
+        assert!(!registry.cancel("request.controlled"));
+        let signal = CancellationSignal::active("cancel.controlled").unwrap();
+        let lease = registry
+            .register_with_signal("request.controlled", signal.clone())
+            .unwrap();
+        assert!(signal.is_cancelled());
+        assert!(lease.token().is_cancelled());
+    }
 
     #[test]
     fn pre_registration_cancellation_is_retained_and_cleanup_is_exact() {
diff --git a/crates/tracedecay-global-db/src/tests/harness.rs b/crates/tracedecay-global-db/src/tests/harness.rs
index 6b9b56c1ad..19b49ac551 100644
--- a/crates/tracedecay-global-db/src/tests/harness.rs
+++ b/crates/tracedecay-global-db/src/tests/harness.rs
@@ -25,8 +25,8 @@ use tracedecay_rusqlite_runtime::repository::RepositoryWriterRuntimeSnapshot;
 #[cfg(test)]
 use tracedecay_store::{
     AnchoredObservationWrite, ObservationPersistOutcome, ObservationProjectionStore,
-    ObservationStore, ObservationWrite, build_observation_resolution_authorization_v1,
-    build_observation_retrieval_anchor,
+    ObservationStore, ObservationWrite, ProjectionPersistOutcome,
+    build_observation_resolution_authorization_v1, build_observation_retrieval_anchor,
 };
 
 pub(super) static TEST_RUNTIME_NONCE: AtomicU64 = AtomicU64::new(1);
@@ -1147,6 +1147,7 @@ pub(crate) async fn seed_projected_messages(
         .unwrap()
     });
     let mut seeded = Vec::with_capacity(indexes.len());
+    let mut writes = Vec::with_capacity(indexes.len());
     for index in indexes {
         let record_id = format!("record.audit-batch-{index}");
         let record = ObservationId::new(record_id.clone()).unwrap();
@@ -1221,17 +1222,39 @@ pub(crate) async fn seed_projected_messages(
         )
         .unwrap();
         let anchored = AnchoredObservationWrite::new(write, anchor.clone(), generation).unwrap();
-        assert!(matches!(
-            store.persist_observation(anchored).await.unwrap(),
-            ObservationPersistOutcome::Committed(_)
-        ));
-        store
-            .project_observation(observation.observation_id())
-            .await
-            .unwrap();
+        writes.push(anchored);
         expected_cursor = Some(next_cursor);
         seeded.push((observation, anchor));
     }
+    if !writes.is_empty() {
+        // Fixture size is the test input; admission and projection use the
+        // same bounded batch authorities as host ingestion.
+        let outcomes = store.persist_observations(writes).await.unwrap();
+        assert_eq!(outcomes.len(), seeded.len());
+        assert!(
+            outcomes.iter().all(|outcome| matches!(
+                outcome.outcome(),
+                ObservationPersistOutcome::Committed(_)
+            ))
+        );
+        let projected = store
+            .project_queued_observations(seeded.len())
+            .await
+            .unwrap()
+            .expect("registered stores support batched projection");
+        assert_eq!(projected.items.len(), seeded.len());
+        for (item, admitted) in projected.items.iter().zip(&outcomes) {
+            assert!(matches!(
+                &item.outcome,
+                ProjectionPersistOutcome::Projected(output) if output.output_count() == 1
+            ));
+            assert_eq!(
+                item.outcome.checkpoint().last_sequence(),
+                admitted.outcome().receipt().sequence(),
+                "fixture projection follows its admitted observations in order"
+            );
+        }
+    }
     seeded
 }
 
diff --git a/crates/tracedecay-maintenance/src/retention/orphan_stores/tests/collection.rs b/crates/tracedecay-maintenance/src/retention/orphan_stores/tests/collection.rs
index 73fd48b1e7..b01d65c921 100644
--- a/crates/tracedecay-maintenance/src/retention/orphan_stores/tests/collection.rs
+++ b/crates/tracedecay-maintenance/src/retention/orphan_stores/tests/collection.rs
@@ -611,15 +611,39 @@ async fn sweep_unregistered_stores_never_deletes_durable_memory_rows() {
     )
     .unwrap();
 
-    let report = sweep_unregistered_stores(&db, &profile_root, 7 * DAY, base, true)
-        .await
-        .unwrap();
-    assert_eq!(report.outcome.collected.len(), 0);
+    let cancellation = CancellationToken::new();
+    let report = sweep_unregistered_store_page(
+        &db,
+        &profile_root,
+        UnregisteredStoreSweepRequestV1 {
+            cursor: None,
+            limit: DEFAULT_UNREGISTERED_STORE_PAGE_LIMIT,
+            retention_secs: 7 * DAY,
+            now: base,
+            apply: true,
+            cancellation: &cancellation,
+            deadline: functional_sweep_deadline(),
+        },
+    )
+    .await
+    .unwrap();
+    assert_eq!(
+        report.completion,
+        UnregisteredSweepCompletionV1::Complete,
+        "functional durable-memory sweep was interrupted: {report:?}"
+    );
+    assert!(report.applied, "complete sweep was not applied: {report:?}");
+    assert!(
+        report.outcome.collected.is_empty(),
+        "durable memory store was collected: {report:?}"
+    );
+    assert_eq!(report.outcome.errors.len(), 1, "{report:?}");
     assert_eq!(
         report.outcome.errors[0].kind,
-        CollectionFailureKind::DurableDataProtected
+        CollectionFailureKind::DurableDataProtected,
+        "{report:?}"
     );
-    assert!(dir.exists());
+    assert!(dir.exists(), "durable memory store was removed: {report:?}");
 }
 
 /// Seeds `branches/feature.db` (carrying a durable memory row) and its WAL the
diff --git a/crates/tracedecay-mcp/src/handlers/hook_runtime/admission.rs b/crates/tracedecay-mcp/src/handlers/hook_runtime/admission.rs
index df2173ffc7..b5bd4023fb 100644
--- a/crates/tracedecay-mcp/src/handlers/hook_runtime/admission.rs
+++ b/crates/tracedecay-mcp/src/handlers/hook_runtime/admission.rs
@@ -478,9 +478,15 @@ const fn intent_gated_lifecycle(producer: tracedecay_domain::NativeHostIdentityV
     )
 }
 
-/// Mirrors the intent pass's own disabled check, so a project the pass would
-/// skip pays no lifecycle write inside the hook's synchronous budget.
-async fn context_scout_intent_enabled(cg: &TraceDecay) -> bool {
+/// Whether this host's lifecycle is enabled under the same policy used by
+/// admission and ledger replay. Claude omits lifecycle writes without intent.
+pub async fn hook_v2_native_lifecycle_enabled(
+    cg: &TraceDecay,
+    producer: tracedecay_domain::NativeHostIdentityV1,
+) -> bool {
+    if !intent_gated_lifecycle(producer) {
+        return true;
+    }
     let Ok(pinned) = cg.configuration_runtime().client().current().await else {
         return false;
     };
@@ -702,8 +708,7 @@ async fn admit_hook_v2_envelope_with_lifecycle_inner(
     if let Some(mount) = mount.as_ref()
         && let Some(native_lifecycle) = mount.lifecycle.as_ref()
     {
-        let intent_enabled =
-            intent_gated_lifecycle(envelope.producer) && context_scout_intent_enabled(cg).await;
+        let lifecycle_enabled = hook_v2_native_lifecycle_enabled(cg, envelope.producer).await;
         let Some(range) = hook_v2_lifecycle_range(envelope, receipt) else {
             return HookV2AdmissionOutcomeV1::Backpressured;
         };
@@ -716,7 +721,7 @@ async fn admit_hook_v2_envelope_with_lifecycle_inner(
             native_lifecycle,
             range,
             envelope.producer,
-            intent_enabled,
+            lifecycle_enabled,
         )
         .await
         {
diff --git a/crates/tracedecay-mcp/src/handlers/hook_runtime/mod.rs b/crates/tracedecay-mcp/src/handlers/hook_runtime/mod.rs
index 318e201f76..3ba5220125 100644
--- a/crates/tracedecay-mcp/src/handlers/hook_runtime/mod.rs
+++ b/crates/tracedecay-mcp/src/handlers/hook_runtime/mod.rs
@@ -24,7 +24,8 @@ mod test_support;
 
 pub use admission::{
     HookV2AdmissionLedgerUnavailable, HookV2AdmissionOutcomeV1, admit_hook_v2_envelope,
-    admit_hook_v2_replayed_envelope_with_lifecycle, hook_v2_pending_work_envelopes,
+    admit_hook_v2_replayed_envelope_with_lifecycle, hook_v2_native_lifecycle_enabled,
+    hook_v2_pending_work_envelopes,
 };
 pub use envelope::daemon_mint_hook_v2_file_id;
 pub use hermes::replay_projectless_hermes_host_admission;
diff --git a/crates/tracedecay-privacy/src/detector_kernel.rs b/crates/tracedecay-privacy/src/detector_kernel.rs
index c855935678..d0c34365a8 100644
--- a/crates/tracedecay-privacy/src/detector_kernel.rs
+++ b/crates/tracedecay-privacy/src/detector_kernel.rs
@@ -192,7 +192,24 @@ pub(crate) fn high_entropy_ranges(text: &str) -> Vec<Range<usize>> {
         let candidate = &text[start..end];
         let macos_temp_path = candidate.starts_with("/private/var/folders/")
             || candidate.starts_with("/var/folders/");
-        if macos_temp_path {
+        let uuid_path = !macos_temp_path
+            && candidate.starts_with('/')
+            && candidate.split('/').any(is_uuid_path_component);
+        let uuid_path_residual = uuid_path.then(|| {
+            candidate
+                .split('/')
+                .filter(|component| !is_uuid_path_component(component))
+                .collect::<Vec<_>>()
+                .join("/")
+        });
+        // UUID directory names are path identity, but the remaining path must
+        // still be scored together so slash-bearing credentials stay visible.
+        if uuid_path_residual
+            .as_deref()
+            .is_some_and(looks_high_entropy_token)
+        {
+            ranges.push(start..end);
+        } else if uuid_path || macos_temp_path {
             let mut component_start = 0usize;
             for component in candidate.split('/') {
                 let component_end = component_start + component.len();
@@ -209,6 +226,17 @@ pub(crate) fn high_entropy_ranges(text: &str) -> Vec<Range<usize>> {
     ranges
 }
 
+fn is_uuid_path_component(component: &str) -> bool {
+    component.len() == 36
+        && component.bytes().enumerate().all(|(index, byte)| {
+            if matches!(index, 8 | 13 | 18 | 23) {
+                byte == b'-'
+            } else {
+                byte.is_ascii_hexdigit()
+            }
+        })
+}
+
 fn is_lcm_payload_ref(candidate: &str) -> bool {
     candidate
         .strip_prefix("ref=")
@@ -405,9 +433,13 @@ mod tests {
 
     #[test]
     fn entropy_kernel_preserves_ordinary_macos_temp_paths() {
-        let path = "/private/var/folders/ab/cd0123456789abcdefghijklmnopqrst/T/";
-        assert!(looks_high_entropy_token(path));
-        assert!(high_entropy_ranges(path).is_empty());
+        for path in [
+            "/private/var/folders/ab/cd0123456789abcdefghijklmnopqrst/T/",
+            "/private/var/folders/ab/cd0123456789abcdefghijklmnopqrst/T/fe600511-e406-4142-b74a-cbf177c5f2cb/",
+        ] {
+            assert!(looks_high_entropy_token(path));
+            assert!(high_entropy_ranges(path).is_empty());
+        }
     }
 
     #[test]
@@ -428,4 +460,57 @@ mod tests {
         assert!(looks_high_entropy_token(secret));
         assert_eq!(high_entropy_ranges(secret), vec![0..secret.len()]);
     }
+
+    #[test]
+    fn entropy_kernel_preserves_absolute_paths_with_uuid_directories() {
+        let observed = "/run/nl/w/fe600511-e406-4142-b74a-cbf177c5f2cb/tmp/";
+        assert!(looks_high_entropy_token(observed));
+        for path in [
+            observed,
+            "/workspace/artifacts/fe600511-e406-4142-b74a-cbf177c5f2cb/project/src/",
+            "/workspace/FE600511-E406-4142-B74A-CBF177C5F2CB/source/",
+        ] {
+            assert!(high_entropy_ranges(path).is_empty(), "{path}");
+        }
+        let malformed = "/run/nl/w/fe600511-e406-4142-b74a-cbf177c5f2cg/tmp/";
+        assert_eq!(high_entropy_ranges(malformed), vec![0..malformed.len()]);
+    }
+
+    #[test]
+    fn entropy_kernel_redacts_secrets_inside_uuid_directory_paths() {
+        let prefix = "/run/nl/w/fe600511-e406-4142-b74a-cbf177c5f2cb/tmp/";
+        for secret in [
+            "Qm9vZ2llV29vZ2llMTIzNDU2Nzg5MGFiY2RlZmdoaWprbG1ub3A4OTc2NTQzMjE",
+            "AbCdEfGhIjKlMnOpQrStUvWxYz01234/56789aBcDeFgHiJkLmNoPqRsTuVwXy",
+        ] {
+            let path = format!("{prefix}{secret}");
+            assert_eq!(high_entropy_ranges(&path), vec![0..path.len()]);
+        }
+    }
+
+    #[test]
+    fn uuid_directory_paths_do_not_dilute_secret_components() {
+        let prefix = format!(
+            "/workspace/fe600511-e406-4142-b74a-cbf177c5f2cb/{}/",
+            "a".repeat(512)
+        );
+        let secret = "Qm9vZ2llV29vZ2llMTIzNDU2Nzg5MGFiY2RlZmdoaWprbG1ub3A4OTc2NTQzMjE";
+        let path = format!("{prefix}{secret}");
+        assert!(!looks_high_entropy_token(&path));
+        assert_eq!(high_entropy_ranges(&path), vec![prefix.len()..path.len()]);
+    }
+
+    #[test]
+    fn structured_metadata_keeps_uuid_paths_but_redacts_path_credentials() {
+        let prefix = "/run/nl/w/fe600511-e406-4142-b74a-cbf177c5f2cb/tmp/";
+        let credential = "ghp_KsY7QwT2mZ4bV9nR6cX1jH8pL3dG5fA0eUwQ";
+        let metadata = json!({
+            "path": format!("{prefix}.tmp4cknUf/project/src/lib.rs"),
+            "source": format!("{prefix}{credential}/source"),
+        });
+        let sanitized = crate::sanitize_provider_metadata_json(&metadata.to_string(), 4096)
+            .expect("structured path metadata");
+        assert_eq!(sanitized["path"], metadata["path"]);
+        assert!(!sanitized["source"].as_str().unwrap().contains(credential));
+    }
 }
diff --git a/crates/tracedecay-query/tests/search_quality_suite/candidate_producers.rs b/crates/tracedecay-query/tests/search_quality_suite/candidate_producers.rs
index 1535b0fc07..f8c901c809 100644
--- a/crates/tracedecay-query/tests/search_quality_suite/candidate_producers.rs
+++ b/crates/tracedecay-query/tests/search_quality_suite/candidate_producers.rs
@@ -1494,7 +1494,7 @@ fn concurrent_artifact_reads_are_bounded_and_keep_the_verified_file() {
                 resident_bytes: bytes,
                 unreclaimable_bytes: bytes,
                 swapped_bytes: 0,
-                cgroup_committed_bytes: None,
+                cgroup_headroom_bytes: None,
             })
         }),
     );
@@ -2685,12 +2685,13 @@ fn reader_refuses_historical_v10_writer_artifact_as_incompatible() {
 
     let directory = tempfile::tempdir().expect("private v10 reopen dir");
     let artifact_path = directory.path().join("lexical-artifact-v10.sqlite");
-    std::fs::copy(checked_in, &artifact_path).expect("copy historical v10 fixture");
+    let bytes = std::fs::read(checked_in).expect("read historical v10 fixture");
+    // Bazel runfiles are read-only; create an owned file without copying their mode.
+    std::fs::write(&artifact_path, &bytes).expect("write historical v10 fixture");
     drop(
         tracedecay_private_fs::make_private_file(&artifact_path)
             .expect("restore private-file protection for content-addressed open"),
     );
-    let bytes = std::fs::read(&artifact_path).expect("read historical v10 fixture");
     let file_size_bytes = u64::try_from(bytes.len()).expect("v10 fixture length");
     let digest = ManifestDigest::new(format!("sha256:{}", hex::encode(Sha256::digest(&bytes))))
         .expect("v10 fixture digest");
diff --git a/crates/tracedecay-runtime-core/src/resident_memory.rs b/crates/tracedecay-runtime-core/src/resident_memory.rs
index 7c8a632c17..34a9c445be 100644
--- a/crates/tracedecay-runtime-core/src/resident_memory.rs
+++ b/crates/tracedecay-runtime-core/src/resident_memory.rs
@@ -301,31 +301,24 @@ pub struct ProcessResidentSampleV1 {
     pub unreclaimable_bytes: u64,
     /// Anonymous pages the kernel moved to swap (`VmSwap`).
     pub swapped_bytes: u64,
-    /// Bytes the kernel charges toward a finite `memory.max`: `memory.current`
-    /// minus `inactive_file`, plus `memory.swap.current`, on the cgroup with
-    /// the tightest finite ceiling.
-    ///
-    /// `None` when no cgroup has a finite `memory.max`. An unlimited cgroup's
-    /// file cache is not a kill line, and inactive file pages are what the
-    /// kernel reclaims before it kills. Active file pages of a mapped store
-    /// stay in this figure because they are not dropped before the kill.
-    pub cgroup_committed_bytes: Option<u64>,
+    /// Remaining committed capacity across every finite ancestor `memory.max`.
+    /// Committed bytes include swap and exclude reclaimable inactive file pages.
+    /// `None` means no finite cgroup ceiling was observed.
+    pub cgroup_headroom_bytes: Option<u64>,
 }
 
 impl ProcessResidentSampleV1 {
-    /// Bytes admission compares with the watermark.
-    ///
-    /// The daemon's anonymous state is the floor, swapped pages included:
-    /// they are live heap the next touch faults back in, so a sample taken
-    /// while the kernel swaps under a cgroup ceiling must not read as room.
-    /// A finite cgroup ceiling also counts its committed working set, so a
-    /// build cannot be admitted while `memory.current` is already at the kill
-    /// line and only the anonymous subset sits under the watermark.
+    /// Occupancy against a process budget, constrained by shared cgroup room.
+    /// Peers consume shared capacity without spending this process's budget.
+    /// Swapped anonymous pages remain live process allocations.
     #[must_use]
-    pub fn admission_bytes(self) -> u64 {
+    pub fn admission_bytes(self, limit_bytes: u64) -> u64 {
         self.unreclaimable_bytes
             .saturating_add(self.swapped_bytes)
-            .max(self.cgroup_committed_bytes.unwrap_or(0))
+            .max(
+                self.cgroup_headroom_bytes
+                    .map_or(0, |room| limit_bytes.saturating_sub(room)),
+            )
     }
 }
 
@@ -349,7 +342,7 @@ fn process_resident_sample_from_status_v1(status: &str) -> Option<ProcessResiden
         resident_bytes: status_kib_field_bytes(status, "VmRSS")?,
         unreclaimable_bytes: anon.checked_add(shmem)?,
         swapped_bytes: status_kib_field_bytes(status, "VmSwap")?,
-        cgroup_committed_bytes: None,
+        cgroup_headroom_bytes: None,
     })
 }
 
@@ -364,24 +357,32 @@ fn memory_stat_field_bytes(stat: &str, field: &str) -> Option<u64> {
     })
 }
 
-/// Working set a finite `memory.max` will kill for: `memory.current` minus
-/// `inactive_file`, plus the cgroup's swapped anonymous pages
-/// (`memory.swap.current`), on the cgroup directory with the tightest finite
-/// ceiling.
-///
-/// `None` when every `memory.max` is absent or `max`. Counting `memory.current`
-/// on an unlimited cgroup treats the machine's page cache as a kill line and
-/// refuses work the kernel can reclaim.
+/// The smallest remaining committed capacity across all finite ancestors.
+/// A larger ancestor can have less room than a smaller child because it also
+/// contains peers. Inactive file pages are reclaimable; swapped heap is not.
 #[cfg(target_os = "linux")]
-fn cgroup_committed_bytes_v1(proc_self_cgroup: &Path, cgroup_root: &Path) -> Option<u64> {
+fn cgroup_headroom_bytes_v1(proc_self_cgroup: &Path, cgroup_root: &Path) -> Option<u64> {
     let mut directory = cgroup_v2_process_directory_v1(proc_self_cgroup, cgroup_root)?;
-    let mut chosen: Option<(std::path::PathBuf, u64)> = None;
+    let mut headroom = None;
     loop {
         if let Some(limit) = finite_cgroup_memory_value_v1(&directory.join("memory.max")) {
-            let tighter = chosen.as_ref().is_none_or(|(_, current)| limit < *current);
-            if tighter {
-                chosen = Some((directory.clone(), limit));
-            }
+            let current = std::fs::read_to_string(directory.join("memory.current"))
+                .ok()?
+                .trim()
+                .parse::<u64>()
+                .ok()?;
+            let inactive_file = std::fs::read_to_string(directory.join("memory.stat"))
+                .ok()
+                .and_then(|stat| memory_stat_field_bytes(&stat, "inactive_file"))
+                .unwrap_or(0);
+            let swapped = std::fs::read_to_string(directory.join("memory.swap.current"))
+                .ok()
+                .and_then(|value| value.trim().parse::<u64>().ok())
+                .unwrap_or(0);
+            let committed = current
+                .saturating_sub(inactive_file)
+                .saturating_add(swapped);
+            headroom = Some(tighten(headroom, limit.saturating_sub(committed)));
         }
         if directory == cgroup_root {
             break;
@@ -392,25 +393,7 @@ fn cgroup_committed_bytes_v1(proc_self_cgroup: &Path, cgroup_root: &Path) -> Opt
         }
         directory = parent.to_path_buf();
     }
-    let (directory, _) = chosen?;
-    let current = std::fs::read_to_string(directory.join("memory.current"))
-        .ok()?
-        .trim()
-        .parse::<u64>()
-        .ok()?;
-    let inactive_file = std::fs::read_to_string(directory.join("memory.stat"))
-        .ok()
-        .and_then(|stat| memory_stat_field_bytes(&stat, "inactive_file"))
-        .unwrap_or(0);
-    let swapped = std::fs::read_to_string(directory.join("memory.swap.current"))
-        .ok()
-        .and_then(|value| value.trim().parse::<u64>().ok())
-        .unwrap_or(0);
-    Some(
-        current
-            .saturating_sub(inactive_file)
-            .saturating_add(swapped),
-    )
+    headroom
 }
 
 /// Sample this process's resident set directly from the kernel.
@@ -439,8 +422,8 @@ pub fn sampled_process_resident_v1() -> Option<ProcessResidentSampleV1> {
         let mut sample = process_resident_sample_from_status_v1(
             &std::fs::read_to_string("/proc/self/status").ok()?,
         )?;
-        sample.cgroup_committed_bytes =
-            cgroup_committed_bytes_v1(Path::new(PROC_SELF_CGROUP_V1), Path::new(CGROUP_V2_ROOT_V1));
+        sample.cgroup_headroom_bytes =
+            cgroup_headroom_bytes_v1(Path::new(PROC_SELF_CGROUP_V1), Path::new(CGROUP_V2_ROOT_V1));
         Some(sample)
     }
     #[cfg(target_os = "windows")]
@@ -482,7 +465,7 @@ pub(crate) fn process_resident_sample_from_macos_vm_info_v1(
         resident_bytes: resident_size.max(unreclaimable_bytes),
         unreclaimable_bytes,
         swapped_bytes: 0,
-        cgroup_committed_bytes: None,
+        cgroup_headroom_bytes: None,
     })
 }
 
@@ -548,7 +531,7 @@ fn process_resident_sample_from_counters_v1() -> Option<ProcessResidentSampleV1>
         resident_bytes: counters.WorkingSetSize as u64,
         unreclaimable_bytes: counters.PrivateUsage as u64,
         swapped_bytes: 0,
-        cgroup_committed_bytes: None,
+        cgroup_headroom_bytes: None,
     })
 }
 
@@ -669,7 +652,7 @@ pub struct ResidentMemoryPressureRegistrationFailureV1;
 /// The measured side of the memory accounting loop.
 ///
 /// One dedicated reader samples the process (`/proc/self/status` on Linux)
-/// and feeds this cell the unreclaimable bytes. Admission re-measures through the same sampler;
+/// and its shared cgroup room. Admission re-measures through the same sampler;
 /// there is no second parser or publisher.
 ///
 /// Every request refused for memory, by any authority on this cell, waits on
@@ -680,8 +663,8 @@ pub struct ResidentMemoryPressureV1 {
     limit_bytes: NonZeroU64,
     high_watermark_bytes: u64,
     low_watermark_bytes: u64,
-    observed_bytes: AtomicU64,
-    observed: AtomicBool,
+    /// Retain kernel measurements so narrower ceilings can derive their own view.
+    observation: Mutex<Option<ProcessResidentSampleV1>>,
     over_budget: AtomicBool,
     headroom: watch::Sender<u64>,
     /// Authorities holding a refused request, settled on every observation.
@@ -754,8 +737,7 @@ impl ResidentMemoryPressureV1 {
             limit_bytes,
             high_watermark_bytes,
             low_watermark_bytes,
-            observed_bytes: AtomicU64::new(0),
-            observed: AtomicBool::new(false),
+            observation: Mutex::new(None),
             over_budget: AtomicBool::new(false),
             headroom: watch::channel(0).0,
             waiting: Mutex::new(Vec::new()),
@@ -773,10 +755,7 @@ impl ResidentMemoryPressureV1 {
         &self,
     ) -> Option<(ProcessResidentSampleV1, ResidentMemoryPressureStateV1)> {
         let sample = (self.sampler)()?;
-        Some((
-            sample,
-            self.publish_observed_resident_bytes(sample.admission_bytes()),
-        ))
+        Some((sample, self.publish_sample(sample)))
     }
 
     /// Publish a fresh admission sample without running pressure reclaimers.
@@ -810,7 +789,7 @@ impl ResidentMemoryPressureV1 {
             Ordering::Release,
         );
         if let Some(sample) = sample {
-            self.publish_observation(sample.admission_bytes());
+            self.publish_observation(sample);
         }
         self.state()
     }
@@ -818,7 +797,7 @@ impl ResidentMemoryPressureV1 {
     /// Refresh pressure after reclaim without running reclaimers recursively.
     fn resample_after_reclaim(&self) -> Option<ResidentMemoryPressureStateV1> {
         let sample = (self.sampler)()?;
-        self.publish_observation(sample.admission_bytes());
+        self.publish_observation(sample);
         Some(self.state())
     }
 
@@ -832,7 +811,7 @@ impl ResidentMemoryPressureV1 {
         self.sample_and_publish().map_or(0, |(sample, state)| {
             state
                 .observed_bytes()
-                .unwrap_or_else(|| sample.admission_bytes())
+                .unwrap_or_else(|| sample.admission_bytes(self.limit_bytes()))
         })
     }
 
@@ -863,16 +842,36 @@ impl ResidentMemoryPressureV1 {
         &self,
         observed_bytes: u64,
     ) -> ResidentMemoryPressureStateV1 {
-        self.publish_observation(observed_bytes);
+        self.publish_sample(ProcessResidentSampleV1 {
+            resident_bytes: observed_bytes,
+            unreclaimable_bytes: observed_bytes,
+            swapped_bytes: 0,
+            cgroup_headroom_bytes: None,
+        })
+    }
+
+    fn publish_sample(&self, sample: ProcessResidentSampleV1) -> ResidentMemoryPressureStateV1 {
+        let observed_bytes = sample.admission_bytes(self.limit_bytes());
+        self.publish_observation(sample);
         if observed_bytes >= self.high_watermark_bytes {
             self.run_pressure_reclaimers(observed_bytes);
         }
         self.state()
     }
 
-    fn publish_observation(&self, observed_bytes: u64) {
-        self.observed_bytes.store(observed_bytes, Ordering::Release);
-        self.observed.store(true, Ordering::Release);
+    fn observed_bytes_for_limit(&self, limit_bytes: u64) -> Option<u64> {
+        self.observation
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .map(|sample| sample.admission_bytes(limit_bytes))
+    }
+
+    fn publish_observation(&self, sample: ProcessResidentSampleV1) {
+        let observed_bytes = sample.admission_bytes(self.limit_bytes());
+        *self
+            .observation
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(sample);
         if observed_bytes >= self.high_watermark_bytes {
             self.over_budget.store(true, Ordering::Release);
         } else if observed_bytes <= self.low_watermark_bytes
@@ -918,7 +917,7 @@ impl ResidentMemoryPressureV1 {
             authority.upgrade().is_some_and(|authority| {
                 let mut state = authority.lock_state();
                 settled |= authority.settle_waiter(&mut state);
-                state.waiting_level.is_some()
+                !state.waiting_requests.is_empty()
             })
         });
         self.any_waiting
@@ -931,10 +930,9 @@ impl ResidentMemoryPressureV1 {
 
     #[must_use]
     pub fn state(&self) -> ResidentMemoryPressureStateV1 {
-        if !self.observed.load(Ordering::Acquire) {
+        let Some(observed_bytes) = self.observed_bytes_for_limit(self.limit_bytes()) else {
             return ResidentMemoryPressureStateV1::Unobserved;
-        }
-        let observed_bytes = self.observed_bytes.load(Ordering::Acquire);
+        };
         if self.over_budget.load(Ordering::Acquire) {
             return ResidentMemoryPressureStateV1::OverBudget {
                 observed_bytes,
@@ -1412,9 +1410,9 @@ struct ResidentMemoryStateV1 {
     used_bytes: u64,
     charges: BTreeMap<ResidentMemoryKeyV1, u64>,
     process_shared_charges: BTreeMap<ResidentMemoryComponentIdV1, u64>,
-    /// The highest occupancy (ledger or measured, whichever is larger) at
-    /// which some refused request fits; `None` while nothing waits.
-    waiting_level: Option<u64>,
+    /// The smallest refused request per ceiling is the first that can fit.
+    /// Shared headroom must be evaluated against each distinct ceiling.
+    waiting_requests: BTreeMap<u64, u64>,
 }
 
 /// The single process ceiling. Callers share one pointer-identical `Arc`.
@@ -1468,12 +1466,16 @@ impl ProcessResidentMemoryV1 {
     /// measured process. The pressure cell's headroom epoch advances once it
     /// would, and every refusal this authority makes registers the same way.
     pub fn wait_for_headroom(self: &Arc<Self>, bytes: u64, ceiling: u64) {
-        let Some(level) = ceiling.checked_sub(bytes) else {
+        if bytes > ceiling {
             return;
-        };
+        }
         {
             let mut state = self.lock_state();
-            state.waiting_level = Some(state.waiting_level.map_or(level, |held| held.max(level)));
+            state
+                .waiting_requests
+                .entry(ceiling)
+                .and_modify(|held| *held = (*held).min(bytes))
+                .or_insert(bytes);
         }
         self.pressure.register_waiting(self);
     }
@@ -1481,15 +1483,18 @@ impl ProcessResidentMemoryV1 {
     /// Clear the waiters once the most satisfiable one fits; each still
     /// refused registers again on its retry.
     fn settle_waiter(&self, state: &mut ResidentMemoryStateV1) -> bool {
-        let Some(level) = state.waiting_level else {
-            return false;
-        };
-        let pressure = self.pressure.state();
-        let occupied = state.used_bytes.max(pressure.observed_bytes().unwrap_or(0));
-        if pressure.is_over_budget() || occupied > level {
+        if self.pressure.state().is_over_budget()
+            || !state.waiting_requests.iter().any(|(ceiling, bytes)| {
+                let observed = self
+                    .pressure
+                    .observed_bytes_for_limit(*ceiling)
+                    .unwrap_or(0);
+                state.used_bytes.max(observed) <= ceiling.saturating_sub(*bytes)
+            })
+        {
             return false;
         }
-        state.waiting_level = None;
+        state.waiting_requests.clear();
         true
     }
 
@@ -1590,7 +1595,12 @@ impl ProcessResidentMemoryV1 {
     /// this authority's limit and the measurement to the pressure cell's, as
     /// [`Self::reserve`] holds them.
     pub fn headroom_below(&self, ceiling: u64) -> ResidentMemoryHeadroomV1 {
-        let observed_bytes = self.pressure.measure_admission_bytes();
+        self.pressure.measure_admission_bytes();
+        let measured_ceiling = ceiling.min(self.pressure.limit_bytes());
+        let observed_bytes = self
+            .pressure
+            .observed_bytes_for_limit(measured_ceiling)
+            .unwrap_or(0);
         let used_bytes = self.lock_state().used_bytes;
         let available_bytes = if self.pressure.state().is_over_budget() {
             0
@@ -1598,9 +1608,7 @@ impl ProcessResidentMemoryV1 {
             let ledger_room = ceiling
                 .min(self.limit_bytes.get())
                 .saturating_sub(used_bytes);
-            let measured_room = ceiling
-                .min(self.pressure.limit_bytes())
-                .saturating_sub(observed_bytes);
+            let measured_room = measured_ceiling.saturating_sub(observed_bytes);
             ledger_room.min(measured_room)
         };
         ResidentMemoryHeadroomV1 {
diff --git a/crates/tracedecay-runtime-core/src/resident_memory/tests.rs b/crates/tracedecay-runtime-core/src/resident_memory/tests.rs
index d9abeffb82..557293b4ec 100644
--- a/crates/tracedecay-runtime-core/src/resident_memory/tests.rs
+++ b/crates/tracedecay-runtime-core/src/resident_memory/tests.rs
@@ -481,7 +481,7 @@ fn additional_reservations_share_identity_but_charge_and_release_independently()
                     resident_bytes: 0,
                     unreclaimable_bytes: 0,
                     swapped_bytes: 0,
-                    cgroup_committed_bytes: None,
+                    cgroup_headroom_bytes: None,
                 })
             }),
         )),
@@ -958,7 +958,7 @@ fn post_reclaim_observation_replaces_pressure_state_without_reentering_reclaimer
                 resident_bytes: after_reclaim,
                 unreclaimable_bytes: after_reclaim,
                 swapped_bytes: 0,
-                cgroup_committed_bytes: None,
+                cgroup_headroom_bytes: None,
             })
         }),
     ));
@@ -1071,7 +1071,7 @@ fn allocator_trim_readmits_growth_when_memory_recovers_inside_checkpoint_throttl
                 resident_bytes: observed,
                 unreclaimable_bytes: observed,
                 swapped_bytes: 0,
-                cgroup_committed_bytes: None,
+                cgroup_headroom_bytes: None,
             })
         }),
     ));
@@ -1197,11 +1197,11 @@ fn process_status_splits_clean_file_pages_from_unreclaimable_bytes() {
             resident_bytes: 3_355_444 * 1024,
             unreclaimable_bytes: 2_548_172 * 1024,
             swapped_bytes: 491_520 * 1024,
-            cgroup_committed_bytes: None,
+            cgroup_headroom_bytes: None,
         })
     );
     assert_eq!(
-        sample.map(super::ProcessResidentSampleV1::admission_bytes),
+        sample.map(|sample| sample.admission_bytes(u64::MAX)),
         Some((2_548_172 + 491_520) * 1024),
         "swapped anonymous pages are the daemon's heap and count toward admission"
     );
@@ -1232,7 +1232,7 @@ fn macos_phys_footprint_is_the_admission_charge() {
     assert_eq!(sample.resident_bytes, 6_400 * 1024 * 1024);
     assert_eq!(sample.unreclaimable_bytes, 5_200 * 1024 * 1024);
     assert_eq!(
-        sample.admission_bytes(),
+        sample.admission_bytes(u64::MAX),
         5_200 * 1024 * 1024,
         "jetsam charges phys_footprint, not the larger resident_size"
     );
@@ -1244,12 +1244,12 @@ fn macos_phys_footprint_is_the_admission_charge() {
     let internal_only = super::process_resident_sample_from_macos_vm_info_v1(8, 0, 5)
         .expect("internal is the fallback when footprint is missing");
     assert_eq!(internal_only.unreclaimable_bytes, 5);
-    assert_eq!(internal_only.admission_bytes(), 5);
+    assert_eq!(internal_only.admission_bytes(u64::MAX), 5);
 }
 
 #[cfg(target_os = "linux")]
 #[test]
-fn cgroup_committed_bytes_refuse_growth_that_unreclaimable_bytes_would_admit() {
+fn cgroup_headroom_bytes_refuse_growth_that_unreclaimable_bytes_would_admit() {
     let limit_bytes = 100 * 1024 * 1024;
     let limit = bytes(limit_bytes);
     let decide = |memory_max: &str, inactive_file: u64| {
@@ -1265,12 +1265,12 @@ fn cgroup_committed_bytes_refuse_growth_that_unreclaimable_bytes_would_admit() {
             format!("inactive_file {inactive_file}\n"),
         )
         .expect("stat");
-        let committed = super::cgroup_committed_bytes_v1(&proc_self_cgroup, &cgroup_root);
+        let committed = super::cgroup_headroom_bytes_v1(&proc_self_cgroup, &cgroup_root);
         let sample = Arc::new(Mutex::new(super::ProcessResidentSampleV1 {
             resident_bytes: 10,
             unreclaimable_bytes: 10,
             swapped_bytes: 0,
-            cgroup_committed_bytes: committed,
+            cgroup_headroom_bytes: committed,
         }));
         let sampled = Arc::clone(&sample);
         let pressure = Arc::new(ResidentMemoryPressureV1::with_sampler(
@@ -1304,6 +1304,90 @@ fn cgroup_committed_bytes_refuse_growth_that_unreclaimable_bytes_would_admit() {
         decide("max\n", 0).is_ok(),
         "an unlimited cgroup does not treat memory.current as a kill line"
     );
+    assert!(
+        decide(&format!("{}\n", 10 * limit_bytes), 0).is_ok(),
+        "a larger shared cgroup with room does not spend the daemon's smaller process budget"
+    );
+}
+
+#[cfg(target_os = "linux")]
+#[test]
+fn cgroup_headroom_respects_a_larger_but_fuller_ancestor() {
+    const MIB: u64 = 1024 * 1024;
+    let (_directory, membership, root) = cgroup_fixture(
+        Some("0::/trace.slice/daemon.scope\n"),
+        Some(&format!("{}\n", 400 * MIB)),
+        None,
+    );
+    let leaf = root.join("trace.slice/daemon.scope");
+    let parent = root.join("trace.slice");
+    fs::write(leaf.join("memory.current"), (20 * MIB).to_string()).unwrap();
+    fs::write(parent.join("memory.max"), (1000 * MIB).to_string()).unwrap();
+    fs::write(parent.join("memory.current"), (990 * MIB).to_string()).unwrap();
+    assert_eq!(
+        super::cgroup_headroom_bytes_v1(&membership, &root),
+        Some(10 * MIB)
+    );
+    fs::write(parent.join("memory.current"), (900 * MIB).to_string()).unwrap();
+    assert_eq!(
+        super::cgroup_headroom_bytes_v1(&membership, &root),
+        Some(100 * MIB)
+    );
+}
+
+#[test]
+fn narrowed_budgets_and_waiters_share_cgroup_capacity_without_charging_peers() {
+    const MIB: u64 = 1024 * 1024;
+    let room = Arc::new(AtomicU64::new(560 * MIB));
+    let sampled = Arc::clone(&room);
+    let pressure = Arc::new(ResidentMemoryPressureV1::with_sampler(
+        bytes(800 * MIB),
+        Arc::new(move || {
+            Some(ProcessResidentSampleV1 {
+                resident_bytes: 10 * MIB,
+                unreclaimable_bytes: 10 * MIB,
+                swapped_bytes: 0,
+                cgroup_headroom_bytes: Some(sampled.load(Ordering::SeqCst)),
+            })
+        }),
+    ));
+    let authority = Arc::new(ProcessResidentMemoryV1::with_pressure(
+        bytes(80 * MIB),
+        Arc::clone(&pressure),
+    ));
+    assert_eq!(authority.headroom_below(80 * MIB).available_bytes, 70 * MIB);
+
+    room.store(10 * MIB, Ordering::SeqCst);
+    assert_eq!(authority.headroom_below(80 * MIB).available_bytes, 0);
+    let epoch = pressure.subscribe_headroom();
+    let before = *epoch.borrow();
+    authority.wait_for_headroom(16 * MIB, 80 * MIB);
+    authority.wait_for_headroom(40 * MIB, 100 * MIB);
+    room.store(20 * MIB, Ordering::SeqCst);
+    pressure.sample_and_publish();
+    // The shared process latch remains raised until its low watermark clears.
+    assert_eq!(*epoch.borrow(), before);
+    room.store(200 * MIB, Ordering::SeqCst);
+    pressure.sample_and_publish();
+    assert!(*epoch.borrow() > before);
+    assert_eq!(authority.headroom_below(80 * MIB).available_bytes, 70 * MIB);
+
+    // With a nominal process latch, a request still waits for its own room.
+    room.store(100 * MIB, Ordering::SeqCst);
+    pressure.sample_and_publish();
+    let before = *epoch.borrow();
+    authority.wait_for_headroom(200 * MIB, 400 * MIB);
+    authority.wait_for_headroom(160 * MIB, 400 * MIB);
+    authority.wait_for_headroom(240 * MIB, 400 * MIB);
+    authority.wait_for_headroom(200 * MIB, 300 * MIB);
+    pressure.sample_and_publish();
+    assert_eq!(*epoch.borrow(), before);
+    room.store(180 * MIB, Ordering::SeqCst);
+    pressure.sample_and_publish();
+    assert!(
+        *epoch.borrow() > before,
+        "the satisfiable request wakes the common epoch"
+    );
 }
 
 /// A daemon the kernel swaps under its cgroup ceiling has small resident
@@ -1367,8 +1451,8 @@ fn admission_counts_swapped_anonymous_pages() {
     fs::write(cgroup.join("memory.stat"), "inactive_file 0\n").expect("stat");
     fs::write(cgroup.join("memory.swap.current"), format!("{resident}\n")).expect("swap");
     assert_eq!(
-        super::cgroup_committed_bytes_v1(&proc_self_cgroup, &cgroup_root),
-        Some(2 * resident),
+        super::cgroup_headroom_bytes_v1(&proc_self_cgroup, &cgroup_root),
+        Some(limit_bytes - 2 * resident),
         "the cgroup's swapped pages are committed to it"
     );
 }
@@ -1392,7 +1476,7 @@ fn checkpoints_read_the_process_once_per_interval_and_then_see_growth() {
                 resident_bytes: resident,
                 unreclaimable_bytes: resident,
                 swapped_bytes: 0,
-                cgroup_committed_bytes: None,
+                cgroup_headroom_bytes: None,
             })
         }),
     );
@@ -1437,7 +1521,7 @@ fn headroom_is_the_ceiling_less_the_larger_of_ledger_and_measurement() {
                     resident_bytes: observed,
                     unreclaimable_bytes: observed,
                     swapped_bytes: 0,
-                    cgroup_committed_bytes: None,
+                    cgroup_headroom_bytes: None,
                 })
             }),
         )),
@@ -1493,7 +1577,7 @@ fn allocator_reclaim_preserves_cgroup_pressure_until_sample_recovers() {
         resident_bytes: 1,
         unreclaimable_bytes: 1,
         swapped_bytes: 0,
-        cgroup_committed_bytes: Some(limit.get()),
+        cgroup_headroom_bytes: Some(0),
     }));
     let sampled = Arc::clone(&sample);
     let pressure = Arc::new(ResidentMemoryPressureV1::with_sampler(
@@ -1520,7 +1604,7 @@ fn allocator_reclaim_preserves_cgroup_pressure_until_sample_recovers() {
         Err(ResidentMemoryAdmissionFailureV1::ObservedOverBudget { .. })
     ));
 
-    sample.lock().expect("sample").cgroup_committed_bytes = Some(1);
+    sample.lock().expect("sample").cgroup_headroom_bytes = Some(limit.get() - 1);
     assert_eq!(pressure.measure_admission_bytes(), 1);
     assert!(matches!(
         pressure.state(),
diff --git a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/registry.rs b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/registry.rs
index a1f3bbc8e7..7a58d40ece 100644
--- a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/registry.rs
+++ b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/registry.rs
@@ -120,6 +120,28 @@ fn cancel_supervisor(
 }
 
 impl SessionTemporalRefreshSchedulerEntry {
+    fn wake_with_history(&self, history: Option<SharedSessionHistoricalIngestor>) {
+        let Some(history) = history else {
+            self.wake.wake();
+            return;
+        };
+        let mut retained = self.history.write().unwrap_or_else(PoisonError::into_inner);
+        let installed = retained.is_none();
+        if installed {
+            *retained = Some(history);
+            self.state.mark_history_pending();
+        }
+        let blocked = !installed && self.state.history_is_blocked();
+        drop(retained);
+        if !blocked {
+            // Publish both kinds of work before notifying the worker.
+            self.state.requeue_projection();
+            self.state.wake_history();
+        } else {
+            self.wake.wake();
+        }
+    }
+
     fn cancel(&self) {
         cancel_supervisor(&self.state, &self.history);
     }
@@ -340,6 +362,10 @@ impl SessionTemporalRefreshSchedulerRegistry {
         history: Option<SharedSessionHistoricalIngestor>,
     ) -> SessionTemporalRefreshSchedulerEntry {
         let state = Arc::new(SessionTemporalRefreshWakeState::default());
+        if history.is_some() {
+            state.mark_history_pending();
+            state.wake_history();
+        }
         let history = Arc::new(std::sync::RwLock::new(history));
         let wake = route.unwrap_or_else(|| state.handle());
         wake.bind(&state);
@@ -583,6 +609,16 @@ impl SessionTemporalRefreshSchedulerRegistry {
         &self,
         owner: StoreOwnerKey,
         database: RegisteredGlobalDbLeaseV1,
+    ) -> SessionTemporalRefreshWake {
+        self.ensure_project_with_retained_history(owner, database, None)
+            .await
+    }
+
+    async fn ensure_project_with_retained_history(
+        &self,
+        owner: StoreOwnerKey,
+        database: RegisteredGlobalDbLeaseV1,
+        history: Option<SharedSessionHistoricalIngestor>,
     ) -> SessionTemporalRefreshWake {
         if self.shutting_down.load(Ordering::Acquire) {
             return inert_session_temporal_refresh_wake();
@@ -613,7 +649,8 @@ impl SessionTemporalRefreshSchedulerRegistry {
                 .history
                 .read()
                 .unwrap_or_else(PoisonError::into_inner)
-                .clone();
+                .clone()
+                .or(history);
             self.retire(vec![(retired_owner.clone(), finished)])
                 .join()
                 .await;
@@ -631,11 +668,12 @@ impl SessionTemporalRefreshSchedulerRegistry {
             return wake;
         }
         if let Some(entry) = project.get(&owner) {
-            entry.wake.wake();
+            // Install history before any wake can start a summary-only pass.
+            entry.wake_with_history(history);
             return entry.wake.clone();
         }
         self.settle_retired(&retired_owner).await;
-        let entry = self.spawn_entry(database, None, None);
+        let entry = self.spawn_entry(database, None, history);
         let wake = entry.wake.clone();
         project.insert(owner, entry);
         wake
@@ -645,6 +683,16 @@ impl SessionTemporalRefreshSchedulerRegistry {
         &self,
         database_path: std::path::PathBuf,
         database: RegisteredGlobalDbLeaseV1,
+    ) -> SessionTemporalRefreshWake {
+        self.ensure_profile_with_retained_history(database_path, database, None)
+            .await
+    }
+
+    async fn ensure_profile_with_retained_history(
+        &self,
+        database_path: std::path::PathBuf,
+        database: RegisteredGlobalDbLeaseV1,
+        history: Option<SharedSessionHistoricalIngestor>,
     ) -> SessionTemporalRefreshWake {
         if self.shutting_down.load(Ordering::Acquire) {
             return inert_session_temporal_refresh_wake();
@@ -665,7 +713,8 @@ impl SessionTemporalRefreshSchedulerRegistry {
                 .history
                 .read()
                 .unwrap_or_else(PoisonError::into_inner)
-                .clone();
+                .clone()
+                .or(history);
             self.retire(vec![(retired_owner.clone(), finished)])
                 .join()
                 .await;
@@ -679,11 +728,11 @@ impl SessionTemporalRefreshSchedulerRegistry {
             return wake;
         }
         if let Some(entry) = profile.get(&database_path) {
-            entry.wake.wake();
+            entry.wake_with_history(history);
             return entry.wake.clone();
         }
         self.settle_retired(&retired_owner).await;
-        let entry = self.spawn_entry(database, None, None);
+        let entry = self.spawn_entry(database, None, history);
         let wake = entry.wake.clone();
         profile.insert(database_path, entry);
         wake
@@ -695,29 +744,9 @@ impl SessionTemporalRefreshSchedulerRegistry {
         database: RegisteredGlobalDbLeaseV1,
         history: SharedSessionHistoricalIngestor,
     ) -> SessionTemporalRefreshWake {
-        let wake = self.ensure_project(owner.clone(), database).await;
-        if session_ingest_disabled() {
-            // Dev/profiling switch: never install a history ingestor, so the
-            // project-open catch-up path cannot ingest either.
-            return wake;
-        }
-        if let Some(entry) = self.project.lock().await.get(&owner) {
-            let mut retained = entry
-                .history
-                .write()
-                .unwrap_or_else(PoisonError::into_inner);
-            let installed = retained.is_none();
-            if installed {
-                *retained = Some(history);
-                entry.state.mark_history_pending();
-            }
-            let blocked = !installed && entry.state.history_is_blocked();
-            drop(retained);
-            if !blocked {
-                entry.state.wake_history();
-            }
-        }
-        wake
+        let history = (!session_ingest_disabled()).then_some(history);
+        self.ensure_project_with_retained_history(owner, database, history)
+            .await
     }
 
     pub async fn ensure_profile_with_history(
@@ -726,29 +755,9 @@ impl SessionTemporalRefreshSchedulerRegistry {
         database: RegisteredGlobalDbLeaseV1,
         history: SharedSessionHistoricalIngestor,
     ) -> SessionTemporalRefreshWake {
-        let wake = self.ensure_profile(database_path.clone(), database).await;
-        if session_ingest_disabled() {
-            // Dev/profiling switch: never install a history ingestor, so the
-            // project-open catch-up path cannot ingest either.
-            return wake;
-        }
-        if let Some(entry) = self.profile.lock().await.get(&database_path) {
-            let mut retained = entry
-                .history
-                .write()
-                .unwrap_or_else(PoisonError::into_inner);
-            let installed = retained.is_none();
-            if installed {
-                *retained = Some(history);
-                entry.state.mark_history_pending();
-            }
-            let blocked = !installed && entry.state.history_is_blocked();
-            drop(retained);
-            if !blocked {
-                entry.state.wake_history();
-            }
-        }
-        wake
+        let history = (!session_ingest_disabled()).then_some(history);
+        self.ensure_profile_with_retained_history(database_path, database, history)
+            .await
     }
 
     pub async fn rekey_project(
diff --git a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/registry_tests.rs b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/registry_tests.rs
index 2d08dd1da0..ad4495ee87 100644
--- a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/registry_tests.rs
+++ b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/registry_tests.rs
@@ -6,6 +6,9 @@ use std::time::Duration;
 use futures_util::FutureExt;
 use tracedecay_global_db::tests::harness::RegisteredGlobalDbHarness;
 use tracedecay_runtime_core::native_runtime;
+use tracedecay_sessions::serving::{
+    SessionProjectionServingState, SessionProjectionServingStatusPort, SessionProjectionStaleReason,
+};
 
 use super::super::history::{
     SessionHistoricalCapacityRelease, SessionHistoricalIngestOutcome, SessionHistoricalIngestPass,
@@ -15,6 +18,44 @@ use super::super::wake::SessionTemporalRefreshWakeState;
 use super::SessionTemporalRefreshSchedulerRegistry;
 use crate::StoreOwnerKey;
 
+/// A retained ingestor must own the first pass, even with every shared permit
+/// occupied. A summary-only first pass would queue ahead of its history and
+/// leave a caller waiting for the history refusal while holding those permits.
+#[tokio::test]
+async fn retained_history_precedes_summary_admission_on_worker_start() {
+    let harness = RegisteredGlobalDbHarness::open("refresh-history-before-summary").await;
+    let registry = SessionTemporalRefreshSchedulerRegistry::default();
+    let admission = registry.historical_ingest_admission();
+    let held = admission
+        .try_acquire_many(u32::try_from(admission.available_permits()).unwrap())
+        .unwrap();
+    let history = Arc::new(HeldHistory::default());
+    let path = harness.registered.db_path().to_path_buf();
+    let entry = registry.spawn_entry(harness.registered.clone(), None, Some(history.clone()));
+    let wake = entry.wake.clone();
+    registry.profile.lock().await.insert(path.clone(), entry);
+
+    tokio::time::timeout(Duration::from_secs(2), registry.wait_profile_idle(&path))
+        .await
+        .expect("history refusal must not wait for summary admission");
+    assert_eq!(
+        wake.serving_status().state,
+        SessionProjectionServingState::Stale {
+            reason: SessionProjectionStaleReason::HistoricalRetry {
+                reason_code: "history_admission_saturated".to_owned(),
+            },
+        }
+    );
+    assert_eq!(history.passes.load(Ordering::Acquire), 0);
+    drop(held);
+    tokio::time::timeout(Duration::from_secs(2), history.entered.notified())
+        .await
+        .expect("the retained history gets the released permit");
+    assert_eq!(history.passes.load(Ordering::Acquire), 1);
+    history.release.notify_one();
+    registry.shutdown().await;
+}
+
 /// A history pass that ignores cancellation until the test releases it. The
 /// pass holds a clone of `held` for as long as the worker owns it, standing in
 /// for the lease clones a live worker retains. A `busy` pass is never released
diff --git a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/wake.rs b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/wake.rs
index 9708b66190..8f37a1a1ed 100644
--- a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/wake.rs
+++ b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/wake.rs
@@ -689,7 +689,10 @@ impl SessionTemporalRefreshWakeState {
 
     #[cfg(any(test, feature = "test-helpers"))]
     pub fn is_idle(&self) -> bool {
-        !self.busy.load(Ordering::Acquire) && !self.has_pending_work()
+        // Consuming a wake follows marking the worker busy. Read the flags
+        // first so observing their consumption also makes that busy state
+        // visible, rather than combining an old idle read with cleared flags.
+        !self.has_pending_work() && !self.busy.load(Ordering::Acquire)
     }
 }
 
diff --git a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs
index 489de73c34..5427dd0fa8 100644
--- a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs
+++ b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs
@@ -1915,13 +1915,12 @@ mod tests {
     use tracedecay_runtime_core::db::engine::params;
     use tracedecay_sessions::runtime::{SessionMessageRecord, SessionRecord};
 
-    /// Wide enough that one item's commit lands inside it on a loaded host.
-    const COMMIT_DEADLINE: Duration = Duration::from_secs(2);
-
-    /// Commits `item_count` items. A group of several items never commits
-    /// before the deadline, and neither does a `slow` item alone; every other
-    /// item alone commits at once. Returns how many commits the deadline cut
-    /// short.
+    /// Commits `item_count` items. A group of several items never commits, and
+    /// neither does a `slow` item alone: its transaction ends in the
+    /// `DeadlineExceeded` that storage reports when its own deadline passes.
+    /// Every other item alone commits. Real writes race no wall-clock deadline
+    /// (`Duration::MAX` starts no timer), so host load cannot cut an item that
+    /// the test means to commit. Returns how many commits were cut short.
     async fn timed_refresh_group_progress(item_count: usize, slow: &[usize]) -> usize {
         let harness = RegisteredGlobalDbHarness::open("refresh-group-progress").await;
         let database = &harness.registered;
@@ -1937,7 +1936,7 @@ mod tests {
 
         let cut_short = std::sync::atomic::AtomicUsize::new(0);
         let cut_short = &cut_short;
-        let outcomes = commit_refresh_groups(item_count, COMMIT_DEADLINE, |range| async move {
+        let outcomes = commit_refresh_groups(item_count, Duration::MAX, |range| async move {
             let transaction = database
                 .begin_write_transaction()
                 .await
@@ -1953,7 +1952,7 @@ mod tests {
             }
             if range.len() > 1 || slow.contains(&range.start) {
                 cut_short.fetch_add(1, Ordering::SeqCst);
-                std::future::pending::<()>().await;
+                return Err(SessionStoreError::DeadlineExceeded);
             }
             transaction
                 .commit()
@@ -2010,6 +2009,28 @@ mod tests {
         );
     }
 
+    /// The deadline itself ends a group that never commits. No I/O races it:
+    /// a commit that is ready at its first poll wins even past the deadline,
+    /// so only the pending group is ever cut, however loaded the host.
+    #[tokio::test]
+    async fn a_refresh_group_that_outlives_its_deadline_is_retried_item_by_item() {
+        let outcomes = commit_refresh_groups(3, Duration::from_millis(10), |range| async move {
+            if range.len() > 1 {
+                std::future::pending::<()>().await;
+            }
+            Ok(range.map(Ok).collect())
+        })
+        .await
+        .unwrap();
+        assert_eq!(
+            outcomes
+                .into_iter()
+                .map(|outcome| outcome.unwrap())
+                .collect::<Vec<usize>>(),
+            [0, 1, 2]
+        );
+    }
+
     #[tokio::test]
     async fn a_timed_out_completion_stays_claimable_by_the_next_pass() {
         let harness = RegisteredGlobalDbHarness::open("refresh-timed-out-completion").await;
diff --git a/crates/tracedecay-session-temporal-store/src/participant_freeze.rs b/crates/tracedecay-session-temporal-store/src/participant_freeze.rs
index 6d86e1d60a..13f0819874 100644
--- a/crates/tracedecay-session-temporal-store/src/participant_freeze.rs
+++ b/crates/tracedecay-session-temporal-store/src/participant_freeze.rs
@@ -531,7 +531,9 @@ mod tests {
         bind_test_session_relation_graph, open_registered_test_database_fixture,
         publish_test_session_relation_projection,
     };
-    use tracedecay_global_db::{RegisteredGlobalDbLeaseV1, RegisteredGlobalDbOwnerV1};
+    use tracedecay_global_db::{
+        RegisteredGlobalDbLeaseV1, RegisteredGlobalDbOwnerV1, RegisteredGlobalDbWriterConnection,
+    };
     use tracedecay_runtime_core::db::TestDatabaseRuntimeScope;
     use tracedecay_runtime_core::db::engine::{Executor, TestConnection};
     use tracedecay_temporal_query::candidates::CandidateChannel;
@@ -656,7 +658,7 @@ mod tests {
 
     /// `execute()` signs its continuation cursors, so a root fixture that stops at
     /// `freeze()` needs no key and one that runs a query needs exactly one.
-    async fn seed_root_cursor_key(connection: &TestConnection) {
+    async fn seed_root_cursor_key(connection: &impl Executor) {
         connection
             .execute(
                 "INSERT INTO session_query_cursor_keys (
@@ -675,7 +677,7 @@ mod tests {
     /// on to publish a real projection must let the publisher mint its own.
     async fn publish_root_relation_projection(
         database: &RegisteredGlobalDbLeaseV1,
-        connection: &TestConnection,
+        connection: &impl Executor,
         session_id: &str,
     ) {
         for table in [
@@ -784,7 +786,7 @@ mod tests {
     }
 
     async fn seed_root_sessions(
-        connection: &TestConnection,
+        connection: &impl Executor,
         count: usize,
         hit_count: usize,
         source_frontier: u64,
@@ -801,7 +803,7 @@ mod tests {
     /// Term overlap across sessions is what separates the lexical tiers, so a
     /// fixture that exercises them has to choose each session's words.
     async fn seed_root_sessions_with_text(
-        connection: &TestConnection,
+        connection: &impl Executor,
         count: usize,
         hit_count: usize,
         source_frontier: u64,
@@ -978,13 +980,11 @@ mod tests {
         }
     }
 
+    /// Opens the registered root without attaching a second writer to its file.
+    /// Callers seed through `writer_connection`, the same lane projection uses.
     async fn open_root_fixture(
         directory: &TempDir,
-    ) -> (
-        RegisteredGlobalDbLeaseV1,
-        RegisteredGlobalDbOwnerV1,
-        TestConnection,
-    ) {
+    ) -> (RegisteredGlobalDbLeaseV1, RegisteredGlobalDbOwnerV1) {
         let database_path = directory.path().join("sessions.db");
         let (database, owner) = open_registered_test_database_fixture(
             &database_path,
@@ -993,12 +993,20 @@ mod tests {
         .await
         .expect("registered schema");
         bind_test_session_relation_graph(&database).expect("session relation graph");
-        (database, owner, TestConnection::open(&database_path))
+        (database, owner)
+    }
+
+    fn root_fixture_writer(
+        database: &RegisteredGlobalDbLeaseV1,
+    ) -> RegisteredGlobalDbWriterConnection<'_> {
+        database
+            .writer_connection()
+            .expect("registered root fixture writer")
     }
 
     /// Groups the first seeded session's matching occurrence together with
     /// `extra_members` further matching occurrences into one span evidence row.
-    async fn seed_root_span_evidence(connection: &TestConnection, extra_members: usize) {
+    async fn seed_root_span_evidence(connection: &impl Executor, extra_members: usize) {
         seed_root_span(connection, extra_members, "needle cohort", "anchor.000").await;
     }
 
@@ -1006,14 +1014,14 @@ mod tests {
     /// happens to contain the one message the query matched. Each member carries
     /// its own anchor and text, so no channel proposes it as a result.
     async fn seed_root_span_over_unmatched_members(
-        connection: &TestConnection,
+        connection: &impl Executor,
         extra_members: usize,
     ) {
         seed_root_span(connection, extra_members, "surrounding chatter", "").await;
     }
 
     async fn seed_root_span(
-        connection: &TestConnection,
+        connection: &impl Executor,
         extra_members: usize,
         text: &str,
         shared_anchor: &str,
@@ -1149,7 +1157,7 @@ mod tests {
     /// query's terms is common and their conjunction is rare. Two sessions carry
     /// all four terms; every other session carries three of them, so an OR clause
     /// proposes the entire root and even a drop-one clause proposes hundreds.
-    async fn seed_root_lexical_cohort(connection: &TestConnection, count: usize) {
+    async fn seed_root_lexical_cohort(connection: &impl Executor, count: usize) {
         seed_root_sessions_with_text(connection, count, count, 1, &|index| {
             match index {
                 0 | 1 => "alpha beta gamma delta",
@@ -1167,7 +1175,7 @@ mod tests {
 
     /// One session holds three of the query's four terms and nothing holds all
     /// four, so the strict tier is a verified zero and the drop-one tier answers.
-    async fn seed_root_relaxable_cohort(connection: &TestConnection) {
+    async fn seed_root_relaxable_cohort(connection: &impl Executor) {
         seed_root_sessions_with_text(connection, 4, 4, 1, &|index| match index {
             0 => "alpha beta gamma".to_string(),
             _ => format!("beta without the rest {index}"),
@@ -1207,7 +1215,8 @@ mod tests {
     #[asupersync::test]
     async fn root_conjunctive_lexical_answers_with_the_messages_every_term_matches() {
         let directory = tempdir().expect("temporary directory");
-        let (database, _owner, connection) = open_root_fixture(&directory).await;
+        let (database, _owner) = open_root_fixture(&directory).await;
+        let connection = root_fixture_writer(&database);
         seed_root_lexical_cohort(&connection, 300).await;
         seed_root_cursor_key(&connection).await;
         for session in ["session.000", "session.001"] {
@@ -1256,7 +1265,8 @@ mod tests {
     #[asupersync::test]
     async fn a_verified_zero_strict_tier_relaxes_one_named_step() {
         let directory = tempdir().expect("temporary directory");
-        let (database, _owner, connection) = open_root_fixture(&directory).await;
+        let (database, _owner) = open_root_fixture(&directory).await;
+        let connection = root_fixture_writer(&database);
         seed_root_relaxable_cohort(&connection).await;
         seed_root_cursor_key(&connection).await;
         publish_root_relation_projection(&database, &connection, "session.000").await;
@@ -1285,7 +1295,8 @@ mod tests {
     #[asupersync::test]
     async fn a_timed_out_strict_tier_refuses_instead_of_relaxing() {
         let directory = tempdir().expect("temporary directory");
-        let (database, _owner, connection) = open_root_fixture(&directory).await;
+        let (database, _owner) = open_root_fixture(&directory).await;
+        let connection = root_fixture_writer(&database);
         seed_root_relaxable_cohort(&connection).await;
         seed_root_cursor_key(&connection).await;
         publish_root_relation_projection(&database, &connection, "session.000").await;
@@ -1313,7 +1324,8 @@ mod tests {
     #[asupersync::test]
     async fn root_span_matched_by_many_members_is_one_candidate_over_300_sessions() {
         let directory = tempdir().expect("temporary directory");
-        let (database, _owner, connection) = open_root_fixture(&directory).await;
+        let (database, _owner) = open_root_fixture(&directory).await;
+        let connection = root_fixture_writer(&database);
         seed_root_sessions(&connection, 300, 1, 1).await;
         seed_root_span_evidence(&connection, 4).await;
 
@@ -1347,7 +1359,8 @@ mod tests {
     #[asupersync::test]
     async fn root_rare_hit_executes_under_the_unchanged_record_ceiling() {
         let directory = tempdir().expect("temporary directory");
-        let (database, _owner, connection) = open_root_fixture(&directory).await;
+        let (database, _owner) = open_root_fixture(&directory).await;
+        let connection = root_fixture_writer(&database);
         seed_root_sessions(&connection, 300, 1, 1).await;
         seed_root_cursor_key(&connection).await;
         publish_root_relation_projection(&database, &connection, "session.000").await;
@@ -1381,7 +1394,8 @@ mod tests {
     #[asupersync::test]
     async fn root_record_read_exhaustion_is_a_typed_budget_refusal_from_execute() {
         let directory = tempdir().expect("temporary directory");
-        let (database, _owner, connection) = open_root_fixture(&directory).await;
+        let (database, _owner) = open_root_fixture(&directory).await;
+        let connection = root_fixture_writer(&database);
         seed_root_sessions(&connection, 300, 8, 1).await;
         seed_root_cursor_key(&connection).await;
         for index in 0..8 {
@@ -1428,7 +1442,8 @@ mod tests {
     #[asupersync::test]
     async fn root_no_hit_over_256_sessions_is_truthful_zero_not_manifest_limit() {
         let directory = tempdir().expect("temporary directory");
-        let (database, _owner, connection) = open_root_fixture(&directory).await;
+        let (database, _owner) = open_root_fixture(&directory).await;
+        let connection = root_fixture_writer(&database);
         seed_root_sessions(&connection, 300, 0, 1).await;
 
         let execution = super::super::RegisteredGlobalDbSessionTemporalExecution::new(&database);
@@ -1467,7 +1482,8 @@ mod tests {
     #[asupersync::test]
     async fn root_no_hit_reports_aggregate_projection_staleness() {
         let directory = tempdir().expect("temporary directory");
-        let (database, _owner, connection) = open_root_fixture(&directory).await;
+        let (database, _owner) = open_root_fixture(&directory).await;
+        let connection = root_fixture_writer(&database);
         seed_root_sessions(&connection, 3, 0, 2).await;
 
         let execution = super::super::RegisteredGlobalDbSessionTemporalExecution::new(&database);
@@ -1486,7 +1502,8 @@ mod tests {
     #[asupersync::test]
     async fn root_rare_hit_over_256_sessions_freezes_only_the_admitted_participant() {
         let directory = tempdir().expect("temporary directory");
-        let (database, _owner, connection) = open_root_fixture(&directory).await;
+        let (database, _owner) = open_root_fixture(&directory).await;
+        let connection = root_fixture_writer(&database);
         seed_root_sessions(&connection, 300, 1, 1).await;
 
         let execution = super::super::RegisteredGlobalDbSessionTemporalExecution::new(&database);
@@ -1511,7 +1528,8 @@ mod tests {
     #[asupersync::test]
     async fn root_strict_population_over_one_window_returns_truthful_partial_coverage() {
         let directory = tempdir().expect("temporary directory");
-        let (database, _owner, connection) = open_root_fixture(&directory).await;
+        let (database, _owner) = open_root_fixture(&directory).await;
+        let connection = root_fixture_writer(&database);
         seed_root_sessions_with_text(&connection, 300, 280, 1, &|index| {
             format!("needle strict-{index:03} cohort")
         })
diff --git a/crates/tracedecay-sessions/src/runtime/git_correlation/backfill/bounded/tests.rs b/crates/tracedecay-sessions/src/runtime/git_correlation/backfill/bounded/tests.rs
index 9b378bc08c..2d1b5c4254 100644
--- a/crates/tracedecay-sessions/src/runtime/git_correlation/backfill/bounded/tests.rs
+++ b/crates/tracedecay-sessions/src/runtime/git_correlation/backfill/bounded/tests.rs
@@ -4,6 +4,7 @@ use std::process::{Command, Stdio};
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::time::Duration;
 
+use futures_util::FutureExt;
 use tracedecay_runtime_core::db::engine::{
     Executor, QueryExecutor, ReadSnapshot, TestConnection, Transaction, TransactionBehavior, params,
 };
@@ -286,24 +287,34 @@ async fn touch(store: &TestStore, session_id: &str) -> i64 {
     }
 }
 
-/// Real Git whose every read first takes two long polls, the way a large
-/// reflog or many worktrees do.
-struct SlowGit;
+/// Real Git whose reads must execute on the native blocking pool.
+#[derive(Default)]
+struct BlockingPoolGit {
+    reads: AtomicUsize,
+}
 
-impl SlowGit {
-    fn stall() {
-        std::thread::sleep(native_runtime::LONG_POLL * 2);
+impl BlockingPoolGit {
+    fn assert_blocking_pool(&self) {
+        // Nested blocking work runs inline on a pool thread. On a native
+        // worker it is queued elsewhere, so either readiness or identity
+        // differs without relying on host-dependent poll timing.
+        assert_eq!(
+            native_runtime::spawn_blocking(|| std::thread::current().id()).now_or_never(),
+            Some(Ok(std::thread::current().id())),
+            "Git reads must run on the native blocking pool"
+        );
+        self.reads.fetch_add(1, Ordering::SeqCst);
     }
 }
 
-impl GitReflogSource for SlowGit {
+impl GitReflogSource for BlockingPoolGit {
     fn reflog(&self, worktree: &Path) -> Option<String> {
-        Self::stall();
+        self.assert_blocking_pool();
         SystemGit.reflog(worktree)
     }
 
     fn current_branch(&self, worktree: &Path) -> Option<String> {
-        Self::stall();
+        self.assert_blocking_pool();
         SystemGit.current_branch(worktree)
     }
 
@@ -312,29 +323,16 @@ impl GitReflogSource for SlowGit {
         worktree: &Path,
         reference: &str,
     ) -> Result<bool, GitCorrelationError> {
-        Self::stall();
+        self.assert_blocking_pool();
         SystemGit.commit_reference_exists(worktree, reference)
     }
 
     fn commit_log(&self, worktree: &Path, branch: &str, since: i64) -> Option<String> {
-        Self::stall();
+        self.assert_blocking_pool();
         SystemGit.commit_log(worktree, branch, since)
     }
 }
 
-#[track_caller]
-fn spawned_here<T: Clone + Send + Sync + 'static>(
-    work: impl Future<Output = T> + Send + 'static,
-) -> (
-    native_runtime::NativeTask<T>,
-    &'static std::panic::Location<'static>,
-) {
-    (
-        native_runtime::NativeTask::spawn(work).expect("native task"),
-        std::panic::Location::caller(),
-    )
-}
-
 /// The session refresh worker runs this pass on a native task. Its Git reads
 /// wait on subprocess pipes, so they must wait on the blocking pool: on the
 /// worker they hold it for every read and stall every task queued behind it.
@@ -342,26 +340,27 @@ fn spawned_here<T: Clone + Send + Sync + 'static>(
 async fn convergence_waits_for_git_off_the_native_worker() {
     let repository = repository_fixture();
     let directory = tempfile::tempdir().unwrap();
-    let store = prepare_store(&directory.path().join("sessions.db"), repository.path()).await;
-
-    let (pass, site) = spawned_here(async move {
-        converge_git_evidence_pass(&store, Arc::new(SlowGit), None)
+    // Keep the fixture's writer teardown outside the native task.
+    let store =
+        Arc::new(prepare_store(&directory.path().join("sessions.db"), repository.path()).await);
+    let git = Arc::new(BlockingPoolGit::default());
+    let task_store = Arc::clone(&store);
+    let task_git: Arc<dyn GitReflogSource> = git.clone();
+
+    let pass = native_runtime::NativeTask::spawn(async move {
+        converge_git_evidence_pass(task_store.as_ref(), task_git, None)
             .await
             .map(|outcome| outcome.pass.backfill)
             .map_err(|error| error.to_string())
-    });
+    })
+    .expect("native task");
     let native_runtime::NativeTaskOutcome::Completed(backfill) = pass.wait().await else {
         panic!("the convergence task did not complete");
     };
     let backfill = backfill.unwrap();
     assert!(backfill.spans_written > 0, "{backfill:?}");
     assert!(backfill.commits_attributed > 0, "{backfill:?}");
-    assert_eq!(
-        native_runtime::long_native_polls().get(site),
-        None,
-        "a Git read held the native worker past {:?}",
-        native_runtime::LONG_POLL
-    );
+    assert!(git.reads.load(Ordering::SeqCst) > 0, "real Git reads ran");
 }
 
 /// The evidence rows, the attribution mark, and the history frontier commit
diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex_app_server.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex_app_server.rs
index d74d64d4d0..7a5537aaf8 100644
--- a/crates/tracedecay-sessions/src/runtime/hosts/codex_app_server.rs
+++ b/crates/tracedecay-sessions/src/runtime/hosts/codex_app_server.rs
@@ -1578,6 +1578,7 @@ mod tests {
         let _process_guard = APP_SERVER_PROCESS_TEST_LOCK
             .lock()
             .unwrap_or_else(std::sync::PoisonError::into_inner);
+        #[cfg(not(target_os = "linux"))]
         unsafe extern "C" {
             fn kill(pid: i32, signal: i32) -> i32;
         }
@@ -1608,6 +1609,30 @@ mod tests {
             .trim()
             .parse()
             .expect("descendant pid");
+        let descendant_is_live = || {
+            // A killed orphan can remain a zombie until the host reaps it;
+            // kill(pid, 0) still succeeds although it has already exited.
+            #[cfg(target_os = "linux")]
+            {
+                let stat = match std::fs::read_to_string(format!("/proc/{descendant_pid}/stat")) {
+                    Ok(stat) => stat,
+                    Err(error) if error.kind() == std::io::ErrorKind::NotFound => return false,
+                    Err(error) => panic!("read descendant process state: {error}"),
+                };
+                stat.rsplit_once(')')
+                    .expect("descendant process stat must contain a command terminator")
+                    .1
+                    .split_whitespace()
+                    .next()
+                    .expect("descendant process stat must contain a state")
+                    != "Z"
+            }
+            #[cfg(not(target_os = "linux"))]
+            unsafe {
+                kill(descendant_pid, 0) == 0
+            }
+        };
+        assert!(descendant_is_live(), "descendant must run before shutdown");
 
         let shutdown = begin_codex_app_server_shutdown();
         let deadline = Instant::now() + Duration::from_secs(1);
@@ -1619,12 +1644,11 @@ mod tests {
             "active child should exit during shutdown"
         );
         let descendant_deadline = Instant::now() + Duration::from_secs(1);
-        while unsafe { kill(descendant_pid, 0) } == 0 && Instant::now() < descendant_deadline {
+        while descendant_is_live() && Instant::now() < descendant_deadline {
             std::thread::sleep(Duration::from_millis(10));
         }
-        assert_ne!(
-            unsafe { kill(descendant_pid, 0) },
-            0,
+        assert!(
+            !descendant_is_live(),
             "app-server descendant should exit during shutdown"
         );
 
diff --git a/crates/tracedecay-store-runtime/src/session_registry/code_graph/sealed_publication_tests.rs b/crates/tracedecay-store-runtime/src/session_registry/code_graph/sealed_publication_tests.rs
index 9632a0e8b4..c7c377edff 100644
--- a/crates/tracedecay-store-runtime/src/session_registry/code_graph/sealed_publication_tests.rs
+++ b/crates/tracedecay-store-runtime/src/session_registry/code_graph/sealed_publication_tests.rs
@@ -266,7 +266,7 @@ fn pressure_from_sample(
             resident_bytes: 0,
             unreclaimable_bytes: 0,
             swapped_bytes: 0,
-            cgroup_committed_bytes: None,
+            cgroup_headroom_bytes: None,
         },
     ));
     let sampled = Arc::clone(&sample);
@@ -288,7 +288,7 @@ fn set_sample(
             resident_bytes: unreclaimable_bytes,
             unreclaimable_bytes,
             swapped_bytes: 0,
-            cgroup_committed_bytes: None,
+            cgroup_headroom_bytes: None,
         };
 }
 
@@ -2206,7 +2206,7 @@ async fn sealed_publication_refuses_over_the_resident_memory_watermark() {
             resident_bytes: 0,
             unreclaimable_bytes: 0,
             swapped_bytes: 0,
-            cgroup_committed_bytes: None,
+            cgroup_headroom_bytes: None,
         },
     ));
     let refuse_once_projecting = Arc::new(AtomicBool::new(true));
@@ -2227,7 +2227,7 @@ async fn sealed_publication_refuses_over_the_resident_memory_watermark() {
                             resident_bytes: limit_bytes,
                             unreclaimable_bytes: limit_bytes,
                             swapped_bytes: 0,
-                            cgroup_committed_bytes: None,
+                            cgroup_headroom_bytes: None,
                         },
                     );
                 }
@@ -2383,7 +2383,7 @@ async fn sealed_publication_samples_resident_memory_per_interval_not_per_row() {
                 resident_bytes: 0,
                 unreclaimable_bytes: 0,
                 swapped_bytes: 0,
-                cgroup_committed_bytes: None,
+                cgroup_headroom_bytes: None,
             })
         }),
     ));
diff --git a/crates/tracedecay/src/daemon/code_index_runtime_graph_activation_tests.rs b/crates/tracedecay/src/daemon/code_index_runtime_graph_activation_tests.rs
index be1502e22d..eab245fe66 100644
--- a/crates/tracedecay/src/daemon/code_index_runtime_graph_activation_tests.rs
+++ b/crates/tracedecay/src/daemon/code_index_runtime_graph_activation_tests.rs
@@ -2616,7 +2616,7 @@ async fn a_refresh_refused_beside_the_serving_graph_publishes_under_status_polli
                 resident_bytes: bytes,
                 unreclaimable_bytes: bytes,
                 swapped_bytes: 0,
-                cgroup_committed_bytes: None,
+                cgroup_headroom_bytes: None,
             })
         }),
     ));
diff --git a/crates/tracedecay/src/daemon/hook_v2_replay_consumer.rs b/crates/tracedecay/src/daemon/hook_v2_replay_consumer.rs
index 7d2e2e7ebc..4935ec3b5f 100644
--- a/crates/tracedecay/src/daemon/hook_v2_replay_consumer.rs
+++ b/crates/tracedecay/src/daemon/hook_v2_replay_consumer.rs
@@ -26,7 +26,8 @@ use tracedecay_hooks::{
 
 use tracedecay_mcp::handlers::hook_runtime::{
     HookV2AdmissionOutcomeV1, admit_hook_v2_envelope,
-    admit_hook_v2_replayed_envelope_with_lifecycle, hook_v2_pending_work_envelopes,
+    admit_hook_v2_replayed_envelope_with_lifecycle, hook_v2_native_lifecycle_enabled,
+    hook_v2_pending_work_envelopes,
 };
 
 mod spool_opener;
@@ -209,10 +210,6 @@ async fn drain_all_hosts(
             Ok(more) => sweep.more_pending |= more,
             Err(cause) => sweep.fail(*host, StatusHookReplaySpoolV1::DeliveryReceipts, cause),
         }
-        Box::pin(redrive_pending_work(
-            graph, data_root, *host, project_id, now, &mut sweep,
-        ))
-        .await;
         let report = match Box::pin(drain_admitted_host_spool(
             *host,
             project_id,
@@ -229,6 +226,11 @@ async fn drain_all_hosts(
                 None
             }
         };
+        // The host spool retains native lifecycle evidence the ledger cannot carry.
+        Box::pin(redrive_pending_work(
+            graph, data_root, *host, project_id, now, &mut sweep,
+        ))
+        .await;
         if let Some(report) = report
             && (report.committed > 0 || report.duplicates > 0 || report.tombstoned > 0)
         {
@@ -295,6 +297,19 @@ async fn redrive_pending_work(
                 .await
             },
             |envelope, native_session_id| async move {
+                // A registered session may belong to an earlier event. It does
+                // not prove that this event's native call/edit lifecycle was
+                // mounted. Only the matching host spool record can finish it.
+                if tracedecay_hooks::NativeContextScoutLifecycleV1::host_mounts_lifecycle(host)
+                    && matches!(
+                        envelope.event,
+                        tracedecay_hooks::HookEventV2::ToolLifecycle { .. }
+                            | tracedecay_hooks::HookEventV2::SavedEdit { .. }
+                    )
+                    && hook_v2_native_lifecycle_enabled(graph, host).await
+                {
+                    return HookV2AdmissionOutcomeV1::Backpressured;
+                }
                 admit_hook_v2_envelope(graph, &envelope, native_session_id, hook_replay_now())
                     .await
             },
@@ -655,6 +670,317 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn pending_edit_waits_for_the_lifecycle_spool_before_settling() {
+        let (fixture, graph, _newer) = replay_owner_fixture().await;
+        let data_root = &graph.hook_store_layout().data_root;
+        let host = NativeHostIdentityV1::KimiCode;
+        let project_id =
+            tracedecay_agent_hosts::hooks::hook_project_id_for_layout(graph.hook_store_layout())
+                .unwrap();
+        let now = hook_replay_now();
+        let hook_runtime = crate::hook_runtime(tracedecay_runtime_core::config::ProfileRoot::new(
+            fixture._temp.path().join("profile"),
+        ));
+        let project = tracedecay_domain::ProjectId::new(
+            graph
+                .hook_store_layout()
+                .identity
+                .project_id
+                .clone()
+                .unwrap(),
+        )
+        .unwrap();
+        let scope = (hook_runtime.scope_resolver)(graph.project_root(), &project).unwrap();
+        let worktree_id = tracedecay_agent_hosts::hooks::hook_worktree_id_for_layout(
+            &hook_runtime,
+            graph.hook_store_layout(),
+        )
+        .unwrap();
+        let binding = published_hook_scope_binding(data_root, worktree_id, host, now).unwrap();
+        let admission_runtime = graph.test_runtime_for_test().unwrap();
+        let project_sessions = admission_runtime
+            .registered_database_arc(tracedecay_sessions::admission::HostAdmissionScope::Project)
+            .unwrap();
+        let background_cpu = admission_runtime.background_cpu();
+        let lifecycle = tracedecay_hooks::NativeContextScoutLifecycleV1::new(
+            "session.replay.edit",
+            "call.replay.edit",
+            [8; 16],
+        )
+        .unwrap();
+        let envelope = tracedecay_hooks::HookEventEnvelopeV2 {
+            schema_version: tracedecay_hooks::HOOK_EVENT_SCHEMA_VERSION,
+            event_id: lifecycle.event_id,
+            producer: host,
+            protected_session_id: tracedecay_agent_hosts::hooks::protected_native_session_id(
+                lifecycle.session_id.as_str(),
+            ),
+            project_id,
+            repository_id: binding.repository_id,
+            worktree_id: binding.worktree_id,
+            worktree_epoch: binding.worktree_epoch,
+            binding_token: binding.binding_token,
+            ordering: tracedecay_hooks::HookOrderingV1::Unknown,
+            observed_at: now,
+            event: tracedecay_hooks::HookEventV2::SavedEdit {
+                file_id: [9; 16],
+                changed_range_count: 1,
+            },
+        };
+        // Commit admission while no orchestrator is mounted, reproducing the
+        // durable state left by a caller cancelled before lifecycle persistence.
+        assert!(matches!(
+            admit_hook_v2_envelope(&graph, &envelope, None, now).await,
+            HookV2AdmissionOutcomeV1::Admitted { .. }
+        ));
+        let pending =
+            || hook_v2_pending_work_envelopes(data_root, host, hook_replay_now()).unwrap();
+        assert_eq!(pending(), vec![envelope.clone()]);
+        assert_eq!(
+            tracedecay_daemon_service::context_scout_lifecycle::register_context_scout_lifecycle_authority(
+                project_id,
+                binding.worktree_id,
+                scope.project_id.clone(),
+                scope.worktree_id.clone(),
+                &project_sessions,
+            ),
+            tracedecay_daemon_service::context_scout_lifecycle::AuthorityRegistrationV1::Registered
+        );
+        let (ran, mut runs) = tokio::sync::mpsc::unbounded_channel();
+        let runtime = tracedecay_daemon_service::BoundedHookOrchestratorV1::new(
+            1,
+            move |request: tracedecay_daemon_service::HookOrchestrationRequestV1, _| {
+                let ran = ran.clone();
+                async move { ran.send(request.lifecycle).unwrap() }
+            },
+        )
+        .unwrap();
+        assert!(
+            tracedecay_daemon_service::register_hook_orchestration_runtime(
+                project_id,
+                binding.worktree_id,
+                &runtime,
+            )
+        );
+        let sweep = drain_all_hosts(
+            &graph,
+            data_root,
+            &fixture.settlements,
+            &project_sessions,
+            &background_cpu,
+        )
+        .await;
+        assert!(
+            sweep.failures.iter().any(|failure| failure.host == host
+                && failure.spool == StatusHookReplaySpoolV1::PendingWork)
+        );
+        assert_eq!(pending(), vec![envelope.clone()]);
+        assert!(
+            runs.try_recv().is_err(),
+            "ledger replay must not execute without lifecycle"
+        );
+
+        let root = hook_v2_spool_root(data_root, host);
+        let (mut spool, _) = HookSpoolV1::open(&root, HookSpoolConfigV1::stock(host), now).unwrap();
+        spool
+            .append_with_native_lifecycle(envelope.clone(), Some(lifecycle.clone()), &binding, now)
+            .unwrap();
+        drop(spool);
+        drain_all_hosts(
+            &graph,
+            data_root,
+            &fixture.settlements,
+            &project_sessions,
+            &background_cpu,
+        )
+        .await;
+        let mounted = tokio::time::timeout(Duration::from_secs(10), runs.recv())
+            .await
+            .unwrap()
+            .unwrap()
+            .expect("spool replay must mount lifecycle");
+        assert_eq!(mounted.session_id, lifecycle.session_id);
+        tokio::time::timeout(Duration::from_secs(10), async {
+            while !pending().is_empty() {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("mounted work must settle its ledger obligation");
+        assert!(
+            runs.try_recv().is_err(),
+            "the mounted edit executes exactly once"
+        );
+        // A known session is NOT proof that another edit's native lifecycle
+        // reached the spool. This is the ledger-only redrive regression.
+        assert_eq!(
+            tracedecay_daemon_service::context_scout_lifecycle::lookup_registered_context_scout_native_session(
+                project_id,
+                binding.worktree_id,
+                envelope.protected_session_id,
+            )
+            .await,
+            Some(lifecycle.session_id.clone())
+        );
+        tracedecay_daemon_service::unregister_hook_orchestration_runtime(
+            project_id,
+            binding.worktree_id,
+            &runtime,
+        );
+        let second_lifecycle = tracedecay_hooks::NativeContextScoutLifecycleV1::new(
+            "session.replay.edit",
+            "call.replay.second",
+            [11; 16],
+        )
+        .unwrap();
+        let second_edit = tracedecay_hooks::HookEventEnvelopeV2 {
+            event_id: second_lifecycle.event_id,
+            observed_at: hook_replay_now(),
+            event: tracedecay_hooks::HookEventV2::SavedEdit {
+                file_id: [11; 16],
+                changed_range_count: 1,
+            },
+            ..envelope.clone()
+        };
+        assert!(matches!(
+            admit_hook_v2_envelope(&graph, &second_edit, None, hook_replay_now()).await,
+            HookV2AdmissionOutcomeV1::Admitted { .. }
+        ));
+        assert!(
+            tracedecay_daemon_service::register_hook_orchestration_runtime(
+                project_id,
+                binding.worktree_id,
+                &runtime,
+            )
+        );
+        drain_all_hosts(
+            &graph,
+            data_root,
+            &fixture.settlements,
+            &project_sessions,
+            &background_cpu,
+        )
+        .await;
+        assert_eq!(pending(), vec![second_edit.clone()]);
+        assert!(
+            runs.try_recv().is_err(),
+            "a known session must not settle an edit without its call evidence"
+        );
+        let (mut second_spool, _) =
+            HookSpoolV1::open(&root, HookSpoolConfigV1::stock(host), hook_replay_now()).unwrap();
+        second_spool
+            .append_with_native_lifecycle(
+                second_edit,
+                Some(second_lifecycle.clone()),
+                &binding,
+                hook_replay_now(),
+            )
+            .unwrap();
+        drop(second_spool);
+        drain_all_hosts(
+            &graph,
+            data_root,
+            &fixture.settlements,
+            &project_sessions,
+            &background_cpu,
+        )
+        .await;
+        let second_mounted = tokio::time::timeout(Duration::from_secs(10), runs.recv())
+            .await
+            .unwrap()
+            .unwrap()
+            .expect("second spool record must mount its own lifecycle");
+        assert_eq!(second_mounted.session_id, second_lifecycle.session_id);
+        tokio::time::timeout(Duration::from_secs(10), async {
+            while !pending().is_empty() {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("second mounted edit must settle its ledger debt");
+        assert!(
+            runs.try_recv().is_err(),
+            "second edit executes exactly once"
+        );
+        // Claude deliberately omits lifecycle writes while intent is disabled;
+        // an acknowledged edit can therefore need ledger-only recovery.
+        tracedecay_daemon_service::unregister_hook_orchestration_runtime(
+            project_id,
+            binding.worktree_id,
+            &runtime,
+        );
+        let claude = NativeHostIdentityV1::ClaudeCode;
+        let claude_binding =
+            published_hook_scope_binding(data_root, binding.worktree_id, claude, hook_replay_now())
+                .unwrap();
+        let disabled_edit = tracedecay_hooks::HookEventEnvelopeV2 {
+            event_id: [10; 16],
+            producer: claude,
+            protected_session_id: tracedecay_agent_hosts::hooks::protected_native_session_id(
+                "session.replay.disabled-edit",
+            ),
+            project_id: claude_binding.project_id,
+            repository_id: claude_binding.repository_id,
+            worktree_id: claude_binding.worktree_id,
+            worktree_epoch: claude_binding.worktree_epoch,
+            binding_token: claude_binding.binding_token,
+            ..envelope
+        };
+        assert!(matches!(
+            admit_hook_v2_envelope(&graph, &disabled_edit, None, hook_replay_now()).await,
+            HookV2AdmissionOutcomeV1::Admitted { .. }
+        ));
+        let disabled_pending =
+            || hook_v2_pending_work_envelopes(data_root, claude, hook_replay_now()).unwrap();
+        assert_eq!(disabled_pending(), vec![disabled_edit]);
+        assert!(
+            tracedecay_daemon_service::register_hook_orchestration_runtime(
+                project_id,
+                binding.worktree_id,
+                &runtime,
+            )
+        );
+        drain_all_hosts(
+            &graph,
+            data_root,
+            &fixture.settlements,
+            &project_sessions,
+            &background_cpu,
+        )
+        .await;
+        assert!(
+            tokio::time::timeout(Duration::from_secs(10), runs.recv())
+                .await
+                .unwrap()
+                .unwrap()
+                .is_none(),
+            "disabled Claude edits settle without mounting a lifecycle"
+        );
+        tokio::time::timeout(Duration::from_secs(10), async {
+            while !disabled_pending().is_empty() {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("disabled Claude work must settle without a spool");
+        assert!(runtime.shutdown().await);
+        assert!(
+            pending().is_empty(),
+            "completed lifecycle work settles its durable obligation"
+        );
+        assert!(!HookSpoolV1::has_records(&root).unwrap());
+        assert!(runs.try_recv().is_err(), "each edit executes exactly once");
+        tracedecay_daemon_service::unregister_hook_orchestration_runtime(
+            project_id,
+            binding.worktree_id,
+            &runtime,
+        );
+        assert!(tracedecay_daemon_service::context_scout_lifecycle::unregister_context_scout_lifecycle_authority(
+            project_id, binding.worktree_id, &project_sessions,
+        ));
+    }
+
     #[test]
     fn only_a_missing_spool_root_is_absent() {
         let dir = tempfile::tempdir().unwrap();
diff --git a/crates/tracedecay/src/daemon/invocation_executor.rs b/crates/tracedecay/src/daemon/invocation_executor.rs
index 87b343f20a..d886c35740 100644
--- a/crates/tracedecay/src/daemon/invocation_executor.rs
+++ b/crates/tracedecay/src/daemon/invocation_executor.rs
@@ -13,8 +13,8 @@ use tracedecay_tool_catalog::ApplicationSurfaceOperation;
 use tracedecay_application::operation_stream::OperationRequestControls;
 use tracedecay_daemon_protocol::DAEMON_TOOL_RESPONSE_GRACE;
 use tracedecay_daemon_service::{
-    DaemonInvocationOperation, DaemonInvocationProblem, ProjectRuntimeRequestLeaseV1,
-    WorkApplicationOutcomeV1,
+    DaemonInvocationOperation, DaemonInvocationPayload, DaemonInvocationProblem,
+    ProjectRuntimeRequestLeaseV1, WorkApplicationOutcomeV1,
 };
 use tracedecay_domain::errors::{Result, TraceDecayError};
 use tracedecay_runtime_core::native_runtime;
@@ -285,7 +285,7 @@ impl InProcessDaemonInvocationExecutor {
                     &self.store_administration,
                     Some(&self.project_path),
                     request,
-                    None,
+                    self.admitted_cancellation.clone(),
                 )
                 .await
         }
@@ -659,10 +659,29 @@ impl tracedecay_daemon_protocol::DaemonInvocationExecutor for InProcessDaemonInv
                     stage: CancellationStage::BeforeAdmission,
                 },
             )?;
-            let executor = self.clone();
+            let mut executor = self.clone();
             let admitted_cancellation = self.admitted_cancellation.clone();
             let request_cancellations = self.invocation.service.request_cancellations().clone();
             let unavailable = |_| tracedecay_daemon_protocol::DaemonInvocationError::Unavailable;
+            // The graph owner must observe the caller's live signal directly:
+            // relaying through two native tasks lets a blocking file walk
+            // continue after the transport has acknowledged cancellation.
+            let cancellation_lease = if executor.admitted_cancellation.is_none()
+                && matches!(&request.payload, DaemonInvocationPayload::GraphTool { .. })
+            {
+                let Some(lease) = request_cancellations
+                    .register_with_signal(&request.request_id, cancellation.clone())
+                else {
+                    return Ok(DaemonInvocationResponse::problem(
+                        request.request_id,
+                        DaemonInvocationProblem::InvalidRequest,
+                    ));
+                };
+                executor.admitted_cancellation = Some(lease.token());
+                Some(lease)
+            } else {
+                None
+            };
             match native_runtime::NativeTask::spawn(async move {
                 let request_id = request.request_id.clone();
                 let invocation =
@@ -671,7 +690,7 @@ impl tracedecay_daemon_protocol::DaemonInvocationExecutor for InProcessDaemonInv
                         tracing::trace_span!("daemon.invocation.invoke_once"),
                     ))
                     .map_err(unavailable)?;
-                settle_in_process_invocation(
+                let result = settle_in_process_invocation(
                     &request_cancellations,
                     &request_id,
                     invocation,
@@ -681,7 +700,9 @@ impl tracedecay_daemon_protocol::DaemonInvocationExecutor for InProcessDaemonInv
                     policy,
                     DAEMON_TOOL_RESPONSE_GRACE,
                 )
-                .await
+                .await;
+                drop(cancellation_lease);
+                result
             })
             .map_err(unavailable)?
             .await
diff --git a/crates/tracedecay/tests/hooks_lsp_suite/intent_pass_journey_test.rs b/crates/tracedecay/tests/hooks_lsp_suite/intent_pass_journey_test.rs
index 5ebc0200c3..99761dfb34 100644
--- a/crates/tracedecay/tests/hooks_lsp_suite/intent_pass_journey_test.rs
+++ b/crates/tracedecay/tests/hooks_lsp_suite/intent_pass_journey_test.rs
@@ -10,6 +10,7 @@ use std::sync::atomic::{AtomicUsize, Ordering};
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
 use serde_json::{Value, json};
+use tracedecay_contracts::retrieval::StatusHookReplayV1;
 use tracedecay_domain::configuration::{
     CONTEXT_SCOUT_SETTINGS_SETTING_KEY, ContextScoutConfigurationStateV1, ContextScoutIntentKindV1,
     ContextScoutIntentSettingsV1, ContextScoutSettingsV1, DecisionEngineSettingsV1,
@@ -501,8 +502,19 @@ fn claude_stop_delivers_the_intent_suggestion_the_daemon_judged_at_the_tool_boun
 
     let stop_event = stop(&project, ENABLED_SESSION, &enabled_transcript);
     let delivers = |context: &String| context.contains(&suggestion_text(YES_INTENT));
-    let mut stop_contexts = Vec::new();
-    while stop_contexts.len() < 5 && !stop_contexts.last().is_some_and(delivers) {
+    let mut stop_contexts = vec![hook_context(&run_hook(
+        home,
+        &project,
+        "hook-stop",
+        &stop_event,
+    ))];
+    if !stop_contexts.last().is_some_and(delivers) {
+        wait_for("the timed-out stop callback's replay to drain", || {
+            let status = run_tool(home, &project, "tracedecay_status", &json!({}));
+            let replay: StatusHookReplayV1 = serde_json::from_value(status["hook_replay"].clone())
+                .unwrap_or_else(|error| panic!("hook replay status ({error}): {status}"));
+            replay == StatusHookReplayV1::Drained && intent_outcomes(&log).len() > before + 1
+        });
         stop_contexts.push(hook_context(&run_hook(
             home,
             &project,
diff --git a/crates/tracedecay/tests/session_suite/session_runtime/retained_history.rs b/crates/tracedecay/tests/session_suite/session_runtime/retained_history.rs
index d5268a00cb..a3d4a8874a 100644
--- a/crates/tracedecay/tests/session_suite/session_runtime/retained_history.rs
+++ b/crates/tracedecay/tests/session_suite/session_runtime/retained_history.rs
@@ -791,6 +791,15 @@ async fn saturated_history_admission_defers_the_pass_and_resumes() {
         SessionHistoricalIngestOutcome::Complete,
     ]));
     let registry = SessionTemporalRefreshSchedulerRegistry::default();
+    registry
+        .ensure_profile(
+            authority.database().db_path().to_path_buf(),
+            authority.database.clone(),
+        )
+        .await;
+    registry
+        .wait_profile_idle(authority.database().db_path())
+        .await;
     let admission = registry.historical_ingest_admission();
     let permits = u32::try_from(admission.available_permits()).unwrap();
     assert!(permits > 0, "the historical ingest admission is bounded");
@@ -805,29 +814,30 @@ async fn saturated_history_admission_defers_the_pass_and_resumes() {
             ingestor.clone(),
         )
         .await;
+    registry
+        .wait_profile_idle(authority.database().db_path())
+        .await;
 
     let saturated = SessionProjectionServingState::Stale {
         reason: SessionProjectionStaleReason::HistoricalRetry {
             reason_code: "history_admission_saturated".to_owned(),
         },
     };
+    let saturated_observed = wait_until(
+        || wake.serving_status().state == saturated,
+        Duration::from_secs(2),
+    )
+    .await;
     assert!(
-        wait_until(
-            || wake.serving_status().state == saturated,
-            Duration::from_secs(2),
-        )
-        .await,
-        "the deferred pass must surface as typed retryable staleness"
+        saturated_observed,
+        "the deferred pass must surface as typed retryable staleness: {:?}",
+        wake.serving_status()
     );
     assert_eq!(
         ingestor.passes.get(),
         0,
         "a saturated admission must not run the historical pass"
     );
-    registry
-        .wait_profile_idle(authority.database().db_path())
-        .await;
-    assert_eq!(ingestor.passes.get(), 0);
 
     drop(held);
     ingestor.passes.wait_for(|passes| *passes >= 1).await;
@@ -851,6 +861,15 @@ async fn a_saturated_history_pass_runs_with_the_permit_its_queue_turn_grants() {
         SessionHistoricalIngestOutcome::Complete,
     ]));
     let registry = SessionTemporalRefreshSchedulerRegistry::default();
+    registry
+        .ensure_profile(
+            authority.database().db_path().to_path_buf(),
+            authority.database.clone(),
+        )
+        .await;
+    registry
+        .wait_profile_idle(authority.database().db_path())
+        .await;
     let admission = registry.historical_ingest_admission();
     let mut held = (0..admission.available_permits())
         .map(|_| {
@@ -867,22 +886,24 @@ async fn a_saturated_history_pass_runs_with_the_permit_its_queue_turn_grants() {
             ingestor.clone(),
         )
         .await;
+    registry
+        .wait_profile_idle(authority.database().db_path())
+        .await;
     let saturated = SessionProjectionServingState::Stale {
         reason: SessionProjectionStaleReason::HistoricalRetry {
             reason_code: "history_admission_saturated".to_owned(),
         },
     };
+    let saturated_observed = wait_until(
+        || wake.serving_status().state == saturated,
+        Duration::from_secs(2),
+    )
+    .await;
     assert!(
-        wait_until(
-            || wake.serving_status().state == saturated,
-            Duration::from_secs(2),
-        )
-        .await
+        saturated_observed,
+        "the queued pass must surface as typed retryable staleness: {:?}",
+        wake.serving_status()
     );
-    registry
-        .wait_profile_idle(authority.database().db_path())
-        .await;
-
     let next_waiter = tokio::spawn(Arc::clone(&admission).acquire_owned());
     tokio::task::yield_now().await;
     held.pop();
diff --git a/dashboard/BUILD.bazel b/dashboard/BUILD.bazel
index 436abbc228..30b68f5741 100644
--- a/dashboard/BUILD.bazel
+++ b/dashboard/BUILD.bazel
@@ -1,3 +1,4 @@
+load("@bazel_lib//lib:copy_to_bin.bzl", "copy_to_bin")
 load("@npm//:defs.bzl", "npm_link_all_packages")
 load("@npm//dashboard:@rsbuild/core/package_json.bzl", rsbuild_bin = "bin")
 load("@npm//dashboard:vitest/package_json.bzl", vitest_bin = "bin")
@@ -6,7 +7,8 @@ package(default_visibility = ["//visibility:public"])
 
 npm_link_all_packages(name = "node_modules")
 
-filegroup(
+# Share source copies across the bundle and tests with different CPU allocations.
+copy_to_bin(
     name = "dashboard_sources",
     srcs = glob(
         [
@@ -28,8 +30,11 @@ vitest_bin.vitest_test(
         "e2e/**",
         "stories/**",
     ]),
-    args = ["run"],
+    # Leave CPU capacity for transforms and the coordinator beside the workers.
+    args = ["run", "--maxWorkers=2"],
     chdir = package_name(),
+    exec_properties = {"cpu_count": "4"},
+    tags = ["cpu:4"],
 )
 
 rsbuild_bin.rsbuild(
diff --git a/dashboard/src/data/auth/authenticatedEventSource.dom.test.ts b/dashboard/src/data/auth/authenticatedEventSource.dom.test.ts
index 9ce6659a5c..ee01189d02 100644
--- a/dashboard/src/data/auth/authenticatedEventSource.dom.test.ts
+++ b/dashboard/src/data/auth/authenticatedEventSource.dom.test.ts
@@ -1,4 +1,7 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import { AuthenticatedEventSource } from './authenticatedEventSource.ts';
+import { installDashboardAuthentication } from './install.ts';
+import { connectEvents } from '../sse/connect.ts';
 
 const TOKEN = 'cd'.repeat(32);
 const BASIC = `Basic ${btoa(`tracedecay:${TOKEN}`)}`;
@@ -33,28 +36,28 @@ async function eventually(check: () => void): Promise<void> {
 
 describe('AuthenticatedEventSource', () => {
   const fetchMock = vi.fn<(input: RequestInfo | URL, init?: RequestInit) => Promise<Response>>();
+  const sources: AuthenticatedEventSource[] = [];
 
   beforeEach(() => {
     sessionStorage.setItem('tracedecay.dashboard.access', TOKEN);
-    vi.resetModules();
     fetchMock.mockReset();
     vi.stubGlobal('fetch', fetchMock);
   });
   afterEach(() => {
+    for (const source of sources.splice(0)) source.close();
     vi.unstubAllGlobals();
+    vi.useRealTimers();
     sessionStorage.clear();
   });
 
-  /** Loads the class, then constructs synchronously so handlers attach before
-   * the first awaited step, as `connectEvents` does. */
-  async function opener(url = '/api/events') {
-    const { AuthenticatedEventSource } = await import('./authenticatedEventSource.ts');
-    return () => new AuthenticatedEventSource(url);
+  function open(url = '/api/events') {
+    const source = new AuthenticatedEventSource(url);
+    sources.push(source);
+    return source;
   }
 
   it('authenticates the stream with a request header and no cookie', async () => {
-    fetchMock.mockResolvedValue(openResponse());
-    const open = await opener();
+    fetchMock.mockImplementation(async () => openResponse());
     const source = open();
     const opened = vi.fn();
     source.onopen = opened;
@@ -78,8 +81,7 @@ describe('AuthenticatedEventSource', () => {
         '\ndata: two\n\n',
       ]),
     );
-    fetchMock.mockResolvedValue(openResponse());
-    const open = await opener();
+    fetchMock.mockImplementation(async () => openResponse());
     const source = open();
     const named = vi.fn();
     const plain = vi.fn();
@@ -93,14 +95,18 @@ describe('AuthenticatedEventSource', () => {
   });
 
   it('reconnects with Last-Event-ID and the server retry hint after the stream drops', async () => {
+    vi.useFakeTimers();
     fetchMock.mockResolvedValueOnce(streamResponse(['retry: 1\nid: 7\ndata: x\n\n']));
-    fetchMock.mockResolvedValue(openResponse());
-    const open = await opener();
+    fetchMock.mockImplementation(async () => openResponse());
     const source = open();
     const errors = vi.fn();
     source.onerror = errors;
 
-    await eventually(() => expect(fetchMock).toHaveBeenCalledTimes(2));
+    await vi.advanceTimersByTimeAsync(0);
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    expect(source.readyState).toBe(AuthenticatedEventSource.CONNECTING);
+    await vi.advanceTimersByTimeAsync(1);
+    expect(fetchMock).toHaveBeenCalledTimes(2);
     expect(new Headers(fetchMock.mock.calls[1]![1]!.headers).get('last-event-id')).toBe('7');
     expect(errors).toHaveBeenCalled();
     source.close();
@@ -108,7 +114,6 @@ describe('AuthenticatedEventSource', () => {
 
   it('closes for good on a refusal instead of retrying with the same credential', async () => {
     fetchMock.mockResolvedValue(new Response('{}', { status: 401 }));
-    const open = await opener();
     const source = open();
     const errors = vi.fn();
     source.onerror = errors;
@@ -120,8 +125,7 @@ describe('AuthenticatedEventSource', () => {
   });
 
   it('close() aborts the in-flight request', async () => {
-    fetchMock.mockResolvedValue(openResponse());
-    const open = await opener();
+    fetchMock.mockImplementation(async () => openResponse());
     const source = open();
     await eventually(() => expect(fetchMock).toHaveBeenCalled());
 
@@ -133,24 +137,24 @@ describe('AuthenticatedEventSource', () => {
 });
 
 describe('installDashboardAuthentication', () => {
+  let connection: ReturnType<typeof connectEvents> | undefined;
   afterEach(() => {
+    connection?.close();
     vi.unstubAllGlobals();
     sessionStorage.clear();
     window.history.replaceState(null, '', '/');
   });
 
   it('routes the app event stream through an authenticated fetch', async () => {
-    vi.resetModules();
     const fetchMock = vi.fn(async () => openResponse());
     vi.stubGlobal('fetch', fetchMock);
     window.history.replaceState(null, '', `/#token=${TOKEN}`);
-    const { installDashboardAuthentication } = await import('./install.ts');
-    const { connectEvents } = await import('../sse/connect.ts');
-
+    vi.stubGlobal('EventSource', globalThis.EventSource);
     installDashboardAuthentication();
-    const connection = connectEvents('/api/events');
+    const active = connectEvents('/api/events');
+    connection = active;
 
-    await eventually(() => expect(connection.state()).toBe('live'));
+    await eventually(() => expect(active.state()).toBe('live'));
     const calls = fetchMock.mock.calls as unknown as Array<[string, RequestInit]>;
     expect(new Headers(calls[0]![1].headers).get('authorization')).toBe(BASIC);
     expect(window.location.hash).toBe('');