refactor(session-refresh): call wake authority directly

crates/tracedecay-session-memory/src/session/mod.rs

@@ -14,8 +14,8 @@ pub use ports::{
 };
 pub use refresh::{
     SessionRefreshConfiguration, SessionRefreshDigest, SessionRefreshHandle, SessionRefreshOutcome,
-    SessionRefreshRequestError, SessionRefreshSchedulerError, SessionRefreshSchedulerPort,
-    SessionRefreshService, SessionRefreshTarget,
+    SessionRefreshRequestError, SessionRefreshSchedulerError, SessionRefreshService,
+    SessionRefreshTarget,
 };
 pub use refresh_service::{
     SessionRefreshAction, SessionRefreshCommand, SessionRefreshCoverageView,

crates/tracedecay-session-memory/src/session/refresh.rs

@@ -199,19 +199,6 @@ impl fmt::Display for SessionRefreshSchedulerError {
 
 impl std::error::Error for SessionRefreshSchedulerError {}
 
-pub trait SessionRefreshSchedulerPort {
-    fn wake(&self) -> Result<(), SessionRefreshSchedulerError>;
-}
-
-impl<T> SessionRefreshSchedulerPort for &T
-where
-    T: SessionRefreshSchedulerPort + ?Sized,
-{
-    fn wake(&self) -> Result<(), SessionRefreshSchedulerError> {
-        (*self).wake()
-    }
-}
-
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub enum SessionRefreshRequestError {
     InvalidProjectorVersion,
@@ -258,7 +245,7 @@ impl<A, S, W> SessionRefreshService<A, S, W>
 where
     A: SessionScopeAuthorizer,
     S: SessionRefreshStore,
-    W: SessionRefreshSchedulerPort,
+    W: Fn() -> Result<(), SessionRefreshSchedulerError>,
 {
     #[hotpath::measure(label = "usecases.session.refresh.begin", future = true)]
     pub async fn begin_or_join(
@@ -336,7 +323,7 @@ where
         // The durable operation is authoritative once the store call returns.
         // Delivery failure must preserve that commit and require reconciliation
         // through the persisted recovery row rather than report plain acceptance.
-        match (receipt.disposition(), self.scheduler.wake()) {
+        match (receipt.disposition(), (self.scheduler)()) {
             (SessionRefreshDispositionV1::Started, Ok(())) => {
                 SessionRefreshOutcome::Started(handle)
             }
@@ -455,7 +442,7 @@ where
         .await
         {
             Ok(Ok(receipt)) => {
-                if self.scheduler.wake().is_err() {
+                if (self.scheduler)().is_err() {
                     SessionRefreshOutcome::CancelledReconciliationRequired(receipt)
                 } else {
                     terminal_outcome(receipt)

crates/tracedecay/src/mcp/server/session_refresh.rs

@@ -6,9 +6,13 @@ use std::collections::HashMap;
 use std::future::Future;
 use std::pin::Pin;
 use std::sync::PoisonError;
+#[cfg(test)]
+use std::time::Duration;
 
 use sha2::{Digest, Sha256};
-use tracedecay_contracts::RequestContext;
+#[cfg(test)]
+use tracedecay_contracts::SessionTemporalRefreshWakeFuture;
+use tracedecay_contracts::{RequestContext, SessionTemporalRefreshWakePort};
 use tracedecay_domain::ProjectId;
 
 use tracedecay_global_db::RegisteredGlobalDbLeaseV1;
@@ -17,9 +21,9 @@ use tracedecay_session_memory::session::{
     SessionRefreshAction, SessionRefreshCommand, SessionRefreshConfiguration,
     SessionRefreshCoverageView, SessionRefreshFrontierView, SessionRefreshHandle,
     SessionRefreshOutcome, SessionRefreshProgressView, SessionRefreshReceiptView,
-    SessionRefreshSchedulerError, SessionRefreshSchedulerPort, SessionRefreshService,
-    SessionRefreshServiceOutcome, SessionRefreshServicePort, SessionRequestBinding,
-    SessionScopeAuthorizationRequest, SessionScopeAuthorizer, utc_micros_value,
+    SessionRefreshSchedulerError, SessionRefreshService, SessionRefreshServiceOutcome,
+    SessionRefreshServicePort, SessionRequestBinding, SessionScopeAuthorizationRequest,
+    SessionScopeAuthorizer, utc_micros_value,
 };
 use tracedecay_session_temporal_store::GlobalDbSessionTemporalStore;
 
@@ -51,23 +55,9 @@ impl SessionScopeAuthorizer for DaemonSessionRefreshAuthorizer<'_> {
     }
 }
 
-#[derive(Clone)]
-struct DaemonSessionRefreshWake(
-    std::sync::Arc<dyn tracedecay_contracts::SessionTemporalRefreshWakePort>,
-);
-
-impl SessionRefreshSchedulerPort for DaemonSessionRefreshWake {
-    fn wake(&self) -> std::result::Result<(), SessionRefreshSchedulerError> {
-        self.0
-            .wake()
-            .then_some(())
-            .ok_or(SessionRefreshSchedulerError)
-    }
-}
-
 pub(crate) struct DaemonSessionRefreshService {
     database: RegisteredGlobalDbLeaseV1,
-    wake: DaemonSessionRefreshWake,
+    wake: std::sync::Arc<dyn SessionTemporalRefreshWakePort>,
     expected_project_id: Option<String>,
     handles: std::sync::Mutex<HashMap<String, SessionRefreshHandle>>,
 }
@@ -81,40 +71,17 @@ enum SessionRefreshHandleLookup {
 impl DaemonSessionRefreshService {
     pub(crate) fn new(
         database: RegisteredGlobalDbLeaseV1,
-        wake: std::sync::Arc<dyn tracedecay_contracts::SessionTemporalRefreshWakePort>,
+        wake: std::sync::Arc<dyn SessionTemporalRefreshWakePort>,
         expected_project_id: Option<String>,
     ) -> Self {
         Self {
             database,
-            wake: DaemonSessionRefreshWake(wake),
+            wake,
             expected_project_id,
             handles: std::sync::Mutex::new(HashMap::new()),
         }
     }
 
-    fn service(
-        &self,
-    ) -> Option<
-        SessionRefreshService<
-            DaemonSessionRefreshAuthorizer<'_>,
-            GlobalDbSessionTemporalStore<'_, tracedecay_global_db::RegisteredGlobalDb>,
-            &DaemonSessionRefreshWake,
-        >,
-    > {
-        Some(SessionRefreshService::new(
-            DaemonSessionRefreshAuthorizer {
-                expected_project_id: self.expected_project_id.as_deref(),
-            },
-            GlobalDbSessionTemporalStore::new(self.database.as_ref()),
-            &self.wake,
-            SessionRefreshConfiguration::new(
-                SESSION_REFRESH_PROJECTOR_VERSION,
-                SESSION_REFRESH_CONFIG_VERSION,
-            )
-            .ok()?,
-        ))
-    }
-
     fn handle(&self, token: &str) -> SessionRefreshHandleLookup {
         if !is_session_refresh_handle_token(token) {
             return missing_session_refresh_handle_lookup(token);
@@ -154,9 +121,20 @@ impl DaemonSessionRefreshService {
         &self,
         command: SessionRefreshCommand,
     ) -> SessionRefreshServiceOutcome {
-        let Some(service) = self.service() else {
+        let Ok(configuration) = SessionRefreshConfiguration::new(
+            SESSION_REFRESH_PROJECTOR_VERSION,
+            SESSION_REFRESH_CONFIG_VERSION,
+        ) else {
             return SessionRefreshServiceOutcome::Unavailable;
         };
+        let service = SessionRefreshService::new(
+            DaemonSessionRefreshAuthorizer {
+                expected_project_id: self.expected_project_id.as_deref(),
+            },
+            GlobalDbSessionTemporalStore::new(self.database.as_ref()),
+            || wake_session_refresh_scheduler(self.wake.as_ref()),
+            configuration,
+        );
         let outcome = match command.action {
             SessionRefreshAction::Begin => {
                 service
@@ -260,6 +238,14 @@ impl DaemonSessionRefreshService {
     }
 }
 
+fn wake_session_refresh_scheduler(
+    wake: &dyn SessionTemporalRefreshWakePort,
+) -> std::result::Result<(), SessionRefreshSchedulerError> {
+    wake.wake()
+        .then_some(())
+        .ok_or(SessionRefreshSchedulerError)
+}
+
 fn is_session_refresh_handle_token(token: &str) -> bool {
     token.strip_prefix("srh_").is_some_and(|digest| {
         digest.len() == 64 && digest.bytes().all(|byte| byte.is_ascii_hexdigit())
@@ -274,6 +260,44 @@ fn missing_session_refresh_handle_lookup(token: &str) -> SessionRefreshHandleLoo
     }
 }
 
+#[cfg(test)]
+#[derive(Clone, Copy)]
+struct FixedSessionTemporalRefreshWake(bool);
+
+#[cfg(test)]
+impl SessionTemporalRefreshWakePort for FixedSessionTemporalRefreshWake {
+    fn wake(&self) -> bool {
+        self.0
+    }
+
+    fn is_unavailable(&self) -> bool {
+        !self.0
+    }
+
+    fn wake_and_wait_until_idle(&self, _timeout: Duration) -> SessionTemporalRefreshWakeFuture<'_> {
+        let accepted = self.0;
+        Box::pin(async move { accepted })
+    }
+}
+
+#[cfg(test)]
+#[test]
+fn accepted_session_refresh_wake_maps_to_typed_success() {
+    assert_eq!(
+        wake_session_refresh_scheduler(&FixedSessionTemporalRefreshWake(true)),
+        Ok(())
+    );
+}
+
+#[cfg(test)]
+#[test]
+fn refused_session_refresh_wake_maps_to_scheduler_error() {
+    assert_eq!(
+        wake_session_refresh_scheduler(&FixedSessionTemporalRefreshWake(false)),
+        Err(SessionRefreshSchedulerError)
+    );
+}
+
 #[cfg(test)]
 #[test]
 fn session_refresh_handle_tokens_are_closed_and_non_leaking() {

crates/tracedecay/src/session_temporal_benchmark.rs

@@ -42,9 +42,9 @@ use tracedecay_session_memory::context::{
 };
 use tracedecay_session_memory::session::{
     AuthorizationGrantId, SessionAuthorizationError, SessionAuthorizationGrant,
-    SessionRefreshSchedulerError, SessionRefreshSchedulerPort, SessionRequestBinding,
-    SessionRetrievalConfiguration, SessionRetrievalOutcome, SessionRetrievalService,
-    SessionScopeAuthorizationRequest, SessionScopeAuthorizer, SessionTemporalQuery,
+    SessionRequestBinding, SessionRetrievalConfiguration, SessionRetrievalOutcome,
+    SessionRetrievalService, SessionScopeAuthorizationRequest, SessionScopeAuthorizer,
+    SessionTemporalQuery,
 };
 use tracedecay_session_temporal_store::RegisteredGlobalDbSessionTemporalExecution;
 use tracedecay_sessions::observation::ObservationCancellation;
@@ -257,15 +257,6 @@ impl SessionScopeAuthorizer for AllowAuthorizer {
     }
 }
 
-#[derive(Clone, Copy, Default)]
-struct NoopWake;
-
-impl SessionRefreshSchedulerPort for NoopWake {
-    fn wake(&self) -> Result<(), SessionRefreshSchedulerError> {
-        Ok(())
-    }
-}
-
 struct Words(&'static str);
 
 impl VersionedTokenEstimator for Words {

crates/tracedecay/src/session_temporal_benchmark/root_relation_fixture.rs

@@ -17,7 +17,7 @@ use tracedecay_temporal_query::context::ContextBudget;
 use tracedecay_temporal_query::ports::ExecutionControl;
 use tracedecay_temporal_query::ranking::DiversityLimits;
 
-use super::{AllowAuthorizer, BenchResult, CONFIG_VERSION, NoopWake, PROJECTOR_VERSION};
+use super::{AllowAuthorizer, BenchResult, CONFIG_VERSION, PROJECTOR_VERSION};
 use tracedecay_session_temporal_store::GlobalDbSessionTemporalStore;
 
 pub(super) const ROOT_RELATION_PARTICIPANT_COUNT: usize = 64;
@@ -57,7 +57,7 @@ pub(super) async fn refresh_sessions(
     let refresh = SessionRefreshService::new(
         AllowAuthorizer,
         GlobalDbSessionTemporalStore::new(db),
-        NoopWake,
+        || Ok(()),
         SessionRefreshConfiguration::new(PROJECTOR_VERSION, CONFIG_VERSION)
             .map_err(|error| format!("root refresh configuration: {error}"))?,
     );