Triage Incidents with LangGraph

Introduction

30 Second Summary

An alert can leave one on-call engineer chasing several clues at once. Delays grow when each clue must be checked one at a time.

In this project, you will build a LangGraph incident-triage CLI that sends logs, deploys, and metrics specialists to investigate in parallel. It saves completed work after a crash and publishes only team-approved fixes to an audited runbook.

What You'll Build

Run one fixture alert and watch three specialists trace its logs, deploys, and metrics before the CLI writes approved fixes and every team decision to runbooks/inc-001.md.

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

• A parallel specialist triage flow that routes INC-2041 to logs, deploys, and metrics investigators. You can watch their completion order vary before their typed findings merge into one diagnosis.
• A crash-safe recovery workflow that keeps completed logs and deploys findings. You can restart the CLI and see only the failed metrics investigation run again.
• A team-owned approval process that remains paused until each service owner decides. The published runbooks/inc-001.md contains approved actions plus a complete decision audit log.
• Secret Mission: Build status.py as a read-only board that shows saved findings, decisions, pending teams, next nodes, and runbook state without resuming the workflow.

Are there any prerequisites?

• Create or sign in to a Google Cloud account. Account creation has no minimum spend, but this project requires billing for usage-based model requests.
• Open Google Cloud Shell. It includes Git, Python, pip, and the Google Cloud CLI.
• Review the Python tutorial and LangGraph overview if you need a refresher. Experience with the durable workflow from Part 1 is recommended.

Before We Start

Before you begin, lock in the system you are building and why it matters. You will turn one alert into a crash-safe investigation with team-owned approvals.

Set Up the Starter Repo

Part 2 depends on Part 1's durable LangGraph workflow. Proving that baseline first keeps later orchestration bugs separate from credential or fixture problems.

You will prepare Google Cloud Shell for Vertex AI. You will install the starter repository before smoke-testing its approval pause.

In this step, get ready to:

• Select a Google Cloud project for Vertex AI.
• Install the starter repository inside an isolated Python environment.
• Prove the inherited durable workflow reaches its approval pause.

Select your Google Cloud project

Vertex AI charges model requests to the active Google Cloud project. Cloud Shell also exposes that selection through the GOOGLE_CLOUD_PROJECT environment variable.

✔️ I have a Google Cloud account

• Sign in through the Google Cloud console.
• Confirm that the project selector appears in the console header.

ⓧ I need a Google Cloud account

Creating the account has no minimum spend. Google may request a payment method, and this project makes usage-based model requests after billing is enabled.

• Open the official Google Cloud signup page.
• Complete the account and billing setup.
• Return to the Google Cloud console.
• Confirm that the project selector appears in the console header.

• Open Google Cloud Shell in a new browser tab.
• Click Activate Cloud Shell if the terminal does not start automatically.
• Approve the authorization prompt if Google Cloud displays one.
• Wait for the command prompt. Initialization normally takes a few seconds.

You now have an authenticated terminal with the Google Cloud CLI available.

Create a new project

A dedicated project gives this build its own billing boundary. Linking billing permits paid Gemini requests.

Billing setup can feel consequential. The supplied project estimate stays under $0.50 for the full build.

• Choose a globally unique Google Cloud project ID.
• Record your project ID here: your Google Cloud project ID.
• Create the dedicated project by running this command:

gcloud projects create [[PROJECT_ID="your Google Cloud project ID"]]

What does this command do?

The gcloud CLI creates a Google Cloud project using the ID you supplied. That ID becomes the project's permanent identifier.

You should see output confirming that the project creation operation finished.

Project creation failed?

If project creation fails, check that the ID uses lowercase letters, numbers, or hyphens and is globally unique. Help me diagnose the project creation error.

• List the billing accounts available to your account by running this command:

gcloud billing accounts list --filter=open=true

What does this command do?

This lists billing accounts whose status is open. You need one account ID for the project link.

You should see at least one open billing account with its account ID.

No open billing account?

An empty list means your account has no open billing account or you cannot view it. Use the official billing setup guide before retrying.

• Record the open billing account ID here: your billing account ID.
• Link billing, select the project, and enable Vertex AI by running these commands:

gcloud billing projects link [[PROJECT_ID="your Google Cloud project ID"]] --billing-account=[[BILLING_ACCOUNT_ID="your billing account ID"]]
gcloud config set project [[PROJECT_ID="your Google Cloud project ID"]]
gcloud services enable aiplatform.googleapis.com

What do these commands do?

• The first command links the open billing account to your project.
• The second command makes that project active in the current Cloud Shell session.
• The third command enables aiplatform.googleapis.com for Vertex AI access.

You should see confirmation that billing is linked. You should also see the service enable operation finish successfully.

Billing or API setup failed?

A permissions error usually means your account cannot link the selected billing account or enable APIs in the project. Confirm both permissions before retrying. Help me diagnose the setup error.

Reuse the Part 1 project

Your Part 1 project already has the durable workflow's cloud boundary. Reusing it keeps that existing configuration available for this project.

• Choose this route only if the Part 1 project still exists.
• Record the existing project ID here: your Google Cloud project ID.
• Select the project and enable Vertex AI by running these commands:

gcloud config set project [[PROJECT_ID="your Google Cloud project ID"]]
gcloud services enable aiplatform.googleapis.com

What do these commands do?

• The first command makes your Part 1 project active in the current Cloud Shell session.
• The second command ensures aiplatform.googleapis.com is enabled for model access.

You should see the selected project in the configuration output. You should also see the service enable operation finish successfully.

Existing project not available?

Confirm that the project ID is the same one used in Part 1. A permissions error means your current account cannot select the project or enable its APIs. Help me diagnose the existing-project setup.

Cloud Shell now propagates the selected project into GOOGLE_CLOUD_PROJECT. The starter app reads this value when it connects to Vertex AI.

Install the starter repository

The starter repository carries Part 1's unchanged CLI plus the incident fixtures used throughout this build. Its pinned dependencies keep every learner on the same Python package versions.

• Clone the starter repository into ~/langgraph-incident-triage-v2 by running this command:

git clone https://github.com/nextwork-projects/langgraph-incident-triage.git ~/langgraph-incident-triage-v2

What does this command do?

Git copies the starter repository into a fixed location in your Cloud Shell home directory. That path remains consistent throughout the project.

You should see Git download the repository files into ~/langgraph-incident-triage-v2.

Clone failed?

Check that your Cloud Shell session has internet access. Confirm that the destination directory does not already contain an incomplete clone.

If the clone still fails, help me diagnose the Git clone problem.

• Enter the repository before creating its virtual environment by running these commands:

cd ~/langgraph-incident-triage-v2
python3 -m venv .venv
source .venv/bin/activate

What do these commands do?

• The first command moves your terminal into the cloned repository.
• The second command creates an isolated environment in .venv.
• The third command activates that environment for the current terminal session.

Your terminal prompt should now indicate that the virtual environment is active.

• Install the pinned project dependencies by running this command:

python3 -m pip install -r requirements.txt

What does this command do?

Python reads the exact package pins from requirements.txt. It installs them inside the active virtual environment.

The installation output should finish without a dependency error.

Dependency installation failed?

Confirm that your terminal is inside ~/langgraph-incident-triage-v2. Confirm that the virtual environment is active before retrying the installation.

If the failure continues, help me troubleshoot the dependency installation.

The Vertex AI API can need a few minutes after enablement. This short wait gives you time to inspect the local evidence specialists use later.

• Click Open Editor in the Cloud Shell toolbar.
• Wait for the Explorer pane to display your home directory.
• Expand langgraph-incident-triage-v2 in the Explorer pane.

• Inspect data/alerts.json in the Cloud Shell file editor for alert records including INC-2041 plus INC-2042.
• Inspect data/logs.jsonl for incident log entries.
• Inspect data/deploys.json for recent deployment records including d-1042.
• Inspect data/metrics.json for service metric time series.
• Inspect data/services.json for service dependencies plus owning teams.

✔️ Awesome, I've got everything!

Your repository is installed with its original starter app. The fixture files remain available for the specialist agents you build next.

ⓧ I'd like to double check the full code

Compare your requirements.txt with this complete pinned file:

langgraph==1.2.14
langgraph-checkpoint-sqlite==3.1.1
langchain-google-genai==4.4.0
langchain==1.4.4

What should match?

LangChain supplies the agent APIs used later. The checkpoint package connects LangGraph to SQLite for durable workflow state.

• The inherited app.py remains the original durable single-workflow approval CLI.
• The start command still accepts the --thread flag plus the --incident flag.
• The data/alerts.json fixture contains INC-2041 plus INC-2042.
• The data/deploys.json fixture contains recent deployment records including d-1042.
• The data/logs.jsonl fixture contains incident log entries.
• The data/metrics.json fixture contains service metric time series.
• The data/services.json fixture maps services to dependencies plus owning teams.

Smoke-test the inherited workflow

