ci_workflow.py

"""Render a GitHub Actions "fine-tuning gate" workflow (v0.71.35).

``soup ci init`` writes ``.github/workflows/soup-gate.yml`` — a CI job that
gates every PR on three Soup checks:

    soup data validate <data>      # dataset format compliance
    soup expect <data> <suite>     # expectations suite (PII / length / refusal)
    soup ship --evidence <ev.json> # SHIP / DON'T-SHIP verdict (exit 2 = block)

The workflow body is a fixed YAML skeleton; only a handful of tokens are
interpolated. Following ``eval_gate_hook.py``: any value placed into a ``run:``
shell step is passed through ``shlex.quote`` (+ control-char rejection) so a
crafted path cannot chain commands, and every path is validated to stay under
the repo root. There is NO top-level torch import.
"""

from __future__ import annotations

import os
import re
import shlex
import tempfile

from soup_cli.utils.paths import enforce_under_cwd_and_no_symlink, is_under_cwd

# GitHub allows quite permissive ref names; we constrain to a safe subset that
# also cannot break the YAML scalar or a shell step.
_BRANCH_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]{0,254}$")
_PY_RE = re.compile(r"^\d+\.\d+$")
_MAX_PATH_LEN = 4096
_MAX_FILE_BYTES = 64 * 1024

_WORKFLOW_TEMPLATE = """\
# Soup fine-tuning gate — generated by `soup ci init` (v0.71.35).
# Gates every PR on: data validate -> expectations -> SHIP verdict.
# Edit the paths below to match your repo. The `soup ship` step needs a
# committed evidence JSON (see `soup ship --evidence`); remove it if unused.
name: Soup Fine-tuning Gate

on:
  push:
    branches: [{branch}]
  pull_request:
    branches: [{branch}]

jobs:
  soup-gate:
    runs-on: ubuntu-latest
    env:
      PYTHONUTF8: "1"
      PYTHONIOENCODING: "utf-8"
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "{python_version}"
      - name: Install Soup CLI
        # Core install is enough: `data validate`, `expect` and
        # `ship --evidence` are all no-torch paths.
        run: pip install soup-cli
      - name: Validate training data
        run: soup data validate {data}
      - name: Run expectations suite
        run: soup expect {data} {suite}
      - name: SHIP / DON'T-SHIP gate
        run: soup ship --evidence {evidence}
"""


def _validate_path(value: str, field: str) -> str:
    if isinstance(value, bool) or not isinstance(value, str):
        raise TypeError(f"{field} must be str")
    if not value:
        raise ValueError(f"{field} must be non-empty")
    if "\x00" in value:
        raise ValueError(f"{field} must not contain NUL")
    if "\n" in value or "\r" in value:
        raise ValueError(f"{field} must be a single line")
    if len(value) > _MAX_PATH_LEN:
        raise ValueError(f"{field} exceeds {_MAX_PATH_LEN} characters")
    if not is_under_cwd(value):
        raise ValueError(f"{field} must stay under the repository root")
    return value


def _safe_shell_quote(value: str) -> str:
    """``shlex.quote`` with a control-char rejection prelude (defence in depth)."""
    if any(ord(ch) < 0x20 for ch in value):
        raise ValueError("value contains control characters")
    return shlex.quote(value)


def render_soup_gate_workflow(
    *,
    data_path: str,
    suite_path: str,
    evidence_path: str,
    python_version: str = "3.11",
    branch: str = "main",
) -> str:
    """Render the workflow YAML body. Deterministic, no I/O.

    Every path is validated to stay under the repo root and shell-quoted before
    it reaches a ``run:`` step; ``python_version`` / ``branch`` are regex-gated.
    """
    data = _validate_path(data_path, "data_path")
    suite = _validate_path(suite_path, "suite_path")
    evidence = _validate_path(evidence_path, "evidence_path")

    if not isinstance(python_version, str) or not _PY_RE.match(python_version):
        raise ValueError("python_version must look like '3.11'")
    if not isinstance(branch, str) or not _BRANCH_RE.match(branch):
        raise ValueError("branch has an unsupported name")

    return _WORKFLOW_TEMPLATE.format(
        branch=branch,
        python_version=python_version,
        data=_safe_shell_quote(data),
        suite=_safe_shell_quote(suite),
        evidence=_safe_shell_quote(evidence),
    )


def write_soup_gate_workflow(
    *,
    data_path: str,
    suite_path: str,
    evidence_path: str,
    python_version: str = "3.11",
    branch: str = "main",
    output_path: str = ".github/workflows/soup-gate.yml",
    overwrite: bool = False,
) -> str:
    """Render + atomically write the workflow. Returns the path written.

    Refuses to clobber an existing file unless ``overwrite=True``; rejects a
    symlink destination (TOCTOU) and any path outside the repo root.
    """
    body = render_soup_gate_workflow(
        data_path=data_path,
        suite_path=suite_path,
        evidence_path=evidence_path,
        python_version=python_version,
        branch=branch,
    )
    if isinstance(output_path, bool) or not isinstance(output_path, str):
        raise TypeError("output_path must be str")
    if not output_path:
        raise ValueError("output_path must be non-empty")
    if "\x00" in output_path:
        raise ValueError("output_path must not contain NUL")
    if not isinstance(overwrite, bool):
        raise TypeError("overwrite must be bool")
    # Single-source-of-truth containment + symlink/reparse-point rejection.
    # A hand-rolled S_ISLNK check misses Windows junctions, so always delegate
    # to the shared helper (v0.71.35 security review).
    enforce_under_cwd_and_no_symlink(output_path, "output_path")
    if os.path.lexists(output_path) and not overwrite:
        raise ValueError("workflow already exists; pass --force to overwrite")
    if len(body.encode("utf-8")) > _MAX_FILE_BYTES:
        raise ValueError("rendered workflow exceeds 64 KiB cap")

    parent = os.path.dirname(os.path.abspath(output_path)) or "."
    os.makedirs(parent, exist_ok=True)
    fd, tmp = tempfile.mkstemp(prefix=".soup-gate.", dir=parent)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            handle.write(body)
        os.replace(tmp, output_path)
    except Exception:
        try:
            os.unlink(tmp)
        except OSError:
            pass
        raise
    return output_path


__all__ = ["render_soup_gate_workflow", "write_soup_gate_workflow"]