feat(bridge): DPLAN-0226 unified TG<->CC bridge — CC-native session discovery, live-proven round-trip (FPLAN-0290/0291/0292)

CHANGELOG.md

@@ -9,6 +9,38 @@ PyPI version — not the changelog header.
 
 ---
 
+## [2026-07-01]
+
+### Added
+
+- **Unified Telegram ↔ Claude Code bridge — CC-native session discovery
+  (DPLAN-0226)** — a Telegram message to a branch's bot now lands directly in
+  that branch's live Claude Code session, and the reply tails back out to
+  Telegram — a full round trip, **live-proven end-to-end from Patrick's own
+  Telegram client** (not just a self-test). The bot's inbound path
+  (`base_bot.ensure_tmux_session`) discovers the active session by enumerating
+  CC-native `~/.claude/sessions/<pid>.json` files (match `cwd`, confirm PID
+  alive, newest by `startedAt`), maps it to a tmux pane by cwd, and injects the
+  message — replacing the old `PRESENCE.central.json` pointer, which is kept but
+  commented out. The outbound path gains a CC-native "Strategy 0" in
+  `_resolve_active_transcript` that prefers the discovered transcript, so
+  assistant replies relay back reliably. Anthropic ToS rules out a cloud peer,
+  so all delivery is local (tmux/PTY). New `session_boot.py` boot wrapper
+  (attach-if-live-else-start-in-tmux; a thin `~/.bashrc claude()` shim delegates
+  to it). Hooks tests 66 green (presence_gate / cc_sessions / session_boot),
+  telegram presence_pointer 42 green. (DPLAN-0226 P1/P2, FPLAN-0290/0291/0292)
+
+### Changed
+
+- **Presence gate re-sourced to CC-native session files (presence_gate v2)** —
+  the single-session guard now sources truth from `~/.claude/sessions/<pid>.json`
+  via a new `cc_sessions` module (`find_occupant`/`find_live_for_cwd`) instead of
+  `PRESENCE.central.json`. Resume-aware (a `/resume` keeps the same PID, so the
+  session is correctly recognized as re-entry, not a duplicate) and exit-aware
+  (CC deletes the file on clean exit). `handle_stop` is now a plain no-op —
+  cleanup is CC's job. The old `presence.py` / `PRESENCE.central.json` are
+  preserved, just no longer sourced. (DPLAN-0226 P1)
+
 ## [2026-06-25]
 
 ### Added

src/aipass/hooks/.seedgo/bypass.json

@@ -998,6 +998,61 @@
       "file": "tests/test_presence_gate.py",
       "standard": "meta",
       "reason": "Test files do not need Version/Modified metadata headers."
