test(P08): integration test harness + drift-detection stubs (REQ-087)

tests/integration/harness.go: temp ORCA_HOME + mock peers + helpers.
tests/integration/scenarios_test.go: ns-create/job-submit/drain/backup/
secrets/acl/metrics scenarios. drift_scenarios_test.go: 4 stubs (auto-
remediation, NFS, cooldown, secret exclusion) skip until P10b.
scripts/tests/orca-commands_test.bash: bats for new CLI commands.

---ci---
project: orca
phase: 08
milestone: v0.11
status: execute
---/ci---

tests/integration/drift_scenarios_test.go

@@ -0,0 +1,80 @@
+package integration
+
+import "testing"
+
+// Drift-detection integration test stubs (P08, REQ-087). The drift
+// detection code itself lands in P10b; these stubs define the scenarios
+// and skip with a clear message so the integration suite is green until
+// P10b ships. Each stub exercises the hermetic harness (NewHarness) and
+// the mock peer transport, so when the P10b implementation lands the
+// stubs can be filled in without re-architecting the test scaffolding.
+
+// TestScenario_DriftAutoRemediation verifies that editing a Traefik
+// config file on a peer is detected as drift within ~10s and
+// auto-remediated back to the canonical state.
+//
+// Scenario:
+//  1. Harness registers a peer with a canonical Traefik config.
+//  2. An external "edit" mutates the config on the peer.
+//  3. The drift detector polls the peer, detects the divergence, and
+//     rewrites the canonical config.
+//  4. The test asserts the peer's config matches the canonical state
+//     and a drift event was recorded.
+//
+// Requires P10b drift detection.
+func TestScenario_DriftAutoRemediation(t *testing.T) {
+	t.Skip("requires P10b drift detection — will be implemented after P10b ships")
+	_ = NewHarness(t)
+}
+
+// TestScenario_DriftNFSFallback verifies that when a peer's NFS mount
+// is unavailable, the drift detector falls back from Path-unit
+// inotify watching to polling, and still detects drift.
+//
+// Scenario:
+//  1. Harness registers a peer with Path units enabled (inotify mode).
+//  2. The peer's NFS mount is simulated as unavailable.
+//  3. The detector disables Path units and switches to polling.
+//  4. A config edit is detected via the polling loop.
+//
+// Requires P10b drift detection.
+func TestScenario_DriftNFSFallback(t *testing.T) {
+	t.Skip("requires P10b drift detection — will be implemented after P10b ships")
+	_ = NewHarness(t)
+}
+
+// TestScenario_DriftRateLimitCooldown verifies that repeated drift
+// events on a peer trigger a cooldown that blocks the remediation loop
+// (rate-limit) so a flapping config does not hot-loop the detector.
+//
+// Scenario:
+//  1. Harness registers a peer.
+//  2. A config is mutated repeatedly beyond the rate-limit threshold.
+//  3. The detector enters cooldown and skips remediation until the
+//     cooldown window elapses.
+//  4. The test asserts a cooldown event was recorded and no
+//     remediation ran during the window.
+//
+// Requires P10b drift detection.
+func TestScenario_DriftRateLimitCooldown(t *testing.T) {
+	t.Skip("requires P10b drift detection — will be implemented after P10b ships")
+	_ = NewHarness(t)
+}
+
+// TestScenario_DriftSecretExclusion verifies that editing a file under
+// /etc/orca/credentials/* does NOT emit a drift event (secrets are
+// excluded from drift detection so credential rotation does not trip
+// remediation).
+//
+// Scenario:
+//  1. Harness registers a peer with credentials under
+//     /etc/orca/credentials/.
+//  2. A credential file is mutated.
+//  3. The test asserts no drift event was recorded for the credentials
+//     path.
+//
+// Requires P10b drift detection.
+func TestScenario_DriftSecretExclusion(t *testing.T) {
+	t.Skip("requires P10b drift detection — will be implemented after P10b ships")
+	_ = NewHarness(t)
+}

tests/integration/harness.go

