Merge pull request #263 from AIOSAI/system/devpulse-s91-cleanup-round-2-s90-stragglers-staged-aimail-c

feat(system): S91 cleanup round 2: S90 stragglers staged (ai_mail contacts/identity, devpulse feedback channel, prax monitor_info)

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

@@ -0,0 +1,142 @@
+# =================== AIPass ====================
+# Name: contacts.py
+# Description: Contacts address book — maps branch names to inbox paths
+# Version: 1.0.0
+# Created: 2026-04-11
+# Modified: 2026-04-11
+# =============================================
+
+"""
+Contacts Address Book Handler
+
+Maps branch names to their inbox paths for fast sender/recipient resolution.
+Solves the BRANCH DETECTION FAILED problem when external projects call drone
+— contacts lookup works even when CWD-walking cannot identify the caller.
+"""
+
+from datetime import datetime
+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
+from aipass.ai_mail.apps.handlers.paths import find_repo_root
+
+CONTACTS_FILE = find_repo_root() / "src/aipass/ai_mail/.ai_mail.local/contacts.json"
+
+
+# =============================================
+# INTERNAL HELPERS
+# =============================================
+
+def _load_contacts() -> Dict:
+    """Load contacts.json from disk with fallback to empty structure.
+
+    Returns:
+        Dict with 'contacts' key mapping branch names to contact info.
+    """
+    if not CONTACTS_FILE.exists():
+        return {"contacts": {}}
+    try:
+        import json
+        with open(CONTACTS_FILE, 'r', encoding='utf-8') as f:
+            data = json.load(f)
+        if not isinstance(data, dict) or "contacts" not in data:
+            return {"contacts": {}}
+        return data
+    except Exception as e:
+        logger.warning("[contacts] _load_contacts() failed: %s", e)
+        return {"contacts": {}}
+
+
+def _save_contacts(data: Dict) -> bool:
+    """Save contacts dict to contacts.json.
+
+    Args:
+        data: Full contacts data dict to save.
+
+    Returns:
+        True on success, False on error.
+    """
+    try:
+        import json
+        CONTACTS_FILE.parent.mkdir(parents=True, exist_ok=True)
+        with open(CONTACTS_FILE, 'w', encoding='utf-8') as f:
+            json.dump(data, f, indent=2, ensure_ascii=False)
+        return True
+    except Exception as e:
+        logger.warning("[contacts] _save_contacts() failed: %s", e)
+        return False
+
+
+# =============================================
+# PUBLIC API
+# =============================================
+
+def get_contact(branch_name: str) -> Optional[Dict]:
+    """Look up a contact by branch name.
+
+    Args:
+        branch_name: Branch name or email (e.g., 'devpulse' or '@devpulse').
+
+    Returns:
+        Contact dict with 'project', 'inbox', 'last_seen' keys, or None if not found.
+    """
+    json_handler.log_operation("contacts_get", {"branch": branch_name})
+    name_key = branch_name.lstrip("@").lower()
+    data = _load_contacts()
+    return data["contacts"].get(name_key)
+
+
+def register_contact(branch_name: str, project: str, inbox_path: str) -> bool:
+    """Add or update a contact in the contacts address book.
+
+    Args:
+        branch_name: Branch name or email (leading @ stripped, lowercased).
+        project: Project name (e.g., 'AIPass', 'VeraStudio').
+        inbox_path: Absolute path to the branch's inbox.json file.
+
+    Returns:
+        True on success, False on error.
+    """
+    json_handler.log_operation("contacts_register", {"branch": branch_name, "inbox": inbox_path})
+    name_key = branch_name.lstrip("@").lower()
+    data = _load_contacts()
+    data["contacts"][name_key] = {
+        "project": project,
+        "inbox": inbox_path,
+        "last_seen": datetime.now().isoformat(),
+    }
+    return _save_contacts(data)
+
+
+def all_contacts() -> Dict:
+    """Return all contacts as a dict keyed by branch name.
+
+    Returns:
+        Dict mapping branch names to contact info dicts.
+    """
+    json_handler.log_operation("contacts_list")
+    data = _load_contacts()
+    return data["contacts"]
+
+
+if __name__ == "__main__":
+    from aipass.cli.apps.modules import console
+
+    console.print("\n" + "=" * 70)
+    console.print("CONTACTS ADDRESS BOOK HANDLER")
+    console.print("=" * 70)
+    console.print(f"\nContacts file: {CONTACTS_FILE}")
+    contacts = all_contacts()
+    if contacts:
+        console.print(f"\n{len(contacts)} registered contact(s):")
+        for name, info in sorted(contacts.items()):
+            console.print(f"  @{name} -> {info.get('inbox', '?')} ({info.get('project', '')})")
+    else:
+        console.print("\nNo contacts registered yet.")
+    console.print("\nFunctions provided:")
+    console.print("  - get_contact(branch_name) -> Optional[Dict]")
+    console.print("  - register_contact(branch_name, project, inbox_path) -> bool")
+    console.print("  - all_contacts() -> Dict")
+    console.print()

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

@@ -0,0 +1,121 @@
+# =================== AIPass ====================
+# Name: identity.py
+# Description: Per-branch self-declaration identity files
+# Version: 1.0.0
+# Created: 2026-04-11
+# Modified: 2026-04-11
+# =============================================
+
+"""
+Per-Branch Identity Handler
+
+Manages identity.json files in each branch's .ai_mail.local/ directory.
+These files allow branches to self-declare their name, project, and inbox
+path — enabling fast sender resolution without CWD-walking.
+"""
+
+import json
+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
+
+
+# =============================================
+# PUBLIC API
+# =============================================
+
+def create_identity(branch_path: Path, branch_name: str, project: str) -> bool:
+    """Write identity.json to branch_path/.ai_mail.local/.
+
+    Args:
+        branch_path: Root directory of the branch.
+        branch_name: Branch name (e.g., 'devpulse').
+        project: Project name (e.g., 'AIPass').
+
+    Returns:
+        True on success, False on error.
+    """
+    json_handler.log_operation("identity_create", {"branch": branch_name, "project": project})
+    try:
+        mail_dir = branch_path / ".ai_mail.local"
+        mail_dir.mkdir(parents=True, exist_ok=True)
+        inbox_path = mail_dir / "inbox.json"
+        identity_data = {
+            "branch": branch_name.lower(),
+            "project": project,
+            "inbox": str(inbox_path),
+        }
+        identity_file = mail_dir / "identity.json"
+        with open(identity_file, 'w', encoding='utf-8') as f:
+            json.dump(identity_data, f, indent=2, ensure_ascii=False)
+        return True
+    except Exception as e:
+        logger.warning("[identity] create_identity(%s) failed: %s", branch_name, e)
+        return False
+
+
+def read_identity(branch_path: Path) -> Optional[Dict]:
+    """Read identity.json from branch_path/.ai_mail.local/.
+
+    Args:
+        branch_path: Root directory of the branch.
+
+    Returns:
+        Identity dict with 'branch', 'project', 'inbox' keys,
+        or None if not found or on error.
+    """
+    json_handler.log_operation("identity_read", {"path": str(branch_path)})
+    identity_file = branch_path / ".ai_mail.local" / "identity.json"
+    if not identity_file.exists():
+        return None
+    try:
+        with open(identity_file, 'r', encoding='utf-8') as f:
+            return json.load(f)
+    except Exception as e:
+        logger.warning("[identity] read_identity(%s) failed: %s", branch_path, e)
+        return None
+
+
+def bootstrap_aipass_identities() -> int:
+    """Create identity.json for all branches registered in the AIPass registry.
+
+    Uses lazy imports to avoid circular dependencies.
+
+    Returns:
+        Number of identity.json files successfully created.
+    """
+    json_handler.log_operation("identity_bootstrap")
+    from aipass.ai_mail.apps.handlers.registry.read import get_all_branches
+    from aipass.ai_mail.apps.handlers.paths import find_repo_root
+
+    repo_root = find_repo_root()
+    branches = get_all_branches()
+    count = 0
+    for branch in branches:
+        path_str = branch.get("path", "")
+        if not path_str:
+            continue
+        branch_path = Path(path_str)
+        if not branch_path.is_absolute():
+            branch_path = (repo_root / branch_path).resolve()
+        name = branch.get("email", "").lstrip("@") or branch.get("name", "").lower()
+        if not name:
+            continue
+        if create_identity(branch_path, name, "AIPass"):
+            count += 1
+    return count
+
+
+if __name__ == "__main__":
+    from aipass.cli.apps.modules import console
+
+    console.print("\n" + "=" * 70)
+    console.print("PER-BRANCH IDENTITY HANDLER")
+    console.print("=" * 70)
+    console.print("\nFunctions provided:")
+    console.print("  - create_identity(branch_path, branch_name, project) -> bool")
+    console.print("  - read_identity(branch_path) -> Optional[Dict]")
+    console.print("  - bootstrap_aipass_identities() -> int")
+    console.print()

src/aipass/ai_mail/tests/test_contacts.py