The smoke test checks your credentials through a real model request. It also proves the inherited checkpoint workflow can still pause for approval.

The first model call can take one or two minutes while Gemini responds. A quiet terminal during that wait does not indicate a failure.

Before you run the smoke test, do you think the workflow will complete immediately or pause for a human decision?

• Start the inherited incident workflow by running this command:

python3 app.py start --thread smoke-001 --incident "Checkout API returns HTTP 500 for 1 in 6 requests after a config deploy."

What does this command do?

The command starts a durable workflow under the thread ID smoke-001. The incident description gives the inherited analyst a concrete failure to assess.

The existing policy gate checks the resulting action before its approval interrupt pauses the workflow.

You should see WORKFLOW PAUSED with an approval payload. That pause proves the credentials plus model access are working.

First run looks noisy?

• Wait a few minutes before retrying if the first call reports 403 PERMISSION_DENIED.
• Treat a LangChainBetaWarning about the v3 streaming protocol as informational.
• Treat a warning that mentions function_call as informational when the workflow still pauses.

If the workflow still cannot reach its pause, help me troubleshoot the Cloud Shell smoke run.

You have your baseline. The inherited durable CLI now reaches its approval payload from the selected Google Cloud project.

Your cloud project plus starter repository are ready. Next, you will split the single analyst into role-specific logs, deploys, and metrics specialists.

Build Specialist Agents

Your inherited workflow can pause for approval. However, one general analyst hides which evidence came from each investigative role.

You will build three independent agents in Google Cloud Shell. Each agent can inspect one evidence source, and every result follows the same typed JSON structure.

In this step, get ready to:

• Connect the specialist module to Gemini through Vertex AI.
• Build logs, deploys, and metrics agents with role-limited tools.
• Run each specialist against INC-2041 and compare the typed findings.

Create the model and fixture tools

Each specialist needs the same model connection. The tools expose narrow slices of the fixture data so each role investigates one evidence source.

• Switch back to the Cloud Shell editor from the previous step.
• Create specialists.py inside ~/langgraph-incident-triage-v2 using the editor file controls.
• Add the imports, shared paths, severity levels, and fixture loader by pasting this code:

# Import the CLI, fixture, and agent helpers used throughout this file.
import argparse
import json
import os
from pathlib import Path

from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_google_genai import ChatGoogleGenerativeAI

# Keep every specialist on one model and one fixture directory.
MODEL_ID = "gemini-3.7-flash"
DATA_DIR = Path(__file__).parent / "data"
LEVELS = {"INFO": 0, "WARN": 1, "ERROR": 2}


def load_json(name: str):
    # Load one fixture file relative to specialists.py.
    return json.loads((DATA_DIR / name).read_text(encoding="utf-8"))

What does this foundation do?

• The imports provide agent creation, tool registration, model access, file paths, and JSON parsing.
• MODEL_ID selects gemini-3.7-flash for every specialist.
• DATA_DIR resolves the fixture folder relative to specialists.py.
• LEVELS gives each log level a numeric threshold.
• load_json() reads one fixture file into Python data.

• Save specialists.py.
• Confirm that specialists.py now appears beside app.py in the editor file pane.

• Add the shared model factory below load_json() by pasting this function:

def build_model():
    # Read the active project selected in Cloud Shell.
    project_id = os.environ.get("GOOGLE_CLOUD_PROJECT")
    if not project_id:
        raise RuntimeError("GOOGLE_CLOUD_PROJECT is not set.")

    # Use Vertex AI with bounded retries and request time.
    return ChatGoogleGenerativeAI(
        model=MODEL_ID,
        project=project_id,
        location="global",
        vertexai=True,
        max_retries=2,
        timeout=60,
    )

How does the model connection work?

build_model() reads GOOGLE_CLOUD_PROJECT from your authenticated Cloud Shell environment. It stops with a direct message when the project value is unavailable.

ChatGoogleGenerativeAI connects to Vertex AI in the global location. The model retries brief request failures up to two times. Each request has a 60-second timeout.

• Save specialists.py.
• Confirm that the saved build_model() function contains the selected project, global location, retry limit, and timeout.

Project value unavailable?

Confirm that Cloud Shell still uses the Google Cloud project selected in the previous step. The model factory deliberately stops when GOOGLE_CLOUD_PROJECT is missing.

Help me diagnose why GOOGLE_CLOUD_PROJECT is unavailable in Cloud Shell.

• Add the logs tool below build_model() by pasting this function:

@tool
def search_logs(service: str, min_level: str = "WARN") -> str:
    """Return log lines for one service at or above a level: INFO, WARN, or ERROR."""
    threshold = LEVELS.get(min_level.upper(), 1)
    lines = []
    for raw in (DATA_DIR / "logs.jsonl").read_text(encoding="utf-8").splitlines():
        entry = json.loads(raw)
        if entry["service"] == service and LEVELS[entry["level"]] >= threshold:
            lines.append(f"{entry['ts']} {entry['level']} {entry['message']}")
    return "\n".join(lines[-25:]) or f"No {min_level} logs for {service}."

What does the logs tool expose?

The decorator registers search_logs as an agent tool. Its docstring describes the accepted inputs.

The function filters data/logs.jsonl by service. It also applies a minimum severity threshold. The final 25 matching entries keep the returned evidence focused.

• Save specialists.py.
• Confirm that search_logs reads from logs.jsonl and returns matching log lines.

• Add the deploys tool below search_logs by pasting this function:

@tool
def recent_deploys(service: str = "") -> str:
    """List recent deploys, newest first. Pass a service name, or leave it empty for every service."""
    deploys = sorted(
        load_json("deploys.json"), key=lambda d: d["deployed_at"], reverse=True
    )
    if service:
        deploys = [d for d in deploys if d["service"] == service]
    return "\n".join(
        f"{d['deployed_at']} {d['id']} {d['service']} ({d['team']}): {d['change']}"
        for d in deploys
    ) or f"No deploys for {service}."

What does the deploys tool expose?

recent_deploys sorts deployment records from newest to oldest. A service value narrows the result to one system.

Each result includes the timestamp, deployment ID, service, owning team, and recorded change. This gives the agent evidence it can compare with the alert timeline.

• Save specialists.py.
• Confirm that recent_deploys reads deploys.json and sorts by deployment time.

• Add the metrics tool below recent_deploys by pasting this function:

@tool
def metric_window(service: str) -> str:
    """Return every metric time series for one service."""
    if os.environ.get("TRIAGE_FAIL") == "metrics":
        raise RuntimeError("Metrics backend timed out.")
    series = load_json("metrics.json").get(service)
    if not series:
        return f"No metrics for {service}."
    return "\n".join(
        f"{name}: " + ", ".join(f"{p['ts'][11:16]}={p['value']}" for p in points)
        for name, points in series.items()
    )

Why include a failure hook?

metric_window returns each stored metric series for one service. Every point is reduced to its time and value.

TRIAGE_FAIL=metrics activates a deliberate backend failure. A later step uses this hook to prove that finished sibling investigations survive a crash.

• Save specialists.py.
• Confirm that the file now contains one fixture tool for each investigative role.

Tool code showing an error?

Check the fixture names against data/logs.jsonl, data/deploys.json, and data/metrics.json. Check each closing parenthesis around the returned string expressions.

Help me compare my three fixture tools with the expected specialists.py code.

Define the role-specific agents

A tool controls which evidence an agent can access. A system prompt gives that agent its investigative role. Structured output gives every result the same Finding shape.

• Add the role map below metric_window by pasting this configuration:

SPECIALISTS = {
    "logs": {
        "tools": [search_logs],
        "prompt": "You are the logs investigator on an incident response team. "
        "Search the logs of the alerting service and its dependencies for "
        "errors that explain the alert, and note when they started.",
    },
    "deploys": {
        "tools": [recent_deploys],
        "prompt": "You are the change investigator on an incident response team. "
        "Find deploys to the alerting service or its dependencies shortly "
        "before the alert, and judge which change could cause it.",
    },
    "metrics": {
        "tools": [metric_window],
        "prompt": "You are the metrics investigator on an incident response team. "
        "Read the metrics of the alerting service and its dependencies and "
        "describe what changed, when, and by how much.",
    },
}

How are the roles isolated?

SPECIALISTS pairs each role with one tool. It also stores one role-specific prompt.

The logs role can inspect only logs through its agent tool list. The same boundary applies to deploys and metrics.

• Save specialists.py.
• Confirm that SPECIALISTS contains the logs, deploys, and metrics roles.

• Add the shared Finding schema below SPECIALISTS by pasting this dictionary:

FINDING_SCHEMA = {
    "title": "Finding",
    "type": "object",
    "properties": {
        "summary": {"type": "string"},
        "evidence": {"type": "array", "items": {"type": "string"}},
        "suspected_cause": {"type": "string"},
        "severity": {"type": "string", "enum": ["SEV1", "SEV2", "SEV3", "SEV4"]},
    },
    "required": ["summary", "evidence", "suspected_cause", "severity"],
}

