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

//! Client-side RPC helpers split out of rpc.rs to keep the parent
//! module under the 1300-line ceiling. Every function here calls
//! [`super::rpc_call`] or opens a bidi stream directly and follows
//! the wire format documented on [`super::Method`].
//!
//! Re-exported through `pub use client::*;` in the parent so external
//! callers keep the `cluster::rpc::call_*` paths they've been using.

use super::*;

// ── Streaming client helpers ─────────────────────────────────────────

/// Upload a blob by streaming from `reader` into the peer's store.
/// Never buffers the whole blob — memory usage is dominated by
/// tokio's default copy buffer. Returns the peer-assigned BlobId,
/// which is NOT locally re-verified (the peer doesn't get the whole
/// content in memory at once so we can't cheaply hash from our side);
/// callers who need cryptographic proof of end-to-end integrity
/// should follow up with `call_blob_stat` and compare hashes.
pub async fn call_blob_put_stream<R>(
    conn: &Connection,
    mut reader: R,
) -> Result<BlobId>
where
    R: AsyncRead + Unpin,
{
    let (mut send, mut recv) = conn
        .open_bi()
        .await
        .context("opening bidi stream for BlobPutStream")?;
    send.write_all(&[Method::BlobPutStream.as_byte()])
        .await
        .context("writing method tag")?;
    tokio::io::copy(&mut reader, &mut send)
        .await
        .context("streaming blob content to peer")?;
    send.finish().context("finishing BlobPutStream send")?;
    let mut status = [0u8; 1];
    recv.read_exact(&mut status)
        .await
        .context("reading BlobPutStream status byte")?;
    if let Some(code) = decode_error(status[0]) {
        bail!("peer replied with error: {}", code.describe());
    }
    if status[0] != STREAM_STATUS_OK {
        bail!(
            "peer replied with unknown status byte 0x{:02x}",
            status[0]
        );
    }
    let mut id_bytes = [0u8; 32];
    recv.read_exact(&mut id_bytes)
        .await
        .context("reading BlobId from BlobPutStream reply")?;
    Ok(BlobId::from_bytes(id_bytes))
}

/// Fetch a blob by streaming its bytes into `writer`. Returns
/// `Ok(false)` when the peer replies [`ErrorCode::NotFound`]; other
/// error codes surface as `Err`. `writer` receives exactly the blob
/// content — no leading status byte, no framing.
pub async fn call_blob_get_stream<W>(
    conn: &Connection,
    id: &BlobId,
    writer: &mut W,
) -> Result<bool>
where
    W: AsyncWrite + Unpin,
{
    let (mut send, mut recv) = conn
        .open_bi()
        .await
        .context("opening bidi stream for BlobGetStream")?;
    let mut req = Vec::with_capacity(33);
    req.push(Method::BlobGetStream.as_byte());
    req.extend_from_slice(id.as_bytes());
    send.write_all(&req)
        .await
        .context("writing BlobGetStream request")?;
    send.finish().context("finishing BlobGetStream send")?;

    let mut status = [0u8; 1];
    recv.read_exact(&mut status)
        .await
        .context("reading BlobGetStream status byte")?;
    if let Some(code) = decode_error(status[0]) {
        if code == ErrorCode::NotFound {
            return Ok(false);
        }
        bail!("peer replied with error: {}", code.describe());
    }
    if status[0] != STREAM_STATUS_OK {
        bail!(
            "peer replied with unknown status byte 0x{:02x}",
            status[0]
        );
    }
    tokio::io::copy(&mut recv, writer)
        .await
        .context("streaming blob content from peer")?;
    writer
        .flush()
        .await
        .context("flushing destination after BlobGetStream")?;
    Ok(true)
}

// `dispatch` is a server-side helper — it lives back in rpc.rs alongside
// `serve_connection`. Tests reach it via `super::dispatch`.

/// Client-side: open a bidi stream, write `method || payload`, read
/// reply. Returns the raw reply bytes; callers deserialise per method.
pub async fn rpc_call(
    conn: &Connection,
    method: Method,
    payload: &[u8],
) -> Result<Vec<u8>> {
    if payload.len() + 1 > MAX_MESSAGE_BYTES {
        bail!(
            "RPC payload {} bytes (+1 for method tag) exceeds cap {}",
            payload.len(),
            MAX_MESSAGE_BYTES
        );
    }
    let (mut send, mut recv) = conn
        .open_bi()
        .await
        .context("opening bidi stream for RPC")?;
    let mut buf = Vec::with_capacity(1 + payload.len());
    buf.push(method.as_byte());
    buf.extend_from_slice(payload);
    send.write_all(&buf).await.context("writing RPC request")?;
    send.finish().context("finishing RPC send stream")?;
    let reply = recv
        .read_to_end(MAX_MESSAGE_BYTES)
        .await
        .context("reading RPC reply")?;
    Ok(reply)
}

