refactor(drone): .trinity/-marker walk-ups replace src/aipass hardcodes — portable resolution + access checks (TDPLAN-0010, FPLAN-0296)

CHANGELOG.md

@@ -81,6 +81,21 @@ PyPI version — not the changelog header.
   degrades gracefully when `@memory` is unavailable (empty meta-tabs, no crash).
   297 tests pass. (built by @spawn, FPLAN-0294, TDPLAN-0010)
 
+- **Drone resolution + access checks made project-portable (TDPLAN-0010
+  foundation)** — five `src/aipass`/fixed-depth self-location hardcodes are
+  replaced with `.trinity/`-marker walk-ups: `rm_handler` sibling protection,
+  `commit_handler` test-gate branch detection, `broker/daemon` allowed-bases,
+  the `handlers/__init__` import-guard access check (now `is_relative_to()`
+  instead of scanning path parts for the literal `aipass`), and
+  `registry_handler`'s `parents[4]` last-resort (now a
+  `.git`/`pyproject.toml`/`setup.py`/`setup.cfg` marker walk). `@name`→path
+  resolution now works for an agent in any project layout via a CWD-first
+  registry walk (AIPASS_HOME only as a last resort when the CWD ancestry has no
+  registry at all). The `_validate_branch_path` containment invariant is
+  untouched — per-project isolation preserved. (Drone uses its own resolver, not
+  the shared `registry_discovery.py`.) 838 tests pass. (built by @drone,
+  FPLAN-0296, TDPLAN-0010)
+
 - **ai_mail routing made project-portable (TDPLAN-0010 foundation)** — the
   fixed-depth `_REPO_ROOT = parents[2].parents[2]` self-location in
   `email.py` / `email_send.py` / `dispatch.py` (4 sites) is replaced with the

src/aipass/drone/apps/handlers/__init__.py

@@ -41,13 +41,15 @@ def _find_real_caller():
     return None, None
 
 
+_BRANCH_ROOT = str(Path(__file__).resolve().parents[2])
+
+
 def _extract_branch_name(filepath: str) -> str:
-    """Extract branch name from a file path."""
-    parts = Path(filepath).parts
-    for i, part in enumerate(parts):
-        if part == "aipass":
-            if i + 1 < len(parts):
-                return parts[i + 1]
+    """Extract branch name from a file path by walking up to .trinity/."""
+    path = Path(filepath)
+    for parent in [path, *path.parents]:
+        if (parent / ".trinity").is_dir():
+            return parent.name
     return "unknown"
 
 
