fix(#149): sign effort-drift POST + regenerate stale status hooks

src/pinky_daemon/agent_registry.py

@@ -113,20 +113,45 @@ def _verify_effort_hook_source() -> str:
 
 def _post_drift(agent: str, expected: str, actual: str, tool_name: str,
                 strict: bool, daemon_url: str) -> None:
+    import base64
+    import hashlib
+    import hmac
+    import time
     import urllib.request
 
+    # HMAC-sign exactly like the daemon's verify_internal_request expects:
+    # SHA256 over the newline-joined agent / METHOD / path / ts, base64url,
+    # with padding stripped.
+    # Prefer the per-agent key — an isolated agent has the global secret both
+    # withheld from its env (#639) and rejected by the daemon (#640), so an
+    # unsigned (or global-secret-signed) effort-drift POST 401s. With no secret
+    # at all there is nothing the daemon would accept, so skip silently.
+    secret = (
+        os.environ.get("PINKY_AGENT_KEY", "").strip()
+        or os.environ.get("PINKY_SESSION_SECRET", "").strip()
+    )
+    if not secret:
+        return
+
     path = f"/agents/{agent}/effort-drift"
     body = json.dumps({
         "expected": expected,
         "actual": actual,
         "tool_name": tool_name,
         "strict": bool(strict),
     }).encode()
+    ts = int(time.time())
+    sig_payload = f"{agent}\\nPOST\\n{path}\\n{ts}".encode()
+    sig = base64.urlsafe_b64encode(
+        hmac.new(secret.encode(), sig_payload, hashlib.sha256).digest()
+    ).decode().rstrip("=")
     req = urllib.request.Request(
         f"{daemon_url}{path}", data=body, method="POST",
     )
     req.add_header("Content-Type", "application/json")
     req.add_header("x-pinky-agent", agent)
+    req.add_header("x-pinky-timestamp", str(ts))
+    req.add_header("x-pinky-signature", sig)
     try:
         urllib.request.urlopen(req, timeout=2)
     except Exception:
@@ -1594,9 +1619,10 @@ def _setup_hooks(work_dir: Path, agent_name: str) -> None:
         and (since #429) effort-drift verification.
 
         Creates ``.claude/`` directory with hook scripts and settings.json.
-        Existing scripts are not overwritten; settings.json is idempotently
-        merged so the verify_effort hook can be added to agents whose
-        settings predate #429 without nuking their existing hooks.
+        PinkyBot-managed hook scripts are rewritten when their content changes
+        (so signing/semantics updates reach agents across releases); settings.json
+        is idempotently merged so the verify_effort hook can be added to agents
+        whose settings predate #429 without nuking their existing hooks.
         """
         # Re-validate even though ``register()`` already did. ``_setup_hooks``
         # writes hook scripts whose paths depend on ``agent_name``; explicit
@@ -1647,17 +1673,23 @@ def _setup_hooks(work_dir: Path, agent_name: str) -> None:
         tmux_post_tool_path = claude_dir / "hook_tmux_post_tool.py"
         tmux_stop_failure_path = claude_dir / "hook_tmux_stop_failure.py"
 
-        if not working_path.exists():
-            working_path.write_text(
-                hook_template.format(agent_name=agent_name, status="working")
-            )
-            _log(f"agent_registry: created hook_working.py for {agent_name}")
-
-        if not idle_path.exists():
-            idle_path.write_text(
-                hook_template.format(agent_name=agent_name, status="idle")
-            )
-            _log(f"agent_registry: created hook_idle.py for {agent_name}")
+        # Rewrite-on-change (not create-if-missing): an agent registered before
+        # per-agent-key signing (#623) keeps a stale hook that signs /status
+        # with only the global secret, which the daemon now rejects for
+        # isolated tenants (#640) → 401. Regenerating on change migrates those
+        # agents to the current template on the next ensure_workspace_hooks.
+        AgentRegistry._write_hook_if_changed(
+            hook_path=working_path,
+            new_source=hook_template.format(agent_name=agent_name, status="working"),
+            hook_filename="hook_working.py",
+            agent_name=agent_name,
+        )
+        AgentRegistry._write_hook_if_changed(
+            hook_path=idle_path,
+            new_source=hook_template.format(agent_name=agent_name, status="idle"),
+            hook_filename="hook_idle.py",
+            agent_name=agent_name,
+        )
 
         # #429: verify_effort hook — compares $CLAUDE_EFFORT (v2.1.133+) to
         # PINKY_EXPECTED_EFFORT (set by daemon at session start). On drift,
@@ -1670,9 +1702,9 @@ def _setup_hooks(work_dir: Path, agent_name: str) -> None:
         #
         # Task #93: PreToolUse + PostToolUse hooks for tmux tool-use tracking.
         #
-        # All five are ALWAYS rewritten (unlike hook_working / hook_idle which
-        # are left alone if present) — they're fully PinkyBot-managed and
-        # getting the latest semantics on disk matters across releases. The
+        # These are ALWAYS rewritten on change — like hook_working / hook_idle
+        # above, they're fully PinkyBot-managed and getting the latest semantics
+        # (e.g. per-agent-key signing) on disk matters across releases. The
         # tmux hooks are installed unconditionally; the daemon endpoint
         # returns ``ok: True, session: None`` for non-tmux runtimes, so each
         # is a cheap no-op for SDK / codex agents (one extra POST per turn).

tests/test_effort_verification.py

@@ -104,6 +104,7 @@ def _run_hook(
     strict: bool = False,
     daemon_url: str = "",
     stdin: str = "",
+    agent_key: str = "alpha-key",
 ) -> subprocess.CompletedProcess:
     env = {**os.environ}
     # Wipe any inherited CLAUDE_EFFORT so test env is hermetic
@@ -112,6 +113,13 @@ def _run_hook(
     env.pop("PINKY_AGENT_NAME", None)
     env.pop("PINKY_STRICT_EFFORT", None)
     env.pop("PINKY_DAEMON_URL", None)
+    # The hook now signs its drift POST and skips when no secret is present, so
+    # control the signing env explicitly: drop any inherited secrets, then set
+    # the per-agent key unless the test asks for the no-secret case.
+    env.pop("PINKY_AGENT_KEY", None)
+    env.pop("PINKY_SESSION_SECRET", None)
+    if agent_key:
+        env["PINKY_AGENT_KEY"] = agent_key
     if actual:
         env["CLAUDE_EFFORT"] = actual
     if expected:
@@ -245,6 +253,28 @@ def test_setup_creates_verify_effort_script(self, tmp_path):
         AgentRegistry._setup_hooks(tmp_path, "alpha")
         assert (tmp_path / ".claude" / "hook_verify_effort.py").exists()
 
+    def test_setup_rewrites_stale_status_hooks(self, tmp_path):
+        """Regression: a pre-existing hook_working.py that signs /status with
+        only the global secret (the old create-if-missing artifact) must be
+        rewritten to the per-agent-key template — otherwise isolated agents
+        401 on /status after the #640 global-secret rejection."""
+        claude_dir = tmp_path / ".claude"
+        claude_dir.mkdir()
+        stale = claude_dir / "hook_working.py"
+        stale.write_text(
+            "#!/usr/bin/env python3\n"
+            "import os\n"
+            "secret = os.environ['PINKY_SESSION_SECRET']  # global secret only\n"
+        )
+        AgentRegistry._setup_hooks(tmp_path, "alpha")
+        rewritten = stale.read_text()
+        # Now prefers the per-agent key, targets this agent, and signs.
+        assert "PINKY_AGENT_KEY" in rewritten
+        assert "/agents/alpha/status" in rewritten
+        assert "x-pinky-signature" in rewritten
+        # hook_idle.py is regenerated the same way.
+        assert "PINKY_AGENT_KEY" in (claude_dir / "hook_idle.py").read_text()
+
     def test_setup_settings_has_verify_hook(self, tmp_path):
         AgentRegistry._setup_hooks(tmp_path, "alpha")
         settings = json.loads(
@@ -459,6 +489,56 @@ def test_mismatch_warn_posts(self, hook_script, fake_daemon):
         assert msg["body"]["actual"] == "medium"
         assert msg["body"]["strict"] is False
 
+    def test_drift_post_is_signed_and_isolated_verifiable(
+        self, hook_script, fake_daemon,
+    ):
+        """The drift POST must carry an HMAC the daemon accepts for an ISOLATED
+        agent — i.e. signed with the per-agent key and verifiable with
+        ``allow_global_secret=False``. Regression guard for the unsigned
+        effort-drift request that 401'd isolated tenants on the Pi deploy.
+        """
+        from pinky_daemon.auth import verify_internal_request
+
+        port, captured = fake_daemon
+        proc = _run_hook(
+            hook_script,
+            expected="high", actual="medium", agent="alpha",
+            agent_key="alpha-per-agent-key",
+            daemon_url=f"http://127.0.0.1:{port}",
+        )
+        assert proc.returncode == 0
+        assert len(captured) == 1, captured
+        hdr = {k.lower(): v for k, v in captured[0]["headers"].items()}
+        assert hdr.get("x-pinky-agent") == "alpha"
+        assert hdr.get("x-pinky-timestamp")
+        assert hdr.get("x-pinky-signature")
+        # Accepted via the per-agent key even with the global secret DISALLOWED
+        # (isolated-tenant policy), and a wrong global secret must not help.
+        assert verify_internal_request(
+            "unrelated-global-secret",
+            agent_name="alpha",
+            method="POST",
+            path="/agents/alpha/effort-drift",
+            timestamp=hdr["x-pinky-timestamp"],
+            signature=hdr["x-pinky-signature"],
+            agent_key="alpha-per-agent-key",
+            allow_global_secret=False,
+        )
+
+    def test_drift_no_secret_skips_post(self, hook_script, fake_daemon):
+        """With neither PINKY_AGENT_KEY nor PINKY_SESSION_SECRET, the hook has
+        nothing the daemon would accept, so it must skip the POST entirely
+        rather than fire an unsigned request that 401s."""
+        port, captured = fake_daemon
+        proc = _run_hook(
+            hook_script,
+            expected="high", actual="medium", agent="alpha",
+            agent_key="",  # no signing secret in env
+            daemon_url=f"http://127.0.0.1:{port}",
+        )
+        assert proc.returncode == 0
+        assert captured == []
+
     def test_mismatch_strict_emits_block(self, hook_script, fake_daemon):
         port, captured = fake_daemon
         proc = _run_hook(