From 109f4608e6168f77ef9598aef47eb4589f151251 Mon Sep 17 00:00:00 2001
From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com>
Date: Sat, 7 Nov 2026 22:13:52 -0800
Subject: [PATCH 01/14] fix(accounting): stop uncommitted monitor token credits

Keep the live monitor's raw tool activity after a local accounting
write fails, but publish zero saved tokens instead of crediting the
uncommitted delta. Only the canonical project-store commit controls
what the monitor reports; optional global projection remains best-effort.

Add a real registered-store and mmap-backed regression covering
successful savings, an overflow refusal, both durable counters and
monitor output. No schema, release or policy changes.

Refs #4447
---
 crates/tracedecay/src/mcp/server/ledger.rs | 93 ++++++++++++++++++++--
 1 file changed, 88 insertions(+), 5 deletions(-)

diff --git a/crates/tracedecay/src/mcp/server/ledger.rs b/crates/tracedecay/src/mcp/server/ledger.rs
index 764a5abedde..2493c44e56c 100644
--- a/crates/tracedecay/src/mcp/server/ledger.rs
+++ b/crates/tracedecay/src/mcp/server/ledger.rs
@@ -89,12 +89,15 @@ pub(crate) struct McpToolErrorAnalyticsRequest<'a> {
 /// Commits a response delta even when startup could not read the display
 /// baseline. The optional cache and global projection consume committed data;
 /// neither can suppress the authoritative project-store write.
+/// Whether the canonical project-store counter committed. A failed optional
+/// global projection cannot undo the local commit; the live monitor credits
+/// only this durable outcome, not the response's optimistic estimate.
 pub(super) async fn commit_token_savings(
     cg: &TraceDecay,
     tokens_saved: Option<&AtomicU64>,
     sink: LedgerSink,
     delta: u64,
-) {
+) -> bool {
     match cg.record_token_savings(delta).await {
         Ok(committed_total) => {
             // Two completions may resume in reverse order after their commits.
@@ -112,6 +115,7 @@ pub(super) async fn commit_token_savings(
                     "background token-accounting projection failed"
                 );
             }
+            true
         }
         Err(error) => {
             tracing::warn!(
@@ -119,6 +123,7 @@ pub(super) async fn commit_token_savings(
                 %error,
                 "background token-accounting commit failed"
             );
+            false
         }
     }
 }