What does the Finding schema guarantee?

Every specialist returns a summary. It also returns an evidence list, a suspected cause, and a severity.

Severity is limited to SEV1 through SEV4. A shared shape lets later graph nodes combine the three findings predictably.

• Save specialists.py.
• Confirm that FINDING_SCHEMA contains four required properties.

• Add the specialist factory below FINDING_SCHEMA by pasting this function:

def build_specialist(name: str):
    spec = SPECIALISTS[name]
    return create_agent(
        build_model(),
        tools=spec["tools"],
        system_prompt=spec["prompt"]
        + " Use your tools before answering. Quote exact evidence. "
        "Rate severity from SEV1 (worst) to SEV4 from what you found.",
        response_format=FINDING_SCHEMA,
        name=f"{name}_specialist",
    )

How is each agent assembled?

build_specialist() looks up the selected role. It passes that role's tool and prompt to create_agent().

response_format applies the Finding schema. The agent name identifies the selected specialist in later graph output.

• Save specialists.py.
• Confirm that create_agent() receives system_prompt, response_format, and name.

Agent factory showing an error?

Check that create_agent comes from langchain.agents. Confirm that the prompt argument is named system_prompt.

Help me debug the build_specialist function and its create_agent arguments.

Add the standalone CLI and test each role

The graph stays unchanged in this step. A standalone CLI proves that every specialist can investigate the same alert through its own evidence source.

• Add the specialist runner and alert formatter below build_specialist() by pasting these functions:

def run_specialist(name: str, brief: str) -> dict:
    result = build_specialist(name).invoke({"messages": [("human", brief)]})
    return result["structured_response"]


def alert_brief(alert_id: str) -> str:
    alert = {a["id"]: a for a in load_json("alerts.json")}[alert_id]
    service = alert["service"]
    related = [service] + load_json("services.json")[service]["depends_on"]
    return (
        f"Alert {alert['id']} fired at {alert['fired_at']} on {service}.\n"
        f"{alert['title']}: {alert['description']}\n"
        f"Services to check: {', '.join(related)}."
    )

How does an alert become an investigation?

run_specialist() sends the alert brief as a human message. It returns the typed structured_response.

alert_brief() combines the selected alert with its service dependencies. Every role receives the same incident scope.

• Save specialists.py.
• Confirm that alert_brief() reads alerts.json and services.json.

• Add the command-line entry point below alert_brief() by pasting this final chunk:

def main():
    parser = argparse.ArgumentParser(description="Run one specialist on one alert.")
    parser.add_argument("specialist", choices=list(SPECIALISTS))
    parser.add_argument("alert")
    args = parser.parse_args()
    print(f"{args.specialist} specialist investigating {args.alert}...")
    finding = run_specialist(args.specialist, alert_brief(args.alert))
    print(json.dumps(finding, indent=2))


if __name__ == "__main__":
    main()

What does the CLI accept?

main() requires a specialist name and an alert ID. The specialist value is restricted to the roles in SPECIALISTS.

The final print formats the typed Finding as indented JSON. This makes the shared output structure visible in the terminal.

• Save specialists.py.
• Return to the Cloud Shell terminal from earlier.

Before you run the checks, do you expect every role to return the same evidence?

• Run the logs specialist against the planted incident by running this command:

python3 specialists.py logs INC-2041

What does this command test?

The command loads INC-2041 from the fixtures. It gives the resulting brief to the logs specialist.

The agent can use only search_logs before returning its Finding object.

You should see an investigation message followed by JSON containing summary, evidence, suspected_cause, and severity.

• Run the deploys specialist against the same incident by running this command:

python3 specialists.py deploys INC-2041

What does this command test?

This runs the change investigator with the same alert. The specialist can inspect only the recent deployment fixture before returning a typed finding.

You should see another Finding JSON object. Its evidence should focus on recent changes. At least one of the three findings should identify d-1042 or the connection-pool change.

• Run the metrics specialist against the same incident by running this command:

python3 specialists.py metrics INC-2041

What does this command test?

This runs the metrics investigator with the same alert. The specialist can inspect only the stored metric time series before returning a typed finding.

You should see a third Finding JSON object based on metric time series. The wording can vary because the model generates each response from the fixture evidence.

Specialist run failed?

Confirm that the terminal still shows the active .venv environment. Check that the terminal is inside ~/langgraph-incident-triage-v2.

A missing project message means GOOGLE_CLOUD_PROJECT is unavailable. A fixture failure usually points to a misspelled file name or alert ID.

Help me diagnose why specialists.py failed for INC-2041.

✔️ Awesome, I've got everything!

That worked. All three roles now return typed findings from their own evidence sources. Keep app.py unchanged for the next step.

ⓧ I'd like to double check the full code

Compare your complete specialists.py with this cumulative version:

import argparse
import json
import os
from pathlib import Path

from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_google_genai import ChatGoogleGenerativeAI

MODEL_ID = "gemini-3.7-flash"
DATA_DIR = Path(__file__).parent / "data"
LEVELS = {"INFO": 0, "WARN": 1, "ERROR": 2}


def load_json(name: str):
    return json.loads((DATA_DIR / name).read_text(encoding="utf-8"))


def build_model():
    project_id = os.environ.get("GOOGLE_CLOUD_PROJECT")
    if not project_id:
        raise RuntimeError("GOOGLE_CLOUD_PROJECT is not set.")

    return ChatGoogleGenerativeAI(
        model=MODEL_ID,
        project=project_id,
        location="global",
        vertexai=True,
        max_retries=2,
        timeout=60,
    )


@tool
def search_logs(service: str, min_level: str = "WARN") -> str:
    """Return log lines for one service at or above a level: INFO, WARN, or ERROR."""
    threshold = LEVELS.get(min_level.upper(), 1)
    lines = []
    for raw in (DATA_DIR / "logs.jsonl").read_text(encoding="utf-8").splitlines():
        entry = json.loads(raw)
        if entry["service"] == service and LEVELS[entry["level"]] >= threshold:
            lines.append(f"{entry['ts']} {entry['level']} {entry['message']}")
    return "\n".join(lines[-25:]) or f"No {min_level} logs for {service}."


@tool
def recent_deploys(service: str = "") -> str:
    """List recent deploys, newest first. Pass a service name, or leave it empty for every service."""
    deploys = sorted(
        load_json("deploys.json"), key=lambda d: d["deployed_at"], reverse=True
    )
    if service:
        deploys = [d for d in deploys if d["service"] == service]
    return "\n".join(
        f"{d['deployed_at']} {d['id']} {d['service']} ({d['team']}): {d['change']}"
        for d in deploys
    ) or f"No deploys for {service}."


@tool
def metric_window(service: str) -> str:
    """Return every metric time series for one service."""
    if os.environ.get("TRIAGE_FAIL") == "metrics":
        raise RuntimeError("Metrics backend timed out.")
    series = load_json("metrics.json").get(service)
    if not series:
        return f"No metrics for {service}."
    return "\n".join(
        f"{name}: " + ", ".join(f"{p['ts'][11:16]}={p['value']}" for p in points)
        for name, points in series.items()
    )


SPECIALISTS = {
    "logs": {
        "tools": [search_logs],
        "prompt": "You are the logs investigator on an incident response team. "
        "Search the logs of the alerting service and its dependencies for "
        "errors that explain the alert, and note when they started.",
    },
    "deploys": {
        "tools": [recent_deploys],
        "prompt": "You are the change investigator on an incident response team. "
        "Find deploys to the alerting service or its dependencies shortly "
        "before the alert, and judge which change could cause it.",
    },
    "metrics": {
        "tools": [metric_window],
        "prompt": "You are the metrics investigator on an incident response team. "
        "Read the metrics of the alerting service and its dependencies and "
        "describe what changed, when, and by how much.",
    },
}

FINDING_SCHEMA = {
    "title": "Finding",
    "type": "object",
    "properties": {
        "summary": {"type": "string"},
        "evidence": {"type": "array", "items": {"type": "string"}},
        "suspected_cause": {"type": "string"},
        "severity": {"type": "string", "enum": ["SEV1", "SEV2", "SEV3", "SEV4"]},
    },
    "required": ["summary", "evidence", "suspected_cause", "severity"],
}


def build_specialist(name: str):
    spec = SPECIALISTS[name]
    return create_agent(
        build_model(),
        tools=spec["tools"],
        system_prompt=spec["prompt"]
        + " Use your tools before answering. Quote exact evidence. "
        "Rate severity from SEV1 (worst) to SEV4 from what you found.",
        response_format=FINDING_SCHEMA,
        name=f"{name}_specialist",
    )


def run_specialist(name: str, brief: str) -> dict:
    result = build_specialist(name).invoke({"messages": [("human", brief)]})
    return result["structured_response"]


def alert_brief(alert_id: str) -> str:
    alert = {a["id"]: a for a in load_json("alerts.json")}[alert_id]
    service = alert["service"]
    related = [service] + load_json("services.json")[service]["depends_on"]
    return (
        f"Alert {alert['id']} fired at {alert['fired_at']} on {service}.\n"
        f"{alert['title']}: {alert['description']}\n"
        f"Services to check: {', '.join(related)}."
    )


def main():
    parser = argparse.ArgumentParser(description="Run one specialist on one alert.")
    parser.add_argument("specialist", choices=list(SPECIALISTS))
    parser.add_argument("alert")
    args = parser.parse_args()
    print(f"{args.specialist} specialist investigating {args.alert}...")
    finding = run_specialist(args.specialist, alert_brief(args.alert))
    print(json.dumps(finding, indent=2))


if __name__ == "__main__":
    main()

Your specialist boundaries now work. Next, you will route alerts into parallel branches and merge the resulting findings into one diagnosis.

Route Alerts in Parallel

Your logs, deploys, and metrics specialists can investigate an alert independently. The inherited workflow still lacks an Incident Commander that chooses the right investigators for each incident.

A router architecture in LangGraph can dispatch selected specialists at the same time. This first parallel run also exposes what happens when several branches update one shared field.

In this step, get ready to:

• Route each alert to the specialists most likely to explain it.
• Fan selected investigations out across parallel graph branches.
• Merge concurrent findings before one deferred synthesis.

Build the routed fan-out

The router reads the alert before specialist work begins. Its structured response contains an initial severity plus the roles needed for that incident.

• Switch back to app.py in the Cloud Shell editor.
• Update the import section so the graph can create branch-specific tasks:

# Import CLI, serialization, path, and typed-state helpers.
import argparse
import json
import os
from pathlib import Path
from typing import TypedDict

# Require checkpoint values to serialize without silent fallbacks.
os.environ.setdefault("LANGGRAPH_STRICT_MSGPACK", "true")

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, Send, interrupt

# Reuse the tested specialists and fixture helpers.
from specialists import SPECIALISTS, alert_brief, build_model, load_json, run_specialist

What changes in these imports?

The Send type carries a destination node plus the state for one branch. The specialist imports let the graph reuse the agents you already tested.

• Replace the existing checkpoint database constant with this project-specific name:

DB_PATH = "triage_checkpoints.db"
SEVERITY_ORDER = ["SEV4", "SEV3", "SEV2", "SEV1"]

Why use a new checkpoint database?

The new database separates this graph from checkpoints created by the inherited workflow. SEVERITY_ORDER gives the graph a consistent ranking from the lowest severity to the highest severity.

• Add the router schema below ANALYSIS_SCHEMA:

ROUTER_SCHEMA = {
    "title": "TriageDecision",
    "type": "object",
    "properties": {
        "severity": {
            "type": "string",
            "enum": ["SEV1", "SEV2", "SEV3", "SEV4"],
        },
        "specialists": {
            "type": "array",
            "items": {"type": "string", "enum": list(SPECIALISTS)},
        },
        "reason": {"type": "string"},
    },
    "required": ["severity", "specialists", "reason"],
}

What does the router return?

• The severity field records the Incident Commander's initial assessment.
• The specialists field contains roles from SPECIALISTS.
• The reason field records why those roles were selected.

Parallel branches will soon return separate findings to one shared state. The first version deliberately gives that field no merge rule.

• Replace the existing IncidentState declaration with this deliberately incomplete parallel version:

class IncidentState(TypedDict, total=False):
    incident_id: str
    incident: str
    alert: dict
    severity: str
    specialists: list[str]
    findings: list[dict]
    analysis: dict
    actions: list[dict]
    approvals: list[dict]
    status: str
    audit_log: list[str]
    runbook_path: str


class SpecialistTask(TypedDict):
    specialist: str
    brief: str

How does branch state travel?

The shared IncidentState stores values that survive across the workflow. Each investigation receives a smaller SpecialistTask containing its role plus the alert brief.

The plain findings field accepts one update per graph step. You will test that limitation before changing it.

• Replace build_analyst() with these structured model builders:

def build_analyst():
    return build_model().with_structured_output(
        schema=ANALYSIS_SCHEMA,
        method="json_schema",
    )


def build_router():
    return build_model().with_structured_output(
        schema=ROUTER_SCHEMA,
        method="json_schema",
    )

Why share the model builder?

Both structured models reuse the verified Vertex AI configuration from specialists.py. Each model receives a different output schema for its role.

• Add the alert briefing helper below safe_incident_id():

def incident_brief(state: IncidentState) -> str:
    return alert_brief(state["alert"]["id"])

What does the briefing helper do?

The helper passes the selected alert ID to alert_brief(). That function expands the fixture into alert details plus related services.

• Add the Incident Commander node below incident_brief():

def route_incident(state: IncidentState):
    # Ask the Incident Commander to classify and route the alert.
    decision = build_router().invoke(
        [
            (
                "system",
                "You are the incident commander. Set a first severity and choose "
                "which specialists investigate: logs, deploys, metrics. For SEV1 "
                "or SEV2, send every specialist. For SEV3 or SEV4, send only the "
                "one or two most likely to explain the symptom.",
            ),
            ("human", incident_brief(state)),
        ]
    )
    print(f"Router: {decision['severity']}, sending {', '.join(decision['specialists'])}")

    # Save the route so later nodes can explain the decision.
    return {
        "severity": decision["severity"],
        "specialists": decision["specialists"],
        "status": "routed",
        "audit_log": state["audit_log"]
        + [f"Router chose {decision['specialists']}: {decision['reason']}"],
    }

How does the router choose branches?

• Severe alerts receive all three specialists.
• Lower-severity alerts receive one or two focused specialists.
• The routing line exposes the selected roles before investigation begins.

• Add the dispatcher plus the shared investigation node below route_incident():

def dispatch_specialists(state: IncidentState):
    # Give every selected branch the same incident brief.
    brief = incident_brief(state)
    return [
        Send("investigate", {"specialist": name, "brief": brief})
        for name in state["specialists"]
    ]


def investigate(task: SpecialistTask):
    # Run one role inside its own parallel branch.
    name = task["specialist"]
    print(f"  {name} specialist investigating...")
    finding = run_specialist(name, task["brief"])
    print(f"  {name} specialist done: {finding['severity']}, {finding['suspected_cause']}")

    # Return a one-item list for the shared findings reducer.
    return {
        "findings": [{"specialist": name, **finding}],
        "severity": finding["severity"],
    }

How does fan-out work?

The dispatcher creates one Send task for every selected role. Every task targets investigate with its own specialist name.

Each branch prints its starting state plus its completed result. The completion lines make runtime scheduling visible.

• Replace the inherited analyze_incident() with this specialist-aware synthesis node:

def analyze_incident(state: IncidentState):
    # Give the analyst every merged finding from the parallel branches.
    analyst = build_analyst()
    findings = json.dumps(state["findings"], indent=2)
    analysis = analyst.invoke(
        [
            (
                "system",
                "You are an SRE incident analyst. Combine the specialists' "
                "findings into one diagnosis and name the most likely cause with "
                "its evidence. Propose one reversible remediation step. Keep "
                "every field concise.",
            ),
            ("human", f"{state['incident']}\n\nFindings:\n{findings}"),
        ]
    )

    # Save one diagnosis and record how much evidence supported it.
    return {
        "analysis": analysis,
        "severity": analysis["severity"],
        "status": "analyzed",
        "audit_log": state["audit_log"]
        + [f"Model combined {len(state['findings'])} findings into one analysis."],
    }

What happens during synthesis?

The analyst receives the selected specialists' typed findings as JSON. Its audit entry records how many findings contributed to the diagnosis.

• In build_graph(), replace the inherited opening path with this routed path:

    builder = StateGraph(IncidentState)
    builder.add_node("route_incident", route_incident)
    builder.add_node("investigate", investigate)
    builder.add_node("analyze_incident", analyze_incident)
    builder.add_node("policy_check", policy_check)
    builder.add_node("approval_gate", approval_gate)
    builder.add_node("publish_runbook", publish_runbook)

    builder.add_edge(START, "route_incident")
    builder.add_conditional_edges(
        "route_incident", dispatch_specialists, ["investigate"]
    )
    builder.add_edge("investigate", "analyze_incident")
    builder.add_edge("analyze_incident", "policy_check")

How does the graph flow?

The graph starts with routing. Its conditional edge expands the returned tasks into parallel investigations.

Every selected investigation points to synthesis. The inherited policy flow continues after analysis.

• Keep the inherited policy edges below the new route.
• Keep the inherited checkpoint compilation at the end of build_graph().

• In parse_args(), find the inherited incident argument:

start_parser.add_argument("--incident", required=True)

What are you replacing?

The inherited CLI accepts free-form incident text. The routed workflow needs an alert ID that maps to a fixture.

• Replace the incident argument with the alert argument:

start_parser.add_argument("--alert", required=True)

What does the alert argument contain?

The argument receives an alert ID such as INC-2041. The workflow uses that ID to load the complete fixture.

• In main(), replace the inherited start input with this fixture-loading block:

        if args.command == "start":
            alerts = {alert["id"]: alert for alert in load_json("alerts.json")}
            alert = alerts[args.alert]
            graph_input = {
                "incident_id": args.thread,
                "incident": f"{alert['title']}: {alert['description']}",
                "alert": alert,
                "status": "new",
                "audit_log": [],
            }

How does the CLI load an incident?

The dictionary indexes every fixture by its alert ID. The selected alert becomes shared state for routing plus specialist investigation.

Watch parallel writes collide

Several investigations can now finish during one graph step. Each branch returns a separate value for the same plain findings field.

• Save app.py.

Before you run the graph, do you think one plain state field can accept all three branch updates?

• Start the first parallel investigation by running:

python3 app.py start --thread parallel-001 --alert INC-2041

This failure is intentional

You will see the router select specialists before the run fails with InvalidUpdateError: At key 'findings': Can receive only one value per step. Use an Annotated key to handle multiple values..

Every branch produced a valid finding. The shared state needs an explicit rule for combining those updates.

Merge results and verify the join

A reducer combines concurrent updates for one state field. The findings reducer joins branch lists while the severity reducer preserves the worst reported level.

• Add the reducer imports near the top of app.py:

import operator
from typing import Annotated, TypedDict

What do these imports provide?

The operator.add function concatenates branch lists. Annotated attaches a reducer to a state field.

• Remove the earlier from typing import TypedDict line.
• Add the severity reducer above IncidentState:

def highest_severity(current: str, new: str) -> str:
    if current not in SEVERITY_ORDER:
        return new
    return max(current, new, key=SEVERITY_ORDER.index)

How does severity reduction work?

The reducer returns the incoming value when no valid severity exists yet. Later updates are compared by their position in SEVERITY_ORDER.

• Replace the plain severity plus findings fields in IncidentState with the reduced versions:

class IncidentState(TypedDict, total=False):
    incident_id: str
    incident: str
    alert: dict
    severity: Annotated[str, highest_severity]
    specialists: list[str]
    findings: Annotated[list[dict], operator.add]
    analysis: dict
    actions: list[dict]
    approvals: list[dict]
    status: str
    audit_log: list[str]
    runbook_path: str

How do the reducers protect state?

• The findings reducer concatenates each branch's one-item list.
• The severity reducer keeps the highest-severity update.
• The other fields keep their inherited single-update behavior.

• In build_graph(), find the current synthesis registration:

builder.add_node("analyze_incident", analyze_incident)

Why change this registration?

A normal downstream node can become eligible while other branch work remains pending. Synthesis needs every selected finding before it runs.

• Replace the registration with this deferred version:

builder.add_node("analyze_incident", analyze_incident, defer=True)

What does deferred execution guarantee?

Deferred execution waits until no selected specialist task remains pending. The graph then runs synthesis once with the merged findings.

• Save app.py.

Before you retry, which specialist completion lines do you expect to appear in an unpredictable order?

Allow time for the agents

A complete start can take 1 to 2 minutes while the router plus selected specialists call the model. A quiet terminal during an investigation is expected.

• Run the repaired workflow with a fresh thread ID:

python3 app.py start --thread parallel-002 --alert INC-2041

What should you see?

You should see the router send the severe alert to logs, deploys, and metrics. Their completion lines can appear in any order.

Synthesis runs once after every selected specialist finishes. The inherited policy flow then continues.

Still seeing a concurrent update error?

Confirm that findings uses Annotated[list[dict], operator.add] in IncidentState.

Confirm that operator plus Annotated are imported near the top of app.py.

Ask for help with my parallel findings reducer.

• Run the severe alert again with another fresh thread ID:

python3 app.py start --thread parallel-003 --alert INC-2041

Compare the second run

You should see the same three roles selected. Their completion order can differ because the branches run independently.

That is the parallel behavior you were looking for. The graph preserves every result despite the changing order.

The Incident Commander can also choose a smaller team for a narrower alert. This keeps specialist work proportional to the incident.

Before you check the second fixture, do you expect the router to select all three specialists again?

• Run the lower-severity fixture with a fresh thread ID:

python3 app.py start --thread parallel-small-001 --alert INC-2042

Check the narrower route

You should see the router choose fewer specialists when one or two roles can explain the symptom. Synthesis still runs once after every selected branch finishes.

✔️ Awesome, I've got everything!

Your router now selects specialist branches. Their concurrent findings merge before one deferred synthesis.

ⓧ I'd like to double check the full code

Compare these routed sections with app.py. Keep the inherited policy, approval, publishing, streaming, and resume sections in the same file.

import operator
from typing import Annotated, TypedDict

from langgraph.types import Command, Send, interrupt

from specialists import SPECIALISTS, alert_brief, build_model, load_json, run_specialist

DB_PATH = "triage_checkpoints.db"
SEVERITY_ORDER = ["SEV4", "SEV3", "SEV2", "SEV1"]


def highest_severity(current: str, new: str) -> str:
    if current not in SEVERITY_ORDER:
        return new
    return max(current, new, key=SEVERITY_ORDER.index)


class IncidentState(TypedDict, total=False):
    incident_id: str
    incident: str
    alert: dict
    severity: Annotated[str, highest_severity]
    specialists: list[str]
    findings: Annotated[list[dict], operator.add]
    analysis: dict
    actions: list[dict]
    approvals: list[dict]
    status: str
    audit_log: list[str]
    runbook_path: str


class SpecialistTask(TypedDict):
    specialist: str
    brief: str

def build_analyst():
    return build_model().with_structured_output(
        schema=ANALYSIS_SCHEMA,
        method="json_schema",
    )


def build_router():
    return build_model().with_structured_output(
        schema=ROUTER_SCHEMA,
        method="json_schema",
    )


def incident_brief(state: IncidentState) -> str:
    return alert_brief(state["alert"]["id"])

def dispatch_specialists(state: IncidentState):
    brief = incident_brief(state)
    return [
        Send("investigate", {"specialist": name, "brief": brief})
        for name in state["specialists"]
    ]


def investigate(task: SpecialistTask):
    name = task["specialist"]
    print(f"  {name} specialist investigating...")
    finding = run_specialist(name, task["brief"])
    print(f"  {name} specialist done: {finding['severity']}, {finding['suspected_cause']}")
    return {
        "findings": [{"specialist": name, **finding}],
        "severity": finding["severity"],
    }

    builder.add_node("route_incident", route_incident)
    builder.add_node("investigate", investigate)
    builder.add_node("analyze_incident", analyze_incident, defer=True)

    builder.add_edge(START, "route_incident")
    builder.add_conditional_edges(
        "route_incident", dispatch_specialists, ["investigate"]
    )
    builder.add_edge("investigate", "analyze_incident")
    builder.add_edge("analyze_incident", "policy_check")

    start_parser = subparsers.add_parser("start")
    start_parser.add_argument("--thread", required=True)
    start_parser.add_argument("--alert", required=True)

        if args.command == "start":
            alerts = {alert["id"]: alert for alert in load_json("alerts.json")}
            alert = alerts[args.alert]
            graph_input = {
                "incident_id": args.thread,
                "incident": f"{alert['title']}: {alert['description']}",
                "alert": alert,
                "status": "new",
                "audit_log": [],
            }

Your Incident Commander now routes alerts through parallel specialist investigations without losing branch results. Next, you will prove that completed investigations remain saved when one specialist crashes.

Resume a Crashed Specialist

Your router now sends logs, deploys, and metrics work down parallel branches. A metrics crash can still interrupt the entire super-step.

LangGraph preserves completed sibling updates as pending writes in SQLite. This step makes that durable state visible before a new process resumes the unfinished branch.

In this step, get ready to:

• Report the specialist findings saved before a workflow failure.
• Allow a failed thread to resume without an approval decision.
• Prove that only the crashed metrics investigation runs again.

Report the saved work after failure

A raw exception hides the useful checkpoint state behind a traceback. A failure summary can show which branches finished plus which node remains pending.

• Return to app.py in the Google Cloud Shell Editor.
• Find the existing display_pending() function.
• Add the failure-reporting function immediately below it using this code:

def display_failure(graph, config, error):
    # Read the durable snapshot after the branch failure.
    snapshot = graph.get_state(config)
    saved = [finding["specialist"] for finding in snapshot.values.get("findings", [])]

    # Separate completed branch output from work that must resume.
    print(f"WORKFLOW FAILED: {error}")
    print(f"Saved findings: {', '.join(saved) or 'none'}")
    print(f"Waiting to re-run: {', '.join(snapshot.next)}")

What does this code do?

• The call to graph.get_state(config) reads the latest checkpoint for the current thread.
• The saved list collects the specialist names already stored under findings.
• The snapshot.next value identifies the graph work that remains pending.

• Scroll to the call to graph.stream_events() inside main().
• Wrap the stream plus its display call with this exception handler:

        try:
            # Persist each graph step before displaying its streamed events.
            stream = graph.stream_events(
                graph_input,
                config=config,
                durability="sync",
                version="v3",
            )
            display_stream(stream)
        except Exception as error:
            # Report the checkpoint instead of hiding it behind a traceback.
            display_failure(graph, config, error)

How does the failure reach the summary?

The try block keeps the existing synchronous checkpoint stream unchanged. The except block passes the same graph plus thread configuration to display_failure().

• Save app.py.

Before you run this, predict which branch will fail. Also predict which sibling findings will survive.

This run often needs 1 to 2 minutes while the specialist agents work, so a quiet terminal is normal.

• Trigger the metrics fixture failure by running this command:

TRIAGE_FAIL=metrics python3 app.py start --thread crash-001 --alert INC-2041

You will see WORKFLOW FAILED after the metrics investigation raises Metrics backend timed out..

The summary lists the completed logs and deploys specialists under Saved findings. It lists investigate under Waiting to re-run.

Missing the saved findings?

Check that you used the fresh thread ID crash-001. An older checkpoint can place the thread at a different graph state.

Confirm that metric_window() still raises the fixture failure when TRIAGE_FAIL=metrics.

Help me diagnose why my failed parallel LangGraph run does not show saved sibling findings.

Resume only the unfinished branch

A resume command can serve two different states. An approval resume carries a decision while a failure resume supplies None so LangGraph retries the pending work.

• Return to parse_args() in app.py.
• Find the argument definition for --decision.
• Replace that definition with this optional argument:

    resume_parser.add_argument(
        "--decision",
        choices=["approve", "reject"],
        help="Leave out to retry a failed run.",
    )

Why make the decision optional?

The existing approval path can still receive approve or reject. Leaving the flag out now represents a failed execution that needs to continue from its checkpoint.

• Scroll to the resume input logic inside main().
• Change the current resume condition to elif args.decision: so the existing Command path handles approval decisions.
• Add this fallback immediately after the decision branch:

        else:
            graph_input = None

What does the fallback change?

The None input tells the graph to continue the saved thread. The checkpoint determines that investigate for metrics is the remaining task.

• Save app.py.
• Clear the metrics failure hook from your shell by running this command:

unset TRIAGE_FAIL

The metrics tool can now read data/metrics.json when the pending branch runs again.

Before you resume, predict whether the logs or deploys specialist will print another investigating message.

• Resume the saved thread from a new Python process by running this command:

python3 app.py resume --thread crash-001

You will see only the metrics specialist investigate before synthesis continues. The logs and deploys investigation messages stay absent because their pending writes were already saved.

That recovery is the key operational win. Your graph can reuse completed model work after a process ends.

Why use checkpoint resume?

Checkpoint resume exposes durable recovery across separate processes. It also avoids repeating successful sibling model calls.

This lesson keeps RetryPolicy, error_handler=, and node timeout= out of the graph. The recovery behavior comes from pending writes plus durability="sync".

🙋‍♀️ Did every specialist run again?

Confirm that the resume command uses the same crash-001 thread ID as the failed run. A different ID has no saved pending work.

Confirm that DB_PATH still points to triage_checkpoints.db from the same repository directory.

Help me find why resuming crash-001 reruns completed specialists or cannot find pending work.

✔️ The metrics branch resumed by itself

Great work. Your checkpoint preserved the completed logs and deploys findings while the new process retried metrics.

ⓧ I'd like to double check the full code

Compare every recovery change in app.py with the reference snippets below.

def display_failure(graph, config, error):
    snapshot = graph.get_state(config)
    saved = [finding["specialist"] for finding in snapshot.values.get("findings", [])]
    print(f"WORKFLOW FAILED: {error}")
    print(f"Saved findings: {', '.join(saved) or 'none'}")
    print(f"Waiting to re-run: {', '.join(snapshot.next)}")

This function reads the persisted snapshot before reporting saved findings plus pending graph work.

    resume_parser.add_argument(
        "--decision",
        choices=["approve", "reject"],
        help="Leave out to retry a failed run.",
    )

This argument keeps approval decisions available while allowing a decision-free recovery command.

        else:
            graph_input = None

This fallback resumes the checkpoint when the command carries no decision.

        try:
            stream = graph.stream_events(
                graph_input,
                config=config,
                durability="sync",
                version="v3",
            )
            display_stream(stream)
        except Exception as error:
            display_failure(graph, config, error)

This handler reports checkpoint state after an exception without changing the synchronous durability or event-stream version.

Your workflow now survives a crashed specialist without discarding completed investigations. Next, you will give each owning team an independent approval decision.

Gate Fixes by Team

Your LangGraph workflow now survives a failed specialist without repeating successful work. That durability makes parallel triage dependable across processes.

A single global approval still gives one person control over fixes owned by different services. This step gives each service team its own interrupt before publication.

In this step, get ready to:

• Assign every proposed action to the team that owns its service.
• Resume one team-specific approval without resolving the other pending approvals.
• Publish a runbook containing approved actions plus a complete decision audit.

Replace the global gate with team-owned approvals

The analysis now proposes separate actions for specific services. The policy layer uses data/services.json to assign each action to its owning team.

How do team-owned gates work?

• The final analysis schema gives each proposed action a service plus a reversible action.
• The policy_check function applies severity rules plus RESTRICTED_TERMS to each action.
• The policy_route function creates one Send("approval_gate", action) branch for every gated action.
• The approvals reducer combines decisions returned by independent branches.

• Update app.py so it matches the complete file in the double-check tab below.
• Save app.py.

✔️ Awesome, I've got everything!

Great. Confirm that app.py is saved before you start the approval workflow.

ⓧ I'd like to double check the full code

Compare your app.py with this final version. Pay close attention to the approval reducer, team-specific resume map, deferred publisher, and audit logic.

import argparse
import json
import os
import operator
from pathlib import Path
from typing import Annotated, TypedDict

os.environ.setdefault("LANGGRAPH_STRICT_MSGPACK", "true")

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, Send, interrupt

from specialists import SPECIALISTS, alert_brief, build_model, load_json, run_specialist

DB_PATH = "triage_checkpoints.db"
SEVERITY_ORDER = ["SEV4", "SEV3", "SEV2", "SEV1"]
RUNBOOK_DIR = Path("runbooks")
RESTRICTED_TERMS = (
    "delete",
    "disable",
    "drop",
    "production",
    "restart",
    "rotate",
)

ANALYSIS_SCHEMA = {
    "title": "IncidentPlan",
    "type": "object",
    "properties": {
        "severity": {
            "type": "string",
            "enum": ["SEV1", "SEV2", "SEV3", "SEV4"],
        },
        "summary": {"type": "string"},
        "likely_cause": {"type": "string"},
        "rationale": {"type": "string"},
        "actions": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "service": {"type": "string"},
                    "action": {"type": "string"},
                },
                "required": ["service", "action"],
            },
        },
    },
    "required": [
        "severity",
        "summary",
        "likely_cause",
        "rationale",
        "actions",
    ],
}

ROUTER_SCHEMA = {
    "title": "TriageDecision",
    "type": "object",
    "properties": {
        "severity": {
            "type": "string",
            "enum": ["SEV1", "SEV2", "SEV3", "SEV4"],
        },
        "specialists": {
            "type": "array",
            "items": {"type": "string", "enum": list(SPECIALISTS)},
        },
        "reason": {"type": "string"},
    },
    "required": ["severity", "specialists", "reason"],
}


def highest_severity(current: str, new: str) -> str:
    if current not in SEVERITY_ORDER:
        return new
    return max(current, new, key=SEVERITY_ORDER.index)


class IncidentState(TypedDict, total=False):
    incident_id: str
    incident: str
    alert: dict
    severity: Annotated[str, highest_severity]
    specialists: list[str]
    findings: Annotated[list[dict], operator.add]
    analysis: dict
    actions: list[dict]
    approvals: Annotated[list[dict], operator.add]
    status: str
    audit_log: list[str]
    runbook_path: str


class SpecialistTask(TypedDict):
    specialist: str
    brief: str


class ActionTask(TypedDict):
    incident_id: str
    team: str
    service: str
    action: str
    policy_reason: str


def build_analyst():
    return build_model().with_structured_output(
        schema=ANALYSIS_SCHEMA,
        method="json_schema",
    )


def build_router():
    return build_model().with_structured_output(
        schema=ROUTER_SCHEMA,
        method="json_schema",
    )


def safe_incident_id(incident_id: str) -> str:
    safe_id = "".join(
        character
        for character in incident_id
        if character.isalnum() or character in {"-", "_"}
    )
    if not any(character.isalnum() for character in safe_id):
        raise ValueError("Incident ID must contain a letter or number.")
    return safe_id


def incident_brief(state: IncidentState) -> str:
    return alert_brief(state["alert"]["id"])


def route_incident(state: IncidentState):
    decision = build_router().invoke(
        [
            (
                "system",
                "You are the incident commander. Set a first severity and choose "
                "which specialists investigate: logs, deploys, metrics. For SEV1 "
                "or SEV2, send every specialist. For SEV3 or SEV4, send only the "
                "one or two most likely to explain the symptom.",
            ),
            ("human", incident_brief(state)),
        ]
    )
    print(f"Router: {decision['severity']}, sending {', '.join(decision['specialists'])}")
    return {
        "severity": decision["severity"],
        "specialists": decision["specialists"],
        "status": "routed",
        "audit_log": state["audit_log"]
        + [f"Router chose {decision['specialists']}: {decision['reason']}"],
    }


def dispatch_specialists(state: IncidentState):
    brief = incident_brief(state)
    return [
        Send("investigate", {"specialist": name, "brief": brief})
        for name in state["specialists"]
    ]


def investigate(task: SpecialistTask):
    name = task["specialist"]
    print(f"  {name} specialist investigating...")
    finding = run_specialist(name, task["brief"])
    print(f"  {name} specialist done: {finding['severity']}, {finding['suspected_cause']}")
    return {
        "findings": [{"specialist": name, **finding}],
        "severity": finding["severity"],
    }


def analyze_incident(state: IncidentState):
    analyst = build_analyst()
    findings = json.dumps(state["findings"], indent=2)
    analysis = analyst.invoke(
        [
            (
                "system",
                "You are an SRE incident analyst. Combine the specialists' "
                "findings into one diagnosis and name the most likely cause with "
                "its evidence. Propose one reversible remediation step for each "
                "service the evidence implicates: the alerting service and every "
                "dependency named in the findings' evidence, at most three. Each "
                "service's owning team must approve its own step. Keep every "
                "field concise.",
            ),
            ("human", f"{state['incident']}\n\nFindings:\n{findings}"),
        ]
    )
    return {
        "analysis": analysis,
        "severity": analysis["severity"],
        "status": "analyzed",
        "audit_log": state["audit_log"]
        + [f"Model combined {len(state['findings'])} findings into one analysis."],
    }


def policy_check(state: IncidentState):
    services = load_json("services.json")
    severity_requires_review = state["severity"] in {"SEV1", "SEV2"}
    actions, approvals = [], []
    for proposed in state["analysis"]["actions"]:
        service = services.get(proposed["service"], {"team": "unowned"})
        searchable_text = f"{state['incident']} {proposed['action']}".lower()
        term_requires_review = any(
            term in searchable_text for term in RESTRICTED_TERMS
        )
        requires_approval = severity_requires_review or term_requires_review
        action = {
            "incident_id": state["incident_id"],
            "team": service["team"],
            "service": proposed["service"],
            "action": proposed["action"],
            "policy_reason": (
                "High severity or restricted operational language detected."
                if requires_approval
                else "No high-risk policy rule matched."
            ),
        }
        actions.append(action | {"requires_approval": requires_approval})
        if not requires_approval:
            approvals.append(action | {"decision": "not required"})
    gated = sum(action["requires_approval"] for action in actions)
    return {
        "actions": actions,
        "approvals": approvals,
        "status": "policy_checked",
        "audit_log": state["audit_log"]
        + [f"Policy check: {gated} of {len(actions)} actions need team approval."],
    }


def policy_route(state: IncidentState):
    gated = [action for action in state["actions"] if action["requires_approval"]]
    if not gated:
        return "publish_runbook"
    return [
        Send(
            "approval_gate",
            {key: value for key, value in action.items() if key != "requires_approval"},
        )
        for action in gated
    ]


def approval_gate(task: ActionTask):
    decision = interrupt(
        {
            "question": f"Approve this action for team {task['team']}?",
            "incident_id": task["incident_id"],
            "team": task["team"],
            "service": task["service"],
            "action": task["action"],
            "policy_reason": task["policy_reason"],
            "allowed_decisions": ["approve", "reject"],
        }
    )
    if decision not in ("approve", "reject"):
        raise ValueError("Decision must be 'approve' or 'reject'.")
    return {"approvals": [task | {"decision": decision}]}


def publish_runbook(state: IncidentState):
    approvals = state.get("approvals", [])
    approved = [a for a in approvals if a["decision"] in ("approve", "not required")]
    decisions = [
        f"Team {a['team']} decision for {a['service']}: {a['decision']}."
        for a in approvals
    ]
    if not approved:
        return {
            "status": "rejected_without_publish",
            "audit_log": state["audit_log"]
            + decisions
            + ["Every action was rejected. Publishing was skipped."],
        }

    RUNBOOK_DIR.mkdir(exist_ok=True)
    path = RUNBOOK_DIR / f"{safe_incident_id(state['incident_id'])}.md"
    analysis = state["analysis"]
    audit_log = state["audit_log"] + decisions + [f"Published runbook to {path}."]
    audit_lines = "\n".join(f"- {entry}" for entry in audit_log)
    finding_lines = "\n".join(
        f"- {f['specialist']} ({f['severity']}): {f['summary']}"
        for f in state["findings"]
    )
    action_lines = "\n".join(
        f"- {a['team']} / {a['service']}: {a['action']} "
        f"({'approved by team' if a['decision'] == 'approve' else 'approval not required'})"
        for a in approved
    )

    path.write_text(
        "# Policy-Controlled Incident Runbook\n\n"
        f"Incident ID: {state['incident_id']}\n\n"
        f"Incident: {state['incident']}\n\n"
        f"Severity: {state['severity']}\n\n"
        f"Summary: {analysis['summary']}\n\n"
        f"Likely cause: {analysis['likely_cause']}\n\n"
        f"Rationale: {analysis['rationale']}\n\n"
        "Findings:\n"
        f"{finding_lines}\n\n"
        "Approved actions:\n"
        f"{action_lines}\n\n"
        "Audit log:\n"
        f"{audit_lines}\n",
        encoding="utf-8",
    )
    return {
        "status": "published",
        "runbook_path": str(path),
        "audit_log": audit_log,
    }


def build_graph(checkpointer: SqliteSaver):
    builder = StateGraph(IncidentState)
    builder.add_node("route_incident", route_incident)
    builder.add_node("investigate", investigate)
    builder.add_node("analyze_incident", analyze_incident, defer=True)
    builder.add_node("policy_check", policy_check)
    builder.add_node("approval_gate", approval_gate)
    builder.add_node("publish_runbook", publish_runbook, defer=True)

    builder.add_edge(START, "route_incident")
    builder.add_conditional_edges(
        "route_incident", dispatch_specialists, ["investigate"]
    )
    builder.add_edge("investigate", "analyze_incident")
    builder.add_edge("analyze_incident", "policy_check")
    builder.add_conditional_edges(
        "policy_check", policy_route, ["approval_gate", "publish_runbook"]
    )
    builder.add_edge("approval_gate", "publish_runbook")
    builder.add_edge("publish_runbook", END)

    return builder.compile(checkpointer=checkpointer)


def display_stream(stream):
    final_state = stream.output
    if stream.interrupted:
        display_pending(stream.interrupts)
        return

    print("WORKFLOW COMPLETE")
    print(f"Status: {final_state['status']}")
    print(f"Runbook: {final_state.get('runbook_path', 'not published')}")


def display_pending(interrupts):
    print(f"WORKFLOW PAUSED: {len(interrupts)} team approvals pending")
    for pending in interrupts:
        request = pending.value
        print(f"- {request['team']} ({request['service']}): {request['action']}")


def display_failure(graph, config, error):
    snapshot = graph.get_state(config)
    saved = [finding["specialist"] for finding in snapshot.values.get("findings", [])]
    print(f"WORKFLOW FAILED: {error}")
    print(f"Saved findings: {', '.join(saved) or 'none'}")
    print(f"Waiting to re-run: {', '.join(snapshot.next)}")


def parse_args():
    parser = argparse.ArgumentParser(
        description="Run a resumable incident-triage workflow."
    )
    subparsers = parser.add_subparsers(dest="command", required=True)

    start_parser = subparsers.add_parser("start")
    start_parser.add_argument("--thread", required=True)
    start_parser.add_argument("--alert", required=True)

    resume_parser = subparsers.add_parser("resume")
    resume_parser.add_argument("--thread", required=True)
    resume_parser.add_argument(
        "--decision",
        choices=["approve", "reject"],
        help="Leave out to retry a failed run.",
    )
    resume_parser.add_argument("--team", help="The team whose action you decide.")

    args = parser.parse_args()
    if args.command == "resume" and args.decision and not args.team:
        parser.error("--decision needs --team.")
    return args


def main():
    args = parse_args()
    config = {"configurable": {"thread_id": args.thread}}

    with SqliteSaver.from_conn_string(DB_PATH) as checkpointer:
        graph = build_graph(checkpointer)

        if args.command == "start":
            alerts = {alert["id"]: alert for alert in load_json("alerts.json")}
            alert = alerts[args.alert]
            graph_input = {
                "incident_id": args.thread,
                "incident": f"{alert['title']}: {alert['description']}",
                "alert": alert,
                "status": "new",
                "audit_log": [],
            }
        elif args.decision:
            pending = graph.get_state(config).interrupts
            team_requests = {
                request.id: args.decision
                for request in pending
                if request.value["team"] == args.team
            }
            if not team_requests:
                teams = sorted({request.value["team"] for request in pending})
                print(
                    f"No pending approval for team {args.team}. "
                    f"Waiting on: {', '.join(teams) or 'nobody'}"
                )
                return
            graph_input = Command(resume=team_requests)
        else:
            graph_input = None

        try:
            stream = graph.stream_events(
                graph_input,
                config=config,
                durability="sync",
                version="v3",
            )
            display_stream(stream)
        except Exception as error:
            display_failure(graph, config, error)


if __name__ == "__main__":
    main()

What changed in the workflow?

The approvals field uses Annotated[list[dict], operator.add] so parallel approval branches can merge their results. Each decision remains attached to its team plus its service.

The publisher uses defer=True so it waits for every selected approval branch. Rejected actions stay in the audit log without appearing under approved actions.

This start can take one to two minutes while the specialists investigate plus the analyst creates service-specific actions.

Before you run this, consider which teams should own the fixes suggested by the planted incident evidence.

• Start the ownership-gated workflow for INC-2041 by running this command:

python3 app.py start --thread inc-001 --alert INC-2041

You should see every selected specialist finish before the workflow reports WORKFLOW PAUSED. The pending list should include separate requests for payments plus platform-data.

Do the pending teams look different?

Model-generated action wording can vary between runs. Confirm that the findings still reference the fixture evidence before checking which services the analysis implicated.

If the workflow fails before pausing, compare app.py with the double-check tab. Focus on the state reducers plus the edges leaving policy_check.

Help me debug my team approval branches.

Resume the payments approval

Every pending interrupt has its own ID. The resume command selects only the interrupt whose payload belongs to the requested team.

The CLI passes that selected ID through Command(resume={interrupt_id: decision}). Other interrupt IDs remain paused in the durable checkpoint.

• Approve the payments action by running this command:

python3 app.py resume --thread inc-001 --team payments --decision approve

You should see WORKFLOW PAUSED again. The remaining pending list should contain platform-data.

What did this decision change?

The payments branch records an approved decision in the checkpoint. The platform-data branch keeps the workflow paused.

The deferred publisher still has pending upstream work. That prevents runbooks/inc-001.md from being created after only one decision.

• Check the project file tree to confirm that runbooks/inc-001.md does not exist yet.

Did the wrong approval resume?

Check that the team value exactly matches a pending team printed by the previous command. The CLI leaves the state unchanged when it cannot find a matching request.

Help me trace the selected interrupt ID.

Reject platform-data and inspect the runbook

The final pending decision releases the deferred publisher. The publisher includes approved or policy-exempt actions while keeping every team decision in the audit log.

Before you run this, consider which action should appear under approved actions after payments approves plus platform-data rejects.

• Reject the platform-data action by running this command:

python3 app.py resume --thread inc-001 --team platform-data --decision reject

You should see WORKFLOW COMPLETE with runbooks/inc-001.md as the published runbook.

• Return to the Cloud Shell Editor.
• Expand runbooks in the Explorer pane.
• Click inc-001.md to open the published runbook.

You should see the planted d-1042 connection-pool cause in the diagnosis. Approved actions should contain the payments fix without the rejected platform-data fix.

The audit log should contain both team decisions. This proves that publication filtered the action list without erasing the rejected decision.

Is the runbook missing or incorrect?

Confirm that the previous output reported a completed workflow. A remaining pending interrupt means the deferred publisher is still waiting.

If a rejected action appears under approved actions, compare the approved filter in publish_runbook with the double-check tab.

Help me debug the published runbook.

You now have separate approval authority for each service owner. Your published runbook preserves the incident evidence plus every decision while exposing only permitted actions.

Secret mission

Build a Read-Only Incident Status Board

Give operators a safe view of a saved incident without advancing it. Build status.py to report findings, decisions, pending teams, next nodes, and runbook state from the durable checkpoint.

Clean Up Your Resources

Clean Up Your Resources

Google Cloud Shell is free to use. Your saved files do not trigger model charges while the workflow is idle.

Decide whether to keep your resources available, pause your work, or delete the resources you no longer need. Future model requests can still incur charges.

Resources you used:

• The dedicated Google Cloud project with the Vertex AI API enabled, if you created a new project in Step 1.
• The ~/langgraph-incident-triage-v2 folder in Cloud Shell. This contains the workflow code, fixture data, and .venv virtual environment.
• The triage_checkpoints.db SQLite database containing checkpoints, pending interrupts, findings, and approval decisions.
• The generated runbooks/ directory, if your workflow published a runbook.

Keep everything running

No action is needed. Choose this if you want to inspect saved incidents or continue experimenting with the workflow.

• Your Google Cloud project remains available for later LangGraph work.
• Your ~/langgraph-incident-triage-v2 folder retains the application code and fixture data.
• Your triage_checkpoints.db database retains every saved thread and pending approval.
• Future Gemini requests can incur charges when you run the workflow again.

Pause - I'll come back to this later

Pause your work by leaving the cloud project and saved workflow state intact. No new model charges occur while you make no new requests.

• Stop running new incident-triage commands.
• End your current Cloud Shell session when you finish.
• Leave the Google Cloud project in place.
• Leave the ~/langgraph-incident-triage-v2 folder in place.
• Return to the saved checkpoint database when you are ready to continue.

Delete - I don't want to use this again

Remove the Cloud Shell artifacts you created for this project. Your cloud cleanup depends on whether Step 1 created a dedicated project or reused the Part 1 project.

I created a dedicated project

This path removes the local workflow files and the dedicated Google Cloud project.

Check Before Deleting

Deleting these resources is permanent. Your saved checkpoints and generated runbooks cannot be recovered after the local folder is removed.

• Exit the active virtual environment and remove the Cloud Shell repository by running these commands:

deactivate
cd ~
rm -rf ~/langgraph-incident-triage-v2

What Do These Commands Do?

• The deactivate command exits the active Python virtual environment.
• The cd ~ command moves the shell outside the repository before deletion.
• The final command deletes ~/langgraph-incident-triage-v2. This also removes .venv, triage_checkpoints.db, and runbooks/.

• Refresh Cloud Shell's file browser.
• Confirm that langgraph-incident-triage-v2 is no longer listed.

Still See the Project Folder?

Check that the command targeted ~/langgraph-incident-triage-v2. A different path leaves the repository untouched.

Help me diagnose why Cloud Shell did not remove my incident-triage folder.

• Open the Manage resources page in the Google Cloud console.
• Select the dedicated project you created in Step 1.
• Click Delete in the toolbar.
• Enter the dedicated project ID when the confirmation dialog requests it.
• Click Shut down to submit the deletion request.
• Confirm that the project enters its pending-deletion state.

Cannot Delete the Cloud Project?

Project deletion can be blocked by account permissions or an organization policy. Confirm that your account has permission to delete the selected project.

Help me diagnose why I cannot delete my dedicated Google Cloud project.

I reused the Part 1 project

This path preserves the shared Google Cloud project. It removes only the local Part 2 repository and its saved workflow artifacts.

Preserve the Shared Project

The reused project may still support your Part 1 workflow. Keep that cloud project in place while you remove this project's Cloud Shell folder.

• Exit the active virtual environment and remove the Cloud Shell repository by running these commands:

deactivate
cd ~
rm -rf ~/langgraph-incident-triage-v2

What Do These Commands Do?

• The deactivate command exits the active Python virtual environment.
• The cd ~ command moves the shell outside the repository before deletion.
• The final command deletes ~/langgraph-incident-triage-v2. This also removes .venv, triage_checkpoints.db, and runbooks/.

• Refresh Cloud Shell's file browser.
• Confirm that langgraph-incident-triage-v2 is no longer listed.
• Confirm that the reused Google Cloud project remains available for your Part 1 resources.

Still See the Project Folder?

Check that the command targeted ~/langgraph-incident-triage-v2. A different path leaves the repository untouched.

Help me diagnose why Cloud Shell did not remove my incident-triage folder.

Nice Work!

Nice Work!

Excellent work! You have completed Part 2 of this LangGraph incident response build. Your system now turns fixture alerts into durable runbooks backed by specialist evidence and team-owned decisions.

You've learned how to:

• Build role-specific specialists that inspect logs, deploys, and metrics before returning typed findings.
• Route alerts through parallel investigations with Send, reducers, deferred synthesis, and checkpoint recovery.
• Enforce team-owned approvals before publishing an audited runbook containing permitted actions.
• Secret Mission: Build a read-only incident status board that inspects saved findings, decisions, pending teams, next nodes, and runbook state.

Ready to quiz yourself?