Skills › Compound Engineering › ce-work › unit_workspace_state.py
unit_workspace_state.py
1070 lines · skills/ce-work/scripts/unit_workspace_state.py at 3449230
1"""Private, crash-recoverable workspace controller for ce-work external units.
3The generic peer-job runner owns process supervision. This controller owns the
4repository-specific transaction: one private run manifest, detached sibling
5worktrees, complete-tree transport commits, canonical integration evidence,
6exact restoration, retention, and explicit cleanup. It never launches a model
7CLI and never commits a worker's output in the canonical checkout.
9Every successful command prints a status word and one compact JSON document.
10Trust failures print only ``UNREADABLE`` and an error on stderr.
11"""
13from __future__ import annotations
15import argparse
16import base64
17import contextlib
18import fcntl
19import hashlib
20import json
21import os
22import re
23import secrets
24import shutil
25import stat
26import subprocess
27import sys
28import tempfile
29import time
30from pathlib import Path
33SCHEMA_VERSION = 1
34PLAN_CHECKPOINT_MESSAGE = "docs(ce-work): checkpoint selected implementation plan"
35_uid_getter = getattr(os, "geteuid", None) or getattr(os, "getuid", None)
36_EFFECTIVE_UID = _uid_getter() if _uid_getter is not None else None
37OWNER_SCRATCH_ROOT = (
38 os.path.join("/tmp", f"compound-engineering-{_EFFECTIVE_UID}")
39 if _EFFECTIVE_UID is not None
40 else None
41)
42MAX_JSON_BYTES = 2 * 1024 * 1024
43MAX_PACKET_BYTES = 200_000
44SAFE_ID = re.compile(r"^[A-Za-z0-9._-]{1,128}$")
45O_NOFOLLOW = getattr(os, "O_NOFOLLOW", 0)
46O_DIRECTORY = getattr(os, "O_DIRECTORY", 0)
47TERMINAL_PROCESS = {"done", "failed", "timeout", "died-without-result"}
48INTEGRATABLE_STATES = {"integration-pending", "integrated", "verified"}
49UNIT_STATES = {
50 "queued", "authoring", "authored", "integration-pending", "integrated",
51 "restoring", "verified", "committed", "preserved", "cleaned", "native-completed",
52}
53GIT_LOCAL_ENV_VARS = frozenset({
54 "GIT_ALTERNATE_OBJECT_DIRECTORIES",
55 "GIT_COMMON_DIR",
56 "GIT_CONFIG",
57 "GIT_CONFIG_COUNT",
58 "GIT_CONFIG_PARAMETERS",
59 "GIT_DIR",
60 "GIT_GRAFT_FILE",
61 "GIT_IMPLICIT_WORK_TREE",
62 "GIT_INDEX_FILE",
63 # Listed by `git rev-parse --local-env-vars` on git < 2.41 (e.g. Debian
64 # bookworm's 2.39); dropped from the list by newer git. Scrub it so an
65 # inherited value cannot leak into unit verification on old git.
66 "GIT_INTERNAL_SUPER_PREFIX",
67 "GIT_NO_REPLACE_OBJECTS",
68 "GIT_OBJECT_DIRECTORY",
69 "GIT_PREFIX",
70 "GIT_REPLACE_REF_BASE",
71 "GIT_SHALLOW_FILE",
72 "GIT_WORK_TREE",
73})
76class Operational(Exception):
77 def __init__(self, word: str, message: str, detail: dict | None = None):
78 super().__init__(message)
79 self.word = word
80 self.detail = detail or {}
83class TrustFailure(Operational):
84 def __init__(self, message: str):
85 super().__init__("UNREADABLE", message)
88def now_iso() -> str:
89 return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
92def test_fault(point: str) -> None:
93 """Deterministic crash-window injection for the repository test suite."""
94 configured = {value.strip() for value in os.environ.get("CE_WORK_TEST_FAULT", "").split(",") if value.strip()}
95 if point in configured:
96 raise Operational("INTERRUPTED", f"injected test interruption at {point}")
99def _private_root_usable(path: str) -> bool:
100 """True when `path` is (or can now be) a directory we own and can write into.
102 Creation is the probe: a sandbox that denies writes under /tmp refuses the
103 mkdir, and one that lets a pre-existing root stand still fails the access
104 check, so both land on the fallback instead of failing at the first run.
105 """
106 try:
107 os.mkdir(path, 0o700)
108 except FileExistsError:
109 pass
110 except OSError:
111 return False
112 try:
113 st = os.lstat(path)
114 except OSError:
115 return False
116 if not stat.S_ISDIR(st.st_mode):
117 return False
118 if _EFFECTIVE_UID is not None and st.st_uid != _EFFECTIVE_UID:
119 return False
120 return os.access(path, os.W_OK)
123def _fallback_scratch_root() -> str:
124 return os.path.join(os.environ.get("TMPDIR") or "/tmp", f"compound-engineering-{_EFFECTIVE_UID}")
127def owner_scratch_root() -> str:
128 """The owner-private scratch root, in the same candidate order as the skills' shell preamble."""
129 if OWNER_SCRATCH_ROOT is None:
130 raise TrustFailure("effective user ID is unavailable; cannot derive the runs root")
131 if _private_root_usable(OWNER_SCRATCH_ROOT):
132 return OWNER_SCRATCH_ROOT
133 return _fallback_scratch_root()
136def runs_root() -> str:
137 configured = os.environ.get("CE_WORK_RUNS_ROOT")
138 if configured:
139 return os.path.abspath(configured)
140 peer_root = os.environ.get("CE_PEER_JOBS_ROOT")
141 if peer_root:
142 return os.path.join(os.path.abspath(peer_root), "ce-work")
143 return os.path.join(owner_scratch_root(), "ce-work")
146def safe_id(value: str, label: str) -> str:
147 if not SAFE_ID.fullmatch(value) or not value.strip("."):
148 raise Operational("REFUSED", f"unsafe {label}: {value!r}")
149 return value
152def digest_bytes(data: bytes) -> str:
153 return hashlib.sha256(data).hexdigest()
156def _valid_git_object_id(value: object) -> bool:
157 if not isinstance(value, str):
158 return False
159 try:
160 raw = bytes.fromhex(value)
161 except ValueError:
162 return False
163 return len(raw) in {20, 32} and raw.hex() == value
166def _native_completion_commit(unit: dict) -> str | None:
167 attempts = unit.get("attempts")
168 if not isinstance(attempts, list) or not attempts or not isinstance(attempts[-1], dict):
169 return None
170 fallback = attempts[-1].get("fallback")
171 if not isinstance(fallback, dict):
172 return None
173 claim = fallback.get("claimed")
174 completion = fallback.get("completed")
175 if not isinstance(claim, dict) or not isinstance(completion, dict) or completion.get("claim") != claim:
176 return None
177 claim_mode = claim.get("mode")
178 if claim_mode not in {"prefer", "require"}:
179 return None
180 accepted_head = completion.get("accepted_head")
181 base = unit.get("workspace", {}).get("base")
182 snapshot = completion.get("snapshot")
183 wave = unit.get("wave", {})
184 changed_paths = completion.get("changed_paths")
185 if not (
186 _valid_git_object_id(accepted_head)
187 and _valid_git_object_id(base)
188 and completion.get("base") == base
189 and isinstance(completion.get("at"), str)
190 and bool(completion["at"])
191 and isinstance(completion.get("summary"), str)
192 and bool(completion["summary"])
193 and isinstance(completion.get("evidence_digest"), str)
194 and len(completion["evidence_digest"]) == 64
195 and _valid_git_object_id(completion["evidence_digest"])
196 and isinstance(snapshot, dict)
197 and snapshot.get("head") == accepted_head
198 and snapshot.get("status_empty") is True
199 and snapshot.get("worktree_index_empty") is True
200 and _valid_git_object_id(snapshot.get("head_tree"))
201 and snapshot.get("head_tree") == snapshot.get("index_tree")
202 and snapshot.get("status_sha256") == digest_bytes(b"")
203 and (
204 not wave.get("id")
205 or (
206 _valid_git_object_id(claim.get("canonical_head"))
207 and isinstance(changed_paths, list)
208 and all(isinstance(path, str) for path in changed_paths)
209 )
210 )
211 ):
212 return None
213 return accepted_head
216def unit_accepted_commit(unit: dict) -> str | None:
217 if unit.get("state") == "native-completed":
218 return _native_completion_commit(unit)
219 if unit.get("state") != "cleaned":
220 return None
221 integration = unit.get("integration")
222 if not isinstance(integration, dict):
223 return None
224 canonical = integration.get("canonical_commit")
225 if not (
226 isinstance(canonical, dict)
227 and all(_valid_git_object_id(canonical.get(field)) for field in ("commit", "parent", "tree"))
228 and isinstance(canonical.get("at"), str)
229 and bool(canonical["at"])
230 ):
231 return None
232 return canonical["commit"]
235def unit_ready_for_run_verification(unit: object) -> bool:
236 return isinstance(unit, dict) and unit_accepted_commit(unit) is not None
239def accepted_unit_commit_snapshot(units: object) -> dict[str, str] | None:
240 if not isinstance(units, dict):
241 return None
242 snapshot: dict[str, str] = {}
243 for unit_id in sorted(units):
244 if not isinstance(unit_id, str) or not SAFE_ID.fullmatch(unit_id):
245 return None
246 unit = units[unit_id]
247 if not isinstance(unit, dict):
248 return None
249 commit = unit_accepted_commit(unit)
250 if commit is None:
251 return None
252 snapshot[unit_id] = commit
253 return snapshot
256def _mode(st: os.stat_result) -> int:
257 return stat.S_IMODE(st.st_mode)
260def _euid() -> int | None:
261 return _EFFECTIVE_UID
264def validate_private_dir(path: str) -> None:
265 try:
266 fd = os.open(path, os.O_RDONLY | O_DIRECTORY | O_NOFOLLOW)
267 except OSError as exc:
268 raise TrustFailure(f"cannot safely open directory {path}: {exc}") from exc
269 try:
270 st = os.fstat(fd)
271 if not stat.S_ISDIR(st.st_mode):
272 raise TrustFailure(f"not a real directory: {path}")
273 if _euid() is not None and st.st_uid != _euid():
274 raise TrustFailure(f"directory is not owned by current user: {path}")
275 if _mode(st) != 0o700:
276 raise TrustFailure(f"directory mode is {_mode(st):04o}, expected 0700: {path}")
277 finally:
278 os.close(fd)
281def ensure_private_dir(path: str) -> None:
282 try:
283 os.mkdir(path, 0o700)
284 except FileExistsError:
285 pass
286 validate_private_dir(path)
289def _owner_root_for_runs(root: str) -> str | None:
290 if OWNER_SCRATCH_ROOT is None:
291 return None
292 for candidate in (OWNER_SCRATCH_ROOT, _fallback_scratch_root()):
293 owner_root = os.path.abspath(candidate)
294 try:
295 if os.path.commonpath([owner_root, os.path.abspath(root)]) == owner_root:
296 return owner_root
297 except ValueError: # different drives on Windows: not under this candidate
298 continue
299 return None
302def _ensure_owner_scratch_root(path: str) -> None:
303 try:
304 os.mkdir(path, 0o700)
305 except FileExistsError:
306 pass
307 try:
308 fd = os.open(path, os.O_RDONLY | O_DIRECTORY | O_NOFOLLOW)
309 except OSError as exc:
310 raise TrustFailure(f"cannot safely open owner scratch root {path}: {exc}") from exc
311 try:
312 current = os.fstat(fd)
313 if not stat.S_ISDIR(current.st_mode):
314 raise TrustFailure(f"owner scratch root is not a real directory: {path}")
315 if _euid() is not None and current.st_uid != _euid():
316 raise TrustFailure(f"owner scratch root is not owned by current user: {path}")
317 if _mode(current) != 0o700:
318 os.fchmod(fd, 0o700)
319 repaired = os.fstat(fd)
320 if repaired.st_uid != current.st_uid or _mode(repaired) != 0o700:
321 raise TrustFailure(f"could not repair owner scratch root mode to 0700: {path}")
322 finally:
323 os.close(fd)
326def ensure_root() -> str:
327 return ensure_runs_root(runs_root())
330def ensure_runs_root(root: str) -> str:
331 """Create or verify one runs root (the creation root, or the other candidate
332 root an existing run was found under) and its private lock directory."""
333 owner_root = _owner_root_for_runs(root)
334 if owner_root is not None:
335 _ensure_owner_scratch_root(owner_root)
336 parent = os.path.dirname(root)
337 # The configured root's ancestors are caller-controlled; the private root
338 # itself and everything below it are the durable confidentiality boundary.
339 os.makedirs(parent, mode=0o700, exist_ok=True)
340 ensure_private_dir(root)
341 ensure_private_dir(os.path.join(root, ".locks"))
342 return root
345def read_private(path: str, cap: int = MAX_JSON_BYTES) -> bytes:
346 try:
347 fd = os.open(path, os.O_RDONLY | O_NOFOLLOW)
348 except OSError as exc:
349 raise TrustFailure(f"cannot safely open state file {path}: {exc}") from exc
350 try:
351 st = os.fstat(fd)
352 if not stat.S_ISREG(st.st_mode):
353 raise TrustFailure(f"state is not a regular file: {path}")
354 if _euid() is not None and st.st_uid != _euid():
355 raise TrustFailure(f"state is not owned by current user: {path}")
356 if _mode(st) != 0o600:
357 raise TrustFailure(f"state mode is {_mode(st):04o}, expected 0600: {path}")
358 if st.st_size > cap:
359 raise TrustFailure(f"state exceeds {cap}-byte limit: {path}")
360 out = bytearray()
361 while len(out) <= cap:
362 part = os.read(fd, min(65536, cap + 1 - len(out)))
363 if not part:
364 break
365 out.extend(part)
366 if len(out) > cap:
367 raise TrustFailure(f"state grew beyond {cap}-byte limit: {path}")
368 return bytes(out)
369 finally:
370 os.close(fd)
373def stat_private_file(path: str) -> os.stat_result:
374 """Validate a private file by descriptor without consuming its content."""
375 try:
376 fd = os.open(path, os.O_RDONLY | O_NOFOLLOW)
377 except OSError as exc:
378 raise TrustFailure(f"cannot safely open state file {path}: {exc}") from exc
379 try:
380 st = os.fstat(fd)
381 if not stat.S_ISREG(st.st_mode):
382 raise TrustFailure(f"state is not a regular file: {path}")
383 if _euid() is not None and st.st_uid != _euid():
384 raise TrustFailure(f"state is not owned by current user: {path}")
385 if _mode(st) != 0o600:
386 raise TrustFailure(f"state mode is {_mode(st):04o}, expected 0600: {path}")
387 return st
388 finally:
389 os.close(fd)
392def read_private_json(path: str) -> dict:
393 try:
394 value = json.loads(read_private(path))
395 except TrustFailure:
396 raise
397 except (ValueError, UnicodeDecodeError) as exc:
398 raise TrustFailure(f"malformed JSON state: {path}") from exc
399 if not isinstance(value, dict):
400 raise TrustFailure(f"JSON state is not an object: {path}")
401 return value
404def create_private(path: str, data: bytes) -> None:
405 try:
406 fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | O_NOFOLLOW, 0o600)
407 except OSError as exc:
408 raise Operational("BLOCKED", f"cannot exclusively create {path}: {exc}") from exc
409 try:
410 os.write(fd, data)
411 os.fsync(fd)
412 finally:
413 os.close(fd)
416def atomic_private_json(path: str, doc: dict) -> None:
417 data = (json.dumps(doc, sort_keys=True, separators=(",", ":")) + "\n").encode()
418 if len(data) > MAX_JSON_BYTES:
419 raise Operational("BLOCKED", "manifest exceeds bounded state size")
420 parent = os.path.dirname(path)
421 fd, tmp = tempfile.mkstemp(prefix=".manifest-", dir=parent)
422 try:
423 os.fchmod(fd, 0o600)
424 with os.fdopen(fd, "wb", closefd=True) as stream:
425 stream.write(data)
426 stream.flush()
427 os.fsync(stream.fileno())
428 os.replace(tmp, path)
429 dfd = os.open(parent, os.O_RDONLY | O_DIRECTORY | O_NOFOLLOW)
430 try:
431 os.fsync(dfd)
432 finally:
433 os.close(dfd)
434 except BaseException:
435 with contextlib.suppress(OSError):
436 os.unlink(tmp)
437 raise
440def candidate_runs_roots() -> list:
441 """Every root an existing run may live under (configured root alone, or the
442 /tmp root and the $TMPDIR fallback, primary first). Creation uses runs_root();
443 lookup must not depend on which root this invocation would create under."""
444 configured = os.environ.get("CE_WORK_RUNS_ROOT")
445 if configured:
446 return [os.path.abspath(configured)]
447 peer_root = os.environ.get("CE_PEER_JOBS_ROOT")
448 if peer_root:
449 return [os.path.join(os.path.abspath(peer_root), "ce-work")]
450 if OWNER_SCRATCH_ROOT is None:
451 raise TrustFailure("effective user ID is unavailable; cannot derive the runs root")
452 roots = [os.path.join(os.path.abspath(OWNER_SCRATCH_ROOT), "ce-work")]
453 fallback = os.path.join(os.path.abspath(_fallback_scratch_root()), "ce-work")
454 if fallback not in roots:
455 roots.append(fallback)
456 return roots
459def run_dir(run_id: str) -> str:
460 rid = safe_id(run_id, "run id")
461 for root in candidate_runs_roots():
462 existing = os.path.join(root, rid)
463 if os.path.isdir(existing) and not os.path.islink(existing):
464 return existing
465 return os.path.join(runs_root(), rid)
468@contextlib.contextmanager
469def locked_manifest(run_id: str, write: bool = False):
470 run_id = safe_id(run_id, "run id")
471 rd = run_dir(run_id)
472 ensure_runs_root(os.path.dirname(rd))
473 validate_private_dir(rd)
474 lock_path = os.path.join(rd, "manifest.lock")
475 try:
476 fd = os.open(lock_path, os.O_RDWR | O_NOFOLLOW)
477 except OSError as exc:
478 raise TrustFailure(f"cannot safely open manifest lock: {exc}") from exc
479 try:
480 st = os.fstat(fd)
481 if not stat.S_ISREG(st.st_mode) or (_euid() is not None and st.st_uid != _euid()) or _mode(st) != 0o600:
482 raise TrustFailure("manifest lock owner/type/mode validation failed")
483 fcntl.flock(fd, fcntl.LOCK_EX if write else fcntl.LOCK_SH)
484 doc = read_private_json(os.path.join(rd, "manifest.json"))
485 if doc.get("schema_version") != SCHEMA_VERSION or doc.get("run_id") != run_id:
486 raise TrustFailure("manifest schema or run identity mismatch")
487 before = json.dumps(doc, sort_keys=True, separators=(",", ":"))
488 yield doc
489 after = json.dumps(doc, sort_keys=True, separators=(",", ":"))
490 if write and after != before:
491 doc["revision"] = int(doc.get("revision", 0)) + 1
492 doc["updated_at"] = now_iso()
493 atomic_private_json(os.path.join(rd, "manifest.json"), doc)
494 finally:
495 with contextlib.suppress(OSError):
496 fcntl.flock(fd, fcntl.LOCK_UN)
497 os.close(fd)
500def sanitized_git_environment(overrides: dict | None = None) -> dict[str, str]:
501 process_env = {key: value for key, value in os.environ.items() if key not in GIT_LOCAL_ENV_VARS}
502 process_env.update(overrides or {})
503 return process_env
506def git(repo: str, *args: str, input_data: bytes | None = None, check: bool = True, env: dict | None = None) -> bytes:
507 proc = subprocess.run(
508 ["git", "-C", repo, *args], input=input_data, capture_output=True,
509 env=sanitized_git_environment(env), check=False,
510 )
511 if check and proc.returncode != 0:
512 message = proc.stderr.decode("utf-8", "replace").strip()
513 raise Operational("BLOCKED", f"git {' '.join(args)} failed: {message}")
514 return proc.stdout
517def git_text(repo: str, *args: str, check: bool = True) -> str:
518 return git(repo, *args, check=check).decode("utf-8", "surrogateescape").strip()
521def commit_index_tree(repo: str, message: str) -> str:
522 """Commit the verified index directly, without invoking repository hooks."""
523 if not message.strip() or "\0" in message:
524 raise Operational("REFUSED", "commit message must be non-empty and contain no NUL")
525 parent = git_text(repo, "rev-parse", "HEAD")
526 branch_ref = git_text(repo, "symbolic-ref", "-q", "HEAD")
527 tree = git_text(repo, "write-tree")
528 commit = git(
529 repo,
530 "commit-tree", tree, "-p", parent,
531 input_data=f"{message.rstrip()}\n".encode("utf-8"),
532 ).decode("ascii", "strict").strip()
533 git(repo, "update-ref", branch_ref, commit, parent)
534 return commit
537def repo_info(repo: str) -> dict:
538 repo = os.path.realpath(repo)
539 top = os.path.realpath(git_text(repo, "rev-parse", "--show-toplevel"))
540 if top != repo:
541 repo = top
542 branch = git_text(repo, "symbolic-ref", "-q", "HEAD", check=False)
543 if not branch:
544 raise Operational("REFUSED", "canonical checkout must be on a branch")
545 git_dir = os.path.realpath(git_text(repo, "rev-parse", "--path-format=absolute", "--absolute-git-dir"))
546 common = os.path.realpath(git_text(repo, "rev-parse", "--path-format=absolute", "--git-common-dir"))
547 st = os.stat(common)
548 roots = sorted(git_text(repo, "rev-list", "--max-parents=0", "HEAD").splitlines())
549 identity = digest_bytes((common + f"\0{st.st_dev}\0{st.st_ino}\0" + "\n".join(roots)).encode())
550 return {
551 "toplevel": repo,
552 "git_dir": git_dir,
553 "common_dir": common,
554 "common_dev": st.st_dev,
555 "common_ino": st.st_ino,
556 "identity_digest": identity,
557 "branch_ref": branch,
558 "head": git_text(repo, "rev-parse", "HEAD"),
559 "head_tree": git_text(repo, "rev-parse", "HEAD^{tree}"),
560 }
563def validate_source(doc: dict) -> None:
564 source = doc.get("source")
565 if source is not None:
566 if not isinstance(source, dict):
567 raise TrustFailure("manifest source record is malformed")
568 kind = source.get("kind")
569 if kind == "prompt":
570 if source.get("storage") != "run" or source.get("path") != "source/bare-prompt.md":
571 raise TrustFailure("prompt source location is malformed")
572 if not isinstance(source.get("digest"), str) or not re.fullmatch(r"[0-9a-f]{64}", source["digest"]):
573 raise TrustFailure("prompt source digest is malformed")
574 data = read_private(os.path.join(run_dir(doc["run_id"]), source["path"]), MAX_PACKET_BYTES)
575 if digest_bytes(data) != source.get("digest"):
576 raise TrustFailure("prompt source digest does not match private content")
577 elif kind == "plan":
578 if source.get("storage") != "repository" or not isinstance(source.get("path"), str):
579 raise TrustFailure("plan source location is malformed")
580 if not isinstance(source.get("digest"), str) or not re.fullmatch(r"[0-9a-f]{64}", source["digest"]):
581 raise TrustFailure("plan source digest is malformed")
582 else:
583 raise TrustFailure("manifest source kind is invalid")
586def validate_repo(doc: dict) -> dict:
587 validate_source(doc)
588 recorded = doc["repository"]
589 current = repo_info(recorded["toplevel"])
590 for key in ("toplevel", "git_dir", "common_dir", "common_dev", "common_ino", "identity_digest"):
591 if current[key] != recorded[key]:
592 raise Operational("BLOCKED", f"canonical repository identity changed ({key})")
593 if current["branch_ref"] != doc["branch"]["ref"]:
594 raise Operational("BLOCKED", "canonical branch changed")
595 return current
598def resolve_plan(repo: str, plan: str) -> tuple[str, str]:
599 supplied = os.path.abspath(plan if os.path.isabs(plan) else os.path.join(repo, plan))
600 try:
601 st = os.lstat(supplied)
602 except OSError as exc:
603 raise Operational("REFUSED", f"selected plan is missing: {exc}") from exc
604 if stat.S_ISLNK(st.st_mode) or not stat.S_ISREG(st.st_mode):
605 raise Operational("REFUSED", "selected plan must be one regular non-symlink file")
606 # OS temp roots may themselves be compatibility symlinks (macOS /var ->
607 # /private/var). Reject a symlink at the selected file, then compare the
608 # resolved file against the already-resolved canonical repository.
609 absolute = os.path.realpath(supplied)
610 if os.path.commonpath([repo, absolute]) != repo:
611 raise Operational("REFUSED", "plan must be inside the canonical repository")
612 return absolute, os.path.relpath(absolute, repo)
615def parse_json_arg(raw: str, label: str) -> dict:
616 try:
617 value = json.loads(raw)
618 except ValueError as exc:
619 raise Operational("REFUSED", f"invalid {label} JSON") from exc
620 if not isinstance(value, dict):
621 raise Operational("REFUSED", f"{label} must be a JSON object")
622 return value
625ROUTE_CONTRACTS = {
626 "codex": {"target": "codex", "harness": "codex", "intermediaries": [], "default_model": "auto", "restriction_posture": "cooperative"},
627 "claude": {"target": "claude", "harness": "claude", "intermediaries": [], "default_model": "auto", "restriction_posture": "cooperative"},
628 "grok-cli": {"target": "grok", "harness": "grok", "intermediaries": [], "default_model": "auto", "restriction_posture": "cooperative"},
629 "cursor": {"target": "cursor", "harness": "cursor-agent", "intermediaries": [], "default_model": "auto", "restriction_posture": "cooperative"},
630 "composer": {"target": "composer", "harness": "cursor-agent", "intermediaries": ["cursor"], "default_model": "composer-2.5[fast=true]", "restriction_posture": "cooperative"},
631 "grok-cursor": {"target": "grok", "harness": "cursor-agent", "intermediaries": ["cursor"], "default_model": "grok-4.7[context=256k,reasoning_effort=high,fast=true]", "restriction_posture": "cooperative"},
632 "opencode": {"target": "opencode", "harness": "opencode", "intermediaries": [], "default_model": "auto", "restriction_posture": "cooperative"},
633 "antigravity": {"target": "antigravity", "harness": "antigravity", "intermediaries": [], "default_model": "gemini-4-argon", "restriction_posture": "cooperative"},
634}
635# Cursor's ACP model ids may carry a bracketed preset; route families apply to
636# the id before the bracket.
637CURSOR_MODEL = re.compile(r"([A-Za-z0-9][A-Za-z0-9._:/-]*)(?:\[[A-Za-z0-9._=,:-]*\])?")
640def route_model_allowed(route: str, model: str) -> bool:
641 if route in {"cursor", "composer", "grok-cursor"}:
642 match = CURSOR_MODEL.fullmatch(model)
643 if not match:
644 return False
645 base = match.group(1)
646 if route == "composer":
647 return bool(re.fullmatch(r"composer-[A-Za-z0-9._-]+", base))
648 if route == "grok-cursor":
649 return bool(re.fullmatch(r"(?:cursor-)?grok-[A-Za-z0-9._-]+", base))
650 lowered = base.lower()
651 return not (lowered in {"composer", "grok"} or lowered.startswith(("composer-", "grok-", "cursor-grok-")))
652 if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:/-]*", model):
653 return False
654 if route == "codex":
655 return model == "auto" or bool(re.fullmatch(r"(?:gpt-[A-Za-z0-9._-]+|o[0-9][A-Za-z0-9._-]*)", model))
656 if route == "claude":
657 return model in {"auto", "fable", "opus", "sonnet", "haiku"} or bool(re.fullmatch(r"claude-[A-Za-z0-9._-]+", model))
658 if route == "grok-cli":
659 return model == "auto" or bool(re.fullmatch(r"grok-[A-Za-z0-9._-]+", model))
660 if route == "opencode":
661 return model == "auto" or bool(re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*/[A-Za-z0-9._-]+", model))
662 if route == "antigravity":
663 # Antigravity also serves Claude and gpt-oss models; this route exists for Gemini.
664 return bool(re.fullmatch(r"gemini-[A-Za-z0-9._-]+", model))
665 return False
668EFFORT_TOKEN = re.compile(r"[A-Za-z0-9][A-Za-z0-9_-]{0,31}")
671def fixed_route_contract(binding: dict, egress: dict, word: str = "BLOCKED") -> dict:
672 if not isinstance(binding, dict) or not isinstance(egress, dict):
673 raise Operational(word, "run binding or egress sanction is malformed")
674 expected_binding_fields = {"mode", "target", "model", "source"}
675 if set(binding) != expected_binding_fields:
676 raise Operational(word, "binding must contain exactly mode, target, model, and source")
677 if binding.get("mode") not in {"prefer", "require"}:
678 raise Operational(word, "binding mode must be 'prefer' or 'require'")
679 source = binding.get("source")
680 if not isinstance(source, str) or not source or "\0" in source or len(source.encode()) > 256:
681 raise Operational(word, "binding source must be a non-empty string of at most 256 bytes")
682 route = egress.get("route")
683 contract = ROUTE_CONTRACTS.get(route)
684 if not contract:
685 allowed = ", ".join(ROUTE_CONTRACTS)
686 raise Operational(word, f"unsupported egress route {route!r}; expected one of: {allowed}")
687 if binding.get("target") != contract["target"]:
688 raise Operational(word, "binding target does not match the sanctioned fixed route")
689 intermediaries = egress.get("intermediaries")
690 if intermediaries != contract["intermediaries"]:
691 raise Operational(word, "egress intermediaries do not match the fixed route")
692 model = binding.get("model")
693 if model is not None and (not isinstance(model, str) or not model):
694 raise Operational(word, "binding model must be null or a non-empty string")
695 requested_model = model or contract["default_model"]
696 if not route_model_allowed(route, requested_model):
697 raise Operational(word, "binding model is not compatible with the sanctioned fixed route")
698 restrictions = egress.get("restrictions", [])
699 if not isinstance(restrictions, list) or not all(isinstance(item, str) for item in restrictions):
700 raise Operational(word, "egress restrictions must be a string list")
701 effort = egress.get("effort")
702 if effort is not None and (not isinstance(effort, str) or not EFFORT_TOKEN.fullmatch(effort)):
703 raise Operational(word, "egress effort must be a short plain token")
704 return contract
707def attempt_authorization(
708 doc: dict,
709 activity_posture: str,
710 unit_id: str,
711 attempt_id: str,
712 packet_digest: str,
713) -> dict:
714 binding = doc.get("binding")
715 egress = doc.get("egress")
716 contract = fixed_route_contract(binding, egress)
717 route = egress.get("route")
718 intermediaries = egress.get("intermediaries")
719 model = binding.get("model")
720 restrictions = egress.get("restrictions", [])
721 authorization = {
722 "schema_version": 1,
723 "run_id": doc["run_id"],
724 "unit_id": unit_id,
725 "attempt_id": attempt_id,
726 "route": route,
727 "target": contract["target"],
728 "harness": contract["harness"],
729 "intermediaries": list(contract["intermediaries"]),
730 "model_requested": model or contract["default_model"],
731 "restriction_posture": contract["restriction_posture"],
732 "restrictions": list(restrictions),
733 "activity_posture": activity_posture,
734 "packet_digest": packet_digest,
735 }
736 # Present only when requested, so an unset run keeps the original key set.
737 if egress.get("effort") is not None:
738 authorization["effort_requested"] = egress["effort"]
739 return authorization
742def read_external_packet(path: str, label: str = "unit packet") -> bytes:
743 supplied = os.path.abspath(path)
744 try:
745 fd = os.open(supplied, os.O_RDONLY | O_NOFOLLOW)
746 except OSError as exc:
747 raise Operational("REFUSED", f"cannot safely open {label}: {exc}") from exc
748 try:
749 st = os.fstat(fd)
750 if not stat.S_ISREG(st.st_mode):
751 raise Operational("REFUSED", f"{label} must be one regular non-symlink file")
752 if st.st_size > MAX_PACKET_BYTES:
753 raise Operational("REFUSED", f"{label} exceeds {MAX_PACKET_BYTES}-byte limit")
754 data = bytearray()
755 while len(data) <= MAX_PACKET_BYTES:
756 part = os.read(fd, min(65536, MAX_PACKET_BYTES + 1 - len(data)))
757 if not part:
758 break
759 data.extend(part)
760 if len(data) > MAX_PACKET_BYTES:
761 raise Operational("REFUSED", f"{label} exceeds {MAX_PACKET_BYTES}-byte limit")
762 return bytes(data)
763 finally:
764 os.close(fd)
767def event(doc: dict, kind: str, unit_id: str | None = None, detail: dict | None = None) -> None:
768 row = {"at": now_iso(), "kind": kind}
769 if unit_id is not None:
770 row["unit_id"] = unit_id
771 if detail:
772 row["detail"] = detail
773 doc.setdefault("events", []).append(row)
776def cmd_init(args) -> tuple[str, dict]:
777 ensure_root()
778 rid = safe_id(args.run_id, "run id")
779 info = repo_info(args.repo)
780 if args.plan:
781 if not args.plan_digest or args.prompt_digest:
782 raise Operational("REFUSED", "plan source requires only --plan-digest")
783 plan_abs, plan_rel = resolve_plan(info["toplevel"], args.plan)
784 source_bytes = Path(plan_abs).read_bytes()
785 source_kind = "plan"
786 supplied_digest = args.plan_digest
787 source_record = {
788 "kind": source_kind,
789 "storage": "repository",
790 "path": plan_rel,
791 "digest": digest_bytes(source_bytes),
792 }
793 else:
794 if not args.prompt_digest or args.plan_digest:
795 raise Operational("REFUSED", "prompt source requires only --prompt-digest")
796 prompt_abs = os.path.realpath(os.path.abspath(args.prompt_brief))
797 if os.path.commonpath([info["toplevel"], prompt_abs]) == info["toplevel"]:
798 raise Operational("REFUSED", "prompt brief must be outside the canonical repository")
799 source_bytes = read_external_packet(args.prompt_brief, "prompt brief")
800 source_kind = "prompt"
801 supplied_digest = args.prompt_digest
802 source_record = {
803 "kind": source_kind,
804 "storage": "run",
805 "path": "source/bare-prompt.md",
806 "digest": digest_bytes(source_bytes),
807 }
808 actual_digest = source_record["digest"]
809 if actual_digest != supplied_digest:
810 raise Operational("REFUSED", f"selected {source_kind} digest does not match content")
811 binding = parse_json_arg(args.binding_json, "binding")
812 egress = parse_json_arg(args.egress_json, "egress")
813 fixed_route_contract(binding, egress, "REFUSED")
814 rd = run_dir(rid)
815 try:
816 os.mkdir(rd, 0o700)
817 except FileExistsError:
818 try:
819 existing = os.lstat(rd)
820 except OSError as exc:
821 raise TrustFailure(f"cannot safely inspect run directory {rd}: {exc}") from exc
822 if stat.S_ISDIR(existing.st_mode) and not os.path.lexists(os.path.join(rd, "manifest.json")):
823 raise Operational(
824 "BLOCKED",
825 "run directory exists without a controller manifest; choose a new run id or remove the directory after confirming no initialization is active",
826 )
827 validate_private_dir(rd)
828 with locked_manifest(rid) as existing:
829 validate_repo(existing)
830 existing_source = existing.get("source")
831 if not isinstance(existing_source, dict):
832 plan = existing.get("plan")
833 existing_source = {
834 "kind": "plan",
835 "storage": "repository",
836 "path": plan.get("path") if isinstance(plan, dict) else None,
837 "digest": plan.get("digest") if isinstance(plan, dict) else None,
838 }
839 if (
840 existing["repository"]["identity_digest"] != info["identity_digest"]
841 or existing_source.get("kind") != source_kind
842 or existing_source.get("digest") != actual_digest
843 ):
844 raise Operational("BLOCKED", "run id already belongs to another repository or source")
845 if existing.get("binding") != binding or existing.get("egress") != egress:
846 raise Operational(
847 "BLOCKED",
848 "run id binding or egress sanction differs from the recorded fixed contract; resume with the recorded contract or choose a new run id",
849 )
850 return "READY", {
851 "run_id": rid,
852 "revision": existing["revision"],
853 "resumed": True,
854 "source_kind": source_kind,
855 "source_digest": actual_digest,
856 "recovery_path": rd,
857 }
858 validate_private_dir(rd)
859 for child in ("units", "jobs", "packets", "source"):
860 ensure_private_dir(os.path.join(rd, child))
861 if source_kind == "prompt":
862 create_private(os.path.join(rd, source_record["path"]), source_bytes)
863 create_private(os.path.join(rd, "manifest.lock"), b"")
864 created = now_iso()
865 doc = {
866 "schema_version": SCHEMA_VERSION,
867 "revision": 0,
868 "run_id": rid,
869 "created_at": created,
870 "updated_at": created,
871 "repository": {k: info[k] for k in ("toplevel", "git_dir", "common_dir", "common_dev", "common_ino", "identity_digest")},
872 "branch": {"ref": info["branch_ref"], "initial_head": info["head"]},
873 "source": source_record,
874 "plan": {
875 "kind": source_kind,
876 "path": plan_rel if source_kind == "plan" else None,
877 "digest": actual_digest,
878 "checkpoint": None,
879 },
880 "binding": binding,
881 "egress": egress,
882 "integration_lock": None,
883 "units": {},
884 "verification_attempts": [],
885 "verifications": [],
886 "blockers": [],
887 "events": [{"at": created, "kind": "run-created"}],
888 }
889 create_private(os.path.join(rd, "manifest.json"), (json.dumps(doc, sort_keys=True, separators=(",", ":")) + "\n").encode())
890 return "READY", {
891 "run_id": rid,
892 "revision": 0,
893 "resumed": False,
894 "source_kind": source_kind,
895 "source_digest": actual_digest,
896 "recovery_path": rd,
897 }
900def status_paths(repo: str) -> set[str]:
901 raw = git(repo, "status", "--porcelain=v1", "-z", "--untracked-files=all")
902 parts = raw.split(b"\0")
903 paths: set[str] = set()
904 i = 0
905 while i < len(parts):
906 entry = parts[i]
907 i += 1
908 if not entry:
909 continue
910 if len(entry) < 4:
911 raise Operational("BLOCKED", "unexpected porcelain status record")
912 code = entry[:2]
913 paths.add(entry[3:].decode("utf-8", "surrogateescape"))
914 if b"R" in code or b"C" in code:
915 if i >= len(parts) or not parts[i]:
916 raise Operational("BLOCKED", "incomplete rename status record")
917 paths.add(parts[i].decode("utf-8", "surrogateescape"))
918 i += 1
919 return paths
922def reconcile_plan_checkpoint(repo: str, doc: dict, info: dict, plan_rel: str) -> dict | None:
923 """Recover the controller's plan commit when its manifest receipt was interrupted."""
924 prior = doc.get("branch", {}).get("initial_head")
925 commit = info["head"]
926 if commit == prior:
927 return None
928 lineage = git_text(repo, "rev-list", "--parents", "-n", "1", commit).split()
929 changed = set(filter(None, git(repo, "diff-tree", "--no-commit-id", "--name-only", "-r", "-z", commit).decode("utf-8", "surrogateescape").split("\0")))
930 message = git(repo, "show", "-s", "--format=%B", commit).decode("utf-8", "surrogateescape").rstrip("\n")
931 plan_bytes = git(repo, "show", f"{commit}:{plan_rel}", check=False)
932 if (
933 not _valid_git_object_id(prior)
934 or lineage != [commit, prior]
935 or changed != {plan_rel}
936 or message != PLAN_CHECKPOINT_MESSAGE
937 or digest_bytes(plan_bytes) != doc["plan"]["digest"]
938 ):
939 raise Operational(
940 "BLOCKED",
941 "canonical HEAD advanced without a recorded matching plan checkpoint",
942 {"expected_prior_head": prior, "head": commit},
943 )
944 committed_at = int(git_text(repo, "show", "-s", "--format=%ct", commit))
945 return {
946 "prior_head": prior,
947 "commit": commit,
948 "tree": info["head_tree"],
949 "path": plan_rel,
950 "digest": doc["plan"]["digest"],
951 "at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(committed_at)),
952 }
955def cmd_checkpoint_plan(args) -> tuple[str, dict]:
956 with locked_manifest(args.run_id, write=True) as doc:
957 info = validate_repo(doc)
958 repo = info["toplevel"]
959 plan = doc.get("plan")
960 if not isinstance(plan, dict) or plan.get("kind", "plan") != "plan" or not plan.get("path"):
961 dirty = status_paths(repo)
962 if dirty:
963 raise Operational("BLOCKED", "prompt-backed external execution requires a clean canonical checkout", {"dirty_paths": sorted(dirty)})
964 return "NOOP", {"checkpoint": None, "head": info["head"], "source_kind": "prompt"}
965 plan_rel = plan["path"]
966 plan_abs, _ = resolve_plan(repo, plan_rel)
967 if digest_bytes(Path(plan_abs).read_bytes()) != doc["plan"]["digest"]:
968 raise Operational("BLOCKED", "selected plan content no longer matches recorded digest")
969 dirty = status_paths(repo)
970 if not dirty:
971 checkpoint = doc["plan"].get("checkpoint")
972 if checkpoint is not None:
973 return "NOOP", {"checkpoint": checkpoint, "head": info["head"]}
974 checkpoint = reconcile_plan_checkpoint(repo, doc, info, plan_rel)
975 if checkpoint is None:
976 return "NOOP", {"checkpoint": None, "head": info["head"]}
977 doc["plan"]["checkpoint"] = checkpoint
978 event(doc, "plan-checkpoint", detail={"commit": checkpoint["commit"], "path": plan_rel})
979 return "CHECKPOINTED", {"checkpoint": checkpoint}
980 if dirty != {plan_rel}:
981 raise Operational("BLOCKED", "canonical dirt is not exactly the selected plan", {"dirty_paths": sorted(dirty)})
982 prior = info["head"]
983 git(repo, "add", "--", plan_rel)
984 staged = set(filter(None, git(repo, "diff", "--cached", "--name-only", "-z").decode("utf-8", "surrogateescape").split("\0")))
985 if staged != {plan_rel}:
986 git(repo, "reset", "--mixed", prior)
987 raise Operational("BLOCKED", "staged paths are not exactly the selected plan")
988 try:
989 commit_index_tree(repo, PLAN_CHECKPOINT_MESSAGE)
990 except Operational:
991 git(repo, "reset", "--mixed", prior, check=False)
992 raise
993 commit = git_text(repo, "rev-parse", "HEAD")
994 test_fault("checkpoint-plan-after-commit")
995 if status_paths(repo):
996 raise Operational("BLOCKED", "checkpoint committed but canonical checkout is not clean")
997 cp = {"prior_head": prior, "commit": commit, "tree": git_text(repo, "rev-parse", "HEAD^{tree}"), "path": plan_rel, "digest": doc["plan"]["digest"], "at": now_iso()}
998 with locked_manifest(args.run_id, write=True) as doc:
999 validate_repo(doc)
1000 doc["plan"]["checkpoint"] = cp
1001 event(doc, "plan-checkpoint", detail={"commit": commit, "path": plan_rel})
1002 return "CHECKPOINTED", {"checkpoint": cp}
1005@contextlib.contextmanager
1006def admin_lock(common_dir: str):
1007 root = ensure_root()
1008 key = digest_bytes(os.path.realpath(common_dir).encode())
1009 path = os.path.join(root, ".locks", f"worktree-{key}.lock")
1010 try:
1011 create_private(path, b"")
1012 except Operational:
1013 pass
1014 data = read_private(path, 64)
1015 del data
1016 fd = os.open(path, os.O_RDWR | O_NOFOLLOW)
1017 try:
1018 fcntl.flock(fd, fcntl.LOCK_EX)
1019 yield
1020 finally:
1021 fcntl.flock(fd, fcntl.LOCK_UN)
1022 os.close(fd)
1025def worktree_rows(repo: str) -> list[dict]:
1026 raw = git_text(repo, "worktree", "list", "--porcelain")
1027 rows, row = [], {}
1028 for line in raw.splitlines() + [""]:
1029 if not line:
1030 if row:
1031 rows.append(row)
1032 row = {}
1033 continue
1034 key, _, value = line.partition(" ")
1035 row[key] = value if value else True
1036 return rows
1039def validate_workspace(doc: dict, unit: dict) -> dict:
1040 repo = doc["repository"]["toplevel"]
1041 workspace = unit["workspace"]["path"]
1042 owned = os.path.join(run_dir(doc["run_id"]), "units", unit["unit_id"])
1043 if os.path.commonpath([os.path.realpath(workspace), os.path.realpath(owned)]) != os.path.realpath(owned):
1044 raise Operational("BLOCKED", "workspace escaped its owned unit directory")
1045 validate_private_dir(workspace)
1046 matches = [r for r in worktree_rows(repo) if os.path.realpath(str(r.get("worktree", ""))) == os.path.realpath(workspace)]
1047 if len(matches) != 1:
1048 raise Operational("BLOCKED", "workspace is not registered exactly once")
1049 if "detached" not in matches[0]:
1050 raise Operational("BLOCKED", "unit workspace is not detached")
1051 common = os.path.realpath(git_text(workspace, "rev-parse", "--path-format=absolute", "--git-common-dir"))
1052 if common != doc["repository"]["common_dir"]:
1053 raise Operational("BLOCKED", "unit workspace belongs to another repository")
1054 return matches[0]
1057def validate_pristine_unit_base(doc: dict, unit: dict) -> dict:
1058 row = validate_workspace(doc, unit)
1059 workspace = unit["workspace"]["path"]
1060 base = unit["workspace"]["base"]
1061 if git_text(workspace, "rev-parse", "HEAD") != base:
1062 raise Operational("BLOCKED", "unit workspace HEAD no longer equals the recorded base")
1063 dirty = status_paths(workspace)
1064 if dirty:
1065 raise Operational(
1066 "BLOCKED",
1067 "unit workspace is dirty before dispatch authorization",
1068 {"dirty_paths": sorted(dirty)},
1069 )
1070 return row