refactor(e2e): hoist generic DDIL harness primitives to shared tests/e2e/harness

Step 1.5 of docs/development/improvements/e2e-go-harness.md. Promotes
Cluster/Node/SSH/DaemonClient/LocalStateRevision/ResetAllNodes/Witness
into the shared package so the LB, multinode, and BM scenarios in
steps 4-5 can build on the same primitives instead of duplicating SSH
+ cluster plumbing.

Moves
- ddil/harness/{cluster,ssh,daemon_client,state,reset,witness}.go
  -> harness/ (plus the witness testdata embed dir).
- ddil/harness/{fault,netem,profiles,snapshot}.go -> ddil/fault/ as a
  new package; fault.PartitionNode/HealNode/KillNATS/StartNATS/
  RestartDaemonOnly/ApplyNetem/ClearNetem/LinkProfile*/TakeSnapshot
  now take harness.Cluster / harness.Node / harness.SSH explicitly.
- ddil/harness/retry.go::Run retry-with-reset extracted as
  harness.RetryWithReset in harness/wait.go; ddil/harness/quarantine.go
  keeps the DDIL-only quarantine wrapper that calls into it.
- WaitForMode moved from fault.go to harness/daemon_client.go since
  it polls daemon state and has no fault-injection concern.
- The pre-hoist harness/ssh.go (PeerSSH, subprocess ssh) is preserved
  as harness/peer_ssh.go so the lb_test suite keeps working unchanged;
  the typed x/crypto/ssh client from DDIL now owns harness/ssh.go.
- New harness/util.go exports ShellQuote (was unexported shellQuote in
  netem.go) so both ddil/fault and reset.go/witness.go can use it.

Env var rename
- SPINIFEX_NODES / SPINIFEX_SSH_USER / SPINIFEX_SSH_KEY replace
  DDIL_NODES / DDIL_SSH_USER / DDIL_SSH_KEY. Old names still honoured
  with a stderr deprecation note (envWithLegacy) so the DDIL CI keeps
  passing until .github/workflows/e2e-ddil.yml drops the old names.
  DDIL_QUARANTINED stays DDIL-specific.

DDIL parity preserved — scenarios still call ddilh.Run for quarantine
semantics; the underlying retry-with-reset is the same code path,
just shared.

Closes mulga-siv-74.

tests/e2e/ddil/fault/fault.go

@@ -1,12 +1,18 @@
 //go:build e2e
 
-package harness
+// Package fault provides DDIL-only fault-injection helpers — iptables-based
+// peer partitioning, NATS/daemon lifecycle toggles, netem link-condition
+// shaping, and instance-state snapshots used to assert that nothing rewound
+// across a partition/heal cycle. Generic cluster/SSH/daemon primitives live
+// in github.com/mulgadc/spinifex/tests/e2e/harness.
+package fault
 
 import (
 	"context"
 	"fmt"
 	"strings"
-	"time"
+
+	"github.com/mulgadc/spinifex/tests/e2e/harness"
 )
 
 // PartitionNode isolates target from peers by installing per-peer iptables
