feat(agent): add SshSyncBackend — push/pull over SSH pipe without pre-running server

SshSyncBackend spawns `clawsync serve <path> --stdio` via SSH for each
push/pull operation, running the full sync protocol over stdin/stdout.
No TCP port or pre-running daemon required on the remote host.

API:
  SshSyncBackend::new(user_host, remote_path, agent_id)
  .with_ssh_command(cmd)   // override ssh binary; defaults to CLAWSYNC_SSH_COMMAND env

SshSyncBackend is re-exported from clawsync_agent crate root alongside
TcpSyncBackend and QuicSyncBackend, completing the transport trifecta.

3 integration tests in clawsync-cli/tests/agent_ssh.rs using fake-SSH shim:
  ssh_backend_push_cold_copy, ssh_backend_pull_cold_copy,
  ssh_backend_incremental_push (verifies delta: only 3/8 revisions sent)

Co-Authored-By: Claude Sonnet 4.6 <[email protected]>

crates/clawsync-agent/src/backend.rs

@@ -17,6 +17,8 @@ use clawsync_onion::selector::SyncSelector;
 use clawsync_transport::protocol::SyncMessage;
 use clawsync_transport::quic::{QuicConfig, quic_connect};
 use clawsync_transport::tcp::TcpConnection;
+use clawsync_transport::{StreamPeer, SyncPeer};
+use tokio::process::Command as TokioCommand;
 use tokio::sync::Semaphore;
 
 /// Maximum number of `LayerPacket`s in-flight before the sender stalls
@@ -526,6 +528,266 @@ impl SyncBackend for QuicSyncBackend {
     }
 }
 
