Hooks and Observability: Extending the Lifecycle Safely

agents
reliability
security

A permission layer from Chapter 8 owns system invariants such as path confinement and approval posture. Hooks add a different kind of control: user-configured code that observes or checks lifecycle boundaries. This chapter follows an event from the in-process EventBus, through the append-only Session journal, to an external HookSystem process. Each boundary gets a contract, a deterministic experiment, and a failure test.

The current package intentionally provides a contained hook runner rather than a full veto protocol. That distinction matters. A hook that crashes should not kill the agent, but a hook whose rejection is silently ignored is not a style gate. We will test both the guarantees that exist and the guarantees that still need implementation.

Event planes

EventBus is an in-process observer. Subscribers receive Python event objects in registration order, one at a time, and a faulty subscriber is logged and contained. HookSystem is an external side-effect boundary. It launches a command or temporary shell script with a lifecycle-specific environment and waits synchronously with a timeout.

Keep the planes separate in the design:

  • use events for telemetry and live consumers that share the process;
  • use hooks for configured integrations that should not import the agent’s internals;
  • never treat either plane as the source of truth for session state. The journal is the replayable state record.

The first check uses a slow subscriber, a faulty subscriber, and a duplicate registration. It asks whether observability changes the agent’s event order or availability.

import asyncio

from agent_harness.events import AgentEvent, EventBus

seen09 = []


async def slow_subscriber09(event):
    await asyncio.sleep(0)
    seen09.append(("slow", event.type.value))


def faulty_subscriber09(event):
    raise RuntimeError("telemetry sink unavailable")


def recording_subscriber09(event):
    seen09.append(("record", event.type.value))


async def event_demo09():
    bus09 = EventBus()
    bus09.subscribe(slow_subscriber09)
    bus09.subscribe(faulty_subscriber09)
    bus09.subscribe(recording_subscriber09)
    bus09.subscribe(recording_subscriber09)  # deduplicated by EventBus
    await bus09.emit(AgentEvent.agent_start("inspect"))
    await bus09.emit(AgentEvent.text_complete("done"))
    return bus09


bus09 = asyncio.run(event_demo09())
print("observed:", seen09)
print("recording registrations are deduplicated:", seen09.count(("record", "agent_start")) == 1)
assert seen09 == [
    ("slow", "agent_start"),
    ("record", "agent_start"),
    ("slow", "text_complete"),
    ("record", "text_complete"),
]
assert bus09.unsubscribe(recording_subscriber09) is True

<1> is delivered to the slow subscriber before the recording subscriber, even though the first handler yields. The faulty subscriber does not prevent later handlers from seeing the event, and registering the same handler twice does not duplicate telemetry. This ordered containment is useful for metrics, but it is not durable: a process crash can still lose events that were never written to the session journal.

Event payloads

High-level event constructors in events.py normalize common lifecycle data. A completed tool event includes success, output, errors, metadata, diffs, truncation, and exit code when the result object provides them. Serializing that payload at the observation boundary makes a trajectory inspectable by the evaluator instead of leaving it as an opaque Python object.

import json

from agent_harness.events import AgentEvent, TokenUsage
from agent_harness.tools.base import ToolResult

success_result09 = ToolResult.success_result(
    "3 matches",
    metadata={"files_searched": 2},
)
completed_event09 = AgentEvent.tool_call_complete(
    "call-1",
    "grep",
    success_result09,
)
end_event09 = AgentEvent.agent_end(
    response="finished",
    usage=TokenUsage(prompt_tokens=12, completion_tokens=4, total_tokens=16),
)
serialized_event09 = json.dumps(
    {"type": completed_event09.type.value, "data": completed_event09.data},
    sort_keys=True,
)
print(serialized_event09)
print("end usage:", end_event09.data["usage"])
assert '"success": true' in serialized_event09
assert end_event09.data["usage"]["total_tokens"] == 16

The serialized event is a compact evidence record, not a replacement for the tool result in the conversation. Keep the model-facing observation and the evaluator-facing event distinct: the former preserves the loop invariant from Chapter 5, while the latter supports cost, safety, and failure attribution.

Event journal

Session records user messages, assistant messages, tool results, usage, turns, and resets as append-only journal entries. Session.replay starts from a fresh system prompt and applies those entries; unknown entry types are ignored for forward compatibility. Replay reconstructs state, while EventBus describes what observers saw during a live run.

import copy
from pathlib import Path

from agent_harness.config import Config
from agent_harness.events import TokenUsage, ToolResultMessage
from agent_harness.session import Session
from agent_harness.tools.base import ToolRegistry