/// Convenience wrapper for [`Method::Ping`]. Sends `payload`, returns
/// the peer's echo (`"pong:" || payload`) with the prefix stripped.
pub async fn call_ping(conn: &Connection, payload: &[u8]) -> Result<Vec<u8>> {
    let reply = rpc_call(conn, Method::Ping, payload).await?;
    if reply.len() == 1 {
        if let Some(code) = decode_error(reply[0]) {
            bail!("peer replied with error: {}", code.describe());
        }
    }
    if let Some(rest) = reply.strip_prefix(b"pong:") {
        Ok(rest.to_vec())
    } else {
        bail!(
            "peer replied with unexpected shape: {} bytes, no 'pong:' prefix",
            reply.len()
        );
    }
}

/// Convenience wrapper for [`Method::PeerStatus`]. Sends an empty
/// payload, deserialises the JSON response.
pub async fn call_peer_status(conn: &Connection) -> Result<PeerStatusReply> {
    let reply = rpc_call(conn, Method::PeerStatus, &[]).await?;
    if reply.len() == 1 {
        if let Some(code) = decode_error(reply[0]) {
            bail!("peer replied with error: {}", code.describe());
        }
    }
    serde_json::from_slice(&reply).context("decoding PeerStatusReply JSON")
}

/// Convenience wrapper for [`Method::DashboardStatus`]. Feeds the
/// dashboard-v2 aggregator. Empty payload → JSON reply with counts
/// + on-disk bytes.
pub async fn call_dashboard_status(conn: &Connection) -> Result<DashboardStatusReply> {
    let reply = rpc_call(conn, Method::DashboardStatus, &[]).await?;
    if reply.len() == 1 {
        if let Some(code) = decode_error(reply[0]) {
            bail!("peer replied with error: {}", code.describe());
        }
    }
    serde_json::from_slice(&reply).context("decoding DashboardStatusReply JSON")
}

pub async fn call_dashboard_storage(conn: &Connection) -> Result<DashboardStorageReply> {
    let reply = rpc_call(conn, Method::DashboardStorage, &[]).await?;
    if reply.len() == 1 {
        if let Some(code) = decode_error(reply[0]) {
            bail!("peer replied with error: {}", code.describe());
        }
    }
    serde_json::from_slice(&reply).context("decoding DashboardStorageReply JSON")
}

/// Phase 9 R1b: convenience wrapper for [`Method::RepoEnsure`].
/// Materializes `(url, git_ref)` on the connected peer under the
/// caller-provided workspace namespace and returns the resulting
/// on-disk path + head sha. Cached reply = the checkout was already
/// present with a valid `.git`.
pub async fn call_repo_ensure(
    conn: &Connection,
    req: &crate::cluster::repo_ensure::RepoEnsureRequest,
) -> Result<crate::cluster::repo_ensure::RepoEnsureReply> {
    let payload = serde_json::to_vec(req).context("encoding RepoEnsureRequest")?;
    let reply = rpc_call(conn, Method::RepoEnsure, &payload).await?;
    if reply.len() == 1 {
        if let Some(code) = decode_error(reply[0]) {
            bail!("peer replied with error: {}", code.describe());
        }
    }
    serde_json::from_slice(&reply).context("decoding RepoEnsureReply JSON")
}

/// Phase 9 R1b: convenience wrapper for [`Method::RepoRelease`].
/// Removes the on-disk checkout for `(url, git_ref)` under the
/// caller's workspace. `removed=false` when nothing was on disk to
/// begin with (still `Ok`).
pub async fn call_repo_release(
    conn: &Connection,
    req: &crate::cluster::repo_ensure::RepoReleaseRequest,
) -> Result<crate::cluster::repo_ensure::RepoReleaseReply> {
    let payload = serde_json::to_vec(req).context("encoding RepoReleaseRequest")?;
    let reply = rpc_call(conn, Method::RepoRelease, &payload).await?;
    if reply.len() == 1 {
        if let Some(code) = decode_error(reply[0]) {
            bail!("peer replied with error: {}", code.describe());
        }
    }
    serde_json::from_slice(&reply).context("decoding RepoReleaseReply JSON")
}

/// Convenience wrapper for [`Method::ShutdownPrepCheck`]. Runs
/// `safe-shutdown-prep.sh --dry-run` on the connected peer and waits
/// for the full report. Never stops anything on the peer.
pub async fn call_shutdown_prep_check(
    conn: &Connection,
) -> Result<crate::cluster::shutdown_prep::ShutdownPrepCheckReply> {
    let req = crate::cluster::shutdown_prep::ShutdownPrepCheckRequest {};
    let payload = serde_json::to_vec(&req).context("encoding ShutdownPrepCheckRequest")?;
    let reply = rpc_call(conn, Method::ShutdownPrepCheck, &payload).await?;
    if reply.len() == 1 {
        if let Some(code) = decode_error(reply[0]) {
            bail!("peer replied with error: {}", code.describe());
        }
    }
    serde_json::from_slice(&reply).context("decoding ShutdownPrepCheckReply JSON")
}

