Merge pull request #260 from AIOSAI/system/devpulse-s90-fix-cross-project-email-sender-detection-calle

feat(system): S90: fix cross-project email — sender detection + caller registry target resolution

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

@@ -33,6 +33,45 @@ _REPO_ROOT = find_repo_root()
 _INBOX_LOCK = None
 
 
+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).
+    """
+    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 {}
+
+
 def _auto_register_contact(email: str, branch_path: Path, inbox_file: Path) -> None:
     """Auto-register a recipient in the contacts address book after successful delivery.
 
@@ -239,10 +278,17 @@ def deliver_email_to_branch(
             if not matched:
                 return False, f"Could not resolve path to email: {to_branch}"
 
-    # Map email address to branch path
+    # Map email address to branch path (AIPass registry + caller's project registry)
     all_branches = get_all_branches()
     branches = {b["email"]: b["path"] for b in all_branches}
 
+    if to_branch not in branches:
+        # Check caller's project registry for local branches (e.g. @strategy in Vera Studio)
+        caller_cwd = os.environ.get("AIPASS_CALLER_CWD", "")
+        if caller_cwd:
+            caller_branches = _load_caller_project_branches(caller_cwd)
+            branches.update(caller_branches)
+
     if to_branch not in branches:
         error_msg = f"Unknown branch email: {to_branch} (available: {len(branches)} branches)"
         return False, error_msg

src/aipass/ai_mail/apps/handlers/users/branch_detection.py

@@ -136,6 +136,26 @@ def detect_branch_from_pwd() -> Optional[Dict]:
             branch_info = _lookup_branch_by_name(caller_branch)
             if branch_info:
                 return branch_info
+            # Last resort: synthesize from env vars (external project, first contact)
+            caller_cwd = os.environ.get("AIPASS_CALLER_CWD", "")
+            if caller_cwd:
+                cwd_path = Path(caller_cwd)
+                name_key = caller_branch.lstrip("@").lower()
+                mailbox = cwd_path / ".ai_mail.local"
+                if not mailbox.exists():
+                    # Walk up to find project-level mailbox
+                    for parent in [cwd_path] + list(cwd_path.parents):
+                        candidate = parent / ".ai_mail.local"
+                        if candidate.exists():
+                            mailbox = candidate
+                            break
+                return {
+                    "name": name_key,
+                    "path": str(cwd_path),
+                    "email": f"@{name_key}",
+                    "status": "active",
+                    "type": "external",
+                }
 
         # Fallback: use caller's CWD for path-based detection (local only)
         caller_cwd = os.environ.get("AIPASS_CALLER_CWD")

src/aipass/ai_mail/tests/test_send_identity.py

@@ -240,18 +240,21 @@ class TestDetectBranchFromPwd:
         result = detect_branch_from_pwd()
         assert result is None
 
-    def test_invalid_branch_name_falls_through(self, clean_env, tmp_path, temp_registry):
-        """Nonexistent branch in AIPASS_CALLER_BRANCH should fall through to CWD.
+    def test_unknown_branch_synthesizes_from_env(self, clean_env, tmp_path, temp_registry):
+        """Unknown branch in AIPASS_CALLER_BRANCH synthesizes info from env vars.
 
-        When env branch name is not in registry, function falls back to
-        AIPASS_CALLER_CWD. If that also fails, result is None.
+        When env branch name is not in registry or contacts, function
+        synthesizes a minimal branch_info dict from the env vars so
+        external project callers can still send mail.
         """
         registry_path, _ = temp_registry
         with patch("aipass.ai_mail.apps.handlers.users.branch_detection.BRANCH_REGISTRY_PATH", registry_path):
-            os.environ["AIPASS_CALLER_BRANCH"] = "totally_fake_branch_xyz"
-            os.environ["AIPASS_CALLER_CWD"] = str(tmp_path)  # Also not a branch
+            os.environ["AIPASS_CALLER_BRANCH"] = "VERA-STUDIO"
+            os.environ["AIPASS_CALLER_CWD"] = str(tmp_path)
             result = detect_branch_from_pwd()
-            assert result is None
+            assert result is not None
+            assert result["name"].lower() == "vera-studio"
+            assert result["email"].lower() == "@vera-studio"
 
 
 # ─── find_branch_root() tests ───────────────────────────

src/aipass/drone/apps/handlers/router_handler.py

@@ -42,7 +42,11 @@ def find_entry_point(branch_path: str, branch_name: str) -> Path:
 
 
 def detect_caller_branch_name(cwd: Path) -> str | None:
-    """Walk up from cwd to find .trinity/passport.json and extract branch name."""
+    """Walk up from cwd to find .trinity/passport.json and extract branch name.
+
+    Falls back to project name from registry if no passport found (external projects
+    calling from project root without a branch-level CWD).
+    """
     current = cwd.resolve()
     for _ in range(10):
         passport = current / ".trinity" / "passport.json"
@@ -64,6 +68,24 @@ def detect_caller_branch_name(cwd: Path) -> str | None:
         if parent == current:
             break
         current = parent
+
+    # Fallback: detect project name from registry file (external projects at root)
+    current = 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)
+                meta = data.get("metadata", {})
+                project_name = meta.get("project_name") or meta.get("name")
+                if project_name:
+                    return project_name.lower().replace(" ", "-")
+            except Exception as exc:
+                logger.info("Failed to read registry %s: %s", reg_file, exc)
+        parent = current.parent
+        if parent == current:
+            break
+        current = parent
     return None