> ## Documentation Index > Fetch the complete documentation index at: https://qitor.mintlify.site/llms.txt > Use this file to discover all available pages before exploring further. # Hook Lifecycle > Master the EngineHook lifecycle to observe, log, and extend QitOS runs at every phase boundary. Hooks are the primary runtime extension point in QitOS. They let you observe and react to every phase transition inside the Engine without modifying the agent itself. This tutorial covers the full hook lifecycle, the context objects hooks receive, and how to write both step-level and tool-level hooks. ## Hooks vs. Critics QitOS has two extension mechanisms that operate at different levels: | Mechanism | Role | Can modify flow? | | ---------- | ---------------------------- | --------------------------------------------- | | **Hook** | Observation and side effects | No -- hooks are notified, they do not control | | **Critic** | Control and gating | Yes -- critics can veto or revise decisions | Use hooks when you want to log, trace, emit metrics, or trigger external notifications. Use critics when you want to constrain or override agent behavior. ## Step 1: The EngineHook Base Class Every hook inherits from `EngineHook`. The base class defines no-op methods for every lifecycle callback, so you only override the ones you need. ```python theme={null} from qitos.engine.hooks import EngineHook ``` The full callback set, in execution order within a single run: | Callback | When it fires | | ----------------------------------- | ------------------------------------------ | | `on_run_start(task, state, engine)` | Before the first step | | `on_before_step(ctx, engine)` | Before a step begins | | `on_before_decide(ctx, engine)` | Before the LLM is called | | `on_after_decide(ctx, engine)` | After the LLM returns a decision | | `on_before_act(ctx, engine)` | Before actions are executed | | `on_after_act(ctx, engine)` | After actions finish executing | | `on_before_critic(ctx, engine)` | Before the critic runs | | `on_after_critic(ctx, engine)` | After the critic finishes | | `on_before_reduce(ctx, engine)` | Before `reduce()` folds results into state | | `on_after_reduce(ctx, engine)` | After `reduce()` completes | | `on_before_check_stop(ctx, engine)` | Before the stop condition is evaluated | | `on_after_check_stop(ctx, engine)` | After the stop condition is evaluated | | `on_after_step(ctx, engine)` | After the step is fully complete | | `on_recover(ctx, engine)` | When the engine recovers from an error | | `on_run_end(result, engine)` | After the final step completes | Additional lifecycle callbacks that fire outside the step loop: | Callback | When it fires | | ---------------------------------------- | ---------------------------------- | | `on_session_start(ctx, engine)` | When an interactive session begins | | `on_session_end(ctx, engine)` | When an interactive session ends | | `on_before_compact(ctx, engine)` | Before context compaction runs | | `on_after_compact(ctx, engine)` | After context compaction completes | | `on_event(event, state, record, engine)` | On every `RuntimeEvent` emission | | `on_step_end(record, state, engine)` | When a `StepRecord` is finalized | All no-op methods return `None`. Override only the callbacks you care about. ## Step 2: HookContext and ToolHookContext Every step-level callback receives a `HookContext` dataclass that carries everything a hook might need: ```python theme={null} from qitos.engine.hooks import HookContext # HookContext fields: # task: str -- the original task string # step_id: int -- current step index # phase: RuntimePhase -- current FSM phase # state: StateSchema -- live agent state # env_view: dict | None -- snapshot of the environment # observation: Any -- latest observation # decision: Any -- latest decision # model_response: dict -- raw model response # action_results: list -- results from action execution # record: StepRecord -- current step record # payload: dict -- arbitrary phase-specific data # error: Exception | None -- error if phase failed # stop_reason: str | None -- reason the run stopped # run_id: str -- unique run identifier # ts: str -- ISO timestamp ``` Tool-level callbacks receive `ToolHookContext`, which extends `HookContext` with tool-specific fields: ```python theme={null} from qitos.engine.hooks import ToolHookContext # ToolHookContext adds: # tool_name: str -- name of the tool being called # tool_args: dict -- arguments passed to the tool # tool_result: Any -- result returned by the tool (on_after_tool_use) # permission_decision: str -- "allowed" or "denied" (on_permission_denied) ``` The `phase` field comes from the `RuntimePhase` enum: ```python theme={null} from qitos.engine.states import RuntimePhase # RuntimePhase values: # INIT, DECIDE, ACT, CRITIC, REDUCE, CHECK_STOP, END, # DECIDE_ERROR, ACT_ERROR, RECOVER, # DELEGATE_START, DELEGATE_END, # HANDOFF_START, HANDOFF_END, # INTERRUPT, FANOUT_START, FANOUT_END, # COMPACT, SESSION_START, SESSION_END ``` ## Step 3: Writing a Custom Logging Hook A common use case is recording every phase transition for later analysis. Here is a `LifecycleRecorderHook` that logs each callback with a timestamp and step ID: ```python theme={null} import logging from qitos.engine.hooks import EngineHook, HookContext from qitos.engine.states import RuntimePhase logger = logging.getLogger("qitos.lifecycle") class LifecycleRecorderHook(EngineHook): """Records every engine lifecycle event into a structured log.""" def on_run_start(self, task, state, engine): logger.info("run_start | task=%s", task[:80]) def on_before_step(self, ctx: HookContext, engine): logger.info("before_step | step=%d phase=%s", ctx.step_id, ctx.phase.value) def on_after_step(self, ctx: HookContext, engine): logger.info("after_step | step=%d", ctx.step_id) def on_before_decide(self, ctx: HookContext, engine): logger.info("before_decide | step=%d", ctx.step_id) def on_after_decide(self, ctx: HookContext, engine): decision = ctx.decision logger.info("after_decide | step=%d decision=%s", ctx.step_id, type(decision).__name__) def on_before_act(self, ctx: HookContext, engine): logger.info("before_act | step=%d", ctx.step_id) def on_after_act(self, ctx: HookContext, engine): logger.info("after_act | step=%d results=%d", ctx.step_id, len(ctx.action_results)) def on_before_critic(self, ctx: HookContext, engine): logger.info("before_critic | step=%d", ctx.step_id) def on_after_critic(self, ctx: HookContext, engine): logger.info("after_critic | step=%d", ctx.step_id) def on_before_reduce(self, ctx: HookContext, engine): logger.info("before_reduce | step=%d", ctx.step_id) def on_after_reduce(self, ctx: HookContext, engine): logger.info("after_reduce | step=%d", ctx.step_id) def on_before_check_stop(self, ctx: HookContext, engine): logger.info("before_check_stop | step=%d", ctx.step_id) def on_after_check_stop(self, ctx: HookContext, engine): logger.info("after_check_stop | step=%d stop_reason=%s", ctx.step_id, ctx.stop_reason) def on_recover(self, ctx: HookContext, engine): logger.warning("recover | step=%d error=%s", ctx.step_id, ctx.error) def on_run_end(self, result, engine): logger.info("run_end | steps=%d", result.state.current_step if hasattr(result, 'state') else '?') ``` Because hooks cannot modify the flow, the `LifecycleRecorderHook` is safe to add to any run without side effects. ## Step 4: Tool-Level Hooks Tool-level hooks fire around individual tool invocations, giving you fine-grained visibility into which tools the agent calls and what they return. ```python theme={null} from qitos.engine.hooks import EngineHook, ToolHookContext class ToolAuditHook(EngineHook): """Audits every tool call: name, args, result, and permission decisions.""" def on_before_tool_use(self, ctx: ToolHookContext, engine): print(f"[TOOL CALL] {ctx.tool_name}({ctx.tool_args})") def on_after_tool_use(self, ctx: ToolHookContext, engine): result_str = str(ctx.tool_result)[:200] # truncate for display print(f"[TOOL RESULT] {ctx.tool_name} -> {result_str}") def on_permission_denied(self, ctx: ToolHookContext, engine): print( f"[PERMISSION DENIED] {ctx.tool_name}({ctx.tool_args}) " f"decision={ctx.permission_decision}" ) ``` The three tool-level callbacks: | Callback | Context | Purpose | | ---------------------- | ---------------------------------------------- | -------------------------------- | | `on_before_tool_use` | `ToolHookContext` (no `tool_result` yet) | Log or validate before execution | | `on_after_tool_use` | `ToolHookContext` (with `tool_result`) | Inspect or record the result | | `on_permission_denied` | `ToolHookContext` (with `permission_decision`) | Track rejected tool calls | Use `on_permission_denied` to monitor security boundaries without modifying the permission system itself. ## Step 5: Registering Hooks with the Engine Hooks are registered on the `Engine` instance before calling `run()`: ```python theme={null} from qitos import Engine engine = Engine(agent=my_agent) engine.add_hook(LifecycleRecorderHook()) engine.add_hook(ToolAuditHook()) result = engine.run(task="Fix the bug in buggy_module.py") ``` You can register multiple hooks. They fire in registration order within each callback. Because hooks are observation-only, order does not affect control flow -- but it does affect log output ordering, which matters for debugging. To inspect currently registered hooks: ```python theme={null} print(engine.hooks) # list of EngineHook instances ``` To remove a hook: ```python theme={null} engine.remove_hook(my_hook_instance) ``` ## Full Lifecycle Diagram ``` on_run_start | v on_before_step |---> on_before_decide |---> on_after_decide |---> on_before_act | |---> on_before_tool_use (per tool call) | |---> on_after_tool_use (per tool call) | |---> on_permission_denied (if denied) |---> on_after_act |---> on_before_critic |---> on_after_critic |---> on_before_reduce |---> on_after_reduce |---> on_before_check_stop |---> on_after_check_stop v on_after_step | ... (loop until stop) | v on_run_end ``` On error recovery, `on_recover` fires instead of the remaining step callbacks for that step, and the engine may retry or abort depending on configuration. Learn how critics differ from hooks and how to use them for control flow and decision gating. Build systems with coordinator and worker agents that dispatch tasks in parallel.