/// Convenience wrapper for [`Method::ShutdownPrepExecute`]. Starts the
/// real shutdown-prep run on the connected peer (detached — this call
/// returns as soon as the peer confirms it started, not when it
/// finishes, since the peer's own daemon stops itself partway
/// through).
pub async fn call_shutdown_prep_execute(
    conn: &Connection,
    confirm_node_name: &str,
) -> Result<crate::cluster::shutdown_prep::ShutdownPrepExecuteReply> {
    let req = crate::cluster::shutdown_prep::ShutdownPrepExecuteRequest {
        confirm_node_name: confirm_node_name.to_string(),
    };
    let payload = serde_json::to_vec(&req).context("encoding ShutdownPrepExecuteRequest")?;
    let reply = rpc_call(conn, Method::ShutdownPrepExecute, &payload).await?;
    if reply.len() == 1 {
        if let Some(code) = decode_error(reply[0]) {
            bail!("peer replied with error: {}", code.describe());
        }
    }
    serde_json::from_slice(&reply).context("decoding ShutdownPrepExecuteReply JSON")
}

/// Recognise a single-byte reply as one of our error codes. Returns
/// `None` for any other single-byte value (which is a valid reply,
/// just an unusually short one).
pub fn decode_error(b: u8) -> Option<ErrorCode> {
    match b {
        0xf0 => Some(ErrorCode::EmptyRequest),
        0xf1 => Some(ErrorCode::UnknownMethod),
        0xf2 => Some(ErrorCode::HandlerFailure),
        0xf3 => Some(ErrorCode::NotFound),
        0xf4 => Some(ErrorCode::InvalidRequest),
        0xf5 => Some(ErrorCode::NotConfigured),
        0xf6 => Some(ErrorCode::AlreadyExists),
        _ => None,
    }
}

// ── Blob RPC client helpers ───────────────────────────────────────────

/// Ask the peer for a blob's size + chunk count. `Ok(None)` when the
/// peer replies [`ErrorCode::NotFound`]; other error codes surface as
/// `Err`.
pub async fn call_blob_stat(
    conn: &Connection,
    id: &BlobId,
) -> Result<Option<BlobStat>> {
    let reply = rpc_call(conn, Method::BlobStat, id.as_bytes()).await?;
    if reply.len() == 1 {
        match decode_error(reply[0]) {
            Some(ErrorCode::NotFound) => return Ok(None),
            Some(code) => bail!("peer replied with error: {}", code.describe()),
            None => {} // single-byte JSON like "1" is technically possible; fall through
        }
    }
    let stat = serde_json::from_slice(&reply).context("decoding BlobStat JSON")?;
    Ok(Some(stat))
}

/// Fetch a blob's raw bytes. `Ok(None)` when the peer replies
/// [`ErrorCode::NotFound`]. Callers that need many-GB transfers should
/// use the streaming variant (Phase 2c); this helper caps at
/// [`MAX_MESSAGE_BYTES`].
pub async fn call_blob_get(
    conn: &Connection,
    id: &BlobId,
) -> Result<Option<Vec<u8>>> {
    let reply = rpc_call(conn, Method::BlobGet, id.as_bytes()).await?;
    if reply.len() == 1 {
        match decode_error(reply[0]) {
            Some(ErrorCode::NotFound) => return Ok(None),
            Some(code) => bail!("peer replied with error: {}", code.describe()),
            None => {} // single-byte blob is legitimate content
        }
    }
    Ok(Some(reply))
}

/// Upload a blob to the peer's store. Returns the BlobId the peer
/// assigned; must match the local hash of `bytes` (proves the peer
/// stored what we sent).
pub async fn call_blob_put(conn: &Connection, bytes: &[u8]) -> Result<BlobId> {
    let reply = rpc_call(conn, Method::BlobPut, bytes).await?;
    if reply.len() == 1 {
        if let Some(code) = decode_error(reply[0]) {
            bail!("peer replied with error: {}", code.describe());
        }
    }
    if reply.len() != 32 {
        bail!(
            "expected 32-byte BlobId in reply, got {} bytes",
            reply.len()
        );
    }
    let mut buf = [0u8; 32];
    buf.copy_from_slice(&reply);
    let assigned = BlobId::from_bytes(buf);
    let expected = BlobId::from_bytes(blake3::hash(bytes).into());
    if assigned != expected {
        bail!(
            "peer returned BlobId {} but content hashes to {}; corruption or protocol drift",
            assigned.to_hex(),
            expected.to_hex()
        );
    }
    Ok(assigned)
}

/// Fetch a blob's manifest (chunk-list + size). `Ok(None)` on NotFound.
pub async fn call_blob_load_manifest(
    conn: &Connection,
    id: &BlobId,
) -> Result<Option<BlobManifest>> {
    let reply = rpc_call(conn, Method::BlobLoadManifest, id.as_bytes()).await?;
    if reply.len() == 1 {
        match decode_error(reply[0]) {
            Some(ErrorCode::NotFound) => return Ok(None),
            Some(code) => bail!("peer replied with error: {}", code.describe()),
            None => {}
        }
    }
    let manifest =
        serde_json::from_slice(&reply).context("decoding BlobManifest JSON")?;
    Ok(Some(manifest))
}

// ── Phase 2d: chunk-level client helpers ─────────────────────────────