journal_config09 = Config(
    cwd=Path.cwd(),
    developer_instructions="Offline journal demonstration.",
)
journal_registry09 = ToolRegistry(journal_config09)
session09 = Session(journal_config09, registry=journal_registry09)
session09.start_turn()
session09.add_user_message("inspect parser.py")
session09.add_assistant_message(
    "I will inspect it.",
    tool_calls=[
        {"id": "call-1", "name": "read_file", "arguments": {"path": "parser.py"}}
    ],
)
session09.add_tool_result(ToolResultMessage("call-1", "1| return token", is_error=False))
session09.track_usage(TokenUsage(prompt_tokens=20, completion_tokens=6, total_tokens=26))
recorded_journal09 = copy.deepcopy(session09.journal)
recorded_journal09.append({"type": "future_event", "detail": "ignored by old replay"})
replayed09 = Session.replay(
    recorded_journal09,
    journal_config09,
    registry=journal_registry09,
)

print("journal types:", [entry["type"] for entry in session09.journal])
print("replayed roles:", [message["role"] for message in replayed09.messages])
print("replayed usage:", replayed09.total_usage.total_tokens)
print("unknown entry tolerated:", len(replayed09.journal) == len(recorded_journal09))
assert replayed09.messages == session09.messages
assert replayed09.turn_count == session09.turn_count
assert replayed09.total_usage == session09.total_usage
assert replayed09.journal == recorded_journal09

<1> copies the journal before adding a future entry, so the live session remains the reference state. Replay reproduces its messages, turn count, and token usage byte-for-byte for the fields represented by the journal; it also retains the unknown entry in the copied journal while ignoring it during application. The timestamp in Session.to_dict() is intentionally not part of this deterministic replay assertion.

This is the observability split to preserve in later chapters: events can be rich and transient, while the journal is the minimal state transition record needed to reconstruct a trajectory.

Hook environment

HookConfig identifies a trigger and exactly one command or inline script. When hooks are enabled, HookSystem runs matching hooks in list order from Config.cwd. The environment contains AI_AGENT_TRIGGER, AI_AGENT_CWD, and context-specific fields such as AI_AGENT_TOOL_NAME, AI_AGENT_TOOL_PARAMS, AI_AGENT_TOOL_RESULT, AI_AGENT_USER_MESSAGE, AI_AGENT_RESPONSE, or AI_AGENT_ERROR.

The current contract uses environment variables and captures stdout/stderr internally. It does not send a JSON document on stdin or interpret a structured allow/deny/modify verdict. The next cell tests the contract that actually exists.

import asyncio
from tempfile import TemporaryDirectory

from agent_harness.config import HookConfig, HookTrigger
from agent_harness.hooks import HookSystem

with TemporaryDirectory() as raw_dir:
    hook_workspace09 = Path(raw_dir)
    hook_config09 = Config(
        cwd=hook_workspace09,
        hooks_enabled=True,
        hooks=[
            HookConfig(
                name="first",
                trigger=HookTrigger.BEFORE_TOOL,
                script=r"printf 'first\n' >> hook-order.log",
            ),
            HookConfig(
                name="second",
                trigger=HookTrigger.BEFORE_TOOL,
                script=r"printf 'second\n' >> hook-order.log",
            ),
            HookConfig(
                name="capture-context",
                trigger=HookTrigger.BEFORE_TOOL,
                script=r'''printf '%s|%s' "$AI_AGENT_TRIGGER" "$AI_AGENT_TOOL_NAME" > hook-env.log''',
            ),
        ],
    )
    hooks09 = HookSystem(hook_config09)
    asyncio.run(hooks09.trigger_before_tool("shell", {"command": "pwd"}))
    order09 = (hook_workspace09 / "hook-order.log").read_text(encoding="utf-8")
    env09 = (hook_workspace09 / "hook-env.log").read_text(encoding="utf-8")
    print("ordered hook output:", repr(order09))
    print("environment contract:", env09)
    assert order09 == "first\nsecond\n"
    assert env09 == "before_tool|shell"

The two log lines establish ordered execution, and the environment file shows that a hook can identify the lifecycle point and tool without importing Python objects. This is enough for a formatter, audit logger, or notification adapter. It is not enough for a hook to return a model-facing rejection, because _run_hook discards process output and does not map nonzero exit codes to a ToolResult.

Hook failures

A user hook is untrusted extension code. It can exit nonzero, raise through a shell command, write hostile output, or wait forever. HookSystem catches exceptions, runs each hook in its own subprocess, starts a new process group, and kills that group when timeout_sec expires. The consequence is graceful degradation: the hook becomes a no-op from the agent’s point of view, and the warning is available to process logs rather than being injected into the model as a fabricated tool result.

import time

from agent_harness.config import HookConfig, HookTrigger
from agent_harness.hooks import HookSystem

