Skip to content

feat(ch12/round4): make configured skills + hooks actually work #595

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ericleepi314 merged 1 commit into from
Jul 2, 2026
Merged

feat(ch12/round4): make configured skills + hooks actually work #595

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
99 changes: 82 additions & 17 deletions src/hooks/config_manager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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():
Expand Down Expand Up @@ -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(
Expand All @@ -327,30 +335,86 @@ 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

@property
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
Expand Down Expand Up @@ -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 (``<cwd>/.claude/settings.json``) + local
(``<cwd>/.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
Expand Down Expand Up @@ -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
Expand Down
106 changes: 68 additions & 38 deletions src/hooks/session_hooks.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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

Expand All @@ -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,
Expand All @@ -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

Expand All @@ -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,
Expand All @@ -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
Loading