@@ -183,9 +188,18 @@ impl McpServer {
             .map(|profile| profile.data_dir().to_path_buf());
         let tool_name = tool_name.to_owned();
         self.spawn_observed_ledger_write(async move {
-            if let Some((tokens_saved, sink)) = persist {
-                commit_token_savings(&cg, tokens_saved.as_deref(), sink, net_saved_tokens).await;
-            }
+            // The monitor is an activity feed, not a second savings authority.
+            // Preserve the activity's raw input size after failure, but never
+            // credit a token delta that the project store could not commit.
+            let committed_saved_tokens = if let Some((tokens_saved, sink)) = persist {
+                if commit_token_savings(&cg, tokens_saved.as_deref(), sink, net_saved_tokens).await {
+                    net_saved_tokens
+                } else {
+                    0
+                }
+            } else {
+                0
+            };
             // The monitor entry opens, locks, and mmaps a file; keep that
             // off the async workers.
             // A server without an owning profile has no monitor ring.
@@ -199,7 +213,7 @@ impl McpServer {
                         &monitor_project_root,
                         "tracedecay",
                         &tool_name,
-                        net_saved_tokens,
+                        committed_saved_tokens,
                         raw_file_tokens,
                     );
                 });
@@ -614,6 +628,75 @@ mod tests {
         assert_eq!(reopened.get_local_counter().await.unwrap(), 35);
     }
 
+    #[tokio::test]
+    async fn live_monitor_reports_only_committed_token_savings() {
+        let (cg, _project, authority) =
+            super::super::writer_test_support::init_indexed_repo().await;
+        let project_root = cg.project_root().to_path_buf();
+        let server = super::super::McpServer::new_with_context(
+            super::super::writer_test_support::registered_context(cg, &authority),
+        )
+        .await;
+        let monitor_dir = server
+            .owner_profile()
+            .expect("fixture has an isolated owner profile")
+            .data_dir()
+            .to_path_buf();
+
+        server.spawn_token_accounting_persist(
+            server.cg_snapshot().await,
+            &project_root,
+            "tracedecay_search",
+            7,
+            20,
+        );
+        server.ledger_writes_settled().await;
+        {
+            let ring = tracedecay_session_memory::monitor_ring::MmapReader::open_at(&monitor_dir)
+                .expect("committed call created monitor");
+            assert_eq!(ring.write_idx(), 1);
+            let entry = ring.entry(0).expect("committed response entry");
+            assert_eq!(entry.delta, 7);
+            assert_eq!(entry.before, 20);
+            assert_eq!(entry.tool_name, "tracedecay_search");
+        }
+        assert_eq!(server.cg_snapshot().await.get_tokens_saved().await.unwrap(), 7);
+
+        // A real committed maximum forces the next recorder's checked-add
+        // refusal, without modifying production code or faking an error.
+        server
+            .cg_snapshot()
+            .await
+            .record_token_savings(u64::MAX - 7)
+            .await
+            .expect("seed a durable lifetime overflow boundary");
+        server.spawn_token_accounting_persist(
+            server.cg_snapshot().await,
+            &project_root,
+            "tracedecay_search",
+            1,
+            30,
+        );
+        server.ledger_writes_settled().await;
+        let ring = tracedecay_session_memory::monitor_ring::MmapReader::open_at(&monitor_dir)
+            .expect("failed accounting still records activity");
+        assert_eq!(ring.write_idx(), 2);
+        let entry = ring.entry(1).expect("failed response activity entry");
+        assert_eq!(entry.delta, 0, "uncommitted savings must not be credited");
+        assert_eq!(entry.before, 30, "raw tool activity remains observable");
+        assert_eq!(entry.tool_name, "tracedecay_search");
+        assert_eq!(
+            server.cg_snapshot().await.get_tokens_saved().await.unwrap(),
+            u64::MAX
+        );
+        assert_eq!(
+            server.cg_snapshot().await.get_local_counter().await.unwrap(),
+            u64::MAX,
+            "the refused delta must not change the resettable counter"
+        );
+        server.shutdown_background_tasks().await;
+    }
+
     #[tokio::test]
     async fn a_failed_accounting_commit_does_not_advance_the_display_cache() {
         let (cg, _project, authority) =

From 48d27c898ca9e32b00b1e788fd08ddaa41e12df6 Mon Sep 17 00:00:00 2001
From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com>
Date: Sun, 8 Nov 2026 02:13:47 -0800
Subject: [PATCH 02/14] style(accounting): apply hosted Rustfmt corrections

Apply the exact source changes requested by the first failing hosted
Bazel Rustfmt gate at the preceding PR head. No behavior or assertions
change; all existing regression tests are preserved.

Refs #4447

Verification: source contexts checked against hosted Rustfmt diffs.
New-head Rust/Bazel tests require hosted CI.
---
 crates/tracedecay/src/mcp/server/ledger.rs | 15 ++++++++++++---
 1 file changed, 12 insertions(+), 3 deletions(-)

diff --git a/crates/tracedecay/src/mcp/server/ledger.rs b/crates/tracedecay/src/mcp/server/ledger.rs
index 2493c44e56c..3dda2e3b1ad 100644
--- a/crates/tracedecay/src/mcp/server/ledger.rs
+++ b/crates/tracedecay/src/mcp/server/ledger.rs
@@ -192,7 +192,8 @@ impl McpServer {
             // Preserve the activity's raw input size after failure, but never
             // credit a token delta that the project store could not commit.
             let committed_saved_tokens = if let Some((tokens_saved, sink)) = persist {
-                if commit_token_savings(&cg, tokens_saved.as_deref(), sink, net_saved_tokens).await {
+                if commit_token_savings(&cg, tokens_saved.as_deref(), sink, net_saved_tokens).await
+                {
                     net_saved_tokens
                 } else {
                     0
@@ -660,7 +661,10 @@ mod tests {
             assert_eq!(entry.before, 20);
             assert_eq!(entry.tool_name, "tracedecay_search");
         }
-        assert_eq!(server.cg_snapshot().await.get_tokens_saved().await.unwrap(), 7);
+        assert_eq!(
+            server.cg_snapshot().await.get_tokens_saved().await.unwrap(),
+            7
+        );
 
         // A real committed maximum forces the next recorder's checked-add
         // refusal, without modifying production code or faking an error.
@@ -690,7 +694,12 @@ mod tests {
             u64::MAX
         );
         assert_eq!(
-            server.cg_snapshot().await.get_local_counter().await.unwrap(),
+            server
+                .cg_snapshot()
+                .await
+                .get_local_counter()
+                .await
+                .unwrap(),
             u64::MAX,
             "the refused delta must not change the resettable counter"
         );

From de8cb3d491c68e9a53befbcb6b9f660374efdb29 Mon Sep 17 00:00:00 2001
From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com>
Date: Sun, 8 Nov 2026 01:52:34 -0800
Subject: [PATCH 03/14] test(automation): bind scheduler fixtures to backend
 identities

Repair three hosted automation_runner_test failures by supplying the
configured executable identity required by the production prefilter.
Use the same identity for the prefilter and retained runner, preserving
manual-schedule, exact-ledger, live-lock and lock-race assertions.

Add a production-prefilter regression for unconfigured refusal,
deduplication and same-occurrence recovery after binding, with zero
backend calls and no effectful terminal. Do not relax backend admission,
change timeouts, or alter the graph publication changes.

Refs #4532
---
 .../tests/automation_runner_test/jobs.rs      | 124 ++++++++++++++++--
 1 file changed, 110 insertions(+), 14 deletions(-)

diff --git a/crates/tracedecay/tests/automation_runner_test/jobs.rs b/crates/tracedecay/tests/automation_runner_test/jobs.rs
index 740283d56e8..7342786381b 100644
--- a/crates/tracedecay/tests/automation_runner_test/jobs.rs
+++ b/crates/tracedecay/tests/automation_runner_test/jobs.rs
@@ -86,6 +86,7 @@ fn enabled_job_config() -> AutomationSettingsV1 {
 struct ContentBackend {
     calls: AtomicUsize,
     content: &'static str,
+    executable: Option<std::path::PathBuf>,
 }
 
 impl ContentBackend {
@@ -93,9 +94,15 @@ impl ContentBackend {
         Self {
             calls: AtomicUsize::new(0),
             content,
+            executable: None,
         }
     }
 
+    fn with_executable(mut self, executable: &Path) -> Self {
+        self.executable = Some(executable.to_path_buf());
+        self
+    }
+
     fn calls(&self) -> usize {
         self.calls.load(Ordering::SeqCst)
     }
@@ -126,7 +133,7 @@ impl AgentTaskBackend for ContentBackend {
     }
 
     fn executable(&self) -> Option<&std::path::Path> {
-        None
+        self.executable.as_deref()
     }
 }
 
@@ -988,6 +995,10 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() {
     let mut job = sample_job("sched-job");
     job.schedule = Some("manual".to_string());
     let config = enabled_job_config();
+    // Reach the schedule gate with a bound identity, not the earlier
+    // backend-unconfigured refusal. This gate never invokes the backend.
+    let executable = tempfile::NamedTempFile::new().unwrap();
+    let backend = ContentBackend::new("unused").with_executable(executable.path());
 
     // Manual schedule: every scheduler tick of the same occurrence reaches
     // the prefilter, which records the skip once.
@@ -996,7 +1007,7 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() {
         let run = evaluate_and_record_scheduler_skip(
             &dashboard_root,
             &config,
-            None,
+            backend.executable(),
             &job,
             "sched-occurrence",
             None,
@@ -1019,7 +1030,7 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() {
     let repeated = evaluate_and_record_scheduler_skip(
         &dashboard_root,
         &config,
-        None,
+        backend.executable(),
         &job,
         "sched-occurrence",
         None,
@@ -1031,6 +1042,76 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() {
         repeated.ledger_record, records[0],
         "an immediate repeat skip must return the exact durable prior row"
     );
+    assert_eq!(backend.calls(), 0);
+}
+
+#[tokio::test]
+async fn scheduler_prefilter_rejects_unconfigured_backend_and_recovers_without_effect() {
+    let temp = tempdir().unwrap();
+    let dashboard_root = temp.path().join("dashboard");
+    let job = sample_job("backend-recovery");
+    let config = enabled_job_config();
+    let occurrence = "backend-recovery-occurrence";
+    let backend = ContentBackend::new("unused");
+
+    let first = evaluate_and_record_scheduler_skip(
+        &dashboard_root,
+        &config,
+        backend.executable(),
+        &job,
+        occurrence,
+        None,
+    )
+    .await
+    .unwrap()
+    .expect("a due job still requires a configured backend");
+    assert_eq!(first.report["reason"], json!("backend_unconfigured"));
+    assert_eq!(first.ledger_record.status, AutomationRunStatus::Skipped);
+    assert_ne!(first.run_id, occurrence);
+    let repeated = evaluate_and_record_scheduler_skip(
+        &dashboard_root,
+        &config,
+        backend.executable(),
+        &job,
+        occurrence,
+        None,
+    )
+    .await
+    .unwrap()
+    .unwrap();
+    assert_eq!(repeated.ledger_record, first.ledger_record);
+
+    let executable = tempfile::NamedTempFile::new().unwrap();
+    let backend = backend.with_executable(executable.path());
+    assert!(
+        evaluate_and_record_scheduler_skip(
+            &dashboard_root,
+            &config,
+            backend.executable(),
+            &job,
+            occurrence,
+            None,
+        )
+        .await
+        .unwrap()
+        .is_none(),
+        "binding the backend must admit the same due occurrence without cooldown"
+    );
+    assert_eq!(backend.calls(), 0);
+    assert_eq!(
+        load_run_records(&dashboard_root, 10).await.unwrap(),
+        vec![first.ledger_record],
+        "advisory admission must not append an effectful terminal"
+    );
+    assert!(
+        tracedecay_automation_runtime::automation::run_ledger::find_run_record_exact_bounded(
+            &dashboard_root,
+            occurrence,
+        )
+        .await
+        .unwrap()
+        .is_none()
+    );
 }
 
 #[tokio::test]
@@ -1206,15 +1287,22 @@ async fn scheduler_prefilter_live_lock_wins_over_not_due_summary() {
 
     let config = enabled_job_config();
     let occurrence = "locked-prefilter-occurrence";
-    let prefilter =
-        evaluate_and_record_scheduler_skip(&dashboard_root, &config, None, &job, occurrence, None)
-            .await
-            .unwrap()
-            .expect("the live job lock must produce a diagnostic");
+    let executable = tempfile::NamedTempFile::new().unwrap();
+    let backend = ContentBackend::new("unused").with_executable(executable.path());
+    let prefilter = evaluate_and_record_scheduler_skip(
+        &dashboard_root,
+        &config,
+        backend.executable(),
+        &job,
+        occurrence,
+        None,
+    )
+    .await
+    .unwrap()
+    .expect("the live job lock must produce a diagnostic");
     assert_eq!(prefilter.report["reason"], json!("scheduler_lock_active"));
     assert_ne!(prefilter.run_id, occurrence);
 
-    let backend = ContentBackend::new("unused");
     let admitted_job = tracedecay_automation_runtime::automation::jobs::admit_job(&job).unwrap();
     let retained = run_user_job_with_backend_for_retained_settlement(
         &dashboard_root,
@@ -1262,12 +1350,21 @@ async fn retained_scheduler_runner_reacquires_after_due_prefilter() {
     job.schedule = Some("every 1h".to_owned());
     let config = enabled_job_config();
     let occurrence = "prefilter-gap-occurrence";
+    let executable = tempfile::NamedTempFile::new().unwrap();
+    let backend = ContentBackend::new("unused").with_executable(executable.path());
 
     assert!(
-        evaluate_and_record_scheduler_skip(&dashboard_root, &config, None, &job, occurrence, None)
-            .await
-            .unwrap()
-            .is_none(),
+        evaluate_and_record_scheduler_skip(
+            &dashboard_root,
+            &config,
+            backend.executable(),
+            &job,
+            occurrence,
+            None,
+        )
+        .await
+        .unwrap()
+        .is_none(),
         "a new interval job is advisory-due before effect admission"
     );
     let _raced_in =
@@ -1276,7 +1373,6 @@ async fn retained_scheduler_runner_reacquires_after_due_prefilter() {
             .unwrap()
             .expect("the advisory prefilter must release its reservation lock");
 
-    let backend = ContentBackend::new("unused");
     let admitted_job = tracedecay_automation_runtime::automation::jobs::admit_job(&job).unwrap();
     let retained = run_user_job_with_backend_for_retained_settlement(
         &dashboard_root,

From 080420c84af1d8b7e271583aeffadc3a44074be1 Mon Sep 17 00:00:00 2001
From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com>
Date: Sun, 8 Nov 2026 13:55:25 -0800
Subject: [PATCH 04/14] fix(sessions): serve live projection during historical
 ingest

---
 .../worker.rs                                 | 73 +++++++++++--------
 .../worker_tests.rs                           | 44 ++++++-----
 2 files changed, 69 insertions(+), 48 deletions(-)

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 5bc41596610..87611566be5 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
@@ -7,6 +7,7 @@ use std::sync::PoisonError;
 use std::sync::atomic::Ordering;
 use std::time::{Duration, Instant};
 
+use futures_util::future::{FusedFuture, FutureExt};
 use tracedecay_lcm::LcmError;
 use tracedecay_store::{
     SessionRefreshBeginOrJoinRequestV1, SessionRefreshCompletionRequestV1,
@@ -246,7 +247,7 @@ pub(super) async fn run_session_temporal_refresh_scheduler(
     let mut history_release = None;
     let mut granted_history_admission = None;
     let mut projection_joins: Vec<SessionProjectionJoin> = Vec::new();
-    let live_projection = LiveProjectionDuringSummary {
+    let live_projection = LiveProjection {
         database: &database,
         state: &state,
         projector: projector.as_ref(),
@@ -264,16 +265,18 @@ pub(super) async fn run_session_temporal_refresh_scheduler(
             // Busy before any wake is consumed: a waiter must never see its
             // wake taken while the worker still reads idle.
             state.mark_worker_busy();
-            let mut projection_requested = state.take_dirty();
             let mut history_requested = state.take_historical_dirty();
             // A wake that arrives while the admitted window is still
-            // unpublished must not start the next history window. That window
-            // is the out-of-scope cursor sweep, and it holds this worker until
-            // it returns, so the newest day never leaves its building generation.
-            if state.history_held_for_projection() {
+            // unpublished must not start the next history window. Publish
+            // the newest day before admitting the out-of-scope cursor sweep.
+            let holding_projection = state.history_held_for_projection();
+            if holding_projection {
                 history_requested = false;
-                projection_requested = true;
             }
+            // A history window polls live projection alongside ingest, so leave
+            // its dirty flag available until that work can actually start.
+            let mut projection_requested =
+                !history_requested && (state.take_dirty() || holding_projection);
             if !projection_requested && !history_requested && !summary_recheck {
                 break;
             }
@@ -289,6 +292,7 @@ pub(super) async fn run_session_temporal_refresh_scheduler(
             let granted = granted_history_admission
                 .take()
                 .filter(|_| history_requested && !history_suppressed);
+            let mut report = SessionTemporalRefreshPassReport::default();
             let history_outcome = if history_requested && !history_suppressed {
                 let history_pass: Pin<Box<dyn Future<Output = HistoryPassReport> + Send + '_>> =
                     Box::pin(session_history_refresh(
@@ -297,11 +301,25 @@ pub(super) async fn run_session_temporal_refresh_scheduler(
                         granted,
                         &state,
                     ));
-                let (outcome, release) = tracing::Instrument::instrument(
+                let history_pass = tracing::Instrument::instrument(
                     history_pass,
                     tracing::trace_span!("daemon.scheduler.session_temporal.history"),
                 )
-                .await;
+                .fuse();
+                tokio::pin!(history_pass);
+                // A hook can arrive before or during this source window. Keep the
+                // admitted ingestor running while publishing its already-durable input.
+                let Some(((outcome, release), _)) = live_projection
+                    .wait_for(&mut history_pass, &mut projection_joins, &mut report, false)
+                    .await
+                else {
+                    // Retirement owns this join. Cancellation has signalled the
+                    // ingestor, but its store and admission stay leased until it exits.
+                    if !history_pass.is_terminated() {
+                        history_pass.await;
+                    }
+                    return;
+                };
                 history_release = release;
                 Some(outcome)
             } else {
@@ -371,10 +389,7 @@ pub(super) async fn run_session_temporal_refresh_scheduler(
                 blocked_history_attempts = 0;
                 blocked_history_until = None;
             }
-            let mut report = if projection_requested
-                || state.has_requests()
-                || history_requires_projection
-            {
+            if projection_requested || state.has_requests() || history_requires_projection {
                 let projection: Pin<
                     Box<dyn Future<Output = SessionTemporalRefreshPassReport> + Send + '_>,
                 > = Box::pin(session_projection_refresh(
@@ -389,14 +404,12 @@ pub(super) async fn run_session_temporal_refresh_scheduler(
                     tracing::trace_span!("daemon.scheduler.session_temporal.projection"),
                 );
                 tokio::pin!(pass);
-                tokio::select! {
+                report = tokio::select! {
                     biased;
                     () = tracing::Instrument::instrument(state.wait_for_cancellation(), tracing::trace_span!("daemon.scheduler.session_temporal.projection_cancel")) => return,
                     report = &mut pass => report,
-                }
-            } else {
-                SessionTemporalRefreshPassReport::default()
-            };
+                };
+            }
             if state.cancelled.load(Ordering::Acquire) {
                 return;
             }
@@ -937,17 +950,16 @@ async fn settle_projection_joins(
     }
 }
 
-/// Keeps one fair summary admission or in-flight summary page alive while the
-/// same worker services live temporal projection. Both futures keep polling:
-/// suspending a summary while it owns a writer could block the projection.
-struct LiveProjectionDuringSummary<'a> {
+/// Keeps history, summary admission, or a summary page in flight while serving
+/// live projection. A stage holding a writer must keep polling so it can release it.
+struct LiveProjection<'a> {
     database: &'a RegisteredGlobalDbLeaseV1,
     state: &'a Arc<SessionTemporalRefreshWakeState>,
     projector: &'a dyn SessionTemporalRefreshProjector,
     policy: SessionTemporalRefreshPolicy,
 }
 
-impl LiveProjectionDuringSummary<'_> {
+impl LiveProjection<'_> {
     async fn wait_for<F>(
         &self,
         stage: F,
@@ -971,7 +983,7 @@ impl LiveProjectionDuringSummary<'_> {
             self.state.requeue_projection();
             retry_at = Some(Instant::now() + session_refresh_retry_delay(class, retry_attempt));
         }
-        let mut summary_pass_owed = false;
+        let mut projected_during_stage = false;
         let result = loop {
             let projection_wake = async {
                 let wake = self.state.wake.notified();
@@ -999,7 +1011,7 @@ impl LiveProjectionDuringSummary<'_> {
             if !self.state.take_dirty() {
                 continue;
             }
-            summary_pass_owed |= !waiting_for_admission;
+            projected_during_stage |= !waiting_for_admission;
             self.state.mark_worker_busy();
             self.state.begin_pass();
             self.state.pass_count.fetch_add(1, Ordering::AcqRel);
@@ -1064,10 +1076,9 @@ impl LiveProjectionDuringSummary<'_> {
                 self.state.idle.notify_waiters();
             }
         };
-        // A running page may have read its inputs before the consumed wake.
-        // Its follow-up needs summary convergence, not another projection of
-        // the frontier this wake already served. Admission has no page yet.
-        Some((result, summary_pass_owed))
+        // A summary page may have read its inputs before this projection,
+        // while a queued admission still has its first page ahead of it.
+        Some((result, projected_during_stage))
     }
 }
 
@@ -2277,7 +2288,7 @@ mod tests {
                         entered.notify_one();
                         finish.notified().await;
                     };
-                    let live_projection = LiveProjectionDuringSummary {
+                    let live_projection = LiveProjection {
                         database: &database,
                         state: &state,
                         projector: &CanonicalSessionTemporalProjector,
@@ -2360,7 +2371,7 @@ mod tests {
                     finish.notified().await;
                     17usize
                 };
-                let live_projection = LiveProjectionDuringSummary {
+                let live_projection = LiveProjection {
                     database: &database,
                     state: &state,
                     projector: &CanonicalSessionTemporalProjector,
diff --git a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker_tests.rs b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker_tests.rs
index bf9b23cf40e..ad619330ce0 100644
--- a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker_tests.rs
+++ b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker_tests.rs
@@ -333,12 +333,9 @@ impl SessionHistoricalIngestor for GatedHistory {
     fn cancel(&self) {}
 }
 
-/// A hook ingest joins the refresh worker with a wake. When that wake lands
-/// while a history pass is running, the pass's projection serves it and no
-/// further pass starts, so the join must end with that pass rather than wait
-/// out its deadline.
+/// An already admitted history scan must not hold a live hook behind its window.
 #[tokio::test]
-async fn a_wake_served_by_the_running_pass_joins_that_pass() {
+async fn a_live_session_publishes_while_a_history_window_is_running() {
     let harness = RegisteredGlobalDbHarness::open("refresh-wake-served-mid-pass").await;
     let state = Arc::new(SessionTemporalRefreshWakeState::default());
     let history = Arc::new(GatedHistory {
@@ -357,23 +354,36 @@ async fn a_wake_served_by_the_running_pass_joins_that_pass() {
     ));
     history.entered.notified().await;
 
+    let session_id = SessionId::new("live-during-history").expect("session id");
+    let store = SessionTemporalStore::new(harness.registered.as_ref());
+    let begun = store
+        .begin_or_join_session_refresh(SessionRefreshBeginOrJoinRequestV1::new(
+            session_id.clone(),
+            SessionRefreshFrontierV1::new(0, 0).expect("empty frontier"),
+        ))
+        .await
+        .expect("running refresh");
     let handle = state.handle();
-    let join = handle.wake_and_wait_for_projection(Duration::from_secs(5));
-    tokio::pin!(join);
-    // The first poll registers the join and delivers the wake.
-    assert!(
-        std::future::poll_fn(|cx| std::task::Poll::Ready(join.as_mut().poll(cx)))
-            .await
-            .is_pending()
-    );
+    let published = handle
+        .wake_and_wait_for_session_projection(&session_id, Duration::from_secs(5))
+        .await;
     history.release.notify_one();
+    state.cancel();
+    worker.await.expect("refresh worker exits on cancellation");
 
     assert!(
-        join.await,
-        "the pass that served the wake must end the join"
+        published,
+        "the live session must publish while history is held"
     );
-    state.cancel();
-    worker.await.expect("refresh worker exits on cancellation");
+    let receipt = store
+        .session_refresh_receipt(SessionRefreshReceiptRequestV1::new(
+            begun.operation_id().clone(),
+            begun.session_id().clone(),
+        ))
+        .await
+        .expect("read publication receipt")
+        .expect("the live generation must be durable");
+    assert_eq!(receipt.state(), SessionRefreshTerminalStateV1::Complete);
 }
 
 /// Every history window advances but waits for the test before returning.

From fcc6072c3adc621edd466c9a93696f8480816553 Mon Sep 17 00:00:00 2001
From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com>
Date: Sun, 8 Nov 2026 13:55:27 -0800
Subject: [PATCH 05/14] fix(lcm): serialize summary publication per store and
 session

---
 .../src/lcm_effects.rs                        | 45 ++++++++++++++++++-
 .../src/lcm_effects/tests.rs                  | 20 +++++++++
 2 files changed, 64 insertions(+), 1 deletion(-)

diff --git a/crates/tracedecay-session-runtime/src/lcm_effects.rs b/crates/tracedecay-session-runtime/src/lcm_effects.rs
index 678df440765..b58bade23c9 100644
--- a/crates/tracedecay-session-runtime/src/lcm_effects.rs
+++ b/crates/tracedecay-session-runtime/src/lcm_effects.rs
@@ -1,9 +1,13 @@
 use std::future::Future;
+use std::path::PathBuf;
 use std::pin::Pin;
-use std::sync::Arc;
+use std::sync::{Arc, LazyLock};
 use std::time::Duration;
 
+use tokio::sync::{Mutex, OwnedMutexGuard};
 use tracedecay_contracts::{CancellationSignal, Deadline};
+use tracedecay_runtime_core::native_runtime::{self, TimeoutError};
+use tracedecay_runtime_core::weak_registry::WeakRegistry;
 use tracedecay_temporal_query::execution::ExecutionControl;
 
 use tracedecay_domain::SessionId;
@@ -110,10 +114,48 @@ impl DaemonLcmEffectService {
         }
     }
 
+    /// Planning and publishing use separate transactions around the provider
+    /// call. A foreground hook and retained convergence for the same session
+    /// must re-plan after the preceding publication before launching a model.
+    async fn compression_turn(
+        &self,
+        request: &LcmCompressionRequest,
+    ) -> Result<OwnedMutexGuard<()>, LcmError> {
+        type SessionTurns = WeakRegistry<(PathBuf, String, String), Mutex<()>>;
+        static TURNS: LazyLock<SessionTurns> = LazyLock::new(WeakRegistry::new);
+        let key = (
+            self.db.db_path().to_path_buf(),
+            request.provider.clone(),
+            request.session_id.clone(),
+        );
+        let (turn, _) = TURNS.get_or_insert_with(key, || Arc::new(Mutex::new(())));
+        let remaining = self.control.remaining()?;
+        if let Ok(guard) = Arc::clone(&turn).try_lock_owned() {
+            return Ok(guard);
+        }
+        let cancelled = async {
+            match &self.control.cancellation {
+                Some(signal) => signal.cancelled().await,
+                None => std::future::pending().await,
+            }
+        };
+        tokio::select! {
+            biased;
+            () = cancelled => Err(LcmError::Cancelled),
+            result = native_runtime::timeout(remaining, turn.lock_owned()) => result.map_err(|error| {
+                match error {
+                    TimeoutError::Elapsed => LcmError::DeadlineExceeded,
+                    TimeoutError::TimerUnavailable(reason) => LcmError::Io(reason),
+                }
+            }),
+        }
+    }
+
     pub(super) async fn compress(
         &self,
         mut request: LcmCompressionRequest,
     ) -> Result<LcmCompressionResponse, LcmError> {
+        let _turn = self.compression_turn(&request).await?;
         // Observation-projected sessions land raw rows without ingest
         // protection; the verified raw loads inside compression reject them.
         // Hydrate the canonical sanitization receipts first so the daemon
@@ -197,6 +239,7 @@ impl DaemonLcmEffectService {
         mut request: LcmCompressionRequest,
         convergence_candidate: &tracedecay_lcm::summary_convergence::LcmSummaryConvergenceCandidate,
     ) -> Result<tracedecay_lcm::summary_convergence::LcmBoundedCompressionResponse, LcmError> {
+        let _turn = self.compression_turn(&request).await?;
         if request.summarizer.has_nonempty_provided_summary() {
             return self
                 .commit_retained_compression(&request, Some(convergence_candidate), None)
diff --git a/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs b/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs
index ee3ab998e87..c268d65ea18 100644
--- a/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs
+++ b/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs
@@ -91,6 +91,26 @@ fn compression_request(session_id: &str) -> LcmCompressionRequest {
     }
 }
 
+#[tokio::test]
+async fn a_queued_compression_honors_cancellation_without_blocking_other_sessions() {
+    let harness = RegisteredGlobalDbHarness::open("lcm-session-compression-turn").await;
+    let signal = CancellationSignal::active("queued-compression").unwrap();
+    let owner = DaemonLcmEffectService::new(harness.registered.clone(), None, None, None);
+    let caller = DaemonLcmEffectService::new(harness.registered.clone(), None, Some(&signal), None);
+    let first = compression_request("first-session");
+    let _held = owner.compression_turn(&first).await.unwrap();
+    let waiting = caller.compress(first);
+    tokio::pin!(waiting);
+    assert!(futures_util::poll!(&mut waiting).is_pending());
+
+    let _independent = owner
+        .compression_turn(&compression_request("other-session"))
+        .await
+        .unwrap();
+    signal.cancel(UtcMicros(1));
+    assert!(matches!(waiting.await, Err(LcmError::Cancelled)));
+}
+
 fn provided_summary(summary_text: impl Into<String>) -> LcmSummarizerMode {
     LcmSummarizerMode::Provided {
         summary_text: summary_text.into(),

From 05ab8b5d5069f482d00f32e122138d6e39cb1239 Mon Sep 17 00:00:00 2001
From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com>
Date: Sun, 8 Nov 2026 14:34:04 -0800
Subject: [PATCH 06/14] test(lsp): measure acknowledgement phases on fleet
 failures

---
 .../src/invocation/dispatch.rs                |  4 +++
 .../src/invocation/lsp.rs                     |  7 ++++
 .../runtime_surface_acceptance.rs             | 35 ++++++++++++++++---
 3 files changed, 41 insertions(+), 5 deletions(-)

diff --git a/crates/tracedecay-daemon-service/src/invocation/dispatch.rs b/crates/tracedecay-daemon-service/src/invocation/dispatch.rs
index 84cbdb3f42a..df06999f06a 100644
--- a/crates/tracedecay-daemon-service/src/invocation/dispatch.rs
+++ b/crates/tracedecay-daemon-service/src/invocation/dispatch.rs
@@ -142,6 +142,8 @@ impl DaemonInvocationService {
     ) -> DaemonInvocationResponse {
         // Keep the admitted dispatch frame behind one allocation for every invocation entry point.
         Box::pin(async move {
+        let ack_measurement = (request.operation() == DaemonInvocationOperation::LspAcknowledge).then(std::time::Instant::now);
+        if ack_measurement.is_some() { tracing::debug!(target: "lsp_ack_measurement", "LSP_ACK_MEASUREMENT phase=dispatch elapsed_us=0"); }
         let request_id = request.request_id.clone();
         let cancellation_lease = if admitted_cancellation.is_none() {
             let Some(lease) = self.request_cancellations.register(&request_id) else {
@@ -179,6 +181,7 @@ impl DaemonInvocationService {
                 .await
             }
         };
+        if let Some(start) = ack_measurement { tracing::debug!(target: "lsp_ack_measurement", "LSP_ACK_MEASUREMENT phase=admitted elapsed_us={}", start.elapsed().as_micros()); }
         let project_runtime_admitted = runtimes.is_admitted();
         let registered_project_root = runtimes.resolved_root;
         let publication = runtimes.publication;
@@ -262,6 +265,7 @@ impl DaemonInvocationService {
         let now_ms = now_millis();
         tracing::Instrument::instrument(self.expire_sessions(now_ms), tracing::trace_span!("daemon.service.invocation.expire_sessions"))
         .await;
+        if let Some(start) = ack_measurement { tracing::debug!(target: "lsp_ack_measurement", "LSP_ACK_MEASUREMENT phase=expired_sessions elapsed_us={}", start.elapsed().as_micros()); }
         let feedback_service = runtimes.feedback_owner;
         let advisory_cycle = runtimes.advisory_cycle;
         let configuration_runtime = runtimes.configuration;
diff --git a/crates/tracedecay-daemon-service/src/invocation/lsp.rs b/crates/tracedecay-daemon-service/src/invocation/lsp.rs
index 7c19fb938f7..135efecca9a 100644
--- a/crates/tracedecay-daemon-service/src/invocation/lsp.rs
+++ b/crates/tracedecay-daemon-service/src/invocation/lsp.rs
@@ -769,10 +769,14 @@ impl DaemonInvocationService {
         session: DaemonLspSessionAccess,
         now_ms: u64,
     ) -> DaemonInvocationResponse {
+        let start = std::time::Instant::now();
+        let phase = |name| tracing::debug!(target: "lsp_ack_measurement", "LSP_ACK_MEASUREMENT phase={name} elapsed_us={}", start.elapsed().as_micros());
+        phase("authenticate");
         let access = match self.authenticate(lsp_registry, session, now_ms).await {
             Ok(access) => access,
             Err(problem) => return DaemonInvocationResponse::problem(request_id, problem),
         };
+        phase("acquire_sessions");
         let mut sessions = self.lsp_sessions.lock().await;
         let Some(session) = sessions.get_mut(access.session_id()) else {
             return DaemonInvocationResponse::problem(
@@ -780,13 +784,16 @@ impl DaemonInvocationService {
                 DaemonInvocationProblem::NotFoundOrNotAuthorized,
             );
         };
+        phase("acknowledge_actor");
         let acknowledged = session.actor.acknowledge_outbound();
         if acknowledged {
+            phase("settle_receipt");
             let _ = session.settle_in_flight_delivery(
                 tracedecay_domain::DeliverySettlementOutcomeV1::Delivered,
                 None,
             );
         }
+        phase("reply");
         DaemonInvocationResponse::with_outcome(
             request_id,
             DaemonInvocationOutcome::LspAcknowledged { acknowledged },
diff --git a/crates/tracedecay/tests/runtime_acceptance_suite/runtime_surface_acceptance.rs b/crates/tracedecay/tests/runtime_acceptance_suite/runtime_surface_acceptance.rs
index f962270f60c..de779470098 100644
--- a/crates/tracedecay/tests/runtime_acceptance_suite/runtime_surface_acceptance.rs
+++ b/crates/tracedecay/tests/runtime_acceptance_suite/runtime_surface_acceptance.rs
@@ -93,6 +93,26 @@ struct RuntimeFixture {
     _environment: common::IsolatedHome,
 }
 
+struct AckTimings(PathBuf);
+
+impl Drop for AckTimings {
+    fn drop(&mut self) {
+        if !std::thread::panicking() {
+            return;
+        }
+        let Ok(events) = std::fs::read_to_string(&self.0) else {
+            return;
+        };
+        let phases = events.lines().filter_map(|line| {
+            line.split_once("LSP_ACK_MEASUREMENT ")
+                .and_then(|(_, phase)| line.split_whitespace().next().map(|time| (time, phase)))
+        });
+        for (time, phase) in phases.rev().take(24).collect::<Vec<_>>().into_iter().rev() {
+            eprintln!("LSP ACK timing {time} {phase}");
+        }
+    }
+}
+
 impl RuntimeFixture {
     fn home(&self) -> &Path {
         self._environment.home()
@@ -221,7 +241,9 @@ async fn lsp_runtime_fixture() -> RuntimeFixture {
     let daemon_log = std::fs::File::create(&daemon_events).expect("create isolated LSP daemon log");
     let daemon = common::spawn_tracedecay_daemon_with(environment.home(), |command| {
         environment.apply_toolchain_env(command);
-        command.env("RUST_LOG", "info").stderr(daemon_log);
+        command
+            .env("RUST_LOG", "info,lsp_ack_measurement=debug")
+            .stderr(daemon_log);
     });
     let output = common::tracedecay_command_with_home(environment.home())
         .arg("init")
@@ -427,10 +449,12 @@ async fn poll_lsp_response(session: &mut DaemonLspSessionClient, response_id: u6
                 let value: Value =
                     serde_json::from_slice(frame.as_slice()).expect("daemon LSP JSON");
                 let (operation_deadline, cancellation) = lsp_control();
-                session
-                    .acknowledge_daemon_frame(operation_deadline, cancellation)
-                    .await
-                    .expect("acknowledge daemon LSP frame");
+                let sent_at = wall_clock_micros().0;
+                let ack_start = std::time::Instant::now();
+                session.acknowledge_daemon_frame(operation_deadline, cancellation).await.unwrap_or_else(|error| {
+                    panic!("acknowledge daemon LSP frame: {error:?}; requested id {response_id}, response id {:?}, notification {}, sent_at_utc_us {sent_at}, client elapsed {:?}",
+                        value.get("id").and_then(Value::as_u64), value.get("method").is_some(), ack_start.elapsed())
+                });
                 if value.get("id").and_then(Value::as_u64) == Some(response_id) {
                     return value;
                 }
@@ -2459,6 +2483,7 @@ async fn stdio_bridge_exits_successfully_after_client_shutdown_and_exit() {
 #[tokio::test(flavor = "multi_thread")]
 async fn production_lsp_negotiates_and_projects_canonical_context() {
     let fixture = lsp_runtime_fixture().await;
+    let _ack_timings = AckTimings(fixture._environment.scratch().join("runtime-daemon.log"));
     let client_project = fixture.client_project_path();
     let root_uri = url::Url::from_directory_path(&client_project)
         .expect("project root URI")

From 06fb3fee4b3203408a308b7a7ec7e545f1e1d475 Mon Sep 17 00:00:00 2001
From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com>
Date: Sun, 8 Nov 2026 15:06:23 -0800
Subject: [PATCH 07/14] revert(lsp): remove temporary acknowledgement
 diagnostics

---
 .../src/invocation/dispatch.rs                |  4 ---
 .../src/invocation/lsp.rs                     |  7 ----
 .../runtime_surface_acceptance.rs             | 35 +++----------------
 3 files changed, 5 insertions(+), 41 deletions(-)

diff --git a/crates/tracedecay-daemon-service/src/invocation/dispatch.rs b/crates/tracedecay-daemon-service/src/invocation/dispatch.rs
index df06999f06a..84cbdb3f42a 100644
--- a/crates/tracedecay-daemon-service/src/invocation/dispatch.rs
+++ b/crates/tracedecay-daemon-service/src/invocation/dispatch.rs
@@ -142,8 +142,6 @@ impl DaemonInvocationService {
     ) -> DaemonInvocationResponse {
         // Keep the admitted dispatch frame behind one allocation for every invocation entry point.
         Box::pin(async move {
-        let ack_measurement = (request.operation() == DaemonInvocationOperation::LspAcknowledge).then(std::time::Instant::now);
-        if ack_measurement.is_some() { tracing::debug!(target: "lsp_ack_measurement", "LSP_ACK_MEASUREMENT phase=dispatch elapsed_us=0"); }
         let request_id = request.request_id.clone();
         let cancellation_lease = if admitted_cancellation.is_none() {
             let Some(lease) = self.request_cancellations.register(&request_id) else {
@@ -181,7 +179,6 @@ impl DaemonInvocationService {
                 .await
             }
         };
-        if let Some(start) = ack_measurement { tracing::debug!(target: "lsp_ack_measurement", "LSP_ACK_MEASUREMENT phase=admitted elapsed_us={}", start.elapsed().as_micros()); }
         let project_runtime_admitted = runtimes.is_admitted();
         let registered_project_root = runtimes.resolved_root;
         let publication = runtimes.publication;
@@ -265,7 +262,6 @@ impl DaemonInvocationService {
         let now_ms = now_millis();
         tracing::Instrument::instrument(self.expire_sessions(now_ms), tracing::trace_span!("daemon.service.invocation.expire_sessions"))
         .await;
-        if let Some(start) = ack_measurement { tracing::debug!(target: "lsp_ack_measurement", "LSP_ACK_MEASUREMENT phase=expired_sessions elapsed_us={}", start.elapsed().as_micros()); }
         let feedback_service = runtimes.feedback_owner;
         let advisory_cycle = runtimes.advisory_cycle;
         let configuration_runtime = runtimes.configuration;
diff --git a/crates/tracedecay-daemon-service/src/invocation/lsp.rs b/crates/tracedecay-daemon-service/src/invocation/lsp.rs
index 135efecca9a..7c19fb938f7 100644
--- a/crates/tracedecay-daemon-service/src/invocation/lsp.rs
+++ b/crates/tracedecay-daemon-service/src/invocation/lsp.rs
@@ -769,14 +769,10 @@ impl DaemonInvocationService {
         session: DaemonLspSessionAccess,
         now_ms: u64,
     ) -> DaemonInvocationResponse {
-        let start = std::time::Instant::now();
-        let phase = |name| tracing::debug!(target: "lsp_ack_measurement", "LSP_ACK_MEASUREMENT phase={name} elapsed_us={}", start.elapsed().as_micros());
-        phase("authenticate");
         let access = match self.authenticate(lsp_registry, session, now_ms).await {
             Ok(access) => access,
             Err(problem) => return DaemonInvocationResponse::problem(request_id, problem),
         };
-        phase("acquire_sessions");
         let mut sessions = self.lsp_sessions.lock().await;
         let Some(session) = sessions.get_mut(access.session_id()) else {
             return DaemonInvocationResponse::problem(
@@ -784,16 +780,13 @@ impl DaemonInvocationService {
                 DaemonInvocationProblem::NotFoundOrNotAuthorized,
             );
         };
-        phase("acknowledge_actor");
         let acknowledged = session.actor.acknowledge_outbound();
         if acknowledged {
-            phase("settle_receipt");
             let _ = session.settle_in_flight_delivery(
                 tracedecay_domain::DeliverySettlementOutcomeV1::Delivered,
                 None,
             );
         }
-        phase("reply");
         DaemonInvocationResponse::with_outcome(
             request_id,
             DaemonInvocationOutcome::LspAcknowledged { acknowledged },
diff --git a/crates/tracedecay/tests/runtime_acceptance_suite/runtime_surface_acceptance.rs b/crates/tracedecay/tests/runtime_acceptance_suite/runtime_surface_acceptance.rs
index de779470098..f962270f60c 100644
--- a/crates/tracedecay/tests/runtime_acceptance_suite/runtime_surface_acceptance.rs
+++ b/crates/tracedecay/tests/runtime_acceptance_suite/runtime_surface_acceptance.rs
@@ -93,26 +93,6 @@ struct RuntimeFixture {
     _environment: common::IsolatedHome,
 }
 
-struct AckTimings(PathBuf);
-
-impl Drop for AckTimings {
-    fn drop(&mut self) {
-        if !std::thread::panicking() {
-            return;
-        }
-        let Ok(events) = std::fs::read_to_string(&self.0) else {
-            return;
-        };
-        let phases = events.lines().filter_map(|line| {
-            line.split_once("LSP_ACK_MEASUREMENT ")
-                .and_then(|(_, phase)| line.split_whitespace().next().map(|time| (time, phase)))
-        });
-        for (time, phase) in phases.rev().take(24).collect::<Vec<_>>().into_iter().rev() {
-            eprintln!("LSP ACK timing {time} {phase}");
-        }
-    }
-}
-
 impl RuntimeFixture {
     fn home(&self) -> &Path {
         self._environment.home()
@@ -241,9 +221,7 @@ async fn lsp_runtime_fixture() -> RuntimeFixture {
     let daemon_log = std::fs::File::create(&daemon_events).expect("create isolated LSP daemon log");
     let daemon = common::spawn_tracedecay_daemon_with(environment.home(), |command| {
         environment.apply_toolchain_env(command);
-        command
-            .env("RUST_LOG", "info,lsp_ack_measurement=debug")
-            .stderr(daemon_log);
+        command.env("RUST_LOG", "info").stderr(daemon_log);
     });
     let output = common::tracedecay_command_with_home(environment.home())
         .arg("init")
@@ -449,12 +427,10 @@ async fn poll_lsp_response(session: &mut DaemonLspSessionClient, response_id: u6
                 let value: Value =
                     serde_json::from_slice(frame.as_slice()).expect("daemon LSP JSON");
                 let (operation_deadline, cancellation) = lsp_control();
-                let sent_at = wall_clock_micros().0;
-                let ack_start = std::time::Instant::now();
-                session.acknowledge_daemon_frame(operation_deadline, cancellation).await.unwrap_or_else(|error| {
-                    panic!("acknowledge daemon LSP frame: {error:?}; requested id {response_id}, response id {:?}, notification {}, sent_at_utc_us {sent_at}, client elapsed {:?}",
-                        value.get("id").and_then(Value::as_u64), value.get("method").is_some(), ack_start.elapsed())
-                });
+                session
+                    .acknowledge_daemon_frame(operation_deadline, cancellation)
+                    .await
+                    .expect("acknowledge daemon LSP frame");
                 if value.get("id").and_then(Value::as_u64) == Some(response_id) {
                     return value;
                 }
@@ -2483,7 +2459,6 @@ async fn stdio_bridge_exits_successfully_after_client_shutdown_and_exit() {
 #[tokio::test(flavor = "multi_thread")]
 async fn production_lsp_negotiates_and_projects_canonical_context() {
     let fixture = lsp_runtime_fixture().await;
-    let _ack_timings = AckTimings(fixture._environment.scratch().join("runtime-daemon.log"));
     let client_project = fixture.client_project_path();
     let root_uri = url::Url::from_directory_path(&client_project)
         .expect("project root URI")

From 20b91e15220ae9f5cb59be3fb75ed6c0abbc64d4 Mon Sep 17 00:00:00 2001
From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com>
Date: Sun, 8 Nov 2026 14:47:10 -0800
Subject: [PATCH 08/14] fix(sessions): reserve history preemption for waiting
 readers

---
 .../wake.rs                                   |   8 ++
 .../worker.rs                                 | 102 ++++++++++++++----
 .../temporal_projection/activation.rs         |  12 ++-
 3 files changed, 99 insertions(+), 23 deletions(-)

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 d4eea3369fb..95bc1c443f5 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
@@ -378,6 +378,14 @@ impl SessionTemporalRefreshWakeState {
         Some(receiver)
     }
 
+    pub(super) fn has_pending_projection_join(&self) -> bool {
+        self.projection_joins
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+            .iter()
+            .any(|join| !join.completion.is_closed())
+    }
+
     /// Take the joins covered by the projection about to start. A join that
     /// arrives during that projection remains queued for the next pass.
     pub(super) fn take_projection_joins(&self) -> Vec<SessionProjectionJoin> {
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 87611566be5..57a7a987f49 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
@@ -310,7 +310,12 @@ pub(super) async fn run_session_temporal_refresh_scheduler(
                 // A hook can arrive before or during this source window. Keep the
                 // admitted ingestor running while publishing its already-durable input.
                 let Some(((outcome, release), _)) = live_projection
-                    .wait_for(&mut history_pass, &mut projection_joins, &mut report, false)
+                    .wait_for(
+                        &mut history_pass,
+                        &mut projection_joins,
+                        &mut report,
+                        BackgroundStage::History,
+                    )
                     .await
                 else {
                     // Retirement owns this join. Cancellation has signalled the
@@ -478,7 +483,12 @@ pub(super) async fn run_session_temporal_refresh_scheduler(
                         state.mark_worker_idle();
                         state.idle.notify_waiters();
                         let Some((permit, _)) = live_projection
-                            .wait_for(&mut admission, &mut projection_joins, &mut report, true)
+                            .wait_for(
+                                &mut admission,
+                                &mut projection_joins,
+                                &mut report,
+                                BackgroundStage::SummaryAdmission,
+                            )
                             .await
                         else {
                             return;
@@ -511,7 +521,12 @@ pub(super) async fn run_session_temporal_refresh_scheduler(
                         );
                         tokio::pin!(page);
                         let Some((result, recheck)) = live_projection
-                            .wait_for(&mut page, &mut projection_joins, &mut report, false)
+                            .wait_for(
+                                &mut page,
+                                &mut projection_joins,
+                                &mut report,
+                                BackgroundStage::SummaryPage,
+                            )
                             .await
                         else {
                             return;
@@ -950,6 +965,13 @@ async fn settle_projection_joins(
     }
 }
 
+#[derive(Clone, Copy, PartialEq, Eq)]
+enum BackgroundStage {
+    History,
+    SummaryAdmission,
+    SummaryPage,
+}
+
 /// Keeps history, summary admission, or a summary page in flight while serving
 /// live projection. A stage holding a writer must keep polling so it can release it.
 struct LiveProjection<'a> {
@@ -960,12 +982,22 @@ struct LiveProjection<'a> {
 }
 
 impl LiveProjection<'_> {
+    fn projection_due(&self, stage: BackgroundStage, joins: &[SessionProjectionJoin]) -> bool {
+        // An admitted history window owns background recovery until its SQL
+        // source frontier settles. Only a waiting reader needs to publish
+        // ahead of that window.
+        self.state.dirty.load(Ordering::Acquire)
+            && (stage != BackgroundStage::History
+                || joins.iter().any(|join| !join.completion.is_closed())
+                || self.state.has_pending_projection_join())
+    }
+
     async fn wait_for<F>(
         &self,
         stage: F,
         joins: &mut Vec<SessionProjectionJoin>,
         report: &mut SessionTemporalRefreshPassReport,
-        waiting_for_admission: bool,
+        background: BackgroundStage,
     ) -> Option<(F::Output, bool)>
     where
         F: Future + Send,
@@ -983,19 +1015,20 @@ impl LiveProjection<'_> {
             self.state.requeue_projection();
             retry_at = Some(Instant::now() + session_refresh_retry_delay(class, retry_attempt));
         }
-        let mut projected_during_stage = false;
+        let mut recheck_summary = false;
         let result = loop {
             let projection_wake = async {
                 let wake = self.state.wake.notified();
+                if !self.projection_due(background, joins) {
+                    wake.await;
+                    return true;
+                }
                 if let Some(deadline) = retry_at {
                     tokio::select! {
                         () = wake => true,
                         fired = native_runtime::sleep_until(deadline) => !stop_without_timer(fired),
                     }
                 } else {
-                    if !self.state.dirty.load(Ordering::Acquire) {
-                        wake.await;
-                    }
                     true
                 }
             };
@@ -1008,10 +1041,10 @@ impl LiveProjection<'_> {
             if !woke {
                 return None;
             }
-            if !self.state.take_dirty() {
+            if !self.projection_due(background, joins) || !self.state.take_dirty() {
                 continue;
             }
-            projected_during_stage |= !waiting_for_admission;
+            recheck_summary |= background == BackgroundStage::SummaryPage;
             self.state.mark_worker_busy();
             self.state.begin_pass();
             self.state.pass_count.fetch_add(1, Ordering::AcqRel);
@@ -1071,14 +1104,14 @@ impl LiveProjection<'_> {
             if let Some(result) = completed_stage {
                 break result;
             }
-            if waiting_for_admission {
+            if background == BackgroundStage::SummaryAdmission {
                 self.state.mark_worker_idle();
                 self.state.idle.notify_waiters();
             }
         };
         // A summary page may have read its inputs before this projection,
         // while a queued admission still has its first page ahead of it.
-        Some((result, projected_during_stage))
+        Some((result, recheck_summary))
     }
 }
 
@@ -2267,12 +2300,45 @@ mod tests {
     use tracedecay_runtime_core::db::engine::params;
     use tracedecay_store::{SessionRefreshReceiptRequestV1, SessionRefreshTerminalStateV1};
 
+    #[tokio::test]
+    async fn a_history_window_leaves_background_projection_queued() {
+        let harness = RegisteredGlobalDbHarness::open("history-retains-background-work").await;
+        let state = Arc::new(SessionTemporalRefreshWakeState::default());
+        state.wake();
+        let live = LiveProjection {
+            database: &harness.registered,
+            state: &state,
+            projector: &CanonicalSessionTemporalProjector,
+            policy: SessionTemporalRefreshPolicy::default(),
+        };
+        let mut joins = Vec::new();
+        let mut report = SessionTemporalRefreshPassReport::default();
+        live.wait_for(
+            tokio::task::yield_now(),
+            &mut joins,
+            &mut report,
+            BackgroundStage::History,
+        )
+        .await
+        .expect("the history window finishes");
+        assert_eq!(state.pass_count.load(Ordering::Acquire), 0);
+        assert!(
+            state.take_dirty(),
+            "background recovery still owns its turn"
+        );
+    }
+
     #[tokio::test]
     async fn a_live_wake_during_summary_retains_one_convergence_pass() {
-        for (name, waiting_for_admission, cancel, owes_convergence) in [
-            ("completed-page", false, false, true),
-            ("completed-admission", true, false, false),
-            ("cancelled-page", false, true, false),
+        for (name, background, cancel, owes_convergence) in [
+            ("completed-page", BackgroundStage::SummaryPage, false, true),
+            (
+                "completed-admission",
+                BackgroundStage::SummaryAdmission,
+                false,
+                false,
+            ),
+            ("cancelled-page", BackgroundStage::SummaryPage, true, false),
         ] {
             let harness = RegisteredGlobalDbHarness::open(name).await;
             let state = Arc::new(SessionTemporalRefreshWakeState::default());
@@ -2297,7 +2363,7 @@ mod tests {
                     let mut joins = Vec::new();
                     let mut report = SessionTemporalRefreshPassReport::default();
                     live_projection
-                        .wait_for(stage, &mut joins, &mut report, waiting_for_admission)
+                        .wait_for(stage, &mut joins, &mut report, background)
                         .await
                 }
             });
@@ -2380,7 +2446,7 @@ mod tests {
                 let mut joins = Vec::new();
                 let mut report = SessionTemporalRefreshPassReport::default();
                 live_projection
-                    .wait_for(stage, &mut joins, &mut report, false)
+                    .wait_for(stage, &mut joins, &mut report, BackgroundStage::SummaryPage)
                     .await
             }
         });
diff --git a/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs b/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs
index 28a1f8d0d5c..340c44db1f5 100644
--- a/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs
+++ b/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs
@@ -1,5 +1,5 @@
 use std::future::Future;
-use std::sync::Arc;
+use std::sync::{Arc, OnceLock};
 use std::time::Duration;
 
 use tracedecay_session_runtime::session_temporal_refresh_scheduler::SessionTemporalRefreshSchedulerRegistry;
@@ -328,14 +328,16 @@ async fn activation_is_pinned_to_the_generation_the_refresh_extends() {
 #[derive(Default)]
 struct GatedHistory {
     entered: tokio::sync::Notify,
-    release: tokio::sync::Notify,
+    release: OnceLock<()>,
 }
 
 impl SessionHistoricalIngestor for GatedHistory {
     fn run_pass(&self) -> SessionHistoricalIngestPass<'_> {
         Box::pin(async move {
             self.entered.notify_one();
-            self.release.notified().await;
+            // Both readers must register before the native worker can capture
+            // the superseded candidate. A yielding history pass serves live joins.
+            self.release.wait();
             SessionHistoricalIngestOutcome::Complete.into()
         })
     }
@@ -345,7 +347,7 @@ impl SessionHistoricalIngestor for GatedHistory {
     }
 
     fn cancel(&self) {
-        self.release.notify_one();
+        let _ = self.release.set(());
     }
 }
 
@@ -409,7 +411,7 @@ async fn reset_requested_during_a_refresh_supersedes_its_candidate() {
         .await,
         (true, true)
     );
-    history.release.notify_one();
+    history.release.set(()).unwrap();
     assert_eq!(
         tokio::join!(named_join, aggregate_join),
         (false, false),

From 8c96a2797f87c39460d4a03a902e9bde34b49a6d Mon Sep 17 00:00:00 2001
From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com>
Date: Sun, 8 Nov 2026 15:48:35 -0800
Subject: [PATCH 09/14] test(sessions): yield refresh history gate instead of
 blocking worker

A synchronous OnceLock wait inside the asynchronous historical-ingest
fixture blocked the worker that must poll live projection concurrently.
Use a durable async watch latch to preserve release ordering while
exercising the real interleaving and supersession journey.

Refs #4481
---
 .../temporal_projection/activation.rs         | 28 +++++++++++++------
 1 file changed, 20 insertions(+), 8 deletions(-)

diff --git a/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs b/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs
index 340c44db1f5..e4562af19a8 100644
--- a/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs
+++ b/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs
@@ -1,5 +1,5 @@
 use std::future::Future;
-use std::sync::{Arc, OnceLock};
+use std::sync::Arc;
 use std::time::Duration;
 
 use tracedecay_session_runtime::session_temporal_refresh_scheduler::SessionTemporalRefreshSchedulerRegistry;
@@ -325,19 +325,31 @@ async fn activation_is_pinned_to_the_generation_the_refresh_extends() {
     );
 }
 
-#[derive(Default)]
 struct GatedHistory {
     entered: tokio::sync::Notify,
-    release: OnceLock<()>,
+    release: tokio::sync::watch::Sender<bool>,
+}
+
+impl Default for GatedHistory {
+    fn default() -> Self {
+        Self {
+            entered: tokio::sync::Notify::new(),
+            release: tokio::sync::watch::Sender::new(false),
+        }
+    }
 }
 
 impl SessionHistoricalIngestor for GatedHistory {
     fn run_pass(&self) -> SessionHistoricalIngestPass<'_> {
+        let mut release = self.release.subscribe();
         Box::pin(async move {
             self.entered.notify_one();
-            // Both readers must register before the native worker can capture
-            // the superseded candidate. A yielding history pass serves live joins.
-            self.release.wait();
+            // The watch latch preserves release even when it precedes this poll,
+            // while yielding the native worker to concurrent live projection.
+            release
+                .wait_for(|released| *released)
+                .await
+                .expect("history release sender remains open");
             SessionHistoricalIngestOutcome::Complete.into()
         })
     }
@@ -347,7 +359,7 @@ impl SessionHistoricalIngestor for GatedHistory {
     }
 
     fn cancel(&self) {
-        let _ = self.release.set(());
+        let _ = self.release.send_replace(true);
     }
 }
 
@@ -411,7 +423,7 @@ async fn reset_requested_during_a_refresh_supersedes_its_candidate() {
         .await,
         (true, true)
     );
-    history.release.set(()).unwrap();
+    assert!(!history.release.send_replace(true));
     assert_eq!(
         tokio::join!(named_join, aggregate_join),
         (false, false),

From c2f0c83b80788050dc1bdd25663920d4af571dc3 Mon Sep 17 00:00:00 2001
From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com>
Date: Sun, 8 Nov 2026 18:15:53 -0800
Subject: [PATCH 10/14] test(sessions): join each superseded publication
 independently

---
 .../temporal_projection/activation.rs         | 46 +++++++++++++------
 1 file changed, 32 insertions(+), 14 deletions(-)

diff --git a/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs b/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs
index e4562af19a8..a7bc3b04760 100644
--- a/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs
+++ b/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs
@@ -1,4 +1,3 @@
-use std::future::Future;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -363,8 +362,22 @@ impl SessionHistoricalIngestor for GatedHistory {
     }
 }
 
+#[derive(Clone, Copy, Debug)]
+enum PublicationScope {
+    Session,
+    Profile,
+}
+
 #[tokio::test]
 async fn reset_requested_during_a_refresh_supersedes_its_candidate() {
+    // Each request wakes the worker immediately. A second request may already
+    // belong to the replacement pass, so exercise each scope on its own reset.
+    for scope in [PublicationScope::Session, PublicationScope::Profile] {
+        assert_superseded_publication(scope).await;
+    }
+}
+
+async fn assert_superseded_publication(scope: PublicationScope) {
     let tmp = TempDir::new().unwrap();
     let (runtime, path, stale) = frontier_digest_fixture(&tmp, true).await;
     let session_id = stale.session_id().clone();
@@ -412,23 +425,28 @@ async fn reset_requested_during_a_refresh_supersedes_its_candidate() {
         )
         .await;
     history.entered.notified().await;
-    let named_join = wake.wake_and_wait_for_session_projection(&session_id, Duration::from_secs(5));
-    let aggregate_join = wake.wake_and_wait_for_projection(Duration::from_secs(5));
-    tokio::pin!(named_join, aggregate_join);
+    let timeout = Duration::from_secs(5);
+    let published = match scope {
+        PublicationScope::Session => {
+            wake.wake_and_wait_for_session_projection(&session_id, timeout)
+                .await
+        }
+        PublicationScope::Profile => wake.wake_and_wait_for_projection(timeout).await,
+    };
+    assert!(
+        !published,
+        "the superseded operation published for {scope:?}"
+    );
     assert_eq!(
-        std::future::poll_fn(|cx| std::task::Poll::Ready((
-            named_join.as_mut().poll(cx).is_pending(),
-            aggregate_join.as_mut().poll(cx).is_pending(),
-        )))
+        rows(
+            &path,
+            "SELECT operation_id FROM session_refresh_operations WHERE state = 'failed'"
+        )
         .await,
-        (true, true)
+        vec![stale.operation_id().as_str()],
+        "{scope:?} must observe the failed operation while history remains parked"
     );
     assert!(!history.release.send_replace(true));
-    assert_eq!(
-        tokio::join!(named_join, aggregate_join),
-        (false, false),
-        "the superseded operation must refuse both waiting publication joins"
-    );
     registry.wait_profile_idle(&path).await;
 
     assert_eq!(

From 63ff93ab9a2e93421aee8c27c64825ed7543ccb7 Mon Sep 17 00:00:00 2001
From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com>
Date: Mon, 9 Nov 2026 01:34:16 -0800
Subject: [PATCH 11/14] fix(accounting): report live monitor I/O failures

---
 .../tests/core_cli_suite/monitor_test.rs      |  3 +-
 .../src/monitor_ring.rs                       | 42 +++++++++++++++----
 crates/tracedecay/src/mcp/server/ledger.rs    | 12 ++++--
 3 files changed, 45 insertions(+), 12 deletions(-)

diff --git a/crates/tracedecay-cli/tests/core_cli_suite/monitor_test.rs b/crates/tracedecay-cli/tests/core_cli_suite/monitor_test.rs
index 27cdd9dab3d..674f012bc11 100644
--- a/crates/tracedecay-cli/tests/core_cli_suite/monitor_test.rs
+++ b/crates/tracedecay-cli/tests/core_cli_suite/monitor_test.rs
@@ -5,7 +5,8 @@ use tempfile::TempDir;
 fn write(dir: &Path, project: &Path, prefix: &str, tool: &str, delta: u64, before: u64) {
     tracedecay_session_memory::monitor_ring::write_entry_to(
         dir, project, prefix, tool, delta, before,
-    );
+    )
+    .expect("write monitor fixture entry");
 }
 
 /// Helper: open reader at a specific mmap dir.
diff --git a/crates/tracedecay-session-memory/src/monitor_ring.rs b/crates/tracedecay-session-memory/src/monitor_ring.rs
index 71140fbbddd..0ad4e350485 100644
--- a/crates/tracedecay-session-memory/src/monitor_ring.rs
+++ b/crates/tracedecay-session-memory/src/monitor_ring.rs
@@ -63,8 +63,8 @@ impl MonitorEntry {
 /// `dir`.
 ///
 /// `project_root` is used to derive the folder name. `prefix` identifies
-/// the tool suite (e.g. "tracedecay"). Best-effort: silently returns on
-/// any failure.
+/// the tool suite (e.g. "tracedecay"). Writer failures are returned to
+/// the caller so accounting can report a degraded live-monitor projection.
 pub fn write_entry_to(
     dir: &Path,
     project_root: &Path,
@@ -72,13 +72,13 @@ pub fn write_entry_to(
     tool_name: &str,
     delta: u64,
     before: u64,
-) {
-    let _ = std::fs::create_dir_all(dir);
+) -> std::io::Result<()> {
+    std::fs::create_dir_all(dir)?;
     let mmap_path = dir.join(MMAP_FILENAME);
     let project = project_root
         .file_name()
         .map_or_default(|n| n.to_string_lossy().to_string());
-    let _ = write_entry_inner(&mmap_path, prefix, &project, tool_name, delta, before);
+    write_entry_inner(&mmap_path, prefix, &project, tool_name, delta, before)
 }
 
 fn write_str(mmap: &mut memmap2::MmapMut, offset: usize, value: &str) {
@@ -326,7 +326,8 @@ mod tests {
     fn write_values(dir: &Path, values: std::ops::RangeInclusive<u64>) {
         let project = Path::new("/projects/alpha");
         for value in values {
-            write_entry_to(dir, project, "tracedecay", "search", value, 0);
+            write_entry_to(dir, project, "tracedecay", "search", value, 0)
+                .expect("write monitor entry");
         }
     }
 
@@ -442,16 +443,41 @@ mod tests {
         );
     }
 
+    #[test]
+    fn writer_reports_unavailable_directory_and_ring_without_clobbering_them() {
+        let dir = tempfile::tempdir().expect("monitor fixture directory");
+        let project = Path::new("/projects/alpha");
+
+        // A non-directory owner path cannot be turned into a monitor cache.
+        let occupied_dir = dir.path().join("occupied");
+        std::fs::write(&occupied_dir, b"owner data").expect("create occupied path");
+        write_entry_to(&occupied_dir, project, "tracedecay", "search", 7, 10)
+            .expect_err("a file used as the monitor directory must be refused");
+        assert_eq!(
+            std::fs::read(&occupied_dir).expect("read original owner file"),
+            b"owner data"
+        );
+
+        // Opening a directory as the mmap is a separate post-mkdir failure.
+        let occupied_ring = dir.path().join(MMAP_FILENAME);
+        std::fs::create_dir(&occupied_ring).expect("create occupied mmap path");
+        write_entry_to(dir.path(), project, "tracedecay", "search", 7, 10)
+            .expect_err("an occupied mmap path must be refused");
+        assert!(occupied_ring.is_dir());
+    }
+
     /// Writers no longer msync, so an already-open reader must still observe
     /// every entry through its existing mapping, without a refresh.
     #[test]
     fn open_reader_observes_every_unflushed_entry() {
         let dir = tempfile::tempdir().expect("monitor dir");
         let project = tempfile::tempdir().expect("project dir");
-        write_entry_to(dir.path(), project.path(), "tracedecay", "first", 1, 2);
+        write_entry_to(dir.path(), project.path(), "tracedecay", "first", 1, 2)
+            .expect("write initial monitor entry");
         let reader = MmapReader::open_at(dir.path()).expect("open monitor");
         for call in 1..40_u64 {
-            write_entry_to(dir.path(), project.path(), "tracedecay", "poll", call, 2);
+            write_entry_to(dir.path(), project.path(), "tracedecay", "poll", call, 2)
+                .expect("write monitor poll entry");
         }
         assert_eq!(reader.write_idx(), 40);
         let last = reader.entry(39).expect("last slot");
diff --git a/crates/tracedecay/src/mcp/server/ledger.rs b/crates/tracedecay/src/mcp/server/ledger.rs
index 3dda2e3b1ad..8bee6bd4bd7 100644
--- a/crates/tracedecay/src/mcp/server/ledger.rs
+++ b/crates/tracedecay/src/mcp/server/ledger.rs
@@ -216,10 +216,16 @@ impl McpServer {
                         &tool_name,
                         committed_saved_tokens,
                         raw_file_tokens,
-                    );
+                    )
                 });
-            if let Err(error) = monitor_write.await {
-                tracing::warn!(error = %error, "live monitor entry write failed");
+            match monitor_write.await {
+                Ok(Ok(())) => {}
+                Ok(Err(error)) => {
+                    tracing::warn!(%error, "live monitor entry I/O failed");
+                }
+                Err(error) => {
+                    tracing::warn!(%error, "live monitor entry task failed");
+                }
             }
         });
     }

From 024ca1c31ce6b0e10164c9bd74f2e0eaa41b44dd Mon Sep 17 00:00:00 2001
From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com>
Date: Mon, 9 Nov 2026 03:23:56 -0800
Subject: [PATCH 12/14] fix(accounting): gate savings history and analytics
 credit on project commit

Refs #4447. Independently scheduled analytics awaits an exact per-response
canonical project-store commit receipt. Failed or abandoned commits preserve
raw activity and cost but cannot add phantom saved-token history or totals.
Add a registered-store overflow and analytics regression.
---
 crates/tracedecay/src/mcp/server/ledger.rs   | 157 ++++++++++++++++++-
 crates/tracedecay/src/mcp/server/requests.rs |  42 ++++-
 2 files changed, 186 insertions(+), 13 deletions(-)

diff --git a/crates/tracedecay/src/mcp/server/ledger.rs b/crates/tracedecay/src/mcp/server/ledger.rs
index 8bee6bd4bd7..84c109710fd 100644
--- a/crates/tracedecay/src/mcp/server/ledger.rs
+++ b/crates/tracedecay/src/mcp/server/ledger.rs
@@ -173,7 +173,16 @@ impl McpServer {
         tool_name: &str,
         net_saved_tokens: u64,
         raw_file_tokens: u64,
-    ) {
+    ) -> Option<tokio::sync::oneshot::Receiver<bool>> {
+        // The independently scheduled analytics writer may report positive
+        // savings only after this exact response's store transaction commits.
+        // A dropped task closes the channel and means refusal, not success.
+        let (commit_sender, commit_receiver) = if net_saved_tokens == 0 {
+            (None, None)
+        } else {
+            let (sender, receiver) = tokio::sync::oneshot::channel();
+            (Some(sender), Some(receiver))
+        };
         // An unreadable baseline suppresses only the approximate display cache.
         // It must not disable future durable deltas if that read failure clears.
         let persist = (net_saved_tokens != 0).then(|| {
@@ -201,6 +210,9 @@ impl McpServer {
             } else {
                 0
             };
+            if let Some(sender) = commit_sender {
+                let _ = sender.send(committed_saved_tokens == net_saved_tokens);
+            }
             // The monitor entry opens, locks, and mmaps a file; keep that
             // off the async workers.
             // A server without an owning profile has no monitor ring.
@@ -228,6 +240,7 @@ impl McpServer {
                 }
             }
         });
+        commit_receiver
     }
 
     /// Resolves once every savings-ledger write spawned so far has
@@ -553,7 +566,7 @@ mod tests {
             .expect("tool call should produce a response");
         server.ledger_writes_settled().await;
         let accounted = tokens_saved.load(Ordering::Acquire);
-        server.spawn_token_accounting_persist(
+        let _ = server.spawn_token_accounting_persist(
             server.cg_snapshot().await,
             &project_root,
             "tracedecay_search",
@@ -588,7 +601,7 @@ mod tests {
         )
         .await;
 
-        first.spawn_token_accounting_persist(
+        let _ = first.spawn_token_accounting_persist(
             first.cg_snapshot().await,
             &project_root,
             "tracedecay_search",
@@ -598,7 +611,7 @@ mod tests {
         first.ledger_writes_settled().await;
         // The second server was created with the same baseline before the
         // first credit. Its own completion must add to the committed total.
-        second.spawn_token_accounting_persist(
+        let _ = second.spawn_token_accounting_persist(
             second.cg_snapshot().await,
             &project_root,
             "tracedecay_search",
@@ -650,7 +663,7 @@ mod tests {
             .data_dir()
             .to_path_buf();
 
-        server.spawn_token_accounting_persist(
+        let _ = server.spawn_token_accounting_persist(
             server.cg_snapshot().await,
             &project_root,
             "tracedecay_search",
@@ -680,7 +693,7 @@ mod tests {
             .record_token_savings(u64::MAX - 7)
             .await
             .expect("seed a durable lifetime overflow boundary");
-        server.spawn_token_accounting_persist(
+        let _ = server.spawn_token_accounting_persist(
             server.cg_snapshot().await,
             &project_root,
             "tracedecay_search",
@@ -712,6 +725,136 @@ mod tests {
         server.shutdown_background_tasks().await;
     }
 
+    #[tokio::test]
+    async fn refused_counter_commit_never_credits_independent_savings_history() {
+        use tracedecay_mcp::tools::response_trailers::ToolTokenAccounting;
+        use tracedecay_runtime_core::db::engine::Executor as _;
+
+        let (cg, _project, authority) =
+            super::super::writer_test_support::init_indexed_repo().await;
+        let project_root = cg.project_root().to_path_buf();
+        let server = super::super::McpServer::new_with_context(
+            super::super::writer_test_support::registered_context(cg, &authority),
+        )
+        .await;
+        let accounting_db = server
+            .accounting_db
+            .as_ref()
+            .expect("registered accounting writer")
+            .clone();
+        let normal = ToolTokenAccounting {
+            raw_file_tokens: 20,
+            response_tokens: 13,
+        };
+        let result = tracedecay_mcp::ToolResult::new(json!({"content": []}), Vec::new());
+
+        let first_commit = server.spawn_token_accounting_persist(
+            server.cg_snapshot().await,
+            &project_root,
+            "tracedecay_search",
+            normal.net_saved_tokens(),
+            normal.raw_file_tokens,
+        );
+        server.spawn_success_analytics(
+            &project_root,
+            "tracedecay_search",
+            &json!("request.first"),
+            &json!({}),
+            &None,
+            true,
+            None,
+            normal,
+            &result,
+            None,
+            None,
+            first_commit,
+        );
+        server.ledger_writes_settled().await;
+        let project_key = project_root.to_str().expect("UTF-8 fixture project");
+        let history = accounting_db
+            .sum_savings(Some(project_key), 0)
+            .await
+            .expect("registered savings history");
+        assert_eq!(history.saved_tokens, 7);
+        assert_eq!(history.calls, 1);
+
+        // A real checked-add failure prevents the second response's positive
+        // durable credit. The independent analytics writer used to race ahead
+        // and persist that failed response as another saved-token event.
+        server
+            .cg_snapshot()
+            .await
+            .record_token_savings(u64::MAX - 7)
+            .await
+            .expect("seed canonical overflow boundary");
+        let refused = ToolTokenAccounting {
+            raw_file_tokens: 30,
+            response_tokens: 29,
+        };
+        let second_commit = server.spawn_token_accounting_persist(
+            server.cg_snapshot().await,
+            &project_root,
+            "tracedecay_search",
+            refused.net_saved_tokens(),
+            refused.raw_file_tokens,
+        );
+        server.spawn_success_analytics(
+            &project_root,
+            "tracedecay_search",
+            &json!("request.refused"),
+            &json!({}),
+            &None,
+            true,
+            None,
+            refused,
+            &result,
+            None,
+            None,
+            second_commit,
+        );
+        server.ledger_writes_settled().await;
+
+        let history = accounting_db
+            .sum_savings(Some(project_key), 0)
+            .await
+            .expect("savings history after refused commit");
+        assert_eq!(history.saved_tokens, 7, "failed commit cannot add savings");
+        assert_eq!(history.calls, 1, "failed commit cannot add a savings row");
+        assert_eq!(
+            server.cg_snapshot().await.get_tokens_saved().await.unwrap(),
+            u64::MAX,
+            "refused accounting cannot alter the canonical project counter"
+        );
+
+        // Cost and raw activity remain independently observable; their
+        // metadata must not claim a saving that failed its canonical commit.
+        let snapshot = accounting_db.read_snapshot().await.expect("analytics snapshot");
+        let mut rows = snapshot
+            .query(
+                "SELECT metadata_json FROM analytics_events
+                 WHERE event_kind = 'mcp_tool_call' AND tool_name = 'tracedecay_search'
+                 ORDER BY id",
+                (),
+            )
+            .await
+            .expect("registered analytics query");
+        let mut metadata = Vec::new();
+        while let Some(row) = rows.next().await.expect("analytics row") {
+            let encoded = row.get::<String>(0).expect("metadata JSON");
+            metadata.push(
+                serde_json::from_str::<serde_json::Value>(&encoded)
+                    .expect("valid analytics metadata"),
+            );
+        }
+        assert_eq!(metadata.len(), 2, "both tool activities remain visible");
+        assert_eq!(metadata[0]["tokens_saved"], json!(7));
+        assert_eq!(metadata[1]["tokens_saved"], json!(0));
+        assert_eq!(metadata[1]["savings_commit"], json!("refused"));
+        assert_eq!(metadata[1]["before_tokens"], json!(30));
+        assert_eq!(metadata[1]["after_tokens"], json!(29));
+        server.shutdown_background_tasks().await;
+    }
+
     #[tokio::test]
     async fn a_failed_accounting_commit_does_not_advance_the_display_cache() {
         let (cg, _project, authority) =
@@ -722,7 +865,7 @@ mod tests {
             super::super::writer_test_support::registered_context(cg, &authority),
         )
         .await;
-        server.spawn_token_accounting_persist(
+        let _ = server.spawn_token_accounting_persist(
             server.cg_snapshot().await,
             &project_root,
             "tracedecay_search",
diff --git a/crates/tracedecay/src/mcp/server/requests.rs b/crates/tracedecay/src/mcp/server/requests.rs
index 805918b00e0..f3ff2afa296 100644
--- a/crates/tracedecay/src/mcp/server/requests.rs
+++ b/crates/tracedecay/src/mcp/server/requests.rs
@@ -1057,21 +1057,21 @@ impl McpServer {
         cg: &Arc<TraceDecay>,
         tool_name: &str,
         result: &mut ToolResult,
-    ) -> ToolTokenAccounting {
+    ) -> (ToolTokenAccounting, Option<tokio::sync::oneshot::Receiver<bool>>) {
         // A result the shared renderer already accounted carries its figures
         // and footer; only persist them.
         let accounting = match result.token_accounting() {
             Some(accounting) => accounting,
             None => self.account_unrendered_result(cg, result).await,
         };
-        self.spawn_token_accounting_persist(
+        let commitment = self.spawn_token_accounting_persist(
             Arc::clone(cg),
             cg.project_root(),
             tool_name,
             accounting.net_saved_tokens(),
             accounting.raw_file_tokens,
         );
-        accounting
+        (accounting, commitment)
     }
 
     async fn account_unrendered_result(
@@ -1096,7 +1096,7 @@ impl McpServer {
     }
 
     #[allow(clippy::too_many_arguments)]
-    fn spawn_success_analytics(
+    pub(super) fn spawn_success_analytics(
         &self,
         accounting_project_root: &Path,
         tool_name: &str,
@@ -1109,6 +1109,7 @@ impl McpServer {
         result: &ToolResult,
         connection_client_name: Option<&str>,
         connection_instance_id: Option<&str>,
+        commit_confirmation: Option<tokio::sync::oneshot::Receiver<bool>>,
     ) {
         let analytics_outcome = if tool_result_has_semantic_error(result) {
             "error"
@@ -1136,7 +1137,7 @@ impl McpServer {
             let failure_reason = (analytics_outcome == "error")
                 .then(|| semantic_failure_reason(result))
                 .flatten();
-            let analytics_event = mcp_tool_analytics_event(McpToolAnalyticsEvent {
+            let mut analytics_event = mcp_tool_analytics_event(McpToolAnalyticsEvent {
                 project_root: accounting_project_root,
                 session_id: analytics_session_id.clone(),
                 tool_name,
@@ -1155,7 +1156,34 @@ impl McpServer {
                 cost: result.cost().as_ref(),
             });
             self.spawn_observed_ledger_write(async move {
+                // Two independently scheduled writes may run in either order.
+                // Only the receipt for this response (not a project total
+                // sampled after another response) proves its durable credit.
+                let committed = match commit_confirmation {
+                    Some(receiver) => receiver.await.unwrap_or(false),
+                    None => net_saved_tokens == 0,
+                };
+                if !committed {
+                    // Keep the observed before/after activity and actual cost,
+                    // but never claim a positive saving in analytics or history.
+                    let Some(metadata_json) = analytics_event.metadata_json.as_mut() else {
+                        tracing::warn!("MCP analytics credit reconciliation lacked metadata");
+                        return;
+                    };
+                    let Ok(mut metadata) = serde_json::from_str::<Value>(metadata_json) else {
+                        tracing::warn!("MCP analytics credit reconciliation rejected metadata");
+                        return;
+                    };
+                    if !metadata.is_object() {
+                        tracing::warn!("MCP analytics credit reconciliation rejected a nonobject");
+                        return;
+                    }
+                    metadata["tokens_saved"] = json!(0);
+                    metadata["savings_commit"] = json!("refused");
+                    *metadata_json = metadata.to_string();
+                }
                 if let Some(registered) = savings_db
+                    && committed
                     && let Err(e) = registered
                         .try_record_savings(
                             &project_path_str,
@@ -1197,7 +1225,8 @@ impl McpServer {
         connection_client_name: Option<&str>,
         connection_instance_id: Option<&str>,
     ) {
-        let accounting = self.apply_token_accounting(cg, tool_name, result).await;
+        let (accounting, commit_confirmation) =
+            self.apply_token_accounting(cg, tool_name, result).await;
         self.spawn_success_analytics(
             accounting_project_root,
             tool_name,
@@ -1210,6 +1239,7 @@ impl McpServer {
             result,
             connection_client_name,
             connection_instance_id,
+            commit_confirmation,
         );
     }
 

From aa3a706d505b22ce62cf40efe0383c23e523a7f8 Mon Sep 17 00:00:00 2001
From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com>
Date: Mon, 9 Nov 2026 05:27:01 -0800
Subject: [PATCH 13/14] fix(accounting): settle project receipts before profile
 projection

---
 .../src/mcp/server/freshness_tests.rs         |  4 +-
 crates/tracedecay/src/mcp/server/ledger.rs    | 82 ++++++++++++++++++-
 2 files changed, 81 insertions(+), 5 deletions(-)

diff --git a/crates/tracedecay/src/mcp/server/freshness_tests.rs b/crates/tracedecay/src/mcp/server/freshness_tests.rs
index 49aefd70556..97588de6b24 100644
--- a/crates/tracedecay/src/mcp/server/freshness_tests.rs
+++ b/crates/tracedecay/src/mcp/server/freshness_tests.rs
@@ -84,7 +84,8 @@ async fn savings_without_a_display_baseline_still_commit_to_the_project_store()
     let (cg, _dir, _authority) = init_indexed_repo().await;
     assert_eq!(cg.get_tokens_saved().await.unwrap(), 0);
 
-    super::ledger::commit_token_savings(&cg, None, super::ledger::LedgerSink::NotMounted, 13).await;
+    super::ledger::commit_token_savings(&cg, None, super::ledger::LedgerSink::NotMounted, 13, None)
+        .await;
     assert_eq!(cg.get_tokens_saved().await.unwrap(), 13);
     assert_eq!(cg.get_local_counter().await.unwrap(), 13);
 
@@ -96,6 +97,7 @@ async fn savings_without_a_display_baseline_still_commit_to_the_project_store()
         Some(&display),
         super::ledger::LedgerSink::NotMounted,
         7,
+        None,
     )
     .await;
     assert_eq!(display.load(Ordering::Relaxed), 20);
diff --git a/crates/tracedecay/src/mcp/server/ledger.rs b/crates/tracedecay/src/mcp/server/ledger.rs
index 84c109710fd..5dfb763d882 100644
--- a/crates/tracedecay/src/mcp/server/ledger.rs
+++ b/crates/tracedecay/src/mcp/server/ledger.rs
@@ -97,6 +97,7 @@ pub(super) async fn commit_token_savings(
     tokens_saved: Option<&AtomicU64>,
     sink: LedgerSink,
     delta: u64,
+    commit_sender: Option<tokio::sync::oneshot::Sender<bool>>,
 ) -> bool {
     match cg.record_token_savings(delta).await {
         Ok(committed_total) => {
@@ -104,6 +105,12 @@ pub(super) async fn commit_token_savings(
             if let Some(tokens_saved) = tokens_saved {
                 tokens_saved.fetch_max(committed_total, Ordering::Relaxed);
             }
+            // A successful project-store transaction is the response's durable
+            // credit. Release its receipt before the optional profile summary:
+            // that projection has a different writer gate and may stall or fail.
+            if let Some(sender) = commit_sender {
+                let _ = sender.send(true);
+            }
             if let LedgerSink::Mounted(gdb) = sink
                 && let Err(error) = gdb
                     .try_upsert_project_tokens(cg.project_root(), committed_total)
@@ -118,6 +125,9 @@ pub(super) async fn commit_token_savings(
             true
         }
         Err(error) => {
+            if let Some(sender) = commit_sender {
+                let _ = sender.send(false);
+            }
             tracing::warn!(
                 project_root = %cg.project_root().display(),
                 %error,
@@ -201,7 +211,14 @@ impl McpServer {
             // Preserve the activity's raw input size after failure, but never
             // credit a token delta that the project store could not commit.
             let committed_saved_tokens = if let Some((tokens_saved, sink)) = persist {
-                if commit_token_savings(&cg, tokens_saved.as_deref(), sink, net_saved_tokens).await
+                if commit_token_savings(
+                    &cg,
+                    tokens_saved.as_deref(),
+                    sink,
+                    net_saved_tokens,
+                    commit_sender,
+                )
+                .await
                 {
                     net_saved_tokens
                 } else {
@@ -210,9 +227,6 @@ impl McpServer {
             } else {
                 0
             };
-            if let Some(sender) = commit_sender {
-                let _ = sender.send(committed_saved_tokens == net_saved_tokens);
-            }
             // The monitor entry opens, locks, and mmaps a file; keep that
             // off the async workers.
             // A server without an owning profile has no monitor ring.
@@ -648,6 +662,66 @@ mod tests {
         assert_eq!(reopened.get_local_counter().await.unwrap(), 35);
     }
 
+    #[tokio::test]
+    async fn profile_projection_writer_cannot_hold_a_committed_response_receipt() {
+        let (cg, _project, authority) =
+            super::super::writer_test_support::init_indexed_repo().await;
+        let project_root = cg.project_root().to_path_buf();
+        let server = super::super::McpServer::new_with_context(
+            super::super::writer_test_support::registered_context(cg, &authority),
+        )
+        .await;
+        let accounting_db = server
+            .accounting_db
+            .as_ref()
+            .expect("registered profile projection")
+            .clone();
+
+        // Hold only the optional profile store's write gate. The canonical
+        // project store is independent and must still settle this response.
+        let held_projection = accounting_db
+            .begin_write_transaction()
+            .await
+            .expect("hold profile writer while project commit proceeds");
+        let receipt = server
+            .spawn_token_accounting_persist(
+                server.cg_snapshot().await,
+                &project_root,
+                "tracedecay_search",
+                7,
+                20,
+            )
+            .expect("nonzero savings need an explicit commit receipt");
+        let credited = tokio::time::timeout(Duration::from_secs(5), receipt)
+            .await
+            .expect("the project receipt cannot depend on the profile projection")
+            .expect("accepted accounting writer must settle the receipt");
+        assert!(credited, "the canonical project-store credit must commit");
+        assert_eq!(
+            server.cg_snapshot().await.get_tokens_saved().await.unwrap(),
+            7
+        );
+
+        held_projection
+            .rollback()
+            .await
+            .expect("release the held profile writer");
+        assert!(
+            server
+                .ledger_writes_settled_within(Duration::from_secs(10))
+                .await,
+            "the optional projection should finish after its writer is released"
+        );
+        assert_eq!(
+            accounting_db
+                .try_get_project_tokens(&project_root)
+                .await
+                .expect("committed profile projection"),
+            7
+        );
+        server.shutdown_background_tasks().await;
+    }
+
     #[tokio::test]
     async fn live_monitor_reports_only_committed_token_savings() {
         let (cg, _project, authority) =

From d0f314b42a84f68a269b7207d6d8894c9737595e Mon Sep 17 00:00:00 2001
From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com>
Date: Mon, 9 Nov 2026 13:19:35 -0800
Subject: [PATCH 14/14] test(accounting): remove unused ledger regression
 import

---
 crates/tracedecay/src/mcp/server/ledger.rs | 4 +---
 1 file changed, 1 insertion(+), 3 deletions(-)

diff --git a/crates/tracedecay/src/mcp/server/ledger.rs b/crates/tracedecay/src/mcp/server/ledger.rs
index 5dfb763d882..e70dc34f45b 100644
--- a/crates/tracedecay/src/mcp/server/ledger.rs
+++ b/crates/tracedecay/src/mcp/server/ledger.rs
@@ -549,6 +549,7 @@ impl McpServer {
 #[cfg(test)]
 mod tests {
     use serde_json::json;
+    use tracedecay_mcp::tools::response_trailers::ToolTokenAccounting;
     use tracedecay_mcp::transport::JsonRpcRequest;
 
     use super::*;
@@ -801,9 +802,6 @@ mod tests {
 
     #[tokio::test]
     async fn refused_counter_commit_never_credits_independent_savings_history() {
-        use tracedecay_mcp::tools::response_trailers::ToolTokenAccounting;
-        use tracedecay_runtime_core::db::engine::Executor as _;
-
         let (cg, _project, authority) =
             super::super::writer_test_support::init_indexed_repo().await;
         let project_root = cg.project_root().to_path_buf();