@@ -20,28 +26,28 @@ import (
 // composes with any prior non-DDIL rules. HealNode's flush is still the
 // correct teardown because the harness owns the target's firewall during a
 // scenario.
-func PartitionNode(ctx context.Context, ssh SSH, target Node, peers []Node) error {
+func PartitionNode(ctx context.Context, ssh harness.SSH, target harness.Node, peers []harness.Node) error {
 	if len(peers) == 0 {
-		return fmt.Errorf("ddil harness: PartitionNode %s: no peers supplied", target.Name)
+		return fmt.Errorf("ddil fault: PartitionNode %s: no peers supplied", target.Name)
 	}
 
 	var lines []string
 	for _, p := range peers {
 		lines = append(lines,
-			fmt.Sprintf("sudo iptables -I INPUT -s %s -j DROP", shellQuote(p.Addr)),
-			fmt.Sprintf("sudo iptables -I OUTPUT -d %s -j DROP", shellQuote(p.Addr)),
+			fmt.Sprintf("sudo iptables -I INPUT -s %s -j DROP", harness.ShellQuote(p.Addr)),
+			fmt.Sprintf("sudo iptables -I OUTPUT -d %s -j DROP", harness.ShellQuote(p.Addr)),
 		)
 	}
 	cmd := strings.Join(lines, " && ")
 	if _, err := ssh.Run(ctx, target, cmd); err != nil {
-		return fmt.Errorf("ddil harness: partition %s: %w", target.Name, err)
+		return fmt.Errorf("ddil fault: partition %s: %w", target.Name, err)
 	}
 
 	// Sanity-check: if our SSH path ran through a peer IP, the rules we just
 	// installed would have severed it. Fail loudly so a future infra change
 	// (e.g. orchestrator moved onto the cluster network) surfaces immediately.
 	if _, err := ssh.Run(ctx, target, "echo ok"); err != nil {
-		return fmt.Errorf("ddil harness: partition %s severed orchestrator SSH "+
+		return fmt.Errorf("ddil fault: partition %s severed orchestrator SSH "+
 			"(orchestrator may share peer IPs): %w", target.Name, err)
 	}
 	return nil
@@ -49,9 +55,9 @@ func PartitionNode(ctx context.Context, ssh SSH, target Node, peers []Node) erro
 
 // HealNode flushes and deletes all iptables rules on target, reversing a
 // prior PartitionNode. Idempotent: safe on a node with no rules installed.
-func HealNode(ctx context.Context, ssh SSH, target Node) error {
+func HealNode(ctx context.Context, ssh harness.SSH, target harness.Node) error {
 	if _, err := ssh.Run(ctx, target, "sudo iptables -F && sudo iptables -X"); err != nil {
-		return fmt.Errorf("ddil harness: heal %s: %w", target.Name, err)
+		return fmt.Errorf("ddil fault: heal %s: %w", target.Name, err)
 	}
 	return nil
 }
@@ -59,17 +65,17 @@ func HealNode(ctx context.Context, ssh SSH, target Node) error {
 // KillNATS stops spinifex-nats on node without touching spinifex-daemon.
 // This is the scenario shape the current happy-path suite cannot express —
 // `systemctl stop spinifex.target` brings both down together.
-func KillNATS(ctx context.Context, ssh SSH, node Node) error {
+func KillNATS(ctx context.Context, ssh harness.SSH, node harness.Node) error {
 	if _, err := ssh.Run(ctx, node, "sudo systemctl stop spinifex-nats"); err != nil {
-		return fmt.Errorf("ddil harness: kill nats on %s: %w", node.Name, err)
+		return fmt.Errorf("ddil fault: kill nats on %s: %w", node.Name, err)
 	}
 	return nil
 }
 
 // StartNATS starts spinifex-nats on node. Paired with KillNATS.
-func StartNATS(ctx context.Context, ssh SSH, node Node) error {
+func StartNATS(ctx context.Context, ssh harness.SSH, node harness.Node) error {
 	if _, err := ssh.Run(ctx, node, "sudo systemctl start spinifex-nats"); err != nil {
-		return fmt.Errorf("ddil harness: start nats on %s: %w", node.Name, err)
+		return fmt.Errorf("ddil fault: start nats on %s: %w", node.Name, err)
 	}
 	return nil
 }
@@ -77,48 +83,9 @@ func StartNATS(ctx context.Context, ssh SSH, node Node) error {
 // RestartDaemonOnly restarts spinifex-daemon without touching spinifex-nats.
 // Used by Scenario B to exercise the daemon-without-NATS startup path
 // (daemon-local-autonomy §1d).
-func RestartDaemonOnly(ctx context.Context, ssh SSH, node Node) error {
+func RestartDaemonOnly(ctx context.Context, ssh harness.SSH, node harness.Node) error {
 	if _, err := ssh.Run(ctx, node, "sudo systemctl restart spinifex-daemon"); err != nil {
-		return fmt.Errorf("ddil harness: restart daemon on %s: %w", node.Name, err)
+		return fmt.Errorf("ddil fault: restart daemon on %s: %w", node.Name, err)
 	}
 	return nil
 }
-
-// WaitForMode polls the daemon's /local/status until it reports the expected
-// mode or timeout expires. Poll interval is 1s — fast enough that a
-// 30-second timeout sees ~30 attempts, slow enough not to flood a recovering
-// daemon.
-//
-// Depends on daemon-local-autonomy §1b. Until that endpoint ships, this
-// function will time out; callers gate on t.Skip.
-func WaitForMode(ctx context.Context, dc *DaemonClient, node Node, want DaemonMode, timeout time.Duration) error {
-	deadline := time.Now().Add(timeout)
-	const interval = 1 * time.Second
-
-	var lastErr error
-	for {
-		if ctx.Err() != nil {
-			return ctx.Err()
-		}
-		status, err := dc.Status(ctx, node)
-		if err == nil && status.Mode == want {
-			return nil
-		}
-		lastErr = err
-
-		if time.Now().After(deadline) {
-			if lastErr != nil {
-				return fmt.Errorf("ddil harness: wait for mode %s on %s: timed out after %s: last error: %w",
-					want, node.Name, timeout, lastErr)
-			}
-			return fmt.Errorf("ddil harness: wait for mode %s on %s: timed out after %s (still reporting another mode)",
-				want, node.Name, timeout)
-		}
-
-		select {
-		case <-ctx.Done():
-			return ctx.Err()
-		case <-time.After(interval):
-		}
-	}
-}

tests/e2e/ddil/fault/netem.go

@@ -1,11 +1,13 @@
 //go:build e2e
 
-package harness
+package fault
 
 import (
 	"context"
 	"fmt"
 	"strings"
+
+	"github.com/mulgadc/spinifex/tests/e2e/harness"
 )
 
 // ApplyNetem installs a tc netem qdisc on iface of node matching profile.
@@ -14,9 +16,9 @@ import (
 // Flapping profiles have no static tc expression — they are driven by a
 // separate orchestrator (not yet implemented). ApplyNetem rejects them with
 // a clear error so callers don't silently apply a no-op.
-func ApplyNetem(ctx context.Context, ssh SSH, node Node, iface string, profile LinkProfile) error {
+func ApplyNetem(ctx context.Context, ssh harness.SSH, node harness.Node, iface string, profile LinkProfile) error {
 	if profile.Flapping != nil {
-		return fmt.Errorf("ddil harness: profile %q is flap-driven; use a flap orchestrator, not ApplyNetem", profile.Name)
+		return fmt.Errorf("ddil fault: profile %q is flap-driven; use a flap orchestrator, not ApplyNetem", profile.Name)
 	}
 
 	var netemArgs []string
@@ -34,15 +36,15 @@ func ApplyNetem(ctx context.Context, ssh SSH, node Node, iface string, profile L
 		netemArgs = append(netemArgs, "rate", profile.Bandwidth)
 	}
 	if len(netemArgs) == 0 {
-		return fmt.Errorf("ddil harness: profile %q has no netem parameters", profile.Name)
+		return fmt.Errorf("ddil fault: profile %q has no netem parameters", profile.Name)
 	}
 
 	// `tc qdisc replace` is idempotent — it adds if absent, replaces if
 	// present — which matches ApplyNetem's contract.
 	cmd := fmt.Sprintf("sudo tc qdisc replace dev %s root netem %s",
-		shellQuote(iface), strings.Join(netemArgs, " "))
+		harness.ShellQuote(iface), strings.Join(netemArgs, " "))
 	if _, err := ssh.Run(ctx, node, cmd); err != nil {
-		return fmt.Errorf("ddil harness: apply netem %s on %s (%s): %w",
+		return fmt.Errorf("ddil fault: apply netem %s on %s (%s): %w",
 			profile.Name, node.Name, iface, err)
 	}
 	return nil
@@ -50,17 +52,10 @@ func ApplyNetem(ctx context.Context, ssh SSH, node Node, iface string, profile L
 
 // ClearNetem removes any root qdisc on iface. Safe to call when no qdisc is
 // present — tc's non-zero exit is swallowed on the remote side.
-func ClearNetem(ctx context.Context, ssh SSH, node Node, iface string) error {
-	cmd := fmt.Sprintf("sudo tc qdisc del dev %s root 2>/dev/null || true", shellQuote(iface))
+func ClearNetem(ctx context.Context, ssh harness.SSH, node harness.Node, iface string) error {
+	cmd := fmt.Sprintf("sudo tc qdisc del dev %s root 2>/dev/null || true", harness.ShellQuote(iface))
 	if _, err := ssh.Run(ctx, node, cmd); err != nil {
-		return fmt.Errorf("ddil harness: clear netem on %s (%s): %w", node.Name, iface, err)
+		return fmt.Errorf("ddil fault: clear netem on %s (%s): %w", node.Name, iface, err)
 	}
 	return nil
 }
-
-// shellQuote wraps s in single quotes and escapes embedded single quotes, so
-// that shell metacharacters in interface names or peer addresses are treated
-// as literals on the remote shell.
-func shellQuote(s string) string {
-	return "'" + strings.ReplaceAll(s, "'", `'\''`) + "'"
-}

tests/e2e/ddil/fault/profiles.go

@@ -1,6 +1,6 @@
 //go:build e2e
 
-package harness
+package fault
 
 import "time"
 

tests/e2e/ddil/fault/snapshot.go

@@ -1,11 +1,13 @@
 //go:build e2e
 
-package harness
+package fault
 
 import (
 	"context"
 	"fmt"
 	"testing"
+
+	"github.com/mulgadc/spinifex/tests/e2e/harness"
 )
 
 // InstanceSnapshot captures a point-in-time view of a single daemon's locally
@@ -13,11 +15,11 @@ import (
 // second snapshot after the cluster heals, then call AssertPreserved to prove
 // the daemon did not lose or regress any instance it had been tracking.
 //
-// Reuses the LocalInstance type from daemon_client.go. Once daemon-local-
-// autonomy §1a lands, that placeholder will be replaced by the daemon's
-// authoritative struct and this alias will pick up the real schema without
-// scenario changes.
-type InstanceSnapshot []LocalInstance
+// Reuses the LocalInstance type from harness/daemon_client.go. Once
+// daemon-local-autonomy §1a lands, that placeholder will be replaced by the
+// daemon's authoritative struct and this alias will pick up the real schema
+// without scenario changes.
+type InstanceSnapshot []harness.LocalInstance
 
 // TakeSnapshot reads /local/instances on the given node and returns the
 // response as an InstanceSnapshot.
@@ -25,10 +27,10 @@ type InstanceSnapshot []LocalInstance
 // The DaemonClient is passed explicitly (rather than through a package-level
 // singleton) so each scenario owns its connection reuse and so dry-run
 // scenarios can stub with a fake client once the endpoint exists.
-func TakeSnapshot(ctx context.Context, d *DaemonClient, node Node) (InstanceSnapshot, error) {
+func TakeSnapshot(ctx context.Context, d *harness.DaemonClient, node harness.Node) (InstanceSnapshot, error) {
 	xs, err := d.Instances(ctx, node)
 	if err != nil {
-		return nil, fmt.Errorf("ddil harness: snapshot %s: %w", node.Name, err)
+		return nil, fmt.Errorf("ddil fault: snapshot %s: %w", node.Name, err)
 	}
 	return InstanceSnapshot(xs), nil
 }
@@ -46,7 +48,7 @@ func TakeSnapshot(ctx context.Context, d *DaemonClient, node Node) (InstanceSnap
 func (pre InstanceSnapshot) AssertPreserved(t *testing.T, post InstanceSnapshot) {
 	t.Helper()
 
-	postByID := make(map[string]LocalInstance, len(post))
+	postByID := make(map[string]harness.LocalInstance, len(post))
 	for _, p := range post {
 		postByID[p.InstanceID] = p
 	}
@@ -68,10 +70,10 @@ func (pre InstanceSnapshot) AssertPreserved(t *testing.T, post InstanceSnapshot)
 		return
 	}
 	if len(missing) > 0 {
-		t.Errorf("ddil harness: instances disappeared from snapshot: %v", missing)
+		t.Errorf("ddil fault: instances disappeared from snapshot: %v", missing)
 	}
 	if len(regressed) > 0 {
-		t.Errorf("ddil harness: instances regressed to a terminal state: %v", regressed)
+		t.Errorf("ddil fault: instances regressed to a terminal state: %v", regressed)
 	}
 	t.FailNow()
 }

tests/e2e/ddil/harness/quarantine.go

@@ -1,13 +1,18 @@
 //go:build e2e
 
+// Package harness is DDIL's thin shim over the shared
+// github.com/mulgadc/spinifex/tests/e2e/harness primitives. It hosts only the
+// quarantine + retry wrapper that DDIL scenarios use; everything generic
+// (Cluster, SSH, DaemonClient, ResetAllNodes, Witness, etc.) lives in the
+// shared harness package.
 package harness
 
 import (
-	"context"
 	"os"
 	"strings"
 	"testing"
-	"time"
+
+	"github.com/mulgadc/spinifex/tests/e2e/harness"
 )
 
 // Run wraps a scenario with reset-and-retry plus quarantine handling. It is
@@ -20,14 +25,9 @@ import (
 // a linked bead tracking its fix; the live status is tracked in
 // tests/e2e/TEST_COVERAGE.md.
 //
-// Retry: a first-attempt failure runs ResetAllNodes and invokes fn once more
-// as a fresh subtest. Both attempts appear in the test output so CI logs
-// preserve the transient-vs-deterministic signal. The first-attempt failure
-// is not hidden — Go's testing package propagates subtest failures to the
-// parent — which is intentional: the retry is a diagnostic aid, not a way
-// to mask flakes. Scenarios crossing the 5% nightly flake threshold are
-// moved to DDIL_QUARANTINED instead.
-func Run(t *testing.T, c *Cluster, ssh SSH, letter string, fn func(*testing.T)) {
+// Retry: delegates to harness.RetryWithReset so the shared package owns the
+// reset-and-rerun mechanics; only quarantine policy stays DDIL-specific.
+func Run(t *testing.T, c *harness.Cluster, ssh harness.SSH, letter string, fn func(*testing.T)) {
 	t.Helper()
 
 	if IsQuarantined(letter) {
@@ -35,18 +35,7 @@ func Run(t *testing.T, c *Cluster, ssh SSH, letter string, fn func(*testing.T))
 		return
 	}
 
-	if t.Run("attempt-1", fn) {
-		return
-	}
-
-	t.Logf("ddil harness: scenario %s failed first attempt, resetting cluster and retrying", letter)
-	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
-	defer cancel()
-	if err := ResetAllNodes(ctx, c, ssh); err != nil {
-		t.Errorf("ddil harness: reset before retry of %s: %v", letter, err)
-		return
-	}
-	t.Run("attempt-2", fn)
+	harness.RetryWithReset(t, c, ssh, "scenario "+letter, fn)
 }
 
 // IsQuarantined reports whether letter appears in DDIL_QUARANTINED.

tests/e2e/ddil/harness/ssh.go

@@ -1,128 +0,0 @@
-//go:build e2e
-
-package harness
-
-import (
-	"bytes"
-	"context"
-	"fmt"
-	"net"
-	"os"
-	"time"
-
-	"golang.org/x/crypto/ssh"
-)
-
-// SSH is the transport scenarios use to run commands on a remote Node.
-//
-// It is an interface so dry-run mode and unit tests can stub it without
-// opening real network connections. Run returns the command's combined
-// stdout+stderr and a non-nil error if the remote command exited non-zero or
-// the session failed.
-type SSH interface {
-	Run(ctx context.Context, node Node, cmd string) ([]byte, error)
-	Close() error
-}
-
-// sshClient is the production SSH transport, backed by
-// golang.org/x/crypto/ssh. One client is opened per (user, key, node) tuple
-// on first use and cached for the harness lifetime.
-type sshClient struct {
-	user       string
-	signer     ssh.Signer
-	conns      map[string]*ssh.Client
-	connectDur time.Duration
-	runDur     time.Duration
-}
-
-var _ SSH = (*sshClient)(nil)
-
-// NewSSH returns an SSH transport configured from a Cluster.
-//
-// The private key at cluster.SSHKeyPath must be in a format understood by
-// ssh.ParsePrivateKey (OpenSSH or PEM). Host keys are not verified — the
-// harness runs against short-lived test infrastructure where TOFU would add
-// noise and no security benefit. Do not reuse this transport for production
-// access.
-func NewSSH(cluster *Cluster) (SSH, error) {
-	keyBytes, err := os.ReadFile(cluster.SSHKeyPath)
-	if err != nil {
-		return nil, fmt.Errorf("ddil harness: read ssh key %s: %w", cluster.SSHKeyPath, err)
-	}
-	signer, err := ssh.ParsePrivateKey(keyBytes)
-	if err != nil {
-		return nil, fmt.Errorf("ddil harness: parse ssh key %s: %w", cluster.SSHKeyPath, err)
-	}
-	return &sshClient{
-		user:       cluster.SSHUser,
-		signer:     signer,
-		conns:      make(map[string]*ssh.Client),
-		connectDur: 10 * time.Second,
-		runDur:     2 * time.Minute,
-	}, nil
-}
-
-func (s *sshClient) dial(node Node) (*ssh.Client, error) {
-	if c, ok := s.conns[node.Addr]; ok {
-		return c, nil
-	}
-	cfg := &ssh.ClientConfig{
-		User:            s.user,
-		Auth:            []ssh.AuthMethod{ssh.PublicKeys(s.signer)},
-		HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec // test harness, not production
-		Timeout:         s.connectDur,
-	}
-	c, err := ssh.Dial("tcp", net.JoinHostPort(node.Addr, "22"), cfg)
-	if err != nil {
-		return nil, fmt.Errorf("ddil harness: ssh dial %s: %w", node.Addr, err)
-	}
-	s.conns[node.Addr] = c
-	return c, nil
-}
-
-func (s *sshClient) Run(ctx context.Context, node Node, cmd string) ([]byte, error) {
-	client, err := s.dial(node)
-	if err != nil {
-		return nil, err
-	}
-	session, err := client.NewSession()
-	if err != nil {
-		return nil, fmt.Errorf("ddil harness: ssh new session %s: %w", node.Name, err)
-	}
-	defer func() { _ = session.Close() }()
-
-	var out bytes.Buffer
-	session.Stdout = &out
-	session.Stderr = &out
-
-	// Propagate context cancellation by signalling the session. ssh.Session
-	// has no Kill; Signal(KILL) is the portable way to interrupt a running
-	// command.
-	done := make(chan struct{})
-	defer close(done)
-	go func() {
-		select {
-		case <-ctx.Done():
-			_ = session.Signal(ssh.SIGKILL)
-			_ = session.Close()
-		case <-done:
-		}
-	}()
-
-	if err := session.Run(cmd); err != nil {
-		return out.Bytes(), fmt.Errorf("ddil harness: ssh run on %s (%q): %w\noutput: %s",
-			node.Name, cmd, err, out.String())
-	}
-	return out.Bytes(), nil
-}
-
-func (s *sshClient) Close() error {
-	var firstErr error
-	for addr, c := range s.conns {
-		if err := c.Close(); err != nil && firstErr == nil {
-			firstErr = fmt.Errorf("ddil harness: ssh close %s: %w", addr, err)
-		}
-		delete(s.conns, addr)
-	}
-	return firstErr
-}

tests/e2e/ddil/scenarios/a_nats_kill_test.go

@@ -7,7 +7,9 @@ import (
 	"testing"
 	"time"
 
-	"github.com/mulgadc/spinifex/tests/e2e/ddil/harness"
+	"github.com/mulgadc/spinifex/tests/e2e/harness"
+	ddilh "github.com/mulgadc/spinifex/tests/e2e/ddil/harness"
+	"github.com/mulgadc/spinifex/tests/e2e/ddil/fault"
 )
 
 // TestScenarioA_NATSKill — stop spinifex-nats on a single node without
@@ -19,7 +21,7 @@ func TestScenarioA_NATSKill(t *testing.T) {
 	deps := requireLiveCluster(t)
 	c, ssh, dc, w := deps.cluster, deps.ssh, deps.dc, deps.witness
 
-	harness.Run(t, c, ssh, "A", func(t *testing.T) {
+	ddilh.Run(t, c, ssh, "A", func(t *testing.T) {
 		ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute)
 		defer cancel()
 
@@ -28,12 +30,12 @@ func TestScenarioA_NATSKill(t *testing.T) {
 		node1, node2, node3 := c.Nodes[0], c.Nodes[1], c.Nodes[2]
 		witnesses := launchWitnesses(ctx, t, w, node1, node2, node3)
 
-		pre, err := harness.TakeSnapshot(ctx, dc, node3)
+		pre, err := fault.TakeSnapshot(ctx, dc, node3)
 		if err != nil {
 			t.Fatalf("pre-fault snapshot on %s: %v", node3.Name, err)
 		}
 
-		if err := harness.KillNATS(ctx, ssh, node3); err != nil {
+		if err := fault.KillNATS(ctx, ssh, node3); err != nil {
 			t.Fatalf("kill nats on %s: %v", node3.Name, err)
 		}
 		// Restore NATS in cleanup so a mid-scenario failure does not leave
@@ -41,7 +43,7 @@ func TestScenarioA_NATSKill(t *testing.T) {
 		t.Cleanup(func() {
 			cctx, ccancel := context.WithTimeout(context.Background(), 1*time.Minute)
 			defer ccancel()
-			_ = harness.StartNATS(cctx, ssh, node3)
+			_ = fault.StartNATS(cctx, ssh, node3)
 		})
 
 		if err := harness.WaitForMode(ctx, dc, node3, harness.ModeStandalone, 30*time.Second); err != nil {
@@ -68,14 +70,14 @@ func TestScenarioA_NATSKill(t *testing.T) {
 			}
 		}
 
-		if err := harness.StartNATS(ctx, ssh, node3); err != nil {
+		if err := fault.StartNATS(ctx, ssh, node3); err != nil {
 			t.Fatalf("restart nats on %s: %v", node3.Name, err)
 		}
 		if err := harness.WaitForMode(ctx, dc, node3, harness.ModeCluster, 60*time.Second); err != nil {
 			t.Fatalf("%s did not return to cluster mode: %v", node3.Name, err)
 		}
 
-		post, err := harness.TakeSnapshot(ctx, dc, node3)
+		post, err := fault.TakeSnapshot(ctx, dc, node3)
 		if err != nil {
 			t.Fatalf("post-heal snapshot on %s: %v", node3.Name, err)
 		}

tests/e2e/ddil/scenarios/b_daemon_restart_test.go

@@ -7,7 +7,9 @@ import (
 	"testing"
 	"time"
 
-	"github.com/mulgadc/spinifex/tests/e2e/ddil/harness"
+	"github.com/mulgadc/spinifex/tests/e2e/harness"
+	ddilh "github.com/mulgadc/spinifex/tests/e2e/ddil/harness"
+	"github.com/mulgadc/spinifex/tests/e2e/ddil/fault"
 )
 
 // TestScenarioB_DaemonRestartWithoutNATS — kill spinifex-nats, restart
@@ -26,7 +28,7 @@ func TestScenarioB_DaemonRestartWithoutNATS(t *testing.T) {
 	deps := requireLiveCluster(t)
 	c, ssh, dc, w := deps.cluster, deps.ssh, deps.dc, deps.witness
 
-	harness.Run(t, c, ssh, "B", func(t *testing.T) {
+	ddilh.Run(t, c, ssh, "B", func(t *testing.T) {
 		ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute)
 		defer cancel()
 
@@ -35,7 +37,7 @@ func TestScenarioB_DaemonRestartWithoutNATS(t *testing.T) {
 		node3 := c.Nodes[2]
 		_ = launchWitnesses(ctx, t, w, node3)
 
-		pre, err := harness.TakeSnapshot(ctx, dc, node3)
+		pre, err := fault.TakeSnapshot(ctx, dc, node3)
 		if err != nil {
 			t.Fatalf("pre-fault snapshot on %s: %v", node3.Name, err)
 		}
@@ -43,20 +45,20 @@ func TestScenarioB_DaemonRestartWithoutNATS(t *testing.T) {
 			t.Fatalf("pre-fault /local/instances on %s empty — witness launch did not register", node3.Name)
 		}
 
-		if err := harness.KillNATS(ctx, ssh, node3); err != nil {
+		if err := fault.KillNATS(ctx, ssh, node3); err != nil {
 			t.Fatalf("kill nats on %s: %v", node3.Name, err)
 		}
 		t.Cleanup(func() {
 			cctx, ccancel := context.WithTimeout(context.Background(), 1*time.Minute)
 			defer ccancel()
-			_ = harness.StartNATS(cctx, ssh, node3)
+			_ = fault.StartNATS(cctx, ssh, node3)
 		})
 
 		// 1d's promise is "daemon comes up without NATS". The 30s budget is
 		// the plan's threshold — if it slips, the new startup path has
 		// regressed back to the 5-minute NATS-wait abort.
 		restartStart := time.Now()
-		if err := harness.RestartDaemonOnly(ctx, ssh, node3); err != nil {
+		if err := fault.RestartDaemonOnly(ctx, ssh, node3); err != nil {
 			t.Fatalf("restart daemon on %s: %v", node3.Name, err)
 		}
 		if err := harness.WaitForMode(ctx, dc, node3, harness.ModeStandalone, 30*time.Second); err != nil {
@@ -64,7 +66,7 @@ func TestScenarioB_DaemonRestartWithoutNATS(t *testing.T) {
 		}
 		t.Logf("daemon on %s reached standalone in %s", node3.Name, time.Since(restartStart))
 
-		post, err := harness.TakeSnapshot(ctx, dc, node3)
+		post, err := fault.TakeSnapshot(ctx, dc, node3)
 		if err != nil {
 			t.Fatalf("post-restart snapshot on %s: %v", node3.Name, err)
 		}
@@ -80,7 +82,7 @@ func TestScenarioB_DaemonRestartWithoutNATS(t *testing.T) {
 			}
 		}
 
-		if err := harness.StartNATS(ctx, ssh, node3); err != nil {
+		if err := fault.StartNATS(ctx, ssh, node3); err != nil {
 			t.Fatalf("restart nats on %s: %v", node3.Name, err)
 		}
 		if err := harness.WaitForMode(ctx, dc, node3, harness.ModeCluster, 60*time.Second); err != nil {

tests/e2e/ddil/scenarios/c_partition_test.go

@@ -7,7 +7,9 @@ import (
 	"testing"
 	"time"
 
-	"github.com/mulgadc/spinifex/tests/e2e/ddil/harness"
+	"github.com/mulgadc/spinifex/tests/e2e/harness"
+	ddilh "github.com/mulgadc/spinifex/tests/e2e/ddil/harness"
+	"github.com/mulgadc/spinifex/tests/e2e/ddil/fault"
 )
 
 // TestScenarioC_CleanPartition — iptables-DROP node3 away from node1 and
@@ -19,7 +21,7 @@ func TestScenarioC_CleanPartition(t *testing.T) {
 	deps := requireLiveCluster(t)
 	c, ssh, dc, w := deps.cluster, deps.ssh, deps.dc, deps.witness
 
-	harness.Run(t, c, ssh, "C", func(t *testing.T) {
+	ddilh.Run(t, c, ssh, "C", func(t *testing.T) {
 		ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute)
 		defer cancel()
 
@@ -31,12 +33,12 @@ func TestScenarioC_CleanPartition(t *testing.T) {
 		// only node-3.
 		witnesses := launchWitnesses(ctx, t, w, node3)
 
-		pre, err := harness.TakeSnapshot(ctx, dc, node3)
+		pre, err := fault.TakeSnapshot(ctx, dc, node3)
 		if err != nil {
 			t.Fatalf("pre-fault snapshot on %s: %v", node3.Name, err)
 		}
 
-		if err := harness.PartitionNode(ctx, ssh, node3, c.Peers(node3)); err != nil {
+		if err := fault.PartitionNode(ctx, ssh, node3, c.Peers(node3)); err != nil {
 			t.Fatalf("partition %s: %v", node3.Name, err)
 		}
 		// HealNode flushes node3's iptables on cleanup; safe even if a later
@@ -44,7 +46,7 @@ func TestScenarioC_CleanPartition(t *testing.T) {
 		t.Cleanup(func() {
 			cctx, ccancel := context.WithTimeout(context.Background(), 1*time.Minute)
 			defer ccancel()
-			_ = harness.HealNode(cctx, ssh, node3)
+			_ = fault.HealNode(cctx, ssh, node3)
 		})
 
 		// The orchestrator dials peers directly from outside the cluster
@@ -74,14 +76,14 @@ func TestScenarioC_CleanPartition(t *testing.T) {
 		// proving QEMU on node3 kept executing despite the network split.
 		harness.AssertProgressed(ctx, t, witnesses[0])
 
-		if err := harness.HealNode(ctx, ssh, node3); err != nil {
+		if err := fault.HealNode(ctx, ssh, node3); err != nil {
 			t.Fatalf("heal %s: %v", node3.Name, err)
 		}
 		if err := harness.WaitForMode(ctx, dc, node3, harness.ModeCluster, 60*time.Second); err != nil {
 			t.Fatalf("%s did not return to cluster mode after heal: %v", node3.Name, err)
 		}
 
-		post, err := harness.TakeSnapshot(ctx, dc, node3)
+		post, err := fault.TakeSnapshot(ctx, dc, node3)
 		if err != nil {
 			t.Fatalf("post-heal snapshot on %s: %v", node3.Name, err)
 		}

tests/e2e/ddil/scenarios/main_test.go

@@ -18,7 +18,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/mulgadc/spinifex/tests/e2e/ddil/harness"
+	"github.com/mulgadc/spinifex/tests/e2e/harness"
 )
 
 // scenarioLetters is the canonical ordered list of scenarios the suite

tests/e2e/harness/cluster.go

@@ -1,10 +1,5 @@
 //go:build e2e
 
-// Package harness exports the DDIL E2E test primitives: typed cluster/node
-// descriptors, an SSH transport, fault-injection helpers, witness-VM helpers,
-// snapshot/state assertions, and the daemon HTTP client used by scenarios.
-//
-// See docs/development/improvements/ddil-e2e-test-harness.md for the design.
 package harness
 
 import (
@@ -16,7 +11,7 @@ import (
 
 // Node identifies a single cluster member addressable over SSH and HTTPS.
 //
-// Index is the 1-based position the node occupies in the DDIL_NODES list;
+// Index is the 1-based position the node occupies in the SPINIFEX_NODES list;
 // scenarios reference nodes by index (node1, node2, node3) so a fixed
 // ordering is required.
 type Node struct {
@@ -37,7 +32,7 @@ type Cluster struct {
 }
 
 // Peers returns every node in the cluster except target. The result is used
-// by PartitionNode to build per-peer iptables DROP rules.
+// by fault.PartitionNode to build per-peer iptables DROP rules.
 func (c *Cluster) Peers(target Node) []Node {
 	out := make([]Node, 0, len(c.Nodes)-1)
 	for _, n := range c.Nodes {
@@ -49,29 +44,29 @@ func (c *Cluster) Peers(target Node) []Node {
 	return out
 }
 
-// ClusterFromEnv builds a Cluster from DDIL_NODES / DDIL_SSH_USER /
-// DDIL_SSH_KEY. Nodes are named node1..nodeN in list order.
+// ClusterFromEnv builds a Cluster from SPINIFEX_NODES / SPINIFEX_SSH_USER /
+// SPINIFEX_SSH_KEY. Nodes are named node1..nodeN in list order.
 //
-// Returns an error if any required variable is missing or if DDIL_NODES
+// Returns an error if any required variable is missing or if SPINIFEX_NODES
 // contains no non-empty entries. main_test.go is expected to call this once
 // and pass the result into every scenario.
 func ClusterFromEnv() (*Cluster, error) {
-	nodesRaw := os.Getenv("DDIL_NODES")
-	user := os.Getenv("DDIL_SSH_USER")
-	key := os.Getenv("DDIL_SSH_KEY")
+	nodesRaw := envWithLegacy("SPINIFEX_NODES", "DDIL_NODES")
+	user := envWithLegacy("SPINIFEX_SSH_USER", "DDIL_SSH_USER")
+	key := envWithLegacy("SPINIFEX_SSH_KEY", "DDIL_SSH_KEY")
 
 	var missing []string
 	if nodesRaw == "" {
-		missing = append(missing, "DDIL_NODES")
+		missing = append(missing, "SPINIFEX_NODES")
 	}
 	if user == "" {
-		missing = append(missing, "DDIL_SSH_USER")
+		missing = append(missing, "SPINIFEX_SSH_USER")
 	}
 	if key == "" {
-		missing = append(missing, "DDIL_SSH_KEY")
+		missing = append(missing, "SPINIFEX_SSH_KEY")
 	}
 	if len(missing) > 0 {
-		return nil, fmt.Errorf("ddil harness: missing required env: %s", strings.Join(missing, ", "))
+		return nil, fmt.Errorf("e2e harness: missing required env: %s", strings.Join(missing, ", "))
 	}
 
 	var nodes []Node
@@ -87,7 +82,7 @@ func ClusterFromEnv() (*Cluster, error) {
 		})
 	}
 	if len(nodes) == 0 {
-		return nil, errors.New("ddil harness: DDIL_NODES contained no addresses")
+		return nil, errors.New("e2e harness: SPINIFEX_NODES contained no addresses")
 	}
 
 	return &Cluster{
@@ -98,7 +93,7 @@ func ClusterFromEnv() (*Cluster, error) {
 }
 
 // NodeFromEnv returns the Nth node (1-based) from the cluster defined by
-// DDIL_NODES. Convenience wrapper for single-node helpers; most scenarios
+// SPINIFEX_NODES. Convenience wrapper for single-node helpers; most scenarios
 // should call ClusterFromEnv once and index c.Nodes directly.
 func NodeFromEnv(index int) (Node, error) {
 	c, err := ClusterFromEnv()
@@ -106,7 +101,22 @@ func NodeFromEnv(index int) (Node, error) {
 		return Node{}, err
 	}
 	if index < 1 || index > len(c.Nodes) {
-		return Node{}, fmt.Errorf("ddil harness: node index %d out of range (have %d nodes)", index, len(c.Nodes))
+		return Node{}, fmt.Errorf("e2e harness: node index %d out of range (have %d nodes)", index, len(c.Nodes))
 	}
 	return c.Nodes[index-1], nil
 }
+
+// envWithLegacy returns os.Getenv(canonical), falling back to legacy on miss.
+// DDIL_* names are accepted for back-compat; rename to SPINIFEX_* once all
+// CI workflows are migrated. A one-line stderr note flags the deprecation
+// so leftover callers surface in CI logs.
+func envWithLegacy(canonical, legacy string) string {
+	if v := os.Getenv(canonical); v != "" {
+		return v
+	}
+	if v := os.Getenv(legacy); v != "" {
+		fmt.Fprintf(os.Stderr, "e2e harness: %s is deprecated; set %s instead\n", legacy, canonical)
+		return v
+	}
+	return ""
+}

tests/e2e/harness/daemon_client.go

@@ -16,6 +16,45 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/types"
 )
 
+// WaitForMode polls the daemon's /local/status until it reports the expected
+// mode or timeout expires. Poll interval is 1s — fast enough that a
+// 30-second timeout sees ~30 attempts, slow enough not to flood a recovering
+// daemon.
+//
+// Depends on daemon-local-autonomy §1b. Until that endpoint ships, this
+// function will time out; callers gate on t.Skip.
+func WaitForMode(ctx context.Context, dc *DaemonClient, node Node, want DaemonMode, timeout time.Duration) error {
+	deadline := time.Now().Add(timeout)
+	const interval = 1 * time.Second
+
+	var lastErr error
+	for {
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
+		status, err := dc.Status(ctx, node)
+		if err == nil && status.Mode == want {
+			return nil
+		}
+		lastErr = err
+
+		if time.Now().After(deadline) {
+			if lastErr != nil {
+				return fmt.Errorf("e2e harness: wait for mode %s on %s: timed out after %s: last error: %w",
+					want, node.Name, timeout, lastErr)
+			}
+			return fmt.Errorf("e2e harness: wait for mode %s on %s: timed out after %s (still reporting another mode)",
+				want, node.Name, timeout)
+		}
+
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(interval):
+		}
+	}
+}
+
 // daemonPort is the cluster-manager HTTPS port the daemon binds for /health
 // and /local/* (see spinifex/daemon/daemon.go ClusterManager, configured via
 // [nodes.<id>.daemon].host). 4432 is the deployed default; 8443 is predastore.
@@ -84,21 +123,21 @@ func (d *DaemonClient) url(node Node, path string) string {
 func (d *DaemonClient) getJSON(ctx context.Context, node Node, path string, out any) error {
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, d.url(node, path), nil)
 	if err != nil {
-		return fmt.Errorf("ddil harness: daemon %s %s: %w", node.Name, path, err)
+		return fmt.Errorf("e2e harness: daemon %s %s: %w", node.Name, path, err)
 	}
 	resp, err := d.http.Do(req)
 	if err != nil {
-		return fmt.Errorf("ddil harness: daemon %s %s: %w", node.Name, path, err)
+		return fmt.Errorf("e2e harness: daemon %s %s: %w", node.Name, path, err)
 	}
 	defer func() { _ = resp.Body.Close() }()
 
 	if resp.StatusCode != http.StatusOK {
 		body, _ := io.ReadAll(resp.Body)
-		return fmt.Errorf("ddil harness: daemon %s %s: status %d: %s",
+		return fmt.Errorf("e2e harness: daemon %s %s: status %d: %s",
 			node.Name, path, resp.StatusCode, string(body))
 	}
 	if err := json.NewDecoder(resp.Body).Decode(out); err != nil {
-		return fmt.Errorf("ddil harness: daemon %s %s decode: %w", node.Name, path, err)
+		return fmt.Errorf("e2e harness: daemon %s %s decode: %w", node.Name, path, err)
 	}
 	return nil
 }

tests/e2e/harness/peer_ssh.go

@@ -0,0 +1,59 @@
+//go:build e2e
+
+package harness
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/exec"
+	"strings"
+	"time"
+)
+
+// PeerSSH is a minimal SSH wrapper that shells out to /usr/bin/ssh with the
+// flags the bash E2E scripts use. It exists so LB/multinode scenarios can
+// drive remote nodes without depending on the DDIL harness. Step 1.5 of
+// docs/development/improvements/e2e-go-harness.md replaces this with the
+// typed x/crypto/ssh client hoisted from ddil/harness/ssh.go.
+type PeerSSH struct {
+	User    string
+	KeyPath string
+	Timeout time.Duration
+}
+
+// NewPeerSSH reads SPINIFEX_SSH_KEY (default ~/.ssh/tf-user-ap-southeast-2)
+// and SPINIFEX_SSH_USER (default tf-user). Matches the bash peer_ssh helper.
+func NewPeerSSH() *PeerSSH {
+	return &PeerSSH{
+		User:    getenv("SPINIFEX_SSH_USER", "tf-user"),
+		KeyPath: getenv("SPINIFEX_SSH_KEY", os.ExpandEnv("$HOME/.ssh/tf-user-ap-southeast-2")),
+		Timeout: 10 * time.Second,
+	}
+}
+
+// Run executes cmd on host over SSH and returns combined stdout+stderr.
+// Non-zero exit becomes a non-nil error with stderr in the message.
+func (p *PeerSSH) Run(ctx context.Context, host, cmd string) ([]byte, error) {
+	args := []string{
+		"-i", p.KeyPath,
+		"-o", "StrictHostKeyChecking=no",
+		"-o", "UserKnownHostsFile=/dev/null",
+		"-o", fmt.Sprintf("ConnectTimeout=%d", int(p.Timeout.Seconds())),
+		"-o", "LogLevel=ERROR",
+		fmt.Sprintf("%s@%s", p.User, host),
+		cmd,
+	}
+	out, err := exec.CommandContext(ctx, "ssh", args...).CombinedOutput()
+	if err != nil {
+		return out, fmt.Errorf("ssh %s@%s: %w: %s", p.User, host, err, strings.TrimSpace(string(out)))
+	}
+	return out, nil
+}
+
+// Ping returns nil if `hostname` succeeds over SSH. Used as a probe before
+// running real remote tests.
+func (p *PeerSSH) Ping(ctx context.Context, host string) error {
+	_, err := p.Run(ctx, host, "hostname")
+	return err
+}

tests/e2e/harness/reset.go

@@ -49,7 +49,7 @@ func resetNode(ctx context.Context, ssh SSH, n Node) error {
 	// iptables -X removes any transient user chains a scenario happens to
 	// leave behind. This matches HealNode's teardown.
 	if _, err := ssh.Run(ctx, n, "sudo iptables -F && sudo iptables -X"); err != nil {
-		return fmt.Errorf("ddil harness: reset iptables on %s: %w", n.Name, err)
+		return fmt.Errorf("e2e harness: reset iptables on %s: %w", n.Name, err)
 	}
 
 	// Remove netem qdiscs only on interfaces that actually carry one.
@@ -60,15 +60,15 @@ func resetNode(ctx context.Context, ssh SSH, n Node) error {
 		return err
 	}
 	for _, iface := range ifaces {
-		cmd := fmt.Sprintf("sudo tc qdisc del dev %s root 2>/dev/null || true", shellQuote(iface))
+		cmd := fmt.Sprintf("sudo tc qdisc del dev %s root 2>/dev/null || true", ShellQuote(iface))
 		if _, err := ssh.Run(ctx, n, cmd); err != nil {
-			return fmt.Errorf("ddil harness: clear netem on %s (%s): %w", n.Name, iface, err)
+			return fmt.Errorf("e2e harness: clear netem on %s (%s): %w", n.Name, iface, err)
 		}
 	}
 
 	// Start services — noop if already active (systemd returns 0).
 	if _, err := ssh.Run(ctx, n, "sudo systemctl start spinifex-nats spinifex-daemon"); err != nil {
-		return fmt.Errorf("ddil harness: start services on %s: %w", n.Name, err)
+		return fmt.Errorf("e2e harness: start services on %s: %w", n.Name, err)
 	}
 	return nil
 }
@@ -81,7 +81,7 @@ func netemIfaces(ctx context.Context, ssh SSH, n Node) ([]string, error) {
 	// root ...` → awk $5 yields the interface name.
 	out, err := ssh.Run(ctx, n, "tc qdisc show | awk '/qdisc netem/ {print $5}' | sort -u")
 	if err != nil {
-		return nil, fmt.Errorf("ddil harness: list netem qdiscs on %s: %w", n.Name, err)
+		return nil, fmt.Errorf("e2e harness: list netem qdiscs on %s: %w", n.Name, err)
 	}
 	return strings.Fields(string(out)), nil
 }
@@ -104,7 +104,7 @@ func waitClusterHealthy(ctx context.Context, c *Cluster, ssh SSH, timeout time.D
 			lastErr = err
 		}
 		if time.Now().After(deadline) {
-			return fmt.Errorf("ddil harness: cluster did not reach healthy within %s: %w", timeout, lastErr)
+			return fmt.Errorf("e2e harness: cluster did not reach healthy within %s: %w", timeout, lastErr)
 		}
 		select {
 		case <-ctx.Done():
@@ -164,11 +164,11 @@ func AssertCleanState(ctx context.Context, t *testing.T, c *Cluster, ssh SSH) {
 	for _, n := range c.Nodes {
 		drops, err := countDropRules(ctx, ssh, n)
 		if err != nil {
-			t.Fatalf("ddil harness: inspect iptables on %s: %v", n.Name, err)
+			t.Fatalf("e2e harness: inspect iptables on %s: %v", n.Name, err)
 		}
 		ifaces, err := netemIfaces(ctx, ssh, n)
 		if err != nil {
-			t.Fatalf("ddil harness: inspect netem on %s: %v", n.Name, err)
+			t.Fatalf("e2e harness: inspect netem on %s: %v", n.Name, err)
 		}
 		if drops > 0 {
 			dirty = append(dirty, fmt.Sprintf("%s: %d iptables DROP rules", n.Name, drops))
@@ -181,9 +181,9 @@ func AssertCleanState(ctx context.Context, t *testing.T, c *Cluster, ssh SSH) {
 	if len(dirty) == 0 {
 		return
 	}
-	t.Logf("ddil harness: dirty baseline before scenario — %s; running ResetAllNodes", strings.Join(dirty, "; "))
+	t.Logf("e2e harness: dirty baseline before scenario — %s; running ResetAllNodes", strings.Join(dirty, "; "))
 	if err := ResetAllNodes(ctx, c, ssh); err != nil {
-		t.Fatalf("ddil harness: reset after dirty baseline: %v", err)
+		t.Fatalf("e2e harness: reset after dirty baseline: %v", err)
 	}
 }
 

tests/e2e/harness/ssh.go

@@ -3,57 +3,126 @@
 package harness
 
 import (
+	"bytes"
 	"context"
 	"fmt"
+	"net"
 	"os"
-	"os/exec"
-	"strings"
 	"time"
+
+	"golang.org/x/crypto/ssh"
 )
 
-// PeerSSH is a minimal SSH wrapper that shells out to /usr/bin/ssh with the
-// flags the bash E2E scripts use. It exists so LB/multinode scenarios can
-// drive remote nodes without depending on the DDIL harness. Step 1.5 of
-// docs/development/improvements/e2e-go-harness.md replaces this with the
-// typed x/crypto/ssh client hoisted from ddil/harness/ssh.go.
-type PeerSSH struct {
-	User    string
-	KeyPath string
-	Timeout time.Duration
+// SSH is the transport scenarios use to run commands on a remote Node.
+//
+// It is an interface so dry-run mode and unit tests can stub it without
+// opening real network connections. Run returns the command's combined
+// stdout+stderr and a non-nil error if the remote command exited non-zero or
+// the session failed.
+type SSH interface {
+	Run(ctx context.Context, node Node, cmd string) ([]byte, error)
+	Close() error
 }
 
-// NewPeerSSH reads SPINIFEX_SSH_KEY (default ~/.ssh/tf-user-ap-southeast-2)
-// and SPINIFEX_SSH_USER (default tf-user). Matches the bash peer_ssh helper.
-func NewPeerSSH() *PeerSSH {
-	return &PeerSSH{
-		User:    getenv("SPINIFEX_SSH_USER", "tf-user"),
-		KeyPath: getenv("SPINIFEX_SSH_KEY", os.ExpandEnv("$HOME/.ssh/tf-user-ap-southeast-2")),
-		Timeout: 10 * time.Second,
-	}
+// sshClient is the production SSH transport, backed by
+// golang.org/x/crypto/ssh. One client is opened per (user, key, node) tuple
+// on first use and cached for the harness lifetime.
+type sshClient struct {
+	user       string
+	signer     ssh.Signer
+	conns      map[string]*ssh.Client
+	connectDur time.Duration
+	runDur     time.Duration
 }
 
-// Run executes cmd on host over SSH and returns combined stdout+stderr.
-// Non-zero exit becomes a non-nil error with stderr in the message.
-func (p *PeerSSH) Run(ctx context.Context, host, cmd string) ([]byte, error) {
-	args := []string{
-		"-i", p.KeyPath,
-		"-o", "StrictHostKeyChecking=no",
-		"-o", "UserKnownHostsFile=/dev/null",
-		"-o", fmt.Sprintf("ConnectTimeout=%d", int(p.Timeout.Seconds())),
-		"-o", "LogLevel=ERROR",
-		fmt.Sprintf("%s@%s", p.User, host),
-		cmd,
-	}
-	out, err := exec.CommandContext(ctx, "ssh", args...).CombinedOutput()
+var _ SSH = (*sshClient)(nil)
+
+// NewSSH returns an SSH transport configured from a Cluster.
+//
+// The private key at cluster.SSHKeyPath must be in a format understood by
+// ssh.ParsePrivateKey (OpenSSH or PEM). Host keys are not verified — the
+// harness runs against short-lived test infrastructure where TOFU would add
+// noise and no security benefit. Do not reuse this transport for production
+// access.
+func NewSSH(cluster *Cluster) (SSH, error) {
+	keyBytes, err := os.ReadFile(cluster.SSHKeyPath)
 	if err != nil {
-		return out, fmt.Errorf("ssh %s@%s: %w: %s", p.User, host, err, strings.TrimSpace(string(out)))
+		return nil, fmt.Errorf("e2e harness: read ssh key %s: %w", cluster.SSHKeyPath, err)
 	}
-	return out, nil
+	signer, err := ssh.ParsePrivateKey(keyBytes)
+	if err != nil {
+		return nil, fmt.Errorf("e2e harness: parse ssh key %s: %w", cluster.SSHKeyPath, err)
+	}
+	return &sshClient{
+		user:       cluster.SSHUser,
+		signer:     signer,
+		conns:      make(map[string]*ssh.Client),
+		connectDur: 10 * time.Second,
+		runDur:     2 * time.Minute,
+	}, nil
 }
 
-// Ping returns nil if `hostname` succeeds over SSH. Used as a probe before
-// running real remote tests.
-func (p *PeerSSH) Ping(ctx context.Context, host string) error {
-	_, err := p.Run(ctx, host, "hostname")
-	return err
+func (s *sshClient) dial(node Node) (*ssh.Client, error) {
+	if c, ok := s.conns[node.Addr]; ok {
+		return c, nil
+	}
+	cfg := &ssh.ClientConfig{
+		User:            s.user,
+		Auth:            []ssh.AuthMethod{ssh.PublicKeys(s.signer)},
+		HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec // test harness, not production
+		Timeout:         s.connectDur,
+	}
+	c, err := ssh.Dial("tcp", net.JoinHostPort(node.Addr, "22"), cfg)
+	if err != nil {
+		return nil, fmt.Errorf("e2e harness: ssh dial %s: %w", node.Addr, err)
+	}
+	s.conns[node.Addr] = c
+	return c, nil
+}
+
+func (s *sshClient) Run(ctx context.Context, node Node, cmd string) ([]byte, error) {
+	client, err := s.dial(node)
+	if err != nil {
+		return nil, err
+	}
+	session, err := client.NewSession()
+	if err != nil {
+		return nil, fmt.Errorf("e2e harness: ssh new session %s: %w", node.Name, err)
+	}
+	defer func() { _ = session.Close() }()
+
+	var out bytes.Buffer
+	session.Stdout = &out
+	session.Stderr = &out
+
+	// Propagate context cancellation by signalling the session. ssh.Session
+	// has no Kill; Signal(KILL) is the portable way to interrupt a running
+	// command.
+	done := make(chan struct{})
+	defer close(done)
+	go func() {
+		select {
+		case <-ctx.Done():
+			_ = session.Signal(ssh.SIGKILL)
+			_ = session.Close()
+		case <-done:
+		}
+	}()
+
+	if err := session.Run(cmd); err != nil {
+		return out.Bytes(), fmt.Errorf("e2e harness: ssh run on %s (%q): %w\noutput: %s",
+			node.Name, cmd, err, out.String())
+	}
+	return out.Bytes(), nil
+}
+
+func (s *sshClient) Close() error {
+	var firstErr error
+	for addr, c := range s.conns {
+		if err := c.Close(); err != nil && firstErr == nil {
+			firstErr = fmt.Errorf("e2e harness: ssh close %s: %w", addr, err)
+		}
+		delete(s.conns, addr)
+	}
+	return firstErr
 }

tests/e2e/harness/state.go

@@ -16,7 +16,7 @@ import (
 func LocalStateRevision(ctx context.Context, d *DaemonClient, node Node) (uint64, error) {
 	s, err := d.Status(ctx, node)
 	if err != nil {
-		return 0, fmt.Errorf("ddil harness: local state revision %s: %w", node.Name, err)
+		return 0, fmt.Errorf("e2e harness: local state revision %s: %w", node.Name, err)
 	}
 	return s.Revision, nil
 }
@@ -28,6 +28,6 @@ func LocalStateRevision(ctx context.Context, d *DaemonClient, node Node) (uint64
 func AssertMonotonic(t *testing.T, before, after uint64) {
 	t.Helper()
 	if after < before {
-		t.Fatalf("ddil harness: local state revision regressed: before=%d after=%d", before, after)
+		t.Fatalf("e2e harness: local state revision regressed: before=%d after=%d", before, after)
 	}
 }

tests/e2e/harness/util.go

@@ -0,0 +1,12 @@
+//go:build e2e
+
+package harness
+
+import "strings"
+
+// ShellQuote wraps s in single quotes and escapes embedded single quotes, so
+// that shell metacharacters in interface names or peer addresses are treated
+// as literals on the remote shell.
+func ShellQuote(s string) string {
+	return "'" + strings.ReplaceAll(s, "'", `'\''`) + "'"
+}

tests/e2e/harness/wait.go

@@ -3,6 +3,7 @@
 package harness
 
 import (
+	"context"
 	"testing"
 	"time"
 )
@@ -44,3 +45,26 @@ func EventuallyErr(t *testing.T, cond func() error, timeout, interval time.Durat
 		time.Sleep(interval)
 	}
 }
+
+// RetryWithReset runs fn as an "attempt-1" subtest; on failure it calls
+// ResetAllNodes and retries fn as "attempt-2". Both attempts surface in the
+// test output so CI logs preserve the transient-vs-deterministic signal —
+// the retry is a diagnostic aid, not a way to mask flakes.
+//
+// Used by DDIL's quarantine wrapper (ddil/harness.Run) and any other scenario
+// that wants the same retry-after-reset shape without taking on DDIL's
+// quarantine semantics.
+func RetryWithReset(t *testing.T, c *Cluster, ssh SSH, label string, fn func(*testing.T)) {
+	t.Helper()
+	if t.Run("attempt-1", fn) {
+		return
+	}
+	t.Logf("e2e harness: %s failed first attempt, resetting cluster and retrying", label)
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancel()
+	if err := ResetAllNodes(ctx, c, ssh); err != nil {
+		t.Errorf("e2e harness: reset before retry of %s: %v", label, err)
+		return
+	}
+	t.Run("attempt-2", fn)
+}

tests/e2e/harness/witness.go

@@ -75,10 +75,10 @@ type Witness struct {
 func NewWitness(cluster *Cluster, transport SSH) (*Witness, error) {
 	region := os.Getenv("AWS_REGION")
 	if region == "" {
-		return nil, errors.New("ddil harness: NewWitness requires AWS_REGION")
+		return nil, errors.New("e2e harness: NewWitness requires AWS_REGION")
 	}
 	if len(cluster.Nodes) == 0 {
-		return nil, errors.New("ddil harness: NewWitness requires a non-empty cluster")
+		return nil, errors.New("e2e harness: NewWitness requires a non-empty cluster")
 	}
 
 	endpoint := "https://" + net.JoinHostPort(cluster.Nodes[0].Addr, strconv.Itoa(gatewayPort))
@@ -93,21 +93,21 @@ func NewWitness(cluster *Cluster, transport SSH) (*Witness, error) {
 		},
 	})
 	if err != nil {
-		return nil, fmt.Errorf("ddil harness: aws session: %w", err)
+		return nil, fmt.Errorf("e2e harness: aws session: %w", err)
 	}
 
 	hostSigner, err := loadSigner(cluster.SSHKeyPath)
 	if err != nil {
-		return nil, fmt.Errorf("ddil harness: host ssh key: %w", err)
+		return nil, fmt.Errorf("e2e harness: host ssh key: %w", err)
 	}
 
 	guestKeyPath := os.Getenv("DDIL_GUEST_SSH_KEY")
 	if guestKeyPath == "" {
-		return nil, errors.New("ddil harness: NewWitness requires DDIL_GUEST_SSH_KEY (path to the private key paired with DDIL_WITNESS_KEY_NAME's registered material)")
+		return nil, errors.New("e2e harness: NewWitness requires DDIL_GUEST_SSH_KEY (path to the private key paired with DDIL_WITNESS_KEY_NAME's registered material)")
 	}
 	guestSigner, err := loadSigner(guestKeyPath)
 	if err != nil {
-		return nil, fmt.Errorf("ddil harness: guest ssh key: %w", err)
+		return nil, fmt.Errorf("e2e harness: guest ssh key: %w", err)
 	}
 
 	return &Witness{
@@ -170,10 +170,10 @@ func LaunchWitnessVM(ctx context.Context, w *Witness, host Node) (*WitnessVM, er
 			UserData:     aws.String(userData),
 		})
 		if err != nil {
-			return nil, fmt.Errorf("ddil harness: RunInstances: %w", err)
+			return nil, fmt.Errorf("e2e harness: RunInstances: %w", err)
 		}
 		if len(out.Instances) == 0 || out.Instances[0].InstanceId == nil {
-			return nil, errors.New("ddil harness: RunInstances returned no instance")
+			return nil, errors.New("e2e harness: RunInstances returned no instance")
 		}
 		id := aws.StringValue(out.Instances[0].InstanceId)
 
@@ -197,7 +197,7 @@ func LaunchWitnessVM(ctx context.Context, w *Witness, host Node) (*WitnessVM, er
 			baseline, err := awaitBaseline(ctx, w, placed, publicIP, 2*time.Minute)
 			if err != nil {
 				_ = w.terminate(ctx, id)
-				return nil, fmt.Errorf("ddil harness: witness baseline read: %w", err)
+				return nil, fmt.Errorf("e2e harness: witness baseline read: %w", err)
 			}
 			return &WitnessVM{
 				InstanceID:      id,
@@ -212,7 +212,7 @@ func LaunchWitnessVM(ctx context.Context, w *Witness, host Node) (*WitnessVM, er
 		lastErr = fmt.Errorf("witness landed on %s, wanted %s", placed.Name, host.Name)
 		_ = w.terminate(ctx, id)
 	}
-	return nil, fmt.Errorf("ddil harness: LaunchWitnessVM: %w after %d attempts", lastErr, maxPlacementAttempts)
+	return nil, fmt.Errorf("e2e harness: LaunchWitnessVM: %w after %d attempts", lastErr, maxPlacementAttempts)
 }
 
 // ReadCounter SSHes the guest at its public IP (tunnelled through the
@@ -235,10 +235,10 @@ func AssertProgressed(ctx context.Context, t *testing.T, v *WitnessVM) {
 	t.Helper()
 	current, err := v.ReadCounter(ctx)
 	if err != nil {
-		t.Fatalf("ddil harness: read witness counter on %s (%s): %v", v.HostNode.Name, v.InstanceID, err)
+		t.Fatalf("e2e harness: read witness counter on %s (%s): %v", v.HostNode.Name, v.InstanceID, err)
 	}
 	if current <= v.BaselineCounter {
-		t.Fatalf("ddil harness: witness %s on %s did not progress: baseline=%d current=%d",
+		t.Fatalf("e2e harness: witness %s on %s did not progress: baseline=%d current=%d",
 			v.InstanceID, v.HostNode.Name, v.BaselineCounter, current)
 	}
 	t.Logf("witness %s on %s progressed %d → %d", v.InstanceID, v.HostNode.Name, v.BaselineCounter, current)
@@ -257,10 +257,10 @@ func (w *Witness) resolveAMI(ctx context.Context) (string, error) {
 		}},
 	})
 	if err != nil {
-		return "", fmt.Errorf("ddil harness: DescribeImages: %w", err)
+		return "", fmt.Errorf("e2e harness: DescribeImages: %w", err)
 	}
 	if len(out.Images) == 0 || out.Images[0].ImageId == nil {
-		return "", errors.New("ddil harness: no ami-ubuntu-* images registered in the cluster")
+		return "", errors.New("e2e harness: no ami-ubuntu-* images registered in the cluster")
 	}
 	w.ami = aws.StringValue(out.Images[0].ImageId)
 	return w.ami, nil
@@ -279,10 +279,10 @@ func (w *Witness) resolveInstanceType(ctx context.Context) (string, error) {
 	}
 	out, err := w.ec2.DescribeInstanceTypesWithContext(ctx, &ec2.DescribeInstanceTypesInput{})
 	if err != nil {
-		return "", fmt.Errorf("ddil harness: DescribeInstanceTypes: %w", err)
+		return "", fmt.Errorf("e2e harness: DescribeInstanceTypes: %w", err)
 	}
 	if len(out.InstanceTypes) == 0 {
-		return "", errors.New("ddil harness: cluster registered no instance types")
+		return "", errors.New("e2e harness: cluster registered no instance types")
 	}
 	sort.Slice(out.InstanceTypes, func(i, j int) bool {
 		a, b := out.InstanceTypes[i], out.InstanceTypes[j]
@@ -307,7 +307,7 @@ func (w *Witness) resolveInstanceType(ctx context.Context) (string, error) {
 	})
 	w.instanceType = aws.StringValue(out.InstanceTypes[0].InstanceType)
 	if w.instanceType == "" {
-		return "", errors.New("ddil harness: smallest instance type has empty name")
+		return "", errors.New("e2e harness: smallest instance type has empty name")
 	}
 	return w.instanceType, nil
 }
@@ -332,7 +332,7 @@ func (w *Witness) waitForRunning(ctx context.Context, id string, timeout time.Du
 			}
 		}
 		if time.Now().After(deadline) {
-			return fmt.Errorf("ddil harness: instance %s did not reach running within %s", id, timeout)
+			return fmt.Errorf("e2e harness: instance %s did not reach running within %s", id, timeout)
 		}
 		select {
 		case <-ctx.Done():
@@ -348,7 +348,7 @@ func (w *Witness) waitForRunning(ctx context.Context, id string, timeout time.Du
 // exactly one process per cluster regardless of net mode (user-hostfwd vs
 // tap-mode bridge).
 func (w *Witness) findHost(ctx context.Context, instanceID string) (Node, error) {
-	cmd := fmt.Sprintf("ps auxw | grep %s | grep qemu-system | grep -v grep", shellQuote(instanceID))
+	cmd := fmt.Sprintf("ps auxw | grep %s | grep qemu-system | grep -v grep", ShellQuote(instanceID))
 	// Give the daemon a short window to actually spawn QEMU after the
 	// EC2 state flip to running, since /aws/ec2 reports "running" before
 	// the daemon has finished forking the process on some nodes.
@@ -367,7 +367,7 @@ func (w *Witness) findHost(ctx context.Context, instanceID string) (Node, error)
 			}
 		}
 		if time.Now().After(deadline) {
-			return Node{}, fmt.Errorf("ddil harness: could not locate QEMU host for %s across %d nodes", instanceID, len(w.cluster.Nodes))
+			return Node{}, fmt.Errorf("e2e harness: could not locate QEMU host for %s across %d nodes", instanceID, len(w.cluster.Nodes))
 		}
 		select {
 		case <-ctx.Done():
@@ -400,7 +400,7 @@ func (w *Witness) resolvePublicIP(ctx context.Context, instanceID string) (strin
 			}
 		}
 		if time.Now().After(deadline) {
-			return "", fmt.Errorf("ddil harness: instance %s has no PublicIpAddress after %s (default subnet MapPublicIpOnLaunch may be off)", instanceID, 30*time.Second)
+			return "", fmt.Errorf("e2e harness: instance %s has no PublicIpAddress after %s (default subnet MapPublicIpOnLaunch may be off)", instanceID, 30*time.Second)
 		}
 		select {
 		case <-ctx.Done():
@@ -415,7 +415,7 @@ func (w *Witness) terminate(ctx context.Context, id string) error {
 		InstanceIds: []*string{aws.String(id)},
 	})
 	if err != nil {
-		return fmt.Errorf("ddil harness: TerminateInstances %s: %w", id, err)
+		return fmt.Errorf("e2e harness: TerminateInstances %s: %w", id, err)
 	}
 	return nil
 }
@@ -435,7 +435,7 @@ func readCounter(ctx context.Context, w *Witness, host Node, publicIP string) (i
 	guestAddr := net.JoinHostPort(publicIP, "22")
 	tunnel, err := hostClient.DialContext(ctx, "tcp", guestAddr)
 	if err != nil {
-		return 0, fmt.Errorf("ddil harness: tunnel %s → %s: %w", host.Name, guestAddr, err)
+		return 0, fmt.Errorf("e2e harness: tunnel %s → %s: %w", host.Name, guestAddr, err)
 	}
 
 	guestCfg := &ssh.ClientConfig{
@@ -447,20 +447,20 @@ func readCounter(ctx context.Context, w *Witness, host Node, publicIP string) (i
 	c, chans, reqs, err := ssh.NewClientConn(tunnel, guestAddr, guestCfg)
 	if err != nil {
 		_ = tunnel.Close()
-		return 0, fmt.Errorf("ddil harness: guest ssh handshake on %s: %w", guestAddr, err)
+		return 0, fmt.Errorf("e2e harness: guest ssh handshake on %s: %w", guestAddr, err)
 	}
 	guestClient := ssh.NewClient(c, chans, reqs)
 	defer func() { _ = guestClient.Close() }()
 
 	session, err := guestClient.NewSession()
 	if err != nil {
-		return 0, fmt.Errorf("ddil harness: guest ssh session: %w", err)
+		return 0, fmt.Errorf("e2e harness: guest ssh session: %w", err)
 	}
 	defer func() { _ = session.Close() }()
 
 	raw, err := session.CombinedOutput("cat /var/lib/counter")
 	if err != nil {
-		return 0, fmt.Errorf("ddil harness: read /var/lib/counter: %w (output: %s)", err, strings.TrimSpace(string(raw)))
+		return 0, fmt.Errorf("e2e harness: read /var/lib/counter: %w (output: %s)", err, strings.TrimSpace(string(raw)))
 	}
 	s := strings.TrimSpace(string(raw))
 	if s == "" {
@@ -468,7 +468,7 @@ func readCounter(ctx context.Context, w *Witness, host Node, publicIP string) (i
 	}
 	n, err := strconv.Atoi(s)
 	if err != nil {
-		return 0, fmt.Errorf("ddil harness: parse /var/lib/counter %q: %w", s, err)
+		return 0, fmt.Errorf("e2e harness: parse /var/lib/counter %q: %w", s, err)
 	}
 	return n, nil
 }
@@ -509,7 +509,7 @@ func awaitBaseline(ctx context.Context, w *Witness, host Node, publicIP string,
 			return 0, err
 		}
 		if time.Now().After(deadline) {
-			return 0, fmt.Errorf("ddil harness: %s baseline not ready after %s: %w", publicIP, timeout, lastErr)
+			return 0, fmt.Errorf("e2e harness: %s baseline not ready after %s: %w", publicIP, timeout, lastErr)
 		}
 		select {
 		case <-ctx.Done():
@@ -529,12 +529,12 @@ func dialHost(ctx context.Context, node Node, user string, signer ssh.Signer) (*
 	d := net.Dialer{Timeout: cfg.Timeout}
 	conn, err := d.DialContext(ctx, "tcp", net.JoinHostPort(node.Addr, "22"))
 	if err != nil {
-		return nil, fmt.Errorf("ddil harness: dial host %s: %w", node.Name, err)
+		return nil, fmt.Errorf("e2e harness: dial host %s: %w", node.Name, err)
 	}
 	c, chans, reqs, err := ssh.NewClientConn(conn, node.Addr, cfg)
 	if err != nil {
 		_ = conn.Close()
-		return nil, fmt.Errorf("ddil harness: host ssh handshake %s: %w", node.Name, err)
+		return nil, fmt.Errorf("e2e harness: host ssh handshake %s: %w", node.Name, err)
 	}
 	return ssh.NewClient(c, chans, reqs), nil
 }