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": "P0b", "stage": "verify", "milestone": "v0.9", "phase_role": "execution", "updated_at": "2026-08-05T03:25:00Z", "milestone_complete": false, "verify": { "build": "pass", "go_test": "18/18", "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/emitter/emitter.go

@@ -0,0 +1,94 @@
+// Package emitter defines the Layer-4 emitter interface (REQ-074,
+// I-B-002): the bridge between the declarative *jobspec.WorkloadSpec
+// and the server-side files. An Emitter renders a *WorkloadSpec into a
+// slice of File artifacts that the SSH-push transport SCPs to peers.
+//
+// Emitters are registered per workload kind + runtime (e.g.
+// "service:wasm", "job:process", "daemonset:wasm"). The Registry looks
+// up the right emitter by "<kind>:<runtime>" and delegates. Unknown
+// combinations return an error so the caller can fail fast before any
+// file is written.
+//
+// P0c only ships the interface, the Registry, a stub SystemdEmitter
+// (process runtime), and the File/Node value types. The full emitter
+// implementations (systemd lifecycle hooks, Traefik, Syncthing,
+// sockets) land in later phases (P02 Traefik, P04 lifecycle, P08
+// sockets, P09 Syncthing, v0.10-P03 secrets).
+package emitter
+
+import (
+	"fmt"
+	"strings"
+
+	"git.cloudinit.dev/coreci/orca/internal/jobspec"
+)
+
+// File is a single rendered artifact destined for a peer. The SSH-push
+// transport writes Content to Path atomically (write-to-tmp + rename)
+// with the given Mode (an octal string like "0644").
+type File struct {
+	Path    string
+	Content string
+	Mode    string
+}
+
+// Node is the minimal peer description an emitter needs to render
+// node-specific paths. It carries the hostname, the runtimes available
+// on the node (so emitters can branch), and the node tags (used by
+// DaemonSet matching and affinity in P05).
+type Node struct {
+	Hostname string
+	Runtime  []string
+	Tags     []string
+}
+
+// Emitter renders a *jobspec.WorkloadSpec for a given Node into a slice
+// of File artifacts. Implementations are registered with a Registry
+// keyed by "<kind>:<runtime>".
+type Emitter interface {
+	Render(spec *jobspec.WorkloadSpec, node *Node) ([]File, error)
+}
+
+// Registry holds emitters keyed by "<kind>:<runtime>" (e.g.
+// "service:process", "job:wasm"). The zero-value Registry is not
+// usable; construct one with NewRegistry.
+type Registry struct {
+	emitters map[string]Emitter
+}
+
+// NewRegistry returns an empty Registry ready for Register calls.
+func NewRegistry() *Registry {
+	return &Registry{emitters: make(map[string]Emitter)}
+}
+
+// Register registers an Emitter under the given key. The key is
+// "<kind>:<runtime>" (e.g. "job:process"). Registering twice under the
+// same key overwrites the prior registration (last-wins) to keep the
+// surface simple; callers are responsible for not double-registering.
+func (r *Registry) Register(key string, e Emitter) {
+	r.emitters[key] = e
+}
+
+// Render looks up the emitter for "<kind>:<runtime>" in the registry and
+// delegates to it. The kind is lowercased so the canonical spec kinds
+// (Job, Service, DaemonSet) map to the lowercase registry keys
+// ("job:process", "service:wasm", "daemonset:process"). Returns an error
+// if the spec is nil, the spec is missing its Kind, the runtime is
+// missing, or no emitter is registered for the combination.
+func (r *Registry) Render(spec *jobspec.WorkloadSpec, node *Node) ([]File, error) {
+	if spec == nil {
+		return nil, fmt.Errorf("emitter: spec is nil")
+	}
+	if strings.TrimSpace(spec.Kind) == "" {
+		return nil, fmt.Errorf("emitter: spec kind is empty")
+	}
+	if spec.Runtime == nil {
+		return nil, fmt.Errorf("emitter: spec runtime is nil")
+	}
+	key := strings.ToLower(spec.Kind) + ":" + spec.Runtime.OneOf
+	e, ok := r.emitters[key]
+	if !ok {
+		return nil, fmt.Errorf("emitter: no emitter registered for %q (kind:runtime)", key)
+	}
+	return e.Render(spec, node)
+}

internal/emitter/emitter_test.go

