refactor(twap-monitor): port poll loop onto keeper

crates/nexum-sdk/src/keeper.rs

@@ -331,6 +331,13 @@ pub trait ConditionalSource<H> {
     /// watch value (the encoded commitment parameters), passed
     /// verbatim so the source owns the decode.
     fn poll(&self, host: &H, watch: WatchRef<'_>, params: &[u8], tick: &Tick) -> Self::Outcome;
+
+    /// Short strategy name compositions prefix shared log lines with
+    /// (for example `"twap"`). Diagnostic only - no behaviour keys
+    /// off it.
+    fn label(&self) -> &'static str {
+        "conditional"
+    }
 }
 
 /// What the retry ledger should do to a watch after a failed

crates/shepherd-sdk/Cargo.toml

@@ -19,8 +19,12 @@ alloy-sol-types.workspace = true
 serde_json.workspace = true
 strum.workspace = true
 thiserror.workspace = true
+tracing.workspace = true
 
 [dev-dependencies]
+# `capture_tracing` observes the run's diagnostics in the
+# acceptance tests.
+nexum-sdk-test = { path = "../nexum-sdk-test" }
 proptest.workspace = true
 # Dev-only cycle (this crate <- shepherd-sdk-test): cargo permits it,
 # and the run acceptance-tests against the composed MockHost

crates/shepherd-sdk/src/cow/run.rs

@@ -11,14 +11,17 @@
 //!
 //! Store faults abort the sweep (the next tick replays it);
 //! submission failures never do - they classify into a
-//! [`RetryAction`](super::RetryAction), the ledger applies the
-//! effect, and the sweep moves on.
+//! [`RetryAction`], the ledger applies the effect, and the sweep
+//! moves on. Diagnostics go through the guest `tracing` facade -
+//! the same channel strategy code logs on - so module tests observe
+//! the composed behaviour with one capture.
 
 use alloy_primitives::{Address, Bytes};
 use cowprotocol::{GPv2OrderData, OrderCreation, Signature};
-use nexum_sdk::Level;
 use nexum_sdk::host::Fault;
