Build an AI Agent Flight Recorder

Trace and debug a simulated support agent with a local timeline and tests.

Introduction

30 Second Summary

A support agent can return a polished answer while hiding every decision that produced it. When that workflow fails, a generic error leaves you guessing where the run broke.

In this project, you will turn a simulated support agent into a traceable workflow with a local flight recorder. You will use correlated JSONL traces to diagnose a routing failure before proving the repair with deterministic tests.

What You'll Build

Your browser timeline will let you open any agent run to pinpoint its route decision, nested operations, timing, and first failed span.

By the end of this project, you'll have:

  • A set of deterministic agent scenarios that reproduce normal runs, routing failures, timeouts, malformed responses, and intermittent-looking failures locally.
  • A local trace timeline powered by FastAPI. You can follow each operation under one trace ID and see exactly where an error started.
  • An observability scorecard that reports trace completeness, failure attribution, span integrity, recovery success, and separate latency measurements. A pytest suite proves those guarantees remain intact.
  • Secret Mission: recover an every-third-run timeout with one bounded retry while preserving the failed attempt as trace evidence.

Are there any prerequisites?

You need the provided starter repository plus Python 3.11 or newer. You also need VS Code and Git.

No API key, cloud account, Docker runtime, or external LLM is required.

Before We Start

A support agent can hide the operation that failed. It can also hide the route decision, timing, or error behind that failure. This project commits you to using correlated execution evidence instead of guessing from source code.

Verify and Launch the Starter Agent

The starter repository is already open in VS Code. A shared Python baseline keeps its behavior consistent across workshop machines.

A virtual environment isolates the pinned FastAPI and pytest packages from your system packages. You will prove the setup by running a known healthy scenario.

In this step, get ready to:
  • Verify Python 3.11 or newer.
  • Isolate the project inside a local .venv environment.
  • Run the opaque starter agent on its healthy password reset scenario.
Verify your Python baseline

This project requires Python 3.11 or newer. Checking the interpreter first prevents an older version from creating an incompatible environment.

  • Open a terminal in VS Code for the starter repository.
  • Check the active Python version by running this command:
python --version

What does this command check?

The command asks your active Python interpreter to print its version. The major and minor numbers identify whether it meets the project baseline.

✔️ I see version 3.11 or higher

Your Python baseline is ready. The interpreter can create the isolated environment for the starter agent.

ⓧ I see an older version

Your default command points to an older interpreter. Check the operating-system-specific Python command before installing another version.

macOS

  • Check the installed Python 3 interpreter by running this command:
python3 --version
Why check Python 3?

macOS can have more than one Python installation. The Python 3 command checks the interpreter used to create .venv in the next substep.

If the output reports Python 3.11 or newer, continue with that interpreter.

  • Use the official Python macOS installation guide if the Python 3 version is still too old.
  • Install a current supported Python release.
  • Start a fresh VS Code terminal session after installation.
  • Repeat the Python 3 version check above.

Windows

  • Check the intended Python 3.11 interpreter by running this command:
py -3.11 --version
Why use the Windows launcher?

The Windows launcher selects Python 3.11 explicitly. This keeps an older default interpreter from creating the project environment.

If the output reports Python 3.11 or newer, continue with the Windows commands in the next substep.

  • Use the official Python Windows installation guide if the launcher cannot find Python 3.11.
  • Install a current supported Python release.
  • Start a fresh VS Code terminal session after installation.
  • Repeat the Windows version check above.

Linux

  • Check the installed Python 3 interpreter by running this command:
python3 --version
Why check Python 3?

Linux distributions can keep an older interpreter as the default Python command. The Python 3 command checks the interpreter used to create .venv in the next substep.

If the output reports Python 3.11 or newer, continue with that interpreter.

  • Use your Linux distribution's supported package manager if the Python 3 version is still too old.
  • Install Python 3.11 or newer.
  • Start a fresh VS Code terminal session after installation.
  • Repeat the Python 3 version check above.

ⓧ Command not found

Your terminal cannot resolve the default Python command. Check the operating-system-specific command before installing Python.

macOS

  • Check whether Python 3 is already available by running this command:
python3 --version
What does this fallback check?

The Python 3 command checks the interpreter name used by the official macOS installation. The next substep uses the same command to create .venv.

If the output reports Python 3.11 or newer, your interpreter is ready.

  • Use the official Python macOS installation guide if this fallback is also unavailable.
  • Install a current supported Python release.
  • Start a fresh VS Code terminal session after installation.
  • Repeat the Python 3 version check above.

Windows

  • Check whether the Python 3.11 interpreter is available through the Windows launcher by running:
py -3.11 --version
What does this fallback check?

The Windows launcher selects the intended Python version directly. The next substep uses that interpreter to create .venv.

If the output reports Python 3.11 or newer, your interpreter is ready.

  • Use the official Python Windows installation guide if the launcher is also unavailable.
  • Install a current supported Python release.
  • Start a fresh VS Code terminal session after installation.
  • Repeat the Windows version check above.

Linux

  • Check whether Python 3 is already available by running this command:
python3 --version
What does this fallback check?

Linux commonly exposes Python 3 through this command. The next substep uses it to create .venv.

If the output reports Python 3.11 or newer, your interpreter is ready.

  • Use your Linux distribution's supported package manager if Python 3 is unavailable.
  • Install Python 3.11 or newer.
  • Start a fresh VS Code terminal session after installation.
  • Repeat the Python 3 version check above.
Create an isolated virtual environment

The .venv directory gives this repository its own interpreter and package set. Activation makes later Python commands use that environment.

The commands differ by operating system. Use the tab that matches your machine.

macOS

  • Create .venv inside the open starter repository by running this command:
python3 -m venv .venv

What does this command create?

Python creates an isolated interpreter and package directory at .venv. Packages installed there stay separate from your system installation.

  • Check the VS Code Explorer sidebar for the new .venv directory.

You should see .venv inside the starter repository.

  • Activate the new environment by running this command:
source .venv/bin/activate

What does activation change?

Activation updates the current terminal session to use the interpreter inside .venv. Later Python commands now use the isolated packages.

Your terminal prompt should now include .venv.

Environment did not activate?

Confirm that the terminal is inside the open starter repository. The activation path only works where the .venv directory exists.

If the directory exists but activation still fails, help me diagnose the macOS virtual environment path.

Windows

  • Create .venv inside the open starter repository by running this command:
py -3.11 -m venv .venv

What does this command create?

The Windows launcher creates .venv with Python 3.11. Packages installed there stay separate from your system installation.

  • Check the VS Code Explorer sidebar for the new .venv directory.

You should see .venv inside the starter repository.

  • Activate the new environment in PowerShell by running this command:
.venv\Scripts\Activate.ps1

What does activation change?

Activation updates the current PowerShell session to use the interpreter inside .venv. Later Python commands now use the isolated packages.

Your PowerShell prompt should now include .venv.

PowerShell blocked activation?

Activation is optional. The next substep includes a direct interpreter path that keeps the dependencies inside .venv.

If the environment directory was not created, help me diagnose the Windows virtual environment command.

Linux

  • Create .venv inside the open starter repository by running this command:
python3 -m venv .venv

What does this command create?

Python creates an isolated interpreter and package directory at .venv. Packages installed there stay separate from your system installation.

  • Check the VS Code Explorer sidebar for the new .venv directory.

You should see .venv inside the starter repository.

  • Activate the new environment by running this command:
source .venv/bin/activate

What does activation change?

Activation updates the current terminal session to use the interpreter inside .venv. Later Python commands now use the isolated packages.

Your terminal prompt should now include .venv.

Environment did not activate?

Confirm that the terminal is inside the open starter repository. The activation path only works where the .venv directory exists.

If the directory exists but activation still fails, help me diagnose the Linux virtual environment path.

Install dependencies and launch the starter agent

The repository already contains requirements.txt with the two pinned packages. Installing from this file gives every learner the same dependency versions.

✔️ .venv is active

Use this path when your terminal prompt includes .venv.

  • Install the pinned repository dependencies by running this command:
python -m pip install -r requirements.txt

What does this command install?

Python runs pip through the active environment. The installer reads requirements.txt to install the exact project pins.

The installation should finish without a dependency error. Your environment now contains FastAPI 0.162.0 plus pytest 9.2.0.

