diff --git a/CHANGELOG.md b/CHANGELOG.md index ddf4ed81..93e93557 100644 --- a/CHANGELOG.md +++ b/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/.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/.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 diff --git a/src/aipass/hooks/.seedgo/bypass.json b/src/aipass/hooks/.seedgo/bypass.json index 56607918..d9e3868e 100644 --- a/src/aipass/hooks/.seedgo/bypass.json +++ b/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": { diff --git a/src/aipass/hooks/apps/handlers/lifecycle/session_boot.py b/src/aipass/hooks/apps/handlers/lifecycle/session_boot.py new file mode 100644 index 00000000..68b583a5 --- /dev/null +++ b/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() diff --git a/src/aipass/hooks/apps/handlers/security/presence_gate.py b/src/aipass/hooks/apps/handlers/security/presence_gate.py index d8614724..36d5a688 100644 --- a/src/aipass/hooks/apps/handlers/security/presence_gate.py +++ b/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/.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/.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 diff --git a/src/aipass/hooks/apps/modules/cc_sessions.py b/src/aipass/hooks/apps/modules/cc_sessions.py new file mode 100644 index 00000000..f8bd872b --- /dev/null +++ b/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/.json). + +CC maintains one JSON file per running session at ~/.claude/sessions/.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 diff --git a/src/aipass/hooks/tests/test_cc_sessions.py b/src/aipass/hooks/tests/test_cc_sessions.py new file mode 100644 index 00000000..2911c687 --- /dev/null +++ b/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 diff --git a/src/aipass/hooks/tests/test_presence_gate.py b/src/aipass/hooks/tests/test_presence_gate.py index e7bed511..a0cf924f 100644 --- a/src/aipass/hooks/tests/test_presence_gate.py +++ b/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() diff --git a/src/aipass/hooks/tests/test_session_boot.py b/src/aipass/hooks/tests/test_session_boot.py new file mode 100644 index 00000000..29a9f168 --- /dev/null +++ b/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) diff --git a/src/aipass/skills/.seedgo/bypass.json b/src/aipass/skills/.seedgo/bypass.json index 33c6c9b4..43527a37 100644 --- a/src/aipass/skills/.seedgo/bypass.json +++ b/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", diff --git a/src/aipass/skills/lib/telegram/apps/handlers/base_bot.py b/src/aipass/skills/lib/telegram/apps/handlers/base_bot.py index 30a07a5b..80d340fa 100644 --- a/src/aipass/skills/lib/telegram/apps/handlers/base_bot.py +++ b/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/.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). diff --git a/src/aipass/skills/lib/telegram/tests/test_presence_pointer.py b/src/aipass/skills/lib/telegram/tests/test_presence_pointer.py index 82d308ac..9ff97bbd 100644 --- a/src/aipass/skills/lib/telegram/tests/test_presence_pointer.py +++ b/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/.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 diff --git a/src/aipass/spawn/templates/builder/.spawn/.template_registry.json b/src/aipass/spawn/templates/builder/.spawn/.template_registry.json index b7943034..01125c39 100644 --- a/src/aipass/spawn/templates/builder/.spawn/.template_registry.json +++ b/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" } }