Phase 4b follow-on: pin --ttl RPC + CLI

Closes Phase 4b by exposing the TTL sidecar written by
Phase 4b primitives on the wire and via `claw-cargo pin`.

Wire additions:
* Method::SetTagExpiry (0x1a) — payload `key_len:u16 || key ||
  expires_at:u64 (LE)`. Reply single-byte OK. `expires_at == 0`
  clears the sidecar.
* Method::GetTagExpiry (0x1b) — payload raw key bytes. Reply 8
  bytes (u64 LE) on hit; NotFound when no sidecar is present.

Both accept writes even when the stamped tag itself is absent,
matching `TagStore::set_stamped_expiry` semantics — the sidecar
takes effect the moment the tag lands.

CLI:
* `claw-cargo pin --ttl <duration>` — humantime-style duration
  (`30d`, `1h30m`, `2w`, ...). Applied to both the primary tag
  and its `.fingerprint` companion so eviction treats them as
  one lifetime. `--ttl 0` / `clear` / `none` clears an existing
  sidecar without touching the value.

Tests: encode/decode roundtrip + malformed-input rejection for
`encode_expiry_record`, method-byte stability, NotConfigured
without a tag store, end-to-end set/get/overwrite/clear over
QUIC, and a real-pin flow that publishes a stamped tag then
attaches TTL. Duration parser is unit-tested for single/compound
forms, case-insensitive units, bad input, and clock alignment.

No new deps — the humantime-style parser is 60 lines in-tree.

Co-Authored-By: Claude Opus 4.7 <[email protected]>

claw-store/src/cluster/rpc/client.rs

@@ -598,6 +598,65 @@ pub async fn call_get_tag_versioned(
     ))
 }
 
+// ── Phase 4b follow-on: TTL client helpers ───────────────────────────
+
+/// Phase 4b follow-on (2026-07-13): attach a TTL sidecar to a stamped
+/// tag on a peer. `expires_at_unix == 0` clears any prior sidecar.
+///
+/// The peer accepts writes even when the stamped tag isn't present
+/// yet — the sidecar sticks around and takes effect once the tag
+/// lands (`TagStore::set_stamped_expiry` semantics).
+pub async fn call_set_tag_expiry(
+    conn: &Connection,
+    key: &str,
+    expires_at_unix: u64,
+) -> Result<()> {
+    let payload = crate::cluster::tags::encode_expiry_record(key, expires_at_unix);
+    let reply = rpc_call(conn, Method::SetTagExpiry, &payload).await?;
+    if reply.len() != 1 {
+        bail!(
+            "expected single-byte SetTagExpiry reply, got {} bytes",
+            reply.len()
+        );
+    }
+    match reply[0] {
+        STREAM_STATUS_OK => Ok(()),
+        code => match decode_error(code) {
+            Some(err) => bail!("peer rejected SetTagExpiry: {}", err.describe()),
+            None => bail!(
+                "peer replied with unknown byte 0x{:02x} for SetTagExpiry",
+                code
+            ),
+        },
+    }
+}
+
+/// Phase 4b follow-on: fetch the TTL sidecar for a stamped tag.
+/// Returns `Ok(None)` when no sidecar is present (never expires or
+/// no such tag).
+pub async fn call_get_tag_expiry(
+    conn: &Connection,
+    key: &str,
+) -> Result<Option<u64>> {
+    let reply = rpc_call(conn, Method::GetTagExpiry, key.as_bytes()).await?;
+    if reply.len() == 1 {
+        match decode_error(reply[0]) {
+            Some(ErrorCode::NotFound) => return Ok(None),
+            Some(err) => bail!("peer replied with error: {}", err.describe()),
+            None => {}
+        }
+    }
+    if reply.len() != 8 {
+        bail!(
+            "expected 8-byte GetTagExpiry reply, got {} bytes",
+            reply.len()
+        );
+    }
+    Ok(Some(u64::from_le_bytes(
+        reply.as_slice().try_into().expect("checked length"),
+    )))
+}
+
 // ── Phase 5g: cache metrics client helper ────────────────────────────
 
 /// Fetch the peer's current cache-metrics snapshot.

