425 lines
13 KiB
Python
425 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""Shared bootstrap library for export, verify, apply, and smoke tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import difflib
|
|
import fnmatch
|
|
import hashlib
|
|
import json
|
|
import shutil
|
|
import stat
|
|
import sys
|
|
import tempfile
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
|
|
BOOTSTRAP_ROOT = Path(__file__).resolve().parent
|
|
DEFAULT_OUTPUT = BOOTSTRAP_ROOT / "exports" / "generic"
|
|
STATE_FILE = ".agent-os-bootstrap-state.json"
|
|
TEXT_DIFF_MAX_BYTES = 100_000
|
|
|
|
|
|
@dataclass
|
|
class Conflict:
|
|
rel_path: str
|
|
target: Path
|
|
managed: bool
|
|
source_hash: str
|
|
target_hash: str
|
|
|
|
|
|
def load_json(path: Path) -> dict:
|
|
with path.open("r", encoding="utf-8") as handle:
|
|
return json.load(handle)
|
|
|
|
|
|
def parse_mode(mode: str) -> int:
|
|
return int(mode, 8)
|
|
|
|
|
|
def load_manifest() -> dict:
|
|
return load_json(BOOTSTRAP_ROOT / "bundle-manifest.json")
|
|
|
|
|
|
def load_rules() -> dict:
|
|
return load_json(BOOTSTRAP_ROOT / "export-rules.json")
|
|
|
|
|
|
def ensure_safe_output(output: Path) -> Path:
|
|
resolved = output.resolve()
|
|
if resolved.parts[-3:] != ("bootstrap", "exports", "generic"):
|
|
raise SystemExit(
|
|
"refusing to write outside a bootstrap/exports/generic directory: "
|
|
f"{resolved}"
|
|
)
|
|
return resolved
|
|
|
|
|
|
def copy_file(source: Path, target: Path, mode: str) -> None:
|
|
if not source.is_file():
|
|
raise FileNotFoundError(f"missing source file: {source}")
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copyfile(source, target)
|
|
target.chmod(parse_mode(mode))
|
|
|
|
|
|
def clean_output(output: Path) -> None:
|
|
ensure_safe_output(output)
|
|
if output.exists():
|
|
shutil.rmtree(output)
|
|
output.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
def create_directories(output: Path, manifest: dict) -> None:
|
|
for directory in manifest["directories"]:
|
|
target = output / directory["path"]
|
|
target.mkdir(parents=True, exist_ok=True)
|
|
if directory.get("keep"):
|
|
keep_file = target / ".gitkeep"
|
|
keep_file.touch()
|
|
keep_file.chmod(stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH)
|
|
|
|
|
|
def export_bundle(output: Path, manifest: dict | None = None, rules: dict | None = None) -> Path:
|
|
manifest = manifest or load_manifest()
|
|
rules = rules or load_rules()
|
|
clean_output(output)
|
|
create_directories(output, manifest)
|
|
|
|
for item in rules["metadata"]:
|
|
copy_file(
|
|
BOOTSTRAP_ROOT / item["source"],
|
|
output / item["target"],
|
|
item.get("mode", "0644"),
|
|
)
|
|
|
|
template_root = BOOTSTRAP_ROOT / rules["template_root"]
|
|
for item in manifest["templates"]:
|
|
copy_file(
|
|
template_root / item["source"],
|
|
output / item["target"],
|
|
item.get("mode", "0644"),
|
|
)
|
|
return output
|
|
|
|
|
|
def verify_required(output: Path, rules: dict) -> list[str]:
|
|
failures: list[str] = []
|
|
for item in rules["verify"]["required"]:
|
|
target = output / item["path"]
|
|
expected_type = item["type"]
|
|
if expected_type == "file" and not target.is_file():
|
|
failures.append(f"missing file: {item['path']}")
|
|
elif expected_type == "directory" and not target.is_dir():
|
|
failures.append(f"missing directory: {item['path']}")
|
|
return failures
|
|
|
|
|
|
def verify_forbidden(output: Path, rules: dict) -> list[str]:
|
|
failures: list[str] = []
|
|
for path in output.rglob("*"):
|
|
rel = path.relative_to(output).as_posix()
|
|
for pattern in rules["verify"].get("forbidden", []):
|
|
if fnmatch.fnmatch(rel, pattern):
|
|
failures.append(f"forbidden path exported: {rel}")
|
|
return failures
|
|
|
|
|
|
def verify_export(output: Path, rules: dict | None = None) -> None:
|
|
rules = rules or load_rules()
|
|
failures = verify_required(output, rules) + verify_forbidden(output, rules)
|
|
if failures:
|
|
for failure in failures:
|
|
print(f"FAIL {failure}", file=sys.stderr)
|
|
raise SystemExit(1)
|
|
|
|
|
|
def count_files(output: Path) -> int:
|
|
return sum(1 for path in output.rglob("*") if path.is_file())
|
|
|
|
|
|
def sha256(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(65536), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def load_state(target_root: Path) -> dict:
|
|
state_path = target_root / STATE_FILE
|
|
if not state_path.is_file():
|
|
return {
|
|
"schema_version": 1,
|
|
"bundle_id": None,
|
|
"bundle_version": None,
|
|
"managed_files": {},
|
|
}
|
|
with state_path.open("r", encoding="utf-8") as handle:
|
|
return json.load(handle)
|
|
|
|
|
|
def save_state(target_root: Path, state: dict) -> None:
|
|
state_path = target_root / STATE_FILE
|
|
state_path.write_text(json.dumps(state, indent=2) + "\n", encoding="utf-8")
|
|
|
|
|
|
def build_bundle(temp_root: Path) -> Path:
|
|
return export_bundle(temp_root / "bootstrap" / "exports" / "generic")
|
|
|
|
|
|
def write_as_new_path(target: Path) -> Path:
|
|
suffix = "".join(target.suffixes)
|
|
stem = target.name[: -len(suffix)] if suffix else target.name
|
|
index = 1
|
|
while True:
|
|
if index == 1:
|
|
candidate_name = f"{stem}.bootstrap-new{suffix}"
|
|
else:
|
|
candidate_name = f"{stem}.bootstrap-new-{index}{suffix}"
|
|
candidate = target.with_name(candidate_name)
|
|
if not candidate.exists():
|
|
return candidate
|
|
index += 1
|
|
|
|
|
|
def copy_into_place(source: Path, target: Path) -> None:
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copyfile(source, target)
|
|
shutil.copymode(source, target)
|
|
|
|
|
|
def iter_bundle_files(bundle_root: Path) -> list[Path]:
|
|
return sorted(path for path in bundle_root.rglob("*") if path.is_file())
|
|
|
|
|
|
def collect_conflicts(bundle_root: Path, target_root: Path, state: dict) -> tuple[list[Path], list[Conflict], list[Path]]:
|
|
create_paths: list[Path] = []
|
|
conflicts: list[Conflict] = []
|
|
unchanged: list[Path] = []
|
|
managed_files = state.get("managed_files", {})
|
|
|
|
for source in iter_bundle_files(bundle_root):
|
|
rel = source.relative_to(bundle_root)
|
|
rel_text = rel.as_posix()
|
|
target = target_root / rel
|
|
source_hash = sha256(source)
|
|
managed = rel_text in managed_files
|
|
|
|
if not target.exists():
|
|
create_paths.append(source)
|
|
continue
|
|
|
|
if not target.is_file():
|
|
raise SystemExit(
|
|
f"refusing to replace non-file path non-destructively: {target}"
|
|
)
|
|
|
|
target_hash = sha256(target)
|
|
if target_hash == source_hash:
|
|
unchanged.append(source)
|
|
continue
|
|
|
|
conflicts.append(
|
|
Conflict(
|
|
rel_path=rel_text,
|
|
target=target,
|
|
managed=managed,
|
|
source_hash=source_hash,
|
|
target_hash=target_hash,
|
|
)
|
|
)
|
|
|
|
return create_paths, conflicts, unchanged
|
|
|
|
|
|
def summarize_conflicts(create_count: int, conflict_count: int, unchanged_count: int) -> str:
|
|
return (
|
|
f"Bootstrap apply summary: create {create_count}, "
|
|
f"conflicts {conflict_count}, unchanged {unchanged_count}"
|
|
)
|
|
|
|
|
|
def is_probably_text(path: Path) -> bool:
|
|
try:
|
|
data = path.read_bytes()
|
|
except OSError:
|
|
return False
|
|
if len(data) > TEXT_DIFF_MAX_BYTES:
|
|
return False
|
|
if b"\x00" in data:
|
|
return False
|
|
try:
|
|
data.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
return False
|
|
return True
|
|
|
|
|
|
def render_diff(source: Path, target: Path, rel_path: str) -> str | None:
|
|
if not (is_probably_text(source) and is_probably_text(target)):
|
|
return None
|
|
source_lines = source.read_text(encoding="utf-8").splitlines(keepends=True)
|
|
target_lines = target.read_text(encoding="utf-8").splitlines(keepends=True)
|
|
diff = difflib.unified_diff(
|
|
target_lines,
|
|
source_lines,
|
|
fromfile=f"{rel_path} (existing)",
|
|
tofile=f"{rel_path} (bootstrap)",
|
|
n=3,
|
|
)
|
|
text = "".join(diff)
|
|
return text or None
|
|
|
|
|
|
def prompt_apply_all(choice_label: str) -> bool:
|
|
prompt = f"Apply '{choice_label}' to all remaining conflicts? [y/N]: "
|
|
print(prompt, end="", file=sys.stderr, flush=True)
|
|
answer = sys.stdin.readline()
|
|
if not answer:
|
|
return False
|
|
return answer.strip().lower() in {"y", "yes"}
|
|
|
|
|
|
def choose_conflict_action(
|
|
source: Path,
|
|
conflict: Conflict,
|
|
apply_to_all: str | None,
|
|
) -> tuple[str, str | None]:
|
|
if apply_to_all is not None:
|
|
return apply_to_all, apply_to_all
|
|
|
|
status = "managed bootstrap file" if conflict.managed else "existing file"
|
|
while True:
|
|
print(
|
|
f"Conflict: {conflict.rel_path} ({status})",
|
|
file=sys.stderr,
|
|
)
|
|
print(
|
|
"Choose [k]eep, [r]eplace, [w]rite-new, or [d]iff: ",
|
|
end="",
|
|
file=sys.stderr,
|
|
flush=True,
|
|
)
|
|
answer = sys.stdin.readline()
|
|
if not answer:
|
|
raise SystemExit(
|
|
"interactive input is required for conflicting files; "
|
|
"rerun with --on-conflict keep|replace|write-new for unattended use"
|
|
)
|
|
choice = answer.strip().lower()
|
|
if choice in {"d", "diff"}:
|
|
text = render_diff(source, conflict.target, conflict.rel_path)
|
|
if text is None:
|
|
print("Diff unavailable for this file.", file=sys.stderr)
|
|
else:
|
|
print(text, file=sys.stderr, end="" if text.endswith("\n") else "\n")
|
|
continue
|
|
if choice in {"k", "keep"}:
|
|
action = "keep"
|
|
elif choice in {"r", "replace"}:
|
|
action = "replace"
|
|
elif choice in {"w", "write-new"}:
|
|
action = "write-new"
|
|
else:
|
|
print("Please enter k, r, w, or d.", file=sys.stderr)
|
|
continue
|
|
|
|
if prompt_apply_all(action):
|
|
return action, action
|
|
return action, None
|
|
|
|
|
|
def normalize_conflict_policy(policy: str | None) -> str | None:
|
|
if policy is None:
|
|
return None
|
|
if policy == "write-new":
|
|
return "write-new"
|
|
if policy in {"keep", "replace"}:
|
|
return policy
|
|
raise SystemExit(f"unsupported conflict policy: {policy}")
|
|
|
|
|
|
def apply_bundle(
|
|
bundle_root: Path,
|
|
target_root: Path,
|
|
conflict_policy: str | None = None,
|
|
) -> dict:
|
|
manifest = load_json(bundle_root / "bundle-manifest.json")
|
|
state = load_state(target_root)
|
|
managed_files = dict(state.get("managed_files", {}))
|
|
create_paths, conflicts, unchanged_paths = collect_conflicts(bundle_root, target_root, state)
|
|
created = 0
|
|
replaced = 0
|
|
kept = 0
|
|
renamed = 0
|
|
unchanged = len(unchanged_paths)
|
|
|
|
print(summarize_conflicts(len(create_paths), len(conflicts), len(unchanged_paths)))
|
|
|
|
if conflicts and conflict_policy is None and not sys.stdin.isatty():
|
|
raise SystemExit(
|
|
"conflicts detected in non-interactive mode; "
|
|
"rerun with --on-conflict keep|replace|write-new"
|
|
)
|
|
|
|
for source in create_paths:
|
|
rel_text = source.relative_to(bundle_root).as_posix()
|
|
target = target_root / rel_text
|
|
copy_into_place(source, target)
|
|
managed_files[rel_text] = {
|
|
"sha256": sha256(target),
|
|
"canonical_path": rel_text,
|
|
}
|
|
created += 1
|
|
|
|
apply_to_all = normalize_conflict_policy(conflict_policy)
|
|
for conflict in conflicts:
|
|
source = bundle_root / conflict.rel_path
|
|
action, apply_to_all = choose_conflict_action(source, conflict, apply_to_all)
|
|
if action == "keep":
|
|
kept += 1
|
|
continue
|
|
if action == "replace":
|
|
copy_into_place(source, conflict.target)
|
|
managed_files[conflict.rel_path] = {
|
|
"sha256": sha256(conflict.target),
|
|
"canonical_path": conflict.rel_path,
|
|
}
|
|
replaced += 1
|
|
continue
|
|
|
|
new_target = write_as_new_path(conflict.target)
|
|
copy_into_place(source, new_target)
|
|
managed_files[new_target.relative_to(target_root).as_posix()] = {
|
|
"sha256": sha256(new_target),
|
|
"canonical_path": conflict.rel_path,
|
|
}
|
|
renamed += 1
|
|
|
|
state.update(
|
|
{
|
|
"schema_version": 1,
|
|
"bundle_id": manifest["bundle"]["id"],
|
|
"bundle_version": manifest["bundle"]["version"],
|
|
"managed_files": managed_files,
|
|
}
|
|
)
|
|
save_state(target_root, state)
|
|
return {
|
|
"created": created,
|
|
"replaced": replaced,
|
|
"kept": kept,
|
|
"renamed": renamed,
|
|
"unchanged": unchanged,
|
|
"conflicts": len(conflicts),
|
|
"target_root": str(target_root),
|
|
}
|
|
|
|
|
|
def with_temp_bundle() -> Path:
|
|
temp_dir = Path(tempfile.mkdtemp(prefix="agent-os-bootstrap-init."))
|
|
return build_bundle(temp_dir)
|