v1_revision.py
python
sha256:ff22b6a82a08c117176c3e2603264a0e49fe8204ce57c3cefe7f5b165a8634f8
docs: recenter public story on governance loop
Human
6 days ago
| 1 | """Explicit local revision selection. Git never imports or launches Muse.""" |
| 2 | from __future__ import annotations |
| 3 | |
| 4 | from dataclasses import dataclass |
| 5 | import json |
| 6 | import os |
| 7 | from pathlib import Path |
| 8 | import re |
| 9 | import selectors |
| 10 | import subprocess |
| 11 | import time |
| 12 | |
| 13 | from cli.v1_io import Refusal |
| 14 | |
| 15 | MUSE_VERSION = "0.2.1rc5" |
| 16 | OID = re.compile(r"sha256:[0-9a-f]{64}\Z") |
| 17 | RESPONSE_LIMIT = 2 * 1024 * 1024 |
| 18 | |
| 19 | |
| 20 | def bounded_run(argv, *, cwd, env, timeout=15, limit=RESPONSE_LIMIT): |
| 21 | """Bound elapsed time and combined output before accumulating a response.""" |
| 22 | with subprocess.Popen(argv, cwd=cwd, env=env, stdout=subprocess.PIPE, |
| 23 | stderr=subprocess.PIPE) as proc: |
| 24 | output = bytearray() |
| 25 | count = 0 |
| 26 | deadline = time.monotonic() + timeout |
| 27 | try: |
| 28 | with selectors.DefaultSelector() as selector: |
| 29 | selector.register(proc.stdout, selectors.EVENT_READ) |
| 30 | selector.register(proc.stderr, selectors.EVENT_READ) |
| 31 | while selector.get_map(): |
| 32 | remaining = deadline - time.monotonic() |
| 33 | if remaining <= 0: |
| 34 | raise Refusal("muse_read_timeout") |
| 35 | for key, _ in selector.select(remaining): |
| 36 | chunk = os.read(key.fileobj.fileno(), 65536) |
| 37 | if not chunk: |
| 38 | selector.unregister(key.fileobj) |
| 39 | continue |
| 40 | count += len(chunk) |
| 41 | if count > limit: |
| 42 | raise Refusal("muse_response_too_large") |
| 43 | if key.fileobj is proc.stdout: |
| 44 | output.extend(chunk) |
| 45 | proc.wait(timeout=max(0.01, deadline - time.monotonic())) |
| 46 | return subprocess.CompletedProcess(argv, proc.returncode, bytes(output), b"") |
| 47 | finally: |
| 48 | if proc.poll() is None: |
| 49 | proc.kill() |
| 50 | proc.wait() |
| 51 | |
| 52 | |
| 53 | def git(root: Path, *args: str, optional: bool = False) -> str: |
| 54 | env = {k: v for k, v in os.environ.items() if not k.startswith("GIT_")} |
| 55 | env.update(PATH="/usr/bin:/bin", GIT_OPTIONAL_LOCKS="0", GIT_CONFIG_NOSYSTEM="1", |
| 56 | GIT_CONFIG_GLOBAL="/dev/null", LC_ALL="C") |
| 57 | proc = subprocess.run(["/usr/bin/git", "-c", "core.fsmonitor=false", "-c", |
| 58 | "core.hooksPath=/dev/null", "-C", str(root), *args], |
| 59 | env=env, capture_output=True, text=True, timeout=15) |
| 60 | if proc.returncode and not optional: |
| 61 | raise Refusal("git_read_failed: " + args[0]) |
| 62 | return proc.stdout.strip() if proc.returncode == 0 else "" |
| 63 | |
| 64 | |
| 65 | def discover(requested: Path) -> Path: |
| 66 | for root in (requested, *requested.parents): |
| 67 | # Stop at the nearest VCS boundary, including invalid markers. Do not |
| 68 | # fall through a Muse checkout to an unrelated parent Git repository. |
| 69 | if any(os.path.lexists(root / p) for p in (".git", ".muse")): |
| 70 | return root |
| 71 | raise Refusal("repository_not_found") |
| 72 | |
| 73 | |
| 74 | def git_branch(root): |
| 75 | value = git(root, "symbolic-ref", "--quiet", "--short", "HEAD", optional=True) |
| 76 | if not value: |
| 77 | raise Refusal("detached_head") |
| 78 | return value |
| 79 | |
| 80 | |
| 81 | def git_head(root): |
| 82 | return git(root, "rev-parse", "--verify", "HEAD", optional=True) or "unborn" |
| 83 | |
| 84 | |
| 85 | def unique_json(pairs): |
| 86 | result = {} |
| 87 | for key, value in pairs: |
| 88 | if key in result: |
| 89 | raise Refusal("muse_response_duplicate_key") |
| 90 | result[key] = value |
| 91 | return result |
| 92 | |
| 93 | |
| 94 | def muse_read(root, python, *, dirty=False): |
| 95 | executable = Path(python) |
| 96 | if not executable.is_absolute() or executable.parent.name != "bin": |
| 97 | raise Refusal("muse_python_requires_absolute_venv_path") |
| 98 | # Preserve the conventional venv Python symlink; resolving it would select |
| 99 | # the base interpreter and lose the separate Muse installation. |
| 100 | executable = executable.parent.resolve() / executable.name |
| 101 | if not (executable.parent.parent / "pyvenv.cfg").is_file(): |
| 102 | raise Refusal("muse_python_requires_separate_venv") |
| 103 | env = {k: v for k, v in os.environ.items() |
| 104 | if not k.startswith(("GIT_", "MUSE_", "PYTHON")) and k != "VIRTUAL_ENV"} |
| 105 | env.update(PATH="/usr/bin:/bin", PYTHONDONTWRITEBYTECODE="1", LC_ALL="C") |
| 106 | reader = Path(__file__).with_name("v1_muse_reader.py") |
| 107 | proc = bounded_run([str(executable), "-I", "-B", str(reader), str(root), |
| 108 | "dirty" if dirty else "revision"], cwd=root, env=env) |
| 109 | if proc.returncode or len(proc.stdout) > RESPONSE_LIMIT: |
| 110 | raise Refusal("muse_read_failed") |
| 111 | try: |
| 112 | value = json.loads(proc.stdout, object_pairs_hook=unique_json) |
| 113 | required = {"root", "store", "repo_id", "branch", "head", "dirty", |
| 114 | "tracked_local", "runtime"} |
| 115 | if not isinstance(value, dict) or set(value) != required: |
| 116 | raise ValueError() |
| 117 | runtime = value["runtime"] |
| 118 | if (not isinstance(runtime, dict) or set(runtime) != {"python", "version", "sha256"} |
| 119 | or runtime["python"] != str(executable) or runtime["version"] != MUSE_VERSION |
| 120 | or not re.fullmatch(r"[0-9a-f]{64}", runtime["sha256"])): |
| 121 | raise ValueError() |
| 122 | if value["root"] != str(root) or not Path(value["store"]).is_absolute(): |
| 123 | raise ValueError() |
| 124 | if not OID.fullmatch(value["repo_id"]): |
| 125 | raise ValueError() |
| 126 | if value["head"] != "unborn" and not OID.fullmatch(value["head"]): |
| 127 | raise ValueError() |
| 128 | branch = value["branch"] |
| 129 | if not isinstance(branch, str) or not branch or len(branch) > 200 or any(ord(c) < 32 for c in branch): |
| 130 | raise ValueError() |
| 131 | if (dirty and type(value["dirty"]) is not bool) or (not dirty and value["dirty"] is not None): |
| 132 | raise ValueError() |
| 133 | if not isinstance(value["tracked_local"], list) or any(not isinstance(p, str) for p in value["tracked_local"]): |
| 134 | raise ValueError() |
| 135 | except (ValueError, TypeError, KeyError, UnicodeError): |
| 136 | raise Refusal("muse_response_invalid") from None |
| 137 | return value |
| 138 | |
| 139 | |
| 140 | @dataclass(frozen=True) |
| 141 | class Revision: |
| 142 | kind: str |
| 143 | branch: str |
| 144 | head: str |
| 145 | source_id: str | None = None |
| 146 | dirty: bool | None = None |
| 147 | tracked_local: tuple[str, ...] = () |
| 148 | |
| 149 | |
| 150 | def revision(root, config, *, dirty=False): |
| 151 | if config["vcs"] == "git": |
| 152 | if not os.path.lexists(root / ".git") or Path(git(root, "rev-parse", "--show-toplevel")).resolve() != root: |
| 153 | raise Refusal("git_repository_mismatch") |
| 154 | return Revision("git", git_branch(root), git_head(root), dirty=( |
| 155 | bool(git(root, "status", "--porcelain", "--untracked-files=normal")) if dirty else None)) |
| 156 | binding = config["muse"] |
| 157 | state = muse_read(root, binding["runtime"]["python"], dirty=dirty) |
| 158 | if state["repo_id"] != binding["repo_id"] or state["store"] != binding["store"]: |
| 159 | raise Refusal("muse_repository_mismatch") |
| 160 | if state["runtime"] != binding["runtime"]: |
| 161 | raise Refusal("muse_runtime_changed: explicitly adopt the reviewed Muse runtime") |
| 162 | return Revision("muse", state["branch"], state["head"], state["repo_id"], |
| 163 | state["dirty"], tuple(state["tracked_local"])) |
File History
1 commit
sha256:ff22b6a82a08c117176c3e2603264a0e49fe8204ce57c3cefe7f5b165a8634f8
docs: recenter public story on governance loop
Human
6 days ago