aaronrene / overseer-kit public
v1_muse_reader.py python
164 lines 7.6 KB
Raw
sha256:ff22b6a82a08c117176c3e2603264a0e49fe8204ce57c3cefe7f5b165a8634f8 docs: recenter public story on governance loop Human 6 days ago
1 """Read-only bridge to Muse 0.2.1rc5, executed in its own conventional venv.
2
3 Do not call the native CLI: require_repo performs GC and some stage readers
4 migrate files. Only object reads, ignore matching and hashing are reused here.
5 Linked worktree HEAD is resolved explicitly; native CLI discovery returns the
6 main store root. This reader never writes refs, caches, hooks or application files.
7 """
8 from __future__ import annotations
9
10 import hashlib
11 import importlib.metadata
12 import json
13 import os
14 from pathlib import Path
15 import re
16 import stat
17 import sys
18 import tomllib
19
20 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
21 from cli.v1_io import Refusal, read
22 from cli.v1_policy import local_path
23 from cli.v1_revision import MUSE_VERSION, OID, unique_json
24
25
26 def readonly(event, args):
27 # A compatibility assertion for pinned Python APIs, not a hostile-code
28 # sandbox. Refuse unexpected writes/network/child processes on this path.
29 if event == "open" and args[2] & (os.O_WRONLY | os.O_RDWR | os.O_CREAT | os.O_TRUNC | os.O_APPEND):
30 raise Refusal("muse_reader_write_attempt")
31 if event in {"os.remove", "os.rename", "os.rmdir", "os.mkdir", "os.chmod",
32 "os.link", "os.symlink", "os.truncate", "subprocess.Popen",
33 "os.system", "socket.connect", "socket.bind"}:
34 raise Refusal("muse_reader_side_effect")
35
36
37 def document(root, path):
38 return json.loads(read(root, path), object_pairs_hook=unique_json)
39
40
41 def layout(root):
42 marker = root / ".muse"
43 if marker.is_symlink():
44 raise Refusal("muse_marker_symlink")
45 if marker.is_dir():
46 return root, ".muse/HEAD", None
47 pointer = read(root, ".muse").decode().strip()
48 if not pointer.startswith("musestore: "):
49 raise Refusal("muse_pointer_invalid")
50 path = Path(pointer[len("musestore: "):])
51 if not path.is_absolute() or path.name != ".muse" or path.resolve() != path or not path.is_dir():
52 raise Refusal("muse_pointer_invalid")
53 store = path.parent
54 matches = []
55 for entry in (path / "worktrees").glob("*.json"):
56 data = document(store, str(entry.relative_to(store)))
57 if data.get("path") == str(root):
58 name = data.get("name")
59 if not isinstance(name, str) or not re.fullmatch(r"[A-Za-z0-9_.-]+", name) or entry.stem != name:
60 raise Refusal("muse_worktree_invalid")
61 matches.append((".muse/worktrees/" + name + ".HEAD", data.get("branch")))
62 if len(matches) != 1:
63 raise Refusal("muse_worktree_unregistered")
64 return store, *matches[0]
65
66
67 def run(root, with_dirty):
68 sys.addaudithook(readonly)
69 if sys.prefix == sys.base_prefix or importlib.metadata.version("muse") != MUSE_VERSION:
70 raise Refusal("muse_version_unsupported")
71 import muse
72 package = Path(muse.__file__).resolve().parent
73 if not package.is_relative_to(Path(sys.prefix).resolve()):
74 raise Refusal("muse_import_origin_mismatch")
75 source_hash = hashlib.sha256()
76 for path in sorted(package.rglob("*.py")):
77 source_hash.update(str(path.relative_to(package)).encode() + b"\0" + path.read_bytes() + b"\0")
78 from muse.core.commits import read_commit
79 from muse.core.snapshots import read_snapshot
80 from muse.core.validation import validate_branch_name
81
82 store, head_path, registered_branch = layout(root)
83 identity = document(store, ".muse/repo.json")
84 if identity.get("domain") != "code" or not OID.fullmatch(identity.get("repo_id", "")):
85 raise Refusal("muse_identity_or_domain_unsupported")
86 head_data = read(store, head_path)
87 symbolic = head_data.decode().strip()
88 prefix = "ref: refs/heads/" if registered_branch is None else "refs/heads/"
89 if not symbolic.startswith(prefix):
90 raise Refusal("muse_detached_or_malformed_head")
91 branch = symbolic[len(prefix):]
92 validate_branch_name(branch)
93 if registered_branch is not None and registered_branch != branch:
94 raise Refusal("muse_worktree_branch_mismatch")
95 ref_path = ".muse/refs/heads/" + branch
96 ref = read(store, ref_path, missing=True)
97 head = "unborn" if ref in (None, b"") else ref.decode().strip()
98 if head != "unborn" and not OID.fullmatch(head):
99 raise Refusal("muse_ref_invalid")
100 for name in ("CHECKOUT_HEAD", "MERGE_STATE.json"):
101 if (store / ".muse" / name).exists():
102 raise Refusal("muse_checkout_or_merge_in_progress")
103 manifest = {}
104 directories = []
105 if head != "unborn":
106 commit = read_commit(store, head)
107 snapshot = read_snapshot(store, commit.snapshot_id) if commit else None
108 if snapshot is None:
109 raise Refusal("muse_revision_objects_missing_or_corrupt")
110 manifest = snapshot.manifest
111 directories = snapshot.directories
112 for path, oid in manifest.items():
113 if Path(path).is_absolute() or ".." in Path(path).parts or not OID.fullmatch(oid):
114 raise Refusal("muse_manifest_invalid")
115 # Never call read_stage: its legacy-format path deletes the old index.
116 stage_raw = read(store, ".muse/code/stage.json", missing=True)
117 stage = json.loads(stage_raw, object_pairs_hook=unique_json) if stage_raw else {"entries": {}}
118 if not isinstance(stage, dict) or not isinstance(stage.get("entries"), dict):
119 raise Refusal("muse_stage_invalid")
120 tracked = sorted(p for p in set(manifest) | set(stage["entries"]) if local_path(p))
121 dirty = None
122 if with_dirty:
123 from muse.core.ignore import is_ignored, resolve_patterns
124 from muse.core.snapshot import _BUILTIN_SECRET_PATTERNS
125 from muse.core.types import hash_file
126 from muse.plugins.code.plugin import _ALWAYS_IGNORE_DIRS
127
128 raw_ignore = read(root, ".museignore", missing=True)
129 ignore = tomllib.loads(raw_ignore.decode()) if raw_ignore else {}
130 patterns = _BUILTIN_SECRET_PATTERNS + resolve_patterns(ignore, "code")
131 current = {}
132 for directory, dirs, files in os.walk(root, followlinks=False):
133 here = Path(directory)
134 rel = here.relative_to(root)
135 dirs[:] = [d for d in dirs if d not in _ALWAYS_IGNORE_DIRS
136 and not (here / d).is_symlink() and not (here / d / ".muse").exists()
137 and not is_ignored((rel / d / "_").as_posix(), patterns)]
138 for name in files:
139 if name in (".muse", ".git"):
140 continue
141 path = here / name
142 relative = path.relative_to(root).as_posix()
143 if is_ignored(relative, patterns):
144 continue
145 if stat.S_ISREG(path.lstat().st_mode):
146 current[relative] = hash_file(path)
147 # Staging is shared by this Muse version. It is reported dirty in any
148 # linked checkout too; do not mistake the primary index for a clean lane.
149 dirty = bool(stage["entries"]) or current != manifest or any(
150 not (root / d).is_dir() for d in directories)
151 if read(store, head_path) != head_data or read(store, ref_path, missing=True) != ref:
152 raise Refusal("muse_revision_changed_during_read")
153 return {"root": str(root), "store": str(store), "repo_id": identity["repo_id"],
154 "branch": branch, "head": head, "dirty": dirty, "tracked_local": tracked,
155 "runtime": {"python": str(Path(sys.prefix).resolve() / "bin" / Path(sys.executable).name),
156 "version": MUSE_VERSION, "sha256": source_hash.hexdigest()}}
157
158
159 if __name__ == "__main__":
160 try:
161 print(json.dumps(run(Path(sys.argv[1]), sys.argv[2] == "dirty")))
162 except Exception as exc:
163 print("muse reader refused: " + (str(exc) if isinstance(exc, Refusal) else type(exc).__name__), file=sys.stderr)
164 raise SystemExit(2)
File History 1 commit
sha256:ff22b6a82a08c117176c3e2603264a0e49fe8204ce57c3cefe7f5b165a8634f8 docs: recenter public story on governance loop Human 6 days ago