Logging agent state transitions for postmortem debugging

When a multi-agent pipeline silently corrupts a task, the difference between a 30-minute fix and a two-day slog is disciplined logging agent state transitions. This guide shows how to instrument state changes with structured events so you can reconstruct execution during postmortem debugging. We will build a minimal but production-grade pattern you can drop into any Python agent loop.

Step 1: Define an explicit state schema

You cannot debug what you have not named. Before writing any logging code, pin down the finite set of states each agent can occupy. A stringly-typed state = "doing stuff" will ruin your postmortem. Use an enum and a typed event container so every emission shares a contract.

import time
from enum import Enum
from pydantic import BaseModel, Field

class AgentState(str, Enum):
    IDLE = "idle"
    PLANNING = "planning"
    TOOL_CALL = "tool_call"
    LLM_WAIT = "llm_wait"
    SYNTHESIZE = "synthesize"
    ERROR = "error"
    DONE = "done"

class TransitionEvent(BaseModel):
    trace_id: str
    agent_id: str
    from_state: AgentState | None
    to_state: AgentState
    ts: float = Field(default_factory=time.time)
    meta: dict = Field(default_factory=dict)

The from_state is nullable because the first transition (boot) has no predecessor. Keep meta free-form but discipline yourself to only put small, queryable facts there (model name, tool id, error code). Never stuff raw prompts or tool outputs into the transition line—hash them if you need correlation.

Step 2: Emit structured JSON at every boundary

Logging agent state transitions means writing one event at the exact moment an agent changes state. Wrap your logger so it outputs one JSON object per line. This is the only format that survives piping through jq, Vector, or a log sink without custom parsers.

import logging, json, sys

logger = logging.getLogger("agent.transitions")
logger.setLevel(logging.INFO)
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(logging.Formatter("%(message)s"))
logger.addHandler(handler)

def emit_transition(event: TransitionEvent):
    logger.info(json.dumps(event.model_dump()))

Call emit_transition inside the agent loop right before you mutate the state variable. If you use a state machine library, hook the on_transition callback. Do not batch these—latency of a single logger.info is microseconds, and you want the timestamp to reflect reality.

Avoid side-effect logging

Do not log the transition after the work is done. If the agent crashes mid-state, the event is lost. Emit first, then act.

Step 3: Propagate trace and parent context

A multi-agent system has concurrent actors. A bare state log without correlation IDs is noise. Use contextvars to thread a trace_id (the whole run) and parent_id (the calling agent) through async or threaded execution.

import contextvars, uuid

trace_id_ctx = contextvars.ContextVar("trace_id")
agent_id_ctx = contextvars.ContextVar("agent_id")

def start_trace(agent_id: str) -> str:
    tid = uuid.uuid4().hex[:16]
    trace_id_ctx.set(tid)
    agent_id_ctx.set(agent_id)
    return tid

def current_trace() -> str:
    return trace_id_ctx.get("no-trace")

When agent A spawns agent B, copy the trace_id but set a new agent_id. This gives you a tree you can rebuild in a postmortem. Logging agent state transitions without this context forces you to guess which planner invoked which executor.

Step 4: Capture LLM call metadata inside the transition

Most agent states hinge on an LLM call. The transition into LLM_WAIT and back out to SYNTHESIZE should carry the model, token usage, and cache status. If you route through n4n.ai, its per-token usage metering and forwarded cache-control hints appear in the response and can be attached directly to the transition event, giving you cost and cache visibility in the same log line.

from openai import OpenAI

client = OpenAI(base_url="https://api.n4n.ai/v1", api_key="sk-your-key")

def call_llm(prompt_hash: str, trace: str, agent: str):
    emit_transition(TransitionEvent(
        trace_id=trace, agent_id=agent,
        from_state=AgentState.PLANNING, to_state=AgentState.LLM_WAIT,
        meta={"model": "gpt-4o-mini", "prompt_hash": prompt_hash}))

    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "..."}],
        extra_headers={"x-cache-control": "read"}
    )

    emit_transition(TransitionEvent(
        trace_id=trace, agent_id=agent,
        from_state=AgentState.LLM_WAIT, to_state=AgentState.SYNTHESIZE,
        meta={"model": resp.model, "usage": resp.usage.model_dump()}))

The prompt_hash is a cheap SHA256 of the prompt—enough to confirm which template fired without bloating the log. Token counts in usage let you spot a runaway loop that re-plans ten times.

Step 5: Ship JSON lines to a queryable store

Local stdout is fine for dev, but postmortem debugging requires retention. Pipe the JSON lines into any system that supports key-value filtering. A minimal production setup is:

python agent_runner.py 2>&1 | jq -c 'select(.trace_id != null)' | \
  tee -a /var/log/agent_transitions.jsonl

For real scale, use a sidecar that tails the file and forwards to OpenSearch or BigQuery. The schema is already flat, so no transformation is needed. Set a retention policy of 14 days; state transitions are high-volume but small.

What not to log

Resist adding stack traces to the transition event. Emit a separate ERROR log line with the exception, and put only error_code in the transition meta. Keeping the transition stream uniform makes replay scripts trivial.

Step 6: Reconstruct a run during postmortem

When a user reports a broken task, pull the trace_id from your request logs and replay the transition sequence. A 20-line helper turns the JSONL file into a timeline:

def replay(trace_id: str, logfile: str):
    events = []
    with open(logfile) as f:
        for line in f:
            try:
                ev = json.loads(line)
            except json.JSONDecodeError:
                continue
            if ev.get("trace_id") == trace_id:
                events.append(ev)
    events.sort(key=lambda e: e["ts"])
    for e in events:
        meta = e.get("meta", {})
        print(f'{e["ts"]:.3f} {e["agent_id"]:12} '
              f'{str(e["from_state"]):10} -> {e["to_state"]:10} '
              f'{meta}')

Run it and you see exactly where the agent stalled—e.g., three LLM_WAIT entries with rising token counts and no SYNTHESIZE means the model looped on a malformed tool response. Logging agent state transitions this way converts a mystery into a stack of evidence.

Step 7: Verify your instrumentation

Instrumentation is worthless if it silently drops events. Before shipping, write a smoke test that forces a known path including an error state.

def test_transitions_emitted():
    tid = start_trace("test-agent")
    emit_transition(TransitionEvent(trace_id=tid, agent_id="test-agent",
                                    from_state=None, to_state=AgentState.IDLE))
    emit_transition(TransitionEvent(trace_id=tid, agent_id="test-agent",
                                    from_state=AgentState.IDLE, to_state=AgentState.ERROR,
                                    meta={"error_code": "forced"}))
    # capture stdout, assert two lines, both parse, trace_id matches

Success criteria: every state change in your agent’s happy path and failure path produces exactly one JSON line with trace_id, agent_id, to_state, and ts. Query the log by trace_id and confirm the ordered list matches your state machine definition. If a transition is missing, your emit call is in the wrong place—move it before the state mutation.

Closing practice notes

Treat the state transition log as a first-class artifact, not a debug afterthought. Review it in code review: if a new agent state is added without a corresponding emit, the PR is incomplete. Over time, logging agent state transitions becomes the backbone of your observability, letting you answer “why did the system do X” by reading a timeline instead of reproducing a race condition.