/// Does the peer already have this chunk? `Ok(true)` on presence,
/// `Ok(false)` on NotFound. Other errors surface as `Err`.
pub async fn call_has_chunk(
    conn: &Connection,
    hash: &ChunkHash,
) -> Result<bool> {
    let reply = rpc_call(conn, Method::HasChunk, hash.as_bytes()).await?;
    if reply.len() != 1 {
        bail!(
            "expected single-byte HasChunk reply, got {} bytes",
            reply.len()
        );
    }
    match reply[0] {
        STREAM_STATUS_OK => Ok(true),
        code => match decode_error(code) {
            Some(ErrorCode::NotFound) => Ok(false),
            Some(err) => bail!("peer replied with error: {}", err.describe()),
            None => bail!(
                "peer replied with unknown byte 0x{:02x} for HasChunk",
                code
            ),
        },
    }
}

/// Upload a single chunk to the peer's store. Peer verifies bytes
/// hash to `hash` before writing; a mismatch surfaces as `Err`.
pub async fn call_put_chunk(
    conn: &Connection,
    hash: &ChunkHash,
    bytes: &[u8],
) -> Result<()> {
    let mut payload = Vec::with_capacity(32 + bytes.len());
    payload.extend_from_slice(hash.as_bytes());
    payload.extend_from_slice(bytes);
    let reply = rpc_call(conn, Method::PutChunk, &payload).await?;
    if reply.len() != 1 {
        bail!(
            "expected single-byte PutChunk reply, got {} bytes",
            reply.len()
        );
    }
    match reply[0] {
        STREAM_STATUS_OK => Ok(()),
        code => match decode_error(code) {
            Some(err) => bail!("peer rejected chunk: {}", err.describe()),
            None => bail!(
                "peer replied with unknown byte 0x{:02x} for PutChunk",
                code
            ),
        },
    }
}

/// Fetch a single chunk from the peer's store. `Ok(None)` on NotFound.
/// Verifies the returned bytes hash to `hash` locally — corruption or
/// protocol drift surfaces as `Err`.
pub async fn call_get_chunk(
    conn: &Connection,
    hash: &ChunkHash,
) -> Result<Option<Vec<u8>>> {
    let reply = rpc_call(conn, Method::GetChunk, hash.as_bytes()).await?;
    if reply.is_empty() {
        bail!("empty GetChunk reply");
    }
    match reply[0] {
        STREAM_STATUS_OK => {
            let bytes = reply[1..].to_vec();
            let recomputed = ChunkHash::from_bytes(blake3::hash(&bytes).into());
            if recomputed != *hash {
                bail!(
                    "GetChunk hash mismatch: requested {}, got bytes hashing to {}",
                    hash.to_hex(),
                    recomputed.to_hex()
                );
            }
            Ok(Some(bytes))
        }
        code => match decode_error(code) {
            Some(ErrorCode::NotFound) => Ok(None),
            Some(err) => bail!("peer replied with error: {}", err.describe()),
            None => {
                if reply.len() == 1 {
                    bail!(
                        "peer replied with unknown byte 0x{:02x} for GetChunk",
                        code
                    );
                }
                // A single legitimate content byte with value 0xf3 is
                // technically distinguishable from NotFound because
                // GetChunk always prefixes with STREAM_STATUS_OK. The
                // unreachable branch stays as belt-and-braces.
                let bytes = reply[1..].to_vec();
                let recomputed = ChunkHash::from_bytes(blake3::hash(&bytes).into());
                if recomputed != *hash {
                    bail!("GetChunk hash mismatch (fallback path)");
                }
                Ok(Some(bytes))
            }
        },
    }
}

/// Commit a manifest whose chunks the peer should already have.
/// Returns the list of chunk hashes the peer is still missing — an
/// empty list means the manifest was persisted; a non-empty list
/// tells the caller which chunks to upload before retrying.
pub async fn call_put_manifest(
    conn: &Connection,
    manifest: &BlobManifest,
) -> Result<Vec<ChunkHash>> {
    let payload =
        serde_json::to_vec(manifest).context("encoding BlobManifest as JSON")?;
    let reply = rpc_call(conn, Method::PutManifest, &payload).await?;
    if reply.len() == 1 {
        if let Some(err) = decode_error(reply[0]) {
            bail!("peer replied with error: {}", err.describe());
        }
    }
    let decoded: PutManifestReply = serde_json::from_slice(&reply)
        .context("decoding PutManifestReply JSON")?;
    if decoded.blob_id != manifest.blob_id {
        bail!(
            "peer echoed blob_id {} but we sent {}",
            decoded.blob_id.to_hex(),
            manifest.blob_id.to_hex()
        );
    }
    Ok(decoded.missing)
}

// ── Phase 5b: reference-store client helpers ─────────────────────────

/// Look up a 32-byte value by 32-byte key. `Ok(None)` on `NotFound`.
pub async fn call_get_ref(
    conn: &Connection,
    key: &RefKey,
) -> Result<Option<RefValue>> {
    call_get_ref_inner(conn, key, Method::GetRef).await
}

/// Ref-forwarding (2026-07-13): strict local-only variant. The peer
/// MUST NOT recurse to its own peers; used by daemons doing ref
/// forwarding to prevent loops.
pub async fn call_get_ref_local(
    conn: &Connection,
    key: &RefKey,
) -> Result<Option<RefValue>> {
    call_get_ref_inner(conn, key, Method::GetRefLocal).await
}

