feat(P09): collector + aggregator (C-11/C-12/C-14) + drift aggregation (REQ-107)

scripts/orca-aggregate.sh: 10s aggregator, cluster.json merge +
drift-events rsync + remediation trigger (P10b stub). scripts/orca-
watchdog.sh: C-11 starvation detection. internal/cli/collector.go:
orca collector start/stop/status. Tests: CLI + bats.

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

internal/cli/collector.go

@@ -0,0 +1,326 @@
+// Package cli: collector.go implements the `orca collector` subcommand
+// (P09, C-12 opt-in). The collector is the lead-side aggregator +
+// watchdog pair: `orca-aggregate.sh` runs every 10s (via a systemd
+// timer) merging per-peer state snapshots into cluster.json, and
+// `orca-watchdog.sh` runs every 30s detecting aggregator starvation
+// (C-11).
+//
+// `orca collector start` emits the scripts + systemd timers/services
+// to the lead and enables them. `orca collector stop` disables and
+// removes them. `orca collector status` reports whether the pair is
+// running.
+//
+// Paths default to the system layout (/etc/orca, /etc/systemd/system);
+// a `--root` flag (default "/") relocates every emitted path under
+// <root> for testability (tests use a temp dir).
+package cli
+
+import (
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+
+	"github.com/spf13/cobra"
+)
+
+var collectorRoot string
+
+const (
+	collectorScriptDir   = "etc/orca/collector"
+	collectorUnitDir      = "etc/systemd/system"
+	collectorAggregateSh  = "orca-aggregate.sh"
+	collectorWatchdogSh   = "orca-watchdog.sh"
+	collectorAggregateSvc = "orca-aggregate.service"
+	collectorAggregateTmr = "orca-aggregate.timer"
+	collectorWatchdogSvc  = "orca-watchdog.service"
+	collectorWatchdogTmr  = "orca-watchdog.timer"
+	collectorStateDir     = "etc/orca/state"
+)
+
+var collectorCmd = &cobra.Command{
+	Use:   "collector",
+	Short: "Manage the lead-side collector (aggregator + watchdog) (P09)",
+	Long: `Manage the lead-side collector: the aggregator (orca-aggregate.sh,
+10s cadence, merges per-peer state into cluster.json + drift-event
+aggregation per REQ-107) and the watchdog (orca-watchdog.sh, 30s
+cadence, detects aggregator starvation per C-11). Opt-in (C-12).`,
+	Args: cobra.NoArgs,
+}
+
+var collectorStartCmd = &cobra.Command{
+	Use:   "start",
+	Short: "Emit the collector scripts + systemd units and enable them",
+	Args:  cobra.NoArgs,
+	RunE: func(cmd *cobra.Command, args []string) error {
+		root, err := collectorResolveRoot()
+		if err != nil {
+			return err
+		}
+		if err := collectorEmit(root); err != nil {
+			return err
+		}
+		if !collectorDryRun {
+			if err := collectorEnable(root); err != nil {
+				return fmt.Errorf("enable: %w", err)
+			}
+		}
+		msg := "collector started"
+		if collectorDryRun {
+			msg = "collector scripts emitted (dry-run, not enabled)"
+		}
+		printResult(msg, map[string]string{"status": "started", "root": root})
+		return nil
+	},
+}
+
+var collectorStopCmd = &cobra.Command{
+	Use:   "stop",
+	Short: "Disable and remove the collector scripts + systemd units",
+	Args:  cobra.NoArgs,
+	RunE: func(cmd *cobra.Command, args []string) error {
+		root, err := collectorResolveRoot()
+		if err != nil {
+			return err
+		}
+		if !collectorDryRun {
+			if err := collectorDisable(root); err != nil {
+				return fmt.Errorf("disable: %w", err)
+			}
+		}
+		if err := collectorRemove(root); err != nil {
+			return err
+		}
+		msg := "collector stopped"
+		if collectorDryRun {
+			msg = "collector artifacts removed (dry-run, not disabled)"
+		}
+		printResult(msg, map[string]string{"status": "stopped", "root": root})
+		return nil
+	},
+}
+
+var collectorStatusCmd = &cobra.Command{
+	Use:   "status",
+	Short: "Report whether the collector (aggregator + watchdog) is running",
+	Args:  cobra.NoArgs,
+	RunE: func(cmd *cobra.Command, args []string) error {
+		root, err := collectorResolveRoot()
+		if err != nil {
+			return err
+		}
+		aggRunning, wdRunning := collectorRunning(root)
+		overall := "running"
+		if !aggRunning && !wdRunning {
+			overall = "stopped"
+		} else if !aggRunning || !wdRunning {
+			overall = "partial"
+		}
+		printResult(
+			fmt.Sprintf("collector: %s (aggregator=%t watchdog=%t)", overall, aggRunning, wdRunning),
+			map[string]any{
+				"status":     overall,
+				"aggregator": aggRunning,
+				"watchdog":   wdRunning,
+				"root":       root,
+			},
+		)
+		return nil
+	},
+}
+
+var collectorDryRun bool
+
+func init() {
+	collectorCmd.PersistentFlags().StringVar(&collectorRoot, "root", "/", "install root for emitted paths (default: /; for testing use a temp dir)")
+	collectorStartCmd.Flags().BoolVar(&collectorDryRun, "dry-run", false, "emit scripts/units without enabling or running systemctl")
+	collectorStopCmd.Flags().BoolVar(&collectorDryRun, "dry-run", false, "remove scripts/units without disabling or running systemctl")
+	collectorCmd.AddCommand(collectorStartCmd)
+	collectorCmd.AddCommand(collectorStopCmd)
+	collectorCmd.AddCommand(collectorStatusCmd)
+	rootCmd.AddCommand(collectorCmd)
+}
+
+func collectorResolveRoot() (string, error) {
+	r := strings.TrimRight(collectorRoot, "/")
+	if r == "" {
+		r = "/"
+	}
+	if !filepath.IsAbs(r) {
+		return "", fmt.Errorf("--root must be absolute, got %q", collectorRoot)
+	}
+	return r, nil
+}
+
+func collectorEmit(root string) error {
+	dirs := []string{
+		filepath.Join(root, collectorScriptDir),
+		filepath.Join(root, collectorUnitDir),
+		filepath.Join(root, collectorStateDir),
+	}
+	for _, d := range dirs {
+		if err := os.MkdirAll(d, 0o755); err != nil {
+			return fmt.Errorf("mkdir %s: %w", d, err)
+		}
+	}
+	files := map[string]struct {
+		Content string
+		Mode    os.FileMode
+	}{
+		filepath.Join(root, collectorScriptDir, collectorAggregateSh):  {collectorAggregateScript, 0o755},
+		filepath.Join(root, collectorScriptDir, collectorWatchdogSh):    {collectorWatchdogScript, 0o755},
+		filepath.Join(root, collectorUnitDir, collectorAggregateSvc):    {collectorAggregateUnit, 0o644},
+		filepath.Join(root, collectorUnitDir, collectorAggregateTmr):    {collectorAggregateTimer, 0o644},
+		filepath.Join(root, collectorUnitDir, collectorWatchdogSvc):     {collectorWatchdogUnit, 0o644},
+		filepath.Join(root, collectorUnitDir, collectorWatchdogTmr):     {collectorWatchdogTimer, 0o644},
+	}
+	for path, f := range files {
+		tmp := path + ".tmp"
+		if err := os.WriteFile(tmp, []byte(f.Content), f.Mode); err != nil {
+			return fmt.Errorf("write %s: %w", path, err)
+		}
+		if err := os.Rename(tmp, path); err != nil {
+			return fmt.Errorf("rename %s: %w", path, err)
+		}
+	}
+	return nil
+}
+
+func collectorRemove(root string) error {
+	paths := []string{
+		filepath.Join(root, collectorScriptDir, collectorAggregateSh),
+		filepath.Join(root, collectorScriptDir, collectorWatchdogSh),
+		filepath.Join(root, collectorUnitDir, collectorAggregateSvc),
+		filepath.Join(root, collectorUnitDir, collectorAggregateTmr),
+		filepath.Join(root, collectorUnitDir, collectorWatchdogSvc),
+		filepath.Join(root, collectorUnitDir, collectorWatchdogTmr),
+	}
+	for _, p := range paths {
+		if err := os.Remove(p); err != nil && !os.IsNotExist(err) {
+			return fmt.Errorf("remove %s: %w", p, err)
+		}
+	}
+	return nil
+}
+
+func collectorRunning(root string) (bool, bool) {
+	aggRunning := fileExists(filepath.Join(root, collectorUnitDir, collectorAggregateSvc)) &&
+		fileExists(filepath.Join(root, collectorScriptDir, collectorAggregateSh))
+	wdRunning := fileExists(filepath.Join(root, collectorUnitDir, collectorWatchdogSvc)) &&
+		fileExists(filepath.Join(root, collectorScriptDir, collectorWatchdogSh))
+	return aggRunning, wdRunning
+}
+
+func fileExists(p string) bool {
+	_, err := os.Stat(p)
+	return err == nil
+}
+
+func collectorEnable(root string) error {
+	if !commandAvailable("systemctl") {
+		return nil
+	}
+	for _, u := range []string{collectorAggregateTmr, collectorWatchdogTmr} {
+		_ = runSystemctl(root, "enable", u)
+		_ = runSystemctl(root, "start", u)
+	}
+	return nil
+}
+
+func collectorDisable(root string) error {
+	if !commandAvailable("systemctl") {
+		return nil
+	}
+	for _, u := range []string{collectorAggregateTmr, collectorWatchdogTmr} {
+		_ = runSystemctl(root, "stop", u)
+		_ = runSystemctl(root, "disable", u)
+	}
+	return nil
+}
+
+func runSystemctl(root, action, unit string) error {
+	args := []string{action, unit}
+	if root != "/" {
+		args = append([]string{"--root", root}, args...)
+	}
+	return runCmd("systemctl", args...)
+}
+
+func commandAvailable(name string) bool {
+	_, err := exec.LookPath(name)
+	return err == nil
+}
+
+func runCmd(name string, args ...string) error {
+	cmd := exec.Command(name, args...)
+	cmd.Stdout = os.Stdout
+	cmd.Stderr = os.Stderr
+	return cmd.Run()
+}
+
+const collectorAggregateScript = `#!/usr/bin/env bash
+set -euo pipefail
+# orca-aggregate.sh — emitted by orca collector start (P09).
+# Placeholder wrapper; the canonical copy lives at scripts/orca-aggregate.sh.
+exec /usr/local/bin/orca-aggregate.sh "$@"
+`
+
+const collectorWatchdogScript = `#!/usr/bin/env bash
+set -euo pipefail
+# orca-watchdog.sh — emitted by orca collector start (P09).
+# Placeholder wrapper; the canonical copy lives at scripts/orca-watchdog.sh.
+exec /usr/local/bin/orca-watchdog.sh "$@"
+`
+
+const collectorAggregateUnit = `[Unit]
+Description=orca aggregator (P09, C-11/C-12)
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=oneshot
+ExecStart=/etc/orca/collector/orca-aggregate.sh
+
+[Install]
+WantedBy=multi-user.target
+`
+
+const collectorAggregateTimer = `[Unit]
+Description=orca aggregator 10s cadence (P09, C-11)
+
+[Timer]
+OnBootSec=10s
+OnUnitActiveSec=10s
+AccuracySec=1s
+Unit=orca-aggregate.service
+
+[Install]
+WantedBy=timers.target
+`
+
+const collectorWatchdogUnit = `[Unit]
+Description=orca watchdog meta-timer (P09, C-11)
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=oneshot
+ExecStart=/etc/orca/collector/orca-watchdog.sh
+
+[Install]
+WantedBy=multi-user.target
+`
+
+const collectorWatchdogTimer = `[Unit]
+Description=orca watchdog 30s cadence (P09, C-11)
+
+[Timer]
+OnBootSec=30s
+OnUnitActiveSec=30s
+AccuracySec=5s
+Unit=orca-watchdog.service
+
+[Install]
+WantedBy=timers.target
+`

