verify(P01): 4-layer verification PASS — REQ-073

---ci---
project: orca
phase: P01
milestone: v0.9
status: verify
---/ci---

.ciagent/CHECKPOINT.json

@@ -1 +1 @@
-{ "phase": "P0c", "stage": "verify", "milestone": "v0.9", "phase_role": "execution", "updated_at": "2026-08-05T03:35:00Z", "milestone_complete": false, "verify": { "build": "pass", "go_test": "20/20", "bats": "20/20", "gofmt": "clean", "verify_reqs": "90 consistent" } }
+{ "phase": "P01", "stage": "verify", "milestone": "v0.9", "phase_role": "execution", "updated_at": "2026-08-05T03:45:00Z", "milestone_complete": false, "verify": { "build": "pass", "go_test": "20/20", "bats": "20/20", "gofmt": "clean", "verify_reqs": "90 consistent" } }

internal/sshpush/doc.go

@@ -0,0 +1,23 @@
+// Package sshpush implements the v0.9 SSH-push transport layer (REQ-073,
+// R-001): the CLI on the operator host SSHes to each peer to render files,
+// apply configs, and run commands. It replaces the v0.8
+// internal/transport mTLS HTTP layer.
+//
+// The Transport reuses one *ssh.Client per peer across multiple
+// operations within a single CLI invocation (I-B-001), retries transient
+// failures with exponential backoff (100ms ×2, cap 5s, max 5 attempts —
+// reimplemented from the v0.8 transport/retry.go pattern, since
+// internal/transport is deprecated and not imported), applies per-call
+// timeouts (10s exec, 30s SCP per I-B-001), and fans out to many peers
+// with bounded concurrency (default 8, errgroup + semaphore).
+//
+// Idempotency is content-addressed (C-18): WriteFile / WriteFileIdempotent
+// compare the remote file's SHA-256 to the local content and skip the
+// write on match — the SSH-push equivalent of the v0.8 X-Orca-Idempotency-Key.
+//
+// Host-key verification reuses proxmox.TOFUHostKeyCallback (D-035), which
+// reads/writes the known_hosts file (certpaths.KnownHostsPath during the
+// v0.9 dual-write window; the move to paths.KnownHostsPath happens in
+// v0.10-P14). The known_hosts file is flock-protected inside the TOFU
+// callback, so the Transport does NOT re-lock.
+package sshpush

internal/sshpush/fanout.go

@@ -0,0 +1,113 @@
+package sshpush
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"sync"
+
+	"golang.org/x/sync/errgroup"
+
+	"git.cloudinit.dev/coreci/orca/internal/emitter"
+)
+
+// DefaultFanoutConcurrency is the default bounded-concurrency limit for
+// fan-out operations (I-B-001). The Transport.ExecAll and WriteAll
+// methods use this when the caller does not override it.
+const DefaultFanoutConcurrency = 8
+
+// ExecAll runs cmd on all peers in parallel with bounded concurrency
+// (default 8, I-B-001). Returns per-peer output and per-peer errors. A
+// nil entry in the errors map means that peer succeeded; the output map
+// contains that peer's stdout. The returned error is non-nil only if
+// the fan-out itself failed (e.g., context cancelled before any peer
+// ran); per-peer failures are in the errors map.
+func (t *Transport) ExecAll(ctx context.Context, peers []string, cmd string) (map[string][]byte, map[string]error) {
+	return t.ExecAllWithConcurrency(ctx, peers, cmd, DefaultFanoutConcurrency)
+}
+
+// ExecAllWithConcurrency is ExecAll with an explicit concurrency limit.
+// A limit <= 0 uses DefaultFanoutConcurrency.
+func (t *Transport) ExecAllWithConcurrency(ctx context.Context, peers []string, cmd string, concurrency int) (map[string][]byte, map[string]error) {
+	if concurrency <= 0 {
+		concurrency = DefaultFanoutConcurrency
+	}
+	out := make(map[string][]byte, len(peers))
+	errs := make(map[string]error, len(peers))
+	var mu sync.Mutex
+	g, gctx := errgroup.WithContext(ctx)
+	g.SetLimit(concurrency)
+	for _, p := range peers {
+		peer := p
+		g.Go(func() error {
+			o, err := t.Exec(gctx, peer, cmd)
+			mu.Lock()
+			defer mu.Unlock()
+			if err != nil {
+				errs[peer] = err
+				return nil // per-peer error; do not cancel the group
+			}
+			out[peer] = o
+			return nil
+		})
+	}
+	_ = g.Wait()
+	return out, errs
+}
+
+// WriteAll writes the given files to each peer in parallel with bounded
+// concurrency (default 8, I-B-001). The files map is keyed by peer; each
+// peer's files are written sequentially (to preserve order and avoid
+// intra-peer races on shared paths). Returns per-peer errors; a peer
+// missing from the map or with a nil entry succeeded. The returned
+// error is non-nil only if the fan-out itself failed (context cancelled).
+func (t *Transport) WriteAll(ctx context.Context, peers []string, files map[string][]emitter.File) map[string]error {
+	return t.WriteAllWithConcurrency(ctx, peers, files, DefaultFanoutConcurrency)
+}
+
+// WriteAllWithConcurrency is WriteAll with an explicit concurrency limit.
+// A limit <= 0 uses DefaultFanoutConcurrency.
+func (t *Transport) WriteAllWithConcurrency(ctx context.Context, peers []string, files map[string][]emitter.File, concurrency int) map[string]error {
+	if concurrency <= 0 {
+		concurrency = DefaultFanoutConcurrency
+	}
+	errs := make(map[string]error, len(peers))
+	var mu sync.Mutex
+	g, gctx := errgroup.WithContext(ctx)
+	g.SetLimit(concurrency)
+	for _, p := range peers {
+		peer := p
+		peerFiles := files[peer]
+		g.Go(func() error {
+			for _, f := range peerFiles {
+				mode := parseMode(f.Mode)
+				if _, err := t.WriteFileIdempotent(gctx, peer, f.Path, []byte(f.Content), mode); err != nil {
+					mu.Lock()
+					errs[peer] = fmt.Errorf("sshpush: write %s on %s: %w", f.Path, peer, err)
+					mu.Unlock()
+					return nil // per-peer error; do not cancel the group
+				}
+			}
+			return nil
+		})
+	}
+	_ = g.Wait()
+	return errs
+}
+
+// parseMode parses an octal mode string like "0644" into an os.FileMode.
+// Returns 0644 on parse failure (a safe default for non-executable
+// config files).
+func parseMode(s string) os.FileMode {
+	var m uint32
+	for _, r := range s {
+		if r < '0' || r > '7' {
+			return 0o644
+		}
+		m = m<<3 | uint32(r-'0')
+	}
+	if m == 0 {
+		return 0o644
+	}
+	return os.FileMode(m)
+}

internal/sshpush/fanout_test.go

