feat: add GEWU installer and service contracts

scripts/install.py

@@ -0,0 +1,315 @@
+"""Install and remove the GEWU instruction pack without replacing user files."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import re
+import shutil
+import tempfile
+from pathlib import Path
+from typing import Any
+
+PACK_ROOT = Path(__file__).resolve().parents[1]
+PACK_DIR = Path(".agents/gewu")
+STATE_PATH = Path(".agents/gewu-install.json")
+MARKER = "<!-- GEWU routing: managed by agents/gewu/scripts/install.py -->"
+ROUTES = {
+    "AGENTS.md": (
+        "For knowledge-base research, ingestion, curation, or synthesis, follow "
+        "[the GEWU profile](.agents/gewu/AGENTS.md) and its applicable kb-* skill. "
+        "Use this routing only for those research tasks."
+    ),
+    "CLAUDE.md": (
+        "For knowledge-base research, ingestion, curation, or synthesis, follow "
+        "[the GEWU profile](.agents/gewu/AGENTS.md) and its applicable kb-* skill."
+    ),
+    "GEMINI.md": (
+        "For knowledge-base research, ingestion, curation, or synthesis, follow "
+        "[the GEWU profile](.agents/gewu/AGENTS.md) and its applicable kb-* skill."
+    ),
+}
+
+
+def digest(data: bytes) -> str:
+    return hashlib.sha256(data).hexdigest()
+
+
+def _safe_regular(path: Path) -> bool:
+    return not path.is_symlink() and (not path.exists() or path.is_file())
+
+
+def _validated_state(state: object) -> dict[str, Any]:
+    if not isinstance(state, dict) or set(state) != {
+        "format",
+        "mode",
+        "pack_path",
+        "files",
+        "routes",
+    }:
+        raise ValueError("unrecognized GEWU installation state")
+    if (
+        state.get("format") != "gewu-install-state@1"
+        or state.get("mode") not in {"integrated", "dedicated"}
+        or state.get("pack_path") != PACK_DIR.as_posix()
+    ):
+        raise ValueError("unrecognized GEWU installation state")
+    files = state.get("files")
+    if not isinstance(files, dict) or "manifest.json" not in files:
+        raise ValueError("GEWU installation file list is invalid")
+    for relative, expected in files.items():
+        if (
+            not isinstance(relative, str)
+            or not relative
+            or "\\" in relative
+            or relative.startswith("/")
+            or any(part in {"", ".", ".."} for part in relative.split("/"))
+            or not isinstance(expected, str)
+            or not re.fullmatch(r"[a-f0-9]{64}", expected)
+        ):
+            raise ValueError("GEWU installation file list is invalid")
+    routes = state.get("routes")
+    if not isinstance(routes, dict) or set(routes) != set(ROUTES):
+        raise ValueError("GEWU installation routes are invalid")
+    for name, route in routes.items():
+        if (
+            not isinstance(name, str)
+            or not isinstance(route, dict)
+            or set(route) != {"block_sha256"}
+            or not isinstance(route.get("block_sha256"), str)
+            or not re.fullmatch(r"[a-f0-9]{64}", route["block_sha256"])
+        ):
+            raise ValueError("GEWU installation routes are invalid")
+    return state
+
+
+def _routing_block(host_file: str, mode: str) -> bytes:
+    qualifier = (
+        "This workspace is dedicated to research; use GEWU for research tasks."
+        if mode == "dedicated"
+        else (
+            "Keep existing development instructions; use GEWU only for research tasks."
+        )
+    )
+    text = f"\n{MARKER}\n{qualifier} {ROUTES[host_file]}\n<!-- /GEWU routing -->\n"
+    return text.encode("utf-8")
+
+
+def _merge_route(existing: bytes, block: bytes) -> bytes:
+    if MARKER.encode() in existing:
+        raise ValueError(
+            "GEWU routing marker already exists; uninstall the recorded installation first"
+        )
+    return existing + block
+
+
+def _pack_files(source: Path) -> dict[str, Path]:
+    result: dict[str, Path] = {}
+    for path in source.rglob("*"):
+        relative = path.relative_to(source)
+        if any(part in {"__pycache__", ".git"} for part in relative.parts):
+            continue
+        if path.is_symlink():
+            raise ValueError(f"pack contains a symlink: {relative}")
+        if path.is_file():
+            result[relative.as_posix()] = path
+    if "manifest.json" not in result:
+        raise ValueError("source is not a GEWU pack (manifest.json missing)")
+    return result
+
+
+def _atomic_write(path: Path, content: bytes) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
+    try:
+        with os.fdopen(fd, "wb") as stream:
+            stream.write(content)
+            stream.flush()
+            os.fsync(stream.fileno())
+        os.replace(temporary, path)
+    finally:
+        if os.path.exists(temporary):
+            os.unlink(temporary)
+
+
+def install(workspace: Path, source: Path = PACK_ROOT, mode: str = "integrated") -> dict[str, Any]:
+    """Copy a pack and append reversible instruction pointers."""
+    workspace = workspace.resolve()
+    if mode not in {"dedicated", "integrated"}:
+        raise ValueError("mode must be dedicated or integrated")
+    destination = workspace / PACK_DIR
+    state_file = workspace / STATE_PATH
+    agents_dir = workspace / ".agents"
+    if agents_dir.is_symlink():
+        raise ValueError(".agents must not be a symlink")
+    if (
+        destination.exists()
+        or destination.is_symlink()
+        or state_file.exists()
+        or state_file.is_symlink()
+    ):
+        raise FileExistsError("GEWU is already installed or the managed destination is occupied")
+    source_files = _pack_files(source.resolve())
+    targets = {name: workspace / name for name in ROUTES}
+    for target in targets.values():
+        if not _safe_regular(target):
+            raise ValueError(f"routing target is not a regular file: {target.name}")
+
+    # Prepare all bytes and validate the destinations before mutating the workspace.
+    copied = {name: path.read_bytes() for name, path in source_files.items()}
+    route_originals = {
+        name: target.read_bytes() if target.exists() else None for name, target in targets.items()
+    }
+    route_blocks = {name: _routing_block(name, mode) for name in ROUTES}
+    route_results = {
+        name: _merge_route(route_originals[name] or b"", route_blocks[name]) for name in ROUTES
+    }
+    state: dict[str, Any] = {
+        "format": "gewu-install-state@1",
+        "mode": mode,
+        "pack_path": PACK_DIR.as_posix(),
+        "files": {name: digest(content) for name, content in copied.items()},
+        "routes": {
+            name: {
+                "block_sha256": digest(route_blocks[name]),
+            }
+            for name in ROUTES
+        },
+    }
+    state_bytes = (json.dumps(state, sort_keys=True, indent=2) + "\n").encode()
+
+    created_destination = False
+    try:
+        destination.mkdir(parents=True)
+        created_destination = True
+        for name, content in copied.items():
+            target = destination / name
+            target.parent.mkdir(parents=True, exist_ok=True)
+            _atomic_write(target, content)
+        for name, content in route_results.items():
+            _atomic_write(targets[name], content)
+        _atomic_write(state_file, state_bytes)
+    except BaseException:
+        # A racing installer that failed to create the destination owns nothing.
+        if created_destination:
+            for name, original in route_originals.items():
+                target = targets[name]
+                if original is None:
+                    target.unlink(missing_ok=True)
+                else:
+                    _atomic_write(target, original)
+            shutil.rmtree(destination, ignore_errors=True)
+            state_file.unlink(missing_ok=True)
+        raise
+    return state
+
+
+def uninstall(workspace: Path) -> None:
+    """Remove unchanged GEWU-owned files and routing blocks; preserve edits."""
+    workspace = workspace.resolve()
+    agents_dir = workspace / ".agents"
+    if agents_dir.is_symlink():
+        raise ValueError(".agents must not be a symlink")
+    state_file = workspace / STATE_PATH
+    if not _safe_regular(state_file) or not state_file.is_file():
+        raise FileNotFoundError("GEWU installation state is missing or unsafe")
+    if state_file.stat().st_size > 2_000_000:
+        raise ValueError("GEWU installation state is too large")
+    state = _validated_state(json.loads(state_file.read_text(encoding="utf-8")))
+    conflicts: list[str] = []
+    destination = workspace / PACK_DIR
+    if destination.is_symlink() or not destination.is_dir():
+        raise ValueError("GEWU pack destination is missing or unsafe")
+    for relative, expected in state.get("files", {}).items():
+        target = destination / relative
+        parent = target.parent
+        unsafe_parent = False
+        while parent != destination.parent:
+            if parent.is_symlink() or not parent.is_dir():
+                unsafe_parent = True
+                break
+            parent = parent.parent
+        if (
+            unsafe_parent
+            or not target.is_file()
+            or target.is_symlink()
+            or digest(target.read_bytes()) != expected
+        ):
+            conflicts.append(f"edited or missing pack file: {PACK_DIR / relative}")
+    route_data: dict[str, bytes] = {}
+    for name, _route in state.get("routes", {}).items():
+        target = workspace / name
+        if not _safe_regular(target):
+            conflicts.append(f"unsafe routing file: {name}")
+            continue
+        current = target.read_bytes() if target.exists() else b""
+        block = _routing_block(name, state["mode"])
+        if not target.exists() or not current.endswith(block):
+            conflicts.append(f"edited routing block: {name}")
+        route_data[name] = current
+    if conflicts:
+        raise ValueError("uninstall refused to remove changed content: " + "; ".join(conflicts))
+
+    # Verify ownership of the exact suffix before removing it.
+    for name, route in state["routes"].items():
+        block = _routing_block(name, state["mode"])
+        current = route_data[name]
+        if digest(block) != route["block_sha256"] or not current.endswith(block):
+            raise ValueError(f"routing ownership check failed: {name}")
+    for name, _route in state["routes"].items():
+        target = workspace / name
+        current = route_data[name]
+        original = current[: -len(_routing_block(name, state["mode"]))]
+        if original:
+            _atomic_write(target, original)
+        else:
+            target.unlink()
+    for relative in sorted(state["files"], key=lambda item: len(Path(item).parts), reverse=True):
+        (destination / relative).unlink()
+    for directory in sorted(
+        (path for path in destination.rglob("*") if path.is_dir() and not path.is_symlink()),
+        key=lambda path: len(path.parts),
+        reverse=True,
+    ):
+        try:
+            directory.rmdir()
+        except OSError:
+            pass
+    try:
+        destination.rmdir()
+    except OSError:
+        pass
+    state_file.unlink()
+    for parent in (workspace / ".agents",):
+        try:
+            parent.rmdir()
+        except OSError:
+            pass
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    commands = parser.add_subparsers(dest="command", required=True)
+    add = commands.add_parser("install")
+    add.add_argument("workspace", type=Path)
+    add.add_argument("--source", type=Path, default=PACK_ROOT)
+    add.add_argument("--mode", choices=("dedicated", "integrated"), default="integrated")
+    remove = commands.add_parser("uninstall")
+    remove.add_argument("workspace", type=Path)
+    arguments = parser.parse_args()
+    try:
+        if arguments.command == "install":
+            install(arguments.workspace, arguments.source, arguments.mode)
+            print(f"GEWU pack and reversible routing installed in {arguments.workspace.resolve()}")
+        else:
+            uninstall(arguments.workspace)
+            print(f"GEWU-owned pack and routing removed from {arguments.workspace.resolve()}")
+    except (OSError, ValueError, json.JSONDecodeError) as error:
+        parser.error(str(error))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/manifest_tool.py

@@ -0,0 +1,115 @@
+"""Verify or refresh the portable GEWU instruction-pack integrity manifest."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import sys
+from pathlib import Path
+from typing import Any
+
+ROOT = Path(__file__).resolve().parents[1]
+MANIFEST = ROOT / "manifest.json"
+
+
+def normalized_bytes(path: Path) -> bytes:
+    text = path.read_text(encoding="utf-8-sig")
+    return text.replace("\r\n", "\n").replace("\r", "\n").encode("utf-8")
+
+
+def sha256(path: Path) -> str:
+    return hashlib.sha256(normalized_bytes(path)).hexdigest()
+
+
+def snapshot_digest(modules: list[dict[str, Any]]) -> str:
+    compact = json.dumps(modules, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+    return hashlib.sha256(compact.encode("utf-8")).hexdigest()
+
+
+def load_manifest() -> dict[str, Any]:
+    manifest = json.loads(MANIFEST.read_text(encoding="utf-8"))
+    if not isinstance(manifest, dict) or not isinstance(manifest.get("modules"), list):
+        raise ValueError("manifest.json has an invalid structure")
+    return manifest
+
+
+def module_path(entry: str) -> Path:
+    path = (ROOT / entry).resolve()
+    if path != ROOT and ROOT not in path.parents:
+        raise ValueError(f"module path escapes pack: {entry}")
+    if not path.is_file():
+        raise ValueError(f"module file is missing: {entry}")
+    return path
+
+
+def verify(manifest: dict[str, Any]) -> None:
+    modules = manifest["modules"]
+    for module in modules:
+        if not isinstance(module, dict) or not isinstance(module.get("entry"), str):
+            raise ValueError("manifest module descriptor is invalid")
+        if module.get("sha256") != sha256(module_path(module["entry"])):
+            raise ValueError(f"module hash mismatch: {module.get('module_id')}")
+    expected_snapshot = manifest.get("snapshot", {}).get("digest")
+    if expected_snapshot != snapshot_digest(modules):
+        raise ValueError("module snapshot digest mismatch")
+    for contract in manifest.get("verified_service_contracts", []):
+        if not isinstance(contract, dict) or not isinstance(contract.get("path"), str):
+            raise ValueError("service contract descriptor is invalid")
+        contract_path = module_path(contract["path"])
+        if contract.get("sha256") != sha256(contract_path):
+            raise ValueError(f"service contract hash mismatch: {contract.get('contract_id')}")
+        service_contract = json.loads(contract_path.read_text(encoding="utf-8"))
+        catalog_path = module_path(str(contract_path.parent / "catalog-v1.json"))
+        catalog = json.loads(catalog_path.read_text(encoding="utf-8"))
+        if service_contract.get("catalog_sha256") != sha256(catalog_path):
+            raise ValueError("bundled service catalog hash mismatch")
+        catalog_capabilities = {
+            operation["capability"]
+            for operation in catalog.get("operations", [])
+            if "mcp" in operation.get("transports", [])
+        }
+        tool_names = {tool.get("name") for tool in service_contract.get("tools", [])}
+        catalog_tool_names = {
+            operation["name"]
+            for operation in catalog.get("operations", [])
+            if "mcp" in operation.get("transports", [])
+        }
+        if tool_names != catalog_tool_names:
+            raise ValueError("MCP tool snapshot differs from the bundled catalog")
+        for module in modules:
+            required = module.get("required_service_capabilities", [])
+            if not isinstance(required, list) or not set(required).issubset(catalog_capabilities):
+                raise ValueError(
+                    f"service capability is not in the pinned catalog: {module.get('module_id')}"
+                )
+        for schema in service_contract.get("record_schemas", []):
+            schema_path = module_path(str(contract_path.parent / schema["path"]))
+            if schema.get("sha256") != sha256(schema_path):
+                raise ValueError(f"bundled record schema hash mismatch: {schema.get('path')}")
+
+
+def refresh(manifest: dict[str, Any]) -> None:
+    modules = manifest["modules"]
+    for module in modules:
+        if not isinstance(module, dict) or not isinstance(module.get("entry"), str):
+            raise ValueError("manifest module descriptor is invalid")
+        module["sha256"] = sha256(module_path(module["entry"]))
+        module["hash_encoding"] = "utf8-no-bom-lf"
+    manifest.setdefault("snapshot", {})["digest"] = snapshot_digest(modules)
+    for contract in manifest.get("verified_service_contracts", []):
+        contract["sha256"] = sha256(module_path(contract["path"]))
+    MANIFEST.write_text(json.dumps(manifest, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
+
+
+def main() -> None:
+    if len(sys.argv) != 2 or sys.argv[1] not in {"refresh", "verify"}:
+        raise SystemExit("usage: python scripts/manifest_tool.py refresh|verify")
+    manifest = load_manifest()
+    if sys.argv[1] == "refresh":
+        refresh(manifest)
+        manifest = load_manifest()
+    verify(manifest)
+
+
+if __name__ == "__main__":
+    main()

scripts/run_checkpoint.py

@@ -0,0 +1,429 @@
+"""Private, atomic checkpoints for host-executed GEWU research runs."""
+
+from __future__ import annotations
+
+import argparse
+import importlib
+import json
+import math
+import os
+import re
+import secrets
+import stat
+import sys
+from datetime import UTC, datetime
+from pathlib import Path
+from types import ModuleType
+from typing import Any
+
+if __package__:
+    manifest_module: ModuleType = importlib.import_module(".manifest_tool", __package__)
+else:
+    sys.path.insert(0, str(Path(__file__).resolve().parent))
+    manifest_module = importlib.import_module("manifest_tool")
+
+load_manifest = manifest_module.load_manifest
+verify = manifest_module.verify
+
+_RUN_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{1,63}$")
+_SHA256 = re.compile(r"^[a-f0-9]{64}$")
+_MUTABLE = {
+    "stage",
+    "sources",
+    "coverage",
+    "budgets",
+    "pending_operations",
+    "proposals",
+    "outputs",
+    "last_failure",
+    "runtime",
+}
+_FORBIDDEN_KEYS = {
+    "token",
+    "secret",
+    "password",
+    "credential",
+    "source_body",
+    "content",
+    "cookie",
+    "authorization",
+    "auth_header",
+}
+
+
+def _reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
+    result: dict[str, Any] = {}
+    for key, value in pairs:
+        if key in result:
+            raise ValueError("checkpoint JSON contains a duplicate key")
+        result[key] = value
+    return result
+
+
+def _safe_payload(value: Any) -> None:
+    if isinstance(value, dict):
+        for key, item in value.items():
+            normalized = str(key).lower().replace("-", "_")
+            if (
+                normalized in _FORBIDDEN_KEYS
+                or any(
+                    part in normalized
+                    for part in ("secret", "password", "credential", "bearer", "api_key")
+                )
+                or normalized.endswith("_token")
+            ):
+                raise ValueError("checkpoint cannot store credentials or source bodies")
+            _safe_payload(item)
+    elif isinstance(value, list):
+        for item in value:
+            _safe_payload(item)
+    elif isinstance(value, float) and not math.isfinite(value):
+        raise ValueError("checkpoint contains a non-finite number")
+    elif not isinstance(value, (str, int, float, bool, type(None))):
+        raise ValueError("checkpoint contains a non-JSON value")
+
+
+def _runs_directory(workspace: Path) -> Path:
+    root = workspace.expanduser().absolute()
+    if root.is_symlink():
+        raise ValueError("workspace root must not be a symbolic link")
+    runs = root / "Runs"
+    if runs.is_symlink():
+        raise ValueError("Runs directory must not be a symbolic link")
+    runs.mkdir(parents=True, exist_ok=True, mode=0o700)
+    if not runs.is_dir() or runs.is_symlink():
+        raise ValueError("Runs path is unavailable")
+    if os.name != "nt":
+        runs.chmod(stat.S_IRWXU)
+    return runs
+
+
+def _checkpoint_path(workspace: Path, run_id: str) -> Path:
+    if not _RUN_ID.fullmatch(run_id):
+        raise ValueError("run ID is invalid")
+    return _runs_directory(workspace) / f"{run_id}.json"
+
+
+def _write_atomic(path: Path, value: dict[str, Any]) -> None:
+    _safe_payload(value)
+    payload = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+    if len(payload.encode("utf-8")) > 1_000_000:
+        raise ValueError("checkpoint exceeds the one-megabyte limit")
+    temporary = path.with_name(f".{path.name}.{secrets.token_hex(8)}.tmp")
+    try:
+        descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+        with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as stream:
+            stream.write(payload + "\n")
+            stream.flush()
+            os.fsync(stream.fileno())
+        os.replace(temporary, path)
+        if os.name != "nt":
+            path.chmod(stat.S_IRUSR | stat.S_IWUSR)
+            directory_fd = os.open(path.parent, os.O_RDONLY)
+            try:
+                os.fsync(directory_fd)
+            finally:
+                os.close(directory_fd)
+    finally:
+        temporary.unlink(missing_ok=True)
+
+
+def _module_pins(manifest: dict[str, Any]) -> list[dict[str, str]]:
+    return [
+        {
+            "module_id": module["module_id"],
+            "entry": module["entry"],
+            "contract_id": module["provided_contract"]["id"],
+            "contract_version": module["provided_contract"]["version"],
+            "implementation_version": module["implementation_version"],
+            "sha256": module["sha256"],
+        }
+        for module in manifest["modules"]
+    ]
+
+
+def create_checkpoint(
+    workspace: Path,
+    *,
+    run_id: str,
+    objective: str,
+    projects: list[str],
+    include_shared: bool,
+    lens: str,
+    budgets: dict[str, int] | None = None,
+) -> dict[str, Any]:
+    manifest = load_manifest()
+    verify(manifest)
+    if not objective.strip() or len(objective) > 4_000:
+        raise ValueError("objective must contain 1 to 4000 characters")
+    if (
+        not projects
+        or len(projects) > 100
+        or len(set(projects)) != len(projects)
+        or any(
+            not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,127}", project) for project in projects
+        )
+    ):
+        raise ValueError("scope must contain distinct explicit project IDs")
+    if not lens.strip() or len(lens) > 2_000:
+        raise ValueError("lens must contain 1 to 2000 characters")
+    path = _checkpoint_path(workspace, run_id)
+    if path.is_symlink() or path.exists():
+        raise ValueError("run checkpoint already exists")
+    now = datetime.now(UTC).isoformat()
+    checkpoint: dict[str, Any] = {
+        "checkpoint_version": "0.1.0",
+        "run_id": run_id,
+        "created_at": now,
+        "updated_at": now,
+        "pack": {
+            "pack_id": manifest["pack_id"],
+            "pack_version": manifest["pack_version"],
+            "snapshot_digest": manifest["snapshot"]["digest"],
+            "modules": _module_pins(manifest),
+            "service_contracts": [
+                {"contract_id": item["contract_id"], "sha256": item["sha256"]}
+                for item in manifest["verified_service_contracts"]
+            ],
+        },
+        "objective": objective,
+        "scope": {"projects": projects, "include_shared": include_shared},
+        "lens": lens,
+        "stage": "context",
+        "sources": [],
+        "coverage": {"selected": [], "included": [], "excluded": [], "skipped": []},
+        "budgets": budgets or {},
+        "runtime": {"host": None, "model": None},
+        "pending_operations": [],
+        "proposals": [],
+        "outputs": [],
+        "last_failure": None,
+    }
+    _validate_checkpoint(checkpoint)
+    _write_atomic(path, checkpoint)
+    return checkpoint
+
+
+def load_checkpoint(workspace: Path, run_id: str) -> dict[str, Any]:
+    manifest = load_manifest()
+    verify(manifest)
+    path = _checkpoint_path(workspace, run_id)
+    if path.is_symlink() or not path.is_file() or path.stat().st_size > 1_000_000:
+        raise ValueError("run checkpoint is unavailable")
+    checkpoint = json.loads(
+        path.read_text(encoding="utf-8"), object_pairs_hook=_reject_duplicate_keys
+    )
+    if not isinstance(checkpoint, dict) or checkpoint.get("run_id") != run_id:
+        raise ValueError("run checkpoint identity is invalid")
+    pinned_pack = checkpoint.get("pack")
+    if not isinstance(pinned_pack, dict) or (
+        pinned_pack.get("snapshot_digest") != manifest["snapshot"]["digest"]
+        or pinned_pack.get("modules") != _module_pins(manifest)
+    ):
+        raise ValueError("pack changed since this run; revalidate before resuming")
+    _safe_payload(checkpoint)
+    _validate_checkpoint(checkpoint)
+    return checkpoint
+
+
+def update_checkpoint(workspace: Path, run_id: str, patch: dict[str, Any]) -> dict[str, Any]:
+    checkpoint = load_checkpoint(workspace, run_id)
+    if not patch or not set(patch).issubset(_MUTABLE):
+        raise ValueError("checkpoint update contains unsupported fields")
+    _safe_payload(patch)
+    updated = {**checkpoint, **patch, "updated_at": datetime.now(UTC).isoformat()}
+    _validate_checkpoint(updated)
+    _write_atomic(_checkpoint_path(workspace, run_id), updated)
+    return updated
+
+
+def _validate_checkpoint(checkpoint: dict[str, Any]) -> None:
+    required = {
+        "checkpoint_version",
+        "run_id",
+        "created_at",
+        "updated_at",
+        "pack",
+        "objective",
+        "scope",
+        "lens",
+        "stage",
+        "sources",
+        "coverage",
+        "budgets",
+        "runtime",
+        "pending_operations",
+        "proposals",
+        "outputs",
+        "last_failure",
+    }
+    if set(checkpoint) != required or checkpoint.get("checkpoint_version") != "0.1.0":
+        raise ValueError("checkpoint fields or version are invalid")
+    if not isinstance(checkpoint["objective"], str) or not checkpoint["objective"].strip():
+        raise ValueError("checkpoint objective is required")
+    if not isinstance(checkpoint["lens"], str) or not checkpoint["lens"].strip():
+        raise ValueError("checkpoint lens is required")
+    scope = checkpoint["scope"]
+    if (
+        not isinstance(scope, dict)
+        or set(scope) != {"projects", "include_shared"}
+        or not isinstance(scope["projects"], list)
+        or not scope["projects"]
+        or len(scope["projects"]) > 100
+        or any(
+            not isinstance(project, str)
+            or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,127}", project)
+            for project in scope["projects"]
+        )
+        or len(set(scope["projects"])) != len(scope["projects"])
+        or type(scope["include_shared"]) is not bool
+    ):
+        raise ValueError("checkpoint scope is invalid")
+    for field in ("sources", "pending_operations", "proposals", "outputs"):
+        if not isinstance(checkpoint[field], list) or len(checkpoint[field]) > 10_000:
+            raise ValueError(f"{field} must be a bounded list")
+    for source in checkpoint["sources"]:
+        if (
+            not isinstance(source, dict)
+            or not isinstance(source.get("source_revision_id"), str)
+            or not isinstance(source.get("status"), str)
+        ):
+            raise ValueError("source records require a revision ID and capture status")
+        extraction_id = source.get("extraction_id")
+        if extraction_id is not None and not _SHA256.fullmatch(str(extraction_id)):
+            raise ValueError("source extraction references must use a SHA-256 ID")
+    if not isinstance(checkpoint["coverage"], dict) or set(checkpoint["coverage"]) != {
+        "selected",
+        "included",
+        "excluded",
+        "skipped",
+    }:
+        raise ValueError("coverage must record selected, included, excluded and skipped items")
+    for key, items in checkpoint["coverage"].items():
+        if not isinstance(items, list) or len(items) > 10_000:
+            raise ValueError("coverage categories must be bounded lists")
+        if key in {"selected", "included"} and any(not isinstance(item, str) for item in items):
+            raise ValueError("selected and included coverage entries must be strings")
+        if key in {"excluded", "skipped"} and any(
+            not isinstance(item, dict)
+            or not isinstance(item.get("item"), str)
+            or not isinstance(item.get("reason"), str)
+            or not item["reason"].strip()
+            for item in items
+        ):
+            raise ValueError("excluded and skipped coverage items require a reason")
+    if not isinstance(checkpoint["budgets"], dict) or len(checkpoint["budgets"]) > 100:
+        raise ValueError("budgets must be a bounded object")
+    if any(type(value) is not int or value < 0 for value in checkpoint["budgets"].values()):
+        raise ValueError("budget values must be non-negative integers")
+    if (
+        not isinstance(checkpoint["runtime"], dict)
+        or set(checkpoint["runtime"]) != {"host", "model"}
+        or any(
+            value is not None and not isinstance(value, str)
+            for value in checkpoint["runtime"].values()
+        )
+    ):
+        raise ValueError(
+            "runtime metadata must identify the host and model or explicitly omit them"
+        )
+    if not isinstance(checkpoint["stage"], str) or not checkpoint["stage"]:
+        raise ValueError("stage must be explicit")
+    for operation in checkpoint["pending_operations"]:
+        if (
+            not isinstance(operation, dict)
+            or not isinstance(operation.get("operation_id"), str)
+            or not isinstance(operation.get("tool_name"), str)
+            or not _SHA256.fullmatch(str(operation.get("input_sha256", "")))
+            or not isinstance(operation.get("status"), str)
+            or operation["status"] not in {"pending", "uncertain", "complete", "failed"}
+        ):
+            raise ValueError("pending operation identity, input hash and status are required")
+    for proposal in checkpoint["proposals"]:
+        if (
+            not isinstance(proposal, dict)
+            or not isinstance(proposal.get("proposal_id"), str)
+            or type(proposal.get("revision")) is not int
+            or proposal["revision"] < 1
+            or not isinstance(proposal.get("status"), str)
+            or proposal["status"]
+            not in {
+                "staged",
+                "evaluated",
+                "eligible",
+                "needs-review",
+                "invalid",
+                "conflict",
+                "publishing",
+                "accepted",
+                "rejected",
+                "withdrawn",
+                "uncertain",
+            }
+            or (
+                proposal.get("status") == "accepted"
+                and not isinstance(proposal.get("receipt"), dict)
+            )
+        ):
+            raise ValueError(
+                "proposal records require an ID, revision, status and accepted receipt"
+            )
+        if proposal["status"] == "accepted":
+            receipt = proposal["receipt"]
+            if (
+                receipt.get("proposal_id") != proposal["proposal_id"]
+                or receipt.get("revision") != proposal["revision"]
+                or receipt.get("state") != "accepted"
+                or not re.fullmatch(
+                    r"(?:[a-f0-9]{40}|[a-f0-9]{64})", str(receipt.get("commit_id", ""))
+                )
+            ):
+                raise ValueError("accepted proposal receipt does not match its revision")
+    for output in checkpoint["outputs"]:
+        if not isinstance(output, dict) or not _SHA256.fullmatch(str(output.get("sha256", ""))):
+            raise ValueError("output records require a SHA-256")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    subparsers = parser.add_subparsers(dest="command", required=True)
+    init = subparsers.add_parser("init")
+    init.add_argument("--workspace", type=Path, default=Path.cwd())
+    init.add_argument("--run-id", required=True)
+    init.add_argument("--objective", required=True)
+    init.add_argument("--projects", required=True)
+    init.add_argument("--include-shared", action="store_true")
+    init.add_argument("--lens", required=True)
+    init.add_argument("--budgets-json", default="{}")
+    update = subparsers.add_parser("update")
+    update.add_argument("--workspace", type=Path, default=Path.cwd())
+    update.add_argument("--run-id", required=True)
+    show = subparsers.add_parser("show")
+    show.add_argument("--workspace", type=Path, default=Path.cwd())
+    show.add_argument("--run-id", required=True)
+    arguments = parser.parse_args()
+    if arguments.command == "init":
+        checkpoint = create_checkpoint(
+            arguments.workspace,
+            run_id=arguments.run_id,
+            objective=arguments.objective,
+            projects=[item for item in arguments.projects.split(",") if item],
+            include_shared=arguments.include_shared,
+            lens=arguments.lens,
+            budgets=json.loads(arguments.budgets_json),
+        )
+    elif arguments.command == "show":
+        checkpoint = load_checkpoint(arguments.workspace, arguments.run_id)
+    else:
+        patch_text = sys.stdin.read(64_001)
+        if len(patch_text.encode("utf-8")) > 64_000:
+            raise SystemExit("checkpoint update exceeds 64 KiB")
+        patch = json.loads(patch_text, object_pairs_hook=_reject_duplicate_keys)
+        if not isinstance(patch, dict):
+            raise SystemExit("checkpoint update must be one JSON object on stdin")
+        checkpoint = update_checkpoint(arguments.workspace, arguments.run_id, patch)
+    print(json.dumps(checkpoint, ensure_ascii=False, sort_keys=True))
+
+
+if __name__ == "__main__":
+    main()