From 6303ba06b1f132d1385d97d8a83b7a545290a2df Mon Sep 17 00:00:00 2001 From: Eric Lee Date: Thu, 2 Jul 2026 09:46:05 -0700 Subject: [PATCH] feat(ch12/round4): make configured skills + hooks actually work (skill delivery, multi-scope + lifecycle hooks) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ch01 round-4's core hook wiring holds; three configured-but-doesn't-work extensibility breaks fixed: - GAP A: skill invocation delivered NOTHING to the model. A discovered, advertised, /skills-listed skill returned only 'Launching skill: foo' when the model invoked it — _run_markdown_skill rendered output['prompt'] but left new_messages unset, so the body never reached the model. Fix: populate new_messages with a meta user message carrying the rendered prompt (TS SkillTool.ts:1111-1117); the executor already consumes it. - GAP B: project/enterprise hooks NEVER loaded. HookConfigManager.load() read only user-scope ~/.clawcodex/settings.json; project (/.clawcodex/settings.json — where teams put PreToolUse hooks), local, and policy hooks never loaded, defeating even the 7 working events. Fix: load() merges user+project+local+policy, tagging each HookConfig with its HookSource. - GAP C: SessionStart/SessionEnd/PreCompact never fired (routers complete, zero callers). Fix: wired the fire sites — SessionStart once before the first real turn, SessionEnd at shutdown, PreCompact before /compact. SECURITY (critic B1, was a BLOCKING RCE): combining GAP B + GAP C would have let a malicious repo's SessionStart command hook run even when the user DECLINED the workspace-trust dialog. The session routers now read the per-context snapshot and apply should_skip_hook_due_to_trust (untrusted → only is_policy hooks survive), mirroring the tool lane + PostSampling. Multi-session-safe (critic M1): the process-global registry holds only USER-scope (cwd-independent) hooks; project/local/policy live in the per-context snapshot so a second session on a different cwd can't wipe/cross-fire the first's project hooks. Docs: my-docs/ch12-extensibility-round4-gap-analysis.md + plan + two facet reports + critic verdicts (REVISE→APPROVE) under my-docs/port-improvement-round-4/. Tests: tests/test_ch12_extensibility_round4.py (9) incl. the B1 trust-gate regression tests (untrusted-project→no-run, policy→run, trusted→run — would have caught B1). Hook/skill/server adjacents (807) + full suite at the pre-existing 9-failure baseline (7491 passed). Deferred: UserPromptSubmit + the general context-injection lane -> ch14; 15 other ABSENT hook events; skill-declared hooks + agent-def frontmatter extras (Phase-3); live-TUI skill slash render; WI-3 fidelity (SessionStart source, SessionEnd /clear+resume, PreCompact auto-compact trigger). Co-Authored-By: Claude Fable 5 --- src/hooks/config_manager.py | 99 +++++++++-- src/hooks/session_hooks.py | 106 +++++++---- src/server/agent_server.py | 54 ++++++ src/tool_system/tools/skill.py | 15 ++ tests/test_ch12_extensibility_round4.py | 224 ++++++++++++++++++++++++ 5 files changed, 443 insertions(+), 55 deletions(-) create mode 100644 tests/test_ch12_extensibility_round4.py diff --git a/src/hooks/config_manager.py b/src/hooks/config_manager.py index d7ef519d3..886633669 100644 --- a/src/hooks/config_manager.py +++ b/src/hooks/config_manager.py @@ -260,7 +260,13 @@ def _translate_legacy_notification_entry( def load_hooks_from_settings( settings_path: str | Path | None = None, + *, + source: HookSource = HookSource.USER_SETTINGS, ) -> HookConfigSnapshot: + # ch12 round-4 WI-2 — ``source`` tags every parsed HookConfig with the + # scope it came from (user/project/local/policy). The trust gate keys + # on ``HookConfig.source.is_policy`` (only policy hooks survive an + # untrusted workspace), so multi-scope loading MUST carry the source. path = Path(settings_path) if settings_path else _get_settings_path() if not path.exists(): @@ -308,11 +314,13 @@ def load_hooks_from_settings( config = _parse_hook_config(inner_raw) if config.type == "command" and not config.command: continue # malformed entry — never execute "" + config.source = source hooks.setdefault(target_event, []).append(config) continue config = _parse_hook_config(hook_raw) if config.type == "command" and not config.command: continue # malformed entry — never execute "" + config.source = source hooks.setdefault(target_event, []).append(config) return HookConfigSnapshot( @@ -327,9 +335,16 @@ def __init__( self, registry: AsyncHookRegistry, settings_path: str | Path | None = None, + *, + cwd: str | Path | None = None, ) -> None: self._registry = registry self._settings_path = Path(settings_path) if settings_path else _get_settings_path() + # ch12 round-4 WI-2 — cwd scopes the project/local settings files. + # When an explicit settings_path is given (tests / SDK), stay in + # single-scope mode for backward compatibility. + self._cwd = str(cwd) if cwd is not None else None + self._single_scope = settings_path is not None self._snapshot: HookConfigSnapshot | None = None self._last_mtime: float = 0.0 @@ -337,20 +352,69 @@ def __init__( def snapshot(self) -> HookConfigSnapshot | None: return self._snapshot - async def load(self) -> HookConfigSnapshot: - snapshot = load_hooks_from_settings(self._settings_path) - self._snapshot = snapshot + def _scope_files(self) -> list[tuple[Path, HookSource]]: + """ch12 round-4 WI-2 — the per-scope hook settings files, low→high + precedence (user < project < local < policy). Single-scope mode + (explicit settings_path) returns just that file as user-scope.""" + if self._single_scope: + return [(self._settings_path, HookSource.USER_SETTINGS)] + from src.permissions.settings_paths import ( + local_settings_path, + project_settings_path, + user_settings_path, + ) + from src.settings.managed_path import resolve_managed_settings_path + + files: list[tuple[Path, HookSource]] = [ + (Path(user_settings_path()), HookSource.USER_SETTINGS), + (Path(project_settings_path(self._cwd)), HookSource.PROJECT_SETTINGS), + (Path(local_settings_path(self._cwd)), HookSource.LOCAL_SETTINGS), + ] + managed = resolve_managed_settings_path() + if managed is not None: + files.append((Path(managed), HookSource.POLICY_SETTINGS)) + return files + async def load(self) -> HookConfigSnapshot: + # ch12 round-4 WI-2 — merge hooks from ALL scopes (was USER only, so + # a repo's .claude/settings.json hooks + enterprise policy hooks + # never fired). Each scope's configs are tagged with their source + # (for the trust gate's policy-survives-distrust rule) and both read + # lanes — the merged snapshot (context lane) + the registry (router + # lane) — are populated per source. + merged: dict[str, list[HookConfig]] = {} + scope_files = self._scope_files() + # ch12 round-4 (critic M1) — the process-GLOBAL registry holds only + # USER-scope hooks (cwd-INDEPENDENT). Project/local/policy hooks are + # cwd-specific, so registering them globally would let a second + # session's load() (different cwd) clear_source + overwrite this + # session's project hooks — cross-session contamination on the + # registry-backed lanes (PostSampling, session routers). They live + # in the per-context SNAPSHOT instead (per-session-safe), and the + # snapshot-first session routers + the snapshot-based tool lane read + # them there. Only USER is cleared+registered on the global registry. await self._registry.clear_source(HookSource.USER_SETTINGS) - for event_name, hook_configs in snapshot.hooks.items(): - for config in hook_configs: - if event_name in ALL_HOOK_EVENTS: - await self._registry.register( - event_name, # type: ignore[arg-type] - config, - HookSource.USER_SETTINGS, - ) + for path, src in scope_files: + snap = load_hooks_from_settings(path, source=src) + for event_name, hook_configs in snap.hooks.items(): + for config in hook_configs: + merged.setdefault(event_name, []).append(config) + if ( + src is HookSource.USER_SETTINGS + and event_name in ALL_HOOK_EVENTS + ): + await self._registry.register( + event_name, # type: ignore[arg-type] + config, + src, + ) + + snapshot = HookConfigSnapshot( + hooks=merged, timestamp=time.time(), + source_path=str(self._settings_path), + ) + self._snapshot = snapshot try: self._last_mtime = self._settings_path.stat().st_mtime @@ -430,12 +494,12 @@ def bootstrap_hook_config_manager( when ``settings.hooks.enabled`` is False (the framework off-switch; unreadable settings fail open to the default of enabled). - ``cwd`` asymmetry, deliberate: ``cwd`` scopes only the - ``hooks.enabled`` knob lookup (settings hierarchy); the hook *entries* - always load from the single user-scope file - (``$CLAUDE_CONFIG_DIR``/``~/.claude/settings.json``) regardless of - cwd. Multi-scope (project/local/policy) entry merging is the ch12 - round-4 subject. + ``cwd`` scopes both the ``hooks.enabled`` knob lookup AND (ch12 + round-4 WI-2) the project/local hook-entry files: ``load()`` now merges + hooks from user + project (``/.claude/settings.json``) + local + (``/.claude/settings.local.json``) + policy (enterprise managed) + scopes, each tagged with its ``HookSource`` for the trust gate. An + explicit ``settings_path`` (tests / SDK) keeps single-scope behavior. """ try: from ..settings.settings import load_settings @@ -467,6 +531,7 @@ def bootstrap_hook_config_manager( manager = HookConfigManager( get_global_hook_registry(), settings_path=settings_path, + cwd=cwd, ) asyncio.run(manager.load()) snapshot = manager.snapshot diff --git a/src/hooks/session_hooks.py b/src/hooks/session_hooks.py index 8e3b7fbab..152dcc86a 100644 --- a/src/hooks/session_hooks.py +++ b/src/hooks/session_hooks.py @@ -34,17 +34,54 @@ COMPACT_EVENT: HookEvent = "PreCompact" +async def _resolve_event_configs( + event: HookEvent, + registry: AsyncHookRegistry | None, + tool_use_context: Any, +) -> list[HookConfig]: + """ch12 round-4 (critic B1+M1) — resolve the HookConfigs to fire for a + lifecycle event, TRUST-GATED and per-session-safe. + + Prefer the per-context SNAPSHOT (per-session, carries all scopes: + user/project/local/policy) when a ``tool_use_context`` is available — + NOT the process-global registry, which is cwd-independent and would + cross-contaminate concurrent sessions (M1). Apply the workspace-trust + filter so an UNtrusted workspace runs only ``is_policy`` hooks — without + this, a malicious repo's ``.clawcodex/settings.json`` SessionStart + command hook executed even when the user DECLINED the trust dialog + (B1: arbitrary code execution). Mirrors the tool lane + (hook_executor.py) and PostSampling (post_sampling_hooks.py). + + Falls back to the global registry (USER-scope only, cwd-independent) + when no context is supplied — the test/back-compat path. + """ + if tool_use_context is not None: + from .hook_executor import _get_hooks_from_snapshot + from .trust_gate import should_skip_hook_due_to_trust + + snapshot = _get_hooks_from_snapshot(tool_use_context) + configs = list(snapshot.get(event, [])) + if should_skip_hook_due_to_trust(tool_use_context): + configs = [c for c in configs if c.source.is_policy] + return configs + + reg = registry or get_global_hook_registry() + hooks = await reg.get_hooks_for_event(event) + return [h.config for h in hooks] + + async def run_session_start_hooks( registry: AsyncHookRegistry | None = None, *, session_id: str | None = None, cwd: str | None = None, + tool_use_context: Any = None, ) -> list[dict[str, Any]]: - reg = registry or get_global_hook_registry() - hooks = await reg.get_hooks_for_event(SESSION_START_EVENT) - + configs = await _resolve_event_configs( + SESSION_START_EVENT, registry, tool_use_context, + ) results: list[dict[str, Any]] = [] - for hook in hooks: + for config in configs: stdin_data = { "hook_event": SESSION_START_EVENT, "session_id": session_id, @@ -55,19 +92,16 @@ async def run_session_start_hooks( from .exec_http_hook import execute_http_hook from .exec_prompt_hook import execute_prompt_hook - if hook.config.type == "command": - result = await _execute_command_hook(hook.config, stdin_data) - elif hook.config.type == "http": - result = await execute_http_hook(hook.config, stdin_data) - elif hook.config.type == "prompt": - result = await execute_prompt_hook(hook.config, stdin_data) + if config.type == "command": + result = await _execute_command_hook(config, stdin_data) + elif config.type == "http": + result = await execute_http_hook(config, stdin_data) + elif config.type == "prompt": + result = await execute_prompt_hook(config, stdin_data) else: continue - results.append({ - "hook": hook.config, - "result": result, - }) + results.append({"hook": config, "result": result}) return results @@ -78,12 +112,13 @@ async def run_session_end_hooks( session_id: str | None = None, total_cost: float | None = None, total_turns: int | None = None, + tool_use_context: Any = None, ) -> list[dict[str, Any]]: - reg = registry or get_global_hook_registry() - hooks = await reg.get_hooks_for_event(SESSION_END_EVENT) - + configs = await _resolve_event_configs( + SESSION_END_EVENT, registry, tool_use_context, + ) results: list[dict[str, Any]] = [] - for hook in hooks: + for config in configs: stdin_data = { "hook_event": SESSION_END_EVENT, "session_id": session_id, @@ -94,17 +129,14 @@ async def run_session_end_hooks( from .hook_executor import _execute_command_hook from .exec_http_hook import execute_http_hook - if hook.config.type == "command": - result = await _execute_command_hook(hook.config, stdin_data) - elif hook.config.type == "http": - result = await execute_http_hook(hook.config, stdin_data) + if config.type == "command": + result = await _execute_command_hook(config, stdin_data) + elif config.type == "http": + result = await execute_http_hook(config, stdin_data) else: continue - results.append({ - "hook": hook.config, - "result": result, - }) + results.append({"hook": config, "result": result}) return results @@ -116,12 +148,13 @@ async def run_compact_hooks( tokens_before: int | None = None, tokens_after: int | None = None, trigger: str = "manual", + tool_use_context: Any = None, ) -> list[dict[str, Any]]: - reg = registry or get_global_hook_registry() - hooks = await reg.get_hooks_for_event(COMPACT_EVENT) - + configs = await _resolve_event_configs( + COMPACT_EVENT, registry, tool_use_context, + ) results: list[dict[str, Any]] = [] - for hook in hooks: + for config in configs: stdin_data = { "hook_event": COMPACT_EVENT, "session_id": session_id, @@ -133,16 +166,13 @@ async def run_compact_hooks( from .hook_executor import _execute_command_hook from .exec_http_hook import execute_http_hook - if hook.config.type == "command": - result = await _execute_command_hook(hook.config, stdin_data) - elif hook.config.type == "http": - result = await execute_http_hook(hook.config, stdin_data) + if config.type == "command": + result = await _execute_command_hook(config, stdin_data) + elif config.type == "http": + result = await execute_http_hook(config, stdin_data) else: continue - results.append({ - "hook": hook.config, - "result": result, - }) + results.append({"hook": config, "result": result}) return results diff --git a/src/server/agent_server.py b/src/server/agent_server.py index 3181f6834..1174f8540 100644 --- a/src/server/agent_server.py +++ b/src/server/agent_server.py @@ -131,6 +131,10 @@ class _AgentSession: # ch11 round-4 WI-1 — SESSION-scoped set of already-surfaced memory # paths, so the LLM recall doesn't re-inject the same memory every turn. _memory_surfaced: set = field(default_factory=set) + # ch12 round-4 WI-3 — SessionStart fires once, lazily, before the first + # real turn (inside the async context; _build_runtime runs sync in an + # executor with no live loop). Guarded so it fires exactly once. + _session_start_fired: bool = False session: Any = None system_prompt: Any = "You are a helpful assistant." _base_system_prompt: Any = None # system prompt before the /plan section is composed in @@ -1281,6 +1285,19 @@ async def _do_compact(self, request_id: object, instructions: object) -> None: if active: self._reply(request_id, {"ok": False, "error": "cannot compact during an active turn"}) return + # ch12 round-4 WI-3 — PreCompact hook fires BEFORE the summarize + # call (TS commands/compact/compact.ts:160). Configured PreCompact + # hooks (e.g. persist state before compaction) never ran because + # the router had no live caller. + try: + from src.hooks.session_hooks import run_compact_hooks + + await run_compact_hooks( + session_id=self.session_id, trigger="manual", + tool_use_context=self.tool_context, + ) + except Exception: # noqa: BLE001 — a hook must not block compaction + logger.debug("[agent-server] PreCompact hooks failed", exc_info=True) try: from src.compact_service.service import compact_conversation @@ -1475,6 +1492,26 @@ def _build_turn_pipeline_config(self, turn_provider: Any) -> Any: exc_info=True) return None + def _fire_session_start_once(self) -> None: + """ch12 round-4 WI-3 — fire SessionStart hooks exactly once, before + the first real turn. Sync wrapper (a tiny asyncio.run) because + _run_turn is a sync worker; the session_hooks router is async.""" + if self._session_start_fired: + return + self._session_start_fired = True + try: + import asyncio as _asyncio + + from src.hooks.session_hooks import run_session_start_hooks + + _asyncio.run(run_session_start_hooks( + session_id=self.session_id, cwd=self.cwd, + tool_use_context=self.tool_context, + )) + except Exception: # noqa: BLE001 — a hook must not block the turn + logger.debug("[agent-server] SessionStart hooks failed", + exc_info=True) + @staticmethod def _parse_turn_budget(prompt: Any) -> int | None: """ch05 round-4 GAP B — the '+500k' auto-continue budget from the @@ -1566,6 +1603,12 @@ def on_message(message: Any) -> None: if env is not None: self._emit(env) + # ch12 round-4 WI-3 — SessionStart fires once, before the first real + # turn (skipped for internal/notification turns so a task + # notification can't count as the session's start). + if not internal: + self._fire_session_start_once() + # /effort: wrap the provider to inject reasoning_effort (default off ⇒ # the real provider is used unchanged). turn_provider = _EffortProvider(self.provider, self._effort) if self._effort else self.provider @@ -1645,6 +1688,17 @@ def on_message(message: Any) -> None: async def shutdown(self) -> None: self._stop.set() + # ch12 round-4 WI-3 — SessionEnd hooks fire at shutdown (TS + # gracefulShutdown.ts:486). Configured cleanup hooks never ran. + try: + from src.hooks.session_hooks import run_session_end_hooks + + await run_session_end_hooks( + session_id=self.session_id, + tool_use_context=self.tool_context, + ) + except Exception: # noqa: BLE001 — a hook must not block shutdown + logger.debug("[agent-server] SessionEnd hooks failed", exc_info=True) # ch10 round-4 WI-2 — stop the eviction sweeper daemon (started in # _build_runtime under single_session). Idempotent; safe if never # started. diff --git a/src/tool_system/tools/skill.py b/src/tool_system/tools/skill.py index 47e6563a8..5fd01bf28 100644 --- a/src/tool_system/tools/skill.py +++ b/src/tool_system/tools/skill.py @@ -285,6 +285,20 @@ def _run_markdown_skill(skill_name: str, args: str, context: ToolContext) -> Too # Build context modifier if skill specifies allowed_tools, model, or effort context_modifier = _build_context_modifier(skill) + # ch12 round-4 WI-1 — DELIVER the rendered skill body to the model. The + # skill tool_result content is only "Launching skill: {name}" + # (_skill_map_result_to_api); the actual instructions must ride as a + # separate meta user message. TS SkillTool.ts:1111-1117 does exactly + # this: newMessages: [createUserMessage({content: finalContent, + # isMeta:true})]. Without it the model was told a skill launched but + # never received its instructions — skills did nothing. The executor + # already consumes ToolResult.new_messages (tool_execution.py:449-451). + new_messages: list[Any] = [] + if prompt and prompt.strip(): + from src.types.messages import create_user_message + + new_messages.append(create_user_message(content=prompt, isMeta=True)) + return ToolResult( name="Skill", output={ @@ -297,6 +311,7 @@ def _run_markdown_skill(skill_name: str, args: str, context: ToolContext) -> Too "model": skill.model, }, context_modifier=context_modifier, + new_messages=new_messages or None, ) diff --git a/tests/test_ch12_extensibility_round4.py b/tests/test_ch12_extensibility_round4.py new file mode 100644 index 000000000..37b44a71f --- /dev/null +++ b/tests/test_ch12_extensibility_round4.py @@ -0,0 +1,224 @@ +"""ch12 round-4 acceptance tests: (WI-1) skill invocation delivers the +rendered body to the model, (WI-2) hooks load from all settings scopes with +correct source tagging, (WI-3) SessionStart/SessionEnd/PreCompact fire sites. + +Covers my-docs/port-improvement-round-4/ch12-extensibility-round4-plan.md. +""" +from __future__ import annotations + +import asyncio +import json +import tempfile +import unittest +from pathlib import Path +from unittest.mock import MagicMock, patch + + +class TestSkillDeliversBody(unittest.TestCase): + """WI-1 — the Skill tool result carries the rendered instructions as a + meta user message (new_messages), not just 'Launching skill: X'.""" + + def test_skill_result_has_new_messages(self): + from src.tool_system.tools.skill import _run_markdown_skill + from src.tool_system.context import ToolContext + + with tempfile.TemporaryDirectory() as tmp: + skill_dir = Path(tmp) / ".claude" / "skills" / "greet" + skill_dir.mkdir(parents=True) + (skill_dir / "SKILL.md").write_text( + "---\nname: greet\ndescription: greet the user\n---\n" + "Say hello warmly and ask how you can help." + ) + ctx = ToolContext(workspace_root=Path(tmp)) + result = _run_markdown_skill("greet", "", ctx) + + self.assertFalse(result.is_error) + self.assertIsNotNone(result.new_messages) + self.assertEqual(len(result.new_messages), 1) + msg = result.new_messages[0] + self.assertTrue(getattr(msg, "isMeta", False)) + self.assertIn("hello", str(msg.content).lower()) + + +class TestMultiScopeHookLoading(unittest.TestCase): + """WI-2 — hooks load from user + project + local scopes, each tagged + with its source (the trust gate keys on source.is_policy).""" + + def test_project_and_local_hooks_load_with_source(self): + from src.hooks.config_manager import HookConfigManager + from src.hooks.hook_types import HookSource + from src.hooks.registry import AsyncHookRegistry + + with tempfile.TemporaryDirectory() as tmp: + proj = Path(tmp) + claude = proj / ".clawcodex" + claude.mkdir() + (claude / "settings.json").write_text(json.dumps({ + "hooks": {"PreToolUse": [ + {"matcher": "Bash", "hooks": [ + {"type": "command", "command": "echo project"}]} + ]} + })) + (claude / "settings.local.json").write_text(json.dumps({ + "hooks": {"PreToolUse": [ + {"matcher": "Bash", "hooks": [ + {"type": "command", "command": "echo local"}]} + ]} + })) + # Point the USER scope at an empty temp file. + user = proj / "user_settings.json" + user.write_text("{}") + + reg = AsyncHookRegistry() + mgr = HookConfigManager(reg, cwd=str(proj)) + with patch("src.permissions.settings_paths.user_settings_path", + return_value=str(user)): + snap = asyncio.run(mgr.load()) + + commands = { + c.command: c.source + for c in snap.hooks.get("PreToolUse", []) + } + self.assertIn("echo project", commands) + self.assertIn("echo local", commands) + self.assertEqual(commands["echo project"], HookSource.PROJECT_SETTINGS) + self.assertEqual(commands["echo local"], HookSource.LOCAL_SETTINGS) + + def test_project_hook_is_policy_false_so_dropped_under_distrust(self): + # SECURITY (critic b): a malicious repo's .clawcodex/settings.json + # PreToolUse hook (source=PROJECT_SETTINGS) is NOT policy, so the + # trust gate drops it in an UNtrusted workspace — it only fires once + # the user trusts the workspace. Loading project hooks is therefore + # safe (trust-then-run). + from src.hooks.hook_types import HookSource + + self.assertFalse(HookSource.PROJECT_SETTINGS.is_policy) + self.assertFalse(HookSource.LOCAL_SETTINGS.is_policy) + self.assertFalse(HookSource.USER_SETTINGS.is_policy) + self.assertTrue(HookSource.POLICY_SETTINGS.is_policy) + + def test_single_scope_mode_unchanged(self): + # An explicit settings_path (tests/SDK) → single user scope. + from src.hooks.config_manager import HookConfigManager + from src.hooks.hook_types import HookSource + from src.hooks.registry import AsyncHookRegistry + + with tempfile.TemporaryDirectory() as tmp: + f = Path(tmp) / "settings.json" + f.write_text(json.dumps({ + "hooks": {"PreToolUse": [ + {"type": "command", "command": "echo user"}]} + })) + reg = AsyncHookRegistry() + mgr = HookConfigManager(reg, settings_path=str(f)) + snap = asyncio.run(mgr.load()) + cfgs = snap.hooks.get("PreToolUse", []) + self.assertEqual(len(cfgs), 1) + self.assertEqual(cfgs[0].source, HookSource.USER_SETTINGS) + + +class TestSessionHookTrustGate(unittest.TestCase): + """critic B1 (BLOCKING security) — the session-lifecycle routers must + NOT run a repo-injected (project) command hook in an UNtrusted + workspace; a policy hook MUST still run.""" + + def _ctx_with_snapshot(self, *, trusted, source): + from src.hooks.config_manager import HookConfigSnapshot + from src.hooks.hook_types import HookConfig + from src.tool_system.context import ToolContext + + ctx = ToolContext(workspace_root=Path("/tmp")) + ctx.workspace_trusted = trusted + + mgr = MagicMock() + mgr.snapshot = HookConfigSnapshot(hooks={ + "SessionStart": [HookConfig( + type="command", command="echo pwned", source=source, + )] + }) + ctx.hook_config_manager = mgr + return ctx + + def _run(self, ctx): + from src.hooks.session_hooks import run_session_start_hooks + + ran = [] + + async def _fake_cmd(config, stdin): + ran.append(config.command) + return {"exit_code": 0} + + with patch("src.hooks.hook_executor._execute_command_hook", _fake_cmd): + asyncio.run(run_session_start_hooks(tool_use_context=ctx)) + return ran + + def test_untrusted_project_hook_does_not_run(self): + from src.hooks.hook_types import HookSource + + ctx = self._ctx_with_snapshot( + trusted=False, source=HookSource.PROJECT_SETTINGS, + ) + self.assertEqual(self._run(ctx), []) # dropped by the trust gate + + def test_untrusted_policy_hook_runs(self): + from src.hooks.hook_types import HookSource + + ctx = self._ctx_with_snapshot( + trusted=False, source=HookSource.POLICY_SETTINGS, + ) + self.assertEqual(self._run(ctx), ["echo pwned"]) # policy survives + + def test_trusted_project_hook_runs(self): + from src.hooks.hook_types import HookSource + + ctx = self._ctx_with_snapshot( + trusted=True, source=HookSource.PROJECT_SETTINGS, + ) + self.assertEqual(self._run(ctx), ["echo pwned"]) # trusted → runs + + +class TestSessionLifecycleFireSites(unittest.TestCase): + """WI-3 — SessionStart/SessionEnd/PreCompact fire at the right sites.""" + + def _session(self): + from src.server.agent_server import AgentServerConfig, _AgentSession + + return _AgentSession( + session_id="s1", cwd="/tmp", + config=AgentServerConfig(single_session=True), + loop=MagicMock(), out_queue=MagicMock(), + ) + + def test_session_start_fires_once(self): + sess = self._session() + calls = [] + + async def _spy(**kw): + calls.append(kw) + return [] + + with patch("src.hooks.session_hooks.run_session_start_hooks", _spy): + sess._fire_session_start_once() + sess._fire_session_start_once() # second call is a no-op + self.assertEqual(len(calls), 1) + self.assertEqual(calls[0].get("session_id"), "s1") + + def test_session_end_fires_on_shutdown(self): + sess = self._session() + calls = [] + + async def _spy(**kw): + calls.append(kw) + return [] + + # shutdown touches other subsystems; patch them to no-ops. + with patch("src.hooks.session_hooks.run_session_end_hooks", _spy), \ + patch("src.tasks.eviction.stop_eviction_sweeper"): + sess._worker = None + sess._current_abort = None + asyncio.run(sess.shutdown()) + self.assertEqual(len(calls), 1) + + +if __name__ == "__main__": + unittest.main()