@@ -0,0 +1,300 @@
+package sshpush
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync/atomic"
+	"testing"
+
+	"golang.org/x/crypto/ssh"
+
+	"git.cloudinit.dev/coreci/orca/internal/emitter"
+)
+
+// --- ExecAll tests ---
+
+func TestExecAll_AllSucceed(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	tr := realTransport(t, srv)
+	defer tr.Close()
+	// Use one server for all peers (same addr).
+	addr := srv.addr()
+	peers := []string{addr, addr, addr}
+	out, errs := tr.ExecAll(context.Background(), peers, "echo hello")
+	for _, p := range peers {
+		if e, ok := errs[p]; ok && e != nil {
+			t.Errorf("peer %s: %v", p, e)
+		}
+		if string(out[p]) != "hello\n" {
+			t.Errorf("out[%s] = %q, want hello\\n", p, out[p])
+		}
+	}
+}
+
+func TestExecAll_OneFails(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	tr := realTransport(t, srv)
+	defer tr.Close()
+	addr := srv.addr()
+	// Peer "bad" returns a permanent error via a mock session.
+	goodPeers := []string{addr}
+	badPeer := "127.0.0.1:1" // unreachable -> transient dial error, retried, fails
+	peers := append(goodPeers, badPeer)
+	out, errs := tr.ExecAll(context.Background(), peers, "echo hello")
+	if string(out[addr]) != "hello\n" {
+		t.Errorf("good peer out = %q, want hello\\n", out[addr])
+	}
+	if errs[badPeer] == nil {
+		t.Error("bad peer should have an error")
+	}
+}
+
+func TestExecAll_WithConcurrency(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	tr := realTransport(t, srv)
+	defer tr.Close()
+	addr := srv.addr()
+	peers := []string{addr, addr, addr, addr}
+	out, errs := tr.ExecAllWithConcurrency(context.Background(), peers, "echo hello", 2)
+	for _, p := range peers {
+		if e := errs[p]; e != nil {
+			t.Errorf("peer %s: %v", p, e)
+		}
+		if string(out[p]) != "hello\n" {
+			t.Errorf("out[%s] = %q", p, out[p])
+		}
+	}
+}
+
+func TestExecAll_EmptyPeers(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	tr := realTransport(t, srv)
+	defer tr.Close()
+	out, errs := tr.ExecAll(context.Background(), nil, "echo hello")
+	if len(out) != 0 || len(errs) != 0 {
+		t.Errorf("empty peers: out=%v errs=%v", out, errs)
+	}
+}
+
+func TestExecAll_ContextCancelled(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	tr := realTransport(t, srv)
+	defer tr.Close()
+	addr := srv.addr()
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	out, errs := tr.ExecAll(ctx, []string{addr, addr}, "echo hello")
+	// With a cancelled context, all peers should fail.
+	for _, p := range []string{addr, addr} {
+		if errs[p] == nil && string(out[p]) == "" {
+			// acceptable: either error or no output
+		}
+	}
+}
+
+// --- WriteAll tests ---
+
+func TestWriteAll_AllSucceed(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	tr := realTransport(t, srv)
+	defer tr.Close()
+	addr := srv.addr()
+	files := map[string][]emitter.File{
+		addr: {
+			{Path: "/w/a", Content: "alpha\n", Mode: "0644"},
+			{Path: "/w/b", Content: "beta\n", Mode: "0644"},
+		},
+	}
+	errs := tr.WriteAll(context.Background(), []string{addr}, files)
+	for p, e := range errs {
+		if e != nil {
+			t.Errorf("peer %s: %v", p, e)
+		}
+	}
+	srv.mu.Lock()
+	if srv.files["/w/a"] != "alpha\n" {
+		t.Errorf("file a = %q", srv.files["/w/a"])
+	}
+	if srv.files["/w/b"] != "beta\n" {
+		t.Errorf("file b = %q", srv.files["/w/b"])
+	}
+	srv.mu.Unlock()
+}
+
+func TestWriteAll_OnePeerFails(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	tr := realTransport(t, srv)
+	defer tr.Close()
+	addr := srv.addr()
+	bad := "127.0.0.1:1"
+	files := map[string][]emitter.File{
+		addr: {{Path: "/ok/f", Content: "ok\n", Mode: "0644"}},
+		bad:  {{Path: "/fail/f", Content: "fail\n", Mode: "0644"}},
+	}
+	errs := tr.WriteAll(context.Background(), []string{addr, bad}, files)
+	if errs[addr] != nil {
+		t.Errorf("good peer should not have error, got %v", errs[addr])
+	}
+	if errs[bad] == nil {
+		t.Error("bad peer should have error")
+	}
+}
+
+func TestWriteAll_EmptyPeers(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	tr := realTransport(t, srv)
+	defer tr.Close()
+	errs := tr.WriteAll(context.Background(), nil, nil)
+	if len(errs) != 0 {
+		t.Errorf("empty peers: errs=%v", errs)
+	}
+}
+
+func TestWriteAll_WithConcurrency(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	tr := realTransport(t, srv)
+	defer tr.Close()
+	addr := srv.addr()
+	files := map[string][]emitter.File{
+		addr: {{Path: "/c/f", Content: "c\n", Mode: "0644"}},
+	}
+	errs := tr.WriteAllWithConcurrency(context.Background(), []string{addr}, files, 4)
+	for _, e := range errs {
+		if e != nil {
+			t.Errorf("peer err: %v", e)
+		}
+	}
+}
+
+func TestWriteAll_Idempotent(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	srv.mu.Lock()
+	srv.files["/i/f"] = "same\n"
+	srv.mu.Unlock()
+	tr := realTransport(t, srv)
+	defer tr.Close()
+	addr := srv.addr()
+	files := map[string][]emitter.File{
+		addr: {{Path: "/i/f", Content: "same\n", Mode: "0644"}},
+	}
+	// Capture writes to verify idempotent skip.
+	var writes int64
+	tr.SetSessionFactory(func(c *ssh.Client) (sshSession, error) {
+		return &writeCountingSession{srv: srv, writes: &writes}, nil
+	})
+	// Pre-populate pool.
+	client, err := tr.dial(addr)
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	tr.pool.Store(addr, client)
+	errs := tr.WriteAll(context.Background(), []string{addr}, files)
+	if errs[addr] != nil {
+		t.Errorf("WriteAll err: %v", errs[addr])
+	}
+	// sha256sum returns a hash that matches -> no write command.
+	srv.mu.Lock()
+	content := srv.files["/i/f"]
+	srv.mu.Unlock()
+	if content != "same\n" {
+		t.Errorf("content changed to %q", content)
+	}
+}
+
+// writeCountingSession counts how many write commands (mkdir + cat >)
+// are issued; returns the server's file content for sha256sum.
+type writeCountingSession struct {
+	srv    *fakeSSHServer
+	writes *int64
+}
+
+func (w *writeCountingSession) CombinedOutput(cmd string) ([]byte, error) {
+	c := trim(cmd)
+	if startsWith(c, "sha256sum ") {
+		path := unquote(trimPrefix(c, "sha256sum "))
+		path = trimSuffix(path, " 2>/dev/null")
+		w.srv.mu.Lock()
+		content, ok := w.srv.files[path]
+		w.srv.mu.Unlock()
+		if !ok {
+			return []byte(""), nil
+		}
+		sum := sha256HexStr([]byte(content))
+		return []byte(sum + "  " + path + "\n"), nil
+	}
+	if startsWith(c, "mkdir -p ") && contains(c, "cat >") {
+		atomic.AddInt64(w.writes, 1)
+		return nil, nil
+	}
+	return nil, nil
+}
+func (w *writeCountingSession) Close() error { return nil }
+
+// Local string helpers to avoid importing strings in a way that
+// conflicts with the test's existing imports.
+func trim(s string) string {
+	for len(s) > 0 && (s[0] == ' ' || s[0] == '\t') {
+		s = s[1:]
+	}
+	for len(s) > 0 && (s[len(s)-1] == ' ' || s[len(s)-1] == '\t') {
+		s = s[:len(s)-1]
+	}
+	return s
+}
+func startsWith(s, prefix string) bool { return len(s) >= len(prefix) && s[:len(prefix)] == prefix }
+func contains(s, sub string) bool {
+	return len(sub) == 0 || (len(s) >= len(sub) && indexOf(s, sub) >= 0)
+}
+func indexOf(s, sub string) int {
+	for i := 0; i+len(sub) <= len(s); i++ {
+		if s[i:i+len(sub)] == sub {
+			return i
+		}
+	}
+	return -1
+}
+func trimPrefix(s, prefix string) string {
+	if startsWith(s, prefix) {
+		return s[len(prefix):]
+	}
+	return s
+}
+func trimSuffix(s, suffix string) string {
+	if len(s) >= len(suffix) && s[len(s)-len(suffix):] == suffix {
+		return s[:len(s)-len(suffix)]
+	}
+	return s
+}
+
+func TestDefaultFanoutConcurrency(t *testing.T) {
+	if DefaultFanoutConcurrency != 8 {
+		t.Errorf("DefaultFanoutConcurrency = %d, want 8", DefaultFanoutConcurrency)
+	}
+}
+
+func TestParseMode_Fanout(t *testing.T) {
+	// parseMode is in fanout.go; sanity-check here too.
+	for _, tc := range []struct{ in, want string }{
+		{"0644", "644"},
+		{"0755", "755"},
+		{"bad", "644"},
+	} {
+		got := fmt.Sprintf("%o", parseMode(tc.in))
+		if got != tc.want {
+			t.Errorf("parseMode(%q) = %s, want %s", tc.in, got, tc.want)
+		}
+	}
+}
+
+var _ = errors.New

internal/sshpush/idempotency.go

