teleo-infrastructure/.agents/skills/teleo-leo-onboarding/scripts/run_clean_context_canary.py

#!/usr/bin/env python3
"""Run the deterministic T2 isolated-install acceptance canary for the skill pack."""

from __future__ import annotations

import argparse
import json
import subprocess
import sys
import tempfile
from pathlib import Path

DEFAULT_MANIFEST = Path("docs/reports/leo-working-state-20260709/skill-pack-manifest.json")

SEMANTIC_CONTRACT = {
    "authority": (
        "teleo-infra-provenance",
        "Canonical code/deployment repository: GitHub `living-ip/teleo-infrastructure`.",
        "GitHub living-ip/teleo-infrastructure",
    ),
    "vps_database": (
        "teleo-leo-onboarding",
        "VPS canonical KB: Docker Postgres `teleo-pg`, database `teleo`.",
        "teleo-pg / teleo",
    ),
    "gcp_database": (
        "teleo-leo-onboarding",
        "GCP canonical KB: private Cloud SQL database `teleo_canonical`.",
        "private Cloud SQL teleo_canonical; persistent copy remains staging",
    ),
    "live_service": (
        "teleo-infra-provenance",
        "Live service: `leoclean-gateway.service`",
        "leoclean-gateway.service",
    ),
    "merged_reconstruction": (
        "teleo-reconstruction-recovery",
        "PR #146 is merged",
        "PR #146 merged deterministic genesis-plus-ledger reconstruction",
    ),
    "merged_reasoning_verifier": (
        "teleo-reconstruction-recovery",
        "PR #147 is merged",
        "PR #147 merged unseen-reasoning verifier hardening",
    ),
    "candidate_only_gcp": (
        "teleo-reconstruction-recovery",
        "PR #148 is a separate, open GCP least-privilege candidate",
        "PR #148 open candidate only; not canonical main",
    ),
    "next_safe_action": (
        "teleo-leo-onboarding",
        ".agents/skills/teleo-leo-onboarding/scripts/validate_skill_pack.py --root .",
        "run the repo-local path validator before any operational action",
    ),
}


def _run(command: list[str]) -> subprocess.CompletedProcess[str]:
    return subprocess.run(command, check=False, capture_output=True, text=True)


def _contains_contract(text: str, required_text: str) -> bool:
    """Compare prose semantically across ordinary Markdown line wrapping."""

    return " ".join(required_text.split()) in " ".join(text.split())


def run_canary(root: Path, manifest_path: Path) -> dict[str, object]:
    manifest = json.loads((root / manifest_path).read_text(encoding="utf-8"))
    installer = root / manifest["installer"]
    selected_skills = sorted({contract[0] for contract in SEMANTIC_CONTRACT.values()})
    result: dict[str, object]

    with tempfile.TemporaryDirectory(prefix="teleo-clean-context-") as temporary_text:
        temporary = Path(temporary_text)
        installed_root = temporary / "skills"
        install_process = _run(
            [str(installer), "--root", str(root), "--manifest", str(manifest_path), "--target", str(installed_root)]
        )
        try:
            install_receipt = json.loads(install_process.stdout)
        except json.JSONDecodeError:
            install_receipt = {"status": "fail", "error": install_process.stdout or install_process.stderr}

        semantic_checks: dict[str, bool] = {}
        answers: dict[str, str] = {}
        evidence_paths: dict[str, str] = {}
        for key, (skill_name, required_text, answer) in SEMANTIC_CONTRACT.items():
            installed_skill = installed_root / skill_name / "SKILL.md"
            text = installed_skill.read_text(encoding="utf-8") if installed_skill.is_file() else ""
            semantic_checks[key] = _contains_contract(text, required_text)
            answers[key] = answer
            evidence_paths[key] = f".agents/skills/{skill_name}/SKILL.md"

        installed_validator = installed_root / "teleo-leo-onboarding" / "scripts" / "validate_skill_pack.py"
        validator_process = _run([str(installed_validator), "--root", str(root), "--manifest", str(manifest_path)])
        try:
            validator_receipt = json.loads(validator_process.stdout)
        except json.JSONDecodeError:
            validator_receipt = {"status": "fail", "error": validator_process.stdout or validator_process.stderr}

        problems: list[dict[str, str]] = []
        if install_process.returncode != 0 or install_receipt.get("status") != "pass":
            problems.append({"kind": "install_failed", "detail": str(install_receipt.get("error", "unknown"))})
        if validator_process.returncode != 0 or validator_receipt.get("status") != "pass":
            problems.append({"kind": "path_validation_failed", "detail": str(validator_receipt.get("problems", []))})
        for key, passed in semantic_checks.items():
            if not passed:
                problems.append({"kind": "semantic_route_missing", "detail": key})

        passed = not problems
        result = {
            "artifact": "leo_teleo_skill_pack_isolated_install_canary",
            "status": "pass" if passed else "fail",
            "required_tier": "T2_runtime",
            "current_tier": "T2_runtime" if passed else "T1_model",
            "proof_scope": "deterministic_isolated_install_and_route_validation",
            "fresh_temporary_skill_root": True,
            "ambient_skill_root_used": False,
            "agent_reasoning_exercised": False,
            "installed_skill_count": install_receipt.get("installed_skill_count", 0),
            "selected_installed_skills": selected_skills,
            "semantic_checks": semantic_checks,
            "expected_answers": answers,
            "evidence_paths": evidence_paths,
            "canary_command": ".agents/skills/teleo-leo-onboarding/scripts/validate_skill_pack.py --root .",
            "canary_exit_status": validator_process.returncode,
            "canary_receipt": validator_receipt,
            "missing_to_reach_required_tier": [] if passed else problems,
            "strongest_claim_allowed": (
                "T2 deterministic isolated local skill install, route-contract, and path validation only; this artifact "
                "does not claim an agent reasoning run, and retained product proofs keep their own VPS, GCP-staging, "
                "isolated-clone, Telegram, and production claim ceilings"
            ),
            "contains_secrets": False,
            "production_mutation_authorized": False,
            "external_runtime_contacted": False,
            "problems": problems,
        }

    result["cleanup_readback"] = {
        "temporary_skill_root_removed": not Path(temporary_text).exists(),
        "orphan_processes_started": False,
    }
    if not result["cleanup_readback"]["temporary_skill_root_removed"]:
        result["status"] = "fail"
        result["current_tier"] = "T1_model"
        result["problems"].append({"kind": "cleanup_failed", "detail": "temporary skill root remains"})
    return result


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--root", type=Path, default=Path.cwd())
    parser.add_argument("--manifest", type=Path, default=DEFAULT_MANIFEST)
    parser.add_argument("--output", type=Path)
    args = parser.parse_args()

    root = args.root.resolve()
    result = run_canary(root, args.manifest)
    payload = json.dumps(result, indent=2, sort_keys=True) + "\n"
    if args.output:
        output = args.output if args.output.is_absolute() else root / args.output
        output.parent.mkdir(parents=True, exist_ok=True)
        output.write_text(payload, encoding="utf-8")
    sys.stdout.write(payload)
    return 0 if result["status"] == "pass" else 1


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