diff --git a/openadapt_flow/__main__.py b/openadapt_flow/__main__.py index 8858d22a..49daee28 100644 --- a/openadapt_flow/__main__.py +++ b/openadapt_flow/__main__.py @@ -7705,6 +7705,10 @@ def _repair_store_flag(rp: argparse.ArgumentParser) -> None: ) pr.set_defaults(func=_cmd_connector) + from openadapt_flow.cli_admit import register_admit_parser + + register_admit_parser(sub) + return parser diff --git a/openadapt_flow/cli_admit.py b/openadapt_flow/cli_admit.py new file mode 100644 index 00000000..504c9b37 --- /dev/null +++ b/openadapt_flow/cli_admit.py @@ -0,0 +1,70 @@ +"""CLI: ``openadapt-flow admit status``.""" + +from __future__ import annotations + +import argparse +import json +import sys + +from openadapt_flow.release_admission import ( + ADMITTED, + DEFAULT_LEDGER_URL, + LedgerError, + load_ledger, + render_status, + status_report, +) + + +def register_admit_parser(sub: argparse._SubParsersAction) -> None: + parser = sub.add_parser( + "admit", + help=( + "Read the published release-admission ledger. " + "Does not sign or issue an admission." + ), + ) + verbs = parser.add_subparsers(dest="admit_cmd", required=True) + status = verbs.add_parser( + "status", + help=( + "Print admitted vs not-admitted for Flow from the public ledger. " + "Never mints a signature." + ), + ) + status.add_argument( + "--ledger", + default=DEFAULT_LEDGER_URL, + help=( + "Public production-lifecycle JSON (URL or path). " + f"Default: {DEFAULT_LEDGER_URL}" + ), + ) + status.add_argument( + "--json", + action="store_true", + dest="as_json", + help="Print the status report as JSON", + ) + status.add_argument( + "--check", + action="store_true", + help="Exit 1 unless Flow currently holds a live admission", + ) + status.set_defaults(func=cmd_admit_status) + + +def cmd_admit_status(args: argparse.Namespace) -> int: + try: + ledger = load_ledger(args.ledger) + report = status_report(ledger) + except LedgerError as exc: + print(f"admit status: {exc}", file=sys.stderr) + return 2 + if args.as_json: + print(json.dumps(report, indent=2, sort_keys=True)) + else: + sys.stdout.write(render_status(report, ledger_source=str(args.ledger))) + if args.check and report["flow"]["state"] != ADMITTED: + return 1 + return 0 diff --git a/openadapt_flow/release_admission.py b/openadapt_flow/release_admission.py new file mode 100644 index 00000000..a406bdff --- /dev/null +++ b/openadapt_flow/release_admission.py @@ -0,0 +1,223 @@ +"""Read the published release-admission ledger. Never mint an admission. + +``openadapt-flow admit status`` answers one question from the public +projection: does this target currently hold a live, non-revoked admission? +An empty, expired, or revoked row is ``not_actively_admitted``. A fetch or +parse failure is also that state. The command does not sign, issue, or +rewrite the ledger. +""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone +from pathlib import Path +from typing import Any +from urllib.error import URLError +from urllib.request import Request, urlopen + +SCHEMA_VERSION = "openadapt.release-admission-status/v1" +DEFAULT_LEDGER_URL = "https://openadapt.ai/production-lifecycle.json" +NOT_ADMITTED = "not_actively_admitted" +ADMITTED = "admitted" +FLOW_TARGET = "flow" +EXPECTED_TARGETS = ( + "agent", + "capture", + "cloud", + "desktop", + "docs", + "flow", + "openadapt", +) +PACK_FIELDS = ( + "wheel_digest", + "substrate_matrix", + "conformance_suite", + "known_fails", + "soak_hours", + "rollback", + "expires_at", + "revocation_path", +) + + +class LedgerError(ValueError): + """The ledger could not be read or did not match the public schema.""" + + +def _now(now: datetime | None) -> datetime: + if now is None: + return datetime.now(timezone.utc) + if now.tzinfo is None: + return now.replace(tzinfo=timezone.utc) + return now.astimezone(timezone.utc) + + +def _parse_time(value: Any) -> datetime | None: + if not isinstance(value, str) or not value: + return None + text = value.replace("Z", "+00:00") + try: + parsed = datetime.fromisoformat(text) + except ValueError: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + +def load_ledger(source: str | Path) -> dict[str, Any]: + """Load a public production-lifecycle projection from a path or URL.""" + + text = str(source) + if Path(text).exists(): + raw = Path(text).read_text(encoding="utf-8") + elif "://" in text: + request = Request(text, headers={"User-Agent": "openadapt-flow-admit-status"}) + try: + with urlopen(request, timeout=15) as response: + raw = response.read().decode("utf-8") + except (URLError, TimeoutError, OSError) as exc: + raise LedgerError(f"failed to fetch ledger {text}: {exc}") from exc + else: + raise LedgerError(f"ledger not found: {text}") + try: + payload = json.loads(raw) + except json.JSONDecodeError as exc: + raise LedgerError(f"ledger is not JSON: {exc}") from exc + if not isinstance(payload, dict): + raise LedgerError("ledger root must be an object") + targets = payload.get("targets") + if not isinstance(targets, list) or not targets: + raise LedgerError("ledger has no targets list") + return payload + + +def _target_map(ledger: dict[str, Any]) -> dict[str, dict[str, Any]]: + out: dict[str, dict[str, Any]] = {} + for row in ledger["targets"]: + if not isinstance(row, dict): + continue + target_id = row.get("id") + if isinstance(target_id, str) and target_id: + out[target_id] = row + return out + + +def classify_admission( + admission: Any, *, now: datetime | None = None +) -> tuple[str, str]: + """Return (state, reason) for one latest_admission value.""" + + clock = _now(now) + if admission is None: + return NOT_ADMITTED, "no live, non-revoked admission in the published ledger" + if not isinstance(admission, dict) or not admission: + return NOT_ADMITTED, "latest_admission is not a signed admission object" + if admission.get("revoked_at"): + return NOT_ADMITTED, "latest admission is revoked" + expires = _parse_time(admission.get("expires_at")) + if expires is None: + return NOT_ADMITTED, "latest admission has no usable expires_at" + if expires <= clock: + return NOT_ADMITTED, "latest admission is expired" + admission_id = admission.get("admission_id") + if not isinstance(admission_id, str) or not admission_id: + return NOT_ADMITTED, "latest admission has no admission_id" + return ADMITTED, f"live admission {admission_id}" + + +def target_status( + ledger: dict[str, Any], target_id: str, *, now: datetime | None = None +) -> dict[str, Any]: + rows = _target_map(ledger) + row = rows.get(target_id) + if row is None: + return { + "id": target_id, + "state": NOT_ADMITTED, + "latest_admission": None, + "reason": f"target {target_id!r} is missing from the ledger", + } + state, reason = classify_admission(row.get("latest_admission"), now=now) + latest = row.get("latest_admission") + return { + "id": target_id, + "display_name": row.get("display_name"), + "state": state, + "latest_admission": latest if isinstance(latest, dict) else None, + "reason": reason, + } + + +def pack_status(flow_row: dict[str, Any]) -> dict[str, Any]: + """Describe the Flow admission pack. Unsigned rows are incomplete.""" + + latest = flow_row.get("latest_admission") + present: list[str] = [] + if isinstance(latest, dict): + artifacts = latest.get("release", {}).get("artifacts") + if isinstance(artifacts, list): + for artifact in artifacts: + if ( + isinstance(artifact, dict) + and artifact.get("kind") == "wheel" + and artifact.get("sha256") + ): + present.append("wheel_digest") + break + if latest.get("expires_at"): + present.append("expires_at") + if latest.get("admission_id"): + present.append("revocation_path") + missing = [field for field in PACK_FIELDS if field not in present] + return { + "complete": not missing, + "present": present, + "missing": missing, + "note": ( + "This CLI reports the published pack. It does not mint a digest, " + "a soak, or a signature." + ), + } + + +def status_report( + ledger: dict[str, Any], *, now: datetime | None = None +) -> dict[str, Any]: + rows = _target_map(ledger) + targets = [ + target_status(ledger, target_id, now=now) for target_id in EXPECTED_TARGETS + ] + flow = next(item for item in targets if item["id"] == FLOW_TARGET) + product_wide = all(item["state"] == ADMITTED for item in targets) and len( + rows + ) >= len(EXPECTED_TARGETS) + return { + "schema_version": SCHEMA_VERSION, + "product_wide_production": product_wide, + "flow": flow, + "targets": targets, + "pack": pack_status(rows.get(FLOW_TARGET, {})), + "derivation": ledger.get("derivation"), + } + + +def render_status(report: dict[str, Any], *, ledger_source: str) -> str: + flow = report["flow"] + product = "yes" if report["product_wide_production"] else "no" + lines = [ + "OpenAdapt release admission", + f"ledger: {ledger_source}", + f"product-wide Production: {product}", + f"flow: {flow['state']}", + f" latest_admission: {flow['latest_admission'] and flow['latest_admission'].get('admission_id') or 'none'}", + f" reason: {flow['reason']}", + ] + pack = report["pack"] + lines.append("pack: complete" if pack["complete"] else "pack: incomplete") + if pack["missing"]: + lines.append(" missing: " + ", ".join(pack["missing"])) + lines.append(" " + pack["note"]) + return "\n".join(lines) + "\n" diff --git a/tests/test_admit_status.py b/tests/test_admit_status.py new file mode 100644 index 00000000..df11b644 --- /dev/null +++ b/tests/test_admit_status.py @@ -0,0 +1,131 @@ +"""``openadapt-flow admit status`` reads the ledger. It never mints one.""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone +from pathlib import Path + +from openadapt_flow.__main__ import main +from openadapt_flow.release_admission import ( + ADMITTED, + EXPECTED_TARGETS, + NOT_ADMITTED, + classify_admission, + load_ledger, + status_report, +) + +NOW = datetime(2026, 8, 30, 12, 0, tzinfo=timezone.utc) + + +def _ledger(admissions: dict[str, dict | None]) -> dict: + targets = [] + for target_id in EXPECTED_TARGETS: + targets.append( + { + "id": target_id, + "display_name": target_id, + "admission_history": [], + "latest_admission": admissions.get(target_id), + } + ) + return { + "schema_version": "openadapt.public-production-lifecycle/v1", + "derivation": { + "mode": "latest_signed_admission_at_read_time", + "static_production_state": False, + }, + "targets": targets, + } + + +def _write(tmp_path: Path, payload: dict) -> Path: + path = tmp_path / "production-lifecycle.json" + path.write_text(json.dumps(payload), encoding="utf-8") + return path + + +def test_empty_ledger_reports_flow_not_admitted(tmp_path: Path) -> None: + path = _write(tmp_path, _ledger({})) + report = status_report(load_ledger(path), now=NOW) + assert report["flow"]["state"] == NOT_ADMITTED + assert report["flow"]["latest_admission"] is None + assert "no live, non-revoked admission" in report["flow"]["reason"] + assert report["product_wide_production"] is False + assert report["pack"]["complete"] is False + assert "wheel_digest" in report["pack"]["missing"] + for row in report["targets"]: + assert row["state"] == NOT_ADMITTED + + +def test_expired_and_revoked_admissions_are_not_admitted() -> None: + expired = { + "admission_id": "production:flow:1", + "expires_at": "2026-08-01T00:00:00Z", + "revoked_at": None, + } + revoked = { + "admission_id": "production:flow:2", + "expires_at": "2026-12-01T00:00:00Z", + "revoked_at": "2026-08-15T00:00:00Z", + } + state, reason = classify_admission(expired, now=NOW) + assert state == NOT_ADMITTED + assert "expired" in reason + state, reason = classify_admission(revoked, now=NOW) + assert state == NOT_ADMITTED + assert "revoked" in reason + state, reason = classify_admission(None, now=NOW) + assert state == NOT_ADMITTED + + +def test_live_admission_is_reported_not_minted() -> None: + live = { + "admission_id": "production:flow:9", + "expires_at": "2026-09-30T00:00:00Z", + "revoked_at": None, + "release": { + "artifacts": [ + {"kind": "wheel", "sha256": "sha256:" + "ab" * 32}, + ] + }, + } + state, reason = classify_admission(live, now=NOW) + assert state == ADMITTED + assert "production:flow:9" in reason + payload = _ledger({target: live for target in EXPECTED_TARGETS}) + report = status_report(payload, now=NOW) + assert report["product_wide_production"] is True + assert report["flow"]["state"] == ADMITTED + assert "wheel_digest" in report["pack"]["present"] + assert report["pack"]["note"].startswith("This CLI reports") + + +def test_cli_status_on_empty_ledger_exits_1_with_check(tmp_path: Path, capsys) -> None: + path = _write(tmp_path, _ledger({})) + code = main(["admit", "status", "--ledger", str(path), "--check"]) + captured = capsys.readouterr() + assert code == 1 + assert "not_actively_admitted" in captured.out + assert "product-wide Production: no" in captured.out + assert "does not mint" in captured.out + + +def test_cli_status_json_does_not_invent_an_admission(tmp_path: Path, capsys) -> None: + path = _write(tmp_path, _ledger({})) + code = main(["admit", "status", "--ledger", str(path), "--json"]) + captured = capsys.readouterr() + assert code == 0 + payload = json.loads(captured.out) + assert payload["flow"]["state"] == NOT_ADMITTED + assert payload["flow"]["latest_admission"] is None + assert payload["product_wide_production"] is False + + +def test_missing_ledger_fails_closed(capsys) -> None: + code = main(["admit", "status", "--ledger", "/no/such/production-lifecycle.json"]) + captured = capsys.readouterr() + assert code == 2 + assert "admit status:" in captured.err + assert captured.out == ""