+    },
+    {
+      "file": "apps/modules/cc_sessions.py",
+      "standard": "json_structure",
+      "reason": "Reads CC-native ~/.claude/sessions/*.json (external files, not branch data) with stdlib json — no JSON file ops needing json_handler."
+    },
+    {
+      "file": "apps/handlers/lifecycle/session_boot.py",
+      "standard": "json_structure",
+      "reason": "Boot wrapper has no JSON file ops — delegates to cc_sessions module for session reads. No json_handler needed."
+    },
+    {
+      "file": "apps/handlers/lifecycle/session_boot.py",
+      "standard": "dead_code",
+      "reason": "Boot wrapper invoked via __main__ entry point from shell shim (tools/install_boot_shim.sh) — not engine-dispatched. boot() and main() called at process entry, not statically imported."
+    },
+    {
+      "file": "apps/handlers/lifecycle/session_boot.py",
+      "standard": "unused_function",
+      "reason": "boot() and main() invoked via __main__ entry point from shell shim — not statically imported. _resolve_claude_binary(), _find_live_sessions(), _find_tmux(), etc. are internal helpers called by boot()."
+    },
+    {
+      "file": "tests/test_cc_sessions.py",
+      "standard": "architecture",
+      "reason": "Test files live in tests/, not in the 3-layer apps structure."
+    },
+    {
+      "file": "tests/test_cc_sessions.py",
+      "standard": "documentation",
+      "reason": "Test methods use descriptive names as documentation per pytest convention."
+    },
+    {
+      "file": "tests/test_cc_sessions.py",
+      "standard": "meta",
+      "reason": "Test files do not need Version/Modified metadata headers."
+    },
+    {
+      "file": "tests/test_session_boot.py",
+      "standard": "architecture",
+      "reason": "Test files live in tests/, not in the 3-layer apps structure."
+    },
+    {
+      "file": "tests/test_session_boot.py",
+      "standard": "documentation",
+      "reason": "Test methods use descriptive names as documentation per pytest convention."
+    },
+    {
+      "file": "tests/test_session_boot.py",
+      "standard": "meta",
+      "reason": "Test files do not need Version/Modified metadata headers."
+    },
+    {
+      "file": "tests/test_session_boot.py",
+      "standard": "encapsulation",
+      "reason": "Tests import handlers directly to test implementation details."
     }
   ],
   "notes": {

src/aipass/hooks/apps/handlers/lifecycle/session_boot.py

@@ -0,0 +1,190 @@
+# =================== AIPass ====================
+# Name: session_boot.py
+# Version: 1.1.0
+# Description: Boot wrapper — attach-if-live or start-in-tmux for Claude Code
+# Branch: hooks
+# Layer: apps/handlers/lifecycle
+# Created: 2026-06-30
+# Modified: 2026-06-30
+# =============================================
+
+"""Boot wrapper for Claude Code sessions.
+
+When Patrick boots Claude in a branch directory, this wrapper:
+  1. If already inside tmux ($TMUX set) → execs claude directly (no nesting).
+  2. Checks CC-native ~/.claude/sessions/ for a live session at this cwd.
+  3. If live AND hosted in a tmux session → attaches to that tmux session.
+  4. If live but NOT in tmux → warns (can't inject into a plain terminal).
+  5. If no live session → starts fresh inside a new tmux session.
+
+Tmux sessions are named after the branch directory (e.g., "hooks", "devpulse").
+All sessions use --permission-mode bypassPermissions (TG user can't answer prompts).
+Remote visibility is preserved (tmux composes with remoteControlAtStartup).
+
+Entry points:
+  drone @hooks boot [claude args...]
+"""
+
+import os
+import shutil
+import subprocess
+import sys
+from pathlib import Path
+
+from aipass.prax.apps.modules.logger import system_logger as logger
+
+_DEFAULT_ARGS = ["--permission-mode", "bypassPermissions"]
+
+
+def _resolve_claude_binary() -> str:
+    """Resolve the REAL claude binary path from PATH.
+
+    Uses shutil.which, which searches the filesystem PATH — it does NOT see
+    shell functions. So even when a claude() shell function shadows the binary,
+    this finds the real one.
+    """
+    path = shutil.which("claude")
+    if path:
+        return path
+    logger.error("[SESSION_BOOT] claude binary not found on PATH")
+    return "claude"
+
+
+def _find_live_sessions(cwd: str) -> list[dict]:
+    """Find live CC sessions for the given cwd via cc_sessions module."""
+    import importlib
+
+    cc_sessions = importlib.import_module("aipass.hooks.apps.modules.cc_sessions")
+    return cc_sessions.find_live_for_cwd(cwd)
+
+
+def _find_tmux() -> str | None:
+    """Find tmux binary on PATH."""
+    return shutil.which("tmux")
+
+
+def _tmux_session_exists(name: str) -> bool:
+    """Check if a tmux session with the given name exists."""
+    result = subprocess.run(
+        ["tmux", "has-session", "-t", name],
+        capture_output=True,
+        text=True,
+    )
+    return result.returncode == 0
+
+
+def _find_tmux_session_for_pid(pid: int) -> str | None:
+    """Find which tmux session hosts the given PID (as a descendant of a pane)."""
+    result = subprocess.run(
+        ["tmux", "list-panes", "-a", "-F", "#{pane_pid} #{session_name}"],
+        capture_output=True,
+        text=True,
+    )
+    if result.returncode != 0:
+        return None
+    for line in result.stdout.strip().splitlines():
+        parts = line.split(None, 1)
+        if len(parts) != 2:
+            continue
+        pane_pid_str, session_name = parts
+        try:
+            pane_pid = int(pane_pid_str)
+        except ValueError:
+            logger.info("[SESSION_BOOT] Non-integer pane PID: %s", pane_pid_str)
+            continue
+        if _is_descendant(pid, pane_pid):
+            return session_name
+    return None
+
+
+def _is_descendant(target_pid: int, ancestor_pid: int) -> bool:
+    """Check if target_pid is a descendant of ancestor_pid via /proc."""
+    pid = target_pid
+    for _ in range(20):
+        if pid == ancestor_pid:
+            return True
+        if pid <= 1:
+            return False
+        try:
+            for line in Path(f"/proc/{pid}/status").read_text().splitlines():
+                if line.startswith("PPid:"):
+                    pid = int(line.split()[1])
+                    break
+            else:
+                return False
+        except OSError as exc:
+            logger.info("[SESSION_BOOT] Cannot read /proc/%d/status: %s", pid, exc)
+            return False
+    return False
+
+
+def boot(cwd: str | None = None, extra_args: list[str] | None = None) -> dict:
+    """Boot Claude Code — attach if live, else start in tmux.
+
+    Args:
+        cwd: Branch directory (defaults to current working directory).
+        extra_args: Additional arguments to pass to claude (after default args).
+
+    Returns:
+        Result dict with action taken and details.
+    """
+    cwd = cwd or str(Path.cwd().resolve())
+    branch = Path(cwd).name
+    claude_bin = _resolve_claude_binary()
+
+    if os.environ.get("TMUX"):
+        logger.info("[SESSION_BOOT] Already inside tmux — running claude directly")
+        claude_cmd = [claude_bin] + _DEFAULT_ARGS
+        if extra_args:
+            claude_cmd.extend(extra_args)
+        os.execvp(claude_bin, claude_cmd)
+        return {"exit_code": 0, "action": "direct", "reason": "already in tmux"}
+
+    tmux = _find_tmux()
+    if not tmux:
+        logger.error("[SESSION_BOOT] tmux not found on PATH")
+        return {"exit_code": 1, "error": "tmux not found — required for session hosting"}
+
+    live = _find_live_sessions(cwd)
+
+    if live:
+        session = live[0]
+        pid = session["pid"]
+        name = session.get("name", "")
+        logger.info("[SESSION_BOOT] Live session found: PID %d%s", pid, f" ({name})" if name else "")
+
+        tmux_session = _find_tmux_session_for_pid(pid)
+        if tmux_session:
+            logger.info("[SESSION_BOOT] Attaching to tmux session '%s'", tmux_session)
+            os.execvp("tmux", ["tmux", "attach-session", "-t", tmux_session])
+            return {"exit_code": 0, "action": "attached", "tmux_session": tmux_session}
+
+        return {
+            "exit_code": 1,
+            "action": "warn",
+            "error": f"Live session at PID {pid} is not in tmux. Kill it first or attach to its terminal.",
+        }
+
+    if _tmux_session_exists(branch):
+        logger.info("[SESSION_BOOT] Killing stale tmux session '%s'", branch)
+        subprocess.run(["tmux", "kill-session", "-t", branch], check=False)
+
+    claude_cmd = [claude_bin] + _DEFAULT_ARGS
+    if extra_args:
+        claude_cmd.extend(extra_args)
+
+    logger.info("[SESSION_BOOT] Starting fresh in tmux session '%s': %s", branch, " ".join(claude_cmd))
+    os.execvp("tmux", ["tmux", "new-session", "-s", branch, "--"] + claude_cmd)
+    return {"exit_code": 0, "action": "started", "tmux_session": branch}
+
+
+def main() -> None:
+    """CLI entry point for the boot wrapper."""
+    result = boot(extra_args=sys.argv[1:] if len(sys.argv) > 1 else None)
+    if result.get("exit_code", 0) != 0:
+        sys.stderr.write(f"{result.get('error', 'boot failed')}\n")
+        sys.exit(result["exit_code"])
+
+
+if __name__ == "__main__":
+    main()

src/aipass/hooks/apps/handlers/security/presence_gate.py

@@ -1,21 +1,29 @@
 # =================== AIPass ====================
 # Name: presence_gate.py
-# Version: 1.0.0
+# Version: 2.0.0
 # Description: Single-session gate — blocks duplicate Claude runtimes per branch
 # Branch: hooks
 # Layer: apps/handlers/security
 # Created: 2026-06-29
-# Modified: 2026-06-29
+# Modified: 2026-06-30
 # =============================================
 
 """Single-session gate — blocks duplicate Claude runtimes per branch.
 
-Fires on UserPromptSubmit: calls presence.claim(). If OCCUPIED by a live PID,
-blocks the prompt. If free or stale, acquires and proceeds.
+Sources truth from CC-native ~/.claude/sessions/<pid>.json (resume-aware,
+exit-aware) instead of PRESENCE.central.json. Resume-aware because /resume
+keeps the same PID; exit-aware because CC deletes the file on clean exit.
 
-Fires on Stop: calls presence.release() to clean up.
+Fires on UserPromptSubmit: checks CC sessions for another live brain in the
+same branch. If occupied by a different PID, blocks. If free, allows.
+
+handle_stop is a no-op (Stop fires every turn, not just session end; CC-native
+session files handle cleanup on exit).
 
 Skips sub-agents and dispatched/daemon session types.
+
+PRESENCE.central.json + presence.py are preserved (not deleted) but the guard
+no longer sources truth from them.
 """
 
 import importlib
@@ -45,11 +53,9 @@ def _resolve_branch(hook_data: dict) -> str:
 def handle(hook_data: dict) -> dict:
     """UserPromptSubmit gate — enforce one live session per branch.
 
-    Args:
-        hook_data: Parsed hook event dict from engine.
-
-    Returns:
-        Result dict with stdout (block JSON or empty) and exit_code.
+    Sources truth from CC-native ~/.claude/sessions/<pid>.json.
+    Resume-aware: /resume keeps the same PID, so exclude_pid correctly
+    identifies re-entry. Exit-aware: CC deletes the file on clean exit.
     """
     try:
         agent_type = hook_data.get("agent_type", "")
@@ -61,21 +67,20 @@ def handle(hook_data: dict) -> dict:
             return _ALLOW
 
         branch = _resolve_branch(hook_data)
-        session_id = os.environ.get("CLAUDE_CODE_SESSION_ID", "")
+        branch_cwd = hook_data.get("cwd", "") or str(Path.cwd())
 
         presence = importlib.import_module("aipass.hooks.apps.modules.presence")
-        result = presence.claim(
-            branch=branch,
-            session_id=session_id,
-            session_type=session_type,
-        )
+        our_pid = presence._resolve_session_pid()
 
-        if result["status"] == "ACQUIRED":
+        cc_sessions = importlib.import_module("aipass.hooks.apps.modules.cc_sessions")
+        occupant = cc_sessions.find_occupant(branch_cwd, exclude_pid=our_pid)
+
+        if occupant is None:
             return _ALLOW
 
-        pid = result.get("pid", "?")
-        holder_type = result.get("session_type", "unknown")
-        reason = f"{branch} already live at PID {pid} (session_type: {holder_type}) — attach, do not spawn."
+        occ_pid = occupant.get("pid", "?")
+        occ_name = occupant.get("name", "")
+        reason = f"{branch} already live at PID {occ_pid}{f' ({occ_name})' if occ_name else ''} — attach, do not spawn."
         logger.warning("[presence_gate] BLOCKED: %s", reason)
         return {
             "exit_code": 2,
@@ -88,10 +93,5 @@ def handle(hook_data: dict) -> dict:
 
 
 def handle_stop(hook_data: dict) -> dict:
-    """No-op on Stop. Presence is NOT released per-turn.
-
-    Stop fires at the end of every assistant turn, not just session end.
-    Releasing each turn would create a gap where a 2nd session wouldn't be blocked.
-    Stale detection (dead claude PID) handles cleanup when the session truly exits.
-    """
+    """No-op on Stop. CC-native session files handle cleanup on exit."""
     return _ALLOW

src/aipass/hooks/apps/modules/cc_sessions.py

@@ -0,0 +1,148 @@
+# =================== AIPass ====================
+# Name: cc_sessions.py
+# Version: 1.0.0
+# Description: Read CC-native session files for active-session discovery
+# Branch: hooks
+# Layer: apps/modules
+# Created: 2026-06-30
+# Modified: 2026-06-30
+# =============================================
+
+"""Read Claude Code native session files (~/.claude/sessions/<pid>.json).
+
+CC maintains one JSON file per running session at ~/.claude/sessions/<pid>.json.
+These are resume-aware (sessionId updated on /resume), exit-aware (deleted on
+clean exit, stale-swept by CC itself), and the authoritative source of which
+sessions are live for a given working directory.
+
+Used by presence_gate to source truth instead of PRESENCE.central.json.
+"""
+
+import json
+import os
+from pathlib import Path
+
+from aipass.cli.apps.modules import err_console
+from aipass.prax.apps.modules.logger import system_logger as logger
+
+CONSOLE = err_console
+
+CC_SESSIONS_DIR = Path.home() / ".claude" / "sessions"
+
+
+def _is_pid_alive(pid: int) -> bool:
+    """Check if a process with the given PID exists."""
+    if pid <= 1:
+        return False
+    try:
+        os.kill(pid, 0)
+        return True
+    except ProcessLookupError:
+        logger.info("[CC_SESSIONS] PID %d not found (dead)", pid)
+        return False
+    except PermissionError:
+        logger.info("[CC_SESSIONS] PID %d exists but permission denied — treating as alive", pid)
+        return True
+    except OSError as exc:
+        logger.info("[CC_SESSIONS] PID %d os.kill failed: %s — treating as dead", pid, exc)
+        return False
+
+
+def read_all_sessions() -> list[dict]:
+    """Read all CC session PID files. Returns list of session dicts."""
+    if not CC_SESSIONS_DIR.is_dir():
+        return []
+    sessions = []
+    for f in CC_SESSIONS_DIR.iterdir():
+        if not f.name.endswith(".json") or not f.stem.isdigit():
+            continue
+        try:
+            data = json.loads(f.read_text(encoding="utf-8"))
+            sessions.append(data)
+        except (json.JSONDecodeError, OSError) as exc:
+            logger.info("[CC_SESSIONS] Failed to read %s: %s", f, exc)
+    return sessions
+
+
+def find_live_for_cwd(cwd: str) -> list[dict]:
+    """Find all live CC sessions whose cwd matches the given directory.
+
+    Compares resolved paths for robustness (symlinks, trailing slashes).
+    Only returns sessions whose PID is still alive.
+    """
+    target = str(Path(cwd).resolve())
+    live = []
+    for session in read_all_sessions():
+        session_cwd = session.get("cwd", "")
+        if not session_cwd:
+            continue
+        if str(Path(session_cwd).resolve()) != target:
+            continue
+        pid = session.get("pid")
+        if pid and _is_pid_alive(pid):
+            live.append(session)
+        else:
+            logger.info(
+                "[CC_SESSIONS] Stale session file for PID %s at %s",
+                pid,
+                session_cwd,
+            )
+    return live
+
+
+def find_occupant(cwd: str, exclude_pid: int | None = None) -> dict | None:
+    """Find a live CC session occupying the given cwd, excluding our own PID.
+
+    Returns the first occupant session dict, or None if the branch is free.
+    This is resume-aware: /resume keeps the same PID, so exclude_pid correctly
+    identifies re-entry.
+    """
+    for session in find_live_for_cwd(cwd):
+        if exclude_pid is not None and session.get("pid") == exclude_pid:
+            continue
+        return session
+    return None
+
+
+# =============================================================================
+# MODULE INTERFACE (drone @hooks routing)
+# =============================================================================
+
+
+def print_introspection():
+    """Print CC session state for drone routing."""
+    CONSOLE.print("[bold cyan]cc_sessions[/bold cyan] Module")
+    CONSOLE.print(f"  Sessions dir: {CC_SESSIONS_DIR}")
+    try:
+        sessions = read_all_sessions()
+        if not sessions:
+            CONSOLE.print("  No CC session files found")
+        else:
+            for s in sessions:
+                pid = s.get("pid", "?")
+                cwd = s.get("cwd", "?")
+                kind = s.get("kind", "?")
+                name = s.get("name", "")
+                alive = _is_pid_alive(pid) if isinstance(pid, int) else False
+                status = "[green]live[/green]" if alive else "[dim]stale[/dim]"
+                label = f" ({name})" if name else ""
+                CONSOLE.print(f"  PID {pid}{label}: {Path(cwd).name}  kind={kind}  {status}")
+    except Exception as exc:
+        logger.info("[CC_SESSIONS] introspection error: %s", exc)
+        CONSOLE.print(f"  Error reading sessions: {exc}")
+
+
+def handle_command(command: str, args: list) -> bool:
+    """Route cc_sessions commands from drone @hooks."""
+    if command in ("--help", "-h", "help"):
+        CONSOLE.print("[bold cyan]cc_sessions[/bold cyan] — CC-native session file reader")
+        CONSOLE.print()
+        CONSOLE.print("  drone @hooks cc_sessions    Show live CC sessions")
+        return True
+
+    if command == "cc_sessions":
+        if not args:
+            print_introspection()
+            return True
+
+    return False

src/aipass/hooks/tests/test_cc_sessions.py

@@ -0,0 +1,152 @@
+"""Tests for CC-native session file reader."""
+
+import json
+import os
+from unittest.mock import patch
+
+from aipass.hooks.apps.modules import cc_sessions
+
+
+class TestIsPidAlive:
+    def test_alive(self):
+        assert cc_sessions._is_pid_alive(os.getpid()) is True
+
+    def test_dead(self):
+        assert cc_sessions._is_pid_alive(999999999) is False
+
+    def test_pid_zero(self):
+        assert cc_sessions._is_pid_alive(0) is False
+
+    def test_pid_one(self):
+        assert cc_sessions._is_pid_alive(1) is False
+
+    def test_permission_error_treated_as_alive(self):
+        with patch("os.kill", side_effect=PermissionError("denied")):
+            assert cc_sessions._is_pid_alive(42) is True
+
+    def test_oserror_treated_as_dead(self):
+        with patch("os.kill", side_effect=OSError("unknown")):
+            assert cc_sessions._is_pid_alive(42) is False
+
+
+class TestReadAllSessions:
+    def test_reads_pid_files(self, tmp_path):
+        session = {"pid": 1234, "sessionId": "abc", "cwd": "/tmp/branch", "kind": "interactive"}
+        (tmp_path / "1234.json").write_text(json.dumps(session))
+        with patch.object(cc_sessions, "CC_SESSIONS_DIR", tmp_path):
+            result = cc_sessions.read_all_sessions()
+        assert len(result) == 1
+        assert result[0]["pid"] == 1234
+
+    def test_skips_non_pid_files(self, tmp_path):
+        (tmp_path / "config.json").write_text("{}")
+        (tmp_path / "abc.json").write_text("{}")
+        with patch.object(cc_sessions, "CC_SESSIONS_DIR", tmp_path):
+            result = cc_sessions.read_all_sessions()
+        assert result == []
+
+    def test_skips_corrupt_json(self, tmp_path):
+        (tmp_path / "999.json").write_text("not json{{{")
+        with patch.object(cc_sessions, "CC_SESSIONS_DIR", tmp_path):
+            result = cc_sessions.read_all_sessions()
+        assert result == []
+
+    def test_empty_dir(self, tmp_path):
+        with patch.object(cc_sessions, "CC_SESSIONS_DIR", tmp_path):
+            result = cc_sessions.read_all_sessions()
+        assert result == []
+
+    def test_missing_dir(self, tmp_path):
+        with patch.object(cc_sessions, "CC_SESSIONS_DIR", tmp_path / "nonexistent"):
+            result = cc_sessions.read_all_sessions()
+        assert result == []
+
+    def test_multiple_sessions(self, tmp_path):
+        for pid in (100, 200, 300):
+            s = {"pid": pid, "sessionId": f"s-{pid}", "cwd": "/tmp", "kind": "interactive"}
+            (tmp_path / f"{pid}.json").write_text(json.dumps(s))
+        with patch.object(cc_sessions, "CC_SESSIONS_DIR", tmp_path):
+            result = cc_sessions.read_all_sessions()
+        assert len(result) == 3
+
+
+class TestFindLiveForCwd:
+    def test_filters_by_cwd(self, tmp_path):
+        s1 = {"pid": os.getpid(), "sessionId": "a", "cwd": "/tmp/hooks", "kind": "interactive"}
+        s2 = {"pid": os.getpid(), "sessionId": "b", "cwd": "/tmp/devpulse", "kind": "interactive"}
+        (tmp_path / f"{os.getpid()}.json").write_text(json.dumps(s1))
+        (tmp_path / "99999.json").write_text(json.dumps(s2))
+        with patch.object(cc_sessions, "CC_SESSIONS_DIR", tmp_path):
+            result = cc_sessions.find_live_for_cwd("/tmp/hooks")
+        assert len(result) == 1
+        assert result[0]["sessionId"] == "a"
+
+    def test_excludes_dead_pids(self, tmp_path):
+        s = {"pid": 999999999, "sessionId": "dead", "cwd": "/tmp/hooks", "kind": "interactive"}
+        (tmp_path / "999999999.json").write_text(json.dumps(s))
+        with patch.object(cc_sessions, "CC_SESSIONS_DIR", tmp_path):
+            result = cc_sessions.find_live_for_cwd("/tmp/hooks")
+        assert result == []
+
+    def test_resolves_paths(self, tmp_path):
+        target = str(tmp_path / "hooks")
+        s = {"pid": os.getpid(), "sessionId": "a", "cwd": target, "kind": "interactive"}
+        (tmp_path / f"{os.getpid()}.json").write_text(json.dumps(s))
+        with patch.object(cc_sessions, "CC_SESSIONS_DIR", tmp_path):
+            result = cc_sessions.find_live_for_cwd(target + "/")
+        assert len(result) == 1
+
+    def test_empty_cwd_skipped(self, tmp_path):
+        s = {"pid": os.getpid(), "sessionId": "a", "cwd": "", "kind": "interactive"}
+        (tmp_path / f"{os.getpid()}.json").write_text(json.dumps(s))
+        with patch.object(cc_sessions, "CC_SESSIONS_DIR", tmp_path):
+            result = cc_sessions.find_live_for_cwd("/tmp/hooks")
+        assert result == []
+
+
+class TestFindOccupant:
+    def test_no_occupant_when_free(self, tmp_path):
+        with patch.object(cc_sessions, "CC_SESSIONS_DIR", tmp_path):
+            result = cc_sessions.find_occupant("/tmp/hooks")
+        assert result is None
+
+    def test_excludes_own_pid(self, tmp_path):
+        my_pid = os.getpid()
+        s = {"pid": my_pid, "sessionId": "mine", "cwd": "/tmp/hooks", "kind": "interactive"}
+        (tmp_path / f"{my_pid}.json").write_text(json.dumps(s))
+        with patch.object(cc_sessions, "CC_SESSIONS_DIR", tmp_path):
+            result = cc_sessions.find_occupant("/tmp/hooks", exclude_pid=my_pid)
+        assert result is None
+
+    def test_finds_other_occupant(self, tmp_path):
+        my_pid = os.getpid()
+        s = {"pid": my_pid, "sessionId": "other", "cwd": "/tmp/hooks", "kind": "interactive"}
+        (tmp_path / f"{my_pid}.json").write_text(json.dumps(s))
+        with patch.object(cc_sessions, "CC_SESSIONS_DIR", tmp_path):
+            result = cc_sessions.find_occupant("/tmp/hooks", exclude_pid=99999)
+        assert result is not None
+        assert result["sessionId"] == "other"
+
+    def test_no_exclude_returns_any_live(self, tmp_path):
+        my_pid = os.getpid()
+        s = {"pid": my_pid, "sessionId": "any", "cwd": "/tmp/hooks", "kind": "interactive"}
+        (tmp_path / f"{my_pid}.json").write_text(json.dumps(s))
+        with patch.object(cc_sessions, "CC_SESSIONS_DIR", tmp_path):
+            result = cc_sessions.find_occupant("/tmp/hooks")
+        assert result is not None
+
+
+class TestIntrospection:
+    def test_print_introspection_no_sessions(self, tmp_path):
+        with patch.object(cc_sessions, "CC_SESSIONS_DIR", tmp_path):
+            cc_sessions.print_introspection()
+
+    def test_handle_command_cc_sessions(self, tmp_path):
+        with patch.object(cc_sessions, "CC_SESSIONS_DIR", tmp_path):
+            assert cc_sessions.handle_command("cc_sessions", []) is True
+
+    def test_handle_command_help(self):
+        assert cc_sessions.handle_command("--help", []) is True
+
+    def test_handle_command_unknown(self):
+        assert cc_sessions.handle_command("unknown", []) is False

src/aipass/hooks/tests/test_presence_gate.py

@@ -7,24 +7,31 @@ from unittest.mock import MagicMock, patch
 from aipass.hooks.apps.handlers.security import presence_gate
 
 
-def _make_presence_mock(claim_result, release_result=True):
-    """Build a mock presence module with given claim/release return values."""
-    mock = MagicMock()
-    mock.claim.return_value = claim_result
-    mock.release.return_value = release_result
-    return mock
+def _make_mocks(our_pid: int | None = 1000, occupant=None):
+    """Build presence + cc_sessions mocks for the gate."""
+    presence_mock = MagicMock()
+    presence_mock._resolve_session_pid.return_value = our_pid
+
+    cc_mock = MagicMock()
+    cc_mock.find_occupant.return_value = occupant
+
+    def import_router(name):
+        if "presence" in name and "cc_sessions" not in name:
+            return presence_mock
+        if "cc_sessions" in name:
+            return cc_mock
+        raise ImportError(name)
+
+    return presence_mock, cc_mock, import_router
 
 
-_ACQUIRED_MOCK = _make_presence_mock({"status": "ACQUIRED"})
-_OCCUPIED_MOCK = _make_presence_mock(
-    {
-        "status": "OCCUPIED",
-        "pid": 5000,
-        "session_id": "existing",
-        "work_dir": "/tmp/branch",
-        "session_type": "interactive",
-    }
-)
+_OCCUPANT = {
+    "pid": 5000,
+    "sessionId": "existing-session",
+    "cwd": "/tmp/branch",
+    "kind": "interactive",
+    "name": "hooks-ab",
+}
 
 
 class TestResolveBranch:
@@ -51,29 +58,40 @@ class TestResolveBranch:
 
 
 class TestHandle:
-    def test_first_prompt_acquired(self):
+    def test_no_occupant_allows(self):
+        _, _, router = _make_mocks(our_pid=1000, occupant=None)
         with patch.dict(os.environ, {"AIPASS_SESSION_TYPE": "interactive"}, clear=True):
-            with patch("importlib.import_module", return_value=_ACQUIRED_MOCK):
+            with patch("importlib.import_module", side_effect=router):
                 result = presence_gate.handle({})
         assert result["exit_code"] == 0
 
-    def test_occupied_blocks(self):
+    def test_occupant_blocks(self):
+        _, _, router = _make_mocks(our_pid=1000, occupant=_OCCUPANT)
         with patch.dict(os.environ, {"AIPASS_SESSION_TYPE": "interactive"}, clear=True):
-            with patch("importlib.import_module", return_value=_OCCUPIED_MOCK):
+            with patch("importlib.import_module", side_effect=router):
                 result = presence_gate.handle({})
         assert result["exit_code"] == 2
         parsed = json.loads(result["stdout"])
         assert parsed["decision"] == "block"
         assert "5000" in parsed["reason"]
 
+    def test_block_includes_session_name(self):
+        _, _, router = _make_mocks(our_pid=1000, occupant=_OCCUPANT)
+        with patch.dict(os.environ, {"AIPASS_SESSION_TYPE": "interactive"}, clear=True):
+            with patch("importlib.import_module", side_effect=router):
+                result = presence_gate.handle({})
+        parsed = json.loads(result["stdout"])
+        assert "hooks-ab" in parsed["reason"]
+
     def test_subagent_skipped(self):
         result = presence_gate.handle({"agent_type": "sub"})
         assert result["exit_code"] == 0
         assert result["stdout"] == ""
 
     def test_main_agent_not_skipped(self):
+        _, _, router = _make_mocks(our_pid=1000, occupant=None)
         with patch.dict(os.environ, {"AIPASS_SESSION_TYPE": "interactive"}, clear=True):
-            with patch("importlib.import_module", return_value=_ACQUIRED_MOCK):
+            with patch("importlib.import_module", side_effect=router):
                 result = presence_gate.handle({"agent_type": "main"})
         assert result["exit_code"] == 0
 
@@ -89,9 +107,18 @@ class TestHandle:
         assert result["exit_code"] == 0
         assert result["stdout"] == ""
 
-    def test_resume_dead_holder_acquires(self):
+    def test_exclude_pid_passed_to_find_occupant(self):
+        _, cc_mock, router = _make_mocks(our_pid=1000, occupant=None)
         with patch.dict(os.environ, {"AIPASS_SESSION_TYPE": "interactive"}, clear=True):
-            with patch("importlib.import_module", return_value=_ACQUIRED_MOCK):
+            with patch("importlib.import_module", side_effect=router):
+                presence_gate.handle({})
+        cc_mock.find_occupant.assert_called_once()
+        assert cc_mock.find_occupant.call_args[1]["exclude_pid"] == 1000
+
+    def test_no_session_pid_allows(self):
+        _, _, router = _make_mocks(our_pid=None, occupant=None)
+        with patch.dict(os.environ, {"AIPASS_SESSION_TYPE": "interactive"}, clear=True):
+            with patch("importlib.import_module", side_effect=router):
                 result = presence_gate.handle({})
         assert result["exit_code"] == 0
 
@@ -99,23 +126,19 @@ class TestHandle:
         branch_dir = tmp_path / "devpulse"
         branch_dir.mkdir()
         (branch_dir / ".trinity").mkdir()
+        _, _, router = _make_mocks(our_pid=1000, occupant=_OCCUPANT)
         with patch.dict(os.environ, {"AIPASS_SESSION_TYPE": "interactive"}, clear=True):
-            with patch("importlib.import_module", return_value=_OCCUPIED_MOCK):
+            with patch("importlib.import_module", side_effect=router):
                 result = presence_gate.handle({"cwd": str(branch_dir)})
         parsed = json.loads(result["stdout"])
         assert "devpulse" in parsed["reason"]
         assert "attach" in parsed["reason"].lower()
 
-    def test_branch_resolved_from_hook_data_cwd(self, tmp_path):
-        branch_dir = tmp_path / "api"
-        branch_dir.mkdir()
-        (branch_dir / ".trinity").mkdir()
-        mock = _make_presence_mock({"status": "ACQUIRED"})
+    def test_gate_error_allows(self):
         with patch.dict(os.environ, {"AIPASS_SESSION_TYPE": "interactive"}, clear=True):
-            with patch("importlib.import_module", return_value=mock):
-                presence_gate.handle({"cwd": str(branch_dir)})
-        mock.claim.assert_called_once()
-        assert mock.claim.call_args[1]["branch"] == "api"
+            with patch("importlib.import_module", side_effect=ImportError("boom")):
+                result = presence_gate.handle({})
+        assert result["exit_code"] == 0
 
 
 class TestHandleStop:
@@ -124,9 +147,10 @@ class TestHandleStop:
         assert result["exit_code"] == 0
         assert result["stdout"] == ""
 
-    def test_stop_does_not_call_presence(self):
-        mock = _make_presence_mock({"status": "ACQUIRED"})
+    def test_stop_does_not_call_anything(self):
+        mock = MagicMock()
         with patch("importlib.import_module", return_value=mock):
             presence_gate.handle_stop({})
         mock.release.assert_not_called()
         mock.claim.assert_not_called()
+        mock.find_occupant.assert_not_called()

src/aipass/hooks/tests/test_session_boot.py

@@ -0,0 +1,212 @@
+"""Tests for session boot wrapper (attach-if-live / start-in-tmux)."""
+
+from unittest.mock import MagicMock, patch
+
+from aipass.hooks.apps.handlers.lifecycle import session_boot
+
+_MOD = "aipass.hooks.apps.handlers.lifecycle.session_boot"
+
+
+class TestResolveClaudeBinary:
+    def test_found_on_path(self):
+        with patch(f"{_MOD}.shutil.which", return_value="/usr/local/bin/claude"):
+            assert session_boot._resolve_claude_binary() == "/usr/local/bin/claude"
+
+    def test_not_found_fallback(self):
+        with patch(f"{_MOD}.shutil.which", return_value=None):
+            assert session_boot._resolve_claude_binary() == "claude"
+
+
+class TestFindTmux:
+    def test_found(self):
+        with patch("shutil.which", return_value="/usr/bin/tmux"):
+            assert session_boot._find_tmux() == "/usr/bin/tmux"
+
+    def test_not_found(self):
+        with patch("shutil.which", return_value=None):
+            assert session_boot._find_tmux() is None
+
+
+class TestTmuxSessionExists:
+    def test_exists(self):
+        with patch(f"{_MOD}.subprocess.run", return_value=MagicMock(returncode=0)):
+            assert session_boot._tmux_session_exists("hooks") is True
+
+    def test_not_exists(self):
+        with patch(f"{_MOD}.subprocess.run", return_value=MagicMock(returncode=1)):
+            assert session_boot._tmux_session_exists("hooks") is False
+
+
+class TestFindTmuxSessionForPid:
+    def test_finds_session(self):
+        output = "1234 hooks\n5678 devpulse\n"
+        with (
+            patch(f"{_MOD}.subprocess.run", return_value=MagicMock(returncode=0, stdout=output)),
+            patch.object(session_boot, "_is_descendant", side_effect=lambda t, a: t == 9999 and a == 1234),
+        ):
+            assert session_boot._find_tmux_session_for_pid(9999) == "hooks"
+
+    def test_not_found(self):
+        output = "1234 hooks\n"
+        with (
+            patch(f"{_MOD}.subprocess.run", return_value=MagicMock(returncode=0, stdout=output)),
+            patch.object(session_boot, "_is_descendant", return_value=False),
+        ):
+            assert session_boot._find_tmux_session_for_pid(9999) is None
+
+    def test_tmux_not_running(self):
+        with patch(f"{_MOD}.subprocess.run", return_value=MagicMock(returncode=1)):
+            assert session_boot._find_tmux_session_for_pid(9999) is None
+
+
+class TestIsDescendant:
+    def test_direct_match(self):
+        assert session_boot._is_descendant(100, 100) is True
+
+    def test_pid_one_not_descendant(self):
+        assert session_boot._is_descendant(1, 999) is False
+
+    def test_proc_walk(self):
+        statuses = {200: "Name:\tpython3\nPPid:\t100\n"}
+        with patch("pathlib.Path.read_text", side_effect=lambda: statuses.get(200, "")):
+            assert session_boot._is_descendant(200, 100) is True
+
+    def test_proc_not_found(self):
+        with patch("pathlib.Path.read_text", side_effect=OSError("no such file")):
+            assert session_boot._is_descendant(200, 100) is False
+
+
+class TestBoot:
+    def test_already_in_tmux_execs_directly(self, tmp_path):
+        with (
+            patch.dict("os.environ", {"TMUX": "/tmp/tmux-1000/default,123,0"}),
+            patch.object(session_boot, "_resolve_claude_binary", return_value="/usr/local/bin/claude"),
+            patch(f"{_MOD}.os.execvp") as mock_exec,
+        ):
+            session_boot.boot(cwd=str(tmp_path))
+        mock_exec.assert_called_once()
+        args = mock_exec.call_args[0]
+        assert args[0] == "/usr/local/bin/claude"
+        assert "--permission-mode" in args[1]
+        assert "bypassPermissions" in args[1]
+
+    def test_already_in_tmux_passes_extra_args(self, tmp_path):
+        with (
+            patch.dict("os.environ", {"TMUX": "/tmp/tmux-1000/default,123,0"}),
+            patch.object(session_boot, "_resolve_claude_binary", return_value="/usr/local/bin/claude"),
+            patch(f"{_MOD}.os.execvp") as mock_exec,
+        ):
+            session_boot.boot(cwd=str(tmp_path), extra_args=["--resume"])
+        args = mock_exec.call_args[0][1]
+        assert "--resume" in args
+
+    def test_no_tmux_errors(self, tmp_path):
+        with (
+            patch.dict("os.environ", {}, clear=True),
+            patch.object(session_boot, "_resolve_claude_binary", return_value="/usr/local/bin/claude"),
+            patch.object(session_boot, "_find_tmux", return_value=None),
+        ):
+            result = session_boot.boot(cwd=str(tmp_path))
+        assert result["exit_code"] == 1
+        assert "tmux not found" in result["error"]
+
+    def test_live_session_in_tmux_attaches(self, tmp_path):
+        live = [{"pid": 1234, "sessionId": "abc", "cwd": str(tmp_path), "name": "hooks-ab"}]
+        with (
+            patch.dict("os.environ", {}, clear=True),
+            patch.object(session_boot, "_resolve_claude_binary", return_value="/usr/local/bin/claude"),
+            patch.object(session_boot, "_find_tmux", return_value="/usr/bin/tmux"),
+            patch.object(session_boot, "_find_live_sessions", return_value=live),
+            patch.object(session_boot, "_find_tmux_session_for_pid", return_value="hooks"),
+            patch(f"{_MOD}.os.execvp") as mock_exec,
+        ):
+            session_boot.boot(cwd=str(tmp_path))
+        mock_exec.assert_called_once_with("tmux", ["tmux", "attach-session", "-t", "hooks"])
+
+    def test_live_session_not_in_tmux_warns(self, tmp_path):
+        live = [{"pid": 1234, "sessionId": "abc", "cwd": str(tmp_path)}]
+        with (
+            patch.dict("os.environ", {}, clear=True),
+            patch.object(session_boot, "_resolve_claude_binary", return_value="/usr/local/bin/claude"),
+            patch.object(session_boot, "_find_tmux", return_value="/usr/bin/tmux"),
+            patch.object(session_boot, "_find_live_sessions", return_value=live),
+            patch.object(session_boot, "_find_tmux_session_for_pid", return_value=None),
+        ):
+            result = session_boot.boot(cwd=str(tmp_path))
+        assert result["exit_code"] == 1
+        assert result["action"] == "warn"
+
+    def test_no_live_session_starts_fresh(self, tmp_path):
+        with (
+            patch.dict("os.environ", {}, clear=True),
+            patch.object(session_boot, "_resolve_claude_binary", return_value="/usr/local/bin/claude"),
+            patch.object(session_boot, "_find_tmux", return_value="/usr/bin/tmux"),
+            patch.object(session_boot, "_find_live_sessions", return_value=[]),
+            patch.object(session_boot, "_tmux_session_exists", return_value=False),
+            patch(f"{_MOD}.os.execvp") as mock_exec,
+        ):
+            session_boot.boot(cwd=str(tmp_path))
+        args = mock_exec.call_args[0]
+        assert args[0] == "tmux"
+        assert "new-session" in args[1]
+        assert "/usr/local/bin/claude" in args[1]
+        assert "--permission-mode" in args[1]
+        assert "bypassPermissions" in args[1]
+
+    def test_stale_tmux_session_killed(self, tmp_path):
+        with (
+            patch.dict("os.environ", {}, clear=True),
+            patch.object(session_boot, "_resolve_claude_binary", return_value="/usr/local/bin/claude"),
+            patch.object(session_boot, "_find_tmux", return_value="/usr/bin/tmux"),
+            patch.object(session_boot, "_find_live_sessions", return_value=[]),
+            patch.object(session_boot, "_tmux_session_exists", return_value=True),
+            patch(f"{_MOD}.subprocess.run") as mock_run,
+            patch(f"{_MOD}.os.execvp"),
+        ):
+            session_boot.boot(cwd=str(tmp_path))
+        kill_calls = [c for c in mock_run.call_args_list if "kill-session" in str(c)]
+        assert len(kill_calls) == 1
+
+    def test_extra_args_passed(self, tmp_path):
+        with (
+            patch.dict("os.environ", {}, clear=True),
+            patch.object(session_boot, "_resolve_claude_binary", return_value="/usr/local/bin/claude"),
+            patch.object(session_boot, "_find_tmux", return_value="/usr/bin/tmux"),
+            patch.object(session_boot, "_find_live_sessions", return_value=[]),
+            patch.object(session_boot, "_tmux_session_exists", return_value=False),
+            patch(f"{_MOD}.os.execvp") as mock_exec,
+        ):
+            session_boot.boot(cwd=str(tmp_path), extra_args=["--resume"])
+        args = mock_exec.call_args[0][1]
+        assert "--resume" in args
+
+
+class TestMain:
+    def test_success(self):
+        with patch.object(session_boot, "boot", return_value={"exit_code": 0, "action": "started"}):
+            session_boot.main()
+
+    def test_failure_exits(self):
+        import pytest
+
+        with (
+            patch.object(session_boot, "boot", return_value={"exit_code": 1, "error": "tmux not found"}),
+            pytest.raises(SystemExit, match="1"),
+        ):
+            session_boot.main()
+
+    def test_passes_sys_argv(self):
+        with (
+            patch.object(session_boot, "boot", return_value={"exit_code": 0}) as mock_boot,
+            patch(f"{_MOD}.sys.argv", ["session_boot", "--resume", "--verbose"]),
+        ):
+            session_boot.main()
+        mock_boot.assert_called_once_with(extra_args=["--resume", "--verbose"])
+
+    def test_no_extra_args_when_no_argv(self):
+        with (
+            patch.object(session_boot, "boot", return_value={"exit_code": 0}) as mock_boot,
+            patch(f"{_MOD}.sys.argv", ["session_boot"]),
+        ):
+            session_boot.main()
+        mock_boot.assert_called_once_with(extra_args=None)

src/aipass/skills/.seedgo/bypass.json

@@ -86,6 +86,16 @@
       "standard": "documentation",
       "reason": "Test file — pytest test methods are self-documenting via class/method names."
     },
+    {
+      "file": "lib/telegram/tests/test_presence_pointer.py",
+      "standard": "architecture",
+      "reason": "Test file — lives in tests/ by convention. Test files are exempt from layer architecture standard."
+    },
+    {
+      "file": "lib/telegram/tests/test_presence_pointer.py",
+      "standard": "encapsulation",
+      "reason": "Test file — imports handler directly for unit testing. Same pattern as test_multi_bot.py and test_attach_only.py."
+    },
     {
       "file": "lib/telegram/tests/test_heartbeat_delivered.py",
       "standard": "architecture",
@@ -186,8 +196,8 @@
     {
       "file": "lib/telegram/apps/handlers/base_bot.py",
       "standard": "unused_function",
-      "functions": ["chunk_text", "on_response"],
-      "reason": "DPLAN-0220 incomplete port — ported-but-unwired from the 9k-line Dev-Pass telegram port. chunk_text (long-message splitting) and on_response (response hook) await wiring; on_response is a pending Wave-2 design call. See README → Ported-but-unwired."
+      "functions": ["chunk_text", "on_response", "_read_transcript_tail", "_extract_assistant_text"],
+      "reason": "DPLAN-0220/DPLAN-0226 — chunk_text + on_response are ported-but-unwired (Dev-Pass). _read_transcript_tail + _extract_assistant_text are DPLAN-0226 OUT relay infrastructure: built and tested, wiring pending end-to-end integration."
     },
     {
       "file": "lib/telegram/apps/handlers/bot_operations.py",

src/aipass/skills/lib/telegram/apps/handlers/base_bot.py

@@ -47,6 +47,7 @@ import argparse
 import atexit
 import json
 import os
+import re
 import signal
 import subprocess
 import sys
@@ -123,6 +124,7 @@ except ImportError:
 # MODULE-LEVEL CONSTANTS
 # =============================================
 
+CC_SESSIONS_DIR = Path.home() / ".claude" / "sessions"
 PENDING_DIR = Path.home() / ".aipass" / "telegram_pending"
 PENDING_TTL = 3600  # 1 hour
 TELEGRAM_CHAR_LIMIT = 4096
@@ -201,6 +203,8 @@ class BaseBot:
         self._mirror_mapping_written = False
         self._last_transcript_path: str | None = None
         self._config_chat_id: int | None = None
+        self._active_session_id: str | None = None
+        self._active_transcript_path: Path | None = None
 
         self.state = {
             "running": True,
@@ -1264,24 +1268,26 @@ class BaseBot:
         Returns:
             True if session is ready
         """
-        # Strategy 1: follow the central presence pointer
-        presence = self._read_presence_pointer()
-        if presence:
-            tmux_target = self._find_tmux_for_presence(presence)
+        # Strategy 1: CC-native session discovery (DPLAN-0226)
+        cc_session = self._discover_cc_session()
+        if cc_session:
+            tmux_target = self._find_tmux_pane_by_cwd()
             if tmux_target:
                 self.session_name = tmux_target
                 self._using_shared_session = True
+                self._active_session_id = cc_session.get("sessionId")
+                self._active_transcript_path = self._resolve_cc_transcript_path(cc_session)
                 logger.info(
-                    "Presence pointer: attaching to '%s' (PID %d, type=%s)",
+                    "CC-native discovery: session '%s' (PID %d) → tmux '%s'",
+                    cc_session.get("name", "?"),
+                    cc_session.get("pid", 0),
                     tmux_target,
-                    presence.get("pid", 0),
-                    presence.get("session_type", "unknown"),
                 )
                 self._write_mirror_mapping()
                 return True
             logger.warning(
-                "Presence pointer found (PID %d) but no matching tmux session",
-                presence.get("pid", 0),
+                "CC session found (PID %d) but no matching tmux pane",
+                cc_session.get("pid", 0),
             )
 
         # Strategy 2: explicit shared-session from config
@@ -1430,6 +1436,8 @@ class BaseBot:
             "processing_message_id": processing_message_id,
             "timestamp": time.time(),
             "transcript_line_after": transcript_line_after,
+            "transcript_path": str(self._active_transcript_path) if self._active_transcript_path else None,
+            "session_id": self._active_session_id,
         }
 
         try:
@@ -1458,11 +1466,14 @@ class BaseBot:
     def _resolve_active_transcript(self) -> tuple[str | None, int]:
         """Identify the ACTIVE Claude JSONL transcript and return its path and line count.
 
-        Uses the tmux session's pane PID to walk the process tree and find which
-        JSONL file the Claude process has open.  Falls back to most-recently-modified
-        JSONL that was touched in the last 5 minutes.  Returns (None, 0) when no
-        active transcript can be identified — safe default that resets the cursor.
+        Prefers the CC-native transcript path when available (set by CC session
+        discovery in ensure_tmux_session). Falls back to PID-based fd scanning
+        and mtime heuristic.
         """
+        # Strategy 0: CC-native transcript path (set by _discover_cc_session)
+        if self._active_transcript_path and self._active_transcript_path.exists():
+            return str(self._active_transcript_path), self._count_file_lines(self._active_transcript_path)
+
         slug = str(self.work_dir).replace("\\", "-").replace("/", "-")
         projects_dir = Path.home() / ".claude" / "projects" / slug
         if not projects_dir.exists():
@@ -1577,76 +1588,61 @@ class BaseBot:
         return count
 
     # =============================================
-    # PRESENCE POINTER
+    # CC-NATIVE SESSION DISCOVERY (DPLAN-0226)
     # =============================================
 
-    def _find_presence_file(self) -> Path | None:
-        """Locate .ai_central/PRESENCE.central.json by walking up from work_dir."""
-        current = self.work_dir.resolve()
-        for _ in range(20):
-            candidate = current / ".ai_central" / "PRESENCE.central.json"
-            if candidate.exists():
-                return candidate
-            if current.parent == current:
-                break
-            current = current.parent
-        return None
+    def _discover_cc_session(self) -> dict | None:
+        """Discover the active CC session for this bot's branch.
 
-    def _read_presence_pointer(self) -> dict | None:
-        """Read the central presence pointer for this bot's branch.
-
-        Returns the presence entry dict if a live session is registered,
-        None if absent/empty/stale/dead.
+        Enumerates ~/.claude/sessions/<pid>.json, filters by cwd matching
+        this bot's work_dir, confirms PID alive. Returns the latest session
+        (by startedAt) or None. CC keeps these current across /resume and
+        deletes on exit, so re-binding is automatic.
         """
-        presence_file = self._find_presence_file()
-        if presence_file is None:
-            return None
-        try:
-            data = json.loads(presence_file.read_text(encoding="utf-8"))
-        except (json.JSONDecodeError, OSError) as exc:
-            logger.info("Failed to read presence file %s: %s", presence_file, exc)
-            return None
-        branch = self.branch_name or self.work_dir.name
-        entry = data.get(branch)
-        if not entry:
-            return None
-        pid = entry.get("pid")
-        if pid is None:
+        if not CC_SESSIONS_DIR.is_dir():
             return None
+        target_cwd = str(self.work_dir.resolve())
+        best = None
+        for f in CC_SESSIONS_DIR.iterdir():
+            if not f.name.endswith(".json") or not f.stem.isdigit():
+                continue
+            try:
+                data = json.loads(f.read_text(encoding="utf-8"))
+            except (json.JSONDecodeError, OSError) as exc:
+                logger.info("Failed to read CC session file %s: %s", f, exc)
+                continue
+            session_cwd = data.get("cwd", "")
+            if not session_cwd:
+                continue
+            if str(Path(session_cwd).resolve()) != target_cwd:
+                continue
+            pid = data.get("pid")
+            if not pid or not self._is_pid_alive(pid):
+                continue
+            if best is None or data.get("startedAt", 0) > best.get("startedAt", 0):
+                best = data
+        return best
+
+    @staticmethod
+    def _is_pid_alive(pid: int) -> bool:
+        """Check if a process with the given PID exists."""
+        if pid <= 1:
+            return False
         try:
             os.kill(pid, 0)
+            return True
         except ProcessLookupError:
-            logger.info("Presence holder PID %d is dead — stale pointer", pid)
-            return None
+            logger.info("PID %d not found (dead)", pid)
+            return False
         except PermissionError:
             logger.info("PID %d exists but permission denied — treating as alive", pid)
+            return True
         except OSError as exc:
             logger.info("PID %d liveness check failed: %s — treating as dead", pid, exc)
-            return None
-        return entry
+            return False
 
-    def _find_tmux_for_presence(self, entry: dict) -> str | None:
-        """Find the tmux session for a presence entry.
-
-        Prefers attach_handle when populated; falls back to scanning tmux
-        sessions for a pane whose CWD matches the entry's work_dir.
-        """
-        handle = entry.get("attach_handle", "")
-        if handle:
-            try:
-                result = subprocess.run(
-                    ["tmux", "has-session", "-t", handle],
-                    capture_output=True,
-                    timeout=5,
-                )
-                if result.returncode == 0:
-                    return handle
-            except (FileNotFoundError, subprocess.TimeoutExpired, OSError) as exc:
-                logger.info("attach_handle '%s' tmux check failed: %s", handle, exc)
-
-        work_dir = entry.get("work_dir", "")
-        if not work_dir:
-            return None
+    def _find_tmux_pane_by_cwd(self) -> str | None:
+        """Find a tmux session with a pane whose CWD matches this bot's work_dir."""
         try:
             result = subprocess.run(
                 ["tmux", "list-panes", "-a", "-F", "#{session_name}:#{pane_current_path}"],
@@ -1656,7 +1652,7 @@ class BaseBot:
             )
             if result.returncode != 0:
                 return None
-            target = str(Path(work_dir).resolve())
+            target = str(self.work_dir.resolve())
             for line in result.stdout.strip().split("\n"):
                 if ":" not in line:
                     continue
@@ -1667,6 +1663,147 @@ class BaseBot:
             logger.info("tmux pane scan failed: %s", exc)
         return None
 
+    @staticmethod
+    def _sanitize_path_for_cc(path_str: str) -> str:
+        """Sanitize a path for CC projects dir (every non-alphanumeric char becomes '-')."""
+        return re.sub(r"[^a-zA-Z0-9]", "-", path_str)
+
+    def _resolve_cc_transcript_path(self, session: dict) -> Path | None:
+        """Resolve the transcript JSONL path from a CC session entry."""
+        session_id = session.get("sessionId")
+        cwd = session.get("cwd", "")
+        if not session_id or not cwd:
+            return None
+        slug = self._sanitize_path_for_cc(cwd)
+        transcript = Path.home() / ".claude" / "projects" / slug / f"{session_id}.jsonl"
+        if transcript.exists():
+            return transcript
+        return None
+
+    @staticmethod
+    def _extract_assistant_text(entry: dict) -> str | None:
+        """Extract text content from a transcript entry if it's an assistant message."""
+        msg = entry.get("message", {})
+        if msg.get("role") != "assistant":
+            return None
+        content = msg.get("content", [])
+        if isinstance(content, str):
+            return content
+        if isinstance(content, list):
+            for block in content:
+                if isinstance(block, dict) and block.get("type") == "text":
+                    return block.get("text", "")
+        return None
+
+    def _read_transcript_tail(self, n_lines: int = 50) -> str | None:
+        """Read the latest assistant text response from the active transcript.
+
+        Scans the last n_lines of the transcript JSONL for the most recent
+        assistant message with text content.
+        """
+        path = self._active_transcript_path
+        if not path or not path.exists():
+            return None
+        try:
+            text = path.read_text(encoding="utf-8").strip()
+            if not text:
+                return None
+            lines = text.split("\n")
+            tail = lines[-n_lines:] if len(lines) > n_lines else lines
+            for line in reversed(tail):
+                try:
+                    entry = json.loads(line)
+                except json.JSONDecodeError:
+                    logger.info("Malformed JSONL line in transcript tail")
+                    continue
+                result = self._extract_assistant_text(entry)
+                if result is not None:
+                    return result
+        except OSError as exc:
+            logger.info("Failed to read transcript tail: %s", exc)
+        return None
+
+    # =============================================
+    # DORMANT: PRESENCE POINTER (DPLAN-0226)
+    # Discovery replaced by CC-native ~/.claude/sessions.
+    # Guard half (presence_gate) stays LIVE in @hooks.
+    # Kept per Patrick's directive — do not delete.
+    # =============================================
+
+    # def _find_presence_file(self) -> Path | None:
+    #     """Locate .ai_central/PRESENCE.central.json by walking up from work_dir."""
+    #     current = self.work_dir.resolve()
+    #     for _ in range(20):
+    #         candidate = current / ".ai_central" / "PRESENCE.central.json"
+    #         if candidate.exists():
+    #             return candidate
+    #         if current.parent == current:
+    #             break
+    #         current = current.parent
+    #     return None
+    #
+    # def _read_presence_pointer(self) -> dict | None:
+    #     """Read the central presence pointer for this bot's branch."""
+    #     presence_file = self._find_presence_file()
+    #     if presence_file is None:
+    #         return None
+    #     try:
+    #         data = json.loads(presence_file.read_text(encoding="utf-8"))
+    #     except (json.JSONDecodeError, OSError):
+    #         return None
+    #     branch = self.branch_name or self.work_dir.name
+    #     entry = data.get(branch)
+    #     if not entry:
+    #         return None
+    #     pid = entry.get("pid")
+    #     if pid is None:
+    #         return None
+    #     try:
+    #         os.kill(pid, 0)
+    #     except ProcessLookupError:
+    #         return None
+    #     except PermissionError:
+    #         pass
+    #     except OSError:
+    #         return None
+    #     return entry
+    #
+    # def _find_tmux_for_presence(self, entry: dict) -> str | None:
+    #     """Find the tmux session for a presence entry."""
+    #     handle = entry.get("attach_handle", "")
+    #     if handle:
+    #         try:
+    #             result = subprocess.run(
+    #                 ["tmux", "has-session", "-t", handle],
+    #                 capture_output=True,
+    #                 timeout=5,
+    #             )
+    #             if result.returncode == 0:
+    #                 return handle
+    #         except (FileNotFoundError, subprocess.TimeoutExpired, OSError):
+    #             pass
+    #     work_dir = entry.get("work_dir", "")
+    #     if not work_dir:
+    #         return None
+    #     try:
+    #         result = subprocess.run(
+    #             ["tmux", "list-panes", "-a", "-F",
+    #              "#{session_name}:#{pane_current_path}"],
+    #             capture_output=True, text=True, timeout=5,
+    #         )
+    #         if result.returncode != 0:
+    #             return None
+    #         target = str(Path(work_dir).resolve())
+    #         for line in result.stdout.strip().split("\n"):
+    #             if ":" not in line:
+    #                 continue
+    #             session_name, pane_path = line.split(":", 1)
+    #             if str(Path(pane_path).resolve()) == target:
+    #                 return session_name
+    #     except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
+    #         pass
+    #     return None
+
     def _write_mirror_mapping(self) -> None:
         """Write persistent mirror mapping file per THE CONTRACT (TDPLAN-0009).
 

src/aipass/skills/lib/telegram/tests/test_presence_pointer.py

@@ -1,19 +1,22 @@
 # =================== AIPass ====================
 # Name: test_presence_pointer.py
-# Description: Tests for FPLAN-0289 P2 — bot follows central presence pointer, never spawns
-# Version: 1.0.0
+# Description: Tests for CC-native session discovery (DPLAN-0226) + transcript tail
+# Version: 2.0.0
 # Created: 2026-06-29
-# Modified: 2026-06-29
+# Modified: 2026-06-30
 # =============================================
 
 """
-Tests for the presence pointer integration (FPLAN-0289 P2).
+Tests for CC-native session discovery (DPLAN-0226).
 
 Covers:
-  - _find_presence_file: walks up from work_dir to locate PRESENCE.central.json
-  - _read_presence_pointer: reads pointer, validates PID liveness, returns entry
-  - _find_tmux_for_presence: attach_handle preference, tmux CWD scan fallback
-  - ensure_tmux_session: presence-first resolution, legacy spawn retired
+  - _discover_cc_session: enumerate ~/.claude/sessions, filter cwd, validate PID
+  - _is_pid_alive: PID liveness check
+  - _find_tmux_pane_by_cwd: tmux pane CWD scan
+  - _sanitize_path_for_cc: path sanitization for CC projects dir
+  - _resolve_cc_transcript_path: transcript JSONL path from session info
+  - _extract_assistant_text / _read_transcript_tail: transcript parsing
+  - ensure_tmux_session: CC-native-first resolution, legacy spawn retired
   - handle_message: no-session error message
 """
 
@@ -52,10 +55,10 @@ def _make_bot(tmp_path, _patch_base_bot_deps, branch_name="devpulse"):
     workdir.mkdir(exist_ok=True)
     with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
         bot = BaseBot(
-            bot_id="presence_test",
+            bot_id="cc_test",
             bot_token="123:FAKETOKEN",
             work_dir=workdir,
-            bot_name="Presence Test Bot",
+            bot_name="CC Discovery Test Bot",
             allowed_user_ids=[111],
             branch_name=branch_name,
         )
@@ -63,272 +66,378 @@ def _make_bot(tmp_path, _patch_base_bot_deps, branch_name="devpulse"):
     return bot
 
 
-def _write_presence(tmp_path, data):
-    """Write a PRESENCE.central.json reachable from bot's work_dir."""
-    ai_central = tmp_path / ".ai_central"
-    ai_central.mkdir(exist_ok=True)
-    presence_file = ai_central / "PRESENCE.central.json"
-    presence_file.write_text(json.dumps(data), encoding="utf-8")
-    return presence_file
+def _write_cc_session(sessions_dir, pid, cwd, session_id="sess-001", started_at=1000, **extra):
+    """Write a CC session file at sessions_dir/<pid>.json."""
+    sessions_dir.mkdir(parents=True, exist_ok=True)
+    data = {
+        "pid": pid,
+        "sessionId": session_id,
+        "cwd": cwd,
+        "startedAt": started_at,
+        "kind": "interactive",
+        "name": "test-session",
+        **extra,
+    }
+    f = sessions_dir / f"{pid}.json"
+    f.write_text(json.dumps(data), encoding="utf-8")
+    return f
+
+
+def _write_transcript(projects_dir, cwd, session_id, lines):
+    """Write a transcript JSONL file at the CC-native path."""
+    slug = BaseBot._sanitize_path_for_cc(cwd)
+    transcript_dir = projects_dir / slug
+    transcript_dir.mkdir(parents=True, exist_ok=True)
+    transcript_file = transcript_dir / f"{session_id}.jsonl"
+    transcript_file.write_text("\n".join(json.dumps(item) for item in lines), encoding="utf-8")
+    return transcript_file
 
 
 # =============================================
-# 1. _find_presence_file
+# 1. _discover_cc_session
 # =============================================
 
 
-class TestFindPresenceFile:
-    """Locate PRESENCE.central.json by walking up from work_dir."""
+class TestDiscoverCcSession:
+    """Enumerate ~/.claude/sessions, filter cwd, validate PID."""
 
-    def test_finds_file_in_parent(self, tmp_path, _patch_base_bot_deps):
-        """Finds PRESENCE.central.json in parent of work_dir."""
+    def test_discovers_live_session(self, tmp_path, _patch_base_bot_deps):
+        """Finds a live session whose cwd matches the bot work directory."""
         bot = _make_bot(tmp_path, _patch_base_bot_deps)
-        _write_presence(tmp_path, {})
-        result = bot._find_presence_file()
+        sessions_dir = tmp_path / "sessions"
+        _write_cc_session(sessions_dir, os.getpid(), str(bot.work_dir.resolve()))
+        with patch("apps.handlers.base_bot.CC_SESSIONS_DIR", sessions_dir):
+            result = bot._discover_cc_session()
         assert result is not None
-        assert result.name == "PRESENCE.central.json"
+        assert result["pid"] == os.getpid()
+        assert result["sessionId"] == "sess-001"
 
-    def test_returns_none_when_absent(self, tmp_path, _patch_base_bot_deps):
-        """Returns None when no .ai_central/ exists."""
+    def test_returns_none_when_no_sessions_dir(self, tmp_path, _patch_base_bot_deps):
+        """Returns None when the sessions directory does not exist."""
         bot = _make_bot(tmp_path, _patch_base_bot_deps)
-        result = bot._find_presence_file()
+        fake_dir = tmp_path / "no_such_dir"
+        with patch("apps.handlers.base_bot.CC_SESSIONS_DIR", fake_dir):
+            result = bot._discover_cc_session()
         assert result is None
 
-    def test_walks_up_multiple_levels(self, tmp_path, _patch_base_bot_deps):
-        """Walks up multiple parent directories to find .ai_central/."""
-        deep_dir = tmp_path / "a" / "b" / "c" / "workdir"
-        deep_dir.mkdir(parents=True)
-        _write_presence(tmp_path, {})
-        with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
-            bot = BaseBot(
-                bot_id="deep_test",
-                bot_token="t",
-                work_dir=deep_dir,
-                branch_name="test",
-            )
-        result = bot._find_presence_file()
-        assert result is not None
-
-
-# =============================================
-# 2. _read_presence_pointer
-# =============================================
-
-
-class TestReadPresencePointer:
-    """Read pointer, validate PID liveness, return entry or None."""
-
-    def test_returns_entry_for_live_pid(self, tmp_path, _patch_base_bot_deps):
-        """Returns entry dict when PID is alive."""
+    def test_returns_none_when_no_matching_cwd(self, tmp_path, _patch_base_bot_deps):
+        """Returns None when no session file has a matching cwd."""
         bot = _make_bot(tmp_path, _patch_base_bot_deps)
-        my_pid = os.getpid()
-        _write_presence(
-            tmp_path,
-            {
-                "devpulse": {
-                    "pid": my_pid,
-                    "session_id": "sess1",
-                    "work_dir": str(tmp_path),
-                    "session_type": "interactive",
-                    "attach_handle": "",
-                    "started": "2026-06-29T10:00:00",
-                    "last_seen": "2026-06-29T10:01:00",
-                }
-            },
-        )
-        result = bot._read_presence_pointer()
-        assert result is not None
-        assert result["pid"] == my_pid
-        assert result["session_type"] == "interactive"
+        sessions_dir = tmp_path / "sessions"
+        _write_cc_session(sessions_dir, os.getpid(), "/some/other/dir")
+        with patch("apps.handlers.base_bot.CC_SESSIONS_DIR", sessions_dir):
+            result = bot._discover_cc_session()
+        assert result is None
 
     def test_returns_none_for_dead_pid(self, tmp_path, _patch_base_bot_deps):
-        """Returns None when the recorded PID is not running."""
+        """Returns None when the session PID is no longer alive."""
         bot = _make_bot(tmp_path, _patch_base_bot_deps)
-        _write_presence(
-            tmp_path,
-            {
-                "devpulse": {
-                    "pid": 99999999,
-                    "work_dir": str(tmp_path),
-                }
-            },
-        )
-        result = bot._read_presence_pointer()
+        sessions_dir = tmp_path / "sessions"
+        _write_cc_session(sessions_dir, 99999999, str(bot.work_dir.resolve()))
+        with patch("apps.handlers.base_bot.CC_SESSIONS_DIR", sessions_dir):
+            result = bot._discover_cc_session()
         assert result is None
 
-    def test_returns_none_when_branch_missing(self, tmp_path, _patch_base_bot_deps):
-        """Returns None when the bot's branch has no entry."""
-        bot = _make_bot(tmp_path, _patch_base_bot_deps, branch_name="devpulse")
-        _write_presence(tmp_path, {"other_branch": {"pid": os.getpid(), "work_dir": str(tmp_path)}})
-        result = bot._read_presence_pointer()
-        assert result is None
-
-    def test_returns_none_when_file_empty(self, tmp_path, _patch_base_bot_deps):
-        """Returns None when presence file is empty JSON object."""
+    def test_picks_latest_when_multiple(self, tmp_path, _patch_base_bot_deps):
+        """Selects the session with the highest startedAt when multiple match."""
         bot = _make_bot(tmp_path, _patch_base_bot_deps)
-        _write_presence(tmp_path, {})
-        result = bot._read_presence_pointer()
-        assert result is None
+        sessions_dir = tmp_path / "sessions"
+        cwd = str(bot.work_dir.resolve())
+        my_pid = os.getpid()
+        _write_cc_session(sessions_dir, my_pid, cwd, session_id="old", started_at=100)
+        _write_cc_session(sessions_dir, my_pid + 1, cwd, session_id="new", started_at=200)
+        with (
+            patch("apps.handlers.base_bot.CC_SESSIONS_DIR", sessions_dir),
+            patch.object(BaseBot, "_is_pid_alive", return_value=True),
+        ):
+            result = bot._discover_cc_session()
+        assert result is not None
+        assert result["sessionId"] == "new"
 
-    def test_returns_none_when_no_presence_file(self, tmp_path, _patch_base_bot_deps):
-        """Returns None when no PRESENCE.central.json exists."""
+    def test_skips_non_json_files(self, tmp_path, _patch_base_bot_deps):
+        """Ignores non-JSON files and files without numeric PID names."""
         bot = _make_bot(tmp_path, _patch_base_bot_deps)
-        result = bot._read_presence_pointer()
+        sessions_dir = tmp_path / "sessions"
+        sessions_dir.mkdir(parents=True)
+        (sessions_dir / "readme.txt").write_text("ignore me")
+        (sessions_dir / "notapid.json").write_text("{}")
+        with patch("apps.handlers.base_bot.CC_SESSIONS_DIR", sessions_dir):
+            result = bot._discover_cc_session()
         assert result is None
 
-    def test_returns_none_when_corrupt_json(self, tmp_path, _patch_base_bot_deps):
-        """Returns None when presence file contains invalid JSON."""
+    def test_skips_corrupt_json(self, tmp_path, _patch_base_bot_deps):
+        """Gracefully skips session files containing invalid JSON."""
         bot = _make_bot(tmp_path, _patch_base_bot_deps)
-        ai_central = tmp_path / ".ai_central"
-        ai_central.mkdir()
-        (ai_central / "PRESENCE.central.json").write_text("not json!", encoding="utf-8")
-        result = bot._read_presence_pointer()
+        sessions_dir = tmp_path / "sessions"
+        sessions_dir.mkdir(parents=True)
+        (sessions_dir / "12345.json").write_text("not json!")
+        with patch("apps.handlers.base_bot.CC_SESSIONS_DIR", sessions_dir):
+            result = bot._discover_cc_session()
         assert result is None
 
-    def test_returns_none_when_no_pid(self, tmp_path, _patch_base_bot_deps):
-        """Returns None when the entry has no pid field."""
-        bot = _make_bot(tmp_path, _patch_base_bot_deps)
-        _write_presence(tmp_path, {"devpulse": {"work_dir": str(tmp_path)}})
-        result = bot._read_presence_pointer()
-        assert result is None
-
-    def test_uses_work_dir_name_when_no_branch_name(self, tmp_path, _patch_base_bot_deps):
-        """Falls back to work_dir.name as branch key when branch_name is None."""
-        bot = _make_bot(tmp_path, _patch_base_bot_deps, branch_name="devpulse")
-        bot.branch_name = None
-        workdir_name = bot.work_dir.name
-        _write_presence(tmp_path, {workdir_name: {"pid": os.getpid(), "work_dir": str(tmp_path)}})
-        result = bot._read_presence_pointer()
+    def test_resolves_symlinked_cwd(self, tmp_path, _patch_base_bot_deps):
+        """Resolved paths match even through symlinks."""
+        real_dir = tmp_path / "real_workdir"
+        real_dir.mkdir()
+        link = tmp_path / "link_workdir"
+        link.symlink_to(real_dir)
+        with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
+            bot = BaseBot(
+                bot_id="sym_test",
+                bot_token="t",
+                work_dir=link,
+                branch_name="test",
+            )
+        sessions_dir = tmp_path / "sessions"
+        _write_cc_session(sessions_dir, os.getpid(), str(real_dir))
+        with patch("apps.handlers.base_bot.CC_SESSIONS_DIR", sessions_dir):
+            result = bot._discover_cc_session()
         assert result is not None
 
-    def test_permission_error_treats_as_alive(self, tmp_path, _patch_base_bot_deps):
-        """PermissionError on os.kill means process exists — treat as alive."""
-        bot = _make_bot(tmp_path, _patch_base_bot_deps)
-        _write_presence(tmp_path, {"devpulse": {"pid": 1, "work_dir": str(tmp_path)}})
+
+# =============================================
+# 2. _is_pid_alive
+# =============================================
+
+
+class TestIsPidAlive:
+    """PID liveness check via os.kill(pid, 0)."""
+
+    def test_own_pid_is_alive(self):
+        """Reports the current process PID as alive."""
+        assert BaseBot._is_pid_alive(os.getpid()) is True
+
+    def test_dead_pid(self):
+        """Reports a non-existent PID as dead."""
+        assert BaseBot._is_pid_alive(99999999) is False
+
+    def test_zero_pid(self):
+        """Reports PID zero as dead."""
+        assert BaseBot._is_pid_alive(0) is False
+
+    def test_negative_pid(self):
+        """Reports a negative PID as dead."""
+        assert BaseBot._is_pid_alive(-1) is False
+
+    def test_permission_error_treated_as_alive(self):
+        """Treats PermissionError from os.kill as evidence the PID is alive."""
         with patch("os.kill", side_effect=PermissionError("denied")):
-            result = bot._read_presence_pointer()
-        assert result is not None
+            assert BaseBot._is_pid_alive(42) is True
+
+    def test_os_error_treated_as_dead(self):
+        """Treats a generic OSError from os.kill as evidence the PID is dead."""
+        with patch("os.kill", side_effect=OSError("some error")):
+            assert BaseBot._is_pid_alive(42) is False
 
 
 # =============================================
-# 3. _find_tmux_for_presence
+# 3. _find_tmux_pane_by_cwd
 # =============================================
 
 
-class TestFindTmuxForPresence:
-    """Find tmux session via attach_handle or CWD scan."""
+class TestFindTmuxPaneByCwd:
+    """Find tmux session with a pane whose CWD matches work_dir."""
 
-    def test_uses_attach_handle_when_present(self, tmp_path, _patch_base_bot_deps):
-        """Returns attach_handle directly when the tmux session exists."""
+    def test_finds_matching_pane(self, tmp_path, _patch_base_bot_deps):
+        """Returns the tmux session name whose pane cwd matches work_dir."""
         bot = _make_bot(tmp_path, _patch_base_bot_deps)
-        entry = {"attach_handle": "my-session", "work_dir": str(tmp_path)}
-        with patch("subprocess.run", return_value=MagicMock(returncode=0)):
-            result = bot._find_tmux_for_presence(entry)
+        workdir = str(bot.work_dir.resolve())
+        pane_output = f"my-session:{workdir}\nother:{tmp_path}\n"
+        with patch(
+            "subprocess.run",
+            return_value=MagicMock(returncode=0, stdout=pane_output),
+        ):
+            result = bot._find_tmux_pane_by_cwd()
         assert result == "my-session"
 
-    def test_attach_handle_falls_back_on_missing_session(self, tmp_path, _patch_base_bot_deps):
-        """Falls back to CWD scan when attach_handle session doesn't exist."""
-        bot = _make_bot(tmp_path, _patch_base_bot_deps)
-        entry = {"attach_handle": "dead-session", "work_dir": str(tmp_path)}
-
-        def _side_effect(cmd, **kwargs):
-            mock = MagicMock()
-            if "has-session" in cmd:
-                mock.returncode = 1
-            elif "list-panes" in cmd:
-                mock.returncode = 0
-                mock.stdout = f"live-session:{tmp_path}\n"
-            return mock
-
-        with patch("subprocess.run", side_effect=_side_effect):
-            result = bot._find_tmux_for_presence(entry)
-        assert result == "live-session"
-
-    def test_cwd_scan_matches_work_dir(self, tmp_path, _patch_base_bot_deps):
-        """CWD scan finds the session whose pane path matches work_dir."""
-        bot = _make_bot(tmp_path, _patch_base_bot_deps)
-        entry = {"attach_handle": "", "work_dir": str(tmp_path / "workdir")}
-        pane_output = f"dev-session:{tmp_path / 'workdir'}\nother:{tmp_path / 'other'}\n"
-        with patch(
-            "subprocess.run",
-            return_value=MagicMock(returncode=0, stdout=pane_output),
-        ):
-            result = bot._find_tmux_for_presence(entry)
-        assert result == "dev-session"
-
     def test_returns_none_when_no_match(self, tmp_path, _patch_base_bot_deps):
-        """Returns None when no tmux pane CWD matches the work_dir."""
+        """Returns None when no tmux pane cwd matches the bot work directory."""
         bot = _make_bot(tmp_path, _patch_base_bot_deps)
-        entry = {"attach_handle": "", "work_dir": "/no/such/dir"}
-        pane_output = f"session1:{tmp_path}\n"
+        pane_output = f"session1:{tmp_path / 'other'}\n"
         with patch(
             "subprocess.run",
             return_value=MagicMock(returncode=0, stdout=pane_output),
         ):
-            result = bot._find_tmux_for_presence(entry)
-        assert result is None
-
-    def test_returns_none_when_no_work_dir(self, tmp_path, _patch_base_bot_deps):
-        """Returns None immediately when work_dir is empty."""
-        bot = _make_bot(tmp_path, _patch_base_bot_deps)
-        entry = {"attach_handle": "", "work_dir": ""}
-        result = bot._find_tmux_for_presence(entry)
+            result = bot._find_tmux_pane_by_cwd()
         assert result is None
 
     def test_returns_none_when_tmux_unavailable(self, tmp_path, _patch_base_bot_deps):
-        """Returns None when tmux is not installed."""
+        """Returns None when the tmux binary is not found."""
         bot = _make_bot(tmp_path, _patch_base_bot_deps)
-        entry = {"attach_handle": "", "work_dir": str(tmp_path)}
         with patch("subprocess.run", side_effect=FileNotFoundError("tmux")):
-            result = bot._find_tmux_for_presence(entry)
+            result = bot._find_tmux_pane_by_cwd()
         assert result is None
 
-    def test_empty_attach_handle_skipped(self, tmp_path, _patch_base_bot_deps):
-        """Empty attach_handle skips has-session check, goes straight to CWD scan."""
+    def test_returns_none_when_tmux_fails(self, tmp_path, _patch_base_bot_deps):
+        """Returns None when the tmux list-panes command exits with failure."""
         bot = _make_bot(tmp_path, _patch_base_bot_deps)
-        entry = {"attach_handle": "", "work_dir": str(tmp_path / "workdir")}
-        pane_output = f"found:{tmp_path / 'workdir'}\n"
-        with patch(
-            "subprocess.run",
-            return_value=MagicMock(returncode=0, stdout=pane_output),
+        with patch("subprocess.run", return_value=MagicMock(returncode=1)):
+            result = bot._find_tmux_pane_by_cwd()
+        assert result is None
+
+
+# =============================================
+# 4. _sanitize_path_for_cc
+# =============================================
+
+
+class TestSanitizePathForCc:
+    """Path sanitization: every non-alphanumeric char becomes '-'."""
+
+    def test_basic_path(self):
+        """Slashes become dashes."""
+        assert BaseBot._sanitize_path_for_cc("/opt/work/project") == "-opt-work-project"
+
+    def test_dots_and_underscores(self):
+        """Dots and underscores become dashes."""
+        assert BaseBot._sanitize_path_for_cc("/opt/work/.my_project") == "-opt-work--my-project"
+
+    def test_backslashes(self):
+        """Backslashes become dashes."""
+        assert BaseBot._sanitize_path_for_cc("D:\\Work\\project") == "D--Work-project"
+
+    def test_alphanumeric_preserved(self):
+        """Alphanumeric characters are kept as-is."""
+        assert BaseBot._sanitize_path_for_cc("abc123XYZ") == "abc123XYZ"
+
+
+# =============================================
+# 5. _resolve_cc_transcript_path
+# =============================================
+
+
+class TestResolveCcTranscriptPath:
+    """Resolve transcript JSONL path from CC session info."""
+
+    def test_resolves_existing_transcript(self, tmp_path, _patch_base_bot_deps):
+        """Returns the transcript path when the JSONL file exists on disk."""
+        bot = _make_bot(tmp_path, _patch_base_bot_deps)
+        cwd = str(bot.work_dir.resolve())
+        session = {"sessionId": "abc-123", "cwd": cwd}
+        slug = BaseBot._sanitize_path_for_cc(cwd)
+        transcript_dir = tmp_path / ".claude" / "projects" / slug
+        transcript_dir.mkdir(parents=True)
+        transcript_file = transcript_dir / "abc-123.jsonl"
+        transcript_file.write_text("{}", encoding="utf-8")
+        with patch("pathlib.Path.home", return_value=tmp_path):
+            result = bot._resolve_cc_transcript_path(session)
+        assert result is not None
+        assert result.name == "abc-123.jsonl"
+
+    def test_returns_none_when_file_missing(self, tmp_path, _patch_base_bot_deps):
+        """Returns None when the transcript JSONL file does not exist."""
+        bot = _make_bot(tmp_path, _patch_base_bot_deps)
+        session = {"sessionId": "nonexistent", "cwd": str(tmp_path)}
+        with patch("pathlib.Path.home", return_value=tmp_path):
+            result = bot._resolve_cc_transcript_path(session)
+        assert result is None
+
+    def test_returns_none_when_no_session_id(self, tmp_path, _patch_base_bot_deps):
+        """Returns None when the session dict lacks a sessionId key."""
+        bot = _make_bot(tmp_path, _patch_base_bot_deps)
+        result = bot._resolve_cc_transcript_path({"cwd": str(tmp_path)})
+        assert result is None
+
+    def test_returns_none_when_no_cwd(self, tmp_path, _patch_base_bot_deps):
+        """Returns None when the session dict lacks a cwd key."""
+        bot = _make_bot(tmp_path, _patch_base_bot_deps)
+        result = bot._resolve_cc_transcript_path({"sessionId": "abc"})
+        assert result is None
+
+
+# =============================================
+# 6. _extract_assistant_text / _read_transcript_tail
+# =============================================
+
+
+class TestTranscriptParsing:
+    """Extract assistant text from transcript JSONL entries."""
+
+    def test_extract_text_from_text_block(self):
+        """Extracts text from a content block with type 'text'."""
+        entry = {"message": {"role": "assistant", "content": [{"type": "text", "text": "Hello world"}]}}
+        assert BaseBot._extract_assistant_text(entry) == "Hello world"
+
+    def test_extract_text_from_string_content(self):
+        """Extracts text when content is a plain string instead of a list."""
+        entry = {"message": {"role": "assistant", "content": "Plain text"}}
+        assert BaseBot._extract_assistant_text(entry) == "Plain text"
+
+    def test_returns_none_for_user_message(self):
+        """Returns None for entries with role 'user'."""
+        entry = {"message": {"role": "user", "content": "question"}}
+        assert BaseBot._extract_assistant_text(entry) is None
+
+    def test_returns_none_for_tool_use_only(self):
+        """Returns None when content contains only tool_use blocks."""
+        entry = {"message": {"role": "assistant", "content": [{"type": "tool_use", "id": "t1"}]}}
+        assert BaseBot._extract_assistant_text(entry) is None
+
+    def test_returns_none_for_empty_entry(self):
+        """Returns None for an empty dict entry."""
+        assert BaseBot._extract_assistant_text({}) is None
+
+    def test_read_transcript_tail_finds_response(self, tmp_path, _patch_base_bot_deps):
+        """Returns the last assistant text from the transcript tail."""
+        bot = _make_bot(tmp_path, _patch_base_bot_deps)
+        transcript = tmp_path / "transcript.jsonl"
+        lines = [
+            json.dumps({"message": {"role": "user", "content": "hi"}}),
+            json.dumps({"message": {"role": "assistant", "content": [{"type": "text", "text": "Hello!"}]}}),
+        ]
+        transcript.write_text("\n".join(lines), encoding="utf-8")
+        bot._active_transcript_path = transcript
+        result = bot._read_transcript_tail()
+        assert result == "Hello!"
+
+    def test_read_transcript_tail_returns_none_when_no_path(self, tmp_path, _patch_base_bot_deps):
+        """Returns None when no active transcript path is set."""
+        bot = _make_bot(tmp_path, _patch_base_bot_deps)
+        bot._active_transcript_path = None
+        assert bot._read_transcript_tail() is None
+
+    def test_read_transcript_tail_returns_none_when_file_missing(self, tmp_path, _patch_base_bot_deps):
+        """Returns None when the transcript file does not exist on disk."""
+        bot = _make_bot(tmp_path, _patch_base_bot_deps)
+        bot._active_transcript_path = tmp_path / "nonexistent.jsonl"
+        assert bot._read_transcript_tail() is None
+
+
+# =============================================
+# 7. ensure_tmux_session — CC-native-first
+# =============================================
+
+
+class TestEnsureWithCcDiscovery:
+    """ensure_tmux_session follows CC-native session discovery first."""
+
+    def test_attaches_via_cc_session(self, tmp_path, _patch_base_bot_deps):
+        """Attaches to a live CC session and sets shared session state."""
+        bot = _make_bot(tmp_path, _patch_base_bot_deps)
+        workdir = str(bot.work_dir.resolve())
+        session = {
+            "pid": os.getpid(),
+            "sessionId": "s1",
+            "cwd": workdir,
+            "startedAt": 1000,
+            "name": "test",
+        }
+        pane_output = f"live-tmux:{workdir}\n"
+        with (
+            patch.object(bot, "_discover_cc_session", return_value=session),
+            patch("subprocess.run", return_value=MagicMock(returncode=0, stdout=pane_output)),
+            patch.object(bot, "_resolve_cc_transcript_path", return_value=tmp_path / "t.jsonl"),
         ):
-            result = bot._find_tmux_for_presence(entry)
-        assert result == "found"
-
-
-# =============================================
-# 4. ensure_tmux_session — presence-first
-# =============================================
-
-
-class TestEnsureWithPresence:
-    """ensure_tmux_session follows presence pointer first."""
-
-    def test_attaches_via_presence_pointer(self, tmp_path, _patch_base_bot_deps):
-        """Attaches to live session found via presence pointer."""
-        bot = _make_bot(tmp_path, _patch_base_bot_deps)
-        _write_presence(
-            tmp_path,
-            {
-                "devpulse": {
-                    "pid": os.getpid(),
-                    "work_dir": str(tmp_path / "workdir"),
-                    "session_type": "interactive",
-                    "attach_handle": "live-session",
-                }
-            },
-        )
-        with patch("subprocess.run", return_value=MagicMock(returncode=0)):
             result = bot.ensure_tmux_session()
         assert result is True
-        assert bot.session_name == "live-session"
+        assert bot.session_name == "live-tmux"
         assert bot._using_shared_session is True
+        assert bot._active_session_id == "s1"
 
-    def test_presence_fallback_to_shared_session(self, tmp_path, _patch_base_bot_deps):
-        """When presence pointer is empty, falls back to shared_session config."""
+    def test_cc_no_tmux_pane_falls_through(self, tmp_path, _patch_base_bot_deps):
+        """CC session found but no tmux pane → falls through to Strategy 2."""
         workdir = tmp_path / "workdir"
         workdir.mkdir(exist_ok=True)
-        _write_presence(tmp_path, {})
         with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
             bot = BaseBot(
                 bot_id="fb_test",
@@ -338,101 +447,91 @@ class TestEnsureWithPresence:
                 shared_session="explicit-session",
             )
         bot.send_message = MagicMock()
-        with patch("subprocess.run", return_value=MagicMock(returncode=0)):
+        session = {"pid": os.getpid(), "sessionId": "s1", "cwd": str(workdir.resolve()), "startedAt": 1000}
+        with (
+            patch.object(bot, "_discover_cc_session", return_value=session),
+            patch.object(bot, "_find_tmux_pane_by_cwd", return_value=None),
+            patch("subprocess.run", return_value=MagicMock(returncode=0)),
+        ):
+            result = bot.ensure_tmux_session()
+        assert result is True
+        assert bot.session_name == "explicit-session"
+
+    def test_fallback_to_shared_session(self, tmp_path, _patch_base_bot_deps):
+        """When CC discovery returns None, falls back to shared_session config."""
+        workdir = tmp_path / "workdir"
+        workdir.mkdir(exist_ok=True)
+        with patch("apps.handlers.base_bot.PENDING_DIR", tmp_path):
+            bot = BaseBot(
+                bot_id="fb_test",
+                bot_token="t",
+                work_dir=workdir,
+                branch_name="devpulse",
+                shared_session="explicit-session",
+            )
+        bot.send_message = MagicMock()
+        with (
+            patch.object(bot, "_discover_cc_session", return_value=None),
+            patch("subprocess.run", return_value=MagicMock(returncode=0)),
+        ):
             result = bot.ensure_tmux_session()
         assert result is True
         assert bot.session_name == "explicit-session"
 
     def test_no_session_returns_false_never_spawns(self, tmp_path, _patch_base_bot_deps):
-        """When no presence and no shared session, returns False — never spawns."""
+        """When no CC session and no shared session, returns False — never spawns."""
         bot = _make_bot(tmp_path, _patch_base_bot_deps)
-        with patch("subprocess.run", return_value=MagicMock(returncode=1)):
+        with (
+            patch.object(bot, "_discover_cc_session", return_value=None),
+            patch("subprocess.run", return_value=MagicMock(returncode=1)),
+        ):
             result = bot.ensure_tmux_session()
         assert result is False
         assert bot._using_shared_session is False
 
-    def test_presence_rebinds_on_pointer_change(self, tmp_path, _patch_base_bot_deps):
-        """Bot re-reads presence on each call, rebinding to new session."""
+    def test_rebinds_on_session_change(self, tmp_path, _patch_base_bot_deps):
+        """Bot re-reads CC sessions on each call, rebinding to new session."""
         bot = _make_bot(tmp_path, _patch_base_bot_deps)
-        my_pid = os.getpid()
+        workdir = str(bot.work_dir.resolve())
+        pane_output = f"tmux-pane:{workdir}\n"
 
-        _write_presence(
-            tmp_path,
-            {
-                "devpulse": {
-                    "pid": my_pid,
-                    "work_dir": str(tmp_path / "workdir"),
-                    "attach_handle": "session-v1",
-                }
-            },
-        )
-        with patch("subprocess.run", return_value=MagicMock(returncode=0)):
+        session_v1 = {"pid": os.getpid(), "sessionId": "v1", "cwd": workdir, "startedAt": 100, "name": "v1"}
+        session_v2 = {"pid": os.getpid(), "sessionId": "v2", "cwd": workdir, "startedAt": 200, "name": "v2"}
+
+        with (
+            patch.object(bot, "_discover_cc_session", return_value=session_v1),
+            patch("subprocess.run", return_value=MagicMock(returncode=0, stdout=pane_output)),
+            patch.object(bot, "_resolve_cc_transcript_path", return_value=None),
+        ):
             bot.ensure_tmux_session()
-        assert bot.session_name == "session-v1"
+        assert bot._active_session_id == "v1"
 
-        _write_presence(
-            tmp_path,
-            {
-                "devpulse": {
-                    "pid": my_pid,
-                    "work_dir": str(tmp_path / "workdir"),
-                    "attach_handle": "session-v2",
-                }
-            },
-        )
-        with patch("subprocess.run", return_value=MagicMock(returncode=0)):
+        with (
+            patch.object(bot, "_discover_cc_session", return_value=session_v2),
+            patch("subprocess.run", return_value=MagicMock(returncode=0, stdout=pane_output)),
+            patch.object(bot, "_resolve_cc_transcript_path", return_value=None),
+        ):
             bot.ensure_tmux_session()
-        assert bot.session_name == "session-v2"
+        assert bot._active_session_id == "v2"
 
-    def test_stale_presence_ignored(self, tmp_path, _patch_base_bot_deps):
-        """Dead PID in presence pointer is treated as absent."""
+    def test_stores_transcript_path(self, tmp_path, _patch_base_bot_deps):
+        """CC discovery stores the resolved transcript path on the bot."""
         bot = _make_bot(tmp_path, _patch_base_bot_deps)
-        _write_presence(
-            tmp_path,
-            {
-                "devpulse": {
-                    "pid": 99999999,
-                    "work_dir": str(tmp_path / "workdir"),
-                    "attach_handle": "dead-session",
-                }
-            },
-        )
-        with patch("subprocess.run", return_value=MagicMock(returncode=1)):
-            result = bot.ensure_tmux_session()
-        assert result is False
-
-    def test_presence_with_tmux_scan_fallback(self, tmp_path, _patch_base_bot_deps):
-        """When attach_handle is empty, bot finds tmux session by CWD scan."""
-        bot = _make_bot(tmp_path, _patch_base_bot_deps)
-        workdir = str(bot.work_dir)
-        _write_presence(
-            tmp_path,
-            {
-                "devpulse": {
-                    "pid": os.getpid(),
-                    "work_dir": workdir,
-                    "attach_handle": "",
-                }
-            },
-        )
-
-        def _side_effect(cmd, **kwargs):
-            mock = MagicMock()
-            if "list-panes" in cmd:
-                mock.returncode = 0
-                mock.stdout = f"found-session:{workdir}\n"
-            else:
-                mock.returncode = 1
-            return mock
-
-        with patch("subprocess.run", side_effect=_side_effect):
-            result = bot.ensure_tmux_session()
-        assert result is True
-        assert bot.session_name == "found-session"
+        workdir = str(bot.work_dir.resolve())
+        session = {"pid": os.getpid(), "sessionId": "s1", "cwd": workdir, "startedAt": 1000, "name": "t"}
+        transcript = tmp_path / "transcript.jsonl"
+        pane_output = f"tmux:{workdir}\n"
+        with (
+            patch.object(bot, "_discover_cc_session", return_value=session),
+            patch("subprocess.run", return_value=MagicMock(returncode=0, stdout=pane_output)),
+            patch.object(bot, "_resolve_cc_transcript_path", return_value=transcript),
+        ):
+            bot.ensure_tmux_session()
+        assert bot._active_transcript_path == transcript
 
 
 # =============================================
-# 5. handle_message — no-session error
+# 8. handle_message — no-session error
 # =============================================
 
 
@@ -440,19 +539,25 @@ class TestHandleMessageNoSession:
     """Error message when no live session is available."""
 
     def test_shows_branch_name_in_error(self, tmp_path, _patch_base_bot_deps):
-        """Error message includes the branch name."""
+        """Includes the branch name in the no-session error message."""
         bot = _make_bot(tmp_path, _patch_base_bot_deps, branch_name="api")
-        with patch("subprocess.run", return_value=MagicMock(returncode=1)):
+        with (
+            patch.object(bot, "_discover_cc_session", return_value=None),
+            patch("subprocess.run", return_value=MagicMock(returncode=1)),
+        ):
             bot.handle_message(42, "hello", {"message_id": 1})
         msg = bot.send_message.call_args[0][1]
         assert "No live Claude session" in msg
         assert "api" in msg
 
     def test_shows_work_dir_name_when_no_branch_name(self, tmp_path, _patch_base_bot_deps):
-        """Falls back to work_dir name in error when branch_name is None."""
+        """Falls back to work_dir name in the error when branch_name is None."""
         bot = _make_bot(tmp_path, _patch_base_bot_deps)
         bot.branch_name = None
-        with patch("subprocess.run", return_value=MagicMock(returncode=1)):
+        with (
+            patch.object(bot, "_discover_cc_session", return_value=None),
+            patch("subprocess.run", return_value=MagicMock(returncode=1)),
+        ):
             bot.handle_message(42, "hello", {"message_id": 1})
         msg = bot.send_message.call_args[0][1]
         assert "No live Claude session" in msg

src/aipass/spawn/templates/builder/.spawn/.template_registry.json

@@ -384,7 +384,7 @@
   },
   "metadata": {
     "description": "Template file tracking registry for ID-based updates",
-    "last_updated": "2026-06-25",
+    "last_updated": "2026-07-01",
     "version": "1.0.0"
   }
 }