aitbc/scripts/qa-cycle.py

#!/usr/bin/env python3
import os, sys, json, subprocess, tempfile, shutil, glob, re
import random, time, logging
from datetime import datetime

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)

AGENT = os.getenv("AGENT_NAME", "aitbc")
LOG_DIR = "/var/log/aitbc"
QA_LOG = os.path.join(LOG_DIR, "qa_cycle.log")

def run_cmd(cmd, capture=True, check=False):
    result = subprocess.run(cmd, shell=True, capture_output=capture, text=True)
    if check and result.returncode != 0:
        logger.error(f"Command failed: {cmd}")
        logger.error(f"stderr: {result.stderr}")
        return None
    return result.stdout if capture else result.returncode

def feature_tests():
    logger.info("=== FEATURE TESTS ===")
    for mod in ["aiohttp", "fastapi", "click", "pydantic"]:
        cmd = f"python3 -c 'import {mod}; print({mod}.__version__)'"
        out = run_cmd(cmd)
        if out is None:
            logger.error(f"Import failed: {mod}")
        else:
            logger.info(f"{mod} import OK: {out.strip()}")
    out = run_cmd("/opt/aitbc/cli_venv/bin/aitbc --help")
    if out:
        logger.info("CLI help works")
    else:
        logger.error("CLI help failed")
    out = run_cmd("curl -s http://localhost:8010/health || true")
    if out and 'ok' in out.lower():
        logger.info("Brother chain health endpoint OK")
    else:
        logger.warning("Brother chain health check inconclusive")

def bug_sentinel():
    logger.info("=== BUG SENTINEL ===")
    logs = glob.glob("/var/log/aitbc/*.log")
    patterns = ["ERROR", "CRITICAL", "FATAL", "Traceback", "Exception"]
    found = []
    for logfile in logs:
        try:
            with open(logfile) as f:
                lines = f.readlines()[-200:]
        except:
            continue
        for line in lines:
            if any(p in line for p in patterns):
                found.append(f"{logfile}: {line.strip()}")
    if found:
        logger.warning(f"Found {len(found)} error patterns (sample):")
        for item in found[:5]:
            logger.warning(f"  {item}")
    else:
        logger.info("No error patterns in recent logs")

def code_review():
    logger.info("=== CODE REVIEW ===")
    py_files = run_cmd("find /opt/aitbc -name '*.py' -type f | head -30").splitlines()
    issues = []
    for f in py_files:
        try:
            with open(f) as fp:
                content = fp.read()
        except:
            continue
        if re.search(r'except\s*:', content):
            issues.append(f"{f}: bare except")
        if re.search(r'def\s+\w+\s*\([^)]*=\s*[\[\{\}]', content):
            issues.append(f"{f}: mutable default argument")
        if 'print(' in content and 'if __name__' not in content:
            issues.append(f"{f}: print statement in library code")
        if re.search(r'(password|secret|key)\s*=\s*[\'"][^\'"]+[\'"]', content, re.IGNORECASE):
            issues.append(f"{f}: possible hardcoded secret")
    if issues:
        logger.warning(f"Found {len(issues)} code quality issues (sample):")
        for i in issues[:5]:
            logger.warning(f"  {i}")
    else:
        logger.info("No obvious code quality issues detected")

def scenario_runner():
    logger.info("=== SCENARIO RUNNER ===")
    tmp = tempfile.mkdtemp(prefix="aitbc_qa_")
    try:
        run_cmd(f"git init {tmp}", check=True)
        run_cmd(f"git -C {tmp} config user.email 'qa@aitbc'", check=True)
        run_cmd(f"git -C {tmp} config user.name 'QA Agent'", check=True)
        run_cmd(f"echo 'test' > {tmp}/test.txt", check=True)
        run_cmd(f"git -C {tmp} add .", check=True)
        run_cmd(f"git -C {tmp} commit -m 'initial'", check=True)
        run_cmd(f"git -C {tmp} checkout -b claim/123", check=True)
        run_cmd(f"git -C {tmp} commit --allow-empty -m 'Claim issue 123'", check=True)
        logger.info("Git workflow test passed")
    except Exception as e:
        logger.error(f"Git workflow test failed: {e}")
    finally:
        shutil.rmtree(tmp, ignore_errors=True)

def main():
    logger.info(f"QA CYCLE STARTED for {AGENT}")
    start = datetime.now()
    # Jitter: 0-900 seconds (15 minutes)
    jitter = random.randint(0, 900)
    logger.info(f"Sleeping {jitter}s before start...")
    time.sleep(jitter)
    feature_tests()
    bug_sentinel()
    code_review()
    scenario_runner()
    elapsed = datetime.now() - start
    logger.info(f"QA CYCLE COMPLETED in {elapsed.total_seconds():.1f}s")

if __name__ == "__main__":
    main()