coord/hook.py

204 lines
6.8 KiB
Python
Raw Permalink Normal View History

#!/usr/bin/env python3
"""Fail-open recognized-write guard and transactional peer-message delivery."""
from __future__ import annotations
import importlib.util
import importlib.machinery
import hashlib
import json
import os
import pathlib
import re
import socket
import sys
import uuid
HERE = pathlib.Path(__file__).resolve().parent
CLI_SPEC = importlib.util.spec_from_loader(
"coord_cli",
importlib.machinery.SourceFileLoader("coord_cli", str(HERE / "coord")),
)
try:
cli = importlib.util.module_from_spec(CLI_SPEC)
assert CLI_SPEC.loader is not None
CLI_SPEC.loader.exec_module(cli)
except Exception as exc: # the hook boundary must fail open, including import
print(f"coord-hook: cannot load coordinator; failing open: {exc}", file=sys.stderr)
raise SystemExit(0)
PATCH_PATH = re.compile(
r"^\*\*\* (?:Add|Update|Delete) File: (.+)$|^\*\*\* Move to: (.+)$",
re.MULTILINE,
)
MAX_MESSAGES = 5
def allow() -> None:
raise SystemExit(0)
def runtime_for(db: object, event: dict) -> str:
agent_id = str(event.get("agent_id") or "")
conversation = str(event.get("session_id") or "")
if agent_id and conversation:
sub_conversation = f"{conversation}:agent:{agent_id}"
exact = str(
uuid.uuid5(
uuid.NAMESPACE_URL, f"fleet-agent-coord:{sub_conversation}"
)
)
try:
return db.resolve_runtime(exact)
except KeyError:
return db.register_instance(
sub_conversation,
runtime_id=exact,
harness="claude-code-subagent",
host=socket.gethostname(),
cwd=str(project_root()),
label=f"subagent:{agent_id}",
)
exact = os.environ.get("COORD_INSTANCE_ID", "")
return db.resolve_runtime(exact or event.get("session_id", ""))
def project_root() -> pathlib.Path:
"""Return the checkout root; the override exists for packaged tests/tools."""
override = os.environ.get("COORD_PROJECT_ROOT", "")
return pathlib.Path(override).resolve() if override else cli.ROOT
def relative_path(raw: str, cwd: pathlib.Path, root: pathlib.Path) -> str | None:
candidate = pathlib.Path(raw)
if not candidate.is_absolute():
candidate = cwd / candidate
try:
return candidate.resolve(strict=False).relative_to(root).as_posix()
except ValueError:
return None
def paths_from(event: dict) -> list[str]:
tool_input = event.get("tool_input") or {}
cwd = pathlib.Path(event.get("cwd") or os.getcwd()).resolve()
root = project_root()
raw_paths: list[str] = []
direct = tool_input.get("file_path") or tool_input.get("path")
if isinstance(direct, str):
raw_paths.append(direct)
body = tool_input.get("command") or tool_input.get("input") or ""
if isinstance(body, str):
for match in PATCH_PATH.finditer(body):
raw_paths.append(match.group(1) or match.group(2))
normalized = (relative_path(path.strip(), cwd, root) for path in raw_paths)
return list(dict.fromkeys(path for path in normalized if path))
def output_context(messages: list[object]) -> None:
lines = []
for item in messages:
sender = cli.runtime_handle(item.sender_runtime_id)
if item.state == "delivered":
digest = hashlib.sha256(item.body.encode("utf-8")).hexdigest()[:10]
lines.append(
f"[{item.message_id}] {sender} pending again "
f"({item.kind}, sha256:{digest}); read with `coord inbox --peek`"
)
else:
lines.append(
f"[{item.message_id}] {sender} ({item.kind}): {item.body}"
)
print(
json.dumps(
{
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"additionalContext": (
"Messages from peer agent runtimes:\n"
+ "\n".join(lines)
+ "\nDelivery repeats until explicitly acknowledged with "
"`coord ack <message-id>`."
),
}
}
)
)
def main() -> None:
try:
event = json.loads(sys.stdin.read())
except Exception:
allow()
db = cli.store()
owner = runtime_for(db, event)
db.heartbeat(
owner,
ttl_seconds=cli.storelib.DEFAULT_INSTANCE_TTL,
renew_lease_ttl_seconds=cli.storelib.DEFAULT_LEASE_TTL,
)
managed_workspace = db.is_managed_workspace(project_root())
leases = db.active_leases()
for path in paths_from(event):
owner_has_lease = False
for lease in leases:
if (
lease["runtime_id"] == owner
and lease["mode"] == "exclusive"
and db.resources_overlap(path, lease["path"])
):
owner_has_lease = True
if lease["runtime_id"] == owner or lease["mode"] != "exclusive":
continue
if not db.resources_overlap(path, lease["path"]):
continue
print(
json.dumps(
{
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "deny",
"permissionDecisionReason": (
f"{path} is leased exclusively by "
f"{cli.runtime_handle(lease['runtime_id'])} for "
f"{lease['purpose'] or 'unstated'}. Coordinate before editing."
),
}
}
)
)
return
if not managed_workspace and not owner_has_lease:
print(
json.dumps(
{
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "deny",
"permissionDecisionReason": (
f"{path} is in the canonical integration checkout. "
"Writable work must use a managed isolated clone, or "
"the integration owner must claim this exact resource."
),
}
}
)
)
return
pending = db.inbox(owner, limit=MAX_MESSAGES, mark_delivered=True)
if pending:
output_context(pending)
if __name__ == "__main__":
try:
main()
except SystemExit:
raise
except Exception as exc: # fail open is load-bearing
print(f"coord-hook: failing open: {exc}", file=sys.stderr)
raise SystemExit(0)