Files
stack/docs/compaction-refresh/probes/p3_d4_focused_run.py
2026-07-18 16:34:36 -05:00

614 lines
24 KiB
Python

#!/usr/bin/env python3
"""D4-only same-PID runtime-generation revocation harness.
AUTHORING NOTE: this file is intentionally not executed until the separately
ratified FIRE authorization. When run later, every invocation creates its own
/tmp fixture and launches the real Pi RPC runtime with only the D4 extension
and ``p3_generation_broker.py``. It does not use the broader Gate0 runner.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import os
import queue
import shutil
import signal
import socket
import subprocess
import sys
import tempfile
import threading
import time
from pathlib import Path
from typing import Any, Callable
HERE = Path(__file__).resolve().parent
# WI-3 is deliberately kept in its reviewed worktree until the release package
# contains the gated launcher. This harness pins the source it will execute;
# falling back to the released `mosaic` binary is forbidden.
GATED_WI_ROOT = HERE.parents[3].parent / "stack-cr-wi3-revoke"
GATED_WI_HEAD = "f400830738998db105107a2a4c69c7f2a2a6fd5d"
GATED_LAUNCHER = GATED_WI_ROOT / "packages/mosaic/framework/tools/lease-broker/launch-runtime.py"
GATED_LAUNCHER_SHA256 = "e950e4224e280f16979d90cabb89aa1896c5ee28bed2df957e14d018d43cda82"
GATED_GENERATION_MODULE = GATED_LAUNCHER.with_name("lease_generation.py")
class PiRpc:
"""Small JSON-RPC client for an isolated real Pi process."""
def __init__(self, command: list[str], cwd: Path, env: dict[str, str]) -> None:
self.process = subprocess.Popen(
command,
cwd=cwd,
env=env,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1,
start_new_session=True,
)
self.events: queue.Queue[dict[str, Any]] = queue.Queue()
self.stderr_lines: list[str] = []
threading.Thread(target=self._read_stdout, daemon=True).start()
threading.Thread(target=self._read_stderr, daemon=True).start()
def _read_stdout(self) -> None:
if self.process.stdout is None:
raise RuntimeError("Pi stdout pipe is unavailable")
for line in self.process.stdout:
try:
self.events.put(json.loads(line))
except json.JSONDecodeError:
continue
def _read_stderr(self) -> None:
if self.process.stderr is None:
raise RuntimeError("Pi stderr pipe is unavailable")
for line in self.process.stderr:
self.stderr_lines.append(line.rstrip("\n"))
def send(self, payload: dict[str, object]) -> None:
if self.process.stdin is None:
raise RuntimeError("Pi stdin pipe is unavailable")
self.process.stdin.write(json.dumps(payload) + "\n")
self.process.stdin.flush()
def wait(
self,
predicate: Callable[[dict[str, Any]], bool],
description: str,
timeout: float = 180,
) -> dict[str, Any]:
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if self.process.poll() is not None and self.events.empty():
detail = " | ".join(self.stderr_lines[-5:])
raise RuntimeError(
f"Pi exited {self.process.returncode} while waiting for {description}: {detail}"
)
try:
event = self.events.get(timeout=0.2)
except queue.Empty:
continue
if predicate(event):
return event
raise TimeoutError(f"timed out waiting for {description}")
def response(self, request_id: str, timeout: float = 180) -> dict[str, Any]:
return self.wait(
lambda event: event.get("type") == "response" and event.get("id") == request_id,
f"response {request_id}",
timeout,
)
def prompt_and_settle(self, request_id: str, message: str) -> None:
self.send({"id": request_id, "type": "prompt", "message": message})
response = self.response(request_id)
if not response.get("success"):
raise RuntimeError(f"prompt rejected: {response}")
self.wait(
lambda event: event.get("type") == "agent_settled",
f"agent_settled {request_id}",
)
def close(self) -> None:
if self.process.poll() is None:
try:
os.killpg(self.process.pid, signal.SIGTERM)
except ProcessLookupError:
pass
try:
self.process.wait(timeout=8)
except subprocess.TimeoutExpired:
os.killpg(self.process.pid, signal.SIGKILL)
self.process.wait(timeout=5)
def wait_path(path: Path, timeout: float = 20) -> None:
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if path.exists():
return
time.sleep(0.05)
raise TimeoutError(f"timed out waiting for {path}")
def request(path: Path, payload: dict[str, object]) -> dict[str, Any]:
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as conn:
conn.connect(str(path))
conn.sendall((json.dumps(payload) + "\n").encode())
reply = conn.makefile("r", encoding="utf-8").readline()
return json.loads(reply)
def jsonl(path: Path) -> list[dict[str, Any]]:
return [json.loads(line) for line in path.read_text().splitlines() if line]
def write_extension(path: Path) -> None:
"""Write the minimal Pi lifecycle bridge into the isolated fixture only."""
path.write_text(
"""import type { ExtensionAPI } from '@earendil-works/pi-coding-agent';
import { Type } from 'typebox';
import { appendFileSync, readFileSync } from 'node:fs';
import net from 'node:net';
const socketPath = process.env['D4_GENERATION_SOCKET'];
const logPath = process.env['D4_PI_LOG'];
function starttime(): number {
const text = readFileSync(`/proc/${process.pid}/stat`, 'utf8');
const close = text.lastIndexOf(')');
return Number(text.slice(close + 2).trim().split(/\\s+/)[19]);
}
function log(event: string, details: Record<string, unknown> = {}): void {
if (!logPath) return;
appendFileSync(logPath, `${JSON.stringify({ event, pid: process.pid, starttime_ticks: starttime(), ...details })}\\n`);
}
function broker(payload: Record<string, unknown>): Promise<Record<string, unknown>> {
if (!socketPath) return Promise.reject(new Error('D4_GENERATION_SOCKET is required'));
return new Promise((resolve, reject) => {
const connection = net.createConnection(socketPath);
let buffer = '';
connection.setEncoding('utf8');
connection.on('connect', () => connection.write(`${JSON.stringify(payload)}\\n`));
connection.on('data', (chunk) => {
buffer += chunk;
const newline = buffer.indexOf('\\n');
if (newline < 0) return;
connection.end();
resolve(JSON.parse(buffer.slice(0, newline)) as Record<string, unknown>);
});
connection.on('error', reject);
});
}
export default function register(pi: ExtensionAPI): void {
let initialStartup = true;
async function lifecycle(
phase: 'start' | 'shutdown',
reason: string,
): Promise<Record<string, unknown>> {
if (!(phase === 'start' && reason === 'startup' && initialStartup)) {
const bump = await broker({ action: 'bump-generation' });
log('generation_state_bump', { phase, reason, bump });
}
initialStartup = false;
return broker({ action: 'lifecycle', phase, reason });
}
pi.on('session_start', async (event) => {
const lifecycleResult = await lifecycle('start', event.reason);
log('session_start', { reason: event.reason, lifecycle: lifecycleResult });
if (event.reason === 'reload') {
const generation = lifecycleResult['new_generation'];
if (typeof generation !== 'number') throw new Error('broker did not return new_generation');
const current = await broker({ action: 'authorize-probe', generation });
const superseded = await broker({ action: 'authorize-probe', generation: generation - 1 });
log('d4_generation_authorization', { generation, current, superseded });
}
});
pi.on('session_shutdown', async (event) => {
const lifecycleResult = await lifecycle('shutdown', event.reason);
log('session_shutdown', { reason: event.reason, lifecycle: lifecycleResult });
});
pi.registerTool({
name: 'd4_fixture_promote',
label: 'D4 Fixture Promotion',
description: 'Promotes only the fixture lease needed for the D4 revocation check.',
parameters: Type.Object({}),
async execute() {
// the promotion step is a D4 test fixture, not a P2 evidence-gathering authorization.
const promotion = await broker({ action: 'promote-probe' });
log('fixture_promotion', { promotion });
return { content: [{ type: 'text', text: 'D4 fixture promotion complete' }] };
},
});
pi.registerCommand('d4-reload', {
description: 'D4-only same-PID reload boundary.',
handler: async (_args, context) => {
await context.reload();
},
});
}
""",
encoding="utf-8",
)
def gated_launcher_precondition(root: Path, socket_path: Path, environment: dict[str, str]) -> None:
"""Refuse to launch unless the exact WI-3 launcher binds this run's fixture."""
if environment.get("MOSAIC_LEASE_BROKER_SOCKET") != str(socket_path):
raise RuntimeError("D4 precondition: lease broker socket is not this fixture")
if environment.get("MOSAIC_LEASE_GENERATION_FILE"):
raise RuntimeError("D4 precondition: inherited generation file is forbidden")
fixture_path_vars = (
"HOME",
"XDG_CONFIG_HOME",
"XDG_CACHE_HOME",
"XDG_STATE_HOME",
"XDG_RUNTIME_DIR",
"TMPDIR",
"D4_PI_LOG",
"MOSAIC_AGENT_WORKDIR",
"MOSAIC_HEARTBEAT_RUN_DIR",
"MOSAIC_HOME",
)
if any(
not (value := environment.get(name)) or not Path(value).is_relative_to(root)
for name in fixture_path_vars
):
raise RuntimeError("D4 precondition: child write path escapes fixture root")
if socket_path.parent != root or root.parent != Path(tempfile.gettempdir()):
raise RuntimeError("D4 precondition: fixture socket is outside this run's temporary root")
try:
head = subprocess.check_output(
["git", "-C", str(GATED_WI_ROOT), "rev-parse", "HEAD"], text=True
).strip()
launcher_bytes = GATED_LAUNCHER.read_bytes()
launcher_source = launcher_bytes.decode("utf-8")
generation_source = GATED_GENERATION_MODULE.read_text(encoding="utf-8")
except (OSError, subprocess.CalledProcessError, UnicodeDecodeError) as error:
raise RuntimeError("D4 precondition: gated WI launcher is unavailable") from error
if head != GATED_WI_HEAD:
raise RuntimeError(f"D4 precondition: gated WI head mismatch: {head}")
if hashlib.sha256(launcher_bytes).hexdigest() != GATED_LAUNCHER_SHA256:
raise RuntimeError("D4 precondition: gated WI launcher hash mismatch")
register = launcher_source.find('"action": "register_anchor"')
initialize = launcher_source.find("initialize_runtime_generation(generation_file, generation)")
execute = launcher_source.find("execute(command[0], command, environment)")
fixture_socket = launcher_source.find('source_environment["MOSAIC_LEASE_BROKER_SOCKET"]')
fixture_state = launcher_source.find('socket_path.parent / f"generation-{session_id}.state"')
exported_state = launcher_source.find('environment["MOSAIC_LEASE_GENERATION_FILE"]')
generation_api = (
"def read_runtime_generation" in generation_source
and "def bump_runtime_generation" in generation_source
)
if (
min(register, initialize, execute, fixture_socket, fixture_state, exported_state) < 0
or register >= initialize
or initialize >= execute
or not generation_api
):
raise RuntimeError("D4 precondition: launcher is not file-generation fixture-bound")
def assert_d4(records: list[dict[str, Any]]) -> dict[str, object]:
def record_where(description: str, candidates: list[dict[str, Any]]) -> dict[str, Any]:
if not candidates:
raise AssertionError(f"missing D4 evidence record: {description}")
return candidates[0]
fixture_listen = record_where(
"fixture listen", [r for r in records if r.get("event") == "listen"]
)
fixture_root = Path(fixture_listen["socket"]).parent
lifecycle = [record for record in records if record.get("event") == "runtime_generation_bump"]
state_bumps = [record for record in records if record.get("event") == "generation_state_bumped"]
promotion = record_where(
"fixture promotion", [r for r in records if r.get("event") == "probe_lease_promoted"]
)
launcher_registration = record_where(
"lease anchor", [r for r in records if r.get("event") == "lease_anchor_registered"]
)
reload_revoke = record_where(
"reload shutdown",
[
r
for r in lifecycle
if r.get("reason") == "reload" and r.get("phase") == "shutdown"
],
)
reload_start = record_where(
"reload start",
[
r
for r in lifecycle
if r.get("reason") == "reload" and r.get("phase") == "start"
],
)
authorization = [
record for record in records if record.get("event") == "generation_authorization"
]
current_generation = reload_start["new_generation"]
current_authorization = record_where(
"current-generation authorization",
[r for r in authorization if r.get("requested_generation") == current_generation],
)
superseded_authorization = record_where(
"superseded-generation authorization",
[r for r in authorization if r.get("requested_generation") == current_generation - 1],
)
identities = {
(record["peercred"]["pid"], record["starttime_ticks"])
for record in [*lifecycle, *state_bumps, promotion, launcher_registration, *authorization]
}
generations = [record["new_generation"] for record in lifecycle]
file_records = [*lifecycle, *state_bumps, promotion, *authorization]
observed_reasons = {record.get("reason") for record in lifecycle}
checks = {
"same_pid_starttime": len(identities) == 1,
"strictly_increasing_generation": all(
previous < current for previous, current in zip(generations, generations[1:])
),
"state_file_drives_lifecycle": [record["generation"] for record in state_bumps]
== generations[1:],
"state_file_source": all(
record.get("generation_source") == "state-file" for record in file_records
),
"state_file_in_fixture_root": all(
Path(record["generation_file"]).parent == fixture_root for record in file_records
)
and Path(launcher_registration["generation_file"]).parent == fixture_root,
"all_lifecycle_boundaries": {"startup", "reload", "fork", "new", "resume"}
<= observed_reasons,
"lease_anchor_fixture": launcher_registration.get("session_id_shape") == "hex-256",
"verified_revoked_on_reload": reload_revoke.get("prior_lease") == "VERIFIED"
and reload_revoke.get("prior_lease_revoked") is True,
"new_generation_unverified": current_authorization.get("code") == "MUTATOR_UNVERIFIED",
"prior_generation_stale": superseded_authorization.get("code") == "STALE_GENERATION",
}
failed = [name for name, passed in checks.items() if not passed]
if failed:
raise AssertionError(f"D4 checks failed: {', '.join(failed)}")
passed = all(checks.values())
if not passed:
raise AssertionError("D4 PASS derivation failed")
return {
"machine_assertions": "PASS" if passed else "FAIL",
"checks": checks,
"same_pid_starttime": next(iter(identities)),
"generations": generations,
"reload_revoke_verified": checks["verified_revoked_on_reload"],
"lease_anchor_fixture": checks["lease_anchor_fixture"],
"file_backed_generation": checks["state_file_drives_lifecycle"],
"new_generation_code": current_authorization["code"],
"superseded_generation_code": superseded_authorization["code"],
}
def isolated_environment(
root: Path, index: int, workspace: Path, socket_path: Path, pi_log: Path
) -> dict[str, str]:
"""Build a write-confined child environment; no inherited path variable survives."""
fixture_home = root / "home"
fixture_config = root / "config"
fixture_cache = root / "cache"
fixture_state = root / "state"
fixture_runtime = root / "runtime"
fixture_tmp = root / "tmp"
fixture_heartbeat = root / "heartbeat"
fixture_mosaic_home = root / "mosaic-home"
for directory in (
fixture_home,
fixture_config,
fixture_cache,
fixture_state,
fixture_runtime,
fixture_tmp,
fixture_heartbeat,
fixture_mosaic_home,
):
directory.mkdir(mode=0o700)
# Authentication/settings are copied into fixture HOME so Pi never writes
# under the operator's HOME. They are not emitted or modified in place.
source_agent = Path.home() / ".pi" / "agent"
target_agent = fixture_home / ".pi" / "agent"
target_agent.mkdir(parents=True, mode=0o700)
for name in ("settings.json", "auth.json", "bin/fd"):
source = source_agent / name
target = target_agent / name
if source.is_file():
target.parent.mkdir(parents=True, mode=0o700)
shutil.copy2(source, target)
environment = {
"HOME": str(fixture_home),
"XDG_CONFIG_HOME": str(fixture_config),
"XDG_CACHE_HOME": str(fixture_cache),
"XDG_STATE_HOME": str(fixture_state),
"XDG_RUNTIME_DIR": str(fixture_runtime),
"TMPDIR": str(fixture_tmp),
"PATH": os.environ.get("PATH", ""),
"LANG": os.environ.get("LANG", "C.UTF-8"),
"TERM": os.environ.get("TERM", "dumb"),
"D4_GENERATION_SOCKET": str(socket_path),
"MOSAIC_LEASE_BROKER_SOCKET": str(socket_path),
"D4_PI_LOG": str(pi_log),
"MOSAIC_AGENT_NAME": f"d4-fixture-{index}",
"MOSAIC_AGENT_WORKDIR": str(workspace),
"MOSAIC_HEARTBEAT_RUN_DIR": str(fixture_heartbeat),
"MOSAIC_HOME": str(fixture_mosaic_home),
"MOSAIC_PI_FORCE_SKILLS": "",
"PI_SKIP_VERSION_CHECK": "1",
}
if "PI_CODING_AGENT" in os.environ:
environment["PI_CODING_AGENT"] = os.environ["PI_CODING_AGENT"]
return environment
def scrub_fixture_credentials(root: Path) -> None:
"""Remove the copied Pi credential/config subtree before retaining evidence."""
copied_agent = root / "home" / ".pi" / "agent"
if copied_agent.exists():
shutil.rmtree(copied_agent)
if copied_agent.exists():
raise RuntimeError("D4 credential scrub failed")
def run_once(index: int) -> Path:
root = Path(tempfile.mkdtemp(prefix=f"gate0-d4-{index}-"))
workspace = root / "workspace"
sessions = root / "sessions"
workspace.mkdir(mode=0o700)
sessions.mkdir(mode=0o700)
socket_path = root / "generation.sock"
generation_log = root / "generation.jsonl"
pi_log = root / "pi.jsonl"
extension = root / "d4_extension.ts"
write_extension(extension)
broker: subprocess.Popen[str] | None = None
pi: PiRpc | None = None
try:
environment = isolated_environment(root, index, workspace, socket_path, pi_log)
# Must run before the fixture broker or Pi process is launched. It proves
# the launcher registers before exec and can only read this fixture socket.
gated_launcher_precondition(root, socket_path, environment)
broker = subprocess.Popen(
[
sys.executable,
str(HERE / "p3_generation_broker.py"),
"--socket",
str(socket_path),
"--log",
str(generation_log),
"--generation-module",
str(GATED_GENERATION_MODULE),
],
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
)
wait_path(socket_path)
pi = PiRpc(
[
sys.executable,
str(GATED_LAUNCHER),
"--runtime",
"pi",
"--",
"pi",
"--mode",
"rpc",
"--session-dir",
str(sessions),
"--no-extensions",
"--no-context-files",
"--no-prompt-templates",
"--model",
"openai-codex/gpt-5.6-sol",
"--thinking",
"medium",
"--extension",
str(extension),
],
workspace,
environment,
)
pi.send({"id": "state", "type": "get_state"})
state = pi.response("state")
original_session = state["data"]["sessionFile"]
pi.prompt_and_settle(
"fixture-promote",
"Call d4_fixture_promote exactly once, then stop.",
)
pi.send({"id": "reload", "type": "prompt", "message": "/d4-reload"})
reload_response = pi.response("reload")
if not reload_response.get("success"):
raise RuntimeError(f"reload failed: {reload_response}")
for request_id, request_payload in [
("clone", {"id": "clone", "type": "clone"}),
("new", {"id": "new", "type": "new_session"}),
(
"resume",
{"id": "resume", "type": "switch_session", "sessionPath": original_session},
),
]:
pi.send(request_payload)
response = pi.response(request_id)
if not response.get("success") or response.get("data", {}).get("cancelled"):
raise RuntimeError(f"{request_id} failed: {response}")
results = assert_d4(jsonl(generation_log))
verdict = results.get("machine_assertions")
if verdict != "PASS":
raise RuntimeError(f"D4 checks did not derive PASS: {verdict}")
(root / "machine-assertions.json").write_text(
json.dumps(results, sort_keys=True, indent=2) + "\n",
encoding="utf-8",
)
print(f"run={index} evidence_dir={root}")
print(f"machine_assertions={verdict}")
print(json.dumps(results, sort_keys=True))
except Exception as error:
(root / "machine-assertions.json").write_text(
json.dumps({"error": f"{type(error).__name__}: {error}"}, sort_keys=True, indent=2)
+ "\n",
encoding="utf-8",
)
print(f"run={index} evidence_dir={root}")
print("machine_assertions=FAIL")
print(f"error={type(error).__name__}: {error}")
raise
finally:
try:
try:
if pi is not None:
pi.close()
finally:
if broker is not None:
try:
request(socket_path, {"action": "shutdown-broker"})
except OSError:
pass
try:
broker.wait(timeout=5)
except subprocess.TimeoutExpired:
broker.kill()
broker.wait()
finally:
scrub_fixture_credentials(root)
return root
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--runs", type=int, default=3, choices=(3,))
args = parser.parse_args()
roots: list[Path] = []
for index in range(1, args.runs + 1):
roots.append(run_once(index))
print("d4_isolation_runs=" + ",".join(str(root) for root in roots))
if __name__ == "__main__":
main()