#!/usr/bin/env python3 """Persistent, narrowly-scoped blunix deployment runner. The builder only creates a durable request. This process owns the switch and therefore is not a child of agent-codex-builder.service. """ import json import fcntl import os import re import subprocess import sys import time import uuid from pathlib import Path def payload_path(directory, job_id): return Path(directory) / f"{job_id}.json" def write_payload(path, payload): path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp") try: with temporary.open("x", encoding="utf-8") as stream: temporary.chmod(0o600) stream.write(json.dumps(payload, indent=2) + "\n") stream.flush() os.fsync(stream.fileno()) temporary.replace(path) fd = os.open(path.parent, os.O_DIRECTORY) try: os.fsync(fd) finally: os.close(fd) finally: temporary.unlink(missing_ok=True) def run_switch(request, environment=None, runner=None): environment = environment or os.environ runner = runner or subprocess.run if request.get("host") != "blunix": raise ValueError("Deployment ist ausschließlich für blunix erlaubt") repo = Path(request["repo"]) expected_repo = Path(environment.get("AGENT_BUILDER_DEPLOY_REPO", "/home/drone/nixos-config")) if repo != expected_repo: raise ValueError(f"Deployment-Repository muss exakt {expected_repo} sein") timeout = int(environment.get("AGENT_BUILDER_DEPLOY_TIMEOUT_SECONDS", "1200")) revision = str(request.get("commit_hash", "")) if not re.fullmatch(r"[0-9a-f]{7,40}", revision): raise ValueError("Deployment requires a reviewed commit hash") resolved = runner( ["git", "-C", str(repo), "rev-parse", "--verify", f"{revision}^{{commit}}"], capture_output=True, text=True, timeout=timeout, check=False, ) exact_revision = resolved.stdout.strip() if resolved.returncode or not re.fullmatch(r"[0-9a-f]{40}", exact_revision): return False, "Reviewed commit is unavailable or ambiguous; no activation attempted." request["resolved_commit_hash"] = exact_revision state_dir = environment.get("AGENT_BUILDER_STATE_DIR") manifest = Path(state_dir) / "last-verified-deployment.json" if state_dir else None if manifest and manifest.exists(): previous = json.loads(manifest.read_text(encoding="utf-8"))["deployed_commit_hash"] ancestry = runner( ["git", "-C", str(repo), "merge-base", "--is-ancestor", previous, exact_revision], capture_output=True, text=True, timeout=timeout, check=False, ) if ancestry.returncode: return False, "Automatic deployment would roll back or diverge from the last verified revision; manual reconciliation required." source = f"git+file://{repo}?rev={exact_revision}" build = runner( ["nix", "build", "--no-link", "--print-out-paths", source + "#nixosConfigurations.blunix.config.system.build.toplevel"], capture_output=True, text=True, timeout=timeout, check=False, ) expected_system = build.stdout.strip() if build.returncode or not re.fullmatch(r"/nix/store/[^/\s]+", expected_system): return False, "Approved revision failed to build; no activation attempted.\n" + build.stderr[-4000:] command = [ environment["AGENT_BUILDER_SUDO_BIN"], environment["AGENT_BUILDER_ENV_BIN"], "HOME=/tmp", environment["AGENT_BUILDER_NIXOS_REBUILD_BIN"], "switch", "--flake", source + "#blunix", ] result = runner( command, capture_output=True, text=True, timeout=int(environment.get("AGENT_BUILDER_DEPLOY_TIMEOUT_SECONDS", "1200")), check=False, ) output = (result.stdout + result.stderr).strip() if result.returncode != 0: return False, output health = runner( [ environment.get( "AGENT_BUILDER_SYSTEMCTL_BIN", "/run/current-system/sw/bin/systemctl", ), "is-active", "board-agent-trigger.service", ], capture_output=True, text=True, timeout=30, check=False, ) health_output = (health.stdout + health.stderr).strip() combined = "\n".join(part for part in (output, health_output) if part) healthy = health.returncode == 0 and health.stdout.strip() == "active" if healthy: generation = runner( ["readlink", "-f", "/run/current-system"], capture_output=True, text=True, timeout=30, check=False, ) closure = generation.stdout.strip() if generation.returncode or closure != expected_system: return False, combined + "\nActivated system does not match the approved revision's built closure." request["activated_system"] = closure request["deployed_commit_hash"] = exact_revision return healthy, combined def process(path, environment=None, runner=None): environment = environment or os.environ with (path.parent / '.delivery.lock').open('a+') as handle: fcntl.flock(handle, fcntl.LOCK_EX) request = json.loads(path.read_text(encoding="utf-8")) write_payload(path.parent / '.claimed' / path.name, {'claimed': True}) state_dir = Path(environment["AGENT_BUILDER_STATE_DIR"]) state_dir.mkdir(parents=True, exist_ok=True, mode=0o700) state_path = state_dir / path.name if state_path.exists(): saved = json.loads(state_path.read_text(encoding="utf-8")) if saved.get("state") in {"deployed", "deploy_failed"}: _write_completion(saved) path.unlink(missing_ok=True) return saved if request.get("state") == "deployed": path.unlink(missing_ok=True) return request max_retries = int(environment.get("AGENT_BUILDER_DEPLOY_MAX_RETRIES", "3")) backoff = float(environment.get("AGENT_BUILDER_DEPLOY_BACKOFF_SECONDS", "5")) retries = int(request.get("retry_count", 0)) ok = False output = "" while retries < max_retries and not ok: retries += 1 request.update(state="deploying", retry_count=retries, updated_at=time.time()) write_payload(path, request) write_payload(state_path, request) try: ok, output = run_switch(request, environment, runner) except (OSError, ValueError, subprocess.TimeoutExpired) as error: output = str(error) if not ok and retries < max_retries and backoff: time.sleep(backoff * retries) if not ok and retries >= max_retries: output = output or "maximale Deployment-Wiederholungen erreicht" request.update( state="deployed" if ok else "deploy_failed", output=output[-4000:], updated_at=time.time(), ) write_payload(state_path, request) if ok: write_payload(state_dir / "last-verified-deployment.json", { "deployed_commit_hash": request["deployed_commit_hash"], "activated_system": request["activated_system"], "job_id": request["job_id"], }) _write_completion(request) path.unlink(missing_ok=True) return request def _write_completion(request): completion_dir = request.get("on_complete_dir") if completion_dir: target = Path(completion_dir) / f"{request['job_id']}.json" if not target.exists(): target.parent.mkdir(parents=True, exist_ok=True, mode=0o700) write_payload( target, { "job_id": request["job_id"], "status": "committed" if request["state"] == "deployed" else "deploy_failed", "deployment_status": request["state"], "deployed_commit_hash": request.get("deployed_commit_hash"), "activated_system": request.get("activated_system"), "failure_detail": request.get("output") if request["state"] != "deployed" else None, "commits": [{"repo": Path(request["repo"]).name, "branch": "dev", "hash": request["commit_hash"]}], "completed_at": time.time(), "runtime": request.get("runtime", "codex"), "pipeline_run_id": request.get("pipeline_run_id"), "board_slug": request.get("board_slug"), "configured_profile": request.get("configured_profile") or {}, "telemetry": request.get("telemetry") or {}, }, ) def main(): queue = Path(os.environ["AGENT_BUILDER_DEPLOY_QUEUE_DIR"]) for path in sorted(queue.glob("*.json")): result = process(path) print(f"{result['job_id']}: {result['state']}", file=sys.stderr) if __name__ == "__main__": main()