multi-agent-mux/.agents/hooks/loop_delegation_guard.sh

#!/usr/bin/env bash
# O-3 — Invocation-Aware Scoped Guard (Rev.2).
#
# Normal mode: the orchestrator IS the Main Creator and may edit files freely.
# While /multi-agent-mux-loop is active it must delegate through run_loop.sh
# instead, so this PreToolUse hook denies direct file mutation and says why.
#
# Contract (agy hooks.json): JSON payload on stdin, JSON decision on stdout.
#   in : {"toolCall":{"name":..., "args":{...}}, "workspacePaths":[...],
#         "transcriptPath":"..."}
#   out: {"decision":"allow"|"deny", "reason":"..."}
#
# Fails OPEN: any internal error emits `allow`. A guard that blocks the agent
# because it could not parse its own input would be worse than the drift.
set -uo pipefail

payload="$(cat)"

exec 3>&1                       # keep the decision channel separate from noise
allow() { printf '{"decision":"allow"}\n' >&3; exit 0; }

MARKER="${MAM_LOOP_GUARD_MARKER:-}"

python3 - "$payload" "$MARKER" >&3 <<'PY' || allow
import json, os, sys, subprocess

payload_raw, marker_override = sys.argv[1], sys.argv[2]

def emit(decision, reason=None):
    out = {"decision": decision}
    if reason:
        out["reason"] = reason
    print(json.dumps(out))
    sys.exit(0)

try:
    p = json.loads(payload_raw)
except Exception:
    emit("allow")                                    # unparseable -> fail open

name = ((p.get("toolCall") or {}).get("name") or "").strip().lower()

# Step-type-derived names (hooks.json matches on these), NOT the model-facing
# tool names. This agy build has CORTEX_STEP_TYPE_FILE_CHANGE / EDIT_NOTEBOOK /
# WRITE_BLOB; there is no REPLACE_FILE_CONTENT step type at all.
MUTATING = {"file_change", "edit_notebook", "write_blob"}
if name not in MUTATING:
    emit("allow")

ws = (p.get("workspacePaths") or [None])[0] or os.getcwd()
marker = marker_override or os.path.join(ws, ".mam", "loop-guard-active")

def _lstart(pid):
    """Process start time, or '' if the process is gone/unknowable."""
    try:
        out = subprocess.run(["ps", "-p", str(pid), "-o", "lstart="],
                             capture_output=True, text=True, timeout=5)
    except Exception:
        return ""
    return " ".join(out.stdout.split())

def _marker_active(path):
    """True only if the marker exists AND its owning process is still alive.

    SIGKILL cannot be trapped, so a trap-based release always has a leak
    window. A stale marker must never block the orchestrator forever, so
    identity (pid + lstart) -- not mere existence -- is the signal.
    """
    if not os.path.exists(path):
        return False
    try:
        with open(path, encoding="utf-8", errors="replace") as f:
            txt = f.read()
    except Exception:
        return False
    fields = {}
    for line in txt.splitlines():
        if "=" in line:
            k, v = line.split("=", 1)
            fields[k.strip()] = v.strip()
    
    try:
        pid = int(fields.get("pid", ""))
    except ValueError:
        pid = None

    if pid is None:
        return True                                  # no pid recorded -> honour it

    recorded_lstart = " ".join(fields.get("lstart", "").split())
    if recorded_lstart:
        # pid + start time is a stable identity. A reused pid always has a
        # different start time, so this closes the rollover livelock: a marker
        # we cannot positively identify must never block the orchestrator.
        return _lstart(pid) == recorded_lstart

    # Legacy marker with no lstart: fall back to liveness, but treat an
    # unidentifiable owner as STALE. Blocking forever is the worse error.
    try:
        os.kill(pid, 0)
        return True
    except ProcessLookupError:
        return False                                 # owner gone -> stale
    except PermissionError:
        return False                                 # different owner -> cannot be our loop

active = _marker_active(marker)

if not active:
    # Best-effort second signal: the skill was invoked but run_loop.sh has not
    # started yet, so no marker exists. Look for the invocation in the tail of
    # the transcript. Absence of a transcript simply means "not active".
    tpath = p.get("transcriptPath") or ""
    try:
        if tpath and os.path.exists(tpath):
            with open(tpath, encoding="utf-8", errors="replace") as f:
                tail = f.readlines()[-200:]
            for line in reversed(tail):
                if "/multi-agent-mux-loop" in line:
                    active = True
                    break
                if "MAM_LOOP_GUARD_RELEASE" in line:
                    break                            # loop finished; stop here
    except Exception:
        pass                                         # transcript unreadable -> not active

if not active:
    emit("allow")

emit("deny",
     "The /multi-agent-mux-loop skill is active, so direct file edits are out "
     "of scope for the orchestrator. Stop editing and delegate instead: run "
     "bash .agents/skills/multi-agent-mux-loop/scripts/run_loop.sh "
     "--creator <session> --task <goal>. "
     "See .agents/MULTI_AGENT_RULES.md #3.2 (Invocation-Aware Scoped Guard).")
PY