scripts.agent_check

Reference › scripts › agent_check

The host-side quality gate for work done by an agent inside an Incus container. It pairs with agent_notify, which the agent runs inside the container when it finishes. A human or host automation runs agent_check; no module or test imports it. It uses resolve_redis_host and resolve_redis_port from redis_factory.

Invocation: --task is required and names the task. --container selects the Incus container (default claude-orchestrata). --listen waits for the done notification on Redis before checking, and --pull skips the wait and checks immediately. --staging (default /tmp/staging) is the host directory that receives the code, and --review-script (default scripts/code-review.py) is the review tool to run. --redis-host, --redis-port and --redis-db override the connection; the database defaults to the REDIS_DB environment variable or 0.

Flow: get_redis_client connects, wait_for_done blocks (default timeout 3600 seconds) until the agent's notification arrives, pull_code copies the container's work into the staging directory, and run_checks runs four checks on it: ruff, mypy, the review script and pytest. generate_report formats the results as Markdown, and main writes it to <staging>/<task>-report.md and prints the path.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/agent_check.py

#!/usr/bin/env python3
"""Agent check script — run ON THE HOST.

Listens for Redis notification from the Incus container, pulls the code,
runs the full quality gate suite, and produces a Claude-facing report.

Usage:
    python agent_check.py --listen --task TASK_ID --container claude-orchestrata
    python agent_check.py --pull --task TASK_ID --container claude-orchestrata
"""

import argparse
import json
import os
import re
import shutil
import subprocess
import sys
import time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

from orchestrata_ultimate.core.redis_factory import (
    resolve_redis_host,
    resolve_redis_port,
)

# Children's output is machine-parsed (regex/substring greps): force color
# OFF for every subprocess this tool spawns. An ambient FORCE_COLOR (e.g.
# exported by the launching terminal/agent harness) made mypy/ruff/pytest
# emit ANSI into captured pipes and silently broke the parsers (gate
# 2026-08-05: mypy "0 issues" on a dirty tree, ruff int(line) crash,
# FAILED-line greps missing their lines).
os.environ["NO_COLOR"] = "1"  # gate-ok: process-bootstrap env routing
os.environ.pop("FORCE_COLOR", None)  # gate-ok: process-bootstrap env routing


def get_redis_client(host: str, port: int, db: int):
    try:
        import redis
    except ImportError:
        print("ERROR: redis package not installed.", file=sys.stderr)
        sys.exit(1)
    return redis.Redis(host=host, port=port, db=db, decode_responses=True)


def wait_for_done(redis_client, task: str, timeout: int = 3600) -> dict:
    """Block until agent sends done notification."""
    channel = f"agent:done:{task}"
    pubsub = redis_client.pubsub()
    pubsub.subscribe(channel)

    print(f"Listening on Redis channel '{channel}'...")
    start = time.time()

    # Cap each poll by the time remaining, not a flat 5s: pubsub.get_message()
    # blocks for its *entire* timeout waiting for a message, so a caller-supplied
    # small `timeout` (e.g. tests exercising fast expiry) was previously stuck
    # inside one 5s poll regardless of the outer deadline.
    while True:
        remaining = timeout - (time.time() - start)
        if remaining <= 0:
            break
        message = pubsub.get_message(timeout=min(5, remaining))
        if message and message["type"] == "message":
            payload = json.loads(message["data"])
            print(f"Received: {payload}")
            return payload

    print("TIMEOUT: No notification received.", file=sys.stderr)
    sys.exit(1)


def pull_code(container: str, task: str, staging: Path) -> Path:
    """Pull code from Incus container to staging directory."""
    src_in_container = f"/home/mohamed/workspace/{task}"
    dest = staging / task

    if dest.exists():
        shutil.rmtree(dest)
    dest.mkdir(parents=True)

    # Use incus file pull to recursively copy
    cmd = [
        "incus",
        "exec",
        container,
        "--",
        "bash",
        "-c",
        f"cd {src_in_container} && tar czf - .",
    ]
    print(f"Pulling code from {container}:{src_in_container} -> {dest}")
    proc = subprocess.Popen(cmd, stdout=subprocess.PIPE)
    assert proc.stdout is not None  # guaranteed by stdout=subprocess.PIPE above
    tar = subprocess.run(["tar", "xzf", "-", "-C", str(dest)], stdin=proc.stdout)
    proc.stdout.close()
    proc.wait()

    if proc.returncode != 0 or tar.returncode != 0:
        print("ERROR: Failed to pull code from container.", file=sys.stderr)
        sys.exit(1)

    print(f"Code staged at {dest}")
    return dest


def run_checks(staged_path: Path, review_script: Path) -> dict:
    """Run the full quality gate suite against staged code."""
    results = {
        "ruff": {},
        "mypy": {},
        "review": {},
        "pytest": {},
    }

    src_path = staged_path / "src"
    tests_path = staged_path / "tests"

    def _ruff() -> None:
        ruff = subprocess.run(
            [
                sys.executable,
                "-m",
                "ruff",
                "check",
                str(src_path),
                str(tests_path),
                "--output-format=json",
            ],
            capture_output=True,
            text=True,
        )
        try:
            results["ruff"]["issues"] = json.loads(ruff.stdout) if ruff.stdout else []
        except json.JSONDecodeError:
            results["ruff"]["issues"] = []
        results["ruff"]["returncode"] = ruff.returncode

    def _mypy() -> None:
        # NOTE: mypy has no built-in JSON output mode (no --output-format flag),
        # unlike ruff. Parse its plain-text "file:line: severity: message" lines
        # instead (a prior version passed --output-format=json, which mypy
        # rejects as an unrecognized argument — it silently produced an empty
        # stdout, so real mypy errors were never surfaced here).
        # Invoke mypy as `python -m mypy` (not the bare "mypy" binary name): the
        # caller's PATH is not guaranteed to include the venv's bin/ (e.g. when
        # this script is spawned as a subprocess whose PATH was inherited from a
        # shell that never activated the venv) — sys.executable always resolves
        # to the interpreter this script itself is running under.
        mypy = subprocess.run(
            [
                sys.executable,
                "-m",
                "mypy",
                str(src_path),
                "--no-error-summary",
                "--no-color-output",
            ],
            capture_output=True,
            text=True,
        )
        mypy_issue_re = re.compile(
            r"^(?P<file>[^:]+):(?P<line>\d+):\s*(?P<severity>\w+):\s*(?P<message>.*)$"
        )
        mypy_issues = []
        for line in mypy.stdout.splitlines():
            match = mypy_issue_re.match(line)
            if match and match.group("severity") == "error":
                mypy_issues.append(
                    {
                        "file": match.group("file"),
                        "line": int(match.group("line")),
                        "message": match.group("message"),
                    }
                )
        results["mypy"]["issues"] = mypy_issues
        results["mypy"]["returncode"] = mypy.returncode

    def _review() -> None:
        # Unified review script (from HOST, not container)
        report_path = staged_path / "review-report.json"
        subprocess.run(
            [
                sys.executable,
                str(review_script),
                str(src_path),
                str(tests_path),
                "--format",
                "json",
                "--output",
                str(report_path),
            ],
            capture_output=True,
            text=True,
        )
        try:
            with open(report_path) as f:
                review_data = json.load(f)
            results["review"] = review_data
        except FileNotFoundError, json.JSONDecodeError:
            results["review"] = {"total_issues": 0, "issues": []}

    def _pytest() -> None:
        # Tests — via the sanctioned threadpool runner. pytest-xdist (-n auto)
        # is BANNED (deadlocks the suite ~78%); run_tests_threadpool.py gives
        # per-module parallelism instead. Use the staged checkout's own copy of
        # the runner so the review reflects that tree.
        # Use THIS host's runner (not the untrusted staged copy) against the staged
        # tree via cwd=staged_path.
        runner = Path(__file__).resolve().parent / "run_tests_threadpool.py"
        pytest = subprocess.run(
            [
                sys.executable,
                str(runner),
                str(tests_path),
                "--pytest-arg=--timeout=120",
                "--pytest-arg=-q",
            ],
            capture_output=True,
            text=True,
            cwd=staged_path,
        )
        results["pytest"]["returncode"] = pytest.returncode
        results["pytest"]["stdout"] = pytest.stdout
        results["pytest"]["stderr"] = pytest.stderr

    with ThreadPoolExecutor(max_workers=4) as pool:
        futures = [
            pool.submit(_ruff),
            pool.submit(_mypy),
            pool.submit(_review),
            pool.submit(_pytest),
        ]
        for fut in futures:
            fut.result()

    print("\n=== Running ruff ===")
    print(f"Ruff: {len(results['ruff']['issues'])} issues")

    print("\n=== Running mypy ===")
    print(f"MyPy: {len(results['mypy']['issues'])} issues")

    print("\n=== Running unified review ===")
    print(f"Review: {results['review'].get('total_issues', 0)} issues")

    print("\n=== Running tests (threadpool runner, no xdist) ===")
    print(f"Pytest: returncode={results['pytest']['returncode']}")

    return results


