#!/usr/bin/env bash
set -euo pipefail

fatal() {
  printf 'install-candidate: %s\n' "$*" >&2
  exit 2
}

TEST_ROOT=${OVERDECK_SUBROUTER_TEST_ROOT:-}
if [[ -z "$TEST_ROOT" && ${EUID:-999} -ne 0 ]]; then
  printf '%s\n' 'install-candidate: root required (invoke through deck-sudo)' >&2
  exit 77
fi
if [[ -n "$TEST_ROOT" ]]; then
  [[ "$TEST_ROOT" == /* && -d "$TEST_ROOT" ]] || {
    printf '%s\n' 'install-candidate: test root must be an existing absolute directory' >&2
    exit 77
  }
  TEST_ROOT=$(realpath -e -- "$TEST_ROOT")
fi

SCRIPT_DIR=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd -P)
ROOT=$(cd -- "$SCRIPT_DIR/../../.." && pwd -P)
MODULE="$ROOT/modules/subrouter"
RUNTIME="${TEST_ROOT}/var/lib/overdeck/subrouter"
RELEASES="$RUNTIME/releases"
CURRENT="$RUNTIME/current"
PREVIOUS="$RUNTIME/previous"
UNIT_TARGET="${TEST_ROOT}/etc/systemd/system/overdeck-subrouter.service"
UNIT_SOURCE="$MODULE/systemd/overdeck-subrouter.service"
OWNER_ARGS=(-o root -g root)
ADMIN_OWNER_ARGS=(-o overdeck-subrouter -g overdeck-subrouter)
CLIENT_UID=${OVERDECK_SUBROUTER_CLIENT_UID:-$(stat -c %u "$ROOT")}
CLIENT_GID=${OVERDECK_SUBROUTER_CLIENT_GID:-$(stat -c %g "$ROOT")}
[[ "$CLIENT_UID" =~ ^[0-9]+$ && "$CLIENT_GID" =~ ^[0-9]+$ ]] || fatal 'workstation client identity is invalid'
CLIENT_OWNER_ARGS=(-o "$CLIENT_UID" -g "$CLIENT_GID")
if [[ -n "$TEST_ROOT" ]]; then OWNER_ARGS=(); ADMIN_OWNER_ARGS=(); CLIENT_OWNER_ARGS=(); fi
LOCK_DIR="${TEST_ROOT}/run/lock"
install -d "${OWNER_ARGS[@]}" -m 0755 "$LOCK_DIR"
exec 9>"$LOCK_DIR/overdeck-subrouter-install.lock"
flock -n 9 || fatal 'another Subrouter install or rollback is active'

CANDIDATE_SNAPSHOT_PARENT=
staging=
old_unit=
unit_staging=
state_staging=
admin_staging=
client_token_staging=
cleanup_exit() {
  if [[ -n "$staging" && -e "$staging" ]]; then chmod -R u+w -- "$staging" 2>/dev/null || true; rm -rf -- "$staging" || true; fi
  [[ -z "$old_unit" ]] || rm -f -- "$old_unit" || true
  [[ -z "$unit_staging" ]] || rm -f -- "$unit_staging" || true
  [[ -z "$state_staging" ]] || rm -f -- "$state_staging" || true
  [[ -z "$admin_staging" ]] || rm -f -- "$admin_staging" || true
  [[ -z "$client_token_staging" ]] || rm -f -- "$client_token_staging" || true
  if [[ -n "$CANDIDATE_SNAPSHOT_PARENT" && -e "$CANDIDATE_SNAPSHOT_PARENT" ]]; then
    chmod -R u+w -- "$CANDIDATE_SNAPSHOT_PARENT" 2>/dev/null || true
    rm -rf -- "$CANDIDATE_SNAPSHOT_PARENT" || true
  fi
}
trap cleanup_exit EXIT

validate_release() {
  local release=$1 store_schema=${2:-}
  python3 - "$release" "$RELEASES" "$store_schema" <<'PY'
import hashlib, json, os, pathlib, re, stat, sys
release = pathlib.Path(sys.argv[1])
releases = pathlib.Path(sys.argv[2]).resolve()
store_schema = int(sys.argv[3]) if sys.argv[3] else None
try:
    if release.parent.resolve() != releases or release.name != release.resolve().name:
        raise ValueError
except (OSError, ValueError):
    raise SystemExit("release escapes release root")
release_mode = release.lstat().st_mode
if not stat.S_ISDIR(release_mode) or release_mode & 0o222: raise SystemExit("release root is not immutable")
if not os.environ.get("OVERDECK_SUBROUTER_TEST_ROOT") and release.stat().st_uid != 0: raise SystemExit("release root is not root-owned")
if not re.fullmatch(r"[0-9a-f]{64}", release.name): raise SystemExit("release digest path is malformed")
receipt_path = release / "receipt.json"
manifest_path = release / "manifest.json"
binary = release / "bin/subrouter"
authority_path = release / "authority-state.json"
patch_path = release / "source.patch"
head_path = release / "source-head"
registry_path = release / "source-registry.json"
unit_path = release / "service.unit"
for required in (receipt_path, manifest_path, binary, authority_path, patch_path, head_path, registry_path, unit_path):
    if not required.is_file() or required.is_symlink(): raise SystemExit("release provenance is incomplete")
r = json.loads(receipt_path.read_text())
m = json.loads(manifest_path.read_text())
a = json.loads(authority_path.read_text())
if r.get("schema") != 1: raise SystemExit("release receipt schema mismatch")
if r.get("candidate_digest") != release.name: raise SystemExit("release candidate digest mismatch")
if hashlib.sha256(patch_path.read_bytes()).hexdigest() != r.get("patch_sha256"): raise SystemExit("release patch digest mismatch")
if head_path.read_text().strip() != r.get("base_commit"): raise SystemExit("release source revision mismatch")
if not re.fullmatch(r"[0-9a-f]{64}", r.get("binary_sha256", "")): raise SystemExit("release binary digest is malformed")
for receipt_key, manifest_key in (("revision", "revision"), ("archive_sha256", "archive_sha256"), ("go_version_required", "go")):
    if r.get(receipt_key) != m.get(manifest_key): raise SystemExit("release manifest mismatch")
if hashlib.sha256(binary.read_bytes()).hexdigest() != r["binary_sha256"]: raise SystemExit("release binary digest mismatch")
candidate_hash = hashlib.sha256()
for value in (r.get("input_digest", ""), m.get("revision", ""), m.get("archive_sha256", ""), r.get("patch_sha256", ""), m.get("go", "")):
    candidate_hash.update(value.encode() + b"\0")
candidate_hash.update(bytes.fromhex(r["binary_sha256"]))
if candidate_hash.hexdigest() != release.name: raise SystemExit("release candidate content digest mismatch")
if not os.access(binary, os.X_OK): raise SystemExit("release binary is not executable")
if a.get("schema") != 1 or a.get("authority_state_schema") != 1 or a.get("rollback_compatible_state_schemas") != [1]:
    raise SystemExit("release authority schema metadata mismatch")
if store_schema is not None and (a["authority_state_schema"] != store_schema or store_schema not in a["rollback_compatible_state_schemas"]):
    raise SystemExit("release authority state schema is incompatible")
phases = ("focused", "static", "module", "full")
records = {}
for phase in phases:
    record_path = release / f"worker-{phase}.json"
    if not record_path.is_file(): raise SystemExit(f"release worker provenance missing: {phase}")
    records[phase] = json.loads(record_path.read_text())
    record = records[phase]
    if record.get("phase") != phase or record.get("input_digest") != r.get("input_digest") or record.get("tree_diff_sha256") != r.get("tree_diff_sha256"):
        raise SystemExit(f"release worker provenance mismatch: {phase}")
    if record.get("worker") != r.get("workers", {}).get(phase) or record.get("exit") != r.get("exits", {}).get(phase):
        raise SystemExit(f"release receipt worker mismatch: {phase}")
    if {name: record.get(name) for name in ("started_epoch", "ended_at", "elapsed_seconds")} != r.get("timing", {}).get(phase):
        raise SystemExit(f"release worker timing mismatch: {phase}")
    if phase != "module":
        dispatch_path = release / f"dispatch-{phase}.json"
        dispatch = json.loads(dispatch_path.read_text())
        if hashlib.sha256(dispatch_path.read_bytes()).hexdigest() != record.get("dispatch_sha256") or record.get("dispatch") != dispatch or r.get("dispatch", {}).get(phase) != dispatch:
            raise SystemExit(f"release dispatch provenance mismatch: {phase}")
        if dispatch.get("registry_sha256") != hashlib.sha256(registry_path.read_bytes()).hexdigest():
            raise SystemExit(f"release dispatch registry mismatch: {phase}")
expected = {"bin", "bin/subrouter", "logs", "receipt.json", "manifest.json", "authority-state.json", "source.patch", "source-head", "source-registry.json", "service.unit"}
expected.update(f"worker-{phase}.json" for phase in phases)
expected.update(f"dispatch-{phase}.json" for phase in ("focused", "static", "full"))
for phase, item in r.get("logs", {}).items():
    path = item.get("path", "")
    if not isinstance(path, str) or not re.fullmatch(r"logs/[^/]+", path): raise SystemExit(f"release log path mismatch: {phase}")
    log = release / path
    if not log.is_file() or hashlib.sha256(log.read_bytes()).hexdigest() != item.get("sha256"):
        raise SystemExit(f"release log digest mismatch: {phase}")
    expected.add(path)
seen = set()
for path in release.rglob("*"):
    relative = path.relative_to(release).as_posix()
    seen.add(relative)
    mode = path.lstat().st_mode
    if mode & 0o222: raise SystemExit(f"writable release path refused: {relative}")
    if stat.S_ISLNK(mode) or not (stat.S_ISDIR(mode) or stat.S_ISREG(mode)):
        raise SystemExit(f"release symlink or special file refused: {relative}")
    if not os.environ.get("OVERDECK_SUBROUTER_TEST_ROOT") and path.stat().st_uid != 0:
        raise SystemExit(f"non-root release path refused: {relative}")
if seen != expected: raise SystemExit("release tree does not exactly match its receipt")
print(a["authority_state_schema"])
PY
}

read_store_schema() {
  local schema_path="$RUNTIME/authority-state-schema.json"
  python3 - "$schema_path" <<'PY'
import json, os, pathlib, stat, sys
path = pathlib.Path(sys.argv[1])
mode = path.lstat().st_mode
if stat.S_ISLNK(mode) or not stat.S_ISREG(mode) or mode & 0o222: raise SystemExit("authority state schema is not immutable")
if not os.environ.get("OVERDECK_SUBROUTER_TEST_ROOT") and path.stat().st_uid != 0: raise SystemExit("authority state schema is not root-owned")
data = json.loads(path.read_text())
if data != {"schema": 1}: raise SystemExit("authority state schema is unsupported")
print(data["schema"])
PY
}

atomic_link() {
  local target=$1 link=$2 temporary
  temporary="${link}.next.$$"
  ln -s "$target" "$temporary"
  mv -Tf "$temporary" "$link"
}

wait_ready() {
  local binary=$1 timeout=20
  [[ -z "$TEST_ROOT" ]] || timeout=1
  local deadline=$((SECONDS + timeout))
  while (( SECONDS < deadline )); do
    if systemctl is-active --quiet overdeck-subrouter.service &&
       "$binary" probe --url http://127.0.0.1:31415 --timeout 2s >/dev/null 2>&1; then
      return 0
    fi
    sleep 1
  done
  return 1
}

rollback_runtime() {
  local old_current=$1 old_previous=$2 old_unit_path=$3 had_unit=$4 was_enabled=$5 failed=0
  if [[ -n "$old_current" ]]; then
    atomic_link "$old_current" "$CURRENT" || failed=1
  else
    rm -f -- "$CURRENT" || failed=1
  fi
  if [[ -n "$old_previous" ]]; then
    atomic_link "$old_previous" "$PREVIOUS" || failed=1
  else
    rm -f -- "$PREVIOUS" || failed=1
  fi
  if [[ "$had_unit" == 1 ]]; then
    install "${OWNER_ARGS[@]}" -m 0644 "$old_unit_path" "$UNIT_TARGET" || failed=1
  else
    rm -f -- "$UNIT_TARGET" || failed=1
  fi
  systemctl daemon-reload || failed=1
  if [[ "$had_unit" == 1 ]]; then
    if [[ "$was_enabled" == 1 ]]; then
      systemctl enable overdeck-subrouter.service || failed=1
    else
      systemctl disable overdeck-subrouter.service || failed=1
    fi
  fi
  if [[ -n "$old_current" ]]; then
    systemctl restart overdeck-subrouter.service || failed=1
  elif [[ "$had_unit" == 1 ]]; then
    systemctl stop overdeck-subrouter.service || failed=1
  else
    systemctl reset-failed overdeck-subrouter.service >/dev/null 2>&1 || true
  fi
  return "$failed"
}

if [[ ${1:-} == --rollback ]]; then
  [[ -L "$PREVIOUS" ]] || fatal 'no previous release is recorded'
  [[ -L "$CURRENT" ]] || fatal 'current release is not recorded'
  target=$(readlink -f "$PREVIOUS")
  old=$(readlink -f "$CURRENT")
  case "$target" in "$RELEASES"/*) ;; *) fatal 'previous release escapes release root' ;; esac
  case "$old" in "$RELEASES"/*) ;; *) fatal 'current release escapes release root' ;; esac
  store_schema=$(read_store_schema) || fatal 'authority state schema is invalid'
  target_schema=$(validate_release "$target" "$store_schema") || fatal 'previous release provenance or compatibility is invalid'
  current_schema=$(validate_release "$old" "$store_schema") || fatal 'current release provenance or compatibility is invalid'
  [[ "$target_schema" == "$current_schema" ]] || fatal 'previous release schema is incompatible with current release'
  old_previous=$target
  atomic_link "$target" "$CURRENT"
  if ! systemctl restart overdeck-subrouter.service || ! wait_ready "$CURRENT/bin/subrouter"; then
    restore_failed=0
    if [[ -n "$old" ]]; then
      atomic_link "$old" "$CURRENT" || restore_failed=1
    else
      rm -f -- "$CURRENT" || restore_failed=1
    fi
    if [[ -n "$old_previous" ]]; then
      atomic_link "$old_previous" "$PREVIOUS" || restore_failed=1
    else
      rm -f -- "$PREVIOUS" || restore_failed=1
    fi
    if [[ -n "$old" ]]; then
      systemctl restart overdeck-subrouter.service || restore_failed=1
    else
      systemctl stop overdeck-subrouter.service || restore_failed=1
    fi
    (( restore_failed == 0 )) || fatal 'rollback candidate failed readiness and the original runtime could not be fully restored'
    fatal 'rollback candidate failed readiness; original runtime restored'
  fi
  if [[ -n "$old" && "$old" != "$target" ]]; then atomic_link "$old" "$PREVIOUS"; fi
  exit 0
fi

candidate_source=${1:-}
[[ -n "$candidate_source" && -d "$candidate_source" ]] || fatal 'candidate directory required'
[[ ! -L "$candidate_source" ]] || fatal 'candidate directory may not be a symlink'
candidate_source=$(realpath -e -- "$candidate_source")
candidate_name=${candidate_source##*/}

# Pin the submitted directory and copy it without following any directory entry. The
# private snapshot is made read-only before any receipt, tree, or source validation.
# /run is noexec on the workstation, so executable candidates cannot be staged beside
# the install lock.
if [[ -n "$TEST_ROOT" ]]; then
  SNAPSHOT_ROOT="$TEST_ROOT/var/tmp"
  install -d -m 0700 "$SNAPSHOT_ROOT"
else
  SNAPSHOT_ROOT=/var/tmp
  [[ -d "$SNAPSHOT_ROOT" && ! -L "$SNAPSHOT_ROOT" ]] || fatal 'candidate snapshot root is unavailable'
fi
CANDIDATE_SNAPSHOT_PARENT=$(mktemp -d "$SNAPSHOT_ROOT/overdeck-subrouter-candidate.XXXXXX")
if [[ -z "$TEST_ROOT" ]]; then chown root:root "$CANDIDATE_SNAPSHOT_PARENT"; fi
python3 - "$candidate_source" "$CANDIDATE_SNAPSHOT_PARENT/$candidate_name" <<'PY' || fatal 'candidate snapshot refused'
import os, pathlib, shutil, stat, sys
source, destination = sys.argv[1:]

def copy_directory(source_path, destination_path):
    source_fd = os.open(source_path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
    try:
        os.mkdir(destination_path, 0o700)
        with os.scandir(source_fd) as entries:
            for entry in entries:
                mode = entry.stat(follow_symlinks=False).st_mode
                source_child = f"/proc/self/fd/{source_fd}/{entry.name}"
                destination_child = os.path.join(destination_path, entry.name)
                if stat.S_ISDIR(mode):
                    copy_directory(source_child, destination_child)
                elif stat.S_ISREG(mode):
                    input_fd = os.open(entry.name, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=source_fd)
                    try:
                        before = os.fstat(input_fd)
                        with os.fdopen(os.open(destination_child, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o400), "wb") as output:
                            with os.fdopen(os.dup(input_fd), "rb") as input_file:
                                shutil.copyfileobj(input_file, output)
                            output.flush()
                            os.fsync(output.fileno())
                        os.chmod(destination_child, 0o500 if before.st_mode & 0o111 else 0o400)
                        after = os.fstat(input_fd)
                        if (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns, before.st_ctime_ns) != (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns, after.st_ctime_ns):
                            raise RuntimeError(f"candidate changed while snapshotting: {entry.name}")
                    finally:
                        os.close(input_fd)
                else:
                    raise RuntimeError(f"candidate symlink or special file refused: {entry.name}")
        os.chmod(destination_path, 0o500)
    finally:
        os.close(source_fd)

copy_directory(source, destination)
PY
candidate="$CANDIDATE_SNAPSHOT_PARENT/$candidate_name"
expected_candidate=$("$MODULE/bin/verify-candidate" path) || fatal 'candidate does not match the current repository inputs'
[[ "$candidate_source" == "$expected_candidate" ]] || fatal 'candidate is not the exact current-tree artifact'
if [[ -n "$TEST_ROOT" && -n ${OVERDECK_SUBROUTER_TEST_AFTER_SNAPSHOT:-} ]]; then
  bash -c "$OVERDECK_SUBROUTER_TEST_AFTER_SNAPSHOT"
fi
head=$(git -c "safe.directory=$ROOT" -C "$ROOT" rev-parse HEAD) || fatal 'could not snapshot source revision'
install -m 0400 "$MODULE/upstream.json" "$CANDIDATE_SNAPSHOT_PARENT/source-manifest.json"
install -m 0400 "$MODULE/patches/0001-overdeck-authority-safety.patch" "$CANDIDATE_SNAPSHOT_PARENT/source.patch"
install -m 0400 "$UNIT_SOURCE" "$CANDIDATE_SNAPSHOT_PARENT/service.unit"
install -m 0400 "$ROOT/modules/workstation/claude/buildbox-hosts.json" "$CANDIDATE_SNAPSHOT_PARENT/source-registry.json"
printf '%s\n' "$head" >"$CANDIDATE_SNAPSHOT_PARENT/source-head"
printf '%s\n' '{"schema":1,"authority_state_schema":1,"rollback_compatible_state_schemas":[1]}' >"$CANDIDATE_SNAPSHOT_PARENT/authority-state.json"
printf '%s\n' '{"schema":1}' >"$CANDIDATE_SNAPSHOT_PARENT/store-schema.json"
chmod -R a-w "$CANDIDATE_SNAPSHOT_PARENT"

validation=$(python3 - "$candidate" "$CANDIDATE_SNAPSHOT_PARENT" "$ROOT" <<'PY'
import hashlib, json, os, pathlib, re, stat, sys
candidate = pathlib.Path(sys.argv[1]).resolve()
source = pathlib.Path(sys.argv[2]).resolve()
root = pathlib.Path(sys.argv[3]).resolve()
receipt_path = candidate / "receipt.json"
artifact_manifest_path = candidate / "manifest.json"
binary = candidate / "bin/subrouter"
manifest_path = source / "source-manifest.json"
patch_path = source / "source.patch"
unit_path = source / "service.unit"
registry_path = source / "source-registry.json"
for path in candidate.rglob("*"):
    mode = path.lstat().st_mode
    if mode & 0o222: raise SystemExit(f"candidate writable path refused: {path}")
    if stat.S_ISLNK(mode): raise SystemExit(f"candidate symlink refused: {path}")
    if not (stat.S_ISDIR(mode) or stat.S_ISREG(mode)): raise SystemExit(f"candidate special file refused: {path}")
if not receipt_path.is_file() or not artifact_manifest_path.is_file() or not binary.is_file(): raise SystemExit("candidate is incomplete")
r = json.loads(receipt_path.read_text())
m = json.loads(manifest_path.read_text())
if json.loads(artifact_manifest_path.read_text()) != m: raise SystemExit("candidate manifest mismatch")
digest = r.get("candidate_digest", "")
if not re.fullmatch(r"[0-9a-f]{64}", digest): raise SystemExit("candidate digest is malformed")
if not re.fullmatch(r"[0-9a-f]{64}", r.get("input_digest", "")): raise SystemExit("input digest is malformed")
if not re.fullmatch(r"[0-9a-f]{64}", r.get("tree_diff_sha256", "")): raise SystemExit("tree diff digest is malformed")
if r.get("schema") != 1: raise SystemExit("receipt schema mismatch")
head = (source / "source-head").read_text().strip()
if r.get("base_commit") != head: raise SystemExit("receipt base commit mismatch")
if candidate.name != digest: raise SystemExit("candidate basename does not match receipt digest")
expected = {
    "complete": True,
    "revision": m["revision"],
    "archive_sha256": m["archive_sha256"],
    "patch_sha256": hashlib.sha256(patch_path.read_bytes()).hexdigest(),
    "go_version_required": m["go"],
}
for key, value in expected.items():
    if r.get(key) != value: raise SystemExit(f"receipt mismatch: {key}")
if r.get("temp_root_policy") != "${XDG_CACHE_HOME:-$HOME/.cache}/overdeck/tests/subrouter/<unique-run>":
    raise SystemExit("receipt temp-root policy mismatch")
module_basetemp = pathlib.Path(r.get("module_basetemp", ""))
if not module_basetemp.is_absolute(): raise SystemExit("receipt module basetemp is not absolute")
try: module_basetemp.resolve().relative_to(root)
except ValueError: pass
else: raise SystemExit("receipt module basetemp is inside checkout")
expected_commands = {
    "focused": "modules/subrouter/bin/verify-candidate focused",
    "static": "modules/subrouter/bin/verify-candidate static",
    "module": "modules/subrouter/bin/verify-candidate module",
    "full": "modules/subrouter/bin/verify-candidate full",
}
try:
    phase_records = {phase: json.loads((candidate / f"worker-{phase}.json").read_text()) for phase in expected_commands}
except (OSError, ValueError):
    raise SystemExit("candidate worker provenance is incomplete")
registry = json.loads(registry_path.read_text())
if set(r.get("logs", {})) != set(expected_commands): raise SystemExit("receipt log phases mismatch")
for phase, command in expected_commands.items():
    record = phase_records[phase]
    if record.get("phase") != phase or record.get("input_digest") != r["input_digest"] or record.get("tree_diff_sha256") != r["tree_diff_sha256"]:
        raise SystemExit(f"worker provenance mismatch: {phase}")
    if r.get("workers", {}).get(phase) != record.get("worker"): raise SystemExit(f"receipt worker mismatch: {phase}")
    if r.get("exits", {}).get(phase) != record.get("exit"): raise SystemExit(f"receipt exit record mismatch: {phase}")
    expected_timing = {name: record.get(name) for name in ("started_epoch", "ended_at", "elapsed_seconds")}
    if r.get("timing", {}).get(phase) != expected_timing: raise SystemExit(f"receipt timing mismatch: {phase}")
    if phase == "module":
        if record.get("basetemp") != r.get("module_basetemp") or record.get("log_sha256") != r["logs"][phase].get("sha256"):
            raise SystemExit("module worker provenance mismatch")
    dispatch_path = candidate / f"dispatch-{phase}.json"
    dispatch = json.loads(dispatch_path.read_text())
    if hashlib.sha256(dispatch_path.read_bytes()).hexdigest() != record.get("dispatch_sha256") or record.get("dispatch") != dispatch:
        raise SystemExit(f"dispatch provenance mismatch: {phase}")
    if r.get("dispatch", {}).get(phase) != dispatch: raise SystemExit(f"receipt dispatch mismatch: {phase}")
    expected_key = f"subrouter-{r['input_digest'][:16]}-{phase}"
    remote = dispatch.get("remote_meta", {})
    raw = dispatch.get("remote_meta_raw", "")
    argv = remote.get("argv", []) if isinstance(remote, dict) else []
    if dispatch.get("schema") != 1 or dispatch.get("dispatch_key") != expected_key or remote.get("key") != expected_key:
        raise SystemExit(f"dispatch key mismatch: {phase}")
    if not isinstance(raw, str) or hashlib.sha256(raw.encode()).hexdigest() != dispatch.get("meta_sha256") or json.loads(raw) != remote:
        raise SystemExit(f"dispatch metadata mismatch: {phase}")
    if hashlib.sha256(json.dumps(argv, separators=(",", ":")).encode()).hexdigest() != dispatch.get("argv_sha256"):
        raise SystemExit(f"dispatch command mismatch: {phase}")
    if dispatch.get("registry_sha256") != hashlib.sha256(registry_path.read_bytes()).hexdigest():
        raise SystemExit(f"dispatch registry mismatch: {phase}")
    registered = next((host for host in registry.get("hosts", []) if host.get("name") == dispatch.get("worker")), None)
    machine_id = registered.get("machine_id", "") if registered else ""
    if not registered or registered.get("state") != "reachable" or dispatch["worker"] not in registry.get("orders", {}).get("build", []):
        raise SystemExit(f"unregistered worker: {phase}")
    if hashlib.sha256(machine_id.encode()).hexdigest() != dispatch.get("machine_id_sha256") or dispatch.get("identity_source") != "machine-id":
        raise SystemExit(f"worker identity mismatch: {phase}")
    if r.get("commands", {}).get(phase) != command: raise SystemExit(f"receipt command mismatch: {phase}")
    if r.get("exits", {}).get(phase) != 0: raise SystemExit(f"receipt phase failed: {phase}")
    tool = r.get("toolchains", {}).get(phase, "")
    if phase == "module":
        if not tool.startswith("Python 3."): raise SystemExit("receipt toolchain mismatch: module")
    elif tool.split()[2:3] != ["go" + m["go"]]:
        raise SystemExit(f"receipt toolchain mismatch: {phase}")
    worker = r.get("workers", {}).get(phase, "")
    if not isinstance(worker, str) or not worker or len(worker) > 128: raise SystemExit(f"receipt worker mismatch: {phase}")
    timing = r.get("timing", {}).get(phase, {})
    if not isinstance(timing.get("started_epoch"), int) or not isinstance(timing.get("elapsed_seconds"), int) or timing["elapsed_seconds"] < 0:
        raise SystemExit(f"receipt timing mismatch: {phase}")
    if not isinstance(timing.get("ended_at"), str) or not timing["ended_at"].endswith("Z"):
        raise SystemExit(f"receipt end time mismatch: {phase}")
    item = r.get("logs", {}).get(phase, {})
    log_path = candidate / item.get("path", "")
    try: log_path.resolve(strict=True).relative_to(candidate)
    except (ValueError, OSError): raise SystemExit(f"receipt log escapes candidate: {phase}")
    if not log_path.is_file() or hashlib.sha256(log_path.read_bytes()).hexdigest() != item.get("sha256"):
        raise SystemExit(f"receipt log mismatch: {phase}")
expected_tree = {"bin", "bin/subrouter", "logs", "receipt.json", "manifest.json"}
expected_tree.update(r["logs"][phase]["path"] for phase in expected_commands)
expected_tree.update(f"worker-{phase}.json" for phase in expected_commands)
expected_tree.update(f"dispatch-{phase}.json" for phase in expected_commands)
actual_tree = {path.relative_to(candidate).as_posix() for path in candidate.rglob("*")}
if actual_tree != expected_tree: raise SystemExit("candidate tree does not exactly match receipt")
deps = r.get("dependency_state", {})
if not re.fullmatch(r"[0-9a-f]{64}", deps.get("go_sum_sha256", "")): raise SystemExit("receipt go.sum digest mismatch")
source_patch_sha = hashlib.sha256((m["archive_sha256"] + "\0" + expected["patch_sha256"]).encode()).hexdigest()
if deps.get("source_archive_plus_patch_sha256") != source_patch_sha: raise SystemExit("receipt source dependency digest mismatch")
actual_binary = hashlib.sha256(binary.read_bytes()).hexdigest()
if actual_binary != r.get("binary_sha256"): raise SystemExit("candidate binary digest mismatch")
candidate_hash = hashlib.sha256()
for value in (r["input_digest"], m["revision"], m["archive_sha256"], expected["patch_sha256"], m["go"]):
    candidate_hash.update(value.encode() + b"\0")
candidate_hash.update(bytes.fromhex(actual_binary))
if candidate_hash.hexdigest() != digest: raise SystemExit("candidate content digest mismatch")
if not os.access(binary, os.X_OK): raise SystemExit("candidate binary is not executable")
print(digest, actual_binary)
PY
) || fatal 'receipt is not a completed exact-revision receipt'
read -r digest binary_sha <<<"$validation"
target="$RELEASES/$digest"
case "$target" in "$RELEASES"/*) ;; *) fatal 'release target escapes release root' ;; esac

if [[ -z "$TEST_ROOT" ]]; then
  getent group overdeck-subrouter >/dev/null 2>&1 || groupadd --system overdeck-subrouter
  if id overdeck-subrouter >/dev/null 2>&1; then
    IFS=: read -r _ _ _ _ _ service_home service_shell < <(getent passwd overdeck-subrouter)
    [[ "$service_home" == "$RUNTIME/state/home" && "$service_shell" == /usr/sbin/nologin ]] || fatal 'existing service identity does not match the authority contract'
    [[ $(id -gn overdeck-subrouter) == overdeck-subrouter ]] || fatal 'existing service identity has the wrong primary group'
  else
    useradd --system --gid overdeck-subrouter --no-create-home --home-dir "$RUNTIME/state/home" --shell /usr/sbin/nologin overdeck-subrouter
  fi
  install -d -o root -g root -m 0755 "$RUNTIME" "$RELEASES" "$(dirname "$UNIT_TARGET")"
  install -d -o overdeck-subrouter -g overdeck-subrouter -m 0700 \
    "$RUNTIME/state" "$RUNTIME/state/home" "$RUNTIME/state/config" \
    "$RUNTIME/state/cache" "$RUNTIME/state/state"
else
  install -d -m 0755 "$RUNTIME" "$RELEASES" "$(dirname "$UNIT_TARGET")"
  install -d -m 0700 "$RUNTIME/state" "$RUNTIME/state/home" "$RUNTIME/state/config" \
    "$RUNTIME/state/cache" "$RUNTIME/state/state"
fi

admin_path="$RUNTIME/state/authority-admin"
client_token_path="${TEST_ROOT}/etc/overdeck/subrouter-authority-client.token"
install -d "${OWNER_ARGS[@]}" -m 0755 "$(dirname "$client_token_path")"
if [[ -e "$admin_path" || -L "$admin_path" ]]; then
  python3 - "$admin_path" <<'PY' || fatal 'authority administration token is invalid'
import os, pathlib, pwd, re, stat, sys
path = pathlib.Path(sys.argv[1])
mode = path.lstat().st_mode
if stat.S_ISLNK(mode) or not stat.S_ISREG(mode) or stat.S_IMODE(mode) != 0o600 or path.stat().st_nlink != 1:
    raise SystemExit("authority administration token must be an unlinked mode-0600 regular file")
if not os.environ.get("OVERDECK_SUBROUTER_TEST_ROOT") and path.stat().st_uid != pwd.getpwnam("overdeck-subrouter").pw_uid:
    raise SystemExit("authority administration token must be service-owned")
body = path.read_text().strip()
if body and not re.fullmatch(r"[0-9a-f]{64}", body):
    raise SystemExit("authority administration token content is invalid")
PY
fi
admin_token=$(python3 - "$admin_path" <<'PY'
import pathlib, re, secrets, sys
path = pathlib.Path(sys.argv[1])
if path.exists():
    body = path.read_text().strip()
    if re.fullmatch(r"[0-9a-f]{64}", body):
        print(body)
        raise SystemExit
print(secrets.token_hex(32))
PY
)
admin_staging="${admin_path}.next.$$"
printf '%s\n' "$admin_token" | install "${ADMIN_OWNER_ARGS[@]}" -m 0600 /dev/stdin "$admin_staging"
sync -f "$admin_staging"
mv -Tf "$admin_staging" "$admin_path"
admin_staging=
client_token_staging="${client_token_path}.next.$$"
printf '%s\n' "$admin_token" | install "${CLIENT_OWNER_ARGS[@]}" -m 0600 /dev/stdin "$client_token_staging"
sync -f "$client_token_staging"
mv -Tf "$client_token_staging" "$client_token_path"
client_token_staging=
sync -f "$RUNTIME/state" "$(dirname "$client_token_path")"
unset admin_token

schema_path="$RUNTIME/authority-state-schema.json"
if [[ -e "$schema_path" || -L "$schema_path" ]]; then
  read_store_schema >/dev/null || fatal 'authority state schema is invalid or unsupported'
else
  state_staging="${schema_path}.next.$$"
  install "${OWNER_ARGS[@]}" -m 0444 "$CANDIDATE_SNAPSHOT_PARENT/store-schema.json" "$state_staging"
  sync -f "$state_staging"
  mv -Tf "$state_staging" "$schema_path"
  state_staging=
  sync -f "$RUNTIME"
fi

if [[ -e "$target" ]]; then
  validate_release "$target" 1 >/dev/null || fatal 'existing release provenance mismatch'
  cmp -s "$candidate/receipt.json" "$target/receipt.json" || fatal 'existing release receipt mismatch'
  cmp -s "$candidate/manifest.json" "$target/manifest.json" || fatal 'existing release manifest mismatch'
else
  staging="$RELEASES/.${digest}.staging.$$"
  install -d "${OWNER_ARGS[@]}" -m 0755 "$staging" "$staging/bin" "$staging/logs"
  install "${OWNER_ARGS[@]}" -m 0555 "$candidate/bin/subrouter" "$staging/bin/subrouter"
  install "${OWNER_ARGS[@]}" -m 0444 "$candidate/receipt.json" "$staging/receipt.json"
  install "${OWNER_ARGS[@]}" -m 0444 "$candidate/manifest.json" "$staging/manifest.json"
  install "${OWNER_ARGS[@]}" -m 0444 "$CANDIDATE_SNAPSHOT_PARENT/authority-state.json" "$staging/authority-state.json"
  install "${OWNER_ARGS[@]}" -m 0444 "$CANDIDATE_SNAPSHOT_PARENT/source.patch" "$staging/source.patch"
  install "${OWNER_ARGS[@]}" -m 0444 "$CANDIDATE_SNAPSHOT_PARENT/source-head" "$staging/source-head"
  install "${OWNER_ARGS[@]}" -m 0444 "$CANDIDATE_SNAPSHOT_PARENT/source-registry.json" "$staging/source-registry.json"
  install "${OWNER_ARGS[@]}" -m 0444 "$CANDIDATE_SNAPSHOT_PARENT/service.unit" "$staging/service.unit"
  for phase in focused static module full; do
    install "${OWNER_ARGS[@]}" -m 0444 "$candidate/worker-$phase.json" "$staging/worker-$phase.json"
  done
  for phase in focused static full; do
    install "${OWNER_ARGS[@]}" -m 0444 "$candidate/dispatch-$phase.json" "$staging/dispatch-$phase.json"
  done
  while IFS= read -r -d '' file; do
    rel=${file#"$candidate/logs/"}
    [[ "$rel" != */* ]] || fatal 'nested candidate log path refused'
    install "${OWNER_ARGS[@]}" -m 0444 "$file" "$staging/logs/$rel"
  done < <(find "$candidate/logs" -mindepth 1 -maxdepth 1 -type f -print0)
  chmod -R a-w "$staging"
  sync -f "$staging/bin/subrouter" "$staging/receipt.json" "$staging/manifest.json" "$staging/authority-state.json" "$staging/source.patch" "$staging/source-head" "$staging/source-registry.json" "$staging/service.unit" "$staging"/worker-*.json "$staging"/dispatch-*.json "$staging/logs" "$staging"
  mv "$staging" "$target"
  staging=
  sync -f "$RELEASES"
