LangBot/plugins/Runner/certification-tests/verify_coding_candidates.py

"""Freeze unsigned official-SDK builds; genuine b5 dependency + RPC gates.

Run with an absolute evidence directory OUTSIDE the repository after committing.
Direct launcher/loopback fixtures are not nsjail or vendor/live certification.
"""

import argparse
import asyncio
import base64
import csv
import hashlib
import importlib.metadata
import io
import json
import os
import subprocess
import sys
import tarfile
import zipfile
from pathlib import Path

from langbot_plugin.entities.io.context import PluginWorkerPolicy
from langbot_plugin.runtime.plugin.artifact import PluginArtifactStore
from langbot_plugin.runtime.plugin.dependency_environment import PluginDependencyEnvironmentStore
from langbot_plugin.runtime.plugin.worker_launcher import PluginWorkerLauncher

ROOT = Path(__file__).resolve().parents[2]
FOLDERS = ("acp-agent-runner", "claude-code-agent", "codex-agent")


async def verify(out):
    assert not out.exists(), "Use a fresh evidence directory; never overwrite frozen candidates"
    assert importlib.metadata.version("langbot-plugin") == "0.7.4"
    assert not subprocess.check_output(["git", "status", "--porcelain"], cwd=ROOT).strip(), "Commit before freezing"
    commit = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip()
    out.mkdir(parents=True)
    dist = importlib.metadata.distribution("langbot-plugin")
    verified = []
    for name, encoded, _ in csv.reader(io.StringIO(dist.read_text("RECORD"))):
        if name.startswith("langbot_plugin/") and name.endswith(".py"):
            digest = hashlib.sha256(Path(dist.locate_file(name)).read_bytes()).digest()
            assert encoded == "sha256=" + base64.urlsafe_b64encode(digest).decode().rstrip("=")
            verified.append(name)
    (out / "sdk-record.json").write_text(
        json.dumps(
            dict(sdk=dist.version, verified_python_files=len(verified), python=sys.version, source_commit=commit),
            indent=2,
        )
    )
    artifacts = PluginArtifactStore(out / "runtime")
    deps = PluginDependencyEnvironmentStore(out / "runtime")
    launcher = PluginWorkerLauncher()
    launcher.configure(
        PluginWorkerPolicy(
            max_cpus=1,
            max_memory_mb=512,
            max_pids=128,
            max_open_files=256,
            max_file_size_mb=512,
            require_hard_limits=False,
        ),
        "oss_dev",
    )
    inventory = []
    for folder in FOLDERS:
        stage = out / "build-source"
        stage.mkdir(exist_ok=True)
        archive = subprocess.check_output(["git", "archive", commit, "Runner/" + folder], cwd=ROOT)
        with tarfile.open(fileobj=io.BytesIO(archive)) as tar:
            tar.extractall(stage, filter="data")
        destination = out / "packages" / folder
        destination.mkdir(parents=True)
        result = subprocess.run(
            [sys.executable, "-m", "langbot_plugin.cli.__init__", "build", "-o", str(destination)],
            cwd=stage / "Runner" / folder,
            text=True,
            capture_output=True,
            timeout=60,
        )
        (out / (folder + "-build.log")).write_text(result.stdout + result.stderr)
        assert result.returncode == 0, result.stdout + result.stderr
        packages = list(destination.glob("*.lbpkg"))
        assert len(packages) == 1
        package = packages[0]
        data = package.read_bytes()
        digest = hashlib.sha256(data).hexdigest()
        with zipfile.ZipFile(package) as z:
            assert not z.comment
            files = {p: hashlib.sha256(z.read(p)).hexdigest() for p in z.namelist() if not p.endswith("/")}
            for path in files:
                if path != "manifest.yaml":
                    assert z.read(path) == (stage / "Runner" / folder / path).read_bytes(), path
        artifact = artifacts.install_package(data, digest)
        assert artifacts.require_verified(digest) == artifact
        env = await launcher.prepare_dependency_environment(deps, artifact)
        assert deps.get_ready(env.digest) == env
        assert await launcher.prepare_dependency_environment(deps, artifact) == env
        assert not list(env.site_packages_path.glob("langbot_plugin*"))
        link = out / "extracted" / folder
        link.parent.mkdir(exist_ok=True)
        link.symlink_to(artifact.code_path, target_is_directory=True)
        row = dict(
            folder=folder,
            author=artifact.plugin_author,
            name=artifact.plugin_name,
            version=artifact.plugin_version,
            package=str(package),
            sha256=digest,
            bytes=len(data),
            source_commit=commit,
            files=files,
            unsigned=True,
            dependency_prepare_passed=True,
            dependency_cache_readback_passed=True,
            sdk_shadow_installed=False,
            code_path=str(artifact.code_path),
            dependency_path=str(env.site_packages_path),
        )
        inventory.append(row)
        (out / "candidate-inventory.json").write_text(json.dumps(inventory, indent=2))
        home = out / "fixture-homes" / folder
        home.mkdir(parents=True)
        child_env = {
            k: v
            for k, v in os.environ.items()
            if not any(s in k.upper() for s in ("TOKEN", "SECRET", "KEY", "PROXY", "DAEMON", "CODEX", "ANTHROPIC"))
        }
        child_env.update(
            HOME=str(home),
            LANGBOT_PLUGIN_FILE_STORAGE_DIR=str(home / "file-transfer"),
            PYTHONDONTWRITEBYTECODE="1",
            PYTHONPATH=os.pathsep.join([str(dist.locate_file("")), str(env.site_packages_path)]),
        )
        probe = Path(__file__).with_name("coding_artifact_rpc_probe.py")
        result = subprocess.run(
            [sys.executable, str(probe), folder, str(out / (folder + "-rpc.json"))],
            cwd=artifact.code_path,
            env=child_env,
            text=True,
            capture_output=True,
            timeout=90,
        )
        (out / (folder + "-rpc.log")).write_text(result.stdout + result.stderr)
        assert result.returncode == 0, result.stdout + result.stderr
        evidence = json.loads((out / (folder + "-rpc.json")).read_text())
        assert evidence["status"] == "initialized"
        assert evidence["invalid_results"][0]["type"] == "run.failed"
        assert evidence["fixture_results"][-1]["type"] == "run.completed"
        assert "FIXTURE_OK" in json.dumps(evidence["fixture_results"])
        assert not any(e["type"] == "run.failed" for e in evidence["fixture_results"])
        row["real_sdk_discovery_rpc_fixture_passed"] = True
        (out / "candidate-inventory.json").write_text(json.dumps(inventory, indent=2))
        print(folder, artifact.plugin_version, digest, flush=True)
    assert len(inventory) == 3


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--evidence", type=Path, required=True)
    asyncio.run(verify(parser.parse_args().evidence.resolve()))