diff --git a/src/aipass/ai_mail/.seedgo/bypass.json b/src/aipass/ai_mail/.seedgo/bypass.json index 12f188ad..8c268bbf 100644 --- a/src/aipass/ai_mail/.seedgo/bypass.json +++ b/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": { diff --git a/src/aipass/ai_mail/apps/handlers/dispatch/wake.py b/src/aipass/ai_mail/apps/handlers/dispatch/wake.py index f7582b1c..46c62eaa 100644 --- a/src/aipass/ai_mail/apps/handlers/dispatch/wake.py +++ b/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 diff --git a/src/aipass/ai_mail/apps/handlers/email/delivery.py b/src/aipass/ai_mail/apps/handlers/email/delivery.py index 973a19a5..d99c82af 100644 --- a/src/aipass/ai_mail/apps/handlers/email/delivery.py +++ b/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: diff --git a/src/aipass/ai_mail/apps/handlers/registry/read.py b/src/aipass/ai_mail/apps/handlers/registry/read.py index 4a3597c2..21725401 100644 --- a/src/aipass/ai_mail/apps/handlers/registry/read.py +++ b/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) diff --git a/src/aipass/ai_mail/tests/test_registry_read.py b/src/aipass/ai_mail/tests/test_registry_read.py index 2464f3ce..9510436b 100644 --- a/src/aipass/ai_mail/tests/test_registry_read.py +++ b/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 diff --git a/src/aipass/ai_mail/tests/test_wake.py b/src/aipass/ai_mail/tests/test_wake.py index d58e0756..e9323ca1 100644 --- a/src/aipass/ai_mail/tests/test_wake.py +++ b/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."""