fi
validate_release "$target" 1 >/dev/null || fatal 'installed release provenance validation failed'

validate_provider_tree() {
  local allow_credentials=${1:-0}
  python3 - "$RUNTIME/state" "$allow_credentials" <<'PY'
import os, pwd, re, stat, sys
state = os.path.abspath(sys.argv[1])
allow_credentials = sys.argv[2] == "1"
roots = (
    os.path.join(state, "providers", "codex", "accounts"),
    os.path.join(state, "providers", "claude"),
)
seen = 0
name_policy = re.compile(r"[A-Za-z0-9._@+-]{1,128}")
if os.environ.get("OVERDECK_SUBROUTER_TEST_ROOT"):
    expected_uid = os.geteuid()
    expected_gid = os.getegid()
else:
    identity = pwd.getpwnam("overdeck-subrouter")
    expected_uid = identity.pw_uid
    expected_gid = identity.pw_gid

def validate_directory(metadata):
    if (
        not stat.S_ISDIR(metadata.st_mode)
        or stat.S_IMODE(metadata.st_mode) != 0o700
        or metadata.st_uid != expected_uid
        or metadata.st_gid != expected_gid
    ):
        raise SystemExit("provider credential tree directory policy refused")

def walk(path, depth):
    global seen
    if depth > 5 or len(path) > 4096:
        raise SystemExit("provider credential tree path policy refused")
    fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
    try:
        validate_directory(os.fstat(fd))
        with os.scandir(fd) as entries:
            for entry in entries:
                seen += 1
                if seen > 256 or not name_policy.fullmatch(entry.name):
                    raise SystemExit("provider credential tree filename policy refused")
                metadata = entry.stat(follow_symlinks=False)
                child = os.path.join(path, entry.name)
                if stat.S_ISDIR(metadata.st_mode):
                    validate_directory(metadata)
                    walk(child, depth + 1)
                elif stat.S_ISREG(metadata.st_mode):
                    if (
                        stat.S_IMODE(metadata.st_mode) != 0o600
                        or metadata.st_nlink != 1
                        or metadata.st_uid != expected_uid
                        or metadata.st_gid != expected_gid
                    ):
                        raise SystemExit("provider credential tree file policy refused")
                    if allow_credentials:
                        continue
                    if depth == 0 and metadata.st_size == 0 and entry.name.endswith(".json.lock"):
                        continue
                    raise SystemExit("provider credentials are not allowed before the first trusted S1 install")
                else:
                    raise SystemExit("provider credential tree type policy refused")
    finally:
        os.close(fd)

for root in roots:
    try:
        root_metadata = os.lstat(root)
    except FileNotFoundError:
        continue
    validate_directory(root_metadata)
    walk(root, 0)
PY
}