@@ -0,0 +1,164 @@
+# =================== AIPass ====================
+# Name: test_contacts.py
+# Description: Tests for the contacts address book handler
+# Version: 1.0.0
+# Created: 2026-04-11
+# Modified: 2026-04-11
+# =============================================
+
+"""Tests for contacts address book handler (DPLAN-0121 Phase 5)."""
+
+import json
+import pytest
+from pathlib import Path
+from unittest.mock import patch
+
+import aipass.ai_mail.apps.handlers.email.contacts as contacts_mod
+from aipass.ai_mail.apps.handlers.email.contacts import (
+    get_contact,
+    register_contact,
+    all_contacts,
+    _load_contacts,
+)
+
+
+# ---- Fixtures ------------------------------------------------
+
+@pytest.fixture(autouse=True)
+def _silence_json_handler():
+    """Prevent log_operation from writing real JSON files during tests."""
+    with patch("aipass.ai_mail.apps.handlers.email.contacts.json_handler") as mock_jh:
+        mock_jh.log_operation.return_value = True
+        yield mock_jh
+
+
+@pytest.fixture
+def isolated_contacts(tmp_path, monkeypatch):
+    """Point CONTACTS_FILE to a tmp_path location for test isolation."""
+    contacts_file = tmp_path / ".ai_mail.local" / "contacts.json"
+    monkeypatch.setattr(contacts_mod, "CONTACTS_FILE", contacts_file)
+    return contacts_file
+
+
+# ---- _load_contacts() tests --------------------------------
+
+def test_load_contacts_missing_file(isolated_contacts):
+    """Missing contacts.json returns default empty structure."""
+    result = _load_contacts()
+    assert result == {"contacts": {}}
+
+
+def test_load_contacts_invalid_json(isolated_contacts):
+    """Corrupted contacts.json returns default empty structure."""
+    isolated_contacts.parent.mkdir(parents=True, exist_ok=True)
+    isolated_contacts.write_text("not json", encoding="utf-8")
+    result = _load_contacts()
+    assert result == {"contacts": {}}
+
+
+def test_load_contacts_missing_key(isolated_contacts):
+    """contacts.json without 'contacts' key returns default empty structure."""
+    isolated_contacts.parent.mkdir(parents=True, exist_ok=True)
+    isolated_contacts.write_text(json.dumps({"other": {}}), encoding="utf-8")
+    result = _load_contacts()
+    assert result == {"contacts": {}}
+
+
+# ---- get_contact() tests -----------------------------------
+
+def test_get_contact_empty(isolated_contacts):
+    """get_contact with no contacts returns None."""
+    result = get_contact("devpulse")
+    assert result is None
+
+
+def test_get_contact_strips_at_sign(isolated_contacts):
+    """get_contact strips leading @ before lookup."""
+    register_contact("devpulse", "AIPass", "/some/inbox.json")
+    result = get_contact("@devpulse")
+    assert result is not None
+    assert result["inbox"] == "/some/inbox.json"
+
+
+def test_get_contact_case_insensitive(isolated_contacts):
+    """get_contact normalises to lowercase for lookup."""
+    register_contact("devpulse", "AIPass", "/some/inbox.json")
+    result = get_contact("DEVPULSE")
+    assert result is not None
+
+
+def test_get_contact_not_found(isolated_contacts):
+    """get_contact returns None for unknown branch."""
+    register_contact("devpulse", "AIPass", "/some/inbox.json")
+    result = get_contact("unknown")
+    assert result is None
+
+
+# ---- register_contact() tests ------------------------------
+
+def test_register_contact_creates_entry(isolated_contacts):
+    """register_contact writes a new entry with correct fields."""
+    ok = register_contact("devpulse", "AIPass", "/path/to/inbox.json")
+    assert ok is True
+
+    result = get_contact("devpulse")
+    assert result is not None
+    assert result["project"] == "AIPass"
+    assert result["inbox"] == "/path/to/inbox.json"
+    assert "last_seen" in result
+
+
+def test_register_contact_updates_existing(isolated_contacts):
+    """register_contact updates last_seen on second registration."""
+    register_contact("devpulse", "AIPass", "/path/to/inbox.json")
+    first = get_contact("devpulse")
+    assert first is not None
+    first_seen = first["last_seen"]
+
+    # Re-register with new path
+    register_contact("devpulse", "AIPass", "/new/path/inbox.json")
+    updated = get_contact("devpulse")
+    assert updated is not None
+    assert updated["inbox"] == "/new/path/inbox.json"
+    # last_seen should be present (may or may not change within same second)
+    assert "last_seen" in updated
+
+
+def test_register_contact_strips_at_sign(isolated_contacts):
+    """register_contact strips leading @ from branch name."""
+    ok = register_contact("@devpulse", "AIPass", "/inbox.json")
+    assert ok is True
+    result = get_contact("devpulse")
+    assert result is not None
+
+
+def test_register_contact_persists_to_disk(isolated_contacts):
+    """register_contact writes JSON to disk that can be read back."""
+    register_contact("testbranch", "TestProject", "/test/inbox.json")
+
+    assert isolated_contacts.exists()
+    with open(isolated_contacts, "r", encoding="utf-8") as f:
+        data = json.load(f)
+
+    assert "contacts" in data
+    assert "testbranch" in data["contacts"]
+    assert data["contacts"]["testbranch"]["project"] == "TestProject"
+
+
+# ---- all_contacts() tests ----------------------------------
+
+def test_all_contacts_empty(isolated_contacts):
+    """all_contacts returns empty dict when no contacts registered."""
+    result = all_contacts()
+    assert result == {}
+
+
+def test_all_contacts_returns_all(isolated_contacts):
+    """all_contacts returns all registered contacts."""
+    register_contact("alpha", "AIPass", "/alpha/inbox.json")
+    register_contact("beta", "AIPass", "/beta/inbox.json")
+
+    result = all_contacts()
+    assert "alpha" in result
+    assert "beta" in result
+    assert len(result) == 2

src/aipass/ai_mail/tests/test_identity.py

@@ -0,0 +1,111 @@
+# =================== AIPass ====================
+# Name: test_identity.py
+# Description: Tests for the per-branch identity handler
+# Version: 1.0.0
+# Created: 2026-04-11
+# Modified: 2026-04-11
+# =============================================
+
+"""Tests for per-branch identity handler (DPLAN-0121 Phase 5).
+
+Bypass entries for architecture and encapsulation are in .seedgo/bypass.json.
+"""
+
+import json
+import pytest
+from pathlib import Path
+from unittest.mock import patch
+
+from aipass.ai_mail.apps.handlers.email.identity import (
+    create_identity,
+    read_identity,
+)
+
+
+# ---- Fixtures ------------------------------------------------
+
+@pytest.fixture(autouse=True)
+def _silence_json_handler():
+    """Prevent log_operation from writing real JSON files during tests."""
+    with patch("aipass.ai_mail.apps.handlers.email.identity.json_handler") as mock_jh:
+        mock_jh.log_operation.return_value = True
+        yield mock_jh
+
+
+# ---- create_identity() tests --------------------------------
+
+def test_create_identity_writes_file(tmp_path):
+    """create_identity writes identity.json with correct fields."""
+    ok = create_identity(tmp_path, "devpulse", "AIPass")
+    assert ok is True
+
+    identity_file = tmp_path / ".ai_mail.local" / "identity.json"
+    assert identity_file.exists()
+
+    with open(identity_file, "r", encoding="utf-8") as f:
+        data = json.load(f)
+
+    assert data["branch"] == "devpulse"
+    assert data["project"] == "AIPass"
+    assert data["inbox"] == str(tmp_path / ".ai_mail.local" / "inbox.json")
+
+
+def test_create_identity_lowercases_name(tmp_path):
+    """create_identity stores branch name in lowercase."""
+    create_identity(tmp_path, "DEVPULSE", "AIPass")
+    identity_file = tmp_path / ".ai_mail.local" / "identity.json"
+    with open(identity_file, "r", encoding="utf-8") as f:
+        data = json.load(f)
+    assert data["branch"] == "devpulse"
+
+
+def test_create_identity_creates_parent_dirs(tmp_path):
+    """create_identity creates .ai_mail.local/ if it does not exist."""
+    branch_path = tmp_path / "nested" / "branch"
+    # Directory does not exist yet
+    assert not branch_path.exists()
+
+    ok = create_identity(branch_path, "newbranch", "AIPass")
+    assert ok is True
+    assert (branch_path / ".ai_mail.local" / "identity.json").exists()
+
+
+def test_create_identity_overwrites_existing(tmp_path):
+    """create_identity overwrites an existing identity.json."""
+    create_identity(tmp_path, "alpha", "OldProject")
+    create_identity(tmp_path, "alpha", "NewProject")
+
+    identity_file = tmp_path / ".ai_mail.local" / "identity.json"
+    with open(identity_file, "r", encoding="utf-8") as f:
+        data = json.load(f)
+
+    assert data["project"] == "NewProject"
+
+
+# ---- read_identity() tests ----------------------------------
+
+def test_read_identity_reads_back(tmp_path):
+    """read_identity returns the dict written by create_identity."""
+    create_identity(tmp_path, "devpulse", "AIPass")
+    result = read_identity(tmp_path)
+
+    assert result is not None
+    assert result["branch"] == "devpulse"
+    assert result["project"] == "AIPass"
+    assert "inbox" in result
+
+
+def test_read_identity_missing_file(tmp_path):
+    """read_identity returns None when identity.json does not exist."""
+    result = read_identity(tmp_path)
+    assert result is None
+
+
+def test_read_identity_corrupted_json(tmp_path):
+    """read_identity returns None when identity.json is invalid JSON."""
+    mail_dir = tmp_path / ".ai_mail.local"
+    mail_dir.mkdir(parents=True)
+    (mail_dir / "identity.json").write_text("not json", encoding="utf-8")
+
+    result = read_identity(tmp_path)
+    assert result is None

