From d84e56344c88eff2497a0ea67c980285184ad80a Mon Sep 17 00:00:00 2001 From: AIOSAI Date: Sun, 26 Apr 2026 14:12:14 -0700 Subject: [PATCH 1/2] =?UTF-8?q?feat(system):=20fix:=20fork-aware=20error?= =?UTF-8?q?=20handling=20(#329)=20+=20hardcoded=20path=20fix=20(#8)=20?= =?UTF-8?q?=E2=80=94=20detect=20403=20push=20errors=20and=20print=20fork?= =?UTF-8?q?=20recovery=20steps,=20replace=20hardcoded=20AIPASS=5FROOT=20wi?= =?UTF-8?q?th=20env=20detection?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: @devpulse --- .claude/hooks/subagent_stop_gate.py | 34 +++++++++++++------ .../drone/apps/handlers/git/pr_handler.py | 24 ++++++++++++- .../apps/plugins/devpulse_ops/pr_plugin.py | 10 +++++- 3 files changed, 56 insertions(+), 12 deletions(-) diff --git a/.claude/hooks/subagent_stop_gate.py b/.claude/hooks/subagent_stop_gate.py index 2fd640df..4140f551 100644 --- a/.claude/hooks/subagent_stop_gate.py +++ b/.claude/hooks/subagent_stop_gate.py @@ -9,20 +9,33 @@ Version: 1.0.0 """ import json +import os import sys import subprocess from pathlib import Path -AIPASS_ROOT = Path.home() / "Projects" / "AIPass" + +def _find_repo_root() -> Path | None: + """Walk up from CWD or AIPASS_HOME to find the git repo root.""" + for start in (os.environ.get("AIPASS_HOME", ""), os.getcwd()): + p = Path(start) + while p != p.parent: + if (p / ".git").exists(): + return p + p = p.parent + return None + + +AIPASS_ROOT = _find_repo_root() def get_modified_py_files() -> list[str]: """Get Python files modified in the working tree (unstaged + staged).""" + if AIPASS_ROOT is None: + return [] try: result = subprocess.run( - ["git", "diff", "--name-only", "HEAD"], - capture_output=True, text=True, timeout=5, - cwd=str(AIPASS_ROOT) + ["git", "diff", "--name-only", "HEAD"], capture_output=True, text=True, timeout=5, cwd=str(AIPASS_ROOT) ) files = [] for line in result.stdout.strip().split("\n"): @@ -38,13 +51,17 @@ def get_modified_py_files() -> list[str]: def run_seedgo_checklist(file_path: str) -> list[str]: """Run seedgo checklist on a single file.""" + if AIPASS_ROOT is None: + return [] if "/.claude/" in file_path: return [] try: result = subprocess.run( ["drone", "@seedgo", "checklist", file_path], - capture_output=True, text=True, timeout=15, - cwd=str(AIPASS_ROOT) + capture_output=True, + text=True, + timeout=15, + cwd=str(AIPASS_ROOT), ) if result.returncode != 0: return [] @@ -86,10 +103,7 @@ def main(): lines.append(f" - {v}") lines.append("\nFix these violations before finishing.") - output = { - "decision": "block", - "reason": "\n".join(lines) - } + output = {"decision": "block", "reason": "\n".join(lines)} print(json.dumps(output)) except Exception: diff --git a/src/aipass/drone/apps/handlers/git/pr_handler.py b/src/aipass/drone/apps/handlers/git/pr_handler.py index f5b7d058..f9c486ef 100644 --- a/src/aipass/drone/apps/handlers/git/pr_handler.py +++ b/src/aipass/drone/apps/handlers/git/pr_handler.py @@ -29,6 +29,24 @@ from aipass.drone.apps.handlers.git.lock_handler import ( ) +def _is_permission_error(stderr: str) -> bool: + """Check if a push failure is a permission/auth error (fork scenario).""" + indicators = ["403", "permission", "denied", "could not read Username"] + lower = stderr.lower() + return any(ind.lower() in lower for ind in indicators) + + +_FORK_RECOVERY = """ +Push failed due to insufficient permissions. You may be working on a fork. + +To contribute from a fork: + 1. Create a fork: gh repo fork AIOSAI/AIPass --remote=false --clone=false + 2. Add fork as remote: git remote add fork + 3. Push to your fork: git push -u fork {branch} + 4. Open cross-repo PR: gh pr create -R AIOSAI/AIPass -H :{branch} -B main +""" + + def _resolve_git_branch(branch_name: str, branch_dir: Path) -> str: """Read git_branch from passport if available, else fall back to citizen/{name}.""" passport_path = branch_dir / ".trinity" / "passport.json" @@ -173,7 +191,11 @@ def create_pr(branch_name: str, description: str, branch_dir: Path) -> dict: cwd=str(repo_root), ) if push.returncode != 0: - result["message"] = f"Push failed: {push.stderr.strip()}" + stderr = push.stderr.strip() + if _is_permission_error(stderr): + result["message"] = _FORK_RECOVERY.format(branch=feature_branch) + else: + result["message"] = f"Push failed: {stderr}" logger.error(result["message"]) return result diff --git a/src/aipass/drone/apps/plugins/devpulse_ops/pr_plugin.py b/src/aipass/drone/apps/plugins/devpulse_ops/pr_plugin.py index 2a590c66..85e87d70 100644 --- a/src/aipass/drone/apps/plugins/devpulse_ops/pr_plugin.py +++ b/src/aipass/drone/apps/plugins/devpulse_ops/pr_plugin.py @@ -32,6 +32,10 @@ from aipass.drone.apps.handlers.git.lock_handler import ( find_repo_root, release_lock, ) +from aipass.drone.apps.handlers.git.pr_handler import ( + _is_permission_error, + _FORK_RECOVERY, +) def slugify(description: str) -> str: @@ -212,7 +216,11 @@ def create_system_pr(description: str, caller: str) -> dict: cwd=str(repo_root), ) if push.returncode != 0: - result["message"] = f"Push failed: {push.stderr.strip()}" + stderr = push.stderr.strip() + if _is_permission_error(stderr): + result["message"] = _FORK_RECOVERY.format(branch=feature_branch) + else: + result["message"] = f"Push failed: {stderr}" logger.error(result["message"]) return result From b6be4a9c687bd49b4e951fbf73b49368764b257a Mon Sep 17 00:00:00 2001 From: AIOSAI Date: Sun, 26 Apr 2026 17:08:22 -0700 Subject: [PATCH 2/2] =?UTF-8?q?feat(system):=20security:=20DPLAN-0155=20Ph?= =?UTF-8?q?ases=202-4=20=E2=80=94=20prompt=20injection=20defense,=20daemon?= =?UTF-8?q?=20path=20confinement,=20JSONL=20writes=20removed?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: @devpulse --- .claude/hooks/pre_edit_gate.py | 40 +++-- .../ai_mail/apps/handlers/dispatch/daemon.py | 35 +--- .../ai_mail/apps/handlers/dispatch/wake.py | 27 --- src/aipass/ai_mail/tests/test_daemon.py | 155 ------------------ src/aipass/ai_mail/tests/test_wake.py | 79 --------- 5 files changed, 33 insertions(+), 303 deletions(-) diff --git a/.claude/hooks/pre_edit_gate.py b/.claude/hooks/pre_edit_gate.py index df49b5a2..93131beb 100644 --- a/.claude/hooks/pre_edit_gate.py +++ b/.claude/hooks/pre_edit_gate.py @@ -59,16 +59,39 @@ def main(): # ------------------------------------------------------------------ fp = Path(file_path) if fp.name == "inbox.json" and ".ai_mail.local" in fp.parts: - _block( - "Direct writes to inbox.json are blocked.\n" - "Use: drone @ai_mail email @ \"Subject\" \"Body\"" - ) + _block('Direct writes to inbox.json are blocked.\nUse: drone @ai_mail email @ "Subject" "Body"') + + # ------------------------------------------------------------------ + # Rule 1.5: Dispatched-agent path confinement (DPLAN-0155 M3) + # Daemon-spawned agents can only write inside their own branch dir. + # Breaks the prompt-injection amplifier chain — even if injected, + # a dispatched agent cannot write to other agents' inboxes or code. + # ------------------------------------------------------------------ + cwd = input_data.get("cwd", "") or os.getcwd() + cwd_branch = _get_branch(cwd) + + session_type = os.environ.get("AIPASS_SESSION_TYPE", "interactive") + if session_type == "daemon" and cwd_branch: + target_branch = _get_branch(str(fp.resolve()) if not fp.is_absolute() else str(fp)) + if target_branch and target_branch != cwd_branch: + _block( + f"Dispatched agent confined to own branch: '{cwd_branch}' " + f"cannot write to '{target_branch}' in daemon mode." + ) + repo_root = None + for parent in Path(cwd).parents: + if (parent / ".git").exists(): + repo_root = parent + break + if repo_root and not target_branch: + allowed_prefix = str(repo_root / "src" / "aipass" / cwd_branch) + resolved = str(fp.resolve()) if not fp.is_absolute() else str(fp) + if not resolved.startswith(allowed_prefix): + _block(f"Dispatched agent restricted to {allowed_prefix}. Cannot write to: {file_path}") # ------------------------------------------------------------------ # Rule 2: Cross-branch write enforcement # ------------------------------------------------------------------ - cwd = input_data.get("cwd", "") or os.getcwd() - cwd_branch = _get_branch(cwd) target_branch = _get_branch(str(fp.resolve()) if not fp.is_absolute() else str(fp)) if cwd_branch and target_branch and cwd_branch != target_branch: @@ -115,10 +138,7 @@ def main(): return error_summary = "\n".join(f" L{e['line']}: {e['message']}" for e in errors[:5]) - _block( - f"Fix {len(errors)} error(s) in {Path(errored_file).name} before editing other files:\n" - f"{error_summary}" - ) + _block(f"Fix {len(errors)} error(s) in {Path(errored_file).name} before editing other files:\n{error_summary}") except Exception: pass # Silent fail → allow diff --git a/src/aipass/ai_mail/apps/handlers/dispatch/daemon.py b/src/aipass/ai_mail/apps/handlers/dispatch/daemon.py index d374fc14..70cf9b97 100644 --- a/src/aipass/ai_mail/apps/handlers/dispatch/daemon.py +++ b/src/aipass/ai_mail/apps/handlers/dispatch/daemon.py @@ -117,31 +117,6 @@ def _write_json(filepath: Path, data: Dict[str, Any]) -> bool: return False -def _set_session_name(branch_path: Path, name: str) -> bool: - """Write custom-title to the most recent Claude session JSONL for a branch. - - Claude stores sessions at ~/.claude/projects/{encoded-cwd}/*.jsonl. - Writing a custom-title entry makes the session identifiable in /resume picker. - """ - encoded_cwd = str(branch_path).replace("/", "-") - projects_dir = Path("~/.claude/projects").expanduser() / encoded_cwd - if not projects_dir.exists(): - return False - jsonl_files = sorted(projects_dir.glob("*.jsonl"), key=lambda f: f.stat().st_mtime, reverse=True) - if not jsonl_files: - return False - latest = jsonl_files[0] - session_id = latest.stem - entry = json.dumps({"type": "custom-title", "customTitle": name, "sessionId": session_id}) - try: - with open(latest, "a", encoding="utf-8") as f: - f.write(entry + "\n") - return True - except OSError as e: - logger.warning("[daemon] Failed to write session name for %s: %s", branch_path, e) - return False - - def _check_lock(branch_path: Path) -> Optional[Dict[str, Any]]: """Check if branch has an active dispatch lock. Returns lock data or None.""" lock_file = branch_path / ".ai_mail.local" / ".dispatch.lock" @@ -362,14 +337,14 @@ def spawn_agent( True if monitor was spawned successfully """ sender = message.get("from", "unknown") - msg_id = message.get("id", "unknown") subject = message.get("subject", "") max_turns = config.get("max_turns_per_wake", 100) lock_file_path = str(branch_path / ".ai_mail.local" / ".dispatch.lock") - # Prompt — no lock cleanup instruction (dispatch_monitor handles it) - prompt = f"Hi. Check inbox for task from {sender} (message ID: {msg_id}). Execute it. Send confirmation when done." + # Prompt — never interpolate sender-controlled content into the prompt. + # The agent reads inbox.json as data, not as instructions (DPLAN-0155 M1). + prompt = "Hi. Check your inbox for new dispatch emails and execute the task. Send confirmation when done." claude_cmd = [ "claude", @@ -409,10 +384,6 @@ def spawn_agent( if key.startswith("CLAUDE") or key == "AIPASS_BOT_ID": spawn_env.pop(key) - # Set session name for /resume picker (daemon always uses -c resume) - spawn_branch_name = branch_email.lstrip("@").upper() - _set_session_name(branch_path, f"{spawn_branch_name}-daemon") - try: process = subprocess.Popen( monitor_cmd, diff --git a/src/aipass/ai_mail/apps/handlers/dispatch/wake.py b/src/aipass/ai_mail/apps/handlers/dispatch/wake.py index 1883b4b6..6536bfbe 100644 --- a/src/aipass/ai_mail/apps/handlers/dispatch/wake.py +++ b/src/aipass/ai_mail/apps/handlers/dispatch/wake.py @@ -210,27 +210,6 @@ def _load_config() -> dict: return config -def _set_session_name(branch_path: Path, name: str) -> bool: - """Write custom-title to the most recent Claude session JSONL for a branch.""" - encoded_cwd = str(branch_path).replace("/", "-") - projects_dir = Path("~/.claude/projects").expanduser() / encoded_cwd - if not projects_dir.exists(): - return False - jsonl_files = sorted(projects_dir.glob("*.jsonl"), key=lambda f: f.stat().st_mtime, reverse=True) - if not jsonl_files: - return False - latest = jsonl_files[0] - session_id = latest.stem - entry = json.dumps({"type": "custom-title", "customTitle": name, "sessionId": session_id}) - try: - with open(latest, "a", encoding="utf-8") as f: - f.write(entry + "\n") - return True - except OSError as e: - logger.warning("[wake] Failed to write session name for %s: %s", branch_path, e) - return False - - def _read_session_type(pid_str: str) -> str: """Read AIPASS_SESSION_TYPE from /proc/{pid}/environ. Returns 'interactive' if unset.""" if sys.platform != "linux": @@ -476,12 +455,6 @@ def wake_branch( "json", ] - # Set session name for /resume picker - branch_name = email.lstrip("@").upper() - session_label = f"{branch_name}-dispatched" - if not fresh: - _set_session_name(branch_path, session_label) - # Step 7: Spawn via dispatch_monitor log_dir = branch_path / "logs" log_dir.mkdir(parents=True, exist_ok=True) diff --git a/src/aipass/ai_mail/tests/test_daemon.py b/src/aipass/ai_mail/tests/test_daemon.py index e21f2412..eaea9f6e 100644 --- a/src/aipass/ai_mail/tests/test_daemon.py +++ b/src/aipass/ai_mail/tests/test_daemon.py @@ -767,13 +767,11 @@ def test_poll_cycle_absolute_path_unchanged(tmp_path, monkeypatch): import os import sys -from pathlib import Path from unittest.mock import MagicMock, mock_open from aipass.ai_mail.apps.handlers.dispatch.daemon import ( _notify_telegram, _handle_signal, - _set_session_name, _check_lock, _acquire_lock, poll_cycle, @@ -872,106 +870,6 @@ def test_handle_signal_sets_shutdown(monkeypatch): assert daemon_mod.SHUTDOWN is True -# ---- _set_session_name tests ------------------------------------ - - -def test_set_session_name_success(tmp_path, monkeypatch): - """Writes custom-title entry to most recent JSONL file.""" - branch_path = tmp_path / "branch" - branch_path.mkdir() - # Redirect ~/.claude/projects to tmp_path so no real filesystem side effects - fake_home = tmp_path / "fakehome" - encoded_cwd = str(branch_path).replace("/", "-") - projects_dir = fake_home / ".claude" / "projects" / encoded_cwd - projects_dir.mkdir(parents=True) - jsonl_file = projects_dir / "session123.jsonl" - jsonl_file.write_text('{"type":"init"}\n', encoding="utf-8") - - _orig_expanduser = Path.expanduser - - def _fake_expanduser(self): - if str(self).startswith("~"): - return fake_home / str(self)[2:] - return _orig_expanduser(self) - - monkeypatch.setattr(Path, "expanduser", _fake_expanduser) - - result = _set_session_name(branch_path, "TEST-daemon") - - assert result is True - content = jsonl_file.read_text(encoding="utf-8") - assert "custom-title" in content - assert "TEST-daemon" in content - - -def test_set_session_name_no_projects_dir(tmp_path, monkeypatch): - """Returns False when projects dir does not exist.""" - branch_path = tmp_path / "nonexistent_branch_xyz_test" - fake_home = tmp_path / "fakehome" - - _orig_expanduser = Path.expanduser - - def _fake_expanduser(self): - if str(self).startswith("~"): - return fake_home / str(self)[2:] - return _orig_expanduser(self) - - monkeypatch.setattr(Path, "expanduser", _fake_expanduser) - - result = _set_session_name(branch_path, "TEST-daemon") - - assert result is False - - -def test_set_session_name_no_jsonl_files(tmp_path, monkeypatch): - """Returns False when projects dir exists but has no JSONL files.""" - branch_path = tmp_path / "branch" - branch_path.mkdir() - fake_home = tmp_path / "fakehome" - encoded_cwd = str(branch_path).replace("/", "-") - projects_dir = fake_home / ".claude" / "projects" / encoded_cwd - projects_dir.mkdir(parents=True) - - _orig_expanduser = Path.expanduser - - def _fake_expanduser(self): - if str(self).startswith("~"): - return fake_home / str(self)[2:] - return _orig_expanduser(self) - - monkeypatch.setattr(Path, "expanduser", _fake_expanduser) - - result = _set_session_name(branch_path, "TEST-daemon") - - assert result is False - - -def test_set_session_name_oserror_on_write(tmp_path, monkeypatch): - """Returns False on OSError when writing to JSONL file.""" - branch_path = tmp_path / "branch" - branch_path.mkdir() - fake_home = tmp_path / "fakehome" - encoded_cwd = str(branch_path).replace("/", "-") - projects_dir = fake_home / ".claude" / "projects" / encoded_cwd - projects_dir.mkdir(parents=True) - jsonl_file = projects_dir / "session456.jsonl" - jsonl_file.write_text('{"type":"init"}\n', encoding="utf-8") - - _orig_expanduser = Path.expanduser - - def _fake_expanduser(self): - if str(self).startswith("~"): - return fake_home / str(self)[2:] - return _orig_expanduser(self) - - monkeypatch.setattr(Path, "expanduser", _fake_expanduser) - - with patch("builtins.open", side_effect=OSError("disk full")): - result = _set_session_name(branch_path, "TEST-daemon") - - assert result is False - - # ---- _check_lock tests ----------------------------------------- @@ -1430,10 +1328,6 @@ def test_spawn_agent_success(tmp_path): "aipass.ai_mail.apps.handlers.dispatch.daemon._acquire_lock", return_value=(True, "Lock acquired"), ), - patch( - "aipass.ai_mail.apps.handlers.dispatch.daemon._set_session_name", - return_value=True, - ), patch("aipass.ai_mail.apps.handlers.dispatch.daemon.log_dispatch"), patch( "aipass.ai_mail.apps.handlers.dispatch.daemon._notify_telegram", @@ -1466,10 +1360,6 @@ def test_spawn_agent_exception(tmp_path): "aipass.ai_mail.apps.handlers.dispatch.daemon.subprocess.Popen", side_effect=OSError("command not found"), ), - patch( - "aipass.ai_mail.apps.handlers.dispatch.daemon._set_session_name", - return_value=True, - ), patch("aipass.ai_mail.apps.handlers.dispatch.daemon.log_dispatch"), patch( "aipass.ai_mail.apps.handlers.dispatch.daemon._notify_telegram", @@ -1512,10 +1402,6 @@ def test_spawn_agent_strips_claude_env_vars(tmp_path, monkeypatch): "aipass.ai_mail.apps.handlers.dispatch.daemon._acquire_lock", return_value=(True, "Lock acquired"), ), - patch( - "aipass.ai_mail.apps.handlers.dispatch.daemon._set_session_name", - return_value=True, - ), patch("aipass.ai_mail.apps.handlers.dispatch.daemon.log_dispatch"), patch( "aipass.ai_mail.apps.handlers.dispatch.daemon._notify_telegram", @@ -1536,47 +1422,6 @@ def test_spawn_agent_strips_claude_env_vars(tmp_path, monkeypatch): assert captured_env.get("AIPASS_SESSION_TYPE") == "daemon" -def test_spawn_agent_sets_session_name(tmp_path): - """Spawn calls _set_session_name with correct branch name.""" - branch_path = tmp_path / "branch" - branch_path.mkdir() - (branch_path / "logs").mkdir() - - message = {"from": "@devpulse", "id": "msg1", "subject": "Test task"} - config = {"max_turns_per_wake": 50} - state = {"daily_counts": {}, "session_cycles": {}} - - mock_process = MagicMock() - mock_process.pid = 54321 - - with ( - patch( - "aipass.ai_mail.apps.handlers.dispatch.daemon.subprocess.Popen", - return_value=mock_process, - ), - patch( - "aipass.ai_mail.apps.handlers.dispatch.daemon._acquire_lock", - return_value=(True, "Lock acquired"), - ), - patch( - "aipass.ai_mail.apps.handlers.dispatch.daemon._set_session_name", - return_value=True, - ) as mock_ssn, - patch("aipass.ai_mail.apps.handlers.dispatch.daemon.log_dispatch"), - patch( - "aipass.ai_mail.apps.handlers.dispatch.daemon._notify_telegram", - return_value=True, - ), - patch( - "aipass.ai_mail.apps.handlers.dispatch.daemon.send_notification", - create=True, - ), - ): - spawn_agent(branch_path, "@testbranch", message, config, state) - - mock_ssn.assert_called_once_with(branch_path, "TESTBRANCH-daemon") - - # ---- run_daemon tests ------------------------------------------- diff --git a/src/aipass/ai_mail/tests/test_wake.py b/src/aipass/ai_mail/tests/test_wake.py index 5ad6a3ac..5cdc38b3 100644 --- a/src/aipass/ai_mail/tests/test_wake.py +++ b/src/aipass/ai_mail/tests/test_wake.py @@ -29,7 +29,6 @@ from aipass.ai_mail.apps.handlers.dispatch.wake import ( DEFAULT_MODEL, _acquire_lock, _load_config, - _set_session_name, _is_branch_occupied, wake_branch, ) @@ -710,84 +709,6 @@ class TestLoadConfig: assert result["max_turns_per_wake"] == 50 -# --- _set_session_name tests -------------------------------------------- - - -class TestSetSessionName: - """Tests for _set_session_name() — writes custom-title to Claude session JSONL.""" - - def test_success_appends_entry(self, tmp_path, monkeypatch): - """Creates expected JSON entry in the most recent session JSONL.""" - encoded_cwd = str(tmp_path).replace("/", "-") - projects_dir = tmp_path / ".claude" / "projects" / encoded_cwd - projects_dir.mkdir(parents=True) - session_file = projects_dir / "abc123.jsonl" - session_file.write_text("", encoding="utf-8") - monkeypatch.setattr( - "aipass.ai_mail.apps.handlers.dispatch.wake.Path.expanduser", - lambda self: tmp_path / ".claude" / "projects" if str(self).endswith("projects") else self, - ) - # We need to mock expanduser properly — override the whole projects_dir lookup - monkeypatch.setattr( - _Path, - "expanduser", - lambda self: tmp_path / str(self).lstrip("~/"), - ) - result = _set_session_name(tmp_path, "TEST-dispatched") - assert result is True - content = session_file.read_text(encoding="utf-8") - entry = json.loads(content.strip()) - assert entry["type"] == "custom-title" - assert entry["customTitle"] == "TEST-dispatched" - assert entry["sessionId"] == "abc123" - - def test_no_projects_dir_returns_false(self, tmp_path, monkeypatch): - """Returns False when ~/.claude/projects/{encoded} does not exist.""" - monkeypatch.setattr( - _Path, - "expanduser", - lambda self: tmp_path / str(self).lstrip("~/"), - ) - result = _set_session_name(tmp_path, "TEST-dispatched") - assert result is False - - def test_no_jsonl_files_returns_false(self, tmp_path, monkeypatch): - """Returns False when projects dir exists but has no .jsonl files.""" - encoded_cwd = str(tmp_path).replace("/", "-") - projects_dir = tmp_path / ".claude" / "projects" / encoded_cwd - projects_dir.mkdir(parents=True) - monkeypatch.setattr( - _Path, - "expanduser", - lambda self: tmp_path / str(self).lstrip("~/"), - ) - result = _set_session_name(tmp_path, "TEST-dispatched") - assert result is False - - def test_os_error_on_write_returns_false(self, tmp_path, monkeypatch): - """Returns False when write to JSONL file raises OSError.""" - encoded_cwd = str(tmp_path).replace("/", "-") - projects_dir = tmp_path / ".claude" / "projects" / encoded_cwd - projects_dir.mkdir(parents=True) - session_file = projects_dir / "abc123.jsonl" - session_file.write_text("", encoding="utf-8") - monkeypatch.setattr( - _Path, - "expanduser", - lambda self: tmp_path / str(self).lstrip("~/"), - ) - - def _fail_open(path, *args, **kwargs): - path_str = str(path) - if path_str.endswith(".jsonl") and "a" in args: - raise OSError("permission denied") - return _REAL_OPEN(path, *args, **kwargs) - - monkeypatch.setattr("builtins.open", _fail_open) - result = _set_session_name(tmp_path, "TEST-dispatched") - assert result is False - - # --- _is_branch_occupied tests ------------------------------------------