modules: re-point ethflow-watcher onto the pool observe path

crates/shepherd-backtest/Cargo.toml

@@ -15,8 +15,10 @@ path = "src/main.rs"
 # (alongside its wasm cdylib) specifically so this crate can drive
 # `strategy::on_chain_logs` directly without an embedded runtime.
 ethflow-watcher = { path = "../../modules/ethflow-watcher" }
+cow-venue = { path = "../cow-venue", features = ["client"] }
 nexum-sdk = { path = "../nexum-sdk" }
-shepherd-sdk-test = { path = "../shepherd-sdk-test" }
+nexum-sdk-test = { path = "../nexum-sdk-test" }
+videre-sdk = { path = "../videre-sdk" }
 
 anyhow.workspace = true
 clap.workspace = true

crates/shepherd-backtest/src/main.rs

@@ -3,8 +3,8 @@
 //! Offline replay harness for Shepherd modules. Loads a fixtures
 //! JSON produced by `tools/backtest-collect/backtest_collect.py`,
 //! drives each on-chain event through the production strategy code
-//! via `shepherd_sdk_test::MockHost`, classifies the result, and
-//! emits a Markdown report at
+//! over `nexum_sdk_test::MockHost` and a recording pool transport,
+//! classifies the result, and emits a Markdown report at
 //! `docs/operations/backtest-reports/backtest-7d-YYYY-MM-DD.md`.
 //!
 //! ## Scope

crates/shepherd-backtest/src/replay.rs

@@ -1,32 +1,37 @@
-//! Per-event replay against `ethflow_watcher::strategy::on_chain_logs`.
+//! Per-event replay against the ethflow-watcher strategy pair.
 //!
-//! Each [`EthFlowFixture`] is driven through the production strategy
-//! exactly the way the live engine does it: a fresh [`MockHost`] is
-//! constructed, a catch-all 200 response is programmed for any
-//! `cow_api_request` call (the observe+verify strategy GETs
-//! `/api/v1/orders/{uid}` to confirm the orderbook has indexed the
-//! order), and `strategy::on_chain_logs` is invoked with an alloy
-//! `Log` reconstructed from the raw `eth_getLogs` payload.
+//! Each [`EthFlowFixture`] is driven the way the live engine does it,
+//! minus the wasm boundary: `strategy::on_chain_logs` runs over a
+//! recording venue transport (the pool seam), then the status
+//! transition the registry would poll for the watched receipt is
+//! delivered through `strategy::on_intent_status`. In backtest context
+//! all fixtures are confirmed real orders, so the simulated transition
+//! is `open`.
 //!
-//! The classification falls into one of the four buckets defined in
-//! the issue:
+//! The classification falls into one of four buckets:
 //!
-//! - `Observed`: the strategy verified the order with exactly one
-//!   `GET /api/v1/orders/{uid}` and wrote `observed:{uid}` to the
-//!   local store. This is the success case under the observe+verify
-//!   strategy.
-//! - `RejectedExpected`: the strategy returned without observing in a
-//!   documented case (reserved for fixtures where the mock returns 404
-//!   — not applicable when all fixtures program 200).
+//! - `Observed`: the strategy registered exactly one `cow` watch whose
+//!   receipt is the fixture's UID and wrote `observed:{uid}` on the
+//!   delivered transition. The success case.
+//! - `RejectedExpected`: reserved for a documented skip path; not
+//!   emitted in the current batch.
 //! - `RejectedUnexpected`: the strategy returned Ok but the observe
-//!   contract was violated (no `observed:{uid}` marker, an unexpected
-//!   orderbook call shape, or a `submit_order` attempt); a follow-up
-//!   should be filed before the report closes.
-//! - `StrategyError`: `on_chain_logs` returned `Err(fault)`. A test
+//!   contract was violated (a wrong watch shape, a non-observe venue
+//!   call, or a missing `observed:{uid}` marker); a follow-up should
+//!   be filed before the report closes.
+//! - `StrategyError`: a strategy call returned `Err(fault)`. A test
 //!   bug or an `unreachable!` we want to investigate.
 