src/aipass/devpulse/apps/handlers/feedback/__init__.py

@@ -0,0 +1,4 @@
+# META
+# module: devpulse.feedback
+# description: Feedback communication handlers package
+# END META

src/aipass/devpulse/apps/handlers/feedback/compose.py

@@ -0,0 +1,217 @@
+# META
+# module: devpulse.feedback
+# description: Compose operations — send feedback and reply to messages
+# END META
+
+"""
+Feedback Compose — sending and replying to feedback messages.
+
+Handles inbound feedback from any agent and devpulse replies.
+Replies are also delivered to the sender's ai_mail inbox.
+"""
+
+import json
+import os
+from datetime import datetime, timezone
+from pathlib import Path
+
+from rich.console import Console
+
+from aipass.prax import logger
+from aipass.devpulse.apps.handlers.feedback.storage import (
+    load_inbox,
+    save_inbox,
+    generate_id,
+)
+
+console = Console(stderr=True)
+
+# AIPass src/aipass/ directory (four levels up from compose.py)
+_AIPASS_ROOT = Path(__file__).resolve().parents[4]
+
+
+def _resolve_sender() -> tuple[str, str]:
+    """Resolve the sender's identity and ai_mail path from drone env vars.
+
+    Drone sets AIPASS_CALLER_BRANCH and AIPASS_CALLER_CWD when routing.
+    Uses these to identify who sent the feedback and where to reply.
+
+    Returns:
+        tuple: (branch_name, ai_mail_path_or_empty_string)
+    """
+    branch = os.environ.get("AIPASS_CALLER_BRANCH", "")
+    caller_cwd = os.environ.get("AIPASS_CALLER_CWD", "")
+
+    if not branch:
+        return "unknown", ""
+
+    # Try to find sender's ai_mail inbox
+    # First: check if caller_cwd points to a branch with ai_mail
+    if caller_cwd:
+        cwd_path = Path(caller_cwd)
+        ai_mail = cwd_path / ".ai_mail.local" / "inbox.json"
+        if ai_mail.exists():
+            return branch, str(ai_mail)
+        # Walk up from CWD to find the branch directory
+        for parent in cwd_path.parents:
+            ai_mail = parent / ".ai_mail.local" / "inbox.json"
+            if ai_mail.exists():
+                return branch, str(ai_mail)
+
+    # Fallback: check AIPass internal path
+    internal_path = _AIPASS_ROOT / branch / ".ai_mail.local" / "inbox.json"
+    if internal_path.exists():
+        return branch, str(internal_path)
+
+    return branch, ""
+
+
+def send_feedback(from_branch: str, subject: str, body: str, ai_mail_path: str = "") -> str:
+    """Add a new feedback message to devpulse's inbox.
+
+    Args:
+        from_branch: Name of the sending branch/agent.
+        subject: Message subject line.
+        body: Message body text.
+        ai_mail_path: Full path to sender's ai_mail inbox for replies.
+
+    Returns:
+        str: The generated message ID.
+    """
+    data = load_inbox()
+    msg_id = generate_id()
+    now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
+
+    message = {
+        "id": msg_id,
+        "from": from_branch,
+        "subject": subject,
+        "body": body,
+        "timestamp": now,
+        "read": False,
+        "thread": [],
+        "reply_path": ai_mail_path,
+    }
+
+    data["messages"].append(message)
+    data["total_messages"] = len(data["messages"])
+    data["unread_count"] = data.get("unread_count", 0) + 1
+
+    save_inbox(data)
+
+    logger.info(f"[FEEDBACK] Received feedback from {from_branch}: {subject}")
+    console.print(f"[green]Feedback received (id: {msg_id}).[/green]")
+
+    return msg_id
+
+
+def reply_to(msg_id: str, body: str) -> bool:
+    """Reply to a feedback message and deliver to sender's ai_mail.
+
+    Adds the reply to the local thread and attempts to deliver
+    a copy to the sender's ai_mail inbox.
+
+    Args:
+        msg_id: The message ID to reply to.
+        body: The reply body text.
+
+    Returns:
+        bool: True if reply was saved (regardless of ai_mail delivery).
+    """
+    data = load_inbox()
+    messages = data.get("messages", [])
+
+    msg = None
+    for m in messages:
+        if m.get("id") == msg_id:
+            msg = m
+            break
+
+    if msg is None:
+        console.print(f"[red]Message {msg_id} not found.[/red]")
+        return False
+
+    now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
+
+    reply = {
+        "from": "devpulse",
+        "body": body,
+        "timestamp": now,
+    }
+
+    msg.setdefault("thread", []).append(reply)
+    save_inbox(data)
+
+    console.print(f"[green]Reply added to thread {msg_id}.[/green]")
+
+    # Deliver to sender's ai_mail using stored reply path
+    sender = msg.get("from", "")
+    reply_path = msg.get("reply_path", "")
+    if sender:
+        _deliver_to_ai_mail(sender, msg.get("subject", ""), body, msg_id, reply_path)
+
+    return True
+
+
+def _deliver_to_ai_mail(
+    to_branch: str, subject: str, body: str, thread_id: str, reply_path: str = ""
+) -> None:
+    """Deliver a reply to the sender's ai_mail inbox.
+
+    Writes directly to the sender's .ai_mail.local/inbox.json.
+    If the path does not exist or delivery fails, logs a warning
+    and skips silently.
+
+    Args:
+        to_branch: Target branch name.
+        subject: Original message subject (prefixed with Re:).
+        body: Reply body text.
+        thread_id: Original feedback message ID for reference.
+    """
+    # Use stored reply_path (works for external projects), fall back to AIPass internal
+    if reply_path:
+        ai_mail_path = Path(reply_path)
+    else:
+        ai_mail_path = _AIPASS_ROOT / to_branch / ".ai_mail.local" / "inbox.json"
+
+    if not ai_mail_path.exists():
+        logger.warning(
+            f"[FEEDBACK] ai_mail inbox not found for {to_branch} at {ai_mail_path} — skipping delivery"
+        )
+        return
+
+    try:
+        with open(ai_mail_path, encoding="utf-8") as f:
+            inbox = json.load(f)
+    except (json.JSONDecodeError, OSError) as e:
+        logger.warning(f"[FEEDBACK] Failed to read {to_branch} ai_mail inbox: {e}")
+        return
+
+    now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
+    mail_id = generate_id()
+
+    mail_message = {
+        "id": mail_id,
+        "from": "devpulse",
+        "to": to_branch,
+        "subject": f"Re: {subject}",
+        "body": body,
+        "timestamp": now,
+        "read": False,
+        "metadata": {
+            "source": "feedback",
+            "thread_id": thread_id,
+        },
+    }
+
+    inbox.setdefault("messages", []).append(mail_message)
+    inbox["total_messages"] = len(inbox["messages"])
+    inbox["unread_count"] = inbox.get("unread_count", 0) + 1
+
+    try:
+        with open(ai_mail_path, "w", encoding="utf-8") as f:
+            json.dump(inbox, f, indent=2)
+            f.write("\n")
+        logger.info(f"[FEEDBACK] Reply delivered to {to_branch} ai_mail")
+    except OSError as e:
+        logger.warning(f"[FEEDBACK] Failed to write to {to_branch} ai_mail: {e}")

src/aipass/devpulse/apps/handlers/feedback/inbox.py