Before you run the agent, do you expect the prepared password reset scenario to succeed or fail?

  • Launch the opaque password reset scenario by running this command:
python flight_recorder.py legacy password_reset

What does this command run?

The legacy command runs the starter agent without structured trace output. The password_reset scenario follows the prepared healthy route.

You should see a recommendation line beginning with status=success. That confirms the opaque starter agent can complete its healthy path.

Starter agent not succeeding?

  • Confirm that the terminal is inside the open starter repository.
  • Confirm that the dependency installation completed without an error.
  • Confirm that the scenario name is password_reset.

If the command still fails, help me debug the starter agent launch.

ⓧ PowerShell blocked activation

PowerShell can invoke the environment interpreter directly. The dependencies remain isolated even when the terminal prompt does not change.

  • Install the pinned repository dependencies through the environment interpreter by running:
.venv\Scripts\python.exe -m pip install -r requirements.txt

What does this fallback install?

The command runs pip through the interpreter stored inside .venv. The packages remain isolated without PowerShell activation.

The installation should finish without a dependency error. The .venv environment now contains FastAPI 0.162.0 plus pytest 9.2.0.

Before you run the agent, do you expect the prepared password reset scenario to succeed or fail?

  • Launch the opaque password reset scenario through the environment interpreter by running:
.venv\Scripts\python.exe flight_recorder.py legacy password_reset

What does this fallback run?

The environment interpreter runs flight_recorder.py with the legacy command. The password_reset scenario follows the prepared healthy route.

You should see a recommendation line beginning with status=success. That confirms the isolated interpreter can run the opaque starter agent.

Direct interpreter command failed?

  • Confirm that .venv exists inside the open starter repository.
  • Confirm that the dependency installation completed without an error.
  • Confirm that the scenario name is password_reset.

If the command still fails, help me debug the PowerShell starter agent launch.

✔️ Awesome, I've got everything!

Your isolated environment contains the pinned dependencies. The starter agent now completes the password reset scenario successfully.

ⓧ I'd like to double check the full code

The VS Code Explorer sidebar should show .venv inside the open starter repository. Compare requirements.txt with the complete pinned file below.

fastapi[standard-no-fastapi-cloud-cli]==0.162.0
pytest==9.2.0

Why these exact pins?

  • The FastAPI line installs version 0.162.0 with its standard local development dependencies.
  • The pytest line installs version 9.2.0 for the regression tests used later.

Your Python environment is isolated. The opaque healthy path works. Next, you will expose how little that success reveals when a support request fails.

Reproduce the Opaque Failure

Your starter agent already returns a successful password-reset recommendation. That baseline proves the local environment can execute the workflow.

Now you need to test what the legacy output reveals when a refund request takes a different path. The crucial question is whether the terminal identifies what happened inside the run.

In this step, get ready to:
  • Reconfirm the healthy password-reset baseline.
  • Run the refund scenario twice to test its repeatability.
  • Audit the diagnostic evidence exposed by the legacy output.
Reconfirm the healthy baseline

A known healthy scenario gives you a comparison point. Its output shows what the legacy interface reports when the workflow reaches a recommendation.

  • Return to the integrated terminal in VS Code from the previous step.
  • Run the known healthy scenario with the active project environment:
python flight_recorder.py legacy password_reset

What Does This Run Prove?

This command reruns the known healthy scenario through the legacy interface. It confirms that your environment remains ready for a controlled comparison.

You should see a line containing status=success. The recommendation should reference knowledge_search evidence for password_reset.

Your baseline is steady: the password-reset path still reaches a recommendation.

Baseline Result Changed?

  • Confirm that you returned to the terminal used in the previous step.
  • Confirm that the terminal folder contains flight_recorder.py.
  • Use the same Python interpreter that produced the earlier successful result.

Ask for help with the unexpected result: help me diagnose why the healthy password-reset scenario stopped succeeding

Trigger the refund path twice

A single run can leave uncertainty about repeatability. Two identical inputs show whether the behavior stays consistent.

Before you run these commands, do you expect both refund requests to produce matching terminal evidence?

  • Run the refund scenario twice by running these commands:
python flight_recorder.py legacy refund_route
python flight_recorder.py legacy refund_route

What Does This Comparison Reveal?

Both executions print only status=error message=Agent failed. The matching result proves that the refund behavior is reproducible.

The generic line gives you no internal evidence about the route or failing operation. Repetition confirms the problem without explaining its cause.

  • Review the two refund outputs in the terminal history.

You should see two matching refund error lines beneath the healthy baseline output.

Refund Outputs Do Not Match?

  • Confirm that both commands use the refund_route scenario.
  • Confirm that both commands use the legacy mode.
  • Rerun only the command whose output differs from the expected generic error.

Ask for help comparing the runs: help me investigate why my two legacy refund runs produce different results

Audit the missing evidence

Useful observability evidence connects a final outcome to the operations that produced it. The legacy error leaves that execution story hidden.

What Is Missing From the Legacy Output?

  • No operation name identifies the failing stage.
  • No selected_tool value reveals the routing decision.
  • No trace correlation connects the work performed during one run.
  • No error_type value distinguishes the failure cause.
  • No simulated_latency_ms value exposes the scenario timing.
  • No wall_duration_ms value exposes the measured local duration.
  • Compare the legacy error line with each evidence category above.

Every category remains unanswered. The legacy line reports only that the agent failed.

Your refund failure is reproducible. Next, you will replace the generic line with correlated JSONL spans that identify the failing operation.

Record Correlated JSONL Spans

Two identical refund failures have proved that the agent can fail reproducibly without producing useful diagnostic evidence. The legacy output still hides the operation that caused the failure.

A trace connects one agent run to its nested spans. You will record each completed span as one structured JSONL row.

In this step, get ready to:
  • Implement the recorder that writes completed spans to a local trace file.
  • Instrument the agent operations with shared trace correlation.
  • Reproduce the refund failure with operation-level evidence.
Build the span recorder

Each line in the trace file represents one finished operation. The recorder writes the line after the operation succeeds or raises an error.

  • In flight_recorder.py, locate the existing JsonlFormatter class.
  • Replace that class with the code below:
class JsonlFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = record.msg if isinstance(record.msg, dict) else {"message": record.getMessage()}
        return json.dumps(payload, sort_keys=True)

What does this formatter do?

  • JsonlFormatter accepts the span dictionary attached to a log record.
  • json.dumps() serializes that dictionary as one sorted JSON object.
  • Each logging call therefore becomes one JSONL row.
  • Locate new_trace() inside the existing FlightRecorder class.
  • Replace new_trace(), span(), and close() with the following methods:
    def new_trace(self) -> str:
        return token_hex(16)
    @contextmanager
    def span(self, operation: str, trace_id: str, parent_span_id: str | None = None, attributes: dict | None = None):
        span_id = token_hex(8)
        started_at = utc_now()
        started_counter = perf_counter()
        data = {"span_id": span_id, "attributes": dict(attributes or {}), "simulated_latency_ms": 0}
        event = {"trace_id": trace_id, "span_id": span_id, "parent_span_id": parent_span_id, "operation": operation, "started_at": started_at, "status": "OK", "attributes": data["attributes"], "error_type": None, "error_message": None}
        try:
            yield data
        except Exception as exc:
            event["status"] = "ERROR"
            event["error_type"] = type(exc).__name__
            event["error_message"] = str(exc)
            raise
        finally:
            event["ended_at"] = utc_now()
            event["wall_duration_ms"] = round((perf_counter() - started_counter) * 1000, 3)
            event["simulated_latency_ms"] = data["simulated_latency_ms"]
            self.logger.info(event)
    def close(self) -> None:
        self.logger.removeHandler(self.handler)
        self.handler.close()

How does a span preserve evidence?

  • new_trace() generates 16 random bytes as a 32-character hexadecimal trace ID.
  • span() generates eight random bytes as a 16-character hexadecimal span ID.
  • yield lets the wrapped operation add attributes and simulated latency.
  • finally records timing evidence for successful operations and failed operations.
  • Save flight_recorder.py.
  • Check the editor for red syntax underlines inside FlightRecorder.

Legacy execution needs the same recorder interface without writing trace rows. A null recorder preserves that interface while discarding the completed span data.

  • Locate the existing NullRecorder class in flight_recorder.py.
  • Replace that class with the code below:
class NullRecorder:
    def new_trace(self) -> str:
        return token_hex(16)
    @contextmanager
    def span(self, operation: str, trace_id: str, parent_span_id: str | None = None, attributes: dict | None = None):
        yield {"span_id": token_hex(8), "attributes": dict(attributes or {}), "simulated_latency_ms": 0}

Why keep a null recorder?

NullRecorder exposes the same new_trace() and span() methods as FlightRecorder. The agent can switch recording behavior without changing its workflow logic.

  • Save flight_recorder.py.
  • Confirm that the legacy happy path still runs by executing this command:
python flight_recorder.py legacy password_reset

You should see a successful recommendation for password_reset. This confirms that the null recorder still supports the original execution path.

Seeing a recorder error?

Confirm that both recorder classes define new_trace() and span(). Check that @contextmanager sits directly above each span() method.

help me compare the FlightRecorder and NullRecorder interfaces

Instrument the agent workflow

The root agent.run span owns one trace ID. Every nested operation receives the root span ID through parent_span_id.

  • Locate the existing run_agent() function in flight_recorder.py.
  • Replace the first part of that function with this code:
def run_agent(scenario_name: str, recorder: FlightRecorder | NullRecorder, repaired: bool = False, run_number: int = 1) -> dict:
    scenario = SCENARIOS[scenario_name]
    trace_id = recorder.new_trace()
    started = perf_counter()
    simulated_total = 0
    try:
        with recorder.span("agent.run", trace_id, attributes={"scenario": scenario_name, "run_number": run_number}) as root:
            try:
                with recorder.span("route_issue", trace_id, root["span_id"]) as route_span:
                    selected_tool = select_tool(scenario["category"], repaired)
                    route_span["attributes"]["selected_tool"] = selected_tool
                    route_span["simulated_latency_ms"] = scenario["latency"][0]
                    simulated_total += scenario["latency"][0]
                with recorder.span("retrieve_context", trace_id, root["span_id"]) as retrieval_span:
                    context = retrieve_context(scenario_name)
                    retrieval_span["attributes"]["context_chars"] = len(context)
                    retrieval_span["simulated_latency_ms"] = scenario["latency"][1]
                    simulated_total += scenario["latency"][1]

How does correlation begin?

  • trace_id remains constant for the complete agent run.
  • root["span_id"] becomes the parent ID for each nested operation.
  • route_issue records the selected tool as a diagnostic attribute.
  • retrieve_context records the number of retrieved context characters.
  • Paste the following continuation directly after the first part of run_agent():
                attempt = 1
                while True:
                    try:
                        with recorder.span("call_tool", trace_id, root["span_id"], {"selected_tool": selected_tool, "attempt": attempt}) as tool_span:
                            tool_span["simulated_latency_ms"] = scenario["latency"][2]
                            simulated_total += scenario["latency"][2]
                            tool_result = call_tool(scenario_name, selected_tool, run_number, attempt)
                        break
                    except AgentFailure as exc:
                        if should_retry(type(exc).__name__, attempt, selected_tool):
                            attempt += 1
                            continue
                        raise
                with recorder.span("compose_recommendation", trace_id, root["span_id"]) as compose_span:
                    if "recommendation" not in tool_result:
                        raise MalformedToolResponse("recommendation is missing")
                    recommendation = tool_result["recommendation"]
                    compose_span["simulated_latency_ms"] = scenario["latency"][3]
                    simulated_total += scenario["latency"][3]
            finally:
                root["simulated_latency_ms"] = simulated_total
        result = {"status": "success", "scenario": scenario_name, "trace_id": trace_id, "recommendation": recommendation}
    except AgentFailure as exc:
        result = {"status": "error", "scenario": scenario_name, "trace_id": trace_id, "error_type": type(exc).__name__}
    result["simulated_latency_ms"] = simulated_total
    result["wall_duration_ms"] = round((perf_counter() - started) * 1000, 3)
    return result

How does the failure become diagnosable?

  • call_tool records the selected tool and the current attempt number.
  • compose_recommendation receives its own child span after a successful tool response.
  • simulated_total accumulates deterministic scenario latency for the root span.
  • perf_counter() measures local elapsed time independently from that scenario data.
  • Save flight_recorder.py.
  • Run the legacy happy path again to check the instrumented workflow:
python flight_recorder.py legacy password_reset

You should still see status=success with a recommendation. The instrumented workflow remains compatible with legacy mode.

Seeing a syntax error in run_agent?

Check that the second code chunk starts inside the inner try block. Confirm that return result aligns with the first statement inside the function.

help me repair the run_agent block structure

Reproduce the traced failure

A controlled reproduction needs a clean trace file and a way to parse its rows. Resetting first ensures that every row belongs to the refund run you are investigating.

  • Locate the existing read_spans() function in flight_recorder.py.
  • Replace that function with the code below:
def read_spans(path: Path = TRACE_PATH) -> list[dict]:
    path = Path(path)
    if not path.exists(): return []
    return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line]

What does the reader return?

read_spans() parses each nonempty JSONL line independently. Later code can group the resulting dictionaries by their shared trace ID.

  • Locate the existing reset_trace() function in flight_recorder.py.
  • Replace that function with the code below:
def reset_trace(path: Path = TRACE_PATH) -> None:
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text("", encoding="utf-8")

Why reset the trace file?

reset_trace() creates the trace directory when needed. It empties the trace file before a controlled reproduction.

  • Save flight_recorder.py.

Before you run the traced scenario, which missing details do you expect the recorder to add to the generic failure?

  • Reset the trace file and reproduce the unrepaired refund failure by running these commands:
python flight_recorder.py reset
python flight_recorder.py trace refund_route

What do these commands prove?

The first command clears traces/agent.jsonl. The second command runs the refund scenario through FlightRecorder.

The terminal now names IncorrectRoute and prints a trace ID. The trace file preserves the operation-level evidence behind that result.

  • Switch back to the Explorer sidebar in VS Code.
  • Select traces/agent.jsonl.
  • Confirm that every row has the same 32-character trace_id.
  • Confirm that every row has a unique 16-character span_id.
  • Compare each child parent_span_id with the agent.run span ID.
  • Find the route_issue row with knowledge_search stored under selected_tool.
  • Find the call_tool row with ERROR status.
  • Confirm that the failing row records IncorrectRoute with attempt 1.

That is the black box opened. You can identify the selected tool and the first failing operation without inspecting the routing source.

Why are there two latency fields?

simulated_latency_ms stores deterministic scenario data. It remains stable across workshop machines.

wall_duration_ms stores measured local elapsed time. It can vary because each computer executes the workflow at a different speed.

Missing trace rows?

Confirm that you saved flight_recorder.py before running the commands. Check that self.logger.info(event) remains inside the finally block.

help me diagnose missing or uncorrelated JSONL spans

✔️ Awesome, I've got everything!

Your refund trace now preserves the selected tool and the first failing operation. Double check that flight_recorder.py is saved before moving on.

ⓧ I'd like to double check the full code

Compare your completed flight_recorder.py with this cumulative version.

import argparse
import json
import logging
import math
import subprocess
import sys
from collections import defaultdict
from contextlib import contextmanager
from datetime import datetime, timezone
from pathlib import Path
from secrets import token_hex
from time import perf_counter
from retry_policy import should_retry

TRACE_PATH = Path("traces/agent.jsonl")
RECOVERY_SCENARIOS = ["password_reset", "shipping_delay", "warranty_question", "refund_route"]
EXPECTED_OPERATIONS = {"agent.run", "route_issue", "retrieve_context", "call_tool", "compose_recommendation"}
SCENARIOS = {
    "password_reset": {"category": "password", "expected_tool": "knowledge_search", "latency": [4, 8, 18, 5]},
    "shipping_delay": {"category": "shipping", "expected_tool": "order_lookup", "latency": [4, 7, 24, 5]},
    "warranty_question": {"category": "warranty", "expected_tool": "knowledge_search", "latency": [3, 9, 16, 6]},
    "refund_route": {"category": "refund", "expected_tool": "billing_lookup", "latency": [4, 8, 20, 5]},
    "timeout_probe": {"category": "password", "expected_tool": "knowledge_search", "latency": [4, 8, 50, 0], "failure": "timeout"},
    "malformed_probe": {"category": "warranty", "expected_tool": "knowledge_search", "latency": [3, 9, 15, 0], "failure": "malformed"},
    "flaky_timeout": {"category": "password", "expected_tool": "knowledge_search", "latency": [3, 6, 30, 4], "failure": "every_third_timeout"},
}

