#!/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())