mindxtrain/deploy/github_push.py

"""GitHub push step builder.

Returns an idempotent list of `Step`s for bootstrapping a git repo, creating
the GitHub remote (via `gh`), and pushing the working tree. Pure builder —
no subprocesses are spawned here. The orchestrator in `_orchestrator.py`
runs the steps via `mindxtrain.operator.runs.spawn_subprocess_streaming` so
each command's stdout streams back over the existing SSE pipeline.

Auth: `GITHUB_TOKEN` is forwarded into the subprocess env as `GH_TOKEN` (the
name `gh` looks for) and as `GITHUB_TOKEN` (for `git push` via the credential
helper). The token never lands in argv or `.git/config`.
"""

from __future__ import annotations

import os
import shutil
from dataclasses import dataclass, field
from pathlib import Path


@dataclass(frozen=True)
class Step:
    """A single shell-out step in a deploy pipeline."""

    label: str
    cmd: list[str]
    env: dict[str, str] = field(default_factory=dict)
    # If `predicate_step` is set, this step only runs when that step's rc
    # matches `predicate_rc_in`. Used to model conditional steps without
    # building a full DAG.
    predicate_step: str | None = None
    predicate_rc_in: tuple[int, ...] = (0,)
    # If `capture_stdout` is True, the orchestrator stashes the step's stdout
    # under `Step.label` so later steps / callers can read it.
    capture_stdout: bool = False
    # If True, the orchestrator continues even on non-zero rc (used for probes).
    allow_failure: bool = False


# Credential helper inline script — feeds the token to git push without
# touching ~/.gitconfig or .git/config. Single-quoted body so $GH_TOKEN is
# expanded by the spawned shell, not by Python's f-string.
_GIT_CRED_HELPER = (
    "credential.helper="
    "!f() { echo username=x-access-token; echo \"password=$GH_TOKEN\"; }; f"
)


@dataclass(frozen=True)
class GithubConfig:
    token: str
    repo: str  # "owner/name"
    branch: str = "main"
    author_name: str = "mindXtrain bot"
    author_email: str = "noreply@pythai.net"


def _step_env(token: str) -> dict[str, str]:
    """Subprocess env: pass GH_TOKEN + GITHUB_TOKEN, scrub nothing else."""
    base = dict(os.environ)
    base["GH_TOKEN"] = token
    base["GITHUB_TOKEN"] = token
    return base


def required_env() -> tuple[str, ...]:
    """Names of env vars the operator must populate for /api/github/push."""
    return ("GITHUB_TOKEN", "GITHUB_REPO")


def missing_env(env: dict[str, str] | None = None) -> list[str]:
    src = env if env is not None else os.environ
    return [k for k in required_env() if not src.get(k)]


def _which_missing() -> list[str]:
    """Binaries the push pipeline depends on, but only if they're absent."""
    out = []
    for binary in ("git", "gh"):
        if shutil.which(binary) is None:
            out.append(binary)
    return out


def status_target(env: dict[str, str] | None = None) -> str:
    src = env if env is not None else os.environ
    return src.get("GITHUB_REPO", "")


def status_missing(env: dict[str, str] | None = None) -> list[str]:
    """Combine env-missing + binary-missing for the /status endpoint."""
    return missing_env(env) + _which_missing()


def bootstrap_steps(
    cfg: GithubConfig,
    *,
    commit_message: str = "mindXtrain initial push",
    force: bool = False,
) -> list[Step]:
    """Idempotent step list. The orchestrator skips conditional steps by
    inspecting earlier steps' rcs."""
    env = _step_env(cfg.token)
    repo_url = f"https://github.com/{cfg.repo}.git"
    push_cmd = [
        "git",
        "-c", _GIT_CRED_HELPER,
        "push", "-u", "origin", f"HEAD:{cfg.branch}",
    ]
    if force:
        push_cmd.append("--force-with-lease")

    return [
        Step(
            label="probe-git",
            cmd=["git", "rev-parse", "--git-dir"],
            env=env,
            allow_failure=True,
        ),
        Step(
            label="git-init",
            cmd=["git", "init", "-b", cfg.branch],
            env=env,
            predicate_step="probe-git",
            predicate_rc_in=(1, 128),  # not a git repo
        ),
        Step(
            label="probe-repo",
            cmd=["gh", "repo", "view", cfg.repo],
            env=env,
            allow_failure=True,
        ),
        Step(
            label="gh-create",
            cmd=["gh", "repo", "create", cfg.repo, "--public",
                 "--source=.", "--remote=origin"],
            env=env,
            predicate_step="probe-repo",
            predicate_rc_in=(1,),  # repo doesn't exist yet
        ),
        Step(
            label="probe-remote",
            cmd=["git", "remote", "get-url", "origin"],
            env=env,
            allow_failure=True,
            capture_stdout=True,
        ),
        Step(
            # If origin already exists but doesn't match, only rewrite when force=True.
            # The orchestrator emits a guidance LogEvent + bails when force=False
            # and probe-remote stdout doesn't match repo_url.
            label="git-remote-add",
            cmd=["git", "remote", "add", "origin", repo_url],
            env=env,
            predicate_step="probe-remote",
            predicate_rc_in=(1, 2, 128),  # remote not configured
        ),
        Step(label="git-add", cmd=["git", "add", "-A"], env=env),
        Step(
            label="probe-stage",
            cmd=["git", "diff", "--cached", "--quiet"],
            env=env,
            allow_failure=True,
        ),
        Step(
            label="git-commit",
            cmd=[
                "git",
                "-c", f"user.email={cfg.author_email}",
                "-c", f"user.name={cfg.author_name}",
                "commit", "-m", commit_message,
            ],
            env=env,
            predicate_step="probe-stage",
            predicate_rc_in=(1,),  # rc=1 means "differences exist" → there's something to commit
        ),
        Step(label="git-push", cmd=push_cmd, env=env),
        Step(
            label="head-sha",
            cmd=["git", "rev-parse", "HEAD"],
            env=env,
            capture_stdout=True,
        ),
    ]


def remote_url(repo: str) -> str:
    return f"https://github.com/{repo}.git"


def write_sha_file(out_dir: Path, sha: str) -> Path:
    """Persist the captured HEAD SHA so emit_receipt callers can pin
    `manifest.git_sha` without re-shelling git."""
    out_dir.mkdir(parents=True, exist_ok=True)
    target = out_dir / "git_sha.txt"
    target.write_text(sha.strip() + "\n")
    return target


__all__ = [
    "GithubConfig",
    "Step",
    "bootstrap_steps",
    "missing_env",
    "remote_url",
    "required_env",
    "status_missing",
    "status_target",
    "write_sha_file",
]