class AgentFailure(RuntimeError): pass
class IncorrectRoute(AgentFailure): pass
class ToolTimeout(AgentFailure): pass
class MalformedToolResponse(AgentFailure): pass

def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")

class JsonlFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = record.msg if isinstance(record.msg, dict) else {"message": record.getMessage()}
        return json.dumps(payload, sort_keys=True)

class FlightRecorder:
    def __init__(self, path: Path = TRACE_PATH):
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.logger = logging.getLogger(f"flight-recorder.{token_hex(4)}")
        self.logger.setLevel(logging.INFO)
        self.logger.propagate = False
        self.handler = logging.FileHandler(self.path, encoding="utf-8")
        self.handler.setFormatter(JsonlFormatter())
        self.logger.addHandler(self.handler)
    def new_trace(self) -> str:
        return token_hex(16)
    @contextmanager
    def span(self, operation: str, trace_id: str, parent_span_id: str | None = None, attributes: dict | None = None):
        span_id = token_hex(8)
        started_at = utc_now()
        started_counter = perf_counter()
        data = {"span_id": span_id, "attributes": dict(attributes or {}), "simulated_latency_ms": 0}
        event = {"trace_id": trace_id, "span_id": span_id, "parent_span_id": parent_span_id, "operation": operation, "started_at": started_at, "status": "OK", "attributes": data["attributes"], "error_type": None, "error_message": None}
        try:
            yield data
        except Exception as exc:
            event["status"] = "ERROR"
            event["error_type"] = type(exc).__name__
            event["error_message"] = str(exc)
            raise
        finally:
            event["ended_at"] = utc_now()
            event["wall_duration_ms"] = round((perf_counter() - started_counter) * 1000, 3)
            event["simulated_latency_ms"] = data["simulated_latency_ms"]
            self.logger.info(event)
    def close(self) -> None:
        self.logger.removeHandler(self.handler)
        self.handler.close()

class NullRecorder:
    def new_trace(self) -> str:
        return token_hex(16)
    @contextmanager
    def span(self, operation: str, trace_id: str, parent_span_id: str | None = None, attributes: dict | None = None):
        yield {"span_id": token_hex(8), "attributes": dict(attributes or {}), "simulated_latency_ms": 0}

def select_tool(category: str, repaired: bool) -> str:
    buggy_routes = {"password": "knowledge_search", "shipping": "order_lookup", "warranty": "knowledge_search", "refund": "knowledge_search"}
    repaired_routes = {**buggy_routes, "refund": "billing_lookup"}
    return (repaired_routes if repaired else buggy_routes)[category]

def retrieve_context(scenario_name: str) -> str:
    return f"support context for {scenario_name}"

def call_tool(scenario_name: str, selected_tool: str, run_number: int, attempt: int) -> dict:
    scenario = SCENARIOS[scenario_name]
    if selected_tool != scenario["expected_tool"]:
        raise IncorrectRoute(f"expected {scenario['expected_tool']}, got {selected_tool}")
    failure = scenario.get("failure")
    if failure == "timeout": raise ToolTimeout("simulated tool deadline exceeded")
    if failure == "malformed": raise MalformedToolResponse("simulated response has no recommendation")
    if failure == "every_third_timeout" and run_number % 3 == 0 and attempt == 1:
        raise ToolTimeout("simulated every-third-run timeout")
    return {"recommendation": f"Use {selected_tool} evidence for {scenario_name}."}

def run_agent(scenario_name: str, recorder: FlightRecorder | NullRecorder, repaired: bool = False, run_number: int = 1) -> dict:
    scenario = SCENARIOS[scenario_name]
    trace_id = recorder.new_trace()
    started = perf_counter()
    simulated_total = 0
    try:
        with recorder.span("agent.run", trace_id, attributes={"scenario": scenario_name, "run_number": run_number}) as root:
            try:
                with recorder.span("route_issue", trace_id, root["span_id"]) as route_span:
                    selected_tool = select_tool(scenario["category"], repaired)
                    route_span["attributes"]["selected_tool"] = selected_tool
                    route_span["simulated_latency_ms"] = scenario["latency"][0]
                    simulated_total += scenario["latency"][0]
                with recorder.span("retrieve_context", trace_id, root["span_id"]) as retrieval_span:
                    context = retrieve_context(scenario_name)
                    retrieval_span["attributes"]["context_chars"] = len(context)
                    retrieval_span["simulated_latency_ms"] = scenario["latency"][1]
                    simulated_total += scenario["latency"][1]
                attempt = 1
                while True:
                    try:
                        with recorder.span("call_tool", trace_id, root["span_id"], {"selected_tool": selected_tool, "attempt": attempt}) as tool_span:
                            tool_span["simulated_latency_ms"] = scenario["latency"][2]
                            simulated_total += scenario["latency"][2]
                            tool_result = call_tool(scenario_name, selected_tool, run_number, attempt)
                        break
                    except AgentFailure as exc:
                        if should_retry(type(exc).__name__, attempt, selected_tool):
                            attempt += 1
                            continue
                        raise
                with recorder.span("compose_recommendation", trace_id, root["span_id"]) as compose_span:
                    if "recommendation" not in tool_result:
                        raise MalformedToolResponse("recommendation is missing")
                    recommendation = tool_result["recommendation"]
                    compose_span["simulated_latency_ms"] = scenario["latency"][3]
                    simulated_total += scenario["latency"][3]
            finally:
                root["simulated_latency_ms"] = simulated_total
        result = {"status": "success", "scenario": scenario_name, "trace_id": trace_id, "recommendation": recommendation}
    except AgentFailure as exc:
        result = {"status": "error", "scenario": scenario_name, "trace_id": trace_id, "error_type": type(exc).__name__}
    result["simulated_latency_ms"] = simulated_total
    result["wall_duration_ms"] = round((perf_counter() - started) * 1000, 3)
    return result

def read_spans(path: Path = TRACE_PATH) -> list[dict]:
    path = Path(path)
    if not path.exists(): return []
    return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line]

def reset_trace(path: Path = TRACE_PATH) -> None:
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text("", encoding="utf-8")

def run_suite(recorder, repaired: bool) -> list[dict]:
    return [run_agent(name, recorder, repaired=repaired) for name in RECOVERY_SCENARIOS]

def nearest_rank_p95(values: list[float]) -> float:
    ordered = sorted(values)
    return ordered[max(0, math.ceil(0.95 * len(ordered)) - 1)]

def format_latency(results: list[dict], key: str) -> tuple[float, float]:
    values = [float(result[key]) for result in results]
    return round(sum(values) / len(values), 3), round(nearest_rank_p95(values), 3)

def group_spans(records: list[dict]) -> dict[str, list[dict]]:
    grouped = defaultdict(list)
    for record in records: grouped[record["trace_id"]].append(record)
    return grouped

def pytest_summary() -> str:
    completed = subprocess.run([sys.executable, "-m", "pytest", "-q"], capture_output=True, text=True, check=False)
    lines = [line.strip() for line in completed.stdout.splitlines() if line.strip()]
    summary = next((line for line in reversed(lines) if "passed" in line), "no summary")
    return summary if completed.returncode == 0 else f"FAILED: {summary}"