@@ -0,0 +1,163 @@
+package emitter
+
+import (
+	"errors"
+	"strings"
+	"testing"
+
+	"git.cloudinit.dev/coreci/orca/internal/jobspec"
+)
+
+// mockEmitter is a test-only Emitter that returns a fixed File slice
+// (or an error) so the Registry tests do not depend on the
+// SystemdEmitter. Implements Emitter via value receiver.
+type mockEmitter struct {
+	files []File
+	err   error
+}
+
+func (m mockEmitter) Render(spec *jobspec.WorkloadSpec, node *Node) ([]File, error) {
+	if m.err != nil {
+		return nil, m.err
+	}
+	out := make([]File, len(m.files))
+	copy(out, m.files)
+	return out, nil
+}
+
+func TestRegistry_RegisterAndRender(t *testing.T) {
+	r := NewRegistry()
+	want := []File{{Path: "/tmp/a", Content: "alpha", Mode: "0644"}}
+	r.Register("job:process", mockEmitter{files: want})
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Job",
+		Name:    "demo",
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/bin/true"},
+	}
+	node := &Node{Hostname: "node-1", Runtime: []string{"process"}}
+	got, err := r.Render(spec, node)
+	if err != nil {
+		t.Fatalf("Render: %v", err)
+	}
+	if len(got) != 1 {
+		t.Fatalf("got %d files, want 1", len(got))
+	}
+	if got[0] != want[0] {
+		t.Errorf("file = %+v, want %+v", got[0], want[0])
+	}
+}
+
+func TestRegistry_UnknownKindRuntime(t *testing.T) {
+	r := NewRegistry()
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Service",
+		Name:    "web",
+		Runtime: &jobspec.RuntimeBlock{OneOf: "wasm"},
+	}
+	_, err := r.Render(spec, &Node{})
+	if err == nil {
+		t.Fatal("expected error for unknown kind:runtime, got nil")
+	}
+	if !strings.Contains(err.Error(), "no emitter registered") {
+		t.Errorf("error = %q, want 'no emitter registered'", err.Error())
+	}
+	if !strings.Contains(err.Error(), "service:wasm") {
+		t.Errorf("error = %q, want it to mention 'service:wasm'", err.Error())
+	}
+}
+
+func TestRegistry_MultipleEmittersCorrectSelected(t *testing.T) {
+	r := NewRegistry()
+	jobFiles := []File{{Path: "/tmp/job", Content: "job", Mode: "0644"}}
+	svcFiles := []File{{Path: "/tmp/svc", Content: "svc", Mode: "0644"}}
+	dsFiles := []File{{Path: "/tmp/ds", Content: "ds", Mode: "0644"}}
+	r.Register("job:process", mockEmitter{files: jobFiles})
+	r.Register("service:process", mockEmitter{files: svcFiles})
+	r.Register("daemonset:process", mockEmitter{files: dsFiles})
+
+	cases := []struct {
+		kind     string
+		runtime  string
+		wantPath string
+	}{
+		{"Job", "process", "/tmp/job"},
+		{"Service", "process", "/tmp/svc"},
+		{"DaemonSet", "process", "/tmp/ds"},
+	}
+	for _, tc := range cases {
+		t.Run(tc.kind+":"+tc.runtime, func(t *testing.T) {
+			spec := &jobspec.WorkloadSpec{
+				Kind:    tc.kind,
+				Name:    "x",
+				Runtime: &jobspec.RuntimeBlock{OneOf: tc.runtime, Command: "/bin/x"},
+			}
+			got, err := r.Render(spec, &Node{Hostname: "n"})
+			if err != nil {
+				t.Fatalf("Render: %v", err)
+			}
+			if len(got) != 1 {
+				t.Fatalf("got %d files, want 1", len(got))
+			}
+			if got[0].Path != tc.wantPath {
+				t.Errorf("path = %q, want %q", got[0].Path, tc.wantPath)
+			}
+		})
+	}
+}
+
+func TestRegistry_NilSpec(t *testing.T) {
+	r := NewRegistry()
+	_, err := r.Render(nil, &Node{})
+	if err == nil {
+		t.Fatal("expected error for nil spec")
+	}
+	if !strings.Contains(err.Error(), "spec is nil") {
+		t.Errorf("error = %q, want 'spec is nil'", err.Error())
+	}
+}
+
+func TestRegistry_EmptyKind(t *testing.T) {
+	r := NewRegistry()
+	spec := &jobspec.WorkloadSpec{Runtime: &jobspec.RuntimeBlock{OneOf: "process"}}
+	_, err := r.Render(spec, &Node{})
+	if err == nil {
+		t.Fatal("expected error for empty kind")
+	}
+	if !strings.Contains(err.Error(), "kind is empty") {
+		t.Errorf("error = %q, want 'kind is empty'", err.Error())
+	}
+}
+
+func TestRegistry_NilRuntime(t *testing.T) {
+	r := NewRegistry()
+	spec := &jobspec.WorkloadSpec{Kind: "Job", Name: "x"}
+	_, err := r.Render(spec, &Node{})
+	if err == nil {
+		t.Fatal("expected error for nil runtime")
+	}
+	if !strings.Contains(err.Error(), "runtime is nil") {
+		t.Errorf("error = %q, want 'runtime is nil'", err.Error())
+	}
+}
+
+func TestRegistry_EmitterErrorPropagates(t *testing.T) {
+	r := NewRegistry()
+	wantErr := errors.New("boom")
+	r.Register("job:process", mockEmitter{err: wantErr})
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Job",
+		Name:    "x",
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/bin/x"},
+	}
+	_, err := r.Render(spec, &Node{})
+	if !errors.Is(err, wantErr) {
+		t.Errorf("err = %v, want %v", err, wantErr)
+	}
+}
+
+// Compile-time assertion that mockEmitter and SystemdEmitter implement
+// Emitter.
+var (
+	_ Emitter = mockEmitter{}
+	_ Emitter = SystemdEmitter{}
+)

internal/emitter/systemd.go