+// ─────────────────────────────────────────────────────────────────────────────
+// SSH implementation
+// ─────────────────────────────────────────────────────────────────────────────
+
+/// `SyncBackend` implementation over an SSH pipe.
+///
+/// Each call to `push` or `pull` spawns:
+/// ```text
+/// <ssh_command> <user_host> clawsync serve <remote_path> --stdio
+/// ```
+/// and runs the sync protocol over the child's stdin/stdout.  No pre-running
+/// `clawsync serve` daemon is required on the remote host — the binary is
+/// started on-demand for every sync.
+///
+/// # SSH command override
+///
+/// By default the system `ssh` binary is used.  Set `CLAWSYNC_SSH_COMMAND` in
+/// the environment (or call [`SshSyncBackend::with_ssh_command`]) to use a
+/// different binary — useful for testing with a fake-SSH shim.
+pub struct SshSyncBackend {
+    /// `user@host` target for the SSH connection.
+    user_host: String,
+    /// Absolute path to the `.onion` / HDF5 sidecar on the remote host.
+    remote_path: String,
+    /// Agent ID used in push/pull protocol messages.
+    agent_id: String,
+    /// SSH binary to invoke (default: `"ssh"`).
+    ssh_command: String,
+}
+
+impl SshSyncBackend {
+    /// Create a new backend.
+    ///
+    /// - `user_host` — SSH target, e.g. `"[email protected]"`
+    /// - `remote_path` — absolute path on the remote, e.g. `"/data/agent.claws"`
+    /// - `agent_id` — identifier sent in protocol messages
+    pub fn new(
+        user_host: impl Into<String>,
+        remote_path: impl Into<String>,
+        agent_id: impl Into<String>,
+    ) -> Self {
+        let ssh_command = std::env::var("CLAWSYNC_SSH_COMMAND").unwrap_or_else(|_| "ssh".into());
+        Self {
+            user_host: user_host.into(),
+            remote_path: remote_path.into(),
+            agent_id: agent_id.into(),
+            ssh_command,
+        }
+    }
+
+    /// Override the SSH binary used to establish connections.
+    pub fn with_ssh_command(mut self, cmd: impl Into<String>) -> Self {
+        self.ssh_command = cmd.into();
+        self
+    }
+
+    /// Spawn the remote `clawsync serve --stdio` process and return a `SyncPeer`
+    /// backed by the child's stdin/stdout.
+    async fn connect(&self) -> Result<SyncPeer, AgentSyncError> {
+        let child = TokioCommand::new(&self.ssh_command)
+            .arg(&self.user_host)
+            .arg("clawsync")
+            .arg("serve")
+            .arg(&self.remote_path)
+            .arg("--stdio")
+            .stdin(std::process::Stdio::piped())
+            .stdout(std::process::Stdio::piped())
+            .stderr(std::process::Stdio::inherit())
+            .spawn()
+            .map_err(|e| AgentSyncError::Io(e))?;
+        Ok(SyncPeer::Stream(StreamPeer::from_child(child)))
+    }
+}
+
+impl SyncBackend for SshSyncBackend {
+    async fn push(
+        &self,
+        local_path: &Path,
+        selector: &SyncSelector,
+    ) -> Result<SyncStats, AgentSyncError> {
+        let local_onion = OnionFile::open(local_path).map_err(AgentSyncError::Onion)?;
+
+        let summaries = local_onion.list_revisions();
+        let local_rev_numbers: Vec<u64> = summaries.iter().map(|s| s.revision).collect();
+
+        let mut peer = self.connect().await?;
+
+        // ── IBLT pre-flight ───────────────────────────────────────────────
+        let sketch = IbltSketch::from_keys(&local_rev_numbers, IBLT_SYNC_SEED);
+        let iblt_manifest = IbltManifest {
+            agent_id: self.agent_id.clone(),
+            file_blake3: [0u8; 32],
+            revision_count: local_rev_numbers.len() as u64,
+            head_revision: local_rev_numbers.last().copied().unwrap_or(0),
+            head_blake3: [0u8; 32],
+            last_write: 0.0,
+            sketch_cells: sketch.cell_count() as u32,
+            sketch: sketch.to_bytes(),
+        };
+        peer.send(&SyncMessage::IbltRequest {
+            sketch: iblt_manifest,
+        })
+        .await
+        .map_err(AgentSyncError::Transport)?;
+
+        let missing_from_remote = match peer.recv().await.map_err(AgentSyncError::Transport)? {
+            SyncMessage::IbltResponse {
+                missing_from_remote,
+                ..
+            } => missing_from_remote,
+            SyncMessage::Error { message } => return Err(AgentSyncError::Remote(message)),
+            other => return Err(AgentSyncError::Protocol(format!("unexpected: {other:?}"))),
+        };
+
+        let all_packets = packets_for_revisions(&local_onion, &missing_from_remote)
+            .map_err(AgentSyncError::SyncOnion)?;
+        let packets = clawsync_onion::selector::filter_packets(all_packets, selector, &local_onion)
+            .map_err(AgentSyncError::SyncOnion)?;
+
+        // ── W=16 pipelined send ───────────────────────────────────────────
+        let sz_map: HashMap<u64, u64> = packets
+            .iter()
+            .map(|p| (p.revision, p.page_data_size() as u64))
+            .collect();
+        let total = packets.len() as u64;
+
+        let (mut read_half, mut write_half) = peer.into_pipe_halves();
+        let sem = Arc::new(Semaphore::new(PIPELINE_WINDOW));
+        let sem_w = sem.clone();
+
+        let write_task = tokio::spawn(async move {
+            for packet in packets {
+                sem_w
+                    .acquire()
+                    .await
+                    .map_err(|_| AgentSyncError::Protocol("semaphore closed".into()))?
+                    .forget();
+                write_half
+                    .send(&SyncMessage::LayerPacket { packet })
+                    .await
+                    .map_err(AgentSyncError::Transport)?;
+            }
+            Ok::<_, AgentSyncError>(write_half)
+        });
+
+        let mut bytes_sent = 0u64;
+        for _ in 0..total {
+            match read_half.recv().await.map_err(AgentSyncError::Transport)? {
+                SyncMessage::Ack { revision } => {
+                    bytes_sent += sz_map.get(&revision).copied().unwrap_or(0);
+                    sem.add_permits(1);
+                }
+                SyncMessage::Error { message } => {
+                    write_task.abort();
+                    return Err(AgentSyncError::Remote(message));
+                }
+                other => {
+                    write_task.abort();
+                    return Err(AgentSyncError::Protocol(format!("unexpected: {other:?}")));
+                }
+            }
+        }
+
+        let mut write_half = write_task
+            .await
+            .map_err(|e| AgentSyncError::Protocol(e.to_string()))??;
+
+        write_half
+            .send(&SyncMessage::SyncComplete {
+                revisions_transferred: total,
+                bytes_transferred: bytes_sent,
+            })
+            .await
+            .map_err(AgentSyncError::Transport)?;
+        write_half
+            .shutdown()
+            .await
+            .map_err(AgentSyncError::Transport)?;
+
+        Ok(SyncStats {
+            revisions_transferred: total,
+            bytes_transferred: bytes_sent,
+            revisions_skipped: 0,
+        })
+    }
+
+    async fn pull(&self, local_path: &Path) -> Result<SyncStats, AgentSyncError> {
+        let local_rev_count = OnionFile::open(local_path)
+            .map(|o| o.revision_count())
+            .unwrap_or(0);
+
+        let mut peer = self.connect().await?;
+
+        peer.send(&SyncMessage::ManifestRequest {
+            agent_id: self.agent_id.clone(),
+            head_revision: local_rev_count.saturating_sub(1),
+            revision_count: local_rev_count,
+        })
+        .await
+        .map_err(AgentSyncError::Transport)?;
+
+        let server_rev_count = match peer.recv().await.map_err(AgentSyncError::Transport)? {
+            SyncMessage::ManifestResponse { manifest } => manifest.revision_count,
+            SyncMessage::Error { message } => return Err(AgentSyncError::Remote(message)),
+            other => return Err(AgentSyncError::Protocol(format!("unexpected: {other:?}"))),
+        };
+
+        if server_rev_count <= local_rev_count {
+            peer.send(&SyncMessage::SyncComplete {
+                revisions_transferred: 0,
+                bytes_transferred: 0,
+            })
+            .await
+            .map_err(AgentSyncError::Transport)?;
+            return Ok(SyncStats::default());
+        }
+
+        let mut local_onion = if local_path.exists() {
+            OnionFile::open(local_path)
+                .or_else(|_| OnionFile::create_auto(local_path))
+                .map_err(AgentSyncError::Onion)?
+        } else {
+            std::fs::write(local_path, b"\x89HDF\r\n\x1a\n").map_err(AgentSyncError::Io)?;
+            OnionFile::create_auto(local_path).map_err(AgentSyncError::Onion)?
+        };
+
+        let mut packets = Vec::new();
+        loop {
+            match peer.recv().await.map_err(AgentSyncError::Transport)? {
+                SyncMessage::LayerPacket { packet } => {
+                    let rev = packet.revision;
+                    peer.send(&SyncMessage::Ack { revision: rev })
+                        .await
+                        .map_err(AgentSyncError::Transport)?;
+                    packets.push(packet);
+                }
+                SyncMessage::SyncComplete {
+                    revisions_transferred,
+                    bytes_transferred,
+                } => {
+                    let stats = merge_packets(&mut local_onion, packets, true)
+                        .map_err(AgentSyncError::SyncOnion)?;
+                    return Ok(SyncStats {
+                        revisions_transferred,
+                        bytes_transferred,
+                        revisions_skipped: stats.revisions_skipped,
+                    });
+                }
+                SyncMessage::Error { message } => return Err(AgentSyncError::Remote(message)),
+                other => return Err(AgentSyncError::Protocol(format!("unexpected: {other:?}"))),
+            }
+        }
+    }
+
+    fn local_manifest(&self, local_path: &Path) -> Result<ClawSyncManifest, AgentSyncError> {
+        let onion = OnionFile::open(local_path).map_err(AgentSyncError::Onion)?;
+        Ok(ClawSyncManifest::from_onion(&self.agent_id, &onion, &[]))
+    }
+}
+
 // ─────────────────────────────────────────────────────────────────────────────
 // Tests
 // ─────────────────────────────────────────────────────────────────────────────