def generate_report(report_path: Path = Path("observability-report.md")) -> str:
    before_results = run_suite(NullRecorder(), repaired=False)
    reset_trace()
    recorder = FlightRecorder()
    after_results = run_suite(recorder, repaired=True)
    timeout_result = run_agent("timeout_probe", recorder, repaired=True)
    recorder.close()
    grouped = group_spans(read_spans())
    scenario_traces = {}
    for spans in grouped.values():
        root = next(span for span in spans if span["operation"] == "agent.run")
        scenario_traces[root["attributes"]["scenario"]] = spans
    expected_total = len(RECOVERY_SCENARIOS) * len(EXPECTED_OPERATIONS)
    present_total = sum(len({span["operation"] for span in scenario_traces[name]} & EXPECTED_OPERATIONS) for name in RECOVERY_SCENARIOS)
    missing_after = expected_total - present_total
    completeness_after = round(100 * present_total / expected_total)
    orphaned = 0
    for spans in grouped.values():
        span_ids = {span["span_id"] for span in spans}
        orphaned += sum(1 for span in spans if span["parent_span_id"] is not None and span["parent_span_id"] not in span_ids)
    attribution_after = any(span["operation"] == "call_tool" and span["status"] == "ERROR" and span["error_type"] == "ToolTimeout" for span in scenario_traces["timeout_probe"])
    before_success = round(100 * sum(result["status"] == "success" for result in before_results) / len(before_results))
    after_success = round(100 * sum(result["status"] == "success" for result in after_results) / len(after_results))
    before_sim_avg, before_sim_p95 = format_latency(before_results, "simulated_latency_ms")
    after_sim_avg, after_sim_p95 = format_latency(after_results, "simulated_latency_ms")
    before_wall_avg, before_wall_p95 = format_latency(before_results, "wall_duration_ms")
    after_wall_avg, after_wall_p95 = format_latency(after_results, "wall_duration_ms")
    tests = pytest_summary()
    report = f"""# AI Agent Flight Recorder Observability Report

The scenario latency fields are deterministic simulation data. Wall-clock fields are local measurements and can vary by machine. p95 uses the nearest-rank value across the four fixed recovery scenarios.

| Measure | Before | After |
|---|---:|---:|
| Trace completeness | 0% | {completeness_after}% |
| Correct failure attribution | 0% | {100 if attribution_after else 0}% |
| Missing expected spans | {expected_total} | {missing_after} |
| Orphaned spans | 0 | {orphaned} |
| Agent execution success rate | {before_success}% | {after_success}% |
| Average simulated latency | {before_sim_avg} ms | {after_sim_avg} ms |
| p95 simulated latency | {before_sim_p95} ms | {after_sim_p95} ms |
| Average measured wall latency | {before_wall_avg} ms | {after_wall_avg} ms |
| p95 measured wall latency | {before_wall_p95} ms | {after_wall_p95} ms |
| Regression tests | Not run | {tests} |

Timeout attribution probe: `{timeout_result['status']}` with `{timeout_result['error_type']}` attributed to `call_tool`.
"""
    report_path.write_text(report, encoding="utf-8")
    return tests

def print_result(result: dict, legacy: bool = False) -> None:
    if result["status"] == "success": print(f"status=success scenario={result['scenario']} trace_id={result['trace_id']} recommendation={result['recommendation']}")
    elif legacy: print("status=error message=Agent failed")
    else: print(f"status=error scenario={result['scenario']} error={result['error_type']} trace_id={result['trace_id']}")

def main() -> None:
    parser = argparse.ArgumentParser(description="Local AI agent flight recorder")
    subparsers = parser.add_subparsers(dest="command", required=True)
    for command in ("legacy", "trace"):
        command_parser = subparsers.add_parser(command)
        command_parser.add_argument("scenario", choices=sorted(SCENARIOS))
        if command == "trace": command_parser.add_argument("--repair", action="store_true")
    subparsers.add_parser("reset")
    subparsers.add_parser("report")
    secret_parser = subparsers.add_parser("secret")
    secret_parser.add_argument("--runs", type=int, default=6)
    args = parser.parse_args()
    if args.command == "reset":
        reset_trace(); print(f"reset={TRACE_PATH}")
    elif args.command == "legacy": print_result(run_agent(args.scenario, NullRecorder()), legacy=True)
    elif args.command == "trace":
        recorder = FlightRecorder(); result = run_agent(args.scenario, recorder, repaired=args.repair); recorder.close(); print_result(result)
    elif args.command == "report": print(f"report=observability-report.md tests={generate_report()}")
    elif args.command == "secret":
        reset_trace(); recorder = FlightRecorder()
        results = [run_agent("flaky_timeout", recorder, repaired=True, run_number=number) for number in range(1, args.runs + 1)]
        recorder.close(); successes = sum(result["status"] == "success" for result in results)
        print(f"runs={args.runs} successes={successes} failures={args.runs - successes} trace_file={TRACE_PATH}")

if __name__ == "__main__": main()

Your retry_policy.py file should still contain the starter no-retry policy.

def should_retry(error_type: str, attempt: int, operation: str) -> bool:
    """Starter policy. The Secret Mission replaces this no-retry behavior."""
    return False

Your pinned dependencies should still match this requirements.txt file.

fastapi[standard-no-fastapi-cloud-cli]==0.162.0
pytest==9.2.0

Your agent now leaves structured evidence behind every traced run. Next, you will turn those JSONL rows into a local timeline that makes the broken route faster to diagnose.

Inspect the Trace Timeline

Your JSONL records now contain enough evidence to explain the refund failure. Reading separate rows still makes the execution hierarchy slow to reconstruct.

This step turns those records into a local FastAPI timeline. You will use the timeline to identify the incorrect route without opening select_tool().

In this step, get ready to:
  • Build a local timeline that groups correlated spans by trace.
  • Launch the timeline at http://127.0.0.1:8000.
  • Diagnose the refund failure from trace evidence.
Build the timeline foundation

The viewer needs a web application plus a small style layer before it can present trace data. It also imports the existing trace reader so the browser always reflects traces/agent.jsonl.

  • Use the file sidebar in VS Code to create viewer.py beside flight_recorder.py.
  • Add the viewer foundation by pasting this code into viewer.py:
import html
import json
from collections import defaultdict
from fastapi import FastAPI
from fastapi.responses import HTMLResponse
from flight_recorder import read_spans

app = FastAPI(title="AI Agent Flight Recorder")
STYLE = """<style>body{font-family:system-ui,sans-serif;margin:2rem;background:#f5f7fb;color:#172033}header{display:flex;justify-content:space-between;align-items:center}.trace{background:white;border-radius:12px;padding:1rem;margin:1rem 0;box-shadow:0 2px 10px #dfe5ef}table{width:100%;border-collapse:collapse}th,td{text-align:left;padding:.55rem;border-bottom:1px solid #e6eaf0;vertical-align:top}.OK{color:#116329;font-weight:700}.ERROR{color:#b42318;font-weight:700}.child{padding-left:1.6rem}code{font-size:.78rem}button{padding:.5rem .8rem}</style>"""

What does the viewer foundation do?

  • The html module provides the escaping function that protects rendered trace content.
  • The json module converts attribute dictionaries into readable strings.
  • The defaultdict collection groups spans under their shared trace IDs.
  • The read_spans() function loads the records already stored in traces/agent.jsonl.
  • The STYLE string gives the timeline its CSS layout.
  • Save viewer.py.
  • Check the repository file list in VS Code.

You should see viewer.py beside flight_recorder.py.

Cannot find viewer.py?

  • Check that you created viewer.py inside the same repository as flight_recorder.py.
  • Confirm the filename ends with .py.

Ask for help with locating the new viewer file.

Render and launch the timeline

The render_timeline() function converts finished spans into trace cards. It sorts operations by their start time so the page reconstructs the execution order.

Trace content can contain characters with special meaning in HTML. Escaping that content keeps scenario data inside the table cells where it belongs.

  • Place the following code below the STYLE definition in viewer.py:
def render_timeline() -> str:
    grouped = defaultdict(list)
    for span in read_spans(): grouped[span["trace_id"]].append(span)
    cards = []
    for trace_id, spans in reversed(list(grouped.items())):
        spans.sort(key=lambda span: span["started_at"])
        root = next(span for span in spans if span["operation"] == "agent.run")
        rows = []
        for span in spans:
            operation_class = "child" if span["parent_span_id"] else "root"
            error = f"{span['error_type']}: {span['error_message']}" if span["error_type"] else ""
            attributes = json.dumps(span["attributes"], sort_keys=True)
            rows.append("<tr>" f"<td class='{operation_class}'>{html.escape(span['operation'])}</td>" f"<td class='{span['status']}'>{span['status']}</td>" f"<td>{span['simulated_latency_ms']}</td>" f"<td>{span['wall_duration_ms']}</td>" f"<td><code>{html.escape(attributes)}</code></td>" f"<td>{html.escape(error)}</td></tr>")
        cards.append("<section class='trace'>" f"<h2>{html.escape(root['attributes']['scenario'])}</h2><p><code>{html.escape(trace_id)}</code></p>" "<table><thead><tr><th>Operation</th><th>Status</th><th>Simulated ms</th><th>Wall ms</th><th>Attributes</th><th>Error</th></tr></thead><tbody>" + "".join(rows) + "</tbody></table></section>")
    content = "".join(cards) if cards else "<p>No traces yet. Run a traced scenario and refresh.</p>"
    return "<!doctype html><html><head><meta charset='utf-8'><meta name='viewport' content='width=device-width,initial-scale=1'><title>AI Agent Flight Recorder</title>" + STYLE + "</head><body><header><div><h1>AI Agent Flight Recorder</h1><p>Structured local execution timeline</p></div><button onclick='location.reload()'>Refresh</button></header>" + content + "</body></html>"

