fix(hooks): presence_gate keys per-branch via hook_data cwd; engine propagates block exit code (FPLAN-0289 P1)

Activation fixes for the single-session presence gate. Two bugs blocked it,
both caught by live testing after all units were green:

1) Wrong branch key. presence_gate used Path.cwd().name, but under the Claude
   Code bridge the hook process cwd is the project root, so every session keyed
   to "AIPass": the gate never enforced one-live-session-per-branch and would
   have rejected sessions project-globally (any 2nd interactive session in any
   branch). Now _resolve_branch(hook_data) reads the event payload's cwd (the
   real session dir) and walks up to the branch root (.trinity/ or apps/),
   mirroring branch_loader. Applied in handle() and handle_stop().

2) Block never reached Claude Code. engine.dispatch() returned only stdout, so
   the bridge could not surface a non-zero exit. dispatch() now returns
   (stdout, exit_code) and the bridge exits with it on a block. Pre-existing gap
   affecting every block hook on every event; now fixed engine-wide. An
   intentional block (exit 2 + {"decision":"block"}) propagates; a crashing hook
   (exit 2, non-JSON stdout) is logged and falls through, so the gate fails open.

Proven: 110 hooks unit tests pass (6 new for branch resolution); seedgo @hooks
100%, no type errors. Live bridge end-to-end (real live holder + real bridge):
duplicate into a held branch -> exit 2 + block reason naming the branch; a
different free branch -> exit 0 (per-branch isolation intact). Gate remains
dormant: not yet wired into provider settings.

Design: DPLAN-0225 / FPLAN-0289 P1 activation. Build by @hooks.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CqoxFdbDMirzkQ5kjRVVos

.aipass/hooks.json

@@ -3,6 +3,11 @@
   "hooks_enabled": true,
 
   "UserPromptSubmit": {
+    "presence_gate": {
+      "enabled": true,
+      "handler": "aipass.hooks.apps.handlers.security.presence_gate.handle",
+      "matcher": ""
+    },
     "identity_injector": {
       "enabled": true,
       "handler": "aipass.hooks.apps.handlers.prompt.identity.handle",
@@ -98,6 +103,11 @@
       "handler": "aipass.hooks.apps.handlers.notification.telegram_response.handle",
       "matcher": "",
       "timeout": 30
+    },
+    "presence_release": {
+      "enabled": true,
+      "handler": "aipass.hooks.apps.handlers.security.presence_gate.handle_stop",
+      "matcher": ""
     }
   },
 

src/aipass/hooks/apps/handlers/bridges/claude.py

@@ -53,9 +53,11 @@ def main() -> None:
         hook_def = full_config.get(event_type, {}).get(hook_filter, {})
         config = {"hooks_enabled": True, event_type: {hook_filter: hook_def}}
 
-    output = dispatch(event_type, stdin_data, config)
+    output, exit_code = dispatch(event_type, stdin_data, config)
     if output:
         sys.stdout.write(output)
+    if exit_code:
+        sys.exit(exit_code)
 
 
 if __name__ == "__main__":

src/aipass/hooks/apps/handlers/security/presence_gate.py

@@ -29,6 +29,19 @@ _ALLOW = {"exit_code": 0, "stdout": ""}
 _NON_BLOCKING_SESSION_TYPES = frozenset({"dispatched", "daemon"})
 
 
