MarketingTool/scripts/final_acceptance.py

#!/usr/bin/env python3
"""Run the deterministic, local Phase 17 final-acceptance gate.

This is a release-evidence collector, not a deployment tool. It performs only
local tests and static/configuration checks; it never sends outreach or calls a
provider. Capacity figures are bounded synthetic smoke measurements, not
production capacity claims.
"""
from __future__ import annotations

import argparse
import json
import re
import shutil
import subprocess
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
API = ROOT / "apps" / "api"
REPORT = ROOT / "docs" / "benchmarks" / "final_acceptance.latest.json"
SCHEMA = ROOT / "docs" / "benchmarks" / "final_acceptance.schema.json"


def deterministic_json(value: object) -> str:
    return json.dumps(value, indent=2, sort_keys=True, separators=(",", ": ")) + "\n"


def build_capacity_smoke() -> dict:
    """Measure fixed, in-memory bounds with no clock or network dependence."""
    total = 1000
    requested = 100
    page = list(range(total))[0:requested]
    batch_input = 5000
    retained = min(batch_input, 100)
    return {
        "bounded": len(page) <= requested and retained <= 100,
        "limits": {"max_page_size": 100, "max_batch_items_retained": 100},
        "pagination": {
            "synthetic_total": total,
            "requested_page_size": requested,
            "returned_items": len(page),
            "has_more": total > requested,
        },
        "large_batch": {
            "input_items": batch_input,
            "retained_items": retained,
            "truncated": batch_input > retained,
        },
        "interpretation": "Deterministic in-memory smoke only; not a production capacity or throughput claim.",
    }


def validate_report(report: dict) -> list[str]:
    required = {"report", "version", "scope", "passed", "checks", "capacity_smoke", "blockers", "limitations"}
    errors = [f"missing:{key}" for key in sorted(required - set(report))]
    if report.get("report") != "phase17-final-acceptance": errors.append("report:const")
    if type(report.get("version")) is not int or report.get("version") != 1: errors.append("version:type-or-const")
    if report.get("scope") != "local-repository-acceptance": errors.append("scope:const")
    if not isinstance(report.get("passed"), bool): errors.append("passed:type")
    if not isinstance(report.get("checks"), dict): errors.append("checks:type")
    if not isinstance(report.get("capacity_smoke"), dict): errors.append("capacity_smoke:type")
    if not isinstance(report.get("limitations"), list) or not all(isinstance(x, str) for x in report.get("limitations", [])):
        errors.append("limitations:type")
    blockers = report.get("blockers", [])
    if not isinstance(blockers, list) or not all(isinstance(x, dict) for x in blockers):
        errors.append("blockers:type")
    else:
        for index, blocker in enumerate(blockers):
            if not {"gate", "status", "reason"}.issubset(blocker): errors.append(f"blockers[{index}]:required")
    return errors


def run_command(name: str, args: list[str], *, cwd: Path = ROOT, timeout: int = 300) -> dict:
    if shutil.which(args[0]) is None:
        return {"passed": False, "status": "unavailable", "command": args, "summary": f"{args[0]} not installed"}
    try:
        completed = subprocess.run(args, cwd=cwd, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=timeout)
    except subprocess.TimeoutExpired:
        return {"passed": False, "status": "timeout", "command": args, "summary": f"{name} exceeded {timeout}s"}
    output = completed.stdout or ""
    match = re.search(r"Ran (\d+) tests?", output)
    summary = f"exit={completed.returncode}"
    if match: summary += f", tests={match.group(1)}"
    if completed.returncode and output:
        summary += ": " + " ".join(output.strip().splitlines()[-2:])[:500]
    return {"passed": completed.returncode == 0, "status": "passed" if completed.returncode == 0 else "failed", "command": args, "summary": summary}


def validate_json_files() -> dict:
    files = [ROOT / "docs" / "benchmarks" / "phase16.latest.json", ROOT / "docs" / "benchmarks" / "phase16.schema.json", SCHEMA]
    results = {}
    parsed = {}
    for path in files:
        key = str(path.relative_to(ROOT))
        try:
            parsed[key] = json.loads(path.read_text(encoding="utf-8"))
            results[key] = {"passed": True, "status": "valid_json"}
        except (OSError, json.JSONDecodeError) as exc:
            results[key] = {"passed": False, "status": "invalid_json", "summary": str(exc)}
    phase16 = parsed.get("docs/benchmarks/phase16.latest.json")
    if isinstance(phase16, dict):
        required = {"benchmark", "version", "offline", "limitations", "acceptance_thresholds", "results", "checks", "passed"}
        schema_ok = required.issubset(phase16) and phase16.get("benchmark") == "phase16" and phase16.get("version") == 1 and phase16.get("offline") is True and isinstance(phase16.get("passed"), bool)
        results["phase16_schema_contract"] = {"passed": schema_ok, "status": "schema_valid" if schema_ok else "schema_invalid"}
    else:
        results["phase16_schema_contract"] = {"passed": False, "status": "schema_invalid"}
    return {"passed": all(item["passed"] for item in results.values()), "files": results}