async fn call_get_ref_inner(
    conn: &Connection,
    key: &RefKey,
    method: Method,
) -> Result<Option<RefValue>> {
    let reply = rpc_call(conn, method, key).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() != 32 {
        bail!("expected 32-byte ref value, got {} bytes", reply.len());
    }
    let mut value = [0u8; 32];
    value.copy_from_slice(&reply);
    Ok(Some(value))
}

/// Set a 32-byte value for a 32-byte key. Overwrites any prior value.
pub async fn call_put_ref(
    conn: &Connection,
    key: &RefKey,
    value: &RefValue,
) -> Result<()> {
    let mut payload = Vec::with_capacity(64);
    payload.extend_from_slice(key);
    payload.extend_from_slice(value);
    let reply = rpc_call(conn, Method::PutRef, &payload).await?;
    if reply.len() != 1 {
        bail!("expected single-byte PutRef reply, got {} bytes", reply.len());
    }
    match reply[0] {
        STREAM_STATUS_OK => Ok(()),
        code => match decode_error(code) {
            Some(err) => bail!("peer rejected PutRef: {}", err.describe()),
            None => bail!(
                "peer replied with unknown byte 0x{:02x} for PutRef",
                code
            ),
        },
    }
}

// ── Phase 3: stamped-ref client helpers ──────────────────────────────

/// Phase 3 (2026-07-13): submit a stamped (CRDT-merge) PutRef.
///
/// Returns `Ok(true)` when the peer merged the write (Merged),
/// `Ok(false)` when the peer rejected it because an equal or
/// higher `(clock, node)` already exists (AlreadyExists). Any
/// other reply is an error.
pub async fn call_put_ref_versioned(
    conn: &Connection,
    key: &crate::cluster::refs::RefKey,
    incoming: &crate::cluster::refs::StampedRef,
) -> Result<bool> {
    let mut payload = Vec::with_capacity(32 + crate::cluster::refs::StampedRef::ENCODED_LEN);
    payload.extend_from_slice(key);
    payload.extend_from_slice(&incoming.to_bytes());
    let reply = rpc_call(conn, Method::PutRefVersioned, &payload).await?;
    if reply.len() != 1 {
        bail!(
            "expected single-byte PutRefVersioned reply, got {} bytes",
            reply.len()
        );
    }
    match reply[0] {
        STREAM_STATUS_OK => Ok(true),
        code => match decode_error(code) {
            Some(ErrorCode::AlreadyExists) => Ok(false),
            Some(err) => bail!("peer rejected PutRefVersioned: {}", err.describe()),
            None => bail!(
                "peer replied with unknown byte 0x{:02x} for PutRefVersioned",
                code
            ),
        },
    }
}

/// Phase 3: fetch a stamped (CRDT-merge) ref.
pub async fn call_get_ref_versioned(
    conn: &Connection,
    key: &crate::cluster::refs::RefKey,
) -> Result<Option<crate::cluster::refs::StampedRef>> {
    call_get_ref_versioned_inner(conn, key, Method::GetRefVersioned).await
}

/// Phase 3b (2026-07-13): strict local-only stamped-ref lookup —
/// the peer MUST NOT recurse. Used by daemons doing ref-forwarding
/// so they never loop.
pub async fn call_get_ref_versioned_local(
    conn: &Connection,
    key: &crate::cluster::refs::RefKey,
) -> Result<Option<crate::cluster::refs::StampedRef>> {
    call_get_ref_versioned_inner(conn, key, Method::GetRefVersionedLocal).await
}

async fn call_get_ref_versioned_inner(
    conn: &Connection,
    key: &crate::cluster::refs::RefKey,
    method: Method,
) -> Result<Option<crate::cluster::refs::StampedRef>> {
    let reply = rpc_call(conn, method, key).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 => {}
        }
    }
    Ok(Some(
        crate::cluster::refs::StampedRef::from_bytes(&reply)
            .context("decoding stamped ref reply")?,
    ))
}

// ── Phase 3c: stamped-tag client helpers ─────────────────────────────

/// Phase 3c (2026-07-13): submit a stamped (CRDT-merge) PutTag.
///
/// Returns `Ok(true)` when the peer merged the write, `Ok(false)`
/// when the peer rejected it because an equal-or-newer version
/// already exists (`AlreadyExists`). Any other reply is an error.
pub async fn call_put_tag_versioned(
    conn: &Connection,
    key: &str,
    incoming: &crate::cluster::tags::StampedTagValue,
) -> Result<bool> {
    let payload = crate::cluster::tags::encode_stamped_record(key, incoming);
    let reply = rpc_call(conn, Method::PutTagVersioned, &payload).await?;
    if reply.len() != 1 {
        bail!(
            "expected single-byte PutTagVersioned reply, got {} bytes",
            reply.len()
        );
    }
    match reply[0] {
        STREAM_STATUS_OK => Ok(true),
        code => match decode_error(code) {
            Some(ErrorCode::AlreadyExists) => Ok(false),
            Some(err) => bail!("peer rejected PutTagVersioned: {}", err.describe()),
            None => bail!(
                "peer replied with unknown byte 0x{:02x} for PutTagVersioned",
                code
            ),
        },
    }
}

