From 411125597d36a9d5356db3265d56a3236e71f4e5 Mon Sep 17 00:00:00 2001 From: Chris Busillo Date: Mon, 8 Jun 2026 17:26:58 -0400 Subject: [PATCH] Add runner maintainer desired-state planner --- control_plane/cli_runner_lanes.py | 144 +++++++ .../contracts/runner_lane_maintainer.py | 405 ++++++++++++++++++ docs/runner-lane-baseline.md | 42 ++ tests/test_runner_lane_maintainer.py | 343 +++++++++++++++ 4 files changed, 934 insertions(+) create mode 100644 control_plane/contracts/runner_lane_maintainer.py create mode 100644 tests/test_runner_lane_maintainer.py diff --git a/control_plane/cli_runner_lanes.py b/control_plane/cli_runner_lanes.py index 4dfd9d81..f8337109 100644 --- a/control_plane/cli_runner_lanes.py +++ b/control_plane/cli_runner_lanes.py @@ -20,6 +20,9 @@ from control_plane.contracts.runner_lane_control import RunnerLaneControlRequest from control_plane.contracts.runner_lane_control import plan_runner_lane_control from control_plane.contracts.runner_lane_inventory import RunnerLaneInventory +from control_plane.contracts.runner_lane_maintainer import RunnerLaneDesiredState +from control_plane.contracts.runner_lane_maintainer import RunnerLaneMaintainerPolicy +from control_plane.contracts.runner_lane_maintainer import plan_runner_lane_maintainer from control_plane.contracts.runner_lane_registration import RunnerLaneRegistrationPolicy from control_plane.contracts.runner_host_hygiene import RunnerHostHygieneObservation from control_plane.contracts.runner_host_hygiene import RunnerHostHygieneApplyAction @@ -81,6 +84,7 @@ def register_runner_lane_commands(work_graph: click.Group) -> None: work_graph.add_command(runner_queue_wait, name="runner-queue-wait") work_graph.add_command(runner_baseline_observe, name="runner-baseline-observe") work_graph.add_command(runner_control_plan, name="runner-control-plan") + work_graph.add_command(runner_maintainer_plan, name="runner-maintainer-plan") work_graph.add_command( runner_lane_registration_executor, name="runner-lane-registration-executor", @@ -493,6 +497,146 @@ def runner_control_plan( ) +@click.command("runner-maintainer-plan") +@click.option("--repository", required=True, help="owner/name repository for the runner lane.") +@click.option("--host-name", required=True, help="Approved runner host name.") +@click.option("--lane-name", required=True, help="Desired runner lane name.") +@click.option( + "--runner-directory", + required=True, + help="Desired absolute lane directory below an allowed registration root.", +) +@click.option("--service-user", required=True, help="Constrained runner service user.") +@click.option( + "--systemd-unit-name", + required=True, + help="Desired systemd unit name, for example launchplane-runner@lane.service.", +) +@click.option("--label", "labels", multiple=True, required=True, help="Desired runner label.") +@click.option( + "--runner-version", + default="latest-approved", + show_default=True, + help="Desired runner version policy label. This command does not download runners.", +) +@click.option( + "--allowed-repository", + "allowed_repositories", + multiple=True, + help="Repository opted into runner lane maintenance. Repeat as needed.", +) +@click.option( + "--approved-host", + "approved_hosts", + multiple=True, + help="Host approved for runner lane maintenance. Repeat as needed.", +) +@click.option( + "--allowed-registration-root", + "allowed_registration_roots", + multiple=True, + help="Absolute root allowed for runner registration. Repeat as needed.", +) +@click.option( + "--allowed-service-user", + "allowed_service_users", + multiple=True, + help="Service user approved for runner lane maintenance. Repeat as needed.", +) +@click.option( + "--required-label", + "required_labels", + multiple=True, + help="Label required by maintainer policy. Defaults to self-hosted, launchplane, launchplane-managed.", +) +@click.option( + "--required-managed-label", + default="launchplane-managed", + show_default=True, + help="Label required on existing lanes before they can be adopted or removed.", +) +@click.option( + "--require-baseline-readiness/--allow-missing-baseline-readiness", + default=True, + show_default=True, + help="Whether baseline readiness must pass before maintainer planning is ready.", +) +@click.option( + "--inventory-file", + type=click.Path(path_type=Path, exists=True, dir_okay=False), + required=True, + help="RunnerLaneInventory JSON from runner-inventory or an equivalent fixture.", +) +@click.option( + "--baseline-readiness-file", + type=click.Path(path_type=Path, exists=True, dir_okay=False), + required=True, + help="RunnerLaneBaselineReadiness JSON or runner-baseline-observe output.", +) +def runner_maintainer_plan( + repository: str, + host_name: str, + lane_name: str, + runner_directory: str, + service_user: str, + systemd_unit_name: str, + labels: tuple[str, ...], + runner_version: str, + allowed_repositories: tuple[str, ...], + approved_hosts: tuple[str, ...], + allowed_registration_roots: tuple[str, ...], + allowed_service_users: tuple[str, ...], + required_labels: tuple[str, ...], + required_managed_label: str, + require_baseline_readiness: bool, + inventory_file: Path, + baseline_readiness_file: Path, +) -> None: + try: + policy = RunnerLaneMaintainerPolicy( + allowed_repositories=allowed_repositories, + approved_hosts=approved_hosts, + allowed_registration_roots=allowed_registration_roots, + allowed_service_users=allowed_service_users, + required_labels=( + required_labels or ("self-hosted", "launchplane", "launchplane-managed") + ), + required_managed_label=required_managed_label, + require_baseline_readiness=require_baseline_readiness, + ) + desired_state = RunnerLaneDesiredState( + repository=repository, + host_name=host_name, + lane_name=lane_name, + runner_directory=runner_directory, + service_user=service_user, + systemd_unit_name=systemd_unit_name, + labels=labels, + runner_version=runner_version, + ) + inventory = _load_runner_lane_inventory(inventory_file) + baseline_readiness = _load_runner_lane_baseline_readiness(baseline_readiness_file) + plan = plan_runner_lane_maintainer( + policy=policy, + desired_state=desired_state, + inventory=inventory, + baseline_readiness=baseline_readiness, + ) + except (OSError, JSONDecodeError, ValidationError, ValueError) as error: + raise click.ClickException(str(error)) from error + click.echo( + json.dumps( + { + "policy": policy.model_dump(mode="json"), + "desired_state": desired_state.model_dump(mode="json"), + "plan": plan.model_dump(mode="json"), + }, + indent=2, + sort_keys=True, + ) + ) + + @click.command("runner-lane-registration-executor") @click.option("--repository", required=True, help="owner/name repository for the runner lane.") @click.option("--host-name", required=True, help="Approved runner host name.") diff --git a/control_plane/contracts/runner_lane_maintainer.py b/control_plane/contracts/runner_lane_maintainer.py new file mode 100644 index 00000000..3d7aebe3 --- /dev/null +++ b/control_plane/contracts/runner_lane_maintainer.py @@ -0,0 +1,405 @@ +from __future__ import annotations + +import re +from typing import Literal + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +from control_plane.contracts.runner_lane_baseline import RunnerLaneBaselineReadiness +from control_plane.contracts.runner_lane_inventory import RunnerLaneInventory +from control_plane.contracts.runner_lane_inventory import RunnerLaneRecord + + +RunnerLaneMaintainerPlanStatus = Literal["ready", "blocked"] +RunnerLaneMaintainerDecision = Literal[ + "recommend_create", + "recommend_verify_adoption", + "recommend_remove_recreate", + "blocked", +] +RunnerLaneMaintainerBlockerKind = Literal["policy", "capability"] +RunnerLaneMaintainerBlockerCode = Literal[ + "approved_host_missing", + "baseline_not_ready", + "desired_label_missing", + "existing_lane_not_managed", + "host_not_allowed", + "inventory_repository_mismatch", + "lane_name_ambiguous", + "repository_not_allowed", + "runner_directory_not_allowed", + "service_user_not_allowed", + "supervised_maintainer_required", +] + + +class RunnerLaneMaintainerPolicy(BaseModel): + model_config = ConfigDict(extra="forbid") + + schema_version: int = Field(default=1, ge=1) + allowed_repositories: tuple[str, ...] = () + approved_hosts: tuple[str, ...] = () + allowed_registration_roots: tuple[str, ...] = () + allowed_service_users: tuple[str, ...] = () + required_labels: tuple[str, ...] = ( + "self-hosted", + "launchplane", + "launchplane-managed", + ) + required_managed_label: str = "launchplane-managed" + require_baseline_readiness: bool = True + + @model_validator(mode="after") + def _normalize_policy(self) -> "RunnerLaneMaintainerPolicy": + self.allowed_repositories = _normalized_repositories(self.allowed_repositories) + self.approved_hosts = _normalized_identifiers(self.approved_hosts, "approved host") + self.allowed_registration_roots = tuple( + sorted( + {_normalized_path(root) for root in self.allowed_registration_roots if root.strip()} + ) + ) + self.allowed_service_users = _normalized_identifiers( + self.allowed_service_users, "service user" + ) + self.required_labels = _normalized_identifiers(self.required_labels, "label") + self.required_managed_label = _normalized_identifier( + self.required_managed_label, "managed label" + ) + if not self.required_labels: + raise ValueError("runner lane maintainer policy requires labels") + return self + + +class RunnerLaneDesiredState(BaseModel): + model_config = ConfigDict(extra="forbid") + + schema_version: int = Field(default=1, ge=1) + repository: str + host_name: str + lane_name: str + runner_directory: str + service_user: str + systemd_unit_name: str + labels: tuple[str, ...] + runner_version: str = "latest-approved" + + @model_validator(mode="after") + def _normalize_state(self) -> "RunnerLaneDesiredState": + self.repository = _normalized_repository(self.repository) + self.host_name = _normalized_identifier(self.host_name, "host_name") + self.lane_name = _normalized_identifier(self.lane_name, "lane_name") + self.runner_directory = _normalized_path(self.runner_directory) + self.service_user = _normalized_identifier(self.service_user, "service_user") + self.systemd_unit_name = _normalized_systemd_unit_name(self.systemd_unit_name) + self.labels = _normalized_identifiers(self.labels, "label") + self.runner_version = _required_text( + self.runner_version, "runner lane maintainer desired state requires runner_version" + ) + if not self.labels: + raise ValueError("runner lane maintainer desired state requires labels") + if self.runner_directory.rsplit("/", 1)[-1] != self.lane_name: + raise ValueError("runner lane maintainer runner_directory must be the lane directory") + expected_unit_name = f"launchplane-runner@{self.lane_name}.service" + if self.systemd_unit_name != expected_unit_name: + raise ValueError("runner lane maintainer systemd_unit_name must match the lane name") + return self + + +class RunnerLaneMaintainerBlocker(BaseModel): + model_config = ConfigDict(extra="forbid") + + code: RunnerLaneMaintainerBlockerCode + message: str + kind: RunnerLaneMaintainerBlockerKind = "policy" + + @model_validator(mode="after") + def _normalize_blocker(self) -> "RunnerLaneMaintainerBlocker": + self.message = _required_text( + self.message, "runner lane maintainer blocker requires message" + ) + return self + + +class RunnerLaneMaintainerPlan(BaseModel): + model_config = ConfigDict(extra="forbid") + + schema_version: int = Field(default=1, ge=1) + status: RunnerLaneMaintainerPlanStatus + decision: RunnerLaneMaintainerDecision + policy_ready: bool + capability_ready: bool + desired_state: RunnerLaneDesiredState + matching_lanes: tuple[RunnerLaneRecord, ...] = () + blockers: tuple[RunnerLaneMaintainerBlocker, ...] = () + next_steps: tuple[str, ...] = () + summary: str + + @model_validator(mode="after") + def _normalize_plan(self) -> "RunnerLaneMaintainerPlan": + self.matching_lanes = tuple( + sorted(self.matching_lanes, key=lambda lane: (lane.name.lower(), lane.github_id)) + ) + self.blockers = tuple(sorted(self.blockers, key=lambda blocker: blocker.code)) + self.next_steps = tuple(step.strip() for step in self.next_steps if step.strip()) + self.summary = _required_text(self.summary, "runner lane maintainer plan requires summary") + if self.status == "ready" and self.blockers: + raise ValueError("ready runner lane maintainer plan cannot include blockers") + if self.status == "blocked" and not self.blockers: + raise ValueError("blocked runner lane maintainer plan requires blockers") + return self + + +def plan_runner_lane_maintainer( + *, + policy: RunnerLaneMaintainerPolicy, + desired_state: RunnerLaneDesiredState, + inventory: RunnerLaneInventory, + baseline_readiness: RunnerLaneBaselineReadiness, +) -> RunnerLaneMaintainerPlan: + blockers: list[RunnerLaneMaintainerBlocker] = [] + inventory_repository = _normalized_repository(inventory.repository) + if ( + not policy.allowed_repositories + or desired_state.repository not in policy.allowed_repositories + ): + blockers.append( + _blocker( + "repository_not_allowed", + f"repository is not opted into runner lane maintenance: {desired_state.repository}", + ) + ) + if inventory_repository != desired_state.repository: + blockers.append( + _blocker( + "inventory_repository_mismatch", + "runner lane inventory repository does not match desired state: " + f"{inventory_repository}", + ) + ) + if not policy.approved_hosts: + blockers.append( + _blocker( + "approved_host_missing", "runner lane maintainer policy requires approved hosts" + ) + ) + elif desired_state.host_name not in policy.approved_hosts: + blockers.append( + _blocker( + "host_not_allowed", + f"runner host is not approved for maintenance: {desired_state.host_name}", + ) + ) + if not _path_under_allowed_roots( + path=desired_state.runner_directory, + allowed_roots=policy.allowed_registration_roots, + ): + blockers.append( + _blocker( + "runner_directory_not_allowed", + "runner directory is not under an allowed registration root: " + f"{desired_state.runner_directory}", + ) + ) + if ( + not policy.allowed_service_users + or desired_state.service_user not in policy.allowed_service_users + ): + blockers.append( + _blocker( + "service_user_not_allowed", + f"runner service user is not approved: {desired_state.service_user}", + ) + ) + for label in policy.required_labels: + if label not in desired_state.labels: + blockers.append( + _blocker( + "desired_label_missing", + f"runner desired state is missing required label: {label}", + ) + ) + if policy.require_baseline_readiness and not baseline_readiness.ready: + blockers.append( + _blocker( + "baseline_not_ready", + f"runner lane baseline is not ready: {baseline_readiness.summary}", + ) + ) + + matching_lanes = tuple( + lane for lane in inventory.lanes if lane.name.strip().lower() == desired_state.lane_name + ) + if len(matching_lanes) > 1: + blockers.append( + _blocker( + "lane_name_ambiguous", + f"runner lane name matches multiple inventory records: {desired_state.lane_name}", + ) + ) + if len(matching_lanes) == 1: + lane = matching_lanes[0] + lane_labels = set(_normalized_identifiers(lane.labels, "label")) + if policy.required_managed_label not in lane_labels: + blockers.append( + _blocker( + "existing_lane_not_managed", + "existing runner lane is not marked as Launchplane-managed: " + f"{desired_state.lane_name}", + ) + ) + + decision = _decision(blockers=tuple(blockers), matching_lanes=matching_lanes) + if decision != "blocked": + blockers.append( + _blocker( + "supervised_maintainer_required", + "runner lane maintainer host mutation requires the supervised maintainer capability", + kind="capability", + ) + ) + status: RunnerLaneMaintainerPlanStatus = "blocked" if blockers else "ready" + return RunnerLaneMaintainerPlan( + status=status, + decision=decision, + policy_ready=not any(blocker.kind == "policy" for blocker in blockers), + capability_ready=not any(blocker.kind == "capability" for blocker in blockers), + desired_state=desired_state, + matching_lanes=matching_lanes, + blockers=tuple(blockers), + next_steps=_next_steps(decision=decision), + summary=_summary(decision=decision, status=status), + ) + + +def _decision( + *, + blockers: tuple[RunnerLaneMaintainerBlocker, ...], + matching_lanes: tuple[RunnerLaneRecord, ...], +) -> RunnerLaneMaintainerDecision: + if blockers: + return "blocked" + if not matching_lanes: + return "recommend_create" + lane = matching_lanes[0] + if lane.status == "online": + return "recommend_verify_adoption" + return "recommend_remove_recreate" + + +def _next_steps(*, decision: RunnerLaneMaintainerDecision) -> tuple[str, ...]: + if decision == "blocked": + return ("resolve blockers before any runner maintainer host mutation",) + if decision == "recommend_create": + return ( + "future maintainer should prepare a supervised systemd-backed runner service under the approved root", + "future maintainer should request a short-lived GitHub registration token inside the maintainer executor", + "future maintainer should verify systemd active state, GitHub online inventory, and baseline readiness", + ) + if decision == "recommend_verify_adoption": + return ( + "future maintainer should verify the existing online runner has a matching Launchplane-owned systemd unit", + "future maintainer should verify service user, runner directory, labels, and baseline evidence match desired state", + ) + return ( + "future maintainer should remove the stale GitHub runner registration through an audited maintainer path", + "future maintainer should recreate the runner only through a supervised systemd-backed service", + "future maintainer should verify post-recreate inventory and baseline evidence before admitting product jobs", + ) + + +def _summary( + *, decision: RunnerLaneMaintainerDecision, status: RunnerLaneMaintainerPlanStatus +) -> str: + if status == "blocked": + return "runner lane maintainer plan is blocked" + if decision == "recommend_create": + return "runner lane maintainer should create a supervised runner service" + if decision == "recommend_verify_adoption": + return "runner lane maintainer should verify existing runner adoption evidence" + return "runner lane maintainer should remove and recreate the stale runner lane" + + +def _blocker( + code: RunnerLaneMaintainerBlockerCode, + message: str, + *, + kind: RunnerLaneMaintainerBlockerKind = "policy", +) -> RunnerLaneMaintainerBlocker: + return RunnerLaneMaintainerBlocker(code=code, message=message, kind=kind) + + +def _path_under_allowed_roots(*, path: str, allowed_roots: tuple[str, ...]) -> bool: + if not allowed_roots: + return False + normalized_path = _normalized_path(path) + for root in allowed_roots: + normalized_root = _normalized_path(root) + if normalized_path == normalized_root or normalized_path.startswith(f"{normalized_root}/"): + return True + return False + + +def _normalized_repositories(values: tuple[str, ...]) -> tuple[str, ...]: + return tuple( + sorted({normalized for value in values if (normalized := _normalized_repository(value))}) + ) + + +def _normalized_repository(value: str) -> str: + normalized_value = value.strip().lower() + if not normalized_value: + return "" + normalized_repository = "/".join(part.strip() for part in normalized_value.split("/")) + if normalized_repository.count("/") != 1: + raise ValueError("runner lane maintainer requires repository formatted as owner/name") + owner, name = normalized_repository.split("/", 1) + if not owner or not name: + raise ValueError("runner lane maintainer requires repository formatted as owner/name") + return normalized_repository + + +def _normalized_identifiers(values: tuple[str, ...], label: str) -> tuple[str, ...]: + return tuple( + sorted({_normalized_identifier(value, label) for value in values if value.strip()}) + ) + + +def _normalized_identifier(value: str, label: str) -> str: + normalized = value.strip().lower() + if not re.fullmatch(r"[a-z0-9][a-z0-9._-]{0,127}", normalized): + raise ValueError( + f"runner lane maintainer {label} must use letters, numbers, dots, underscores, or hyphens" + ) + return normalized + + +def _normalized_systemd_unit_name(value: str) -> str: + normalized = value.strip().lower() + if not re.fullmatch(r"launchplane-runner@[a-z0-9][a-z0-9._-]{0,127}\.service", normalized): + raise ValueError( + "runner lane maintainer systemd_unit_name must match launchplane-runner@.service" + ) + return normalized + + +def _normalized_path(value: str) -> str: + normalized = value.strip().rstrip("/") + if not normalized.startswith("/"): + raise ValueError("runner lane maintainer paths must be absolute") + segments: list[str] = [] + for segment in normalized.split("/"): + if segment in {"", "."}: + continue + if segment == "..": + if segments: + segments.pop() + continue + segments.append(segment) + return "/" + "/".join(segments) + + +def _required_text(value: str, message: str) -> str: + normalized_value = value.strip() + if not normalized_value: + raise ValueError(message) + return normalized_value diff --git a/docs/runner-lane-baseline.md b/docs/runner-lane-baseline.md index 068ac9a8..d0a3f4cc 100644 --- a/docs/runner-lane-baseline.md +++ b/docs/runner-lane-baseline.md @@ -168,6 +168,48 @@ baseline readiness JSON. `--mutate` records the operator's explicit mutation intent in the request so the planner can tell whether a future host adapter would be allowed to proceed, but this CLI command itself remains a dry-run surface. +Runner maintainer planning is also read-only. The typed contract in +`control_plane.contracts.runner_lane_maintainer` captures the desired durable +runner state for a repository, host, lane, registration root, service user, +systemd unit name, labels, and runner-version policy. It compares that desired +state against saved GitHub runner inventory and saved baseline readiness, then +returns the future maintainer decision without contacting GitHub or the host. + +To build the first desired-state plan for a product lane, run: + +```bash +uv run launchplane work-graph runner-maintainer-plan \ + --repository owner/name \ + --host-name chris-testing \ + --lane-name product-runner-1 \ + --runner-directory \ + /home/launchplane-runner-hygiene/actions-runners/product-runner-1 \ + --service-user launchplane-runner-hygiene \ + --systemd-unit-name launchplane-runner@product-runner-1.service \ + --label self-hosted \ + --label launchplane \ + --label launchplane-managed \ + --allowed-repository owner/name \ + --approved-host chris-testing \ + --allowed-registration-root /home/launchplane-runner-hygiene/actions-runners \ + --allowed-service-user launchplane-runner-hygiene \ + --inventory-file runner-inventory.json \ + --baseline-readiness-file runner-baseline.json +``` + +The plan decision can be `recommend_create`, `recommend_verify_adoption`, +`recommend_remove_recreate`, or `blocked`. A recommendation names the durable +action a future maintainer should take; it does not authorize this command to +perform that action. Until the supervised host maintainer exists, +otherwise-valid create, adoption-verification, and remove/recreate plans remain +`status: blocked` with the typed capability blocker +`supervised_maintainer_required`. These packets set `policy_ready: true` and +`capability_ready: false`. Policy blockers such as an unapproved host, unsafe +runner directory, missing managed label, duplicate lane name, repository +mismatch, or failed baseline readiness produce `decision: blocked` with +`policy_ready: false` and must be resolved before any host mutation can be +considered. + ## Registration Executor The first narrow host adapter for creating a repo-scoped runner lane is the diff --git a/tests/test_runner_lane_maintainer.py b/tests/test_runner_lane_maintainer.py new file mode 100644 index 00000000..b4b88c20 --- /dev/null +++ b/tests/test_runner_lane_maintainer.py @@ -0,0 +1,343 @@ +from __future__ import annotations + +import json +import unittest +from pathlib import Path +from tempfile import TemporaryDirectory +from typing import cast + +from click import Command +from click.testing import CliRunner +from click.testing import Result + +from control_plane.cli import main +from control_plane.contracts.runner_lane_baseline import RunnerLaneBaselineReadiness +from control_plane.contracts.runner_lane_inventory import RunnerLaneInventory +from control_plane.contracts.runner_lane_inventory import RunnerLaneRecord +from control_plane.contracts.runner_lane_inventory import build_runner_lane_inventory +from control_plane.contracts.runner_lane_maintainer import RunnerLaneDesiredState +from control_plane.contracts.runner_lane_maintainer import RunnerLaneMaintainerPolicy +from control_plane.contracts.runner_lane_maintainer import plan_runner_lane_maintainer + + +CLI_MAIN = cast(Command, main) + + +class RunnerLaneMaintainerPlanTests(unittest.TestCase): + def test_plan_decides_create_when_inventory_is_empty(self) -> None: + plan = plan_runner_lane_maintainer( + policy=_policy(), + desired_state=_desired_state(), + inventory=_inventory(lanes=()), + baseline_readiness=_ready_baseline(), + ) + + self.assertEqual(plan.status, "blocked") + self.assertEqual(plan.decision, "recommend_create") + self.assertTrue(plan.policy_ready) + self.assertFalse(plan.capability_ready) + self.assertEqual( + [blocker.code for blocker in plan.blockers], ["supervised_maintainer_required"] + ) + self.assertIn("systemd-backed", " ".join(plan.next_steps)) + + def test_plan_decides_verify_adoption_for_online_managed_lane(self) -> None: + plan = plan_runner_lane_maintainer( + policy=_policy(), + desired_state=_desired_state(), + inventory=_inventory(lanes=(_lane(status="online"),)), + baseline_readiness=_ready_baseline(), + ) + + self.assertEqual(plan.status, "blocked") + self.assertEqual(plan.decision, "recommend_verify_adoption") + self.assertTrue(plan.policy_ready) + self.assertFalse(plan.capability_ready) + self.assertEqual( + [blocker.code for blocker in plan.blockers], ["supervised_maintainer_required"] + ) + self.assertIn("systemd unit", " ".join(plan.next_steps)) + + def test_plan_decides_remove_recreate_for_offline_managed_lane(self) -> None: + plan = plan_runner_lane_maintainer( + policy=_policy(), + desired_state=_desired_state(), + inventory=_inventory(lanes=(_lane(status="offline", github_id=21),)), + baseline_readiness=_ready_baseline(), + ) + + self.assertEqual(plan.status, "blocked") + self.assertEqual(plan.decision, "recommend_remove_recreate") + self.assertEqual(plan.matching_lanes[0].github_id, 21) + self.assertTrue(plan.policy_ready) + self.assertFalse(plan.capability_ready) + self.assertEqual( + [blocker.code for blocker in plan.blockers], ["supervised_maintainer_required"] + ) + self.assertIn("recreate", " ".join(plan.next_steps)) + + def test_plan_blocks_existing_unmanaged_lane(self) -> None: + plan = plan_runner_lane_maintainer( + policy=_policy(), + desired_state=_desired_state(), + inventory=_inventory(lanes=(_lane(labels=("self-hosted", "launchplane")),)), + baseline_readiness=_ready_baseline(), + ) + + self.assertEqual(plan.status, "blocked") + self.assertEqual(plan.decision, "blocked") + self.assertFalse(plan.policy_ready) + self.assertTrue(plan.capability_ready) + self.assertEqual([blocker.code for blocker in plan.blockers], ["existing_lane_not_managed"]) + + def test_plan_blocks_without_baseline_readiness(self) -> None: + plan = plan_runner_lane_maintainer( + policy=_policy(), + desired_state=_desired_state(), + inventory=_inventory(lanes=()), + baseline_readiness=RunnerLaneBaselineReadiness( + ready=False, + observed_lanes=0, + compliant_lanes=0, + violations=(), + summary="no runner lane baseline observations supplied", + ), + ) + + self.assertEqual(plan.status, "blocked") + self.assertFalse(plan.policy_ready) + self.assertTrue(plan.capability_ready) + self.assertEqual([blocker.code for blocker in plan.blockers], ["baseline_not_ready"]) + + def test_plan_blocks_unsafe_or_out_of_policy_state(self) -> None: + plan = plan_runner_lane_maintainer( + policy=_policy(approved_hosts=("chris-testing",)), + desired_state=_desired_state(host_name="other-host"), + inventory=_inventory(lanes=()), + baseline_readiness=_ready_baseline(), + ) + + self.assertEqual(plan.status, "blocked") + self.assertEqual([blocker.code for blocker in plan.blockers], ["host_not_allowed"]) + + def test_plan_blocks_ambiguous_lane_name(self) -> None: + plan = plan_runner_lane_maintainer( + policy=_policy(), + desired_state=_desired_state(), + inventory=_inventory( + lanes=( + _lane(status="offline", github_id=21), + _lane(status="online", github_id=22, name="CM-WEBSITE-CHRIS-TESTING"), + ) + ), + baseline_readiness=_ready_baseline(), + ) + + self.assertEqual(plan.status, "blocked") + self.assertEqual(plan.decision, "blocked") + self.assertFalse(plan.policy_ready) + self.assertTrue(plan.capability_ready) + self.assertEqual([blocker.code for blocker in plan.blockers], ["lane_name_ambiguous"]) + + def test_plan_requires_explicit_allowed_service_user(self) -> None: + plan = plan_runner_lane_maintainer( + policy=RunnerLaneMaintainerPolicy( + allowed_repositories=("cbusillo/odoo-tenant-cm-website",), + approved_hosts=("chris-testing",), + allowed_registration_roots=("/home/launchplane-runner-hygiene/actions-runners",), + allowed_service_users=(), + ), + desired_state=_desired_state(), + inventory=_inventory(lanes=()), + baseline_readiness=_ready_baseline(), + ) + + self.assertEqual(plan.status, "blocked") + self.assertFalse(plan.policy_ready) + self.assertTrue(plan.capability_ready) + self.assertEqual([blocker.code for blocker in plan.blockers], ["service_user_not_allowed"]) + + def test_desired_state_requires_systemd_template_unit(self) -> None: + with self.assertRaisesRegex(ValueError, "launchplane-runner"): + _desired_state(systemd_unit_name="actions.runner.cm.service") + + def test_desired_state_requires_full_runner_directory(self) -> None: + with self.assertRaisesRegex(ValueError, "lane directory"): + RunnerLaneDesiredState( + repository="cbusillo/odoo-tenant-cm-website", + host_name="chris-testing", + lane_name="cm-website-chris-testing", + runner_directory="/home/launchplane-runner-hygiene/actions-runners", + service_user="launchplane-runner-hygiene", + systemd_unit_name=_UNIT_NAME, + labels=("self-hosted", "launchplane", "launchplane-managed"), + ) + + def test_desired_state_requires_systemd_unit_for_lane(self) -> None: + with self.assertRaisesRegex(ValueError, "lane name"): + _desired_state(systemd_unit_name="launchplane-runner@other-lane.service") + + +class RunnerLaneMaintainerCliTests(unittest.TestCase): + def test_cli_emits_create_decision_for_empty_inventory(self) -> None: + result = _invoke_cli(lanes=()) + + self.assertEqual(result.exit_code, 0, result.output) + payload = json.loads(result.output) + self.assertEqual(payload["plan"]["status"], "blocked") + self.assertEqual(payload["plan"]["decision"], "recommend_create") + self.assertTrue(payload["plan"]["policy_ready"]) + self.assertFalse(payload["plan"]["capability_ready"]) + self.assertEqual(payload["plan"]["blockers"][0]["code"], "supervised_maintainer_required") + self.assertEqual(payload["desired_state"]["systemd_unit_name"], _UNIT_NAME) + + def test_cli_emits_remove_recreate_decision_for_offline_lane(self) -> None: + result = _invoke_cli(lanes=(_lane(status="offline", github_id=21),)) + + self.assertEqual(result.exit_code, 0, result.output) + payload = json.loads(result.output) + self.assertEqual(payload["plan"]["decision"], "recommend_remove_recreate") + self.assertEqual(payload["plan"]["matching_lanes"][0]["github_id"], 21) + + def test_cli_uses_nested_baseline_readiness_envelope(self) -> None: + result = _invoke_cli(lanes=(), nested_baseline=True) + + self.assertEqual(result.exit_code, 0, result.output) + payload = json.loads(result.output) + self.assertEqual(payload["plan"]["decision"], "recommend_create") + + +_UNIT_NAME = "launchplane-runner@cm-website-chris-testing.service" + + +def _policy(*, approved_hosts: tuple[str, ...] = ("chris-testing",)) -> RunnerLaneMaintainerPolicy: + return RunnerLaneMaintainerPolicy( + allowed_repositories=("cbusillo/odoo-tenant-cm-website",), + approved_hosts=approved_hosts, + allowed_registration_roots=("/home/launchplane-runner-hygiene/actions-runners",), + allowed_service_users=("launchplane-runner-hygiene",), + ) + + +def _desired_state( + *, + host_name: str = "chris-testing", + systemd_unit_name: str = _UNIT_NAME, +) -> RunnerLaneDesiredState: + return RunnerLaneDesiredState( + repository="cbusillo/odoo-tenant-cm-website", + host_name=host_name, + lane_name="cm-website-chris-testing", + runner_directory="/home/launchplane-runner-hygiene/actions-runners/cm-website-chris-testing", + service_user="launchplane-runner-hygiene", + systemd_unit_name=systemd_unit_name, + labels=( + "self-hosted", + "launchplane", + "launchplane-managed", + "chris-testing", + "cm-website", + ), + ) + + +def _ready_baseline() -> RunnerLaneBaselineReadiness: + return RunnerLaneBaselineReadiness( + ready=True, + observed_lanes=1, + compliant_lanes=1, + violations=(), + summary="runner lane baseline satisfied", + ) + + +def _inventory(*, lanes: tuple[RunnerLaneRecord, ...]) -> RunnerLaneInventory: + return build_runner_lane_inventory( + repository="cbusillo/odoo-tenant-cm-website", + observed_at="2026-06-08T21:15:00Z", + lanes=lanes, + ) + + +def _lane( + *, + status: str = "online", + github_id: int = 1, + name: str = "cm-website-chris-testing", + labels: tuple[str, ...] = ( + "self-hosted", + "launchplane", + "launchplane-managed", + ), +) -> RunnerLaneRecord: + return RunnerLaneRecord( + github_id=github_id, + name=name, + repository="cbusillo/odoo-tenant-cm-website", + status=status, + busy=False, + labels=labels, + host_hint="chris-testing", + observed_at="2026-06-08T21:15:00Z", + ) + + +def _invoke_cli(*, lanes: tuple[RunnerLaneRecord, ...], nested_baseline: bool = False) -> Result: + with TemporaryDirectory() as temp_dir: + temp_path = Path(temp_dir) + inventory_file = temp_path / "inventory.json" + baseline_file = temp_path / "baseline.json" + inventory_file.write_text( + json.dumps(_inventory(lanes=lanes).model_dump(mode="json")), + encoding="utf-8", + ) + baseline_payload: dict[str, object] = _ready_baseline().model_dump(mode="json") + if nested_baseline: + baseline_payload = {"readiness": baseline_payload} + baseline_file.write_text(json.dumps(baseline_payload), encoding="utf-8") + + return CliRunner().invoke( + CLI_MAIN, + [ + "work-graph", + "runner-maintainer-plan", + "--repository", + "cbusillo/odoo-tenant-cm-website", + "--host-name", + "chris-testing", + "--lane-name", + "cm-website-chris-testing", + "--runner-directory", + "/home/launchplane-runner-hygiene/actions-runners/cm-website-chris-testing", + "--service-user", + "launchplane-runner-hygiene", + "--systemd-unit-name", + _UNIT_NAME, + "--label", + "self-hosted", + "--label", + "launchplane", + "--label", + "launchplane-managed", + "--label", + "chris-testing", + "--label", + "cm-website", + "--allowed-repository", + "cbusillo/odoo-tenant-cm-website", + "--approved-host", + "chris-testing", + "--allowed-registration-root", + "/home/launchplane-runner-hygiene/actions-runners", + "--allowed-service-user", + "launchplane-runner-hygiene", + "--inventory-file", + str(inventory_file), + "--baseline-readiness-file", + str(baseline_file), + ], + ) + + +if __name__ == "__main__": + unittest.main()