@@ -0,0 +1,165 @@
+# META
+# module: devpulse.feedback
+# description: Inbox operations — list, view, clear feedback messages
+# END META
+
+"""
+Feedback Inbox — reading and managing feedback messages.
+
+Provides list, view, clear, and summary operations for
+devpulse's personal feedback mailbox.
+"""
+
+from rich.console import Console
+from rich.table import Table
+
+from aipass.prax import logger
+from aipass.devpulse.apps.handlers.feedback.storage import load_inbox, save_inbox
+
+console = Console(stderr=True)
+
+
+def list_messages() -> None:
+    """Display a Rich table of all feedback messages.
+
+    Shows id, from, subject, date, and read status for each message.
+    """
+    data = load_inbox()
+    messages = data.get("messages", [])
+
+    if not messages:
+        console.print("[dim]No feedback messages.[/dim]")
+        return
+
+    table = Table(title="Feedback Inbox", show_lines=False)
+    table.add_column("ID", style="cyan", width=10)
+    table.add_column("From", style="green", width=14)
+    table.add_column("Subject", style="white", min_width=20)
+    table.add_column("Date", style="dim", width=19)
+    table.add_column("Status", width=8)
+
+    for msg in messages:
+        status = "[dim]read[/dim]" if msg.get("read") else "[bold yellow]NEW[/bold yellow]"
+        table.add_row(
+            msg.get("id", "?"),
+            msg.get("from", "?"),
+            msg.get("subject", "(no subject)"),
+            msg.get("timestamp", "?"),
+            status,
+        )
+
+    console.print(table)
+
+
+def view_message(msg_id: str) -> None:
+    """Display a message and its thread, marking it as read.
+
+    Args:
+        msg_id: The 8-char hex message ID to view.
+    """
+    data = load_inbox()
+    messages = data.get("messages", [])
+
+    msg = _find_message(messages, msg_id)
+    if msg is None:
+        console.print(f"[red]Message {msg_id} not found.[/red]")
+        return
+
+    # Mark as read
+    if not msg.get("read"):
+        msg["read"] = True
+        data["unread_count"] = max(0, data.get("unread_count", 1) - 1)
+        save_inbox(data)
+
+    # Display message
+    console.print(f"\n[bold cyan]From:[/bold cyan] {msg.get('from', '?')}")
+    console.print(f"[bold cyan]Subject:[/bold cyan] {msg.get('subject', '(no subject)')}")
+    console.print(f"[bold cyan]Date:[/bold cyan] {msg.get('timestamp', '?')}")
+    console.print(f"[bold cyan]ID:[/bold cyan] {msg.get('id', '?')}")
+    console.print(f"\n{msg.get('body', '')}")
+
+    # Display thread
+    thread = msg.get("thread", [])
+    if thread:
+        console.print(f"\n[bold]Thread ({len(thread)} replies):[/bold]")
+        for reply in thread:
+            console.print(f"\n  [dim]{reply.get('timestamp', '?')}[/dim] [green]{reply.get('from', '?')}[/green]:")
+            console.print(f"  {reply.get('body', '')}")
+
+    console.print()
+
+
+def clear_message(msg_id: str) -> None:
+    """Remove a single message from the inbox.
+
+    Args:
+        msg_id: The 8-char hex message ID to remove.
+    """
+    data = load_inbox()
+    messages = data.get("messages", [])
+
+    msg = _find_message(messages, msg_id)
+    if msg is None:
+        console.print(f"[red]Message {msg_id} not found.[/red]")
+        return
+
+    was_unread = not msg.get("read")
+    data["messages"] = [m for m in messages if m.get("id") != msg_id]
+    data["total_messages"] = len(data["messages"])
+    if was_unread:
+        data["unread_count"] = max(0, data.get("unread_count", 1) - 1)
+
+    save_inbox(data)
+    console.print(f"[green]Cleared message {msg_id}.[/green]")
+
+
+def clear_all_read() -> None:
+    """Remove all read messages from the inbox."""
+    data = load_inbox()
+    messages = data.get("messages", [])
+
+    before_count = len(messages)
+    data["messages"] = [m for m in messages if not m.get("read")]
+    after_count = len(data["messages"])
+    removed = before_count - after_count
+
+    data["total_messages"] = after_count
+
+    save_inbox(data)
+
+    if removed == 0:
+        console.print("[dim]No read messages to clear.[/dim]")
+    else:
+        console.print(f"[green]Cleared {removed} read message(s).[/green]")
+
+
+def get_summary() -> str:
+    """Return a summary string of inbox status.
+
+    Returns:
+        str: Summary like "3 messages, 2 unread" or "No feedback messages."
+    """
+    data = load_inbox()
+    total = data.get("total_messages", 0)
+    unread = data.get("unread_count", 0)
+
+    if total == 0:
+        return "No feedback messages."
+
+    return f"{total} message{'s' if total != 1 else ''}, {unread} unread"
+
+
+def _find_message(messages: list, msg_id: str) -> dict | None:
+    """Find a message by ID in the messages list.
+
+    Args:
+        messages: List of message dicts.
+        msg_id: The message ID to find.
+
+    Returns:
+        The message dict if found, None otherwise.
+    """
+    for msg in messages:
+        if msg.get("id") == msg_id:
+            return msg
+    return None

src/aipass/devpulse/apps/handlers/feedback/storage.py

@@ -0,0 +1,92 @@
+# META
+# module: devpulse.feedback
+# description: JSON persistence layer for feedback inbox
+# END META
+
+"""
+Feedback Storage — JSON persistence for devpulse feedback inbox.
+
+Handles reading, writing, and ID generation for feedback messages.
+Data lives in devpulse/.feedback.local/inbox.json.
+"""
+
+import json
+import secrets
+from pathlib import Path
+
+from aipass.prax import logger
+
+# devpulse/ root (three levels up from this file: handlers/feedback/storage.py -> apps/ -> devpulse/)
+_DEVPULSE_ROOT = Path(__file__).resolve().parents[3]
+
+FEEDBACK_DIR = _DEVPULSE_ROOT / ".feedback.local"
+
+
+def get_inbox_path() -> Path:
+    """Return the path to the feedback inbox.json file."""
+    return FEEDBACK_DIR / "inbox.json"
+
+
+def _ensure_dir() -> None:
+    """Ensure the .feedback.local/ directory exists."""
+    FEEDBACK_DIR.mkdir(parents=True, exist_ok=True)
+
+
+def load_inbox() -> dict:
+    """Load the feedback inbox from disk.
+
+    Creates an empty inbox if the file does not exist.
+
+    Returns:
+        dict: The inbox data with mailbox, total_messages, unread_count, and messages.
+    """
+    _ensure_dir()
+    inbox_path = get_inbox_path()
+
+    if not inbox_path.exists():
+        empty_inbox = {
+            "mailbox": "feedback",
+            "total_messages": 0,
+            "unread_count": 0,
+            "messages": [],
+        }
+        save_inbox(empty_inbox)
+        return empty_inbox
+
+    try:
+        with open(inbox_path, encoding="utf-8") as f:
+            return json.load(f)
+    except (json.JSONDecodeError, OSError) as e:
+        logger.error(f"[FEEDBACK] Failed to load inbox: {e}")
+        return {
+            "mailbox": "feedback",
+            "total_messages": 0,
+            "unread_count": 0,
+            "messages": [],
+        }
+
+
+def save_inbox(data: dict) -> None:
+    """Write the feedback inbox to disk.
+
+    Args:
+        data: The inbox dict to persist.
+    """
+    _ensure_dir()
+    inbox_path = get_inbox_path()
+
+    try:
+        with open(inbox_path, "w", encoding="utf-8") as f:
+            json.dump(data, f, indent=2)
+            f.write("\n")
+    except OSError as e:
+        logger.error(f"[FEEDBACK] Failed to save inbox: {e}")
+
+
+def generate_id() -> str:
+    """Generate an 8-character hex message ID.
+
+    Returns:
+        str: An 8-character lowercase hexadecimal string.
+    """
+    return secrets.token_hex(4)

src/aipass/devpulse/apps/modules/feedback.py