@@ -60,14 +62,13 @@ def _guard_branch_access():
     """
     caller_file, import_line = _find_real_caller()
 
-    # DEBUG: Print what we found
     import os
 
     if os.environ.get("AIPASS_DEBUG_GUARD"):
         import sys
 
-        print(f"[GUARD DEBUG] caller_file = {caller_file}", file=sys.stderr)
-        print(f"[GUARD DEBUG] import_line = {import_line}", file=sys.stderr)
+        sys.stderr.write(f"[GUARD DEBUG] caller_file = {caller_file}\n")
+        sys.stderr.write(f"[GUARD DEBUG] import_line = {import_line}\n")
 
     if caller_file is None:
         # Can't determine caller from real files
@@ -79,10 +80,7 @@ def _guard_branch_access():
                 return  # Allow command-line Python through
         return  # Allow if truly can't determine
 
-    # Check if caller is from our branch
-    # MY_BRANCH is "aipass.drone" (dotted), but filesystem uses "/aipass/drone/"
-    branch_path = "/" + MY_BRANCH.replace(".", "/") + "/"
-    if branch_path in caller_file.replace("\\", "/"):
+    if Path(caller_file).is_relative_to(_BRANCH_ROOT):
         return  # Same branch, allowed
 
     # External caller - block access
@@ -90,6 +88,7 @@ def _guard_branch_access():
     caller_filename = Path(caller_file).name
     blocked_import = import_line if import_line else "unknown"
 
+    module_api = f"{MY_BRANCH}.apps.modules"
     raise ImportError(
         f"\n{'=' * 60}\n"
         f"ACCESS DENIED: Cross-branch handler import blocked\n"
@@ -99,11 +98,7 @@ def _guard_branch_access():
         f"  Blocked:       {blocked_import}\n"
         f"\n"
         f"  Handlers are internal to their branch.\n"
-        f"  Use the module API instead:\n"
-        f"    from {MY_BRANCH}.apps.modules.<module> import <function>\n"
-        f"\n"
-        f"  Example:\n"
-        f"    from {MY_BRANCH}.apps.modules.logger import logger\n"
+        f"  Use the module API instead: {module_api}.<module>\n"
         f"\n"
         f"  For full standards guide:\n"
         f"    drone @seedgo handlers\n"

src/aipass/drone/apps/handlers/broker/daemon.py

@@ -27,6 +27,7 @@ from __future__ import annotations
 import hashlib
 import hmac as hmac_mod
 import json
+import os
 import secrets
 import socket
 import threading
@@ -50,8 +51,6 @@ _TMP_BASES = (Path("/tmp"), Path("/var/tmp"))
 
 def _find_project_root() -> Path | None:
     """Walk up from CWD to find *_REGISTRY.json; return its parent as project root."""
-    import os
-
     cwd = Path.cwd()
     for parent in [cwd, *cwd.parents]:
         if list(parent.glob("*_REGISTRY.json")):
@@ -166,11 +165,24 @@ class BrokerDaemon:
         if identity == "devpulse":
             bases.append(self._repo_root)
             return bases
-        branch_dir = self._repo_root / "src" / "aipass" / identity
-        if branch_dir.is_dir():
+        branch_dir = self._resolve_branch_dir(identity)
+        if branch_dir and branch_dir.is_dir():
             bases.append(branch_dir)
         return bases
 
+    def _resolve_branch_dir(self, identity: str) -> Path | None:
+        """Find a branch directory by name via .trinity/ marker walk."""
+        if self._repo_root is None:
+            return None
+        for root, dirs, _files in os.walk(self._repo_root):
+            depth = len(Path(root).relative_to(self._repo_root).parts)
+            if depth > 3:
+                dirs.clear()
+                continue
+            if Path(root).name == identity and (Path(root) / ".trinity").is_dir():
+                return Path(root).resolve()
+        return None
+
     def _handle_identify(self, req: BrokerRequest) -> tuple[BrokerResponse, str | None]:
         """Verify HMAC and bind identity to the connection."""
         audit_entry: dict = {

src/aipass/drone/apps/handlers/git/commit_handler.py

@@ -18,6 +18,18 @@ from aipass.drone.apps.handlers.json import json_handler
 from aipass.drone.apps.handlers.git.lock_handler import find_repo_root
 
 
+def _find_branch_for_path(filepath: str, repo_root: Path) -> tuple[str, Path] | None:
+    """Find which branch a changed file belongs to by walking up to .trinity/."""
+    abs_path = (repo_root / filepath).resolve()
+    root = repo_root.resolve()
+    for parent in [abs_path.parent, *abs_path.parent.parents]:
+        if not parent.is_relative_to(root):
+            break
+        if (parent / ".trinity").is_dir():
+            return parent.name, parent
+    return None
+
+
 def _run_test_gate(repo_root: Path) -> dict | None:
     """Run pytest for changed branches. Returns error dict if tests fail, None if all pass."""
     status_result = subprocess.run(
@@ -26,21 +38,22 @@ def _run_test_gate(repo_root: Path) -> dict | None:
         text=True,
         cwd=str(repo_root),
     )
-    changed_branches: set[str] = set()
+    changed_branches: dict[str, Path] = {}
     for line in status_result.stdout.splitlines():
         if len(line) < 4:
             continue
         filepath = line[3:].split(" -> ")[-1]
-        parts = Path(filepath).parts
-        if len(parts) >= 3 and parts[0] == "src" and parts[1] == "aipass":
-            changed_branches.add(parts[2])
+        result = _find_branch_for_path(filepath, repo_root)
+        if result:
+            name, branch_path = result
+            changed_branches[name] = branch_path
 
     venv_python = repo_root / ".venv" / "bin" / "python"
     python_bin = str(venv_python) if venv_python.exists() else "python3"
 
     failed_branches: list[tuple[str, str]] = []
     for branch_name in sorted(changed_branches):
-        test_dir = repo_root / "src" / "aipass" / branch_name / "tests"
+        test_dir = changed_branches[branch_name] / "tests"
         if not test_dir.is_dir():
             continue
         try:

src/aipass/drone/apps/handlers/registry_handler.py

@@ -144,13 +144,16 @@ def find_registry() -> Path:
         if hit is not None:
             return hit
 
-    # Fallback — use package-relative path; glob there too
-    fallback_dir = Path(__file__).resolve().parents[4]
+    # Fallback — walk up from this file to the project root (marker-based)
+    _markers = (".git", "pyproject.toml", "setup.py", "setup.cfg")
+    fallback_dir = Path(__file__).resolve().parent
+    for parent in [fallback_dir, *fallback_dir.parents]:
+        if any((parent / m).exists() for m in _markers):
+            fallback_dir = parent
+            break
     hit = _first_registry_in(fallback_dir)
     if hit is not None:
         return hit
-    # Ultimate fallback: return a conventional name so the caller
-    # gets a clear "not found" path in the error message.
     return fallback_dir / "AIPASS_REGISTRY.json"
 
 

src/aipass/drone/apps/handlers/rm_handler.py

@@ -67,16 +67,23 @@ def get_allowed_roots() -> list[Path]:
     return roots
 
 
+def _find_branch_root(path: Path, project_root: Path) -> Path | None:
+    """Walk up from *path* looking for .trinity/; return branch dir or None."""
+    root = project_root.resolve()
+    for parent in [path, *path.parents]:
+        if not parent.is_relative_to(root):
+            break
+        if (parent / ".trinity").is_dir():
+            return parent
+    return None
+
+
 def _detect_current_branch(project_root: Path | None) -> str | None:
     """Return the branch name the CWD lives in, or None."""
     if project_root is None:
         return None
-    cwd = Path.cwd().resolve()
-    aipass_src = project_root / "src" / "aipass"
-    if not cwd.is_relative_to(aipass_src):
-        return None
-    rel = cwd.relative_to(aipass_src)
-    return rel.parts[0] if rel.parts else None
+    branch_root = _find_branch_root(Path.cwd().resolve(), project_root)
+    return branch_root.name if branch_root else None
 
 
 def _resolve_git_dir(path: Path) -> Path | None:
@@ -109,14 +116,11 @@ def check_carveouts(resolved: Path, project_root: Path | None) -> tuple[bool, st
                 return True, "Protected: path resolves inside .git worktree gitdir"
 
     if project_root is not None:
-        aipass_src = project_root / "src" / "aipass"
-        if resolved.is_relative_to(aipass_src):
-            rel = resolved.relative_to(aipass_src)
-            if rel.parts:
-                target_branch = rel.parts[0]
-                current_branch = _detect_current_branch(project_root)
-                if current_branch is None or target_branch != current_branch:
-                    return True, (f"Protected: path is inside sibling branch src/aipass/{target_branch}/")
+        target_branch_root = _find_branch_root(resolved, project_root)
+        if target_branch_root is not None:
+            current_branch = _detect_current_branch(project_root)
+            if current_branch is None or target_branch_root.name != current_branch:
+                return True, f"Protected: path is inside sibling branch {target_branch_root.name}/"
 
     return False, ""
 

src/aipass/drone/tests/test_broker.py

@@ -90,11 +90,12 @@ def _send_identified(broker: BrokerDaemon, branch: str, req: BrokerRequest) -> B
 
 @pytest.fixture()
 def repo_root(tmp_path: Path) -> Path:
-    """Set up a mock repo root with branch directories."""
+    """Set up a mock repo root with branch directories marked by .trinity/."""
     root = tmp_path / "repo"
     root.mkdir()
     branch = root / "src" / "aipass" / "testbranch"
     branch.mkdir(parents=True)
+    (branch / ".trinity").mkdir()
     (branch / "deleteme.txt").write_text("delete me", encoding="utf-8")
     (branch / "subdir").mkdir()
     (branch / "subdir" / "nested.txt").write_text("nested", encoding="utf-8")
@@ -102,6 +103,7 @@ def repo_root(tmp_path: Path) -> Path:
     (branch / ".git" / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
     sibling = root / "src" / "aipass" / "sibling"
     sibling.mkdir(parents=True)
+    (sibling / ".trinity").mkdir()
     (sibling / "important.txt").write_text("don't delete", encoding="utf-8")
     return root
 

src/aipass/drone/tests/test_git_access.py

@@ -387,8 +387,11 @@ class TestCommitChanges:
             stderr="",
         )
 
-        test_dir = repo_dir / "src" / "aipass" / "drone" / "tests"
-        test_dir.mkdir(parents=True)
+        drone_dir = repo_dir / "src" / "aipass" / "drone"
+        drone_dir.mkdir(parents=True)
+        (drone_dir / ".trinity").mkdir()
+        test_dir = drone_dir / "tests"
+        test_dir.mkdir()
 
         with (
             patch("shutil.which", return_value="/usr/bin/ruff"),
@@ -417,8 +420,11 @@ class TestCommitChanges:
         mock_diff = MagicMock(returncode=1, stdout="", stderr="")
         mock_commit = MagicMock(returncode=0, stdout="[main abc999] green commit", stderr="")
 
-        test_dir = repo_dir / "src" / "aipass" / "drone" / "tests"
-        test_dir.mkdir(parents=True)
+        drone_dir = repo_dir / "src" / "aipass" / "drone"
+        drone_dir.mkdir(parents=True)
+        (drone_dir / ".trinity").mkdir()
+        test_dir = drone_dir / "tests"
+        test_dir.mkdir()
 
         with (
             patch("shutil.which", return_value="/usr/bin/ruff"),

src/aipass/drone/tests/test_rm.py

@@ -36,10 +36,11 @@ def project_dir(tmp_path):
 
 @pytest.fixture()
 def project_with_branches(project_dir):
-    """Project root with src/aipass/<branch> layout for sibling tests."""
+    """Project root with branch dirs containing .trinity/ markers."""
     for branch in ("drone", "api", "flow"):
         d = project_dir / "src" / "aipass" / branch
         d.mkdir(parents=True)
+        (d / ".trinity").mkdir()
         (d / "README.md").write_text(f"# {branch}")
     return project_dir