@@ -0,0 +1,101 @@
+package sshpush
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"fmt"
+	"math/rand"
+	"os"
+	"strings"
+)
+
+// WriteFileIdempotent writes content to peer:path atomically (write-to-tmp
+// + mv, REQ-074) with mode, but only if the remote file's SHA-256 differs
+// from the local content's SHA-256 (C-18 content-addressed idempotency —
+// the SSH-push equivalent of the v0.8 X-Orca-Idempotency-Key).
+//
+// Returns written=true if the file was written, written=false if the
+// content already matched (skip). The default per-SCP timeout is
+// SCPTimeout (I-B-001).
+//
+// Atomicity: the content is written to a temp file in the same directory
+// as the target, then `mv`'d into place. The temp file is mode-appended
+// (e.g. `/etc/orca/foo.conf.orca-tmp-<rand>`) so the rename is atomic on
+// POSIX filesystems.
+func (t *Transport) WriteFileIdempotent(ctx context.Context, peer string, path string, content []byte, mode os.FileMode) (bool, error) {
+	localHash := sha256Hex(content)
+	remoteHash, err := t.remoteSHA256(ctx, peer, path)
+	if err == nil && remoteHash != "" && strings.EqualFold(remoteHash, localHash) {
+		return false, nil
+	}
+	if err := t.writeFile(ctx, peer, path, content, mode); err != nil {
+		return false, err
+	}
+	return true, nil
+}
+
+// writeFile writes content to peer:path atomically (write-to-tmp + mv).
+// It writes the content via a single SSH exec (cat heredoc + chmod + mv),
+// keeping the transfer in one round-trip. The temp file lives next to the
+// target so the rename is atomic.
+func (t *Transport) writeFile(ctx context.Context, peer string, path string, content []byte, mode os.FileMode) error {
+	dir, base := splitDir(path)
+	tmpName := fmt.Sprintf(".orca-tmp-%s", randomToken(8))
+	tmpPath := base + "/" + tmpName
+	if dir == "" {
+		tmpPath = tmpName
+	}
+	// Build the remote command: mkdir -p <dir> && cat > <tmp> <<'EOF'
+	// ... EOF && chmod <mode> <tmp> && mv <tmp> <path>. The heredoc
+	// delimiter is chosen to not appear in the content (we use a fixed
+	// marker; content with the marker would break, but the marker is
+	// sufficiently unusual).
+	const eof = "ORCA_PUSH_EOF_a1b2c3"
+	modeStr := fmt.Sprintf("%04o", uint32(mode.Perm()))
+	cmd := fmt.Sprintf(
+		"mkdir -p %s && cat > %s <<'%s'\n%s\n%s\nchmod %s %s && mv -f %s %s",
+		shellQuote(base),
+		shellQuote(tmpPath),
+		eof,
+		string(content),
+		eof,
+		modeStr,
+		shellQuote(tmpPath),
+		shellQuote(tmpPath),
+		shellQuote(path),
+	)
+	execCtx, cancel := context.WithTimeout(ctx, SCPTimeout)
+	defer cancel()
+	if _, err := t.execWithRetry(execCtx, peer, cmd, true); err != nil {
+		return fmt.Errorf("sshpush: write %s: %w", path, err)
+	}
+	return nil
+}
+
+// sha256Hex returns the lowercase hex SHA-256 digest of b.
+func sha256Hex(b []byte) string {
+	sum := sha256.Sum256(b)
+	return hex.EncodeToString(sum[:])
+}
+
+// splitDir returns the directory and the directory itself (for mkdir).
+// For "/etc/orca/foo.conf" it returns ("/etc/orca", "/etc/orca"). For
+// "foo.conf" it returns ("", ".").
+func splitDir(path string) (dir, base string) {
+	idx := strings.LastIndex(path, "/")
+	if idx < 0 {
+		return "", "."
+	}
+	return path[:idx], path[:idx]
+}
+
+// randomToken returns a random hex token of the given byte length. Used
+// for temp-file naming to avoid collisions under parallel fan-out.
+func randomToken(n int) string {
+	b := make([]byte, n)
+	for i := range b {
+		b[i] = byte(rand.Intn(256))
+	}
+	return hex.EncodeToString(b)
+}

internal/sshpush/idempotency_test.go

@@ -0,0 +1,348 @@
+package sshpush
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"os"
+	"strings"
+	"testing"
+
+	"golang.org/x/crypto/ssh"
+)
+
+func TestSha256Hex(t *testing.T) {
+	got := sha256Hex([]byte("hello"))
+	want := "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"
+	if got != want {
+		t.Errorf("sha256Hex = %q, want %q", got, want)
+	}
+}
+
+func TestSplitDir(t *testing.T) {
+	dir, base := splitDir("/etc/orca/foo.conf")
+	if dir != "/etc/orca" || base != "/etc/orca" {
+		t.Errorf("splitDir(/etc/orca/foo.conf) = (%q,%q), want (/etc/orca,/etc/orca)", dir, base)
+	}
+	dir, base = splitDir("foo.conf")
+	if dir != "" || base != "." {
+		t.Errorf("splitDir(foo.conf) = (%q,%q), want (\"\",.)", dir, base)
+	}
+}
+
+func TestRandomToken(t *testing.T) {
+	a := randomToken(8)
+	b := randomToken(8)
+	if a == b {
+		t.Error("randomToken returned same value twice")
+	}
+	if len(a) != 16 { // 8 bytes hex = 16 chars
+		t.Errorf("randomToken(8) len = %d, want 16", len(a))
+	}
+}
+
+// --- WriteFileIdempotent tests via the fake SSH server ---
+
+func TestWriteFileIdempotent_WritesWhenFileMissing(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	tr := realTransport(t, srv)
+	defer tr.Close()
+	content := []byte("first content\n")
+	written, err := tr.WriteFileIdempotent(context.Background(), srv.addr(), "/etc/orca/a.conf", content, 0o644)
+	if err != nil {
+		t.Fatalf("WriteFileIdempotent: %v", err)
+	}
+	if !written {
+		t.Error("written=false, want true (file was missing)")
+	}
+	srv.mu.Lock()
+	got := srv.files["/etc/orca/a.conf"]
+	srv.mu.Unlock()
+	if got != string(content) {
+		t.Errorf("remote file = %q, want %q", got, string(content))
+	}
+}
+
+func TestWriteFileIdempotent_SkipsWhenContentMatches(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	content := []byte("same content\n")
+	srv.mu.Lock()
+	srv.files["/etc/orca/b.conf"] = string(content)
+	srv.mu.Unlock()
+	tr := realTransport(t, srv)
+	defer tr.Close()
+	written, err := tr.WriteFileIdempotent(context.Background(), srv.addr(), "/etc/orca/b.conf", content, 0o644)
+	if err != nil {
+		t.Fatalf("WriteFileIdempotent: %v", err)
+	}
+	if written {
+		t.Error("written=true, want false (content matched)")
+	}
+}
+
+func TestWriteFileIdempotent_WritesWhenContentDiffers(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	srv.mu.Lock()
+	srv.files["/etc/orca/c.conf"] = "old content\n"
+	srv.mu.Unlock()
+	tr := realTransport(t, srv)
+	defer tr.Close()
+	newContent := []byte("new content\n")
+	written, err := tr.WriteFileIdempotent(context.Background(), srv.addr(), "/etc/orca/c.conf", newContent, 0o644)
+	if err != nil {
+		t.Fatalf("WriteFileIdempotent: %v", err)
+	}
+	if !written {
+		t.Error("written=false, want true (content differed)")
+	}
+	srv.mu.Lock()
+	got := srv.files["/etc/orca/c.conf"]
+	srv.mu.Unlock()
+	if got != string(newContent) {
+		t.Errorf("remote file = %q, want %q", got, string(newContent))
+	}
+}
+
+func TestWriteFile_DelegatesToIdempotent(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	tr := realTransport(t, srv)
+	defer tr.Close()
+	content := []byte("delegated\n")
+	if err := tr.WriteFile(context.Background(), srv.addr(), "/etc/orca/d.conf", content, 0o600); err != nil {
+		t.Fatalf("WriteFile: %v", err)
+	}
+	srv.mu.Lock()
+	got := srv.files["/etc/orca/d.conf"]
+	srv.mu.Unlock()
+	if got != string(content) {
+		t.Errorf("remote file = %q, want %q", got, string(content))
+	}
+}
+
+// --- Pure-logic idempotency tests via mock session (no SSH server) ---
+
+// mockHashSession returns the hash of the file matching the sha256sum
+// command's path argument; for write commands (mkdir + cat >), it
+// records the write. This lets us test the idempotency decision logic
+// without a real SSH server.
+type mockHashSession struct {
+	files     map[string]string
+	out       []byte
+	err       error
+	cmd       string
+	writeHook func(cmd string)
+}
+
+func (m *mockHashSession) CombinedOutput(cmd string) ([]byte, error) {
+	m.cmd = cmd
+	c := strings.TrimSpace(cmd)
+	if strings.HasPrefix(c, "sha256sum ") {
+		rest := strings.TrimSpace(strings.TrimPrefix(c, "sha256sum "))
+		rest = strings.TrimSuffix(rest, " 2>/dev/null")
+		rest = strings.TrimSpace(rest)
+		path := unquote(rest)
+		content, ok := m.files[path]
+		if !ok {
+			return []byte(""), nil
+		}
+		sum := sha256.Sum256([]byte(content))
+		return []byte(hex.EncodeToString(sum[:]) + "  " + path + "\n"), nil
+	}
+	if strings.HasPrefix(c, "mkdir -p ") && strings.Contains(c, "cat >") {
+		if m.writeHook != nil {
+			m.writeHook(c)
+		}
+		return nil, nil
+	}
+	return m.out, m.err
+}
+func (m *mockHashSession) Close() error { return nil }
+
+func TestWriteFileIdempotent_MockSkip(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	tr := realTransport(t, srv)
+	defer tr.Close()
+	files := map[string]string{"/x/f": "match"}
+	var writes int
+	tr.SetSessionFactory(func(c *ssh.Client) (sshSession, error) {
+		return &mockHashSession{
+			files:     files,
+			writeHook: func(string) { writes++ },
+		}, nil
+	})
+	client, err := tr.dial(srv.addr())
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	tr.pool.Store(srv.addr(), client)
+	written, err := tr.WriteFileIdempotent(context.Background(), srv.addr(), "/x/f", []byte("match"), 0o644)
+	if err != nil {
+		t.Fatalf("WriteFileIdempotent: %v", err)
+	}
+	if written {
+		t.Error("written=true, want false (hash matched)")
+	}
+	if writes != 0 {
+		t.Errorf("writes = %d, want 0 (no write on hash match)", writes)
+	}
+}
+
+func TestWriteFileIdempotent_MockWrite(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	tr := realTransport(t, srv)
+	defer tr.Close()
+	files := map[string]string{"/x/f": "old"}
+	var writes int
+	tr.SetSessionFactory(func(c *ssh.Client) (sshSession, error) {
+		return &mockHashSession{
+			files:     files,
+			writeHook: func(string) { writes++ },
+		}, nil
+	})
+	client, err := tr.dial(srv.addr())
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	tr.pool.Store(srv.addr(), client)
+	written, err := tr.WriteFileIdempotent(context.Background(), srv.addr(), "/x/f", []byte("new"), 0o644)
+	if err != nil {
+		t.Fatalf("WriteFileIdempotent: %v", err)
+	}
+	if !written {
+		t.Error("written=false, want true (hash differed)")
+	}
+	if writes != 1 {
+		t.Errorf("writes = %d, want 1", writes)
+	}
+}
+
+func TestWriteFileIdempotent_MockMissingFile(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	tr := realTransport(t, srv)
+	defer tr.Close()
+	files := map[string]string{}
+	var writes int
+	tr.SetSessionFactory(func(c *ssh.Client) (sshSession, error) {
+		return &mockHashSession{
+			files:     files,
+			writeHook: func(string) { writes++ },
+		}, nil
+	})
+	client, err := tr.dial(srv.addr())
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	tr.pool.Store(srv.addr(), client)
+	written, err := tr.WriteFileIdempotent(context.Background(), srv.addr(), "/x/new", []byte("fresh"), 0o644)
+	if err != nil {
+		t.Fatalf("WriteFileIdempotent: %v", err)
+	}
+	if !written {
+		t.Error("written=false, want true (file missing)")
+	}
+	if writes != 1 {
+		t.Errorf("writes = %d, want 1", writes)
+	}
+}
+
+func TestRemoteSHA256_ParsesDigest(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	srv.mu.Lock()
+	srv.files["/x/h"] = "abc"
+	srv.mu.Unlock()
+	tr := realTransport(t, srv)
+	defer tr.Close()
+	got, err := tr.remoteSHA256(context.Background(), srv.addr(), "/x/h")
+	if err != nil {
+		t.Fatalf("remoteSHA256: %v", err)
+	}
+	want := fmt.Sprintf("%x", sha256.Sum256([]byte("abc")))
+	if got != want {
+		t.Errorf("remoteSHA256 = %q, want %q", got, want)
+	}
+}
+
+func TestRemoteSHA256_MissingFile(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	tr := realTransport(t, srv)
+	defer tr.Close()
+	got, err := t_remoteSHA256_noErr(t, tr, srv.addr(), "/missing")
+	if err != nil {
+		t.Fatalf("remoteSHA256: %v", err)
+	}
+	if got != "" {
+		t.Errorf("remoteSHA256 = %q, want empty for missing file", got)
+	}
+}
+
+func t_remoteSHA256_noErr(t *testing.T, tr *Transport, peer, path string) (string, error) {
+	t.Helper()
+	return tr.remoteSHA256(context.Background(), peer, path)
+}
+
+func TestWriteFile_ModeApplied(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	tr := realTransport(t, srv)
+	defer tr.Close()
+	if err := tr.WriteFile(context.Background(), srv.addr(), "/m/f", []byte("mode"), 0o755); err != nil {
+		t.Fatalf("WriteFile: %v", err)
+	}
+	srv.mu.Lock()
+	got := srv.files["/m/f"]
+	srv.mu.Unlock()
+	if got != "mode" {
+		t.Errorf("content = %q, want mode", got)
+	}
+}
+
+func TestWriteFileIdempotent_ExecError(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	tr := realTransport(t, srv)
+	defer tr.Close()
+	tr.SetSessionFactory(func(c *ssh.Client) (sshSession, error) {
+		return &mockSession{err: fmt.Errorf("%w: boom", ErrPermanent)}, nil
+	})
+	client, err := tr.dial(srv.addr())
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	tr.pool.Store(srv.addr(), client)
+	_, err = tr.WriteFileIdempotent(context.Background(), srv.addr(), "/x/f", []byte("z"), 0o644)
+	if err == nil {
+		t.Fatal("expected error, got nil")
+	}
+}
+
+func TestParseMode(t *testing.T) {
+	for _, tc := range []struct {
+		in   string
+		want os.FileMode
+	}{
+		{"0644", 0o644},
+		{"0755", 0o755},
+		{"0600", 0o600},
+		{"bad", 0o644},
+		{"", 0o644},
+		{"0", 0o644},
+	} {
+		got := parseMode(tc.in)
+		if got != tc.want {
+			t.Errorf("parseMode(%q) = %o, want %o", tc.in, got, tc.want)
+		}
+	}
+}
+
+var _ = errors.New