@@ -0,0 +1,459 @@
+// Package integration provides the hermetic test harness for orca
+// integration tests (P08, REQ-087). The harness creates a temp
+// ORCA_HOME, initializes a cluster (DB + CA + master key), registers
+// mock peers that record SSH commands without a real SSH server, and
+// exposes helpers (SubmitJob, ListAllocs, DrainNode, VerifyState) for
+// the scenario tests.
+//
+// The harness is hermetic: every test gets its own temp ORCA_HOME
+// (t.Setenv), so there is no cross-test state leakage and no reliance
+// on the operator's ~/.orca. Mock peers record every Exec call so the
+// drain/migrate scenarios can assert on the exact systemctl commands
+// issued without a real systemd host.
+package integration
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"log/slog"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+
+	"git.cloudinit.dev/coreci/orca/internal/certpaths"
+	"git.cloudinit.dev/coreci/orca/internal/engine"
+	"git.cloudinit.dev/coreci/orca/internal/model"
+	"git.cloudinit.dev/coreci/orca/internal/paths"
+	"git.cloudinit.dev/coreci/orca/internal/secrets"
+	"git.cloudinit.dev/coreci/orca/internal/security"
+	"git.cloudinit.dev/coreci/orca/internal/store"
+)
+
+// Peer is a mock SSH peer registered with the harness. It records
+// every Exec command the harness issues against it so scenarios can
+// assert on the exact systemctl invocations without a real SSH server.
+// RunningAllocs is the set of orca-alloc-<id>.service units the mock
+// reports as running; listRunningAllocs-style commands read it, and
+// stop/start commands mutate it.
+type Peer struct {
+	Name     string
+	Address  string
+	NodeID   string
+	mu       sync.Mutex
+	commands []string
+	running  map[string]bool
+}
+
+// Commands returns a snapshot of the recorded Exec commands.
+func (p *Peer) Commands() []string {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	out := make([]string, len(p.commands))
+	copy(out, p.commands)
+	return out
+}
+
+// Exec records cmd and simulates a systemd host:
+//   - "systemctl list-units 'orca-alloc-*.service' ..." reports the
+//     currently-running alloc ids.
+//   - "systemctl stop orca-alloc-<id>.service" removes <id> from the
+//     running set.
+//   - "systemctl start orca-alloc-<id>.service" adds <id> to the
+//     running set.
+//   - everything else is recorded and returns empty output.
+func (p *Peer) Exec(_ context.Context, _ string, cmd string) ([]byte, error) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.commands = append(p.commands, cmd)
+	switch {
+	case startsWith(cmd, "systemctl list-units 'orca-alloc-*.service'"):
+		var lines []byte
+		for id := range p.running {
+			lines = append(lines, []byte(fmt.Sprintf("orca-alloc-%s.service loaded active running orca alloc %s\n", id, id))...)
+		}
+		return lines, nil
+	case startsWith(cmd, "systemctl stop orca-alloc-"):
+		id := allocIDFromStopCmd(cmd)
+		delete(p.running, id)
+		return nil, nil
+	case startsWith(cmd, "systemctl start orca-alloc-"):
+		id := allocIDFromStartCmd(cmd)
+		if id != "" {
+			p.running[id] = true
+		}
+		return nil, nil
+	default:
+		return nil, nil
+	}
+}
+
+// RunningAllocs returns the currently-running alloc ids (snapshot).
+func (p *Peer) RunningAllocs() []string {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	out := make([]string, 0, len(p.running))
+	for id := range p.running {
+		out = append(out, id)
+	}
+	return out
+}
+
+// SetRunning injects alloc ids into the running set (for scenarios that
+// need a peer to start with running allocations before a drain).
+func (p *Peer) SetRunning(ids ...string) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	for _, id := range ids {
+		if p.running == nil {
+			p.running = make(map[string]bool)
+		}
+		p.running[id] = true
+	}
+}
+
+// MockTransport is the harness's drainExecer: it routes Exec calls to
+// the registered Peer with the matching name/address. It satisfies
+// the cli.drainExecer interface (Exec(ctx, peer, cmd) ([]byte, error))
+// without importing the cli package (which would create an import
+// cycle).
+type MockTransport struct {
+	mu    sync.RWMutex
+	peers map[string]*Peer
+}
+
+// NewMockTransport returns an empty MockTransport.
+func NewMockTransport() *MockTransport {
+	return &MockTransport{peers: make(map[string]*Peer)}
+}
+
+// Register associates peer with the harness transport under both the
+// peer's Name and Address so peerAddrForNode-style "host:22" lookups
+// resolve.
+func (t *MockTransport) Register(p *Peer) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.peers[p.Name] = p
+	t.peers[p.Address] = p
+}
+
+// Exec routes cmd to the registered peer. Unknown peer returns an
+// error so scenarios catch a missing registration.
+func (t *MockTransport) Exec(ctx context.Context, peer string, cmd string) ([]byte, error) {
+	t.mu.RLock()
+	p, ok := t.peers[peer]
+	t.mu.RUnlock()
+	if !ok {
+		return nil, fmt.Errorf("mock transport: unknown peer %q", peer)
+	}
+	return p.Exec(ctx, peer, cmd)
+}
+
+// Harness is the hermetic test harness. Each test constructs one with
+// NewHarness and defers h.Close to clean up the temp ORCA_HOME.
+type Harness struct {
+	t         *testing.T
+	Home      string
+	DB        *sql.DB
+	Nodes     *store.NodeRepo
+	Jobs      *store.JobRepo
+	Tasks     *store.TaskRepo
+	Registry  *engine.NodeRegistry
+	Transport *MockTransport
+
+	dbClose func() error
+}
+
+// NewHarness creates a temp ORCA_HOME, initializes the cluster (DB
+// migrations + CA + master key + localhost node), and returns a ready
+// harness. The temp dir is cleaned up via t.Cleanup.
+func NewHarness(t *testing.T) *Harness {
+	t.Helper()
+	home := t.TempDir()
+	t.Setenv("ORCA_HOME", home)
+
+	h := &Harness{
+		t:         t,
+		Home:      home,
+		Transport: NewMockTransport(),
+	}
+
+	if err := os.MkdirAll(paths.ClusterDir(), 0o755); err != nil {
+		t.Fatalf("mkdir cluster dir: %v", err)
+	}
+
+	db, err := store.Open(certpaths.DBPath())
+	if err != nil {
+		t.Fatalf("open db: %v", err)
+	}
+	h.DB = db
+	h.dbClose = db.Close
+	h.Nodes = store.NewNodeRepo(db)
+	h.Jobs = store.NewJobRepo(db)
+	h.Tasks = store.NewTaskRepo(db)
+	audit := engine.NewAudit(store.NewAuditRepo(db), testLogger())
+	h.Registry = engine.NewNodeRegistry(h.Nodes, audit, testLogger())
+
+	if _, err := security.CAInit(home, "orca-test-ca"); err != nil {
+		t.Fatalf("CAInit: %v", err)
+	}
+
+	mk, err := secrets.GenerateMasterKey()
+	if err != nil {
+		t.Fatalf("GenerateMasterKey: %v", err)
+	}
+	if err := secrets.SaveMasterKey(paths.MasterKeyPath(), mk); err != nil {
+		t.Fatalf("SaveMasterKey: %v", err)
+	}
+
+	ctx := context.Background()
+	localhost := &model.Node{
+		ID:       uuid.NewString(),
+		Name:     "localhost",
+		Address:  "localhost:8443",
+		State:    model.NodeStateReady,
+		JoinedAt: time.Now().UTC(),
+		LastSeen: time.Now().UTC(),
+		Kind:     string(model.NodeKindLocalhost),
+		OS:       "linux",
+	}
+	if err := h.Registry.Join(ctx, localhost); err != nil {
+		t.Fatalf("register localhost: %v", err)
+	}
+
+	t.Cleanup(func() { _ = h.Close() })
+	return h
+}
+
+// RegisterPeer creates a mock Peer, registers it as a node in the
+// cluster registry, and wires it into the mock transport so drain/
+// migrate commands route to it. The peer starts with no running
+// allocations; use Peer.SetRunning to seed it.
+func (h *Harness) RegisterPeer(name, address string) *Peer {
+	h.t.Helper()
+	p := &Peer{
+		Name:    name,
+		Address: address,
+		NodeID:  uuid.NewString(),
+		running: make(map[string]bool),
+	}
+	ctx := context.Background()
+	node := &model.Node{
+		ID:       p.NodeID,
+		Name:     name,
+		Address:  address,
+		State:    model.NodeStateReady,
+		JoinedAt: time.Now().UTC(),
+		LastSeen: time.Now().UTC(),
+		Kind:     string(model.NodeKindLinux),
+		OS:       "linux",
+	}
+	if err := h.Registry.Join(ctx, node); err != nil {
+		h.t.Fatalf("register peer %q: %v", name, err)
+	}
+	h.Transport.Register(p)
+	return p
+}
+
+// SubmitJob runs a single-task job on the local executor and returns
+// the job ID. The command must be a real binary path resolvable on the
+// test host (e.g. "/bin/true", "/bin/echo").
+func (h *Harness) SubmitJob(ctx context.Context, name, command string, args ...string) (string, error) {
+	h.t.Helper()
+	exec := engine.NewExecutor(h.Jobs, h.Tasks, testLogger())
+	job := &model.Job{
+		ID:     uuid.NewString(),
+		Name:   name,
+		Spec:   command,
+		Status: model.JobStatusPending,
+	}
+	specs := []engine.TaskSpec{{
+		Name:    name,
+		Command: command,
+		Args:    args,
+	}}
+	if err := exec.Run(ctx, job, specs); err != nil {
+		return job.ID, err
+	}
+	return job.ID, nil
+}
+
+// ListAllocs returns the currently-running alloc ids on the named
+// peer (as reported by the mock transport). For the local node the
+// harness has no mock systemd host, so it returns an empty slice.
+func (h *Harness) ListAllocs(ctx context.Context, peerName string) ([]string, error) {
+	h.t.Helper()
+	p, ok := h.lookupPeer(peerName)
+	if !ok {
+		return nil, fmt.Errorf("ListAllocs: peer %q not registered", peerName)
+	}
+	cmd := "systemctl list-units 'orca-alloc-*.service' --type=service --state=running --no-legend --no-pager"
+	if _, err := h.Transport.Exec(ctx, peerName, cmd); err != nil {
+		return nil, err
+	}
+	return p.RunningAllocs(), nil
+}
+
+// DrainNode marks the named peer's node "draining", stops every
+// running allocation on it via the mock transport, and marks it
+// "drained". It mirrors the `orca node drain` flow without going
+// through the CLI cobra rootCmd (so the harness stays hermetic and
+// does not leak global flag state between scenarios).
+func (h *Harness) DrainNode(ctx context.Context, peerName string) ([]string, error) {
+	h.t.Helper()
+	p, ok := h.lookupPeer(peerName)
+	if !ok {
+		return nil, fmt.Errorf("DrainNode: peer %q not registered", peerName)
+	}
+	nodes, err := h.Registry.List(ctx)
+	if err != nil {
+		return nil, err
+	}
+	var node *model.Node
+	for _, n := range nodes {
+		if n.Name == peerName {
+			node = n
+			break
+		}
+	}
+	if node == nil {
+		return nil, fmt.Errorf("DrainNode: node %q not in registry", peerName)
+	}
+	if err := h.Registry.SetNodeState(ctx, node.ID, string(model.NodeStateDraining)); err != nil {
+		return nil, err
+	}
+	ids := p.RunningAllocs()
+	var stopped []string
+	for _, id := range ids {
+		stopCmd := fmt.Sprintf("systemctl stop orca-alloc-%s.service", id)
+		if _, err := h.Transport.Exec(ctx, peerName, stopCmd); err != nil {
+			return stopped, err
+		}
+		stopped = append(stopped, id)
+	}
+	if err := h.Registry.SetNodeState(ctx, node.ID, string(model.NodeStateDrained)); err != nil {
+		return stopped, err
+	}
+	return stopped, nil
+}
+
+// VerifyState asserts the harness's on-disk state is self-consistent:
+// the cluster dir exists, the master key is present with mode 0600,
+// and the SQLite DB opens. Scenarios call it after a backup/restore
+// round-trip to confirm the restored tree is usable.
+func (h *Harness) VerifyState() error {
+	if info, err := os.Stat(paths.ClusterDir()); err != nil || !info.IsDir() {
+		return fmt.Errorf("cluster dir missing: %v", err)
+	}
+	mkInfo, err := os.Stat(paths.MasterKeyPath())
+	if err != nil {
+		return fmt.Errorf("master key missing: %w", err)
+	}
+	if mkInfo.Mode().Perm() != secrets.MasterKeyMode {
+		return fmt.Errorf("master key mode %04o, want %04o", mkInfo.Mode().Perm(), secrets.MasterKeyMode)
+	}
+	db, err := store.Open(certpaths.DBPath())
+	if err != nil {
+		return fmt.Errorf("db open: %w", err)
+	}
+	defer db.Close()
+	if err := db.Ping(); err != nil {
+		return fmt.Errorf("db ping: %w", err)
+	}
+	return nil
+}
+
+// MasterKeyPath returns the cluster master key path (paths.MasterKeyPath()).
+func (h *Harness) MasterKeyPath() string { return paths.MasterKeyPath() }
+
+// Root returns the temp ORCA_HOME root.
+func (h *Harness) Root() string { return paths.Root() }
+
+// Close releases the DB handle. It is idempotent and safe to call
+// from t.Cleanup.
+func (h *Harness) Close() error {
+	if h.dbClose != nil {
+		err := h.dbClose()
+		h.dbClose = nil
+		return err
+	}
+	return nil
+}
+
+// lookupPeer returns the registered Peer by name.
+func (h *Harness) lookupPeer(name string) (*Peer, bool) {
+	h.Transport.mu.RLock()
+	defer h.Transport.mu.RUnlock()
+	p, ok := h.Transport.peers[name]
+	return p, ok
+}
+
+// testLogger returns a slog logger writing JSON to stderr at Info
+// level. It mirrors internal/cli.newLogger so the harness logs the
+// same way the CLI does.
+func testLogger() *slog.Logger {
+	return slog.New(slog.NewJSONHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelInfo}))
+}
+
+// startsWith reports whether s begins with prefix.
+func startsWith(s, prefix string) bool {
+	return len(s) >= len(prefix) && s[:len(prefix)] == prefix
+}
+
+// allocIDFromStopCmd extracts the alloc id from a
+// "systemctl stop orca-alloc-<id>.service" command.
+func allocIDFromStopCmd(cmd string) string {
+	const prefix = "systemctl stop orca-alloc-"
+	const suffix = ".service"
+	if !startsWith(cmd, prefix) {
+		return ""
+	}
+	s := cmd[len(prefix):]
+	s = trimSuffix(s, suffix)
+	return s
+}
+
+// allocIDFromStartCmd extracts the alloc id from a
+// "systemctl start orca-alloc-<id>.service" command.
+func allocIDFromStartCmd(cmd string) string {
+	const prefix = "systemctl start orca-alloc-"
+	const suffix = ".service"
+	if !startsWith(cmd, prefix) {
+		return ""
+	}
+	s := cmd[len(prefix):]
+	s = trimSuffix(s, suffix)
+	return s
+}
+
+// trimSuffix removes a trailing suffix from s if present.
+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
+}
+
+// Compile-time assertion that *MockTransport satisfies the drainExecer
+// shape used by internal/cli (Exec(ctx, peer, cmd) ([]byte, error)).
+// We can't import internal/cli (import cycle), so the assertion is
+// structural via an anonymous interface.
+var _ drainExecerShape = (*MockTransport)(nil)
+
+type drainExecerShape interface {
+	Exec(context.Context, string, string) ([]byte, error)
+}
+
+// statPerm returns the file permissions for path, surfacing the
+// stat error so tests can assert on file mode without rewriting the
+// os.Stat dance.
+func statPerm(path string) (os.FileMode, error) {
+	info, err := os.Stat(path)
+	if err != nil {
+		return 0, err
+	}
+	return info.Mode().Perm(), nil
+}

