feat(lease): add single-turn Claude promotion trigger
This commit is contained in:
Jason Woltje
@@ -0,0 +1,306 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Claude Stop hook that completes a pending operator-triggered promotion."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import fcntl
|
||||
import json
|
||||
import os
|
||||
import secrets
|
||||
import stat
|
||||
import subprocess
|
||||
import sys
|
||||
from collections.abc import Callable, Mapping
|
||||
from pathlib import Path
|
||||
from typing import Final, NamedTuple, TextIO
|
||||
|
||||
MAX_FRAME: Final = 64 * 1024
|
||||
PROMOTER_TIMEOUT_SECONDS: Final = 10.0
|
||||
PROMOTER: Final = Path(__file__).resolve().with_name("lease_promote.py")
|
||||
PENDING_DIRECTORY: Final = "mosaic-lease"
|
||||
LOCK_FILE: Final = "promotion.lock"
|
||||
TERMINAL_FAILURE_CODES: Final = frozenset(
|
||||
{
|
||||
"RECEIPT_REPLAY",
|
||||
"RECEIPT_MISMATCH",
|
||||
"INVALID_LEASE_TRANSITION",
|
||||
"PROMOTION_TOKEN_INVALID",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
class PendingChallenge(NamedTuple):
|
||||
value: str
|
||||
device: int
|
||||
inode: int
|
||||
|
||||
|
||||
def reject_duplicate_json_keys(pairs: list[tuple[str, object]]) -> dict[str, object]:
|
||||
value: dict[str, object] = {}
|
||||
for key, item in pairs:
|
||||
if key in value:
|
||||
raise ValueError("duplicate promoter JSON key")
|
||||
value[key] = item
|
||||
return value
|
||||
|
||||
|
||||
def session_pending_name(environ: Mapping[str, str]) -> tuple[Path, str]:
|
||||
runtime_dir = Path(environ["XDG_RUNTIME_DIR"])
|
||||
session_id = environ["MOSAIC_LEASE_SESSION_ID"]
|
||||
if not runtime_dir.is_absolute():
|
||||
raise ValueError("XDG_RUNTIME_DIR must be absolute")
|
||||
if len(session_id) != 64 or any(character not in "0123456789abcdef" for character in session_id):
|
||||
raise ValueError("invalid lease session id")
|
||||
return runtime_dir, f"pending-{session_id}"
|
||||
|
||||
|
||||
def open_pending_directory(runtime_dir: Path) -> int | None:
|
||||
directory_flags = (
|
||||
os.O_RDONLY
|
||||
| getattr(os, "O_CLOEXEC", 0)
|
||||
| getattr(os, "O_DIRECTORY", 0)
|
||||
| getattr(os, "O_NOFOLLOW", 0)
|
||||
)
|
||||
try:
|
||||
runtime_descriptor = os.open(runtime_dir, directory_flags)
|
||||
except FileNotFoundError:
|
||||
return None
|
||||
try:
|
||||
runtime_metadata = os.fstat(runtime_descriptor)
|
||||
if (
|
||||
not stat.S_ISDIR(runtime_metadata.st_mode)
|
||||
or runtime_metadata.st_uid != os.getuid()
|
||||
or stat.S_IMODE(runtime_metadata.st_mode) != 0o700
|
||||
):
|
||||
raise ValueError("unsafe XDG runtime directory")
|
||||
try:
|
||||
descriptor = os.open(PENDING_DIRECTORY, directory_flags, dir_fd=runtime_descriptor)
|
||||
except FileNotFoundError:
|
||||
return None
|
||||
finally:
|
||||
os.close(runtime_descriptor)
|
||||
|
||||
metadata = os.fstat(descriptor)
|
||||
if (
|
||||
not stat.S_ISDIR(metadata.st_mode)
|
||||
or metadata.st_uid != os.getuid()
|
||||
or stat.S_IMODE(metadata.st_mode) != 0o700
|
||||
):
|
||||
os.close(descriptor)
|
||||
raise ValueError("unsafe promotion pending directory")
|
||||
return descriptor
|
||||
|
||||
|
||||
def acquire_lock(directory_descriptor: int) -> int:
|
||||
flags = (
|
||||
os.O_RDWR
|
||||
| os.O_CREAT
|
||||
| getattr(os, "O_CLOEXEC", 0)
|
||||
| getattr(os, "O_NOFOLLOW", 0)
|
||||
)
|
||||
descriptor = os.open(LOCK_FILE, flags, 0o600, dir_fd=directory_descriptor)
|
||||
metadata = os.fstat(descriptor)
|
||||
if (
|
||||
not stat.S_ISREG(metadata.st_mode)
|
||||
or metadata.st_uid != os.getuid()
|
||||
or stat.S_IMODE(metadata.st_mode) != 0o600
|
||||
):
|
||||
os.close(descriptor)
|
||||
raise ValueError("unsafe promotion lock file")
|
||||
try:
|
||||
fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
except BlockingIOError:
|
||||
os.close(descriptor)
|
||||
raise
|
||||
return descriptor
|
||||
|
||||
|
||||
def read_pending(directory_descriptor: int, name: str) -> PendingChallenge | None:
|
||||
flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
|
||||
try:
|
||||
descriptor = os.open(name, flags, dir_fd=directory_descriptor)
|
||||
except FileNotFoundError:
|
||||
return None
|
||||
try:
|
||||
metadata = os.fstat(descriptor)
|
||||
if (
|
||||
not stat.S_ISREG(metadata.st_mode)
|
||||
or metadata.st_uid != os.getuid()
|
||||
or stat.S_IMODE(metadata.st_mode) != 0o600
|
||||
or metadata.st_size <= 0
|
||||
or metadata.st_size > MAX_FRAME
|
||||
):
|
||||
raise ValueError("unsafe promotion pending file")
|
||||
raw = os.read(descriptor, MAX_FRAME + 1)
|
||||
finally:
|
||||
os.close(descriptor)
|
||||
if len(raw) > MAX_FRAME:
|
||||
raise ValueError("oversized promotion challenge")
|
||||
challenge = raw.decode("utf-8")
|
||||
if (
|
||||
len(challenge) != 64
|
||||
or any(character not in "0123456789abcdef" for character in challenge)
|
||||
):
|
||||
raise ValueError("invalid promotion challenge")
|
||||
return PendingChallenge(challenge, metadata.st_dev, metadata.st_ino)
|
||||
|
||||
|
||||
def delete_pending_if_unchanged(
|
||||
directory_descriptor: int,
|
||||
name: str,
|
||||
pending: PendingChallenge,
|
||||
error_stream: TextIO,
|
||||
) -> None:
|
||||
quarantine = f".{name}.delete-{secrets.token_hex(8)}"
|
||||
try:
|
||||
os.rename(
|
||||
name,
|
||||
quarantine,
|
||||
src_dir_fd=directory_descriptor,
|
||||
dst_dir_fd=directory_descriptor,
|
||||
)
|
||||
except FileNotFoundError:
|
||||
return
|
||||
except OSError as error:
|
||||
print(f"Mosaic promotion could not quarantine pending file: {error}", file=error_stream)
|
||||
return
|
||||
|
||||
try:
|
||||
moved = os.stat(
|
||||
quarantine,
|
||||
dir_fd=directory_descriptor,
|
||||
follow_symlinks=False,
|
||||
)
|
||||
if (moved.st_dev, moved.st_ino) == (pending.device, pending.inode):
|
||||
os.unlink(quarantine, dir_fd=directory_descriptor)
|
||||
os.fsync(directory_descriptor)
|
||||
return
|
||||
|
||||
print("Mosaic promotion pending file changed; preserving replacement.", file=error_stream)
|
||||
try:
|
||||
os.link(
|
||||
quarantine,
|
||||
name,
|
||||
src_dir_fd=directory_descriptor,
|
||||
dst_dir_fd=directory_descriptor,
|
||||
follow_symlinks=False,
|
||||
)
|
||||
except FileExistsError:
|
||||
print(
|
||||
f"Mosaic promotion preserved replacement as {quarantine}.",
|
||||
file=error_stream,
|
||||
)
|
||||
else:
|
||||
os.unlink(quarantine, dir_fd=directory_descriptor)
|
||||
os.fsync(directory_descriptor)
|
||||
except OSError as error:
|
||||
print(f"Mosaic promotion could not resolve pending file: {error}", file=error_stream)
|
||||
|
||||
|
||||
def parse_reply(completed: subprocess.CompletedProcess[str]) -> dict[str, object] | None:
|
||||
if completed.returncode != 0:
|
||||
return None
|
||||
try:
|
||||
value = json.loads(
|
||||
completed.stdout,
|
||||
object_pairs_hook=reject_duplicate_json_keys,
|
||||
)
|
||||
except (json.JSONDecodeError, RecursionError, TypeError, ValueError):
|
||||
return None
|
||||
if not isinstance(value, dict):
|
||||
return None
|
||||
if set(value) == {"stage", "ok", "state"}:
|
||||
if (
|
||||
value.get("stage") == "promote_lease"
|
||||
and value.get("ok") is True
|
||||
and value.get("state") == "VERIFIED"
|
||||
):
|
||||
return value
|
||||
return None
|
||||
if set(value) == {"stage", "ok", "code"}:
|
||||
if (
|
||||
value.get("stage") in {"observe_receipt", "promote_lease"}
|
||||
and value.get("ok") is False
|
||||
and isinstance(value.get("code"), str)
|
||||
and value.get("code")
|
||||
):
|
||||
return value
|
||||
return None
|
||||
|
||||
|
||||
def main(
|
||||
*,
|
||||
environ: Mapping[str, str] | None = None,
|
||||
stderr: TextIO | None = None,
|
||||
run: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run,
|
||||
) -> int:
|
||||
source_environment = os.environ if environ is None else environ
|
||||
error_stream = sys.stderr if stderr is None else stderr
|
||||
directory_descriptor: int | None = None
|
||||
lock_descriptor: int | None = None
|
||||
|
||||
try:
|
||||
runtime_dir, pending_name = session_pending_name(source_environment)
|
||||
directory_descriptor = open_pending_directory(runtime_dir)
|
||||
if directory_descriptor is None:
|
||||
return 0
|
||||
try:
|
||||
lock_descriptor = acquire_lock(directory_descriptor)
|
||||
except (BlockingIOError, FileNotFoundError):
|
||||
print("Mosaic promotion completion deferred: promotion is in progress.", file=error_stream)
|
||||
return 0
|
||||
pending = read_pending(directory_descriptor, pending_name)
|
||||
if pending is None:
|
||||
return 0
|
||||
completed = run(
|
||||
[
|
||||
sys.executable,
|
||||
"-I",
|
||||
"-S",
|
||||
"-B",
|
||||
str(PROMOTER),
|
||||
"--complete",
|
||||
pending.value,
|
||||
],
|
||||
check=False,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=dict(source_environment),
|
||||
timeout=PROMOTER_TIMEOUT_SECONDS,
|
||||
)
|
||||
reply = parse_reply(completed)
|
||||
if reply is not None and reply.get("ok") is True:
|
||||
delete_pending_if_unchanged(
|
||||
directory_descriptor,
|
||||
pending_name,
|
||||
pending,
|
||||
error_stream,
|
||||
)
|
||||
print("Mosaic lease promotion completed.", file=error_stream)
|
||||
return 0
|
||||
|
||||
if reply is not None:
|
||||
code = str(reply["code"])
|
||||
print(f"Mosaic promotion incomplete: {code}.", file=error_stream)
|
||||
if code in TERMINAL_FAILURE_CODES:
|
||||
delete_pending_if_unchanged(
|
||||
directory_descriptor,
|
||||
pending_name,
|
||||
pending,
|
||||
error_stream,
|
||||
)
|
||||
else:
|
||||
diagnostic = completed.stderr.strip() or f"promoter exit {completed.returncode}"
|
||||
print(f"Mosaic promotion retryable failure: {diagnostic}.", file=error_stream)
|
||||
except (KeyError, OSError, RecursionError, UnicodeError, ValueError, subprocess.SubprocessError) as error:
|
||||
print(f"Mosaic promotion completion deferred: {type(error).__name__}: {error}", file=error_stream)
|
||||
finally:
|
||||
if lock_descriptor is not None:
|
||||
os.close(lock_descriptor)
|
||||
if directory_descriptor is not None:
|
||||
os.close(directory_descriptor)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user