pi-agent-config/shared/lib/py/tests/test_pi_rpc_smoke.py

"""Zero-token smoke test for pi_rpc against a real pi process.

Sends only `get_state`, so no model call is made. Verifies that the launch
contract produces a live process, that the isolation flags land, and that the
process group is cleaned up.

Run:  python3 shared/lib/py/tests/test_pi_rpc_smoke.py
"""
from __future__ import annotations

import asyncio
import os
import signal
import os
import subprocess
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from dataclasses import replace  # noqa: E402
import pathlib  # noqa: E402
import tempfile  # noqa: E402

from pi_rpc import PiLaunchConfig, PiRpcClient, PiRpcError  # noqa: E402

REPO = Path(__file__).resolve().parents[4]
HARNESS = REPO / "docs" / "evidence" / "probe-harness"
PI_BIN = os.environ.get("PI_BIN", "/home/claw/.npm-global/bin/pi")


def check(label: str, ok: bool, detail: str = "") -> bool:
    print(f"  {'PASS' if ok else 'FAIL'}  {label}{(' — ' + detail) if detail else ''}")
    return ok


async def main() -> int:
    failures = 0

    print("== build_args reflects the isolation baseline ==")
    cfg = PiLaunchConfig(
        pi_bin=PI_BIN,
        workspace=HARNESS,
        session_dir=Path("/tmp/pi-rpc-smoke-sessions"),
        provider="zenmux",
        model="openai/gpt-5.6-luna",
        thinking="low",
        display_name="pi_rpc smoke",
        extensions=(HARNESS / "probe-ext.ts",),
        skills=(HARNESS / ".pi" / "skills" / "probe-skill",),
        session_id_prefix="",  # ephemeral for the smoke test
        receipt_tools=frozenset({"probe_plain_schema"}),
        turn_deadline_seconds=45.0,
    )
    args = cfg.build_args(None)
    for flag in ("--no-builtin-tools", "--no-extensions", "--no-skills",
                 "--no-prompt-templates", "--no-themes", "--approve", "--no-session"):
        failures += not check(f"{flag} present", flag in args)
    failures += not check("--tools absent (registry allowlist would block dynamic tools)",
                          "--tools" not in args)

    print("== system prompt and the skills/read interaction ==")
    sp = pathlib.Path(tempfile.mkdtemp()) / "SYSTEM.md"
    sp.write_text("marker", encoding="utf-8")
    withprompt = replace(cfg, system_prompt=sp, append_system_prompt=sp)
    prompt_args = withprompt.build_args(None)
    failures += not check("--system-prompt passed through", "--system-prompt" in prompt_args)
    failures += not check("--append-system-prompt passed through",
                          "--append-system-prompt" in prompt_args)

    # A dedicated agent whose tools are all domain-specific has no 'read', and pi
    # then discards every --skill argument silently. Measured, not assumed.
    noread = replace(cfg, skills=(sp.parent,), no_builtin_tools=True,
                     extension_registers_read=False)
    failures += not check("skills without a read tool are flagged as ineffective",
                          not noread._read_reachable())
    failures += not check("an extension that registers read is trusted",
                          replace(noread, extension_registers_read=True)._read_reachable())
    failures += not check("an explicit allowlist naming read counts",
                          replace(noread, tools=("read", "counts"))._read_reachable())
    failures += not check("an allowlist without read does not",
                          not replace(noread, tools=("counts",))._read_reachable())

    print("== env is minimal ==")
    os.environ["SMOKE_FAKE_SECRET"] = "must-not-propagate"
    env = cfg.build_env()
    failures += not check("secret withheld", "SMOKE_FAKE_SECRET" not in env)
    failures += not check("PATH forwarded", "PATH" in env)
    failures += not check(f"small env ({len(env)} vars)", len(env) < 20)

    print("== live process: start, get_state, stop ==")
    client = PiRpcClient(cfg)
    try:
        await client.start()
        pid = client.pid
        failures += not check("process running", client.running, f"pid={pid}")
        state = await client.get_state()
        failures += not check("get_state returned a model", bool(state.get("model")),
                              str(state.get("model", {}).get("id", "?")))
        failures += not check("not streaming", state.get("isStreaming") is False)
        failures += not check("thinkingLevel honoured", state.get("thinkingLevel") == "low",
                              str(state.get("thinkingLevel")))
        stats = await client.get_session_stats()
        failures += not check("get_session_stats works", "tokens" in stats)
    finally:
        await client.stop()

    failures += not check("process stopped", not client.running)
    if pid:
        alive = subprocess.run(["ps", "-p", str(pid)], capture_output=True).returncode == 0
        failures += not check("no orphan process", not alive)

    # A process that dies mid-command must be reported immediately, not at the
    # turn deadline. Before the sentinel this sat on the event queue for the full
    # 180 s budget; a fallback that launches a fresh pi then hid the death.
    print("== process exit is detected immediately ==")
    client2 = PiRpcClient(cfg)
    await client2.start()
    import time as _time
    try:
        os.killpg(os.getpgid(client2.pid), signal.SIGKILL)
        started = _time.monotonic()
        try:
            await asyncio.wait_for(client2.get_session_stats(), timeout=10)
            failures += not check("dead process reported", False, "command succeeded after SIGKILL")
        except PiRpcError:
            elapsed = _time.monotonic() - started
            failures += not check("dead process reported promptly", elapsed < 10,
                                  f"{elapsed:.1f}s (must be well under the 180 s budget)")
    finally:
        try:
            await client2.stop()
        except Exception:
            pass

    failures += not check("no orphan from killed client", not client2.running)

    print()
    print("RESULT:", "all checks passed" if failures == 0 else f"{failures} check(s) failed")
    return 1 if failures else 0


if __name__ == "__main__":
    raise SystemExit(asyncio.run(main()))