@app.get("/", response_class=HTMLResponse)
def timeline() -> str:
    return render_timeline()

How does the timeline work?

  • The shared trace_id places every operation from one agent run into the same card.
  • The started_at sort restores the operation order.
  • The parent_span_id field determines which operations receive the indented child style.
  • The html.escape() calls protect scenario names plus trace values before rendering them.
  • The / endpoint returns the complete timeline through HTMLResponse.
  • Save viewer.py.

Before you start the viewer, which operation do you expect the timeline to mark as the first failure?

  • Start the local timeline server from the existing project environment by running:
fastapi dev viewer.py

What does this command do?

The command loads viewer.py in development mode. The server exposes the timeline at http://127.0.0.1:8000.

  • Open http://127.0.0.1:8000 in your browser.

You should see an AI Agent Flight Recorder page with a refund_route card. Its table shows the operation order plus status, simulated timing, wall timing, attributes, and error evidence.

That is the raw refund trace turned into a readable execution timeline.

Timeline does not load?

  • Return to the terminal that uses your existing local Python environment.
  • Confirm no other local server is using port 8000.
  • Compare viewer.py with the complete file below for missing quotes or parentheses.

Ask for help with debugging the local FastAPI timeline.

✔️ Awesome, I've got everything!

Your timeline is running. Keep the server terminal active while you inspect the refund trace.

ⓧ I'd like to double check the full code

Compare your viewer.py with this complete version.

import html
import json
from collections import defaultdict
from fastapi import FastAPI
from fastapi.responses import HTMLResponse
from flight_recorder import read_spans

app = FastAPI(title="AI Agent Flight Recorder")
STYLE = """<style>body{font-family:system-ui,sans-serif;margin:2rem;background:#f5f7fb;color:#172033}header{display:flex;justify-content:space-between;align-items:center}.trace{background:white;border-radius:12px;padding:1rem;margin:1rem 0;box-shadow:0 2px 10px #dfe5ef}table{width:100%;border-collapse:collapse}th,td{text-align:left;padding:.55rem;border-bottom:1px solid #e6eaf0;vertical-align:top}.OK{color:#116329;font-weight:700}.ERROR{color:#b42318;font-weight:700}.child{padding-left:1.6rem}code{font-size:.78rem}button{padding:.5rem .8rem}</style>"""

def render_timeline() -> str:
    grouped = defaultdict(list)
    for span in read_spans(): grouped[span["trace_id"]].append(span)
    cards = []
    for trace_id, spans in reversed(list(grouped.items())):
        spans.sort(key=lambda span: span["started_at"])
        root = next(span for span in spans if span["operation"] == "agent.run")
        rows = []
        for span in spans:
            operation_class = "child" if span["parent_span_id"] else "root"
            error = f"{span['error_type']}: {span['error_message']}" if span["error_type"] else ""
            attributes = json.dumps(span["attributes"], sort_keys=True)
            rows.append("<tr>" f"<td class='{operation_class}'>{html.escape(span['operation'])}</td>" f"<td class='{span['status']}'>{span['status']}</td>" f"<td>{span['simulated_latency_ms']}</td>" f"<td>{span['wall_duration_ms']}</td>" f"<td><code>{html.escape(attributes)}</code></td>" f"<td>{html.escape(error)}</td></tr>")
        cards.append("<section class='trace'>" f"<h2>{html.escape(root['attributes']['scenario'])}</h2><p><code>{html.escape(trace_id)}</code></p>" "<table><thead><tr><th>Operation</th><th>Status</th><th>Simulated ms</th><th>Wall ms</th><th>Attributes</th><th>Error</th></tr></thead><tbody>" + "".join(rows) + "</tbody></table></section>")
    content = "".join(cards) if cards else "<p>No traces yet. Run a traced scenario and refresh.</p>"
    return "<!doctype html><html><head><meta charset='utf-8'><meta name='viewport' content='width=device-width,initial-scale=1'><title>AI Agent Flight Recorder</title>" + STYLE + "</head><body><header><div><h1>AI Agent Flight Recorder</h1><p>Structured local execution timeline</p></div><button onclick='location.reload()'>Refresh</button></header>" + content + "</body></html>"

@app.get("/", response_class=HTMLResponse)
def timeline() -> str:
    return render_timeline()
Diagnose the refund trace

A useful timeline identifies the decision that led to a failure. It also separates deterministic scenario data from measurements that change across machines.

Before you inspect the rows, which field do you expect will expose the wrong route?

  • Find the refund_route trace card.
  • Inspect the route_issue row.
  • Read the selected_tool attribute.
  • Locate the first child operation with an ERROR status.
  • Compare the simulated timing column with the wall timing column.

You should see that route_issue selected knowledge_search even though the refund scenario expected billing_lookup.

The first failing child operation is call_tool. Its error field reports IncorrectRoute.

The simulated_latency_ms values come from fixed scenario data. The wall_duration_ms values measure local execution and can vary by machine.

Your timeline now exposes the route decision plus the exact failing operation. Next, you will repair the route and prove the improvement with regression tests plus an observability report.

Repair the Route and Prove Observability

Your timeline has exposed the refund defect from trace evidence. You know the wrong tool was selected before the failure reached call_tool.

A plausible diagnosis needs repeatable proof. In this step, you will exercise the repaired route before locking the trace contract into tests and generating a before-and-after scorecard.

In this step, get ready to:
  • Run the corrected refund route and verify the successful trace.
  • Build six regression tests for the observability contract.
  • Generate the scorecard and document the simulation boundary.
Prove the repaired route

The select_tool() function keeps the original route table beside a corrected route table. The --repair option selects the corrected mapping while preserving the broken version for comparison.

  • In flight_recorder.py, locate select_tool().
  • Confirm the function contains this repaired refund mapping:
def select_tool(category: str, repaired: bool) -> str:
    buggy_routes = {"password": "knowledge_search", "shipping": "order_lookup", "warranty": "knowledge_search", "refund": "knowledge_search"}
    repaired_routes = {**buggy_routes, "refund": "billing_lookup"}
    return (repaired_routes if repaired else buggy_routes)[category]

How does the repair stay measurable?

  • The buggy_routes dictionary preserves the failure you diagnosed.
  • The repaired_routes dictionary changes only the refund mapping.
  • The repaired argument lets the same scenario prove both states.

Before you run the repaired scenario, do you expect the trace to stop at call_tool or continue through recommendation composition?

  • Return to the terminal you used for trace commands while leaving the local server running.
  • Run the repaired refund scenario with tracing enabled by running:
python flight_recorder.py trace refund_route --repair

You should see a successful result for refund_route. The recommendation should reference billing_lookup.

  • Return to the browser tab containing the FastAPI timeline.
  • Refresh the timeline page.

You should see a new refund trace with every operation marked OK. The route attributes should show selected_tool as billing_lookup.

Still seeing the broken route?

Check that your command includes --repair. Refresh the browser after the command finishes so the latest trace appears.

Confirm that repaired_routes maps refund to billing_lookup.

help me diagnose why my repaired refund trace still selects knowledge_search

Build the regression suite

A regression test turns each observability promise into an executable contract. Your pytest suite will cover completeness, identity, attribution, hierarchy, recovery, and timing semantics.

  • Create tests/test_observability.py from the file tree in VS Code.
  • Add the imports and shared traced-run helper by copying:
from flight_recorder import FlightRecorder, read_spans, run_agent

def traced_run(tmp_path, scenario: str, repaired: bool = False):
    path = tmp_path / "agent.jsonl"; recorder = FlightRecorder(path)
    result = run_agent(scenario, recorder, repaired=repaired); recorder.close()
    return result, read_spans(path)

What does the helper provide?

  • The tmp_path fixture gives each test its own trace file.
  • The helper runs one scenario through a real FlightRecorder.
  • The return value keeps the agent result beside its recorded spans.
  • Save tests/test_observability.py.
  • Confirm test_observability.py is visible inside the tests folder.

