Merge pull request #80 from AIOSAI/feat/ai_mail-seedgo-compliance-jsonhandler-and-intros

feat(ai_mail): seedgo compliance — json_handler and introspection updates

src/aipass/ai_mail/apps/handlers/central_writer.py

@@ -36,6 +36,8 @@ from pathlib import Path
 from datetime import datetime
 from typing import Dict, Any, List, Tuple
 
+from aipass.ai_mail.apps.handlers.json import json_handler
+
 
 # =============================================================================
 # CONSTANTS
@@ -290,6 +292,8 @@ def update_central() -> Dict[str, Any]:
         >>> print(stats["system_totals"]["total_unread"])
         5
     """
+    json_handler.log_operation("update_central", {"target": str(CENTRAL_FILE)})
+
     # Aggregate statistics from all branches
     branch_stats = aggregate_branch_stats()
 

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

@@ -32,6 +32,7 @@ from urllib.request import Request, urlopen
 from urllib.error import URLError
 
 from aipass.prax.apps.modules.logger import system_logger as logger
+from aipass.ai_mail.apps.handlers.json import json_handler
 
 
 def _find_repo_root() -> Path:
@@ -600,6 +601,8 @@ def run_daemon() -> None:
 
     Exits gracefully on SIGTERM/SIGINT or kill switch.
     """
+    json_handler.log_operation("run_daemon", {"pid": os.getpid()})
+
     if not _write_pid_file():
         return
 

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

@@ -28,6 +28,7 @@ import time
 from pathlib import Path
 
 from aipass.prax.apps.modules.logger import system_logger as logger