claw-store/src/cluster/rpc/tests_phase4b_ttl.rs

@@ -0,0 +1,177 @@
+//! Phase 4b follow-on (2026-07-13): SetTagExpiry / GetTagExpiry RPC
+//! end-to-end. Kept in its own file to stay under the 1300-line
+//! ceiling and to keep TTL wiring in one place.
+
+use super::*;
+use crate::cluster::transport::{NodeIdentity, QuicClient, QuicServer};
+use crate::config::ClusterConfig;
+use std::net::SocketAddr;
+use std::sync::atomic::{AtomicU16, Ordering};
+use std::time::Duration;
+
+static NEXT_PORT: AtomicU16 = AtomicU16::new(47001);
+fn next_port() -> u16 {
+    NEXT_PORT.fetch_add(1, Ordering::Relaxed)
+}
+
+fn loopback(port: u16) -> SocketAddr {
+    format!("127.0.0.1:{port}").parse().unwrap()
+}
+
+async fn bootstrap_gossip(name: &str, port: u16) -> Arc<ClusterGossip> {
+    let cfg = ClusterConfig {
+        zone: "fabric-10g".into(),
+        bind_lan: Some(loopback(port)),
+        ..Default::default()
+    };
+    Arc::new(ClusterGossip::bootstrap(&cfg, name).await.unwrap())
+}
+
+async fn router_with_full_stack(name: &str, port: u16) -> (tempfile::TempDir, Arc<RpcRouter>) {
+    use crate::cluster::refs::RefStore;
+    use crate::cluster::tags::TagStore;
+    let gossip = bootstrap_gossip(name, port).await;
+    let tmp = tempfile::TempDir::new().unwrap();
+    let blob_store =
+        Arc::new(crate::cluster::blob::BlobStore::open(tmp.path().join("blobs")).unwrap());
+    let ref_store = Arc::new(RefStore::open(tmp.path().join("refs-db")).unwrap());
+    let tag_store = Arc::new(TagStore::open(tmp.path().join("tags-db")).unwrap());
+    let router = Arc::new(
+        RpcRouter::new(gossip, name.into(), "fabric-10g".into())
+            .with_blob_store(blob_store)
+            .with_ref_store(ref_store)
+            .with_tag_store(tag_store),
+    );
+    (tmp, router)
+}
+
+#[test]
+fn ttl_method_bytes_stable() {
+    assert_eq!(Method::SetTagExpiry.as_byte(), 0x1a);
+    assert_eq!(Method::GetTagExpiry.as_byte(), 0x1b);
+    for m in [Method::SetTagExpiry, Method::GetTagExpiry] {
+        assert_eq!(Method::from_byte(m.as_byte()), Some(m));
+    }
+}
+
+#[tokio::test]
+async fn ttl_rpcs_return_not_configured_without_store() {
+    let gossip = bootstrap_gossip("solo", next_port()).await;
+    let router = RpcRouter::new(gossip, "solo".into(), "z".into());
+    // SetTagExpiry needs a valid record; GetTagExpiry needs a key.
+    let payload = crate::cluster::tags::encode_expiry_record("k", 42);
+    let mut set_req = vec![Method::SetTagExpiry.as_byte()];
+    set_req.extend_from_slice(&payload);
+    assert_eq!(
+        dispatch(&router, &set_req).await,
+        vec![ErrorCode::NotConfigured.as_byte()]
+    );
+
+    let mut get_req = vec![Method::GetTagExpiry.as_byte()];
+    get_req.extend_from_slice(b"k");
+    assert_eq!(
+        dispatch(&router, &get_req).await,
+        vec![ErrorCode::NotConfigured.as_byte()]
+    );
+}
+
+#[tokio::test]
+async fn end_to_end_set_and_get_tag_expiry() {
+    let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
+    let (_tmp_a, router_a) = router_with_full_stack("a", next_port()).await;
+
+    let server_a = QuicServer::bind(loopback(0), id_a).unwrap();
+    let addr = server_a.local_addr().unwrap();
+    let ra = router_a.clone();
+    let acc = tokio::spawn(async move {
+        while let Some(Ok(conn)) = server_a.accept().await {
+            let r = ra.clone();
+            tokio::spawn(async move {
+                let _ = serve_connection(conn, r).await;
+            });
+        }
+    });
+    let client = QuicClient::new(loopback(0), id_b).unwrap();
+    let conn = client.connect(addr, "a").await.unwrap();
+
+    let key = "clawverse:main:latest";
+
+    // No sidecar yet.
+    assert_eq!(call_get_tag_expiry(&conn, key).await.unwrap(), None);
+
+    // Set an absolute expiry — no requirement that the stamped tag
+    // already exist (matches TagStore::set_stamped_expiry semantics).
+    call_set_tag_expiry(&conn, key, 1_800_000_000).await.unwrap();
+    assert_eq!(
+        call_get_tag_expiry(&conn, key).await.unwrap(),
+        Some(1_800_000_000)
+    );
+
+    // Overwrite with a later value.
+    call_set_tag_expiry(&conn, key, 1_900_000_000).await.unwrap();
+    assert_eq!(
+        call_get_tag_expiry(&conn, key).await.unwrap(),
+        Some(1_900_000_000)
+    );
+
+    // Clear (expires_at == 0).
+    call_set_tag_expiry(&conn, key, 0).await.unwrap();
+    assert_eq!(call_get_tag_expiry(&conn, key).await.unwrap(), None);
+
+    conn.close(quinn::VarInt::from_u32(0), b"done");
+    client.shutdown().await;
+    tokio::time::sleep(Duration::from_millis(50)).await;
+    acc.abort();
+}
+
+#[tokio::test]
+async fn ttl_survives_process_boundary_via_pin_flow() {
+    // Real pin flow: PutTagVersioned, then SetTagExpiry, then read
+    // both back through the same connection. Exercises the exact
+    // sequence the pin --ttl CLI will emit.
+    use crate::cluster::refs::node_stamp_for;
+    use crate::cluster::tags::StampedTagValue;
+
+    let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
+    let (_tmp_a, router_a) = router_with_full_stack("a", next_port()).await;
+
+    let server_a = QuicServer::bind(loopback(0), id_a).unwrap();
+    let addr = server_a.local_addr().unwrap();
+    let ra = router_a.clone();
+    let acc = tokio::spawn(async move {
+        while let Some(Ok(conn)) = server_a.accept().await {
+            let r = ra.clone();
+            tokio::spawn(async move {
+                let _ = serve_connection(conn, r).await;
+            });
+        }
+    });
+    let client = QuicClient::new(loopback(0), id_b).unwrap();
+    let conn = client.connect(addr, "a").await.unwrap();
+
+    let key = "clawverse:main:pr-42";
+    let stamped = StampedTagValue {
+        value: [0xAB; 32],
+        clock: 42,
+        node: node_stamp_for("runner-x"),
+    };
+    assert!(call_put_tag_versioned(&conn, key, &stamped).await.unwrap());
+
+    let expires_at = 2_000_000_000u64;
+    call_set_tag_expiry(&conn, key, expires_at).await.unwrap();
+
+    // Both surfaces round-trip.
+    assert_eq!(
+        call_get_tag_versioned(&conn, key).await.unwrap(),
+        Some(stamped)
+    );
+    assert_eq!(
+        call_get_tag_expiry(&conn, key).await.unwrap(),
+        Some(expires_at)
+    );
+
+    conn.close(quinn::VarInt::from_u32(0), b"done");
+    client.shutdown().await;
+    tokio::time::sleep(Duration::from_millis(50)).await;
+    acc.abort();
+}