Publishing guard stops only a second helper in the same session

engine/hooks/publish-act-guard/README.md

@@ -1,16 +1,28 @@
 # publish-act-guard
 
-Refuses a publishing command issued from inside a subagent while a live Invoker
-owner is reachable, so publishing work routes through Invoker instead of
-fanning out across parallel subagents.
+Refuses a publishing command from a second subagent in the same session,
+within 30 minutes of another subagent's publish, while a live Invoker owner is
+reachable. Parallel publishing work routes through Invoker; a single subagent
+that finishes its work and pushes is not blocked.
+
+## Why only the second publisher
+
+The incident behind this gate (2026-09-09) was eight subagents each publishing
+from one session. The first version blocked every subagent publish while
+Invoker answered. Over 2026-09-14..25 it blocked 25 times across 7 sessions,
+and each block moved a single subagent's final push back to the parent
+session, which is never gated, often through a hand-off to the user. The
+parallel case is the one worth stopping, so the gate now records the first
+publisher per session and refuses a different one inside the window.
 
 ## Fires on
 
 A shell-like PreToolUse call whose command, parsed at command position, performs
 one of: `git push`, `gh pr create|merge|ready`, a mutating `gh api` call against
 `/pulls`, `mergify stack push`, `create-pr.mjs`, `safe-stack-push.mjs --execute`.
-All three must hold: the caller is a subagent, the act is one of those, and the
-liveness probe reports a reachable owner.
+All four must hold: the caller is a subagent, the act is one of those, a
+different subagent in the same session published within the last 30 minutes,
+and the liveness probe reports a reachable owner.
 
 ## Silent on
 
@@ -37,14 +49,22 @@ is a typed field, and never the prompt.
   (set `PUBLISH_ACT_GUARD_DEBUG=1` to see it).
 - Liveness unreadable (probe timed out or errored): open, and the block message
   says UNCHECKED so the report has to name it.
-- Owner reachable and a subagent is publishing: closed.
+- No session id in the payload, or the publisher ledger unreadable: open, UNCHECKED.
+- Owner reachable and a second subagent in the session publishing inside the
+  window: closed.
+
+## Publisher ledger
+
+`$TMPDIR/publish-act-guard-publishers.json` maps session id to subagent id to
+the time of its first allowed publish. Entries older than the window are
+dropped on each write.
 
 ## Block message and escape hatch
 
-The refusal names the act and points at the `invoker-plan-to-invoker` skill. It
-clears on its own when no live owner answers. The user clears it deliberately by
-saying "do it locally" or "don't use invoker", which routes the work here rather
-than to Invoker.
+The refusal names the act and the other subagent, and points at the
+`invoker-plan-to-invoker` skill. It also says the subagent can finish without
+publishing and hand the commit to the parent session, which is never gated. It
+clears on its own when no live owner answers or when the window passes.
 
 ## Liveness probe
 

engine/hooks/publish-act-guard/detect.py

@@ -1,21 +1,32 @@
-"""Refuse a publishing command run from inside a subagent while a live
+"""Refuse a second subagent's publishing command in one session while a live
 Invoker owner is reachable.
 
 The predecessor, agent-routing-guard, classified the spawn prompt with
 regexes over prose: "carrying commits" parsed as an action because the noun
 test required a determiner, and deleting the word cleared the block without
 changing what the subagent would do. This detector never reads prose. It
 parses the command the tool is about to run, which is a typed field, and
-decides from three facts: the act, the caller, and whether Invoker can take
-the work.
+decides from four facts: the act, the caller, whether another subagent in
+the same session already published within the window, and whether Invoker
+can take the work.
+
+The incident behind this gate was eight subagents publishing in parallel
+from one session. A single subagent that finishes its work and pushes is not
+that incident, and blocking it only moved the push back to the parent
+session, which this gate never covers. So the first publishing subagent in a
+session goes through and is recorded; a different subagent publishing within
+PARALLEL_WINDOW_SECONDS of it is refused.
 
 Fail directions, one per read:
   - unparsable payload, non-shell tool, empty command, non-subagent caller,
     command that is not a publishing act: open, with the reason recorded.
   - Invoker liveness could not be determined: open, and the reason is
     printed, because a probe that cannot run must not hold up publishing
     when Invoker may itself be down.