/// Phase 3c: fetch a stamped tag value.
pub async fn call_get_tag_versioned(
    conn: &Connection,
    key: &str,
) -> Result<Option<crate::cluster::tags::StampedTagValue>> {
    let reply = rpc_call(conn, Method::GetTagVersioned, 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 => {}
        }
    }
    Ok(Some(
        crate::cluster::tags::StampedTagValue::from_bytes(&reply)
            .context("decoding stamped tag reply")?,
    ))
}

// ── 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.
pub async fn call_get_metrics(conn: &Connection) -> Result<MetricsReply> {
    let reply = rpc_call(conn, Method::GetMetrics, &[]).await?;
    if reply.len() == 1 {
        if let Some(err) = decode_error(reply[0]) {
            bail!("peer replied with error: {}", err.describe());
        }
    }
    serde_json::from_slice(&reply).context("decoding MetricsReply JSON")
}

// ── Phase 5d: tag-store client helpers ───────────────────────────────

/// Publish (or overwrite) a named tag pointing at a 32-byte value.
pub async fn call_put_tag(
    conn: &Connection,
    key: &str,
    value: &[u8; 32],
) -> Result<()> {
    let payload = crate::cluster::tags::encode_record(key, value);
    let reply = rpc_call(conn, Method::PutTag, &payload).await?;
    if reply.len() != 1 {
        bail!("expected single-byte PutTag reply, got {} bytes", reply.len());
    }
    match reply[0] {
        STREAM_STATUS_OK => Ok(()),
        code => match decode_error(code) {
            Some(err) => bail!("peer rejected PutTag: {}", err.describe()),
            None => bail!(
                "peer replied with unknown byte 0x{:02x} for PutTag",
                code
            ),
        },
    }
}

