Connect the Triage Workflow to Linear
Connect a LangGraph triage workflow to Linear with inspectable approvals.
Introduction
30 Second Summary
When checkout breaks and customers can't pay, the urgent fix is often approved in a terminal and never written into the team's tracker. The next person on call can't tell what was approved, rejected, or already done.
In this project, you will extend a supplied Python starter built with LangGraph. You will run it in Google Cloud Shell and connect it to Linear, where the incident, findings, approvals, and runbook stay visible.
What You'll Build
You will turn the supplied INC-2041 fixture into a Linear incident. As the terminal workflow runs, you will see its state, findings, approval sub-issues, and final runbook update in one place.
By the end of this project, you'll have:
- A visible incident journey through Triage, Investigating, Awaiting Approval, and Done, even when model wording varies.
- An inspectable approval record with one findings comment and one reusable sub-issue for each gated action.
- A synchronized runbook outcome where CLI approvals update sub-issues, complete the parent incident, and publish one final runbook comment.
- Secret Mission: Reject every pending team on a fresh incident, return the parent to Triage, and prove that no runbook is published.
Are there any prerequisites?
- Complete Triage Incidents with LangGraph and Observe Agent Workflows with Langfuse before starting.
- Keep your existing Google Cloud Shell environment ready with the Google Cloud and Langfuse configuration from those projects.
Before We Start
Incident decisions become hard to trust when they remain scattered across terminal output and memory. Before the hands-on work, commit to building a workflow that records each decision in Linear, the shared issue tracker for this project.
Set Up Linear and the Starter Repository
Small setup differences can break an incident workflow. Starting with tested code makes later problems easier to find.
The supplied langgraph-triage-linear starter repository contains the existing LangGraph workflow and the files needed to run it. By the end, the starter test suite passes in Google Cloud Shell, and the INC-2041 fixture alert appears as an incident in Linear.
In this step, get ready to:
- Clone the starter repository and pass its existing tests.
- Create a Linear Free workspace and connect a personal API key.
- Configure the workspace and seed the first Linear incident.
Clone and test the starter
Google Cloud Shell provides the Python runtime and persistent home storage needed for this project. The starter repository uses Git to preserve the compatibility-tested dependency pins from the earlier workflow.
- Open the Google Cloud console in your browser.
- Click Activate Cloud Shell in the top toolbar.
- If an Authorize Cloud Shell dialog appears, click Authorize.
- Wait for the terminal prompt. Startup usually takes a few seconds while Google prepares the session.
You now have an authenticated terminal connected to your existing Google Cloud environment.
- Check the project ID shown in yellow in your terminal prompt. It should be the Google Cloud project you used in Observe Agent Workflows with Langfuse.
- If the prompt shows a different project, switch to it by running:
gcloud config set project [[PROJECT_ID="your Google Cloud project ID"]]Why does the project matter?
Later in this step you copy GOOGLE_CLOUD_PROJECT from your earlier .env file. Matching the terminal to that project keeps your Vertex AI calls and costs in one place.
- Clone the starter repository and move into it by running these commands:
cd ~
git clone https://github.com/nextwork-projects/langgraph-triage-linear.git
cd langgraph-triage-linearWhat do these commands do?
- cd ~ moves the terminal to Google Cloud Shell's persistent home directory.
- git clone downloads the starter into a new langgraph-triage-linear folder.
- The final command moves the terminal into that folder so later commands use the correct files.
- Check that your terminal prompt now shows the langgraph-triage-linear folder.
Repository did not clone?
Confirm that the repository URL matches the command exactly. A network interruption can also stop the download before the folder is created.
Ask for help with the clone failure.
A virtual environment keeps this repository's Python packages separate from Cloud Shell's system packages. Activating it ensures every installation and test uses the same isolated environment.
- Create the .venv virtual environment and activate it by running:
python3 -m venv .venv && source .venv/bin/activateWhat does this command do?
Python creates an isolated environment inside .venv. The activation step updates the current shell so Python packages install into that environment.
- Check that your terminal prompt now begins with (.venv).
Virtual environment not active?
Make sure the terminal is still inside langgraph-triage-linear. Check that the .venv creation command completed before activation ran.
Ask for help activating the environment.
The repository pins the package versions tested with its starter workflow and observability code. Dependency installation can take a few minutes while Cloud Shell downloads those packages.
- Install the pinned dependencies from requirements.txt by running:
python3 -m pip install -r requirements.txtWhy use the requirements file?
The requirements.txt file defines the compatibility-tested environment for this repository. Keeping those pins avoids introducing package changes that the starter tests do not cover.
The installation finishes when the terminal returns to the (.venv) prompt without a package installation failure.
Dependency installation failed?
Confirm that (.venv) appears at the start of your prompt. A temporary package download failure can be retried with the same installation command.
Ask for help with the installation output.
Before you test the starter, what result would prove that the starter workflow still works in this fresh environment?
- Run the starter test suite by running:
python3 -m pytestWhat does this test prove?
The test suite checks the starter workflow against its expected starter behavior. A passing run confirms that the repository and installed dependencies work together.
You should see a test summary showing that all starter tests passed. Solid start. The starter workflow is working inside your new virtual environment.
Tests did not pass?
Confirm that you ran the tests from the langgraph-triage-linear folder. Confirm that (.venv) still appears in the terminal prompt.
Ask for help interpreting the failing test.
Create the workspace and connect your key
A personal API key authenticates the supplied Linear client. Creating this key is the sensitive part of the setup, so you paste it only into the local .env file.
Have a Linear account
- Open Linear in your browser and sign in.
- Select your workspace name in the top-left corner.
- Hover over Switch workspace and select Create or join a workspace.
- Create a new workspace named Triage Workflow on the Free plan.
Need a Linear account
Linear Free costs $0 and includes 2 teams and 250 issues, which covers this project.
- Open Linear's pricing page in your browser.
- Choose the Free plan.
- Create your Linear account.
- Name the new workspace Triage Workflow.
Why a new workspace?
The setup script creates teams, so it needs a workspace where you're an admin. On the Free plan, every member is an admin. A new Triage Workflow workspace also gets a default team with the TRI key, which the script renames to Incidents. A work workspace can block team creation.
- Open your Linear Security & Access settings.
- In the Personal API keys section, create a key named Triage Workflow.
- Give the key full access so the setup script can create teams, workflow states, labels, issues, and comments.
- Copy the key and keep this tab open until you paste it into .env. Don't capture the key in a screenshot.
- Switch back to the active Google Cloud Shell terminal.
- Create the local .env file from the repository template by running:
cp .env.example .envWhy copy the environment template?
The template provides the environment variable names expected by the project. Your local .env copy holds the values used by this Google Cloud Shell repository.
- Open .env in the Google Cloud Shell Editor by running this command:
cloudshell edit .envWhat does this command do?
The command opens .env in the Google Cloud Shell Editor while keeping the terminal session available.
You should see the environment variable template from the repository.
Environment file did not open?
Confirm that the terminal is still inside langgraph-triage-linear and that .env exists beside .env.example.
Help me open the environment file without exposing its contents.
Your Langfuse project's .env file already holds three values this project needs: your Google Cloud project ID and your two Langfuse keys.
- Open the earlier project's .env in a second editor tab by running:
cloudshell edit ~/langgraph-triage-observability/.envDeleted your Langfuse keys or the earlier folder?
If you revoked your keys during the earlier project's clean up:
- Open your project in Langfuse Cloud US
- Go to Settings > API Keys, and create a new key pair
- Paste the public and secret keys into .env
- Copy GOOGLE_CLOUD_PROJECT, LANGFUSE_PUBLIC_KEY, and LANGFUSE_SECRET_KEY from that file into the same lines of your new .env.
- Keep LANGFUSE_BASE_URL and LANGFUSE_TRACING_ENVIRONMENT as they are. The template already has the right values.
- Leave LINEAR_WEBHOOK_SECRET, DATABASE_URL, GITHUB_TOKEN, GITHUB_REPO, and LINEAR_READONLY_KEY unchanged for this project.
- Save .env.
The editor should now show the retained configuration as saved.
Changes did not stay saved?
Confirm that the edited value remains visible after you move the cursor to another line. If it disappears, save .env again before adding the Linear key.
Help me verify that the environment file is saved safely.
Keep the API key and .env out of screenshots. Store the key only in .env inside your Google Cloud Shell project folder.
- Paste your Linear key after LINEAR_API_KEY=.
- Save .env.
- Export every value from .env into the current Google Cloud Shell session by running:
set -a && source .env && set +aWhat does this command do?
The command temporarily enables automatic exporting while the shell reads .env. It then disables that behavior so later shell variables keep their normal behavior.
The terminal should return to the (.venv) prompt without displaying your credentials.
Environment values did not export?
Check .env for spaces around variable names or unmatched quotes. Keep the key value on the same line as LINEAR_API_KEY=.
Help me diagnose the environment export without sharing credentials.
Shape and seed the workspace
The workspace needs known teams and states before the graph can make deterministic updates. The setup script creates that shared structure before any incident enters the workflow.
- Configure the Linear workspace by running:
python3 linear_setup.pyWhat does the setup script create?
The script creates or reuses the Incidents team and the Approvals team. It also prepares the workflow states and labels used by later graph nodes.
You should see setup output confirming that the required workspace items were created or reused.
Workspace setup stopped?
If you see LINEAR_API_KEY is not set., the current shell did not receive the value from .env. Check that the key uses the exact LINEAR_API_KEY name.
If you see Access denied, Linear refused to create a team. Check that your key was created in the Triage Workflow workspace and has full access.
Ask for help without sharing the key.
- Switch back to your Linear workspace.
- Select the Incidents team in the sidebar, then select Team settings under Go to.
- Select Issue statuses & automations.
- Confirm that the workflow includes Triage, Investigating, Awaiting Approval, and Done. You'll also see Fixing, which a later project in this series uses, and Linear's default states. Leave them as they are.
- Open the Approvals team's settings the same way, then select Issue statuses & automations.
- Confirm that the workflow includes Needs Approval, Approved, and Rejected, alongside Linear's default states.
- Open Settings, then select Labels under Workspace.
- Confirm that the label list includes the severity and team label groups.
Before you seed the incident, what three details do you expect the script to print for the new Linear issue?
- Create a fresh incident from the INC-2041 fixture by running:
python3 seed_incidents.py INC-2041What does the seed command do?
The script creates a new Linear incident from the selected fixture. Each run creates a fresh issue that can travel through the workflow independently.
You should see an issue ID such as TRI-5, the fixture alert ID INC-2041, and a Linear URL in the terminal output.
No incident URL printed?
Confirm that the workspace setup completed successfully. Confirm that the current shell still has the .env values exported.
Ask for help with the seed command.
- Record the printed TRI issue ID here: printed TRI issue ID.
- Export that issue ID for later steps by running this command:
export ISSUE_ID=[[ISSUE_ID="printed TRI issue ID"]]Why save the issue ID?
The ISSUE_ID environment variable keeps the new issue ID available to later commands in this shell. It becomes the stable reference for starting and resuming the incident workflow.
- Open the Linear URL printed by the seed command.
- Confirm that the incident is in Triage.
- Read the incident description to confirm that it contains the INC-2041 fixture alert metadata.
Incident not visible in Triage?
Confirm that the URL belongs to the same workspace configured by linear_setup.py. If the issue is missing, rerun the seed command and open the newly printed URL.
Help me find the seeded incident without sharing credentials.
Setup complete. Your tested repository can now address a seeded incident through a configured Linear workspace.
Setup checkpoint
Your active Google Cloud Shell session is inside langgraph-triage-linear with .venv activated. The starter dependencies are installed with the existing tests passing.
Your .env configuration is exported without exposing its credentials. Linear now contains the configured teams and a seeded incident referenced by ISSUE_ID.
Your deterministic system of record is ready. Next, you will teach the graph to pick up this Linear incident and move it into active investigation.
Pick Up a Linear Incident
Your starter project and Linear incident are ready. The workflow still cannot open that incident because it only accepts sample alerts.
You will update app.py, the main workflow file, so LangGraph can read the Linear issue and start investigating it. Success means the parent moves from Triage to Investigating, the terminal reaches WORKFLOW PAUSED:, and no approval sub-issues exist yet.
In this step, get ready to:
- Extend the graph state with Linear issue context.
- Translate a Linear incident into graph input.
- Run the incident until the approval gates pause.
Extend the graph state and client boundary
The graph needs Linear's internal issue ID for external writes, while the issue ID such as TRI-5 names the checkpoint thread used by terminal commands. This separation lets deterministic graph nodes update the correct external issue.
Why deterministic writes?
The supplied client sends controlled GraphQL operations for issue reads and state changes. Keeping these writes in graph nodes makes every external change inspectable.
- Open app.py in the Google Cloud Shell Editor by running this command:
cloudshell edit app.py- Find the import telemetry line near the top of app.py.
- Replace that line with this import group:
import telemetry
import linear as linear_api
from linear import APPROVALS_TEAM, INCIDENTS_TEAM, Linear, alert_fieldsWhat do these imports provide?
- linear_api gives the graph access to the supplied deterministic client factory.
- INCIDENTS_TEAM identifies the Linear team whose incident state changes in this step.
- Linear documents the client returned by linear_client().
- alert_fields extracts fixture metadata from a seeded issue description.
- Find the existing IncidentState class in app.py.
- Replace the class with this Linear-aware version:
class IncidentState(TypedDict, total=False):
incident_id: str
issue_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
skip_specialists: list[str]What does this state change do?
issue_id stores Linear's internal issue ID for API writes. It differs from the issue ID, such as TRI-5, that names the checkpoint thread.
- Save app.py.
- Confirm the state change preserves the starter behavior by running:
python3 -m pytestWhat should happen?
The starter tests should pass. This proves the extra state field has not changed the existing fixture workflow.
🙋♀️ Tests failing after the state edit?
Check that issue_id: str sits directly below incident_id: str. Keep every existing IncidentState field in the class.
Use this prompt if the failure continues: help me compare my IncidentState class with the required app.py structure
Approval tasks also need space for external issue references. These fields stay optional because the current terminal gates still support fixture-only runs.
- Find the existing ActionTask class in app.py.
- Replace ActionTask with this version.
- Add linear_client() directly below the class:
class ActionTask(TypedDict, total=False):
incident_id: str
team: str
service: str
action: str
policy_reason: str
approval_issue_id: str
approval_issue: str
def linear_client() -> Linear:
return linear_api.client()What does this boundary do?
ActionTask can now carry approval issue references once those issues exist. linear_client() gives graph nodes one reusable route to the authenticated client.
- Save app.py.
- Verify the client boundary preserves the existing workflow by running:
python3 -m pytestWhat should happen now?
The tests should pass again. Your graph state can now represent Linear-backed work without requiring it for fixture runs.
🙋♀️ Seeing an import or type error?
Check that both Linear import lines sit immediately below import telemetry. Confirm that ActionTask includes total=False.
Use this prompt for a focused comparison: help me debug the Linear imports and ActionTask fields in app.py
Turn a Linear issue into graph input
A Linear issue contains the fixture alert ID inside its description. Two input helpers preserve the graph's existing alert shape while adding the parent's internal issue ID.
- Find main() near the bottom of app.py.
- Add incident_input() directly below main():
def incident_input(
thread: str,
alert: dict,
issue: dict | None = None,
skip=(),
) -> dict:
return {
"incident_id": thread,
"issue_id": issue["id"] if issue else "",
"incident": f"{alert['title']}: {alert['description']}",
"alert": alert,
"status": "new",
"audit_log": [],
"skip_specialists": list(skip),
}What does this helper preserve?
- thread stores the durable checkpoint thread.
- issue_id stores Linear's internal issue ID when an issue is supplied.
- alert keeps the fixture alert shape expected by the router and specialists.
- skip_specialists preserves the existing fixture-only testing option.
- Save app.py.
- Check the new helper against the existing tests by running:
python3 -m pytestWhat should happen?
The tests should pass. The new helper centralizes graph input without changing how the existing fields are populated.
🙋♀️ Seeing a syntax error in incident_input?
Check the parentheses around the function parameters. Confirm that every dictionary entry remains inside the returned braces.
Ask for help with the exact function: help me find the syntax problem in my incident_input function
- Add issue_input() directly below incident_input():
def issue_input(
issue_ref: str,
thread: str | None = None,
skip=(),
) -> tuple[str, dict]:
issue = linear_client().issue(issue_ref)
alert_id = alert_fields(issue["description"])["alert"]
alert = {
candidate["id"]: candidate for candidate in load_json("alerts.json")
}[alert_id]
print(f"Picked up {issue['identifier']}: {issue['title']} ({alert_id})")
thread = thread or issue["identifier"]
return thread, incident_input(thread, alert, issue, skip)How does issue_input work?
- The client loads the referenced Linear issue.
- alert_fields() reads the seeded alert ID from its description.
- The fixture lookup restores the full alert object expected by the existing graph.
- The issue ID names the default checkpoint thread.
- Save app.py.
- Confirm the issue translation helper is valid by running:
python3 -m pytestWhat should happen now?
The tests should pass. The issue path can now produce the same input shape as the fixture path.
🙋♀️ Issue helper tests not passing?
Confirm that alert_fields() receives issue["description"]. Check that the alert lookup still reads alerts.json.
Use this prompt to inspect the data translation: help me trace how issue_input turns Linear metadata into incident_input
The command parser needs two mutually exclusive entry points. A run can now start from one Linear issue or one fixture alert.
- Find the start command arguments inside parse_args().
- Replace the required thread and alert arguments with this source selector:
start_parser = subparsers.add_parser("start")
start_parser.add_argument("--thread", help="Defaults to the issue identifier.")
source = start_parser.add_mutually_exclusive_group(required=True)
source.add_argument("--issue", help="A Linear incident, such as TRI-5.")
source.add_argument("--alert", help="A fixture alert, without Linear.")What does the source selector enforce?
The mutually exclusive group requires exactly one source. --issue activates Linear-backed input while --alert preserves the previous fixture route.
- Find the validation logic near the end of parse_args().
- Add these checks before return args:
if args.command == "resume" and args.decision and not args.team:
parser.error("--decision needs --team.")
if args.command == "start" and not (args.thread or args.issue):
parser.error("--alert needs --thread.")
if args.command == "start" and not args.thread:
args.thread = args.issue
return argsWhy default the thread?
An issue ID is stable across the start and resume commands. Using it to name the checkpoint thread ties durable checkpoints to the incident operators can see.
- Save app.py.
- Verify both command routes still parse correctly by running:
python3 -m pytestWhat should happen?
The tests should pass. The parser now accepts the Linear route without removing the fixture route.
🙋♀️ Parser tests failing?
Check that --issue and --alert belong to the same mutually exclusive group. Keep --thread on start_parser as an optional argument.
Use this prompt for parser help: help me debug the mutually exclusive start arguments in parse_args
The execution logic is easier to reuse when streaming and scoring live in one helper. This keeps source selection separate from graph execution.
- Add the first part of run_graph() directly below issue_input():
def run_graph(graph, thread: str, command: str, graph_input) -> dict:
config = {"configurable": {"thread_id": thread}}
before = graph.get_state(config).values
started = graph_input if isinstance(graph_input, dict) else before
alert_id = started["alert"]["id"]
trace_extra, handler = telemetry.trace_config(
thread,
command,
alert_id,
started.get("skip_specialists", []),
)
try:
stream = graph.stream_events(
graph_input,
config={**config, **trace_extra},
durability="sync",
version="v3",
)
display_stream(stream)
except Exception as error:
display_failure(graph, config, error)What does this execution layer handle?
run_graph() loads the selected thread before streaming the graph. It also keeps the existing trace configuration and failure snapshot behavior around every command.
- Save app.py.
- Confirm the first execution layer is valid by running:
python3 -m pytestWhat should happen?
The tests should pass. The helper is valid even before the existing scoring logic moves into it.
🙋♀️ Seeing an indentation error?
Confirm that except Exception as error: aligns with try:. Keep the stream configuration nested inside the try block.
Use this prompt to inspect the block structure: help me fix the indentation in run_graph
- Add the scoring section directly below the display_failure() line inside run_graph():
values = graph.get_state(config).values
if handler and handler.last_trace_id:
url = telemetry.trace_url(handler.last_trace_id)
print(f"Trace: {url}")
if values.get("issue_id") and url:
linear_client().attach(
values["issue_id"],
url,
f"Langfuse trace: {command}",
thread,
)
if "analysis" in values and "analysis" not in before:
correct = score_run(
telemetry.langfuse_client(),
handler.last_trace_id,
alert_id,
values["analysis"],
)
print(f"Score root_cause_correct: {correct}")
return valuesThe trace link now also lands on the Linear incident, so a responder can open the run's trace from the issue. Each start or resume command attaches its own trace, titled with the command.
What stays unchanged?
The helper still prints the Langfuse trace link after execution. A new analysis still receives the existing root_cause_correct score.
- Save app.py.
- Verify the completed execution helper by running:
python3 -m pytestWhat should happen now?
The tests should pass. Streaming and scoring now share one execution path for start and resume commands.
🙋♀️ Scoring tests failing?
Check that values is loaded after the exception handler. Confirm that return values remains inside run_graph().
Use this prompt for a focused review: help me debug the scoring section in run_graph
- Find the source-selection section inside run_command().
- Replace the fixture-only start branch with this dual-source branch:
if args.command == "start" and args.issue:
args.thread, graph_input = issue_input(
args.issue,
args.thread,
args.skip,
)
elif args.command == "start":
alerts = {
alert["id"]: alert for alert in load_json("alerts.json")
}
graph_input = incident_input(
args.thread,
alerts[args.alert],
skip=args.skip,
)How are both start paths preserved?
The first branch loads a Linear issue through issue_input(). The second branch still loads a fixture and passes it through the shared incident_input() helper.
- Save app.py.
- Check the source-selection edit by running:
python3 -m pytestWhat should happen?
The tests should pass. Existing fixture starts still work while the new issue branch is ready for its graph node.
🙋♀️ Fixture tests failing now?
Confirm that the fixture branch still reads alerts.json. Pass args.alert into the lookup before calling incident_input().
Use this prompt to compare the branches: help me debug the Linear and fixture branches in run_command
- Remove the old tracing and streaming section from the bottom of run_command(), from alert_id = ... to the end of the function.
- Delete the line before = graph.get_state(config).values near the top of run_command(). run_graph() now reads that state itself.
- Add this shared execution call at the end of the function:
run_graph(graph, args.thread, args.command, graph_input)Why use one execution call?
Every command now reaches the same checkpoint, tracing, streaming, and scoring logic. Source selection only decides what enters the graph.
- Save app.py.
- Verify the completed input path by running:
python3 -m pytestWhat should happen now?
The full starter test suite should pass. Your CLI can now build graph input from either supported source.
🙋♀️ Seeing duplicate trace output?
Check that the old telemetry.trace_config() and graph.stream_events() section no longer remains inside run_command(). Those calls now belong only to run_graph().
Use this prompt to find duplicated execution logic: help me remove duplicate streaming logic from run_command
Pick up the incident and inspect the pause
The graph still begins at the router, so Linear cannot show that work has started. A pickup node creates that external state change before any specialist runs.
- Find incident_brief() in app.py.
- Add pick_up_incident() directly below it:
def pick_up_incident(state: IncidentState):
if state.get("issue_id"):
linear_client().move(state["issue_id"], INCIDENTS_TEAM, "Investigating")
return {"status": "investigating"}What does the pickup node guarantee?
Linear-backed runs move the parent incident to Investigating. Fixture-only runs skip the external write while receiving the same internal status.
- Find the first node registrations inside build_graph().
- Add the pickup node before the router registration:
builder = StateGraph(IncidentState)
builder.add_node("pick_up_incident", pick_up_incident)
builder.add_node("route_incident", route_incident)What does node registration do?
Registration gives the graph a named deterministic operation. The next edit places that operation on the path from START to the router.
- Save app.py.
- Check the node definition and registration by running:
python3 -m pytestWhat should happen?
The tests should pass. The node exists without changing execution order until its edges are connected.
🙋♀️ Graph compilation failing?
Confirm that the registered name is pick_up_incident. Check that the second argument references the function without parentheses.
Use this prompt to inspect the registration: help me debug the pick_up_incident node registration
- Find the existing edge from START to route_incident.
- Replace that edge with this two-edge path:
builder.add_edge(START, "pick_up_incident")
builder.add_edge("pick_up_incident", "route_incident")How does the graph start now?
Every run enters pick_up_incident() first. The node performs a Linear move only when issue_id contains an external issue reference.
- Save app.py.
- Confirm the completed graph still passes its tests by running:
python3 -m pytestWhat should happen?
The tests should pass. The existing investigation path now has one deterministic pickup node in front of it.
🙋♀️ Tests stop before the router?
Check that the first edge ends at pick_up_incident. Confirm that the second edge begins there and ends at route_incident.
Use this prompt to inspect the edge order: help me fix the START path through pick_up_incident
Before you start the Linear-backed run, what external state do you expect the pickup node to change before the workflow pauses?
- Start the seeded Linear incident by running:
python3 app.py start --issue "$ISSUE_ID"What should you see?
The terminal first prints a line beginning with Picked up and names your TRI issue. Specialist progress follows before the run reaches WORKFLOW PAUSED:.
The run also prints its trace and score after creating a durable local checkpoint. Under WORKFLOW PAUSED:, look for the pending approvals, such as one for payments and one for platform-data. The teams and the analysis wording can vary with the model's output.
LangChainBetaWarning lines and a warning about non-text parts are expected. You can ignore them.
🙋♀️ The issue run did not reach the pause?
If the client reports LINEAR_API_KEY is not set., return to the exported .env session from the previous step. If the issue cannot be found, confirm that $ISSUE_ID still contains the seeded TRI issue ID.
Use this prompt with your terminal output: help me troubleshoot why my Linear-backed workflow did not pause
- Switch back to the seeded incident in Linear.
- Confirm its current workflow state is Investigating.
- Confirm the incident shows a Langfuse trace: start link under its description.
- Select the Approvals team in Linear.
- Confirm its issue list contains no approval sub-issues for this incident.
That is the entry boundary working: your local graph has picked up a visible Linear incident and paused on its durable approval gates.
The empty Approvals team is the intended shortfall. The graph still keeps team decisions inside terminal interrupts, so external approval work does not exist yet.
✔️ Awesome, I've got everything!
Great work. Save app.py and keep the paused $ISSUE_ID thread for the next step.
ⓧ I'd like to double check the full code
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
import telemetry
import linear as linear_api
from linear import APPROVALS_TEAM, INCIDENTS_TEAM, Linear, alert_fields
from prompts import analyst_messages
from score import score_run
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
issue_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
skip_specialists: list[str]
class SpecialistTask(TypedDict):
specialist: str
brief: str
class ActionTask(TypedDict, total=False):
incident_id: str
team: str
service: str
action: str
policy_reason: str
approval_issue_id: str
approval_issue: str
def linear_client() -> Linear:
return linear_api.client()
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 pick_up_incident(state: IncidentState):
if state.get("issue_id"):
linear_client().move(state["issue_id"], INCIDENTS_TEAM, "Investigating")
return {"status": "investigating"}
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)),
]
)
skipped = state.get("skip_specialists", [])
decision["specialists"] = [
name for name in decision["specialists"] if name not in skipped
]
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)
prompt, messages = analyst_messages(state["incident"], findings)
analysis = analyst.invoke(
messages,
config={"metadata": {"langfuse_prompt": prompt}} if prompt else None,
)
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 = [
approval
for approval in approvals
if approval["decision"] in ("approve", "not required")
]
decisions = [
f"Team {approval['team']} decision for {approval['service']}: {approval['decision']}."
for approval 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"- {finding['specialist']} ({finding['severity']}): {finding['summary']}"
for finding in state["findings"]
)
action_lines = "\n".join(
f"- {approval['team']} / {approval['service']}: {approval['action']} "
f"({'approved by team' if approval['decision'] == 'approve' else 'approval not required'})"
for approval 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("pick_up_incident", pick_up_incident)
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, "pick_up_incident")
builder.add_edge("pick_up_incident", "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", help="Defaults to the issue identifier.")
source = start_parser.add_mutually_exclusive_group(required=True)
source.add_argument("--issue", help="A Linear incident, such as TRI-5.")
source.add_argument("--alert", help="A fixture alert, without Linear.")
start_parser.add_argument(
"--skip",
nargs="*",
default=[],
choices=list(SPECIALISTS),
help="Specialists to hold back, to test that the score catches a weak run.",
)
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.")
if args.command == "start" and not (args.thread or args.issue):
parser.error("--alert needs --thread.")
if args.command == "start" and not args.thread:
args.thread = args.issue
return args
def main():
args = parse_args()
config = {"configurable": {"thread_id": args.thread}}
try:
run_command(args, config)
finally:
telemetry.shutdown()
def incident_input(
thread: str,
alert: dict,
issue: dict | None = None,
skip=(),
) -> dict:
return {
"incident_id": thread,
"issue_id": issue["id"] if issue else "",
"incident": f"{alert['title']}: {alert['description']}",
"alert": alert,
"status": "new",
"audit_log": [],
"skip_specialists": list(skip),
}
def issue_input(
issue_ref: str,
thread: str | None = None,
skip=(),
) -> tuple[str, dict]:
issue = linear_client().issue(issue_ref)
alert_id = alert_fields(issue["description"])["alert"]
alert = {
candidate["id"]: candidate for candidate in load_json("alerts.json")
}[alert_id]
print(f"Picked up {issue['identifier']}: {issue['title']} ({alert_id})")
thread = thread or issue["identifier"]
return thread, incident_input(thread, alert, issue, skip)
def run_graph(graph, thread: str, command: str, graph_input) -> dict:
config = {"configurable": {"thread_id": thread}}
before = graph.get_state(config).values
started = graph_input if isinstance(graph_input, dict) else before
alert_id = started["alert"]["id"]
trace_extra, handler = telemetry.trace_config(
thread,
command,
alert_id,
started.get("skip_specialists", []),
)
try:
stream = graph.stream_events(
graph_input,
config={**config, **trace_extra},
durability="sync",
version="v3",
)
display_stream(stream)
except Exception as error:
display_failure(graph, config, error)
values = graph.get_state(config).values
if handler and handler.last_trace_id:
url = telemetry.trace_url(handler.last_trace_id)
print(f"Trace: {url}")
if values.get("issue_id") and url:
linear_client().attach(
values["issue_id"],
url,
f"Langfuse trace: {command}",
thread,
)
if "analysis" in values and "analysis" not in before:
correct = score_run(
telemetry.langfuse_client(),
handler.last_trace_id,
alert_id,
values["analysis"],
)
print(f"Score root_cause_correct: {correct}")
return values
def run_command(args, config):
with SqliteSaver.from_conn_string(DB_PATH) as checkpointer:
graph = build_graph(checkpointer)
if args.command == "start" and args.issue:
args.thread, graph_input = issue_input(
args.issue,
args.thread,
args.skip,
)
elif args.command == "start":
alerts = {
alert["id"]: alert for alert in load_json("alerts.json")
}
graph_input = incident_input(
args.thread,
alerts[args.alert],
skip=args.skip,
)
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)
print(f"Recorded {args.decision} for {args.team}.")
else:
graph_input = None
run_graph(graph, args.thread, args.command, graph_input)
if __name__ == "__main__":
main()Compare the changed imports, state types, input helpers, parser, pickup node, graph edges, and command routing with your file.
Your workflow now starts from a real Linear incident while preserving the fixture path. Next, you will turn those hidden terminal gates into visible approval work under the parent incident.
Publish Findings and Approval Work
Your workflow can now open the Linear incident and start investigating it. However, the findings and approval requests only appear in the terminal when LangGraph pauses.
You will update app.py so Linear shows one findings comment and one reusable approval sub-issue for every gated action. Success means the parent is Awaiting Approval with a severity label, one findings comment, and the expected approval sub-issues.
In this step, get ready to:
- Create one reusable approval sub-issue for every gated action.
- Publish the computed findings to the parent incident.
- Run a fresh incident until Linear shows the approval work.
Create reusable approval sub-issues
LangGraph can retry a node within a checkpoint thread. Approval creation therefore needs an idempotent marker derived from a stable approval key, so the workflow can reuse an existing approval sub-issue.
- Open app.py in the Cloud Shell Editor by running this command:
cloudshell edit app.py- In app.py, add import hashlib directly below import argparse.
- Add import re directly below import operator.
- Find policy_check(), then add these approval helpers directly below it:
APPROVAL_LINE = re.compile(r"^Approval: (\w+)$", re.MULTILINE)
def approval_key(action: dict) -> str:
text = f"{action['team']}|{action['service']}|{action['action']}"
return hashlib.sha1(text.encode()).hexdigest()[:8]
def short_title(text: str, limit: int = 120) -> str:
if len(text) <= limit:
return text
return text[: limit - 1].rsplit(" ", 1)[0].rstrip(",;:") + "…"
def approval_description(state: IncidentState, action: dict) -> str:
return (
f"Approval: {approval_key(action)}\n"
f"Incident: {state['incident_id']}\n"
f"Team: {action['team']}\n"
f"Service: {action['service']}\n\n"
f"**Action:** {action['action']}\n\n"
f"**Why it needs approval:** {action['policy_reason']}\n\n"
"Move this issue to **Approved** or **Rejected** to decide."
)How Does the Approval Marker Work?
- The SHA-1 digest turns the team, service, and proposed action into a stable eight-character approval key.
- The approval marker combines the Approval: prefix with that key inside each approval sub-issue description.
- A retry searches for the marker before creating an approval sub-issue, which prevents duplicates.
- The title helper shortens the visible issue title without changing the full action in the description.
- Add the first part of create_approval_subissues() directly below approval_description().
def create_approval_subissues(state: IncidentState):
gated = [action for action in state["actions"] if action["requires_approval"]]
if not state.get("issue_id") or not gated:
return {}
linear = linear_client()
existing = {}
for child in linear.sub_issues(state["issue_id"]):
match = APPROVAL_LINE.search(child["description"] or "")
if match:
existing[match.group(1)] = child
approvals_team = linear.team(APPROVALS_TEAM)
actions = []Why Look Up Approval Sub-Issues First?
The lookup builds a map from approval keys to existing approval sub-issues. A retry can then reuse a matching issue instead of creating a duplicate.
- Continue create_approval_subissues() directly after actions = [] with this code:
for action in state["actions"]:
if not action["requires_approval"]:
actions.append(action)
continue
child = existing.get(approval_key(action)) or linear.create_issue(
teamId=approvals_team["id"],
parentId=state["issue_id"],
title=short_title(f"Approve for {action['team']}: {action['action']}"),
description=approval_description(state, action),
stateId=linear.state_id(APPROVALS_TEAM, "Needs Approval"),
labelIds=[linear.label_id(action["team"], "team")],
)
actions.append(
action
| {
"approval_issue_id": child["id"],
"approval_issue": child["identifier"],
}
)
identifiers = [
action["approval_issue"]
for action in actions
if action.get("approval_issue")
]
print(f"Approval sub-issues: {', '.join(identifiers)}")
return {"actions": actions}What Does This Node Preserve?
- Gated actions receive the approval sub-issue's internal issue ID for later API writes.
- Gated actions also receive the approval sub-issue's issue ID, such as APR-1, shown to responders.
- Non-gated actions stay in graph state so mixed policy results remain complete.
- Save app.py.
Before you run the tests, do you expect the existing fixture workflow to remain stable after adding this unused node?
- Check the existing behavior by running:
python3 -m pytestWhy Run the Tests Here?
The node is defined but has not entered the graph yet. A passing test run confirms that the new imports and function definitions load without breaking the starter workflow.
You should see the existing test suite pass. That proves the approval creation code loads cleanly before you connect it to the graph.
Seeing a Python Error?
Check that both imports sit with the other imports at the top of app.py. Confirm that the second code block remains indented inside create_approval_subissues().
Compare the quotes around Approval: with the code above if the regular expression fails to load.
Help me debug the approval sub-issue code in app.py.
Report findings to Linear
The approval sub-issues show who must decide. The parent incident also needs the evidence behind those decisions.
- Add findings_comment() directly below create_approval_subissues():
def findings_comment(state: IncidentState) -> str:
analysis = state["analysis"]
finding_lines = "\n".join(
f"- **{finding['specialist']}** ({finding['severity']}): {finding['summary']}"
for finding in state["findings"]
)
action_lines = "\n".join(
f"- {action['team']} / {action['service']}: {action['action']} "
+ (
f"(needs team approval in {action['approval_issue']})"
if action.get("approval_issue")
else "(needs team approval)"
if action["requires_approval"]
else "(no approval needed)"
)
for action in state["actions"]
)
return (
f"**Triage findings: {state['severity']}**\n\n"
f"**Summary:** {analysis['summary']}\n\n"
f"**Likely cause:** {analysis['likely_cause']}\n\n"
f"**Rationale:** {analysis['rationale']}\n\n"
f"**Specialists**\n{finding_lines}\n\n"
f"**Proposed actions**\n{action_lines}"
)What Does the Findings Comment Show?
- The heading records the computed severity at the time of the run.
- The specialist section combines the evidence collected by each selected specialist.
- The action section links gated proposals to the issue IDs of their approval sub-issues.
- Add report_findings() directly below findings_comment():
def report_findings(state: IncidentState):
gated = any(action["requires_approval"] for action in state["actions"])
if state.get("issue_id"):
linear = linear_client()
linear.set_group_label(state["issue_id"], "severity", state["severity"])
if gated:
linear.move(state["issue_id"], INCIDENTS_TEAM, "Awaiting Approval")
linear.comment(state["issue_id"], findings_comment(state))
return {"status": "awaiting_approval" if gated else "reported"}Why Post the Comment Last?
The severity label and state transition are repeatable writes. The findings comment comes last so a failure during an earlier write does not leave duplicate comments after a retry.
These writes stay inside deterministic graph nodes instead of an unrestricted agent tool. The workflow controls every change made to Linear.
- Save app.py.
Before you test again, do you expect the new reporting functions to change any existing run before they become graph nodes?
- Confirm the starter workflow still passes by running:
python3 -m pytestWhat Does This Test Prove?
A passing run confirms that both reporting functions load alongside the existing application. The next substep connects them to the execution path.
You should see the test suite pass again. Your application can now format the parent comment without changing the active graph.
Tests Failing After the Reporting Code?
Check the nested conditional expression in findings_comment() against the code above. Confirm that report_findings() sits at the top level of app.py.
Help me debug the findings reporting functions.
Wire the nodes and run a fresh incident
The new writes need to finish before the interrupt fan-out. This order guarantees that every pending terminal decision already has matching work in Linear.
- In build_graph(), add these node registrations directly below the policy_check registration:
builder.add_node("create_approval_subissues", create_approval_subissues)
builder.add_node("report_findings", report_findings)What Do These Registrations Do?
The registrations make both deterministic write functions available as graph nodes. Their edge order decides when Linear changes happen.
- Find the conditional edge that currently starts from policy_check.
- Replace that edge with this sequence:
builder.add_edge("policy_check", "create_approval_subissues")
builder.add_edge("create_approval_subissues", "report_findings")
builder.add_conditional_edges(
"report_findings", policy_route, ["approval_gate", "publish_runbook"]
)How Does the New Order Protect the Record?
Approval creation runs first so the findings comment can reference the issue IDs of real approval sub-issues. Reporting runs next so the parent reaches Awaiting Approval before the graph pauses.
- Save app.py.
✔️ Awesome, I've got everything!
Your approval creation, findings reporting, and graph wiring are ready. Continue with the fresh incident run.
ⓧ I'd like to double check the full code
Compare your cumulative app.py with this reference. It includes the Linear entry point from the previous step plus this step's approval and reporting nodes.
import argparse
import hashlib
import json
import os
import operator
import re
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
import telemetry
import linear as linear_api
from linear import APPROVALS_TEAM, INCIDENTS_TEAM, Linear, alert_fields
from prompts import analyst_messages
from score import score_run
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
issue_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
skip_specialists: list[str]
class SpecialistTask(TypedDict):
specialist: str
brief: str
class ActionTask(TypedDict, total=False):
incident_id: str
team: str
service: str
action: str
policy_reason: str
approval_issue_id: str
approval_issue: str
def linear_client() -> Linear:
return linear_api.client()
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 pick_up_incident(state: IncidentState):
if state.get("issue_id"):
linear_client().move(state["issue_id"], INCIDENTS_TEAM, "Investigating")
return {"status": "investigating"}
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)),
]
)
skipped = state.get("skip_specialists", [])
decision["specialists"] = [
name for name in decision["specialists"] if name not in skipped
]
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)
prompt, messages = analyst_messages(state["incident"], findings)
analysis = analyst.invoke(
messages,
config={"metadata": {"langfuse_prompt": prompt}} if prompt else None,
)
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."],
}
APPROVAL_LINE = re.compile(r"^Approval: (\w+)$", re.MULTILINE)
def approval_key(action: dict) -> str:
text = f"{action['team']}|{action['service']}|{action['action']}"
return hashlib.sha1(text.encode()).hexdigest()[:8]
def short_title(text: str, limit: int = 120) -> str:
if len(text) <= limit:
return text
return text[: limit - 1].rsplit(" ", 1)[0].rstrip(",;:") + "…"
def approval_description(state: IncidentState, action: dict) -> str:
return (
f"Approval: {approval_key(action)}\n"
f"Incident: {state['incident_id']}\n"
f"Team: {action['team']}\n"
f"Service: {action['service']}\n\n"
f"**Action:** {action['action']}\n\n"
f"**Why it needs approval:** {action['policy_reason']}\n\n"
"Move this issue to **Approved** or **Rejected** to decide."
)
def create_approval_subissues(state: IncidentState):
gated = [action for action in state["actions"] if action["requires_approval"]]
if not state.get("issue_id") or not gated:
return {}
linear = linear_client()
existing = {}
for child in linear.sub_issues(state["issue_id"]):
match = APPROVAL_LINE.search(child["description"] or "")
if match:
existing[match.group(1)] = child
approvals_team = linear.team(APPROVALS_TEAM)
actions = []
for action in state["actions"]:
if not action["requires_approval"]:
actions.append(action)
continue
child = existing.get(approval_key(action)) or linear.create_issue(
teamId=approvals_team["id"],
parentId=state["issue_id"],
title=short_title(f"Approve for {action['team']}: {action['action']}"),
description=approval_description(state, action),
stateId=linear.state_id(APPROVALS_TEAM, "Needs Approval"),
labelIds=[linear.label_id(action["team"], "team")],
)
actions.append(
action
| {
"approval_issue_id": child["id"],
"approval_issue": child["identifier"],
}
)
identifiers = [
action["approval_issue"]
for action in actions
if action.get("approval_issue")
]
print(f"Approval sub-issues: {', '.join(identifiers)}")
return {"actions": actions}
def findings_comment(state: IncidentState) -> str:
analysis = state["analysis"]
finding_lines = "\n".join(
f"- **{finding['specialist']}** ({finding['severity']}): {finding['summary']}"
for finding in state["findings"]
)
action_lines = "\n".join(
f"- {action['team']} / {action['service']}: {action['action']} "
+ (
f"(needs team approval in {action['approval_issue']})"
if action.get("approval_issue")
else "(needs team approval)"
if action["requires_approval"]
else "(no approval needed)"
)
for action in state["actions"]
)
return (
f"**Triage findings: {state['severity']}**\n\n"
f"**Summary:** {analysis['summary']}\n\n"
f"**Likely cause:** {analysis['likely_cause']}\n\n"
f"**Rationale:** {analysis['rationale']}\n\n"
f"**Specialists**\n{finding_lines}\n\n"
f"**Proposed actions**\n{action_lines}"
)
def report_findings(state: IncidentState):
gated = any(action["requires_approval"] for action in state["actions"])
if state.get("issue_id"):
linear = linear_client()
linear.set_group_label(state["issue_id"], "severity", state["severity"])
if gated:
linear.move(state["issue_id"], INCIDENTS_TEAM, "Awaiting Approval")
linear.comment(state["issue_id"], findings_comment(state))
return {"status": "awaiting_approval" if gated else "reported"}
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 = [
approval
for approval in approvals
if approval["decision"] in ("approve", "not required")
]
decisions = [
f"Team {approval['team']} decision for {approval['service']}: {approval['decision']}."
for approval 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"- {finding['specialist']} ({finding['severity']}): {finding['summary']}"
for finding in state["findings"]
)
action_lines = "\n".join(
f"- {approval['team']} / {approval['service']}: {approval['action']} "
f"({'approved by team' if approval['decision'] == 'approve' else 'approval not required'})"
for approval 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("pick_up_incident", pick_up_incident)
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("create_approval_subissues", create_approval_subissues)
builder.add_node("report_findings", report_findings)
builder.add_node("approval_gate", approval_gate)
builder.add_node("publish_runbook", publish_runbook, defer=True)
builder.add_edge(START, "pick_up_incident")
builder.add_edge("pick_up_incident", "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_edge("policy_check", "create_approval_subissues")
builder.add_edge("create_approval_subissues", "report_findings")
builder.add_conditional_edges(
"report_findings", 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", help="Defaults to the issue identifier.")
source = start_parser.add_mutually_exclusive_group(required=True)
source.add_argument("--issue", help="A Linear incident, such as TRI-5.")
source.add_argument("--alert", help="A fixture alert, without Linear.")
start_parser.add_argument(
"--skip",
nargs="*",
default=[],
choices=list(SPECIALISTS),
help="Specialists to hold back, to test that the score catches a weak run.",
)
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.")
if args.command == "start" and not (args.thread or args.issue):
parser.error("--alert needs --thread.")
if args.command == "start" and not args.thread:
args.thread = args.issue
return args
def main():
args = parse_args()
config = {"configurable": {"thread_id": args.thread}}
try:
run_command(args, config)
finally:
telemetry.shutdown()
def incident_input(
thread: str,
alert: dict,
issue: dict | None = None,
skip=(),
) -> dict:
return {
"incident_id": thread,
"issue_id": issue["id"] if issue else "",
"incident": f"{alert['title']}: {alert['description']}",
"alert": alert,
"status": "new",
"audit_log": [],
"skip_specialists": list(skip),
}
def issue_input(
issue_ref: str,
thread: str | None = None,
skip=(),
) -> tuple[str, dict]:
issue = linear_client().issue(issue_ref)
alert_id = alert_fields(issue["description"])["alert"]
alert = {
candidate["id"]: candidate for candidate in load_json("alerts.json")
}[alert_id]
print(f"Picked up {issue['identifier']}: {issue['title']} ({alert_id})")
thread = thread or issue["identifier"]
return thread, incident_input(thread, alert, issue, skip)
def run_graph(graph, thread: str, command: str, graph_input) -> dict:
config = {"configurable": {"thread_id": thread}}
before = graph.get_state(config).values
started = graph_input if isinstance(graph_input, dict) else before
alert_id = started["alert"]["id"]
trace_extra, handler = telemetry.trace_config(
thread,
command,
alert_id,
started.get("skip_specialists", []),
)
try:
stream = graph.stream_events(
graph_input,
config={**config, **trace_extra},
durability="sync",
version="v3",
)
display_stream(stream)
except Exception as error:
display_failure(graph, config, error)
values = graph.get_state(config).values
if handler and handler.last_trace_id:
url = telemetry.trace_url(handler.last_trace_id)
print(f"Trace: {url}")
if values.get("issue_id") and url:
linear_client().attach(
values["issue_id"],
url,
f"Langfuse trace: {command}",
thread,
)
if "analysis" in values and "analysis" not in before:
correct = score_run(
telemetry.langfuse_client(),
handler.last_trace_id,
alert_id,
values["analysis"],
)
print(f"Score root_cause_correct: {correct}")
return values
def run_command(args, config):
with SqliteSaver.from_conn_string(DB_PATH) as checkpointer:
graph = build_graph(checkpointer)
if args.command == "start" and args.issue:
args.thread, graph_input = issue_input(
args.issue,
args.thread,
args.skip,
)
elif args.command == "start":
alerts = {
alert["id"]: alert for alert in load_json("alerts.json")
}
graph_input = incident_input(
args.thread,
alerts[args.alert],
skip=args.skip,
)
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)
print(f"Recorded {args.decision} for {args.team}.")
else:
graph_input = None
run_graph(graph, args.thread, args.command, graph_input)
if __name__ == "__main__":
main()How to Use This Reference
Use this file to compare identifiers, function order, and graph edges. Keep your own environment values outside the source file.
Your earlier issue ID already names a checkpoint thread with saved progress. A fresh incident receives a new issue ID, so the updated graph starts with an empty checkpoint state.
- Seed a fresh INC-2041 incident by running:
python3 seed_incidents.py INC-2041What Does This Command Create?
The seeding script creates a new parent incident from the INC-2041 fixture alert. It prints the incident's issue ID and URL.
You should see a new issue ID beside INC-2041, followed by the issue URL.
- Record the new issue ID here: printed TRI issue ID.
- Export the new issue ID for this checkpoint thread by running:
export ISSUE_ID=[[ISSUE_ID="printed TRI issue ID"]]Why Replace the Issue Variable?
ISSUE_ID now contains the fresh issue ID used by the start command. The workflow also uses that issue ID to name the checkpoint thread.
Before you start the workflow, how many approval sub-issues do you expect for each gated action?
- Run the updated workflow against the fresh incident:
python3 app.py start --issue "$ISSUE_ID"What Happens During This Run?
The graph investigates the fixture alert before evaluating its proposed actions. It creates missing approval sub-issues before reporting the findings to the parent incident.
The workflow then reaches the human approval interrupts. The saved checkpoint state now contains each approval sub-issue's internal issue ID and issue ID.
You should see Approval sub-issues: followed by one or more APR issue IDs for the approval sub-issues created in Linear. The run then reaches WORKFLOW PAUSED: with the pending teams listed below it.
- Return to the fresh parent incident in the Linear workspace from earlier.
- Confirm the parent state is Awaiting Approval.
- Confirm the parent has a computed severity label.
- Confirm the timeline contains exactly one comment beginning with Triage findings: SEV.
- Select the approval sub-issue list under the parent incident.
- Count one approval sub-issue for each pending gated action shown in the terminal.
- Select one approval sub-issue.
- Confirm its state is Needs Approval.
That closes the visibility gap: every gated action now has durable approval work in Linear. The parent incident also carries the evidence that responders need before deciding.
Missing Approval Sub-Issues or Findings?
Confirm that you exported the newly printed issue ID before starting the workflow. Reusing the earlier issue ID resumes its saved checkpoint state instead of running the new nodes from the beginning.
Check that report_findings is the source of the conditional edge. Confirm that create_approval_subissues runs directly after policy_check.
Help me diagnose a missing Linear approval sub-issue or findings comment.
Your workflow now leaves a complete approval trail before it pauses. Next, you will synchronize each CLI decision with its matching Linear approval sub-issue and publish the final outcome.
Sync Decisions and Publish the Runbook
Your Linear incident now shows its findings and approval requests. Decisions still appear only in the terminal, so the approval sub-issues can show the wrong result.
You will update app.py so each terminal decision updates its approval sub-issue before LangGraph saves the next checkpoint state to SQLite. After approving every team, each approval sub-issue should be Approved, the parent should be Done, and one final comment should begin Runbook published.
In this step, get ready to:
- Synchronize each terminal decision with its matching Linear approval sub-issue.
- Generate the final runbook or the safe rejection outcome.
- Approve every pending team and verify the completed incident in Linear.
Synchronize each approval decision
Each interrupt already carries the approval sub-issue references created before the workflow paused. The approval gate uses the internal issue ID for the API update while preserving the issue ID for responders.
- Open app.py in the Cloud Shell Editor by running this command:
cloudshell edit app.py- Find the existing approval_gate() function.
- Replace the entire function with the code below:
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"],
"approval_issue_id": task.get("approval_issue_id", ""),
"approval_issue": task.get("approval_issue", ""),
"allowed_decisions": ["approve", "reject"],
}
)
if decision not in ("approve", "reject"):
raise ValueError("Decision must be 'approve' or 'reject'.")
if task.get("approval_issue_id"):
outcome = "Approved" if decision == "approve" else "Rejected"
linear_client().move(task["approval_issue_id"], APPROVALS_TEAM, outcome)
return {"approvals": [task | {"decision": decision}]}What Does This Code Do?
- The interrupt payload preserves the approval sub-issue's issue ID and internal issue ID.
- The selected decision maps to the matching Linear workflow state.
- The approval sub-issue moves before the decision enters the saved checkpoint state.
- Save app.py.
- Confirm the synchronization logic passes the repository tests by running:
python3 -m pytestWhat Does This Check Prove?
The starter tests don't call Linear, so a pass confirms your edit kept the existing graph working. The two resumes in Complete the approved workflow are the real check that each decision moves its approval sub-issue.
You should see the test suite complete successfully. That passing suite confirms the approval gate now updates both sides of the decision boundary.
Tests Not Passing?
Check that both approval sub-issue fields remain inside the interrupt dictionary. Confirm that the Linear move happens before the function returns its approvals update.
Use the first traceback location to compare your function with the block above. Help me debug the approval gate test failure.
Build the final Linear outcome
The final node needs one formatter for two publication destinations. Linear-backed runs publish the runbook as a parent-issue comment. Fixture-only runs publish it as a local Markdown file. The diagram below shows every path publish_runbook() can take, including the rejection path that publishes nothing.
- In app.py, find the existing publish_runbook() function.
- Insert the first part of runbook_markdown() directly above it:
def runbook_markdown(
state: IncidentState,
heading: str,
approved: list[dict],
audit_log: list[str],
) -> str:
analysis = state["analysis"]
finding_lines = "\n".join(
f"- {finding['specialist']} ({finding['severity']}): {finding['summary']}"
for finding in state["findings"]
)
action_lines = "\n".join(
f"- {action['team']} / {action['service']}: {action['action']} "
f"({'approved by team' if action['decision'] == 'approve' else 'approval not required'})"
for action in approved
)
audit_lines = "\n".join(f"- {entry}" for entry in audit_log)How Does the Formatter Start?
This first part converts findings and approved actions into Markdown list lines. It also prepares every audit entry for the final section.
- Continue directly below the audit_lines line with the formatter output:
return (
f"{heading}\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"
)What Does the Formatter Return?
The returned Markdown combines the incident identity, analysis, findings, approved actions, and audit log. The caller supplies the heading so a Linear comment and a local Markdown file can use the same body.
- Save app.py.
- Verify the new formatter preserves the tested workflow by running:
python3 -m pytestWhat Does This Check Prove?
A passing suite confirms that adding the helper preserved the existing incident flow. You'll see the formatter's output when the workflow publishes the runbook to Linear.
You should see the tests pass before you connect this formatter to the final publication branches.
- Select the existing publish_runbook() function through the line immediately before build_graph().
- Replace that selection with the rejection branch below:
def publish_runbook(state: IncidentState):
approvals = state.get("approvals", [])
approved = [
approval
for approval in approvals
if approval["decision"] in ("approve", "not required")
]
decisions = [
f"Team {approval['team']} decision for {approval['service']}: {approval['decision']}."
for approval in approvals
]
if not approved:
if state.get("issue_id"):
linear = linear_client()
linear.move(state["issue_id"], INCIDENTS_TEAM, "Triage")
linear.comment(
state["issue_id"],
"**Every proposed action was rejected.** Nothing was published, "
"and the incident is back in Triage.\n\n"
+ "\n".join(f"- {line}" for line in decisions),
)
return {
"status": "rejected_without_publish",
"audit_log": state["audit_log"]
+ decisions
+ ["Every action was rejected. Publishing was skipped."],
}How Does the Workflow Fail Closed?
The function first separates executable actions from rejected actions. An empty approved list returns the parent incident to Triage and records that publication was skipped.
- Continue directly below the rejection branch with the publication paths:
if state.get("issue_id"):
location = f"Linear issue {state['incident_id']}"
audit_log = state["audit_log"] + decisions + [f"Published runbook to {location}."]
linear = linear_client()
linear.move(state["issue_id"], INCIDENTS_TEAM, "Done")
linear.comment(
state["issue_id"],
runbook_markdown(state, "**Runbook published**", approved, audit_log),
)
else:
RUNBOOK_DIR.mkdir(exist_ok=True)
path = RUNBOOK_DIR / f"{safe_incident_id(state['incident_id'])}.md"
location = str(path)
audit_log = state["audit_log"] + decisions + [f"Published runbook to {path}."]
path.write_text(
runbook_markdown(
state,
"# Policy-Controlled Incident Runbook",
approved,
audit_log,
),
encoding="utf-8",
)
return {
"status": "published",
"runbook_path": location,
"audit_log": audit_log,
}How Do the Publication Paths Differ?
- A Linear-backed run moves the parent incident to Done and publishes the formatted runbook as one comment.
- A fixture-only run publishes a local Markdown runbook inside the runbooks directory.
- Both successful paths record the publication destination and completed audit log in the saved checkpoint state.
- Save app.py.
- Confirm both final outcome branches pass the repository tests by running:
python3 -m pytestWhat Does the Final Test Cover?
The tests confirm the change didn't break the graph. They don't call Linear, so you prove the outcomes live: the published runbook in Complete the approved workflow, and the all-rejected outcome in the Secret Mission.
You should see the full suite pass. Your application now has a deterministic final write for every decision outcome.
Final Outcome Tests Failing?
Check the indentation where the publication branch follows the rejection return. Confirm that the Linear comment calls receive the parent issue_id.
Compare the first failing assertion with the matching branch above. Help me debug the final publication tests.
✔️ Awesome, I've got everything!
Your final app.py is saved and the test suite passes.
ⓧ I'd like to double check the full code
import argparse
import hashlib
import json
import os
import operator
import re
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
import telemetry
import linear as linear_api
from linear import APPROVALS_TEAM, INCIDENTS_TEAM, Linear, alert_fields
from prompts import analyst_messages
from score import score_run
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
issue_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
skip_specialists: list[str]
class SpecialistTask(TypedDict):
specialist: str
brief: str
class ActionTask(TypedDict, total=False):
incident_id: str
team: str
service: str
action: str
policy_reason: str
approval_issue_id: str
approval_issue: str
def linear_client() -> Linear:
return linear_api.client()
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 pick_up_incident(state: IncidentState):
if state.get("issue_id"):
linear_client().move(state["issue_id"], INCIDENTS_TEAM, "Investigating")
return {"status": "investigating"}
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)),
]
)
skipped = state.get("skip_specialists", [])
decision["specialists"] = [
name for name in decision["specialists"] if name not in skipped
]
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)
prompt, messages = analyst_messages(state["incident"], findings)
analysis = analyst.invoke(
messages,
config={"metadata": {"langfuse_prompt": prompt}} if prompt else None,
)
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."],
}
APPROVAL_LINE = re.compile(r"^Approval: (\w+)$", re.MULTILINE)
def approval_key(action: dict) -> str:
text = f"{action['team']}|{action['service']}|{action['action']}"
return hashlib.sha1(text.encode()).hexdigest()[:8]
def short_title(text: str, limit: int = 120) -> str:
if len(text) <= limit:
return text
return text[: limit - 1].rsplit(" ", 1)[0].rstrip(",;:") + "…"
def approval_description(state: IncidentState, action: dict) -> str:
return (
f"Approval: {approval_key(action)}\n"
f"Incident: {state['incident_id']}\n"
f"Team: {action['team']}\n"
f"Service: {action['service']}\n\n"
f"**Action:** {action['action']}\n\n"
f"**Why it needs approval:** {action['policy_reason']}\n\n"
"Move this issue to **Approved** or **Rejected** to decide."
)
def create_approval_subissues(state: IncidentState):
gated = [action for action in state["actions"] if action["requires_approval"]]
if not state.get("issue_id") or not gated:
return {}
linear = linear_client()
existing = {}
for child in linear.sub_issues(state["issue_id"]):
match = APPROVAL_LINE.search(child["description"] or "")
if match:
existing[match.group(1)] = child
approvals_team = linear.team(APPROVALS_TEAM)
actions = []
for action in state["actions"]:
if not action["requires_approval"]:
actions.append(action)
continue
child = existing.get(approval_key(action)) or linear.create_issue(
teamId=approvals_team["id"],
parentId=state["issue_id"],
title=short_title(f"Approve for {action['team']}: {action['action']}"),
description=approval_description(state, action),
stateId=linear.state_id(APPROVALS_TEAM, "Needs Approval"),
labelIds=[linear.label_id(action["team"], "team")],
)
actions.append(
action
| {
"approval_issue_id": child["id"],
"approval_issue": child["identifier"],
}
)
identifiers = [
action["approval_issue"]
for action in actions
if action.get("approval_issue")
]
print(f"Approval sub-issues: {', '.join(identifiers)}")
return {"actions": actions}
def findings_comment(state: IncidentState) -> str:
analysis = state["analysis"]
finding_lines = "\n".join(
f"- **{finding['specialist']}** ({finding['severity']}): {finding['summary']}"
for finding in state["findings"]
)
action_lines = "\n".join(
f"- {action['team']} / {action['service']}: {action['action']} "
+ (
f"(needs team approval in {action['approval_issue']})"
if action.get("approval_issue")
else "(needs team approval)"
if action["requires_approval"]
else "(no approval needed)"
)
for action in state["actions"]
)
return (
f"**Triage findings: {state['severity']}**\n\n"
f"**Summary:** {analysis['summary']}\n\n"
f"**Likely cause:** {analysis['likely_cause']}\n\n"
f"**Rationale:** {analysis['rationale']}\n\n"
f"**Specialists**\n{finding_lines}\n\n"
f"**Proposed actions**\n{action_lines}"
)
def report_findings(state: IncidentState):
gated = any(action["requires_approval"] for action in state["actions"])
if state.get("issue_id"):
linear = linear_client()
linear.set_group_label(state["issue_id"], "severity", state["severity"])
if gated:
linear.move(state["issue_id"], INCIDENTS_TEAM, "Awaiting Approval")
linear.comment(state["issue_id"], findings_comment(state))
return {"status": "awaiting_approval" if gated else "reported"}
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"],
"approval_issue_id": task.get("approval_issue_id", ""),
"approval_issue": task.get("approval_issue", ""),
"allowed_decisions": ["approve", "reject"],
}
)
if decision not in ("approve", "reject"):
raise ValueError("Decision must be 'approve' or 'reject'.")
if task.get("approval_issue_id"):
outcome = "Approved" if decision == "approve" else "Rejected"
linear_client().move(task["approval_issue_id"], APPROVALS_TEAM, outcome)
return {"approvals": [task | {"decision": decision}]}
def runbook_markdown(
state: IncidentState,
heading: str,
approved: list[dict],
audit_log: list[str],
) -> str:
analysis = state["analysis"]
finding_lines = "\n".join(
f"- {finding['specialist']} ({finding['severity']}): {finding['summary']}"
for finding in state["findings"]
)
action_lines = "\n".join(
f"- {action['team']} / {action['service']}: {action['action']} "
f"({'approved by team' if action['decision'] == 'approve' else 'approval not required'})"
for action in approved
)
audit_lines = "\n".join(f"- {entry}" for entry in audit_log)
return (
f"{heading}\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"
)
def publish_runbook(state: IncidentState):
approvals = state.get("approvals", [])
approved = [
approval
for approval in approvals
if approval["decision"] in ("approve", "not required")
]
decisions = [
f"Team {approval['team']} decision for {approval['service']}: {approval['decision']}."
for approval in approvals
]
if not approved:
if state.get("issue_id"):
linear = linear_client()
linear.move(state["issue_id"], INCIDENTS_TEAM, "Triage")
linear.comment(
state["issue_id"],
"**Every proposed action was rejected.** Nothing was published, "
"and the incident is back in Triage.\n\n"
+ "\n".join(f"- {line}" for line in decisions),
)
return {
"status": "rejected_without_publish",
"audit_log": state["audit_log"]
+ decisions
+ ["Every action was rejected. Publishing was skipped."],
}
if state.get("issue_id"):
location = f"Linear issue {state['incident_id']}"
audit_log = state["audit_log"] + decisions + [f"Published runbook to {location}."]
linear = linear_client()
linear.move(state["issue_id"], INCIDENTS_TEAM, "Done")
linear.comment(
state["issue_id"],
runbook_markdown(state, "**Runbook published**", approved, audit_log),
)
else:
RUNBOOK_DIR.mkdir(exist_ok=True)
path = RUNBOOK_DIR / f"{safe_incident_id(state['incident_id'])}.md"
location = str(path)
audit_log = state["audit_log"] + decisions + [f"Published runbook to {path}."]
path.write_text(
runbook_markdown(
state,
"# Policy-Controlled Incident Runbook",
approved,
audit_log,
),
encoding="utf-8",
)
return {
"status": "published",
"runbook_path": location,
"audit_log": audit_log,
}
def build_graph(checkpointer: SqliteSaver):
builder = StateGraph(IncidentState)
builder.add_node("pick_up_incident", pick_up_incident)
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("create_approval_subissues", create_approval_subissues)
builder.add_node("report_findings", report_findings)
builder.add_node("approval_gate", approval_gate)
builder.add_node("publish_runbook", publish_runbook, defer=True)
builder.add_edge(START, "pick_up_incident")
builder.add_edge("pick_up_incident", "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_edge("policy_check", "create_approval_subissues")
builder.add_edge("create_approval_subissues", "report_findings")
builder.add_conditional_edges(
"report_findings", 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", help="Defaults to the issue identifier.")
source = start_parser.add_mutually_exclusive_group(required=True)
source.add_argument("--issue", help="A Linear incident, such as TRI-5.")
source.add_argument("--alert", help="A fixture alert, without Linear.")
start_parser.add_argument(
"--skip",
nargs="*",
default=[],
choices=list(SPECIALISTS),
help="Specialists to hold back, to test that the score catches a weak run.",
)
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.")
if args.command == "start" and not (args.thread or args.issue):
parser.error("--alert needs --thread.")
if args.command == "start" and not args.thread:
args.thread = args.issue
return args
def main():
args = parse_args()
config = {"configurable": {"thread_id": args.thread}}
try:
run_command(args, config)
finally:
telemetry.shutdown()
def incident_input(
thread: str,
alert: dict,
issue: dict | None = None,
skip=(),
) -> dict:
return {
"incident_id": thread,
"issue_id": issue["id"] if issue else "",
"incident": f"{alert['title']}: {alert['description']}",
"alert": alert,
"status": "new",
"audit_log": [],
"skip_specialists": list(skip),
}
def issue_input(
issue_ref: str,
thread: str | None = None,
skip=(),
) -> tuple[str, dict]:
issue = linear_client().issue(issue_ref)
alert_id = alert_fields(issue["description"])["alert"]
alert = {
candidate["id"]: candidate for candidate in load_json("alerts.json")
}[alert_id]
print(f"Picked up {issue['identifier']}: {issue['title']} ({alert_id})")
thread = thread or issue["identifier"]
return thread, incident_input(thread, alert, issue, skip)
def run_graph(graph, thread: str, command: str, graph_input) -> dict:
config = {"configurable": {"thread_id": thread}}
before = graph.get_state(config).values
started = graph_input if isinstance(graph_input, dict) else before
alert_id = started["alert"]["id"]
trace_extra, handler = telemetry.trace_config(
thread,
command,
alert_id,
started.get("skip_specialists", []),
)
try:
stream = graph.stream_events(
graph_input,
config={**config, **trace_extra},
durability="sync",
version="v3",
)
display_stream(stream)
except Exception as error:
display_failure(graph, config, error)
values = graph.get_state(config).values
if handler and handler.last_trace_id:
url = telemetry.trace_url(handler.last_trace_id)
print(f"Trace: {url}")
if values.get("issue_id") and url:
linear_client().attach(
values["issue_id"],
url,
f"Langfuse trace: {command}",
thread,
)
if "analysis" in values and "analysis" not in before:
correct = score_run(
telemetry.langfuse_client(),
handler.last_trace_id,
alert_id,
values["analysis"],
)
print(f"Score root_cause_correct: {correct}")
return values
def run_command(args, config):
with SqliteSaver.from_conn_string(DB_PATH) as checkpointer:
graph = build_graph(checkpointer)
if args.command == "start" and args.issue:
args.thread, graph_input = issue_input(
args.issue,
args.thread,
args.skip,
)
elif args.command == "start":
alerts = {
alert["id"]: alert for alert in load_json("alerts.json")
}
graph_input = incident_input(
args.thread,
alerts[args.alert],
skip=args.skip,
)
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)
print(f"Recorded {args.decision} for {args.team}.")
else:
graph_input = None
run_graph(graph, args.thread, args.command, graph_input)
if __name__ == "__main__":
main()Complete the approved workflow
The saved checkpoint state still contains every team listed under WORKFLOW PAUSED:. Reusing the same issue ID resumes the matching checkpoint thread.
Before you resume the first team, which Linear records do you expect this decision to change?
- Record the first team listed under the earlier WORKFLOW PAUSED: output here: pending team name.
- Approve that team by running this command:
python3 app.py resume --thread "$ISSUE_ID" --decision approve --team [[TEAM="pending team name"]]What Does This Command Do?
The command uses $ISSUE_ID to find the checkpoint thread and resumes only the interrupt for the selected team. The approval gate moves that team's approval sub-issue to Approved before adding the decision to the saved checkpoint state.
You should see the decision recorded for that team. If other gates remain, you should also see the workflow pause with a shorter pending list.
- Replace pending team name with the next pending team name.
- Repeat the same command once for every remaining team.
The final resume should print WORKFLOW COMPLETE and Status: published. The runbook destination names the Linear issue that now stores the runbook as a comment.
Decision Not Recorded?
If the terminal says there is no pending approval for the team, compare your spelling with the current pending list. Team names must match the saved pending interrupt exactly.
Confirm that $ISSUE_ID still contains the issue ID for this paused run. Help me troubleshoot a resume command that cannot find the pending team.
- Return to the Linear incident referenced by $ISSUE_ID.
- Confirm the parent incident state shows Done.
- Confirm every approval sub-issue shows a green check. Hover over the check, or open the sub-issue, to see Approved.
- Confirm the timeline still contains one findings comment.
- Confirm the timeline contains exactly one final comment beginning **Runbook published**.
The final comment contains the incident details, computed severity, analysis, findings, approved actions, and audit log. These records prove that the saved checkpoint state and Linear outcome finished in sync.
You closed the full approval loop. Every approval sub-issue now shows Approved, while the saved checkpoint state records the same decision and the parent incident holds one inspectable runbook comment.
Secret mission
Prove the Rejection Path
Drive a fresh incident through every rejection gate. You will prove that the workflow returns the incident to Triage without publishing a runbook.
Clean Up Your Resources
Clean Up Your Resources
Choose whether to preserve the completed workflow, pause your session, or remove its test data and local secrets. Keeping the current state adds no ongoing infrastructure cost.
Resources you used:
- Completed Linear incident identified by $ISSUE_ID with its Approved approval sub-issues and timeline comments.
- Rejected Linear incident identified by $REJECTED_ISSUE_ID with its Rejected approval sub-issues and rejection comment.
- Linear personal API key created in Security & access.
- Credential file .env in Google Cloud Shell.
- Local SQLite checkpoint database triage_checkpoints.db with both completed thread histories.
Keep everything running
Keep this state if you plan to continue the series by deploying this workflow to Cloud Run. The next project picks up from this langgraph-triage-linear folder, its .env file, and your Linear key. Your approved path and rejection path remain ready to demonstrate.
- Leave both completed Linear incidents in their current states.
- Keep .env private because it contains credentials.
- Retain triage_checkpoints.db to preserve both workflow outcomes.
- Your Linear Free workspace remains $0 within plan limits.
- Future Vertex AI calls can still incur usage charges under your existing account policies.
Pause - I'll come back to this later
Pausing ends the active Google Cloud Shell process. Files under $HOME remain available for your next session.
- Close the active Google Cloud Shell session.
- Leave the langgraph-triage-linear folder under $HOME.
- Leave both Linear incidents in their current states.
Closing the session clears its exported environment values while preserving the files in persistent storage.
Delete - I don't want to use this again
Remove the project test data and local runtime state if you want to start fresh.
Before You Delete
Deleting both Linear incidents removes the visible record of the approved outcome and the rejected outcome.
Revoking the personal API key cuts off this project's Linear access. Save anything you want to reference before you continue.
Remove Linear test data:
- Remove every approval sub-issue linked to $ISSUE_ID from the Approvals team in Linear.
- Remove every approval sub-issue linked to $REJECTED_ISSUE_ID from the Approvals team in Linear.
- Remove the parent issue identified by $ISSUE_ID from the Incidents team.
- Remove the parent issue identified by $REJECTED_ISSUE_ID from the Incidents team.
- Confirm the Incidents team no longer lists either test incident.
Revoke Linear access:
- Return to Linear's Security & access settings to locate the personal API key created for this project.
- Revoke the personal API key created for this project.
- Confirm the project key is no longer active.
Remove local credentials and checkpoints:
- In the langgraph-triage-linear folder from earlier, remove the stored environment values and checkpoint database by running this command:
rm -f .env triage_checkpoints.dbWhat Does This Remove?
The command deletes the two named files from the current repository. Removing .env clears every stored environment value from this copy.
Removing triage_checkpoints.db deletes both local thread histories and their resume state.
The command returns to the prompt without output when the removal succeeds.
- Close the active Google Cloud Shell session to clear the values already exported into its process.
The reusable Incidents and Approvals workspace configuration remains in Linear.
Nice Work!
Nice Work!
Strong finish. Your triage workflow now carries incidents from pickup to an inspectable approval outcome in Linear.
You've learned how to:
- Use a Linear-backed entry point to start a LangGraph thread from an incident. Watch the parent move from Triage through approval to Done.
- Keep deterministic Linear writes inside graph nodes. Publish one findings comment. Create one reusable approval sub-issue for each gated action.
- Resume every team gate from durable SQLite checkpoints. Synchronize each CLI decision with the matching sub-issue. Finish an approved incident with exactly one published runbook.
- Secret Mission: Prove unanimous rejection returns a fresh incident to Triage. Confirm every approval sub-issue is Rejected. Verify no runbook is published.
You can now demonstrate an agentic triage workflow whose operational state stays inspectable at every gate.
Ready to quiz yourself?