← Files Build StewardARCHIVED FILE
skills/build-space/scripts/build_steward.py
77.2 KB · Oct 2, 2026 · 00:34 UTC
#!/usr/bin/env python3
"""Build Steward: local, fail-closed build-residue inventory and reclaim.
The command never reads candidate payload contents, never uses the network, and
never accepts shell fragments or glob patterns. The optional app inventory reads
only bundle identity metadata from ``Contents/Info.plist``. ``audit`` either
emits a strict read-only summary without creating local state or creates a
mode-0600 plan without clobbering an existing path. ``apply`` requires the
plan digest plus exact item IDs and revalidates every candidate.
"""
from __future__ import annotations
import argparse
import contextlib
import ctypes
import datetime as dt
import errno
import hashlib
import html
import json
import os
import plistlib
import secrets
import shutil
import stat
import subprocess
import sys
import tempfile
import time
from pathlib import Path
from typing import Any, Iterable, Iterator
PLAN_SCHEMA = "build-steward.plan.v2"
RECEIPT_SCHEMA = "build-steward.receipt.v1"
LEASE_EVENT_SCHEMA = "build-steward.lease-event.v1"
VERSION = "0.1.1"
DEFAULT_MIN_AGE_HOURS = 24.0
DEFAULT_TTL_MINUTES = 30
DEFAULT_MAX_CANDIDATES = 100
DEFAULT_MAX_ENTRIES = 100_000
DEFAULT_SCAN_SECONDS = 5.0
DEFAULT_TOTAL_ENTRIES = 500_000
DEFAULT_TOTAL_SCAN_SECONDS = 45.0
DEFAULT_MIN_SIZE_MIB = 64
DEFAULT_RECEIPT_RESERVE_MIB = 2
ELIGIBLE_KINDS = {"xcode-derived-data", "xcode-temp-pipeline"}
XCODE_PRODUCERS = (
"xcode",
"xcodebuild",
"xcbuild",
"xctest",
"xctrunner",
"swift",
"swift-build",
"swift-driver",
"swift-frontend",
"swiftc",
"clang",
"clang++",
"ld",
)
TEMP_HINTS = (
"build",
"claude",
"codex",
"derived",
"operator",
"session",
"worktree",
"xcode",
)
APP_RESIDUE_HINTS = ("backup", "candidate", "failed", "old", "rollback")
class StewardError(RuntimeError):
"""Expected fail-closed error."""
def utc_now() -> dt.datetime:
return dt.datetime.now(dt.timezone.utc)
def isoformat(value: dt.datetime) -> str:
return value.astimezone(dt.timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
def parse_time(value: str) -> dt.datetime:
return dt.datetime.fromisoformat(value.replace("Z", "+00:00"))
def canonical_json(value: Any) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8")
def digest_value(value: Any) -> str:
return "sha256:" + hashlib.sha256(canonical_json(value)).hexdigest()
def normalize_path(raw: str | os.PathLike[str]) -> str:
expanded = os.path.expandvars(os.path.expanduser(os.fspath(raw)))
return os.path.normpath(os.path.abspath(expanded))
def allocated_bytes(st: os.stat_result) -> int:
blocks = getattr(st, "st_blocks", None)
return int(blocks * 512) if blocks is not None else int(st.st_size)
def human_bytes(value: int) -> str:
size = float(max(0, value))
units = ("B", "KiB", "MiB", "GiB", "TiB")
for unit in units:
if size < 1024.0 or unit == units[-1]:
return f"{size:.1f} {unit}" if unit != "B" else f"{int(size)} B"
size /= 1024.0
return f"{int(value)} B"
def atomic_create_json(path: str | os.PathLike[str], payload: Any) -> None:
"""Atomically create a private JSON file; never replace an existing path."""
destination = Path(path)
destination.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
fd, temporary = tempfile.mkstemp(prefix=f".{destination.name}.", dir=str(destination.parent))
try:
os.fchmod(fd, 0o600)
with os.fdopen(fd, "w", encoding="utf-8") as handle:
json.dump(payload, handle, indent=2, sort_keys=True)
handle.write("\n")
handle.flush()
os.fsync(handle.fileno())
try:
os.link(temporary, destination)
except FileExistsError as error:
raise StewardError("refusing to overwrite an existing output path") from error
except OSError as error:
raise StewardError("could not create output with no-clobber guarantees") from error
os.unlink(temporary)
directory_fd = os.open(destination.parent, os.O_RDONLY)
try:
os.fsync(directory_fd)
finally:
os.close(directory_fd)
except BaseException:
with contextlib.suppress(FileNotFoundError):
os.unlink(temporary)
raise
def atomic_create_text(path: str | os.PathLike[str], value: str) -> None:
"""Atomically create a private UTF-8 text file; never replace a path."""
destination = Path(path)
destination.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
fd, temporary = tempfile.mkstemp(prefix=f".{destination.name}.", dir=str(destination.parent))
try:
os.fchmod(fd, 0o600)
with os.fdopen(fd, "w", encoding="utf-8") as handle:
handle.write(value)
handle.flush()
os.fsync(handle.fileno())
try:
os.link(temporary, destination)
except FileExistsError as error:
raise StewardError("refusing to overwrite an existing output path") from error
except OSError as error:
raise StewardError("could not create output with no-clobber guarantees") from error
os.unlink(temporary)
directory_fd = os.open(destination.parent, os.O_RDONLY)
try:
os.fsync(directory_fd)
finally:
os.close(directory_fd)
except BaseException:
with contextlib.suppress(FileNotFoundError):
os.unlink(temporary)
raise
def read_json(path: str | os.PathLike[str]) -> dict[str, Any]:
nofollow = getattr(os, "O_NOFOLLOW", None)
if nofollow is None:
raise StewardError("no-follow control-file reads are unavailable")
flags = os.O_RDONLY | nofollow | getattr(os, "O_CLOEXEC", 0)
try:
descriptor = os.open(path, flags)
except OSError as error:
raise StewardError("control file could not be opened safely") from error
try:
control_stat = os.fstat(descriptor)
owner = os.getuid() if hasattr(os, "getuid") else int(control_stat.st_uid)
if (
not stat.S_ISREG(control_stat.st_mode)
or int(control_stat.st_uid) != owner
or int(control_stat.st_nlink) != 1
or int(control_stat.st_size) > 16 * 1024 * 1024
):
raise StewardError("control file identity or size is unsafe")
with os.fdopen(descriptor, "r", encoding="utf-8") as handle:
descriptor = -1
value = json.load(handle)
finally:
if descriptor >= 0:
os.close(descriptor)
if not isinstance(value, dict):
raise StewardError("JSON root must be an object")
return value
def create_or_confirm_json(path: str | os.PathLike[str], payload: Any) -> None:
if os.path.lexists(path):
existing = read_json(path)
if canonical_json(existing) != canonical_json(payload):
raise StewardError("existing output does not match the completed receipt")
return
atomic_create_json(path, payload)
def mode_code(mode: int) -> str:
if stat.S_ISDIR(mode):
return "d"
if stat.S_ISREG(mode):
return "f"
if stat.S_ISLNK(mode):
return "l"
return "o"
def snapshot_path(path: str, *, max_entries: int, max_seconds: float) -> dict[str, Any]:
"""Hash metadata only; never follow symlinks or read file contents."""
started = time.monotonic()
root = normalize_path(path)
root_stat = os.lstat(root)
root_device = int(root_stat.st_dev)
owner = os.getuid() if hasattr(os, "getuid") else None
tree_hash = hashlib.sha256()
total_allocated = 0
entry_count = 0
latest_mtime_ns = 0
latest_ctime_ns = 0
crossed_mount = False
foreign_owner = False
contains_vcs = False
truncated = False
error_count = 0
symlink_count = 0
stack: list[tuple[str, str]] = [(root, ".")]
while stack:
if entry_count >= max_entries or time.monotonic() - started > max_seconds:
truncated = True
break
current, relative = stack.pop()
try:
current_stat = os.lstat(current)
except OSError:
error_count += 1
continue
entry_count += 1
total_allocated += allocated_bytes(current_stat)
latest_mtime_ns = max(latest_mtime_ns, int(current_stat.st_mtime_ns))
if relative != ".":
latest_ctime_ns = max(latest_ctime_ns, int(current_stat.st_ctime_ns))
if int(current_stat.st_dev) != root_device:
crossed_mount = True
if owner is not None and int(current_stat.st_uid) != owner:
foreign_owner = True
if os.path.basename(current) in {".git", ".hg", ".svn"}:
contains_vcs = True
link_target = ""
if stat.S_ISLNK(current_stat.st_mode):
symlink_count += 1
try:
link_target = os.readlink(current)
except OSError:
error_count += 1
metadata = (
relative,
mode_code(current_stat.st_mode),
int(current_stat.st_dev),
int(current_stat.st_ino),
int(current_stat.st_mode),
int(current_stat.st_size),
int(current_stat.st_mtime_ns),
int(current_stat.st_ctime_ns) if relative != "." else 0,
int(current_stat.st_uid),
link_target,
)
tree_hash.update(canonical_json(metadata))
tree_hash.update(b"\n")
if (
stat.S_ISDIR(current_stat.st_mode)
and not stat.S_ISLNK(current_stat.st_mode)
and int(current_stat.st_dev) == root_device
):
remaining_capacity = max(0, max_entries - entry_count - len(stack))
try:
entries = []
with os.scandir(current) as iterator:
for entry in iterator:
if time.monotonic() - started > max_seconds or len(entries) >= remaining_capacity:
truncated = True
break
entries.append((entry.name, entry.path))
entries.sort(key=lambda item: item[0], reverse=True)
except OSError:
error_count += 1
continue
for entry_name, entry_path in entries:
child_relative = entry_name if relative == "." else f"{relative}/{entry_name}"
stack.append((entry_path, child_relative))
return {
"device": root_device,
"inode": int(root_stat.st_ino),
"mode": int(root_stat.st_mode),
"uid": int(root_stat.st_uid),
"mtime_ns": int(root_stat.st_mtime_ns),
"ctime_ns": int(root_stat.st_ctime_ns),
"latest_mtime_ns": latest_mtime_ns,
"latest_ctime_ns": latest_ctime_ns,
"allocated_bytes": total_allocated,
"entry_count": entry_count,
"tree_digest": "sha256:" + tree_hash.hexdigest(),
"crossed_mount": crossed_mount,
"foreign_owner": foreign_owner,
"contains_vcs": contains_vcs,
"symlink_count": symlink_count,
"truncated": truncated,
"error_count": error_count,
}
def process_state() -> dict[str, Any]:
try:
completed = subprocess.run(
["ps", "-axo", "comm="],
check=False,
capture_output=True,
text=True,
timeout=5,
)
except (OSError, subprocess.SubprocessError):
return {"status": "unavailable", "known_producers_running": []}
if completed.returncode != 0 or completed.stderr.strip():
return {"status": "unavailable", "known_producers_running": []}
names = {os.path.basename(line.strip()).lower() for line in completed.stdout.splitlines() if line.strip()}
matched = [producer for producer in XCODE_PRODUCERS if producer.lower() in names]
return {"status": "available", "known_producers_running": matched}
def check_open_handles(path: str) -> str:
"""Return clear, open, or unavailable without exposing process details."""
try:
candidate_stat = os.lstat(path)
except OSError:
return "unavailable"
if stat.S_ISLNK(candidate_stat.st_mode):
return "unavailable"
command = ["lsof", "-nP"]
if stat.S_ISDIR(candidate_stat.st_mode):
command.extend(["+D", path])
else:
command.extend(["--", path])
try:
completed = subprocess.run(
command,
check=False,
capture_output=True,
text=True,
timeout=20,
)
except (OSError, subprocess.SubprocessError):
return "unavailable"
if completed.stderr.strip():
return "unavailable"
if completed.returncode == 0:
lines = [line for line in completed.stdout.splitlines() if line.strip()]
return "open" if len(lines) > 1 else "unavailable"
if completed.returncode == 1 and not completed.stdout.strip():
return "clear"
return "unavailable"
def app_identity(path: str) -> dict[str, str]:
plist_path = os.path.join(path, "Contents", "Info.plist")
try:
with open(plist_path, "rb") as handle:
data = plistlib.load(handle)
except (OSError, plistlib.InvalidFileException):
return {}
result: dict[str, str] = {}
for output_key, plist_key in (
("bundle_id", "CFBundleIdentifier"),
("version", "CFBundleShortVersionString"),
("build", "CFBundleVersion"),
):
value = data.get(plist_key)
if value is not None:
result[output_key] = str(value)
return result
def immediate_children(root: str) -> list[str]:
try:
with os.scandir(root) as entries:
return [entry.path for entry in sorted(entries, key=lambda item: item.name)]
except OSError:
return []
def is_within(path: str, root: str) -> bool:
candidate = normalize_path(path)
boundary = normalize_path(root)
try:
return os.path.commonpath((candidate, boundary)) == boundary and candidate != boundary
except ValueError:
return False
def is_same_or_within(path: str, root: str) -> bool:
candidate = normalize_path(path)
boundary = normalize_path(root)
try:
return os.path.commonpath((candidate, boundary)) == boundary
except ValueError:
return False
def control_path_touches_candidate(control_path: str, candidate_path: str) -> bool:
"""Check lexical and resolved paths so a parent symlink cannot hide overlap."""
return is_same_or_within(control_path, candidate_path) or is_same_or_within(
os.path.realpath(control_path), os.path.realpath(candidate_path)
)
def capture_directory_chain(path: str) -> dict[str, Any]:
"""Bind every existing path component without following symlinks."""
normalized = normalize_path(path)
current = os.path.sep
entries: list[dict[str, Any]] = []
try:
parts = Path(normalized).parts[1:]
for part in parts:
current = os.path.join(current, part)
component = os.lstat(current)
entries.append(
{
"path": current,
"device": int(component.st_dev),
"inode": int(component.st_ino),
"mode": int(component.st_mode),
"uid": int(component.st_uid),
}
)
if stat.S_ISLNK(component.st_mode) or not stat.S_ISDIR(component.st_mode):
return {"status": "unsafe", "entries": entries}
except OSError:
return {"status": "unavailable", "entries": entries}
return {"status": "bound", "entries": entries}
def verify_directory_chain(saved: dict[str, Any]) -> bool:
if saved.get("status") != "bound":
return False
entries = saved.get("entries")
if not isinstance(entries, list) or not entries:
return False
for expected in entries:
try:
current = os.lstat(expected["path"])
except (OSError, KeyError, TypeError):
return False
observed = {
"device": int(current.st_dev),
"inode": int(current.st_ino),
"mode": int(current.st_mode),
"uid": int(current.st_uid),
}
if any(observed[key] != expected.get(key) for key in observed):
return False
if stat.S_ISLNK(current.st_mode) or not stat.S_ISDIR(current.st_mode):
return False
return True
def candidate_id(kind: str, path: str, snapshot: dict[str, Any]) -> str:
material = f"{kind}\0{normalize_path(path)}\0{snapshot['device']}\0{snapshot['inode']}"
return hashlib.sha256(material.encode("utf-8")).hexdigest()[:24]
def classification_for(
kind: str,
snapshot: dict[str, Any],
*,
age_seconds: float,
minimum_age_seconds: float,
processes: dict[str, Any],
root_is_symlink: bool,
root_is_bound: bool = True,
) -> tuple[str, list[str]]:
reasons: list[str] = []
if root_is_symlink:
reasons.append("candidate-is-symlink")
if snapshot["crossed_mount"]:
reasons.append("filesystem-boundary-crossed")
if snapshot["foreign_owner"]:
reasons.append("foreign-owner-present")
if snapshot["truncated"]:
reasons.append("scan-budget-exhausted")
if snapshot["error_count"]:
reasons.append("scan-errors")
if snapshot["contains_vcs"]:
reasons.append("version-control-state-present")
if not root_is_bound:
reasons.append("approved-root-not-identity-bound")
if kind in ELIGIBLE_KINDS and not stat.S_ISDIR(int(snapshot.get("mode", 0))):
reasons.append("eligible-class-must-be-directory")
if root_is_symlink or snapshot["crossed_mount"] or snapshot["foreign_owner"] or not root_is_bound:
return "protected", reasons
if snapshot["contains_vcs"]:
return "protected", reasons
if snapshot["truncated"] or snapshot["error_count"]:
return "review", reasons
if kind in ELIGIBLE_KINDS and not stat.S_ISDIR(int(snapshot.get("mode", 0))):
return "review", reasons
if kind not in ELIGIBLE_KINDS:
reasons.append("not-proven-rebuildable")
return "review", reasons
if age_seconds < minimum_age_seconds:
reasons.append("younger-than-policy")
return "active", reasons
if processes["status"] != "available":
reasons.append("process-visibility-unavailable")
return "review", reasons
if kind.startswith("xcode-") and processes["known_producers_running"]:
reasons.append("known-producer-running")
return "active", reasons
reasons.append("inactive-rebuildable-candidate")
return "eligible", reasons
def discover_specs(args: argparse.Namespace) -> tuple[list[dict[str, str]], list[dict[str, str]]]:
"""Return candidate specs and approved roots without traversing file contents."""
specs: list[dict[str, str]] = []
roots: list[dict[str, str]] = []
def add_root(raw: str, role: str) -> str:
path = normalize_path(raw)
if not any(item["path"] == path for item in roots):
roots.append({"path": path, "role": role})
return path
def add(path: str, kind: str, root: str, source: str) -> None:
normalized = normalize_path(path)
if len(specs) >= args.max_candidates:
return
existing = next((item for item in specs if item["path"] == normalized), None)
priority = {"generic": 0, "app-bundle": 1, "simulator-state": 1, "xcode-temp-pipeline": 2, "xcode-derived-data": 2, "explicit-rebuildable": 3}
if existing:
if priority.get(kind, 0) > priority.get(existing["kind"], 0):
existing.update({"kind": kind, "root": root, "source": source})
return
specs.append({"path": normalized, "kind": kind, "root": root, "source": source})
for raw in args.root:
root = add_root(raw, "inspect")
if os.path.isdir(root) and not os.path.islink(root):
for child in immediate_children(root):
add(child, "generic", root, "explicit-root")
elif os.path.lexists(root):
parent = add_root(os.path.dirname(root), "inspect")
add(root, "generic", parent, "explicit-root")
for raw in args.rebuildable:
path = normalize_path(raw)
root = add_root(os.path.dirname(path), "explicit-rebuildable")
if os.path.lexists(path):
add(path, "explicit-rebuildable", root, "explicit-rebuildable")
if args.profile == "macos-power-user":
home = Path.home()
derived_data = add_root(str(home / "Library/Developer/Xcode/DerivedData"), "xcode-derived-data")
for child in immediate_children(derived_data):
add(child, "xcode-derived-data", derived_data, "macos-power-user")
temp_roots = []
environment_temp = os.environ.get("TMPDIR")
if environment_temp:
temp_roots.append(environment_temp)
temp_roots.append("/private/tmp")
for raw_temp in temp_roots:
raw_temp = os.path.realpath(normalize_path(raw_temp))
temp_root = add_root(raw_temp, "agent-and-xcode-temp")
for child in immediate_children(temp_root):
name = os.path.basename(child).lower()
if name.startswith("xcodedistpipeline.~~~"):
add(child, "xcode-temp-pipeline", temp_root, "macos-power-user")
elif raw_temp == "/private/tmp" and any(hint in name for hint in TEMP_HINTS):
add(child, "generic", temp_root, "macos-power-user")
simulator_root = add_root(str(home / "Library/Developer/CoreSimulator/Devices"), "simulator-state")
for child in immediate_children(simulator_root):
add(child, "simulator-state", simulator_root, "macos-power-user")
applications_root = add_root("/Applications", "application-bundles")
app_paths = [path for path in immediate_children(applications_root) if path.lower().endswith(".app")]
identities = {path: app_identity(path) for path in app_paths}
bundle_counts: dict[str, int] = {}
for identity in identities.values():
bundle_id = identity.get("bundle_id")
if bundle_id:
bundle_counts[bundle_id] = bundle_counts.get(bundle_id, 0) + 1
for path, identity in identities.items():
name = os.path.basename(path).lower()
bundle_id = identity.get("bundle_id")
if any(hint in name for hint in APP_RESIDUE_HINTS) or (bundle_id and bundle_counts.get(bundle_id, 0) > 1):
add(path, "app-bundle", applications_root, "macos-power-user")
return specs, roots
def build_plan(args: argparse.Namespace) -> dict[str, Any]:
created = utc_now()
audit_started = time.monotonic()
minimum_age_seconds = float(args.min_age_hours) * 3600.0
processes = process_state()
lease_state_path = getattr(args, "state_dir", None) or str(Path.home() / ".build-steward")
leases = lease_root_context(lease_state_path)
specs, roots = discover_specs(args)
for root in roots:
root["chain"] = capture_directory_chain(root["path"])
roots_by_path = {root["path"]: root for root in roots}
candidates: list[dict[str, Any]] = []
minimum_size = int(args.min_size_mib) * 1024 * 1024
total_entry_limit = int(getattr(args, "total_entries", DEFAULT_TOTAL_ENTRIES))
total_time_limit = float(getattr(args, "total_scan_seconds", DEFAULT_TOTAL_SCAN_SECONDS))
total_entries = 0
considered_specs = 0
for spec in specs:
remaining_seconds = total_time_limit - (time.monotonic() - audit_started)
remaining_entries = total_entry_limit - total_entries
if remaining_seconds <= 0 or remaining_entries <= 0:
break
considered_specs += 1
path = spec["path"]
if not os.path.lexists(path):
continue
try:
root_is_symlink = stat.S_ISLNK(os.lstat(path).st_mode)
snapshot = snapshot_path(
path,
max_entries=min(int(args.max_entries), remaining_entries),
max_seconds=min(float(args.scan_seconds), remaining_seconds),
)
except OSError:
continue
total_entries += int(snapshot["entry_count"])
if snapshot["allocated_bytes"] < minimum_size:
continue
latest_seconds = max(snapshot["latest_mtime_ns"], snapshot["latest_ctime_ns"]) / 1_000_000_000
age_seconds = max(0.0, time.time() - latest_seconds)
classification, reasons = classification_for(
spec["kind"],
snapshot,
age_seconds=age_seconds,
minimum_age_seconds=minimum_age_seconds,
processes=processes,
root_is_symlink=root_is_symlink,
root_is_bound=roots_by_path.get(spec["root"], {}).get("chain", {}).get("status") == "bound",
)
if any(
is_same_or_within(path, lease_root) or is_same_or_within(lease_root, path)
for lease_root in leases.get("unresolved", [])
):
classification = "protected"
reasons.append("incomplete-session-reservation")
elif any(
is_same_or_within(path, lease_root) or is_same_or_within(lease_root, path)
for lease_root in leases.get("active", [])
):
classification = "active"
reasons.append("active-session-reservation")
identifier = candidate_id(spec["kind"], path, snapshot)
metadata: dict[str, Any] = {}
if spec["kind"] == "app-bundle":
metadata = app_identity(path)
candidates.append(
{
"id": identifier,
"label": f"{spec['kind']}:{identifier[:8]}",
"path": path,
"approved_root": spec["root"],
"kind": spec["kind"],
"source": spec["source"],
"classification": classification,
"reason_codes": reasons,
"age_seconds": int(age_seconds),
"snapshot": snapshot,
"metadata": metadata,
}
)
candidates.sort(key=lambda item: (-int(item["snapshot"]["allocated_bytes"]), item["id"]))
plan: dict[str, Any] = {
"schema_version": PLAN_SCHEMA,
"tool_version": VERSION,
"created_at": isoformat(created),
"expires_at": isoformat(created + dt.timedelta(minutes=int(args.ttl_minutes))),
"platform": sys.platform,
"profile": args.profile,
"policy": {
"minimum_age_seconds": int(minimum_age_seconds),
"minimum_size_bytes": minimum_size,
"max_candidates": int(args.max_candidates),
"max_entries_per_candidate": int(args.max_entries),
"scan_seconds_per_candidate": float(args.scan_seconds),
"total_entry_limit": total_entry_limit,
"total_scan_seconds": total_time_limit,
},
"audit": {
"status": "complete" if considered_specs == len(specs) else "partial",
"candidate_specs_total": len(specs),
"candidate_specs_considered": considered_specs,
"entries_scanned": total_entries,
"elapsed_seconds": round(time.monotonic() - audit_started, 3),
},
"process_check": processes,
"reservation_check": {
"status": leases.get("status", "unknown"),
"active_count": int(leases.get("active_lease_count", 0)),
"incomplete_count": int(leases.get("unresolved_lease_count", 0)),
},
"approved_roots": roots,
"candidates": candidates,
}
plan["plan_digest"] = digest_value(plan)
return plan
def verify_plan(plan: dict[str, Any]) -> str:
if plan.get("schema_version") != PLAN_SCHEMA:
raise StewardError("unsupported or missing plan schema")
supplied = plan.get("plan_digest")
if not isinstance(supplied, str):
raise StewardError("plan digest is missing")
unsigned = dict(plan)
unsigned.pop("plan_digest", None)
computed = digest_value(unsigned)
if supplied != computed:
raise StewardError("plan digest does not match plan contents")
return supplied
def summarize_plan(plan: dict[str, Any], *, saved_plan: bool = True) -> str:
digest = verify_plan(plan)
counts = {key: 0 for key in ("eligible", "review", "active", "protected")}
sizes = {key: 0 for key in counts}
for candidate in plan.get("candidates", []):
classification = candidate.get("classification", "review")
if classification not in counts:
classification = "review"
counts[classification] += 1
sizes[classification] += int(candidate.get("snapshot", {}).get("allocated_bytes", 0))
lines = [
f"Plan {digest}" if saved_plan else "Read-only audit summary",
f"Safe to reclaim: {counts['eligible']} item(s), estimated {human_bytes(sizes['eligible'])}",
f"Needs review: {counts['review']} item(s), measured {human_bytes(sizes['review'])}",
f"Active/recent: {counts['active']} item(s), measured {human_bytes(sizes['active'])}",
f"Protected: {counts['protected']} item(s), measured {human_bytes(sizes['protected'])}",
]
audit = plan.get("audit", {})
if audit.get("status") != "complete":
remaining = max(
0,
int(audit.get("candidate_specs_total", 0)) - int(audit.get("candidate_specs_considered", 0)),
)
lines.append(f"Audit status: partial; {remaining} candidate(s) were not scanned before the global limit.")
eligible = [item for item in plan.get("candidates", []) if item.get("classification") == "eligible"]
if eligible:
lines.append("Eligible item IDs (still require approval):")
for item in eligible[:20]:
lines.append(
f"- {item['id']} · {item['kind']} · estimated {human_bytes(int(item['snapshot']['allocated_bytes']))}"
)
if len(eligible) > 20:
lines.append(f"- {len(eligible) - 20} more eligible item(s) remain in the local plan")
if saved_plan:
lines.append("Exact paths remain only in the local mode-0600 plan.")
else:
lines.append("No plan, output file, or local state was created.")
return "\n".join(lines)
@contextlib.contextmanager
def executor_lock(state_dir: Path) -> Iterator[None]:
normalized = normalize_path(state_dir)
if os.path.realpath(normalized) != normalized:
raise StewardError("state directory may not contain symlink components")
state_dir = Path(normalized)
state_dir.mkdir(mode=0o700, parents=True, exist_ok=True)
state_stat = os.lstat(state_dir)
owner = os.getuid() if hasattr(os, "getuid") else int(state_stat.st_uid)
if stat.S_ISLNK(state_stat.st_mode) or not stat.S_ISDIR(state_stat.st_mode):
raise StewardError("state path is not a real directory")
if int(state_stat.st_uid) != owner or stat.S_IMODE(state_stat.st_mode) & 0o022:
raise StewardError("state directory ownership or permissions are unsafe")
lock_path = state_dir / "executor.lock"
nofollow = getattr(os, "O_NOFOLLOW", None)
if nofollow is None:
raise StewardError("no-follow file locking is unavailable")
flags = os.O_RDWR | os.O_CREAT | nofollow | getattr(os, "O_CLOEXEC", 0)
try:
descriptor = os.open(lock_path, flags, 0o600)
except OSError as error:
raise StewardError("executor lock could not be opened safely") from error
try:
lock_stat = os.fstat(descriptor)
if (
not stat.S_ISREG(lock_stat.st_mode)
or int(lock_stat.st_uid) != owner
or int(lock_stat.st_nlink) != 1
or stat.S_IMODE(lock_stat.st_mode) & 0o077
):
raise StewardError("executor lock identity is unsafe")
try:
import fcntl
fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
except (ImportError, BlockingIOError, OSError) as error:
raise StewardError("another Build Steward executor is active, or locking is unavailable") from error
yield
finally:
with contextlib.suppress(Exception):
import fcntl
fcntl.flock(descriptor, fcntl.LOCK_UN)
os.close(descriptor)
def ensure_private_subdirectory(state_dir: Path, name: str) -> Path:
if not name or name in {".", ".."} or os.path.sep in name:
raise StewardError("invalid private state directory name")
destination = state_dir / name
destination.mkdir(mode=0o700, parents=False, exist_ok=True)
observed = os.lstat(destination)
owner = os.getuid() if hasattr(os, "getuid") else int(observed.st_uid)
if (
stat.S_ISLNK(observed.st_mode)
or not stat.S_ISDIR(observed.st_mode)
or int(observed.st_uid) != owner
or stat.S_IMODE(observed.st_mode) & 0o077
):
raise StewardError("private state directory identity or permissions are unsafe")
return destination
def load_lease_states(leases_dir: Path) -> dict[str, dict[str, Any]]:
entries = []
with os.scandir(leases_dir) as iterator:
for entry in iterator:
if not entry.name.endswith(".json"):
continue
if len(entries) >= 10_000:
raise StewardError("lease event limit exceeded")
if not entry.is_file(follow_symlinks=False):
raise StewardError("lease state contains a non-file entry")
entries.append(entry.path)
latest: dict[str, dict[str, Any]] = {}
for path in sorted(entries):
event = read_json(path)
lease_id = event.get("lease_id")
sequence = event.get("sequence")
if (
event.get("schema_version") != LEASE_EVENT_SCHEMA
or not isinstance(lease_id, str)
or len(lease_id) != 24
or any(character not in "0123456789abcdef" for character in lease_id)
or not isinstance(sequence, int)
or event.get("event") not in {"started", "heartbeat", "closed"}
):
raise StewardError("lease state is malformed")
previous = latest.get(lease_id)
if previous is None or sequence > int(previous["sequence"]):
latest[lease_id] = event
return latest
def append_lease_event(leases_dir: Path, event: dict[str, Any]) -> None:
suffix = secrets.token_hex(4)
filename = f"{int(event['sequence']):020d}-{event['lease_id']}-{event['event']}-{suffix}.json"
atomic_create_json(leases_dir / filename, event)
def active_reserved_bytes(states: dict[str, dict[str, Any]], device: int) -> int:
return sum(
int(event.get("expected_peak_bytes", 0))
for event in states.values()
if event.get("event") != "closed" and int(event.get("volume_device", -1)) == int(device)
)
def normalized_attributed_roots(values: Iterable[str], workspace: str) -> list[dict[str, Any]]:
roots: list[dict[str, Any]] = []
for raw in values:
path = normalize_path(raw)
identity: dict[str, Any] = {"status": "missing"}
if os.path.lexists(path):
observed = os.lstat(path)
identity = {
"status": "bound",
"device": int(observed.st_dev),
"inode": int(observed.st_ino),
"mode": int(observed.st_mode),
"uid": int(observed.st_uid),
}
roots.append(
{
"path": path,
"workspace_relative": os.path.relpath(path, workspace) if is_same_or_within(path, workspace) else None,
"identity": identity,
}
)
return roots
def lease_start(args: argparse.Namespace) -> dict[str, Any]:
workspace = normalize_path(args.workspace)
workspace_stat = os.lstat(workspace)
owner = os.getuid() if hasattr(os, "getuid") else int(workspace_stat.st_uid)
if (
stat.S_ISLNK(workspace_stat.st_mode)
or not stat.S_ISDIR(workspace_stat.st_mode)
or int(workspace_stat.st_uid) != owner
):
raise StewardError("workspace identity or ownership is unsafe")
state_dir = Path(normalize_path(args.state_dir or str(Path.home() / ".build-steward")))
requested = int(float(args.expected_peak_gib) * 1024**3)
floor = int(float(args.hard_floor_gib) * 1024**3)
if requested <= 0 or floor < 0:
raise StewardError("lease size and floor must be non-negative")
with executor_lock(state_dir):
leases_dir = ensure_private_subdirectory(state_dir, "leases")
states = load_lease_states(leases_dir)
processes = process_state()
free = free_bytes_for(workspace)
reserved = active_reserved_bytes(states, int(workspace_stat.st_dev))
remaining = free - reserved - requested
blocked_reason = None
warning = None
if remaining < floor:
blocked_reason = "headroom-floor-not-met"
elif processes["status"] != "available":
warning = "process-visibility-unavailable"
elif processes["known_producers_running"] and not any(
event.get("event") != "closed" for event in states.values()
):
blocked_reason = "unleased-build-producer-running"
if blocked_reason:
return {
"schema_version": "build-steward.lease-result.v1",
"status": "blocked",
"reason": blocked_reason,
"free_bytes": free,
"active_reserved_bytes": reserved,
"requested_peak_bytes": requested,
"hard_floor_bytes": floor,
"remaining_after_reservations_bytes": remaining,
}
lease_id = secrets.token_hex(12)
sequence = time.time_ns()
workspace_material = (
f"{workspace}\0{int(workspace_stat.st_dev)}\0{int(workspace_stat.st_ino)}"
).encode("utf-8")
event = {
"schema_version": LEASE_EVENT_SCHEMA,
"event": "started",
"sequence": sequence,
"at": isoformat(utc_now()),
"lease_id": lease_id,
"client": args.client,
"workspace": workspace,
"workspace_fingerprint": "sha256:" + hashlib.sha256(workspace_material).hexdigest(),
"workspace_device": int(workspace_stat.st_dev),
"workspace_inode": int(workspace_stat.st_ino),
"volume_device": int(workspace_stat.st_dev),
"expected_peak_bytes": requested,
"hard_floor_bytes": floor,
"free_bytes_at_start": free,
"process_check": processes,
"warning": warning,
"cache_roots": normalized_attributed_roots(args.cache_root, workspace),
"work_roots": normalized_attributed_roots(args.work_root, workspace),
"outcome": None,
}
append_lease_event(leases_dir, event)
return {
"schema_version": "build-steward.lease-result.v1",
"status": "active",
"lease_id": lease_id,
"free_bytes": free,
"active_reserved_bytes": reserved + requested,
"hard_floor_bytes": floor,
"remaining_after_reservations_bytes": remaining,
"warning": warning,
}
def lease_transition(args: argparse.Namespace, event_name: str) -> dict[str, Any]:
state_dir = Path(normalize_path(args.state_dir or str(Path.home() / ".build-steward")))
with executor_lock(state_dir):
leases_dir = ensure_private_subdirectory(state_dir, "leases")
states = load_lease_states(leases_dir)
current = states.get(args.lease)
if current is None:
raise StewardError("unknown lease ID")
if current.get("event") == "closed":
if event_name == "closed":
return current
raise StewardError("lease is already closed")
event = dict(current)
event["event"] = event_name
event["sequence"] = time.time_ns()
event["at"] = isoformat(utc_now())
if event_name == "closed":
free_after = free_bytes_for(event["workspace"])
event["free_bytes_at_close"] = free_after
event["observed_free_delta_bytes"] = free_after - int(event["free_bytes_at_start"])
event["outcome"] = args.outcome
append_lease_event(leases_dir, event)
return event
def lease_root_context(state_path: str) -> dict[str, Any]:
state_dir = Path(normalize_path(state_path))
if not os.path.lexists(state_dir):
return {"status": "none", "active": [], "unresolved": []}
if os.path.realpath(state_dir) != str(state_dir):
raise StewardError("lease state path contains symlink components")
state_stat = os.lstat(state_dir)
owner = os.getuid() if hasattr(os, "getuid") else int(state_stat.st_uid)
if (
not stat.S_ISDIR(state_stat.st_mode)
or stat.S_ISLNK(state_stat.st_mode)
or int(state_stat.st_uid) != owner
or stat.S_IMODE(state_stat.st_mode) & 0o022
):
raise StewardError("lease state directory is unsafe")
leases_dir = state_dir / "leases"
if not os.path.lexists(leases_dir):
return {"status": "none", "active": [], "unresolved": []}
leases_stat = os.lstat(leases_dir)
if (
not stat.S_ISDIR(leases_stat.st_mode)
or stat.S_ISLNK(leases_stat.st_mode)
or int(leases_stat.st_uid) != owner
or stat.S_IMODE(leases_stat.st_mode) & 0o077
):
raise StewardError("lease event directory is unsafe")
states = load_lease_states(leases_dir)
active: list[str] = []
unresolved: list[str] = []
for event in states.values():
roots = [
item.get("path")
for item in [*event.get("cache_roots", []), *event.get("work_roots", [])]
if isinstance(item, dict) and isinstance(item.get("path"), str)
]
if event.get("event") != "closed":
active.extend(roots)
elif event.get("outcome") in {"failed", "interrupted"}:
unresolved.extend(roots)
return {
"status": "loaded",
"active": sorted(set(active)),
"unresolved": sorted(set(unresolved)),
"active_lease_count": sum(1 for event in states.values() if event.get("event") != "closed"),
"unresolved_lease_count": sum(
1
for event in states.values()
if event.get("event") == "closed" and event.get("outcome") in {"failed", "interrupted"}
),
"active_lease_ids": sorted(
lease_id for lease_id, event in states.items() if event.get("event") != "closed"
),
}
@contextlib.contextmanager
def open_bound_root(root_record: dict[str, Any]) -> Iterator[int]:
chain = root_record.get("chain", {})
if not verify_directory_chain(chain):
raise StewardError("approved root identity changed")
nofollow = getattr(os, "O_NOFOLLOW", None)
directory = getattr(os, "O_DIRECTORY", None)
if nofollow is None or directory is None:
raise StewardError("secure directory handles are unavailable")
flags = os.O_RDONLY | directory | nofollow | getattr(os, "O_CLOEXEC", 0)
try:
descriptor = os.open(root_record["path"], flags)
except OSError as error:
raise StewardError("approved root could not be opened securely") from error
try:
observed = os.fstat(descriptor)
expected = chain["entries"][-1]
if (
int(observed.st_dev) != int(expected["device"])
or int(observed.st_ino) != int(expected["inode"])
or int(observed.st_mode) != int(expected["mode"])
or int(observed.st_uid) != int(expected["uid"])
or not stat.S_ISDIR(observed.st_mode)
):
raise StewardError("approved root handle does not match the plan")
yield descriptor
finally:
os.close(descriptor)
def rename_no_replace_at(parent_fd: int, source_name: str, destination_name: str) -> None:
"""Atomically quarantine one direct child without following or replacing."""
if (
not source_name
or not destination_name
or source_name in {".", ".."}
or destination_name in {".", ".."}
or os.path.sep in source_name
or os.path.sep in destination_name
):
raise StewardError("unsafe quarantine entry name")
library = ctypes.CDLL(None, use_errno=True)
source = os.fsencode(source_name)
destination = os.fsencode(destination_name)
if sys.platform == "darwin" and hasattr(library, "renameatx_np"):
operation = library.renameatx_np
operation.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint]
operation.restype = ctypes.c_int
result = operation(parent_fd, source, parent_fd, destination, 0x00000004 | 0x00000010)
elif sys.platform.startswith("linux") and hasattr(library, "renameat2"):
operation = library.renameat2
operation.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint]
operation.restype = ctypes.c_int
result = operation(parent_fd, source, parent_fd, destination, 0x00000001)
else:
raise StewardError("atomic no-replace quarantine is unavailable")
if result != 0:
error_number = ctypes.get_errno()
if error_number == errno.EEXIST:
raise StewardError("quarantine destination already exists")
raise StewardError("atomic quarantine failed") from OSError(error_number, os.strerror(error_number))
def secure_clear_directory(directory_fd: int, expected_device: int) -> None:
"""Remove entries beneath an anchored directory without crossing devices."""
nofollow = getattr(os, "O_NOFOLLOW", None)
directory_flag = getattr(os, "O_DIRECTORY", None)
if nofollow is None or directory_flag is None:
raise StewardError("secure recursive deletion is unavailable")
duplicate = os.dup(directory_fd)
try:
with os.scandir(duplicate) as iterator:
names = sorted(entry.name for entry in iterator)
finally:
with contextlib.suppress(OSError):
os.close(duplicate)
for name in names:
try:
before = os.stat(name, dir_fd=directory_fd, follow_symlinks=False)
except OSError as error:
raise StewardError("quarantined entry changed during deletion") from error
if int(before.st_dev) != int(expected_device):
raise StewardError("refusing to cross a filesystem boundary during deletion")
if stat.S_ISDIR(before.st_mode):
flags = os.O_RDONLY | directory_flag | nofollow | getattr(os, "O_CLOEXEC", 0)
try:
child_fd = os.open(name, flags, dir_fd=directory_fd)
except OSError as error:
raise StewardError("quarantined directory could not be opened securely") from error
try:
opened = os.fstat(child_fd)
if (
int(opened.st_dev) != int(before.st_dev)
or int(opened.st_ino) != int(before.st_ino)
or not stat.S_ISDIR(opened.st_mode)
):
raise StewardError("quarantined directory identity changed")
secure_clear_directory(child_fd, expected_device)
finally:
os.close(child_fd)
os.rmdir(name, dir_fd=directory_fd)
else:
os.unlink(name, dir_fd=directory_fd)
def secure_remove_quarantined(parent_fd: int, tombstone_name: str, expected: dict[str, Any]) -> None:
nofollow = getattr(os, "O_NOFOLLOW", None)
directory_flag = getattr(os, "O_DIRECTORY", None)
if nofollow is None or directory_flag is None:
raise StewardError("secure recursive deletion is unavailable")
flags = os.O_RDONLY | directory_flag | nofollow | getattr(os, "O_CLOEXEC", 0)
try:
tombstone_fd = os.open(tombstone_name, flags, dir_fd=parent_fd)
except OSError as error:
raise StewardError("quarantine could not be opened securely") from error
try:
observed = os.fstat(tombstone_fd)
if (
int(observed.st_dev) != int(expected["device"])
or int(observed.st_ino) != int(expected["inode"])
or int(observed.st_uid) != int(expected["uid"])
or not stat.S_ISDIR(observed.st_mode)
):
raise StewardError("quarantine identity does not match the approved item")
secure_clear_directory(tombstone_fd, int(expected["device"]))
finally:
os.close(tombstone_fd)
os.rmdir(tombstone_name, dir_fd=parent_fd)
os.fsync(parent_fd)
def snapshots_match(before: dict[str, Any], after: dict[str, Any]) -> bool:
keys = (
"device",
"inode",
"mode",
"uid",
"latest_mtime_ns",
"latest_ctime_ns",
"allocated_bytes",
"entry_count",
"tree_digest",
"crossed_mount",
"foreign_owner",
"contains_vcs",
"symlink_count",
"truncated",
"error_count",
)
return all(before.get(key) == after.get(key) for key in keys)
def trusted_automatic_candidate(candidate: dict[str, Any], root: str) -> bool:
if candidate.get("source") != "macos-power-user":
return False
path = normalize_path(candidate["path"])
if os.path.dirname(path) != root:
return False
kind = candidate.get("kind")
if kind == "xcode-derived-data":
expected = normalize_path(Path.home() / "Library/Developer/Xcode/DerivedData")
return root == expected
if kind == "xcode-temp-pipeline":
allowed = {normalize_path("/private/tmp")}
environment_temp = os.environ.get("TMPDIR")
if environment_temp:
allowed.add(os.path.realpath(normalize_path(environment_temp)))
return root in allowed and os.path.basename(path).lower().startswith("xcodedistpipeline.~~~")
return False
def validate_candidate_boundary(candidate: dict[str, Any], approved_roots: dict[str, dict[str, Any]]) -> None:
path = normalize_path(candidate["path"])
root = normalize_path(candidate["approved_root"])
root_record = approved_roots.get(root)
if root_record is None:
raise StewardError(f"item {candidate['id']} references an unapproved root")
if not is_within(path, root):
raise StewardError(f"item {candidate['id']} escapes or equals its approved root")
if os.path.dirname(path) != root:
raise StewardError(f"item {candidate['id']} is not a direct child of its approved root")
if any(character in path for character in ("*", "?", "[", "]")):
raise StewardError(f"item {candidate['id']} contains glob metacharacters")
if not verify_directory_chain(root_record.get("chain", {})):
raise StewardError(f"item {candidate['id']} has a changed or unsafe approved root")
if int(candidate.get("snapshot", {}).get("device", -1)) != int(
root_record.get("chain", {}).get("entries", [{}])[-1].get("device", -2)
):
raise StewardError(f"item {candidate['id']} is on an unexpected filesystem")
if not trusted_automatic_candidate(candidate, root):
raise StewardError(f"item {candidate['id']} is not a trusted automatic cache class")
def free_bytes_for(path: str) -> int:
probe = path if os.path.exists(path) else os.path.dirname(path)
return int(shutil.disk_usage(probe).free)
def volume_free_snapshot(roots: Iterable[str]) -> dict[str, dict[str, Any]]:
"""Measure each underlying device once, even when several roots share it."""
result: dict[str, dict[str, Any]] = {}
for root in sorted({normalize_path(value) for value in roots}):
device = str(int(os.stat(root).st_dev))
if device not in result:
result[device] = {"probe_root": root, "free_bytes": free_bytes_for(root)}
return result
def action_key(plan_digest: str, item_ids: Iterable[str]) -> str:
material = {"plan_digest": plan_digest, "item_ids": sorted(set(item_ids))}
return hashlib.sha256(canonical_json(material)).hexdigest()
def approval_digest(plan_digest: str, selected: Iterable[dict[str, Any]]) -> str:
items = list(selected)
material = {
"schema_version": "build-steward.approval.v2",
"plan_digest": plan_digest,
"item_ids": sorted(item["id"] for item in items),
"estimated_allocated_bytes": sum(int(item["snapshot"]["allocated_bytes"]) for item in items),
}
return digest_value(material)
def local_review_document(plan: dict[str, Any], selected: list[dict[str, Any]]) -> str:
digest = verify_plan(plan)
approval = approval_digest(digest, selected)
rows = []
for item in selected:
rows.append(
"<tr>"
f"<td><code>{html.escape(item['id'])}</code></td>"
f"<td>{html.escape(item['kind'])}</td>"
f"<td>{html.escape(item['path'])}</td>"
f"<td>{html.escape(human_bytes(int(item['snapshot']['allocated_bytes'])))}</td>"
f"<td>{int(item['age_seconds']) // 3600} h</td>"
"</tr>"
)
return """<!doctype html>
<meta charset="utf-8">
<meta name="robots" content="noindex,nofollow">
<title>Build Steward local approval review</title>
<style>body{font:16px system-ui;max-width:1100px;margin:40px auto;padding:0 20px;color:#17211d}table{border-collapse:collapse;width:100%}th,td{border:1px solid #b8c6c0;padding:9px;text-align:left}code{word-break:break-all}.warning{background:#fff4cc;padding:12px}</style>
<h1>Build Steward local approval review</h1>
<p class="warning"><strong>Private local file:</strong> verify every exact path. Do not upload or paste this page into a chat.</p>
<p>Plan: <code>""" + html.escape(digest) + """</code></p>
<p>Action approval: <code>""" + html.escape(approval) + """</code></p>
<table><thead><tr><th>Item ID</th><th>Proven class</th><th>Exact path</th><th>Estimated allocation</th><th>Age</th></tr></thead><tbody>""" + "".join(rows) + """</tbody></table>
<p>Approval applies only to these item IDs under this unexpired plan. The executor will revalidate everything.</p>
"""
def select_eligible(plan: dict[str, Any], item_ids: Iterable[str]) -> list[dict[str, Any]]:
requested = list(dict.fromkeys(item_ids))
if not requested:
raise StewardError("at least one exact item ID is required")
candidates_by_id = {item["id"]: item for item in plan.get("candidates", [])}
if any(item_id not in candidates_by_id for item_id in requested):
raise StewardError("one or more requested item IDs are not in this plan")
selected = [candidates_by_id[item_id] for item_id in requested]
if any(item.get("classification") != "eligible" for item in selected):
raise StewardError("only items classified as eligible may be prepared or applied")
return selected
def apply_plan(args: argparse.Namespace) -> dict[str, Any]:
plan = read_json(args.plan)
plan_digest = verify_plan(plan)
if utc_now() > parse_time(plan["expires_at"]):
raise StewardError("plan expired; run a fresh audit")
selected = select_eligible(plan, args.item)
action_approval = approval_digest(plan_digest, selected)
if args.approve != action_approval:
raise StewardError("approval digest does not match this exact item selection")
approved_roots = {
normalize_path(item["path"]): item
for item in plan.get("approved_roots", [])
if isinstance(item, dict) and isinstance(item.get("path"), str)
}
for candidate in selected:
validate_candidate_boundary(candidate, approved_roots)
state_dir = Path(normalize_path(args.state_dir or str(Path.home() / ".build-steward")))
control_paths = [normalize_path(args.plan), normalize_path(args.receipt), str(state_dir)]
for candidate in selected:
if any(control_path_touches_candidate(path, candidate["path"]) for path in control_paths):
raise StewardError("plan, receipt, and state paths must stay outside every selected item")
receipts_dir = state_dir / "receipts"
key = action_key(plan_digest, [item["id"] for item in selected])
canonical_receipt = receipts_dir / f"{key}.json"
pending_receipt = receipts_dir / f"{key}.pending.json"
with executor_lock(state_dir):
receipts_dir.mkdir(mode=0o700, parents=False, exist_ok=True)
receipts_stat = os.lstat(receipts_dir)
owner = os.getuid() if hasattr(os, "getuid") else int(receipts_stat.st_uid)
if (
stat.S_ISLNK(receipts_stat.st_mode)
or not stat.S_ISDIR(receipts_stat.st_mode)
or int(receipts_stat.st_uid) != owner
or stat.S_IMODE(receipts_stat.st_mode) & 0o022
):
raise StewardError("receipt state directory identity or permissions are unsafe")
if os.path.lexists(canonical_receipt):
receipt = read_json(canonical_receipt)
create_or_confirm_json(args.receipt, receipt)
return receipt
if os.path.lexists(pending_receipt):
raise StewardError("an incomplete prior action exists; inspect its private pending receipt before retrying")
if os.path.lexists(args.receipt):
raise StewardError("refusing to overwrite an existing receipt path")
current_processes = process_state()
if current_processes["status"] != "available":
raise StewardError("process visibility is unavailable; no items were changed")
if any(item["kind"].startswith("xcode-") for item in selected) and current_processes["known_producers_running"]:
raise StewardError("a known build producer is running; no items were changed")
reserve_bytes = int(args.receipt_reserve_mib) * 1024 * 1024
if free_bytes_for(str(state_dir)) < reserve_bytes:
raise StewardError("insufficient space on the receipt volume")
for candidate in selected:
if free_bytes_for(candidate["approved_root"]) < reserve_bytes:
raise StewardError("insufficient space to persist recovery receipts safely")
preflight_results: list[dict[str, Any]] = []
policy = plan["policy"]
for candidate in selected:
path = candidate["path"]
if not os.path.lexists(path):
preflight_results.append({"id": candidate["id"], "status": "already-absent"})
continue
root_stat = os.lstat(path)
if stat.S_ISLNK(root_stat.st_mode):
preflight_results.append({"id": candidate["id"], "status": "blocked-symlink"})
continue
current_snapshot = snapshot_path(
path,
max_entries=int(policy["max_entries_per_candidate"]),
max_seconds=float(policy["scan_seconds_per_candidate"]),
)
if not snapshots_match(candidate["snapshot"], current_snapshot):
preflight_results.append({"id": candidate["id"], "status": "stale-plan"})
continue
handles = check_open_handles(path)
if handles != "clear":
preflight_results.append({"id": candidate["id"], "status": f"blocked-open-handles-{handles}"})
continue
preflight_results.append({"id": candidate["id"], "status": "ready"})
if any(result["status"] != "ready" for result in preflight_results):
raise StewardError("one or more approved items changed or became active; no items were changed")
started_at = utc_now()
before_free = volume_free_snapshot(item["approved_root"] for item in selected)
pre_receipt = {
"schema_version": RECEIPT_SCHEMA,
"tool_version": VERSION,
"action_key": key,
"status": "prepared",
"plan_digest": plan_digest,
"approval_digest": action_approval,
"started_at": isoformat(started_at),
"selected_items": [
{
"id": item["id"],
"path": item["path"],
"tombstone": os.path.join(
item["approved_root"], f".build-steward-reclaim-{item['id']}"
),
"kind": item["kind"],
"estimated_allocated_bytes": int(item["snapshot"]["allocated_bytes"]),
}
for item in selected
],
"preflight": preflight_results,
"free_bytes_before": before_free,
"effects": [],
}
atomic_create_json(pending_receipt, pre_receipt)
effects: list[dict[str, Any]] = []
interrupted = False
for candidate in selected:
identifier = candidate["id"]
path = candidate["path"]
root = normalize_path(candidate["approved_root"])
source_name = os.path.basename(path)
tombstone_name = f".build-steward-reclaim-{identifier}"
tombstone = os.path.join(root, tombstone_name)
try:
validate_candidate_boundary(candidate, approved_roots)
immediate_snapshot = snapshot_path(
path,
max_entries=int(policy["max_entries_per_candidate"]),
max_seconds=float(policy["scan_seconds_per_candidate"]),
)
except (OSError, StewardError) as error:
effects.append(
{"id": identifier, "status": "skipped", "reason": type(error).__name__}
)
break
if not snapshots_match(candidate["snapshot"], immediate_snapshot):
effects.append({"id": identifier, "status": "skipped", "reason": "changed-before-reclaim"})
continue
if check_open_handles(path) != "clear":
effects.append({"id": identifier, "status": "skipped", "reason": "open-handle-state-changed"})
break
try:
with open_bound_root(approved_roots[root]) as root_fd:
source_stat = os.stat(source_name, dir_fd=root_fd, follow_symlinks=False)
if (
int(source_stat.st_dev) != int(candidate["snapshot"]["device"])
or int(source_stat.st_ino) != int(candidate["snapshot"]["inode"])
or not stat.S_ISDIR(source_stat.st_mode)
):
raise StewardError("candidate identity changed before quarantine")
rename_no_replace_at(root_fd, source_name, tombstone_name)
os.fsync(root_fd)
tombstone_stat = os.lstat(tombstone)
if (
int(tombstone_stat.st_dev) != int(candidate["snapshot"]["device"])
or int(tombstone_stat.st_ino) != int(candidate["snapshot"]["inode"])
or not stat.S_ISDIR(tombstone_stat.st_mode)
):
raise StewardError("quarantine identity changed unexpectedly")
atomic_create_json(
receipts_dir / f"{key}.{identifier}.quarantined.json",
{
"schema_version": RECEIPT_SCHEMA,
"action_key": key,
"item_id": identifier,
"state": "quarantined",
"at": isoformat(utc_now()),
"tombstone": tombstone,
},
)
quarantined_snapshot = snapshot_path(
tombstone,
max_entries=int(policy["max_entries_per_candidate"]),
max_seconds=float(policy["scan_seconds_per_candidate"]),
)
if not snapshots_match(candidate["snapshot"], quarantined_snapshot):
raise StewardError("quarantined item changed; manual recovery is required")
if check_open_handles(tombstone) != "clear":
raise StewardError("quarantined item has an open handle; manual recovery is required")
with open_bound_root(approved_roots[root]) as root_fd:
secure_remove_quarantined(root_fd, tombstone_name, candidate["snapshot"])
atomic_create_json(
receipts_dir / f"{key}.{identifier}.reclaimed.json",
{
"schema_version": RECEIPT_SCHEMA,
"action_key": key,
"item_id": identifier,
"state": "reclaimed",
"at": isoformat(utc_now()),
},
)
effects.append(
{
"id": identifier,
"status": "reclaimed",
"estimated_allocated_bytes": int(candidate["snapshot"]["allocated_bytes"]),
}
)
except (Exception, KeyboardInterrupt) as error: # receipt the partial state, then stop widening effects
interrupted = isinstance(error, KeyboardInterrupt)
effects.append(
{
"id": identifier,
"status": "failed",
"reason": type(error).__name__,
"recovery_tombstone": tombstone if os.path.lexists(tombstone) else None,
}
)
break
after_free = {
device: {
"probe_root": measurement["probe_root"],
"free_bytes": free_bytes_for(measurement["probe_root"]),
}
for device, measurement in before_free.items()
}
actual_delta = {
device: after_free[device]["free_bytes"] - before_free[device]["free_bytes"]
for device in before_free
}
failed = any(item["status"] == "failed" for item in effects)
reclaimed = sum(1 for item in effects if item["status"] == "reclaimed")
final_receipt = dict(pre_receipt)
status_value = "interrupted" if interrupted else (
"partial" if failed or reclaimed < len(selected) else "complete"
)
final_receipt.update(
{
"status": status_value,
"completed_at": isoformat(utc_now()),
"effects": effects,
"free_bytes_after": after_free,
"actual_free_delta_bytes": actual_delta,
}
)
final_receipt["requested_receipt_copy"] = "written"
try:
atomic_create_json(args.receipt, final_receipt)
except StewardError:
final_receipt["requested_receipt_copy"] = "blocked-no-clobber"
atomic_create_json(canonical_receipt, final_receipt)
with contextlib.suppress(FileNotFoundError):
os.unlink(pending_receipt)
return final_receipt
def command_preflight(args: argparse.Namespace) -> int:
path = normalize_path(args.path)
free = free_bytes_for(path)
required = int(float(args.required_gib) * 1024**3)
status_value = "ready" if free >= required else "blocked"
payload = {
"schema_version": "build-steward.preflight.v1",
"status": status_value,
"free_bytes": free,
"required_bytes": required,
"shortfall_bytes": max(0, required - free),
}
if args.json:
print(json.dumps(payload, sort_keys=True))
else:
print(f"Build preflight: {status_value}")
print(f"Free: {human_bytes(free)}")
print(f"Required: {human_bytes(required)}")
if payload["shortfall_bytes"]:
print(f"Shortfall: {human_bytes(payload['shortfall_bytes'])}")
return 0 if status_value == "ready" else 2
def command_reserve(args: argparse.Namespace) -> int:
result = lease_start(args)
print(f"Build reservation: {result['status']}")
print(f"Free now: {human_bytes(int(result['free_bytes']))}")
print(f"Reserved after this request: {human_bytes(int(result['active_reserved_bytes']))}")
print(f"Hard floor: {human_bytes(int(result['hard_floor_bytes']))}")
projected = int(result["remaining_after_reservations_bytes"])
if projected >= 0:
print(f"Projected free after reservations: {human_bytes(projected)}")
else:
print(f"Projected disk deficit: {human_bytes(-projected)}")
if projected < int(result["hard_floor_bytes"]):
print(
"Headroom shortfall: "
f"{human_bytes(int(result['hard_floor_bytes']) - projected)}"
)
if result["status"] == "active":
print(f"Reservation ID: {result['lease_id']}")
if result.get("warning"):
print(f"Warning: {result['warning']}")
return 0
print(f"Reason: {result['reason']}")
return 2
def command_release(args: argparse.Namespace) -> int:
result = lease_transition(args, "closed")
print(f"Build reservation released: {result['lease_id']}")
print(f"Outcome: {result.get('outcome', args.outcome)}")
if "observed_free_delta_bytes" in result:
print(
"Observed free-space change during reservation: "
f"{human_bytes(int(result['observed_free_delta_bytes']))}"
)
return 0
def command_reservations(args: argparse.Namespace) -> int:
state_path = args.state_dir or str(Path.home() / ".build-steward")
context = lease_root_context(state_path)
active_ids = context.get("active_lease_ids", [])
print(f"Active build reservations: {len(active_ids)}")
for lease_id in active_ids:
print(f"- {lease_id}")
print(f"Incomplete build sessions protected: {int(context.get('unresolved_lease_count', 0))}")
return 0
def command_audit(args: argparse.Namespace) -> int:
if not args.root and not args.rebuildable and args.profile == "none":
raise StewardError("choose --profile macos-power-user, --root, or --rebuildable")
plan = build_plan(args)
if bool(getattr(args, "read_only", False)):
print(summarize_plan(plan, saved_plan=False))
return 0
output = getattr(args, "output", None)
if not isinstance(output, str) or not output:
raise StewardError("choose exactly one audit result mode: --read-only or --output")
for candidate in plan.get("candidates", []):
if control_path_touches_candidate(output, candidate["path"]):
raise StewardError("plan output must stay outside every inventoried item")
atomic_create_json(output, plan)
print(summarize_plan(plan))
print("Plan saved locally with mode 0600; no candidate was changed.")
return 0
def command_summarize(args: argparse.Namespace) -> int:
print(summarize_plan(read_json(args.plan)))
return 0
def command_prepare(args: argparse.Namespace) -> int:
plan = read_json(args.plan)
plan_digest = verify_plan(plan)
if utc_now() > parse_time(plan["expires_at"]):
raise StewardError("plan expired; run a fresh audit")
selected = select_eligible(plan, args.item)
roots = {
normalize_path(item["path"]): item
for item in plan.get("approved_roots", [])
if isinstance(item, dict) and isinstance(item.get("path"), str)
}
for candidate in selected:
validate_candidate_boundary(candidate, roots)
if control_path_touches_candidate(args.review_output, candidate["path"]):
raise StewardError("local review output must stay outside every selected item")
atomic_create_text(args.review_output, local_review_document(plan, selected))
total = sum(int(item["snapshot"]["allocated_bytes"]) for item in selected)
print(f"Plan {plan_digest}")
print(f"Prepared selection: {len(selected)} item(s), estimated {human_bytes(total)}")
print(f"Action approval {approval_digest(plan_digest, selected)}")
print("Private local review created with mode 0600. Inspect every exact path outside the chat before approval.")
return 0
def command_apply(args: argparse.Namespace) -> int:
receipt = apply_plan(args)
effects = receipt.get("effects", [])
reclaimed = sum(1 for item in effects if item.get("status") == "reclaimed")
skipped = sum(1 for item in effects if item.get("status") == "skipped")
failed = sum(1 for item in effects if item.get("status") == "failed")
actual = sum(int(value) for value in receipt.get("actual_free_delta_bytes", {}).values())
print(f"Reclaim status: {receipt.get('status', 'unknown')}")
print(f"Reclaimed: {reclaimed}; skipped: {skipped}; failed: {failed}")
print(f"Observed free-space change: {human_bytes(actual)}")
print("Exact effects and recovery details remain in the local mode-0600 receipt.")
if receipt.get("status") == "interrupted":
return 130
if failed:
return 1
if skipped or receipt.get("status") != "complete" or receipt.get("requested_receipt_copy") != "written":
return 2
return 0
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="build-steward",
description="Audit and reclaim proven build residue with exact, expiring plans.",
)
parser.add_argument("--version", action="version", version=VERSION)
subparsers = parser.add_subparsers(dest="command", required=True)
preflight = subparsers.add_parser("preflight", help="Check a build free-space floor without writing")
preflight.add_argument("--path", default=".")
preflight.add_argument("--required-gib", type=float, default=20.0)
preflight.add_argument("--json", action="store_true")
preflight.set_defaults(handler=command_preflight)
reserve = subparsers.add_parser("reserve", help="Reserve shared disk headroom before a heavy build")
reserve.add_argument("--workspace", default=".")
reserve.add_argument("--client", choices=("codex", "claude", "chatgpt", "other"), required=True)
reserve.add_argument("--expected-peak-gib", type=float, default=10.0)
reserve.add_argument("--hard-floor-gib", type=float, default=20.0)
reserve.add_argument("--cache-root", action="append", default=[])
reserve.add_argument("--work-root", action="append", default=[])
reserve.add_argument("--state-dir")
reserve.set_defaults(handler=command_reserve)
release = subparsers.add_parser("release", help="Release a shared build reservation")
release.add_argument("--lease", required=True)
release.add_argument("--outcome", choices=("succeeded", "failed", "interrupted"), required=True)
release.add_argument("--state-dir")
release.set_defaults(handler=command_release)
reservations = subparsers.add_parser("reservations", help="List active reservation IDs without paths")
reservations.add_argument("--state-dir")
reservations.set_defaults(handler=command_reservations)
audit = subparsers.add_parser("audit", help="Run a bounded audit; save a plan or emit a strict read-only summary")
audit_mode = audit.add_mutually_exclusive_group(required=True)
audit_mode.add_argument("--output", help="New private path for an expiring plan")
audit_mode.add_argument("--read-only", action="store_true", help="Emit a redacted summary without creating files")
audit.add_argument("--profile", choices=("none", "macos-power-user"), default="none")
audit.add_argument("--root", action="append", default=[], help="Inspect immediate children of this exact root")
audit.add_argument("--rebuildable", action="append", default=[], help="Assert one exact path is rebuildable")
audit.add_argument("--min-age-hours", type=float, default=DEFAULT_MIN_AGE_HOURS)
audit.add_argument("--ttl-minutes", type=int, default=DEFAULT_TTL_MINUTES)
audit.add_argument("--max-candidates", type=int, default=DEFAULT_MAX_CANDIDATES)
audit.add_argument("--max-entries", type=int, default=DEFAULT_MAX_ENTRIES)
audit.add_argument("--scan-seconds", type=float, default=DEFAULT_SCAN_SECONDS)
audit.add_argument("--total-entries", type=int, default=DEFAULT_TOTAL_ENTRIES)
audit.add_argument("--total-scan-seconds", type=float, default=DEFAULT_TOTAL_SCAN_SECONDS)
audit.add_argument("--min-size-mib", type=int, default=DEFAULT_MIN_SIZE_MIB)
audit.add_argument("--state-dir", help="Shared reservation state; defaults to ~/.build-steward")
audit.set_defaults(handler=command_audit)
summarize = subparsers.add_parser("summarize", help="Print a redacted plan summary")
summarize.add_argument("--plan", required=True)
summarize.set_defaults(handler=command_summarize)
prepare = subparsers.add_parser("prepare", help="Create a private exact-path review for selected item IDs")
prepare.add_argument("--plan", required=True)
prepare.add_argument("--item", action="append", default=[], help="Exact eligible item ID; repeat as needed")
prepare.add_argument("--review-output", required=True)
prepare.set_defaults(handler=command_prepare)
apply = subparsers.add_parser("apply", help="Apply approved eligible item IDs after revalidation")
apply.add_argument("--plan", required=True)
apply.add_argument("--approve", required=True, help="Exact selection-specific approval digest from prepare")
apply.add_argument("--item", action="append", default=[], help="Exact eligible item ID; repeat as needed")
apply.add_argument("--receipt", required=True)
apply.add_argument("--state-dir")
apply.add_argument("--receipt-reserve-mib", type=int, default=DEFAULT_RECEIPT_RESERVE_MIB)
apply.set_defaults(handler=command_apply)
return parser
def main(argv: list[str] | None = None) -> int:
parser = build_parser()
args = parser.parse_args(argv)
try:
return int(args.handler(args))
except StewardError as error:
print(f"Build Steward stopped without claiming success: {error}", file=sys.stderr)
return 3
except KeyboardInterrupt:
print("Build Steward interrupted; inspect private action receipts before retrying", file=sys.stderr)
return 130
except (OSError, ValueError, TypeError, KeyError, json.JSONDecodeError):
print("Build Steward stopped without claiming success: local input or filesystem validation failed", file=sys.stderr)
return 4
if __name__ == "__main__":
raise SystemExit(main())
SHA-256: 982f4e92e71b59e96f8d77f947fdd494d94ac899bae690fb76173126febea541