orchestrata_ultimate.promotion.promoter

Reference › orchestrata_ultimate › promotion › promoter

The Promoter class is the front of the promotion pipeline (dev, then test, then prod): it validates an environment, runs the suite against it, promotes (deploys) to it, and keeps a JSON ledger of what happened, by default reports/promotion-state.json under the repository. It is what the promote_env command drives. Deployment itself is delegated to the runner; profile checks come from env_profile. The promotion narrative explains why promotion means deployment.

The stages and what each refuses:

• validate: the profile must exist and define every required key, every infrastructure host must be on the environment's own subnet, the repository's validate_env.py preflight must pass with the profile as its environment, and Redis and the graph must answer on TCP (unless the promoter was built with probe off). Any failure raises PromotionError. Success writes a validate PASS record.
• test: refused outright for prod, because the suite writes test data and prod is live. For other environments it validates first, runs scripts/run_tests.py --json pinned to the environment, and records the commit sha, the ship source id from reports/.ship-verified when it names the same commit, the totals, the reused ids and PASS or FAIL.
• promote: gate_for checks the previous stage. Promotion to test needs the dev validation to have passed. Promotion to prod needs a PASS test record whose recorded sha equals the current HEAD; a record without a sha counts as absent. The target is then validated, and prod additionally needs explicit confirmation (--yes) unless it is a dry run. The test target goes to promoter_targets; other targets run deploy with streamed progress, record the report, and raise PromotionError naming the failed stage.

drift_line and status print the ledger per environment with the recorded sha and whether the target still serves it.

At a glance

Class diagram

Dependency diagram

API

Classes

Promoter (dataclass)

Decorators: dataclass · Line 54

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/promotion/promoter.py

"""The stateful promoter — env_dir / ledger_path injected for testability.

Tests construct a Promoter over a tmp ledger + a tmp env dir; production uses
the defaults. No os.environ mutation, no monkeypatch — pure dependency
injection (the suite-wide test-standard seam). Split out of
``scripts/promote_env.py`` (RUL-021 file budget) — behaviour-neutral.
"""

from __future__ import annotations

import json
import os
import subprocess
import sys
from dataclasses import dataclass, field
from pathlib import Path

from dotenv import dotenv_values

from orchestrata_ultimate.core.paths import get_project_root
from orchestrata_ultimate.promotion import promoter_targets
from orchestrata_ultimate.promotion.env_profile import (
    ENV_SUBNET,
    PIPELINE,
    PromotionError,
    _as_utc,
    _utc_now,
    check_subnet,
    missing_required,
    probe_infra,
)
from orchestrata_ultimate.promotion.runner import deploy, drift
from orchestrata_ultimate.promotion.stages import DeployContext

#: Resolved through ``get_project_root`` (``ORCHESTRATA_ROOT``, else the nearest
#: ``pyproject.toml`` above this file), never a fixed ``__file__`` parent count.
#: Since the proper install (dd919d55c, 2026-10-20) this module lives in
#: ``.venv/lib/python3.14t/site-packages/``, where three parents up is
#: ``.venv/lib/python3.14t``: ship's promote read
#: ``.venv/lib/python3.14t/env/test.env`` there and refused (2026-10-22).
_REPO_ROOT = get_project_root()


def _parse_run_document(stdout: str) -> dict:
    """The JSON document `run_tests.py --json` prints; {} when stdout has none."""
    try:
        doc = json.loads(stdout)
    except ValueError:
        return {}
    return doc if isinstance(doc, dict) else {}


