#681 owner-cap PART4: watchdog+feedback gate on sealed-registry owner (is_owner), cross-project. The old cwd.name=='devpulse' check was a NO-OP through drone (routed module runs cwd=branch_path, so Path.cwd() is always devpulse; a @flow caller sailed through). New shared handlers/owner/guard.py resolves the REAL caller via AIPASS_CALLER_BRANCH/CWD and checks the frozen is_owner(email,start_path) contract — portable to any project. feedback send stays open (inbound channel); mailbox mgmt owner-only. Fail-safe: legacy devpulse-path heuristic when no owner sealed / resolver import fails. Live-verified owner-allow + flow-deny (both tools) + send-open. 18 tests (15 guard + 3 gate), branch audit 100%.

CHANGELOG.md

@@ -63,6 +63,25 @@ PyPI version — not the changelog header.
 
 ### Fixed
 
+- **Owner-capability PART 4 — devpulse's `watchdog` + `feedback` now gate on the
+  sealed-registry owner, and cross-project (issue #681).** Closes the
+  owner-capability model (#678): the last two owner-only tools were still gated
+  by a hardcoded `cwd.name == "devpulse"` check — which, it turns out, was a
+  **no-op through drone**: drone runs a routed module with `cwd=<branch_path>`,
+  so the module's own `Path.cwd()` is *always* the devpulse tree and can't
+  identify the caller (a `@flow` caller sailed straight through). A new shared
+  `handlers/owner/guard.py` resolves the *real* caller from the env drone sets
+  (`AIPASS_CALLER_BRANCH` / `AIPASS_CALLER_CWD`) and checks it against the sealed
+  owner via the frozen `is_owner(email, start_path)` contract — so it works in
+  any project (devpulse in AIPass, whoever owns elsewhere), not a hardcoded name.
+  `feedback send` stays open (it's the inbound channel any agent uses to drop
+  feedback to the owner); every mailbox read/manage verb is owner-only. Fail-safe:
+  if no owner is sealed yet (old/partial install) or the resolver can't import,
+  it falls back to the legacy devpulse-path heuristic so existing installs never
+  hard-break. Live-verified end-to-end: owner allowed, `@flow` denied on both
+  tools, `send` open. 18 new tests (15 guard + 3 gate), branch audit 100%.
+  (built + verified by devpulse)
+
 - **seedgo `json_structure` now sanctions `custom_config/` for operator-editable
   config (issue #643).** The standard said "`{branch}_json/` root, one directory,
   no splits" and the checker ignored subdirs, so `custom_config/` (home of

src/aipass/devpulse/.seedgo/bypass.json

@@ -10,6 +10,16 @@
       "file": "devpulse_json/compass",
       "reason": "compass/ is the devpulse-owned Compass decision store (SQLite/FTS5 — db + wal + shm) which needs its own directory. Legitimate data subdir, not operator-config (custom_config/) nor auto-gen root data. Sanctioned exception per #643."
     },
+    {
+      "standard": "handlers",
+      "file": "apps/handlers/owner/guard.py",
+      "reason": "Lazy-imports is_owner/get_owner from spawn.apps.handlers.registry inside _owner_decision() — is_owner is the frozen shared owner-capability contract (#191, TDPLAN-0012); spawn is its sole home, no modules re-export. Same authorized cross-branch primitive pattern as ai_mail dispatch_monitor. Import failure falls back to the legacy heuristic. #681."
+    },
+    {
+      "standard": "encapsulation",
+      "file": "apps/handlers/owner/guard.py",
+      "reason": "Lazy cross-branch import of the is_owner/get_owner resolver from spawn.apps.handlers.registry — the frozen owner-capability contract consumed identically by hooks + ai_mail. No spawn modules-level re-export exists; this is the sanctioned entry point. #681."
+    },
     {
       "standard": "architecture",
       "reason": "No 'manager' citizen_class template in spawn. Devpulse is the only manager branch."
@@ -50,6 +60,16 @@
       "file": "tests/test_git_gate.py",
       "reason": "Test imports git_gate.py from ~/.claude/hooks/ via importlib — external hook, not a branch module."
     },
+    {
+      "standard": "encapsulation",
+      "file": "tests/test_feedback_module.py",
+      "reason": "Router test imports the feedback storage handler directly to arrange/assert inbox state (save_inbox/load_inbox) — standard test-fixture access, same pattern as test_feedback_storage.py / test_compass_store.py. #681."
+    },
+    {
+      "standard": "encapsulation",
+      "file": "tests/test_owner_guard.py",
+      "reason": "Unit test imports the owner guard handler (its SUT) and patches spawn.registry's get_owner/is_owner — the guard is a shared handler primitive with no apps/modules/ command entry point. Same direct-SUT pattern as test_compass_store.py. #681."
+    },
     {
       "standard": "encapsulation",
       "file": "tools/spot_check.py",

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

@@ -0,0 +1,7 @@
+# =================== AIPass ====================
+# Name: __init__.py
+# Description: Owner-capability guard handlers package
+# Version: 1.0.0
+# Created: 2026-07-10
+# Modified: 2026-07-10
+# =============================================

src/aipass/devpulse/apps/handlers/owner/guard.py

@@ -0,0 +1,115 @@
+# =================== AIPass ====================
+# Name: guard.py
+# Description: Owner-capability caller guard for devpulse owner-only tools
+# Version: 1.0.0
+# Created: 2026-07-10
+# Modified: 2026-07-10
+# =============================================
+
+"""
+Owner-capability guard for devpulse's owner-only tools.
+
+watchdog and feedback's mailbox-management verbs are the project OWNER's
+tools. The catch: drone runs a routed module with ``cwd=<branch_path>`` (see
+drone router_handler), so the module's own ``Path.cwd()`` is ALWAYS the
+devpulse tree and can't identify who called. The real caller lives in the env
+drone sets — ``AIPASS_CALLER_BRANCH`` / ``AIPASS_CALLER_CWD``. This resolves
+that caller and checks it against the sealed-registry owner via ``is_owner``.
+
+Portable by construction: ``is_owner`` reads each project's OWN sealed
+registry, so "owner" is devpulse in AIPass and whoever owns elsewhere (e.g.
+@vera in Vera Studio) — no hardcoded name, no per-project scaffolding.
+
+Fail-safe: if the owner resolver is unavailable (import fails, or a project
+has no sealed owner yet — an old/partial install), it falls back to the legacy
+devpulse-path heuristic so existing installs never hard-break. Concretely #681.
+
+Returns a plain bool — the calling MODULE owns user-facing output (handlers
+don't print). Denials are audit-logged here.
+"""
+
+import os
+from pathlib import Path
+
+from aipass.prax import logger
+from aipass.devpulse.apps.handlers.json import json_handler
+
+
+def _resolve_caller() -> tuple[str, Path]:
+    """Resolve ``(caller_email, caller_cwd)`` from the env drone sets.
+
+    ``caller_email`` is ``@<branch>`` — a branch's address is ``@`` + its
+    directory name (matches ai_mail's identity resolution). Prefers the
+    ``AIPASS_CALLER_BRANCH`` env var; otherwise walks up ``AIPASS_CALLER_CWD``
+    for a ``.trinity/passport.json`` and uses that directory's name. Falls back
+    to the process cwd when no caller env is set (direct, non-drone invocation).
+
+    Returns:
+        tuple: (caller_email_or_empty, caller_cwd_path)
+    """
+    caller_cwd_env = os.environ.get("AIPASS_CALLER_CWD", "")
+    caller_cwd = Path(caller_cwd_env) if caller_cwd_env else Path.cwd()
+
+    branch = os.environ.get("AIPASS_CALLER_BRANCH", "")
+    if not branch:
+        for candidate in [caller_cwd, *caller_cwd.parents]:
+            if (candidate / ".trinity" / "passport.json").exists():
+                branch = candidate.name
+                break
+
+    email = f"@{branch.lstrip('@').lower()}" if branch else ""
+    return email, caller_cwd
+
+
+def _legacy_devpulse_heuristic(caller_cwd: Path) -> bool:
+    """Pre-owner behavior: allow only a caller standing in the devpulse tree.
+
+    Used solely as the fail-safe when the owner resolver can't decide, so
+    existing AIPass installs keep working before the sealed owner is present.
+    """
+    return caller_cwd.name == "devpulse" or any(p.name == "devpulse" for p in caller_cwd.parents)
+
+
+def _owner_decision(email: str, caller_cwd: Path) -> bool:
+    """Decide whether ``email`` is the owner of the project at ``caller_cwd``.
+
+    Owner check runs against the caller's OWN project registry (start_path =
+    caller_cwd), so cross-project calls resolve the caller's owner, not
+    AIPass's. Falls back to the legacy heuristic when the resolver is
+    unavailable or the project has no sealed owner yet.
+    """
+    try:
+        # is_owner is the frozen shared owner-capability contract (spawn is its
+        # sole home; no modules re-export). Lazy import keeps cold start fast and
+        # enables the fail-safe fallback below.
+        from aipass.spawn.apps.handlers.registry import get_owner, is_owner
+    except ImportError as exc:
+        logger.warning("[owner_guard] owner resolver unavailable (%s) — legacy heuristic", exc)
+        return _legacy_devpulse_heuristic(caller_cwd)
+
+    if get_owner(start_path=caller_cwd) is None:
+        # No sealed owner in this project -> resolver can't decide -> legacy path check.
+        logger.info("[owner_guard] no sealed owner at %s — legacy heuristic", caller_cwd)
+        return _legacy_devpulse_heuristic(caller_cwd)
+
+    return bool(email) and is_owner(email, start_path=caller_cwd)
+
+
+def guard_owner_caller(tool: str) -> bool:
+    """Gate an owner-only tool.
+
+    Args:
+        tool: Name of the calling tool (e.g. 'watchdog', 'feedback') for the
+            audit line. The caller is responsible for any user-facing message.
+
+    Returns:
+        bool: True to allow the call; False (after audit-logging the denial) to
+        reject a non-owner caller.
+    """
+    email, caller_cwd = _resolve_caller()
+    if _owner_decision(email, caller_cwd):
+        return True
+
+    json_handler.log_operation("owner_guard_denied", {"tool": tool, "caller": email or "unknown"})
+    logger.info("[owner_guard] %s denied non-owner caller=%s", tool, email or "unknown")
+    return False

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

@@ -27,7 +27,7 @@ from aipass.devpulse.apps.handlers.feedback.compose import (
 )
 
 from aipass.prax import logger
-from aipass.cli.apps.modules import err_console, error
+from aipass.cli.apps.modules import err_console, error, warning
 from aipass.devpulse.apps.handlers.json import json_handler
 
 console = err_console
@@ -47,6 +47,21 @@ HELP_TEXT = """\
 """
 
 
+def _guard_caller() -> bool:
+    """Owner-only gate for mailbox reads/management (see handlers.owner.guard).
+
+    The mailbox belongs to the project owner. `send` and `--help` stay open
+    (send is the inbound channel any agent uses to drop feedback here); every
+    other verb reads or mutates the owner's mail and is owner-gated. #681.
+    """
+    from aipass.devpulse.apps.handlers.owner.guard import guard_owner_caller
+
+    if guard_owner_caller("feedback"):
+        return True
+    warning("feedback mailbox management is owner-only — refusing non-owner call")
+    return False
+
+
 def print_introspection() -> None:
     """Display module introspection info."""
     console.print()
@@ -73,6 +88,20 @@ def handle_command(command: str, args: list[str]) -> bool:
     if command != "feedback":
         return False
 
+    # Open verbs (no owner gate): help + `send`. `send` is the inbound channel
+    # any agent uses to drop feedback into the owner's mailbox. Everything else
+    # reads or manages that mailbox -> owner-only (#681).
+    if args and args[0] in ("--help", "-h", "help"):
+        console.print(HELP_TEXT)
+        return True
+
+    if args and args[0] == "send":
+        json_handler.log_operation("feedback_command", {"subcommand": "send"})
+        return _handle_send(args[1:])
+
+    if not _guard_caller():
+        return True
+
     if not args:
         print_introspection()
         summary = get_summary()
@@ -83,10 +112,6 @@ def handle_command(command: str, args: list[str]) -> bool:
     sub_args = args[1:]
     json_handler.log_operation("feedback_command", {"subcommand": subcommand})
 
-    if subcommand in ("--help", "-h", "help"):
-        console.print(HELP_TEXT)
-        return True
-
     if subcommand == "inbox":
         list_messages()
         return True
@@ -107,9 +132,6 @@ def handle_command(command: str, args: list[str]) -> bool:
         reply_to(msg_id, body)
         return True
 
-    if subcommand == "send":
-        return _handle_send(sub_args)
-
     if subcommand == "clear":
         if not sub_args:
             error("Usage: feedback clear <id> | feedback clear --all")

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

@@ -25,7 +25,6 @@ See FPLAN-0186 for the build plan and DPLAN-0130 for the design record.
 """
 
 import importlib
-from pathlib import Path
 from typing import List
 
 from aipass.prax.apps.modules.logger import system_logger as logger
@@ -113,11 +112,18 @@ def print_introspection() -> None:
 
 
 def _guard_caller() -> bool:
-    """Reject cross-branch invocation. Devpulse-only tool."""
-    cwd = Path.cwd()
-    if cwd.name == "devpulse" or any(p.name == "devpulse" for p in cwd.parents):
+    """Reject non-owner invocation. Owner-only tool.
+
+    Gates on the PROJECT OWNER (sealed registry) via the shared owner guard, so
+    it works across projects — not a hardcoded 'devpulse' name. (Drone runs a
+    routed module with cwd=<branch_path>, so Path.cwd() can't identify the real
+    caller; the guard reads the AIPASS_CALLER_* env drone sets.) #681.
+    """
+    from aipass.devpulse.apps.handlers.owner.guard import guard_owner_caller
+
+    if guard_owner_caller("watchdog"):
         return True
-    warning("watchdog is a devpulse-only module — refusing cross-branch call")
+    warning("watchdog is an owner-only module — refusing non-owner call")
     return False
 
 

src/aipass/devpulse/tests/test_feedback_module.py

@@ -1,7 +1,10 @@
-# META
-# module: devpulse.feedback
-# description: Tests for feedback module command routing
-# END META
+# =================== AIPass ====================
+# Name: test_feedback_module.py
+# Description: Tests for feedback module command routing
+# Version: 1.0.0
+# Created: 2026-04-11
+# Modified: 2026-07-10
+# =============================================
 
 """Tests for feedback module — command routing via handle_command()."""
 
@@ -13,6 +16,13 @@ from aipass.devpulse.apps.handlers.feedback import storage
 from aipass.devpulse.apps.modules import feedback as feedback_module
 
 
+@pytest.fixture(autouse=True)
+def _bypass_caller_guard():
+    """Force _guard_caller to pass so routing tests don't depend on owner env."""
+    with patch.object(feedback_module, "_guard_caller", return_value=True):
+        yield
+
+
 @pytest.fixture
 def mock_feedback_dir(tmp_path):
     """Patch FEEDBACK_DIR to use tmp_path for isolation."""
@@ -183,6 +193,35 @@ class TestCommandRouting:
         assert result is True  # Handled (shows error + hint)
 
 
+class TestOwnerGate:
+    """Owner gate wraps mailbox management; send + help stay open (#681)."""
+
+    def test_management_blocked_for_non_owner(self, populated_inbox):
+        """A denied guard blocks a management verb — view does not mark read."""
+        with patch.object(feedback_module, "_guard_caller", return_value=False):
+            result = feedback_module.handle_command("feedback", ["view", "aaa11111"])
+        assert result is True  # command still "handled" (clean refusal)
+
+        data = storage.load_inbox()
+        msg = next(m for m in data["messages"] if m["id"] == "aaa11111")
+        assert msg["read"] is False  # action was gated out
+
+    def test_send_open_for_non_owner(self, empty_inbox):
+        """send bypasses the owner gate — any agent can drop feedback."""
+        with patch.object(feedback_module, "_guard_caller", return_value=False):
+            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  # send bypassed the gate
+
+    def test_help_open_for_non_owner(self, empty_inbox):
+        """--help bypasses the owner gate."""
+        with patch.object(feedback_module, "_guard_caller", return_value=False):
+            result = feedback_module.handle_command("feedback", ["--help"])
+        assert result is True
+
+
 class TestHandleCommandHasCorrectSignature:
     """Verify handle_command meets auto-discovery requirements."""
 

src/aipass/devpulse/tests/test_owner_guard.py

@@ -0,0 +1,178 @@
+# =================== AIPass ====================
+# Name: test_owner_guard.py
+# Description: Tests for the shared owner-capability caller guard (#681)
+# Version: 1.0.0
+# Created: 2026-07-10
+# Modified: 2026-07-10
+# =============================================
+
+"""Tests for the shared owner-capability caller guard (handlers/owner/guard.py).
+
+Covers caller resolution from the drone env, the owner decision against a
+(patched) sealed registry — including cross-project ownership — and the
+legacy fail-safe fallback when no owner is sealed.
+"""
+
+from pathlib import Path
+
+import pytest
+
+from aipass.devpulse.apps.handlers.owner import guard as guard_mod
+
+
+@pytest.fixture(autouse=True)
+def _clear_caller_env(monkeypatch):
+    """Start each test with no caller env; tests set exactly what they need."""
+    monkeypatch.delenv("AIPASS_CALLER_BRANCH", raising=False)
+    monkeypatch.delenv("AIPASS_CALLER_CWD", raising=False)
+
+
+def _patch_registry(monkeypatch, owner_email):
+    """Patch spawn's owner resolver. owner_email=None => no sealed owner."""
+    import aipass.spawn.apps.handlers.registry as reg
+
+    def fake_get_owner(start_path=None):
+        """Stand-in for get_owner: owner entry dict, or None when unsealed."""
+        return {"email": owner_email} if owner_email else None
+
+    def _norm(email):
+        """Normalize an email to lowercase with a leading '@'."""
+        return (email if email.startswith("@") else f"@{email}").lower()
+
+    def fake_is_owner(email, start_path=None):
+        """Stand-in for is_owner: True iff email matches the sealed owner."""
+        if not owner_email or not email:
+            return False
+        return _norm(email) == _norm(owner_email)
+
+    monkeypatch.setattr(reg, "get_owner", fake_get_owner)
+    monkeypatch.setattr(reg, "is_owner", fake_is_owner)
+
+
+# ------------------------------------------------------------------
+# _resolve_caller
+# ------------------------------------------------------------------
+
+
+class TestResolveCaller:
+    """Caller identity resolution from the env drone sets."""
+
+    def test_uses_branch_env(self, monkeypatch, tmp_path):
+        """AIPASS_CALLER_BRANCH becomes the '@branch' email directly."""
+        monkeypatch.setenv("AIPASS_CALLER_BRANCH", "flow")
+        monkeypatch.setenv("AIPASS_CALLER_CWD", str(tmp_path))
+        email, cwd = guard_mod._resolve_caller()
+        assert email == "@flow"
+        assert cwd == tmp_path
+
+    def test_strips_and_lowercases(self, monkeypatch, tmp_path):
+        """A '@Mixed' branch env normalizes to lowercase, single leading '@'."""
+        monkeypatch.setenv("AIPASS_CALLER_BRANCH", "@DevPulse")
+        monkeypatch.setenv("AIPASS_CALLER_CWD", str(tmp_path))
+        email, _ = guard_mod._resolve_caller()
+        assert email == "@devpulse"
+
+    def test_walks_up_for_passport(self, monkeypatch, tmp_path):
+        """With no branch env, walk up the caller cwd to the passport dir name."""
+        branch = tmp_path / "mybranch"
+        (branch / ".trinity").mkdir(parents=True)
+        (branch / ".trinity" / "passport.json").write_text("{}", encoding="utf-8")
+        sub = branch / "apps" / "modules"
+        sub.mkdir(parents=True)
+        monkeypatch.setenv("AIPASS_CALLER_CWD", str(sub))
+        email, cwd = guard_mod._resolve_caller()
+        assert email == "@mybranch"
+        assert cwd == sub
+
+    def test_empty_when_no_branch_and_no_passport(self, monkeypatch, tmp_path):
+        """No branch env and no passport anywhere above => empty email."""
+        monkeypatch.setenv("AIPASS_CALLER_CWD", str(tmp_path))
+        email, _ = guard_mod._resolve_caller()
+        assert email == ""
+
+
+# ------------------------------------------------------------------
+# _legacy_devpulse_heuristic
+# ------------------------------------------------------------------
+
+
+class TestLegacyHeuristic:
+    """The pre-owner fail-safe: allow only a caller in the devpulse tree."""
+
+    def test_allows_devpulse_subtree(self, tmp_path):
+        """A path with a 'devpulse' ancestor is allowed."""
+        assert guard_mod._legacy_devpulse_heuristic(tmp_path / "devpulse" / "apps") is True
+
+    def test_allows_devpulse_leaf(self, tmp_path):
+        """A path whose leaf dir is 'devpulse' is allowed."""
+        assert guard_mod._legacy_devpulse_heuristic(tmp_path / "devpulse") is True
+
+    def test_rejects_other_branch(self, tmp_path):
+        """A path with no 'devpulse' component is rejected."""
+        assert guard_mod._legacy_devpulse_heuristic(tmp_path / "flow" / "apps") is False
+
+
+# ------------------------------------------------------------------
+# _owner_decision (patched resolver)
+# ------------------------------------------------------------------
+
+
+class TestOwnerDecision:
+    """The core owner check against a (patched) sealed registry."""
+
+    def test_allows_owner(self, monkeypatch, tmp_path):
+        """The sealed owner's email is allowed."""
+        _patch_registry(monkeypatch, "@devpulse")
+        assert guard_mod._owner_decision("@devpulse", tmp_path) is True
+
+    def test_rejects_non_owner(self, monkeypatch, tmp_path):
+        """A non-owner email is rejected when an owner is sealed."""
+        _patch_registry(monkeypatch, "@devpulse")
+        assert guard_mod._owner_decision("@flow", tmp_path) is False
+
+    def test_cross_project_owner(self, monkeypatch, tmp_path):
+        """Owner is per-project: @vera owns elsewhere, devpulse does not."""
+        _patch_registry(monkeypatch, "@vera")
+        assert guard_mod._owner_decision("@vera", tmp_path) is True
+        assert guard_mod._owner_decision("@devpulse", tmp_path) is False
+
+    def test_no_sealed_owner_falls_back_to_heuristic(self, monkeypatch):
+        """No sealed owner => legacy devpulse-path heuristic decides."""
+        _patch_registry(monkeypatch, None)  # get_owner -> None
+        assert guard_mod._owner_decision("@anyone", Path("/x/devpulse/y")) is True
+        assert guard_mod._owner_decision("@anyone", Path("/x/flow/y")) is False
+
+    def test_empty_email_rejected_when_owner_sealed(self, monkeypatch, tmp_path):
+        """An unresolved caller (empty email) is rejected when owner is sealed."""
+        _patch_registry(monkeypatch, "@devpulse")
+        assert guard_mod._owner_decision("", tmp_path) is False
+
+
+# ------------------------------------------------------------------
+# guard_owner_caller / is_owner_caller (end to end via env)
+# ------------------------------------------------------------------
+
+
+class TestGuardOwnerCaller:
+    """End-to-end gate behavior driven by the drone caller env."""
+
+    def test_allows_owner(self, monkeypatch, tmp_path):
+        """Owner caller => guard allows."""
+        _patch_registry(monkeypatch, "@devpulse")
+        monkeypatch.setenv("AIPASS_CALLER_BRANCH", "devpulse")
+        monkeypatch.setenv("AIPASS_CALLER_CWD", str(tmp_path))
+        assert guard_mod.guard_owner_caller("watchdog") is True
+
+    def test_denies_non_owner(self, monkeypatch, tmp_path):
+        """Non-owner caller => guard denies (and audit-logs)."""
+        _patch_registry(monkeypatch, "@devpulse")
+        monkeypatch.setenv("AIPASS_CALLER_BRANCH", "flow")
+        monkeypatch.setenv("AIPASS_CALLER_CWD", str(tmp_path))
+        assert guard_mod.guard_owner_caller("feedback") is False
+
+    def test_cross_project_owner_end_to_end(self, monkeypatch, tmp_path):
+        """A non-devpulse owner (@vera) is allowed in its own project."""
+        _patch_registry(monkeypatch, "@vera")
+        monkeypatch.setenv("AIPASS_CALLER_BRANCH", "vera")
+        monkeypatch.setenv("AIPASS_CALLER_CWD", str(tmp_path))
+        assert guard_mod.guard_owner_caller("watchdog") is True