#!/usr/bin/env python3 """deploy_stage.py — stage + verify one CI-qualified revision for deployment. Called by .gitea/workflows/deploy.yml. Stages src/ into /mnt/workdrive/rune-code/, stamps _build_identity.json, verifies the staged tree (import + SHA assert + authority smoke), then promotes through the MONOTONIC promotion authority (deploy_promotion.py): an older candidate finishing after a newer promoted one is fenced out (STALE_SUPERSEDED) and reports truthfully — build/stage success != promotion success. Exit 0 = PROMOTED and verified, or NOOP_ALREADY_CURRENT, or a truthful STALE_SUPERSEDED (newer build already current; nothing mutated). Non-zero = deployment refused (divergent/unknown ordering, verification failure); the previous deployment stays current. """ from __future__ import annotations import json import os import shutil import subprocess import sys import time from pathlib import Path DEPLOY_ROOT = Path("/mnt/workdrive/rune-code") REPO_ROOT = Path(__file__).resolve().parent.parent VENV_PY = "/home/nick/.venvs/memory/bin/python3" sys.path.insert(0, str(REPO_ROOT / "scripts")) sys.path.insert(0, str(REPO_ROOT / "src")) import deploy_promotion # noqa: E402 from rune.release_notes import ( # noqa: E402 ReleaseNote as _ReleaseNote, auto_initial_note as _auto_note, converged_deployed_sha as _authority_sha, convergence_refresh as _refresh, fetch_release as _fetch_release, publish_release as _publish, sha_from_body as _sha_from_body, summary_from_body as _summary_from_body, ) def publish_out_of_lock(short: str, sha: str, *, deploy_root: Path = DEPLOY_ROOT, scaffold_store: Path | None = None) -> None: """§13D/§14B: release-history publication AFTER the promotion lock is released. Network I/O never holds the flock. §14B INVARIANT: publication does NOT trust the originating job's SHA. The CURRENT deployment authority is re-read immediately before publishing; the note is constructed from CURRENT authority. A stale job whose candidate was superseded converges the NEW authority (or refuses a version it cannot truthfully describe) — it may help converge B, but may never reassert itself as deployed. Failure degrades to STALE (software valid, history lags; retry converges).""" try: server = os.environ.get("GITEA_SERVER", "").rstrip("/") repo = os.environ.get("GITEA_REPO", "") token = os.environ.get("GITEA_TOKEN", "") if not (server and repo and token): return # §14B: re-read CURRENT authority — never trust the job's SHA. authority_sha = _authority_sha(str(deploy_root)) if not authority_sha: print("WARNING: release-history convergence skipped: " "deployment authority unreadable — STALE, retry converges") return store = str(scaffold_store or (REPO_ROOT / "store" / "scaffold" / "current.json")) identity = deploy_root / authority_sha[:12] / "src" / "rune" / "_build_identity.json" try: version = json.loads(identity.read_text()).get("version", "") except (OSError, ValueError): version = "" if not version: print("WARNING: release-history convergence skipped: " "no version on current authority — STALE, retry converges") return tag = f"v{version}" existing = _fetch_release(server, repo, token, tag) if existing is None: note = _auto_note( version=version, summary=(f"Rune v{version} — evidence-derived release " f"history (automatic convergence)"), scaffold_store=store, release_sha=authority_sha, deployed_sha=authority_sha, provenance=["automatic convergence publish " "(deploy workflow, authority-derived)"]) else: note = _refresh( _ReleaseNote.from_dict({ "version": version, "summary": _summary_from_body(existing.get("body", "")), "release_sha": _sha_from_body(existing.get("body", ""), "tag") or authority_sha, "provenance": [], }), deployed_sha=authority_sha, scaffold_store=store, extra_provenance="automatic convergence publish " "(deploy workflow, authority-derived)") # preserve the tag SHA the stored body carries tag_sha = _sha_from_body(existing.get("body", ""), "tag") if tag_sha: note.release_sha = tag_sha result = _publish(server, repo, token, note) print(f"release-history convergence: " f"{result.get('action', 'unknown')} " f"(authority {authority_sha[:12]})") except Exception as conv_exc: # noqa: BLE001 — history STALE, software valid print(f"WARNING: release-history convergence failed: " f"{conv_exc!s:.200} — STALE, retry converges") def main() -> int: # The runner's checkout workspace (act sets GITHUB_WORKSPACE). Never a # hardcoded dev path: the runner user has its own checkout and git # refuses "dubious ownership" on foreign working copies. repo = Path(os.environ.get("GITHUB_WORKSPACE", os.getcwd())).resolve() sha = subprocess.run( ["git", "-C", str(repo), "-c", f"safe.directory={repo}", "rev-parse", "HEAD"], capture_output=True, text=True, ).stdout.strip() if not sha: print(f"FATAL: cannot resolve HEAD in {repo}", file=sys.stderr) return 1 short = sha[:12] # --- stage --------------------------------------------------------- stage = DEPLOY_ROOT / short tmp_stage = DEPLOY_ROOT / f".staging-{short}-{int(time.time())}" shutil.copytree(repo / "src", tmp_stage / "src") # shared/ is required at runtime: src/rune bare imports (prompt_sync -> # build_pipeline_policy, memory_store -> shard_manager, etc.) resolve # via the ../../shared fallback from src/rune/*. Its absence broke the # deployed TUI launch (ModuleNotFoundError: build_pipeline_policy). if (repo / "shared").exists(): shutil.copytree(repo / "shared", tmp_stage / "shared") for extra in ("pyproject.toml", "README.md"): if (repo / extra).exists(): shutil.copy2(repo / extra, tmp_stage / extra) # --- build identity ------------------------------------------------- sys.path.insert(0, str(repo / "src")) import rune as rune_dev # noqa: E402 — read the source version identity = { "commit": sha, "version": rune_dev.__version__, "deployed_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), } (tmp_stage / "src" / "rune" / "_build_identity.json").write_text(json.dumps(identity, indent=2)) # --- verify the STAGED tree (before promotion) ---------------------- env = dict(os.environ, RUNE_HOME="/mnt/workdrive/rune", RUNE_MEMORY_DIR="/home/nick/.config/opencode/memory") verify = subprocess.run( [ VENV_PY, "-c", f""" import json, sys sys.path.insert(0, {str(tmp_stage / "src")!r}) import rune info = rune.build_info() assert info["source"] == "deployed", f"identity file missing: {{info}}" assert info["commit"] == {sha!r}, f"SHA mismatch: {{info['commit']}}" assert info["version"], "empty version" from rune import adaptive_router, builder_bridge, context_broker, rune_operator # noqa: F401 print("VERIFIED:", json.dumps(info)) """, ], capture_output=True, text=True, env=env, timeout=120, ) print(verify.stdout.strip()) if verify.returncode != 0: print("VERIFY FAILED — deployment refused:\n" + verify.stderr[-800:], file=sys.stderr) shutil.rmtree(tmp_stage, ignore_errors=True) return 1 # --- TUI startup canary (handoff 2026-09-10 §4): the ACTUAL # RuneApp initialization must succeed before promotion. A staged # revision whose TUI cannot initialize never becomes current. canary = subprocess.run( [ VENV_PY, str(REPO_ROOT / "scripts" / "tui_startup_canary.py"), "--mode", "candidate", "--stage", str(tmp_stage), ], capture_output=True, text=True, env={**env, "PYTHONPATH": str(tmp_stage / "src")}, timeout=180, cwd=str(tmp_stage), ) print(canary.stdout.strip()) if canary.returncode != 0: print("TUI STARTUP CANARY FAILED — deployment refused:\n" + canary.stderr[-800:], file=sys.stderr) shutil.rmtree(tmp_stage, ignore_errors=True) return 1 # --- promote: FENCED through the promotion authority ----------------- # SCAFFOLD contract DEPLOY-MONOTONIC-001: promotion is monotonic. The # ancestry check + lock live in deploy_promotion.py; this script only # handles the truthful outcome reporting. A stale candidate must NOT # claim promotion success — build/stage success != promotion success. # # §14C: NO separate convergence-intent ledger. The durable pending- # convergence state IS the deployment authority itself (atomically # swapped `current` symlink + stamped build identity): publication # derives from it, idempotently, outside the lock. The old intent # file had exactly one writer and zero readers, non-atomic writes, # and a bogus recorded_at (SECONDS is a bash builtin, never exported # — always "0"). Deriving from authority gives atomicity, restart # readability, corruption immunity, monotonicity (stale publishers # read CURRENT authority, not their own job) and idempotent retry # for free. REMOVE more complexity than you add. outcome, detail = deploy_promotion.promote( sha, DEPLOY_ROOT / short, tmp_stage, repo) # §13D: publish release history OUTSIDE the promotion lock (network # I/O must not hold the flock). Only for PROMOTED outcomes. if outcome == deploy_promotion.OUTCOME_PROMOTED: publish_out_of_lock(short, sha) if outcome == deploy_promotion.OUTCOME_PROMOTED: print(f"PROMOTED: {short}") print("current ->", os.readlink(DEPLOY_ROOT / "current")) if (DEPLOY_ROOT / "last-good").is_symlink(): print("last-good ->", os.readlink(DEPLOY_ROOT / "last-good")) return 0 if outcome == deploy_promotion.OUTCOME_NOOP: print(f"NOOP_ALREADY_CURRENT: {short} is the deployed commit") print(detail) return 0 if outcome == deploy_promotion.OUTCOME_STALE: print(f"STALE_SUPERSEDED: {short} is older than the deployed commit — promotion refused (monotonic promotion)") print(detail) print("current unchanged:", os.readlink(DEPLOY_ROOT / "current")) return 0 # truthful non-error terminal state: the newer build won # divergent / unknown-current: fail closed print(f"REFUSED ({outcome}): {short} — {detail}", file=sys.stderr) return 1 if __name__ == "__main__": sys.exit(main())