zemion
2ffdb23f69
Verified with the coordinated workspace changes by devkit full run 2026-09-08T225814-186389-0000-3e3ed7cd (all seven phases passed). This shared UI pass does not mark the individual module reviews complete.
1033 lines
39 KiB
Python
Executable File
1033 lines
39 KiB
Python
Executable File
"""Explicit-path Git maintenance with frozen plans and interruption receipts.
|
||
|
||
This is deliberately narrower than a Git frontend: no staging all files,
|
||
amending, force-pushing, hook bypasses, filter execution or automatic rollback.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from contextlib import contextmanager
|
||
import hashlib
|
||
import os
|
||
from pathlib import Path
|
||
import re
|
||
import shutil
|
||
import stat
|
||
import tempfile
|
||
import uuid
|
||
|
||
from .common import (
|
||
atomic_json,
|
||
digest,
|
||
identifier,
|
||
now,
|
||
read_json,
|
||
resource_lock,
|
||
state_root,
|
||
)
|
||
from .process import require_capture
|
||
from .workspace import load_project, selected_repositories
|
||
|
||
MAX_FILE_BYTES = 64 * 1024 * 1024
|
||
MAX_GIT_OUTPUT_BYTES = 32 * 1024 * 1024
|
||
|
||
|
||
def _git(
|
||
repo: Path,
|
||
*argv: str,
|
||
allowed: tuple[int, ...] = (0,),
|
||
timeout: int = 30,
|
||
input_bytes: bytes | None = None,
|
||
index_file: Path | None = None,
|
||
) -> bytes:
|
||
# Namespace/common-dir overrides can redirect both push and its verification
|
||
# to the same unintended ref. Unknown overrides are refused, not scrubbed.
|
||
# GIT_PAGER is inert because every command explicitly passes --no-pager.
|
||
safe_git_environment = {"GIT_OPTIONAL_LOCKS", "GIT_TERMINAL_PROMPT", "GIT_PAGER"}
|
||
unsafe = {
|
||
key
|
||
for key in os.environ
|
||
if key.startswith("GIT_") and key not in safe_git_environment
|
||
}
|
||
if unsafe:
|
||
raise ValueError(
|
||
"Git environment overrides are unsupported for frozen maintenance plans"
|
||
)
|
||
env = {
|
||
**os.environ,
|
||
"GIT_TERMINAL_PROMPT": "0",
|
||
"GIT_OPTIONAL_LOCKS": "0",
|
||
"GIT_LITERAL_PATHSPECS": "1",
|
||
}
|
||
if index_file is not None:
|
||
env["GIT_INDEX_FILE"] = str(index_file)
|
||
if input_bytes is not None and len(input_bytes) > MAX_FILE_BYTES:
|
||
raise ValueError("Git input exceeds maintenance bounds")
|
||
result = require_capture(
|
||
["git", "--no-pager", "-C", str(repo), *argv],
|
||
env=env,
|
||
timeout=timeout,
|
||
max_stdout=MAX_GIT_OUTPUT_BYTES,
|
||
max_stderr=1024 * 1024,
|
||
input_bytes=input_bytes,
|
||
)
|
||
if result.returncode not in allowed:
|
||
# Git stderr may contain credentials, hook output or source data.
|
||
raise ValueError(
|
||
f"Git {argv[0]} failed (exit {result.returncode}); inspect the repository locally"
|
||
)
|
||
return result.stdout
|
||
|
||
|
||
def _config(repo: Path) -> tuple[dict[str, list[str]], str]:
|
||
raw = _git(repo, "config", "--null", "--list")
|
||
values: dict[str, list[str]] = {}
|
||
for item in raw.split(b"\0"):
|
||
if not item:
|
||
continue
|
||
key, _, value = item.partition(b"\n")
|
||
values.setdefault(os.fsdecode(key).lower(), []).append(os.fsdecode(value))
|
||
for key, entries in values.items():
|
||
if key in {
|
||
"core.hookspath",
|
||
"core.sshcommand",
|
||
"core.gitproxy",
|
||
"core.alternaterefscommand",
|
||
"extensions.partialclone",
|
||
"push.pushoption",
|
||
} or re.fullmatch(r"remote\..*\.(?:receivepack|uploadpack|vcs)", key):
|
||
raise ValueError(f"Unsupported external Git configuration: {key}")
|
||
if key in {
|
||
"core.fsmonitor",
|
||
"commit.gpgsign",
|
||
"push.gpgsign",
|
||
"remote.origin.mirror",
|
||
} and entries[-1].lower() not in {"false", "no", "off", "0"}:
|
||
raise ValueError(f"Unsupported Git maintenance configuration: {key}")
|
||
if (
|
||
re.fullmatch(r"remote\..*\.promisor", key)
|
||
or key in {"push.recursesubmodules", "submodule.recurse"}
|
||
) and entries[-1].lower() not in {"false", "no", "off", "0"}:
|
||
raise ValueError(
|
||
"Partial clone or recursive submodule operations require the normal Git workflow"
|
||
)
|
||
hooks_raw = os.fsdecode(_git(repo, "rev-parse", "--git-path", "hooks")).strip()
|
||
hooks = Path(hooks_raw) if Path(hooks_raw).is_absolute() else repo / hooks_raw
|
||
if hooks.is_symlink():
|
||
raise ValueError("Symlinked Git hooks are unsupported")
|
||
# Conservative discovery also covers post-index-change and future hook
|
||
# names; Git's shipped *.sample files are never invoked as hooks.
|
||
if hooks.exists() and any(
|
||
not entry.name.endswith(".sample") and os.access(entry, os.X_OK)
|
||
for entry in hooks.iterdir()
|
||
):
|
||
raise ValueError(
|
||
"Active Git hooks require the normal reviewed Git workflow; they are never bypassed"
|
||
)
|
||
return values, digest(raw.hex())
|
||
|
||
|
||
def _remote_urls(repo: Path) -> tuple[list[str], list[str]]:
|
||
names = os.fsdecode(_git(repo, "remote")).splitlines()
|
||
if "origin" not in names:
|
||
return [], []
|
||
fetch = os.fsdecode(_git(repo, "remote", "get-url", "--all", "origin")).splitlines()
|
||
push = os.fsdecode(
|
||
_git(repo, "remote", "get-url", "--push", "--all", "origin")
|
||
).splitlines()
|
||
return fetch, push
|
||
|
||
|
||
def _safe_remote(url: str) -> bool:
|
||
# Explicit standard Git transports only; never an ext:: or custom helper.
|
||
return (
|
||
bool(
|
||
re.match(r"^(?:https://|ssh://|file:///|/)", url)
|
||
or re.fullmatch(r"(?:[A-Za-z0-9_.-]+@)?[A-Za-z0-9_.-]+:[^\s]+", url)
|
||
)
|
||
and "::" not in url
|
||
and "\n" not in url
|
||
)
|
||
|
||
|
||
def _index(
|
||
repo: Path, *, index_file: Path | None = None
|
||
) -> tuple[dict[str, dict], str]:
|
||
raw = _git(repo, "ls-files", "--stage", "-z", index_file=index_file)
|
||
entries = {}
|
||
for entry in raw.split(b"\0"):
|
||
if not entry:
|
||
continue
|
||
metadata, _, encoded_path = entry.partition(b"\t")
|
||
mode, object_id, stage = metadata.decode("ascii").split()
|
||
if stage != "0":
|
||
raise ValueError(
|
||
"Unmerged index entries require manual conflict resolution"
|
||
)
|
||
entries[os.fsdecode(encoded_path)] = {"mode": mode, "blob": object_id}
|
||
return entries, hashlib.sha256(raw).hexdigest()
|
||
|
||
|
||
def _paths(
|
||
repo: Path,
|
||
paths: list[str],
|
||
config: dict[str, list[str]],
|
||
*,
|
||
allow_missing: bool = False,
|
||
) -> list[dict]:
|
||
if not paths or len(paths) > 256:
|
||
raise ValueError(
|
||
"Select 1–256 explicit file paths; directories and globs are not expanded"
|
||
)
|
||
entries, _ = _index(repo)
|
||
result = []
|
||
for name in sorted(set(paths)):
|
||
relative = Path(name)
|
||
if (
|
||
not isinstance(name, str)
|
||
or not name
|
||
or relative.is_absolute()
|
||
or name != relative.as_posix()
|
||
or any(part in {".", "..", ".git"} for part in name.split("/"))
|
||
or "\0" in name
|
||
or "\n" in name
|
||
):
|
||
raise ValueError("Maintenance paths must be explicit relative file paths")
|
||
target = repo / relative
|
||
if any(
|
||
part.is_symlink()
|
||
for part in (target, *target.parents)
|
||
if part != repo.parent
|
||
):
|
||
raise ValueError("Symlinked selected paths are unsupported")
|
||
if not target.resolve().is_relative_to(repo.resolve()):
|
||
raise ValueError("Selected path escapes its repository")
|
||
if entries.get(name, {}).get("mode") == "160000":
|
||
raise ValueError("Submodule paths require the normal reviewed Git workflow")
|
||
if not target.exists():
|
||
if (
|
||
not allow_missing
|
||
and name not in entries
|
||
and name not in _tree_entries(repo, "HEAD")
|
||
):
|
||
raise ValueError("Selected file is neither present nor tracked")
|
||
result.append(
|
||
{
|
||
"path": name,
|
||
"exists": False,
|
||
"sha256": None,
|
||
"mode": None,
|
||
"blob": None,
|
||
}
|
||
)
|
||
continue
|
||
data, metadata = _read_selected_bytes(target)
|
||
attributes = _git(
|
||
repo, "check-attr", "-z", "filter", "working-tree-encoding", "--", name
|
||
).split(b"\0")
|
||
if any(
|
||
value not in {b"", b"unspecified", b"unset"} for value in attributes[2::3]
|
||
):
|
||
raise ValueError(
|
||
"Selected filter/encoding attributes require the normal Git workflow"
|
||
)
|
||
blob = os.fsdecode(
|
||
_git(repo, "hash-object", "--path", name, "--stdin", input_bytes=data)
|
||
).strip()
|
||
mode = "100755" if metadata.st_mode & 0o111 else "100644"
|
||
if (
|
||
config.get("core.filemode", ["true"])[-1].lower() == "false"
|
||
and name in entries
|
||
):
|
||
mode = entries[name]["mode"]
|
||
result.append(
|
||
{
|
||
"path": name,
|
||
"exists": True,
|
||
"sha256": hashlib.sha256(data).hexdigest(),
|
||
"mode": mode,
|
||
"blob": blob,
|
||
}
|
||
)
|
||
return result
|
||
|
||
|
||
def snapshot(repo: Path, paths: list[str], *, allow_missing: bool = False) -> dict:
|
||
config, config_hash = _config(repo)
|
||
root = Path(
|
||
os.fsdecode(_git(repo, "rev-parse", "--show-toplevel")).strip()
|
||
).resolve()
|
||
if root != repo.resolve():
|
||
raise ValueError(
|
||
"Select the registered repository root, not an embedded directory"
|
||
)
|
||
if _git(repo, "rev-parse", "--shared-index-path").strip():
|
||
raise ValueError("Split indexes require the normal reviewed Git workflow")
|
||
head = os.fsdecode(_git(repo, "rev-parse", "--verify", "HEAD")).strip()
|
||
branch = os.fsdecode(
|
||
_git(repo, "symbolic-ref", "--quiet", "--short", "HEAD")
|
||
).strip()
|
||
|
||
def git_path(name):
|
||
return Path(os.fsdecode(_git(repo, "rev-parse", "--git-path", name)).strip())
|
||
|
||
for marker in (
|
||
"MERGE_HEAD",
|
||
"CHERRY_PICK_HEAD",
|
||
"REVERT_HEAD",
|
||
"rebase-merge",
|
||
"rebase-apply",
|
||
"BISECT_LOG",
|
||
):
|
||
location = git_path(marker)
|
||
if (location if location.is_absolute() else repo / location).exists():
|
||
raise ValueError(
|
||
"An in-progress Git operation requires the normal reviewed Git workflow"
|
||
)
|
||
grafts = git_path("info/grafts")
|
||
if (grafts if grafts.is_absolute() else repo / grafts).exists() or _git(
|
||
repo, "for-each-ref", "--format=%(refname)", "refs/replace/"
|
||
).strip():
|
||
raise ValueError(
|
||
"Git replacement/graft history requires the normal reviewed Git workflow"
|
||
)
|
||
flags = _git(repo, "ls-files", "-v", "-z")
|
||
if any(
|
||
item and (chr(item[0]).islower() or item[:1] == b"S")
|
||
for item in flags.split(b"\0")
|
||
):
|
||
raise ValueError(
|
||
"Assume-unchanged/skip-worktree entries are unsupported for frozen plans"
|
||
)
|
||
fetch, push = _remote_urls(repo)
|
||
index_entries, index_hash = _index(repo)
|
||
# A globally installed but unused LFS driver is harmless. Refuse active
|
||
# filters/encodings across tracked files too: Git may refresh their index
|
||
# metadata during commit, even when they are not selected.
|
||
attr_paths = sorted(set(index_entries) | set(paths))
|
||
attr_input = b"\0".join(os.fsencode(name) for name in attr_paths) + b"\0"
|
||
attrs = _git(
|
||
repo,
|
||
"check-attr",
|
||
"-z",
|
||
"--stdin",
|
||
"filter",
|
||
"working-tree-encoding",
|
||
input_bytes=attr_input,
|
||
).split(b"\0")
|
||
if any(value not in {b"", b"unspecified", b"unset"} for value in attrs[2::3]):
|
||
raise ValueError(
|
||
"Active Git filter/encoding attributes require the normal reviewed Git workflow"
|
||
)
|
||
return {
|
||
"head": head,
|
||
"branch": branch,
|
||
"origin_fetch_sha256": digest(fetch),
|
||
"origin_push_sha256": digest(push),
|
||
"origin_present": bool(fetch),
|
||
"config_sha256": config_hash,
|
||
"index_sha256": index_hash,
|
||
"unselected_index_sha256": digest(
|
||
{name: entry for name, entry in index_entries.items() if name not in paths}
|
||
),
|
||
"files": _paths(repo, paths, config, allow_missing=allow_missing),
|
||
}
|
||
|
||
|
||
def _receipt_path(args, plan_id: str) -> Path:
|
||
identifier(plan_id)
|
||
return (
|
||
state_root(Path(args.workspace_root), getattr(args, "state_dir", None))
|
||
/ "maintenance"
|
||
/ plan_id
|
||
)
|
||
|
||
|
||
def _seal(value: dict) -> dict:
|
||
return {
|
||
**value,
|
||
"integrity_sha256": digest(
|
||
{key: item for key, item in value.items() if key != "integrity_sha256"}
|
||
),
|
||
}
|
||
|
||
|
||
def _read(args, plan_id: str) -> tuple[dict, dict, Path]:
|
||
directory = _receipt_path(args, plan_id)
|
||
plan, state = (
|
||
read_json(directory / "plan.json"),
|
||
read_json(directory / "state.json"),
|
||
)
|
||
for value in (plan, state):
|
||
if (
|
||
not isinstance(value, dict)
|
||
or value.get("integrity_sha256") != _seal(value)["integrity_sha256"]
|
||
):
|
||
raise ValueError("Maintenance receipt integrity mismatch")
|
||
if value.get("plan_id") != plan_id:
|
||
raise ValueError("Maintenance receipt identity mismatch")
|
||
if (
|
||
plan.get("workspace_root") != str(Path(args.workspace_root).resolve())
|
||
or state.get("plan_sha256") != plan["integrity_sha256"]
|
||
):
|
||
raise ValueError("Maintenance receipt belongs to another workspace or plan")
|
||
project = load_project(Path(args.workspace_root), getattr(args, "project", None))
|
||
matched = selected_repositories(project, [plan["repository"]])
|
||
if len(matched) != 1 or str(matched[0].path) != plan["repository_path"]:
|
||
raise ValueError("The registered repository identity changed")
|
||
return plan, state, directory
|
||
|
||
|
||
def _write_state(directory: Path, state: dict, **updates) -> dict:
|
||
result = _seal({**state, **updates, "updated_at": now()})
|
||
atomic_json(directory / "state.json", result)
|
||
return result
|
||
|
||
|
||
def _result(plan: dict, state: dict, *lines: str) -> dict:
|
||
return {
|
||
"plan": plan,
|
||
"state": state,
|
||
"summary": [
|
||
f"Git maintenance {plan['plan_id']}: {state['status']}",
|
||
*lines,
|
||
"Only explicit selected paths are in scope; no hooks/filters bypass, force push, staging-all or automatic rollback.",
|
||
],
|
||
"_exit_code": 0,
|
||
}
|
||
|
||
|
||
@contextmanager
|
||
def _locked(args, repo: Path):
|
||
root = state_root(Path(args.workspace_root), getattr(args, "state_dir", None))
|
||
with resource_lock(root / "locks", "git-maintenance:" + str(repo.resolve())):
|
||
yield
|
||
|
||
|
||
def plan(args) -> dict:
|
||
project = load_project(Path(args.workspace_root), getattr(args, "project", None))
|
||
selected = selected_repositories(project, [args.repo])
|
||
if len(selected) != 1:
|
||
raise ValueError("Select exactly one repository")
|
||
repo = selected[0]
|
||
message = args.message
|
||
if (
|
||
not isinstance(message, str)
|
||
or not message.strip()
|
||
or "\0" in message
|
||
or len(message.encode()) > 16384
|
||
):
|
||
raise ValueError("A bounded, nonempty commit message is required")
|
||
frozen = snapshot(repo.path, args.path)
|
||
entries, _ = _index(repo.path)
|
||
# A partially staged selected file is ambiguous: users must first decide
|
||
# whether to commit its index or working-tree version through normal Git.
|
||
head_entries = _tree_entries(repo.path, frozen["head"])
|
||
for item in frozen["files"]:
|
||
current = entries.get(item["path"])
|
||
base = head_entries.get(item["path"])
|
||
desired = (
|
||
{"mode": item["mode"], "blob": item["blob"]} if item["exists"] else None
|
||
)
|
||
if current != base and current != desired:
|
||
raise ValueError(
|
||
"A selected file has different staged and working-tree changes; reconcile it explicitly first"
|
||
)
|
||
if all(
|
||
head_entries.get(item["path"])
|
||
== ({"mode": item["mode"], "blob": item["blob"]} if item["exists"] else None)
|
||
for item in frozen["files"]
|
||
):
|
||
raise ValueError("Selected paths contain no commit changes")
|
||
plan_id = "git-" + uuid.uuid4().hex
|
||
payload = _seal(
|
||
{
|
||
"schema_version": 1,
|
||
"plan_id": plan_id,
|
||
"created_at": now(),
|
||
"workspace_root": str(Path(args.workspace_root).resolve()),
|
||
"repository": repo.name,
|
||
"repository_path": str(repo.path),
|
||
"message": message,
|
||
"snapshot": frozen,
|
||
}
|
||
)
|
||
state = _seal(
|
||
{
|
||
"schema_version": 1,
|
||
"plan_id": plan_id,
|
||
"plan_sha256": payload["integrity_sha256"],
|
||
"status": "planned",
|
||
"updated_at": now(),
|
||
}
|
||
)
|
||
if getattr(args, "apply", False):
|
||
with _locked(args, repo.path):
|
||
if snapshot(repo.path, args.path) != frozen:
|
||
raise ValueError(
|
||
"Repository changed while the maintenance plan was being prepared"
|
||
)
|
||
directory = _receipt_path(args, plan_id)
|
||
if directory.exists():
|
||
raise ValueError("Maintenance plan already exists")
|
||
atomic_json(directory / "plan.json", payload)
|
||
atomic_json(directory / "state.json", state)
|
||
return _result(
|
||
payload,
|
||
state,
|
||
"Frozen plan saved. Commit and push each require their own explicit --apply.",
|
||
)
|
||
return _result(
|
||
payload,
|
||
{**state, "status": "preview"},
|
||
"Dry run: no plan file, index change, commit or network request was made.",
|
||
)
|
||
|
||
|
||
def _tree_entries(repo: Path, commit: str) -> dict:
|
||
entries = {}
|
||
for row in _git(repo, "ls-tree", "-r", "-z", commit).split(b"\0"):
|
||
if row:
|
||
metadata, _, name = row.partition(b"\t")
|
||
mode, kind, object_id = metadata.decode("ascii").split()
|
||
entries[os.fsdecode(name)] = {"mode": mode, "blob": object_id}
|
||
return entries
|
||
|
||
|
||
def _validate_commit(repo: Path, plan: dict, candidate: str) -> None:
|
||
frozen = plan["snapshot"]
|
||
parents = (
|
||
os.fsdecode(_git(repo, "rev-list", "--parents", "-n", "1", candidate))
|
||
.strip()
|
||
.split()
|
||
)
|
||
if parents != [candidate, frozen["head"]]:
|
||
raise ValueError("HEAD is not the single planned child commit")
|
||
raw = _git(repo, "cat-file", "commit", candidate)
|
||
message = raw.partition(b"\n\n")[2]
|
||
if message.rstrip(b"\n") != plan["message"].encode().rstrip(b"\n"):
|
||
raise ValueError("Commit message does not match the frozen plan")
|
||
before, after = _tree_entries(repo, frozen["head"]), _tree_entries(repo, candidate)
|
||
wanted = {item["path"] for item in frozen["files"]}
|
||
changed = {
|
||
name
|
||
for name in before.keys() | after.keys()
|
||
if before.get(name) != after.get(name)
|
||
}
|
||
if not changed or not changed <= wanted:
|
||
raise ValueError("Commit contains changes outside the approved paths")
|
||
for item in frozen["files"]:
|
||
expected = (
|
||
{"mode": item["mode"], "blob": item["blob"]} if item["exists"] else None
|
||
)
|
||
if after.get(item["path"]) != expected:
|
||
raise ValueError("Committed file content does not match the frozen plan")
|
||
|
||
|
||
def _index_path(repo: Path) -> Path:
|
||
raw = Path(os.fsdecode(_git(repo, "rev-parse", "--git-path", "index")).strip())
|
||
path = raw if raw.is_absolute() else repo / raw
|
||
if (
|
||
path.is_symlink()
|
||
or not path.is_file()
|
||
or path.stat().st_size > MAX_GIT_OUTPUT_BYTES
|
||
):
|
||
raise ValueError("The real Git index must be a bounded regular file")
|
||
return path
|
||
|
||
|
||
@contextmanager
|
||
def _index_lock(repo: Path):
|
||
path = _index_path(repo)
|
||
lock = Path(str(path) + ".lock")
|
||
try:
|
||
descriptor = os.open(lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
|
||
except FileExistsError:
|
||
raise ValueError(
|
||
"Git index is locked; inspect the owning operation, never delete an unverified lock automatically"
|
||
) from None
|
||
identity = os.fstat(descriptor)
|
||
try:
|
||
yield path, lock, descriptor
|
||
finally:
|
||
os.close(descriptor)
|
||
if lock.exists():
|
||
current = lock.lstat()
|
||
if (current.st_dev, current.st_ino) == (identity.st_dev, identity.st_ino):
|
||
lock.unlink()
|
||
|
||
|
||
def _index_info(files: list[dict]) -> bytes:
|
||
return b"".join(
|
||
(
|
||
f"{item['mode']} {item['blob']}\t{item['path']}\0"
|
||
if item["exists"]
|
||
else f"0 {'0' * 40}\t{item['path']}\0"
|
||
).encode()
|
||
for item in files
|
||
)
|
||
|
||
|
||
def _read_selected_bytes(path: Path):
|
||
descriptor = os.open(
|
||
path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0)
|
||
)
|
||
with os.fdopen(descriptor, "rb") as handle:
|
||
metadata = os.fstat(handle.fileno())
|
||
if not stat.S_ISREG(metadata.st_mode) or metadata.st_size > MAX_FILE_BYTES:
|
||
raise ValueError("Selected content changed or exceeds its bound")
|
||
data = handle.read(MAX_FILE_BYTES + 1)
|
||
if len(data) > MAX_FILE_BYTES:
|
||
raise ValueError("Selected content exceeds its bound")
|
||
return data, metadata
|
||
|
||
|
||
def _capture_blob(repo: Path, item: dict) -> None:
|
||
if not item["exists"]:
|
||
return
|
||
data, _ = _read_selected_bytes(repo / item["path"])
|
||
if len(data) > MAX_FILE_BYTES or hashlib.sha256(data).hexdigest() != item["sha256"]:
|
||
raise ValueError(
|
||
"Selected content changed before it could be captured; prepare another plan"
|
||
)
|
||
# The same captured bytes are hashed and written. No second working-tree
|
||
# read can substitute newer edits into the approved blob.
|
||
blob = os.fsdecode(
|
||
_git(
|
||
repo,
|
||
"hash-object",
|
||
"-w",
|
||
"--path",
|
||
item["path"],
|
||
"--stdin",
|
||
input_bytes=data,
|
||
)
|
||
).strip()
|
||
if blob != item["blob"]:
|
||
raise ValueError(
|
||
"Git's captured blob differs from the frozen plan; no commit was published"
|
||
)
|
||
|
||
|
||
def _selected_index(repo: Path, source: Path, target: Path, files: list[dict]) -> str:
|
||
shutil.copyfile(source, target)
|
||
_git(
|
||
repo,
|
||
"update-index",
|
||
"-z",
|
||
"--index-info",
|
||
input_bytes=_index_info(files),
|
||
index_file=target,
|
||
)
|
||
return _index(repo, index_file=target)[1]
|
||
|
||
|
||
def _publish_index(repo: Path, locked, prepared: Path, expected_before: str) -> None:
|
||
index, lock, descriptor = locked
|
||
if _index(repo)[1] != expected_before:
|
||
raise ValueError(
|
||
"The real index changed concurrently; its new contents were preserved"
|
||
)
|
||
data = prepared.read_bytes()
|
||
if len(data) > MAX_GIT_OUTPUT_BYTES:
|
||
raise ValueError("Prepared index exceeds its bound")
|
||
os.fchmod(descriptor, stat.S_IMODE(index.stat().st_mode))
|
||
os.ftruncate(descriptor, 0)
|
||
os.lseek(descriptor, 0, os.SEEK_SET)
|
||
view = memoryview(data)
|
||
while view:
|
||
written = os.write(descriptor, view)
|
||
view = view[written:]
|
||
os.fsync(descriptor)
|
||
if _index(repo)[1] != expected_before:
|
||
raise ValueError(
|
||
"The real index changed concurrently; its new contents were preserved"
|
||
)
|
||
os.replace(lock, index)
|
||
directory = os.open(index.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
|
||
try:
|
||
os.fsync(directory)
|
||
finally:
|
||
os.close(directory)
|
||
|
||
|
||
def commit(args) -> dict:
|
||
payload, state, directory = _read(args, args.plan_id)
|
||
repo = Path(payload["repository_path"])
|
||
with _locked(args, repo) if getattr(args, "apply", False) else _no_lock():
|
||
payload, state, directory = _read(args, args.plan_id)
|
||
if state["status"] == "committed":
|
||
return _result(
|
||
payload, state, "Commit already recorded; no duplicate commit created."
|
||
)
|
||
if state["status"] != "planned":
|
||
raise ValueError(
|
||
"Commit requires a planned receipt; reconcile interrupted/uncertain state first"
|
||
)
|
||
paths = [item["path"] for item in payload["snapshot"]["files"]]
|
||
if snapshot(repo, paths) != payload["snapshot"]:
|
||
raise ValueError(
|
||
"HEAD, branch, origin, config, index or selected file content changed; prepare a new plan"
|
||
)
|
||
if not getattr(args, "apply", False):
|
||
return _result(
|
||
payload,
|
||
{**state, "status": "preview"},
|
||
"Dry run: the frozen selected-path commit is ready; no index or commit was changed.",
|
||
)
|
||
state = _write_state(directory, state, status="committing")
|
||
try:
|
||
with (
|
||
_index_lock(repo) as locked,
|
||
tempfile.TemporaryDirectory(
|
||
prefix=".commit-index-", dir=directory
|
||
) as scratch,
|
||
):
|
||
if _index(repo)[1] != payload["snapshot"]["index_sha256"]:
|
||
raise ValueError("Real index changed before its lock was acquired")
|
||
private = Path(scratch) / "commit.index"
|
||
target = Path(scratch) / "selected.index"
|
||
for item in payload["snapshot"]["files"]:
|
||
_capture_blob(repo, item)
|
||
_git(repo, "read-tree", payload["snapshot"]["head"], index_file=private)
|
||
_git(
|
||
repo,
|
||
"update-index",
|
||
"-z",
|
||
"--index-info",
|
||
input_bytes=_index_info(payload["snapshot"]["files"]),
|
||
index_file=private,
|
||
)
|
||
tree = os.fsdecode(_git(repo, "write-tree", index_file=private)).strip()
|
||
target_hash = _selected_index(
|
||
repo, locked[0], target, payload["snapshot"]["files"]
|
||
)
|
||
current = snapshot(repo, paths, allow_missing=True)
|
||
if any(
|
||
current[key] != payload["snapshot"][key]
|
||
for key in (
|
||
"head",
|
||
"branch",
|
||
"origin_fetch_sha256",
|
||
"origin_push_sha256",
|
||
"config_sha256",
|
||
"index_sha256",
|
||
)
|
||
):
|
||
raise ValueError(
|
||
"Repository identity or real index changed during commit preparation"
|
||
)
|
||
candidate = os.fsdecode(
|
||
_git(
|
||
repo,
|
||
"commit-tree",
|
||
tree,
|
||
"-p",
|
||
payload["snapshot"]["head"],
|
||
"-m",
|
||
payload["message"],
|
||
)
|
||
).strip()
|
||
_validate_commit(repo, payload, candidate)
|
||
state = _write_state(
|
||
directory,
|
||
state,
|
||
status="committing",
|
||
phase="prepared",
|
||
commit=candidate,
|
||
prepared_tree=tree,
|
||
target_index_sha256=target_hash,
|
||
)
|
||
# Atomic compare-and-swap cannot overwrite another commit that
|
||
# advanced this branch after our snapshot.
|
||
_git(
|
||
repo,
|
||
"update-ref",
|
||
"--no-deref",
|
||
"-m",
|
||
"devkit selected-path commit",
|
||
f"refs/heads/{payload['snapshot']['branch']}",
|
||
candidate,
|
||
payload["snapshot"]["head"],
|
||
)
|
||
if os.fsdecode(_git(repo, "rev-parse", "HEAD")).strip() != candidate:
|
||
raise ValueError(
|
||
"HEAD changed while publishing the prepared commit; inspect before reconciliation"
|
||
)
|
||
_publish_index(
|
||
repo, locked, target, payload["snapshot"]["index_sha256"]
|
||
)
|
||
committed_snapshot = snapshot(repo, paths, allow_missing=True)
|
||
if (
|
||
committed_snapshot["head"] != candidate
|
||
or committed_snapshot["branch"] != payload["snapshot"]["branch"]
|
||
):
|
||
raise ValueError(
|
||
"HEAD or branch changed before the exact commit result could be recorded"
|
||
)
|
||
if (
|
||
committed_snapshot["unselected_index_sha256"]
|
||
!= payload["snapshot"]["unselected_index_sha256"]
|
||
):
|
||
raise ValueError(
|
||
"Unrelated index entries changed during the commit; reconcile before proceeding"
|
||
)
|
||
state = _write_state(
|
||
directory,
|
||
state,
|
||
status="committed",
|
||
commit=candidate,
|
||
committed_snapshot=committed_snapshot,
|
||
)
|
||
except BaseException:
|
||
_write_state(directory, state, status="needs_reconcile", operation="commit")
|
||
raise
|
||
return _result(
|
||
payload,
|
||
state,
|
||
"Selected paths committed. Unrelated staged/dirty files were left untouched; nothing was pushed.",
|
||
)
|
||
|
||
|
||
@contextmanager
|
||
def _no_lock():
|
||
yield
|
||
|
||
|
||
def _remote_head(repo: Path, url: str, branch: str) -> str | None:
|
||
raw = _git(repo, "ls-remote", "--refs", url, f"refs/heads/{branch}", timeout=60)
|
||
entries = raw.splitlines()
|
||
if not entries:
|
||
return None
|
||
if len(entries) != 1:
|
||
raise ValueError("Remote branch response is ambiguous")
|
||
object_id, _, ref = entries[0].partition(b"\t")
|
||
if ref.decode() != f"refs/heads/{branch}" or not re.fullmatch(
|
||
rb"[0-9a-f]{40,64}", object_id
|
||
):
|
||
raise ValueError("Remote branch response is malformed")
|
||
return object_id.decode()
|
||
|
||
|
||
def push(args) -> dict:
|
||
payload, state, directory = _read(args, args.plan_id)
|
||
repo = Path(payload["repository_path"])
|
||
with _locked(args, repo) if getattr(args, "apply", False) else _no_lock():
|
||
payload, state, directory = _read(args, args.plan_id)
|
||
if state["status"] == "pushed":
|
||
return _result(
|
||
payload,
|
||
state,
|
||
"Push already recorded; no duplicate network mutation attempted.",
|
||
)
|
||
if state["status"] != "committed":
|
||
raise ValueError(
|
||
"Push requires a verified recorded commit; reconcile uncertain state first"
|
||
)
|
||
paths = [item["path"] for item in payload["snapshot"]["files"]]
|
||
current = snapshot(repo, paths, allow_missing=True)
|
||
if current != state["committed_snapshot"] or current["head"] != state["commit"]:
|
||
raise ValueError(
|
||
"Repository changed after its recorded commit; push is blocked"
|
||
)
|
||
_validate_commit(repo, payload, state["commit"])
|
||
_, urls = _remote_urls(repo)
|
||
if len(urls) != 1 or not _safe_remote(urls[0]):
|
||
raise ValueError(
|
||
"Push requires exactly one origin push URL using a supported standard Git transport"
|
||
)
|
||
if not getattr(args, "apply", False):
|
||
return _result(
|
||
payload,
|
||
{**state, "status": "preview"},
|
||
"Dry run: normal push of the recorded commit/branch only; no network request made.",
|
||
)
|
||
state = _write_state(directory, state, status="pushing")
|
||
try:
|
||
_git(
|
||
repo,
|
||
"push",
|
||
"--porcelain",
|
||
"--no-follow-tags",
|
||
"--recurse-submodules=no",
|
||
"origin",
|
||
f"{state['commit']}:refs/heads/{payload['snapshot']['branch']}",
|
||
timeout=180,
|
||
)
|
||
remote_head = _remote_head(repo, urls[0], payload["snapshot"]["branch"])
|
||
if remote_head != state["commit"]:
|
||
raise ValueError(
|
||
"Remote head does not match the recorded commit; reconcile before retrying"
|
||
)
|
||
state = _write_state(
|
||
directory, state, status="pushed", remote_commit=remote_head
|
||
)
|
||
except BaseException:
|
||
_write_state(directory, state, status="needs_reconcile", operation="push")
|
||
raise
|
||
return _result(
|
||
payload,
|
||
state,
|
||
"Normal push verified at the exact recorded branch. No force push or extra tags were requested.",
|
||
)
|
||
|
||
|
||
def status(args) -> dict:
|
||
payload, state, _ = _read(args, args.plan_id)
|
||
return _result(
|
||
payload,
|
||
state,
|
||
"Recorded local maintenance state; no remote request or repository mutation made.",
|
||
)
|
||
|
||
|
||
def reconcile(args) -> dict:
|
||
payload, state, directory = _read(args, args.plan_id)
|
||
repo = Path(payload["repository_path"])
|
||
if state["status"] not in {"committing", "pushing", "needs_reconcile"}:
|
||
raise ValueError(
|
||
"Only interrupted/uncertain maintenance requires reconciliation"
|
||
)
|
||
if not getattr(args, "apply", False):
|
||
return _result(
|
||
payload,
|
||
{**state, "status": "preview"},
|
||
"Dry run: reconciliation can record an exact existing commit/push, never create either. Push reconciliation reads the frozen remote only with --apply.",
|
||
)
|
||
with _locked(args, repo):
|
||
payload, state, directory = _read(args, args.plan_id)
|
||
if state["status"] in {"committed", "pushed", "not_committed"}:
|
||
return _result(
|
||
payload,
|
||
state,
|
||
"Another reconciliation already recorded this result; no replay or downgrade occurred.",
|
||
)
|
||
if state["status"] not in {"committing", "pushing", "needs_reconcile"}:
|
||
raise ValueError("Receipt is no longer awaiting reconciliation")
|
||
paths = [item["path"] for item in payload["snapshot"]["files"]]
|
||
current = snapshot(repo, paths, allow_missing=True)
|
||
if any(
|
||
current[key] != payload["snapshot"][key]
|
||
for key in (
|
||
"branch",
|
||
"origin_fetch_sha256",
|
||
"origin_push_sha256",
|
||
"config_sha256",
|
||
"unselected_index_sha256",
|
||
)
|
||
):
|
||
raise ValueError(
|
||
"Repository/remote identity changed; reconciliation cannot claim a result"
|
||
)
|
||
if state.get("operation") == "push" or state["status"] == "pushing":
|
||
_, urls = _remote_urls(repo)
|
||
if (
|
||
len(urls) != 1
|
||
or not _safe_remote(urls[0])
|
||
or current["head"] != state.get("commit")
|
||
):
|
||
raise ValueError("Frozen push identity cannot be established")
|
||
remote = _remote_head(repo, urls[0], payload["snapshot"]["branch"])
|
||
if remote != state["commit"]:
|
||
raise ValueError(
|
||
"Push is not verifiably complete; inspect the remote before preparing another operation"
|
||
)
|
||
state = _write_state(
|
||
directory, state, status="pushed", remote_commit=remote
|
||
)
|
||
else:
|
||
if current["head"] == payload["snapshot"]["head"]:
|
||
state = _write_state(directory, state, status="not_committed")
|
||
return _result(
|
||
payload,
|
||
state,
|
||
"The planned commit is not current on this branch. Nothing was rolled back; inspect any prepared objects and create a new plan.",
|
||
)
|
||
if current["head"] != state.get("commit"):
|
||
raise ValueError(
|
||
"HEAD is not the recorded prepared commit; reconciliation cannot adopt another commit"
|
||
)
|
||
_validate_commit(repo, payload, current["head"])
|
||
if current["index_sha256"] != state.get("target_index_sha256"):
|
||
if current["index_sha256"] != payload["snapshot"]["index_sha256"]:
|
||
raise ValueError(
|
||
"The real index changed independently; preserve it and resolve selected staging through normal Git"
|
||
)
|
||
with (
|
||
_index_lock(repo) as locked,
|
||
tempfile.TemporaryDirectory(
|
||
prefix=".reconcile-index-", dir=directory
|
||
) as scratch,
|
||
):
|
||
prepared = Path(scratch) / "selected.index"
|
||
target = _selected_index(
|
||
repo, locked[0], prepared, payload["snapshot"]["files"]
|
||
)
|
||
if target != state.get("target_index_sha256"):
|
||
raise ValueError(
|
||
"Reconstructed index differs from the recorded prepared index"
|
||
)
|
||
checked = snapshot(repo, paths, allow_missing=True)
|
||
if (
|
||
checked["head"] != state["commit"]
|
||
or checked["branch"] != payload["snapshot"]["branch"]
|
||
):
|
||
raise ValueError(
|
||
"HEAD or branch changed during index reconciliation; no result was adopted"
|
||
)
|
||
_publish_index(
|
||
repo, locked, prepared, payload["snapshot"]["index_sha256"]
|
||
)
|
||
current = snapshot(repo, paths, allow_missing=True)
|
||
if (
|
||
current["head"] != state["commit"]
|
||
or current["branch"] != payload["snapshot"]["branch"]
|
||
):
|
||
raise ValueError(
|
||
"HEAD or branch changed before the exact reconciled result could be recorded"
|
||
)
|
||
_validate_commit(repo, payload, state["commit"])
|
||
state = _write_state(
|
||
directory,
|
||
state,
|
||
status="committed",
|
||
commit=state["commit"],
|
||
committed_snapshot=current,
|
||
)
|
||
return _result(
|
||
payload,
|
||
state,
|
||
"The already-existing result was verified and recorded; no commit or push was executed.",
|
||
)
|
||
|
||
|
||
def register(subparsers) -> None:
|
||
parser = subparsers.add_parser(
|
||
"git", help="Frozen explicit-path Git maintenance; all writes require --apply"
|
||
)
|
||
commands = parser.add_subparsers(dest="git_command", required=True)
|
||
create = commands.add_parser(
|
||
"plan",
|
||
help="Preview/save exact repository, paths, content, index and commit message",
|
||
)
|
||
create.add_argument("--repo", required=True)
|
||
create.add_argument("--path", action="append", required=True)
|
||
create.add_argument("--message", required=True)
|
||
create.add_argument(
|
||
"--apply",
|
||
action="store_true",
|
||
help="Save an immutable local plan; never commits or pushes",
|
||
)
|
||
create.set_defaults(handler=plan)
|
||
for name, handler in (
|
||
("commit", commit),
|
||
("push", push),
|
||
("status", status),
|
||
("reconcile", reconcile),
|
||
):
|
||
command = commands.add_parser(name)
|
||
command.add_argument("plan_id")
|
||
if name != "status":
|
||
command.add_argument("--apply", action="store_true")
|
||
command.set_defaults(handler=handler)
|