fix(code): refresh hooks after cwd switches

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.

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,28 +369,30 @@ 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",
                     message=f"Legacy hooks list missing at {path}",
                     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(

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

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(),
         )