From 83b2595d23009dcb462c119775d3a68a8918580c 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] 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(),