@dataclass
class Promoter:
    repo_root: Path = _REPO_ROOT
    env_dir: Path = field(default=None)  # type: ignore[assignment]
    ledger_path: Path = field(default=None)  # type: ignore[assignment]
    # Infra reachability is probed by default; tests pass probe=False to keep
    # the gate logic hermetic (the live TCP probe is exercised separately).
    probe: bool = True

    def __post_init__(self) -> None:
        if self.env_dir is None:
            self.env_dir = self.repo_root / "env"
        if self.ledger_path is None:
            self.ledger_path = self.repo_root / "reports" / "promotion-state.json"

    # ---- profile loading -------------------------------------------------- #
    def profile_path(self, env: str) -> Path:
        return self.env_dir / f"{env}.env"

    def load_profile(self, env: str) -> dict[str, str]:
        """Load ``env/<env>.env`` as a plain dict — no os.environ read, no
        defaults. Raises PromotionError on a missing file or empty required
        key."""
        if env not in PIPELINE:
            raise PromotionError(
                f"unknown env {env!r}; expected one of {', '.join(PIPELINE)}"
            )
        path = self.profile_path(env)
        if not path.exists():
            raise PromotionError(f"env profile not found: {path}")
        raw = dotenv_values(path)
        profile = {k: v for k, v in raw.items() if v is not None}
        missing = missing_required(profile)
        if missing:
            raise PromotionError(
                f"{path.name} is missing required infra keys {missing} "
                f"(RUL-003: no literal fallback — the stage cannot guess an "
                f"endpoint)"
            )
        return profile

    # ---- ledger ----------------------------------------------------------- #
    def load_ledger(self) -> dict:
        if self.ledger_path.exists():
            loaded = json.loads(self.ledger_path.read_text())
            if not isinstance(loaded, dict):
                raise PromotionError(f"ledger {self.ledger_path} is not a JSON object")
            return loaded
        return {}

    def _write_ledger(self, ledger: dict) -> None:
        self.ledger_path.parent.mkdir(parents=True, exist_ok=True)
        self.ledger_path.write_text(json.dumps(ledger, indent=2, sort_keys=True) + "\n")

    def record(self, env: str, stage: str, payload: dict) -> None:
        ledger = self.load_ledger()
        ledger.setdefault(env, {})[stage] = {"at": _utc_now(), **payload}
        self._write_ledger(ledger)

    # ---- gates ------------------------------------------------------------ #
    def head_sha(self) -> str:
        """The commit of this checkout — what `test` tests and `promote` deploys.

        `promote` resolves its commit the same way (stages_resolve.resolve_commit:
        `git rev-parse HEAD` in ctx.repo_root, which is this repo_root).
        """
        proc = subprocess.run(
            ["git", "rev-parse", "HEAD"],
            cwd=self.repo_root,
            capture_output=True,
            text=True,
        )
        if proc.returncode != 0:
            raise PromotionError(
                f"cannot resolve HEAD of {self.repo_root}: {proc.stderr.strip()}"
            )
        return proc.stdout.strip()

    def ship_source_id(self, sha: str) -> str | None:
        """ship's source id for `sha`, from ship's verified stamp.

        deploy/ship.sh writes `<source-id> <commit>` to reports/.ship-verified
        when its suite step passes. The id is returned only when the stamp's
        commit is `sha`; a stamp of another commit says nothing about this one.
        """
        stamp = self.repo_root / "reports" / ".ship-verified"
        try:
            fields = stamp.read_text().split()
        except OSError:
            return None
        if len(fields) >= 2 and fields[1] == sha:
            return fields[0]
        return None

    def gate_for(self, target: str) -> None:
        """Raise PromotionError unless the prior stage's gate is satisfied."""
        idx = PIPELINE.index(target)
        if idx == 0:
            return  # dev is the entry point — no upstream gate
        prior = PIPELINE[idx - 1]
        prior_rec = self.load_ledger().get(prior, {})
        if target == "prod":
            test_rec = prior_rec.get("test")
            if not test_rec or test_rec.get("result") != "PASS":
                raise PromotionError(
                    "cannot promote to prod: no GREEN test-env suite run on "
                    "record (run `promote_env.py test test` first). Current "
                    f"test record: {test_rec}"
                )
            # The GREEN run must be of the commit being promoted. A record
            # without a sha cannot say what it tested, so it counts as absent.
            tested = test_rec.get("sha")
            promoted = self.head_sha()
            if not tested or tested != promoted:
                raise PromotionError(
                    "cannot promote to prod: the GREEN test-env run is not of "
                    f"the commit being promoted (tested {tested or 'none'}, "
                    f"promoting {promoted}); run `promote_env.py test test` "
                    "on this commit first"
                )
        else:
            val_rec = prior_rec.get("validate")
            if not val_rec or val_rec.get("result") != "PASS":
                raise PromotionError(
                    f"cannot promote to {target}: {prior} has not validated "
                    f"(run `promote_env.py validate {prior}` first)"
                )

    # ---- commands --------------------------------------------------------- #
    def validate(self, env: str) -> int:
        """Profile present + required keys + right subnet + infra reachable."""
        profile = self.load_profile(env)
        subnet_violations = check_subnet(env, profile)
        if subnet_violations:
            for v in subnet_violations:
                print(f"FAIL subnet: {v}", file=sys.stderr)
            raise PromotionError(
                f"{env} profile is on the wrong subnet — refusing (mis-edit guard)"
            )
        # Authoritative preflight: run scripts/validate_env.py with the profile
        # as the process env so its exact REQUIRED_KEYS contract gates us too.
        preflight = subprocess.run(
            [
                sys.executable,
                str(self.repo_root / "scripts" / "validate_env.py"),
                "--quiet",
            ],
            env={**os.environ, **profile},
            cwd=self.repo_root,
        )
        if preflight.returncode != 0:
            raise PromotionError(
                f"validate_env preflight failed for {env} (rc={preflight.returncode})"
            )
        if self.probe:
            infra_failures = probe_infra(profile)
            if infra_failures:
                for f in infra_failures:
                    print(f"FAIL reachability: {f}", file=sys.stderr)
                raise PromotionError(f"{env} infra is not reachable — refusing")
        self.record(env, "validate", {"result": "PASS"})
        print(
            f"PASS validate {env}: profile OK, subnet {ENV_SUBNET[env]}x, "
            f"redis+graph reachable"
        )
        return 0

    def test(self, env: str, cmd_override: list[str] | None = None) -> int:
        """Run the suite pinned to <env>. Refused for prod."""
        if env == "prod":
            raise PromotionError(
                "refusing to run the test suite against prod — the suite writes "
                "test data and prod is the live cluster. Promote to prod with "
                "`promote --to prod`, which gates on the GREEN test-env run."
            )
        self.validate(env)  # a stage never tests an unvalidated env
        cmd = cmd_override or [sys.executable, "scripts/run_tests.py", "--json"]
        print(f"RUN test {env}: {' '.join(cmd)} (ORCHESTRATA_ENV_PROFILE={env})")
        sha = self.head_sha()
        # stdout is the run's JSON document (stderr streams through); parsing
        # it, not reports/run-tests-latest.json, keeps a nested run that
        # overwrites the shared report from changing this record.
        proc = subprocess.run(
            cmd,
            env={**os.environ, "ORCHESTRATA_ENV_PROFILE": env},
            cwd=self.repo_root,
            stdout=subprocess.PIPE,
            text=True,
        )
        sys.stdout.write(proc.stdout or "")
        doc = _parse_run_document(proc.stdout or "")
        reused = doc.get("reused")
        ids = reused.get("ids") if isinstance(reused, dict) else None
        result = "PASS" if proc.returncode == 0 else "FAIL"
        self.record(
            env,
            "test",
            {
                "sha": sha,
                "source_id": self.ship_source_id(sha),
                "totals": doc.get("test_totals"),
                "reused": {"ids": ids if isinstance(ids, list) else []},
                "result": result,
                "returncode": proc.returncode,
                "command": " ".join(cmd),
            },
        )
        print(f"{result} test {env} (rc={proc.returncode})")
        return proc.returncode

    def promote(
        self,
        target: str,
        assume_yes: bool = False,
        *,
        execute: bool = True,
        skip_graphs: bool = False,
        skip_ingest: bool = False,
    ) -> int:
        """Gated DEPLOYMENT to <target>: advance it, restart it, prove it.

        This used to write a receipt — gate satisfied, target reachable,
        timestamp — and stop. That made PROMOTED a claim with no referent:
        measured 2026-08-10, the ledger read PROMOTED while prod served a tree
        76 commits behind, and nothing in the system could tell. Promotion now
        means the target actually runs the promoted commit, and the ledger
        records WHICH commit so the question has an answer.
        """
        self.gate_for(target)
        self.validate(target)  # target must itself be reachable
        if target == "prod" and not assume_yes and execute:
            print(
                "About to deploy to PROD (the live 10.53.1.x cluster): advance "
                "the tree, restart the services, verify. Re-run with --yes.",
                file=sys.stderr,
            )
            raise PromotionError("prod promotion requires explicit --yes")

        if target == "test":
            return self._promote_test_container(execute=execute)

        ctx = self._deploy_context(target)
        # Stream each stage as it starts and finishes (flush per line — this
        # process is usually watched through a log file). The buffered
        # end-of-run report alone left the slowest stage running in total
        # silence for 15+ minutes on the first live promote, indistinguishable
        # from a hang (user ruling 2026-08-14: gates must notify).
        report = deploy(
            ctx,
            execute=execute,
            skip_graphs=skip_graphs,
            skip_ingest=skip_ingest,
            progress=lambda line: print(line, flush=True),
        )
        print(report.render())

        if execute:
            self.record(target, "promote", report.as_dict())
        if not report.ok:
            failure = report.failure
            raise PromotionError(
                f"deployment to {target} FAILED at stage "
                f"{failure.name if failure else '?'}: "
                f"{failure.detail if failure else 'unknown'}"
            )
        return 0

    def _promote_test_container(self, *, execute: bool = True) -> int:
        return promoter_targets.promote_test_container(self, execute=execute)

    def _deploy_context(self, target: str) -> DeployContext:
        return promoter_targets.deploy_context(self, target)

    def drift_line(self, env: str) -> str:
        """One line saying whether the ledger still describes reality."""
        ledger = self.load_ledger().get(env, {}).get("promote", {})
        sha = str(ledger.get("sha", "")) if isinstance(ledger, dict) else ""
        try:
            result = drift(self._deploy_context(env), sha)
        except Exception as exc:  # profile missing, tree absent
            return f"  drift: unknown ({exc})"
        return f"  drift: {result.detail}"

    def status(self) -> int:
        ledger = self.load_ledger()
        if not ledger:
            print("no promotion records yet")
            return 0
        for env in PIPELINE:
            rec = ledger.get(env, {})
            line = [f"{env:5s}"]
            for stage in ("validate", "test", "promote"):
                s = rec.get(stage)
                if not s:
                    line.append(f"{stage}=-")
                    continue
                cell = f"{stage}={s.get('result')}@{_as_utc(s.get('at', ''))}"
                # The sha is the fact that makes a promotion record checkable;
                # without it "PROMOTED" cannot be compared against anything.
                if stage == "promote" and s.get("sha"):
                    cell += f" sha={str(s['sha'])[:12]}"
                line.append(cell)
            print("  ".join(line))
            if rec.get("promote"):
                print(self.drift_line(env))
        return 0