fix(wake): resolve_branch falls back to caller project registry for cross-project dispatch

src/aipass/ai_mail/.seedgo/bypass.json

@@ -28,7 +28,7 @@
     {
       "file": "apps/handlers/dispatch/wake.py",
       "standard": "deep_nesting",
-      "reason": "2 functions: _check_lock() depth 4 (lock validation with PID checks and age), _is_branch_occupied() depth 5 (process inspection, filters by session type — must check /proc)"
+      "reason": "3 functions: _check_lock() depth 4 (lock validation with PID checks and age), _is_branch_occupied() depth 5 (process inspection, filters by session type — must check /proc), resolve_branch() depth 4 (AIPass registry scan + caller-registry fallback for cross-project dispatch)"
     },
     {
       "file": "apps/handlers/email/send.py",
@@ -68,7 +68,7 @@
     {
       "file": "apps/handlers/email/delivery.py",
       "standard": "handlers",
-      "reason": "Imports json_utils.json_handler (load_json, save_json) — shared handler utility for JSON I/O, same-branch cross-handler import. Also imports registry.read.get_all_branches — consolidated from dual implementation per DPLAN-0036. Also lazy-imports email.contacts.register_contact inside _auto_register_contact() and _auto_register_sender() for post-delivery contact registration (DPLAN-0121 Phase 5)."
+      "reason": "Imports json_utils.json_handler (load_json, save_json) — shared handler utility for JSON I/O, same-branch cross-handler import. Also imports registry.read.get_all_branches — consolidated from dual implementation per DPLAN-0036. Also lazy-imports email.contacts.register_contact inside _auto_register_contact() and _auto_register_sender() for post-delivery contact registration (DPLAN-0121 Phase 5). Also lazy-imports registry.read.get_caller_project_branches inside _load_caller_project_branches() to delegate to shared cross-project registry implementation (issue #283)."
     },
     {
       "file": "apps/handlers/email/inbox_cleanup.py",
@@ -98,7 +98,7 @@
     {
       "file": "apps/handlers/dispatch/wake.py",
       "standard": "handlers",
-      "reason": "Imports notify.send_notification — same-branch cross-handler import for wake completion notifications."
+      "reason": "Imports notify.send_notification — same-branch cross-handler import for wake completion notifications. Lazy-imports registry.read.get_caller_project_branches inside resolve_branch() for cross-project branch resolution. Same cross-handler pattern as reply.py, delivery.py, and branch_detection.py."
     },
     {
       "file": "apps/handlers/dispatch/dispatch_monitor.py",
@@ -220,6 +220,11 @@
       "standard": "handlers",
       "reason": "Imports paths.find_repo_root — shared utility consolidated from 8 copies per DPLAN-0036."
     },
+    {
+      "file": "apps/handlers/registry/read.py",
+      "standard": "deep_nesting",
+      "reason": "get_caller_project_branches() depth 4 — walks directory tree with per-registry-file parsing, list/dict branch format detection, and path resolution. Same pattern as delivery.py _load_caller_project_branches()."
+    },
     {
       "file": "apps/handlers/email/format.py",
       "standard": "handlers",
@@ -289,6 +294,36 @@
       "file": "tests/test_identity.py",
       "standard": "encapsulation",
       "reason": "Unit tests must import handlers directly to exercise create_identity, read_identity. Module entry-point-only rule does not apply to tests."
+    },
+    {
+      "file": "tests/test_registry_read.py",
+      "standard": "architecture",
+      "reason": "Test file lives in tests/ directory — not subject to 3-layer app structure rule."
+    },
+    {
+      "file": "tests/test_registry_read.py",
+      "standard": "encapsulation",
+      "reason": "Unit tests must import handlers directly to exercise internal functions (_derive_email_from_branch_name, get_all_branches, get_branch_by_email, get_caller_project_branches). Module entry-point-only rule does not apply to tests."
+    },
+    {
+      "file": "tests/test_registry_read.py",
+      "standard": "documentation",
+      "reason": "Test fixture functions (registry_file) are private test infrastructure — docstring requirement does not apply to pytest fixtures."
+    },
+    {
+      "file": "tests/test_wake.py",
+      "standard": "architecture",
+      "reason": "Test file lives in tests/ directory — not subject to 3-layer app structure rule."
+    },
+    {
+      "file": "tests/test_wake.py",
+      "standard": "encapsulation",
+      "reason": "Unit tests must import handlers directly to exercise internal functions (_read_json, _check_lock, _check_pid_alive, resolve_branch, DispatchStatus, etc.). Module entry-point-only rule does not apply to tests."
+    },
+    {
+      "file": "tests/test_wake.py",
+      "standard": "documentation",
+      "reason": "Test helper functions (_fake_open_factory, _raise_process_lookup, repo_root fixture) are private test infrastructure — docstring requirement does not apply to test helpers."
     }
   ],
   "notes": {

src/aipass/ai_mail/apps/handlers/dispatch/wake.py

@@ -320,19 +320,39 @@ def _check_pid_alive(pid: int) -> bool:
 # ─── Branch Resolution ──────────────────────────────────
 
 def resolve_branch(branch_email: str) -> Optional[Tuple[Path, str]]:
-    """Resolve a branch email to its absolute filesystem path."""
+    """Resolve a branch email to its absolute filesystem path.
+
+    Checks the AIPass registry first, then falls back to the caller's
+    project registry via AIPASS_CALLER_CWD for cross-project dispatch.
+    """
     email = f"@{branch_email.lstrip('@').lower()}"
+
+    # Step 1: AIPass registry (local branches)
     registry = _read_json(BRANCH_REGISTRY)
-    if registry is None:
-        return None
-    for branch in registry.get("branches", []):
-        if branch.get("email", "").lower() == email:
-            path = Path(branch.get("path", ""))
-            if not path.is_absolute():
-                path = _REPO_ROOT / path
-            if path.exists():
-                return path, email
-            return None
+    if registry is not None:
+        for branch in registry.get("branches", []):
+            if branch.get("email", "").lower() == email:
+                path = Path(branch.get("path", ""))
+                if not path.is_absolute():
+                    path = _REPO_ROOT / path
+                if path.exists():
+                    return path, email
+                return None  # Found but path missing — definitive failure
+
+    # Step 2: Caller's project registry (cross-project dispatch)
+    caller_cwd = os.environ.get("AIPASS_CALLER_CWD", "")
+    if caller_cwd:
+        try:
+            from aipass.ai_mail.apps.handlers.registry.read import get_caller_project_branches
+            caller_branches = get_caller_project_branches(caller_cwd)
+            branch_path_str = caller_branches.get(email, "")
+            if branch_path_str:
+                branch_path = Path(branch_path_str)
+                if branch_path.exists():
+                    return branch_path, email
+        except Exception as e:
+            logger.warning("[wake] resolve_branch caller registry fallback failed: %s", e)
+
     return None
 
 

src/aipass/ai_mail/apps/handlers/email/delivery.py

@@ -36,40 +36,11 @@
 def _load_caller_project_branches(caller_cwd: str) -> Dict[str, str]:
     """Load branches from the caller's project registry.
 
-    Walks up from caller_cwd to find a *_REGISTRY.json file, then extracts
-    branch email→path mappings. Used when the target branch isn't in the
-    AIPass registry (e.g. @strategy in Vera Studio).
+    Delegates to registry.read.get_caller_project_branches — shared
+    implementation used by both delivery and wake for cross-project resolution.
     """
-    current = Path(caller_cwd).resolve()
-    for _ in range(10):
-        for reg_file in current.glob("*_REGISTRY.json"):
-            try:
-                with open(reg_file, "r", encoding="utf-8") as f:
-                    data = json.load(f)
-                result = {}
-                branches = data.get("branches", [])
-                if isinstance(branches, list):
-                    for b in branches:
-                        email = b.get("email", f"@{b.get('name', '').lower()}")
-                        path = b.get("path", "")
-                        if path and not Path(path).is_absolute():
-                            path = str((reg_file.parent / path).resolve())
-                        result[email] = path
-                elif isinstance(branches, dict):
-                    for name, info in branches.items():
-                        email = info.get("email", f"@{name}")
-                        path = info.get("path", "")
-                        if path and not Path(path).is_absolute():
-                            path = str((reg_file.parent / path).resolve())
-                        result[email] = path
-                return result
-            except Exception as exc:
-                logger.warning("Failed reading caller registry %s: %s", reg_file, exc)
-        parent = current.parent
-        if parent == current:
-            break
-        current = parent
-    return {}
+    from aipass.ai_mail.apps.handlers.registry.read import get_caller_project_branches
+    return get_caller_project_branches(caller_cwd)
 
 
 def _auto_register_contact(email: str, branch_path: Path, inbox_file: Path) -> None:

src/aipass/ai_mail/apps/handlers/registry/read.py

@@ -21,6 +21,7 @@
 """
 
 import json
+from pathlib import Path
 from typing import List, Dict, Optional
 
 from aipass.prax.apps.modules.logger import system_logger as logger
@@ -146,6 +147,57 @@ def get_branch_by_email(email: str) -> Optional[Dict]:
     return None
 
 
+def get_caller_project_branches(caller_cwd: str) -> Dict[str, str]:
+    """Load branch email→path mappings from the caller's project registry.
+
+    Walks up from caller_cwd to find a *_REGISTRY.json file (e.g.
+    VERA_REGISTRY.json), then extracts branch email→path mappings.
+    Used for cross-project dispatch when the target branch is not in
+    the AIPass registry.
+
+    Args:
+        caller_cwd: Working directory of the calling project (typically
+                    from AIPASS_CALLER_CWD env var).
+
+    Returns:
+        Dict mapping email address to absolute path string.
+        Empty dict if no registry found or on error.
+    """
+    current = Path(caller_cwd).resolve()
+    for _ in range(10):
+        for reg_file in current.glob("*_REGISTRY.json"):
+            try:
+                with open(reg_file, "r", encoding="utf-8") as f:
+                    data = json.load(f)
+                result: Dict[str, str] = {}
+                branches = data.get("branches", [])
+                if isinstance(branches, list):
+                    for b in branches:
+                        email = b.get("email", f"@{b.get('name', '').lower()}")
+                        path = b.get("path", "")
+                        if path and not Path(path).is_absolute():
+                            path = str((reg_file.parent / path).resolve())
+                        if email and path:
+                            result[email] = path
+                elif isinstance(branches, dict):
+                    for name, info in branches.items():
+                        email = info.get("email", f"@{name}")
+                        path = info.get("path", "")
+                        if path and not Path(path).is_absolute():
+                            path = str((reg_file.parent / path).resolve())
+                        if email and path:
+                            result[email] = path
+                if result:
+                    return result
+            except Exception as exc:
+                logger.warning("[registry] get_caller_project_branches: failed reading %s: %s", reg_file, exc)
+        parent = current.parent
+        if parent == current:
+            break
+        current = parent
+    return {}
+
+
 if __name__ == "__main__":
     from aipass.cli.apps.modules import console
     console.print("\n" + "="*70)

src/aipass/ai_mail/tests/test_registry_read.py

@@ -18,6 +18,7 @@
     _derive_email_from_branch_name,
     get_all_branches,
     get_branch_by_email,
+    get_caller_project_branches,
 )
 
 
@@ -148,3 +149,78 @@ def test_get_branch_by_email_not_found(registry_file):
     registry_file.write_text(json.dumps(SAMPLE_REGISTRY), encoding="utf-8")
     result = get_branch_by_email("@nonexistent")
     assert result is None
+
+
+# --- get_caller_project_branches() tests -----------------------------
+
+
+class TestGetCallerProjectBranches:
+    """Tests for get_caller_project_branches()."""
+
+    def test_finds_registry_in_cwd(self, tmp_path):
+        """Returns email->path mapping from a *_REGISTRY.json in caller_cwd."""
+        branch_path = tmp_path / "src" / "strategy"
+        branch_path.mkdir(parents=True)
+        registry = {
+            "branches": [
+                {"name": "STRATEGY", "email": "@strategy", "path": str(branch_path)}
+            ]
+        }
+        (tmp_path / "VERA_REGISTRY.json").write_text(json.dumps(registry), encoding="utf-8")
+        result = get_caller_project_branches(str(tmp_path))
+        assert result == {"@strategy": str(branch_path)}
+
+    def test_finds_registry_in_parent(self, tmp_path):
+        """Walks up from caller_cwd to find registry in parent."""
+        branch_path = tmp_path / "src" / "strategy"
+        branch_path.mkdir(parents=True)
+        registry = {
+            "branches": [
+                {"name": "STRATEGY", "email": "@strategy", "path": str(branch_path)}
+            ]
+        }
+        (tmp_path / "VERA_REGISTRY.json").write_text(json.dumps(registry), encoding="utf-8")
+        subdir = tmp_path / "src" / "strategy" / "apps"
+        subdir.mkdir(parents=True)
+        result = get_caller_project_branches(str(subdir))
+        assert result == {"@strategy": str(branch_path)}
+
+    def test_resolves_relative_paths(self, tmp_path):
+        """Resolves relative paths in registry relative to the registry file."""
+        branch_path = tmp_path / "src" / "strategy"
+        branch_path.mkdir(parents=True)
+        registry = {
+            "branches": [
+                {"name": "STRATEGY", "email": "@strategy", "path": "src/strategy"}
+            ]
+        }
+        (tmp_path / "VERA_REGISTRY.json").write_text(json.dumps(registry), encoding="utf-8")
+        result = get_caller_project_branches(str(tmp_path))
+        assert result == {"@strategy": str(branch_path)}
+
+    def test_handles_dict_format(self, tmp_path):
+        """Handles dict-format branches (AIPass format)."""
+        branch_path = tmp_path / "src" / "quality"
+        branch_path.mkdir(parents=True)
+        registry = {
+            "branches": {
+                "quality": {"email": "@quality", "path": str(branch_path)}
+            }
+        }
+        (tmp_path / "AIPASS_REGISTRY.json").write_text(json.dumps(registry), encoding="utf-8")
+        result = get_caller_project_branches(str(tmp_path))
+        assert result == {"@quality": str(branch_path)}
+
+    def test_returns_empty_when_no_registry(self, tmp_path):
+        """Returns empty dict when no *_REGISTRY.json exists."""
+        result = get_caller_project_branches(str(tmp_path))
+        assert result == {}
+
+    def test_skips_aipass_registry_name(self, tmp_path):
+        """Also works with AIPASS_REGISTRY.json -- doesn't skip it (delivery uses all)."""
+        branch_path = tmp_path / "branch"
+        branch_path.mkdir()
+        registry = {"branches": [{"name": "TEST", "email": "@test", "path": str(branch_path)}]}
+        (tmp_path / "AIPASS_REGISTRY.json").write_text(json.dumps(registry), encoding="utf-8")
+        result = get_caller_project_branches(str(tmp_path))
+        assert "@test" in result

src/aipass/ai_mail/tests/test_wake.py

@@ -508,6 +508,65 @@ def test_falls_back_to_name_when_not_found(self, monkeypatch, tmp_path):
         assert result == "claude"
 
 
+class TestResolveBranchCallerRegistry:
+    """resolve_branch() falls back to caller's project registry for cross-project wake."""
+
+    def test_resolves_from_caller_registry(self, tmp_path, monkeypatch):
+        """Branch not in AIPass registry is found via AIPASS_CALLER_CWD."""
+        # External branch path
+        branch_path = tmp_path / "src" / "strategy"
+        branch_path.mkdir(parents=True)
+        (branch_path / ".ai_mail.local").mkdir()
+
+        # External registry
+        registry = {
+            "branches": [
+                {"name": "STRATEGY", "email": "@strategy", "path": str(branch_path)}
+            ]
+        }
+        (tmp_path / "VERA_REGISTRY.json").write_text(json.dumps(registry), encoding="utf-8")
+
+        # AIPass registry has no @strategy
+        aipass_registry = tmp_path / "AIPASS_REGISTRY.json"
+        aipass_registry.write_text(json.dumps({"branches": []}), encoding="utf-8")
+        monkeypatch.setattr(wake_mod, "BRANCH_REGISTRY", aipass_registry)
+        monkeypatch.setenv("AIPASS_CALLER_CWD", str(tmp_path))
+
+        result = resolve_branch("@strategy")
+        assert result is not None
+        resolved_path, email = result
+        assert email == "@strategy"
+        assert resolved_path == branch_path
+
+    def test_aipass_registry_takes_precedence(self, tmp_path, monkeypatch):
+        """AIPass registry result is returned before checking caller registry."""
+        branch_path = tmp_path / "src" / "drone"
+        branch_path.mkdir(parents=True)
+
+        aipass_registry = tmp_path / "AIPASS_REGISTRY.json"
+        aipass_registry.write_text(json.dumps({
+            "branches": [{"name": "DRONE", "email": "@drone", "path": str(branch_path)}]
+        }), encoding="utf-8")
+        monkeypatch.setattr(wake_mod, "_REPO_ROOT", tmp_path)
+        monkeypatch.setattr(wake_mod, "BRANCH_REGISTRY", aipass_registry)
+        monkeypatch.delenv("AIPASS_CALLER_CWD", raising=False)
+
+        result = resolve_branch("@drone")
+        assert result is not None
+        assert result[1] == "@drone"
+
+    def test_returns_none_when_not_found_anywhere(self, tmp_path, monkeypatch):
+        """Returns None when branch is missing from both registries."""
+        aipass_registry = tmp_path / "AIPASS_REGISTRY.json"
+        aipass_registry.write_text(json.dumps({"branches": []}), encoding="utf-8")
+        monkeypatch.setattr(wake_mod, "BRANCH_REGISTRY", aipass_registry)
+        monkeypatch.setenv("AIPASS_CALLER_CWD", str(tmp_path))
+        # No *_REGISTRY.json in tmp_path other than the aipass one (which has no @nonexistent)
+
+        result = resolve_branch("@nonexistent")
+        assert result is None
+
+
 class TestWakeBranchSpawnEnv:
     """Ensure wake_branch() spawn_env includes ~/.local/bin for restricted-PATH envs."""