@@ -0,0 +1,147 @@
+# META
+# module: devpulse.feedback
+# description: Command routing module for feedback mailbox
+# END META
+
+"""
+Feedback Module — command routing for devpulse's personal feedback mailbox.
+
+Auto-discovered by devpulse.py via handle_command() convention.
+Routes feedback subcommands to the appropriate handler functions.
+"""
+
+import shlex
+
+from rich.console import Console
+
+from aipass.prax import logger
+from aipass.devpulse.apps.handlers.feedback.inbox import (
+    list_messages,
+    view_message,
+    clear_message,
+    clear_all_read,
+    get_summary,
+)
+from aipass.devpulse.apps.handlers.feedback.compose import (
+    send_feedback,
+    reply_to,
+    _resolve_sender,
+)
+
+console = Console(stderr=True)
+
+HELP_TEXT = """\
+[bold cyan]feedback[/bold cyan] — DevPulse personal feedback mailbox
+
+[bold]Usage:[/bold]
+  feedback                        Inbox summary (count, unread)
+  feedback inbox                  List all messages
+  feedback view <id>              Read message + thread
+  feedback reply <id> "message"   Reply to sender
+  feedback send "subject" "body"  Receive feedback from agent
+  feedback clear <id>             Remove a message
+  feedback clear --all            Remove all read messages
+  feedback --help                 Show this help
+"""
+
+
+def handle_command(command: str, args: list[str]) -> bool:
+    """Route feedback commands to handler functions.
+
+    Auto-discovered by devpulse.py module loader.
+
+    Args:
+        command: The primary command string.
+        args: Additional arguments after the command.
+
+    Returns:
+        bool: True if the command was handled, False otherwise.
+    """
+    if command != "feedback":
+        return False
+
+    # No subcommand — show summary
+    if not args:
+        summary = get_summary()
+        console.print(f"[bold cyan]Feedback:[/bold cyan] {summary}")
+        return True
+
+    subcommand = args[0]
+    sub_args = args[1:]
+
+    if subcommand in ("--help", "-h", "help"):
+        console.print(HELP_TEXT)
+        return True
+
+    if subcommand == "inbox":
+        list_messages()
+        return True
+
+    if subcommand == "view":
+        if not sub_args:
+            console.print("[red]Usage: feedback view <id>[/red]")
+            return True
+        view_message(sub_args[0])
+        return True
+
+    if subcommand == "reply":
+        if len(sub_args) < 2:
+            console.print('[red]Usage: feedback reply <id> "message"[/red]')
+            return True
+        msg_id = sub_args[0]
+        body = " ".join(sub_args[1:])
+        reply_to(msg_id, body)
+        return True
+
+    if subcommand == "send":
+        return _handle_send(sub_args)
+
+    if subcommand == "clear":
+        if not sub_args:
+            console.print("[red]Usage: feedback clear <id> | feedback clear --all[/red]")
+            return True
+        if sub_args[0] == "--all":
+            clear_all_read()
+        else:
+            clear_message(sub_args[0])
+        return True
+
+    console.print(f"[red]Unknown feedback subcommand: {subcommand}[/red]")
+    console.print("Use [bold]feedback --help[/bold] for usage.")
+    return True
+
+
+def _handle_send(args: list[str]) -> bool:
+    """Handle the send subcommand, parsing from_branch, subject, and body.
+
+    Expected format: send "subject" "body"
+    The from_branch is extracted from the first arg or defaults to 'unknown'.
+
+    Args:
+        args: Arguments after 'send'.
+
+    Returns:
+        bool: Always True (command was handled).
+    """
+    if len(args) < 2:
+        console.print('[red]Usage: feedback send "subject" "body"[/red]')
+        console.print("[dim]Tip: from_branch is auto-detected or pass as first arg.[/dim]")
+        return True
+
+    # Auto-detect sender from drone env vars, fall back to arg parsing
+    auto_branch, auto_path = _resolve_sender()
+
+    if len(args) >= 3 and not args[0].startswith('"'):
+        # Explicit from_branch provided as first arg
+        from_branch = args[0]
+        subject = args[1]
+        body = " ".join(args[2:])
+        ai_mail_path = auto_path if from_branch == auto_branch else ""
+    else:
+        from_branch = auto_branch
+        subject = args[0]
+        body = " ".join(args[1:])
+        ai_mail_path = auto_path
+
+    send_feedback(from_branch, subject, body, ai_mail_path)
+    return True

src/aipass/devpulse/tests/test_feedback_compose.py

@@ -0,0 +1,216 @@
+# META
+# module: devpulse.feedback
+# description: Tests for feedback compose operations
+# END META
+
+"""Tests for feedback compose — send, reply, ai_mail delivery."""
+
+import json
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from aipass.devpulse.apps.handlers.feedback import storage, compose
+
+
+@pytest.fixture
+def mock_feedback_dir(tmp_path):
+    """Patch FEEDBACK_DIR to use tmp_path for isolation."""
+    feedback_dir = tmp_path / ".feedback.local"
+    with patch.object(storage, "FEEDBACK_DIR", feedback_dir):
+        yield feedback_dir
+
+
+@pytest.fixture
+def mock_aipass_root(tmp_path):
+    """Patch _AIPASS_ROOT to use tmp_path for ai_mail delivery tests."""
+    with patch.object(compose, "_AIPASS_ROOT", tmp_path):
+        yield tmp_path
+
+
+@pytest.fixture
+def empty_inbox(mock_feedback_dir):
+    """Start with an empty feedback inbox."""
+    storage.save_inbox({
+        "mailbox": "feedback",
+        "total_messages": 0,
+        "unread_count": 0,
+        "messages": [],
+    })
+
+
+class TestSendFeedback:
+    """Tests for send_feedback()."""
+
+    def test_creates_message(self, empty_inbox, mock_feedback_dir):
+        """Should add a message to the inbox."""
+        msg_id = compose.send_feedback("seedgo", "Test subject", "Test body")
+
+        data = storage.load_inbox()
+        assert data["total_messages"] == 1
+        assert data["unread_count"] == 1
+        assert len(data["messages"]) == 1
+
+        msg = data["messages"][0]
+        assert msg["id"] == msg_id
+        assert msg["from"] == "seedgo"
+        assert msg["subject"] == "Test subject"
+        assert msg["body"] == "Test body"
+        assert msg["read"] is False
+        assert msg["thread"] == []
+
+    def test_returns_message_id(self, empty_inbox, mock_feedback_dir):
+        """Should return a valid 8-char hex ID."""
+        msg_id = compose.send_feedback("prax", "Subject", "Body")
+        assert isinstance(msg_id, str)
+        assert len(msg_id) == 8
+
+    def test_increments_counts(self, empty_inbox, mock_feedback_dir):
+        """Should increment total_messages and unread_count."""
+        compose.send_feedback("a", "s1", "b1")
+        compose.send_feedback("b", "s2", "b2")
+
+        data = storage.load_inbox()
+        assert data["total_messages"] == 2
+        assert data["unread_count"] == 2
+
+    def test_timestamp_format(self, empty_inbox, mock_feedback_dir):
+        """Should set an ISO-format timestamp."""
+        compose.send_feedback("flow", "Subject", "Body")
+
+        data = storage.load_inbox()
+        ts = data["messages"][0]["timestamp"]
+        # Should be ISO format: YYYY-MM-DDTHH:MM:SS
+        assert "T" in ts
+        assert len(ts) == 19
+
+
+class TestReplyTo:
+    """Tests for reply_to()."""
+
+    @pytest.fixture
+    def inbox_with_message(self, mock_feedback_dir):
+        """Create inbox with a single message to reply to."""
+        storage.save_inbox({
+            "mailbox": "feedback",
+            "total_messages": 1,
+            "unread_count": 1,
+            "messages": [
+                {
+                    "id": "aaa11111",
+                    "from": "seedgo",
+                    "subject": "Test feedback",
+                    "body": "Original message.",
+                    "timestamp": "2026-04-11T10:00:00",
+                    "read": True,
+                    "thread": [],
+                },
+            ],
+        })
+
+    def test_adds_reply_to_thread(self, inbox_with_message, mock_aipass_root):
+        """Should append reply to the message thread."""
+        result = compose.reply_to("aaa11111", "Great point, thanks!")
+
+        assert result is True
+        data = storage.load_inbox()
+        msg = data["messages"][0]
+        assert len(msg["thread"]) == 1
+        assert msg["thread"][0]["from"] == "devpulse"
+        assert msg["thread"][0]["body"] == "Great point, thanks!"
+
+    def test_reply_to_nonexistent(self, inbox_with_message, mock_aipass_root):
+        """Should return False for nonexistent message."""
+        result = compose.reply_to("zzz99999", "Reply")
+        assert result is False
+
+    def test_multiple_replies_build_thread(self, inbox_with_message, mock_aipass_root):
+        """Should accumulate replies in thread order."""
+        compose.reply_to("aaa11111", "Reply 1")
+        compose.reply_to("aaa11111", "Reply 2")
+
+        data = storage.load_inbox()
+        thread = data["messages"][0]["thread"]
+        assert len(thread) == 2
+        assert thread[0]["body"] == "Reply 1"
+        assert thread[1]["body"] == "Reply 2"
+
+
+class TestAiMailDelivery:
+    """Tests for ai_mail delivery on reply."""
+
+    @pytest.fixture
+    def inbox_with_message(self, mock_feedback_dir):
+        """Create inbox with a message from seedgo."""
+        storage.save_inbox({
+            "mailbox": "feedback",
+            "total_messages": 1,
+            "unread_count": 0,
+            "messages": [
+                {
+                    "id": "aaa11111",
+                    "from": "seedgo",
+                    "subject": "Test feedback",
+                    "body": "Original.",
+                    "timestamp": "2026-04-11T10:00:00",
+                    "read": True,
+                    "thread": [],
+                },
+            ],
+        })
+
+    def test_delivers_to_ai_mail(self, inbox_with_message, mock_aipass_root):
+        """Should write reply to sender's ai_mail inbox."""
+        # Create sender's ai_mail directory and inbox
+        ai_mail_dir = mock_aipass_root / "seedgo" / ".ai_mail.local"
+        ai_mail_dir.mkdir(parents=True)
+        ai_mail_inbox = ai_mail_dir / "inbox.json"
+        with open(ai_mail_inbox, "w", encoding="utf-8") as f:
+            json.dump({
+                "mailbox": "inbox",
+                "total_messages": 0,
+                "unread_count": 0,
+                "messages": [],
+            }, f)
+
+        compose.reply_to("aaa11111", "Thanks for the feedback!")
+
+        with open(ai_mail_inbox, encoding="utf-8") as f:
+            mail_data = json.load(f)
+
+        assert mail_data["total_messages"] == 1
+        assert mail_data["unread_count"] == 1
+
+        mail_msg = mail_data["messages"][0]
+        assert mail_msg["from"] == "devpulse"
+        assert mail_msg["to"] == "seedgo"
+        assert mail_msg["subject"] == "Re: Test feedback"
+        assert mail_msg["body"] == "Thanks for the feedback!"
+        assert mail_msg["metadata"]["source"] == "feedback"
+        assert mail_msg["metadata"]["thread_id"] == "aaa11111"
+
+    def test_skips_delivery_when_no_ai_mail(self, inbox_with_message, mock_aipass_root):
+        """Should log warning and skip when sender has no ai_mail inbox."""
+        # Don't create sender's ai_mail directory
+        result = compose.reply_to("aaa11111", "Reply without delivery")
+
+        # Reply should still succeed locally
+        assert result is True
+        data = storage.load_inbox()
+        assert len(data["messages"][0]["thread"]) == 1
+
+    def test_skips_delivery_on_corrupt_ai_mail(self, inbox_with_message, mock_aipass_root):
+        """Should skip delivery when sender's ai_mail inbox is corrupt."""
+        ai_mail_dir = mock_aipass_root / "seedgo" / ".ai_mail.local"
+        ai_mail_dir.mkdir(parents=True)
+        ai_mail_inbox = ai_mail_dir / "inbox.json"
+        with open(ai_mail_inbox, "w", encoding="utf-8") as f:
+            f.write("{corrupt json!!!")
+
+        result = compose.reply_to("aaa11111", "Reply with corrupt ai_mail")
+
+        # Reply should still succeed locally
+        assert result is True
+        data = storage.load_inbox()
+        assert len(data["messages"][0]["thread"]) == 1