+use std::cell::RefCell;
+use std::rc::Rc;
+
+use cow_venue::client::CowClient;
 use ethflow_watcher::strategy;
-use shepherd_sdk_test::MockHost;
+use nexum_sdk::status_body::{IntentStatus, StatusBody};
+use nexum_sdk_test::{CapturedEvents, MockHost, capture_tracing};
+use videre_sdk::client::{VenueId, VenueTransport};
+use videre_sdk::rt::complete;
+use videre_sdk::{Quotation, SubmitOutcome, VenueFault};
 
 use crate::fixtures::{EthFlowFixture, parse_address};
 
@@ -45,9 +50,8 @@ pub struct ReplayOutcome {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum Classification {
     Observed,
-    /// Reserved for any documented skip path (e.g. a fixture where the mock
-    /// returns 404 for an un-indexed order). Not emitted in the current batch;
-    /// retained so the acceptance-ratio formula is complete.
+    /// Reserved for any documented skip path. Not emitted in the current
+    /// batch; retained so the acceptance-ratio formula is complete.
     #[allow(dead_code)]
     RejectedExpected(String),
     RejectedUnexpected(String),
@@ -74,16 +78,94 @@ impl Classification {
     }
 }
 
-/// Replay one EthFlow fixture through the production strategy.
+/// One recorded transport call: the verb, and for `observe` the routed
+/// venue and receipt.
+#[derive(Clone, Debug, PartialEq, Eq)]
+enum Call {
+    Quote,
+    Submit,
+    Observe(String, Vec<u8>),
+    Status,
+    Cancel,
+}
+
+impl Call {
+    fn label(&self) -> &'static str {
+        match self {
+            Call::Quote => "quote",
+            Call::Submit => "submit",
+            Call::Observe(..) => "observe",
+            Call::Status => "status",
+            Call::Cancel => "cancel",
+        }
+    }
+}
+
+/// Records every pool call; `observe` accepts, the other verbs refuse.
+/// Cloneable over shared state so the replay keeps a handle after one
+/// moves into the client.
+#[derive(Clone, Default)]
+struct RecordingVenues {
+    calls: Rc<RefCell<Vec<Call>>>,
+}
+
+impl RecordingVenues {
+    fn calls(&self) -> Vec<Call> {
+        self.calls.borrow().clone()
+    }
+}
+
+impl videre_sdk::client::sealed::SealedTransport for RecordingVenues {}
+
+impl VenueTransport for RecordingVenues {
+    async fn quote(&self, _venue: &VenueId, _body: Vec<u8>) -> Result<Quotation, VenueFault> {
+        self.calls.borrow_mut().push(Call::Quote);
+        Err(VenueFault::Unsupported)
+    }
+
+    async fn submit(&self, _venue: &VenueId, _body: Vec<u8>) -> Result<SubmitOutcome, VenueFault> {
+        self.calls.borrow_mut().push(Call::Submit);
+        Err(VenueFault::Unsupported)
+    }
+
+    async fn observe(&self, venue: &VenueId, receipt: &[u8]) -> Result<(), VenueFault> {
+        self.calls
+            .borrow_mut()
+            .push(Call::Observe(venue.to_string(), receipt.to_vec()));
+        Ok(())
+    }
+
+    async fn status(
+        &self,
+        _venue: &VenueId,
+        _receipt: &[u8],
+    ) -> Result<videre_sdk::IntentStatus, VenueFault> {
+        self.calls.borrow_mut().push(Call::Status);
+        Err(VenueFault::Unsupported)
+    }
+
+    async fn cancel(&self, _venue: &VenueId, _receipt: &[u8]) -> Result<(), VenueFault> {
+        self.calls.borrow_mut().push(Call::Cancel);
+        Err(VenueFault::Unsupported)
+    }
+}
+
+/// The `open` transition the registry would report for an indexed order.
+fn open_status() -> Result<Vec<u8>, String> {
+    StatusBody {
+        status: IntentStatus::Open,
+        proof: None,
+        reason: None,
+    }
+    .encode()
+    .map_err(|e| format!("status body encode: {e}"))
+}
+
+/// Replay one EthFlow fixture through the production strategy pair.
 pub fn replay_ethflow(fx: &EthFlowFixture, chain_id: u64) -> ReplayOutcome {
     let host = MockHost::new();
-
-    // Program a catch-all 200 response for any cow_api_request. In
-    // the observe+verify strategy the module GETs
-    // `/api/v1/orders/{uid}` to confirm the orderbook has indexed the
-    // order. In backtest context all fixtures are confirmed real orders,
-    // so the mock orderbook always returns 200 (indexed).
-    host.cow_api.respond_to_request(Ok("{}".to_string()));
+    let venues = RecordingVenues::default();
+    let client = CowClient::with_transport(venues.clone());
 
     // Reconstruct the log fields. Topics + data come straight from the
     // collector's `raw_log`; the contract address is the EthFlow
@@ -120,18 +202,32 @@ pub fn replay_ethflow(fx: &EthFlowFixture, chain_id: u64) -> ReplayOutcome {
     }
     .into();
 
-    // Drive the strategy.
-    let result = strategy::on_chain_logs(&host, chain_id, &[log]);
-    let log_lines: Vec<String> = host
-        .logging
-        .lines()
-        .into_iter()
-        .map(|l| format!("[{:?}] {}", l.level, l.message))
-        .collect();
+    // Drive the strategy: register the watch, then deliver the status
+    // transition the registry would poll for the watched receipt.
+    let (result, logs) = capture_tracing(|| {
+        complete(strategy::on_chain_logs(&host, &client, chain_id, &[log]))
+            .ok_or_else(|| "strategy future suspended".to_owned())
+            .and_then(|r| r.map_err(|e| e.to_string()))?;
+        let calls = venues.calls();
+        let [Call::Observe(venue, receipt)] = calls.as_slice() else {
+            let shapes: Vec<&str> = calls.iter().map(Call::label).collect();
+            return Err(format!(
+                "expected exactly one observe; saw [{}]",
+                shapes.join(", ")
+            ));
+        };
+        if venue != "cow" {
+            return Err(format!("watch routed to venue `{venue}`, not `cow`"));
+        }
+        strategy::on_intent_status(&host, venue, receipt, &open_status()?)
+            .map_err(|e| e.to_string())?;
+        Ok(receipt.clone())
+    });
+    let log_lines = log_lines(&logs);
 
     let class = match result {
-        Err(e) => Classification::StrategyError(e.to_string()),
-        Ok(()) => classify_ok(&host, &fx.uid, &log_lines),
+        Err(detail) => classify_err(&venues, detail),
+        Ok(receipt) => classify_ok(&host, &receipt, &fx.uid, &log_lines),
     };
 
     ReplayOutcome {
@@ -143,6 +239,13 @@ pub fn replay_ethflow(fx: &EthFlowFixture, chain_id: u64) -> ReplayOutcome {
     }
 }
 
+fn log_lines(logs: &CapturedEvents) -> Vec<String> {
+    logs.events()
+        .into_iter()
+        .map(|e| format!("[{:?}] {}", e.level, e.message))
+        .collect()
+}
+
 fn error_outcome(fx: &EthFlowFixture, reason: String) -> ReplayOutcome {
     ReplayOutcome {
         uid: fx.uid.clone(),
@@ -153,31 +256,40 @@ fn error_outcome(fx: &EthFlowFixture, reason: String) -> ReplayOutcome {
     }
 }
 
-/// Classify an `Ok(())` replay by inspecting the mock's recorded
-/// side effects, independent of the strategy's own logging.
+/// A failed replay is an observe-contract violation when the strategy
+/// itself returned Ok but touched the pool wrongly; otherwise a
+/// strategy error.
+fn classify_err(venues: &RecordingVenues, detail: String) -> Classification {
+    if venues
+        .calls()
+        .iter()
+        .any(|c| !matches!(c, Call::Observe(..)))
+    {
+        return Classification::RejectedUnexpected(format!(
+            "strategy called a non-observe venue verb; observe-only must never submit ({detail})"
+        ));
+    }
+    if detail.starts_with("expected exactly one observe")
+        || detail.starts_with("watch routed to venue")
+    {
+        return Classification::RejectedUnexpected(detail);
+    }
+    Classification::StrategyError(detail)
+}
+
+/// Classify an `Ok` replay by inspecting the recorded side effects,
+/// independent of the strategy's own logging.
 ///
 /// `Observed` demands the full observe contract, not just the marker:
-/// exactly one orderbook call, shaped `GET /api/v1/orders/{uid}`,
-/// zero `submit_order` attempts, and the `observed:{uid}` store key.
-/// The exact-UID match (never an `observed:` prefix scan) means a
-/// compute_uid divergence between module and collector cannot produce
-/// a false `Observed`.
-fn classify_ok(host: &MockHost, uid: &str, log_lines: &[String]) -> Classification {
-    if host.cow_api.call_count() > 0 {
-        return Classification::RejectedUnexpected(
-            "strategy called submit_order; observe+verify must never submit".into(),
-        );
-    }
-    let requests = host.cow_api.request_calls();
-    let expected_path = format!("/api/v1/orders/{uid}");
-    if requests.len() != 1 || requests[0].method != "GET" || requests[0].path != expected_path {
-        let shapes: Vec<String> = requests
-            .iter()
-            .map(|r| format!("{} {}", r.method, r.path))
-            .collect();
+/// the one watch's receipt renders to exactly the fixture UID (never a
+/// prefix scan, so a compute_uid divergence between module and
+/// collector cannot produce a false `Observed`), and the delivered
+/// transition wrote the `observed:{uid}` store key.
+fn classify_ok(host: &MockHost, receipt: &[u8], uid: &str, log_lines: &[String]) -> Classification {
+    let receipt_hex = format!("0x{}", hex::encode(receipt));
+    if receipt_hex != uid {
         return Classification::RejectedUnexpected(format!(
-            "expected exactly one GET {expected_path}; saw [{}]",
-            shapes.join(", ")
+            "watched receipt {receipt_hex} does not match the collector UID"
         ));
     }
     if host

crates/shepherd-backtest/src/report.rs

@@ -37,11 +37,12 @@ pub fn render(fx: &Fixtures, outcomes: &[ReplayOutcome], threshold: f64) -> Stri
     ));
     out.push_str(
         "Replays every collected EthFlow `OrderPlacement` event through the production \
-         `ethflow_watcher::strategy::on_chain_logs` code path via `shepherd_sdk_test::MockHost`. \
-         The orderbook is **never hit**: the MockHost programs a catch-all 200 for all \
-         `cow_api_request` calls so the observe+verify strategy sees every fixture as \
-         already indexed. Success is measured by whether the strategy wrote the exact \
-         `observed:{uid}` marker to the local store after the 200 confirmation.\n\n",
+         `ethflow_watcher::strategy` pair (`on_chain_logs` then `on_intent_status`) over \
+         `nexum_sdk_test::MockHost` and a recording pool transport. The orderbook is \
+         **never hit**: the transport accepts every watch and the replay delivers the \
+         `open` transition the registry would poll for it, so every fixture reads as \
+         indexed. Success is measured by whether the strategy watched exactly the \
+         fixture UID and wrote the exact `observed:{uid}` marker on the transition.\n\n",
     );
     out.push_str("## Run metadata\n\n");
     out.push_str("| Field | Value |\n|---|---|\n");

crates/shepherd-cow-host/Cargo.toml

@@ -50,4 +50,3 @@ toml.workspace = true
 tokio.workspace = true
 wiremock.workspace = true
 tempfile.workspace = true
-alloy-rpc-types-eth.workspace = true