diff --git a/Cargo.lock b/Cargo.lock
index 51ba4e86b..ceb289478 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1429,7 +1429,7 @@ dependencies = [
  "opentelemetry",
  "opentelemetry-otlp",
  "opentelemetry_sdk",
- "pallas",
+ "pallas 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
  "paste",
  "protoc-wkt",
  "rayon",
@@ -1477,7 +1477,7 @@ dependencies = [
  "itertools 0.14.0",
  "num-bigint",
  "num-rational",
- "pallas",
+ "pallas 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
  "paste",
  "proptest",
  "rayon",
@@ -1498,7 +1498,7 @@ dependencies = [
  "hex",
  "itertools 0.14.0",
  "opentelemetry",
- "pallas",
+ "pallas 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
  "proptest",
  "rayon",
  "regex",
@@ -1520,7 +1520,7 @@ dependencies = [
  "dolos-core",
  "dolos-testing",
  "fjall",
- "pallas",
+ "pallas 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
  "tempfile",
  "thiserror 2.0.18",
  "tracing",
@@ -1549,7 +1549,7 @@ dependencies = [
  "num-bigint",
  "num-rational",
  "num-traits",
- "pallas",
+ "pallas 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
  "rayon",
  "reqwest 0.12.28",
  "serde",
@@ -1573,7 +1573,7 @@ dependencies = [
  "dolos-testing",
  "hex",
  "http-body-util",
- "pallas",
+ "pallas 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
  "serde",
  "serde_json",
  "tokio",
@@ -1593,7 +1593,7 @@ dependencies = [
  "futures-util",
  "hex",
  "itertools 0.14.0",
- "pallas",
+ "pallas 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
  "redb",
  "redb-extras",
  "serde",
@@ -1620,7 +1620,7 @@ dependencies = [
  "futures-util",
  "hex",
  "itertools 0.14.0",
- "pallas",
+ "pallas 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
  "rand 0.9.3",
  "tokio",
  "tokio-stream",
@@ -1639,7 +1639,7 @@ dependencies = [
  "itertools 0.14.0",
  "jsonrpsee",
  "opentelemetry",
- "pallas",
+ "pallas 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
  "rand 0.9.3",
  "serde",
  "serde_json",
@@ -3330,7 +3330,7 @@ version = "0.50.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5"
 dependencies = [
- "windows-sys 0.59.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -3590,17 +3590,34 @@ version = "1.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "846ff0e3269c223e6ed6202eca1dc98399bc85f7db9be4e4574fab778cfa981b"
 dependencies = [
- "pallas-addresses",
- "pallas-codec",
- "pallas-configs",
- "pallas-crypto",
+ "pallas-addresses 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-codec 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-configs 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-crypto 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-network 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-primitives 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-traverse 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-txbuilder 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-utxorpc 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-validate 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "pallas"
+version = "1.1.1"
+source = "git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c#79dbeeba3a5d3894d06f87c2e894c21a8ec2619c"
+dependencies = [
+ "pallas-addresses 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-codec 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-configs 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-crypto 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
  "pallas-hardano",
- "pallas-network",
- "pallas-primitives",
- "pallas-traverse",
- "pallas-txbuilder",
- "pallas-utxorpc",
- "pallas-validate",
+ "pallas-network 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-primitives 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-traverse 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-txbuilder 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-utxorpc 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-validate 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
 ]
 
 [[package]]
@@ -3614,8 +3631,23 @@ dependencies = [
  "crc",
  "cryptoxide 0.4.4",
  "hex",
- "pallas-codec",
- "pallas-crypto",
+ "pallas-codec 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-crypto 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "thiserror 2.0.18",
+]
+
+[[package]]
+name = "pallas-addresses"
+version = "1.1.1"
+source = "git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c#79dbeeba3a5d3894d06f87c2e894c21a8ec2619c"
+dependencies = [
+ "base58",
+ "bech32 0.11.1",
+ "crc",
+ "cryptoxide 0.4.4",
+ "hex",
+ "pallas-codec 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-crypto 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
  "thiserror 2.0.18",
 ]
 
@@ -3631,6 +3663,17 @@ dependencies = [
  "thiserror 2.0.18",
 ]
 
+[[package]]
+name = "pallas-codec"
+version = "1.1.1"
+source = "git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c#79dbeeba3a5d3894d06f87c2e894c21a8ec2619c"
+dependencies = [
+ "hex",
+ "minicbor 0.26.4",
+ "serde",
+ "thiserror 2.0.18",
+]
+
 [[package]]
 name = "pallas-configs"
 version = "1.1.1"
@@ -3639,9 +3682,24 @@ checksum = "9413d9d84c67434857c08ec2d92c96452671c68fb1921c0c1999a7a699587186"
 dependencies = [
  "base64 0.22.1",
  "num-rational",
- "pallas-addresses",
- "pallas-crypto",
- "pallas-primitives",
+ "pallas-addresses 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-crypto 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-primitives 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "serde",
+ "serde_json",
+ "serde_with 3.21.0",
+]
+
+[[package]]
+name = "pallas-configs"
+version = "1.1.1"
+source = "git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c#79dbeeba3a5d3894d06f87c2e894c21a8ec2619c"
+dependencies = [
+ "base64 0.22.1",
+ "num-rational",
+ "pallas-addresses 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-crypto 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-primitives 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
  "serde",
  "serde_json",
  "serde_with 3.21.0",
@@ -3655,7 +3713,20 @@ checksum = "5abe09448466b168f50eb0beaae4e38a16e49bdfe6b2fba5748ea17a025b15b1"
 dependencies = [
  "cryptoxide 0.4.4",
  "hex",
- "pallas-codec",
+ "pallas-codec 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "rand_core 0.10.1",
+ "serde",
+ "thiserror 2.0.18",
+]
+
+[[package]]
+name = "pallas-crypto"
+version = "1.1.1"
+source = "git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c#79dbeeba3a5d3894d06f87c2e894c21a8ec2619c"
+dependencies = [
+ "cryptoxide 0.4.4",
+ "hex",
+ "pallas-codec 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
  "rand_core 0.10.1",
  "serde",
  "thiserror 2.0.18",
@@ -3664,16 +3735,15 @@ dependencies = [
 [[package]]
 name = "pallas-hardano"
 version = "1.1.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "adf818b49428697b0d0d5e680fecffa94020b141460dc6ebfe45295b43c8a6b1"
+source = "git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c#79dbeeba3a5d3894d06f87c2e894c21a8ec2619c"
 dependencies = [
  "binary-layout",
  "hex",
- "pallas-addresses",
- "pallas-codec",
- "pallas-crypto",
- "pallas-network",
- "pallas-traverse",
+ "pallas-addresses 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-codec 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-crypto 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-network 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-traverse 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
  "serde",
  "serde_json",
  "serde_with 3.21.0",
@@ -3691,8 +3761,25 @@ dependencies = [
  "byteorder",
  "hex",
  "itertools 0.14.0",
- "pallas-codec",
- "pallas-crypto",
+ "pallas-codec 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-crypto 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "rand 0.10.1",
+ "socket2 0.6.3",
+ "thiserror 2.0.18",
+ "tokio",
+ "tracing",
+]
+
+[[package]]
+name = "pallas-network"
+version = "1.1.1"
+source = "git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c#79dbeeba3a5d3894d06f87c2e894c21a8ec2619c"
+dependencies = [
+ "byteorder",
+ "hex",
+ "itertools 0.14.0",
+ "pallas-codec 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-crypto 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
  "rand 0.10.1",
  "socket2 0.6.3",
  "thiserror 2.0.18",
@@ -3707,8 +3794,20 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "35c2b627b97f708bbb0252834e300d56b1ccea9c982c0e4189d1a414c1b53c18"
 dependencies = [
  "hex",
- "pallas-codec",
- "pallas-crypto",
+ "pallas-codec 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-crypto 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "serde",
+ "serde_json",
+]
+
+[[package]]
+name = "pallas-primitives"
+version = "1.1.1"
+source = "git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c#79dbeeba3a5d3894d06f87c2e894c21a8ec2619c"
+dependencies = [
+ "hex",
+ "pallas-codec 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-crypto 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
  "serde",
  "serde_json",
 ]
@@ -3721,10 +3820,26 @@ checksum = "ba24ccce6561ab01d910aa8a89abf7f949f4009c0e36771e763f09d3efba1eed"
 dependencies = [
  "hex",
  "itertools 0.14.0",
- "pallas-addresses",
- "pallas-codec",
- "pallas-crypto",
- "pallas-primitives",
+ "pallas-addresses 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-codec 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-crypto 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-primitives 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "paste",
+ "serde",
+ "thiserror 2.0.18",
+]
+
+[[package]]
+name = "pallas-traverse"
+version = "1.1.1"
+source = "git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c#79dbeeba3a5d3894d06f87c2e894c21a8ec2619c"
+dependencies = [
+ "hex",
+ "itertools 0.14.0",
+ "pallas-addresses 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-codec 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-crypto 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-primitives 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
  "paste",
  "serde",
  "thiserror 2.0.18",
@@ -3737,11 +3852,27 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "24573093e2c342a63cda545ef799c51248a9ccb07061a9de0f2126b694830697"
 dependencies = [
  "hex",
- "pallas-addresses",
- "pallas-codec",
- "pallas-crypto",
- "pallas-primitives",
- "pallas-traverse",
+ "pallas-addresses 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-codec 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-crypto 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-primitives 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-traverse 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "serde",
+ "serde_json",
+ "thiserror 2.0.18",
+]
+
+[[package]]
+name = "pallas-txbuilder"
+version = "1.1.1"
+source = "git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c#79dbeeba3a5d3894d06f87c2e894c21a8ec2619c"
+dependencies = [
+ "hex",
+ "pallas-addresses 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-codec 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-crypto 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-primitives 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-traverse 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
  "serde",
  "serde_json",
  "thiserror 2.0.18",
@@ -3753,11 +3884,25 @@ version = "1.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bb8a5432ea38e559378d5872d012c5cf34df95e0ed7e5a224aa2645ce57f9675"
 dependencies = [
- "pallas-codec",
- "pallas-crypto",
- "pallas-primitives",
- "pallas-traverse",
- "pallas-validate",
+ "pallas-codec 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-crypto 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-primitives 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-traverse 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-validate 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "prost-types 0.13.5",
+ "utxorpc-spec 0.19.0",
+]
+
+[[package]]
+name = "pallas-utxorpc"
+version = "1.1.1"
+source = "git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c#79dbeeba3a5d3894d06f87c2e894c21a8ec2619c"
+dependencies = [
+ "pallas-codec 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-crypto 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-primitives 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-traverse 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-validate 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
  "prost-types 0.13.5",
  "utxorpc-spec 0.19.0",
 ]
@@ -3767,16 +3912,34 @@ name = "pallas-validate"
 version = "1.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "dfaac47eb61c648c32cb98bd9fc74414bd173ee57b17c59412242731d0365964"
+dependencies = [
+ "chrono",
+ "hex",
+ "itertools 0.14.0",
+ "pallas-addresses 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-codec 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-crypto 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-primitives 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-traverse 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "serde",
+ "thiserror 2.0.18",
+ "tracing",
+]
+
+[[package]]
+name = "pallas-validate"
+version = "1.1.1"
+source = "git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c#79dbeeba3a5d3894d06f87c2e894c21a8ec2619c"
 dependencies = [
  "amaru-uplc",
  "chrono",
  "hex",
  "itertools 0.14.0",
- "pallas-addresses",
- "pallas-codec",
- "pallas-crypto",
- "pallas-primitives",
- "pallas-traverse",
+ "pallas-addresses 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-codec 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-crypto 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-primitives 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
+ "pallas-traverse 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
  "serde",
  "thiserror 2.0.18",
  "tracing",
@@ -4083,7 +4246,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf"
 dependencies = [
  "heck",
- "itertools 0.14.0",
+ "itertools 0.12.1",
  "log",
  "multimap",
  "once_cell",
@@ -4116,7 +4279,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d"
 dependencies = [
  "anyhow",
- "itertools 0.14.0",
+ "itertools 0.12.1",
  "proc-macro2",
  "quote",
  "syn",
@@ -4129,7 +4292,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf"
 dependencies = [
  "anyhow",
- "itertools 0.14.0",
+ "itertools 0.12.1",
  "proc-macro2",
  "quote",
  "syn",
@@ -4715,7 +4878,7 @@ dependencies = [
  "security-framework 3.7.0",
  "security-framework-sys",
  "webpki-root-certs",
- "windows-sys 0.59.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -6059,7 +6222,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2a581b7e2051be3ce376d896e3a05e5ac4f74f486f9de1848f3df2c308776d21"
 dependencies = [
  "hex",
- "pallas",
+ "pallas 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "serde",
  "thiserror 2.0.18",
  "trait-variant",
@@ -6098,8 +6261,8 @@ dependencies = [
  "cryptoxide 0.4.4",
  "ed25519-bip32",
  "hex",
- "pallas-addresses",
- "pallas-crypto",
+ "pallas-addresses 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pallas-crypto 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "reqwest 0.12.28",
  "serde",
  "serde_json",
@@ -6977,7 +7140,7 @@ dependencies = [
  "dolos-cardano",
  "dolos-core",
  "hex",
- "pallas",
+ "pallas 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)",
  "postgres",
  "postgres-native-tls",
  "serde",
diff --git a/Cargo.toml b/Cargo.toml
index d01cf19b8..b3eafadc0 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -207,8 +207,8 @@ readme = "README.md"
 authors = ["TxPipe <hello@txpipe.io>"]
 
 [workspace.dependencies]
-pallas = { version = "1.1.1", features = ["hardano", "phase2", "unstable"] }
-# pallas = { git = "https://github.com/txpipe/pallas.git", rev = "a7b5a86", features = ["hardano", "phase2", "unstable"] }
+# pallas = { version = "1.1.1", features = ["hardano", "phase2", "unstable"] }
+pallas = { git = "https://github.com/Javieracost/pallas.git", rev = "79dbeeba3a5d3894d06f87c2e894c21a8ec2619c", features = ["hardano", "phase2", "unstable"] }
 # pallas = { path = "../pallas/pallas", features = ["hardano", "phase2", "unstable"] }
 
 thiserror = "2.0.12"
diff --git a/crates/core/src/crawl.rs b/crates/core/src/crawl.rs
index c62754e3f..6ed458557 100644
--- a/crates/core/src/crawl.rs
+++ b/crates/core/src/crawl.rs
@@ -36,11 +36,18 @@ impl<D: Domain> Batch<D> {
     }
 
     fn from_archive(last_point: ChainPoint, domain: &D) -> Result<Self, DomainError> {
+        // Origin is not a block, so there's nothing already seen to skip; for
+        // any concrete point, the block at the point itself was already
+        // delivered (or acknowledged via intersection) and must be skipped.
+        let seen = match &last_point {
+            ChainPoint::Origin => 0,
+            _ => 1,
+        };
+
         let page = domain
             .archive()
             .get_range(Some(last_point.slot()), None)?
-            // skip the last point, we already seen it
-            .skip(1)
+            .skip(seen)
             .take(LOAD_BATCH_SIZE)
             .map(|(slot, body)| (ChainPoint::Slot(slot), Arc::new(body)))
             .collect::<VecDeque<_>>();
diff --git a/src/serve/o7s_unix/chainsync.rs b/src/serve/o7s_unix/chainsync.rs
index a8f9a570f..0c37d62cd 100644
--- a/src/serve/o7s_unix/chainsync.rs
+++ b/src/serve/o7s_unix/chainsync.rs
@@ -4,14 +4,14 @@ use pallas::network::miniprotocols::{
     chainsync::{BlockContent, ClientRequest, N2CServer, Tip},
     Point,
 };
-use tracing::{debug, info};
+use tracing::{debug, info, warn};
 
 use crate::prelude::*;
 
 pub struct Session<D: Domain> {
     domain: D,
     crawler: Option<ChainCrawler<D>>,
-    is_new_intersection: bool,
+    pending_rollback: Option<ChainPoint>,
     connection: N2CServer,
 }
 
@@ -41,7 +41,9 @@ impl<D: Domain> Session<D> {
     async fn send_intersect_found(&mut self, point: ChainPoint) -> Result<(), Error> {
         debug!("sending intersection found");
 
-        self.is_new_intersection = true;
+        // chain-sync always starts by rolling the client back to the
+        // negotiated intersection before any forward event
+        self.pending_rollback = Some(point.clone());
 
         let tip = self.prepare_tip()?;
 
@@ -63,38 +65,29 @@ impl<D: Domain> Session<D> {
 
         debug!("waiting for tip change notification");
 
-        let crawler = self.assume_crawler();
-
-        let event = crawler.next_tip().await;
-
-        self.send_tip_event(event).await?;
-
-        Ok(())
+        // after an await-reply the client is owed exactly one chain-sync
+        // message; undo events have no wire counterpart (the mark that
+        // follows them becomes the rollback), so keep waiting through them
+        loop {
+            let event = self.assume_crawler().next_tip().await;
+
+            match event {
+                TipEvent::Apply(_, block) => return self.send_forward(block).await,
+                TipEvent::Mark(point) => return self.send_rollback(point).await,
+                _ => continue,
+            }
+        }
     }
 
-    async fn send_forward(&mut self, point: ChainPoint, block: RawBlock) -> Result<(), Error> {
+    async fn send_forward(&mut self, block: RawBlock) -> Result<(), Error> {
         debug!("sending forward event");
 
         let tip = self.prepare_tip()?;
 
-        let point = Point::try_from(point).map_err(|_| Error::custom("invalid point"))?;
-
-        // Ouroboros chain-sync always starts by sending the intersection point as an
-        // initial rollback event. The `is_new_intersection`` flag allows us to track if
-        // we have already sent that initial rollback or not
-        if self.is_new_intersection {
-            self.connection
-                .send_roll_backward(point, tip)
-                .await
-                .map_err(Error::server)?;
-
-            self.is_new_intersection = false;
-        } else {
-            self.connection
-                .send_roll_forward(BlockContent((*block).clone()), tip)
-                .await
-                .map_err(Error::server)?;
-        }
+        self.connection
+            .send_roll_forward(BlockContent((*block).clone()), tip)
+            .await
+            .map_err(Error::server)?;
 
         Ok(())
     }
@@ -112,24 +105,22 @@ impl<D: Domain> Session<D> {
             .map_err(Error::server)
     }
 
-    async fn send_tip_event(&mut self, log: TipEvent) -> Result<(), Error> {
-        match log {
-            TipEvent::Apply(p, b) => self.send_forward(p, b).await,
-            TipEvent::Mark(p) => self.send_rollback(p).await,
-            // we skip undo events and expect a Mark to come after
-            _ => Ok(()),
-        }
-    }
-
     async fn handle_next_request(&mut self) -> Result<(), Error> {
         debug!("handling next request");
 
+        // the rollback to the negotiated intersection is the reply to the
+        // first next-request; no block is consumed by it
+        if let Some(point) = self.pending_rollback.take() {
+            self.send_rollback(point).await?;
+            return Ok(());
+        }
+
         let crawler = self.assume_crawler();
 
         let next = crawler.next_block();
 
-        if let Some((point, block)) = next {
-            self.send_forward(point, block).await?;
+        if let Some((_, block)) = next {
+            self.send_forward(block).await?;
         } else {
             self.send_await_and_next().await?;
         }
@@ -196,11 +187,15 @@ pub async fn handle_session<D: Domain, C: CancelToken>(
         domain,
         connection,
         crawler: None,
-        is_new_intersection: false,
+        pending_rollback: None,
     };
 
     tokio::select! {
-        _ = session.process_requests() => {
+        result = session.process_requests() => {
+            if let Err(e) = result {
+                warn!(?e, "chainsync session error");
+                return Err(ServeError::Internal(e.into()));
+            }
             info!("client ended protocol");
         },
         _ = cancel.cancelled() => {
@@ -210,3 +205,147 @@ pub async fn handle_session<D: Domain, C: CancelToken>(
 
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    use tokio_util::sync::CancellationToken;
+
+    use dolos_core::ChainPoint;
+    use dolos_testing::blocks::make_conway_block;
+    use dolos_testing::slot_to_chainpoint;
+    use dolos_testing::toy_domain::ToyDomain;
+
+    use pallas::network::facades::{NodeClient, NodeServer};
+    use pallas::network::miniprotocols::chainsync::NextResponse;
+
+    use crate::serve::CancelTokenImpl;
+
+    fn spawn_server(
+        domain: ToyDomain,
+        listener: tokio::net::UnixListener,
+    ) -> tokio::task::JoinHandle<()> {
+        tokio::spawn(async move {
+            let connection = NodeServer::accept(&listener, 0).await.unwrap();
+
+            let NodeServer {
+                plexer, chainsync, ..
+            } = connection;
+
+            let cancel = CancelTokenImpl(CancellationToken::new());
+
+            handle_session(domain, chainsync, cancel).await.unwrap();
+
+            plexer.abort().await;
+        })
+    }
+
+    fn seed_archive_blocks(
+        domain: &ToyDomain,
+        slots: std::ops::Range<u64>,
+    ) -> Vec<(ChainPoint, Vec<u8>)> {
+        let writer = domain.archive().start_writer().unwrap();
+
+        let mut blocks = vec![];
+
+        for slot in slots {
+            let (point, block) = make_conway_block(slot);
+            writer.apply(&point, &block).unwrap();
+            blocks.push((point, (*block).clone()));
+        }
+
+        writer.commit().unwrap();
+
+        blocks
+    }
+
+    #[tokio::test]
+    async fn chainsync_starts_with_rollback_to_intersection() {
+        let domain = ToyDomain::new(None, None);
+
+        let tempdir = tempfile::tempdir().unwrap();
+        let socket = tempdir.path().join("node.socket");
+        let listener = tokio::net::UnixListener::bind(&socket).unwrap();
+
+        let server = spawn_server(domain.clone(), listener);
+
+        let mut client = NodeClient::connect(&socket, 0).await.unwrap();
+
+        let (point, _) = client
+            .chainsync()
+            .find_intersect(vec![Point::Origin])
+            .await
+            .unwrap();
+
+        assert_eq!(point, Some(Point::Origin));
+
+        // the reply to the first next-request is the rollback to the
+        // negotiated intersection
+        let next = client.chainsync().request_next().await.unwrap();
+        assert!(matches!(next, NextResponse::RollBackward(Point::Origin, _)));
+
+        // an empty chain then awaits at tip; undo events carry no chain-sync
+        // message of their own, so the reply must come from the mark that
+        // follows them
+        let next = client.chainsync().request_next().await.unwrap();
+        assert!(matches!(next, NextResponse::Await));
+
+        let (undone_point, undone_block) = make_conway_block(9);
+        domain.notify_tip(dolos_core::TipEvent::Undo(undone_point, undone_block));
+        domain.notify_tip(dolos_core::TipEvent::Mark(slot_to_chainpoint(7)));
+
+        let next = client.chainsync().recv_while_must_reply().await.unwrap();
+        assert!(matches!(next, NextResponse::RollBackward(_, _)));
+
+        client.chainsync().send_done().await.unwrap();
+
+        server.await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn chainsync_streams_archive_blocks_without_loss() {
+        let domain = ToyDomain::new(None, None);
+
+        let blocks = seed_archive_blocks(&domain, 1..15);
+
+        let tempdir = tempfile::tempdir().unwrap();
+        let socket = tempdir.path().join("node.socket");
+        let listener = tokio::net::UnixListener::bind(&socket).unwrap();
+
+        let server = spawn_server(domain.clone(), listener);
+
+        let mut client = NodeClient::connect(&socket, 0).await.unwrap();
+
+        let intersect = Point::try_from(blocks[0].0.clone()).unwrap();
+
+        let (point, _) = client
+            .chainsync()
+            .find_intersect(vec![intersect.clone()])
+            .await
+            .unwrap();
+
+        assert_eq!(point, Some(intersect.clone()));
+
+        let next = client.chainsync().request_next().await.unwrap();
+        match next {
+            NextResponse::RollBackward(p, _) => assert_eq!(p, intersect),
+            other => panic!("expected initial rollback, got {other:?}"),
+        }
+
+        // every block after the intersection must arrive, in order, starting
+        // right after the intersection point and across the internal batch
+        // boundary
+        for (_, expected) in &blocks[1..13] {
+            let next = client.chainsync().request_next().await.unwrap();
+            match next {
+                NextResponse::RollForward(content, _) => assert_eq!(&content.0, expected),
+                other => panic!("expected roll forward, got {other:?}"),
+            }
+        }
+
+        client.chainsync().send_done().await.unwrap();
+
+        server.await.unwrap();
+    }
+}
diff --git a/src/serve/o7s_unix/mod.rs b/src/serve/o7s_unix/mod.rs
index 39232f4fe..66f5c3eae 100644
--- a/src/serve/o7s_unix/mod.rs
+++ b/src/serve/o7s_unix/mod.rs
@@ -8,6 +8,7 @@ use crate::prelude::*;
 
 mod chainsync;
 mod statequery;
+mod txmonitor;
 mod utils;
 
 #[derive(Clone)]
@@ -29,6 +30,7 @@ async fn handle_session<D: Domain, C: CancelToken>(
         plexer,
         chainsync,
         statequery,
+        txmonitor,
         ..
     } = connection;
 
@@ -44,13 +46,21 @@ async fn handle_session<D: Domain, C: CancelToken>(
         cancel.clone(),
     ));
 
-    let result = tokio::try_join!(chainsync_task, statequery_task);
+    let txmonitor_task = tokio::spawn(txmonitor::handle_session(
+        domain.clone(),
+        txmonitor,
+        cancel.clone(),
+    ));
+
+    let result = tokio::try_join!(chainsync_task, statequery_task, txmonitor_task);
     plexer.abort().await;
 
-    let (chainsync_res, statequery_res) = result.map_err(|e| ServeError::Internal(e.into()))?;
+    let (chainsync_res, statequery_res, txmonitor_res) =
+        result.map_err(|e| ServeError::Internal(e.into()))?;
 
     chainsync_res?;
     statequery_res?;
+    txmonitor_res?;
 
     Ok(())
 }
diff --git a/src/serve/o7s_unix/statequery.rs b/src/serve/o7s_unix/statequery.rs
index d891cb27a..0a6d9697c 100644
--- a/src/serve/o7s_unix/statequery.rs
+++ b/src/serve/o7s_unix/statequery.rs
@@ -67,20 +67,33 @@ impl<D: Domain> Session<D> {
         Ok(())
     }
 
+    /// Returns whether the acquire succeeded. On failure, the protocol is
+    /// back in the idle state and the client is expected to acquire again.
     async fn handle_acquire(
         &mut self,
         point: Option<pallas::network::miniprotocols::Point>,
-    ) -> Result<(), Error> {
+    ) -> Result<bool, Error> {
         debug!(?point, "handling acquire request");
 
-        let chain_point = match point {
-            Some(p) => ChainPoint::from(p),
-            None => {
-                // None means acquire the latest point
-                self.tip_cursor()?
-            }
+        let Some(p) = point else {
+            // None means acquire the volatile tip, which always succeeds at
+            // the current cursor. The archive may still be catching up to the
+            // cursor while the node syncs, so it can't be used to validate
+            // the node's own tip.
+            let cursor = self
+                .domain
+                .state()
+                .read_cursor()
+                .map_err(Error::server)?
+                .unwrap_or(ChainPoint::Origin);
+
+            self.acquired_point = Some(cursor);
+            self.send_acquired().await?;
+            return Ok(true);
         };
 
+        let chain_point = ChainPoint::from(p);
+
         let exists = match &chain_point {
             ChainPoint::Origin => true,
             ChainPoint::Specific(slot, hash) => {
@@ -103,12 +116,12 @@ impl<D: Domain> Session<D> {
         if exists {
             self.acquired_point = Some(chain_point);
             self.send_acquired().await?;
+            Ok(true)
         } else {
             self.send_failure(localstate::AcquireFailure::PointNotOnChain)
                 .await?;
+            Ok(false)
         }
-
-        Ok(())
     }
 
     /// Decode cardano-cli tagged query format.
@@ -348,10 +361,12 @@ impl<D: Domain> Session<D> {
         Ok(())
     }
 
+    /// Returns whether the re-acquire succeeded. On failure, the protocol is
+    /// back in the idle state, not the acquired state.
     async fn handle_reacquire(
         &mut self,
         point: Option<pallas::network::miniprotocols::Point>,
-    ) -> Result<(), Error> {
+    ) -> Result<bool, Error> {
         debug!(?point, "handling reacquire request");
         self.handle_acquire(point).await
     }
@@ -370,7 +385,10 @@ impl<D: Domain> Session<D> {
                 .await
                 .map_err(Error::server)?
             {
-                self.handle_acquire(req.0).await?;
+                if !self.handle_acquire(req.0).await? {
+                    // failed acquire leaves the protocol idle
+                    continue;
+                }
             } else {
                 break;
             }
@@ -386,9 +404,10 @@ impl<D: Domain> Session<D> {
                         self.handle_query(query).await?;
                     }
                     localstate::ClientQueryRequest::ReAcquire(point) => {
-                        self.handle_reacquire(point).await?;
-                        // After reacquire, we stay in acquired state but with
-                        // new point
+                        if !self.handle_reacquire(point).await? {
+                            // failed reacquire leaves the protocol idle
+                            break;
+                        }
                     }
                     localstate::ClientQueryRequest::Release => {
                         self.handle_release().await?;
@@ -431,3 +450,94 @@ pub async fn handle_session<D: Domain, C: CancelToken>(
 
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    use tokio_util::sync::CancellationToken;
+
+    use dolos_testing::toy_domain::ToyDomain;
+
+    use pallas::network::facades::{NodeClient, NodeServer};
+    use pallas::network::miniprotocols::localstate::ClientError;
+    use pallas::network::miniprotocols::Point;
+
+    use crate::serve::CancelTokenImpl;
+
+    fn spawn_server(
+        domain: ToyDomain,
+        listener: tokio::net::UnixListener,
+    ) -> tokio::task::JoinHandle<()> {
+        tokio::spawn(async move {
+            let connection = NodeServer::accept(&listener, 0).await.unwrap();
+
+            let NodeServer {
+                plexer, statequery, ..
+            } = connection;
+
+            let cancel = CancelTokenImpl(CancellationToken::new());
+
+            handle_session(domain, statequery, cancel).await.unwrap();
+
+            plexer.abort().await;
+        })
+    }
+
+    #[tokio::test]
+    async fn statequery_acquires_volatile_tip() {
+        let domain = ToyDomain::new(None, None);
+
+        let tempdir = tempfile::tempdir().unwrap();
+        let socket = tempdir.path().join("node.socket");
+        let listener = tokio::net::UnixListener::bind(&socket).unwrap();
+
+        let server = spawn_server(domain, listener);
+
+        let mut client = NodeClient::connect(&socket, 0).await.unwrap();
+
+        client.statequery().acquire(None).await.unwrap();
+
+        client.statequery().send_release().await.unwrap();
+        client.statequery().send_done().await.unwrap();
+
+        server.await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn statequery_recovers_from_failed_acquire() {
+        let domain = ToyDomain::new(None, None);
+
+        let tempdir = tempfile::tempdir().unwrap();
+        let socket = tempdir.path().join("node.socket");
+        let listener = tokio::net::UnixListener::bind(&socket).unwrap();
+
+        let server = spawn_server(domain, listener);
+
+        let mut client = NodeClient::connect(&socket, 0).await.unwrap();
+
+        let bogus = Point::Specific(1, vec![0xab; 32]);
+
+        let failure = client.statequery().acquire(Some(bogus.clone())).await;
+        assert!(matches!(failure, Err(ClientError::AcquirePointNotFound)));
+
+        // the session must survive a failed acquire and accept a new one
+        client.statequery().acquire(None).await.unwrap();
+
+        // same for a failed re-acquire from the acquired state
+        client
+            .statequery()
+            .send_reacquire(Some(bogus))
+            .await
+            .unwrap();
+        let failure = client.statequery().recv_while_acquiring().await;
+        assert!(matches!(failure, Err(ClientError::AcquirePointNotFound)));
+
+        client.statequery().acquire(None).await.unwrap();
+
+        client.statequery().send_release().await.unwrap();
+        client.statequery().send_done().await.unwrap();
+
+        server.await.unwrap();
+    }
+}
diff --git a/src/serve/o7s_unix/txmonitor.rs b/src/serve/o7s_unix/txmonitor.rs
new file mode 100644
index 000000000..34421ca6c
--- /dev/null
+++ b/src/serve/o7s_unix/txmonitor.rs
@@ -0,0 +1,453 @@
+use futures_util::StreamExt as _;
+use std::collections::BTreeSet;
+
+use pallas::codec::utils::TagWrap;
+use pallas::network::miniprotocols::txmonitor::{
+    self, ClientQueryRequest, MempoolMeasures, MempoolSizeAndCapacity, SizeAndCapacity,
+};
+use tracing::{debug, info, warn};
+
+use dolos_cardano::load_effective_pparams;
+use dolos_core::{MempoolStore as _, StateStore as _, TipSubscription as _};
+
+use crate::prelude::*;
+
+// HACK: the tx era number differs from the block era number, we subtract 1 to
+// make them match.
+fn to_n2c_era(era: u16) -> u8 {
+    (era - 1) as u8
+}
+
+struct Snapshot {
+    slot: u64,
+    txs: Vec<MempoolTx>,
+    cursor: usize,
+}
+
+impl Snapshot {
+    fn fingerprint(&self) -> (u64, BTreeSet<TxHash>) {
+        (self.slot, self.txs.iter().map(|tx| tx.hash).collect())
+    }
+
+    fn size_in_bytes(&self) -> u64 {
+        self.txs
+            .iter()
+            .map(|tx| tx.payload.cbor().len() as u64)
+            .sum()
+    }
+}
+
+pub struct Session<D: Domain> {
+    domain: D,
+    connection: txmonitor::Server,
+    snapshot: Option<Snapshot>,
+}
+
+impl<D: Domain> Session<D> {
+    fn take_snapshot(&self) -> Result<Snapshot, Error> {
+        let point = self
+            .domain
+            .state()
+            .read_cursor()
+            .map_err(Error::server)?
+            .unwrap_or(ChainPoint::Origin);
+
+        let mempool = self.domain.mempool();
+
+        // confirmed txs are already on-chain, a node mempool wouldn't hold them
+        let txs = mempool
+            .peek_pending()
+            .into_iter()
+            .chain(mempool.peek_inflight())
+            .filter(|tx| !matches!(tx.stage, MempoolTxStage::Confirmed))
+            .collect();
+
+        Ok(Snapshot {
+            slot: point.slot(),
+            txs,
+            cursor: 0,
+        })
+    }
+
+    fn acquired(&self) -> &Snapshot {
+        self.snapshot
+            .as_ref()
+            .expect("txmonitor request handled without an acquired snapshot")
+    }
+
+    async fn handle_acquire(&mut self) -> Result<(), Error> {
+        let snapshot = self.take_snapshot()?;
+        let slot = snapshot.slot;
+
+        debug!(slot, txs = snapshot.txs.len(), "acquired mempool snapshot");
+
+        self.snapshot = Some(snapshot);
+
+        self.connection
+            .send_acquired(slot)
+            .await
+            .map_err(Error::server)?;
+
+        Ok(())
+    }
+
+    async fn handle_await_acquire(&mut self) -> Result<(), Error> {
+        let previous = self.acquired().fingerprint();
+
+        let mut mempool_updates = self.domain.mempool().subscribe();
+        let mut mempool_open = true;
+
+        let mut tip_updates = self
+            .domain
+            .watch_tip(None)
+            .map_err(|e| Error::server(e.to_string()))?;
+
+        loop {
+            let snapshot = self.take_snapshot()?;
+
+            if snapshot.fingerprint() != previous {
+                let slot = snapshot.slot;
+
+                debug!(slot, txs = snapshot.txs.len(), "acquired changed snapshot");
+
+                self.snapshot = Some(snapshot);
+
+                self.connection
+                    .send_acquired(slot)
+                    .await
+                    .map_err(Error::server)?;
+
+                return Ok(());
+            }
+
+            tokio::select! {
+                update = mempool_updates.next(), if mempool_open => {
+                    match update {
+                        Some(Err(e)) => return Err(Error::server(e)),
+                        Some(Ok(_)) => (),
+                        // a closed stream must not busy-loop the select; keep
+                        // waking on tip changes only
+                        None => mempool_open = false,
+                    }
+                }
+                _ = tip_updates.next_tip() => (),
+            }
+        }
+    }
+
+    async fn handle_next_tx(&mut self) -> Result<(), Error> {
+        let snapshot = self
+            .snapshot
+            .as_mut()
+            .expect("txmonitor request handled without an acquired snapshot");
+
+        let tx = snapshot.txs.get(snapshot.cursor).map(|tx| {
+            let EraCbor(era, cbor) = &tx.payload;
+            (to_n2c_era(*era), TagWrap::new(cbor.clone().into()))
+        });
+
+        if tx.is_some() {
+            snapshot.cursor += 1;
+        }
+
+        self.connection
+            .send_next_tx(tx)
+            .await
+            .map_err(Error::server)?;
+
+        Ok(())
+    }
+
+    async fn handle_has_tx(&mut self, id: txmonitor::TxId) -> Result<(), Error> {
+        // peers wrap the same hash in every plausible era and combine the
+        // answers, so we match on the hash alone and ignore the era tag
+        let (_era, hash) = &id;
+
+        let has = self
+            .acquired()
+            .txs
+            .iter()
+            .any(|tx| tx.hash.as_slice() == hash.as_slice());
+
+        self.connection
+            .send_has_tx(has)
+            .await
+            .map_err(Error::server)?;
+
+        Ok(())
+    }
+
+    fn capacity_in_bytes(&self) -> Result<u64, Error> {
+        // cardano-node sizes its mempool at twice the max block body by
+        // default; report the same so clients get a familiar reference
+        let pparams = load_effective_pparams::<D>(self.domain.state())
+            .map_err(|e| Error::server(e.to_string()))?;
+
+        Ok(pparams.max_block_body_size_or_default() * 2)
+    }
+
+    async fn handle_get_sizes(&mut self) -> Result<(), Error> {
+        let capacity = self.capacity_in_bytes()?;
+        let snapshot = self.acquired();
+
+        let sizes = MempoolSizeAndCapacity {
+            capacity_in_bytes: capacity as u32,
+            size_in_bytes: snapshot.size_in_bytes() as u32,
+            number_of_txs: snapshot.txs.len() as u32,
+        };
+
+        self.connection
+            .send_size_and_capacity(sizes)
+            .await
+            .map_err(Error::server)?;
+
+        Ok(())
+    }
+
+    async fn handle_get_measures(&mut self) -> Result<(), Error> {
+        let capacity = self.capacity_in_bytes()?;
+        let snapshot = self.acquired();
+
+        let measures = MempoolMeasures {
+            tx_count: snapshot.txs.len() as u32,
+            measures: vec![(
+                "transaction_bytes".to_string(),
+                SizeAndCapacity {
+                    size: snapshot.size_in_bytes(),
+                    capacity,
+                },
+            )],
+        };
+
+        self.connection
+            .send_measures(measures)
+            .await
+            .map_err(Error::server)?;
+
+        Ok(())
+    }
+
+    async fn process_requests(&mut self) -> Result<(), Error> {
+        loop {
+            if self
+                .connection
+                .recv_while_idle()
+                .await
+                .map_err(Error::server)?
+                .is_none()
+            {
+                break;
+            }
+
+            self.handle_acquire().await?;
+
+            loop {
+                match self
+                    .connection
+                    .recv_while_acquired()
+                    .await
+                    .map_err(Error::server)?
+                {
+                    ClientQueryRequest::AwaitAcquire => {
+                        self.handle_await_acquire().await?;
+                    }
+                    ClientQueryRequest::NextTx => {
+                        self.handle_next_tx().await?;
+                    }
+                    ClientQueryRequest::HasTx(id) => {
+                        self.handle_has_tx(id).await?;
+                    }
+                    ClientQueryRequest::GetSizes => {
+                        self.handle_get_sizes().await?;
+                    }
+                    ClientQueryRequest::GetMeasures => {
+                        self.handle_get_measures().await?;
+                    }
+                    ClientQueryRequest::Release => {
+                        self.snapshot = None;
+                        break;
+                    }
+                }
+            }
+        }
+
+        Ok(())
+    }
+}
+
+pub async fn handle_session<D: Domain, C: CancelToken>(
+    domain: D,
+    connection: txmonitor::Server,
+    cancel: C,
+) -> Result<(), ServeError> {
+    let mut session = Session {
+        domain,
+        connection,
+        snapshot: None,
+    };
+
+    info!("txmonitor session started");
+
+    tokio::select! {
+        result = session.process_requests() => {
+            if let Err(e) = result {
+                warn!(?e, "txmonitor session error");
+                return Err(ServeError::Internal(e.into()));
+            }
+            info!("txmonitor client ended protocol");
+        },
+        _ = cancel.cancelled() => {
+            info!("txmonitor protocol was cancelled");
+        }
+    }
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    use tokio_util::sync::CancellationToken;
+
+    use dolos_testing::mempool::make_test_mempool_tx;
+    use dolos_testing::slot_to_chainpoint;
+    use dolos_testing::toy_domain::ToyDomain;
+
+    use pallas::network::facades::{NodeClient, NodeServer};
+
+    use crate::serve::CancelTokenImpl;
+
+    fn spawn_server(
+        domain: ToyDomain,
+        listener: tokio::net::UnixListener,
+    ) -> tokio::task::JoinHandle<()> {
+        tokio::spawn(async move {
+            let connection = NodeServer::accept(&listener, 0).await.unwrap();
+
+            let NodeServer {
+                plexer, txmonitor, ..
+            } = connection;
+
+            let cancel = CancelTokenImpl(CancellationToken::new());
+
+            handle_session(domain, txmonitor, cancel).await.unwrap();
+
+            plexer.abort().await;
+        })
+    }
+
+    #[tokio::test]
+    async fn txmonitor_serves_snapshot_queries() {
+        let domain = ToyDomain::new(None, None);
+
+        let seeded_hash = TxHash::from([0xab; 32]);
+
+        domain
+            .mempool()
+            .receive(make_test_mempool_tx(seeded_hash))
+            .unwrap();
+
+        let tempdir = tempfile::tempdir().unwrap();
+        let socket = tempdir.path().join("node.socket");
+        let listener = tokio::net::UnixListener::bind(&socket).unwrap();
+
+        let server = spawn_server(domain, listener);
+
+        let mut client = NodeClient::connect(&socket, 0).await.unwrap();
+
+        client.monitor().acquire().await.unwrap();
+
+        let sizes = client.monitor().query_size_and_capacity().await.unwrap();
+        assert_eq!(sizes.number_of_txs, 1);
+        assert_eq!(sizes.size_in_bytes, 1);
+        assert!(sizes.capacity_in_bytes > 0);
+
+        let (era, body) = client.monitor().query_next_tx().await.unwrap().unwrap();
+        assert_eq!(era, 6);
+        assert_eq!(body.0.len(), 1);
+
+        let next = client.monitor().query_next_tx().await.unwrap();
+        assert!(next.is_none());
+
+        let has = client
+            .monitor()
+            .query_has_tx((6, seeded_hash.to_vec().into()))
+            .await
+            .unwrap();
+        assert!(has);
+
+        // the era wrapper is not part of the identity check
+        let has = client
+            .monitor()
+            .query_has_tx((4, seeded_hash.to_vec().into()))
+            .await
+            .unwrap();
+        assert!(has);
+
+        let has = client
+            .monitor()
+            .query_has_tx((6, vec![0xcd; 32].into()))
+            .await
+            .unwrap();
+        assert!(!has);
+
+        let measures = client.monitor().query_measures().await.unwrap();
+        assert_eq!(measures.tx_count, 1);
+
+        client.monitor().release().await.unwrap();
+        client.monitor().done().await.unwrap();
+
+        server.await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn txmonitor_await_acquire_wakes_on_change() {
+        let domain = ToyDomain::new(None, None);
+
+        let tempdir = tempfile::tempdir().unwrap();
+        let socket = tempdir.path().join("node.socket");
+        let listener = tokio::net::UnixListener::bind(&socket).unwrap();
+
+        let server = spawn_server(domain.clone(), listener);
+
+        let mut client = NodeClient::connect(&socket, 0).await.unwrap();
+
+        client.monitor().acquire().await.unwrap();
+
+        let new_hash = TxHash::from([0x11; 32]);
+
+        let trigger = tokio::spawn({
+            let domain = domain.clone();
+
+            async move {
+                tokio::time::sleep(std::time::Duration::from_millis(300)).await;
+
+                domain
+                    .mempool()
+                    .receive(make_test_mempool_tx(new_hash))
+                    .unwrap();
+
+                // the toy mempool emits no events; a tip event provides the
+                // wake-up, mirroring a new block reaching the node
+                domain.notify_tip(TipEvent::Mark(slot_to_chainpoint(1)));
+            }
+        });
+
+        client.monitor().await_acquire().await.unwrap();
+
+        let has = client
+            .monitor()
+            .query_has_tx((6, new_hash.to_vec().into()))
+            .await
+            .unwrap();
+        assert!(has);
+
+        trigger.await.unwrap();
+
+        client.monitor().release().await.unwrap();
+        client.monitor().done().await.unwrap();
+
+        server.await.unwrap();
+    }
+}