fix(sessions): rediscover refreshes superseded by reset

crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs

@@ -1057,7 +1057,15 @@ async fn reject_staged_completion<'a>(
             &recovery,
             durable_projector_failure_code(REFRESH_COMPLETION_REFUSED),
         ) {
-            Some(request) => apply_fail_effect(store, state, &recovery, request, report).await,
+            Some(request) => {
+                let disposition = match error {
+                    SessionStoreError::RefreshSupersededByReset => {
+                        RefreshFailureDisposition::Rediscover
+                    }
+                    _ => RefreshFailureDisposition::SuppressDiscovery,
+                };
+                apply_fail_effect(store, state, &recovery, request, disposition, report).await;
+            }
             None => report.terminal_errors += 1,
         }
         return;
@@ -1431,7 +1439,15 @@ pub async fn apply_refresh_effect(
                         durable_projector_failure_code(REFRESH_PROGRESS_REFUSED),
                     ) {
                         Some(request) => {
-                            apply_fail_effect(store, state, recovery, request, report).await;
+                            apply_fail_effect(
+                                store,
+                                state,
+                                recovery,
+                                request,
+                                RefreshFailureDisposition::SuppressDiscovery,
+                                report,
+                            )
+                            .await;
                         }
                         None => report.terminal_errors += 1,
                     }
@@ -1446,7 +1462,15 @@ pub async fn apply_refresh_effect(
             }
         }
         SessionTemporalRefreshEffect::Fail(request) => {
-            apply_fail_effect(store, state, recovery, request, report).await;
+            apply_fail_effect(
+                store,
+                state,
+                recovery,
+                request,
+                RefreshFailureDisposition::SuppressDiscovery,
+                report,
+            )
+            .await;
         }
         SessionTemporalRefreshEffect::Deferred => report.deferred += 1,
     }
@@ -1530,7 +1554,15 @@ async fn apply_prepared_refresh_effect(
                         SessionRefreshDispositionV1::Joined => report.joined += 1,
                     }
                     if receipt.operation_id() == prepared.recovery.operation_id() {
-                        apply_fail_effect(store, state, &prepared.recovery, request, report).await;
+                        apply_fail_effect(
+                            store,
+                            state,
+                            &prepared.recovery,
+                            request,
+                            RefreshFailureDisposition::SuppressDiscovery,
+                            report,
+                        )
+                        .await;
                     } else {
                         report.terminal_errors += 1;
                     }
@@ -1550,11 +1582,17 @@ async fn apply_prepared_refresh_effect(
     }
 }
 
