rein-aharness/rein_aharness/smoke.py

"""Deterministic remote smoke without an LLM session.

Proves packaging, git write to the target repo, kaizen metrics, and hub
reporting on Railiance. Full agentic sessions still need Claude Code (or a
hosted adapter); this path is the T06 end-to-end gate when the CLI is absent.
"""

from __future__ import annotations

import subprocess
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path

from rein_aharness.runner import RunResult, run_task
from rein_aharness.taskspec import TaskSpec


@dataclass
class SmokeResult:
    run: RunResult
    pushed: bool
    push_reason: str = ""


class _CommittingSmokeAdapter:
    """Minimal adapter: write SMOKE.md and commit (no network, no push)."""

    def __init__(self, repo: Path, stamp: str):
        self.repo = repo
        self.stamp = stamp
        self.prompts: list[str] = []

    def execute_prompt(self, prompt, config):
        self.prompts.append(prompt)
        path = self.repo / "SMOKE.md"
        path.write_text(
            f"# rein-aharness smoke\n\nstamp: {self.stamp}\n",
            encoding="utf-8",
        )
        subprocess.run(["git", "add", "SMOKE.md"], cwd=self.repo, check=True)
        subprocess.run(
            [
                "git",
                "-c",
                "user.email=rein-aharness@railiance.local",
                "-c",
                "user.name=rein-aharness",
                "commit",
                "-qm",
                f"harness smoke: {self.stamp}",
            ],
            cwd=self.repo,
            check=True,
        )
        from llm_connect.models import LLMResponse

        return LLMResponse(
            content=f"smoke committed {self.stamp}",
            model="smoke-adapter",
            usage={"input_tokens": 0, "output_tokens": 0},
            finish_reason="stop",
        )


def _git(repo: Path, *args: str, check: bool = True) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        ["git", "-C", str(repo), *args],
        capture_output=True,
        text=True,
        timeout=120,
        check=check,
    )


def ensure_sandbox_clone(
    dest: Path,
    *,
    remote: str = "ssh://git@forgejo-agent-harness/coulomb/executor-sandbox.git",
) -> Path:
    """Clone or fetch the executor-sandbox repo at *dest*."""
    dest = Path(dest).expanduser()
    if (dest / ".git").is_dir():
        _git(dest, "fetch", "origin", check=False)
        _git(dest, "checkout", "main", check=False)
        _git(dest, "pull", "--ff-only", "origin", "main", check=False)
        return dest
    dest.parent.mkdir(parents=True, exist_ok=True)
    subprocess.run(
        ["git", "clone", remote, str(dest)],
        check=True,
        capture_output=True,
        text=True,
        timeout=120,
    )
    return dest


def run_smoke(
    target_repo: Path,
    *,
    report_to_hub: bool = True,
    push: bool = True,
    agent: str = "coach",
) -> SmokeResult:
    target_repo = Path(target_repo).expanduser().resolve()
    if not (target_repo / ".git").is_dir():
        raise RuntimeError(f"not a git repo: {target_repo}")

    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    adapter = _CommittingSmokeAdapter(target_repo, stamp)
    spec = TaskSpec(
        title=f"harness smoke {stamp}",
        description=(
            "Deterministic smoke: write SMOKE.md and commit. "
            "No LLM session (Railiance packaging gate)."
        ),
        target_repo=target_repo,
        agent=agent,
        labels=["harness", "smoke", "railiance"],
        completion_event_type="harness_smoke",
        timeout_seconds=120,
    )
    result = run_task(
        spec,
        adapter=adapter,
        report_to_hub=report_to_hub,
        write_metrics=True,
    )

    pushed = False
    push_reason = ""
    if push and result.ok:
        push_proc = _git(target_repo, "push", "origin", "HEAD", check=False)
        if push_proc.returncode == 0:
            pushed = True
        else:
            push_reason = (push_proc.stderr or push_proc.stdout or "push failed").strip()[
                :300
            ]
    elif not push:
        push_reason = "push skipped"

    return SmokeResult(run=result, pushed=pushed, push_reason=push_reason)