zemion
2ffdb23f69
Verified with the coordinated workspace changes by devkit full run 2026-09-08T225814-186389-0000-3e3ed7cd (all seven phases passed). This shared UI pass does not mark the individual module reviews complete.
367 lines
20 KiB
Python
Executable File
367 lines
20 KiB
Python
Executable File
"""Headless adapter to the existing, receipt-bound release console lifecycle.
|
|
|
|
The ASGI application runs in this process; no HTTP socket or background server
|
|
is started. Critical release orchestration remains in the existing service.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import importlib
|
|
from pathlib import Path
|
|
import re
|
|
import secrets
|
|
import sys
|
|
from typing import Any
|
|
from urllib.parse import quote
|
|
|
|
from .common import redact
|
|
|
|
|
|
META_ROOT = Path(__file__).resolve().parents[3]
|
|
RELEASE_ROOT = META_ROOT / "tools/release"
|
|
_REPOSITORY = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}\Z")
|
|
_VERSION = re.compile(r"[0-9]+\.[0-9]+\.[0-9]+(?:[-+][0-9A-Za-z][0-9A-Za-z.-]*)?\Z")
|
|
_RUN = re.compile(r"rr-(?:[0-9]{8}T[0-9]{6}Z-[0-9a-f]{12}|request-[0-9a-f]{64})\Z")
|
|
_STEP = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:@-]{0,159}\Z")
|
|
_REQUEST = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:@/-]{7,127}\Z")
|
|
|
|
|
|
def _typed(pattern: re.Pattern[str], label: str):
|
|
def parse(value: str) -> str:
|
|
if not pattern.fullmatch(value):
|
|
raise argparse.ArgumentTypeError(f"Invalid {label}.")
|
|
return value
|
|
return parse
|
|
|
|
|
|
def _limit(value: str) -> int:
|
|
try:
|
|
result = int(value)
|
|
except ValueError as exc:
|
|
raise argparse.ArgumentTypeError("Limit must be an integer from 1 to 100.") from exc
|
|
if not 1 <= result <= 100:
|
|
raise argparse.ArgumentTypeError("Limit must be an integer from 1 to 100.")
|
|
return result
|
|
|
|
|
|
def _planning_options(parser: argparse.ArgumentParser, *, selection: bool) -> None:
|
|
if selection:
|
|
parser.add_argument("--repo", action="append", default=[], type=_typed(_REPOSITORY, "repository name"))
|
|
parser.add_argument("--repo-version", action="append", default=[], metavar="REPO=VERSION")
|
|
parser.add_argument("--target-version", type=_typed(_VERSION, "target version"))
|
|
parser.add_argument("--channel", default="stable")
|
|
parser.add_argument("--online", action="store_true", help="Allow the existing remote/catalog checks.")
|
|
parser.add_argument("--remote-tags", action="store_true", help="Explicitly inspect remote Git tags.")
|
|
parser.add_argument("--public-catalog", action="store_true", help="Explicitly inspect the public catalog.")
|
|
parser.add_argument("--include-migrations", action="store_true", help="Run migration audits; never applies migrations.")
|
|
|
|
|
|
def register(subparsers: Any) -> None:
|
|
"""Register release commands without importing FastAPI, HTTPX or Core."""
|
|
parser = subparsers.add_parser("release", help="Plan and operate durable GovOPlaN release runs.")
|
|
commands = parser.add_subparsers(dest="release_command", required=True)
|
|
for name in ("plan", "status", "create"):
|
|
command = commands.add_parser(name)
|
|
_planning_options(command, selection=name != "status")
|
|
if name == "status":
|
|
command.add_argument("--include-website", action="store_true")
|
|
if name == "create":
|
|
command.add_argument("--request-id", required=True, type=_typed(_REQUEST, "request ID"))
|
|
command.add_argument("--apply", action="store_true", help="Persist the frozen run; otherwise preview only.")
|
|
command.set_defaults(handler=handle)
|
|
|
|
listing = commands.add_parser("list", help="List bounded, workspace-scoped durable run history.")
|
|
listing.add_argument("--limit", type=_limit, default=20)
|
|
listing.add_argument("--cursor")
|
|
listing.set_defaults(handler=handle)
|
|
for name in ("show", "preview", "execute", "resume", "retry", "reconcile"):
|
|
command = commands.add_parser(name)
|
|
command.add_argument("run_id", type=_typed(_RUN, "run ID"))
|
|
if name in {"preview", "execute", "retry", "reconcile"}:
|
|
command.add_argument("step_id", type=_typed(_STEP, "step ID"))
|
|
if name in {"execute", "resume", "retry", "reconcile"}:
|
|
command.add_argument("--request-id", required=True, type=_typed(_REQUEST, "request ID"))
|
|
command.add_argument("--apply", action="store_true", help="Apply this explicit durable transition; otherwise preview only.")
|
|
if name in {"execute", "reconcile"}:
|
|
command.add_argument("--confirm", default="", help="Exact confirmation required by the existing release service.")
|
|
if name == "execute":
|
|
command.add_argument("--signing-key", action="append", default=[], metavar="KEY_ID=PRIVATE_KEY_FILE")
|
|
if name == "reconcile":
|
|
command.add_argument("--outcome", required=True, choices=("effect_absent", "effect_succeeded", "unresolved"))
|
|
command.set_defaults(handler=handle)
|
|
|
|
|
|
def _selection(args: argparse.Namespace) -> tuple[list[str], dict[str, str]]:
|
|
versions: dict[str, str] = {}
|
|
for item in args.repo_version:
|
|
repo, separator, version = item.partition("=")
|
|
if not separator or not _REPOSITORY.fullmatch(repo) or not _VERSION.fullmatch(version):
|
|
raise ValueError("--repo-version must be REPO=VERSION with a valid repository and version.")
|
|
if repo in versions and versions[repo] != version:
|
|
raise ValueError(f"Conflicting target versions for {repo}.")
|
|
versions[repo] = version
|
|
repos = list(dict.fromkeys([*args.repo, *versions]))
|
|
if not repos:
|
|
raise ValueError("Select at least one --repo or --repo-version explicitly.")
|
|
for repo in repos:
|
|
if repo not in versions and args.target_version:
|
|
versions[repo] = args.target_version
|
|
if args.release_command == "create" and any(repo not in versions for repo in repos):
|
|
raise ValueError("Creating a run requires an explicit version for every selected repository.")
|
|
return repos, versions
|
|
|
|
|
|
def _planning_query(args: argparse.Namespace) -> dict[str, Any]:
|
|
return {
|
|
"channel": args.channel,
|
|
"online": args.online,
|
|
"remote_tags": args.remote_tags,
|
|
"public_catalog": args.public_catalog or args.online,
|
|
"include_migrations": args.include_migrations,
|
|
**({"target_version": args.target_version} if args.target_version else {}),
|
|
}
|
|
|
|
|
|
def _load_application(args: argparse.Namespace) -> tuple[Any, str]:
|
|
# The generic CLI/help path stays dependency-light. The backend still checks
|
|
# its operator-controlled runtime and registered source origins on apply.
|
|
for name, loaded in list(sys.modules.items()):
|
|
if name in {"server", "govoplan_release"} or name.startswith(("server.", "govoplan_release.")):
|
|
expected = RELEASE_ROOT / ("server" if name.startswith("server") else "govoplan_release")
|
|
source = getattr(loaded, "__file__", None)
|
|
if not isinstance(source, str) or not Path(source).resolve().is_relative_to(expected.resolve()):
|
|
raise ValueError("A foreign module shadows the trusted GovOPlaN release service.")
|
|
# Reprioritize even when the path was added previously below an unrelated
|
|
# working directory. Validate cached packages before importing any submodule.
|
|
sys.path[:] = [str(RELEASE_ROOT), *(path for path in sys.path if path != str(RELEASE_ROOT))]
|
|
module = importlib.import_module("server.app")
|
|
if Path(module.__file__).resolve() != (RELEASE_ROOT / "server/app.py").resolve():
|
|
raise ValueError("A foreign server.app module shadows the GovOPlaN release service.")
|
|
token = secrets.token_urlsafe(32)
|
|
state_dir = getattr(args, "state_dir", None)
|
|
app = module.create_app(
|
|
workspace_root=Path(args.workspace_root).expanduser().resolve(),
|
|
token=token,
|
|
run_state_root=Path(state_dir) / "release-console" if state_dir is not None else None,
|
|
)
|
|
return app, token
|
|
|
|
|
|
def _error_detail(payload: Any) -> str:
|
|
detail = payload.get("detail") if isinstance(payload, dict) else None
|
|
if isinstance(detail, str):
|
|
return detail
|
|
if isinstance(detail, list):
|
|
# Validation input may contain signing-key arguments. Never echo it.
|
|
return "; ".join(
|
|
".".join(str(part) for part in item.get("loc", [])) + ": " + str(item.get("msg", "Invalid request"))
|
|
for item in detail if isinstance(item, dict)
|
|
)
|
|
return "The release service did not return a valid successful response."
|
|
|
|
|
|
class _ServiceError(Exception):
|
|
def __init__(self, status: int, payload: Any):
|
|
super().__init__(_error_detail(payload))
|
|
self.status = status
|
|
|
|
|
|
def _brief(value: Any, *, limit: int = 240) -> str:
|
|
"""Bound display-only fields; never serialize an executor or its arguments."""
|
|
if not isinstance(value, (str, int, float, bool)):
|
|
return "unknown"
|
|
text = " ".join(redact(str(value)).split())
|
|
return text if len(text) <= limit else text[:limit - 1] + "…"
|
|
|
|
|
|
def _status(payload: dict[str, Any]) -> str:
|
|
status = payload.get("status")
|
|
for key in ("summary", "state"):
|
|
if isinstance(payload.get(key), dict):
|
|
status = payload[key].get("status", status)
|
|
if isinstance(payload.get("state_step"), dict):
|
|
status = payload["state_step"].get("state", status)
|
|
# Return the semantic value unchanged: display redaction must never affect
|
|
# failure exit codes, even when an environment secret happens to equal it.
|
|
return status if isinstance(status, str) else "ok"
|
|
|
|
|
|
def _summary_lines(name: str, payload: dict[str, Any], note: str | None = None) -> list[str]:
|
|
"""Compact, bounded projection; the unchanged JSON result retains details."""
|
|
lines = [f"Release {name}: {_brief(_status(payload))}."]
|
|
if note:
|
|
lines.append(note)
|
|
plan = payload.get("immutable", {}).get("plan", {}) if isinstance(payload.get("immutable"), dict) else payload
|
|
if not isinstance(plan, dict):
|
|
plan = {}
|
|
if isinstance(plan.get("source_preflight_ready"), bool):
|
|
prefix = "Frozen plan source" if "immutable" in payload else "Source"
|
|
lines.append(f"{prefix} preflight ready: {str(plan['source_preflight_ready']).lower()}.")
|
|
units = plan.get("units", [])
|
|
if isinstance(units, list):
|
|
for unit in units[:12]:
|
|
if isinstance(unit, dict):
|
|
lines.append(f"{_brief(unit.get('repo'))}: {_brief(unit.get('status', 'planned'))}; target {_brief(unit.get('target_version'))}.")
|
|
if len(units) > 12:
|
|
lines.append(f"{len(units) - 12} more selected repositories; use --json for every repository.")
|
|
dashboard = payload.get("summary")
|
|
if isinstance(dashboard, dict):
|
|
counts = [f"{dashboard[key]} {label}" for key, label in (
|
|
("repository_count", "repositories"), ("missing_count", "missing"),
|
|
("dirty_count", "dirty"), ("ahead_count", "ahead"),
|
|
("behind_count", "behind"), ("error_count", "errors"),
|
|
) if isinstance(dashboard.get(key), int)]
|
|
if counts:
|
|
lines.append("Repository status: " + ", ".join(counts) + ".")
|
|
findings = plan.get("gate_findings", payload.get("collection_errors", []))
|
|
if isinstance(findings, list):
|
|
for finding in findings[:4]:
|
|
if isinstance(finding, dict):
|
|
scope = f" ({_brief(finding['repo'])})" if finding.get("repo") else ""
|
|
lines.append(f"Gate {_brief(finding.get('code'))}{scope}: {_brief(finding.get('message'))}")
|
|
if len(findings) > 4:
|
|
lines.append(f"{len(findings) - 4} more gate findings; use --json for details.")
|
|
state = payload.get("state", {})
|
|
steps = state.get("steps", []) if isinstance(state, dict) else []
|
|
if isinstance(payload.get("state_step"), dict):
|
|
steps = [payload["state_step"]]
|
|
if isinstance(steps, list) and steps:
|
|
counts: dict[str, int] = {}
|
|
for step in steps:
|
|
if isinstance(step, dict):
|
|
status = _brief(step.get("state", "unknown"))
|
|
counts[status] = counts.get(status, 0) + 1
|
|
lines.append("Steps: " + ", ".join(f"{count} {state}" for state, count in sorted(counts.items())) + ".")
|
|
relevant = [step for step in steps if isinstance(step, dict) and step.get("state") != "succeeded"]
|
|
for step in relevant[:3]:
|
|
lines.append(f"Step {_brief(step.get('id'))}: {_brief(step.get('state'))}." +
|
|
(f" {_brief(step['disabled_reason'])}" if step.get("disabled_reason") else ""))
|
|
execution = payload.get("execution_result")
|
|
if isinstance(execution, dict):
|
|
lines.append(f"Executor result: {_brief(execution.get('status', 'recorded'))}.")
|
|
recommendation = payload.get("recommended_next", plan.get("recommended_action"))
|
|
if isinstance(recommendation, dict) and recommendation:
|
|
step = f" [{_brief(recommendation['step_id'])}]" if recommendation.get("step_id") else ""
|
|
lines.append(f"Next: {_brief(recommendation.get('id'))}{step} — {_brief(recommendation.get('title'))}.")
|
|
if recommendation.get("remediation"):
|
|
lines.append(_brief(recommendation["remediation"]))
|
|
runs = payload.get("runs")
|
|
if isinstance(runs, list):
|
|
lines.append(f"{len(runs)} release runs in this page.")
|
|
for run in runs[:12]:
|
|
if isinstance(run, dict):
|
|
lines.append(f"{_brief(run.get('run_id'))}: {_brief(_status(run))}.")
|
|
if payload.get("next_cursor"):
|
|
lines.append("More history available; use --json for the next cursor.")
|
|
return lines
|
|
|
|
|
|
async def _run(args: argparse.Namespace) -> dict[str, Any]:
|
|
# Resolve CLI-only validation before loading the service or creating state.
|
|
if getattr(args, "project", None) is not None:
|
|
raise ValueError("Release uses the authoritative GovOPlaN catalog and does not accept --project overrides; select the registered --workspace-root instead.")
|
|
name = args.release_command
|
|
selection = _selection(args) if name in {"plan", "create"} else None
|
|
keys = getattr(args, "signing_key", [])
|
|
if len(keys) > 8 or any(not re.fullmatch(r"[A-Za-z0-9._-]{1,128}=.+", key) or "-----BEGIN" in key or "\n" in key or "\r" in key or len(key) > 4096 for key in keys):
|
|
raise ValueError("Provide at most eight --signing-key KEY_ID=PRIVATE_KEY_FILE arguments, never key material.")
|
|
import httpx
|
|
app, token = _load_application(args)
|
|
metadata = {
|
|
"workspace_root": str(app.state.workspace_root),
|
|
"state_location": str(app.state.release_runs.root),
|
|
"candidate_location": str(app.state.release_candidate_root),
|
|
}
|
|
transport = httpx.ASGITransport(app=app, raise_app_exceptions=False)
|
|
async with httpx.AsyncClient(
|
|
transport=transport, base_url="http://govoplan-devkit.invalid",
|
|
headers={"X-Release-Console-Token": token}, timeout=None,
|
|
follow_redirects=False,
|
|
) as client:
|
|
async def request(method: str, path: str, **kwargs: Any) -> dict[str, Any]:
|
|
response = await client.request(method, path, **kwargs)
|
|
try:
|
|
payload = response.json()
|
|
except ValueError:
|
|
payload = None
|
|
if response.status_code >= 400 or not isinstance(payload, dict):
|
|
raise _ServiceError(response.status_code, payload)
|
|
return payload
|
|
|
|
def result(payload: dict[str, Any], *, dry_run: bool = False, summary: str | None = None) -> dict[str, Any]:
|
|
status = _status(payload)
|
|
execution_failed = isinstance(payload.get("execution_result"), dict) and payload["execution_result"].get("status") == "failed"
|
|
return {
|
|
**metadata, "operation": name, "dry_run": dry_run,
|
|
"result": payload,
|
|
"_exit_code": 1 if execution_failed or status in {"blocked", "failed", "interrupted"} else 0,
|
|
"summary": [*_summary_lines(name, payload, summary), f"Durable state: {metadata['state_location']}"],
|
|
}
|
|
|
|
async def preview(run: dict[str, Any], step_id: str) -> dict[str, Any]:
|
|
plan = run.get("immutable", {}).get("plan", {})
|
|
plan_step = next((step for step in plan.get("dry_run_steps", []) if step.get("id") == step_id), None)
|
|
state_step = next((step for step in run.get("state", {}).get("steps", []) if step.get("id") == step_id), None)
|
|
if plan_step is None or state_step is None:
|
|
raise _ServiceError(404, {"detail": "Release run step was not found."})
|
|
if state_step.get("executor", {}).get("kind") == "catalog_publish":
|
|
return await request("POST", f"{run_path}/steps/{quote(step_id, safe='')}/preview", json={"remote": "origin"})
|
|
return {
|
|
"run_id": run["run_id"], "plan_step": plan_step, "state_step": state_step,
|
|
"note": "Frozen-plan inspection only; no executor was called and no live preflight is claimed.",
|
|
}
|
|
|
|
if name == "status":
|
|
payload = await request("GET", "/api/dashboard", params={**_planning_query(args), "include_website": args.include_website})
|
|
return result(payload)
|
|
if name == "list":
|
|
params = {"limit": args.limit, **({"cursor": args.cursor} if args.cursor else {})}
|
|
return result(await request("GET", "/api/release-runs", params=params))
|
|
if name in {"plan", "create"}:
|
|
assert selection is not None
|
|
repos, versions = selection
|
|
query = {
|
|
**_planning_query(args), "repos": ",".join(repos),
|
|
"repo_versions": ",".join(f"{repo}={version}" for repo, version in versions.items()),
|
|
}
|
|
if name == "plan" or not args.apply:
|
|
payload = await request("GET", "/api/selective-plan", params=query)
|
|
return result(payload, dry_run=True, summary="Release plan inspected; no run was created and no release step executed.")
|
|
body = {key: value for key, value in _planning_query(args).items() if key != "target_version"}
|
|
payload = await request("POST", "/api/release-runs", json={**body, "request_id": args.request_id, "repo_versions": versions})
|
|
return result(payload)
|
|
|
|
run_path = "/api/release-runs/" + quote(args.run_id, safe="")
|
|
if name == "show":
|
|
return result(await request("GET", run_path))
|
|
if name == "preview" or not args.apply:
|
|
run = await request("GET", run_path)
|
|
payload = await preview(run, args.step_id) if name in {"preview", "execute"} else run
|
|
return result(payload, dry_run=True, summary=f"Release {name} inspected; no durable transition or executor was invoked.")
|
|
body = {"request_id": args.request_id}
|
|
if name in {"execute", "reconcile"}:
|
|
body["confirm"] = args.confirm
|
|
if name == "execute":
|
|
body.update({"remote": "origin", "signing_keys": keys})
|
|
if name == "reconcile":
|
|
body["outcome"] = args.outcome
|
|
path = f"{run_path}/resume" if name == "resume" else f"{run_path}/steps/{quote(args.step_id, safe='')}/{name}"
|
|
return result(await request("POST", path, json=body))
|
|
|
|
|
|
def handle(args: argparse.Namespace) -> dict[str, Any]:
|
|
"""Return the common devkit JSON/summary envelope; never retry mutations."""
|
|
try:
|
|
return asyncio.run(_run(args))
|
|
except _ServiceError as exc:
|
|
summary = [f"Release {args.release_command} failed (HTTP {exc.status}): {exc}"]
|
|
if args.release_command in {"create", "execute", "resume", "retry", "reconcile"}:
|
|
summary.append("No automatic retry occurred. Inspect the run; reuse the same request ID for a known replay, or resume/reconcile an uncertain effect before a new attempt.")
|
|
return {"_exit_code": 1, "status": "error", "http_status": exc.status, "summary": summary}
|
|
except ModuleNotFoundError as exc:
|
|
return {"_exit_code": 2, "status": "unavailable", "summary": [f"Release commands need the GovOPlaN development dependencies ({exc.name} is unavailable)."]}
|
|
except ValueError as exc:
|
|
return {"_exit_code": 2, "status": "invalid", "summary": [str(exc)]}
|