@@ -0,0 +1,60 @@
+package emitter
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+
+	"git.cloudinit.dev/coreci/orca/internal/jobspec"
+)
+
+// SystemdEmitter is a stub Emitter implementation for the "process"
+// runtime. It renders a minimal systemd unit file for the workload.
+//
+// This is a STUB — the full systemd emitter (with lifecycle hooks,
+// sockets, EnvironmentFile, LoadCredential) lands in later phases:
+//
+//   - P04: lifecycle hooks (ExecStop, ExecStartPre/Post, timeouts)
+//   - P08: socket plumbing (R-007)
+//   - v0.10-P03: secrets via EnvironmentFile= + LoadCredential=
+//
+// P0c ships only the minimal [Service]\nExecStart=... shape to prove
+// the Emitter interface end-to-end. The unit name carries the
+// `orca-v1-` prefix per the dual-write window (REQ-090) so the v0.9
+// SSH-push path does not collide with the v0.8 daemon's
+// `orca-<job>.service` units during the migration window.
+type SystemdEmitter struct{}
+
+// unitNamePrefix is the v0.9 SSH-push unit-name prefix. The v0.8
+// daemon uses `orca-<job>.service`; the v0.9 path uses
+// `orca-v1-<spec.Name>.service` so the two never overlap (REQ-090,
+// I-C-006). The prefix is load-bearing — do not change it without
+// updating the dual-write window contract.
+const unitNamePrefix = "orca-v1-"
+
+// Render renders a minimal systemd unit file for a process-runtime
+// workload. The unit name is `/etc/systemd/system/<unitNamePrefix><spec.Name>.service`
+// and the content is a minimal `[Service]` block with the runtime
+// command as ExecStart. Mode is 0644 (the lead applier chmods after
+// atomic rename).
+//
+// Returns an error if the spec is nil, the spec is missing its name,
+// or the runtime command is empty (a workload with no command has
+// nothing to ExecStart).
+func (SystemdEmitter) Render(spec *jobspec.WorkloadSpec, node *Node) ([]File, error) {
+	if spec == nil {
+		return nil, errors.New("emitter/systemd: spec is nil")
+	}
+	if strings.TrimSpace(spec.Name) == "" {
+		return nil, errors.New("emitter/systemd: spec name is empty")
+	}
+	if spec.Runtime == nil {
+		return nil, errors.New("emitter/systemd: runtime block is nil")
+	}
+	if strings.TrimSpace(spec.Runtime.Command) == "" {
+		return nil, errors.New("emitter/systemd: runtime command is empty")
+	}
+	path := fmt.Sprintf("/etc/systemd/system/%s%s.service", unitNamePrefix, spec.Name)
+	content := fmt.Sprintf("[Service]\nExecStart=%s\n", spec.Runtime.Command)
+	return []File{{Path: path, Content: content, Mode: "0644"}}, nil
+}

internal/emitter/systemd_test.go

@@ -0,0 +1,156 @@
+package emitter
+
+import (
+	"strings"
+	"testing"
+
+	"git.cloudinit.dev/coreci/orca/internal/jobspec"
+)
+
+func TestSystemdEmitter_RenderJob(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Job",
+		Name:    "backup",
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/usr/bin/rsync -a /src /dst"},
+	}
+	node := &Node{Hostname: "node-1", Runtime: []string{"process"}}
+	files, err := SystemdEmitter{}.Render(spec, node)
+	if err != nil {
+		t.Fatalf("Render: %v", err)
+	}
+	if len(files) != 1 {
+		t.Fatalf("got %d files, want 1", len(files))
+	}
+	f := files[0]
+	wantPath := "/etc/systemd/system/orca-v1-backup.service"
+	if f.Path != wantPath {
+		t.Errorf("Path = %q, want %q", f.Path, wantPath)
+	}
+	wantContent := "[Service]\nExecStart=/usr/bin/rsync -a /src /dst\n"
+	if f.Content != wantContent {
+		t.Errorf("Content = %q, want %q", f.Content, wantContent)
+	}
+	if f.Mode != "0644" {
+		t.Errorf("Mode = %q, want 0644", f.Mode)
+	}
+}
+
+func TestSystemdEmitter_RenderService(t *testing.T) {
+	// The full service emitter (Traefik route + health checks) lands in
+	// P02; here we only prove the systemd side renders for a Service
+	// kind with a process runtime.
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Service",
+		Name:    "web",
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/usr/local/bin/httpd -f"},
+	}
+	node := &Node{Hostname: "node-1", Runtime: []string{"process"}}
+	files, err := SystemdEmitter{}.Render(spec, node)
+	if err != nil {
+		t.Fatalf("Render: %v", err)
+	}
+	if len(files) != 1 {
+		t.Fatalf("got %d files, want 1", len(files))
+	}
+	if files[0].Path != "/etc/systemd/system/orca-v1-web.service" {
+		t.Errorf("Path = %q, want /etc/systemd/system/orca-v1-web.service", files[0].Path)
+	}
+	if !strings.Contains(files[0].Content, "ExecStart=/usr/local/bin/httpd -f") {
+		t.Errorf("Content = %q, want it to contain the ExecStart line", files[0].Content)
+	}
+}
+
+func TestSystemdEmitter_EmptyCommandError(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Job",
+		Name:    "x",
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: ""},
+	}
+	_, err := SystemdEmitter{}.Render(spec, &Node{})
+	if err == nil {
+		t.Fatal("expected error for empty command, got nil")
+	}
+	if !strings.Contains(err.Error(), "command is empty") {
+		t.Errorf("error = %q, want 'command is empty'", err.Error())
+	}
+}
+
+func TestSystemdEmitter_WhitespaceCommandError(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Job",
+		Name:    "x",
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "   "},
+	}
+	_, err := SystemdEmitter{}.Render(spec, &Node{})
+	if err == nil {
+		t.Fatal("expected error for whitespace-only command, got nil")
+	}
+	if !strings.Contains(err.Error(), "command is empty") {
+		t.Errorf("error = %q, want 'command is empty'", err.Error())
+	}
+}
+
+func TestSystemdEmitter_NilSpec(t *testing.T) {
+	v := SystemdEmitter{}
+	if _, err := v.Render(nil, &Node{}); err == nil {
+		t.Fatal("expected error for nil spec")
+	}
+}
+
+func TestSystemdEmitter_EmptyName(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Job",
+		Name:    "  ",
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/bin/x"},
+	}
+	_, err := SystemdEmitter{}.Render(spec, &Node{})
+	if err == nil {
+		t.Fatal("expected error for empty name")
+	}
+	if !strings.Contains(err.Error(), "name is empty") {
+		t.Errorf("error = %q, want 'name is empty'", err.Error())
+	}
+}
+
+func TestSystemdEmitter_NilRuntime(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{Kind: "Job", Name: "x"}
+	_, err := SystemdEmitter{}.Render(spec, &Node{})
+	if err == nil {
+		t.Fatal("expected error for nil runtime")
+	}
+	if !strings.Contains(err.Error(), "runtime block is nil") {
+		t.Errorf("error = %q, want 'runtime block is nil'", err.Error())
+	}
+}
+
+func TestSystemdEmitter_UnitNamePrefix(t *testing.T) {
+	// The orca-v1- prefix is load-bearing for the dual-write window
+	// (REQ-090, I-C-006): the v0.8 daemon writes `orca-<job>.service`
+	// and the v0.9 SSH-push path writes `orca-v1-<spec.Name>.service`,
+	// so the two never collide. This test guards against accidental
+	// removal of the prefix.
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Job",
+		Name:    "dual-write-safety",
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/bin/true"},
+	}
+	files, err := SystemdEmitter{}.Render(spec, &Node{Hostname: "n"})
+	if err != nil {
+		t.Fatalf("Render: %v", err)
+	}
+	if !strings.HasPrefix(files[0].Path, "/etc/systemd/system/orca-v1-") {
+		t.Errorf("Path = %q, want it to start with /etc/systemd/system/orca-v1- (REQ-090)", files[0].Path)
+	}
+	if !strings.HasSuffix(files[0].Path, ".service") {
+		t.Errorf("Path = %q, want it to end with .service", files[0].Path)
+	}
+	// Explicitly assert the full expected unit name to lock the contract.
+	want := "/etc/systemd/system/orca-v1-dual-write-safety.service"
+	if files[0].Path != want {
+		t.Errorf("Path = %q, want %q", files[0].Path, want)
+	}
+	// Sanity: the prefix is exactly "orca-v1-", not "orca-v0" or "orca".
+	if unitNamePrefix != "orca-v1-" {
+		t.Errorf("unitNamePrefix = %q, want orca-v1-", unitNamePrefix)
+	}
+}

