feat(system): S90: ai_mail contacts-based resolution (DPLAN-0121 Phase 5) — replaces env var branch detection

Co-Authored-By: @devpulse <devpulse@aipass>
This commit is contained in:
AIOSAI
2026-04-11 22:49:47 -07:00
co-authored by @devpulse
parent cb9c39f400
commit 46d80093d1
4 changed files with 141 additions and 3 deletions
+32 -2
View File
@@ -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."
"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)."
},
{
"file": "apps/handlers/email/inbox_cleanup.py",
@@ -233,7 +233,17 @@
{
"file": "apps/handlers/users/branch_detection.py",
"standard": "handlers",
"reason": "Imports paths.find_repo_root — shared utility consolidated from 8 copies per DPLAN-0036."
"reason": "Imports paths.find_repo_root — shared utility consolidated from 8 copies per DPLAN-0036. Also lazy-imports email.contacts.get_contact inside _get_contact_info() for contacts-first sender detection (DPLAN-0121 Phase 5)."
},
{
"file": "apps/handlers/email/contacts.py",
"standard": "handlers",
"reason": "Imports paths.find_repo_root — shared utility for CONTACTS_FILE path resolution. Same pattern as registry/read.py and other handlers per DPLAN-0036."
},
{
"file": "apps/handlers/email/identity.py",
"standard": "handlers",
"reason": "bootstrap_aipass_identities() lazy-imports registry.read.get_all_branches and paths.find_repo_root inside function body to avoid circular imports. Same cross-handler pattern as reply.py."
},
{
"file": "tests/test_delivery.py",
@@ -259,6 +269,26 @@
"file": "tests/test_send_identity.py",
"standard": "documentation",
"reason": "Test helper functions (_load_active_source, fixtures) are private/internal — docstring requirement does not apply to test infrastructure."
},
{
"file": "tests/test_contacts.py",
"standard": "architecture",
"reason": "Test file lives in tests/ directory — not subject to 3-layer app structure rule."
},
{
"file": "tests/test_contacts.py",
"standard": "encapsulation",
"reason": "Unit tests must import handlers directly to exercise internal functions (_load_contacts, get_contact, register_contact, all_contacts). Module entry-point-only rule does not apply to tests."
},
{
"file": "tests/test_identity.py",
"standard": "architecture",
"reason": "Test file lives in tests/ directory — not subject to 3-layer app structure rule."
},
{
"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."
}
],
"notes": {
@@ -33,6 +33,47 @@ _REPO_ROOT = find_repo_root()
_INBOX_LOCK = None
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.
Non-critical: failures are logged and silently ignored.
Args:
email: Recipient email address (e.g., '@devpulse').
branch_path: Resolved path to the branch root directory.
inbox_file: Path to the branch's inbox.json file.
"""
try:
from aipass.ai_mail.apps.handlers.email.contacts import register_contact
name_key = email.lstrip("@").lower()
register_contact(name_key, "AIPass", str(inbox_file))
except Exception as e:
logger.warning("[delivery] _auto_register_contact(%s) failed: %s", email, e)
def _auto_register_sender(branch_name: str, caller_cwd: str) -> None:
"""Auto-register a sender in contacts when called from an external project.
Walks up from caller_cwd to find .ai_mail.local/inbox.json.
Non-critical: failures are logged and silently ignored.
Args:
branch_name: Sender branch name or email (e.g., 'vera' or '@vera').
caller_cwd: Working directory of the calling project.
"""
try:
candidate = Path(caller_cwd)
for path in [candidate] + list(candidate.parents)[:5]:
inbox_file = path / ".ai_mail.local" / "inbox.json"
if inbox_file.exists():
from aipass.ai_mail.apps.handlers.email.contacts import register_contact
name_key = branch_name.lstrip("@").lower()
register_contact(name_key, "", str(inbox_file))
return
except Exception as e:
logger.warning("[delivery] _auto_register_sender(%s) failed: %s", branch_name, e)
def _get_inbox_lock():
"""Lazy import inbox_lock context manager."""
global _INBOX_LOCK
@@ -301,6 +342,15 @@ def deliver_email_to_branch(
logger.warning("[delivery] failed to acquire inbox lock for %s: %s", to_branch, e)
return False, f"Failed to acquire inbox lock: {e}"
# Auto-register recipient in contacts for future fast lookup
_auto_register_contact(to_branch, branch_path, inbox_file)
# Auto-register sender if external project called with AIPASS_CALLER_BRANCH
caller_branch = os.environ.get("AIPASS_CALLER_BRANCH", "")
caller_cwd = os.environ.get("AIPASS_CALLER_CWD", "")
if caller_branch and caller_cwd:
_auto_register_sender(caller_branch, caller_cwd)
# Send desktop notification for new email
_send_desktop_notification(email_data['from'], to_branch, email_data['subject'], email_data.get('message', ''))
@@ -31,6 +31,37 @@ from aipass.ai_mail.apps.handlers.paths import find_repo_root
BRANCH_REGISTRY_PATH = find_repo_root() / "AIPASS_REGISTRY.json"
def _get_contact_info(branch_name: str) -> Optional[Dict]:
"""Look up branch info from the contacts address book.
Fastest path for sender detection — works for external projects that
have previously registered via contacts, bypassing registry/CWD walk.
Args:
branch_name: Branch name or email (e.g., 'devpulse' or '@devpulse').
Returns:
Synthetic branch info dict compatible with registry format, or None.
"""
try:
from aipass.ai_mail.apps.handlers.email.contacts import get_contact
contact = get_contact(branch_name)
if not contact:
return None
inbox_path = Path(contact["inbox"])
branch_path = inbox_path.parent.parent # .ai_mail.local -> branch root
name_key = branch_name.lstrip("@").lower()
return {
"name": name_key.upper(),
"email": "@" + name_key,
"path": str(branch_path),
"project": contact.get("project", ""),
}
except Exception as e:
logger.warning("[identity] _get_contact_info(%s) failed: %s", branch_name, e)
return None
def _find_caller_registry() -> Optional[Path]:
"""Find the caller's AIPASS_REGISTRY.json by walking up from AIPASS_CALLER_CWD.
@@ -97,6 +128,11 @@ def detect_branch_from_pwd() -> Optional[Dict]:
# Primary: use explicit branch name passed by drone (works in Docker + local)
caller_branch = os.environ.get("AIPASS_CALLER_BRANCH")
if caller_branch:
# Try contacts first (fastest, works for external projects)
contact = _get_contact_info(caller_branch)
if contact:
return contact
# Fall back to registry lookup
branch_info = _lookup_branch_by_name(caller_branch)
if branch_info:
return branch_info
+23 -1
View File
@@ -119,7 +119,7 @@ def handle_command(command: str, args: List[str]) -> bool:
if command in ("--help", "-h"):
print_help()
return True
valid = ["send", "email", "inbox", "view", "close", "reply", "sent", "contacts", "read"]
valid = ["send", "email", "inbox", "view", "close", "reply", "sent", "contacts", "read", "register"]
if command not in valid:
return False
if args and args[0] in ['--help', '-h', 'help']:
@@ -131,6 +131,7 @@ def handle_command(command: str, args: List[str]) -> bool:
"inbox": handle_inbox, "view": handle_view,
"close": handle_close, "reply": handle_reply, "read": handle_view,
"sent": handle_sent, "contacts": handle_contacts,
"register": handle_register,
}
return dispatch[command](args)
@@ -471,6 +472,27 @@ def handle_contacts(args: List[str]) -> bool:
return True
def handle_register(args: List[str]) -> bool:
"""Register a branch in the contacts address book.
Usage: drone @ai_mail register @branch /path/to/inbox [project]
"""
json_handler.log_operation("register_contact_initiated", {"args": args})
if len(args) < 2:
error("Usage: register @branch /path/to/inbox [project]")
return True
branch_name = args[0].lstrip("@")
inbox_path = args[1]
project = args[2] if len(args) > 2 else ""
from aipass.ai_mail.apps.handlers.email.contacts import register_contact
ok = register_contact(branch_name, project, inbox_path)
if ok:
console.print(f"[green]Registered @{branch_name} -> {inbox_path}[/green]")
else:
error(f"Failed to register @{branch_name}")
return True
def print_introspection():
"""Display module introspection info."""
console.print()