+def _resolve_branch(hook_data: dict) -> str:
+    """Resolve the branch name from hook_data's cwd (session dir, not process cwd)."""
+    cwd = hook_data.get("cwd", "") or str(Path.cwd())
+    search = Path(cwd).resolve()
+    while search.parent != search:
+        if (search / ".trinity").is_dir() or (search / "apps").is_dir():
+            return search.name
+        if (search / "pyproject.toml").exists() or (search / ".git").is_dir():
+            break
+        search = search.parent
+    return Path(cwd).name
+
+
 def handle(hook_data: dict) -> dict:
     """UserPromptSubmit gate — enforce one live session per branch.
 
@@ -47,7 +60,7 @@ def handle(hook_data: dict) -> dict:
         if session_type in _NON_BLOCKING_SESSION_TYPES:
             return _ALLOW
 
-        branch = Path.cwd().name
+        branch = _resolve_branch(hook_data)
         session_id = os.environ.get("CLAUDE_CODE_SESSION_ID", "")
 
         presence = importlib.import_module("aipass.hooks.apps.modules.presence")
@@ -83,7 +96,7 @@ def handle_stop(hook_data: dict) -> dict:
     Returns:
         Result dict (always allows — Stop is informational).
     """
-    branch = Path.cwd().name
+    branch = _resolve_branch(hook_data)
     try:
         presence = importlib.import_module("aipass.hooks.apps.modules.presence")
         released = presence.release(branch)

src/aipass/hooks/apps/modules/engine.py

@@ -89,17 +89,17 @@ def _matches(matcher: str, value: str) -> bool:
     return value in matcher.split("|")
 
 
-def dispatch(event_type: str, stdin_data: str, config: dict) -> str:
+def dispatch(event_type: str, stdin_data: str, config: dict) -> tuple[str, int]:
-    """Core dispatch — run hooks for event, return merged stdout."""
+    """Core dispatch — run hooks for event, return (merged_stdout, exit_code)."""
     if not config.get("hooks_enabled", True):
         logger.info("[HOOKS] all hooks disabled")
         _log({"ts": time.time(), "event": event_type, "action": "all_hooks_disabled"})
-        return ""
+        return "", 0
 
     event_hooks = config.get(event_type, {})
     if not event_hooks:
         _log({"ts": time.time(), "event": event_type, "action": "no_hooks_configured"})
-        return ""
+        return "", 0
 
     match_value = ""
     parsed = {}
@@ -196,7 +196,7 @@ def dispatch(event_type: str, stdin_data: str, config: dict) -> str:
                         "total_ms": round(total_ms, 1),
                     }
                 )
-                return result["stdout"]
+                return result["stdout"], 2
 
             logger.error(
                 "[HOOKS] %s.%s CRASHED exit=2: %s",
@@ -229,7 +229,7 @@ def dispatch(event_type: str, stdin_data: str, config: dict) -> str:
         }
     )
 
-    return "\n".join(outputs)
+    return "\n".join(outputs), 0
 
 
 # =============================================================================

src/aipass/hooks/tests/test_engine.py

@@ -107,13 +107,15 @@ class TestDispatch:
         config = {"hooks_enabled": False}
         with patch("aipass.hooks.apps.modules.engine._log"):
             result = dispatch("UserPromptSubmit", "{}", config)
-        assert result == ""
+        assert result[0] == ""
+        assert result[1] == 0
 
     def test_no_hooks_for_event_returns_empty(self, mock_logger):
         config = {"hooks_enabled": True}
         with patch("aipass.hooks.apps.modules.engine._log"):
             result = dispatch("UnknownEvent", "{}", config)
