113 lines
5.5 KiB
Python
Executable File
113 lines
5.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""No-model native Pi smoke: project discovery, footer, /goal and Alt+G."""
|
|
import fcntl
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import pty
|
|
import re
|
|
import select
|
|
import signal
|
|
import struct
|
|
import subprocess
|
|
import tempfile
|
|
import termios
|
|
import time
|
|
import uuid
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
ANSI = re.compile(r"\x1b(?:\[[0-?]*[ -/]*[@-~]|\][^\x07]*(?:\x07|\x1b\\))")
|
|
TEXT = "native recall " + "full acceptance criteria " * 16 + "END-OF-GOAL"
|
|
|
|
# A fresh clone has no ignored development installation. Exercise the same
|
|
# fail-closed sync used by the launcher before testing Pi discovery.
|
|
subprocess.run([str(ROOT / "scripts/sync-dev-extensions.sh")], cwd=ROOT, check=True)
|
|
|
|
|
|
def receive(fd, pattern, timeout=12):
|
|
data = b""
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
ready, _, _ = select.select([fd], [], [], max(0, deadline - time.monotonic()))
|
|
if not ready:
|
|
break
|
|
try:
|
|
data += os.read(fd, 65536)
|
|
except OSError:
|
|
break
|
|
clean = ANSI.sub("", data.decode(errors="replace"))
|
|
if pattern in clean:
|
|
return data
|
|
raise AssertionError(f"missing {pattern!r}: {data[-4000:]!r}")
|
|
|
|
|
|
with tempfile.TemporaryDirectory(prefix="ng-native-pi-") as home:
|
|
env = {**os.environ, "PI_CODING_AGENT_DIR": home, "PI_OFFLINE": "1", "TERM": "xterm-256color"}
|
|
env.pop("MOSAIC_LAUNCH_INCARNATION", None)
|
|
rpc = subprocess.Popen(["pi", "--approve", "--offline", "--mode", "rpc", "--no-session", "--no-tools"], cwd=ROOT, env=env, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
|
|
try:
|
|
rpc.stdin.write('{"id":"commands","type":"get_commands"}\n')
|
|
rpc.stdin.flush()
|
|
deadline = time.monotonic() + 15
|
|
while True:
|
|
assert time.monotonic() < deadline, "RPC discovery timed out"
|
|
response = json.loads(rpc.stdout.readline())
|
|
if response.get("id") == "commands":
|
|
break
|
|
commands = [c for c in response["data"]["commands"] if c["name"] == "goal"]
|
|
assert len(commands) == 1, commands
|
|
assert str(ROOT / ".pi/extensions/goal/index.ts") in json.dumps(commands[0]), commands
|
|
print("PASS native project discovery: exactly one local /goal", flush=True)
|
|
finally:
|
|
rpc.terminate()
|
|
rpc.communicate(timeout=10)
|
|
|
|
for status, label, width, no_color, timed in [("paused", "Paused", 120, False, False), ("blocked", "Blocked", 45, False, False), ("none", "Complete", 120, True, False), ("active", "Waiting", 45, False, True), ("active", "Waiting", 120, False, False)]:
|
|
incarnation = "ng-native-test-" + uuid.uuid4().hex
|
|
statefile = ROOT / ".pi/state/goal" / f"goal-state.{incarnation}.json"
|
|
statefile.parent.mkdir(parents=True, exist_ok=True)
|
|
state = dict(version=1, text=TEXT, status=status, checks=0, maxChecks=25, noProgressReports=0, maxNoProgressReports=3, workEventSinceReport=False, setAt="2026-09-06T00:00:00.000Z")
|
|
if status == "none":
|
|
state.update(text="", lastOutcome=dict(text=TEXT, status="complete", evidence="native fixture", at="2026-09-06T00:00:00.000Z"))
|
|
if status == "active":
|
|
state.update(activeWait=dict(owner="native fixture", nextCheck="manual acceptance"))
|
|
if timed:
|
|
state.update(waitTimeoutSeconds=3600, waitWakeUsed=False)
|
|
state["activeWait"].update(deadlineAt=int(time.time() * 1000) + 3600000, wakeSent=False)
|
|
statefile.write_text(json.dumps(state))
|
|
tty_env = {**env, "MOSAIC_LAUNCH_INCARNATION": incarnation}
|
|
tty_env.pop("NO_COLOR", None)
|
|
if no_color:
|
|
tty_env["NO_COLOR"] = "1"
|
|
master, slave = pty.openpty()
|
|
fcntl.ioctl(slave, termios.TIOCSWINSZ, struct.pack("HHHH", 40, width, 0, 0))
|
|
child = subprocess.Popen(["pi", "--approve", "--offline", "--no-session", "--no-tools", "--no-skills", "--no-context-files"], cwd=ROOT, env=tty_env, stdin=slave, stdout=slave, stderr=slave, start_new_session=True)
|
|
os.close(slave)
|
|
try:
|
|
startup = receive(master, f"Goal: {label}")
|
|
Path("/tmp/ng-goal-native-startup.log").write_bytes(startup)
|
|
os.write(master, b"\x1b[1;1R\x1b[?1;2c\x1b[?0u")
|
|
# Let terminal capability negotiation and editor initialization settle.
|
|
while select.select([master], [], [], 0.5)[0]:
|
|
os.read(master, 65536)
|
|
os.write(master, b"\x1b[103;3u")
|
|
receive(master, "END-OF-GOAL")
|
|
os.write(master, b"\x1b[200~/goal --max 0\x1b[201~")
|
|
receive(master, "/goal --max 0")
|
|
os.write(master, b"\x1b[13u")
|
|
receive(master, 'integer, got "0"')
|
|
os.write(master, b"\x1b[200~/goal\x1b[201~")
|
|
entered = receive(master, "/goal")
|
|
Path("/tmp/ng-goal-native-entered.log").write_bytes(entered)
|
|
os.write(master, b"\x1b[13u")
|
|
receive(master, "END-OF-GOAL")
|
|
if status == "active":
|
|
current = json.loads(statefile.read_text())
|
|
assert current == state, "waiting startup/recall must not inject checks or mutate the fixture"
|
|
print(f"PASS native {label}, width={width}, NO_COLOR={no_color}, timed={timed}: footer + /goal + Alt+G; zero wait checks", flush=True)
|
|
finally:
|
|
os.killpg(child.pid, signal.SIGTERM)
|
|
child.wait(timeout=10)
|
|
os.close(master)
|
|
statefile.unlink(missing_ok=True)
|