+from aipass.ai_mail.apps.handlers.json import json_handler
 
 
 def _send_bounce(branch_email: str, reason: str, sender: str,
@@ -98,6 +99,8 @@ def main():
         logger.warning("[monitor] No claude command after --")
         sys.exit(1)
 
+    json_handler.log_operation("dispatch_monitor_start", {"branch": branch_email, "sender": sender})
+
     # Open stderr log for claude output
     try:
         stderr_fh = open(stderr_log, 'a', encoding='utf-8')

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

@@ -19,6 +19,8 @@ from pathlib import Path
 from datetime import datetime
 from typing import Dict, Any, List, Optional
 
+from aipass.ai_mail.apps.handlers.json import json_handler
+
 # Dispatch log location (package-relative)
 _AI_MAIL_DIR = Path(__file__).resolve().parents[3]  # ai_mail/
 DISPATCH_LOG_FILE = _AI_MAIL_DIR / ".ai_mail.local" / "dispatch_log.json"
@@ -71,6 +73,8 @@ def log_dispatch(branch: str, pid: Optional[int], status: str, error_msg: Option
     Returns:
         True if logged successfully
     """
+    json_handler.log_operation("log_dispatch", {"branch": branch, "status": status})
+
     dispatches = load_dispatch_log()
 
     entry: Dict[str, Any] = {

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

@@ -25,6 +25,7 @@ from pathlib import Path
 from typing import Optional, Tuple, List
 
 from aipass.prax.apps.modules.logger import system_logger as logger
+from aipass.ai_mail.apps.handlers.json import json_handler
 
 
 def _find_repo_root() -> Path:
@@ -316,6 +317,8 @@ def wake_branch(branch_email: str, custom_message: Optional[str] = None,
     Returns:
         Tuple of (DispatchStatus with all steps, overall success bool)
     """
+    json_handler.log_operation("wake_branch", {"branch": branch_email, "fresh": fresh, "auto": auto})
+
     status = DispatchStatus()
 
     # Step 1: Pause check (auto-dispatch only)

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

@@ -17,6 +17,7 @@ from pathlib import Path
 from typing import List, Tuple, Callable, Optional
 
 from aipass.prax import logger
+from aipass.ai_mail.apps.handlers.json import json_handler
 
 
 def batch_close(
@@ -36,6 +37,7 @@ def batch_close(
         Tuple of (results_list, closed_count, failed_count)
         results_list contains (message_id, success, message) tuples
     """
+    json_handler.log_operation("batch_close", {"branch_path": str(branch_path), "count": len(message_ids)})
     batch_mode = len(message_ids) > 1
     results = []
     closed_count = 0

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

@@ -16,9 +16,10 @@ Independent handler - no module dependencies.
 import json
 from pathlib import Path
 from datetime import datetime
-from typing import Dict
+from typing import Dict, Optional
 
 from aipass.prax.apps.modules.logger import system_logger as logger
+from aipass.ai_mail.apps.handlers.json import json_handler
 
 # Lazy imports
 _append_footer = None
@@ -52,6 +53,7 @@ def create_email_file(to_branch: str, subject: str, message: str, user_info: Dic
     Returns:
         Path to created email file in sent folder
     """
+    json_handler.log_operation("create_email_file", {"to": to_branch, "subject": subject})
     timestamp = datetime.now()
     timestamp_str = timestamp.strftime(user_info.get("timestamp_format", "%Y-%m-%d %H:%M:%S"))
 
@@ -110,7 +112,7 @@ def _trigger_sent_purge(mailbox_path: Path) -> None:
         logger.warning("[create] _trigger_sent_purge() failed: %s", e)
 
 
-def load_email_file(email_file: Path) -> Dict | None:
+def load_email_file(email_file: Path) -> Optional[Dict]:
     """
     Load email data from file.
 
@@ -158,7 +160,7 @@ if __name__ == "__main__":
     c.print()
     c.print("FUNCTIONS PROVIDED:")
     c.print("  - create_email_file(to_branch, subject, message, user_info) -> Path")
-    c.print("  - load_email_file(email_file) -> Dict | None")
+    c.print("  - load_email_file(email_file) -> Optional[Dict]")
     c.print("  - sanitize_subject(subject, max_length) -> str")
     c.print()
     c.print("HANDLER CHARACTERISTICS:")

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

@@ -22,6 +22,7 @@ from datetime import datetime
 from typing import Dict, Optional
 
 from aipass.prax.apps.modules.logger import system_logger as logger
+from aipass.ai_mail.apps.handlers.json import json_handler
 
 # Lazy-loaded write_section reference
 _write_section = None
@@ -155,6 +156,7 @@ def push_dashboard_update(branch_path: Path) -> bool:
     Returns:
         True if update succeeded, False on any error (never raises)
     """
+    json_handler.log_operation("dashboard_sync", {"branch_path": str(branch_path)})
     try:
         branch_path = Path(branch_path)
 

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

@@ -21,6 +21,7 @@ from typing import Dict, Tuple, List, Optional, Callable
 
 from aipass.prax.apps.modules.logger import system_logger as logger
 from aipass.ai_mail.apps.handlers.json_utils.json_handler import load_json, save_json
+from aipass.ai_mail.apps.handlers.json import json_handler
 
 
 def _find_repo_root() -> Path:
@@ -242,6 +243,8 @@ def deliver_email_to_branch(
         Tuple of (success: bool, error_message: str)
         error_message is empty string if successful
     """
+    json_handler.log_operation("deliver_email", {"to": to_branch, "subject": email_data.get("subject", "")})
+
     # Handle path input from DRONE's @ resolution
     if to_branch.startswith('/'):
         branches_list = get_all_branches()

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

@@ -18,6 +18,7 @@ from datetime import datetime
 from typing import Dict, Any, Callable, Optional
 
 from aipass.prax import logger
+from aipass.ai_mail.apps.handlers.json import json_handler
 
 
 def build_error_report(to_branch: str, subject: str, error_msg: str) -> Dict[str, Any]:
@@ -75,6 +76,8 @@ def dispatch_send_error(
     Returns:
         True if error dispatched successfully, False otherwise.
     """
+    json_handler.log_operation("dispatch_send_error", {"to_branch": to_branch, "subject": subject})
+
     try:
         email_data = build_error_report(to_branch, subject, error_msg)
         deliver_fn("@drone", email_data)

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

@@ -14,6 +14,8 @@ Reminds branches of process steps: Seed audit, memory update, FPLAN close, confi
 Independent handler - no module dependencies.
 """
 
+from aipass.ai_mail.apps.handlers.json import json_handler
+
 # Standard footer for all outgoing emails
 STANDARD_FOOTER = """
 ---
@@ -47,6 +49,7 @@ def append_footer(message: str) -> str:
     Returns:
         Message with footer appended
     """
+    json_handler.log_operation("append_footer", {"message_length": len(message)})
     return message + get_footer()
 
 

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

@@ -18,6 +18,8 @@ from pathlib import Path
 from typing import Dict, Optional
 
 from aipass.prax.apps.modules.logger import system_logger as logger
+from aipass.ai_mail.apps.handlers.json import json_handler
+
 
 def _find_repo_root() -> Path:
     """Walk up from this file to find AIPASS_REGISTRY.json (repo root)."""
@@ -104,6 +106,7 @@ def format_email_header(email_data: Dict) -> str:
     Returns:
         Formatted header string
     """
+    json_handler.log_operation("format_email_header", {"subject": email_data.get("subject", "")})
     sender = format_sender_display(
         email_data.get('from_name', 'Unknown'),
         email_data.get('from', 'unknown')

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

@@ -8,6 +8,8 @@
 
 from pathlib import Path
 
+from aipass.ai_mail.apps.handlers.json import json_handler
+
 """
 Email Header Handler
 
@@ -68,6 +70,7 @@ def prepend_dispatch_header(message: str, no_memory_save: bool = False) -> str:
     Returns:
         Message with dispatch header prepended
     """
+    json_handler.log_operation("prepend_dispatch_header", {"no_memory_save": no_memory_save})
     return get_dispatch_header(no_memory_save=no_memory_save) + message
 
 

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

@@ -21,6 +21,7 @@ from datetime import datetime
 from typing import Dict, Tuple, Optional, Any
 
 from aipass.prax.apps.modules.logger import system_logger as logger
+from aipass.ai_mail.apps.handlers.json import json_handler
 
 
 # Lazy import for inbox file lock
@@ -45,7 +46,7 @@ def _get_console() -> Any:
 def _get_update_section() -> Any:
     """Lazy import update_section."""
     try:
-        from aipass.devpulse.apps.modules.dashboard import update_section
+        from aipass.devpulse.apps.modules.dashboard import update_section  # type: ignore[import-not-found]
         return update_section
     except ImportError:
         return None
@@ -154,6 +155,7 @@ def mark_read_and_archive(branch_path: Path, message_id: str) -> Tuple[bool, str
     Returns:
         Tuple of (success: bool, message: str)
     """
+    json_handler.log_operation("inbox_cleanup", {"branch_path": str(branch_path), "message_id": message_id})
     mailbox_path = branch_path / ".ai_mail.local"
     inbox_file = mailbox_path / "inbox.json"
 

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

@@ -23,6 +23,8 @@ import sys
 from pathlib import Path
 from contextlib import contextmanager
 
+from aipass.ai_mail.apps.handlers.json import json_handler
+
 # fcntl is POSIX-only (Linux/macOS). On Windows, use msvcrt for locking.
 if sys.platform == "win32":
     import msvcrt
@@ -49,6 +51,7 @@ def inbox_lock(inbox_file: Path):
     Raises:
         OSError: If lock cannot be acquired
     """
+    json_handler.log_operation("inbox_lock", {"inbox_file": str(inbox_file)})
     lock_file = inbox_file.parent / ".inbox.lock"
     lock_fd = None
 

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

@@ -18,6 +18,7 @@ from pathlib import Path
 from typing import Dict
 
 from aipass.prax.apps.modules.logger import system_logger as logger
+from aipass.ai_mail.apps.handlers.json import json_handler
 
 
 
@@ -34,6 +35,7 @@ def load_inbox(inbox_file: Path) -> Dict:
     Raises:
         Exception: If file cannot be read or parsed
     """
+    json_handler.log_operation("load_inbox", {"inbox_file": str(inbox_file)})
     if not inbox_file.exists():
         return {"messages": []}
 

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

@@ -16,6 +16,8 @@ Independent handler - no module or display dependencies.
 from pathlib import Path
 from typing import Dict, Optional, Any, Callable, Tuple
 
+from aipass.ai_mail.apps.handlers.json import json_handler
+
 
 def resolve_inbox_target(
     args_first: Optional[str],
@@ -40,6 +42,7 @@ def resolve_inbox_target(
             target_branch: str | None (the explicit target, or None for current)
             error: str | None (set when success is False)
     """
+    json_handler.log_operation("resolve_inbox_target", {"target": args_first})
     target_branch = None
     if args_first and args_first.startswith("@"):
         target_branch = args_first

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

@@ -27,6 +27,7 @@ from datetime import datetime
 from typing import Dict, List, Any
 
 from aipass.prax.apps.modules.logger import system_logger as logger
+from aipass.ai_mail.apps.handlers.json import json_handler
 
 # Purge configuration
 MAX_EMAILS = 10
@@ -275,6 +276,7 @@ def run_purge(mailbox_path: Path) -> Dict[str, Any]:
     Returns:
         Dict with combined results
     """
+    json_handler.log_operation("run_purge", {"mailbox_path": str(mailbox_path)})
     sent_result = purge_sent_folder(mailbox_path)
     deleted_result = purge_deleted_folder(mailbox_path)
 

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

@@ -19,6 +19,7 @@ from typing import Dict, Tuple, Optional
 from datetime import datetime
 
 from aipass.prax.apps.modules.logger import system_logger as logger
+from aipass.ai_mail.apps.handlers.json import json_handler
 
 # Services imported in __main__ only (handlers should not display)
 
@@ -70,6 +71,7 @@ def send_reply(
     Returns:
         Tuple of (success: bool, message: str, reply_id: str or None)
     """
+    json_handler.log_operation("send_reply", {"from_branch": str(from_branch_path), "reply_to": original_email.get("from", "unknown")})
     # Import here to avoid circular imports
     from aipass.ai_mail.apps.handlers.email.delivery import deliver_email_to_branch, get_all_branches
     from aipass.ai_mail.apps.handlers.email.inbox_cleanup import mark_as_closed_and_archive

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

@@ -18,7 +18,7 @@ from pathlib import Path
 from typing import Optional, Tuple, List, Dict, Any
 
 from aipass.prax import logger
-# logger imported from aipass.prax
+from aipass.ai_mail.apps.handlers.json import json_handler
 
 
 def resolve_sender_info(
@@ -41,6 +41,7 @@ def resolve_sender_info(
     Returns:
         Dict with email_address, display_name, mailbox_path, timestamp_format.
     """
+    json_handler.log_operation("resolve_sender", {"from_branch": from_branch})
     if from_branch:
         email_addr = f"@{from_branch.lstrip('@').lower()}"
         branch_info = get_branch_by_email_fn(email_addr)
@@ -81,13 +82,14 @@ def send_to_broadcast(
     on_delivered_callback,
     log_operation_fn,
     update_central_fn,
-) -> Tuple[bool, int, int, Optional[str]]:
+) -> Tuple[bool, int, int, Any]:
     """
     Execute broadcast send to all branches.
 
     Returns:
-        Tuple of (success, success_count, total_count, error_msg).
+        Tuple of (success, success_count, total_count, results_or_error).
-        error_msg is set if the email file could not be loaded.
+        On failure: 4th element is an error string.
+        On success: 4th element is a list of (branch_name, success, error_msg) tuples.
     """
     email_file = create_email_file_fn("all", subject, message, user_info, reply_to=reply_to, dispatched_to=dispatched_to)
     email_data = load_email_file_fn(email_file)

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

@@ -16,6 +16,8 @@ Independent handler - no module or display dependencies.
 from pathlib import Path
 from typing import List, Dict, Optional, Any
 
+from aipass.ai_mail.apps.handlers.json import json_handler
+
 
 def parse_send_args(args: List[str]) -> Dict[str, Any]:
     """
@@ -35,6 +37,7 @@ def parse_send_args(args: List[str]) -> Dict[str, Any]:
             mode: 'direct' | 'interactive' | 'error'
             error: str | None (set when mode=='error')
     """
+    json_handler.log_operation("parse_send_args", {"arg_count": len(args)})
     working_args = list(args)
 
     # Extract --dispatch / --auto-execute

src/aipass/ai_mail/apps/handlers/json/json_handler.py

@@ -24,4 +24,5 @@ from aipass.ai_mail.apps.handlers.json_utils.json_handler import (  # noqa: F401
     save_json,
     ensure_json_exists,
     get_json_path,
+    log_operation,
 )

src/aipass/ai_mail/apps/handlers/monitoring/errors.py

@@ -26,6 +26,8 @@ import re
 from pathlib import Path
 from typing import Optional, Dict, Tuple
 
+from aipass.ai_mail.apps.handlers.json import json_handler
+
 
 def _find_repo_root() -> Path:
     """Walk up from this file to find AIPASS_REGISTRY.json (repo root)."""
@@ -63,6 +65,8 @@ def parse_error_log_line(log_line: str) -> Optional[Dict]:
         Dict with timestamp, logger_name, module_name, level, message
         None if not an ERROR line or parsing fails
     """
+    json_handler.log_operation("parse_error_log_line", {"log_line_length": len(log_line)})
+
     match = re.match(LOG_PATTERN, log_line.strip())
 
     if not match:

src/aipass/ai_mail/apps/handlers/monitoring/memory.py

@@ -23,6 +23,8 @@ Architecture:
 # =============================================
 from pathlib import Path
 
+from aipass.ai_mail.apps.handlers.json import json_handler
+
 
 # =============================================
 # CONSTANTS
@@ -107,6 +109,8 @@ def get_health_info(file_path: Path | str) -> dict:
     Returns:
         Dict with line_count, status, needs_email
     """
+    json_handler.log_operation("get_health_info", {"file_path": str(file_path)})
+
     line_count = count_file_lines(file_path)
 
     return {

src/aipass/ai_mail/apps/handlers/notify.py

@@ -20,6 +20,7 @@ import subprocess
 import sys
 
 from aipass.prax.apps.modules.logger import system_logger as logger
+from aipass.ai_mail.apps.handlers.json import json_handler
 
 
 def send_notification(title: str, body: str, source: str = "ai_mail",
@@ -36,6 +37,8 @@ def send_notification(title: str, body: str, source: str = "ai_mail",
     Returns:
         True if sent, False on failure
     """
+    json_handler.log_operation("send_notification", {"title": title, "source": source})
+
     # Primary: dbus direct (bypasses Portal, supports stacking)
     if _send_via_dbus(title, body, source, icon):
         return True

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

@@ -14,6 +14,8 @@ import json
 from pathlib import Path
 from typing import Dict
 
+from aipass.ai_mail.apps.handlers.json import json_handler
+
 
 def load_registry(registry_file: Path) -> Dict:
     """
@@ -35,6 +37,8 @@ def load_registry(registry_file: Path) -> Dict:
             }
         }
     """
+    json_handler.log_operation("load_registry", {"registry_file": str(registry_file)})
+
     if not registry_file.exists():
         return {
             "last_updated": "",

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

@@ -22,7 +22,9 @@ Handler Independence:
 
 import json
 from pathlib import Path
-from typing import List, Dict
+from typing import List, Dict, Optional
+
+from aipass.ai_mail.apps.handlers.json import json_handler
 
 
 # Constants
@@ -53,6 +55,8 @@ def get_all_branches() -> List[Dict]:
     Note:
         Returns empty list if registry not found or on error.
     """
+    json_handler.log_operation("get_all_branches", {"registry_path": str(BRANCH_REGISTRY_PATH)})
+
     branches = []
 
     if not BRANCH_REGISTRY_PATH.exists():
@@ -118,7 +122,7 @@ def _derive_email_from_branch_name(branch_name: str) -> str:
     return f"@{email_part}"
 
 
-def get_branch_by_email(email: str) -> Dict | None:
+def get_branch_by_email(email: str) -> Optional[Dict]:
     """
     Get branch information by email address.
 
@@ -170,7 +174,7 @@ if __name__ == "__main__":
     console.print()
     console.print("FUNCTIONS PROVIDED:")
     console.print("  - get_all_branches() -> List[Dict]")
-    console.print("  - get_branch_by_email(email) -> Dict | None")
+    console.print("  - get_branch_by_email(email) -> Optional[Dict]")
     console.print("  - get_branch_email_map() -> Dict[str, str]")
     console.print("  - get_branch_path_map() -> Dict[str, str]")
     console.print()

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

@@ -25,6 +25,8 @@ from pathlib import Path
 from datetime import datetime
 from typing import Dict, Tuple
 
+from aipass.ai_mail.apps.handlers.json import json_handler
+
 
 # Constants
 MODULE_NAME = "registry.update"
@@ -56,6 +58,8 @@ def ping_registry(
     Returns:
         True if registry updated successfully, False otherwise
     """
+    json_handler.log_operation("ping_registry", {"branch_name": branch_name, "branch_path": str(branch_path)})
+
     try:
         # Ensure registry directory exists
         REGISTRY_PATH.parent.mkdir(parents=True, exist_ok=True)

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

@@ -19,9 +19,10 @@ Walks up directory tree to find branch root (has .trinity/passport.json).
 import os
 import json
 from pathlib import Path
-from typing import Dict
+from typing import Dict, Optional
 
 from aipass.prax.apps.modules.logger import system_logger as logger
+from aipass.ai_mail.apps.handlers.json import json_handler
 
 # =============================================
 # CONSTANTS
@@ -55,7 +56,7 @@ def _get_branches_list(registry: dict) -> list:
 # BRANCH DETECTION FUNCTIONS
 # =============================================
 
-def detect_branch_from_pwd() -> Dict | None:
+def detect_branch_from_pwd() -> Optional[Dict]:
     """
     Detect which branch is calling based on current working directory.
 
@@ -73,6 +74,8 @@ def detect_branch_from_pwd() -> Dict | None:
         }
         None if no branch detected
     """
+    json_handler.log_operation("detect_branch_from_pwd", {"cwd": str(Path.cwd())})
+
     try:
         # Primary: use explicit branch name passed by drone (works in Docker + local)
         caller_branch = os.environ.get("AIPASS_CALLER_BRANCH")
@@ -102,7 +105,7 @@ def detect_branch_from_pwd() -> Dict | None:
         return None
 
 
-def _lookup_branch_by_name(branch_name: str) -> Dict | None:
+def _lookup_branch_by_name(branch_name: str) -> Optional[Dict]:
     """
     Look up branch in the registry by name (case-insensitive).
 
@@ -135,7 +138,7 @@ def _lookup_branch_by_name(branch_name: str) -> Dict | None:
         return None
 
 
-def find_branch_root(start_path: Path) -> Path | None:
+def find_branch_root(start_path: Path) -> Optional[Path]:
     """
     Walk up directory tree to find branch root.
 
@@ -165,7 +168,7 @@ def find_branch_root(start_path: Path) -> Path | None:
     return None
 
 
-def get_branch_info_from_registry(branch_path: Path) -> Dict | None:
+def get_branch_info_from_registry(branch_path: Path) -> Optional[Dict]:
     """
     Look up branch information in BRANCH_REGISTRY.json by path.
 
@@ -250,9 +253,9 @@ if __name__ == "__main__":
     console.print("  Walks up directory tree to find branch root")
     console.print()
     console.print("FUNCTIONS PROVIDED:")
-    console.print("  - detect_branch_from_pwd() -> Dict | None")
+    console.print("  - detect_branch_from_pwd() -> Optional[Dict]")
-    console.print("  - find_branch_root(start_path) -> Path | None")
+    console.print("  - find_branch_root(start_path) -> Optional[Path]")
-    console.print("  - get_branch_info_from_registry(branch_path) -> Dict | None")
+    console.print("  - get_branch_info_from_registry(branch_path) -> Optional[Dict]")
     console.print("  - get_branch_display_name(branch_info) -> str")
     console.print("  - get_local_config_path(branch_path) -> Path")
     console.print()

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

@@ -19,9 +19,10 @@ PHILOSOPHY: Fail hard if detection fails. Fallbacks hide bugs.
 # IMPORTS
 # =============================================
 from pathlib import Path
-from typing import Dict
+from typing import Dict, Optional
 
 from aipass.prax.apps.modules.logger import system_logger as logger
+from aipass.ai_mail.apps.handlers.json import json_handler
 
 # Import branch detection functions
 from .branch_detection import detect_branch_from_pwd
@@ -49,6 +50,8 @@ def get_current_user() -> Dict:
     Raises:
         RuntimeError: If branch detection fails (not called from a branch directory)
     """
+    json_handler.log_operation("get_current_user", {"cwd": str(Path.cwd())})
+
     # Detect branch from PWD
     branch_info = detect_branch_from_pwd()
 
@@ -93,7 +96,7 @@ def get_current_user() -> Dict:
     }
 
 
-def get_user_by_email(email: str) -> Dict | None:
+def get_user_by_email(email: str) -> Optional[Dict]:
     """
     Get user information by email address from BRANCH_REGISTRY.json
 

src/aipass/ai_mail/apps/modules/branch_ping.py

@@ -31,7 +31,7 @@ from aipass.ai_mail.apps.handlers.registry.update import (
     get_branch_context,
     update_json_memory_health
 )
-from aipass.ai_mail.apps.handlers.json_utils.json_handler import log_operation
+from aipass.ai_mail.apps.handlers.json import json_handler
 
 MODULE_NAME = "branch_ping"
 THRESHOLDS = {"green": (0, 400), "yellow": (401, 550), "red": (551, float('inf'))}
@@ -55,7 +55,7 @@ def handle_ping(verbose: bool = False) -> bool:
         obs_status = {"line_count": obs_count, "status": obs_status_code}
         ping_registry(branch_name, cwd, local_status, obs_status)
 
-        log_operation("ping_executed", {"branch": branch_name, "local_count": local_count, "obs_count": obs_count})
+        json_handler.log_operation("ping_executed", {"branch": branch_name, "local_count": local_count, "obs_count": obs_count})
 
         if verbose:
             console.print(f"Ping successful for {branch_name}")
@@ -172,6 +172,10 @@ def handle_command(command: str, args: List[str]) -> bool:
         print_help()
         return True
 
+    if not args:
+        print_introspection()
+        return True
+
     if command == "ping":
         return handle_ping("--verbose" in args or "-v" in args)
     elif command == "status":

src/aipass/ai_mail/apps/modules/dispatch.py

@@ -19,6 +19,7 @@ from typing import List
 
 from aipass.prax.apps.modules.logger import system_logger as logger
 from aipass.cli.apps.modules import console, error
+from aipass.ai_mail.apps.handlers.json import json_handler
 from aipass.ai_mail.apps.handlers.dispatch.status import (
     load_dispatch_log,
     check_pid_status,
@@ -81,11 +82,13 @@ def handle_command(command: str, args: List[str]) -> bool:
         return True
 
     if not args:
-        print_help()
+        print_introspection()
         return True
 
     subcommand = args[0]
 
+    json_handler.log_operation("dispatch_command", {"subcommand": subcommand})
+
     if subcommand == "status":
         return _orchestrate_status()
     elif subcommand == "daemon":

src/aipass/ai_mail/apps/modules/email.py

@@ -45,7 +45,7 @@ from aipass.ai_mail.apps.handlers.email.reply import get_email_by_id, send_reply
 from aipass.ai_mail.apps.handlers.email.header import prepend_dispatch_header
 from aipass.ai_mail.apps.handlers.users.user import get_current_user
 from aipass.ai_mail.apps.handlers.registry.read import get_all_branches, get_branch_by_email
-from aipass.ai_mail.apps.handlers.json_utils.json_handler import log_operation
+from aipass.ai_mail.apps.handlers.json import json_handler
 from aipass.ai_mail.apps.handlers.email.send import (
     resolve_sender_info, send_to_broadcast, send_to_single, collect_interactive_input
 )
@@ -116,7 +116,7 @@ def handle_command(command: str, args: List[str]) -> bool:
 
 def handle_send(args: List[str]) -> bool:
     """Orchestrate email sending workflow."""
-    log_operation("send_email_initiated", {"args_count": len(args)})
+    json_handler.log_operation("send_email_initiated", {"args_count": len(args)})
     parsed = parse_send_args(args)
 
     if parsed["mode"] == "error":
@@ -192,7 +192,7 @@ def _send_direct(to_branch, subject, message, auto_execute=False,
         success, error_msg = send_to_single(
             to_branch, subject, message, user_info, auto_execute, no_memory_save,
             reply_to, dispatched_to, create_email_file, load_email_file,
-            deliver_email_to_branch, _delivery_callback, log_operation, update_central)
+            deliver_email_to_branch, _delivery_callback, json_handler.log_operation, update_central)
 
         if success:
             label = f"\\[dispatch: queued for daemon]" if auto_execute else ""
@@ -205,7 +205,7 @@ def _send_direct(to_branch, subject, message, auto_execute=False,
             return True
         else:
             error(f"Failed to deliver: {error_msg}")
-            dispatch_send_error(to_branch, subject, error_msg, deliver_email_to_branch)
+            dispatch_send_error(to_branch, subject, error_msg or "", deliver_email_to_branch)
             return False
     except BrokenPipeError:
         logger.info("[email] Send: broken pipe (stdout closed early)")
@@ -224,11 +224,11 @@ def _send_broadcast(subject, message, user_info, auto_execute, no_memory_save, r
     ok, success_count, total, results = send_to_broadcast(
         subject, message, user_info, auto_execute, no_memory_save, reply_to, dispatched_to,
         branches, create_email_file, load_email_file, deliver_email_to_branch,
-        _delivery_callback, log_operation, update_central)
+        _delivery_callback, json_handler.log_operation, update_central)
-    if isinstance(results, str):
+    if isinstance(results, str) or results is None:
         error("Failed to load email file for broadcast")
         return False
-    for name, ok, err in results:
+    for name, ok, err in results:  # type: ignore[union-attr]
         if ok:
             console.print(f"  [green]OK[/green] {name}")
         else:
@@ -239,7 +239,7 @@ def _send_broadcast(subject, message, user_info, auto_execute, no_memory_save, r
 
 def handle_inbox(args: List[str]) -> bool:
     """Orchestrate inbox viewing."""
-    log_operation("inbox_viewed")
+    json_handler.log_operation("inbox_viewed")
     try:
         first_arg = args[0] if args else None
         ok, info = resolve_inbox_target(first_arg, _REPO_ROOT, get_branch_by_email, get_current_user)
@@ -278,14 +278,14 @@ def handle_inbox(args: List[str]) -> bool:
 
 def handle_view(args: List[str]) -> bool:
     """View email content and mark as opened."""
-    log_operation("view_email_initiated", {"args": args})
+    json_handler.log_operation("view_email_initiated", {"args": args})
     if not args:
         error("Usage: drone @ai_mail view <message_id>")
         return False
     try:
         branch_path = Path(get_current_user()["mailbox_path"]).parent
         success, message, email_data = mark_as_opened(branch_path, args[0])
-        if not success:
+        if not success or email_data is None:
             error(message)
             return False
         console.print(f"\n{'='*60}")
@@ -296,7 +296,7 @@ def handle_view(args: List[str]) -> bool:
         console.print(f"[dim]Status: opened | ID: {args[0]}[/dim]")
         console.print(f"[dim]To reply: drone @ai_mail reply {args[0]} \"your message\"[/dim]")
         console.print(f"[dim]To close: drone @ai_mail close {args[0]}[/dim]")
-        log_operation("email_viewed", {"message_id": args[0]})
+        json_handler.log_operation("email_viewed", {"message_id": args[0]})
         return True
     except BrokenPipeError:
         return True
@@ -308,7 +308,7 @@ def handle_view(args: List[str]) -> bool:
 
 def handle_close(args: List[str]) -> bool:
     """Close email(s) and archive to deleted."""
-    log_operation("close_email_initiated", {"args": args})
+    json_handler.log_operation("close_email_initiated", {"args": args})
     if not args:
         error("Usage: drone @ai_mail close <id> [id2 ...] | close all")
         return False
@@ -321,7 +321,7 @@ def handle_close(args: List[str]) -> bool:
             else:
                 error(message)
             if success:
-                log_operation("email_closed_all", {"count": count})
+                json_handler.log_operation("email_closed_all", {"count": count})
             return success
 
         results, closed, failed = batch_close(branch_path, args, mark_as_closed_and_archive)
@@ -331,7 +331,7 @@ def handle_close(args: List[str]) -> bool:
             else:
                 error(message)
             if success:
-                log_operation("email_closed", {"message_id": msg_id})
+                json_handler.log_operation("email_closed", {"message_id": msg_id})
 
         if len(args) > 1 and closed > 0:
             try:
@@ -350,7 +350,7 @@ def handle_close(args: List[str]) -> bool:
 
 def handle_reply(args: List[str]) -> bool:
     """Reply to an email."""
-    log_operation("reply_email_initiated", {"args": args})
+    json_handler.log_operation("reply_email_initiated", {"args": args})
     if len(args) < 2:
         error("Usage: drone @ai_mail reply <message_id> \"your message\"")
         return False
@@ -367,7 +367,7 @@ def handle_reply(args: List[str]) -> bool:
         else:
             error(message)
         if success:
-            log_operation("email_replied", {"message_id": args[0], "reply_id": reply_id})
+            json_handler.log_operation("email_replied", {"message_id": args[0], "reply_id": reply_id})
         return success
     except Exception as e:
         logger.error(f"[email] Reply failed: {e}")
@@ -377,7 +377,7 @@ def handle_reply(args: List[str]) -> bool:
 
 def handle_sent(args: List[str]) -> bool:
     """View sent messages."""
-    log_operation("sent_viewed")
+    json_handler.log_operation("sent_viewed")
     try:
         sent_folder = Path(get_current_user()["mailbox_path"]) / "sent"
         if not sent_folder.exists():
@@ -402,7 +402,7 @@ def handle_sent(args: List[str]) -> bool:
 
 def handle_contacts(args: List[str]) -> bool:
     """View contacts."""
-    log_operation("contacts_viewed")
+    json_handler.log_operation("contacts_viewed")
     try:
         branches = get_all_branches()
         if not branches: