Build an Incident Commander Agent

Introduction

30 Second Summary

When an online checkout suddenly fails, pressure rises with every rejected order. A rushed recovery can turn one failure into several.

In this project, you will build a CLI incident commander with LangGraph and FastMCP for a synthetic checkout outage. The workflow keeps a human in control through process restarts and model failures.

What You'll Build

You'll watch your terminal-based incident commander investigate a checkout outage before stopping at a human approval checkpoint for its simulated recovery action.

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

• A bounded investigation through MCP that shows the evidence behind every proposed action.
• A durable approval thread that survives a process restart before allowing simulated execution.
• Resilience metrics that expose Ollama-first routing, OpenAI fallback, Redis cache hits, and projected cost for 100 runs. A GitHub Actions gate blocks evaluation regressions.
• Secret Mission: Run a controlled Ollama outage drill that proves circuit opening, OpenAI fallback, and local recovery.

Are there any prerequisites?

You need Python 3.10 through 3.14, uv, VS Code, Colima, Ollama, OpenAI API access, and a GitHub repository. You should also be comfortable with local Docker services, Langfuse evaluators, the existing run_evals script, and GitHub Actions.

Before We Start

Before any hands-on work, define the Incident Commander's safety boundary for autonomous investigation of telemetry, logs, resources, runbooks, and prompts. The simulated restart through execute_remediation always requires explicit human Decision approval so the agent cannot turn evidence into operational action by itself.

Prepare the Local Runtime

A drifting dependency or unavailable service can look like a graph orchestration bug. That ambiguity makes later failures harder to diagnose.

A reproducible runtime gives every graph run the same package versions. Visible service checks expose infrastructure problems before they can hide inside agent behavior.

In this step, get ready to:

• Create a pinned Python project in VS Code.
• Run PostgreSQL plus Redis through Docker Compose.
• Prepare Ollama models plus an OpenAI API credential.

Create the pinned project files

The project manifest defines the supported Python range. It also pins every package used by the Incident Commander.

• Press Cmd+Space on macOS or the Windows key on Windows to open system search.
• Type VS Code into system search.
• Press Enter to open VS Code.

You will see the VS Code window. The editor is ready for your project directory.

• Click File in the top menu bar.
• Select Open Folder.
• Navigate to your Desktop in the folder picker.
• Create a folder named incident-commander with the folder picker's New Folder control.
• Select the incident-commander folder.
• Click Open.
• Confirm trust for the folder if VS Code displays a workspace trust request.

You will see incident-commander at the top of the Explorer sidebar. VS Code is now working inside the correct directory.

• Select the New Folder control in the Explorer sidebar.
• Enter incident_commander as the folder name.

You will see the incident_commander package directory in the Explorer sidebar.

• Select the incident_commander folder in the Explorer sidebar.
• Select the New File control.
• Enter __init__.py as the file name.

You will see __init__.py inside the package directory. This gives later module commands a stable package path.

• Add the package initializer by replacing the contents of incident_commander/__init__.py with this code:

"""Resilient incident commander."""

What does this initializer do?

The module docstring records the package's purpose. The initializer also makes the package visible as a concrete Python module.

• Save incident_commander/__init__.py.

The file remains visible under incident_commander with no unsaved-change indicator in its tab.

Is the package file missing?

• Check that the directory name uses an underscore in incident_commander.
• Check that __init__.py has two underscores on each side of init.

Help me fix the package path.

• Select the top-level incident-commander directory in the Explorer sidebar.
• Select the New File control.
• Enter pyproject.toml as the file name.
• Add the pinned project manifest by replacing the file contents with this code:

[project]
name = "incident-commander"
version = "0.1.0"
requires-python = ">=3.10,<3.15"
dependencies = [
  "circuitbreaker==2.1.3",
  "fastmcp==4.1.3",
  "langchain[mcp]==1.4.7",
  "langchain-ollama==1.1.0",
  "langchain-openai==1.7.2",
  "langgraph==1.2.16",
  "langgraph-checkpoint-postgres==3.2.0",
  "ollama==0.6.3",
  "psycopg[binary,pool]==3.3.6",
  "redisvl[ollama]==0.28.0",
]
[dependency-groups]
dev = ["pytest==9.1.1"]
[tool.pytest.ini_options]
testpaths = ["tests"]

What does this manifest control?

• The project table identifies the package plus its release.
• The requires-python value supports Python 3.10 through 3.14.
• The dependencies array pins each runtime package to an exact release.
• The development group pins the test runner used by later contract checks.

• Save pyproject.toml.

You will see pyproject.toml at the top level of incident-commander with no unsaved-change indicator.

Does the manifest show a problem?

• Check that every dependency value has matching double quotes.
• Check that the dependencies array has matching square brackets.
• Confirm the file is named pyproject.toml.

Help me inspect the manifest.

A lock file records the complete dependency resolution behind the manifest. Committing it keeps future environment builds aligned.

• Open VS Code's integrated terminal from the incident-commander workspace.
• Generate uv.lock by running this command:

uv lock

You will see dependency resolution activity in the terminal. The Explorer sidebar will list uv.lock beside pyproject.toml.

Was the lock file skipped?

• Confirm the terminal is inside the incident-commander directory.
• Confirm pyproject.toml was saved before the lock command ran.
• Check that your Python version falls within the manifest's supported range.

Help me generate the lock file.

The first synchronization may take a few minutes while packages download. A quiet stretch during this command is expected.

• Synchronize the project environment from uv.lock by running this command:

uv sync

You will see the environment finish synchronizing from the locked dependency set. The project now has the pinned packages available locally.

Did synchronization fail?

• Confirm uv.lock exists beside pyproject.toml.
• Check that the terminal remains inside incident-commander.
• Review the terminal output to identify which pinned package failed to install.

Help me diagnose synchronization.

• Commit uv.lock through your existing Git workflow.

Your repository history now contains the generated lock file. Future synchronizations can reproduce the same resolved environment.

Start the data services

The Compose configuration fixes each service image plus its localhost-only port. A named PostgreSQL volume preserves the approval checkpoints added later.

• Select the top-level incident-commander directory in the Explorer sidebar.
• Select the New File control.
• Enter compose.yaml as the file name.
• Add the local service configuration by replacing the file contents with this code:

services:
  postgres:
    image: postgres:18.6-alpine
    environment:
      POSTGRES_USER: incident
      POSTGRES_PASSWORD: incident
      POSTGRES_DB: incident
    ports: ["127.0.0.1:5432:5432"]
    volumes: ["incident-postgres:/var/lib/postgresql"]
  redis:
    image: redis:8.10.2-alpine
    ports: ["127.0.0.1:6379:6379"]
volumes:
  incident-postgres:

What does this configuration create?

• The postgres service uses postgres:18.6-alpine.
• The PostgreSQL service uses incident for its database credentials.
• The PostgreSQL port binds to 127.0.0.1:5432.
• The redis service uses redis:8.10.2-alpine on 127.0.0.1:6379.
• The incident-postgres volume preserves PostgreSQL data across container restarts.

• Save compose.yaml.

You will see compose.yaml beside pyproject.toml in the Explorer sidebar.

The first container start may take a few minutes while Colima downloads the pinned images. Terminal download activity means the services are being prepared.

• Start PostgreSQL plus Redis in the background by running this command:

docker compose up -d

You will see activity for the network plus the incident-postgres volume. You will also see both containers start.

Did a container fail to start?

• Confirm Colima is running before retrying the Compose command.
• Check that the terminal remains inside incident-commander.
• Check whether another local process already uses port 5432 or port 6379.

Help me start the containers.

Use this checkpoint to compare every hand-written project file. The generated uv.lock remains the dependency resolver's output.

✔️ Awesome, I've got everything!

Your three hand-written project files are complete. Confirm each file is saved before continuing.

ⓧ I'd like to double check the full code

[project]
name = "incident-commander"
version = "0.1.0"
requires-python = ">=3.10,<3.15"
dependencies = [
  "circuitbreaker==2.1.3",
  "fastmcp==4.1.3",
  "langchain[mcp]==1.4.7",
  "langchain-ollama==1.1.0",
  "langchain-openai==1.7.2",
  "langgraph==1.2.16",
  "langgraph-checkpoint-postgres==3.2.0",
  "ollama==0.6.3",
  "psycopg[binary,pool]==3.3.6",
  "redisvl[ollama]==0.28.0",
]
[dependency-groups]
dev = ["pytest==9.1.1"]
[tool.pytest.ini_options]
testpaths = ["tests"]

"""Resilient incident commander."""

services:
  postgres:
    image: postgres:18.6-alpine
    environment:
      POSTGRES_USER: incident
      POSTGRES_PASSWORD: incident
      POSTGRES_DB: incident
    ports: ["127.0.0.1:5432:5432"]
    volumes: ["incident-postgres:/var/lib/postgresql"]
  redis:
    image: redis:8.10.2-alpine
    ports: ["127.0.0.1:6379:6379"]
volumes:
  incident-postgres:

The uv.lock file is generated from this exact pyproject.toml manifest. It should already be committed to your repository.

Prepare model access and verify the runtime

The local chat model handles the default generation path. The embedding model supports the semantic cache introduced later.

Each model download may take a few minutes depending on your connection. Steady progress output means Ollama is still working.

• Download the local chat model by running this command:

ollama pull llama3.2:3b

You will see the model layers download. The command completes after Ollama stores llama3.2:3b locally.

Did the chat model download stop?

• Confirm the Ollama application is running.
• Check your network connection before retrying the pull.
• Keep the llama3.2:3b model tag unchanged.

Help me download the chat model.

• Download the embedding model by running this command:

ollama pull nomic-embed-text:v1.5

You will see another set of model layers download. The command completes after Ollama stores nomic-embed-text:v1.5 locally.

Did the embedding model download stop?

• Confirm the Ollama application remains running.
• Check your network connection before retrying the pull.
• Keep the nomic-embed-text:v1.5 model tag unchanged.

Help me download the embedding model.

Protect your API key

Credentials deserve extra care. Keep your OpenAI key in the current shell instead of a project file.

Exporting the key does not call OpenAI. Charges begin only when the application or evaluators call its API.

• Export OPENAI_API_KEY in the current shell through your existing secure credential workflow.
• Keep the credential out of pyproject.toml.
• Keep the credential out of compose.yaml.
• Continue using the same terminal session for the remaining checks.

Before you run the checks, which runtime layer do you expect each command to confirm?

• Confirm the pinned dependencies import successfully by running this command:

uv run python -c "import fastmcp, langgraph, redisvl; print('dependencies ready')"

You will see dependencies ready. This proves the synchronized Python environment can import the core packages.

• Confirm both data services are running by running this command:

docker compose ps

You will see the postgres plus redis services listed with running states. This proves both local data services are available.

• Confirm both local models are available by running this command:

ollama ls

You will see llama3.2:3b plus nomic-embed-text:v1.5 in the model list. This proves both model assets are stored locally.

Is one runtime check failing?

• Synchronize the environment again if the dependency import check fails.
• Restart the Compose services if either configured service is missing.
• Pull the missing model again if its exact tag is absent from the model list.

Help me isolate the failed runtime layer.

That is the local foundation complete. Your dependencies import successfully while both data services plus both local models are available.

Your runtime is pinned plus observable. Next, you will publish three bounded capability servers with FastMCP.

Publish MCP Capability Boundaries

Your synchronized runtime is ready. Now the incident agent needs strict Model Context Protocol boundaries for every capability it can reach.

A visible FastMCP response proves each contract before an agent loop obscures its source. In this step, you will publish three capability servers.

In this step, get ready to:

• Define read-only telemetry tools plus the current incident resource.
• Separate runbook guidance from simulated change execution.
• Verify every capability through in-memory clients.

Publish the telemetry boundary

A capability boundary controls which operations a client can discover. The telemetry server exposes evidence without permitting a change.

• Create mcp_servers.py inside the incident_commander folder with the file control above VS Code's Explorer tree.
• Add the telemetry boundary by pasting this code into mcp_servers.py:

import asyncio
import json
import sys
from fastmcp import Client, FastMCP

telemetry_mcp = FastMCP("Telemetry")
runbooks_mcp = FastMCP("Runbooks")
change_mcp = FastMCP("ChangeControl")
INCIDENT = {"service": "checkout-api", "symptom": "HTTP 500 after deployment", "error_rate": 18.4, "latency_p95_ms": 920}
READ_ONLY = {"readOnlyHint": True, "idempotentHint": True, "openWorldHint": False}

@telemetry_mcp.tool(annotations=READ_ONLY)
def get_service_health(service: str) -> dict:
    return {"service": service, "status": "degraded" if service == "checkout-api" else "unknown", "error_rate": INCIDENT["error_rate"]}

@telemetry_mcp.tool(annotations=READ_ONLY)
def query_logs(service: str, severity: str = "ERROR") -> list[dict]:
    return [] if service != "checkout-api" else [{"severity": severity, "message": "database connection pool exhausted"}, {"severity": severity, "message": "request failed with HTTP 500"}]

@telemetry_mcp.resource("incident://current", mime_type="application/json")
def current_incident() -> str:
    return json.dumps(INCIDENT)

What does this code do?

• The three server objects create independent namespaces for telemetry capabilities.
• The INCIDENT dictionary provides deterministic evidence for the synthetic checkout outage.
• The READ_ONLY annotations tell clients that telemetry operations do not mutate state.
• The incident://current resource publishes the complete incident snapshot as JSON.

• Save incident_commander/mcp_servers.py.
• Confirm mcp_servers.py is listed beneath incident_commander in the Explorer tree.

Good progress. Your first boundary now defines health checks plus log evidence without exposing a write operation.

Telemetry file missing or incomplete?

• Check that mcp_servers.py sits inside incident_commander.
• Compare the three telemetry function names with get_service_health, query_logs, plus current_incident.

Need help checking the telemetry boundary?

Separate runbooks from change control

Runbook access remains read-only because it supplies guidance. Change execution belongs to a separate server because it represents the operation that requires approval.

• Append the runbook tool plus resource plus prompt below current_incident() by pasting this code:

@runbooks_mcp.tool(annotations=READ_ONLY)
def search_runbook(symptom: str) -> dict:
    return {"match": "checkout pool exhaustion", "symptom": symptom, "action": "restart checkout-api after approval", "rollback": "restore previous deployment"}

@runbooks_mcp.resource("runbook://checkout-api", mime_type="application/json")
def checkout_runbook() -> str:
    return json.dumps({"diagnosis": "confirm pool exhaustion", "action": "restart checkout-api after approval", "validation": "error rate below 2 percent"})

@runbooks_mcp.prompt
def diagnose_incident(evidence: str) -> str:
    return "Use only this evidence. Propose one reversible action, risk, and validation check:\n" + evidence

How does the runbook boundary work?

• The search_runbook tool matches symptoms to a reversible response.
• The runbook://checkout-api resource publishes structured diagnosis plus action plus validation guidance.
• The diagnose_incident prompt turns supplied evidence into a constrained planning request.
• The shared READ_ONLY annotations keep search separate from execution.

• Save incident_commander/mcp_servers.py.
• Confirm the editor now contains search_runbook plus checkout_runbook plus diagnose_incident.

Runbook capability missing?

• Check that every decorator uses runbooks_mcp.
• Check that the resource URI is exactly runbook://checkout-api.
• Check that the prompt accepts an evidence argument.

Need help repairing the runbook capability?

The ChangeControl server carries the risky capability. Its annotations make the simulated write boundary visible to any client that inspects the tool.

• Append the simulated remediation tool below diagnose_incident() by pasting this code:

@change_mcp.tool(annotations={"readOnlyHint": False, "destructiveHint": True, "idempotentHint": True, "openWorldHint": False})
def execute_remediation(action: str) -> dict:
    return {"status": "simulated", "action": action, "changed_real_system": False}

Why mark a simulated tool as destructive?

The annotations describe operational intent. The tool represents a state-changing action even though this project keeps the result simulated.

The changed_real_system value remains False. This makes the safety boundary observable in the response.

• Save incident_commander/mcp_servers.py.
• Confirm execute_remediation returns status as simulated.

That separation is now explicit. Evidence plus guidance remain read-only while simulated execution sits behind its own capability server.

Change tool looks read-only?

• Check that readOnlyHint is False on execute_remediation.
• Check that destructiveHint is True.
• Check that the function belongs to change_mcp.

Need help checking the change annotations?

Smoke-test every boundary

An agent loop can hide which server produced a result. Direct in-memory clients expose each tool plus resource plus prompt before orchestration begins.

• Append the server registry plus smoke test below execute_remediation() by pasting this code:

SERVERS = {"telemetry": telemetry_mcp, "runbooks": runbooks_mcp, "change": change_mcp}

async def smoke() -> None:
    async with Client(telemetry_mcp) as c:
        print((await c.call_tool("get_service_health", {"service": "checkout-api"})).data)
        print((await c.read_resource("incident://current"))[0].text)
    async with Client(runbooks_mcp) as c:
        print((await c.call_tool("search_runbook", {"symptom": "HTTP 500"})).data)
        print((await c.get_prompt("diagnose_incident", {"evidence": "pool exhausted"})).messages)
    async with Client(change_mcp) as c:
        print((await c.call_tool("execute_remediation", {"action": "restart checkout-api"})).data)

What does the smoke test prove?

• Each Client connects directly to one in-memory server.
• The telemetry client calls a tool plus reads a resource.
• The runbook client calls a tool plus renders a prompt.
• The change client invokes only the simulated remediation tool.

• Save incident_commander/mcp_servers.py.
• Confirm the smoke function contains three separate Client contexts.

The final entry point chooses between the smoke test plus the three stdio server modes. This lets the same module act as a verifier or a capability process.

• Append the command entry point below smoke() by pasting this code:

def main() -> None:
    target = sys.argv[1] if len(sys.argv) == 2 else ""
    if target == "smoke":
        asyncio.run(smoke()); return
    if target not in SERVERS:
        raise SystemExit("usage: python -m incident_commander.mcp_servers telemetry|runbooks|change|smoke")
    SERVERS[target].run(show_banner=False)

if __name__ == "__main__":
    main()

How does the entry point route requests?

• The smoke target runs the in-memory verification path.
• The telemetry plus runbooks plus change targets start their selected servers through stdio.
• An unsupported target exits with the accepted choices.

• Save incident_commander/mcp_servers.py.

Before you run this, consider whether read-only evidence or the simulated change appears first.

• Exercise all three servers through in-memory clients by running this command:

uv run python -m incident_commander.mcp_servers smoke

What does this command verify?

The command runs the module's smoke path inside the synchronized environment. It calls every capability without starting a subprocess transport.

You will see degraded checkout health first. The following outputs show the incident resource plus runbook match plus rendered prompt plus simulated change result.

The final result reports changed_real_system as False. Your capability boundaries work without touching a real service.

Smoke test did not complete?

• Check that every code chunk is saved in incident_commander/mcp_servers.py.
• Check that SERVERS maps telemetry plus runbooks plus change to their matching server objects.
• Check that the smoke target reaches asyncio.run(smoke()).

Need help diagnosing the smoke test?

✔️ Awesome, I've got everything!

Your three capability servers are saved. The in-memory smoke test now proves their contracts.

ⓧ I'd like to double check the full code

import asyncio
import json
import sys
from fastmcp import Client, FastMCP

telemetry_mcp = FastMCP("Telemetry")
runbooks_mcp = FastMCP("Runbooks")
change_mcp = FastMCP("ChangeControl")
INCIDENT = {"service": "checkout-api", "symptom": "HTTP 500 after deployment", "error_rate": 18.4, "latency_p95_ms": 920}
READ_ONLY = {"readOnlyHint": True, "idempotentHint": True, "openWorldHint": False}

@telemetry_mcp.tool(annotations=READ_ONLY)
def get_service_health(service: str) -> dict:
    return {"service": service, "status": "degraded" if service == "checkout-api" else "unknown", "error_rate": INCIDENT["error_rate"]}

@telemetry_mcp.tool(annotations=READ_ONLY)
def query_logs(service: str, severity: str = "ERROR") -> list[dict]:
    return [] if service != "checkout-api" else [{"severity": severity, "message": "database connection pool exhausted"}, {"severity": severity, "message": "request failed with HTTP 500"}]

@telemetry_mcp.resource("incident://current", mime_type="application/json")
def current_incident() -> str:
    return json.dumps(INCIDENT)

@runbooks_mcp.tool(annotations=READ_ONLY)
def search_runbook(symptom: str) -> dict:
    return {"match": "checkout pool exhaustion", "symptom": symptom, "action": "restart checkout-api after approval", "rollback": "restore previous deployment"}

@runbooks_mcp.resource("runbook://checkout-api", mime_type="application/json")
def checkout_runbook() -> str:
    return json.dumps({"diagnosis": "confirm pool exhaustion", "action": "restart checkout-api after approval", "validation": "error rate below 2 percent"})

@runbooks_mcp.prompt
def diagnose_incident(evidence: str) -> str:
    return "Use only this evidence. Propose one reversible action, risk, and validation check:\n" + evidence

@change_mcp.tool(annotations={"readOnlyHint": False, "destructiveHint": True, "idempotentHint": True, "openWorldHint": False})
def execute_remediation(action: str) -> dict:
    return {"status": "simulated", "action": action, "changed_real_system": False}

SERVERS = {"telemetry": telemetry_mcp, "runbooks": runbooks_mcp, "change": change_mcp}

async def smoke() -> None:
    async with Client(telemetry_mcp) as c:
        print((await c.call_tool("get_service_health", {"service": "checkout-api"})).data)
        print((await c.read_resource("incident://current"))[0].text)
    async with Client(runbooks_mcp) as c:
        print((await c.call_tool("search_runbook", {"symptom": "HTTP 500"})).data)
        print((await c.get_prompt("diagnose_incident", {"evidence": "pool exhausted"})).messages)
    async with Client(change_mcp) as c:
        print((await c.call_tool("execute_remediation", {"action": "restart checkout-api"})).data)

def main() -> None:
    target = sys.argv[1] if len(sys.argv) == 2 else ""
    if target == "smoke":
        asyncio.run(smoke()); return
    if target not in SERVERS:
        raise SystemExit("usage: python -m incident_commander.mcp_servers telemetry|runbooks|change|smoke")
    SERVERS[target].run(show_banner=False)

if __name__ == "__main__":
    main()

Your three MCP boundaries now publish visible evidence plus guidance plus simulated execution. Next, you will connect them to specialist agents with tightly scoped access.

Orchestrate the Specialist Agents

Your three MCP boundaries now expose telemetry evidence, runbook guidance, and a guarded change capability. The next challenge is coordinating those capabilities without giving every role unrestricted access.

A LangGraph workflow can investigate, plan, review, and pause for an operator. This step tests whether that approval pause can survive a planned process restart.

In this step, get ready to:

• Build the model layer used by the specialist roles.
• Orchestrate investigator, planner, and reviewer subgraphs with bounded MCP access.
• Reach the approval interrupt and test a restart with an in-memory checkpoint.

Build the model routing layer

Each specialist needs a consistent interface for invoking a model and recording usage. The router provides that interface while keeping provider details outside the graph.

Why keep models outside the graph?

The graph should describe operational control flow. A separate router keeps model selection, fallback behavior, caching, and cost measurements behind one boundary.

This separation lets every role use the same response and metric shape. The graph stays focused on incident decisions.

• In the Explorer sidebar of VS Code from earlier, create models.py inside the incident_commander folder.
• Add the imports, pricing constants, and usage helpers by pasting this first chunk:

import json
import os
from typing import Any
from circuitbreaker import circuit
from langchain_ollama import ChatOllama
from langchain_openai import ChatOpenAI
from redisvl.extensions.cache.llm import SemanticCache
from redisvl.utils.vectorize import OllamaTextVectorizer

IN_RATE, OUT_RATE = 0.25, 2.00

def usage(message: Any) -> dict:
    data = getattr(message, "usage_metadata", None) or {}
    return {"input_tokens": int(data.get("input_tokens", 0)), "output_tokens": int(data.get("output_tokens", 0))}

def cost(data: dict) -> float:
    return (data["input_tokens"] * IN_RATE + data["output_tokens"] * OUT_RATE) / 1_000_000

What do these helpers do?

• The imports bring in the local and remote chat models, the semantic cache, the vectorizer, and the circuit-breaker decorator.
• The usage() helper normalizes token metadata into input and output counts.
• The cost() helper converts remote token usage into a dollar amount.

• Save incident_commander/models.py.
• Confirm the helper layer imports by running this command in the terminal panel from earlier:

uv run python -c "import incident_commander.models; print('model helpers ready')"

What should I see?

You should see model helpers ready. This proves the new module can load every pinned dependency.

• Add the router constructor and its provider-specific protected methods below cost() by pasting this chunk:

class ResilientModelRouter:
    def __init__(self) -> None:
        self.local_model = ChatOllama(model="llama3.2:3b", temperature=0)
        self.remote_model = ChatOpenAI(model="gpt-5-mini", max_retries=0)
        self.force_openai = os.getenv("MODEL_MODE") == "openai"
        self.cache = None if os.getenv("DISABLE_SEMANTIC_CACHE") == "1" else SemanticCache(name="incident_cache_v1", redis_url=os.getenv("REDIS_URL", "redis://localhost:6379"), ttl=3600, distance_threshold=0.1, vectorizer=OllamaTextVectorizer(model="nomic-embed-text:v1.5"))

    @circuit(failure_threshold=2, recovery_timeout=15, expected_exception=Exception, name="ollama_generation")
    async def local(self, prompt: str) -> Any:
        return await self.local_model.ainvoke(prompt)

    @circuit(failure_threshold=2, recovery_timeout=15, expected_exception=Exception, name="openai_generation")
    async def remote(self, prompt: str) -> Any:
        return await self.remote_model.ainvoke(prompt)

    @circuit(failure_threshold=2, recovery_timeout=15, expected_exception=Exception, name="ollama_agent")
    async def local_agent(self, agent: Any, incident: str) -> dict:
        result = await agent.ainvoke({"messages": [{"role": "user", "content": incident}]})
        if not any(getattr(m, "type", "") == "tool" for m in result["messages"]):
            raise RuntimeError("no local tool evidence")
        return result

    @circuit(failure_threshold=2, recovery_timeout=15, expected_exception=Exception, name="openai_agent")
    async def remote_agent(self, agent: Any, incident: str) -> dict:
        result = await agent.ainvoke({"messages": [{"role": "user", "content": incident}]})
        if not any(getattr(m, "type", "") == "tool" for m in result["messages"]):
            raise RuntimeError("no remote tool evidence")
        return result

How is provider access bounded?

• The constructor creates an Ollama model for local work and an OpenAI API model for remote work.
• Each provider has separate generation and agent circuit protection, so failures stay isolated by operation.
• The agent methods reject results that contain no tool evidence, which stops an investigator from succeeding through unsupported guesses.
• The Redis semantic cache is configured here, but only the generation method decides what may enter it.

• Save incident_commander/models.py.
• Confirm the router can be constructed by running:

uv run python -c "from incident_commander.models import ResilientModelRouter; print(type(ResilientModelRouter()).__name__)"

What should I see?

You should see ResilientModelRouter. The providers and cache configuration are now available to the graph.

• Add the evidence-aware agent routing method inside ResilientModelRouter by pasting this chunk below remote_agent():

    async def run_agent(self, local_agent: Any, remote_agent: Any, incident: str) -> tuple[dict, dict]:
        provider, reason = "ollama", None
        try:
            if self.force_openai: raise RuntimeError("MODEL_MODE=openai")
            result = await self.local_agent(local_agent, incident)
        except Exception as error:
            provider, reason = "openai", type(error).__name__
            result = await self.remote_agent(remote_agent, incident)
        totals = {"input_tokens": 0, "output_tokens": 0}
        tools = []
        for message in result["messages"]:
            current = usage(message)
            totals = {key: totals[key] + current[key] for key in totals}
            if getattr(message, "type", "") == "tool": tools.append(str(getattr(message, "name", "unknown")))
        metric = {"phase": "investigator", "provider": provider, "cache_hit": False, "tools": tools, **totals, "cost_usd": cost(totals) if provider == "openai" else 0.0}
        if reason: metric["fallback_reason"] = reason
        return result, metric

What does agent routing record?

• The local investigator gets the first attempt unless MODEL_MODE selects the remote provider.
• A failed local attempt moves the same incident to the remote investigator.
• The returned metric records the provider, tools, token counts, cache state, cost, and fallback reason.

• Save incident_commander/models.py.
• Confirm the router exposes agent orchestration by running:

uv run python -c "from incident_commander.models import ResilientModelRouter; print(hasattr(ResilientModelRouter, 'run_agent'))"

What should I see?

You should see True. The investigator can now return evidence and an audit metric through one call.

• Finish ResilientModelRouter by adding role generation and cost summarization below run_agent():

    async def generate(self, role: str, prompt: str, prefer_remote: bool = False) -> tuple[str, dict]:
        key = f"incident-v1|{role}|{prompt}"
        if self.cache:
            hits = await self.cache.acheck(prompt=key)
            if hits:
                metric = json.loads(hits[0]["response"]); metric["cache_hit"] = True; metric["cost_usd"] = 0.0
                return metric["text"], metric
        provider, reason = "openai" if prefer_remote or self.force_openai else "ollama", None
        try:
            message = await (self.remote(prompt) if provider == "openai" else self.local(prompt))
        except Exception as error:
            provider, reason = "openai", type(error).__name__
            message = await self.remote(prompt)
        totals = usage(message)
        text = message.content if isinstance(message.content, str) else str(message.content)
        metric = {"phase": role, "provider": provider, "cache_hit": False, "tools": [], **totals, "cost_usd": cost(totals) if provider == "openai" else 0.0, "text": text}
        if reason: metric["fallback_reason"] = reason
        if self.cache: await self.cache.astore(prompt=key, response=json.dumps(metric), metadata={"role": role, "prompt_version": "v1"})
        return text, metric

def cost_summary(metrics: list[dict]) -> dict:
    spent = sum(float(m.get("cost_usd", 0)) for m in metrics)
    return {"run_cost_usd": round(spent, 8), "cost_per_100_runs_usd": round(spent * 100, 6), "cache_hits": sum(bool(m.get("cache_hit")) for m in metrics)}

What gets cached?

• The cache key includes a prompt version and role, which separates planner results from reviewer results.
• The method caches generated text from those roles. Investigator tool decisions and change execution never pass through this method.
• The summary converts all recorded remote spending into run cost and projected cost per 100 runs.

• Save incident_commander/models.py.
• Confirm the completed router exposes role generation by running:

uv run python -c "from incident_commander.models import ResilientModelRouter; print(hasattr(ResilientModelRouter, 'generate'))"

What should I see?

You should see True. Your specialist roles now share one model and metrics boundary.

Seeing an import or connection error?

• Check that models.py sits inside the existing incident_commander folder.
• Check that the imports and indentation match the reference tab below.
• Confirm the PostgreSQL and Redis containers from the previous step are still running if the cache reports a connection problem.

Ask for help with debugging the model router.

✔️ Awesome, I've got everything!

Great. Save incident_commander/models.py before building the graph.

ⓧ I'd like to double check the full code

import json
import os
from typing import Any
from circuitbreaker import circuit
from langchain_ollama import ChatOllama
from langchain_openai import ChatOpenAI
from redisvl.extensions.cache.llm import SemanticCache
from redisvl.utils.vectorize import OllamaTextVectorizer

IN_RATE, OUT_RATE = 0.25, 2.00

def usage(message: Any) -> dict:
    data = getattr(message, "usage_metadata", None) or {}
    return {"input_tokens": int(data.get("input_tokens", 0)), "output_tokens": int(data.get("output_tokens", 0))}

def cost(data: dict) -> float:
    return (data["input_tokens"] * IN_RATE + data["output_tokens"] * OUT_RATE) / 1_000_000

class ResilientModelRouter:
    def __init__(self) -> None:
        self.local_model = ChatOllama(model="llama3.2:3b", temperature=0)
        self.remote_model = ChatOpenAI(model="gpt-5-mini", max_retries=0)
        self.force_openai = os.getenv("MODEL_MODE") == "openai"
        self.cache = None if os.getenv("DISABLE_SEMANTIC_CACHE") == "1" else SemanticCache(name="incident_cache_v1", redis_url=os.getenv("REDIS_URL", "redis://localhost:6379"), ttl=3600, distance_threshold=0.1, vectorizer=OllamaTextVectorizer(model="nomic-embed-text:v1.5"))

    @circuit(failure_threshold=2, recovery_timeout=15, expected_exception=Exception, name="ollama_generation")
    async def local(self, prompt: str) -> Any:
        return await self.local_model.ainvoke(prompt)

    @circuit(failure_threshold=2, recovery_timeout=15, expected_exception=Exception, name="openai_generation")
    async def remote(self, prompt: str) -> Any:
        return await self.remote_model.ainvoke(prompt)

    @circuit(failure_threshold=2, recovery_timeout=15, expected_exception=Exception, name="ollama_agent")
    async def local_agent(self, agent: Any, incident: str) -> dict:
        result = await agent.ainvoke({"messages": [{"role": "user", "content": incident}]})
        if not any(getattr(m, "type", "") == "tool" for m in result["messages"]):
            raise RuntimeError("no local tool evidence")
        return result

    @circuit(failure_threshold=2, recovery_timeout=15, expected_exception=Exception, name="openai_agent")
    async def remote_agent(self, agent: Any, incident: str) -> dict:
        result = await agent.ainvoke({"messages": [{"role": "user", "content": incident}]})
        if not any(getattr(m, "type", "") == "tool" for m in result["messages"]):
            raise RuntimeError("no remote tool evidence")
        return result

    async def run_agent(self, local_agent: Any, remote_agent: Any, incident: str) -> tuple[dict, dict]:
        provider, reason = "ollama", None
        try:
            if self.force_openai: raise RuntimeError("MODEL_MODE=openai")
            result = await self.local_agent(local_agent, incident)
        except Exception as error:
            provider, reason = "openai", type(error).__name__
            result = await self.remote_agent(remote_agent, incident)
        totals = {"input_tokens": 0, "output_tokens": 0}
        tools = []
        for message in result["messages"]:
            current = usage(message)
            totals = {key: totals[key] + current[key] for key in totals}
            if getattr(message, "type", "") == "tool": tools.append(str(getattr(message, "name", "unknown")))
        metric = {"phase": "investigator", "provider": provider, "cache_hit": False, "tools": tools, **totals, "cost_usd": cost(totals) if provider == "openai" else 0.0}
        if reason: metric["fallback_reason"] = reason
        return result, metric

    async def generate(self, role: str, prompt: str, prefer_remote: bool = False) -> tuple[str, dict]:
        key = f"incident-v1|{role}|{prompt}"
        if self.cache:
            hits = await self.cache.acheck(prompt=key)
            if hits:
                metric = json.loads(hits[0]["response"]); metric["cache_hit"] = True; metric["cost_usd"] = 0.0
                return metric["text"], metric
        provider, reason = "openai" if prefer_remote or self.force_openai else "ollama", None
        try:
            message = await (self.remote(prompt) if provider == "openai" else self.local(prompt))
        except Exception as error:
            provider, reason = "openai", type(error).__name__
            message = await self.remote(prompt)
        totals = usage(message)
        text = message.content if isinstance(message.content, str) else str(message.content)
        metric = {"phase": role, "provider": provider, "cache_hit": False, "tools": [], **totals, "cost_usd": cost(totals) if provider == "openai" else 0.0, "text": text}
        if reason: metric["fallback_reason"] = reason
        if self.cache: await self.cache.astore(prompt=key, response=json.dumps(metric), metadata={"role": role, "prompt_version": "v1"})
        return text, metric

def cost_summary(metrics: list[dict]) -> dict:
    spent = sum(float(m.get("cost_usd", 0)) for m in metrics)
    return {"run_cost_usd": round(spent, 8), "cost_per_100_runs_usd": round(spent * 100, 6), "cache_hits": sum(bool(m.get("cache_hit")) for m in metrics)}

Orchestrate the specialist subgraphs

The parent graph controls the incident lifecycle. Each role subgraph receives only the state and capabilities required for its job.

Why use role subgraphs?

Subgraphs make each role boundary visible in the control flow. The investigator gathers evidence, the planner proposes an action, and the reviewer checks safety.

The parent graph owns the approval interrupt. That keeps execution authority outside every model role.

• Create graph.py inside the incident_commander folder from the Explorer sidebar.
• Add the graph imports, typed state, MCP configuration, and subgraph helper by pasting this chunk:

import sys
from typing import Any, TypedDict
from fastmcp import Client
from langchain.agents import create_agent
from langchain.mcp import MCPAdapter
from langgraph.graph import END, START, StateGraph
from langgraph.types import interrupt
from pydantic import BaseModel
from incident_commander.mcp_servers import runbooks_mcp, telemetry_mcp
from incident_commander.models import ResilientModelRouter

class IncidentState(TypedDict):
    incident: str; evidence: list[str]; plan: str; review: str
    approved: bool | None; execution: dict[str, Any]; metrics: list[dict]
class Decision(BaseModel):
    approved: bool
    note: str | None = None

CONFIG = {"mcpServers": {name: {"command": sys.executable, "args": ["-m", "incident_commander.mcp_servers", name]} for name in ("telemetry", "runbooks", "change")}}

def role_graph(node: Any) -> Any:
    graph = StateGraph(IncidentState); graph.add_node("act", node); graph.add_edge(START, "act"); graph.add_edge("act", END)
    return graph.compile()

What establishes the control contract?

• The IncidentState contract carries evidence, plans, reviews, approval, execution results, and metrics between roles.
• The Decision model gives the approval interrupt a typed response shape.
• The CONFIG mapping launches each existing FastMCP server over standard input and output.
• The role_graph() helper wraps one role node in its own start-to-end subgraph.

• Save incident_commander/graph.py.
• Confirm the graph contracts import by running:

uv run python -c "from incident_commander.graph import Decision, IncidentState; print('graph types ready')"

What should I see?

You should see graph types ready. The state and approval contracts are valid Python types.

• Start build_incident_graph() below role_graph() by adding the MCP tool boundary and investigator role:

async def build_incident_graph(router: ResilientModelRouter, checkpointer: Any) -> Any:
    async with MCPAdapter(CONFIG) as adapter: tools = await adapter.list_tools()
    read_tools = [t for t in tools if t.name.startswith(("telemetry_", "runbooks_"))]
    change_tool = next(t for t in tools if t.name == "change_execute_remediation")
    prompt = "Use telemetry health, logs, and runbook search before concluding. Return concise evidence."
    local_agent = create_agent(router.local_model, read_tools, system_prompt=prompt, name="local_investigator")
    remote_agent = create_agent(router.remote_model, read_tools, system_prompt=prompt, name="remote_investigator")

    async def investigate(state: IncidentState) -> dict:
        result, metric = await router.run_agent(local_agent, remote_agent, state["incident"])
        evidence = [str(m.content) for m in result["messages"] if getattr(m, "type", "") == "tool"]
        evidence.append(str(result["messages"][-1].content))
        return {"evidence": evidence, "metrics": state["metrics"] + [metric]}

How is investigator access restricted?

• The MCP adapter discovers tools from all three subprocess servers.
• The read_tools filter gives both investigator agents only telemetry and runbook capabilities.
• The change tool is stored separately for the execution node, so the investigator cannot invoke it.
• The investigator collects tool messages as evidence before adding the model's conclusion.

• Save incident_commander/graph.py.
• Confirm the investigator builder is available by running:

uv run python -c "from incident_commander.graph import build_incident_graph; print('investigator ready')"

What should I see?

You should see investigator ready. The graph can now construct an investigator with read-only MCP access.

• Add the planner and reviewer role nodes inside build_incident_graph() below investigate():

    async def plan(state: IncidentState) -> dict:
        async with Client(telemetry_mcp) as c: incident = (await c.read_resource("incident://current"))[0].text
        async with Client(runbooks_mcp) as c:
            runbook = (await c.read_resource("runbook://checkout-api"))[0].text
            rendered = await c.get_prompt("diagnose_incident", {"evidence": "\n".join(state["evidence"])})
        text, metric = await router.generate("planner", f"{state['incident']}\n{incident}\n{runbook}\n{rendered.messages}")
        return {"plan": text, "metrics": state["metrics"] + [metric]}

    async def review(state: IncidentState) -> dict:
        text, metric = await router.generate("reviewer", f"Check grounding, reversibility, and approval safety.\nEvidence:{state['evidence']}\nPlan:{state['plan']}", prefer_remote=True)
        return {"review": text, "metrics": state["metrics"] + [metric]}

What can each role see?

• The planner reads the current incident resource, the checkout runbook resource, and the rendered diagnostic prompt.
• The planner receives the incident plus gathered evidence. It does not receive the change tool.
• The reviewer receives only evidence and the proposed plan. Its prompt checks grounding, reversibility, and approval safety.

• Save incident_commander/graph.py.
• Confirm the expanded graph module still imports by running:

uv run python -c "import incident_commander.graph; print('specialist roles ready')"

What should I see?

You should see specialist roles ready. The planner and reviewer definitions now load with the investigator.

• Finish build_incident_graph() by adding the approval interrupt, guarded execution, and graph edges:

    def approval(state: IncidentState) -> dict:
        decision = interrupt({"question": "Approve simulated remediation?", "plan": state["plan"], "review": state["review"]}, response_schema=Decision)
        return {"approved": decision.approved}

    async def execute(state: IncidentState) -> dict:
        return {"execution": await change_tool.ainvoke({"action": state["plan"]})}

    graph = StateGraph(IncidentState)
    graph.add_node("investigator", role_graph(investigate)); graph.add_node("planner", role_graph(plan)); graph.add_node("reviewer", role_graph(review))
    graph.add_node("approval", approval); graph.add_node("execute", execute)
    graph.add_edge(START, "investigator"); graph.add_edge("investigator", "planner"); graph.add_edge("planner", "reviewer"); graph.add_edge("reviewer", "approval")
    graph.add_conditional_edges("approval", lambda s: "execute" if s["approved"] else END); graph.add_edge("execute", END)
    return graph.compile(checkpointer=checkpointer)

How does approval protect execution?

• The interrupt exposes the plan and review before accepting a typed Decision.
• A rejected decision routes directly to END.
• An approved decision reaches execute(), which is the only node holding the change tool.
• The supplied checkpointer stores graph state at the interrupt boundary.

• Save incident_commander/graph.py.
• Confirm the completed graph module loads by running:

uv run python -c "from incident_commander.graph import build_incident_graph; print('incident graph ready')"

What should I see?

You should see incident graph ready. The full specialist workflow is ready for a checkpointed run.

Graph module not loading?

• Check that every nested role function remains indented inside build_incident_graph().
• Check that the MCP tool prefixes remain telemetry_ and runbooks_.
• Compare the final edges with the reference tab if Python reports an undefined node.

Ask for help with troubleshooting the specialist graph.

✔️ Awesome, I've got everything!

Your investigator, planner, reviewer, approval, and execution nodes are connected. Save incident_commander/graph.py.

ⓧ I'd like to double check the full code

import sys
from typing import Any, TypedDict
from fastmcp import Client
from langchain.agents import create_agent
from langchain.mcp import MCPAdapter
from langgraph.graph import END, START, StateGraph
from langgraph.types import interrupt
from pydantic import BaseModel
from incident_commander.mcp_servers import runbooks_mcp, telemetry_mcp
from incident_commander.models import ResilientModelRouter

class IncidentState(TypedDict):
    incident: str; evidence: list[str]; plan: str; review: str
    approved: bool | None; execution: dict[str, Any]; metrics: list[dict]
class Decision(BaseModel):
    approved: bool
    note: str | None = None

CONFIG = {"mcpServers": {name: {"command": sys.executable, "args": ["-m", "incident_commander.mcp_servers", name]} for name in ("telemetry", "runbooks", "change")}}

def role_graph(node: Any) -> Any:
    graph = StateGraph(IncidentState); graph.add_node("act", node); graph.add_edge(START, "act"); graph.add_edge("act", END)
    return graph.compile()

async def build_incident_graph(router: ResilientModelRouter, checkpointer: Any) -> Any:
    async with MCPAdapter(CONFIG) as adapter: tools = await adapter.list_tools()
    read_tools = [t for t in tools if t.name.startswith(("telemetry_", "runbooks_"))]
    change_tool = next(t for t in tools if t.name == "change_execute_remediation")
    prompt = "Use telemetry health, logs, and runbook search before concluding. Return concise evidence."
    local_agent = create_agent(router.local_model, read_tools, system_prompt=prompt, name="local_investigator")
    remote_agent = create_agent(router.remote_model, read_tools, system_prompt=prompt, name="remote_investigator")

    async def investigate(state: IncidentState) -> dict:
        result, metric = await router.run_agent(local_agent, remote_agent, state["incident"])
        evidence = [str(m.content) for m in result["messages"] if getattr(m, "type", "") == "tool"]
        evidence.append(str(result["messages"][-1].content))
        return {"evidence": evidence, "metrics": state["metrics"] + [metric]}

    async def plan(state: IncidentState) -> dict:
        async with Client(telemetry_mcp) as c: incident = (await c.read_resource("incident://current"))[0].text
        async with Client(runbooks_mcp) as c:
            runbook = (await c.read_resource("runbook://checkout-api"))[0].text
            rendered = await c.get_prompt("diagnose_incident", {"evidence": "\n".join(state["evidence"])})
        text, metric = await router.generate("planner", f"{state['incident']}\n{incident}\n{runbook}\n{rendered.messages}")
        return {"plan": text, "metrics": state["metrics"] + [metric]}

    async def review(state: IncidentState) -> dict:
        text, metric = await router.generate("reviewer", f"Check grounding, reversibility, and approval safety.\nEvidence:{state['evidence']}\nPlan:{state['plan']}", prefer_remote=True)
        return {"review": text, "metrics": state["metrics"] + [metric]}

    def approval(state: IncidentState) -> dict:
        decision = interrupt({"question": "Approve simulated remediation?", "plan": state["plan"], "review": state["review"]}, response_schema=Decision)
        return {"approved": decision.approved}

    async def execute(state: IncidentState) -> dict:
        return {"execution": await change_tool.ainvoke({"action": state["plan"]})}

    graph = StateGraph(IncidentState)
    graph.add_node("investigator", role_graph(investigate)); graph.add_node("planner", role_graph(plan)); graph.add_node("reviewer", role_graph(review))
    graph.add_node("approval", approval); graph.add_node("execute", execute)
    graph.add_edge(START, "investigator"); graph.add_edge("investigator", "planner"); graph.add_edge("planner", "reviewer"); graph.add_edge("reviewer", "approval")
    graph.add_conditional_edges("approval", lambda s: "execute" if s["approved"] else END); graph.add_edge("execute", END)
    return graph.compile(checkpointer=checkpointer)

Test the process-local approval checkpoint

A checkpoint makes the approval pause resumable only when its storage remains available. The first test uses process memory so you can see what happens across a real restart.

• Create cli.py inside the incident_commander folder from the Explorer sidebar.
• Add the command parser and initial graph state by pasting this chunk:

import argparse, asyncio, json, os
from typing import Any
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
from incident_commander.graph import build_incident_graph
from incident_commander.models import ResilientModelRouter, cost_summary

def get_args() -> argparse.Namespace:
    root = argparse.ArgumentParser(); commands = root.add_subparsers(dest="command", required=True)
    start = commands.add_parser("start"); start.add_argument("--thread", required=True); start.add_argument("--incident", required=True)
    resume = commands.add_parser("resume"); resume.add_argument("--thread", required=True)
    choice = resume.add_mutually_exclusive_group(required=True); choice.add_argument("--approve", action="store_true"); choice.add_argument("--reject", action="store_true")
    resume.add_argument("--note", default=None); return root.parse_args()

def initial(incident: str) -> dict:
    return {"incident": incident, "evidence": [], "plan": "", "review": "", "approved": None, "execution": {}, "metrics": []}

What does the CLI accept?

• The start command requires a thread identifier and incident description.
• The resume command requires the same thread identifier plus an approval or rejection.
• The initial state starts with empty role outputs and no approval decision.

• Add the graph invocation function below initial() by pasting this chunk:

async def invoke(args: argparse.Namespace, saver: Any) -> None:
    graph = await build_incident_graph(ResilientModelRouter(), saver); config = {"configurable": {"thread_id": args.thread}}
    value = initial(args.incident) if args.command == "start" else Command(resume={"approved": bool(args.approve), "note": args.note})
    result = await graph.ainvoke(value, config)
    if "__interrupt__" in result:
        print(json.dumps(result["__interrupt__"][0].value, indent=2)); print("Resume with the same --thread value."); return
    print(json.dumps(result, indent=2, default=str)); print(json.dumps(cost_summary(result["metrics"]), indent=2))

How does a thread pause and resume?

• The thread identifier becomes part of the graph configuration used by the checkpointer.
• A start command sends a fresh incident state into the graph.
• A resume command sends a typed approval payload through Command.
• An interrupt prints the approval question, plan, and review before the process exits.

• Finish the process-local CLI below invoke() by pasting this chunk:

async def run() -> None:
    args = get_args(); os.environ.setdefault("LANGGRAPH_STRICT_MSGPACK", "true")
    if os.getenv("CHECKPOINT_BACKEND") == "memory": await invoke(args, InMemorySaver()); return

def main() -> None: asyncio.run(run())
if __name__ == "__main__": main()

What checkpoint does this use?

The CHECKPOINT_BACKEND setting selects InMemorySaver for this test. The saver belongs to the current Python process.

Strict message packing is enabled before graph invocation. The CLI then starts or resumes the selected thread.

• Save incident_commander/cli.py.

✔️ Awesome, I've got everything!

The process-local CLI is ready. Keep incident_commander/cli.py saved for the restart test.

ⓧ I'd like to double check the full code

import argparse, asyncio, json, os
from typing import Any
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
from incident_commander.graph import build_incident_graph
from incident_commander.models import ResilientModelRouter, cost_summary

def get_args() -> argparse.Namespace:
    root = argparse.ArgumentParser(); commands = root.add_subparsers(dest="command", required=True)
    start = commands.add_parser("start"); start.add_argument("--thread", required=True); start.add_argument("--incident", required=True)
    resume = commands.add_parser("resume"); resume.add_argument("--thread", required=True)
    choice = resume.add_mutually_exclusive_group(required=True); choice.add_argument("--approve", action="store_true"); choice.add_argument("--reject", action="store_true")
    resume.add_argument("--note", default=None); return root.parse_args()

def initial(incident: str) -> dict:
    return {"incident": incident, "evidence": [], "plan": "", "review": "", "approved": None, "execution": {}, "metrics": []}

async def invoke(args: argparse.Namespace, saver: Any) -> None:
    graph = await build_incident_graph(ResilientModelRouter(), saver); config = {"configurable": {"thread_id": args.thread}}
    value = initial(args.incident) if args.command == "start" else Command(resume={"approved": bool(args.approve), "note": args.note})
    result = await graph.ainvoke(value, config)
    if "__interrupt__" in result:
        print(json.dumps(result["__interrupt__"][0].value, indent=2)); print("Resume with the same --thread value."); return
    print(json.dumps(result, indent=2, default=str)); print(json.dumps(cost_summary(result["metrics"]), indent=2))

async def run() -> None:
    args = get_args(); os.environ.setdefault("LANGGRAPH_STRICT_MSGPACK", "true")
    if os.getenv("CHECKPOINT_BACKEND") == "memory": await invoke(args, InMemorySaver()); return

def main() -> None: asyncio.run(run())
if __name__ == "__main__": main()

Before you start the incident, what evidence do you expect the three specialist roles to produce before the approval question appears?

• Start the in-memory incident thread by running:

CHECKPOINT_BACKEND=memory uv run python -m incident_commander.cli start --thread incident-memory --incident "checkout-api returns HTTP 500 after a deployment"

What should I see?

You should see an approval payload containing Approve simulated remediation?, a plan, and a review. You should also see Resume with the same --thread value..

That is a real graph interrupt. No simulated change has executed because the workflow is waiting for a decision.

That is the first orchestration milestone reached: the investigator, planner, and reviewer completed their work without crossing the approval boundary.

Before you resume from a new command, do you think the new Python process can recover the paused thread from the previous process?

• Attempt to approve the same thread from a new process by running:

CHECKPOINT_BACKEND=memory uv run python -m incident_commander.cli resume --thread incident-memory --approve --note "Evidence supports a simulated restart"

This failure is expected

The resume attempt fails because the new process creates a new InMemorySaver. That saver has no checkpoint for incident-memory.

The approval pause worked inside one process. The restart exposed the missing durability required for an operator handoff.

Did the first command fail before approval?

• Confirm the existing Ollama service is running with the required local model available.
• Confirm PostgreSQL and Redis from the earlier runtime step are still running.
• Check that your existing terminal session still has the OpenAI credential exported before the reviewer runs.
• Compare models.py, graph.py, and cli.py with their full-code tabs.

Ask for help with debugging the approval run.

Your specialist graph now investigates, plans, reviews, and stops before execution. Next, you will replace the process-local checkpoint with durable storage so the same approval thread can survive a restart.

Make Approval Durable with PostgreSQL

The previous step proved that your LangGraph workflow can pause before a change. Its process-local checkpoint disappeared when Python exited.

A PostgreSQL checkpointer stores graph state outside the Python process. This converts approval into a durable operational control that another process can resume.

In this step, get ready to:

• Create the command-line interface for starting or resuming an incident thread.
• Initialize the PostgreSQL checkpoint tables with strict serialization.
• Resume an approved thread from a new Python process.

Create the durable CLI

The CLI gives operators separate commands for starting an incident or resuming a paused one. Its arguments preserve the thread identity and capture an explicit approval decision.

• Return to VS Code from earlier.
• Select the incident_commander folder in the Explorer sidebar.

The folder is now the destination for the new CLI module.

• Click the New File icon in the Explorer toolbar.
• Type cli.py into the filename field.
• Press Enter to create the file.

You will see cli.py beneath the incident_commander folder. The new editor tab is ready for the command parser.

• Add the command parser and initial graph state by copying this code into cli.py:

import argparse, asyncio, json, os
from typing import Any
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from langgraph.types import Command
from incident_commander.graph import build_incident_graph
from incident_commander.models import ResilientModelRouter, cost_summary

def get_args() -> argparse.Namespace:
    root = argparse.ArgumentParser(); commands = root.add_subparsers(dest="command", required=True)
    start = commands.add_parser("start"); start.add_argument("--thread", required=True); start.add_argument("--incident", required=True)
    resume = commands.add_parser("resume"); resume.add_argument("--thread", required=True)
    choice = resume.add_mutually_exclusive_group(required=True); choice.add_argument("--approve", action="store_true"); choice.add_argument("--reject", action="store_true")
    resume.add_argument("--note", default=None); return root.parse_args()

def initial(incident: str) -> dict:
    return {"incident": incident, "evidence": [], "plan": "", "review": "", "approved": None, "execution": {}, "metrics": []}

What does this code do?

• The imports connect the CLI to both checkpoint backends and the existing incident graph.
• The start command requires a thread identifier and an incident description.
• The resume command requires exactly one approval choice.
• The initial() function creates every field required by IncidentState.

• Save incident_commander/cli.py.
• Confirm that cli.py remains listed beneath the incident_commander folder in the Explorer sidebar.

Is the CLI file missing?

• Check that you selected the incident_commander folder before creating the file.
• Confirm that the filename is exactly cli.py.

Get help checking the first half of cli.py.

Persist the approval interrupt

Durability comes from compiling the parent graph with AsyncPostgresSaver. The same saver stores the interrupt and restores it when a later process supplies the matching thread identifier.

• Add durable graph invocation beneath initial() by copying this code:

async def invoke(args: argparse.Namespace, saver: Any) -> None:
    graph = await build_incident_graph(ResilientModelRouter(), saver); config = {"configurable": {"thread_id": args.thread}}
    value = initial(args.incident) if args.command == "start" else Command(resume={"approved": bool(args.approve), "note": args.note})
    result = await graph.ainvoke(value, config)
    if "__interrupt__" in result:
        print(json.dumps(result["__interrupt__"][0].value, indent=2)); print("Resume with the same --thread value."); return
    print(json.dumps(result, indent=2, default=str)); print(json.dumps(cost_summary(result["metrics"]), indent=2))

async def run() -> None:
    args = get_args(); os.environ.setdefault("LANGGRAPH_STRICT_MSGPACK", "true")
    if os.getenv("CHECKPOINT_BACKEND") == "memory": await invoke(args, InMemorySaver()); return
    url = os.getenv("DATABASE_URL", "postgresql://incident:incident@localhost:5432/incident?sslmode=disable")
    async with AsyncPostgresSaver.from_conn_string(url) as saver:
        await saver.setup(); await invoke(args, saver)

def main() -> None: asyncio.run(run())
if __name__ == "__main__": main()

How does persistence work?

• The thread_id becomes the stable key for the incident's checkpoint history.
• A resumed command passes approved and note through Command to the typed interrupt.
• The LANGGRAPH_STRICT_MSGPACK setting enables strict MessagePack serialization for checkpoint data.
• The setup() call initializes the checkpoint tables before the graph starts.
• The memory backend remains available for reproducing the process-local behavior from the previous step.

• Save incident_commander/cli.py.

API usage reminder

The reviewer uses OpenAI, so this run can consume API credits. The process stops after printing the approval request.

Before you run the incident, consider whether the process will return to your shell after saving the pause.

• Start the durable incident thread by running this command:

uv run python -m incident_commander.cli start --thread incident-postgres --incident "checkout-api returns HTTP 500 after a deployment"

You will see the approval question with the generated plan and review. You will also see Resume with the same --thread value. before the process exits.

The returned shell prompt proves that Python has exited. PostgreSQL now holds the paused incident-postgres thread.

Does the thread miss approval?

• Confirm that the PostgreSQL container from the local runtime is still running.
• Check that DATABASE_URL has not been changed to another database.
• Review the terminal output for the model phase that stopped before the interrupt.

Get help tracing the durable start command.

✔️ Awesome, I've got everything!

Your CLI now starts incidents with PostgreSQL persistence and preserves the approval interrupt after process exit.

ⓧ I'd like to double check the full code

import argparse, asyncio, json, os
from typing import Any
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from langgraph.types import Command
from incident_commander.graph import build_incident_graph
from incident_commander.models import ResilientModelRouter, cost_summary

def get_args() -> argparse.Namespace:
    root = argparse.ArgumentParser(); commands = root.add_subparsers(dest="command", required=True)
    start = commands.add_parser("start"); start.add_argument("--thread", required=True); start.add_argument("--incident", required=True)
    resume = commands.add_parser("resume"); resume.add_argument("--thread", required=True)
    choice = resume.add_mutually_exclusive_group(required=True); choice.add_argument("--approve", action="store_true"); choice.add_argument("--reject", action="store_true")
    resume.add_argument("--note", default=None); return root.parse_args()

def initial(incident: str) -> dict:
    return {"incident": incident, "evidence": [], "plan": "", "review": "", "approved": None, "execution": {}, "metrics": []}

async def invoke(args: argparse.Namespace, saver: Any) -> None:
    graph = await build_incident_graph(ResilientModelRouter(), saver); config = {"configurable": {"thread_id": args.thread}}
    value = initial(args.incident) if args.command == "start" else Command(resume={"approved": bool(args.approve), "note": args.note})
    result = await graph.ainvoke(value, config)
    if "__interrupt__" in result:
        print(json.dumps(result["__interrupt__"][0].value, indent=2)); print("Resume with the same --thread value."); return
    print(json.dumps(result, indent=2, default=str)); print(json.dumps(cost_summary(result["metrics"]), indent=2))

async def run() -> None:
    args = get_args(); os.environ.setdefault("LANGGRAPH_STRICT_MSGPACK", "true")
    if os.getenv("CHECKPOINT_BACKEND") == "memory": await invoke(args, InMemorySaver()); return
    url = os.getenv("DATABASE_URL", "postgresql://incident:incident@localhost:5432/incident?sslmode=disable")
    async with AsyncPostgresSaver.from_conn_string(url) as saver:
        await saver.setup(); await invoke(args, saver)

def main() -> None: asyncio.run(run())
if __name__ == "__main__": main()

Resume the same thread

The stored interrupt waits for a value matching Decision. Reusing incident-postgres lets a new Python process load the exact paused graph state.

Before you resume the thread, consider which graph path an explicit approval should unlock.

• Approve the paused incident by running this command:

uv run python -m incident_commander.cli resume --thread incident-postgres --approve --note "Evidence supports a simulated restart"

You will see "approved": true in the resumed state. The execution result includes "status": "simulated" and "changed_real_system": false.

That is the durable control boundary working. A new process recovered the paused thread and executed only the simulated remediation after explicit approval.

The existing graph routes rejection to END. Approval invokes only change_execute_remediation.

Does the resume command miss the checkpoint?

• Confirm that the resume command uses the exact thread value incident-postgres.
• Confirm that the start command reached the approval interrupt before its process exited.
• Confirm that the PostgreSQL container stayed available between the two commands.

Get help recovering the paused thread.

Your approval workflow now survives process restarts without widening the change boundary. Next, you will measure provider resilience, cache behavior, cost, and evaluation quality.

Prove Resilience and Evaluation Gates

Your approval workflow now survives a process restart. It also keeps every simulated change behind an explicit operator decision.

A durable agent still needs visible signals for provider failures, spend, cache behavior, and quality regressions. This step turns each concern into a measurable pass or fail result.

In this step, get ready to:

• Compare forced OpenAI routing with the default Ollama-first path.
• Create a golden dataset contract with measurable evaluation thresholds.
• Add contract tests and a GitHub Actions regression gate.

Compare provider and cache behavior

The model router uses separate circuit breakers for Ollama and OpenAI. Each breaker opens after 2 failures. A recovery probe becomes available after 15 seconds.

The RedisVL semantic cache stores versioned planner and reviewer responses in incident_cache_v1. Investigator tool decisions and simulated changes always run through their live paths.

OpenAI usage can cost money

The forced baseline calls OpenAI and can consume paid credits. Keep this comparison to one short incident thread.

The standard rate for gpt-5-mini is $0.25 per 1M input tokens. Output costs $2.00 per 1M tokens.

Before the forced run, which provider do you expect every model phase to report?

• Switch back to the terminal session from the durable approval step.
• Set MODEL_MODE to openai in that terminal session.
• Set DISABLE_SEMANTIC_CACHE to 1 in the same terminal session.
• Run the earlier CLI start pattern for thread incident-openai using incident checkout-api p95 latency increased after deployment.
• Resume thread incident-openai with the approval flag from the previous step.

You should see provider reported as openai across the model phases. The final summary includes run_cost_usd and cost_per_100_runs_usd.

How is the cost projection calculated?

Each OpenAI message contributes its input and output token counts. The router applies the configured rates to those totals.

The cost_per_100_runs_usd value multiplies the observed run cost by 100. It gives you a concrete planning estimate from the current execution.

The default route reveals a different pattern. The investigator and planner prefer local generation while the reviewer uses OpenAI for its safety review.

• Remove MODEL_MODE from the terminal session.
• Remove DISABLE_SEMANTIC_CACHE from the terminal session.
• Run the earlier CLI start pattern for thread incident-cache-a using incident checkout-api p95 latency increased after deployment.
• Resume thread incident-cache-a with the approval flag.

You should see Ollama handle the investigator and planner work. The reviewer should report OpenAI when it generates a fresh review.

Before the repeated run, which phases do you expect the cache to serve?

• Run the same incident through a new thread named incident-cache-b.
• Resume thread incident-cache-b with the approval flag.

You should see cache hits for matching planner and reviewer prompts. The investigator still gathers live tool evidence while cache_hits increases in the cost summary.

Missing the expected provider or cache metrics?

• Check that the forced baseline had MODEL_MODE set to openai.
• Check that both cache runs used identical incident text.
• Confirm the second cache run used a new thread ID.

Help me compare the provider and cache metrics from my Incident Commander runs.

Build evaluation contracts and gate

A golden dataset gives every evaluator the same incidents and expected behavior. The result contract then reduces each run to scores that a deterministic gate can check.

• Return to the project in VS Code from earlier.
• Create an evals folder inside the Incident Commander project with the Explorer sidebar's New Folder control.

You should see evals listed beside incident_commander in the Explorer sidebar.

• Add the three evaluation cases to evals/golden.jsonl by pasting this content:

{"id":"pool-exhaustion","incident":"checkout-api returns HTTP 500 after deployment","expected_tools":["telemetry_get_service_health","telemetry_query_logs","runbooks_search_runbook"],"expected_action":"restart checkout-api after approval"}
{"id":"latency-spike","incident":"checkout-api p95 latency increased after deployment","expected_tools":["telemetry_get_service_health","telemetry_query_logs","runbooks_search_runbook"],"expected_action":"restart checkout-api after approval"}
{"id":"unknown-service","incident":"catalog-api is unhealthy without supporting telemetry","expected_tools":["telemetry_get_service_health","telemetry_query_logs","runbooks_search_runbook"],"expected_action":"do not execute an unsupported change"}

What does this dataset define?

• The pool-exhaustion case expects health, log, and runbook evidence before a restart plan.
• The latency-spike case checks the same tool discipline against a different symptom.
• The unknown-service case expects the agent to avoid an unsupported change.

• Save evals/golden.jsonl.
• Confirm the editor shows exactly three JSON Lines records.

Does the golden dataset show JSON errors?

Keep each case on one complete line. Check that every property name and string retains its double quotes.

Help me find the invalid record in my golden JSON Lines dataset.

• Add the passing result fixture to evals/results.example.json by pasting this content:

{"cases":[{"id":"pool-exhaustion","faithfulness":0.92,"relevance":0.90,"tool_correctness":1.0},{"id":"latency-spike","faithfulness":0.88,"relevance":0.86,"tool_correctness":1.0},{"id":"unknown-service","faithfulness":0.86,"relevance":0.84,"tool_correctness":1.0}]}

What does the result contract measure?

• The faithfulness score represents grounding in available evidence.
• The relevance score represents how directly the response addresses the incident.
• The tool_correctness score checks whether the required tools were used correctly.

• Save evals/results.example.json.
• Confirm the file contains one cases array with three result objects.

Does the fixture fail to parse?

Check the closing braces around the cases array. Confirm every score remains a number without surrounding quotes.

Help me correct the JSON structure in my evaluation result fixture.

• Create incident_commander/eval_gate.py with the Explorer sidebar's New File control.
• Add the threshold checks by pasting this code:

import json, sys
from pathlib import Path
PER_CASE = {"faithfulness": .80, "relevance": .80, "tool_correctness": 1.0}
AGGREGATE = {"faithfulness": .85, "relevance": .85, "tool_correctness": 1.0}
def evaluate_gate(path: Path) -> list[str]:
    cases = json.loads(path.read_text())["cases"]; failures = []
    for case in cases:
        for metric, threshold in PER_CASE.items():
            if float(case[metric]) < threshold: failures.append(f"{case['id']} {metric} below {threshold}")
    for metric, threshold in AGGREGATE.items():
        average = sum(float(case[metric]) for case in cases) / len(cases)
        if average < threshold: failures.append(f"aggregate {metric}={average:.3f} below {threshold}")
    return failures
def main() -> None:
    failures = evaluate_gate(Path(sys.argv[1]))
    if failures:
        print("Evaluation gate failed:\n" + "\n".join(failures)); raise SystemExit(1)
    print("Evaluation gate passed")
if __name__ == "__main__": main()

How does the gate protect quality?

Every case needs faithfulness and relevance scores of at least 0.80. Every case also needs a tool correctness score of 1.0.

Aggregate faithfulness and relevance must reach 0.85. Aggregate tool correctness must remain at 1.0.

Any threshold failure produces a message and exits with status 1. That exit status becomes the CI failure signal.

• Save incident_commander/eval_gate.py.

Before you run the gate, do you expect the example scores to pass every threshold?

• Check the example fixture against the evaluation gate by running:

uv run python -m incident_commander.eval_gate evals/results.example.json

What does this check prove?

The command reads the shared result contract and applies every per-case threshold. It also calculates the three aggregate scores.

You should see Evaluation gate passed. A score below its threshold makes the command exit non-zero.

Does the evaluation gate fail unexpectedly?

• Confirm the command points to evals/results.example.json.
• Check that every case contains faithfulness, relevance, and tool_correctness.
• Compare the threshold dictionaries with the reference code below.

Help me diagnose why my passing evaluation fixture is failing the gate.

Double-check the complete evaluation contract files before adding automation.

✔️ Awesome, I've got everything!

Great. Confirm all three files are saved before continuing.

ⓧ I'd like to double check the full code

{"id":"pool-exhaustion","incident":"checkout-api returns HTTP 500 after deployment","expected_tools":["telemetry_get_service_health","telemetry_query_logs","runbooks_search_runbook"],"expected_action":"restart checkout-api after approval"}
{"id":"latency-spike","incident":"checkout-api p95 latency increased after deployment","expected_tools":["telemetry_get_service_health","telemetry_query_logs","runbooks_search_runbook"],"expected_action":"restart checkout-api after approval"}
{"id":"unknown-service","incident":"catalog-api is unhealthy without supporting telemetry","expected_tools":["telemetry_get_service_health","telemetry_query_logs","runbooks_search_runbook"],"expected_action":"do not execute an unsupported change"}

{"cases":[{"id":"pool-exhaustion","faithfulness":0.92,"relevance":0.90,"tool_correctness":1.0},{"id":"latency-spike","faithfulness":0.88,"relevance":0.86,"tool_correctness":1.0},{"id":"unknown-service","faithfulness":0.86,"relevance":0.84,"tool_correctness":1.0}]}

import json, sys
from pathlib import Path
PER_CASE = {"faithfulness": .80, "relevance": .80, "tool_correctness": 1.0}
AGGREGATE = {"faithfulness": .85, "relevance": .85, "tool_correctness": 1.0}
def evaluate_gate(path: Path) -> list[str]:
    cases = json.loads(path.read_text())["cases"]; failures = []
    for case in cases:
        for metric, threshold in PER_CASE.items():
            if float(case[metric]) < threshold: failures.append(f"{case['id']} {metric} below {threshold}")
    for metric, threshold in AGGREGATE.items():
        average = sum(float(case[metric]) for case in cases) / len(cases)
        if average < threshold: failures.append(f"aggregate {metric}={average:.3f} below {threshold}")
    return failures
def main() -> None:
    failures = evaluate_gate(Path(sys.argv[1]))
    if failures:
        print("Evaluation gate failed:\n" + "\n".join(failures)); raise SystemExit(1)
    print("Evaluation gate passed")
if __name__ == "__main__": main()

Add contract tests and CI gate

Local checks protect your current branch. A GitHub Actions workflow applies the same checks to pull requests and manual evaluation runs.

The workflow treats evals/results.json as the handoff between your existing evaluator and the deterministic gate. This isolates any evaluator-specific adjustment to one command.

• Create a tests folder inside the project with the Explorer sidebar's New Folder control.

You should see the new tests folder beside evals.

• Add the MCP and gate checks to tests/test_contracts.py by pasting this code:

import asyncio
from pathlib import Path
from fastmcp import Client
from incident_commander.eval_gate import evaluate_gate
from incident_commander.mcp_servers import change_mcp, telemetry_mcp

def test_contracts() -> None:
    async def run() -> None:
        async with Client(telemetry_mcp) as c:
            assert {t.name for t in await c.list_tools()} == {"get_service_health", "query_logs"}
        async with Client(change_mcp) as c:
            result = await c.call_tool("execute_remediation", {"action": "restart checkout-api"})
            assert result.data["changed_real_system"] is False
    asyncio.run(run())

def test_gate() -> None:
    assert evaluate_gate(Path("evals/results.example.json")) == []

What do these tests protect?

The contract test confirms that telemetry exposes only the two expected tools. It also proves that remediation reports changed_real_system as False.

The gate test runs the example scores through evaluate_gate(). An empty failure list proves that the fixture satisfies the contract.

• Save tests/test_contracts.py.

Before you run the tests, do you expect both the MCP boundary and evaluation fixture to pass?

• Run the contract tests through the locked project environment with:

uv run python -m pytest

What does this test run prove?

The test run loads the same pinned environment used by the application. It checks the MCP safety contract and the passing evaluation fixture.

You should see both tests pass. That confirms the change tool remains simulated and the example scores satisfy the gate.

Do the contract tests fail?

• Confirm tests/test_contracts.py is saved inside the project.
• Check that evals/results.example.json still contains all three passing cases.
• Confirm the change server still returns changed_real_system as False.

Help me troubleshoot the failing Incident Commander contract tests.

• Create a .github folder inside the project with the Explorer sidebar's New Folder control.

You should see .github at the top level of the project.

• Create a workflows folder inside .github with the same Explorer control.

You should now have the path .github/workflows in the Explorer sidebar.

• Add the regression workflow to .github/workflows/evals.yml by pasting this configuration:

name: Agent evaluation gate
on: [pull_request, workflow_dispatch]
permissions:
  contents: read
jobs:
  unit-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v7.0.1
      - uses: astral-sh/setup-uv@v10.3.2
        with: {python-version: "3.12", enable-cache: true}
      - run: uv sync --locked
      - run: uv run python -m pytest
  live-evals:
    if: ${{ github.event_name != 'pull_request' || github.event.pull_request.head.repo.fork == false }}
    runs-on: ubuntu-latest
    env:
      OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
      MODEL_MODE: openai
      DISABLE_SEMANTIC_CACHE: "1"
    steps:
      - uses: actions/checkout@v7.0.1
      - uses: astral-sh/setup-uv@v10.3.2
        with: {python-version: "3.12", enable-cache: true}
      - run: uv sync --locked
      - run: uv run python run_evals.py --dataset evals/golden.jsonl --output evals/results.json
      - run: uv run python -m incident_commander.eval_gate evals/results.json

How does the workflow enforce quality?

The unit-tests job installs the locked environment and runs the contract tests. This catches MCP boundary changes before live evaluation begins.

The live-evals job forces OpenAI routing and disables semantic caching. Your evaluator writes fresh scores to evals/results.json.

The final command passes those scores to the deterministic gate. Any reported regression makes the GitHub Actions job fail.

Live evaluations can spend credits

The reviewer, forced baseline, and model-based evaluators can call OpenAI. Keep the initial golden set at three cases while you validate the workflow.

Repository secrets stay unavailable to workflows triggered from forks. The workflow skips the live job for those pull requests.

• Save .github/workflows/evals.yml.

Before the final checks, do you expect the local contracts and example evaluation scores to pass together?

• Rerun the contract suite with:

uv run python -m pytest

Why rerun the tests?

This command mirrors the unit-test command in the workflow. A local pass confirms that CI receives the same test entry point.

You should see both contract tests pass. The result proves that the MCP boundaries and example fixture remain valid.

• Run the example scores through the same gate used by CI with:

uv run python -m incident_commander.eval_gate evals/results.example.json

Why run the gate again?

This command mirrors the final workflow check while using the stable example fixture. It verifies the handoff from result JSON to process exit status.

You should see Evaluation gate passed after the passing test output. Scores below the configured thresholds make this command return a non-zero status.

Does the final verification fail?

• Compare the local commands with the corresponding run entries in .github/workflows/evals.yml.
• Confirm the workflow points the evaluator output to evals/results.json.
• Align the evaluator command if your existing runner uses a different invocation while preserving the same result contract.

Help me troubleshoot my evaluation workflow and local regression gate.

Double-check the complete test and workflow files before finishing the step.

✔️ Awesome, I've got everything!

Great. Confirm both files are saved and both local checks pass.

ⓧ I'd like to double check the full code

import asyncio
from pathlib import Path
from fastmcp import Client
from incident_commander.eval_gate import evaluate_gate
from incident_commander.mcp_servers import change_mcp, telemetry_mcp

def test_contracts() -> None:
    async def run() -> None:
        async with Client(telemetry_mcp) as c:
            assert {t.name for t in await c.list_tools()} == {"get_service_health", "query_logs"}
        async with Client(change_mcp) as c:
            result = await c.call_tool("execute_remediation", {"action": "restart checkout-api"})
            assert result.data["changed_real_system"] is False
    asyncio.run(run())

def test_gate() -> None:
    assert evaluate_gate(Path("evals/results.example.json")) == []

name: Agent evaluation gate
on: [pull_request, workflow_dispatch]
permissions:
  contents: read
jobs:
  unit-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v7.0.1
      - uses: astral-sh/setup-uv@v10.3.2
        with: {python-version: "3.12", enable-cache: true}
      - run: uv sync --locked
      - run: uv run python -m pytest
  live-evals:
    if: ${{ github.event_name != 'pull_request' || github.event.pull_request.head.repo.fork == false }}
    runs-on: ubuntu-latest
    env:
      OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
      MODEL_MODE: openai
      DISABLE_SEMANTIC_CACHE: "1"
    steps:
      - uses: actions/checkout@v7.0.1
      - uses: astral-sh/setup-uv@v10.3.2
        with: {python-version: "3.12", enable-cache: true}
      - run: uv sync --locked
      - run: uv run python run_evals.py --dataset evals/golden.jsonl --output evals/results.json
      - run: uv run python -m incident_commander.eval_gate evals/results.json

That completes the reliability layer. Your Incident Commander now exposes provider failures, cache behavior, token usage, projected cost, and evaluation regressions as observable signals.

Secret mission

Run an Ollama Failure Drill

Create a controlled outage for the Incident Commander's preferred local model. You will capture OpenAI fallback metrics, prove that the local circuit opens after two failures, and confirm recovery after the 15-second timeout.

Clean Up Your Resources

Clean Up Your Resources

Your PostgreSQL, Redis, and Ollama services run locally, so idle local resources do not create a cloud bill. Calls to the OpenAI API can still cost money whenever the application or evaluation workflow sends a request.

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Cost warning

Keeping the local containers or models does not create an API usage charge. Future gpt-5-mini calls from the application or GitHub Actions workflow can consume OpenAI credits.

Pause the local services to free system memory. Delete the workflow access if you have finished using paid model calls.

Resources you used:

• The postgres container managed through Docker Compose.
• The redis container holding the semantic cache index.
• The incident-postgres volume holding durable checkpoint tables.
• The llama3.2:3b Ollama model.
• The nomic-embed-text:v1.5 Ollama embedding model.
• The local project repository containing uv.lock, the source files, the tests, and the uv-managed project environment.
• The .github/workflows/evals.yml evaluation workflow in your GitHub repository.
• The repository secret named OPENAI_API_KEY used by live evaluations.

Keep everything running

No action needed. Choose this if you are still testing incidents or demonstrating the approval workflow.

• Leave the postgres container running so existing incident threads remain resumable.
• Leave the redis container running so cached planner and reviewer responses remain available.
• Keep Ollama running so the router can continue using the local provider.
• Retain the GitHub Actions workflow if you still want evaluation regressions checked.
• Monitor your OpenAI credits whenever live evaluations or fallback requests run.

Pause - I'll come back to this later

Stop the running containers to free system memory. Your checkpoint volume, downloaded models, project files, workflow, and secret remain available.

• Return to the terminal in the repository folder that contains compose.yaml.
• Stop the PostgreSQL and Redis containers by running this command:

docker compose stop

What does this command pause?

Docker Compose stops the postgres and redis containers. The incident-postgres volume stays on disk, so your durable incident threads remain available.

• Confirm the terminal reports that the postgres service stopped.
• Confirm the terminal reports that the redis service stopped.

Containers not stopping?

• Check that the current folder contains compose.yaml.
• Confirm that the Docker runtime provided by Colima is available.

Help me diagnose why docker compose stop failed.

When you return, the same Compose configuration restores both services. Your saved checkpoint data remains in the named volume.

• Return to the repository folder that contains compose.yaml.
• Restart the PostgreSQL and Redis containers by running this command:

docker compose up -d

What does this command restore?

Docker Compose starts both services in the background. PostgreSQL reconnects to the existing incident-postgres volume, so durable incident threads remain available.

• Confirm the terminal reports that the postgres service started.
• Confirm the terminal reports that the redis service started.

Services not restarting?

• Check that Colima is providing an available Docker runtime.
• Check that ports 5432 and 6379 are available on your machine.

Help me diagnose why docker compose up failed.

Delete - I don't want to use this again

Remove the project resources when you have finished testing the Incident Commander. Deleting the checkpoint volume is permanent, so your saved incident threads cannot be resumed afterward.

• Return to the terminal in the repository folder that contains compose.yaml.
• Remove the PostgreSQL container, the Redis container, and the checkpoint volume by running this command:

docker compose down -v

What does this command remove?

Docker Compose removes the postgres and redis containers. It also deletes the incident-postgres volume declared in compose.yaml.

The durable checkpoints disappear with the volume. The semantic cache disappears with the Redis container.

• Confirm the terminal reports that the postgres service was removed.
• Confirm the terminal reports that the redis service was removed.
• Confirm the terminal reports that the incident-postgres volume was removed.

Resources not being removed?

• Check that the current folder contains the project copy of compose.yaml.
• Confirm that the Docker runtime provided by Colima is available.

Help me diagnose why docker compose down failed.

The two Ollama models can support other local projects. Remove them only when no other project depends on them.

• In Ollama, remove the local model tagged llama3.2:3b if no other project uses it.
• In Ollama, remove the local model tagged nomic-embed-text:v1.5 if no other project uses it.
• Check Ollama's local model list to confirm that both tags are absent.

The remote workflow can still start paid evaluations after local resources are gone. Remove its workflow file and repository secret to close that path.

• In the GitHub repository's file browser, select .github/workflows/evals.yml.
• Delete .github/workflows/evals.yml from the default branch.
• Remove OPENAI_API_KEY from the repository's Actions secret settings.
• Confirm the evaluation workflow no longer appears in the repository.
• Confirm the repository secret is no longer listed.

Your remaining local repository contains the source code, tests, generated lockfile, and project environment. Delete the whole repository folder only if you no longer need any of them.

• Close VS Code to release the local project files.
• Use Finder on macOS or File Explorer on Windows to locate the repository folder containing compose.yaml.
• Delete that repository folder from your machine.
• Confirm that compose.yaml and uv.lock are no longer present on your machine.

Nice Work!

Nice Work!

You did it! Your production-shaped Incident Commander now investigates a checkout outage through bounded capabilities while preserving operator control over every simulated change.

You've learned how to:

• Publish three bounded FastMCP capability servers that expose tools, resources, prompts, and safety annotations.
• Orchestrate investigator, planner, and reviewer roles with LangGraph. Resume durable operator approval from PostgreSQL after process exit.
• Measure Ollama-first routing, OpenAI fallback, Redis cache hits, and cost per 100 runs. Block evaluation regressions with a golden dataset in GitHub Actions.
• Secret Mission: Prove fallback, circuit opening, and recovery during a controlled Ollama outage.

Ready to quiz yourself?