def generate_report(results: dict, task: str) -> str:
    """Generate a Claude-facing failure report."""
    lines = [f"# Quality Gate Report for Task: {task}", ""]

    # Count errors
    review_issues = results["review"].get("issues", [])
    errors = [i for i in review_issues if i.get("severity") == "ERROR"]
    warnings = [i for i in review_issues if i.get("severity") == "WARNING"]
    ruff_count = len(results["ruff"]["issues"])
    mypy_count = len(results["mypy"]["issues"])
    pytest_rc = results["pytest"]["returncode"]

    lines.append("## Summary")
    lines.append(f"- Review ERRORs: {len(errors)}")
    lines.append(f"- Review WARNINGs: {len(warnings)}")
    lines.append(f"- Ruff issues: {ruff_count}")
    lines.append(f"- MyPy issues: {mypy_count}")
    lines.append(f"- Pytest: {'PASS' if pytest_rc == 0 else 'FAIL'} (rc={pytest_rc})")
    lines.append("")

    if errors:
        lines.append("## CLAUDE_FIX_REQUIRED (ERRORs)")
        for issue in errors[:50]:
            lines.append(
                f"- {issue.get('file', '?')}:{issue.get('line', '?')} "
                f"[{issue.get('code', '?')}] {issue.get('message', '?')}"
            )
        if len(errors) > 50:
            lines.append(f"- ... and {len(errors) - 50} more ERRORs")
        lines.append("")

    if ruff_count:
        lines.append("## RUFF_ISSUES")
        for issue in results["ruff"]["issues"][:20]:
            lines.append(
                f"- {issue.get('filename', '?')}:{issue.get('location', {}).get('row', '?')} "
                f"[{issue.get('code', '?')}] {issue.get('message', '?')}"
            )
        lines.append("")

    if mypy_count:
        lines.append("## MYPY_ISSUES")
        for issue in results["mypy"]["issues"][:20]:
            line = issue.get("line", "?")
            msg = issue.get("message", "?")
            lines.append(f"- {issue.get('file', '?')}:{line} {msg}")
        lines.append("")

    if pytest_rc != 0:
        lines.append("## PYTEST_FAILURES")
        lines.append("```")
        lines.append(
            results["pytest"]["stdout"][-2000:]
            if len(results["pytest"]["stdout"]) > 2000
            else results["pytest"]["stdout"]
        )
        lines.append("```")
        lines.append("")

    # Leaked issues check
    lines.append("## LEAK_CHECK")
    lines.append(
        "Verify no test files were deleted, no # noqa added, no review script modified."
    )
    lines.append("")

    return "\n".join(lines)


def main() -> None:
    parser = argparse.ArgumentParser(description="Agent quality gate checker")
    parser.add_argument("--task", required=True, help="Task identifier")
    parser.add_argument(
        "--container", default="claude-orchestrata", help="Incus container name"
    )
    parser.add_argument(
        "--listen",
        action="store_true",
        help="Wait for Redis notification before checking",
    )
    parser.add_argument(
        "--pull", action="store_true", help="Pull code and run checks immediately"
    )
    parser.add_argument(
        "--staging", default="/tmp/staging", help="Staging directory on host"
    )
    parser.add_argument(
        "--review-script",
        default="scripts/code-review.py",
        help="Path to review script on host",
    )
    parser.add_argument("--redis-host", default=None)
    parser.add_argument("--redis-port", type=int, default=None)
    parser.add_argument(
        "--redis-db", type=int, default=int(os.environ.get("REDIS_DB", "0"))
    )
    args = parser.parse_args()

    redis_host = args.redis_host or resolve_redis_host()
    redis_port = args.redis_port or resolve_redis_port()
    redis_client = get_redis_client(redis_host, redis_port, args.redis_db)

    if args.listen:
        wait_for_done(redis_client, args.task)

    if args.listen or args.pull:
        staged = pull_code(args.container, args.task, Path(args.staging))
        results = run_checks(staged, Path(args.review_script))
        report = generate_report(results, args.task)

        report_path = Path(args.staging) / f"{args.task}-report.md"
        report_path.write_text(report)
        print(f"\nReport written to {report_path}")
        print(report)

        # Exit non-zero if there are errors
        total_errors = len(
            [
                i
                for i in results["review"].get("issues", [])
                if i.get("severity") == "ERROR"
            ]
        )
        total_errors += len(results["ruff"]["issues"])
        total_errors += len(results["mypy"]["issues"])
        if results["pytest"]["returncode"] != 0:
            total_errors += 1

        if total_errors > 0:
            print(f"\n❌ QUALITY GATE FAILED: {total_errors} total issues found.")
            sys.exit(1)
        else:
            print("\n✅ QUALITY GATE PASSED: Zero issues.")


if __name__ == "__main__":
    main()