internal/cli/collector_test.go

@@ -0,0 +1,160 @@
+package cli
+
+import (
+	"bytes"
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+func TestCollectorCmdRegistered(t *testing.T) {
+	found := false
+	for _, cmd := range rootCmd.Commands() {
+		if cmd.Name() == "collector" {
+			found = true
+			break
+		}
+	}
+	if !found {
+		t.Fatal("collector command not registered on root")
+	}
+}
+
+func TestCollectorSubcommandsRegistered(t *testing.T) {
+	want := map[string]bool{"start": false, "stop": false, "status": false}
+	for _, cmd := range collectorCmd.Commands() {
+		if _, ok := want[cmd.Name()]; ok {
+			want[cmd.Name()] = true
+		}
+	}
+	for name, found := range want {
+		if !found {
+			t.Errorf("collector subcommand %q not registered", name)
+		}
+	}
+}
+
+func runCollectorCmd(t *testing.T, root string, dryRun bool, args ...string) (string, error) {
+	t.Helper()
+	resetRootFlags(t)
+	full := append([]string{"collector"}, args...)
+	if root != "" {
+		full = append(full, "--root", root)
+	}
+	if dryRun && (len(args) > 0 && (args[0] == "start" || args[0] == "stop")) {
+		full = append(full, "--dry-run")
+	}
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	rootCmd.SetArgs(full)
+	err := rootCmd.Execute()
+	return buf.String(), err
+}
+
+func TestCollectorStartEmitsArtifacts(t *testing.T) {
+	root := t.TempDir()
+	out, err := runCollectorCmd(t, root, true, "start")
+	if err != nil {
+		t.Fatalf("orca collector start: %v\n%s", err, out)
+	}
+	wantFiles := []string{
+		filepath.Join(root, collectorScriptDir, collectorAggregateSh),
+		filepath.Join(root, collectorScriptDir, collectorWatchdogSh),
+		filepath.Join(root, collectorUnitDir, collectorAggregateSvc),
+		filepath.Join(root, collectorUnitDir, collectorAggregateTmr),
+		filepath.Join(root, collectorUnitDir, collectorWatchdogSvc),
+		filepath.Join(root, collectorUnitDir, collectorWatchdogTmr),
+	}
+	for _, p := range wantFiles {
+		if _, err := os.Stat(p); err != nil {
+			t.Errorf("expected emitted file %s: %v", p, err)
+		}
+	}
+	for _, p := range []string{
+		filepath.Join(root, collectorScriptDir, collectorAggregateSh),
+		filepath.Join(root, collectorScriptDir, collectorWatchdogSh),
+	} {
+		info, err := os.Stat(p)
+		if err != nil {
+			t.Fatalf("stat %s: %v", p, err)
+		}
+		if perm := info.Mode().Perm(); perm&0o111 == 0 {
+			t.Errorf("expected executable bit on %s, got %o", p, perm)
+		}
+	}
+}
+
+func TestCollectorStartStatusRunning(t *testing.T) {
+	root := t.TempDir()
+	if _, err := runCollectorCmd(t, root, true, "start"); err != nil {
+		t.Fatalf("start: %v", err)
+	}
+	agg, wd := collectorRunning(root)
+	if !agg || !wd {
+		t.Errorf("expected both running, got agg=%t wd=%t", agg, wd)
+	}
+	out, err := runCollectorCmd(t, root, true, "status")
+	if err != nil {
+		t.Fatalf("status: %v", err)
+	}
+	if !contains(out, "running") {
+		t.Errorf("status output should say running, got %q", out)
+	}
+}
+
+func TestCollectorStopRemovesArtifacts(t *testing.T) {
+	root := t.TempDir()
+	if _, err := runCollectorCmd(t, root, true, "start"); err != nil {
+		t.Fatalf("start: %v", err)
+	}
+	out, err := runCollectorCmd(t, root, true, "stop")
+	if err != nil {
+		t.Fatalf("stop: %v\n%s", err, out)
+	}
+	wantFiles := []string{
+		filepath.Join(root, collectorScriptDir, collectorAggregateSh),
+		filepath.Join(root, collectorScriptDir, collectorWatchdogSh),
+		filepath.Join(root, collectorUnitDir, collectorAggregateSvc),
+		filepath.Join(root, collectorUnitDir, collectorAggregateTmr),
+		filepath.Join(root, collectorUnitDir, collectorWatchdogSvc),
+		filepath.Join(root, collectorUnitDir, collectorWatchdogTmr),
+	}
+	for _, p := range wantFiles {
+		if _, err := os.Stat(p); !os.IsNotExist(err) {
+			t.Errorf("expected %s removed, got %v", p, err)
+		}
+	}
+}
+
+func TestCollectorStatusWhenStopped(t *testing.T) {
+	root := t.TempDir()
+	out, err := runCollectorCmd(t, root, true, "status")
+	if err != nil {
+		t.Fatalf("status: %v", err)
+	}
+	if !contains(out, "stopped") {
+		t.Errorf("expected stopped, got %q", out)
+	}
+	agg, wd := collectorRunning(root)
+	if agg || wd {
+		t.Errorf("expected neither running, got agg=%t wd=%t", agg, wd)
+	}
+}
+
+func TestCollectorResolveRootRejectsRelative(t *testing.T) {
+	collectorRoot = "tmp/relative"
+	_, err := collectorResolveRoot()
+	if err == nil {
+		t.Error("expected error for relative root")
+	}
+	collectorRoot = "/"
+	r, err := collectorResolveRoot()
+	if err != nil || r != "/" {
+		t.Errorf("expected / for default, got %q err=%v", r, err)
+	}
+}
+
+func contains(haystack, needle string) bool {
+	return bytes.Contains([]byte(haystack), []byte(needle))
+}

internal/cli/namespace_test.go

@@ -43,6 +43,8 @@ func resetCommandFlags() {
 	backupOutPath, restoreInPath, restoreTargetDir = "", "", ""
 	restoreForce = false
 	restoreDryRun = false
+	collectorRoot = "/"
+	collectorDryRun = false
 	resetNSFlags()
 	// Reset per-command output writers so tests that polluted them
 	// (e.g. daemon tests calling cmd.SetOut(&buf)) don't leak into

scripts/orca-aggregate.sh

@@ -0,0 +1,172 @@
+#!/usr/bin/env bash
+# orca-aggregate.sh — lead-side aggregator (P09, C-11/C-12/C-14, REQ-107, D-237).
+#
+# Runs every 10s via orca-aggregate.timer. For each peer in
+# /etc/orca/peers.env: SSHs in, reads /run/orca/state/<latest>.json (the
+# peer's state snapshot), and merges all peer states into a single
+# /etc/orca/state/cluster.json on the lead. Writes atomically (temp +
+# rename). Structured logging via logger -t orca-aggregate (syslog).
+#
+# Drift-event aggregation extension (REQ-107, D-237): rsyncs each
+# peer's /etc/orca/state/drift-events/ to a temp dir, validates each
+# event's hash against /etc/orca/state/applied/<txn>/manifest.json,
+# triggers orca-remediate.sh <peer> <txn> for auto-remediable paths
+# (the remediation script itself lands in P10b — the call is emitted
+# but the script may not exist yet), then consumes (deletes) the event
+# file on the peer after processing. Aggregated drift state is written
+# to /etc/orca/state/drift-events-aggregated.json. The drift section
+# is skipped gracefully (with a log line) when the drift-events
+# directory does not exist on the lead or on any peer.
+#
+# The aggregator is opt-in (C-12): the operator enables it with
+# `orca cluster config --aggregator=true`, which installs this script
+# + the systemd units via `orca collector start`.
+
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+# shellcheck source=lib/orca-log.sh
+. "$SCRIPT_DIR/lib/orca-log.sh"
+
+ORCA_LOG_ACTOR="spiffe://orca/cli/aggregator"
+
+PEERS_ENV="${ORCA_PEERS_ENV:-/etc/orca/peers.env}"
+STATE_DIR="${ORCA_STATE_DIR:-/etc/orca/state}"
+CLUSTER_JSON="$STATE_DIR/cluster.json"
+DRIFT_AGG_JSON="$STATE_DIR/drift-events-aggregated.json"
+REMOTE_STATE_DIR="${ORCA_REMOTE_STATE_DIR:-/run/orca/state}"
+REMOTE_DRIFT_DIR="${ORCA_REMOTE_DRIFT_DIR:-/etc/orca/state/drift-events}"
+APPLIED_DIR="${ORCA_APPLIED_DIR:-/etc/orca/state/applied}"
+REMEDIATE_SCRIPT="${ORCA_REMEDIATE_SCRIPT:-/usr/local/sbin/orca-remediate.sh}"
+SSH_OPTS="${ORCA_SSH_OPTS:--o BatchMode=yes -o StrictHostKeyChecking=accept-new -o ConnectTimeout=5}"
+RSYNC_OPTS_RSYNC="${ORCA_RSYNC_OPTS_RSYNC:-}"
+AGG_TMP="$(mktemp -d)"
+trap 'rm -rf "$AGG_TMP"' EXIT
+
+mkdir -p "$STATE_DIR"
+
+read_peers() {
+	if [ ! -f "$PEERS_ENV" ]; then
+		return 0
+	fi
+	awk -F= '/^[[:space:]]*#/ {next} /^[[:space:]]*$/ {next} {sub(/^[[:space:]]*PEER_[^=]*=/,""); print}' "$PEERS_ENV"
+}
+
+merge_tmp="$AGG_TMP/merge.jsonl"
+: >"$merge_tmp"
+
+peer_count=0
+while IFS= read -r peer; do
+	[ -z "$peer" ] && continue
+	peer_count=$((peer_count + 1))
+	latest_json="$(ssh $SSH_OPTS "$peer" "ls -1 $REMOTE_STATE_DIR/*.json 2>/dev/null | sort | tail -1" 2>/dev/null || true)"
+	if [ -z "$latest_json" ]; then
+		orca_log_warn "aggregate" "$peer" "skipped" "no state snapshot on peer"
+		continue
+	fi
+	snapshot="$(ssh $SSH_OPTS "$peer" "cat '$latest_json'" 2>/dev/null || true)"
+	if [ -z "$snapshot" ]; then
+		orca_log_warn "aggregate" "$peer" "skipped" "failed to read state snapshot"
+		continue
+	fi
+	printf '{"peer":"%s","state":%s}\n' "$peer" "$snapshot" >>"$merge_tmp"
+	orca_log_info "aggregate" "$peer" "ok" "snapshot=$latest_json"
+done < <(read_peers)
+
+ts="$(date -u +%Y-%m-%dT%H:%M:%S.%3NZ)"
+{
+	printf '{"ts":"%s","peers":[' "$ts"
+	first=1
+	while IFS= read -r line; do
+		[ -z "$line" ] && continue
+		if [ "$first" -eq 1 ]; then
+			first=0
+		else
+			printf ','
+		fi
+		printf '%s' "$line"
+	done <"$merge_tmp"
+	printf ']}'
+} >"$AGG_TMP/cluster.json.new"
+mv "$AGG_TMP/cluster.json.new" "$CLUSTER_JSON"
+orca_log_info "aggregate" "-" "ok" "peers=$peer_count cluster_json=$CLUSTER_JSON"
+
+if [ ! -d "$REMOTE_DRIFT_DIR" ] && [ ! -d "$APPLIED_DIR" ]; then
+	orca_log_info "drift" "-" "skipped" "drift-events + applied dirs absent (P10b not shipped)"
+else
+	drift_tmp="$AGG_TMP/drift-events"
+	mkdir -p "$drift_tmp"
+	drift_agg_tmp="$AGG_TMP/drift-agg.jsonl"
+	: >"$drift_agg_tmp"
+	while IFS= read -r peer; do
+		[ -z "$peer" ] && continue
+		peer_drift_dir="$drift_tmp/$peer"
+		mkdir -p "$peer_drift_dir"
+		if ! rsync -a --quiet $RSYNC_OPTS_RSYNC "$peer:$REMOTE_DRIFT_DIR/" "$peer_drift_dir/" 2>/dev/null; then
+			orca_log_info "drift" "$peer" "skipped" "rsync failed or dir absent"
+			continue
+		fi
+		shopt -s nullglob
+		event_files=("$peer_drift_dir"/*.json)
+		shopt -u nullglob
+		if [ "${#event_files[@]}" -eq 0 ]; then
+			orca_log_info "drift" "$peer" "ok" "no events"
+			continue
+		fi
+		for event_file in "${event_files[@]}"; do
+			event_id="$(basename "$event_file" .json)"
+			path_val="$(grep -o '"path"[[:space:]]*:[[:space:]]*"[^"]*"' "$event_file" | head -1 | sed 's/.*: *"\([^"]*\)"/\1/')"
+			new_sha="$(grep -o '"new_sha256"[[:space:]]*:[[:space:]]*"[^"]*"' "$event_file" | head -1 | sed 's/.*: *"\([^"]*\)"/\1/')"
+			txn_val="$(grep -o '"latest_txn"[[:space:]]*:[[:space:]]*"[^"]*"' "$event_file" | head -1 | sed 's/.*: *"\([^"]*\)"/\1/')"
+			status_val="$(grep -o '"status"[[:space:]]*:[[:space:]]*"[^"]*"' "$event_file" | head -1 | sed 's/.*: *"\([^"]*\)"/\1/')"
+			if [ -z "$txn_val" ]; then
+				orca_log_warn "drift" "$peer" "skipped" "event=$event_id no latest_txn"
+				continue
+			fi
+			manifest="$APPLIED_DIR/$txn_val/manifest.json"
+			drift_confirmed=0
+			if [ -f "$manifest" ]; then
+				manifest_sha="$(grep -o "\"$path_val\"[^\"]*\"sha256\"[[:space:]]*:[[:space:]]*\"[^\"]*\"" "$manifest" 2>/dev/null | grep -o '"sha256"[[:space:]]*:[[:space:]]*"[^"]*"' | head -1 | sed 's/.*: *"\([^"]*\)"/\1/')"
+				if [ -n "$manifest_sha" ] && [ "$manifest_sha" != "$new_sha" ]; then
+					drift_confirmed=1
+				fi
+			else
+				orca_log_warn "drift" "$peer" "skipped" "txn=$txn_val manifest absent"
+			fi
+			if [ "$drift_confirmed" -eq 1 ]; then
+				if [ -x "$REMEDIATE_SCRIPT" ]; then
+					orca_log_info "drift" "$peer" "remediate" "txn=$txn_val path=$path_val"
+					"$REMEDIATE_SCRIPT" "$peer" "$txn_val" "$path_val" \
+						|| orca_log_error "drift" "$peer" "remediate-failed" "txn=$txn_val path=$path_val"
+				else
+					orca_log_warn "drift" "$peer" "remediate-stub" "orca-remediate.sh not installed (P10b)"
+				fi
+				printf '{"peer":"%s","event_id":"%s","txn":"%s","path":"%s","status":"%s","remediated":true}\n' \
+					"$peer" "$event_id" "$txn_val" "$path_val" "$status_val" >>"$drift_agg_tmp"
+			else
+				printf '{"peer":"%s","event_id":"%s","txn":"%s","path":"%s","status":"%s","remediated":false}\n' \
+					"$peer" "$event_id" "$txn_val" "$path_val" "$status_val" >>"$drift_agg_tmp"
+			fi
+			ssh $SSH_OPTS "$peer" "rm -f '$REMOTE_DRIFT_DIR/$event_id.json'" 2>/dev/null \
+				|| orca_log_warn "drift" "$peer" "consume-failed" "event=$event_id"
+		done
+	done < <(read_peers)
+	{
+		printf '{"ts":"%s","events":[' "$ts"
+		first=1
+		while IFS= read -r line; do
+			[ -z "$line" ] && continue
+			if [ "$first" -eq 1 ]; then
+				first=0
+			else
+				printf ','
+			fi
+			printf '%s' "$line"
+		done <"$drift_agg_tmp"
+		printf ']}'
+	} >"$AGG_TMP/drift-agg.json.new"
+	mv "$AGG_TMP/drift-agg.json.new" "$DRIFT_AGG_JSON"
+	orca_log_info "drift" "-" "ok" "aggregated=$DRIFT_AGG_JSON"
+fi
+
+orca_log_info "aggregate" "-" "done" "peers=$peer_count"

scripts/orca-watchdog.sh

@@ -0,0 +1,42 @@
+#!/usr/bin/env bash
+# orca-watchdog.sh — lead-side watchdog meta-timer (P09, C-11).
+#
+# Runs every 30s via orca-watchdog.timer. Checks whether
+# orca-aggregate.sh has run in the last N seconds (default 30s — 3x the
+# 10s cadence). If not: fires a structured alert via
+# `logger -t orca-watchdog -p user.err "aggregator starvation detected"`.
+# The check compares the mtime of /etc/orca/state/cluster.json to the
+# current time.
+
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+# shellcheck source=lib/orca-log.sh
+. "$SCRIPT_DIR/lib/orca-log.sh"
+
+ORCA_LOG_ACTOR="spiffe://orca/cli/watchdog"
+
+STATE_DIR="${ORCA_STATE_DIR:-/etc/orca/state}"
+CLUSTER_JSON="$STATE_DIR/cluster.json"
+STARVATION_THRESHOLD="${ORCA_WATCHDOG_THRESHOLD:-30}"
+
+alert() {
+	logger -t orca-watchdog -p user.err "aggregator starvation detected: $1"
+}
+
+if [ ! -f "$CLUSTER_JSON" ]; then
+	alert "cluster.json absent at $CLUSTER_JSON"
+	orca_log_error "watchdog" "-" "starved" "cluster.json absent"
+	exit 0
+fi
+
+now="$(date +%s)"
+mtime="$(stat -c %Y "$CLUSTER_JSON" 2>/dev/null || stat -f %m "$CLUSTER_JSON" 2>/dev/null || echo 0)"
+age=$((now - mtime))
+
+if [ "$age" -ge "$STARVATION_THRESHOLD" ]; then
+	alert "cluster.json age ${age}s >= threshold ${STARVATION_THRESHOLD}s"
+	orca_log_error "watchdog" "-" "starved" "age=${age}s threshold=${STARVATION_THRESHOLD}s"
+else
+	orca_log_info "watchdog" "-" "ok" "age=${age}s threshold=${STARVATION_THRESHOLD}s"
+fi

scripts/tests/aggregator_test.bash

@@ -0,0 +1,126 @@
+#!/usr/bin/env bats
+# Tests for scripts/orca-aggregate.sh + orca-watchdog.sh (P09, C-11,
+# REQ-107). Hermetic: exercises the aggregator against a fake peers.env
+# + state dir in a temp dir; the SSH path is exercised only when a
+# fake ssh shim is on PATH. The watchdog is exercised with a stale
+# cluster.json to confirm starvation detection.
+
+load test_helper
+
+@test "orca-aggregate.sh exists and is executable" {
+	[ -f "$SCRIPTS_DIR/orca-aggregate.sh" ]
+	[ -x "$SCRIPTS_DIR/orca-aggregate.sh" ]
+}
+
+@test "orca-watchdog.sh exists and is executable" {
+	[ -f "$SCRIPTS_DIR/orca-watchdog.sh" ]
+	[ -x "$SCRIPTS_DIR/orca-watchdog.sh" ]
+}
+
+@test "orca-aggregate.sh handles missing peers.env gracefully" {
+	tmp="$(mktemp -d)"
+	run env \
+		ORCA_PEERS_ENV="$tmp/nonexistent.env" \
+		ORCA_STATE_DIR="$tmp/state" \
+		ORCA_REMOTE_STATE_DIR="$tmp/remote-state" \
+		ORCA_REMOTE_DRIFT_DIR="$tmp/remote-drift" \
+		ORCA_APPLIED_DIR="$tmp/applied" \
+		bash "$SCRIPTS_DIR/orca-aggregate.sh"
+	assert_status 0 "$status"
+	# cluster.json written atomically with an empty peers array.
+	[ -f "$tmp/state/cluster.json" ]
+	assert_contains "$(cat "$tmp/state/cluster.json")" '"peers":[]'
+}
+
+@test "orca-aggregate.sh skips drift section when drift-events absent" {
+	tmp="$(mktemp -d)"
+	# No applied dir and no remote drift dir on lead → drift section
+	# must skip (no drift-events-aggregated.json written).
+	run env \
+		ORCA_PEERS_ENV="$tmp/nonexistent.env" \
+		ORCA_STATE_DIR="$tmp/state" \
+		ORCA_REMOTE_STATE_DIR="$tmp/remote-state" \
+		ORCA_REMOTE_DRIFT_DIR="$tmp/absent-drift" \
+		ORCA_APPLIED_DIR="$tmp/absent-applied" \
+		bash "$SCRIPTS_DIR/orca-aggregate.sh"
+	assert_status 0 "$status"
+	[ -f "$tmp/state/cluster.json" ]
+	# Drift aggregation skipped → aggregated file must NOT exist.
+	[ ! -f "$tmp/state/drift-events-aggregated.json" ]
+}
+
+@test "orca-watchdog.sh detects starvation with old cluster.json" {
+	tmp="$(mktemp -d)"
+	mkdir -p "$tmp/state"
+	cluster="$tmp/state/cluster.json"
+	echo '{}' >"$cluster"
+	# Set mtime to 120s ago so age (120) >= threshold (default 30).
+	old_ts="$(date -d '120 seconds ago' +%Y%m%d%H%M.%S)"
+	touch -t "$old_ts" "$cluster"
+	# logger is available in CI; capture the alert by overriding PATH
+	# with a fake logger that records to a file.
+	fake_bin="$tmp/bin"
+	mkdir -p "$fake_bin"
+	cat >"$fake_bin/logger" <<'LOGGER_EOF'
+#!/usr/bin/env bash
+echo "logger: $*" >>"$LOGGER_OUT"
+LOGGER_EOF
+	chmod +x "$fake_bin/logger"
+	env \
+		ORCA_STATE_DIR="$tmp/state" \
+		ORCA_WATCHDOG_THRESHOLD=30 \
+		LOGGER_OUT="$tmp/logger.out" \
+		PATH="$fake_bin:$PATH" \
+		bash "$SCRIPTS_DIR/orca-watchdog.sh"
+	[ -f "$tmp/logger.out" ]
+	out="$(cat "$tmp/logger.out")"
+	assert_contains "$out" "aggregator starvation detected"
+}
+
+@test "orca-watchdog.sh passes when cluster.json is fresh" {
+	tmp="$(mktemp -d)"
+	mkdir -p "$tmp/state"
+	echo '{}' >"$tmp/state/cluster.json"
+	# Fresh mtime (now); age 0 < threshold.
+	fake_bin="$tmp/bin"
+	mkdir -p "$fake_bin"
+	cat >"$fake_bin/logger" <<'LOGGER_EOF'
+#!/usr/bin/env bash
+echo "logger: $*" >>"$LOGGER_OUT"
+LOGGER_EOF
+	chmod +x "$fake_bin/logger"
+	env \
+		ORCA_STATE_DIR="$tmp/state" \
+		ORCA_WATCHDOG_THRESHOLD=30 \
+		LOGGER_OUT="$tmp/logger.out" \
+		PATH="$fake_bin:$PATH" \
+		bash "$SCRIPTS_DIR/orca-watchdog.sh"
+	if [ -f "$tmp/logger.out" ]; then
+		out="$(cat "$tmp/logger.out")"
+		! assert_contains "$out" "aggregator starvation detected" || {
+			echo "unexpected starvation alert for fresh cluster.json" >&2
+			return 1
+		}
+	fi
+}
+
+@test "orca-watchdog.sh alerts when cluster.json absent" {
+	tmp="$(mktemp -d)"
+	fake_bin="$tmp/bin"
+	mkdir -p "$fake_bin"
+	cat >"$fake_bin/logger" <<'LOGGER_EOF'
+#!/usr/bin/env bash
+echo "logger: $*" >>"$LOGGER_OUT"
+LOGGER_EOF
+	chmod +x "$fake_bin/logger"
+	env \
+		ORCA_STATE_DIR="$tmp/state" \
+		ORCA_WATCHDOG_THRESHOLD=30 \
+		LOGGER_OUT="$tmp/logger.out" \
+		PATH="$fake_bin:$PATH" \
+		bash "$SCRIPTS_DIR/orca-watchdog.sh"
+	[ -f "$tmp/logger.out" ]
+	out="$(cat "$tmp/logger.out")"
+	assert_contains "$out" "aggregator starvation detected"
+	assert_contains "$out" "cluster.json absent"
+}