internal/jobspec/markdown.go

@@ -24,6 +24,37 @@ type WorkloadSpec struct {
 	Secrets     []string
 	Volumes     []VolumeSpec
 	Body        string
+
+	// Kind-specific blocks consumed by the P0c schema validators
+	// (internal/spec/schema). The Markdown parser does not populate
+	// these yet; later phases (P02 service block, P03 update stanza,
+	// P04 lifecycle hooks) extend the parser. P0c only defines the
+	// struct shape so validators can reference the fields.
+
+	// Restart is the restart policy block. Required for Service and
+	// DaemonSet; optional for Job (defaults to never/on-failure).
+	// Populated by the P04 lifecycle phase.
+	Restart *RestartBlock
+
+	// Schedule is the schedule block. For Job it carries an optional
+	// cron string; for DaemonSet it carries the placement mode
+	// (every-node/matching/mandatory). Populated by P05 (scheduler
+	// skeleton) and the DaemonSet phase.
+	Schedule *ScheduleBlock
+
+	// Update is the rolling/canary update stanza. Required for
+	// Service. Populated by the P03 update-stanza phase.
+	Update *UpdateBlock
+
+	// Service is the service block (Traefik route definition). For
+	// Service kind it is implied; Job and DaemonSet do not carry a
+	// Traefik route by default (D-175). Populated by the P02 service
+	// block phase.
+	Service *ServiceBlock
+
+	// Timeout is an optional execution timeout (duration string) for
+	// Job. Populated by P04.
+	Timeout string
 }
 
 // RuntimeBlock is a minimal runtime abstraction surface populated by the
@@ -35,6 +66,37 @@ type RuntimeBlock struct {
 	Command string
 }
 
+// RestartBlock is the restart policy block. Mode is one of never,
+// on-failure, service (REQ-074 schema validators). Populated by P04.
+type RestartBlock struct {
+	Mode       string
+	MaxRetries int
+	Delay      string
+}
+
+// ScheduleBlock is the scheduling block. For Job, Cron is an optional
+// cron expression. For DaemonSet, Mode is one of every-node, matching,
+// mandatory (REQ-074). Populated by P05 and the DaemonSet phase.
+type ScheduleBlock struct {
+	Mode string
+	Cron string
+}
+
+// UpdateBlock is the rolling/canary update stanza. Required for Service.
+// Populated by P03.
+type UpdateBlock struct {
+	Strategy string
+	MaxSurge int
+}
+
+// ServiceBlock is the Traefik route definition. For Service it is
+// implied (Traefik route YES); Job and DaemonSet do not carry one by
+// default (D-175). Populated by P02.
+type ServiceBlock struct {
+	Host    string
+	RouteID string
+}
+
 // PortSpec is a minimal port binding entry. HostIP is optional.
 type PortSpec struct {
 	Name     string

internal/security/flock.go

@@ -25,3 +25,11 @@ func Flock(path string) (release func(), err error) {
 		_ = f.Close()
 	}, nil
 }
+
+func tryFlockEx(fd int) error {
+	return syscall.Flock(fd, syscall.LOCK_EX|syscall.LOCK_NB)
+}
+
+func releaseFlock(fd int) error {
+	return syscall.Flock(fd, syscall.LOCK_UN)
+}