-  - Invoker reachable and a subagent is publishing: closed.
+  - no session id, or the publisher ledger could not be read: open, and the
+    reason is printed.
+  - Invoker reachable and a different subagent in this session published
+    within the window: closed.
 """
 from __future__ import annotations
 
@@ -35,6 +46,11 @@
     os.environ.get("TMPDIR", "/tmp"), "publish-act-guard-liveness.json"
 )
 DEBUG_ENV = "PUBLISH_ACT_GUARD_DEBUG"
+PUBLISHERS_PATH = os.path.join(
+    os.environ.get("TMPDIR", "/tmp"), "publish-act-guard-publishers.json"
+)
+PARALLEL_WINDOW_SECONDS = 30 * 60
+SESSION_ID_KEYS = ("session_id", "sessionId")
 
 SUBAGENT_ID_KEYS = ("subagent_id", "subagentId", "agent_id", "agentId", "sub_agent_id")
 
@@ -49,12 +65,14 @@
 
 BLOCK_MESSAGE = (
     "publish-act-guard: this subagent is about to run a publishing command "
-    "({act}) while a live Invoker owner is reachable.\n"
-    "Publishing work routes through Invoker, not through parallel subagents: "
-    "follow the installed {skill} skill, then submit the plan.\n"
+    "({act}), and subagent {other} in the same session already published "
+    "{minutes} min ago while a live Invoker owner is reachable.\n"
+    "Several subagents publishing in parallel is work for Invoker: follow the "
+    "installed {skill} skill, then submit the plan. Or finish without "
+    "publishing and report the commit: the parent session is never gated and "
+    "can publish it.\n"
     "This gate reads the command, never the prompt. It goes quiet on its own "
-    "when no live owner answers, and the user clears it by saying "
-    "\"do it locally\" or \"don't use invoker\"."
+    "when no live owner answers or {window} min after the other publish."
 )
 
 
@@ -97,11 +115,16 @@ def command_text(payload: dict) -> str:
     return ""
 
 
+def _first_string(payload: dict, keys: tuple[str, ...]) -> str:
+    for key in keys:
+        value = payload.get(key)
+        if isinstance(value, str) and value.strip():
+            return value.strip()
+    return ""
+
+
 def in_subagent(payload: dict) -> bool:
-    return any(
-        isinstance(payload.get(key), str) and payload[key].strip()
-        for key in SUBAGENT_ID_KEYS
-    )
+    return bool(_first_string(payload, SUBAGENT_ID_KEYS))
 
 
 def _words(command: str) -> list[str]:
@@ -255,7 +278,47 @@ def _probe() -> int | None:
     return completed.returncode
 
 
-def decide(payload: dict, runner=None) -> str | None:
+def _ledger_read() -> tuple[dict | None, str]:
+    try:
+        with open(PUBLISHERS_PATH, encoding="utf-8") as handle:
+            ledger = json.load(handle)
+    except FileNotFoundError:
+        return {}, ""
+    except (PermissionError, json.JSONDecodeError, OSError) as exc:
+        return None, f"publisher ledger unreadable ({type(exc).__name__}: {exc})"
+    if not isinstance(ledger, dict):
+        return None, "publisher ledger is not an object"
+    return ledger, ""
+
+
+def _ledger_record(ledger: dict, session: str, agent: str, now: float) -> None:
+    fresh = {
+        sid: {aid: stamp for aid, stamp in agents.items()
+              if isinstance(stamp, (int, float)) and now - stamp <= PARALLEL_WINDOW_SECONDS}
+        for sid, agents in ledger.items() if isinstance(agents, dict)
+    }
+    fresh.setdefault(session, {})[agent] = now
+    fresh = {sid: agents for sid, agents in fresh.items() if agents}
+    try:
+        with open(PUBLISHERS_PATH, "w", encoding="utf-8") as handle:
+            json.dump(fresh, handle)
+    except (PermissionError, OSError) as exc:
+        _warn(f"publisher ledger not written ({type(exc).__name__}: {exc})")
+
+
+def other_recent_publisher(ledger: dict, session: str, agent: str, now: float) -> tuple[str, float] | None:
+    agents = ledger.get(session, {})
+    if not isinstance(agents, dict):
+        _warn(f"publisher ledger entry for session {session} is not an object; treating it as empty")
+        return None
+    recent = [
+        (aid, stamp) for aid, stamp in agents.items()
+        if aid != agent and isinstance(stamp, (int, float)) and 0 <= now - stamp <= PARALLEL_WINDOW_SECONDS
+    ]
+    return max(recent, key=lambda item: item[1]) if recent else None
+
+
+def decide(payload: dict, runner=None, now: float | None = None) -> str | None:
     """The refusal to print, or None to let the command run."""
     if not isinstance(payload, dict):
         return _silent("payload is not an object")
@@ -269,6 +332,9 @@ def decide(payload: dict, runner=None) -> str | None:
     act = publishing_act(command)
     if act is None:
         return _silent("command performs no publishing act")
+    stamp = time.time() if now is None else now
+    session = _first_string(payload, SESSION_ID_KEYS)
+    agent = _first_string(payload, SUBAGENT_ID_KEYS)
     state, reason = invoker_state(runner=runner)
     if state == DOWN:
         return _silent(f"no live Invoker owner; {act} may proceed here")
@@ -277,4 +343,26 @@ def decide(payload: dict, runner=None) -> str | None:
             f"publish-act-guard: UNCHECKED: could not tell whether a live Invoker owner "
             f"is reachable ({reason}); allowing {act}. Say so in the report."
         )
-    return BLOCK_MESSAGE.format(act=act, skill=ROUTING_SKILL)
+    if not session:
+        return (
+            f"publish-act-guard: UNCHECKED: the payload carries no session id, so parallel "
+            f"publishers cannot be told apart; allowing {act}. Say so in the report."
+        )
+    ledger, ledger_reason = _ledger_read()
+    if ledger is None:
+        return (
+            f"publish-act-guard: UNCHECKED: {ledger_reason}; allowing {act}. "
+            f"Say so in the report."
+        )
+    other = other_recent_publisher(ledger, session, agent, stamp)
+    if other is not None:
+        other_agent, other_stamp = other
+        return BLOCK_MESSAGE.format(
+            act=act,
+            other=other_agent,
+            minutes=int((stamp - other_stamp) // 60),
+            window=PARALLEL_WINDOW_SECONDS // 60,
+            skill=ROUTING_SKILL,
+        )
+    _ledger_record(ledger, session, agent, stamp)
+    return _silent(f"first publishing subagent in session {session}; {act} may proceed")

engine/hooks/publish-act-guard/tests/test_detect.py

@@ -7,10 +7,12 @@
 import detect
 
 
-def payload(command, *, subagent=True, tool="Bash"):
+def payload(command, *, subagent=True, tool="Bash", agent="a4b7b959ce73515c9", session="s-1"):
     data = {"tool_name": tool, "tool_input": {"command": command}}
+    if session is not None:
+        data["session_id"] = session
     if subagent:
-        data["subagent_id"] = "a4b7b959ce73515c9"
+        data["subagent_id"] = agent
     return data
 
 
@@ -53,12 +55,61 @@ def test_dry_run_push_is_not_an_act(self):
         self.assertIsNone(detect.publishing_act("git push --dry-run origin HEAD"))
 
 
-class Decide(unittest.TestCase):
-    def test_blocks_subagent_publish_when_owner_is_live(self):
-        refusal = detect.decide(payload("git push -u origin HEAD"), runner=LIVE)
+class PublisherLedgerCase(unittest.TestCase):
+    def setUp(self):
+        self.ledger = detect.PUBLISHERS_PATH if hasattr(detect, "PUBLISHERS_PATH") else None
+        self.path = os.path.join(os.environ.get("TMPDIR", "/tmp"), f"publish-act-guard-test-{os.getpid()}.json")
+        if os.path.exists(self.path):
+            os.unlink(self.path)
+        detect.PUBLISHERS_PATH = self.path
+
+    def tearDown(self):
+        if os.path.exists(self.path):
+            os.unlink(self.path)
+        if self.ledger is not None:
+            detect.PUBLISHERS_PATH = self.ledger
+
+
+class Decide(PublisherLedgerCase):
+    def test_first_subagent_publish_in_a_session_is_allowed_when_owner_is_live(self):
+        self.assertIsNone(detect.decide(payload("git push -u origin HEAD"), runner=LIVE))
+
+    def test_same_subagent_may_publish_again(self):
+        self.assertIsNone(detect.decide(payload("git push -u origin HEAD"), runner=LIVE))
+        self.assertIsNone(detect.decide(payload("gh api -X PATCH repos/o/r/pulls/1 -f title=x"), runner=LIVE))
+
+    def test_blocks_a_second_subagent_publishing_in_the_same_session(self):
+        self.assertIsNone(detect.decide(payload("git push -u origin HEAD", agent="first"), runner=LIVE))
+        refusal = detect.decide(payload("mergify stack push", agent="second"), runner=LIVE)
         self.assertIsNotNone(refusal)
-        self.assertIn("git push", refusal)
-        self.assertIn("publish-act-guard", refusal)
+        self.assertIn("mergify stack push", refusal)
+        self.assertIn("first", refusal)
+        self.assertIn("parent session", refusal)
+
+    def test_second_subagent_is_allowed_once_the_window_has_passed(self):
+        self.assertIsNone(detect.decide(payload("git push", agent="first"), runner=LIVE, now=1000.0))
+        later = 1000.0 + detect.PARALLEL_WINDOW_SECONDS + 1
+        self.assertIsNone(detect.decide(payload("git push", agent="second"), runner=LIVE, now=later))
+
+    def test_other_sessions_do_not_count(self):
+        self.assertIsNone(detect.decide(payload("git push", agent="first", session="s-1"), runner=LIVE))
+        self.assertIsNone(detect.decide(payload("git push", agent="second", session="s-2"), runner=LIVE))
+
+    def test_parallel_publishers_are_allowed_when_no_live_owner(self):
+        self.assertIsNone(detect.decide(payload("git push", agent="first"), runner=DOWN))
+        self.assertIsNone(detect.decide(payload("git push", agent="second"), runner=DOWN))
+
+    def test_unreadable_ledger_reports_and_allows(self):
+        with open(self.path, "w", encoding="utf-8") as handle:
+            handle.write("{not json")
+        refusal = detect.decide(payload("git push", agent="second"), runner=LIVE)
+        self.assertIsNotNone(refusal)
+        self.assertIn("UNCHECKED", refusal)
+
+    def test_missing_session_id_reports_and_allows(self):
+        refusal = detect.decide(payload("git push", session=None), runner=LIVE)
+        self.assertIsNotNone(refusal)
+        self.assertIn("UNCHECKED", refusal)
 
     def test_allows_when_no_live_owner(self):
         self.assertIsNone(detect.decide(payload("git push -u origin HEAD"), runner=DOWN))
@@ -83,9 +134,10 @@ def test_unreadable_liveness_reports_and_allows(self):
         self.assertIn("mergify stack push", refusal)
 
     def test_prompt_wording_cannot_trigger_or_clear_the_gate(self):
-        wording = payload("echo 'this subagent is carrying commits and will push a PR'")
+        self.assertIsNone(detect.decide(payload("git push", agent="first"), runner=LIVE))
+        wording = payload("echo 'this subagent is carrying commits and will push a PR'", agent="second")
         self.assertIsNone(detect.decide(wording, runner=LIVE))
-        real = payload("git push -u origin HEAD")
+        real = payload("git push -u origin HEAD", agent="second")
         self.assertIsNotNone(detect.decide(real, runner=LIVE))