containment_config09 = Config(
    cwd=Path.cwd(),
    hooks_enabled=True,
    hooks=[
        HookConfig(
            name="crash",
            trigger=HookTrigger.BEFORE_AGENT,
            command="exit 7",
        ),
        HookConfig(
            name="hang",
            trigger=HookTrigger.BEFORE_AGENT,
            command="sleep 2",
            timeout_sec=0.05,
        ),
    ],
)
started09 = time.monotonic()
asyncio.run(HookSystem(containment_config09).trigger_before_agent("offline check"))
elapsed09 = time.monotonic() - started09
print("trigger returned:", True)
print("contained in under one second:", elapsed09 < 1.0)
assert elapsed09 < 1.0

The check establishes liveness, not correctness of the hook’s policy. A crashing or hanging hook cannot break the loop, but the current API also does not emit a durable hook-failure event or return a structured verdict to the caller. A production dispatcher should record hook name, trigger, exit status, timeout, and bounded diagnostics as an event while keeping those diagnostics out of the model-facing observation unless deliberately converted into a safe result.

Observability metrics

Observability is useful only when it answers a question. For a trajectory, count lifecycle events, tool successes and failures, token usage, hook failures, and budget stops separately. A single “run completed” metric is a proxy: it can be green while the agent edited the wrong file or while telemetry silently dropped a tool result.

The following synthetic record is deterministic, but it uses the same AgentEvent constructors and ToolResult payload shape that a live Agent emits.

from collections import Counter

from agent_harness.events import AgentEvent

trajectory_events09 = [
    AgentEvent.agent_start("fix parser"),
    AgentEvent.tool_call_complete(
        "call-1",
        "read_file",
        ToolResult.success_result("12 lines"),
    ),
    AgentEvent.tool_call_complete(
        "call-2",
        "edit",
        ToolResult.error_result("old_string found 2 times"),
    ),
    AgentEvent.agent_error("turn budget exhausted"),
]
event_counts09 = Counter(event.type.value for event in trajectory_events09)
tool_events09 = [
    event for event in trajectory_events09 if event.type.value == "tool_call_complete"
]
failed_tools09 = sum(not event.data["success"] for event in tool_events09)
summary09 = {
    "event_counts": dict(sorted(event_counts09.items())),
    "tool_calls": len(tool_events09),
    "failed_tools": failed_tools09,
    "budget_or_agent_errors": event_counts09["agent_error"],
}
print(summary09)
assert summary09 == {
    "event_counts": {
        "agent_error": 1,
        "agent_start": 1,
        "tool_call_complete": 2,
    },
    "tool_calls": 2,
    "failed_tools": 1,
    "budget_or_agent_errors": 1,
}

The metric vector distinguishes a failed edit from a failed run and a run that ended because of a budget. Those distinctions support the reliability lens from the course plan: correction failure can be counted from repeated tool errors, proxy optimization can be found when “agent end” rises without task checks, and observability failure can be found when the journal and event counts disagree.

Hook configuration

Hooks relocate specification from library code into user configuration. That is useful for project-specific style gates, but it makes configuration validation part of the safety boundary. HookConfig rejects an empty hook and rejects a definition that supplies both a command and an inline script.

from agent_harness.config import HookConfig, HookTrigger

for invalid_kwargs09 in (
    {"name": "empty", "trigger": HookTrigger.BEFORE_TOOL},
    {
        "name": "ambiguous",
        "trigger": HookTrigger.BEFORE_TOOL,
        "command": "true",
        "script": "true",
    },
):
    try:
        HookConfig(**invalid_kwargs09)
    except ValueError as error09:
        print(type(error09).__name__, ":", str(error09))
    else:
        raise AssertionError("invalid hook configuration was accepted")

Validation prevents an ambiguous hook from reaching the process launcher. It does not validate the shell command’s intent, so a command can still be syntactically valid and unsafe. That is why hook policy and permission policy remain distinct: the former is an extension contract, while the latter is a system-owned capability boundary.

The current HookSystem also has no blocking semantics. A command that prints deny or exits with status 7 is contained and ignored; it cannot replace a tool result. Implementing the plan’s JSON-on-stdin and allow/deny/modify verdicts would require a new API contract, bounded output parsing, and an explicit rule for what happens when that parser fails. This notebook records the gap instead of presenting a non-existent veto as a guarantee.

Hook guarantees

The existing APIs now have executable evidence for four claims:

  1. In-process subscribers are ordered, deduplicated, and failure-contained.
  2. Session journals replay conversation state independently of transient observers.
  3. External hooks run in order with a documented environment and bounded lifetime.
  4. Event payloads can be reduced to metrics that preserve tool, error, and budget distinctions.

The remaining uncertainty is equally concrete: nonzero hook exits are not vetoes, hook output is not a structured model-facing result, and hook failures are not yet first-class journal entries. Those are design decisions for the next implementation pass, not assumptions to hide in prose. The next course chapter can therefore delegate work with a clear audit trail rather than adding another opaque extension point.

Back to top