/// Look up a named tag. `Ok(None)` on `NotFound`.
pub async fn call_get_tag(
    conn: &Connection,
    key: &str,
) -> Result<Option<[u8; 32]>> {
    if key.is_empty() {
        bail!("tag key cannot be empty");
    }
    let reply = rpc_call(conn, Method::GetTag, 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() != 32 {
        bail!("expected 32-byte tag value, got {} bytes", reply.len());
    }
    let mut value = [0u8; 32];
    value.copy_from_slice(&reply);
    Ok(Some(value))
}

/// Delete a named tag. `Ok(false)` when no such tag existed.
pub async fn call_delete_tag(conn: &Connection, key: &str) -> Result<bool> {
    if key.is_empty() {
        bail!("tag key cannot be empty");
    }
    let reply = rpc_call(conn, Method::DeleteTag, key.as_bytes()).await?;
    if reply.len() != 1 {
        bail!(
            "expected single-byte DeleteTag reply, got {} bytes",
            reply.len()
        );
    }
    match reply[0] {
        STREAM_STATUS_OK => Ok(true),
        code => match decode_error(code) {
            Some(ErrorCode::NotFound) => Ok(false),
            Some(err) => bail!("peer replied with error: {}", err.describe()),
            None => bail!(
                "peer replied with unknown byte 0x{:02x} for DeleteTag",
                code
            ),
        },
    }
}

/// List every tag stored on the peer. Sorted by key.
pub async fn call_list_tags(conn: &Connection) -> Result<Vec<TagEntry>> {
    let reply = rpc_call(conn, Method::ListTags, &[]).await?;
    if reply.len() == 1 {
        if let Some(err) = decode_error(reply[0]) {
            bail!("peer replied with error: {}", err.describe());
        }
    }
    serde_json::from_slice(&reply).context("decoding TagEntry list JSON")
}

/// High-level partial-sync helper: replicate a local blob to a peer
/// by uploading only chunks the peer is missing.
///
/// Flow:
///   1. Load the local manifest.
///   2. For each chunk, ask the peer if it already has it.
///   3. Upload the missing chunks.
///   4. Send the manifest so the peer commits.
///
/// Returns `(uploaded_chunks, total_chunks)` — the difference is what
/// dedup saved. On a warm peer with an identical prior build both
/// numbers are equal to `manifest.chunks.len()` minus the count that
/// were already present, and only the missing-chunk bytes cross the
/// wire.
pub async fn push_blob_missing_chunks(
    conn: &Connection,
    local: &BlobStore,
    id: &BlobId,
) -> Result<(usize, usize)> {
    let manifest = local
        .load_manifest(id)
        .await?
        .with_context(|| format!("blob {} not present locally", id.to_hex()))?;
    let total = manifest.chunks.len();
    let mut uploaded = 0usize;
    for hash in &manifest.chunks {
        if !call_has_chunk(conn, hash).await? {
            let bytes = local
                .read_chunk(hash)
                .await?
                .with_context(|| format!("chunk {} missing locally", hash.to_hex()))?;
            call_put_chunk(conn, hash, &bytes).await?;
            uploaded += 1;
        }
    }
    // Commit the manifest. `missing` MUST be empty by now — if the
    // peer still reports missing chunks after we uploaded them, the
    // most likely cause is a store crash on their side; surface as Err.
    let still_missing = call_put_manifest(conn, &manifest).await?;
    if !still_missing.is_empty() {
        bail!(
            "peer still reports {} missing chunks after we uploaded {} — retry",
            still_missing.len(),
            uploaded
        );
    }
    Ok((uploaded, total))
}

/// Phase 5h: chunk-level cross-peer prewarm with bounded memory.
///
/// Unlike [`push_blob_missing_chunks`] this doesn't need a local
/// `BlobStore` — it forwards chunk data directly from `upstream` to
/// `downstream` one chunk at a time. Memory ceiling is one chunk
/// (4 MiB by default), independent of blob size, so multi-GB blobs
/// don't blow up RAM on the mediator.
///
/// Flow:
///   1. Load the manifest from `upstream`.
///   2. For each chunk, ask `downstream` if it already has it.
///   3. For missing chunks: `GetChunk` upstream → `PutChunk` downstream.
///      Bytes are held only for the duration of the pair of calls.
///   4. Commit the manifest downstream via `PutManifest`. If the peer
///      still reports missing chunks (concurrent eviction, storage
///      failure), we retry each once; a second failure surfaces as
///      `Err`.
///
/// Returns `(uploaded_chunks, total_chunks)` — the difference is the
/// dedup save. When `downstream` is already fully warm the returned
/// `uploaded` is 0 and no chunk bytes crossed the wire.
///
/// Note: this streams chunks sequentially. Fan-out (N concurrent
/// chunk transfers) would speed multi-GB prewarms but complicates
/// error handling; a future revision can layer parallelism on top
/// without changing the memory story per in-flight chunk.
pub async fn prewarm_missing_chunks_between(
    upstream: &Connection,
    downstream: &Connection,
    id: &BlobId,
) -> Result<(usize, usize)> {
    let manifest = call_blob_load_manifest(upstream, id)
        .await?
        .with_context(|| format!("upstream missing blob {}", id.to_hex()))?;
    let total = manifest.chunks.len();
    let mut uploaded = 0usize;
    for hash in &manifest.chunks {
        if call_has_chunk(downstream, hash).await? {
            continue;
        }
        let bytes = call_get_chunk(upstream, hash)
            .await?
            .with_context(|| {
                format!(
                    "upstream manifest referenced chunk {} but GetChunk returned NotFound",
                    hash.to_hex()
                )
            })?;
        call_put_chunk(downstream, hash, &bytes).await?;
        uploaded += 1;
        // `bytes` is dropped here — memory ceiling is one chunk at a
        // time. Reassigned on the next iteration.
    }
    let still_missing = call_put_manifest(downstream, &manifest).await?;
    if !still_missing.is_empty() {
        // One retry pass: fetch + push each chunk the peer still lacks.
        // Guards against a narrow race where the peer evicted a chunk
        // between our HasChunk and its PutManifest verification pass.
        for hash in &still_missing {
            let bytes = call_get_chunk(upstream, hash).await?.with_context(|| {
                format!("retry fetch of chunk {} failed", hash.to_hex())
            })?;
            call_put_chunk(downstream, hash, &bytes).await?;
            uploaded += 1;
        }
        let final_missing = call_put_manifest(downstream, &manifest).await?;
        if !final_missing.is_empty() {
            bail!(
                "downstream still missing {} chunks after retry; storage may be failing",
                final_missing.len()
            );
        }
    }
    Ok((uploaded, total))
}

/// Field finding 2026-07-12 (Pi restore = 18s single-stream): fetch a
/// blob by pulling its chunks in parallel and reassembling in memory.
/// Faster than `call_blob_get_stream` on connections with per-stream
/// throughput ceilings — parallel streams stack their contributions.
///
/// Flow:
///   1. `LoadManifest` upstream (small).
///   2. Spawn N concurrent `GetChunk` tasks bounded by a semaphore.
///   3. Assemble the results in manifest order into a `Vec<u8>` sized
///      to `manifest.total_size`.
///
/// `concurrency <= 1` degrades to sequential (matches
/// `call_blob_get_stream` semantics but keeps the code path uniform).
/// Memory ceiling: `total_size + 4 MiB × in-flight` — dominated by
/// the reassembled blob buffer itself.
pub async fn call_blob_get_parallel(
    conn: &Connection,
    id: &BlobId,
    concurrency: usize,
) -> Result<Option<Vec<u8>>> {
    let manifest = match call_blob_load_manifest(conn, id).await? {
        Some(m) => m,
        None => return Ok(None),
    };
    let concurrency = concurrency.max(1);
    let sem = std::sync::Arc::new(tokio::sync::Semaphore::new(concurrency));
    let mut set = tokio::task::JoinSet::new();
    for (idx, hash) in manifest.chunks.iter().copied().enumerate() {
        let permit = sem
            .clone()
            .acquire_owned()
            .await
            .context("acquiring get_parallel semaphore permit")?;
        let conn = conn.clone();
        set.spawn(async move {
            let _permit = permit;
            let bytes = call_get_chunk(&conn, &hash).await?.with_context(|| {
                format!(
                    "manifest referenced chunk {} but GetChunk returned NotFound",
                    hash.to_hex()
                )
            })?;
            Ok::<(usize, Vec<u8>), anyhow::Error>((idx, bytes))
        });
    }
    // Assemble in manifest order. Pre-size the outer vec so we can
    // slot each chunk's bytes at the right offset without copying.
    let mut out = vec![0u8; manifest.total_size as usize];
    // Chunk boundaries: chunk i starts at i * CHUNK_SIZE.
    let chunk_size = crate::cluster::blob::CHUNK_SIZE;
    let mut first_err: Option<anyhow::Error> = None;
    while let Some(join) = set.join_next().await {
        match join {
            Ok(Ok((idx, bytes))) => {
                let start = idx * chunk_size;
                let end = start + bytes.len();
                if end > out.len() {
                    if first_err.is_none() {
                        first_err = Some(anyhow::anyhow!(
                            "chunk {} at idx {} would overflow reassembly buffer \
                             (end {}, total_size {})",
                            manifest.chunks[idx].to_hex(),
                            idx,
                            end,
                            manifest.total_size
                        ));
                    }
                    continue;
                }
                out[start..end].copy_from_slice(&bytes);
            }
            Ok(Err(e)) => {
                if first_err.is_none() {
                    first_err = Some(e);
                }
            }
            Err(join_err) => {
                if first_err.is_none() {
                    first_err = Some(anyhow::Error::from(join_err));
                }
            }
        }
    }
    if let Some(e) = first_err {
        return Err(e).context("parallel chunk fetch failed");
    }
    Ok(Some(out))
}

/// Phase 5k: parallel-fanout variant of [`prewarm_missing_chunks_between`].
///
/// Runs the has→get→put pipeline for each chunk concurrently, bounded
/// by `concurrency`. Pilot 2026-07-12 measured **109 MiB/s** on the
/// sequential path — ~11% of a 10G link. Parallelism pushes toward
/// the link cap; on the same clawverse workload (249 chunks) we
/// expect a several-× speedup with `concurrency = 8`.
///
/// `concurrency = 0` or `1` degrades to the sequential path.
///
/// Memory: one 4 MiB chunk buffer × in-flight requests. `concurrency
/// = 8` → 32 MiB peak; `concurrency = 32` → 128 MiB.
///
/// `quinn::Connection` is `Clone` (internal `Arc`) so we can share it
/// across the spawned tasks without wrapping in an outer Arc.
pub async fn prewarm_missing_chunks_between_parallel(
    upstream: &Connection,
    downstream: &Connection,
    id: &BlobId,
    concurrency: usize,
) -> Result<(usize, usize)> {
    // Fall through to the sequential path when parallelism disabled.
    if concurrency <= 1 {
        return prewarm_missing_chunks_between(upstream, downstream, id).await;
    }
    let manifest = call_blob_load_manifest(upstream, id)
        .await?
        .with_context(|| format!("upstream missing blob {}", id.to_hex()))?;
    let total = manifest.chunks.len();

    let sem = std::sync::Arc::new(tokio::sync::Semaphore::new(concurrency));
    let mut set = tokio::task::JoinSet::new();
    for hash in manifest.chunks.iter().copied() {
        let permit = sem
            .clone()
            .acquire_owned()
            .await
            .context("acquiring prewarm semaphore permit")?;
        let up = upstream.clone();
        let down = downstream.clone();
        set.spawn(async move {
            // Permit held for the whole pipeline — released on drop.
            let _permit = permit;
            if call_has_chunk(&down, &hash).await? {
                return Ok::<bool, anyhow::Error>(false);
            }
            let bytes = call_get_chunk(&up, &hash).await?.with_context(|| {
                format!(
                    "upstream manifest referenced chunk {} but GetChunk returned NotFound",
                    hash.to_hex()
                )
            })?;
            call_put_chunk(&down, &hash, &bytes).await?;
            Ok(true)
        });
    }

    let mut uploaded = 0usize;
    let mut first_err: Option<anyhow::Error> = None;
    while let Some(join) = set.join_next().await {
        match join {
            Ok(Ok(pushed)) => {
                if pushed {
                    uploaded += 1;
                }
            }
            Ok(Err(e)) => {
                if first_err.is_none() {
                    first_err = Some(e);
                }
            }
            Err(join_err) => {
                if first_err.is_none() {
                    first_err = Some(anyhow::Error::from(join_err));
                }
            }
        }
    }
    if let Some(e) = first_err {
        return Err(e).context("prewarm chunk task failed");
    }

    // Commit + one retry pass — same shape as the sequential variant.
    let still_missing = call_put_manifest(downstream, &manifest).await?;
    if !still_missing.is_empty() {
        for hash in &still_missing {
            let bytes = call_get_chunk(upstream, hash)
                .await?
                .with_context(|| format!("retry fetch of chunk {} failed", hash.to_hex()))?;
            call_put_chunk(downstream, hash, &bytes).await?;
            uploaded += 1;
        }
        let final_missing = call_put_manifest(downstream, &manifest).await?;
        if !final_missing.is_empty() {
            bail!(
                "downstream still missing {} chunks after retry; storage may be failing",
                final_missing.len()
            );
        }
    }
    Ok((uploaded, total))
}