src/aipass/devpulse/tests/test_feedback_inbox.py

@@ -0,0 +1,235 @@
+# META
+# module: devpulse.feedback
+# description: Tests for feedback inbox operations
+# END META
+
+"""Tests for feedback inbox — list, view, clear, summary."""
+
+import json
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from aipass.devpulse.apps.handlers.feedback import storage, inbox
+
+
+@pytest.fixture
+def mock_feedback_dir(tmp_path):
+    """Patch FEEDBACK_DIR to use tmp_path for isolation."""
+    feedback_dir = tmp_path / ".feedback.local"
+    with patch.object(storage, "FEEDBACK_DIR", feedback_dir):
+        yield feedback_dir
+
+
+@pytest.fixture
+def populated_inbox(mock_feedback_dir):
+    """Create an inbox with sample messages."""
+    data = {
+        "mailbox": "feedback",
+        "total_messages": 3,
+        "unread_count": 2,
+        "messages": [
+            {
+                "id": "aaa11111",
+                "from": "seedgo",
+                "subject": "Audit suggestion",
+                "body": "Consider adding a new checker.",
+                "timestamp": "2026-04-11T10:00:00",
+                "read": False,
+                "thread": [],
+            },
+            {
+                "id": "bbb22222",
+                "from": "prax",
+                "subject": "Logger update",
+                "body": "New log format available.",
+                "timestamp": "2026-04-11T11:00:00",
+                "read": True,
+                "thread": [
+                    {
+                        "from": "devpulse",
+                        "body": "Thanks, will review.",
+                        "timestamp": "2026-04-11T11:30:00",
+                    }
+                ],
+            },
+            {
+                "id": "ccc33333",
+                "from": "flow",
+                "subject": "Plan status",
+                "body": "FPLAN-042 completed.",
+                "timestamp": "2026-04-11T12:00:00",
+                "read": False,
+                "thread": [],
+            },
+        ],
+    }
+    storage.save_inbox(data)
+    return data
+
+
+class TestListMessages:
+    """Tests for list_messages()."""
+
+    def test_empty_inbox(self, mock_feedback_dir, capsys):
+        """Should print 'no messages' for empty inbox."""
+        storage.save_inbox({
+            "mailbox": "feedback",
+            "total_messages": 0,
+            "unread_count": 0,
+            "messages": [],
+        })
+        list_messages_output = inbox.list_messages()
+        # list_messages prints to stderr via Rich Console
+        # We just verify it doesn't raise
+
+    def test_lists_all_messages(self, populated_inbox):
+        """Should not raise when listing populated inbox."""
+        # list_messages outputs to stderr via Rich Console; no assertion on output
+        # Just verify it runs without error
+        inbox.list_messages()
+
+
+class TestViewMessage:
+    """Tests for view_message()."""
+
+    def test_view_marks_as_read(self, populated_inbox):
+        """Should mark an unread message as read."""
+        inbox.view_message("aaa11111")
+
+        data = storage.load_inbox()
+        msg = next(m for m in data["messages"] if m["id"] == "aaa11111")
+        assert msg["read"] is True
+
+    def test_view_decrements_unread_count(self, populated_inbox):
+        """Should decrement unread_count when marking as read."""
+        inbox.view_message("aaa11111")
+
+        data = storage.load_inbox()
+        assert data["unread_count"] == 1  # Was 2, now 1
+
+    def test_view_already_read_no_change(self, populated_inbox):
+        """Should not change unread_count for already-read messages."""
+        inbox.view_message("bbb22222")
+
+        data = storage.load_inbox()
+        assert data["unread_count"] == 2  # Unchanged
+
+    def test_view_nonexistent_message(self, populated_inbox):
+        """Should handle nonexistent message ID gracefully."""
+        # Should not raise
+        inbox.view_message("zzz99999")
+
+    def test_view_message_with_thread(self, populated_inbox):
+        """Should display thread replies without error."""
+        inbox.view_message("bbb22222")
+
+
+class TestClearMessage:
+    """Tests for clear_message()."""
+
+    def test_clear_removes_message(self, populated_inbox):
+        """Should remove the specified message."""
+        inbox.clear_message("aaa11111")
+
+        data = storage.load_inbox()
+        ids = [m["id"] for m in data["messages"]]
+        assert "aaa11111" not in ids
+        assert data["total_messages"] == 2
+
+    def test_clear_unread_decrements_count(self, populated_inbox):
+        """Should decrement unread_count when clearing an unread message."""
+        inbox.clear_message("aaa11111")
+
+        data = storage.load_inbox()
+        assert data["unread_count"] == 1
+
+    def test_clear_read_message_no_unread_change(self, populated_inbox):
+        """Should not change unread_count when clearing a read message."""
+        inbox.clear_message("bbb22222")
+
+        data = storage.load_inbox()
+        assert data["unread_count"] == 2  # Was unread=2, cleared a read msg
+
+    def test_clear_nonexistent_message(self, populated_inbox):
+        """Should handle nonexistent message ID gracefully."""
+        inbox.clear_message("zzz99999")
+
+        data = storage.load_inbox()
+        assert data["total_messages"] == 3  # Unchanged
+
+
+class TestClearAllRead:
+    """Tests for clear_all_read()."""
+
+    def test_removes_read_messages(self, populated_inbox):
+        """Should remove all messages with read=True."""
+        inbox.clear_all_read()
+
+        data = storage.load_inbox()
+        assert data["total_messages"] == 2
+        ids = [m["id"] for m in data["messages"]]
+        assert "bbb22222" not in ids  # Was read
+        assert "aaa11111" in ids  # Was unread
+        assert "ccc33333" in ids  # Was unread
+
+    def test_no_read_messages(self, mock_feedback_dir):
+        """Should handle inbox with no read messages."""
+        storage.save_inbox({
+            "mailbox": "feedback",
+            "total_messages": 1,
+            "unread_count": 1,
+            "messages": [
+                {"id": "x", "from": "a", "subject": "b", "body": "c",
+                 "timestamp": "2026-04-11T10:00:00", "read": False, "thread": []},
+            ],
+        })
+        inbox.clear_all_read()
+
+        data = storage.load_inbox()
+        assert data["total_messages"] == 1
+
+    def test_empty_inbox_clear_all(self, mock_feedback_dir):
+        """Should handle empty inbox gracefully."""
+        storage.save_inbox({
+            "mailbox": "feedback",
+            "total_messages": 0,
+            "unread_count": 0,
+            "messages": [],
+        })
+        inbox.clear_all_read()
+
+
+class TestGetSummary:
+    """Tests for get_summary()."""
+
+    def test_empty_inbox_summary(self, mock_feedback_dir):
+        """Should return 'No feedback messages.' for empty inbox."""
+        storage.save_inbox({
+            "mailbox": "feedback",
+            "total_messages": 0,
+            "unread_count": 0,
+            "messages": [],
+        })
+        result = inbox.get_summary()
+        assert result == "No feedback messages."
+
+    def test_populated_summary(self, populated_inbox):
+        """Should return correct count string."""
+        result = inbox.get_summary()
+        assert result == "3 messages, 2 unread"
+
+    def test_single_message_summary(self, mock_feedback_dir):
+        """Should use singular 'message' for count of 1."""
+        storage.save_inbox({
+            "mailbox": "feedback",
+            "total_messages": 1,
+            "unread_count": 0,
+            "messages": [
+                {"id": "x", "from": "a", "subject": "b", "body": "c",
+                 "timestamp": "2026-04-11T10:00:00", "read": True, "thread": []},
+            ],
+        })
+        result = inbox.get_summary()
+        assert result == "1 message, 0 unread"