-        assert result == ""
+        assert result[0] == ""
+        assert result[1] == 0
 
     def test_disabled_hook_skipped(self, mock_logger):
         config = {
@@ -130,7 +132,8 @@ class TestDispatch:
             with patch("aipass.hooks.apps.modules.engine._run_hook") as mock_run:
                 result = dispatch("PreToolUse", '{"tool_name":"Edit"}', config)
         mock_run.assert_not_called()
-        assert result == ""
+        assert result[0] == ""
+        assert result[1] == 0
 
     def test_matcher_filters_hooks(self, mock_logger):
         config = {
@@ -165,7 +168,8 @@ class TestDispatch:
                 mock_run.return_value = {"exit_code": 0, "stdout": "edit_output", "stderr": "", "elapsed_ms": 10}
                 result = dispatch("PreToolUse", '{"tool_name":"Edit"}', config)
         mock_run.assert_called_once()
-        assert "edit_output" in result
+        assert "edit_output" in result[0]
+        assert result[1] == 0
 
     def test_multiple_hooks_concatenate_output(self, mock_logger):
         config = {
@@ -182,8 +186,9 @@ class TestDispatch:
                     {"exit_code": 0, "stdout": "output_B", "stderr": "", "elapsed_ms": 10},
                 ]
                 result = dispatch("UserPromptSubmit", '{"user_prompt":"test"}', config)
-        assert "output_A" in result
+        assert "output_A" in result[0]
-        assert "output_B" in result
+        assert "output_B" in result[0]
+        assert result[1] == 0
 
     def test_exit2_with_block_json_bails(self, mock_logger):
         config = {
@@ -199,8 +204,9 @@ class TestDispatch:
                 mock_run.return_value = {"exit_code": 2, "stdout": block_json, "stderr": "", "elapsed_ms": 10}
                 result = dispatch("PreToolUse", '{"tool_name":"Edit"}', config)
         assert mock_run.call_count == 1
-        parsed = json.loads(result)
+        parsed = json.loads(result[0])
         assert parsed["decision"] == "block"
+        assert result[1] == 2
 
     def test_exit2_without_json_is_crash_not_block(self, mock_logger):
         config = {
@@ -218,7 +224,8 @@ class TestDispatch:
                 ]
                 result = dispatch("PreToolUse", '{"tool_name":"Edit"}', config)
         assert mock_run.call_count == 2
-        assert "survived" in result
+        assert "survived" in result[0]
+        assert result[1] == 0
 
     def test_hook_with_custom_timeout(self, mock_logger):
         config = {
@@ -261,7 +268,8 @@ class TestDispatch:
             with patch("aipass.hooks.apps.modules.engine._run_hook") as mock_run:
                 mock_run.return_value = {"exit_code": 0, "stdout": "ok", "stderr": "", "elapsed_ms": 5}
                 result = dispatch("UserPromptSubmit", "not json at all{{{", config)
-        assert "ok" in result
+        assert "ok" in result[0]
+        assert result[1] == 0
 
 
 class TestFindProjectConfig:
@@ -448,7 +456,7 @@ class TestExceptionContracts:
     def test_dispatch_with_none_config_event_returns_empty(self, mock_logger):
         with patch("aipass.hooks.apps.modules.engine._log"):
             result = dispatch("Stop", "{}", {"hooks_enabled": True})
-        assert result == ""
+        assert result == ("", 0)
 
     def test_run_hook_timeout_returns_negative_exit(self, mock_subprocess, mock_logger):
         mock_subprocess.side_effect = subprocess.TimeoutExpired("cmd", 30)
@@ -508,10 +516,13 @@ class TestInitProvisioning:
         assert len(lines) == 2
         assert json.loads(lines[0])["existing"] is True
 
-    def test_dispatch_returns_string(self, mock_logger):
+    def test_dispatch_returns_tuple(self, mock_logger):
         with patch("aipass.hooks.apps.modules.engine._log"):
             result = dispatch("Stop", "{}", {"hooks_enabled": True})
-        assert isinstance(result, str)
+        assert isinstance(result, tuple)
+        assert isinstance(result[0], str)
+        assert isinstance(result[1], int)
+        assert result == ("", 0)
 
 
 class TestConftest:
@@ -666,7 +677,8 @@ class TestErrorResilienceExtended:
             with patch("aipass.hooks.apps.modules.engine._run_hook") as mock_run:
                 mock_run.return_value = {"exit_code": 0, "stdout": "ok", "stderr": "", "elapsed_ms": 5}
                 result = dispatch("Stop", "", config)
-        assert "ok" in result
+        assert "ok" in result[0]
+        assert result[1] == 0
 
     def test_config_with_nonexistent_dir(self, temp_test_dir, mock_logger):
         nonexistent = temp_test_dir / "does_not_exist"

src/aipass/hooks/tests/test_presence_gate.py

@@ -27,6 +27,29 @@ _OCCUPIED_MOCK = _make_presence_mock(
 )
 
 
+class TestResolveBranch:
+    def test_uses_hook_data_cwd(self, tmp_path):
+        branch_dir = tmp_path / "devpulse"
+        branch_dir.mkdir()
+        (branch_dir / ".trinity").mkdir()
+        assert presence_gate._resolve_branch({"cwd": str(branch_dir)}) == "devpulse"
+
+    def test_walks_up_to_branch_root(self, tmp_path):
+        branch_dir = tmp_path / "hooks"
+        (branch_dir / "apps" / "modules").mkdir(parents=True)
+        sub = branch_dir / "apps" / "modules"
+        assert presence_gate._resolve_branch({"cwd": str(sub)}) == "hooks"
+
+    def test_stops_at_repo_root(self, tmp_path):
+        (tmp_path / ".git").mkdir()
+        assert presence_gate._resolve_branch({"cwd": str(tmp_path)}) == tmp_path.name
+
+    def test_fallback_to_path_cwd_when_no_cwd_in_hook_data(self):
+        result = presence_gate._resolve_branch({})
+        assert isinstance(result, str)
+        assert len(result) > 0
+
+
 class TestHandle:
     def test_first_prompt_acquired(self):
         with patch.dict(os.environ, {"AIPASS_SESSION_TYPE": "interactive"}, clear=True):
@@ -72,17 +95,28 @@ class TestHandle:
                 result = presence_gate.handle({})
         assert result["exit_code"] == 0
 
-    def test_block_message_includes_branch(self):
+    def test_block_message_includes_branch(self, tmp_path):
-        mock_cwd = MagicMock()
+        branch_dir = tmp_path / "devpulse"
-        mock_cwd.name = "devpulse"
+        branch_dir.mkdir()
+        (branch_dir / ".trinity").mkdir()
         with patch.dict(os.environ, {"AIPASS_SESSION_TYPE": "interactive"}, clear=True):
             with patch("importlib.import_module", return_value=_OCCUPIED_MOCK):
-                with patch.object(presence_gate.Path, "cwd", return_value=mock_cwd):
+                result = presence_gate.handle({"cwd": str(branch_dir)})
-                    result = presence_gate.handle({})
         parsed = json.loads(result["stdout"])
         assert "devpulse" in parsed["reason"]
         assert "attach" in parsed["reason"].lower()
 
+    def test_branch_resolved_from_hook_data_cwd(self, tmp_path):
+        branch_dir = tmp_path / "api"
+        branch_dir.mkdir()
+        (branch_dir / ".trinity").mkdir()
+        mock = _make_presence_mock({"status": "ACQUIRED"})
+        with patch.dict(os.environ, {"AIPASS_SESSION_TYPE": "interactive"}, clear=True):
+            with patch("importlib.import_module", return_value=mock):
+                presence_gate.handle({"cwd": str(branch_dir)})
+        mock.claim.assert_called_once()
+        assert mock.claim.call_args[1]["branch"] == "api"
+
 
 class TestHandleStop:
     def test_stop_releases(self):
@@ -102,3 +136,12 @@ class TestHandleStop:
         with patch("importlib.import_module", side_effect=ImportError("no module")):
             result = presence_gate.handle_stop({})
         assert result["exit_code"] == 0
+
+    def test_stop_uses_hook_data_cwd(self, tmp_path):
+        branch_dir = tmp_path / "skills"
+        branch_dir.mkdir()
+        (branch_dir / ".trinity").mkdir()
+        mock = _make_presence_mock({"status": "ACQUIRED"})
+        with patch("importlib.import_module", return_value=mock):
+            presence_gate.handle_stop({"cwd": str(branch_dir)})
+        mock.release.assert_called_once_with("skills")