diff --git a/src/aipass/ai_mail/apps/handlers/email/delivery.py b/src/aipass/ai_mail/apps/handlers/email/delivery.py index 18c5bafe..973a19a5 100644 --- a/src/aipass/ai_mail/apps/handlers/email/delivery.py +++ b/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 diff --git a/src/aipass/ai_mail/apps/handlers/users/branch_detection.py b/src/aipass/ai_mail/apps/handlers/users/branch_detection.py index ff54fd8b..bcc72033 100644 --- a/src/aipass/ai_mail/apps/handlers/users/branch_detection.py +++ b/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") diff --git a/src/aipass/ai_mail/tests/test_send_identity.py b/src/aipass/ai_mail/tests/test_send_identity.py index 63df4168..d3e0ecc8 100755 --- a/src/aipass/ai_mail/tests/test_send_identity.py +++ b/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 ─────────────────────────── diff --git a/src/aipass/drone/apps/handlers/router_handler.py b/src/aipass/drone/apps/handlers/router_handler.py index ee1c77ab..5841ac96 100644 --- a/src/aipass/drone/apps/handlers/router_handler.py +++ b/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