src/aipass/devpulse/tests/test_feedback_module.py

@@ -0,0 +1,201 @@
+# META
+# module: devpulse.feedback
+# description: Tests for feedback module command routing
+# END META
+
+"""Tests for feedback module — command routing via handle_command()."""
+
+import json
+from unittest.mock import patch, MagicMock
+
+import pytest
+
+from aipass.devpulse.apps.handlers.feedback import storage
+from aipass.devpulse.apps.modules import feedback as feedback_module
+
+
+@pytest.fixture
+def mock_feedback_dir(tmp_path):
+    """Patch FEEDBACK_DIR to use tmp_path for isolation."""
+    feedback_dir = tmp_path / ".feedback.local"
+    with patch.object(storage, "FEEDBACK_DIR", feedback_dir):
+        yield feedback_dir
+
+
+@pytest.fixture
+def empty_inbox(mock_feedback_dir):
+    """Start with an empty feedback inbox."""
+    storage.save_inbox({
+        "mailbox": "feedback",
+        "total_messages": 0,
+        "unread_count": 0,
+        "messages": [],
+    })
+
+
+@pytest.fixture
+def populated_inbox(mock_feedback_dir):
+    """Create an inbox with sample messages."""
+    storage.save_inbox({
+        "mailbox": "feedback",
+        "total_messages": 2,
+        "unread_count": 1,
+        "messages": [
+            {
+                "id": "aaa11111",
+                "from": "seedgo",
+                "subject": "Test",
+                "body": "Body text.",
+                "timestamp": "2026-04-11T10:00:00",
+                "read": False,
+                "thread": [],
+            },
+            {
+                "id": "bbb22222",
+                "from": "prax",
+                "subject": "Already read",
+                "body": "Old message.",
+                "timestamp": "2026-04-11T09:00:00",
+                "read": True,
+                "thread": [],
+            },
+        ],
+    })
+
+
+class TestCommandRouting:
+    """Tests for handle_command() routing."""
+
+    def test_ignores_non_feedback_commands(self):
+        """Should return False for non-feedback commands."""
+        assert feedback_module.handle_command("status", []) is False
+        assert feedback_module.handle_command("help", []) is False
+        assert feedback_module.handle_command("mail", []) is False
+
+    def test_bare_feedback_shows_summary(self, empty_inbox):
+        """Should show summary for bare 'feedback' command."""
+        result = feedback_module.handle_command("feedback", [])
+        assert result is True
+
+    def test_feedback_help(self, empty_inbox):
+        """Should show help text."""
+        result = feedback_module.handle_command("feedback", ["--help"])
+        assert result is True
+
+    def test_feedback_help_alias(self, empty_inbox):
+        """Should accept 'help' as alias for --help."""
+        result = feedback_module.handle_command("feedback", ["help"])
+        assert result is True
+
+    def test_feedback_inbox(self, populated_inbox):
+        """Should list messages."""
+        result = feedback_module.handle_command("feedback", ["inbox"])
+        assert result is True
+
+    def test_feedback_view(self, populated_inbox):
+        """Should view a specific message."""
+        result = feedback_module.handle_command("feedback", ["view", "aaa11111"])
+        assert result is True
+
+        # Verify message was marked as read
+        data = storage.load_inbox()
+        msg = next(m for m in data["messages"] if m["id"] == "aaa11111")
+        assert msg["read"] is True
+
+    def test_feedback_view_no_id(self, populated_inbox):
+        """Should handle view without ID gracefully."""
+        result = feedback_module.handle_command("feedback", ["view"])
+        assert result is True
+
+    def test_feedback_clear(self, populated_inbox):
+        """Should clear a specific message."""
+        result = feedback_module.handle_command("feedback", ["clear", "bbb22222"])
+        assert result is True
+
+        data = storage.load_inbox()
+        ids = [m["id"] for m in data["messages"]]
+        assert "bbb22222" not in ids
+
+    def test_feedback_clear_all(self, populated_inbox):
+        """Should clear all read messages."""
+        result = feedback_module.handle_command("feedback", ["clear", "--all"])
+        assert result is True
+
+        data = storage.load_inbox()
+        assert all(not m["read"] for m in data["messages"])
+
+    def test_feedback_clear_no_args(self, populated_inbox):
+        """Should handle clear without args gracefully."""
+        result = feedback_module.handle_command("feedback", ["clear"])
+        assert result is True
+
+    def test_feedback_send(self, empty_inbox):
+        """Should accept feedback from an agent."""
+        result = feedback_module.handle_command(
+            "feedback", ["send", "seedgo", "Bug report", "Found an issue"]
+        )
+        assert result is True
+
+        data = storage.load_inbox()
+        assert data["total_messages"] == 1
+        assert data["messages"][0]["from"] == "seedgo"
+        assert data["messages"][0]["subject"] == "Bug report"
+
+    def test_feedback_send_without_from(self, empty_inbox):
+        """Should handle send with just subject and body."""
+        # When args start with a quoted-looking string, from defaults to 'unknown'
+        result = feedback_module.handle_command(
+            "feedback", ["send", "Subject here", "Body text"]
+        )
+        assert result is True
+
+        data = storage.load_inbox()
+        assert data["total_messages"] == 1
+
+    def test_feedback_send_too_few_args(self, empty_inbox):
+        """Should handle send with insufficient args."""
+        result = feedback_module.handle_command("feedback", ["send"])
+        assert result is True  # Handled (shows usage)
+
+        data = storage.load_inbox()
+        assert data["total_messages"] == 0
+
+    @patch.object(feedback_module, "reply_to")
+    def test_feedback_reply(self, mock_reply, populated_inbox):
+        """Should route reply command correctly."""
+        mock_reply.return_value = True
+        result = feedback_module.handle_command(
+            "feedback", ["reply", "aaa11111", "Good point!"]
+        )
+        assert result is True
+        mock_reply.assert_called_once_with("aaa11111", "Good point!")
+
+    def test_feedback_reply_no_args(self, populated_inbox):
+        """Should handle reply with insufficient args."""
+        result = feedback_module.handle_command("feedback", ["reply"])
+        assert result is True  # Handled (shows usage)
+
+    def test_feedback_reply_no_body(self, populated_inbox):
+        """Should handle reply with ID but no body."""
+        result = feedback_module.handle_command("feedback", ["reply", "aaa11111"])
+        assert result is True  # Handled (shows usage)
+
+    def test_unknown_subcommand(self, empty_inbox):
+        """Should handle unknown subcommands gracefully."""
+        result = feedback_module.handle_command("feedback", ["nonexistent"])
+        assert result is True  # Handled (shows error + hint)
+
+
+class TestHandleCommandHasCorrectSignature:
+    """Verify handle_command meets auto-discovery requirements."""
+
+    def test_handle_command_exists(self):
+        """Module must have handle_command function."""
+        assert hasattr(feedback_module, "handle_command")
+        assert callable(feedback_module.handle_command)
+
+    def test_handle_command_takes_two_args(self):
+        """handle_command must accept (command, args) signature."""
+        import inspect
+        sig = inspect.signature(feedback_module.handle_command)
+        assert len(sig.parameters) == 2

src/aipass/devpulse/tests/test_feedback_storage.py

