Something went wrong. Try again.
Personal fork of prime agent. Mirror of: https://mnemosyne.sigilzero.dev/sigilzero/prime-agent
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420#!/usr/bin/env python3"""Run the full workspace gate with private state and no inherited credentials.
Enter `nix develop --no-update-lock-file` first. Dependency downloads are allowed;provider/MCP credentials are not. One target/cache is reused, never per-lane copies."""import argparseimport tomllibimport fcntlimport hashlibimport jsonimport osfrom pathlib import Pathimport signalimport shutilimport shleximport subprocessimport sysimport tempfileimport timefrom urllib.parse import unquote, urlparse
# Keep the pinned shell's compiler/linker inputs, not the caller's credentials.TOOL_ENV = ( "PATH", "LANG", "LC_ALL", "TZ", "NIX_CC", "NIX_BINTOOLS", "NIX_CFLAGS_COMPILE", "NIX_LDFLAGS", "NIX_SSL_CERT_FILE", "SSL_CERT_FILE", "SSL_CERT_DIR", "PKG_CONFIG_PATH", "PKG_CONFIG_LIBDIR", "LIBRARY_PATH", "LD_LIBRARY_PATH", "RUST_SRC_PATH", "PA_WORKSPACE_BOOTSTRAP_PYTHON", "PA_TS_BINARY", "PA_TS_REFERENCE", "PA_WORKSPACE_TEST_SLOTS",)KERNEL_IMPORTS = ( "rlm", "dill", "requests", "httpx", "yaml", "tomli", "dotenv", "pandas", "numpy", "scipy", "bs4", "lxml", "pydantic", "tyro", "mcp",)
# Every bundled Python package, including skills used by scripted Rust fixtures.SKILL_APIS = { "agent_message": ("send",), "agent_observe": ("list_agents", "get_agent", "recent_messages"), "attach_image": ("run",), "compact": ("status", "run"), "edit": ("run",), "goal": ("get", "create", "complete"), "rlm_heartbeat": ("list", "create", "update", "delete"), "websearch": ("run",),}
def bundled_skills(repo): skills = [] for manifest in sorted((repo / "skills").glob("*/pyproject.toml")): data = tomllib.loads(manifest.read_text()) packages = data["tool"]["hatch"]["build"]["targets"]["wheel"]["packages"] for package in packages: module = Path(package).name if module not in SKILL_APIS: raise ValueError(f"bundled skill needs an API qualification contract: {module}") skills.append((data["project"]["name"], module, manifest.parent, package)) if {module for _, module, _, _ in skills} != set(SKILL_APIS): raise ValueError("bundled Python skill manifests do not match qualification contracts") return skills
def skill_identity(source, package): digest = hashlib.sha256() digest.update((Path(source) / "pyproject.toml").read_bytes()) digest.update(b"\0") digest.update(package_identity(Path(source) / package).encode()) return digest.hexdigest()
def probe_kernel(python, code, repo, env, deadline): # -B suppresses writes, not reads. A fresh cache prefix also prevents # timestamp-valid stale bytecode in an approved venv from being loaded. private = Path(env["PRIME_AGENT_KERNEL_VENV"]).parent with tempfile.TemporaryDirectory(prefix="probe-pycache-", dir=private) as cache: return run([str(python), "-I", "-B", "-X", f"pycache_prefix={cache}", "-c", code], repo=repo, env=env, deadline=deadline, capture=True)
def verify_skills(repo, python, env, deadline, skills): # Isolated imports suppress user-site/PYTHONPATH and bytecode writes even for # approved environments. Read metadata and the module that Python loaded. contracts = [(distribution, module, SKILL_APIS[module]) for distribution, module, _, _ in skills] code = f"""import importlib, importlib.metadata, jsonfrom pathlib import Pathrows = {{}}for distribution, name, apis in {contracts!r}: module = importlib.import_module(name) if not all(callable(getattr(module, api, None)) for api in apis): raise ValueError("bundled skill API missing: " + name) metadata = importlib.metadata.distribution(distribution) rows[name] = {{"direct_url": json.loads(metadata.read_text("direct_url.json")), "package_root": str(Path(module.__file__).resolve().parent)}}print(json.dumps(rows))""" observations = json.loads(probe_kernel(python, code, repo, env, deadline)) for _, module, expected, package in skills: observation = observations[module] url = urlparse(observation["direct_url"]["url"]) if url.scheme != "file" or url.netloc not in ("", "localhost"): raise ValueError(f"bundled skill {module} requires local-source installation metadata") source = Path(unquote(url.path)) if skill_identity(source, package) != skill_identity(expected, package): raise ValueError(f"bundled skill {module} source identity mismatch") if package_identity(observation["package_root"]) != package_identity(expected / package): raise ValueError(f"actually imported bundled skill {module} bytes mismatch")
def runtime_identity(source): source = Path(source) files = [source / "pyproject.toml", source / "uv.lock"] files.extend((source / "src/rlm").rglob("*.py")) machines = source / "src/rlm/machines" if machines.is_dir(): files.extend(path for path in machines.rglob("*") if path.is_file() and path.suffix != ".py") digest = hashlib.sha256() for path in sorted(files): digest.update(path.relative_to(source).as_posix().encode()) digest.update(b"\0") digest.update(path.read_bytes()) digest.update(b"\0") return digest.hexdigest()
def package_identity(package): """Hash imported package bytes independently of editable/wheel installation layout.""" package = Path(package) if not (package / "__init__.py").is_file(): raise ValueError("imported kernel runtime package is missing __init__.py") files = set(package.rglob("*.py")) machines = package / "machines" if machines.is_dir(): files.update(path for path in machines.rglob("*") if path.is_file()) digest = hashlib.sha256() for path in sorted(files): relative = path.relative_to(package) if "__pycache__" in relative.parts or path.suffix == ".pyc": continue digest.update(relative.as_posix().encode()) digest.update(b"\0") digest.update(path.read_bytes()) digest.update(b"\0") return digest.hexdigest()
def isolated_environment(repo, state, target, cargo_home, ambient): env = {key: ambient[key] for key in TOOL_ENV if key in ambient} home = state / "home" kernel = home / ".prime/agent/kernel-venv" env.update({ "HOME": str(home), "USER": "workspace-fixture", "LOGNAME": "workspace-fixture", "TMPDIR": str(state / "tmp"), "TMP": str(state / "tmp"), "TEMP": str(state / "tmp"), "XDG_CONFIG_HOME": str(state / "config"), "XDG_CACHE_HOME": str(state / "cache"), "XDG_DATA_HOME": str(state / "data"), "XDG_STATE_HOME": str(state / "state"), "XDG_RUNTIME_DIR": str(state / "run"), "PRIME_AGENT_CODING_AGENT_DIR": str(state / "agent"), "PRIME_AGENT_KERNEL_VENV": str(kernel), "PRIME_AGENT_DAEMON_SOCKET": str(state / "run/daemon.sock"), "DO_NOT_TRACK": "1", "CARGO_TARGET_DIR": str(target), "CARGO_HOME": str(cargo_home), "CARGO_BUILD_JOBS": "8", "CARGO_INCREMENTAL": "0", "CARGO_PROFILE_DEV_DEBUG": "0", "CARGO_PROFILE_TEST_DEBUG": "0", "RUST_TEST_THREADS": "4", "PYTHONDONTWRITEBYTECODE": "1", "UV_CACHE_DIR": str(cargo_home / "uv-cache"), "UV_PYTHON_INSTALL_DIR": str(cargo_home / "uv-python"), "UV_PROJECT_ENVIRONMENT": str(kernel), }) return env
def require_ts_reference(env): binary = Path(env.get("PA_TS_BINARY", "")) if not binary.is_absolute() or not binary.is_file() or not os.access(binary, os.X_OK): raise ValueError("PA_TS_BINARY must name an existing absolute executable for required parity") reference = Path(env.get("PA_TS_REFERENCE", "")) catalog = reference / "packages/coding-agent/src/core/prime-inference-model-catalog.ts" if not reference.is_absolute() or not catalog.is_file(): raise ValueError("PA_TS_REFERENCE must name the TypeScript source checkout for required parity")
def run(command, *, repo, env, deadline, capture=False): print("+ " + " ".join(map(str, command)), flush=True) started = time.monotonic() child = subprocess.Popen(command, cwd=repo, env=env, start_new_session=True, stdout=subprocess.PIPE if capture else None, text=True) try: output, _ = child.communicate(timeout=max(0.01, deadline - time.monotonic())) except (subprocess.TimeoutExpired, KeyboardInterrupt): os.killpg(child.pid, signal.SIGTERM) try: child.communicate(timeout=5) except subprocess.TimeoutExpired: os.killpg(child.pid, signal.SIGKILL) child.communicate() raise print(f"Elapsed {time.monotonic() - started:.2f}s (exit {child.returncode}): " + " ".join(map(str, command)), flush=True) if child.returncode: raise subprocess.CalledProcessError(child.returncode, command) return output
def prepare_kernel(repo, env, deadline, approved_python=None): kernel = Path(env["PRIME_AGENT_KERNEL_VENV"]) kernel.parent.mkdir(parents=True, exist_ok=True) skills = bundled_skills(repo) if approved_python is None: if kernel.exists() or kernel.is_symlink(): raise ValueError("cold qualification requires a NEW disposable kernel directory") declared_python = env.get("PA_WORKSPACE_BOOTSTRAP_PYTHON") if not declared_python: raise ValueError("PA_WORKSPACE_BOOTSTRAP_PYTHON must name the Nix dev shell's Python 3.11 executable") bootstrap_python = Path(declared_python) if not bootstrap_python.is_absolute() or not bootstrap_python.is_file() or not os.access(bootstrap_python, os.X_OK): raise ValueError(f"PA_WORKSPACE_BOOTSTRAP_PYTHON must be an existing absolute executable: {declared_python}") run(["uv", "sync", "--locked", "--no-default-groups", "--group", "kernel", "--no-config", "--project", str(repo / "prime-agent-runtime"), "--python", str(bootstrap_python), "--no-python-downloads"], repo=repo, env=env, deadline=deadline) python = kernel / "bin/python" # The locked kernel group must provide every skill dependency first. # Install only exact checkout skill code; do not resolve dependencies or # reinstall the already-provenanced runtime outside its locked sync. run(["uv", "pip", "install", "--no-config", "--no-python-downloads", "--python", str(python), "--no-deps", *[str(source) for _, _, source, _ in skills]], repo=repo, env=env, deadline=deadline) else: python = approved_python.absolute() # Dependency consistency check is read-only in both cold and approved paths. run(["uv", "pip", "check", "--no-config", "--python", str(python)], repo=repo, env=env, deadline=deadline) # -I ignores PYTHONPATH and user site packages. A noneditable install can # retain matching direct_url metadata while its copied package is stale. # Check the bytes actually imported, not just the recorded source directory. code = ("import importlib, importlib.metadata, json; from pathlib import Path; " f"[importlib.import_module(name) for name in {KERNEL_IMPORTS!r}]; " "d=importlib.metadata.distribution('prime-agent-runtime'); " "print(json.dumps({'direct_url': json.loads(d.read_text('direct_url.json')), " "'package_root': str(Path(importlib.import_module('rlm').__file__).resolve().parent)}))") observation = json.loads(probe_kernel(python, code, repo, env, deadline)) url = urlparse(observation["direct_url"]["url"]) if url.scheme != "file" or url.netloc not in ("", "localhost"): raise ValueError("kernel runtime must have local-source installation metadata") source = Path(unquote(url.path)) if runtime_identity(source) != runtime_identity(repo / "prime-agent-runtime"): raise ValueError("approved kernel runtime does not match this workspace's source identity") if package_identity(observation["package_root"]) != package_identity(repo / "prime-agent-runtime/src/rlm"): raise ValueError("actually imported kernel runtime does not match this workspace's package bytes") verify_skills(repo, python, env, deadline, skills) if approved_python is not None: # Tests without an override still find the same approved interpreter. kernel.symlink_to(python.parent.parent, target_is_directory=True) env["PRIME_AGENT_KERNEL_PYTHON"] = str(kernel / "bin/python") env["PA_E2E_KERNEL_PYTHON"] = str(kernel / "bin/python") print("Verified runtime and bundled skill source, imported bytes, and API contracts", flush=True)
VENDOR_MLUA_FEATURES = "lua54,vendored,async,send,serialize"VENDOR_MLUA_CONTRACTS = ( "route_migration_forwards_to_latest_parent_and_clears_on_drop", "real_wake_debt_before_during_after_ack_is_not_consumed_by_forwarding", "nested_waker_guards_restore_both_bindings", "never_ready_no_wake_future_does_not_schedule_marker_spin", "async_thread_migration_routes_registered_proxy_without_rust_repoll", "concurrent_route_registration_and_real_wake_forward_once", "actual_rust_poll_ack_preserves_concurrent_notification",)VENDOR_MLUA_HOOKS = ( "zero_result_hook_yields_preserve_concat_arguments_and_frame_locals", "async_stream_keeps_ordinary_yield_payloads_and_initial_arguments", "async_c_yield_preserves_initial_arguments_and_return_values", "counted_hook_async_handoff_timeline", "counted_hook_timer_control",)
def vendor_tree_identity(source): if source.is_symlink() or not source.is_dir(): raise ValueError("vendored mlua source root must be a real directory") rows = {} for path in sorted(source.rglob("*")): if path.is_symlink(): raise ValueError("vendored mlua source must not contain symlinks") if path.is_file(): rows[path.relative_to(source).as_posix()] = hashlib.sha256(path.read_bytes()).hexdigest() if not rows or "Cargo.toml" not in rows: raise ValueError("vendored mlua source tree is incomplete") return rows
def prepare_vendor_mlua(repo, state, env, deadline): # Keep the copied package outside the Rust workspace without manifest mutation. if state.resolve().is_relative_to(repo.resolve()): raise ValueError("vendor qualification state must be outside the workspace") source = repo / "vendor/mlua" expected = vendor_tree_identity(source) lock = repo / "vendor/mlua/Cargo.lock" provenance = json.loads((repo / "scripts/vendor-mlua-lock.json").read_text()) if (provenance.get("schema") != 1 or provenance.get("backend") != "lua54" or provenance.get("features") != VENDOR_MLUA_FEATURES.split(",") or provenance.get("manifest_sha256") != expected["Cargo.toml"] or provenance.get("lock_sha256") != hashlib.sha256(lock.read_bytes()).hexdigest()): raise ValueError("vendored mlua committed lock provenance mismatch") copied = state / "vendor-mlua" shutil.copytree(source, copied) if vendor_tree_identity(copied) != expected or vendor_tree_identity(source) != expected: raise ValueError("vendored mlua copied source identity mismatch") (state / "vendor-mlua-source-identities.json").write_text(json.dumps(expected, sort_keys=True)) run(["cargo", "fetch", "--locked", "--manifest-path", str(copied / "Cargo.toml")], repo=repo, env=env, deadline=deadline) return copied, expected
def verify_vendor_mlua_copy(repo, copied, expected): if (copied / "Cargo.lock").read_bytes() != (repo / "vendor/mlua/Cargo.lock").read_bytes(): raise ValueError("vendored mlua copied lock identity mismatch") observed = vendor_tree_identity(copied) # Build outputs are in the existing shared target, never in copied source. if observed != expected or vendor_tree_identity(repo / "vendor/mlua") != expected: raise ValueError("vendored mlua source changed after preparation")
def check_vendor_mlua(repo, copied, expected, env, deadline): verify_vendor_mlua_copy(repo, copied, expected) command = ["cargo", "test", "--locked", "--offline", "--manifest-path", str(copied / "Cargo.toml"), "--no-default-features", "--features", VENDOR_MLUA_FEATURES] listing = run(command + ["--lib", "--", "--list"], repo=repo, env=env, deadline=deadline, capture=True) for name in VENDOR_MLUA_CONTRACTS: if "thread::wake_debt_tests::" + name + ": test" not in listing.splitlines(): raise ValueError("vendored mlua contract test missing: " + name) hooks = run(command + ["--test", "hooks", "--", "--list"], repo=repo, env=env, deadline=deadline, capture=True) for name in VENDOR_MLUA_HOOKS: if name + ": test" not in hooks.splitlines(): raise ValueError("vendored mlua hook regression missing: " + name) for name in VENDOR_MLUA_CONTRACTS: run(command + ["--lib", "thread::wake_debt_tests::" + name, "--", "--exact"], repo=repo, env=env, deadline=deadline) # Full Lua54 hook compatibility, not only the newly introduced regressions. run(command + ["--test", "hooks"], repo=repo, env=env, deadline=deadline) run(command + ["--test", "async"], repo=repo, env=env, deadline=deadline) run(command + ["--test", "thread"], repo=repo, env=env, deadline=deadline) verify_vendor_mlua_copy(repo, copied, expected)
def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--state-root", type=Path, help="new private directory; retained for logs/diagnosis") parser.add_argument("--target-dir", type=Path, help="one existing Cargo target to reuse (default: workspace target)") parser.add_argument("--cargo-home", type=Path, help="dedicated credential-free Cargo cache (default: target/workspace-cargo)") parser.add_argument("--kernel-python", type=Path, help="approved test venv; source identity and kernel extras are checked") parser.add_argument("--require-ts-reference", action="store_true", help="fail unless explicit TS binary and source reference are provisioned") parser.add_argument("--timeout", type=float, default=10800, help="total deadline in seconds (default: 10800)") parser.add_argument("--skip-lint", action="store_true", help="the caller already ran formatting, hook and Clippy gates") args = parser.parse_args() if args.timeout <= 0: parser.error("--timeout must be positive") repo = Path(__file__).resolve().parents[1] target = (args.target_dir or repo / "target").absolute() cargo_home = (args.cargo_home or target / "workspace-cargo").absolute() if args.state_root: state = args.state_root.absolute() state.mkdir(mode=0o700) # Refuse reuse of an old state/credential tree. else: state = Path(tempfile.mkdtemp(prefix="pa-workspace-", dir="/tmp")) print(f"Workspace state: {state}\nCargo target: {target}\nCargo cache: {cargo_home}", flush=True) target.mkdir(parents=True, exist_ok=True) cargo_home.mkdir(parents=True, exist_ok=True) if any((cargo_home / name).exists() for name in ("credentials", "credentials.toml", "config", "config.toml")): raise ValueError("Cargo cache must not contain credentials or caller-owned registry configuration") with (target / ".downstream-workspace-check.lock").open("a") as lock: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) env = isolated_environment(repo, state, target, cargo_home, os.environ) if args.require_ts_reference: require_ts_reference(env) for key in ("HOME", "TMPDIR", "XDG_RUNTIME_DIR", "PRIME_AGENT_CODING_AGENT_DIR"): Path(env[key]).mkdir(parents=True, exist_ok=True, mode=0o700) deadline = time.monotonic() + args.timeout prepare_kernel(repo, env, deadline, args.kernel_python) # Fetch the complete lockfile, including test-only dependencies, before # the offline all-target Clippy and test commands run in a cold CI cache. run(["cargo", "fetch", "--locked"], repo=repo, env=env, deadline=deadline) vendor_copy, vendor_identity = prepare_vendor_mlua(repo, state, env, deadline) run(["cargo", "build", "--locked", "--workspace", "--bins"], repo=repo, env=env, deadline=deadline) test_env = dict(env, CARGO_NET_OFFLINE="true", UV_OFFLINE="1") check_vendor_mlua(repo, vendor_copy, vendor_identity, test_env, deadline) test_env["WORKSPACE_TEST_COMMAND"] = shlex.join([ "python3", "-B", "scripts/check_workspace_tests.py", "--state-root", str(state / "test-targets"), "--deadline", str(deadline), ]) gate = "workspace-tests" if args.skip_lint else "workspace-check" run(["make", gate], repo=repo, env=test_env, deadline=deadline) print("Full workspace gate passed", flush=True) return 0
if __name__ == "__main__": try: raise SystemExit(main()) except (OSError, ValueError, subprocess.SubprocessError) as error: print(f"Workspace gate failed: {error}", file=sys.stderr) raise SystemExit(1) from error