authority_status() {
  local binary=$1
  if [[ -n "$TEST_ROOT" ]]; then
    "$binary" authority-status --state-dir "$RUNTIME/state"
  else
    /usr/sbin/runuser --user overdeck-subrouter -- /usr/bin/env -i \
      HOME="$RUNTIME/state/home" \
      XDG_CONFIG_HOME="$RUNTIME/state/config" \
      XDG_CACHE_HOME="$RUNTIME/state/cache" \
      XDG_STATE_HOME="$RUNTIME/state/state" \
      SUBROUTER_STATE_DIR="$RUNTIME/state" \
      OVERDECK_SUBROUTER_SERVICE_IDENTITY=overdeck-subrouter \
      "$binary" authority-status --state-dir "$RUNTIME/state"
  fi
}
old_current=
trusted_upgrade=0
if [[ -L "$CURRENT" ]]; then
  old_current=$(readlink -f "$CURRENT") || fatal 'current release link is invalid'
  case "$old_current" in "$RELEASES"/*) ;; *) fatal 'current release escapes release root' ;; esac
  validate_release "$old_current" 1 >/dev/null || fatal 'current release provenance or authority schema is invalid'
  trusted_upgrade=1
elif [[ -e "$CURRENT" ]]; then
  fatal 'current release is not a symlink'
fi
validate_provider_tree "$trusted_upgrade" || fatal 'provider credential tree metadata policy refused'
pre_status=$(authority_status "$target/bin/subrouter") \
  || fatal 'candidate could not inspect the isolated authority state as the service identity'
grep -Eq '^Provider credentials: [0-9]+$' <<<"$pre_status" || fatal 'candidate provider credential count is invalid'
grep -Eq '^Authority routes: [0-9]+$' <<<"$pre_status" || fatal 'candidate authority route count is invalid'
grep -Eq '^Proxy grants: [0-9]+$' <<<"$pre_status" || fatal 'candidate proxy grant count is invalid'
grep -qx 'Unresolved attempts: 0' <<<"$pre_status" || fatal 'S0 refuses a state directory containing unresolved refresh attempts'
if [[ "$trusted_upgrade" == 0 ]]; then
  grep -qx 'Provider credentials: 0' <<<"$pre_status" || fatal 'S0 refuses a state directory containing provider credentials before the first trusted install'
fi
pre_provider_count=$(grep -E '^Provider credentials: [0-9]+$' <<<"$pre_status")
pre_route_count=$(grep -E '^Authority routes: [0-9]+$' <<<"$pre_status")
pre_grant_count=$(grep -E '^Proxy grants: [0-9]+$' <<<"$pre_status")

old_previous=$(readlink -f "$PREVIOUS" 2>/dev/null || true)
old_unit=$(mktemp)
had_unit=0
was_enabled=0
if [[ -f "$UNIT_TARGET" ]]; then cp --preserve=mode,timestamps "$UNIT_TARGET" "$old_unit"; had_unit=1; fi
if systemctl is-enabled --quiet overdeck-subrouter.service >/dev/null 2>&1; then was_enabled=1; fi

unit_staging="${UNIT_TARGET}.next.$$"
install "${OWNER_ARGS[@]}" -m 0644 "$target/service.unit" "$unit_staging"
post_status=
activate_candidate() {
  if [[ -n "$old_current" && "$old_current" != "$target" ]]; then
    atomic_link "$old_current" "$PREVIOUS" || return
  fi
  atomic_link "$target" "$CURRENT" || return
  mv -Tf "$unit_staging" "$UNIT_TARGET" || return
  sync -f "$CURRENT" "$UNIT_TARGET" "$RUNTIME" "$(dirname "$UNIT_TARGET")" || return
  systemctl daemon-reload || return
  systemctl restart overdeck-subrouter.service || return
  wait_ready "$CURRENT/bin/subrouter" || return
  validate_provider_tree "$trusted_upgrade" || return
  post_status=$(authority_status "$CURRENT/bin/subrouter") || return
  grep -qx 'Store lease: held' <<<"$post_status" || return
  grep -qx "$pre_provider_count" <<<"$post_status" || return
  grep -qx "$pre_route_count" <<<"$post_status" || return
  grep -qx "$pre_grant_count" <<<"$post_status" || return
  grep -qx 'Unresolved attempts: 0' <<<"$post_status" || return
  grep -qx 'Gateway: ready' <<<"$post_status" || return
  systemctl enable overdeck-subrouter.service || return
}

if ! activate_candidate; then
  if ! rollback_runtime "$old_current" "$old_previous" "$old_unit" "$had_unit" "$was_enabled"; then
    fatal 'candidate activation failed and the previous runtime could not be fully restored'
  fi
  fatal 'candidate failed activation or readiness; previous runtime restored'
fi

printf '%s\n' "$post_status"