+enum RefreshFailureDisposition {
+    SuppressDiscovery,
+    Rediscover,
+}
+
 async fn apply_fail_effect(
     store: &SessionTemporalStore<'_, tracedecay_global_db::RegisteredGlobalDb>,
     state: &SessionTemporalRefreshWakeState,
     recovery: &SessionRefreshRecoveryV1,
     request: SessionRefreshFailureRequestV1,
+    disposition: RefreshFailureDisposition,
     report: &mut SessionTemporalRefreshPassReport,
 ) {
     if !state.claim_terminal_attempt(recovery) {
@@ -1571,7 +1609,9 @@ async fn apply_fail_effect(
                 "session temporal refresh failed durably"
             );
             report.failed += 1;
-            state.record_terminal_discovery_failure(recovery);
+            if matches!(disposition, RefreshFailureDisposition::SuppressDiscovery) {
+                state.record_terminal_discovery_failure(recovery);
+            }
         }
         Err(error) if is_retryable_storage(&error) => {
             report.last_error = Some(format!("{error:?}"));

crates/tracedecay-session-temporal-store/src/refresh.rs

@@ -3225,14 +3225,8 @@ async fn complete_session_refresh_in_transaction(
         COMPLETE_REFRESH,
     )
     .await?;
-    // A begin applies any pending reset, so one pending now arrived while
-    // this refresh ran: its candidate covers effects the reset retired and can
-    // never match the live frontier. Refusing it terminally ends the operation
-    // so the next begin can apply the reset.
     if session_reset_is_pending(transaction, request.session_id()).await? {
-        return Err(SessionStoreError::InvalidStateTransition {
-            context: "refresh candidate was superseded by a requested session reset",
-        });
+        return Err(SessionStoreError::RefreshSupersededByReset);
     }
     let progress = require_exact_terminal_progress(
         transaction,

crates/tracedecay-store/src/session/common.rs

@@ -274,6 +274,8 @@ pub enum SessionStoreError {
     },
     #[error("session temporal idempotency conflict in {context}")]
     IdempotencyConflict { context: &'static str },
+    #[error("refresh candidate was superseded by a requested session reset")]
+    RefreshSupersededByReset,
     #[error("invalid session temporal state transition in {context}")]
     InvalidStateTransition { context: &'static str },
     #[error("session temporal capability {capability:?} is unsupported")]

crates/tracedecay/tests/session_suite/temporal_projection/activation.rs

@@ -1,3 +1,5 @@
+use tracedecay_session_runtime::session_temporal_refresh_scheduler::SessionTemporalRefreshSchedulerRegistry;
+
 use super::*;
 
 const OMITTED_LINEAGE_REFUSAL: &str = "session-temporal storage operation activate session temporal generation failed: candidate omits canonical typed assertion lineage through the frozen frontier";
@@ -315,11 +317,6 @@ async fn activation_is_pinned_to_the_generation_the_refresh_extends() {
     );
 }
 
-/// A source rewrite that retires an observation a running refresh already
-/// projected asks for a session reset. That refresh can never prove its
-/// frontier again, so completion refuses it as a terminal state transition
-/// (never a retryable storage fault that would block the reset forever), and
-/// the next refresh applies the reset and activates the rewritten history.
 #[tokio::test]
 async fn reset_requested_during_a_refresh_supersedes_its_candidate() {
     let tmp = TempDir::new().unwrap();
@@ -343,8 +340,6 @@ async fn reset_requested_during_a_refresh_supersedes_its_candidate() {
             ],
         )
         .unwrap();
-    // The reset request `request_session_temporal_reset` records for a
-    // retired observation that produced session output.
     let requested = rusqlite::Connection::open(&path)
         .unwrap()
         .execute(
@@ -359,36 +354,42 @@ async fn reset_requested_during_a_refresh_supersedes_its_candidate() {
         .unwrap();
     assert_eq!(requested, 1, "the retired observation requests one reset");
 
-    let store = runtime
-        .session_temporal_store(HostAdmissionScope::Profile)
-        .unwrap();
-    let refused = complete_candidate(&store, &stale).await;
-    assert!(
-        matches!(
-            refused,
-            Err(SessionStoreError::InvalidStateTransition {
-                context: "refresh candidate was superseded by a requested session reset"
-            })
-        ),
-        "{refused:?}"
-    );
-    assert_eq!(active_generations(&path).await, vec!["1:active"]);
+    let registry = SessionTemporalRefreshSchedulerRegistry::default();
+    registry
+        .ensure_profile(
+            path.clone(),
+            runtime
+                .registered_database_arc(HostAdmissionScope::Profile)
+                .unwrap(),
+        )
+        .await;
+    registry.wait_profile_idle(&path).await;
+    registry.shutdown().await;
 
-    fail_candidate(&store, &stale).await;
-    let rebuilt = begin_candidate(&store, &session_id, 2).await;
-    assert_ne!(rebuilt.operation_id(), stale.operation_id());
-    let first = occurrence(&session_id, &observation(&session_id, 0, "one"));
-    persist_batch(
-        &store,
-        &rebuilt,
-        batch(&rebuilt, vec![first], vec![], vec![]),
+    assert_eq!(
+        rows(
+            &path,
+            "SELECT state FROM session_refresh_operations ORDER BY created_at, operation_id"
+        )
+        .await,
+        vec!["failed", "complete"]
+    );
+    let replacement_generation = rows(
+        &path,
+        "SELECT binding.generation || ':active'
+         FROM session_refresh_bindings AS binding
+         JOIN session_refresh_operations AS operation
+           ON operation.session_id = binding.session_id
+          AND operation.operation_id = binding.operation_id
+         WHERE operation.state = 'complete'",
     )
-    .await
-    .unwrap();
-    complete_candidate(&store, &rebuilt).await.unwrap();
+    .await;
+    assert_eq!(replacement_generation.len(), 1);
+    assert_ne!(replacement_generation, vec!["1:active"]);
+    assert_eq!(active_generations(&path).await, replacement_generation);
     assert_eq!(
-        active_generations(&path).await,
-        vec![format!("{}:active", rebuilt.candidate_generation().value())]
+        rows(&path, "SELECT CAST(MAX(observation_sequence) AS TEXT) FROM session_temporal_observation_effects").await,
+        vec!["2"]
     );
     assert_eq!(
         rows(
@@ -399,6 +400,80 @@ async fn reset_requested_during_a_refresh_supersedes_its_candidate() {
         .await,
         vec!["1"]
     );
+    assert_eq!(
+        rows(
+            &path,
+            "SELECT occurrence.message_id FROM session_occurrences AS occurrence
+             JOIN session_temporal_generations AS generation
+               ON generation.session_id = occurrence.session_id
+              AND generation.generation = occurrence.generation
+             WHERE generation.state = 'active' ORDER BY occurrence.message_id"
+        )
+        .await,
+        vec![format!("message.temporal.{}.0", session_id.as_str())]
+    );
+}
+
+#[tokio::test]
+async fn ambiguous_candidate_failure_does_not_repeat_for_an_unchanged_frontier() {
+    let tmp = TempDir::new().unwrap();
+    let runtime = profile_runtime(&tmp).await;
+    let path = runtime
+        .database_path(HostAdmissionScope::Profile)
+        .unwrap()
+        .to_path_buf();
+    let session_id = session("session.temporal.ambiguous-discovery");
+    let observation_store = runtime
+        .observation_store(HostAdmissionScope::Profile)
+        .unwrap();
+    let store = runtime
+        .session_temporal_store(HostAdmissionScope::Profile)
+        .unwrap();
+    let duplicate = "message.temporal.duplicate";
+    let mut occurrences = Vec::new();
+    for ordinal in 0..2 {
+        let observation = persist_custom_observation(
+            &observation_store,
+            observation_with_message_ids(&session_id, ordinal, "duplicate", duplicate, None),
+        )
+        .await;
+        occurrences.push(occurrence_with_message_id(
+            &session_id,
+            &observation,
+            duplicate,
+        ));
+    }
+    let candidate = begin_candidate(&store, &session_id, 2).await;
+    persist_batch(
+        &store,
+        &candidate,
+        batch(&candidate, occurrences, vec![], vec![]),
+    )
+    .await
+    .unwrap();
+    let registry = SessionTemporalRefreshSchedulerRegistry::default();
+    let wake = registry
+        .ensure_profile(
+            path.clone(),
+            runtime
+                .registered_database_arc(HostAdmissionScope::Profile)
+                .unwrap(),
+        )
+        .await;
+    registry.wait_profile_idle(&path).await;
+    assert_eq!(
+        rows(&path, "SELECT state FROM session_refresh_operations").await,
+        vec!["failed"]
+    );
+
+    wake.wake();
+    registry.wait_profile_idle(&path).await;
+    registry.shutdown().await;
+    assert_eq!(
+        rows(&path, "SELECT state FROM session_refresh_operations").await,
+        vec!["failed"]
+    );
+    assert_eq!(active_generations(&path).await, vec!["1:active"]);
 }
 
 /// Persists observations `one` and `two` in a fresh profile and projects the