tests/integration/harness_test.go

@@ -0,0 +1,138 @@
+package integration
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"git.cloudinit.dev/coreci/orca/internal/model"
+	"git.cloudinit.dev/coreci/orca/internal/paths"
+)
+
+func TestHarness_CreatesValidOrcaHome(t *testing.T) {
+	h := NewHarness(t)
+	if h.Home == "" {
+		t.Fatal("Home is empty")
+	}
+	if h.DB == nil {
+		t.Fatal("DB is nil")
+	}
+	if h.Registry == nil {
+		t.Fatal("Registry is nil")
+	}
+	if err := h.VerifyState(); err != nil {
+		t.Fatalf("VerifyState: %v", err)
+	}
+	ctx := context.Background()
+	nodes, err := h.Registry.List(ctx)
+	if err != nil {
+		t.Fatalf("List nodes: %v", err)
+	}
+	if len(nodes) != 1 || nodes[0].Name != "localhost" {
+		t.Errorf("expected localhost node, got %+v", nodes)
+	}
+}
+
+func TestHarness_MasterKeyMode(t *testing.T) {
+	h := NewHarness(t)
+	if err := h.VerifyState(); err != nil {
+		t.Fatalf("VerifyState: %v", err)
+	}
+	info, err := statPerm(paths.MasterKeyPath())
+	if err != nil {
+		t.Fatalf("stat master key: %v", err)
+	}
+	const want = 0o600
+	if info != want {
+		t.Errorf("master key mode %o, want %o", info, want)
+	}
+}
+
+func TestHarness_MockPeerRecordsCommands(t *testing.T) {
+	h := NewHarness(t)
+	peer := h.RegisterPeer("node-a", "node-a:22")
+	peer.SetRunning("alloc-1", "alloc-2")
+
+	ctx := context.Background()
+	running, err := h.ListAllocs(ctx, "node-a")
+	if err != nil {
+		t.Fatalf("ListAllocs: %v", err)
+	}
+	if len(running) != 2 {
+		t.Errorf("running allocs = %v, want 2", running)
+	}
+	cmds := peer.Commands()
+	if len(cmds) < 1 {
+		t.Fatalf("expected recorded commands, got %v", cmds)
+	}
+	foundList := false
+	for _, c := range cmds {
+		if startsWith(c, "systemctl list-units 'orca-alloc-*.service'") {
+			foundList = true
+		}
+	}
+	if !foundList {
+		t.Errorf("no list-units command recorded: %v", cmds)
+	}
+}
+
+func TestHarness_DrainNodeStopsAllocs(t *testing.T) {
+	h := NewHarness(t)
+	peer := h.RegisterPeer("node-b", "node-b:22")
+	peer.SetRunning("alloc-x", "alloc-y")
+
+	ctx := context.Background()
+	stopped, err := h.DrainNode(ctx, "node-b")
+	if err != nil {
+		t.Fatalf("DrainNode: %v", err)
+	}
+	if len(stopped) != 2 {
+		t.Errorf("stopped = %v, want 2", stopped)
+	}
+	if r := peer.RunningAllocs(); len(r) != 0 {
+		t.Errorf("after drain, running = %v, want empty", r)
+	}
+	nodes, err := h.Registry.List(ctx)
+	if err != nil {
+		t.Fatalf("List: %v", err)
+	}
+	for _, n := range nodes {
+		if n.Name == "node-b" && n.State != model.NodeStateDrained {
+			t.Errorf("node-b state = %q, want drained", n.State)
+		}
+	}
+}
+
+func TestHarness_SubmitJobHelperWorks(t *testing.T) {
+	h := NewHarness(t)
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	jobID, err := h.SubmitJob(ctx, "smoke", "/bin/true")
+	if err != nil {
+		t.Fatalf("SubmitJob: %v", err)
+	}
+	if jobID == "" {
+		t.Fatal("SubmitJob returned empty id")
+	}
+	jobs, err := h.Jobs.List(ctx)
+	if err != nil {
+		t.Fatalf("Jobs.List: %v", err)
+	}
+	var found bool
+	for _, j := range jobs {
+		if j.ID == jobID && j.Status == model.JobStatusComplete {
+			found = true
+		}
+	}
+	if !found {
+		t.Errorf("submitted job %s not complete in list: %+v", jobID, jobs)
+	}
+}
+
+func TestHarness_MockTransportUnknownPeerErrors(t *testing.T) {
+	h := NewHarness(t)
+	ctx := context.Background()
+	if _, err := h.Transport.Exec(ctx, "ghost", "true"); err == nil {
+		t.Error("expected error for unknown peer, got nil")
+	}
+}