def safety_checks() -> dict:
    compose = (ROOT / "docker-compose.yml").read_text(encoding="utf-8")
    config = (API / "app" / "config.py").read_text(encoding="utf-8")
    main = (API / "app" / "main.py").read_text(encoding="utf-8")
    tests = "\n".join(p.read_text(encoding="utf-8") for p in (API / "tests").glob("test_*.py"))
    checks = {
        "outreach_disabled": 'AUTOMATED_OUTREACH_ENABLED: "false"' in compose and '"false"' in config and 'outreach_enabled' in main,
        "no_send_network": '"network_send": False' in main and "urllib.request" not in main and "smtplib" not in main,
        "tenant_routes": "organization_id=?" in main and "session_user" in main and "require_auth" in main and "tenant" in tests.lower(),
    }
    return {"passed": all(checks.values()), "checks": checks}


def git_state() -> dict:
    result = run_command("git-state", ["git", "status", "--short", "--branch"])
    head = subprocess.run(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True, capture_output=True, check=False)
    porcelain = subprocess.run(["git", "status", "--porcelain"], cwd=ROOT, text=True, capture_output=True, check=False)
    return {
        "passed": result["passed"] and head.returncode == 0 and bool(head.stdout.strip()),
        "status": result["status"],
        "branch_status": result.get("summary", ""),
        "head": head.stdout.strip(),
        "working_tree_clean": porcelain.returncode == 0 and not porcelain.stdout.strip(),
    }


def build_report() -> dict:
    checks = {
        "api_tests": run_command("api-tests", [sys.executable, "-m", "unittest", "discover", "-v", "-s", "apps/api/tests", "-t", "apps/api"]),
        "py_compile": run_command("py_compile", [sys.executable, "-m", "py_compile", *[str(p.relative_to(ROOT)) for p in sorted((ROOT / "scripts").glob("*.py"))], *[str(p.relative_to(ROOT)) for p in sorted((API / "app").glob("*.py"))]]),
        "benchmark_no_latency": run_command("benchmark-no-latency", [sys.executable, "scripts/benchmark_phase16.py", "--no-latency", "--output", "/tmp/prospect-phase16-acceptance.json"]),
        "shell_syntax": run_command("shell-syntax", ["bash", "-n", *[str(p.relative_to(ROOT)) for p in sorted((ROOT / "scripts").glob("*.sh"))]]),
        "json_validation": validate_json_files(),
        "compose_config": run_command("compose-config", ["docker", "compose", "-f", "docker-compose.yml", "config", "--quiet"]),
        "git_state": git_state(),
        "safety_invariants": safety_checks(),
    }
    blockers = [
        {"gate": "remote_auth", "status": "blocked", "reason": "Remote repository authentication and branch permission are not available to this local acceptance run."},
        {"gate": "deployment_access", "status": "blocked", "reason": "No production host, Docker/Compose, DNS/TLS, secrets, or deployment access is available; no deployment was attempted."},
        {"gate": "real_provider_legal", "status": "blocked", "reason": "Real source/provider enablement, consent/legal basis, terms, and operational approval remain explicit gates; outreach stays disabled."},
    ]
    return {
        "report": "phase17-final-acceptance", "version": 1,
        "scope": "local-repository-acceptance",
        "passed": all(item.get("passed", False) for item in checks.values()),
        "checks": checks,
        "capacity_smoke": build_capacity_smoke(),
        "blockers": blockers,
        "limitations": ["Synthetic capacity smoke measurements do not establish production throughput, concurrency, durability, or availability."],
    }


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--output", type=Path, default=REPORT)
    args = parser.parse_args()
    report = build_report()
    errors = validate_report(report)
    if errors: raise SystemExit("invalid acceptance report: " + ", ".join(errors))
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(deterministic_json(report), encoding="utf-8")
    print(deterministic_json(report), end="")
    return 0 if report["passed"] else 1


if __name__ == "__main__":
    raise SystemExit(main())