feat(system): S89: ai_mail view latest + ANSI TTY detection + auto-watchdog hook (DPLAN-0119)

Co-Authored-By: @devpulse <devpulse@aipass>

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

@@ -220,10 +220,7 @@ def _send_direct(to_branch, subject, message, auto_execute=False,
             label = f"\\[dispatch: queued for daemon]" if auto_execute else ""
             console.print(f"[green]Email sent to {to_branch} {label}[/green]")
             if auto_execute:
-                try:
-                    trigger.fire('email_dispatched', to=to_branch, subject=subject)
-                except Exception as e:
-                    logger.warning("[email] trigger fire for email_dispatched failed: %s", e)
+                _fire_dispatch_trigger(to_branch, subject)
             return True
         else:
             error(f"Failed to deliver: {error_msg}")
@@ -239,6 +236,14 @@ def _send_direct(to_branch, subject, message, auto_execute=False,
         return False
 
 
+def _fire_dispatch_trigger(to_branch: str, subject: str) -> None:
+    """Fire email_dispatched trigger event if auto_execute enabled."""
+    try:
+        trigger.fire('email_dispatched', to=to_branch, subject=subject)
+    except Exception as e:
+        logger.warning("[email] trigger fire for email_dispatched failed: %s", e)
+
+
 def _send_broadcast(subject, message, user_info, auto_execute, no_memory_save, reply_to, dispatched_to) -> bool:
     """Broadcast send to all branches - display wrapper."""
     branches = get_all_branches()
@@ -312,11 +317,26 @@ def handle_view(args: List[str]) -> bool:
     """View email content and mark as opened."""
     json_handler.log_operation("view_email_initiated", {"args": args})
     if not args:
-        error("Usage: drone @ai_mail view <message_id>")
+        error("Usage: drone @ai_mail view <message_id> | drone @ai_mail view latest")
         return True
     try:
         branch_path = _resolve_branch_path()
-        success, message, email_data = mark_as_opened(branch_path, args[0])
+        message_id = args[0]
+
+        # Handle "latest" shortcut — get the most recent message ID
+        if message_id.lower() == "latest":
+            inbox_file = branch_path / ".ai_mail.local" / "inbox.json"
+            inbox_data = load_inbox(inbox_file)
+            if not inbox_data or not inbox_data.get("messages"):
+                error("Inbox is empty")
+                return True
+            # Get the most recent message (last in the list)
+            message_id = inbox_data["messages"][-1].get("id")
+            if not message_id:
+                error("Could not find latest message")
+                return True
+
+        success, message, email_data = mark_as_opened(branch_path, message_id)
         if not success or email_data is None:
             error(message)
             return True

src/aipass/cli/apps/modules/display.py

@@ -35,10 +35,14 @@ from aipass.cli.apps.handlers.json import json_handler
 # NOTE: Cannot import prax here — circular import (prax depends on cli)
 # Silent catches in this file are bypassed via .seedgo/bypass.json
 
+# Detect if output is a TTY (interactive terminal) vs piped/redirected
+# When piped, disable force_terminal to let Rich auto-detect and strip ANSI codes
+_IS_TTY = sys.stdout.isatty()
+
 # Initialize Rich console (lowercase follows service instance pattern)
-CONSOLE = Console(force_terminal=True)  # Internal constant — force_terminal ensures ANSI colors even when piped
+CONSOLE = Console(force_terminal=_IS_TTY)  # TTY=colors, piped=plain
 console = CONSOLE  # Primary export (lowercase service instance pattern)
-err_console = Console(stderr=True, force_terminal=True)  # Stderr console for error/warning output
+err_console = Console(stderr=True, force_terminal=sys.stderr.isatty())  # Stderr auto-detect
 
 # Trigger loaded lazily to avoid circular import
 _TRIGGER = None

src/aipass/prax/apps/handlers/monitoring/branch_detector.py

@@ -127,8 +127,8 @@ class BranchDetector:
                 for item in project_dir.iterdir():
                     if item.is_file() and item.name.endswith('_REGISTRY.json'):
                         return item.stem.replace('_REGISTRY', '')
-            except (OSError, PermissionError):
-                pass
+            except (OSError, PermissionError) as e:
+                logger.info(f"[branch_detector] Cannot read project dir {project_dir}: {e}")
             return project_part.upper()
         return project_part.upper()
 
@@ -198,15 +198,21 @@ class BranchDetector:
 
         # Internal AIPass: path contains -projects-aipass-src-aipass-
         if '-projects-aipass-src-aipass-' in folder_lower:
-            # Simple hyphen-to-slash decode works (no hyphens in AIPass branch names)
-            project_path = '/' + project_folder.replace('-', '/')
+            # Strip leading dash before decode — avoids double slash and preserves
+            # normalization that treats - and _ as equivalent (handles ai_mail→ai-mail).
+            name_part = project_folder[1:] if project_folder.startswith('-') else project_folder
+            project_path = '/' + name_part.replace('-', '/')
             for registered_path, branch_name in self.branch_map.items():
                 reg_norm = registered_path.replace('_', '/')
                 proj_norm = project_path.replace('_', '/')
                 if reg_norm == proj_norm or registered_path == project_path:
                     return f"AIPASS/{branch_name}{sub_suffix}"
-            # Fallback: last segment after aipass-
+            # Fallback: scan segments for known branch names (handles multi-word: ai_mail)
             segs = [s for s in project_folder.split('-') if s]
+            for n in range(min(3, len(segs)), 0, -1):
+                candidate = '_'.join(segs[-n:]).upper()
+                if candidate in self.known_branches:
+                    return f"AIPASS/{candidate}{sub_suffix}"
             if segs:
                 return f"AIPASS/{segs[-1].upper()}{sub_suffix}"
             return None
@@ -257,7 +263,7 @@ class BranchDetector:
         try:
             rel = path.relative_to(projects_base)
         except ValueError:
-            return None
+            return None  # Path is not under ~/Projects/
 
         parts = rel.parts
         if not parts: