Kai
b5a29b05e1
Two findings that changed the plan rather than confirming it:
23. Skills require a tool literally named `read`. Curator's tools are all
domain-specific, so every --skill argument was discarded in silence. The
planned split into curator-core / video-arr / books-ingest was inert before
it was written; the policy stays in APPEND_SYSTEM.md. memo-inbox is
unaffected because it registers a restricted `read` override, which is why
the earlier note generalised wrongly from it.
24. A long-lived session is worth far more than the startup it saves: 99.97% of
input read from cache on a continuing conversation against 0% on a new one.
That is what makes the generated tool list necessary rather than merely
tidy -- anything varying at the front of the prompt destroys it -- and it
makes rotation a cost to be bounded rather than applied eagerly.
profile.toml now describes the phase-3 configuration that is actually deployed,
including that the empty `skills` list is a finding and not an oversight.
pi_rpc gains --system-prompt support and no longer guesses whether a `read` tool
will exist; extension_registers_read has to be stated.
harness-layering.md records what transfers from a widely-shared account of
building a personal coding harness on pi, and what does not. The layering frame
holds and the cache-hit figure was the useful part. Its central recommendation --
installing third-party packages -- is disqualifying for an unattended agent
holding tracker credentials, and its discipline layer (AGENTS.md) is precisely
what we block, because it is discovered from every parent directory.
117 lines
4.9 KiB
Python
117 lines
4.9 KiB
Python
"""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 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 # 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)
|
|
|
|
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()))
|