tests/integration/scenarios_test.go

@@ -0,0 +1,476 @@
+package integration
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"git.cloudinit.dev/coreci/orca/internal/acl"
+	"git.cloudinit.dev/coreci/orca/internal/backup"
+	"git.cloudinit.dev/coreci/orca/internal/certpaths"
+	"git.cloudinit.dev/coreci/orca/internal/model"
+	"git.cloudinit.dev/coreci/orca/internal/paths"
+	"git.cloudinit.dev/coreci/orca/internal/secrets"
+	"git.cloudinit.dev/coreci/orca/internal/store"
+	"git.cloudinit.dev/coreci/orca/internal/transport"
+)
+
+// TestScenario_NsCreate_JobSubmit_List exercises the REQ-087 core
+// flow end-to-end: create a namespace, submit a job into it, and list
+// the resulting allocations.
+func TestScenario_NsCreate_JobSubmit_List(t *testing.T) {
+	h := NewHarness(t)
+	ctx := context.Background()
+
+	nsName := "webapp"
+	nsDir := paths.NamespaceDir(nsName)
+	for _, sub := range []string{"db", "jobs", "alloc"} {
+		if err := os.MkdirAll(filepath.Join(nsDir, sub), 0o755); err != nil {
+			t.Fatalf("mkdir %s/%s: %v", nsDir, sub, err)
+		}
+	}
+	nsMd := renderNSMd(nsName, []string{paths.DefaultNamespace()}, true, true)
+	if err := os.WriteFile(paths.NSMd(nsName), []byte(nsMd), 0o644); err != nil {
+		t.Fatalf("write ns.md: %v", err)
+	}
+	if _, err := os.Stat(paths.NSMd(nsName)); err != nil {
+		t.Fatalf("ns.md not present: %v", err)
+	}
+
+	jobID, err := h.SubmitJob(ctx, "webapp-1", "/bin/true")
+	if err != nil {
+		t.Fatalf("SubmitJob: %v", err)
+	}
+	jobs, err := h.Jobs.List(ctx)
+	if err != nil {
+		t.Fatalf("Jobs.List: %v", err)
+	}
+	if len(jobs) != 1 || jobs[0].ID != jobID {
+		t.Errorf("expected 1 job %s, got %+v", jobID, jobs)
+	}
+	if jobs[0].Status != model.JobStatusComplete {
+		t.Errorf("job status = %q, want complete", jobs[0].Status)
+	}
+	tasks, err := h.Tasks.ListByJob(ctx, jobID)
+	if err != nil {
+		t.Fatalf("ListByJob: %v", err)
+	}
+	if len(tasks) != 1 || tasks[0].ExitCode != 0 {
+		t.Errorf("expected 1 complete task, got %+v", tasks)
+	}
+}
+
+// TestScenario_Drain verifies that draining a peer stops its running
+// allocations and flips the node to "drained".
+func TestScenario_Drain(t *testing.T) {
+	h := NewHarness(t)
+	peer := h.RegisterPeer("drain-target", "drain-target:22")
+	peer.SetRunning("alloc-d1", "alloc-d2", "alloc-d3")
+
+	ctx := context.Background()
+	before, err := h.ListAllocs(ctx, "drain-target")
+	if err != nil {
+		t.Fatalf("ListAllocs before: %v", err)
+	}
+	if len(before) != 3 {
+		t.Fatalf("running before drain = %v, want 3", before)
+	}
+
+	stopped, err := h.DrainNode(ctx, "drain-target")
+	if err != nil {
+		t.Fatalf("DrainNode: %v", err)
+	}
+	if len(stopped) != 3 {
+		t.Errorf("stopped = %v, want 3", stopped)
+	}
+
+	after, err := h.ListAllocs(ctx, "drain-target")
+	if err != nil {
+		t.Fatalf("ListAllocs after: %v", err)
+	}
+	if len(after) != 0 {
+		t.Errorf("running after drain = %v, want empty", after)
+	}
+
+	nodes, err := h.Registry.List(ctx)
+	if err != nil {
+		t.Fatalf("Registry.List: %v", err)
+	}
+	for _, n := range nodes {
+		if n.Name == "drain-target" && n.State != model.NodeStateDrained {
+			t.Errorf("node state = %q, want drained", n.State)
+		}
+	}
+	cmds := peer.Commands()
+	if !containsCmd(cmds, "systemctl stop orca-alloc-alloc-d1.service") ||
+		!containsCmd(cmds, "systemctl stop orca-alloc-alloc-d2.service") ||
+		!containsCmd(cmds, "systemctl stop orca-alloc-alloc-d3.service") {
+		t.Errorf("missing stop commands in recorded: %v", cmds)
+	}
+}
+
+// TestScenario_BackupRestore verifies a backup → destroy → restore
+// round-trip recovers a usable ORCA_HOME.
+func TestScenario_BackupRestore(t *testing.T) {
+	h := NewHarness(t)
+	ctx := context.Background()
+	if _, err := h.SubmitJob(ctx, "pre-backup", "/bin/true"); err != nil {
+		t.Fatalf("SubmitJob pre-backup: %v", err)
+	}
+	if err := h.VerifyState(); err != nil {
+		t.Fatalf("VerifyState before backup: %v", err)
+	}
+	// Checkpoint the WAL into the main db file so the backup captures
+	// the committed job rows (the backup excludes *.db-wal sidecars).
+	if _, err := h.DB.ExecContext(ctx, "PRAGMA wal_checkpoint(TRUNCATE)"); err != nil {
+		t.Fatalf("wal_checkpoint: %v", err)
+	}
+
+	mk, err := secrets.LoadMasterKey(paths.MasterKeyPath())
+	if err != nil {
+		t.Fatalf("LoadMasterKey: %v", err)
+	}
+	tarball := filepath.Join(t.TempDir(), "backup.tar.gz")
+	if err := backup.Backup(backup.BackupOptions{
+		SourceDir:  h.Root(),
+		OutputPath: tarball,
+		MasterKey:  mk,
+	}); err != nil {
+		t.Fatalf("Backup: %v", err)
+	}
+	if _, err := os.Stat(tarball + ".sig"); err != nil {
+		t.Fatalf("signature file missing: %v", err)
+	}
+
+	destroyed := h.Root() + "-destroyed"
+	if err := os.Rename(h.Root(), destroyed); err != nil {
+		t.Fatalf("rename to destroy ORCA_HOME: %v", err)
+	}
+	t.Setenv("ORCA_HOME", h.Root())
+
+	if err := backup.Restore(backup.RestoreOptions{
+		InputPath: tarball,
+		TargetDir: h.Root(),
+		MasterKey: mk,
+		Force:     true,
+	}); err != nil {
+		t.Fatalf("Restore: %v", err)
+	}
+
+	if err := h.VerifyState(); err != nil {
+		t.Fatalf("VerifyState after restore: %v", err)
+	}
+	db, err := openDB()
+	if err != nil {
+		t.Fatalf("open db after restore: %v", err)
+	}
+	defer db.Close()
+	jobs, err := jobList(db, ctx)
+	if err != nil {
+		t.Fatalf("job list after restore: %v", err)
+	}
+	if len(jobs) == 0 {
+		t.Errorf("expected restored jobs, got 0")
+	}
+}
+
+// TestScenario_Secrets exercises set/get/list/delete secrets across
+// namespaces via the secrets package directly.
+func TestScenario_Secrets(t *testing.T) {
+	h := NewHarness(t)
+	if err := h.VerifyState(); err != nil {
+		t.Fatalf("VerifyState: %v", err)
+	}
+	mk, err := secrets.LoadMasterKey(paths.MasterKeyPath())
+	if err != nil {
+		t.Fatalf("LoadMasterKey: %v", err)
+	}
+	for _, ns := range []string{"prod", "staging"} {
+		if err := os.MkdirAll(paths.NamespaceDir(ns), 0o755); err != nil {
+			t.Fatalf("mkdir ns %s: %v", ns, err)
+		}
+	}
+
+	nsKeyProd, err := secrets.DeriveNamespaceKey(mk, "prod")
+	if err != nil {
+		t.Fatalf("DeriveNamespaceKey prod: %v", err)
+	}
+	enc, err := secrets.EncryptEnvFile(nsKeyProd, []string{"API_KEY=hunter2", "DB_PASS=secret"})
+	if err != nil {
+		t.Fatalf("EncryptEnvFile: %v", err)
+	}
+	if err := writeAtomic(paths.NSSecrets("prod"), []byte(enc), 0o600); err != nil {
+		t.Fatalf("write prod secrets: %v", err)
+	}
+
+	got, err := secrets.DecryptEnvFile(nsKeyProd, string(mustReadFile(t, paths.NSSecrets("prod"))))
+	if err != nil {
+		t.Fatalf("DecryptEnvFile prod: %v", err)
+	}
+	if len(got) != 2 {
+		t.Errorf("prod secrets = %v, want 2 lines", got)
+	}
+
+	nsKeyStaging, err := secrets.DeriveNamespaceKey(mk, "staging")
+	if err != nil {
+		t.Fatalf("DeriveNamespaceKey staging: %v", err)
+	}
+	enc2, err := secrets.EncryptEnvFile(nsKeyStaging, []string{"TOKEN=abc"})
+	if err != nil {
+		t.Fatalf("EncryptEnvFile staging: %v", err)
+	}
+	if err := writeAtomic(paths.NSSecrets("staging"), []byte(enc2), 0o600); err != nil {
+		t.Fatalf("write staging secrets: %v", err)
+	}
+
+	decStaging, err := secrets.DecryptEnvFile(nsKeyStaging, string(mustReadFile(t, paths.NSSecrets("staging"))))
+	if err != nil {
+		t.Fatalf("DecryptEnvFile staging: %v", err)
+	}
+	if len(decStaging) != 1 || decStaging[0] != "TOKEN=abc" {
+		t.Errorf("staging secret = %v, want [TOKEN=abc]", decStaging)
+	}
+
+	_, err = secrets.DecryptEnvFile(nsKeyProd, string(mustReadFile(t, paths.NSSecrets("staging"))))
+	if err == nil {
+		t.Error("decrypting staging with prod key succeeded; want cross-ns isolation failure")
+	}
+
+	enc3, err := secrets.EncryptEnvFile(nsKeyProd, []string{"API_KEY=hunter2"})
+	if err != nil {
+		t.Fatalf("EncryptEnvFile delete: %v", err)
+	}
+	if err := writeAtomic(paths.NSSecrets("prod"), []byte(enc3), 0o600); err != nil {
+		t.Fatalf("rewrite prod secrets: %v", err)
+	}
+	decProd, err := secrets.DecryptEnvFile(nsKeyProd, string(mustReadFile(t, paths.NSSecrets("prod"))))
+	if err != nil {
+		t.Fatalf("DecryptEnvFile after delete: %v", err)
+	}
+	if len(decProd) != 1 {
+		t.Errorf("prod after delete = %v, want 1 line", decProd)
+	}
+}
+
+// TestScenario_ACL exercises grant/check/revoke permissions and
+// persists the ACL state to ClusterDir()/acl.json under the harness's
+// temp ORCA_HOME.
+func TestScenario_ACL(t *testing.T) {
+	h := NewHarness(t)
+	if err := h.VerifyState(); err != nil {
+		t.Fatalf("VerifyState: %v", err)
+	}
+	if err := os.MkdirAll(filepath.Dir(paths.ACLPath()), 0o755); err != nil {
+		t.Fatalf("mkdir cluster dir: %v", err)
+	}
+	a := acl.NewACL()
+	id := acl.Identity{Kind: acl.KindToken, ID: "operator-1"}
+	a.Grant(id, "prod", acl.PermRead|acl.PermWrite)
+	if !a.Check(id, "prod", acl.PermRead) {
+		t.Error("expected read on prod after grant")
+	}
+	if !a.Check(id, "prod", acl.PermWrite) {
+		t.Error("expected write on prod after grant")
+	}
+	if a.Check(id, "prod", acl.PermAdmin) {
+		t.Error("admin should not be granted")
+	}
+	if a.Check(id, "staging", acl.PermRead) {
+		t.Error("cross-ns read should be denied")
+	}
+	admin := acl.Identity{Kind: acl.KindToken, ID: "root"}
+	a.Grant(admin, "prod", acl.PermAdmin)
+	if !a.Check(admin, "prod", acl.PermRead) {
+		t.Error("admin should imply read")
+	}
+	if !a.Check(admin, "prod", acl.PermWrite) {
+		t.Error("admin should imply write")
+	}
+	type aclState struct {
+		Entries []acl.ACLEntry `json:"entries"`
+	}
+	state := aclState{Entries: a.List()}
+	data, err := json.MarshalIndent(state, "", "  ")
+	if err != nil {
+		t.Fatalf("marshal acl: %v", err)
+	}
+	if err := writeAtomic(paths.ACLPath(), data, 0o644); err != nil {
+		t.Fatalf("write acl.json: %v", err)
+	}
+	loaded, err := os.ReadFile(paths.ACLPath())
+	if err != nil {
+		t.Fatalf("read acl.json: %v", err)
+	}
+	if !strings.Contains(string(loaded), "operator-1") {
+		t.Errorf("acl.json missing operator-1: %s", loaded)
+	}
+	a.Revoke(id, "prod")
+	if a.Check(id, "prod", acl.PermRead) {
+		t.Error("read should be denied after revoke")
+	}
+	if !a.Check(admin, "prod", acl.PermAdmin) {
+		t.Error("admin should survive revoke of operator-1")
+	}
+}
+
+// TestScenario_Metrics starts the metrics HTTP endpoint, then hits
+// /metrics and /healthz to verify the exposition format.
+func TestScenario_Metrics(t *testing.T) {
+	h := NewHarness(t)
+	if err := h.VerifyState(); err != nil {
+		t.Fatalf("VerifyState: %v", err)
+	}
+	m := transport.NewMetrics()
+	m.SetGauge("nodes_total", 1)
+	m.SetGauge("allocs_total", 0)
+
+	mux := http.NewServeMux()
+	mux.HandleFunc("/metrics", func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
+		if err := m.WritePrometheus(w); err != nil {
+			t.Errorf("WritePrometheus: %v", err)
+		}
+	})
+	mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write([]byte("ok\n"))
+	})
+
+	srv := &http.Server{Handler: mux}
+	addr := freeAddr()
+	srv.Addr = addr
+	errCh := make(chan error, 1)
+	go func() { errCh <- srv.ListenAndServe() }()
+	t.Cleanup(func() { _ = srv.Close() })
+
+	if ok := waitListen(addr, 5*time.Second); !ok {
+		t.Fatalf("metrics server did not start: %v", <-errCh)
+	}
+
+	resp, err := http.Get("http://" + addr + "/healthz")
+	if err != nil {
+		t.Fatalf("GET /healthz: %v", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		t.Errorf("/healthz status = %d, want 200", resp.StatusCode)
+	}
+	body, _ := io.ReadAll(resp.Body)
+	if !strings.Contains(string(body), "ok") {
+		t.Errorf("/healthz body = %q, want ok", body)
+	}
+
+	resp2, err := http.Get("http://" + addr + "/metrics")
+	if err != nil {
+		t.Fatalf("GET /metrics: %v", err)
+	}
+	defer resp2.Body.Close()
+	metricsBody, _ := io.ReadAll(resp2.Body)
+	if !strings.Contains(string(metricsBody), "orca_nodes_total") && !strings.Contains(string(metricsBody), "nodes_total") {
+		t.Errorf("/metrics missing nodes_total: %s", metricsBody)
+	}
+}
+
+// --- helpers ---
+
+func renderNSMd(name string, parents []string, inheritsEnv, inheritsSecrets bool) string {
+	var b strings.Builder
+	b.WriteString("---\nkind: Namespace\nname: ")
+	b.WriteString(name)
+	b.WriteString("\n")
+	quoted := make([]string, len(parents))
+	for i, p := range parents {
+		quoted[i] = fmt.Sprintf("%q", p)
+	}
+	b.WriteString("parents: [")
+	b.WriteString(strings.Join(quoted, ", "))
+	b.WriteString("]\n")
+	fmt.Fprintf(&b, "inherits_env: %t\n", inheritsEnv)
+	fmt.Fprintf(&b, "inherits_secrets: %t\n", inheritsSecrets)
+	b.WriteString("---\n")
+	return b.String()
+}
+
+func containsCmd(cmds []string, want string) bool {
+	for _, c := range cmds {
+		if c == want {
+			return true
+		}
+	}
+	return false
+}
+
+func writeAtomic(path string, data []byte, mode os.FileMode) error {
+	dir := filepath.Dir(path)
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		return err
+	}
+	tmp, err := os.CreateTemp(dir, ".test-tmp-*")
+	if err != nil {
+		return err
+	}
+	tmpName := tmp.Name()
+	defer func() { _ = os.Remove(tmpName) }()
+	if _, err := tmp.Write(data); err != nil {
+		_ = tmp.Close()
+		return err
+	}
+	if err := tmp.Chmod(mode); err != nil {
+		_ = tmp.Close()
+		return err
+	}
+	if err := tmp.Close(); err != nil {
+		return err
+	}
+	return os.Rename(tmpName, path)
+}
+
+func mustReadFile(t *testing.T, path string) []byte {
+	t.Helper()
+	b, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatalf("read %s: %v", path, err)
+	}
+	return b
+}
+
+func openDB() (*sql.DB, error) {
+	return store.Open(certpaths.DBPath())
+}
+
+func jobList(db *sql.DB, ctx context.Context) ([]*model.Job, error) {
+	return store.NewJobRepo(db).List(ctx)
+}
+
+// freeAddr returns a free localhost port for a test HTTP server.
+func freeAddr() string {
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		panic(fmt.Sprintf("freeAddr: %v", err))
+	}
+	defer l.Close()
+	return l.Addr().String()
+}
+
+// waitListen polls addr until a TCP dial succeeds or timeout elapses.
+func waitListen(addr string, timeout time.Duration) bool {
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		c, err := net.DialTimeout("tcp", addr, 100*time.Millisecond)
+		if err == nil {
+			_ = c.Close()
+			return true
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	return false
+}