internal/sshpush/transport.go

@@ -0,0 +1,483 @@
+package sshpush
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"math/rand"
+	"net"
+	"os"
+	"strings"
+	"sync"
+	"time"
+
+	"golang.org/x/crypto/ssh"
+
+	"git.cloudinit.dev/coreci/orca/internal/proxmox"
+)
+
+// Default timeouts and retry parameters (REQ-073, I-B-001).
+const (
+	// ExecTimeout is the default per-exec timeout for a single SSH
+	// command (I-B-001).
+	ExecTimeout = 10 * time.Second
+	// SCPTimeout is the default per-SCP timeout for a single file
+	// transfer (I-B-001).
+	SCPTimeout = 30 * time.Second
+	// DialTimeout is the default SSH dial timeout.
+	DialTimeout = 15 * time.Second
+	// RetryInitial is the first backoff interval (v0.8 transport/retry.go).
+	RetryInitial = 100 * time.Millisecond
+	// RetryMax is the cap on backoff between attempts.
+	RetryMax = 5 * time.Second
+	// RetryMaxAttempts is the total attempt count (including the first).
+	RetryMaxAttempts = 5
+)
+
+// Sentinel errors. ErrTransient marks a transient failure worth
+// retrying; ErrPermanent marks a non-retryable failure (auth, host-key
+// mismatch, validation). These mirror the v0.8 transport sentinels
+// (reimplemented here since internal/transport is not imported).
+var (
+	ErrTransient    = errors.New("sshpush: transient error")
+	ErrPermanent    = errors.New("sshpush: permanent error")
+	ErrNotConnected = errors.New("sshpush: not connected")
+)
+
+// Transport is the SSH-push transport (REQ-073). It reuses one
+// *ssh.Client per peer across multiple operations within a single CLI
+// invocation (I-B-001). The zero value is NOT usable; construct one with
+// NewTransport.
+type Transport struct {
+	// pool caches *ssh.Client per peer address ("host:port").
+	pool sync.Map
+	// keyPath is the SSH private key path (Ed25519, D-037).
+	keyPath string
+	// knownHostsPath is the v0.9 known_hosts path (paths.KnownHostsPath()
+	// = ClusterDir()/known_hosts). It is stored for the v0.10-P14 migration
+	// when proxmox.TOFUHostKeyCallback will accept a path parameter; today
+	// the callback reads certpaths.KnownHostsPath() (the v0.8 flat layout)
+	// directly, so this field is not yet read by dial(). Tests set
+	// $ORCA_HOME so certpaths.KnownHostsPath() resolves under the temp dir.
+	knownHostsPath string
+	// user is the remote SSH user (default "orca", D-037).
+	user string
+	// signer is the parsed SSH private key signer, set lazily on first
+	// dial.
+	signer  ssh.Signer
+	signErr error
+	// signerOnce guards signer initialization.
+	signerOnce sync.Once
+
+	// dialer is the SSH dialer. Tests override it to inject a mock
+	// server. The default uses ssh.DialContext via the context-aware
+	// wrapper.
+	dialer sshDialer
+
+	// sessionFactory returns a new session for a given client. Tests
+	// override it to inject mock sessions without a real *ssh.Client.
+	// When nil, the default (*ssh.Client).NewSession is used.
+	sessionFactory func(*ssh.Client) (sshSession, error)
+
+	// mu guards the closed flag (pool iteration is sync.Map.Range).
+	closed bool
+	mu     sync.Mutex
+}
+
+// sshSession is the minimal *ssh.Session surface the transport uses.
+// It lets tests substitute a mock without a real SSH server.
+type sshSession interface {
+	CombinedOutput(cmd string) ([]byte, error)
+	Close() error
+}
+
+// sshDialer is the SSH dialer interface (mirrors proxmox.sshDialerType).
+// The default uses ssh.Dial; tests inject mocks that return a fake
+// *ssh.Client or an error.
+type sshDialer interface {
+	DialContext(ctx context.Context, network, addr string, config *ssh.ClientConfig) (*ssh.Client, error)
+}
+
+// defaultSSHDialer wraps ssh.Dial with a context-aware connect timeout.
+type defaultSSHDialer struct{}
+
+func (defaultSSHDialer) DialContext(ctx context.Context, network, addr string, config *ssh.ClientConfig) (*ssh.Client, error) {
+	d := net.Dialer{Timeout: config.Timeout}
+	if d.Timeout == 0 {
+		d.Timeout = DialTimeout
+	}
+	conn, err := d.DialContext(ctx, network, addr)
+	if err != nil {
+		return nil, err
+	}
+	sshConn, chans, reqs, err := ssh.NewClientConn(conn, addr, config)
+	if err != nil {
+		_ = conn.Close()
+		return nil, err
+	}
+	return ssh.NewClient(sshConn, chans, reqs), nil
+}
+
+// NewTransport returns a Transport configured with the given SSH
+// private key path and known_hosts path. The known_hosts path is the v0.9
+// location (paths.KnownHostsPath); it is stored for the v0.10-P14
+// migration when the TOFU callback will accept a path parameter. Today
+// dial() delegates host-key verification to proxmox.TOFUHostKeyCallback,
+// which reads certpaths.KnownHostsPath() (the v0.8 flat layout under
+// $ORCA_HOME) directly — so callers must ensure $ORCA_HOME points at the
+// cluster root (the CLI sets this up). The remote user defaults to
+// "orca" (D-037); override with SetUser. The dialer defaults to the
+// real ssh.Dial-based dialer; tests call SetDialer to inject a mock.
+func NewTransport(keyPath, knownHostsPath string) *Transport {
+	return &Transport{
+		keyPath:        keyPath,
+		knownHostsPath: knownHostsPath,
+		user:           "orca",
+		dialer:         defaultSSHDialer{},
+	}
+}
+
+// SetUser overrides the remote SSH user (default "orca").
+func (t *Transport) SetUser(user string) {
+	if user != "" {
+		t.user = user
+	}
+}
+
+// SetDialer overrides the SSH dialer (for tests).
+func (t *Transport) SetDialer(d sshDialer) {
+	if d != nil {
+		t.dialer = d
+	}
+}
+
+// SetSessionFactory overrides the session factory (for tests). The
+// factory is called per-exec/write/read to obtain a fresh session; it
+// must close the session when the test mock is done, or the transport
+// will call Close on the returned session.
+func (t *Transport) SetSessionFactory(f func(*ssh.Client) (sshSession, error)) {
+	t.sessionFactory = f
+}
+
+// dial returns the cached *ssh.Client for peer, dialing and caching on
+// first use (I-B-001 connection pooling). Returns an error if the dial
+// fails or the transport is closed.
+func (t *Transport) dial(peer string) (*ssh.Client, error) {
+	t.mu.Lock()
+	if t.closed {
+		t.mu.Unlock()
+		return nil, ErrPermanent
+	}
+	t.mu.Unlock()
+	if c, ok := t.pool.Load(peer); ok {
+		return c.(*ssh.Client), nil
+	}
+	// Lazily parse the private key signer (once across all dials).
+	t.signerOnce.Do(func() {
+		keyBytes, err := os.ReadFile(t.keyPath)
+		if err != nil {
+			t.signErr = fmt.Errorf("sshpush: read key %s: %w", t.keyPath, err)
+			return
+		}
+		s, err := ssh.ParsePrivateKey(keyBytes)
+		if err != nil {
+			t.signErr = fmt.Errorf("sshpush: parse key: %w", err)
+			return
+		}
+		t.signer = s
+	})
+	if t.signErr != nil {
+		return nil, t.signErr
+	}
+	// Host-key verification reuses the v0.8 TOFU wrapper (D-035). The
+	// known_hosts file is flock-protected inside the callback on
+	// first-connect capture, so we do NOT re-lock here.
+	cb, err := proxmox.TOFUHostKeyCallback(peer, nil)
+	if err != nil {
+		return nil, fmt.Errorf("sshpush: host-key callback: %w", err)
+	}
+	config := &ssh.ClientConfig{
+		User:            t.user,
+		Auth:            []ssh.AuthMethod{ssh.PublicKeys(t.signer)},
+		HostKeyCallback: cb,
+		Timeout:         DialTimeout,
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), DialTimeout)
+	defer cancel()
+	client, err := t.dialer.DialContext(ctx, "tcp", peer, config)
+	if err != nil {
+		return nil, classifyDialErr(err)
+	}
+	// Race: two goroutines dialing the same peer concurrently both
+	// create a client. Last-wins; the loser is closed. This is rare
+	// (dial is rare and the pool hit short-circuits) and harmless.
+	if existing, loaded := t.pool.LoadOrStore(peer, client); loaded {
+		_ = client.Close()
+		return existing.(*ssh.Client), nil
+	}
+	return client, nil
+}
+
+// Exec runs cmd on peer over SSH and returns its combined output. The
+// default per-exec timeout is ExecTimeout (I-B-001); override by
+// passing a context with a shorter deadline. Transient failures are
+// retried with exponential backoff (100ms ×2, cap 5s, max 5 attempts —
+// the v0.8 transport/retry.go pattern, reimplemented here).
+func (t *Transport) Exec(ctx context.Context, peer string, cmd string) ([]byte, error) {
+	return t.execWithRetry(ctx, peer, cmd, true)
+}
+
+// execWithRetry runs the exec with retry. exec is treated as
+// idempotent (read-only) for retry purposes; the idempotency helpers
+// (WriteFileIdempotent) handle writes.
+func (t *Transport) execWithRetry(ctx context.Context, peer string, cmd string, idempotent bool) ([]byte, error) {
+	var lastErr error
+	for attempt := 1; attempt <= RetryMaxAttempts; attempt++ {
+		if err := ctx.Err(); err != nil {
+			return nil, err
+		}
+		out, err := t.execOnce(ctx, peer, cmd)
+		if err == nil {
+			return out, nil
+		}
+		if errors.Is(err, ErrPermanent) {
+			return nil, err
+		}
+		lastErr = err
+		if attempt == RetryMaxAttempts {
+			break
+		}
+		if !isTransient(err) {
+			return nil, err
+		}
+		wait := backoff(RetryInitial, RetryMax, attempt)
+		timer := time.NewTimer(wait)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return nil, ctx.Err()
+		case <-timer.C:
+		}
+	}
+	return nil, lastErr
+}
+
+// execOnce runs the command a single time against peer.
+func (t *Transport) execOnce(ctx context.Context, peer string, cmd string) ([]byte, error) {
+	client, err := t.dial(peer)
+	if err != nil {
+		return nil, classifyDialErr(err)
+	}
+	sess, err := t.newSession(client)
+	if err != nil {
+		return nil, fmt.Errorf("sshpush: new session: %w", err)
+	}
+	defer sess.Close()
+	type result struct {
+		out []byte
+		err error
+	}
+	ch := make(chan result, 1)
+	go func() {
+		out, err := sess.CombinedOutput(cmd)
+		ch <- result{out, err}
+	}()
+	timeout := ExecTimeout
+	if dl, ok := ctx.Deadline(); ok {
+		if remaining := time.Until(dl); remaining > 0 && remaining < timeout {
+			timeout = remaining
+		}
+	}
+	select {
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	case <-time.After(timeout):
+		return nil, fmt.Errorf("sshpush: exec timeout after %s: %w", timeout, ErrTransient)
+	case r := <-ch:
+		if r.err != nil {
+			return r.out, classifyExecErr(r.err)
+		}
+		return r.out, nil
+	}
+}
+
+// newSession returns a session for client, using the override factory
+// when set (tests), otherwise the real *ssh.Client.NewSession.
+func (t *Transport) newSession(client *ssh.Client) (sshSession, error) {
+	if t.sessionFactory != nil {
+		return t.sessionFactory(client)
+	}
+	s, err := client.NewSession()
+	if err != nil {
+		return nil, err
+	}
+	return &realSession{Session: s}, nil
+}
+
+// realSession wraps *ssh.Session to satisfy the sshSession interface.
+type realSession struct {
+	*ssh.Session
+}
+
+func (r *realSession) CombinedOutput(cmd string) ([]byte, error) {
+	return r.Session.CombinedOutput(cmd)
+}
+
+// WriteFile SCPs content to peer:path atomically (write-to-tmp + mv,
+// REQ-074). The default per-SCP timeout is SCPTimeout (I-B-001).
+// Idempotency: if the file already exists with the same SHA-256, the
+// write is skipped (C-18). Use WriteFileIdempotent for the explicit
+// written/skipped result.
+func (t *Transport) WriteFile(ctx context.Context, peer string, path string, content []byte, mode os.FileMode) error {
+	_, err := t.WriteFileIdempotent(ctx, peer, path, content, mode)
+	return err
+}
+
+// ReadFile reads the file at peer:path via SSH cat.
+func (t *Transport) ReadFile(ctx context.Context, peer string, path string) ([]byte, error) {
+	cmd := fmt.Sprintf("cat %s", shellQuote(path))
+	out, err := t.Exec(ctx, peer, cmd)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+// Close closes all pooled SSH clients (REQ-073). Safe to call
+// multiple times; subsequent calls are no-ops.
+func (t *Transport) Close() error {
+	t.mu.Lock()
+	if t.closed {
+		t.mu.Unlock()
+		return nil
+	}
+	t.closed = true
+	t.mu.Unlock()
+	var firstErr error
+	t.pool.Range(func(key, value any) bool {
+		if c, ok := value.(*ssh.Client); ok {
+			if err := c.Close(); err != nil && firstErr == nil {
+				firstErr = err
+			}
+		}
+		t.pool.Delete(key)
+		return true
+	})
+	return firstErr
+}
+
+// backoff returns the wait duration for the n-th attempt (1-indexed).
+// Formula: min(Initial * 2^(n-1), Max), with up to 25% jitter (matches
+// v0.8 transport/retry.go).
+func backoff(initial, max time.Duration, n int) time.Duration {
+	d := initial
+	for i := 1; i < n; i++ {
+		d *= 2
+		if d > max {
+			d = max
+			break
+		}
+	}
+	if d <= 0 {
+		return 0
+	}
+	jitter := time.Duration(rand.Int63n(int64(d) / 2))
+	d = d - d/4 + jitter
+	if d < 0 {
+		d = 0
+	}
+	return d
+}
+
+// isTransient reports whether err looks like a transient failure worth
+// retrying (mirrors v0.8 transport.IsTransient, reimplemented here).
+func isTransient(err error) bool {
+	if err == nil {
+		return false
+	}
+	if errors.Is(err, ErrTransient) {
+		return true
+	}
+	if errors.Is(err, ErrPermanent) {
+		return false
+	}
+	s := err.Error()
+	for _, sub := range []string{
+		"connection refused", "i/o timeout", "EOF",
+		"no such host", "connection reset", "timeout",
+		"deadline exceeded", "temporarily unavailable",
+	} {
+		if strings.Contains(s, sub) {
+			return true
+		}
+	}
+	return false
+}
+
+// classifyDialErr converts a raw ssh.Dial error into a transport error
+// (transient vs permanent). Auth failures and host-key mismatches are
+// permanent; everything else is transient.
+func classifyDialErr(err error) error {
+	if err == nil {
+		return nil
+	}
+	s := err.Error()
+	if strings.Contains(s, "unable to authenticate") || strings.Contains(s, "handshake failed") {
+		return fmt.Errorf("%w: %v", ErrPermanent, err)
+	}
+	if strings.Contains(s, "host key") && strings.Contains(s, "mismatch") {
+		return fmt.Errorf("%w: %v", ErrPermanent, err)
+	}
+	if strings.Contains(s, "knownhosts") {
+		return fmt.Errorf("%w: %v", ErrPermanent, err)
+	}
+	return fmt.Errorf("%w: %v", ErrTransient, err)
+}
+
+// classifyExecErr converts a raw session exec error into a transport
+// error. Non-zero exit codes are NOT transient (the command ran; the
+// failure is logical, not network). Session-creation failures and
+// network-level errors are transient.
+func classifyExecErr(err error) error {
+	if err == nil {
+		return nil
+	}
+	var exitErr *ssh.ExitError
+	if errors.As(err, &exitErr) {
+		return fmt.Errorf("%w: exit %d", ErrPermanent, exitErr.ExitStatus())
+	}
+	s := err.Error()
+	for _, sub := range []string{"EOF", "session closed", "channel closed"} {
+		if strings.Contains(s, sub) {
+			return fmt.Errorf("%w: %v", ErrTransient, err)
+		}
+	}
+	return fmt.Errorf("%w: %v", ErrPermanent, err)
+}
+
+// shellQuote single-quotes a path for safe shell interpolation. It
+// escapes embedded single-quotes via the standard '\” idiom.
+func shellQuote(s string) string {
+	return "'" + strings.ReplaceAll(s, "'", "'\\''") + "'"
+}
+
+// remoteSHA256 returns the SHA-256 of the file at peer:path via SSH
+// `sha256sum`, or ("", error) if the file is missing or the command
+// fails. The returned hash is the hex digest (lowercase, no filename).
+func (t *Transport) remoteSHA256(ctx context.Context, peer string, path string) (string, error) {
+	cmd := fmt.Sprintf("sha256sum %s 2>/dev/null", shellQuote(path))
+	out, err := t.execWithRetry(ctx, peer, cmd, true)
+	if err != nil {
+		return "", err
+	}
+	out = bytes.TrimSpace(out)
+	if len(out) == 0 {
+		return "", nil
+	}
+	fields := strings.Fields(string(out))
+	if len(fields) == 0 {
+		return "", nil
+	}
+	return fields[0], nil
+}

internal/sshpush/transport_test.go

@@ -0,0 +1,664 @@
+package sshpush
+
+import (
+	"context"
+	"crypto/ed25519"
+	"crypto/rand"
+	"crypto/sha256"
+	"crypto/x509"
+	"encoding/pem"
+	"errors"
+	"fmt"
+	"net"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"golang.org/x/crypto/ssh"
+	"golang.org/x/crypto/ssh/knownhosts"
+)
+
+// --- fakeSSHServer: a minimal in-process SSH server for hermetic tests.
+
+type fakeSSHServer struct {
+	listener net.Listener
+	config   *ssh.ServerConfig
+	done     chan struct{}
+
+	mu        sync.Mutex
+	files     map[string]string
+	hostKey   ssh.Signer
+	cmdCount  int64
+	execDelay time.Duration
+}
+
+func newFakeSSHServer(t *testing.T) *fakeSSHServer {
+	t.Helper()
+	_, priv, err := ed25519.GenerateKey(rand.Reader)
+	if err != nil {
+		t.Fatalf("ed25519 gen: %v", err)
+	}
+	signer, err := ssh.NewSignerFromKey(priv)
+	if err != nil {
+		t.Fatalf("ssh signer: %v", err)
+	}
+	config := &ssh.ServerConfig{
+		NoClientAuth: true,
+	}
+	config.AddHostKey(signer)
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	srv := &fakeSSHServer{
+		listener: ln,
+		config:   config,
+		done:     make(chan struct{}),
+		files:    make(map[string]string),
+		hostKey:  signer,
+	}
+	go srv.serve()
+	return srv
+}
+
+func (s *fakeSSHServer) addr() string                 { return s.listener.Addr().String() }
+func (s *fakeSSHServer) hostPublicKey() ssh.PublicKey { return s.hostKey.PublicKey() }
+
+func (s *fakeSSHServer) close() {
+	_ = s.listener.Close()
+	<-s.done
+}
+
+func (s *fakeSSHServer) setExecDelay(d time.Duration) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.execDelay = d
+}
+
+func (s *fakeSSHServer) serve() {
+	for {
+		conn, err := s.listener.Accept()
+		if err != nil {
+			close(s.done)
+			return
+		}
+		go s.handle(conn)
+	}
+}
+
+func (s *fakeSSHServer) handle(netConn net.Conn) {
+	defer netConn.Close()
+	_, chans, reqs, err := ssh.NewServerConn(netConn, s.config)
+	if err != nil {
+		return
+	}
+	go ssh.DiscardRequests(reqs)
+	for newChan := range chans {
+		if newChan.ChannelType() != "session" {
+			newChan.Reject(ssh.UnknownChannelType, "only session")
+			continue
+		}
+		go s.handleSession(newChan)
+	}
+}
+
+func (s *fakeSSHServer) handleSession(newChan ssh.NewChannel) {
+	ch, reqs, err := newChan.Accept()
+	if err != nil {
+		return
+	}
+	defer ch.Close()
+	for req := range reqs {
+		if req.Type != "exec" {
+			req.Reply(false, nil)
+			continue
+		}
+		var execReq struct{ Command string }
+		if err := ssh.Unmarshal(req.Payload, &execReq); err != nil {
+			req.Reply(false, nil)
+			continue
+		}
+		req.Reply(true, nil)
+		atomic.AddInt64(&s.cmdCount, 1)
+		s.mu.Lock()
+		delay := s.execDelay
+		s.mu.Unlock()
+		if delay > 0 {
+			time.Sleep(delay)
+		}
+		out, code := s.runCommand(execReq.Command)
+		_, _ = ch.Write(out)
+		_, _ = ch.SendRequest("exit-status", false, ssh.Marshal(struct{ Code uint32 }{uint32(code)}))
+		_ = ch.Close()
+		return
+	}
+}
+
+// runCommand implements the minimal command surface the transport uses:
+// echo (for exec tests), sha256sum (for idempotency), cat (read), and the
+// heredoc-based write (cat > tmp <<EOF ... EOF && chmod ... && mv ...).
+func (s *fakeSSHServer) runCommand(cmd string) ([]byte, int) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	trimmed := strings.TrimSpace(cmd)
+	switch {
+	case trimmed == "echo hello":
+		return []byte("hello\n"), 0
+	case strings.HasPrefix(trimmed, "sha256sum "):
+		// Format: sha256sum '/path' 2>/dev/null
+		rest := strings.TrimSpace(strings.TrimPrefix(trimmed, "sha256sum "))
+		rest = strings.TrimSuffix(rest, " 2>/dev/null")
+		rest = strings.TrimSpace(rest)
+		path := unquote(rest)
+		content, ok := s.files[path]
+		if !ok {
+			// `2>/dev/null` swallows the error; sha256sum exits 1 but
+			// stderr is suppressed. The transport treats empty output as
+			// "file missing" (no hash), so return ("", 0).
+			return []byte(""), 0
+		}
+		sum := sha256HexStr([]byte(content))
+		return []byte(sum + "  " + path + "\n"), 0
+	case strings.HasPrefix(trimmed, "cat '"):
+		path := unquote(strings.TrimPrefix(trimmed, "cat "))
+		content, ok := s.files[path]
+		if !ok {
+			return []byte("cat: " + path + ": No such file or directory\n"), 1
+		}
+		return []byte(content), 0
+	case strings.HasPrefix(trimmed, "mkdir -p ") && strings.Contains(trimmed, "cat >"):
+		return s.handleWrite(trimmed)
+	default:
+		return []byte("sh: command not found\n"), 127
+	}
+}
+
+// handleWrite parses the heredoc write command produced by writeFile.
+// Command format:
+//
+//	mkdir -p '<dir>' && cat > '<tmp>' <<'ORCA_PUSH_EOF_a1b2c3'
+//	<content>
+//	ORCA_PUSH_EOF_a1b2c3
+//	chmod <mode> '<tmp>' && mv -f '<tmp>' '<path>'
+func (s *fakeSSHServer) handleWrite(cmd string) ([]byte, int) {
+	const eof = "ORCA_PUSH_EOF_a1b2c3"
+	// Find the opening heredoc line: ... <<'EOF'\n
+	openerIdx := strings.Index(cmd, "<<'"+eof+"'")
+	if openerIdx < 0 {
+		return []byte("sh: no heredoc opener\n"), 1
+	}
+	// Body starts after the opener line's newline.
+	rest := cmd[openerIdx+len("<<'"+eof+"'"):]
+	nl := strings.Index(rest, "\n")
+	if nl < 0 {
+		return []byte("sh: no body start\n"), 1
+	}
+	body := rest[nl+1:]
+	// Body ends at the closing EOF marker on its own line.
+	closeIdx := strings.Index(body, "\n"+eof+"\n")
+	if closeIdx < 0 {
+		// Maybe EOF is at the end without trailing newline.
+		closeIdx = strings.Index(body, "\n"+eof)
+		if closeIdx < 0 {
+			return []byte("sh: no heredoc close\n"), 1
+		}
+		body = body[:closeIdx]
+	} else {
+		body = body[:closeIdx]
+	}
+	// Find the mv target: last quoted arg of "mv -f 'tmp' 'path'".
+	mvIdx := strings.LastIndex(cmd, "mv -f ")
+	if mvIdx < 0 {
+		return []byte("sh: no mv\n"), 1
+	}
+	tail := cmd[mvIdx+len("mv -f "):]
+	parts := splitQuoted(tail)
+	if len(parts) < 2 {
+		return []byte("sh: bad mv args\n"), 1
+	}
+	target := parts[1]
+	s.files[target] = body
+	return nil, 0
+}
+
+// splitQuoted splits a string of the form "'a' 'b'" into ["a","b"].
+func splitQuoted(s string) []string {
+	var out []string
+	var cur strings.Builder
+	in := false
+	for _, r := range s {
+		if r == '\'' {
+			if in {
+				out = append(out, cur.String())
+				cur.Reset()
+			}
+			in = !in
+			continue
+		}
+		if in {
+			cur.WriteRune(r)
+		}
+	}
+	return out
+}
+
+func unquote(s string) string {
+	s = strings.TrimSpace(s)
+	if len(s) >= 2 && s[0] == '\'' && s[len(s)-1] == '\'' {
+		return s[1 : len(s)-1]
+	}
+	if len(s) >= 2 && s[0] == '"' && s[len(s)-1] == '"' {
+		return s[1 : len(s)-1]
+	}
+	return s
+}
+
+// sha256HexStr is a test-local copy of the sha256Hex helper.
+func sha256HexStr(b []byte) string {
+	sum := sha256.Sum256(b)
+	return fmt.Sprintf("%x", sum[:])
+}
+
+// --- test helpers for transport setup ---
+
+// setupORCAHome creates a temp ORCA_HOME with an empty known_hosts (at
+// the v0.8 flat location $ORCA_HOME/known_hosts, which is where
+// proxmox.TOFUHostKeyCallback reads via certpaths.KnownHostsPath()) and a
+// generated Ed25519 SSH key, returns the key path.
+func setupORCAHome(t *testing.T) (keyPath string) {
+	t.Helper()
+	dir := t.TempDir()
+	t.Setenv("ORCA_HOME", dir)
+	// certpaths.KnownHostsPath() = paths.Root()/known_hosts = $ORCA_HOME/known_hosts.
+	knownHosts := filepath.Join(dir, "known_hosts")
+	if err := os.WriteFile(knownHosts, []byte{}, 0o600); err != nil {
+		t.Fatalf("create known_hosts: %v", err)
+	}
+	_, priv, err := ed25519.GenerateKey(rand.Reader)
+	if err != nil {
+		t.Fatalf("ed25519 gen: %v", err)
+	}
+	der, err := x509.MarshalPKCS8PrivateKey(priv)
+	if err != nil {
+		t.Fatalf("marshal key: %v", err)
+	}
+	pemBytes := pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: der})
+	keyPath = filepath.Join(dir, "orca_ssh_key")
+	if err := os.WriteFile(keyPath, pemBytes, 0o600); err != nil {
+		t.Fatalf("write key: %v", err)
+	}
+	return keyPath
+}
+
+// realTransport returns a Transport wired to use the real SSH dialer
+// against a fake SSH server, with the server's host key pre-populated in
+// known_hosts (so the TOFU callback matches on first dial — no first-
+// connect write race in tests).
+func realTransport(t *testing.T, srv *fakeSSHServer) *Transport {
+	t.Helper()
+	keyPath := setupORCAHome(t)
+	tr := NewTransport(keyPath, "")
+	tr.SetUser("root")
+	addr := srv.addr()
+	line := knownhosts.Line([]string{knownhosts.Normalize(addr)}, srv.hostPublicKey())
+	home := os.Getenv("ORCA_HOME")
+	kh := filepath.Join(home, "known_hosts")
+	if err := os.WriteFile(kh, []byte(line+"\n"), 0o600); err != nil {
+		t.Fatalf("pre-pop known_hosts: %v", err)
+	}
+	return tr
+}
+
+// --- mock dialer + mock session for pure-logic tests (no real SSH) ---
+
+type mockDialer struct {
+	client *ssh.Client
+	err    error
+	calls  int
+}
+
+func (m *mockDialer) DialContext(ctx context.Context, network, addr string, cfg *ssh.ClientConfig) (*ssh.Client, error) {
+	m.calls++
+	if m.err != nil {
+		return nil, m.err
+	}
+	return m.client, nil
+}
+
+type mockSession struct {
+	out []byte
+	err error
+	cmd string
+}
+
+func (m *mockSession) CombinedOutput(cmd string) ([]byte, error) {
+	m.cmd = cmd
+	return m.out, m.err
+}
+func (m *mockSession) Close() error { return nil }
+
+// --- tests ---
+
+func TestNewTransport_Defaults(t *testing.T) {
+	tr := NewTransport("/tmp/key", "/tmp/kh")
+	if tr.keyPath != "/tmp/key" {
+		t.Errorf("keyPath = %q", tr.keyPath)
+	}
+	if tr.user != "orca" {
+		t.Errorf("default user = %q, want orca", tr.user)
+	}
+	if tr.dialer == nil {
+		t.Error("dialer is nil")
+	}
+}
+
+func TestSetUser(t *testing.T) {
+	tr := NewTransport("/tmp/key", "/tmp/kh")
+	tr.SetUser("root")
+	if tr.user != "root" {
+		t.Errorf("user = %q, want root", tr.user)
+	}
+	tr.SetUser("")
+	if tr.user != "root" {
+		t.Errorf("user = %q, want root", tr.user)
+	}
+}
+
+func TestBackoff(t *testing.T) {
+	// Without jitter, attempt 1 -> 100ms, 2 -> 200ms, ... up to 5s cap.
+	// The jittered result is in [d/4, 3d/4) where d is the capped base,
+	// so for high attempts the result can reach 3*d/4 < 1.5*d. We bound
+	// the upper end at 2x the cap to allow jitter headroom.
+	for _, tc := range []struct {
+		attempt int
+		max     time.Duration
+	}{
+		{1, 200 * time.Millisecond},
+		{2, 400 * time.Millisecond},
+		{6, 2 * RetryMax},
+		{10, 2 * RetryMax},
+	} {
+		got := backoff(RetryInitial, RetryMax, tc.attempt)
+		if got < 0 || got > tc.max {
+			t.Errorf("backoff(%d) = %s, want in [0, %s]", tc.attempt, got, tc.max)
+		}
+	}
+}
+
+func TestIsTransient(t *testing.T) {
+	if isTransient(nil) {
+		t.Error("nil should not be transient")
+	}
+	if !isTransient(ErrTransient) {
+		t.Error("ErrTransient should be transient")
+	}
+	if isTransient(ErrPermanent) {
+		t.Error("ErrPermanent should not be transient")
+	}
+	if !isTransient(errors.New("connection refused")) {
+		t.Error("connection refused should be transient")
+	}
+	if !isTransient(errors.New("i/o timeout")) {
+		t.Error("i/o timeout should be transient")
+	}
+	if isTransient(errors.New("some other error")) {
+		t.Error("unknown error should not be transient")
+	}
+}
+
+func TestClassifyDialErr(t *testing.T) {
+	if got := classifyDialErr(nil); got != nil {
+		t.Errorf("nil -> nil, got %v", got)
+	}
+	perm := classifyDialErr(errors.New("ssh: unable to authenticate"))
+	if !errors.Is(perm, ErrPermanent) {
+		t.Errorf("auth failure should be permanent, got %v", perm)
+	}
+	perm2 := classifyDialErr(errors.New("host key mismatch"))
+	if !errors.Is(perm2, ErrPermanent) {
+		t.Errorf("host key mismatch should be permanent, got %v", perm2)
+	}
+	trans := classifyDialErr(errors.New("connection refused"))
+	if !errors.Is(trans, ErrTransient) {
+		t.Errorf("connection refused should be transient, got %v", trans)
+	}
+}
+
+func TestClassifyExecErr(t *testing.T) {
+	exitErr := ssh.ExitError{}
+	perm := classifyExecErr(&exitErr)
+	if !errors.Is(perm, ErrPermanent) {
+		t.Errorf("ExitError should be permanent, got %v", perm)
+	}
+	trans := classifyExecErr(errors.New("session closed"))
+	if !errors.Is(trans, ErrTransient) {
+		t.Errorf("session closed should be transient, got %v", trans)
+	}
+}
+
+func TestShellQuote(t *testing.T) {
+	got := shellQuote("/etc/orca/foo.conf")
+	if got != "'/etc/orca/foo.conf'" {
+		t.Errorf("shellQuote = %q", got)
+	}
+	got = shellQuote("it's a path")
+	if got != "'it'\\''s a path'" {
+		t.Errorf("shellQuote with quote = %q", got)
+	}
+}
+
+func TestTransport_ExecSuccess_RealSSH(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	tr := realTransport(t, srv)
+	defer tr.Close()
+	out, err := tr.Exec(context.Background(), srv.addr(), "echo hello")
+	if err != nil {
+		t.Fatalf("Exec: %v", err)
+	}
+	if strings.TrimSpace(string(out)) != "hello" {
+		t.Errorf("out = %q, want hello", out)
+	}
+}
+
+func TestTransport_ExecRetry_TransientFailure(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	tr := realTransport(t, srv)
+	defer tr.Close()
+	var calls int32
+	tr.SetSessionFactory(func(c *ssh.Client) (sshSession, error) {
+		n := atomic.AddInt32(&calls, 1)
+		if n < 3 {
+			return &mockSession{err: errors.New("EOF")}, nil
+		}
+		return &mockSession{out: []byte("ok\n")}, nil
+	})
+	client, err := tr.dial(srv.addr())
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	tr.pool.Store(srv.addr(), client)
+	out, err := tr.Exec(context.Background(), srv.addr(), "echo ok")
+	if err != nil {
+		t.Fatalf("Exec: %v (calls=%d)", err, atomic.LoadInt32(&calls))
+	}
+	if string(out) != "ok\n" {
+		t.Errorf("out = %q, want ok\\n", out)
+	}
+	if got := atomic.LoadInt32(&calls); got < 3 {
+		t.Errorf("calls = %d, want >= 3 (retried)", got)
+	}
+}
+
+func TestTransport_ExecPermanentError_NoRetry(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	tr := realTransport(t, srv)
+	defer tr.Close()
+	var calls int32
+	tr.SetSessionFactory(func(c *ssh.Client) (sshSession, error) {
+		atomic.AddInt32(&calls, 1)
+		return &mockSession{err: &ssh.ExitError{}}, nil
+	})
+	client, err := tr.dial(srv.addr())
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	tr.pool.Store(srv.addr(), client)
+	_, err = tr.Exec(context.Background(), srv.addr(), "exit 1")
+	if err == nil {
+		t.Fatal("expected error, got nil")
+	}
+	if !errors.Is(err, ErrPermanent) {
+		t.Errorf("err should be permanent, got %v", err)
+	}
+	if got := atomic.LoadInt32(&calls); got != 1 {
+		t.Errorf("calls = %d, want 1 (no retry on permanent)", got)
+	}
+}
+
+func TestTransport_ExecTimeout(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	srv.setExecDelay(500 * time.Millisecond)
+	tr := realTransport(t, srv)
+	defer tr.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
+	defer cancel()
+	_, err := tr.Exec(ctx, srv.addr(), "echo hello")
+	if err == nil {
+		t.Fatal("expected timeout error, got nil")
+	}
+	if !errors.Is(err, context.DeadlineExceeded) && !errors.Is(err, ErrTransient) {
+		t.Errorf("err should be timeout/transient, got %v", err)
+	}
+}
+
+func TestTransport_ConnectionPoolReuse(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	tr := realTransport(t, srv)
+	defer tr.Close()
+	addr := srv.addr()
+	c1, err := tr.dial(addr)
+	if err != nil {
+		t.Fatalf("first dial: %v", err)
+	}
+	c2, err := tr.dial(addr)
+	if err != nil {
+		t.Fatalf("second dial: %v", err)
+	}
+	if c1 != c2 {
+		t.Error("pool did not reuse client for same peer")
+	}
+}
+
+func TestTransport_CloseClosesAllClients(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	tr := realTransport(t, srv)
+	if _, err := tr.dial(srv.addr()); err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	count := 0
+	tr.pool.Range(func(_, _ any) bool {
+		count++
+		return true
+	})
+	if count != 1 {
+		t.Fatalf("pool has %d entries, want 1", count)
+	}
+	if err := tr.Close(); err != nil {
+		t.Errorf("Close: %v", err)
+	}
+	_, err := tr.dial(srv.addr())
+	if !errors.Is(err, ErrPermanent) {
+		t.Errorf("dial after Close should be ErrPermanent, got %v", err)
+	}
+	if err := tr.Close(); err != nil {
+		t.Errorf("second Close: %v", err)
+	}
+}
+
+func TestTransport_ReadFile_RealSSH(t *testing.T) {
+	srv := newFakeSSHServer(t)
+	defer srv.close()
+	srv.mu.Lock()
+	srv.files["/etc/orca/test.conf"] = "content-line\n"
+	srv.mu.Unlock()
+	tr := realTransport(t, srv)
+	defer tr.Close()
+	out, err := tr.ReadFile(context.Background(), srv.addr(), "/etc/orca/test.conf")
+	if err != nil {
+		t.Fatalf("ReadFile: %v", err)
+	}
+	if string(out) != "content-line\n" {
+		t.Errorf("out = %q", out)
+	}
+}
+
+func TestTransport_DialKeyParseFailure(t *testing.T) {
+	dir := t.TempDir()
+	t.Setenv("ORCA_HOME", dir)
+	if err := os.WriteFile(filepath.Join(dir, "known_hosts"), []byte{}, 0o600); err != nil {
+		t.Fatalf("kh: %v", err)
+	}
+	keyPath := filepath.Join(dir, "bad_key")
+	if err := os.WriteFile(keyPath, []byte("not a key"), 0o600); err != nil {
+		t.Fatalf("write key: %v", err)
+	}
+	tr := NewTransport(keyPath, "")
+	tr.SetUser("root")
+	_, err := tr.dial("127.0.0.1:1")
+	if err == nil {
+		t.Fatal("expected parse error, got nil")
+	}
+	if !strings.Contains(err.Error(), "parse key") {
+		t.Errorf("err should mention parse key, got %v", err)
+	}
+}
+
+func TestTransport_DialKeyMissing(t *testing.T) {
+	dir := t.TempDir()
+	t.Setenv("ORCA_HOME", dir)
+	if err := os.WriteFile(filepath.Join(dir, "known_hosts"), []byte{}, 0o600); err != nil {
+		t.Fatalf("kh: %v", err)
+	}
+	tr := NewTransport(filepath.Join(dir, "missing_key"), "")
+	tr.SetUser("root")
+	_, err := tr.dial("127.0.0.1:1")
+	if err == nil {
+		t.Fatal("expected read error, got nil")
+	}
+	if !strings.Contains(err.Error(), "read key") {
+		t.Errorf("err should mention read key, got %v", err)
+	}
+}
+
+func TestTransport_DialMockFailure(t *testing.T) {
+	dir := t.TempDir()
+	t.Setenv("ORCA_HOME", dir)
+	if err := os.WriteFile(filepath.Join(dir, "known_hosts"), []byte{}, 0o600); err != nil {
+		t.Fatalf("kh: %v", err)
+	}
+	_, priv, _ := ed25519.GenerateKey(rand.Reader)
+	der, _ := x509.MarshalPKCS8PrivateKey(priv)
+	pemBytes := pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: der})
+	keyPath := filepath.Join(dir, "orca_ssh_key")
+	_ = os.WriteFile(keyPath, pemBytes, 0o600)
+	tr := NewTransport(keyPath, "")
+	tr.SetUser("root")
+	tr.SetDialer(&mockDialer{err: errors.New("connection refused")})
+	_, err := tr.dial("127.0.0.1:1")
+	if err == nil {
+		t.Fatal("expected dial error")
+	}
+	if !errors.Is(err, ErrTransient) {
+		t.Errorf("connection refused should be transient, got %v", err)
+	}
+}