훅은 루프가 의미 있는 경계에서 발화하는 이벤트에 핸들러를 붙이는 방법입니다. 관측, 비용 집계, 감사 로그 같은 횡단 관심사를 루프 코드를 건드리지 않고 추가할 때 씁니다. 가장 흔한 함정은 핸들러를 작성하고도 bootstrap에 등록하지 않는 것입니다. 핸들러가 존재한다고 발화하지는 않습니다.
1. 이벤트를 고릅니다
발화 지점은 core/hooks/system.py의 HookEvent enum에 모두 정의되어 있습니다. 도구 실행을 듣고 싶으면 TOOL_EXEC_STARTED / TOOL_EXEC_ENDED / TOOL_EXEC_FAILED, 세션 경계는 SESSION_STARTED / SESSION_ENDED, LLM 호출은 LLM_CALL_STARTED / LLM_CALL_ENDED를 고릅니다. 없는 이벤트가 필요하면 enum에 멤버를 추가하고 발화 지점도 같은 PR에 함께 넣습니다. 예약만 하고 emit-site를 미루면 발화하지 않는 죽은 이벤트가 됩니다.
2. 핸들러를 작성합니다
핸들러 시그니처는 (event: HookEvent, data: dict) -> None입니다. fire-and-forget 관측자는 None을 돌려주고, 권고 값을 호출자에게 돌려주는 피드백 훅은 dict을 돌려줍니다. 핸들러 안의 예외는 HookSystem.trigger가 잡아서 경고 로그로 남기므로 다른 핸들러를 막지 않습니다. 동기·비동기 둘 다 지원합니다.
from core.hooks import HookEvent
def on_tool_failed(event: HookEvent, data: dict) -> None:
tool_name = data.get("tool_name", "")
error = data.get("error", "")
# observe only — no return value needed
log.warning("tool %s failed: %s", tool_name, error)3. bootstrap에서 등록합니다
이 단계가 핵심입니다. core/wiring/bootstrap.py의 build_hooks()가 HookSystem을 만들고 모든 핵심 핸들러를 붙이는 곳입니다. 거기서 hooks.register(event, handler, name=..., priority=...)를 호출합니다. name은 겹치는 exact/prefix 범위에서 고유해야 합니다. 다른 핸들러를 같은 이름으로 등록하면 fail-loud하며, 의도적 교체만 replace=True를 사용합니다.
# core/wiring/bootstrap.py — build_hooks()
hooks.register(
HookEvent.TOOL_EXEC_FAILED,
on_tool_failed,
name="tool_failure_observer",
priority=60,
)한 핸들러를 이벤트 family에 붙이려면 register_prefix("SUBAGENT", handler, ...)를 씁니다. 운영 이벤트 저장은 일반 핸들러가 아니라 post-dispatch sink가 담당합니다.
4. 우선순위 등급을 정합니다
priority는 낮을수록 먼저 실행됩니다(기본 100). 데이터를 보강하는 인터셉터는 낮게, 단순 관측자는 높게 둡니다. 기존 bootstrap의 등급을 기준으로 삼으면 됩니다. metrics는 45, agent_runtime_state 기록은 55, 세션 라이프사이클 로거는 90입니다.trigger_interceptor() 경로의 핸들러는 {"block": True} 또는 {"modify": {...}}를 돌려줘 체인을 막거나 데이터를 수정할 수 있습니다.
확인
발화되는지 직접 트리거해서 확인합니다. list_hooks()는 와일드카드 구독자까지 합쳐 어떤 핸들러가 실제로 발화될지 보여줍니다.
from core.hooks import HookEvent, HookSystem
hooks = HookSystem()
hooks.register(HookEvent.TOOL_EXEC_FAILED, on_tool_failed, name="tool_failure_observer")
print(hooks.list_hooks(HookEvent.TOOL_EXEC_FAILED))
# {'tool_exec_failed': ['tool_failure_observer']}
results = hooks.trigger(HookEvent.TOOL_EXEC_FAILED, {"tool_name": "web_fetch", "error": "timeout"})
print([r.success for r in results]) # [True]참조: Hook system, Agentic loop.