#!/usr/bin/env python3 """t15_builder_session.py — T15 §15Q: REAL event-driven builder proof. The difference from T14, enforced by construction: launch → register WAITING_EVENT → YIELD (select on the bridge FIFO) → builder exit event → resume immediately → reload canonical state → readiness/exit canaries → independent verification → continuation NO sleep/poll loops anywhere on the normal path. The watchdog is only a failure detector (lost event / dead process), never the scheduler. Modes: start classify → activate → contract → real builder → event wait → exit canary → verification → evidence-backed close interrupt like start, but the builder process group is SIGKILLed at the first source file → PROCESS_LOST → canonical state persists for a fresh builder (run `resume` next) resume FRESH controller + fresh builder from canonical state only status print canonical construction state The builder is the real model process (mimo run — Verboo per the standing provider policy). Builder identity is recorded, never trusted. """ from __future__ import annotations import json import os import signal import subprocess import sys import time from pathlib import Path ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT / "src")) sys.path.insert(0, str(ROOT / "scripts")) import t14_builder_session as t14 # noqa: E402 — reuse the proven pieces from rune import construction_interface as ci # noqa: E402 from rune import event_wait as ew # noqa: E402 from rune import scaffold as sc # noqa: E402 from rune import structural_store as ss # noqa: E402 from rune.event_wait import DEFAULT_DAEMON_URL # noqa: E402 from rune.provider_context import ProviderUnresolved, require_model # noqa: E402 # T16 §16F: NO silent default provider — resolve from the ambient provider # context (T15_BUILDER_MODEL, falling back to MIMO_MODEL); loud failure # with exit code 3 when the provider layer has not selected one. try: BUILDER_MODEL = os.environ.get("T15_BUILDER_MODEL", "").strip() \ or require_model() except ProviderUnresolved as _exc: import sys as _sys _sys.stderr.write(str(_exc) + "\n") raise SystemExit(3) # 3 = PROVIDER_UNRESOLVED (no silent fallback) DAEMON_URL = os.environ.get("T15_DAEMON_URL", DEFAULT_DAEMON_URL) def log(msg: str) -> None: print(f"[t15] {msg}", flush=True) def readiness_canary(toolchain_hint: str) -> dict: """§15F: may the build stage begin? Executable checks, not optimism. Runtime checks derive from the verifier (15R generalisation).""" checks = [] runtimes = (["node"] if "node" in VERIFIER else []) + (["python3"] if "python" in VERIFIER else []) for rt in runtimes: r = subprocess.run([rt, "--version"], capture_output=True, text=True) checks.append({"name": f"{rt}_present", "passed": r.returncode == 0, "evidence": (r.stdout or "").strip()[:30]}) r2 = subprocess.run(["mimo", "--version"], capture_output=True, text=True) checks.append({"name": "builder_cli_present", "passed": r2.returncode == 0, "evidence": (r2.stdout or "").strip()[:40]}) try: import urllib.request with urllib.request.urlopen(DAEMON_URL + "/health", timeout=3) as resp: ok = resp.status == 200 checks.append({"name": "event_bridge_healthy", "passed": ok, "evidence": DAEMON_URL}) except Exception as e: # noqa: BLE001 checks.append({"name": "event_bridge_healthy", "passed": False, "evidence": str(e)[:80]}) verdict = "READY" if all(c["passed"] for c in checks) else "BLOCKED" return {"canary": "readiness", "verdict": verdict, "checks": checks} # 15R: the driver is construction INFRASTRUCTURE — the request/scope are # parameters, never scheduler-specific constants. Defaults preserve the # proven fixture. REQUEST = os.environ.get( "T15_REQUEST", "Build me a small standalone practice scheduler that can create, " "list and complete practice items.") SCOPE = [s for s in os.environ.get( "T15_SCOPE", "scheduler.js,scheduler.test.js,package.json").split(",") if s] PROJECT_NAME = os.environ.get("T15_PROJECT_NAME", "") SPEC_CONTRACT = os.environ.get("T15_SPEC_CONTRACT", "contract:practice-items-spec") TOOLCHAIN = os.environ.get("T15_TOOLCHAIN", "node >= 18 (node:test built-in runner, ES modules)") VERIFIER = os.environ.get("T15_VERIFIER", "node --test") def setup_construction(store: Path, project_dir: Path) -> tuple[str, str]: """classify → activate → seed spec → contract → transitions. Reuses the proven T14 canonical path unchanged.""" staged = t14.classify_and_activate(store, REQUEST) project = staged["activation"]["project"] assert staged["activation"]["mode"] == "PROJECT_BUILD" _seed_spec(store, project) iface = ci.ConstructionInterface( builder=BUILDER_MODEL, project_store=str(store / "projects" / project), project_id=project) sub = iface.submit_need( REQUEST, origin="rune_native", constraints={"environment": {"os": "linux", "runtime": TOOLCHAIN, "runner": VERIFIER}, "contracts": [SPEC_CONTRACT]}, affected_surface=list(SCOPE), evidence_refs=["t15-user-request", "node-v22-installed"], approval_context="t15 user-approved fixture") assert sub.get("success"), sub.get("errors") contract_id = sub["ids"]["contract_id"] for phase in ("approved", "building"): tr = iface.transition(contract_id, phase, evidence_ref=f"t15 {phase}") assert tr.get("success"), tr.get("errors") log(f"contract {contract_id} → building") return project, contract_id def _seed_spec(store: Path, project: str) -> str: """Govern the requirement as a contract on the project surface (generic — request/scope come from parameters).""" proj = sc.Scaffold(str(store / "projects" / project), project) doc = proj.load() if ss.entity(doc, SPEC_CONTRACT) is None: ss.add_entity(doc, id=SPEC_CONTRACT, kind="contract", identities={"objective": REQUEST}, evidence_refs=["t15-request"]) ss.add_entity(doc, id="capability:planned-" + project[:24], kind="capability", maturity="DECLARED", evidence_refs=["t15-request"]) ss.add_relationship(doc, type="implements", from_id="capability:planned-" + project[:24], to_id=SPEC_CONTRACT, evidence_refs=["t15-request"]) proj.commit(doc, what="seed requirement spec (user request)") return SPEC_CONTRACT def run_build_via_events(project_dir: Path, operation_id: str, projection: dict, contract_id: str, interrupt_after_first_file: bool) -> dict: """THE 15Q core: real builder under the event notifier; the driver YIELDS on the bridge FIFO. No sleep. Watchdog = failure detector.""" ew.register_process_wait( operation_id, contract_id=contract_id, project="t15-fixture", objective=REQUEST, watched_for="builder process exit event", watchdog_secs=900) # op marker rides the prompt (last argv) so cmdline liveness can find # the build even if node regroups its children (T15Q live finding) builder_cmd = ["mimo", "run", "-m", BUILDER_MODEL, "--format", "json", t14.builder_prompt(projection, resumed=False) + f" [op:{operation_id}]"] notifier = ew.run_builder_under_events(builder_cmd, operation_id, daemon_url=DAEMON_URL) log(f"builder launched (pgid {notifier.pid}); driver YIELDS on the " f"event bridge — no polling") if interrupt_after_first_file: # kill at first source file: the interruption is real, the event # never arrives (the notifier dies with the group) — the watchdog # must classify PROCESS_LOST, never silently wait. # TEST_TIMING: this 0.5s poll exists ONLY on the deliberate # interrupt path (killing at a checkpoint), never on the normal # continuation path. while True: files = [p for p in project_dir.iterdir() if p.suffix in (".mjs", ".js", ".json")] if files: break time.sleep(0.5) log("INTERRUPTING builder (first file landed) — SIGKILL group") os.killpg(notifier.pid, signal.SIGKILL) result = ew.await_resume_event(operation_id, watchdog_secs=900, pgid=notifier.pid) return result def finish_via_canaries(store: Path, project: str, contract_id: str, project_dir: Path, before: dict, resume_note: str) -> int: """Reload canonical state → scope audit → INDEPENDENT verification (15G test truth) → evidence-backed lifecycle.""" iface = ci.ConstructionInterface( builder=BUILDER_MODEL, project_store=str(store / "projects" / project), project_id=project) ctx = iface.construction_context(contract_id) assert ctx["state"]["active"] or ctx["state"]["phase"] == "closed" log(f"canonical state: phase={ctx['state']['phase']} " f"scope={ctx['state']['scope']}") # exit canary: scope audit with provenance-aware residue (15O) audit = t14.audit_scope(project_dir, before, list(SCOPE)) # 15G: verification truth — node --test must actually run tests # 15G test truth via the CONTRACT's verifier (parameterized — 15R # generalisation: python unittest/pytest, node --test, etc.) truth = subprocess.run( ["bash", "-c", VERIFIER], cwd=str(project_dir), capture_output=True, text=True, timeout=300) import re count = None for pat in (r"# pass (\d+)", # node --test TAP summary r"(\d+) passed", # pytest r"Ran (\d+) tests"): # python unittest m = re.search(pat, (truth.stdout or "") + (truth.stderr or "")) if m: count = int(m.group(1)) break test_count = count if count is not None else 0 log(f"verification: exit={truth.returncode} tests={test_count}") if test_count == 0: log("FALSE GREEN REFUSED: zero tests executed (15G)") return 1 if audit["scope_violations"]: log(f"closure REFUSED: scope violations {audit['scope_violations']}") return 1 builder_status = "UNKNOWN" result = t14.complete_lifecycle(iface, contract_id, audit, {"cmd": "node --test", "exit_code": truth.returncode}, builder_status) summary = { "event_driven": True, "builder_model": BUILDER_MODEL, "resume_note": resume_note, "scope_audit": audit, "verification": {"exit_code": truth.returncode, "test_count": test_count}, "lifecycle": result, } print(json.dumps(summary, indent=1)) return 0 if result.get("closed") else 1 def discover(store: Path) -> tuple[str, str] | None: """§15U: delegates to the canonical interface discovery (via the t14 shim — one hop, no duplication).""" return t14.discover_contract_id(store) def cmd_start(store: Path, project_dir: Path, interrupt: bool) -> int: ready = readiness_canary("node") assert ready["verdict"] == "READY", f"readiness BLOCKED: {ready['checks']}" log(f"readiness READY ({len(ready['checks'])} checks)") before = t14.snapshot_files(project_dir) project, contract_id = setup_construction(store, project_dir) operation_id = f"t15-{contract_id}" iface = ci.ConstructionInterface( builder=BUILDER_MODEL, project_store=str(store / "projects" / project), project_id=project) projection = t14.build_projection(iface, contract_id, phase_note="initial build") result = run_build_via_events(project_dir, operation_id, projection, contract_id, interrupt) if result["kind"] == "watchdog": # §15E canonical-state-wins: process-table absence is NOT truth # (detached builders re-parent their workers — T15Q live finding). # The durable waitstore decides: a recorded verdict means the # builder completed and the event WAS processed. rec = ew.resume_or_recover(operation_id) if rec["kind"] == "event": log(f"watchdog fired but canonical state shows the build " f"completed (verdict={rec['wait'].get('verdict')}) — " f"continuing from authority") return finish_via_canaries(store, project, contract_id, project_dir, before, "watchdog→canonical recovery") log("PROCESS_LOST: no event, no durable verdict — fresh builder " "resumes from authority only") return 3 wait = result.get("wait", {}) log(f"event-driven resume: verdict={wait.get('verdict')} " f"exit={wait.get('process_event_evidence', {}).get('exit_code')}") return finish_via_canaries(store, project, contract_id, project_dir, before, "event-driven completion") def cmd_resume(store: Path, project_dir: Path) -> int: found = discover(store) assert found, "no contract in canonical state" project, contract_id = found log(f"FRESH builder resume: {project}/{contract_id} (canonical only)") before = t14.snapshot_files(project_dir) iface = ci.ConstructionInterface( builder=BUILDER_MODEL + "-resumed", project_store=str(store / "projects" / project), project_id=project) projection = t14.build_projection( iface, contract_id, phase_note="resumed after real interruption") operation_id = f"t15-resume-{contract_id}" ew.register_process_wait( operation_id, contract_id=contract_id, project=project, objective=REQUEST, watched_for="builder exit event", watchdog_secs=900) builder_cmd = ["mimo", "run", "-m", BUILDER_MODEL, "--format", "json", t14.builder_prompt(projection, resumed=True) + f" [op:{operation_id}]"] notifier = ew.run_builder_under_events(builder_cmd, operation_id, daemon_url=DAEMON_URL) log("driver YIELDS on the bridge — no polling") result = ew.await_resume_event(operation_id, watchdog_secs=900, pgid=notifier.pid) if result["kind"] == "watchdog": rec = ew.resume_or_recover(operation_id) if rec["kind"] == "event": log(f"watchdog fired but canonical state shows the resume " f"build completed (verdict={rec['wait'].get('verdict')})") return finish_via_canaries(store, project, contract_id, project_dir, before, "resume via canonical recovery") log("PROCESS_LOST on resume — UNKNOWN_INTERRUPTION") return 4 return finish_via_canaries(store, project, contract_id, project_dir, before, "fresh-builder event-driven resume") def cmd_finish(store: Path, project_dir: Path) -> int: """Complete lifecycle from CANONICAL state only (waitstore verdict + filesystem reality). Used after a watchdog→canonical recovery.""" found = discover(store) assert found, "no contract in canonical state" project, contract_id = found before = t14.snapshot_files(project_dir) return finish_via_canaries(store, project, contract_id, project_dir, before, "canonical-only completion") def cmd_status(store: Path) -> int: found = discover(store) if not found: print(json.dumps({"active": False})) return 0 project, contract_id = found iface = ci.ConstructionInterface( builder="inspector", project_store=str(store / "projects" / project), project_id=project) print(json.dumps(iface.construction_context(contract_id)["state"], indent=1, default=str)) return 0 def main() -> int: import argparse ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("mode", choices=["start", "resume", "finish", "status"]) ap.add_argument("--store", required=True) ap.add_argument("--project-dir", required=False) ap.add_argument("--interrupt-after-first-file", action="store_true") args = ap.parse_args() store = Path(args.store).resolve() project_dir = (Path(args.project_dir).resolve() if args.project_dir else None) if project_dir: project_dir.mkdir(parents=True, exist_ok=True) if args.mode == "start": return cmd_start(store, project_dir, args.interrupt_after_first_file) if args.mode == "resume": return cmd_resume(store, project_dir) if args.mode == "finish": return cmd_finish(store, project_dir) return cmd_status(store) if __name__ == "__main__": sys.exit(main())