-use nexum_sdk::keeper::{ConditionalSource, Gates, Journal, Retrier, Tick, WatchRef, WatchSet};
+use nexum_sdk::keeper::{
+    ConditionalSource, Gates, Journal, Retrier, RetryAction, Tick, WatchRef, WatchSet,
+};
 
 use super::{
     CowApiError, CowHost, PollOutcome, classify_submit_error, gpv2_to_order_data,
@@ -47,7 +50,7 @@ where
         };
         match source.poll(host, watch, &params, tick) {
             PollOutcome::Ready { order, signature } => {
-                submit_ready(host, watch, &order, signature, tick)?;
+                submit_ready(host, watch, &order, signature, tick, source.label())?;
             }
             PollOutcome::TryNextBlock => {}
             PollOutcome::TryOnBlock(block) => gates.set_next_block(watch, block)?,
@@ -71,14 +74,12 @@ fn submit_ready<H: CowHost>(
     order: &GPv2OrderData,
     signature: Bytes,
     tick: &Tick,
+    label: &str,
 ) -> Result<(), Fault> {
     let Ok(owner) = watch.owner_hex().parse::<Address>() else {
-        host.log(
-            Level::WARN,
-            &format!(
-                "watch {} carries an unparseable owner; skipping submit",
-                watch.key(),
-            ),
+        tracing::warn!(
+            "watch {} carries an unparseable owner; skipping submit",
+            watch.key(),
         );
         return Ok(());
     };
@@ -88,30 +89,21 @@ fn submit_ready<H: CowHost>(
     if let Some(uid) = client_uid.as_deref()
         && journal.contains(uid)?
     {
-        host.log(
-            Level::INFO,
-            &format!("{uid} already submitted; skipping re-submit"),
-        );
+        tracing::info!("{label} {uid} already submitted; skipping re-submit");
         return Ok(());
     }
 
     let creation = match build_order_creation(order, signature, owner) {
         Ok(creation) => creation,
         Err(message) => {
-            host.log(
-                Level::WARN,
-                &format!("submit skipped for {owner:#x}: {message}"),
-            );
+            tracing::warn!("{label} submit skipped for {owner:#x}: {message}");
             return Ok(());
         }
     };
     let body = match serde_json::to_vec(&creation) {
         Ok(body) => body,
         Err(e) => {
-            host.log(
-                Level::ERROR,
-                &format!("OrderCreation JSON encode failed: {e}"),
-            );
+            tracing::error!("OrderCreation JSON encode failed: {e}");
             return Ok(());
         }
     };
@@ -127,23 +119,17 @@ fn submit_ready<H: CowHost>(
             // Log and carry on - the already-submitted arm keeps the
             // next tick's re-post idempotent.
             if let Err(fault) = journal.record(marker) {
-                host.log(
-                    Level::ERROR,
-                    &format!("submitted {marker} but journal write failed: {fault}"),
-                );
+                tracing::error!("submitted {marker} but journal write failed: {fault}");
             }
             if let Some(client) = client_uid.as_deref()
                 && client != server_uid
             {
-                host.log(
-                    Level::WARN,
-                    &format!(
-                        "UID divergence: client={client} server={server_uid} \
-                         (marker keyed on the client UID)"
-                    ),
+                tracing::warn!(
+                    "{label} UID divergence: client={client} server={server_uid} \
+                     (marker keyed on the client UID)"
                 );
             }
-            host.log(Level::INFO, &format!("submitted {marker}"));
+            tracing::info!("submitted {marker}");
         }
         Err(CowApiError::Rejected(rejection)) if is_already_submitted(&rejection) => {
             // Success wearing an error status: the orderbook already
@@ -154,27 +140,29 @@ fn submit_ready<H: CowHost>(
             if let Some(uid) = client_uid.as_deref()
                 && let Err(fault) = journal.record(uid)
             {
-                host.log(
-                    Level::ERROR,
-                    &format!("orderbook already holds {uid} but journal write failed: {fault}"),
-                );
+                tracing::error!("orderbook already holds {uid} but journal write failed: {fault}");
             }
-            host.log(
-                Level::INFO,
-                &format!(
-                    "orderbook already holds this order ({}); receipt recorded",
-                    rejection.error_type,
-                ),
+            tracing::info!(
+                "orderbook already holds this order ({}); receipt recorded",
+                rejection.error_type,
             );
         }
         Err(err) => {
             let action = classify_submit_error(&err);
             Retrier::new(host).apply(watch, action, tick.epoch_s)?;
-            let label: &'static str = action.into();
-            host.log(
-                Level::WARN,
-                &format!("submit failed ({err}); retry action {label}"),
-            );
+            match action {
+                RetryAction::TryNextBlock => tracing::warn!("submit retry-next-block: {err}"),
+                RetryAction::Backoff { seconds } => {
+                    tracing::warn!("submit backoff {seconds}s: {err}");
+                }
+                RetryAction::Drop => tracing::warn!("submit dropped watch: {err}"),
+                // `RetryAction` is non-exhaustive; the ledger already
+                // ran the effect, so the log needs only the name.
+                other => {
+                    let action_label: &'static str = other.into();
+                    tracing::warn!("submit retry action {action_label}: {err}");
+                }
+            }
         }
     }
     Ok(())

crates/shepherd-sdk/tests/run.rs

@@ -10,6 +10,7 @@ use alloy_primitives::{Address, B256, U256, address, hex, keccak256};
 use cowprotocol::{BuyTokenDestination, GPv2OrderData, OrderKind, SellTokenSource};
 use nexum_sdk::host::{Fault, LocalStoreHost as _, RateLimit};
 use nexum_sdk::keeper::{ConditionalSource, Gates, Journal, Tick, WatchRef, WatchSet};
+use nexum_sdk_test::capture_tracing;
 use shepherd_sdk::cow::{CowApiError, OrderRejection, PollOutcome, order_uid_hex, run};
 use shepherd_sdk_test::MockHost;
 
@@ -253,15 +254,16 @@ fn ready_marker_keys_on_the_client_uid_when_the_server_diverges() {
         let order = order.clone();
         src(move |_, _, _, _| ready_outcome(&order))
     };
-    run(&host, &source, &sample_tick()).unwrap();
+    let (result, logs) = capture_tracing(|| run(&host, &source, &sample_tick()));
+    result.unwrap();
 
     let snapshot = host.store.snapshot();
     assert!(snapshot.contains_key(&format!("submitted:{}", client_uid(&order))));
     assert!(
         !snapshot.contains_key("submitted:0xfeedface"),
         "marker must key on the client UID, not the divergent server UID",
     );
-    assert!(host.logging.contains("UID divergence"));
+    assert!(logs.any(|e| e.message.contains("UID divergence")));
 }
 
 #[test]

modules/twap-monitor/Cargo.toml

@@ -14,12 +14,10 @@ shepherd-sdk = { path = "../../crates/shepherd-sdk" }
 cowprotocol = { version = "0.2.0", default-features = false }
 alloy-primitives = { version = "1.6", default-features = false, features = ["std"] }
 alloy-sol-types = { version = "1.6", default-features = false, features = ["std"] }
-serde_json = { version = "1", default-features = false, features = ["alloc"] }
-strum = { version = "0.28", default-features = false, features = ["derive"] }
-thiserror = "2"
 tracing = { version = "0.1", default-features = false }
 wit-bindgen = { version = "0.59", default-features = false, features = ["macros", "realloc"] }
 
 [dev-dependencies]
+serde_json = { version = "1", default-features = false, features = ["alloc"] }
 shepherd-sdk-test = { path = "../../crates/shepherd-sdk-test" }
 nexum-sdk-test = { path = "../../crates/nexum-sdk-test" }