diff --git a/libs/code/deepagents_code/app.py b/libs/code/deepagents_code/app.py index e57fa015839..f29cb210ee5 100644 --- a/libs/code/deepagents_code/app.py +++ b/libs/code/deepagents_code/app.py @@ -3043,12 +3043,7 @@ def __init__( self._register_custom_themes() self._hook_trust: WorkspaceTrust | None = hook_trust - """Project-hook trust policy, forwarded verbatim to `HooksManager`. - - Carried rather than applied: session state — and therefore the hooks - coordinator that owns and resolves this policy — cannot be built until - after construction. The app never inspects it. - """ + """Project-hook trust shared across pending and live manager state.""" self.theme = _load_theme_preference() """Active Textual theme name. @@ -4723,6 +4718,92 @@ async def _reload_hooks( await self._hooks.reload(cwd=Path(self._cwd), plugins=plugins) + async def _retarget_hooks_after_cwd_switch( + self, + *, + reload_manager: bool = True, + ) -> None: + """Resolve project-hook trust after a successful cwd switch. + + Args: + reload_manager: Whether to activate the target workspace's runtime + immediately. In-session resumes defer activation until the outgoing + runtime has received `SessionEnd`. + """ + from deepagents_code._env_vars import EXPERIMENTAL, is_env_truthy + + if not is_env_truthy(EXPERIMENTAL): + return + + from deepagents_code.hooks.loading import project_hooks_path + from deepagents_code.hooks.trust import project_root_for, trust_project_hooks + from deepagents_code.tui.widgets.cwd_switch import HookTrustScreen + + state = self._session_state + trust = ( + state.hooks.trust + if state is not None + else self._hook_trust or self._hooks.trust + ) + try: + root = await asyncio.to_thread(project_root_for, Path(self._cwd)) + config_path = project_hooks_path(root) + has_project_hooks = await asyncio.to_thread(config_path.is_file) + except (OSError, ValueError): + logger.warning( + "Could not inspect project hooks after cwd switch", + exc_info=True, + ) + else: + if has_project_hooks and not await asyncio.to_thread(trust.allows, root): + trust = await asyncio.to_thread(trust.without_session_grant, root) + if trust.consult_store: + prompt_grant = await asyncio.to_thread( + trust.with_session_grant, root + ) + if self.is_running: + refreshed = asyncio.Event() + if self.call_after_refresh(refreshed.set): + await refreshed.wait() + try: + choice = await self._push_screen_wait( + HookTrustScreen( + project_root=str(root), + config_path=str(config_path), + ) + ) + except Exception: + logger.warning( + "Project hooks trust prompt failed after cwd switch", + exc_info=True, + ) + self.notify( + "Project hooks were not loaded because the trust prompt " + "failed.", + severity="warning", + markup=False, + ) + choice = None + + if choice in {"allow_once", "always_allow"}: + trust = prompt_grant + if choice == "always_allow" and not await asyncio.to_thread( + trust_project_hooks, + root, + store_path=trust.store_path, + ): + self.notify( + "Approved for this session, but the decision could not " + "be saved — you may be asked again next time.", + severity="warning", + markup=False, + ) + + self._hook_trust = trust + self._hooks.trust = trust + if reload_manager: + await self._reload_hooks() + async def _run_session_start_hook(self, cause: SessionStartCause) -> bool: """Run `SessionStart`, surfacing a stop as a chat message. @@ -24508,9 +24589,15 @@ async def _offer_thread_cwd_switch( "on the current thread.", ) ) + return outcome + await self._retarget_hooks_after_cwd_switch(reload_manager=False) return outcome self._preserve_launch_relative_server_paths(Path(self._cwd)) self._switch_process_cwd(target) + # Cross-agent resumes reload after the outgoing `SessionEnd`. + await self._retarget_hooks_after_cwd_switch( + reload_manager=abort != "thread_switch" + ) return "continue" self.notify( @@ -24559,6 +24646,8 @@ async def _restore_cwd_after_failed_thread_switch(self, previous_cwd: Path) -> N "rollback", previous_cwd, ) + return + await self._reload_hooks() return try: @@ -24577,6 +24666,8 @@ async def _restore_cwd_after_failed_thread_switch(self, previous_cwd: Path) -> N timeout=15, markup=False, ) + return + await self._reload_hooks() async def _resume_thread(self, thread_id: str) -> None: """Resume a previously saved thread. diff --git a/libs/code/deepagents_code/hooks/loading.py b/libs/code/deepagents_code/hooks/loading.py index fd8206dad63..0e3c0b7e130 100644 --- a/libs/code/deepagents_code/hooks/loading.py +++ b/libs/code/deepagents_code/hooks/loading.py @@ -136,6 +136,9 @@ class LoadedHooksConfig: canonical deduplication (symlinks / shared config dirs can alias paths). """ + project_source_fingerprint: str | None = None + """SHA-256 fingerprint of the exact project source bytes that were loaded.""" + def project_hooks_path(project_root: Path) -> Path: """Return the project-scoped hooks configuration path. @@ -193,20 +196,22 @@ def load_hooks_config( merged: dict[HookEvent, list[SourcedGroup]] = {} loaded_paths: list[Path] = [] project_source_loaded = False + project_source_fingerprint: str | None = None def _merge(document: HooksConfig, source: HooksSource) -> None: for event, groups in document.hooks.items(): merged.setdefault(event, []).extend((source, group) for group in groups) def _ingest(path: Path, *, as_project: bool) -> None: - nonlocal project_source_loaded + nonlocal project_source_fingerprint, project_source_loaded resolved = path.expanduser().resolve(strict=False) - document, file_diagnostics = _read_hooks_document(resolved) + document, file_diagnostics, fingerprint = _read_hooks_document(resolved) diagnostics.extend(file_diagnostics) if document is None: return if as_project: project_source_loaded = True + project_source_fingerprint = fingerprint loaded_paths.append(resolved) _merge(document, FileHooksSource(location=str(resolved))) @@ -250,6 +255,7 @@ def _ingest(path: Path, *, as_project: bool) -> None: snapshot_id=compute_snapshot_id(config, groups=groups), groups=groups, project_source_loaded=project_source_loaded, + project_source_fingerprint=project_source_fingerprint, ) @@ -334,25 +340,21 @@ def _canonical_group(group: MatcherGroup, *, source: HooksSource) -> dict[str, o def read_hooks_json( path: Path, -) -> tuple[bool, JsonValue, tuple[HookDiagnostic, ...]]: - """Decode one hooks JSON document, reporting read failures as diagnostics. - - Shared by the project/user file loader and by callers that must transform a - document before it is validated, so every hooks document on disk reports - unreadable bytes, invalid UTF-8, and malformed JSON the same way instead of - raising into an unrelated caller. +) -> tuple[bool, JsonValue, tuple[HookDiagnostic, ...], str | None]: + """Decode one hooks document and fingerprint the exact bytes read. Args: path: Document path. Returns: - Whether decoding succeeded, the decoded JSON value, and any diagnostics. - An absent file is not a diagnostic. + Whether decoding succeeded, the decoded document, diagnostics, and the + exact-byte SHA-256 fingerprint. An absent file is not a diagnostic. """ if not path.is_file(): - return False, None, () + return False, None, (), None try: - decoded: JsonValue = json.loads(path.read_text(encoding="utf-8")) + content = path.read_bytes() + decoded: JsonValue = json.loads(content.decode("utf-8")) except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: message = f"Failed to read hooks config at {path}: {exc}" logger.warning(message) @@ -367,21 +369,22 @@ def read_hooks_json( field=str(path), ), ), + None, ) - return True, decoded, () + return True, decoded, (), hashlib.sha256(content).hexdigest() def _read_hooks_document( path: Path, -) -> tuple[HooksConfig | None, tuple[HookDiagnostic, ...]]: - decoded, data, read_diagnostics = read_hooks_json(path) +) -> tuple[HooksConfig | None, tuple[HookDiagnostic, ...], str | None]: + decoded, data, read_diagnostics, fingerprint = read_hooks_json(path) if not decoded: - return None, read_diagnostics + return None, read_diagnostics, None if is_legacy_hooks_document(data): hooks = data.get("hooks", []) if isinstance(data, dict) else [] if not isinstance(hooks, list): - return None, ( + diagnostics = ( HookDiagnostic( code="invalid_config", severity="warning", @@ -389,6 +392,7 @@ def _read_hooks_document( field=str(path), ), ) + return None, diagnostics, fingerprint legacy_entries: list[dict[str, object]] = [ {str(key): value for key, value in item.items()} for item in hooks @@ -404,7 +408,7 @@ def _read_hooks_document( "migrate to Hooks v2" ) ) - return migrated, ( + diagnostics = ( HookDiagnostic( code="legacy_deprecated", severity="warning", @@ -421,8 +425,10 @@ def _read_hooks_document( field=str(path), ), ) + return migrated, diagnostics, fingerprint - return _validate_hooks_document(data, path) + document, diagnostics = _validate_hooks_document(data, path) + return document, diagnostics, fingerprint def _validate_hooks_document( diff --git a/libs/code/deepagents_code/hooks/manager.py b/libs/code/deepagents_code/hooks/manager.py index c7837cfd22e..559f7b7e562 100644 --- a/libs/code/deepagents_code/hooks/manager.py +++ b/libs/code/deepagents_code/hooks/manager.py @@ -630,13 +630,31 @@ def _load_runtime( project_dir=project_context.project_root or project_context.user_cwd, plugins=plugins, ) - return HooksRuntime.create( + runtime = HooksRuntime.create( cwd=cwd, workspace_trusted=trust.allows(cwd), presenter=presenter, plugin_sources=plugin_sources, plugin_diagnostics=plugin_diagnostics, ) + if runtime.project_hooks_loaded and ( + runtime.project_hooks_fingerprint is None + or not trust.allows( + cwd, + project_hooks_fingerprint=runtime.project_hooks_fingerprint, + ) + ): + logger.warning( + "Project hooks changed while loading; reloading without project hooks" + ) + runtime = HooksRuntime.create( + cwd=cwd, + workspace_trusted=False, + presenter=presenter, + plugin_sources=plugin_sources, + plugin_diagnostics=plugin_diagnostics, + ) except Exception: logger.exception("Failed to load hook configuration; hooks disabled") return None + return runtime diff --git a/libs/code/deepagents_code/hooks/runtime.py b/libs/code/deepagents_code/hooks/runtime.py index d5c96d69d36..70aa506fe76 100644 --- a/libs/code/deepagents_code/hooks/runtime.py +++ b/libs/code/deepagents_code/hooks/runtime.py @@ -67,6 +67,9 @@ class HooksRuntime: """ project_hooks_loaded: bool + project_hooks_fingerprint: str | None + """SHA-256 fingerprint of the exact project-hook bytes in the snapshot.""" + presenter: HookPresenter fulfillments: HookFulfillmentLedger @@ -133,6 +136,7 @@ def create( cwd=project_context.user_cwd, workspace_trusted=workspace_trusted, project_hooks_loaded=loaded.project_source_loaded, + project_hooks_fingerprint=loaded.project_source_fingerprint, presenter=presenter if presenter is not None else HookPresenter(), fulfillments=HookFulfillmentLedger(), ) diff --git a/libs/code/deepagents_code/hooks/trust.py b/libs/code/deepagents_code/hooks/trust.py index 7b2f7a45ead..4b8379ee1a3 100644 --- a/libs/code/deepagents_code/hooks/trust.py +++ b/libs/code/deepagents_code/hooks/trust.py @@ -2,13 +2,14 @@ from __future__ import annotations +import hashlib import json import logging import os import tempfile import threading from contextlib import contextmanager, suppress -from dataclasses import dataclass +from dataclasses import dataclass, replace from datetime import UTC, datetime from pathlib import Path from typing import TYPE_CHECKING, Literal @@ -337,6 +338,21 @@ def project_root_for(cwd: Path | str) -> Path: return context.project_root or context.user_cwd +def _project_hooks_fingerprint(project_root: Path) -> str | None: + """Return a content fingerprint for a workspace's project hooks file.""" + from deepagents_code.hooks.loading import project_hooks_path + + try: + content = project_hooks_path(project_root).read_bytes() + except OSError: + logger.warning( + "Could not fingerprint project hooks for session trust", + exc_info=True, + ) + return None + return hashlib.sha256(content).hexdigest() + + @dataclass(frozen=True, slots=True) class WorkspaceTrust: """Decides whether project-scoped hooks may run in a given directory. @@ -350,12 +366,8 @@ class WorkspaceTrust: every reload; nothing upstream needs to hold or reinterpret the decision. """ - session_grants: frozenset[str] = frozenset() - """Canonical workspace roots trusted for this session only. - - Populated from an explicit CLI grant or an in-session `allow once` choice, - neither of which is persisted to the trust store. - """ + session_grants: frozenset[tuple[str, str]] = frozenset() + """Canonical workspace roots and hook fingerprints trusted for this session.""" consult_store: bool = True """Whether persisted trust may satisfy the policy. @@ -397,10 +409,8 @@ def for_session( A policy that grants `cwd`'s workspace root for this session and defers to the persisted store everywhere else. """ - return cls( - session_grants=cls._grants_for(cwd) if granted else frozenset(), - store_path=store_path, - ) + policy = cls(store_path=store_path) + return policy.with_session_grant(cwd) if granted else policy @classmethod def explicit_only( @@ -426,45 +436,91 @@ def explicit_only( A policy that allows `cwd`'s workspace root only when `granted`, and allows nothing otherwise. """ - return cls( - session_grants=cls._grants_for(cwd) if granted else frozenset(), - consult_store=False, - store_path=store_path, - ) + policy = cls(consult_store=False, store_path=store_path) + return policy.with_session_grant(cwd) if granted else policy + + def with_session_grant(self, cwd: Path | str) -> WorkspaceTrust: + """Return a policy with a content-bound session grant for `cwd`. - @staticmethod - def _grants_for(cwd: Path | str) -> frozenset[str]: + Args: + cwd: Directory whose workspace should be granted. + + Returns: + A replacement policy preserving persisted-store posture. Fingerprint + failures leave the workspace ungranted. + """ try: - return frozenset({_project_key(project_root_for(cwd))}) + root = project_root_for(cwd) except (OSError, ValueError): logger.warning( "Could not resolve workspace root for session hook trust", exc_info=True, ) - return frozenset() + return self + fingerprint = _project_hooks_fingerprint(root) + if fingerprint is None: + return self.without_session_grant(root) + grants = dict(self.session_grants) + grants[_project_key(root)] = fingerprint + return replace(self, session_grants=frozenset(grants.items())) + + def without_session_grant(self, cwd: Path | str) -> WorkspaceTrust: + """Return a policy without any session grant for `cwd`. + + Args: + cwd: Directory whose workspace grant should be removed. - def allows(self, cwd: Path | str) -> bool: + Returns: + A replacement policy preserving persisted-store posture. + """ + try: + key = _project_key(project_root_for(cwd)) + except (OSError, ValueError): + logger.warning( + "Could not resolve workspace root while revoking session hook trust", + exc_info=True, + ) + return self + grants = dict(self.session_grants) + grants.pop(key, None) + return replace(self, session_grants=frozenset(grants.items())) + + def allows( + self, + cwd: Path | str, + *, + project_hooks_fingerprint: str | None = None, + ) -> bool: """Return whether project hooks may run for a working directory. Args: cwd: Directory to resolve trust for. + project_hooks_fingerprint: Fingerprint of project-hook bytes already + loaded from `cwd`. When omitted, the file is read to resolve a + prospective load. Returns: - `True` when the enclosing workspace root was granted for this - session, or is recorded in the trust store and `consult_store` is - set. Unresolvable directories fail closed. + `True` when the enclosing workspace root has an unchanged session + grant, or is recorded in the trust store and `consult_store` is set. + Unresolvable directories fail closed. """ try: root = project_root_for(cwd) except (OSError, ValueError): - # The raised exception carries the offending path; don't restate it. logger.warning( "Could not resolve workspace root; treating project hooks as untrusted", exc_info=True, ) return False - if _project_key(root) in self.session_grants: - return True + granted_fingerprint = dict(self.session_grants).get(_project_key(root)) + if granted_fingerprint is not None: + current_fingerprint = ( + project_hooks_fingerprint + if project_hooks_fingerprint is not None + else _project_hooks_fingerprint(root) + ) + if granted_fingerprint == current_fingerprint: + return True if not self.consult_store: return False return is_project_hooks_trusted(root, store_path=self.store_path) diff --git a/libs/code/deepagents_code/plugins/adapters/hooks.py b/libs/code/deepagents_code/plugins/adapters/hooks.py index 4fac82dd929..7ca45ef2352 100644 --- a/libs/code/deepagents_code/plugins/adapters/hooks.py +++ b/libs/code/deepagents_code/plugins/adapters/hooks.py @@ -46,7 +46,7 @@ def _plugin_documents( documents: list[tuple[Path, JsonValue]] = [] diagnostics: list[HookDiagnostic] = [] for path in plugin.inventory.hook_files: - decoded, document, read_diagnostics = read_hooks_json(path) + decoded, document, read_diagnostics, _fingerprint = read_hooks_json(path) diagnostics.extend(read_diagnostics) if decoded: documents.append((path, document)) diff --git a/libs/code/deepagents_code/tui/widgets/cwd_switch.py b/libs/code/deepagents_code/tui/widgets/cwd_switch.py index 7116a4b01f5..44fe2087f80 100644 --- a/libs/code/deepagents_code/tui/widgets/cwd_switch.py +++ b/libs/code/deepagents_code/tui/widgets/cwd_switch.py @@ -6,6 +6,7 @@ from textual.binding import Binding, BindingType from textual.containers import Vertical +from textual.content import Content from textual.screen import ModalScreen from textual.widgets import Static @@ -240,3 +241,83 @@ def action_quit_or_interrupt(self) -> None: def action_quit_app(self) -> None: """Delegate Ctrl+D to the app-level quit handler.""" cast("DeepAgentsApp", self.app).action_quit_app() + + +HookTrustChoice = Literal["allow_once", "always_allow", "deny"] + + +class HookTrustScreen(ModalScreen[HookTrustChoice]): + """Ask how project hooks in a newly entered workspace should be trusted.""" + + can_focus = True + can_focus_children = False + + BINDINGS: ClassVar[list[BindingType]] = [ + Binding("enter", "allow_once", "Allow once", show=False, priority=True), + Binding("a", "always_allow", "Always allow", show=False, priority=True), + Binding("escape", "deny", "Deny", show=False, priority=True), + ] + + CSS = CwdSwitchPromptScreen.CSS.replace( + "CwdSwitchPromptScreen", "HookTrustScreen" + ).replace("width: 72;", "width: 76;") + + def __init__(self, *, project_root: str, config_path: str) -> None: + """Initialize the project-hooks trust prompt. + + Args: + project_root: Workspace root governing the trust decision. + config_path: Project hooks file that may execute commands. + """ + super().__init__() + self._project_root = project_root + self._config_path = config_path + + def compose(self) -> ComposeResult: + """Compose the project-hooks trust dialog. + + Yields: + Title, warning body, and keyboard help widgets. + """ + with Vertical(): + yield Static( + "Project hooks can execute commands", + classes="cwd-switch-title", + markup=False, + ) + yield Static( + Content.from_markup( + "The workspace [bold]$root[/bold] contains project hooks at " + "[bold]$path[/bold]. Only allow hooks for projects you trust. " + "Always allow also trusts future edits to this file.", + root=self._project_root, + path=self._config_path, + ), + classes="cwd-switch-body", + markup=False, + ) + yield Static( + "Enter: allow once · A: always allow · Esc: deny", + classes="cwd-switch-help", + markup=False, + ) + + def on_mount(self) -> None: + """Focus the modal so its bindings receive keyboard input.""" + self.focus() + + def action_allow_once(self) -> None: + """Approve the current file contents for this session.""" + self.dismiss("allow_once") + + def action_always_allow(self) -> None: + """Approve this workspace persistently.""" + self.dismiss("always_allow") + + def action_deny(self) -> None: + """Deny project hooks in this workspace.""" + self.dismiss("deny") + + def action_cancel(self) -> None: + """Treat app-level cancellation as deny.""" + self.action_deny() diff --git a/libs/code/tests/unit_tests/hooks/test_trust.py b/libs/code/tests/unit_tests/hooks/test_trust.py index 9be4ef19b72..4a4c1a8ebc3 100644 --- a/libs/code/tests/unit_tests/hooks/test_trust.py +++ b/libs/code/tests/unit_tests/hooks/test_trust.py @@ -7,7 +7,7 @@ from concurrent.futures import ThreadPoolExecutor from dataclasses import replace from typing import TYPE_CHECKING -from unittest.mock import MagicMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -30,6 +30,8 @@ if TYPE_CHECKING: from pathlib import Path + from deepagents_code.app import DeepAgentsApp + @pytest.fixture(autouse=True) def _enable_hooks_v2(monkeypatch: pytest.MonkeyPatch) -> None: @@ -37,13 +39,13 @@ def _enable_hooks_v2(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv(EXPERIMENTAL, "1") -def _write_project_hooks(root: Path) -> Path: - (root / ".git").mkdir(parents=True) +def _write_project_hooks(root: Path, *, event: str = "Stop") -> Path: + (root / ".git").mkdir(parents=True, exist_ok=True) hooks_dir = root / ".deepagents" - hooks_dir.mkdir() + hooks_dir.mkdir(exist_ok=True) (hooks_dir / "hooks.json").write_text( json.dumps( - {"hooks": {"Stop": [{"hooks": [{"type": "command", "command": "true"}]}]}} + {"hooks": {event: [{"hooks": [{"type": "command", "command": "true"}]}]}} ), encoding="utf-8", ) @@ -133,6 +135,23 @@ def test_session_grant_does_not_extend_to_other_workspaces(tmp_path: Path) -> No assert not is_project_hooks_trusted(granted, store_path=store) +@pytest.mark.parametrize("persisted", [False, True]) +def test_project_hook_edits_invalidate_only_session_grants( + persisted: bool, + tmp_path: Path, +) -> None: + root = _write_project_hooks(tmp_path / "project") + store = tmp_path / "state" / "hooks_trust.json" + if persisted: + assert trust_project_hooks(root, store_path=store) + policy = WorkspaceTrust.for_session(root, granted=True, store_path=store) + assert policy.allows(root) + + _write_project_hooks(root, event="SessionEnd") + + assert policy.allows(root) is persisted + + def test_explicit_only_policy_ignores_persisted_trust(tmp_path: Path) -> None: """Headless runs must not inherit a grant made in an interactive session.""" root = _write_project_hooks(tmp_path / "project") @@ -276,7 +295,7 @@ async def test_textual_app_forwards_hook_trust( ) with patch( "deepagents_code.hooks.runtime.HooksRuntime.create", - return_value=MagicMock(), + return_value=MagicMock(project_hooks_loaded=False), ) as create: await app._init_session_state() @@ -298,7 +317,7 @@ async def test_textual_app_defaults_to_untrusted_without_a_policy( app = DeepAgentsApp(agent=MagicMock(), thread_id="thread") with patch( "deepagents_code.hooks.runtime.HooksRuntime.create", - return_value=MagicMock(), + return_value=MagicMock(project_hooks_loaded=False), ) as create: await app._init_session_state() @@ -323,6 +342,34 @@ def _manager(cwd: Path, trust: WorkspaceTrust) -> HooksManager: ) +def test_manager_rejects_project_hooks_changed_after_trust_check( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from deepagents_code.hooks import trust as trust_module + + _isolate_hook_config(tmp_path, monkeypatch) + root = _write_project_hooks(tmp_path / "project") + policy = WorkspaceTrust.for_session(root, granted=True) + fingerprint = trust_module._project_hooks_fingerprint + + def fingerprint_then_replace(project_root: Path) -> str | None: + result = fingerprint(project_root) + _write_project_hooks(project_root, event="SessionEnd") + return result + + monkeypatch.setattr( + trust_module, + "_project_hooks_fingerprint", + fingerprint_then_replace, + ) + + manager = _manager(root, policy) + + assert not manager.has_handlers(HookEvent.STOP) + assert not manager.has_handlers(HookEvent.SESSION_END) + + async def test_reload_drops_project_hooks_when_leaving_trusted_workspace( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, @@ -381,3 +428,167 @@ def test_headless_manager_ignores_persisted_trust( assert not opted_out.has_handlers(HookEvent.STOP) assert opted_in.has_handlers(HookEvent.STOP) + + +async def _textual_app(cwd: Path, trust: WorkspaceTrust) -> DeepAgentsApp: + from deepagents_code.app import DeepAgentsApp + + app = DeepAgentsApp( + agent=MagicMock(), + thread_id="thread", + cwd=cwd, + hook_trust=trust, + ) + await app._init_session_state() + return app + + +async def test_cwd_retarget_without_project_hooks_reloads_without_prompt( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + _isolate_hook_config(tmp_path, monkeypatch) + current = _write_project_hooks(tmp_path / "current") + target = tmp_path / "target" + target.mkdir() + app = await _textual_app( + current, + WorkspaceTrust.for_session(current, granted=True), + ) + assert app._hooks.has_handlers(HookEvent.STOP) + app._cwd = str(target) + prompt = AsyncMock(return_value="deny") + monkeypatch.setattr(app, "_push_screen_wait", prompt) + + await app._retarget_hooks_after_cwd_switch() + + assert not app._hooks.has_handlers(HookEvent.STOP) + prompt.assert_not_awaited() + + +async def test_launch_cwd_retarget_updates_policy_before_session_state_exists( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from deepagents_code.app import DeepAgentsApp + + _isolate_hook_config(tmp_path, monkeypatch) + target = _write_project_hooks(tmp_path / "target") + app = DeepAgentsApp( + agent=MagicMock(), + thread_id="thread", + cwd=target, + hook_trust=WorkspaceTrust(), + ) + monkeypatch.setattr( + app, + "_push_screen_wait", + AsyncMock(return_value="allow_once"), + ) + + await app._retarget_hooks_after_cwd_switch() + + assert app._session_state is None + assert app._hook_trust is not None + assert app._hook_trust.allows(target) + await app._init_session_state() + assert app._hooks.has_handlers(HookEvent.STOP) + + +@pytest.mark.parametrize("choice", ["allow_once", "always_allow"]) +async def test_cwd_retarget_grants_project_hooks_from_prompt( + choice: str, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + _isolate_hook_config(tmp_path, monkeypatch) + current = tmp_path / "current" + current.mkdir() + target = _write_project_hooks(tmp_path / "target") + store = tmp_path / "state" / "hooks_trust.json" + app = await _textual_app(current, WorkspaceTrust(store_path=store)) + app._cwd = str(target) + monkeypatch.setattr(app, "_push_screen_wait", AsyncMock(return_value=choice)) + + await app._retarget_hooks_after_cwd_switch() + + assert app._hooks.has_handlers(HookEvent.STOP) + assert app._hooks.trust.allows(target) + assert is_project_hooks_trusted(target, store_path=store) is ( + choice == "always_allow" + ) + + +async def test_cwd_retarget_rejects_allow_once_when_file_changes_during_prompt( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + _isolate_hook_config(tmp_path, monkeypatch) + current = tmp_path / "current" + current.mkdir() + target = _write_project_hooks(tmp_path / "target") + app = await _textual_app(current, WorkspaceTrust()) + app._cwd = str(target) + + def mutate_before_allow(_screen: object) -> str: + _write_project_hooks(target, event="SessionEnd") + return "allow_once" + + monkeypatch.setattr( + app, + "_push_screen_wait", + AsyncMock(side_effect=mutate_before_allow), + ) + + await app._retarget_hooks_after_cwd_switch() + + assert not app._hooks.trust.allows(target) + assert not app._hooks.has_handlers(HookEvent.SESSION_END) + + +async def test_cwd_retarget_prompt_failure_fails_closed( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + _isolate_hook_config(tmp_path, monkeypatch) + current = tmp_path / "current" + current.mkdir() + target = _write_project_hooks(tmp_path / "target") + app = await _textual_app(current, WorkspaceTrust()) + app._cwd = str(target) + monkeypatch.setattr( + app, + "_push_screen_wait", + AsyncMock(side_effect=RuntimeError("screen unavailable")), + ) + notify = MagicMock() + monkeypatch.setattr(app, "notify", notify) + + await app._retarget_hooks_after_cwd_switch() + + assert not app._hooks.has_handlers(HookEvent.STOP) + assert not app._hooks.trust.allows(target) + notify.assert_called_once() + + +async def test_cwd_retarget_explicit_only_policy_never_prompts( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + _isolate_hook_config(tmp_path, monkeypatch) + current = tmp_path / "current" + current.mkdir() + target = _write_project_hooks(tmp_path / "target") + app = await _textual_app( + current, + WorkspaceTrust.explicit_only(current, granted=False), + ) + app._cwd = str(target) + prompt = AsyncMock(return_value="allow_once") + monkeypatch.setattr(app, "_push_screen_wait", prompt) + + await app._retarget_hooks_after_cwd_switch() + + prompt.assert_not_awaited() + assert app._hooks.trust.consult_store is False + assert not app._hooks.has_handlers(HookEvent.STOP) diff --git a/libs/code/tests/unit_tests/test_app.py b/libs/code/tests/unit_tests/test_app.py index 92d1a75056b..6a06539a52d 100644 --- a/libs/code/tests/unit_tests/test_app.py +++ b/libs/code/tests/unit_tests/test_app.py @@ -32460,12 +32460,18 @@ async def test_deferred_reveal_timer_fires_after_delay( class TestResumeThreadCwdSwitch: """Tests for cwd mismatch handling while resuming threads.""" + @pytest.mark.parametrize( + ("abort", "reload_manager"), + [(None, True), ("thread_switch", False)], + ) async def test_offer_switch_changes_process_cwd_and_widgets( self, + abort: Literal["thread_switch"] | None, + reload_manager: bool, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: - """Accepting the prompt switches process and UI cwd before startup.""" + """Accepting the prompt switches process and UI cwd.""" from deepagents_code.config import settings current = tmp_path / "current" @@ -32486,6 +32492,8 @@ async def test_offer_switch_changes_process_cwd_and_widgets( status_bar = MagicMock() status_bar.cwd = str(current) app._status_bar = status_bar + retarget = AsyncMock() + monkeypatch.setattr(app, "_retarget_hooks_after_cwd_switch", retarget) reload_calls: list[Path | None] = [] def reload_from_environment(*, start_path: Path | None = None) -> list[str]: @@ -32502,7 +32510,9 @@ def reload_from_environment(*, start_path: Path | None = None) -> list[str]: patch("deepagents_code.sessions.get_thread_cwd", return_value=str(target)), patch("deepagents_code.model_config.clear_caches") as clear_caches, ): - ok = await app._offer_thread_cwd_switch("thread-1", restart_server=False) + ok = await app._offer_thread_cwd_switch( + "thread-1", restart_server=False, abort=abort + ) assert ok == "continue" assert Path.cwd() == target @@ -32513,6 +32523,7 @@ def reload_from_environment(*, start_path: Path | None = None) -> list[str]: clear_caches.assert_called_once_with() screen = push_wait.call_args.args[0] assert screen._project_settings_change_detected is True + retarget.assert_awaited_once_with(reload_manager=reload_manager) async def test_offer_switch_preserves_launch_relative_server_paths( self, @@ -32662,6 +32673,8 @@ async def test_offer_switch_failure_returns_abort( app._replace_server_after_cwd_switch = replace # ty: ignore[invalid-assignment] mount = AsyncMock() app._mount_message = mount # ty: ignore[invalid-assignment] + retarget = AsyncMock() + monkeypatch.setattr(app, "_retarget_hooks_after_cwd_switch", retarget) with patch("deepagents_code.sessions.get_thread_cwd", return_value=str(target)): outcome = await app._offer_thread_cwd_switch( @@ -32670,6 +32683,7 @@ async def test_offer_switch_failure_returns_abort( assert outcome == "abort" replace.assert_awaited_once() + retarget.assert_not_awaited() # The failed switch must be surfaced as a durable message, distinct from # the transient toast `_replace_server_after_cwd_switch` already raised. mount.assert_awaited_once() @@ -32765,6 +32779,8 @@ async def test_resume_prefetch_failure_restores_server_backed_cwd_switch( set_spinner = AsyncMock() app._set_spinner = set_spinner # ty: ignore[invalid-assignment] app._update_status = MagicMock() # ty: ignore[invalid-assignment] + reload_hooks = AsyncMock() + monkeypatch.setattr(app, "_reload_hooks", reload_hooks) replace_calls: list[Path] = [] def replace_server(cwd: Path) -> str: @@ -32786,6 +32802,7 @@ def replace_server(cwd: Path) -> str: assert app._lc_thread_id == "old-thread" set_spinner.assert_has_awaits([call("Loading thread"), call(None)]) load_thread_history.assert_not_awaited() + reload_hooks.assert_awaited_once_with() async def test_threads_switch_offers_abort_and_cancels( self,