feat(ai_mail): daemon: recognize AIPASS-TEST ping token — auto-ack without spawning (Phase 3.5 redo on fresh main) (#358)

Co-authored-by: @ai_mail <ai_mail@aipass>

src/aipass/ai_mail/.seedgo/bypass.json

@@ -344,6 +344,16 @@
       "file": "tests/test_wake_blocklist.py",
       "standard": "encapsulation",
       "reason": "Unit tests must import wake handlers directly to verify blocklist behavior. Module entry-point-only rule does not apply to tests."
+    },
+    {
+      "file": "tests/test_daemon.py",
+      "standard": "architecture",
+      "reason": "Test file lives in tests/ directory \u2014 not subject to 3-layer app structure rule."
+    },
+    {
+      "file": "tests/test_daemon.py",
+      "standard": "encapsulation",
+      "reason": "Unit tests must import daemon internals directly to verify behavior. Module entry-point-only rule does not apply to tests."
     }
   ],
   "notes": {

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

@@ -54,6 +54,10 @@ SCHEDULER_CONFIG = _REPO_ROOT / ".aipass" / "scheduler_config.json"
 # Graceful shutdown
 SHUTDOWN = False
 
+# AIPASS-TEST ping token — exact string that marks a test-only email.
+# daemon intercepts these before dispatch scan and auto-acks without spawning.
+TEST_TOKEN = "[AIPASS-TEST — do not update memories, do not execute, reply 'ack' only]"
+
 
 def _notify_telegram(message: str) -> bool:
     """Send a notification to Patrick's Telegram via the scheduler bot."""
@@ -517,6 +521,98 @@ def _is_branch_occupied(branch_path: Path) -> bool:
     return False
 
 
+def _has_test_token(body: str) -> bool:
+    """Return True if body contains the AIPASS-TEST token outside a code fence.
+
+    Code-fence-aware: lines between ``` markers are skipped so the token
+    inside a quoted example does not trigger auto-ack. Detection is
+    case-sensitive and line-anchored (strips whitespace before comparing).
+    """
+    in_fence = False
+    for line in body.splitlines():
+        stripped = line.strip()
+        if stripped.startswith("```"):
+            in_fence = not in_fence
+            continue
+        if not in_fence and stripped == TEST_TOKEN:
+            return True
+    return False
+
+
+def _auto_ack_test_email(branch_path: Path, branch_email: str, message: Dict[str, Any]) -> bool:
+    """Send an 'ack' reply to a test-token email and close it.
+
+    Runs drone commands with cwd=branch_path so sender identity resolves
+    to the target branch, not @ai_mail.
+
+    Returns True if both reply and close succeed.
+    """
+    msg_id = message.get("id", "")
+    sender = message.get("from_email") or message.get("from", "")
+    subject = message.get("subject", "")
+    if not msg_id or not sender:
+        logger.warning("[daemon] _auto_ack_test_email: missing id or sender in message")
+        return False
+
+    reply_subject = f"Re: {subject}" if subject else "Re: (no subject)"
+    try:
+        result = subprocess.run(
+            ["drone", "@ai_mail", "email", sender, reply_subject, "ack"],
+            cwd=str(branch_path),
+            capture_output=True,
+            text=True,
+            timeout=30,
+        )
+        if result.returncode != 0:
+            logger.warning("[daemon] _auto_ack_test_email: reply failed for %s: %s", msg_id, result.stderr)
+            return False
+    except (OSError, subprocess.TimeoutExpired) as exc:
+        logger.warning("[daemon] _auto_ack_test_email: reply subprocess error: %s", exc)
+        return False
+
+    try:
+        result = subprocess.run(
+            ["drone", "@ai_mail", "close", msg_id],
+            cwd=str(branch_path),
+            capture_output=True,
+            text=True,
+            timeout=30,
+        )
+        if result.returncode != 0:
+            logger.warning("[daemon] _auto_ack_test_email: close failed for %s: %s", msg_id, result.stderr)
+            return False
+    except (OSError, subprocess.TimeoutExpired) as exc:
+        logger.warning("[daemon] _auto_ack_test_email: close subprocess error: %s", exc)
+        return False
+
+    logger.info("[daemon] auto-acked test email %s at %s", msg_id, branch_email)
+    return True
+
+
+def scan_and_ack_test_emails(branch_path: Path, branch_email: str) -> int:
+    """Scan a branch inbox for AIPASS-TEST tokens and auto-ack each one.
+
+    Called at the start of poll_cycle per-branch, before check_inbox_for_dispatch,
+    so test emails are consumed and never reach the dispatch scanner.
+
+    Returns the count of test emails acked.
+    """
+    inbox_file = branch_path / ".ai_mail.local" / "inbox.json"
+    inbox_data = _read_json(inbox_file)
+    if inbox_data is None:
+        return 0
+
+    acked = 0
+    for msg in inbox_data.get("messages", []):
+        if msg.get("status") not in ("new", "opened"):
+            continue
+        body = msg.get("body", "")
+        if _has_test_token(body):
+            if _auto_ack_test_email(branch_path, branch_email, msg):
+                acked += 1
+    return acked
+
+
 def poll_cycle(config: Dict[str, Any], state: Dict[str, Any]) -> int:
     """
     Run one poll cycle across all registered branches.
@@ -551,6 +647,9 @@ def poll_cycle(config: Dict[str, Any], state: Dict[str, Any]) -> int:
             logger.info(f"SKIP {branch_email}: daily limit reached ({daily_count}/{max_daily})")
             continue
 
+        # Intercept AIPASS-TEST ping emails before dispatch scan
+        scan_and_ack_test_emails(branch_path, branch_email)
+
         # Always check/clean stale locks (even without dispatch emails)
         existing_lock = _check_lock(branch_path)
         if existing_lock is not None:

src/aipass/ai_mail/tests/test_daemon.py

@@ -507,3 +507,148 @@ def test_is_protected_branch_other():
 def test_is_protected_branch_empty_string():
     """Empty string is not protected."""
     assert is_protected_branch("") is False
+
+
+# ---- _has_test_token tests -----------------------------------
+
+from aipass.ai_mail.apps.handlers.dispatch.daemon import (  # noqa: E402
+    _has_test_token,
+    _auto_ack_test_email,
+    scan_and_ack_test_emails,
+    TEST_TOKEN,
+)
+
+
+def test_has_test_token_plain_body():
+    """Token on its own line is detected."""
+    body = f"Some text\n{TEST_TOKEN}\nMore text"
+    assert _has_test_token(body) is True
+
+
+def test_has_test_token_only_token():
+    """Body containing only the token is detected."""
+    assert _has_test_token(TEST_TOKEN) is True
+
+
+def test_has_test_token_absent():
+    """Body without token returns False."""
+    assert _has_test_token("Hello, please process inbox.") is False
+
+
+def test_has_test_token_inside_code_fence_ignored():
+    """Token inside a code fence is not detected."""
+    body = f"Example:\n```\n{TEST_TOKEN}\n```\nEnd"
+    assert _has_test_token(body) is False
+
+
+def test_has_test_token_after_code_fence():
+    """Token after a closing fence is still detected."""
+    body = f"```\nsome code\n```\n{TEST_TOKEN}"
+    assert _has_test_token(body) is True
+
+
+def test_has_test_token_whitespace_stripped():
+    """Leading/trailing whitespace is stripped before comparison."""
+    body = f"  {TEST_TOKEN}  "
+    assert _has_test_token(body) is True
+
+
+def test_has_test_token_partial_match_not_detected():
+    """A partial token string does not match."""
+    assert _has_test_token("[AIPASS-TEST]") is False
+
+
+# ---- _auto_ack_test_email tests ------------------------------
+
+
+def test_auto_ack_test_email_success(tmp_path):
+    """Successful reply + close returns True."""
+    branch_path = tmp_path / "testbranch"
+    branch_path.mkdir()
+    message = {"id": "abc123", "from_email": "@devpulse", "subject": "Ping"}
+
+    with patch("aipass.ai_mail.apps.handlers.dispatch.daemon.subprocess.run") as mock_run:
+        mock_run.return_value.returncode = 0
+        result = _auto_ack_test_email(branch_path, "@testbranch", message)
+
+    assert result is True
+    assert mock_run.call_count == 2
+
+
+def test_auto_ack_test_email_reply_failure(tmp_path):
+    """Failed reply returns False without attempting close."""
+    branch_path = tmp_path / "testbranch"
+    branch_path.mkdir()
+    message = {"id": "abc123", "from_email": "@devpulse", "subject": "Ping"}
+
+    with patch("aipass.ai_mail.apps.handlers.dispatch.daemon.subprocess.run") as mock_run:
+        mock_run.return_value.returncode = 1
+        mock_run.return_value.stderr = "error"
+        result = _auto_ack_test_email(branch_path, "@testbranch", message)
+
+    assert result is False
+    assert mock_run.call_count == 1
+
+
+def test_auto_ack_test_email_missing_id(tmp_path):
+    """Message without id returns False immediately."""
+    message = {"from_email": "@devpulse", "subject": "Ping"}
+    result = _auto_ack_test_email(tmp_path, "@testbranch", message)
+    assert result is False
+
+
+def test_auto_ack_test_email_missing_sender(tmp_path):
+    """Message without from_email or from returns False immediately."""
+    message = {"id": "abc123", "subject": "Ping"}
+    result = _auto_ack_test_email(tmp_path, "@testbranch", message)
+    assert result is False
+
+
+# ---- scan_and_ack_test_emails tests -------------------------
+
+
+def test_scan_and_ack_test_emails_acks_matching(tmp_path):
+    """Returns count of acked test emails."""
+    branch_path = tmp_path / "testbranch"
+    ai_mail_local = branch_path / ".ai_mail.local"
+    ai_mail_local.mkdir(parents=True)
+    inbox = {
+        "messages": [
+            {"id": "t1", "status": "new", "from_email": "@devpulse", "subject": "test", "body": TEST_TOKEN},
+            {"id": "n1", "status": "new", "from_email": "@devpulse", "subject": "work", "body": "do something"},
+        ]
+    }
+    (ai_mail_local / "inbox.json").write_text(json.dumps(inbox))
+
+    with patch("aipass.ai_mail.apps.handlers.dispatch.daemon._auto_ack_test_email", return_value=True) as mock_ack:
+        count = scan_and_ack_test_emails(branch_path, "@testbranch")
+
+    assert count == 1
+    mock_ack.assert_called_once()
+
+
+def test_scan_and_ack_test_emails_skips_closed(tmp_path):
+    """Closed messages are not scanned."""
+    branch_path = tmp_path / "testbranch"
+    ai_mail_local = branch_path / ".ai_mail.local"
+    ai_mail_local.mkdir(parents=True)
+    inbox = {
+        "messages": [
+            {"id": "t1", "status": "closed", "from_email": "@devpulse", "subject": "test", "body": TEST_TOKEN},
+        ]
+    }
+    (ai_mail_local / "inbox.json").write_text(json.dumps(inbox))
+
+    with patch("aipass.ai_mail.apps.handlers.dispatch.daemon._auto_ack_test_email") as mock_ack:
+        count = scan_and_ack_test_emails(branch_path, "@testbranch")
+
+    assert count == 0
+    mock_ack.assert_not_called()
+
+
+def test_scan_and_ack_test_emails_no_inbox(tmp_path):
+    """Missing inbox returns 0."""
+    branch_path = tmp_path / "testbranch"
+    branch_path.mkdir()
+    count = scan_and_ack_test_emails(branch_path, "@testbranch")
+    assert count == 0