@@ -0,0 +1,149 @@
+# META
+# module: devpulse.feedback
+# description: Tests for feedback storage layer
+# END META
+
+"""Tests for feedback storage — load, save, generate_id, directory creation."""
+
+import json
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from aipass.devpulse.apps.handlers.feedback import storage
+
+
+@pytest.fixture
+def mock_feedback_dir(tmp_path):
+    """Patch FEEDBACK_DIR to use tmp_path for isolation."""
+    feedback_dir = tmp_path / ".feedback.local"
+    with patch.object(storage, "FEEDBACK_DIR", feedback_dir):
+        yield feedback_dir
+
+
+class TestGetInboxPath:
+    """Tests for get_inbox_path()."""
+
+    def test_returns_path_object(self, mock_feedback_dir):
+        """Should return a Path instance."""
+        result = storage.get_inbox_path()
+        assert isinstance(result, Path)
+
+    def test_returns_inbox_json_under_feedback_dir(self, mock_feedback_dir):
+        """Should point to inbox.json inside the feedback directory."""
+        result = storage.get_inbox_path()
+        assert result == mock_feedback_dir / "inbox.json"
+
+
+class TestLoadInbox:
+    """Tests for load_inbox()."""
+
+    def test_creates_empty_inbox_when_missing(self, mock_feedback_dir):
+        """Should create a new inbox.json with empty state if file does not exist."""
+        result = storage.load_inbox()
+
+        assert result["mailbox"] == "feedback"
+        assert result["total_messages"] == 0
+        assert result["unread_count"] == 0
+        assert result["messages"] == []
+
+        # File should now exist on disk
+        inbox_path = mock_feedback_dir / "inbox.json"
+        assert inbox_path.exists()
+
+    def test_creates_directory_if_missing(self, mock_feedback_dir):
+        """Should create the .feedback.local/ directory if it does not exist."""
+        assert not mock_feedback_dir.exists()
+        storage.load_inbox()
+        assert mock_feedback_dir.exists()
+
+    def test_loads_existing_inbox(self, mock_feedback_dir):
+        """Should load and return existing inbox data."""
+        mock_feedback_dir.mkdir(parents=True, exist_ok=True)
+        inbox_data = {
+            "mailbox": "feedback",
+            "total_messages": 1,
+            "unread_count": 1,
+            "messages": [{"id": "abcd1234", "from": "seedgo", "subject": "test"}],
+        }
+        inbox_path = mock_feedback_dir / "inbox.json"
+        with open(inbox_path, "w", encoding="utf-8") as f:
+            json.dump(inbox_data, f)
+
+        result = storage.load_inbox()
+        assert result["total_messages"] == 1
+        assert result["messages"][0]["id"] == "abcd1234"
+
+    def test_handles_corrupt_json(self, mock_feedback_dir):
+        """Should return empty inbox when JSON is corrupt."""
+        mock_feedback_dir.mkdir(parents=True, exist_ok=True)
+        inbox_path = mock_feedback_dir / "inbox.json"
+        with open(inbox_path, "w", encoding="utf-8") as f:
+            f.write("{invalid json!!!")
+
+        result = storage.load_inbox()
+        assert result["total_messages"] == 0
+        assert result["messages"] == []
+
+
+class TestSaveInbox:
+    """Tests for save_inbox()."""
+
+    def test_saves_data_to_disk(self, mock_feedback_dir):
+        """Should write inbox data as JSON to disk."""
+        data = {
+            "mailbox": "feedback",
+            "total_messages": 2,
+            "unread_count": 1,
+            "messages": [{"id": "a"}, {"id": "b"}],
+        }
+        storage.save_inbox(data)
+
+        inbox_path = mock_feedback_dir / "inbox.json"
+        assert inbox_path.exists()
+
+        with open(inbox_path, encoding="utf-8") as f:
+            loaded = json.load(f)
+        assert loaded["total_messages"] == 2
+        assert len(loaded["messages"]) == 2
+
+    def test_creates_directory_on_save(self, mock_feedback_dir):
+        """Should create the directory if it does not exist during save."""
+        assert not mock_feedback_dir.exists()
+        storage.save_inbox({"mailbox": "feedback", "total_messages": 0, "unread_count": 0, "messages": []})
+        assert mock_feedback_dir.exists()
+
+    def test_overwrites_existing_file(self, mock_feedback_dir):
+        """Should overwrite existing data on save."""
+        storage.save_inbox({"mailbox": "feedback", "total_messages": 1, "unread_count": 0, "messages": [{"id": "x"}]})
+        storage.save_inbox({"mailbox": "feedback", "total_messages": 0, "unread_count": 0, "messages": []})
+
+        result = storage.load_inbox()
+        assert result["total_messages"] == 0
+        assert result["messages"] == []
+
+
+class TestGenerateId:
+    """Tests for generate_id()."""
+
+    def test_returns_string(self):
+        """Should return a string."""
+        result = storage.generate_id()
+        assert isinstance(result, str)
+
+    def test_returns_8_chars(self):
+        """Should return exactly 8 characters."""
+        result = storage.generate_id()
+        assert len(result) == 8
+
+    def test_returns_hex_string(self):
+        """Should return only hexadecimal characters."""
+        result = storage.generate_id()
+        assert all(c in "0123456789abcdef" for c in result)
+
+    def test_returns_unique_ids(self):
+        """Should generate different IDs on successive calls."""
+        ids = {storage.generate_id() for _ in range(100)}
+        # With 8 hex chars (32 bits), 100 IDs should all be unique
+        assert len(ids) == 100

src/aipass/prax/apps/modules/monitor_info.py

@@ -0,0 +1,104 @@
+# =================== AIPass ====================
+# Name: monitor_info.py
+# Description: Monitor Introspection & Help Display
+# Version: 0.1.0
+# Created: 2026-04-11
+# Modified: 2026-04-11
+# =============================================
+
+"""
+Monitor Info - Introspection and Help Display
+
+Extracted from monitor.py to keep the orchestration module under size limits.
+Contains pure display functions with no dependency on monitor module state.
+"""
+
+from aipass.cli.apps.modules import console
+from aipass.prax.apps.handlers.json import json_handler
+
+
+def print_introspection():
+    """Display module introspection - shows connected handlers and architecture"""
+    json_handler.log_operation("print_introspection", {"module": "monitor"})
+    console.print()
+    console.print("[bold cyan]PRAX Monitor Module[/bold cyan]")
+    console.print()
+    console.print("[yellow]Purpose:[/yellow]")
+    console.print("  Mission Control for autonomous branch monitoring")
+    console.print("  Unified console for file changes, logs, and module activity")
+    console.print()
+
+    console.print("[yellow]Connected Handlers (apps/handlers/monitoring/):[/yellow]")
+    console.print()
+    console.print("  [cyan]1. unified_stream.py[/cyan]")
+    console.print("     [dim]→ print_event() - Terminal output formatting[/dim]")
+    console.print()
+    console.print("  [cyan]2. branch_detector.py[/cyan]")
+    console.print("     [dim]→ detect_branch_from_path() - Path-to-branch mapping[/dim]")
+    console.print()
+    console.print("  [cyan]3. interactive_filter.py[/cyan]")
+    console.print("     [dim]→ FilterState, parse_command() - Runtime filtering[/dim]")
+    console.print()
+    console.print("  [cyan]4. monitoring_filters.py[/cyan]")
+    console.print("     [dim]→ should_monitor(), get_priority() - Event filtering[/dim]")
+    console.print()
+    console.print("  [cyan]5. event_queue.py[/cyan]")
+    console.print("     [dim]→ MonitoringEvent, MonitoringQueue - Event buffering[/dim]")
+    console.print()
+    console.print("  [cyan]6. module_tracker.py[/cyan]")
+    console.print("     [dim]→ ModuleTracker - Module execution tracking[/dim]")
+    console.print()
+    console.print("  [cyan]7. file watcher (threaded)[/cyan]")
+    console.print("     [dim]→ Real-time file change detection using watchdog[/dim]")
+    console.print("     [green]STATUS: Active - monitors ECOSYSTEM_ROOT recursively[/green]")
+    console.print()
+    console.print("  [cyan]8. log monitor (threaded)[/cyan]")
+    console.print("     [dim]→ Log stream processing from SYSTEM_LOGS_DIR[/dim]")
+    console.print("     [green]STATUS: Active - watches *.log files for new entries[/green]")
+    console.print()
+
+    console.print("[dim]Run 'drone @prax monitor --help' for usage[/dim]")
+    console.print()
+
+
+def print_help():
+    """Drone-compliant help output - command syntax and examples"""
+    console.print()
+    console.print("[bold cyan]PRAX Monitor - Unified Branch Monitoring[/bold cyan]")
+    console.print()
+
+    console.print("[yellow]Commands:[/yellow]")
+    console.print()
+    console.print("  [cyan]drone @prax monitor[/cyan]")
+    console.print("    Show module introspection")
+    console.print()
+    console.print("  [cyan]drone @prax monitor run[/cyan]")
+    console.print("    Start monitoring all branches")
+    console.print()
+    console.print("  [cyan]drone @prax monitor run all[/cyan]")
+    console.print("    Explicit all-branches monitoring")
+    console.print()
+    console.print("  [cyan]drone @prax monitor run [branches][/cyan]")
+    console.print("    Monitor specific branches (comma-separated)")
+    console.print("    Example: drone @prax monitor run seedgo,cli,flow")
+    console.print()
+    console.print("  [cyan]drone @prax monitor --help[/cyan]")
+    console.print("    Show this help")
+    console.print()
+
+    console.print("[yellow]Interactive Mode Commands:[/yellow]")
+    console.print()
+    console.print("  [cyan]help[/cyan]          Show available commands")
+    console.print("  [cyan]status[/cyan]        Display current monitoring state")
+    console.print("  [cyan]filter [branches][/cyan]  Adjust branch filter")
+    console.print("  [cyan]quit/exit[/cyan]     Stop monitoring")
+    console.print()
+
+    console.print("[yellow]Examples:[/yellow]")
+    console.print()
+    console.print("  [dim]# Monitor all branches[/dim]")
+    console.print("  $ drone @prax monitor run")
+    console.print()
+    console.print("  [dim]# Monitor specific branches[/dim]")
+    console.print("  $ drone @prax monitor run seedgo,cli,flow")
+    console.print()