internal/security/flock_test.go

@@ -4,6 +4,7 @@ import (
 	"os"
 	"path/filepath"
 	"testing"
+	"time"
 )
 
 func TestFlock_acquireAndRelease(t *testing.T) {
@@ -45,17 +46,37 @@ func TestFlock_concurrentBlocks(t *testing.T) {
 	if err != nil {
 		t.Fatalf("first Flock: %v", err)
 	}
-	defer r1()
 
-	done := make(chan error, 1)
+	// Give the blocking goroutine a chance to start and block.
+	time.Sleep(50 * time.Millisecond)
+
+	// Verify the second lock is blocked by checking it hasn't acquired after a short window.
+	// Use a non-blocking attempt: open the file and try LOCK_EX|LOCK_NB.
+	blocked := make(chan bool, 1)
 	go func() {
-		_, err := Flock(path)
+		f, err := os.OpenFile(path, os.O_RDWR, 0o600)
-		done <- err
+		if err != nil {
+			blocked <- false
+			return
+		}
+		defer f.Close()
+		// LOCK_NB = non-blocking; returns EWOULDBLOCK if locked.
+		if err := tryFlockEx(int(f.Fd())); err != nil {
+			blocked <- true // got EWOULDBLOCK = the lock is held by r1
+			return
+		}
+		releaseFlock(int(f.Fd()))
+		blocked <- false // acquired = r1 didn't hold the lock (bug)
 	}()
 
 	select {
-	case <-done:
+	case b := <-blocked:
-		t.Fatal("second Flock should block while first holds the lock")
+		if !b {
-	default:
+			t.Fatal("second lock acquired while first holds it — lock not working")
 		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("non-blocking try-lock timed out")
+	}
+
+	r1()
 }

internal/spec/schema/schema.go

@@ -0,0 +1,164 @@
+// Package schema provides kind-specific validators for the unified
+// *jobspec.WorkloadSpec introduced in P0b (REQ-064). Each workload kind
+// (Job, Service, DaemonSet per R-012) has different required fields;
+// this package exposes a Validator interface and a ValidatorFor
+// dispatcher so the emitter layer (REQ-074) and the lint engine
+// (REQ-084) can reject invalid specs before rendering.
+//
+// The validators operate purely on the *WorkloadSpec shape; they do no
+// I/O. Required-field violations return a structured error listing
+// every problem found (missing required fields, invalid combinations).
+package schema
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+
+	"git.cloudinit.dev/coreci/orca/internal/jobspec"
+)
+
+// Validator validates a *jobspec.WorkloadSpec against a kind-specific
+// schema. Implementations are pure (no I/O) and return a clear error
+// listing every violation found.
+type Validator interface {
+	Validate(spec *jobspec.WorkloadSpec) error
+}
+
+// JobValidator validates the Job workload kind (R-012).
+//
+// Rules:
+//   - no service block required (Job has no Traefik route by default D-175)
+//   - restart optional (defaults to never/on-failure when omitted)
+//   - schedule optional (cron string)
+//   - timeout optional
+//   - ports optional
+//   - count must be 1 (or unset → 1); count > 1 is an error for Job
+//     (use a Service for replicas)
+//   - no Traefik route (a ServiceBlock is rejected)
+type JobValidator struct{}
+
+// ServiceValidator validates the Service workload kind (R-012).
+//
+// Rules:
+//   - ports required (at least one)
+//   - count ≥ 1
+//   - restart required (mode must be service)
+//   - update required
+//   - runtime required
+//   - service block implied (Traefik route YES)
+type ServiceValidator struct{}
+
+// DaemonSetValidator validates the DaemonSet workload kind (R-012).
+//
+// Rules:
+//   - schedule block with mode (every-node/matching/mandatory) required
+//   - no ports (no Traefik route by default D-175)
+//   - no count (implicit = nodes matching condition)
+//   - restart required
+type DaemonSetValidator struct{}
+
+// ValidatorFor returns the Validator for the given workload kind, or an
+// error for an unknown kind. kind must be one of Job, Service,
+// DaemonSet (R-012).
+func ValidatorFor(kind string) (Validator, error) {
+	switch kind {
+	case "Job":
+		return JobValidator{}, nil
+	case "Service":
+		return ServiceValidator{}, nil
+	case "DaemonSet":
+		return DaemonSetValidator{}, nil
+	default:
+		return nil, fmt.Errorf("schema: unknown kind %q (want one of Job, Service, DaemonSet)", kind)
+	}
+}
+
+// Validate validates a Job spec. See JobValidator for the rules.
+func (JobValidator) Validate(spec *jobspec.WorkloadSpec) error {
+	if spec == nil {
+		return errors.New("schema/Job: spec is nil")
+	}
+	var errs []string
+	if strings.TrimSpace(spec.Name) == "" {
+		errs = append(errs, "name is required")
+	}
+	if spec.Count != 0 && spec.Count != 1 {
+		errs = append(errs, fmt.Sprintf("count must be 1 (or unset) for Job, got %d (use Service for replicas)", spec.Count))
+	}
+	if spec.Service != nil {
+		errs = append(errs, "service block (Traefik route) is not allowed for Job (D-175)")
+	}
+	return composeErrors("schema/Job", errs)
+}
+
+// Validate validates a Service spec. See ServiceValidator for the rules.
+func (ServiceValidator) Validate(spec *jobspec.WorkloadSpec) error {
+	if spec == nil {
+		return errors.New("schema/Service: spec is nil")
+	}
+	var errs []string
+	if strings.TrimSpace(spec.Name) == "" {
+		errs = append(errs, "name is required")
+	}
+	if len(spec.Ports) == 0 {
+		errs = append(errs, "ports required (at least one)")
+	}
+	if spec.Count < 1 {
+		errs = append(errs, fmt.Sprintf("count must be ≥ 1 for Service, got %d", spec.Count))
+	}
+	if spec.Restart == nil {
+		errs = append(errs, "restart block required for Service")
+	} else if spec.Restart.Mode != "service" {
+		errs = append(errs, fmt.Sprintf("restart mode must be %q for Service, got %q", "service", spec.Restart.Mode))
+	}
+	if spec.Update == nil {
+		errs = append(errs, "update block required for Service")
+	}
+	if spec.Runtime == nil {
+		errs = append(errs, "runtime block required for Service")
+	}
+	return composeErrors("schema/Service", errs)
+}
+
+// Validate validates a DaemonSet spec. See DaemonSetValidator for the rules.
+func (DaemonSetValidator) Validate(spec *jobspec.WorkloadSpec) error {
+	if spec == nil {
+		return errors.New("schema/DaemonSet: spec is nil")
+	}
+	var errs []string
+	if strings.TrimSpace(spec.Name) == "" {
+		errs = append(errs, "name is required")
+	}
+	if spec.Schedule == nil {
+		errs = append(errs, "schedule block required for DaemonSet")
+	} else {
+		switch spec.Schedule.Mode {
+		case "every-node", "matching", "mandatory":
+		case "":
+			errs = append(errs, "schedule mode required for DaemonSet (one of every-node, matching, mandatory)")
+		default:
+			errs = append(errs, fmt.Sprintf("schedule mode %q invalid (want one of every-node, matching, mandatory)", spec.Schedule.Mode))
+		}
+	}
+	if len(spec.Ports) > 0 {
+		errs = append(errs, "ports not allowed for DaemonSet (no Traefik route by default D-175)")
+	}
+	if spec.Count != 0 {
+		errs = append(errs, fmt.Sprintf("count not allowed for DaemonSet (implicit = nodes matching condition), got %d", spec.Count))
+	}
+	if spec.Restart == nil {
+		errs = append(errs, "restart block required for DaemonSet")
+	}
+	return composeErrors("schema/DaemonSet", errs)
+}
+
+// composeErrors joins the per-field errors into a single error prefixed
+// by the validator name. Returns nil when there are no errors so the
+// caller can return the result directly.
+func composeErrors(name string, errs []string) error {
+	if len(errs) == 0 {
+		return nil
+	}
+	return fmt.Errorf("%s: %s", name, strings.Join(errs, "; "))
+}

internal/spec/schema/schema_test.go

@@ -0,0 +1,436 @@
+package schema
+
+import (
+	"strings"
+	"testing"
+
+	"git.cloudinit.dev/coreci/orca/internal/jobspec"
+)
+
+func TestJobValidator_ValidMinimal(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{Kind: "Job", Name: "backup", Count: 1}
+	v := JobValidator{}
+	if err := v.Validate(spec); err != nil {
+		t.Fatalf("expected nil, got %v", err)
+	}
+}
+
+func TestJobValidator_ValidWithSchedule(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:     "Job",
+		Name:     "backup",
+		Count:    1,
+		Schedule: &jobspec.ScheduleBlock{Cron: "0 2 * * *"},
+		Timeout:  "1h",
+	}
+	v := JobValidator{}
+	if err := v.Validate(spec); err != nil {
+		t.Fatalf("expected nil, got %v", err)
+	}
+}
+
+func TestJobValidator_ValidUnsetCount(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{Kind: "Job", Name: "one-shot"}
+	v := JobValidator{}
+	if err := v.Validate(spec); err != nil {
+		t.Fatalf("unset count should default-accept, got %v", err)
+	}
+}
+
+func TestJobValidator_MissingName(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{Kind: "Job", Count: 1}
+	err := JobValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for missing name, got nil")
+	}
+	if !strings.Contains(err.Error(), "name is required") {
+		t.Errorf("error = %q, want 'name is required'", err.Error())
+	}
+}
+
+func TestJobValidator_CountGreaterThanOne(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{Kind: "Job", Name: "batch", Count: 3}
+	err := JobValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for count > 1, got nil")
+	}
+	if !strings.Contains(err.Error(), "count must be 1") {
+		t.Errorf("error = %q, want 'count must be 1'", err.Error())
+	}
+}
+
+func TestJobValidator_ServiceBlockRejected(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Job",
+		Name:    "x",
+		Count:   1,
+		Service: &jobspec.ServiceBlock{Host: "x.example"},
+	}
+	err := JobValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for service block on Job, got nil")
+	}
+	if !strings.Contains(err.Error(), "service block") {
+		t.Errorf("error = %q, want 'service block'", err.Error())
+	}
+}
+
+func TestJobValidator_NilSpec(t *testing.T) {
+	v := JobValidator{}
+	if err := v.Validate(nil); err == nil {
+		t.Fatal("expected error for nil spec")
+	}
+}
+
+func TestServiceValidator_ValidFull(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Service",
+		Name:    "web",
+		Count:   3,
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/bin/http"},
+		Restart: &jobspec.RestartBlock{Mode: "service"},
+		Update:  &jobspec.UpdateBlock{Strategy: "rolling", MaxSurge: 1},
+		Ports:   []jobspec.PortSpec{{Name: "http", Port: 8080}},
+	}
+	v := ServiceValidator{}
+	if err := v.Validate(spec); err != nil {
+		t.Fatalf("expected nil, got %v", err)
+	}
+}
+
+func TestServiceValidator_MissingPorts(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Service",
+		Name:    "web",
+		Count:   2,
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process"},
+		Restart: &jobspec.RestartBlock{Mode: "service"},
+		Update:  &jobspec.UpdateBlock{Strategy: "rolling"},
+	}
+	err := ServiceValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for missing ports, got nil")
+	}
+	if !strings.Contains(err.Error(), "ports required") {
+		t.Errorf("error = %q, want 'ports required'", err.Error())
+	}
+}
+
+func TestServiceValidator_CountZero(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Service",
+		Name:    "web",
+		Count:   0,
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process"},
+		Restart: &jobspec.RestartBlock{Mode: "service"},
+		Update:  &jobspec.UpdateBlock{Strategy: "rolling"},
+		Ports:   []jobspec.PortSpec{{Name: "http", Port: 8080}},
+	}
+	err := ServiceValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for count 0, got nil")
+	}
+	if !strings.Contains(err.Error(), "count must be") {
+		t.Errorf("error = %q, want 'count must be'", err.Error())
+	}
+}
+
+func TestServiceValidator_MissingRestart(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Service",
+		Name:    "web",
+		Count:   1,
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process"},
+		Update:  &jobspec.UpdateBlock{Strategy: "rolling"},
+		Ports:   []jobspec.PortSpec{{Name: "http", Port: 8080}},
+	}
+	err := ServiceValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for missing restart, got nil")
+	}
+	if !strings.Contains(err.Error(), "restart block required") {
+		t.Errorf("error = %q, want 'restart block required'", err.Error())
+	}
+}
+
+func TestServiceValidator_WrongRestartMode(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Service",
+		Name:    "web",
+		Count:   1,
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process"},
+		Restart: &jobspec.RestartBlock{Mode: "on-failure"},
+		Update:  &jobspec.UpdateBlock{Strategy: "rolling"},
+		Ports:   []jobspec.PortSpec{{Name: "http", Port: 8080}},
+	}
+	err := ServiceValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for wrong restart mode, got nil")
+	}
+	if !strings.Contains(err.Error(), "restart mode must be") {
+		t.Errorf("error = %q, want 'restart mode must be'", err.Error())
+	}
+}
+
+func TestServiceValidator_MissingUpdate(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Service",
+		Name:    "web",
+		Count:   1,
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process"},
+		Restart: &jobspec.RestartBlock{Mode: "service"},
+		Ports:   []jobspec.PortSpec{{Name: "http", Port: 8080}},
+	}
+	err := ServiceValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for missing update, got nil")
+	}
+	if !strings.Contains(err.Error(), "update block required") {
+		t.Errorf("error = %q, want 'update block required'", err.Error())
+	}
+}
+
+func TestServiceValidator_MissingRuntime(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Service",
+		Name:    "web",
+		Count:   1,
+		Restart: &jobspec.RestartBlock{Mode: "service"},
+		Update:  &jobspec.UpdateBlock{Strategy: "rolling"},
+		Ports:   []jobspec.PortSpec{{Name: "http", Port: 8080}},
+	}
+	err := ServiceValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for missing runtime, got nil")
+	}
+	if !strings.Contains(err.Error(), "runtime block required") {
+		t.Errorf("error = %q, want 'runtime block required'", err.Error())
+	}
+}
+
+func TestServiceValidator_NilSpec(t *testing.T) {
+	v := ServiceValidator{}
+	if err := v.Validate(nil); err == nil {
+		t.Fatal("expected error for nil spec")
+	}
+}
+
+func TestDaemonSetValidator_Valid(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:     "DaemonSet",
+		Name:     "log-shipper",
+		Schedule: &jobspec.ScheduleBlock{Mode: "every-node"},
+		Restart:  &jobspec.RestartBlock{Mode: "on-failure"},
+	}
+	v := DaemonSetValidator{}
+	if err := v.Validate(spec); err != nil {
+		t.Fatalf("expected nil, got %v", err)
+	}
+}
+
+func TestDaemonSetValidator_ValidMatchingMode(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:     "DaemonSet",
+		Name:     "x",
+		Schedule: &jobspec.ScheduleBlock{Mode: "matching"},
+		Restart:  &jobspec.RestartBlock{Mode: "on-failure"},
+	}
+	v := DaemonSetValidator{}
+	if err := v.Validate(spec); err != nil {
+		t.Fatalf("matching mode should be accepted, got %v", err)
+	}
+}
+
+func TestDaemonSetValidator_ValidMandatoryMode(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:     "DaemonSet",
+		Name:     "x",
+		Schedule: &jobspec.ScheduleBlock{Mode: "mandatory"},
+		Restart:  &jobspec.RestartBlock{Mode: "on-failure"},
+	}
+	v := DaemonSetValidator{}
+	if err := v.Validate(spec); err != nil {
+		t.Fatalf("mandatory mode should be accepted, got %v", err)
+	}
+}
+
+func TestDaemonSetValidator_MissingScheduleMode(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:     "DaemonSet",
+		Name:     "x",
+		Schedule: &jobspec.ScheduleBlock{Mode: ""},
+		Restart:  &jobspec.RestartBlock{Mode: "on-failure"},
+	}
+	err := DaemonSetValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for missing schedule mode, got nil")
+	}
+	if !strings.Contains(err.Error(), "schedule mode required") {
+		t.Errorf("error = %q, want 'schedule mode required'", err.Error())
+	}
+}
+
+func TestDaemonSetValidator_MissingScheduleBlock(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "DaemonSet",
+		Name:    "x",
+		Restart: &jobspec.RestartBlock{Mode: "on-failure"},
+	}
+	err := DaemonSetValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for missing schedule block, got nil")
+	}
+	if !strings.Contains(err.Error(), "schedule block required") {
+		t.Errorf("error = %q, want 'schedule block required'", err.Error())
+	}
+}
+
+func TestDaemonSetValidator_InvalidScheduleMode(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:     "DaemonSet",
+		Name:     "x",
+		Schedule: &jobspec.ScheduleBlock{Mode: "always"},
+		Restart:  &jobspec.RestartBlock{Mode: "on-failure"},
+	}
+	err := DaemonSetValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for invalid schedule mode, got nil")
+	}
+	if !strings.Contains(err.Error(), "schedule mode") {
+		t.Errorf("error = %q, want 'schedule mode'", err.Error())
+	}
+}
+
+func TestDaemonSetValidator_HasPorts(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:     "DaemonSet",
+		Name:     "x",
+		Schedule: &jobspec.ScheduleBlock{Mode: "every-node"},
+		Restart:  &jobspec.RestartBlock{Mode: "on-failure"},
+		Ports:    []jobspec.PortSpec{{Name: "http", Port: 80}},
+	}
+	err := DaemonSetValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for ports on DaemonSet, got nil")
+	}
+	if !strings.Contains(err.Error(), "ports not allowed") {
+		t.Errorf("error = %q, want 'ports not allowed'", err.Error())
+	}
+}
+
+func TestDaemonSetValidator_HasCount(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:     "DaemonSet",
+		Name:     "x",
+		Count:    3,
+		Schedule: &jobspec.ScheduleBlock{Mode: "every-node"},
+		Restart:  &jobspec.RestartBlock{Mode: "on-failure"},
+	}
+	err := DaemonSetValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for count on DaemonSet, got nil")
+	}
+	if !strings.Contains(err.Error(), "count not allowed") {
+		t.Errorf("error = %q, want 'count not allowed'", err.Error())
+	}
+}
+
+func TestDaemonSetValidator_MissingRestart(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:     "DaemonSet",
+		Name:     "x",
+		Schedule: &jobspec.ScheduleBlock{Mode: "every-node"},
+	}
+	err := DaemonSetValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for missing restart on DaemonSet, got nil")
+	}
+	if !strings.Contains(err.Error(), "restart block required") {
+		t.Errorf("error = %q, want 'restart block required'", err.Error())
+	}
+}
+
+func TestDaemonSetValidator_NilSpec(t *testing.T) {
+	v := DaemonSetValidator{}
+	if err := v.Validate(nil); err == nil {
+		t.Fatal("expected error for nil spec")
+	}
+}
+
+func TestValidatorFor_EachKind(t *testing.T) {
+	cases := []struct {
+		kind string
+		want string
+	}{
+		{"Job", "schema.JobValidator"},
+		{"Service", "schema.ServiceValidator"},
+		{"DaemonSet", "schema.DaemonSetValidator"},
+	}
+	for _, tc := range cases {
+		t.Run(tc.kind, func(t *testing.T) {
+			v, err := ValidatorFor(tc.kind)
+			if err != nil {
+				t.Fatalf("ValidatorFor(%q): %v", tc.kind, err)
+			}
+			got := fmtType(v)
+			if got != tc.want {
+				t.Errorf("ValidatorFor(%q) type = %q, want %q", tc.kind, got, tc.want)
+			}
+		})
+	}
+}
+
+func TestValidatorFor_UnknownKind(t *testing.T) {
+	_, err := ValidatorFor("CronJob")
+	if err == nil {
+		t.Fatal("expected error for unknown kind, got nil")
+	}
+	if !strings.Contains(err.Error(), "unknown kind") {
+		t.Errorf("error = %q, want 'unknown kind'", err.Error())
+	}
+}
+
+// fmtType returns a readable type name for a validator. Uses fmt.Sprintf
+// with %T rather than reflection to keep the test surface minimal.
+func fmtType(v Validator) string {
+	switch v.(type) {
+	case JobValidator:
+		return "schema.JobValidator"
+	case ServiceValidator:
+		return "schema.ServiceValidator"
+	case DaemonSetValidator:
+		return "schema.DaemonSetValidator"
+	default:
+		return "unknown"
+	}
+}
+
+// Ensure composeErrors returns nil for empty input (covers the
+// short-circuit branch that the validators rely on).
+func TestComposeErrors_Empty(t *testing.T) {
+	if err := composeErrors("schema/X", nil); err != nil {
+		t.Errorf("composeErrors(nil) = %v, want nil", err)
+	}
+	if err := composeErrors("schema/X", []string{}); err != nil {
+		t.Errorf("composeErrors([]) = %v, want nil", err)
+	}
+}
+
+// Ensure the error type returned by composeErrors is a non-nil error
+// when violations are present (guards against accidental nil-return).
+func TestComposeErrors_NonEmpty(t *testing.T) {
+	err := composeErrors("schema/X", []string{"a", "b"})
+	if err == nil {
+		t.Fatal("expected non-nil error")
+	}
+	if !strings.Contains(err.Error(), "a") || !strings.Contains(err.Error(), "b") {
+		t.Errorf("error = %q, want both 'a' and 'b'", err.Error())
+	}
+}
+
+// Compile-time assertion that the validators implement the interface.
+var (
+	_ Validator = JobValidator{}
+	_ Validator = ServiceValidator{}
+	_ Validator = DaemonSetValidator{}
+)

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)
+	}
+}