Cannot find the test file?

Confirm the file path is tests/test_observability.py. A file created beside the tests folder will not be collected from the intended location.

help me place my pytest file in the correct repository folder

  • Add the successful trace completeness test below traced_run() by copying:
def test_happy_trace_is_complete(tmp_path):
    result, spans = traced_run(tmp_path, "password_reset")
    assert result["status"] == "success"
    assert {span["operation"] for span in spans} == {"agent.run", "route_issue", "retrieve_context", "call_tool", "compose_recommendation"}
    assert all(span["status"] == "OK" for span in spans)

What does completeness mean?

This test requires the happy path to include all five expected operations. It also requires every recorded span to finish successfully.

  • Save tests/test_observability.py.
  • Verify the first trace contract by running:
python -m pytest -q

You should see one passing test. Your recorder now has an executable completeness contract.

Is the first test failing?

Check the indentation inside test_happy_trace_is_complete(). Confirm the expected operation names match the snippet exactly.

help me debug the happy trace completeness test

  • Add the trace and span identifier test below the first test by copying:
def test_trace_and_span_ids_have_expected_shape(tmp_path):
    _, spans = traced_run(tmp_path, "password_reset")
    assert len({span["trace_id"] for span in spans}) == 1
    assert all(len(span["trace_id"]) == 32 for span in spans)
    assert all(len(span["span_id"]) == 16 for span in spans)
    assert len({span["span_id"] for span in spans}) == len(spans)

What identity contract is enforced?

Every span in one run must share a single 32-character trace_id. Each operation must receive a unique 16-character span_id.

  • Save tests/test_observability.py.
  • Verify both trace contracts by running:
python -m pytest -q

You should see two passing tests. Trace correlation and span uniqueness are now protected.

Is the identifier test failing?

Check that new_trace() produces one trace identifier per run. Check that span() creates a new span identifier for each operation.

help me debug the trace and span identifier test

  • Add the failure attribution test below the identifier test by copying:
def test_failure_is_attributed_to_call_tool(tmp_path):
    result, spans = traced_run(tmp_path, "refund_route")
    failures = [span for span in spans if span["status"] == "ERROR" and span["operation"] != "agent.run"]
    assert result["error_type"] == "IncorrectRoute"
    assert [(span["operation"], span["error_type"]) for span in failures] == [("call_tool", "IncorrectRoute")]

Why test the first failing operation?

The root span reports the overall failure. This test requires the actionable cause to remain attributed to call_tool as IncorrectRoute.

  • Save tests/test_observability.py.
  • Verify failure attribution by running:
python -m pytest -q

You should see three passing tests. The suite can now distinguish a generic failed run from its operation-level cause.

Is attribution attached to the wrong span?

Confirm that the unrepaired refund_route scenario is used in this test. The failure list should exclude the root agent.run span.

help me debug IncorrectRoute attribution in my JSONL spans

  • Add the parent integrity test below the attribution test by copying:
def test_trace_has_no_orphaned_children(tmp_path):
    _, spans = traced_run(tmp_path, "refund_route"); span_ids = {span["span_id"] for span in spans}
    assert all(span["parent_span_id"] is None or span["parent_span_id"] in span_ids for span in spans)

What is an orphaned span?

A child becomes orphaned when its parent_span_id points to a span missing from the same trace. This test protects the hierarchy used by the timeline.

  • Save tests/test_observability.py.
  • Verify parent integrity by running:
python -m pytest -q

You should see four passing tests. Every recorded child now has a verifiable parent.

Did the parent integrity test fail?

Inspect the child spans for a parent_span_id value that is absent from the trace's span identifiers. Each operation created inside agent.run should reference the root span.

help me find an orphaned child span in my trace

  • Add the repaired route recovery test below the parent integrity test by copying:
def test_repaired_route_recovers_refund_scenario(tmp_path):
    result, spans = traced_run(tmp_path, "refund_route", repaired=True)
    route = next(span for span in spans if span["operation"] == "route_issue")
    assert result["status"] == "success"
    assert route["attributes"]["selected_tool"] == "billing_lookup"
    assert all(span["status"] == "OK" for span in spans)

What makes recovery provable?

A successful result alone could hide an incorrect route. This test also checks the route_issue evidence and every span status.

  • Save tests/test_observability.py.
  • Verify the repaired route contract by running:
python -m pytest -q

You should see five passing tests. The refund repair is now guarded against regression.

Is the repaired route test failing?

Confirm that repaired=True reaches run_agent(). Confirm the corrected refund mapping selects billing_lookup.

help me debug the repaired refund route test

  • Add the latency semantics test below the recovery test by copying:
def test_simulated_and_wall_latency_are_separate(tmp_path):
    result, spans = traced_run(tmp_path, "password_reset")
    assert result["simulated_latency_ms"] == 35
    assert result["wall_duration_ms"] >= 0
    assert all("simulated_latency_ms" in span and "wall_duration_ms" in span for span in spans)

Why protect both latency fields?

The scenario's simulated latency remains deterministic at 35 milliseconds. The measured wall duration remains machine-dependent while preserving a separate field.

Before you run the complete suite, do you expect all six contracts to agree with the traces produced by the recorder?

  • Save tests/test_observability.py.
  • Run the complete regression suite by running:
python -m pytest -q

You should see six passing tests. That is the observability contract secured across success, failure, hierarchy, repair, identity, and latency.

Are fewer than six tests passing?

Confirm every test function begins with test_. Compare the failing assertion with the corresponding trace field before changing recorder behavior.

help me diagnose my failing observability regression suite

✔️ Awesome, I've got everything!

Your six-test suite passes. Save tests/test_observability.py before continuing.

ⓧ I'd like to double check the full code

Compare your file with this complete regression suite.

from flight_recorder import FlightRecorder, read_spans, run_agent

def traced_run(tmp_path, scenario: str, repaired: bool = False):
    path = tmp_path / "agent.jsonl"; recorder = FlightRecorder(path)
    result = run_agent(scenario, recorder, repaired=repaired); recorder.close()
    return result, read_spans(path)

def test_happy_trace_is_complete(tmp_path):
    result, spans = traced_run(tmp_path, "password_reset")
    assert result["status"] == "success"
    assert {span["operation"] for span in spans} == {"agent.run", "route_issue", "retrieve_context", "call_tool", "compose_recommendation"}
    assert all(span["status"] == "OK" for span in spans)

def test_trace_and_span_ids_have_expected_shape(tmp_path):
    _, spans = traced_run(tmp_path, "password_reset")
    assert len({span["trace_id"] for span in spans}) == 1
    assert all(len(span["trace_id"]) == 32 for span in spans)
    assert all(len(span["span_id"]) == 16 for span in spans)
    assert len({span["span_id"] for span in spans}) == len(spans)

def test_failure_is_attributed_to_call_tool(tmp_path):
    result, spans = traced_run(tmp_path, "refund_route")
    failures = [span for span in spans if span["status"] == "ERROR" and span["operation"] != "agent.run"]
    assert result["error_type"] == "IncorrectRoute"
    assert [(span["operation"], span["error_type"]) for span in failures] == [("call_tool", "IncorrectRoute")]

def test_trace_has_no_orphaned_children(tmp_path):
    _, spans = traced_run(tmp_path, "refund_route"); span_ids = {span["span_id"] for span in spans}
    assert all(span["parent_span_id"] is None or span["parent_span_id"] in span_ids for span in spans)

def test_repaired_route_recovers_refund_scenario(tmp_path):
    result, spans = traced_run(tmp_path, "refund_route", repaired=True)
    route = next(span for span in spans if span["operation"] == "route_issue")
    assert result["status"] == "success"
    assert route["attributes"]["selected_tool"] == "billing_lookup"
    assert all(span["status"] == "OK" for span in spans)

def test_simulated_and_wall_latency_are_separate(tmp_path):
    result, spans = traced_run(tmp_path, "password_reset")
    assert result["simulated_latency_ms"] == 35
    assert result["wall_duration_ms"] >= 0
    assert all("simulated_latency_ms" in span and "wall_duration_ms" in span for span in spans)
Generate the scorecard and documentation

The test suite proves individual contracts. The report converts those results into a Markdown scorecard that compares the opaque agent with the observable repaired workflow.

  • Create README.md beside flight_recorder.py in the repository file tree.
  • Document the local workflow and simulation boundary by copying:
# AI Agent Flight Recorder

A local, deterministic observability lab for a simulated support agent. It records operations as correlated JSONL spans, renders a FastAPI timeline, reproduces failures, and evaluates reliability with pytest plus a before-and-after scorecard.

## Engineering lesson

An agent that produces an answer is not necessarily observable or debuggable. Useful traces preserve correlation, hierarchy, timing semantics, status, and failure attribution.

## Run locally

1. Create and activate `.venv` using the workshop instructions.
2. Run `python -m pip install -r requirements.txt`.
3. Run `python flight_recorder.py trace password_reset`.
4. Run `fastapi dev viewer.py` and open `http://127.0.0.1:8000`.
5. Run `python -m pytest -q`.
6. Run `python flight_recorder.py report`.

## Portfolio evidence

- `traces/agent.jsonl`: correlated execution evidence
- Local timeline: operation order, route decision, status, error, and latency
- `observability-report.md`: before-and-after metrics and test results
- `tests/`: trace contract and recovery regression coverage

## Simulation boundary

Agent responses, tools, errors, and `simulated_latency_ms` values are deterministic educational simulations. `wall_duration_ms` is measured locally and can vary by machine. The Secret Mission retries only the read-only `knowledge_search` operation, never an operation that could create duplicate side effects. This teaches the trace model directly. It is not an OTLP implementation or production observability backend.

Why document the simulation boundary?

The README separates deterministic scenario data from measurements collected on the learner's machine. That distinction keeps the project evidence technically honest.

The workflow also points reviewers to the trace file, timeline, scorecard, and regression suite.

  • Save README.md.
  • Confirm README.md appears beside flight_recorder.py in the repository file tree.

Is the README formatting broken?

Check that every heading begins on its own line. Keep a blank line between each heading and its content.

help me compare my README with the project documentation template

✔️ Awesome, I've got everything!

Your README now explains how to run the project and interpret its evidence.

ⓧ I'd like to double check the full code

Compare your documentation with this complete README.md.

# AI Agent Flight Recorder

A local, deterministic observability lab for a simulated support agent. It records operations as correlated JSONL spans, renders a FastAPI timeline, reproduces failures, and evaluates reliability with pytest plus a before-and-after scorecard.

## Engineering lesson

An agent that produces an answer is not necessarily observable or debuggable. Useful traces preserve correlation, hierarchy, timing semantics, status, and failure attribution.

## Run locally

1. Create and activate `.venv` using the workshop instructions.
2. Run `python -m pip install -r requirements.txt`.
3. Run `python flight_recorder.py trace password_reset`.
4. Run `fastapi dev viewer.py` and open `http://127.0.0.1:8000`.
5. Run `python -m pytest -q`.
6. Run `python flight_recorder.py report`.

## Portfolio evidence

- `traces/agent.jsonl`: correlated execution evidence
- Local timeline: operation order, route decision, status, error, and latency
- `observability-report.md`: before-and-after metrics and test results
- `tests/`: trace contract and recovery regression coverage

## Simulation boundary

Agent responses, tools, errors, and `simulated_latency_ms` values are deterministic educational simulations. `wall_duration_ms` is measured locally and can vary by machine. The Secret Mission retries only the read-only `knowledge_search` operation, never an operation that could create duplicate side effects. This teaches the trace model directly. It is not an OTLP implementation or production observability backend.

Before you generate the report, do you expect the after-state scorecard to show complete traces and successful recovery across all four fixed scenarios?

  • Generate the observability report by running:
python flight_recorder.py report

You should see output naming observability-report.md and reporting six passing tests. The command also replaces the trace file with evidence from the repaired recovery suite and timeout probe.

How is the scorecard calculated?

The recovery suite runs four fixed scenarios before and after repair. The after-state traces are checked for expected operations, valid parents, failure attribution, and successful execution.

The p95 latency uses the nearest-rank value from those four scenarios. Simulated values remain fixed while measured wall values can differ across machines.

  • Open observability-report.md from the repository file tree.
  • Confirm the after-state trace completeness is 100%.
  • Confirm the after-state failure attribution is 100%.
  • Confirm the after-state missing expected span count is 0.
  • Confirm the after-state orphaned span count is 0.
  • Confirm the after-state execution success rate is 100%.
  • Confirm the report includes average and p95 simulated latency rows.
  • Confirm the report includes average and p95 measured wall latency rows.

You should also see the timeout probe attributed to call_tool as ToolTimeout. The regression summary should report six passing tests.

Does the report show weaker after-state results?

Run the six-test suite again before regenerating the report. Check the repaired refund trace if completeness or recovery is below the expected result.

help me diagnose unexpected values in observability-report.md

The report and tests now form one reviewable project snapshot. Use Git to preserve that evidence together.

  • Stage the project evidence and create a commit by running these commands:
git add flight_recorder.py retry_policy.py viewer.py requirements.txt tests/test_observability.py README.md observability-report.md
git commit -m "Prove agent observability"

What does this commit preserve?

The first command stages the recorder, viewer, tests, report, requirements, retry policy, and README. The second command records those staged files as one project snapshot.

You should see a commit summary listing the saved files. Your repository now preserves the repair beside the evidence that proves it.

Could not create the commit?

Confirm the terminal is inside the repository. If Git requests author details, configure the identity used for your local commits before retrying.

help me troubleshoot my Git commit without changing project files

You have repaired the incorrect route and proved the result through traces, tests, and measured reporting. The project can now explain both how a run succeeded and where a run failed.

Secret mission

Recover the Every-Third-Run Timeout

A timeout strikes every third retry-safe lookup. Use trace evidence to uncover the pattern, recover with one bounded retry, and preserve the failed attempt for diagnosis.

Clean Up Your Resources

Clean Up Your Resources

All execution stays on your machine. This project has no ongoing service costs.

Choose whether to leave the timeline running. You can stop it for later or remove the local project contents.

Resources you used:

  • The local FastAPI development process that serves the timeline at http://127.0.0.1:8000.
  • The .venv virtual environment that holds the pinned project dependencies.
  • The project repository open in VS Code. It contains your source code plus all generated evidence.

Keep everything running

Leave everything as it is while you inspect the trace timeline or extend the retry policy.

  • Leave the FastAPI process running so the timeline stays available at http://127.0.0.1:8000.
  • Keep the project artifacts in the open repository.
  • Keep .venv so the pinned dependencies remain available.

Pause - I'll come back to this later

Free port 8000 by ending the local timeline process. Your trace evidence and virtual environment stay ready for your return.

  • Switch back to the terminal panel where the FastAPI timeline process is running.
  • Press Ctrl+C to stop the process.
  • Return to the existing timeline browser tab.
  • Refresh the browser tab to confirm the local timeline no longer responds.

The terminal prompt returns after the process stops. Every project file remains in the repository.

Delete - I don't want to use this again

Remove the running process and every visible repository item when you want a clean local reset. Deletion is permanent for uncommitted local evidence.

  • Switch back to the terminal panel where the FastAPI timeline process is running.
  • Press Ctrl+C to stop the process.
  • Return to the Explorer sidebar in VS Code.
  • Select every top-level file and folder shown for the repository.
  • Right-click the selected items.
  • Choose Delete from the context menu.
  • Confirm the deletion when VS Code asks.
  • Confirm the Explorer sidebar no longer lists the project files or .venv.

Your machine is clear of the local flight recorder. No cloud resources or paid services remain.

Nice Work!

Nice Work!

Outstanding work! You transformed an opaque simulated support agent into a workflow whose failures can be reconstructed from correlated JSONL evidence.

You've learned how to:

  • Capture a complete agent execution as correlated spans with parent-child relationships. The resulting trace preserves execution hierarchy. It also preserves the diagnostic evidence required to reconstruct a failure.
  • Use a local FastAPI timeline to group operations by trace ID. Diagnose the incorrect refund route from call_tool failure evidence. Repair the selected operation to billing_lookup.
  • Protect the trace contract with deterministic pytest regression tests. Quantify the before-and-after improvement in observability-report.md.
  • Secret Mission: Apply a bounded retry to every-third-run ToolTimeout failures from the read-only knowledge_search operation. Preserve the failed first attempt under its original trace. Reject retries for unsafe operations.

Ready to quiz yourself?