846 lines
32 KiB
Python
846 lines
32 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 ast
|
|
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 dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, Callable
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
# WI-3 remains in a reviewed worktree until the release package contains the
|
|
# gated launcher. The probe resolves that worktree portably and never falls
|
|
# back to the released `mosaic` binary.
|
|
GATED_WI_ROOT_OVERRIDE = os.environ.get("GATED_WI_ROOT")
|
|
GATED_WI_BRANCH = "refs/heads/feat/830-compaction-revoke"
|
|
GATED_WI_HEAD = "f400830738998db105107a2a4c69c7f2a2a6fd5d"
|
|
GATED_WI_ANCESTOR = "66b1e0a0"
|
|
GATED_BROKER_HEAD = "23c0caca9b5d44002e6184cd7f2b6c837e8795b2"
|
|
LEASE_BROKER_DIRECTORY = "packages/mosaic/framework/tools/lease-broker"
|
|
BROKER_RELATIVE_PATH = "docs/compaction-refresh/probes/p3_generation_broker.py"
|
|
GATED_LAUNCHER_SHA256 = "e950e4224e280f16979d90cabb89aa1896c5ee28bed2df957e14d018d43cda82"
|
|
GATED_GENERATION_SHA256 = "061625402f08488eac47acd23272904e71fd1a71fd15b3bdab158632c801be4c"
|
|
GATED_BROKER_SHA256 = "4db4fef1ac6658a8ca79ad5091cefc901d2aa26003265c3d6726c294cf895cad"
|
|
|
|
|
|
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 repository_root() -> Path:
|
|
for candidate in HERE.parents:
|
|
if (candidate / ".git").exists():
|
|
return candidate
|
|
raise RuntimeError("D4 precondition: probe repository root is unavailable")
|
|
|
|
|
|
def resolve_gated_wi_root() -> Path:
|
|
"""Resolve an explicit override or the unique checked-out WI-3 branch."""
|
|
|
|
if GATED_WI_ROOT_OVERRIDE:
|
|
candidate = Path(GATED_WI_ROOT_OVERRIDE).expanduser()
|
|
candidates = [candidate]
|
|
else:
|
|
try:
|
|
listing = subprocess.check_output(
|
|
["git", "-C", str(repository_root()), "worktree", "list", "--porcelain"],
|
|
text=True,
|
|
)
|
|
except (OSError, subprocess.CalledProcessError) as error:
|
|
raise RuntimeError("D4 precondition: cannot enumerate WI-3 worktrees") from error
|
|
candidates = []
|
|
worktree: Path | None = None
|
|
head: str | None = None
|
|
branch: str | None = None
|
|
for line in [*listing.splitlines(), ""]:
|
|
if line.startswith("worktree "):
|
|
worktree = Path(line.removeprefix("worktree "))
|
|
head = None
|
|
branch = None
|
|
elif line.startswith("HEAD "):
|
|
head = line.removeprefix("HEAD ")
|
|
elif line.startswith("branch "):
|
|
branch = line.removeprefix("branch ")
|
|
elif not line and worktree is not None:
|
|
if head == GATED_WI_HEAD and branch == GATED_WI_BRANCH:
|
|
candidates.append(worktree)
|
|
worktree = None
|
|
if len(candidates) != 1:
|
|
raise RuntimeError("D4 precondition: WI-3 worktree is ambiguous or unavailable")
|
|
|
|
gated_root = candidates[0]
|
|
try:
|
|
if not gated_root.is_dir():
|
|
raise RuntimeError("D4 precondition: GATED_WI_ROOT is not a directory")
|
|
is_worktree = subprocess.check_output(
|
|
["git", "-C", str(gated_root), "rev-parse", "--is-inside-work-tree"],
|
|
text=True,
|
|
).strip()
|
|
head = subprocess.check_output(
|
|
["git", "-C", str(gated_root), "rev-parse", "HEAD"], text=True
|
|
).strip()
|
|
except (OSError, subprocess.CalledProcessError) as error:
|
|
raise RuntimeError("D4 precondition: GATED_WI_ROOT is not a git worktree") from error
|
|
if is_worktree != "true":
|
|
raise RuntimeError("D4 precondition: GATED_WI_ROOT is not a git worktree")
|
|
if head != GATED_WI_HEAD:
|
|
raise RuntimeError(f"D4 precondition: gated WI head mismatch: {head}")
|
|
try:
|
|
forward_contains = subprocess.run(
|
|
[
|
|
"git",
|
|
"-C",
|
|
str(gated_root),
|
|
"merge-base",
|
|
"--is-ancestor",
|
|
GATED_WI_ANCESTOR,
|
|
GATED_WI_HEAD,
|
|
],
|
|
check=False,
|
|
).returncode == 0
|
|
except OSError as error:
|
|
raise RuntimeError("D4 precondition: cannot verify WI-3 ancestry") from error
|
|
if not forward_contains:
|
|
raise RuntimeError("D4 precondition: gated WI lacks required ancestor")
|
|
return gated_root
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PinnedClosure:
|
|
launcher: Path
|
|
generation: Path
|
|
broker: Path
|
|
|
|
|
|
def git_object_bytes(git_root: Path, commit: str, relative_path: str) -> bytes:
|
|
try:
|
|
return subprocess.check_output(
|
|
["git", "-C", str(git_root), "show", f"{commit}:{relative_path}"]
|
|
)
|
|
except (OSError, subprocess.CalledProcessError) as error:
|
|
raise RuntimeError(f"D4 precondition: missing pinned source {relative_path}") from error
|
|
|
|
|
|
def closure_import_guard(member_sources: dict[str, str]) -> None:
|
|
"""Refuse an incomplete project-code closure before materializing it."""
|
|
|
|
allowed_nonstdlib = {"lease_generation"}
|
|
stdlib = getattr(sys, "stdlib_module_names", frozenset())
|
|
for name, source in member_sources.items():
|
|
try:
|
|
tree = ast.parse(source, filename=name)
|
|
except SyntaxError as error:
|
|
raise RuntimeError(f"D4 precondition: pinned {name} does not parse") from error
|
|
for node in ast.walk(tree):
|
|
module: str | None = None
|
|
if isinstance(node, ast.Import):
|
|
for alias in node.names:
|
|
module = alias.name.split(".", maxsplit=1)[0]
|
|
if module not in stdlib and module not in allowed_nonstdlib:
|
|
raise RuntimeError(f"D4 precondition: unpinned import {module} in {name}")
|
|
elif isinstance(node, ast.ImportFrom):
|
|
if node.level:
|
|
raise RuntimeError(f"D4 precondition: relative import in {name}")
|
|
if node.module:
|
|
module = node.module.split(".", maxsplit=1)[0]
|
|
if module not in stdlib and module not in allowed_nonstdlib:
|
|
raise RuntimeError(f"D4 precondition: unpinned import {module} in {name}")
|
|
|
|
|
|
def write_pinned_file(path: Path, data: bytes) -> None:
|
|
descriptor = os.open(
|
|
path,
|
|
os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC,
|
|
0o600,
|
|
)
|
|
try:
|
|
remaining = memoryview(data)
|
|
while remaining:
|
|
written = os.write(descriptor, remaining)
|
|
if written <= 0:
|
|
raise OSError("pinned write made no progress")
|
|
remaining = remaining[written:]
|
|
finally:
|
|
os.close(descriptor)
|
|
|
|
|
|
def materialize_closure(root: Path, gated_root: Path, gate0_root: Path) -> PinnedClosure:
|
|
"""Pin the complete project-authored runtime closure inside this fixture."""
|
|
|
|
launcher_relative = f"{LEASE_BROKER_DIRECTORY}/launch-runtime.py"
|
|
generation_relative = f"{LEASE_BROKER_DIRECTORY}/lease_generation.py"
|
|
members = (
|
|
(
|
|
"launch-runtime.py",
|
|
gated_root,
|
|
GATED_WI_HEAD,
|
|
launcher_relative,
|
|
GATED_LAUNCHER_SHA256,
|
|
),
|
|
(
|
|
"lease_generation.py",
|
|
gated_root,
|
|
GATED_WI_HEAD,
|
|
generation_relative,
|
|
GATED_GENERATION_SHA256,
|
|
),
|
|
(
|
|
"p3_generation_broker.py",
|
|
gate0_root,
|
|
GATED_BROKER_HEAD,
|
|
BROKER_RELATIVE_PATH,
|
|
GATED_BROKER_SHA256,
|
|
),
|
|
)
|
|
member_bytes: dict[str, bytes] = {}
|
|
member_sources: dict[str, str] = {}
|
|
for name, git_root, commit, relative_path, digest in members:
|
|
data = git_object_bytes(git_root, commit, relative_path)
|
|
if hashlib.sha256(data).hexdigest() != digest:
|
|
raise RuntimeError(f"D4 precondition: {name} hash mismatch")
|
|
try:
|
|
member_sources[name] = data.decode("utf-8")
|
|
except UnicodeDecodeError as error:
|
|
raise RuntimeError(f"D4 precondition: pinned {name} is not UTF-8") from error
|
|
member_bytes[name] = data
|
|
closure_import_guard(member_sources)
|
|
|
|
pinned = root / "pinned"
|
|
pinned.mkdir(mode=0o700)
|
|
paths = {name: pinned / name for name, *_ in members}
|
|
for name, path in paths.items():
|
|
write_pinned_file(path, member_bytes[name])
|
|
return PinnedClosure(
|
|
launcher=paths["launch-runtime.py"],
|
|
generation=paths["lease_generation.py"],
|
|
broker=paths["p3_generation_broker.py"],
|
|
)
|
|
|
|
|
|
def gated_launcher_precondition(
|
|
root: Path, socket_path: Path, environment: dict[str, str]
|
|
) -> PinnedClosure:
|
|
"""Verify and materialize the full WI-3/probe closure before execution."""
|
|
|
|
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")
|
|
|
|
gated_root = resolve_gated_wi_root()
|
|
closure = materialize_closure(root, gated_root, repository_root())
|
|
launcher_source = closure.launcher.read_text(encoding="utf-8")
|
|
generation_source = closure.generation.read_text(encoding="utf-8")
|
|
# Exact hashes in materialize_closure are the trust anchor. These marker
|
|
# checks are belt-and-suspenders diagnostics only.
|
|
behavior_markers = (
|
|
'"action": "register_anchor"',
|
|
"initialize_runtime_generation(generation_file, generation)",
|
|
"execute(command[0], command, environment)",
|
|
'source_environment["MOSAIC_LEASE_BROKER_SOCKET"]',
|
|
'socket_path.parent / f"generation-{session_id}.state"',
|
|
'environment["MOSAIC_LEASE_GENERATION_FILE"]',
|
|
)
|
|
if not all(marker in launcher_source for marker in behavior_markers) or not (
|
|
"def read_runtime_generation" in generation_source
|
|
and "def bump_runtime_generation" in generation_source
|
|
):
|
|
raise RuntimeError("D4 precondition: pinned launcher lacks file-generation markers")
|
|
return closure
|
|
|
|
|
|
def reject_pinned_bytecode(pinned_directory: Path) -> None:
|
|
cache_directory = pinned_directory / "__pycache__"
|
|
if cache_directory.exists() or any(pinned_directory.rglob("*.pyc")):
|
|
raise RuntimeError("D4 precondition: pinned bytecode cache is forbidden")
|
|
|
|
|
|
def launch_verified_pi(
|
|
launcher: Path,
|
|
workspace: Path,
|
|
sessions: Path,
|
|
extension: Path,
|
|
environment: dict[str, str],
|
|
) -> PiRpc:
|
|
command = [
|
|
sys.executable,
|
|
"-I",
|
|
"-B",
|
|
str(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),
|
|
]
|
|
reject_pinned_bytecode(launcher.parent)
|
|
# This is deliberately the statement immediately before Popen (inside
|
|
# PiRpc): the fixture-pinned launcher bytes are re-hashed then executed.
|
|
if hashlib.sha256(launcher.read_bytes()).hexdigest() != GATED_LAUNCHER_SHA256:
|
|
raise RuntimeError("D4 precondition: adjacent launcher hash mismatch")
|
|
return PiRpc(command, workspace, environment)
|
|
|
|
|
|
def launch_verified_broker(
|
|
broker_path: Path,
|
|
generation_path: Path,
|
|
socket_path: Path,
|
|
log_path: Path,
|
|
environment: dict[str, str],
|
|
) -> subprocess.Popen[str]:
|
|
command = [
|
|
sys.executable,
|
|
"-I",
|
|
"-B",
|
|
str(broker_path),
|
|
"--socket",
|
|
str(socket_path),
|
|
"--log",
|
|
str(log_path),
|
|
"--generation-module",
|
|
str(generation_path),
|
|
]
|
|
reject_pinned_bytecode(broker_path.parent)
|
|
if hashlib.sha256(broker_path.read_bytes()).hexdigest() != GATED_BROKER_SHA256:
|
|
raise RuntimeError("D4 precondition: pinned broker hash mismatch")
|
|
# The final helper re-hash is immediately adjacent to the broker Popen.
|
|
if hashlib.sha256(generation_path.read_bytes()).hexdigest() != GATED_GENERATION_SHA256:
|
|
raise RuntimeError("D4 precondition: pinned helper hash mismatch")
|
|
return subprocess.Popen(
|
|
command,
|
|
env=environment,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
)
|
|
|
|
|
|
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",
|
|
"PYTHONDONTWRITEBYTECODE": "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.
|
|
closure = gated_launcher_precondition(root, socket_path, environment)
|
|
broker = launch_verified_broker(
|
|
closure.broker, closure.generation, socket_path, generation_log, environment
|
|
)
|
|
wait_path(socket_path)
|
|
pi = launch_verified_pi(closure.launcher, workspace, sessions, extension, 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()
|