crates/clawsync-agent/src/lib.rs

@@ -32,7 +32,9 @@ pub mod negotiator;
 pub mod onion_memory;
 pub mod scheduler;
 
-pub use backend::{PeerCapabilities, QuicSyncBackend, SyncBackend, SyncStats, TcpSyncBackend};
+pub use backend::{
+    PeerCapabilities, QuicSyncBackend, SshSyncBackend, SyncBackend, SyncStats, TcpSyncBackend,
+};
 pub use error::AgentSyncError;
 pub use negotiator::{SessionCapabilities, assert_compatible, negotiate};
 pub use onion_memory::{MemorySnapshot, OnionMemory};

crates/clawsync-cli/Cargo.toml

@@ -32,6 +32,7 @@ notify    = { workspace = true }
 [dev-dependencies]
 clawhdf5           = { workspace = true }
 clawhdf5-onion     = { workspace = true }
+clawsync-agent     = { workspace = true }
 clawsync-onion     = { workspace = true }
 clawsync-transport = { workspace = true }
 tempfile           = "3"

crates/clawsync-cli/tests/agent_ssh.rs

@@ -0,0 +1,172 @@
+//! Integration tests for `SshSyncBackend` in `clawsync-agent`.
+//!
+//! Uses a fake-SSH shell script that strips the `user@host` and `clawsync`
+//! arguments then exec's the real `clawsync` binary with the remaining args,
+//! so the full SSH dispatch path is exercised without a real SSH daemon.
+
+use std::path::{Path, PathBuf};
+
+use clawhdf5_onion::writer::OnionFile;
+use clawsync_agent::{SshSyncBackend, SyncBackend};
+use clawsync_onion::selector::SyncSelector;
+use tempfile::TempDir;
+
+const BIN: &str = env!("CARGO_BIN_EXE_clawsync");
+const H5_MAGIC: &[u8] = b"\x89HDF\r\n\x1a\n";
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Helpers
+// ─────────────────────────────────────────────────────────────────────────────
+
+fn make_onion(n: u8) -> (TempDir, PathBuf, OnionFile) {
+    let dir = TempDir::new().unwrap();
+    let h5 = dir.path().join("data.h5");
+    std::fs::write(&h5, H5_MAGIC).unwrap();
+    let mut onion = OnionFile::create(&h5, 4096).unwrap();
+    for i in 0..n {
+        let mut s = onion.begin_session(None).unwrap();
+        s.record_page(0, &vec![i; 4096]);
+        onion.commit_session(s, Some(&format!("rev {i}"))).unwrap();
+    }
+    onion.flush().unwrap();
+    (dir, h5, onion)
+}
+
+/// Fake-SSH script: drops `user@host` + `clawsync`, then exec's BIN with the
+/// remaining args (`serve <path> --stdio`).
+#[cfg(unix)]
+fn write_fake_ssh(path: &Path) {
+    use std::os::unix::fs::PermissionsExt;
+    let script = format!("#!/bin/sh\nshift 2\nexec {BIN} \"$@\"\n");
+    std::fs::write(path, script.as_bytes()).unwrap();
+    let mut perms = std::fs::metadata(path).unwrap().permissions();
+    perms.set_mode(0o755);
+    std::fs::set_permissions(path, perms).unwrap();
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Tests
+// ─────────────────────────────────────────────────────────────────────────────
+
+/// Push 5 revisions to an empty remote via fake-SSH pipe.
+#[tokio::test]
+#[cfg(unix)]
+async fn ssh_backend_push_cold_copy() {
+    let tmp = TempDir::new().unwrap();
+    let fake_ssh = tmp.path().join("fake-ssh");
+    write_fake_ssh(&fake_ssh);
+
+    let (_src_dir, src_h5, _) = make_onion(5);
+
+    let dst_dir = TempDir::new().unwrap();
+    let dst_h5 = dst_dir.path().join("remote.h5");
+    std::fs::write(&dst_h5, H5_MAGIC).unwrap();
+
+    let user = std::env::var("USER").unwrap_or_else(|_| "testuser".to_string());
+    let backend = SshSyncBackend::new(
+        format!("{user}@localhost"),
+        dst_h5.to_str().unwrap(),
+        "ssh-test-agent",
+    )
+    .with_ssh_command(fake_ssh.to_str().unwrap());
+
+    let stats = backend
+        .push(&src_h5, &SyncSelector::All)
+        .await
+        .expect("SSH push failed");
+
+    assert_eq!(stats.revisions_transferred, 5);
+
+    // Give the server process a moment to flush.
+    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+
+    let dst_onion = OnionFile::open(&dst_h5).expect("cannot open dst onion");
+    assert_eq!(dst_onion.revision_count() as u64, 5);
+}
+
+/// Pull 5 revisions from a remote to a new local file via fake-SSH.
+#[tokio::test]
+#[cfg(unix)]
+async fn ssh_backend_pull_cold_copy() {
+    let tmp = TempDir::new().unwrap();
+    let fake_ssh = tmp.path().join("fake-ssh");
+    write_fake_ssh(&fake_ssh);
+
+    let (_src_dir, src_h5, _) = make_onion(5);
+
+    let local_dir = TempDir::new().unwrap();
+    let local_h5 = local_dir.path().join("local.h5");
+
+    let user = std::env::var("USER").unwrap_or_else(|_| "testuser".to_string());
+    let backend = SshSyncBackend::new(
+        format!("{user}@localhost"),
+        src_h5.to_str().unwrap(),
+        "ssh-test-agent",
+    )
+    .with_ssh_command(fake_ssh.to_str().unwrap());
+
+    let stats = backend
+        .pull(&local_h5)
+        .await
+        .expect("SSH pull failed");
+
+    assert_eq!(stats.revisions_transferred, 5);
+
+    let local_onion = OnionFile::open(&local_h5).expect("cannot open local onion");
+    assert_eq!(local_onion.revision_count() as u64, 5);
+}
+
+/// Push 5 revisions, then push 3 more — only the delta transfers.
+#[tokio::test]
+#[cfg(unix)]
+async fn ssh_backend_incremental_push() {
+    let tmp = TempDir::new().unwrap();
+    let fake_ssh = tmp.path().join("fake-ssh");
+    write_fake_ssh(&fake_ssh);
+
+    let (_src_dir, src_h5, mut src_onion) = make_onion(5);
+
+    let dst_dir = TempDir::new().unwrap();
+    let dst_h5 = dst_dir.path().join("remote.h5");
+    std::fs::write(&dst_h5, H5_MAGIC).unwrap();
+
+    let user = std::env::var("USER").unwrap_or_else(|_| "testuser".to_string());
+    let make_backend = || {
+        SshSyncBackend::new(
+            format!("{user}@localhost"),
+            dst_h5.to_str().unwrap(),
+            "ssh-test-agent",
+        )
+        .with_ssh_command(fake_ssh.to_str().unwrap())
+    };
+
+    // Cold push: 5 revisions.
+    let s1 = make_backend()
+        .push(&src_h5, &SyncSelector::All)
+        .await
+        .expect("first push failed");
+    assert_eq!(s1.revisions_transferred, 5);
+    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+
+    // Add 3 more revisions.
+    for i in 5u8..8 {
+        let mut s = src_onion.begin_session(None).unwrap();
+        s.record_page(0, &vec![i; 4096]);
+        src_onion
+            .commit_session(s, Some(&format!("rev {i}")))
+            .unwrap();
+    }
+    src_onion.flush().unwrap();
+
+    // Incremental push: only 3 new revisions transferred.
+    let s2 = make_backend()
+        .push(&src_h5, &SyncSelector::All)
+        .await
+        .expect("second push failed");
+    assert_eq!(s2.revisions_transferred, 3, "expected only 3 new revisions");
+
+    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+
+    let dst_onion = OnionFile::open(&dst_h5).expect("cannot open dst onion");
+    assert_eq!(dst_onion.revision_count() as u64, 8);
+}