Fix automatic compaction closure races

apps/signalboxd/src/context_guard.rs

@@ -828,17 +828,19 @@ async fn close_failed_compaction_turn(
             .commit_compaction_failure_preview(preview.clone(), model_calls, identities)
             .await
         {
-            Err(CommitActivationPreviewError::Activation(
-                StartEligibleTurnRepositoryError::IdentityCollision(_),
-            ))
-            | Err(CommitActivationPreviewError::ModelCall(
-                ModelCallRepositoryError::IdentityCollision(_),
-            )) => {}
+            Err(error) if compaction_failure_closure_collision_is_retryable(&error) => {}
             outcome => return outcome,
         }
     }
 }
 
+fn compaction_failure_closure_collision_is_retryable(error: &CommitActivationPreviewError) -> bool {
+    matches!(
+        error,
+        CommitActivationPreviewError::ModelCall(ModelCallRepositoryError::IdentityCollision(_))
+    )
+}
+
 #[cfg(test)]
 mod tests {
     use std::{
@@ -850,10 +852,17 @@ mod tests {
     use signalbox_domain::{ActivatedTurn, TurnId};
     use signalbox_persistence::{
         context_compaction::ContextCompactionRepositoryError,
-        start_eligible_turn::StartEligibleTurnRepositoryError,
+        model_execution::{ModelCallIdentityCollision, ModelCallRepositoryError},
+        start_eligible_turn::{
+            CommitActivationPreviewError, StartEligibleTurnIdentityCollision,
+            StartEligibleTurnRepositoryError,
+        },
     };
 
-    use super::{ContextGuardedTurnPassError, guarded_failure_stage, report_guarded_ambiguity};
+    use super::{
+        ContextGuardedTurnPassError, compaction_failure_closure_collision_is_retryable,
+        guarded_failure_stage, report_guarded_ambiguity,
+    };
     use crate::{
         ActivatedTurnExecution, FatalExecutionSignal, FatalExecutionSupervisor,
         TurnPassExecutionStage, process_runtime::AutomaticContextCompactionError,
@@ -863,6 +872,26 @@ mod tests {
     #[derive(Clone, Copy, Debug, Eq, PartialEq)]
     struct CommitAmbiguousFailure;
 
+    #[test]
+    fn reminted_compaction_failure_identity_collision_is_retryable() {
+        let error = CommitActivationPreviewError::ModelCall(
+            ModelCallRepositoryError::IdentityCollision(ModelCallIdentityCollision::SemanticEntry),
+        );
+
+        assert!(compaction_failure_closure_collision_is_retryable(&error));
+    }
+
+    #[test]
+    fn immutable_activation_identity_collision_is_not_retryable() {
+        let error = CommitActivationPreviewError::Activation(
+            StartEligibleTurnRepositoryError::IdentityCollision(
+                StartEligibleTurnIdentityCollision::StartingFrontier,
+            ),
+        );
+
+        assert!(!compaction_failure_closure_collision_is_retryable(&error));
+    }
+
     impl fmt::Display for CommitAmbiguousFailure {
         fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
             formatter.write_str("commit acknowledgement was lost")

crates/domain/src/model_execution.rs

@@ -8132,7 +8132,14 @@ mod tests {
     #[test]
     fn s03_inv015_automatic_compaction_failure_closes_call_free_turn() {
         let failure_entry = semantic_transcript_entry_id(10);
-        let failed = active_execution()
+        let execution = active_execution();
+        let session = execution.session();
+        let starting_entry = execution
+            .current_snapshot
+            .ordered_entries()
+            .next()
+            .expect("fixture activation carries its origin");
+        let failed = execution
             .fail_automatic_context_compaction(FailedModelCallTurnIdentities::new(
                 failure_entry,
                 context_frontier_id(11),
@@ -8152,11 +8159,8 @@ mod tests {
                 .ordered_entries()
                 .collect::<Vec<_>>(),
             vec![
-                SemanticTranscriptEntryRef::from_source(
-                    session_id(1),
-                    semantic_transcript_entry_id(5),
-                ),
-                SemanticTranscriptEntryRef::from_source(session_id(1), failure_entry),
+                starting_entry,
+                SemanticTranscriptEntryRef::from_source(session, failure_entry),
             ]
         );
     }

crates/persistence/src/start_eligible_turn.rs

@@ -27,7 +27,10 @@ use crate::{
         defaults_version_to_numeric, input_position_to_numeric, positive_u64_from_numeric,
         session_id_to_uuid, turn_id_to_uuid,
     },
-    model_execution::{SnapshotAppend, SnapshotAppendError, insert_snapshot_append},
+    model_execution::{
+        SnapshotAppend, SnapshotAppendError, insert_snapshot_append,
+        lock_delegated_child_endpoint_sessions,
+    },
     outbox::{self, OutboxEvent},
     session::{SessionCorruption, SessionRepositoryError, load_session_from_connection},
     submit_input::{
@@ -417,6 +420,12 @@ impl StartEligibleTurnRepository {
             .await
             .map_err(StartEligibleTurnRepositoryError::from)
             .map_err(CommitActivationPreviewError::Activation)?;
+        // A delegated-child failure publishes into its parent session. Keep
+        // that endpoint pair ahead of the child scheduler in the global lock
+        // order, matching every other delegated terminalization path.
+        lock_delegated_child_endpoint_sessions(&mut transaction, session)
+            .await
+            .map_err(CommitActivationPreviewError::ModelCall)?;
         let session_uuid = session_id_to_uuid(session);
         let (session_exists, scheduler_session) =
             sqlx::query_as::<_, (bool, Option<Uuid>)>(crate::lock_inventory::START_ELIGIBLE_TURN)