훅 등록

훅은 루프가 의미 있는 경계에서 발화하는 이벤트에 핸들러를 붙이는 방법입니다. 관측, 비용 집계, 감사 로그 같은 횡단 관심사를 루프 코드를 건드리지 않고 추가할 때 씁니다. 가장 흔한 함정은 핸들러를 작성하고도 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은 고유해야 하며, 같은 이름으로 재등록하면 기존 핸들러를 덮어써서 명시 등록과 파일시스템 디스커버리가 겹쳐도 이중 등록이 안 생깁니다.

# core/wiring/bootstrap.py — build_hooks()
hooks.register(
    HookEvent.TOOL_EXEC_FAILED,
    on_tool_failed,
    name="tool_failure_observer",
    priority=60,
)

한 핸들러를 여러 이벤트(또는 모든 이벤트)에 한 번에 붙이려면 register_prefix("*", handler, ...)를 씁니다. run log writer가 바로 이 방식으로 "*" 와일드카드를 구독해서, 새HookEvent를 추가해도 자동으로 로그에 잡힙니다.

4. 우선순위 등급을 정합니다

priority는 낮을수록 먼저 실행됩니다(기본 100). 데이터를 보강하는 인터셉터는 낮게, 단순 관측자는 높게 둡니다. 기존 bootstrap의 등급을 기준으로 삼으면 됩니다. run log writer는 50, agent_runtime_state 기록은 55, stuck detector는 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.