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.

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:

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:

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:

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:

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.

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():

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:

print(engine.hooks)  # list of EngineHook instances

To remove a hook:

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.

Related guide: Critics

Learn how critics differ from hooks and how to use them for control flow and decision gating.

Next tutorial: Multi-Agent Systems

Build systems with coordinator and worker agents that dispatch tasks in parallel.