Pipeline Firefighter LangGraph
blueprint · 7 agents · 4 MCP servers
1425★ stars
208forks
4.8rating
What it does
A triager reads the job failure and states what broke. Three probes then run in parallel: a data probe inspects the upstream tables for nulls and row-count anomalies, a historian lists the commits that recently touched the failing model, and a metrics probe reads the freshness and volume dashboards. A fixer proposes a patch and a verifier dry-runs it against the warehouse with explain; they loop until it holds or the attempt budget is spent. A reporter opens a pull request with the diagnosis and posts a short note to the data channel.
The cast · 7 agents
Fixerfixercoding
Propose the smallest change that addresses the diagnosis: a corrected query, a guard on a null source, or a rollback of a suspect commit. If you received a verifier note, address it and try again. Never guess when the probes disagree; say what you would check.
no tools · reasoning only
Triagertriagerfast
Read the job failure and state plainly what broke: which model or query, at which step, and the literal error. Do not guess the cause yet.
no tools · reasoning only
Reporterreporterfast
Open a pull request with the patch, the diagnosis, and what the probes found. Post a one-paragraph summary to the data channel linking the PR. If the loop did not converge, say so and hand off to a human with what was tried.
code.open_prchat.post_message
Verifierverifierreasoning
Dry-run the patched query with explain against the warehouse. Decide whether it would run correctly and at reasonable cost, without applying it. Return an object with a 'verified' boolean and specific notes.
db.explain
Data ProbedataProbereasoning
Inspect the upstream tables the failing model reads. Check for nulls in required columns, sudden row-count drops, and stale partitions. Report only read-only findings.
db.list_indicesdb.query_dsl
Historianhistorianfast
List the commits merged in the last 72 hours that touched the failing model or its sources. Report author, message, and time for each.
code.list_commits
Metrics ProbemetricsProbefast
Read the freshness and row-volume dashboards for the affected dataset around the failure time. Report what changed and when it started.
metrics.query_range
Flow
→ Triager → incident
⇉ parallel
→ Data Probe → dataFindings
→ Historian → codeFindings
→ Metrics Probe → metricFindings
↻ loop until fix.verified == true · max 3
→ Fixer → patch
→ Verifier → fix
→ Reporter → resolution
Interface
Inputs
failureobject
The failed job: model or query name, the error text, and the run time.
Outputs
resolutionstring
The diagnosis and the pull request that was opened, as posted to Slack.
Depends on · 4 MCP servers
io.github.supabase/postgres-mcpnot in the registry yetdb
io.github.github/github-mcpnot in the registry yetcode
io.github.grafana/grafana-mcpnot in the registry yetmetrics
io.github.slack/slack-mcpnot in the registry yetchat⤓ Export runnable code
# Generated by @socketcat/compiler for target: langgraph
# blueprint: com.socketcat/pipeline-firefighter v1.0.0 schema: socketcat.dev/blueprint/v0
# This code is yours. Edit it freely. The socketcat_runtime helper is optional and can be vendored.
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
import socketcat_runtime as _rt
class State(TypedDict, total=False):
failure: object
resolution: object
incident: object
dataFindings: object
codeFindings: object
metricFindings: object
__loop_6: object
patch: object
fix: object
def node_AgentInvoke_0(state):
return _rt.run_agent("triager", state, {"failure":"failure"}, "incident")
def node_Fork_1(state):
return {}
def node_Join_2(state):
return {}
def node_AgentInvoke_3(state):
return _rt.run_agent("dataProbe", state, {"incident":"incident"}, "dataFindings")
def node_AgentInvoke_4(state):
return _rt.run_agent("historian", state, {"incident":"incident"}, "codeFindings")
def node_AgentInvoke_5(state):
return _rt.run_agent("metricsProbe", state, {"incident":"incident"}, "metricFindings")
def node_LoopHeader_6(state):
return {}
def node_LoopTick_7(state):
return {"__loop_6": state.get("__loop_6", 0) + 1}
def node_Nop_8(state):
return {}
def node_AgentInvoke_9(state):
return _rt.run_agent("fixer", state, {"fix":"fix","incident":"incident","codeFindings":"codeFindings","dataFindings":"dataFindings","metricFindings":"metricFindings"}, "patch")
def node_AgentInvoke_10(state):
return _rt.run_agent("verifier", state, {"patch":"patch"}, "fix")
def node_AgentInvoke_11(state):
return _rt.run_agent("reporter", state, {"fix":"fix","patch":"patch"}, "resolution")
def route_LoopHeader_6(state):
count = state.get("__loop_6", 0)
return "exit" if (_rt.cond("fix.verified == true", state) or count >= 3) else "loop"
def build():
b = StateGraph(State)
b.add_node("AgentInvoke_0", node_AgentInvoke_0)
b.add_node("Fork_1", node_Fork_1)
b.add_node("Join_2", node_Join_2)
b.add_node("AgentInvoke_3", node_AgentInvoke_3)
b.add_node("AgentInvoke_4", node_AgentInvoke_4)
b.add_node("AgentInvoke_5", node_AgentInvoke_5)
b.add_node("LoopHeader_6", node_LoopHeader_6)
b.add_node("LoopTick_7", node_LoopTick_7)
b.add_node("Nop_8", node_Nop_8)
b.add_node("AgentInvoke_9", node_AgentInvoke_9)
b.add_node("AgentInvoke_10", node_AgentInvoke_10)
b.add_node("AgentInvoke_11", node_AgentInvoke_11)
b.add_edge(START, "AgentInvoke_0")
b.add_edge("AgentInvoke_0", "Fork_1")
b.add_edge("Fork_1", "AgentInvoke_3")
b.add_edge("Fork_1", "AgentInvoke_4")
b.add_edge("Fork_1", "AgentInvoke_5")
b.add_edge("Join_2", "AgentInvoke_9")
b.add_edge("AgentInvoke_3", "Join_2")
b.add_edge("AgentInvoke_4", "Join_2")
b.add_edge("AgentInvoke_5", "Join_2")
b.add_conditional_edges("LoopHeader_6", route_LoopHeader_6, {"loop": "AgentInvoke_9", "exit": "Nop_8"})
b.add_edge("LoopTick_7", "LoopHeader_6")
b.add_edge("Nop_8", "AgentInvoke_11")
b.add_edge("AgentInvoke_9", "AgentInvoke_10")
b.add_edge("AgentInvoke_10", "LoopTick_7")
b.add_edge("AgentInvoke_11", END)
return b.compile()
INPUTS = ["failure"]
OUTPUTS = ["resolution"]
if __name__ == "__main__":
_rt.main(build, INPUTS, OUTPUTS)
▸blueprint.json (the portable format)
{
"id": "com.socketcat/pipeline-firefighter",
"flow": {
"type": "sequence",
"blocks": [
{
"in": {
"failure": "failure"
},
"out": "incident",
"use": "triager",
"type": "agent"
},
{
"type": "parallel",
"blocks": [
{
"in": {
"incident": "incident"
},
"out": "dataFindings",
"use": "dataProbe",
"type": "agent"
},
{
"in": {
"incident": "incident"
},
"out": "codeFindings",
"use": "historian",
"type": "agent"
},
{
"in": {
"incident": "incident"
},
"out": "metricFindings",
"use": "metricsProbe",
"type": "agent"
}
]
},
{
"max": 3,
"type": "loop",
"until": "fix.verified == true",
"blocks": [
{
"in": {
"fix": "fix",
"incident": "incident",
"codeFindings": "codeFindings",
"dataFindings": "dataFindings",
"metricFindings": "metricFindings"
},
"out": "patch",
"use": "fixer",
"type": "agent"
},
{
"in": {
"patch": "patch"
},
"out": "fix",
"use": "verifier",
"type": "agent"
}
]
},
{
"in": {
"fix": "fix",
"patch": "patch"
},
"out": "resolution",
"use": "reporter",
"type": "agent"
}
]
},
"tags": [
"data-engineering",
"pipelines",
"orchestrator-workers",
"on-call"
],
"title": "Pipeline Firefighter",
"agents": {
"fixer": {
"model": {
"hint": "coding"
},
"title": "Fixer",
"instructions": "Propose the smallest change that addresses the diagnosis: a corrected query, a guard on a null source, or a rollback of a suspect commit. If you received a verifier note, address it and try again. Never guess when the probes disagree; say what you would check."
},
"triager": {
"model": {
"hint": "fast"
},
"title": "Triager",
"instructions": "Read the job failure and state plainly what broke: which model or query, at which step, and the literal error. Do not guess the cause yet."
},
"reporter": {
"model": {
"hint": "fast"
},
"title": "Reporter",
"tools": [
"code.open_pr",
"chat.post_message"
],
"instructions": "Open a pull request with the patch, the diagnosis, and what the probes found. Post a one-paragraph summary to the data channel linking the PR. If the loop did not converge, say so and hand off to a human with what was tried."
},
"verifier": {
"model": {
"hint": "reasoning"
},
"title": "Verifier",
"tools": [
"db.explain"
],
"output": {
"type": "object",
"required": [
"verified"
],
"properties": {
"notes": {
"type": "string"
},
"verified": {
"type": "boolean"
}
}
},
"instructions": "Dry-run the patched query with explain against the warehouse. Decide whether it would run correctly and at reasonable cost, without applying it. Return an object with a 'verified' boolean and specific notes."
},
"dataProbe": {
"model": {
"hint": "reasoning"
},
"title": "Data Probe",
"tools": [
"db.list_indices",
"db.query_dsl"
],
"instructions": "Inspect the upstream tables the failing model reads. Check for nulls in required columns, sudden row-count drops, and stale partitions. Report only read-only findings."
},
"historian": {
"model": {
"hint": "fast"
},
"title": "Historian",
"tools": [
"code.list_commits"
],
"instructions": "List the commits merged in the last 72 hours that touched the failing model or its sources. Report author, message, and time for each."
},
"metricsProbe": {
"model": {
"hint": "fast"
},
"title": "Metrics Probe",
"tools": [
"metrics.query_range"
],
"instructions": "Read the freshness and row-volume dashboards for the affected dataset around the failure time. Report what changed and when it started."
}
},
"estate": {
"fix": {
"type": "object",
"description": "The verifier's result, including a 'verified' flag."
},
"patch": {
"type": "object",
"description": "The fixer's current proposed change."
},
"incident": {
"type": "object",
"description": "The triager's plain reading of what failed and where."
},
"codeFindings": {
"type": "object",
"description": "The historian's list of recent commits touching the model."
},
"dataFindings": {
"type": "object",
"description": "The data probe's report on upstream table health."
},
"metricFindings": {
"type": "object",
"description": "The metrics probe's read of freshness and volume around the failure."
}
},
"$schema": "socketcat.dev/blueprint/v0",
"authors": [
{
"url": "https://socketcat.com",
"name": "SocketCat"
}
],
"license": "MIT",
"servers": [
{
"ref": "io.github.supabase/postgres-mcp",
"alias": "db"
},
{
"ref": "io.github.github/github-mcp",
"alias": "code"
},
{
"ref": "io.github.grafana/grafana-mcp",
"alias": "metrics"
},
{
"ref": "io.github.slack/slack-mcp",
"alias": "chat"
}
],
"summary": "Diagnoses a failed data pipeline job from three angles at once, then loops a fixer against a verifier until the patch passes a dry run.",
"targets": [
"langgraph",
"*"
],
"version": "1.0.0",
"interface": {
"inputs": {
"failure": {
"type": "object",
"description": "The failed job: model or query name, the error text, and the run time."
}
},
"outputs": {
"resolution": {
"type": "string",
"description": "The diagnosis and the pull request that was opened, as posted to Slack."
}
}
},
"extensions": {
"dev.langgraph": {
"checkpointer": "memory"
}
},
"description": "A triager reads the job failure and states what broke. Three probes then run in parallel: a data probe inspects the upstream tables for nulls and row-count anomalies, a historian lists the commits that recently touched the failing model, and a metrics probe reads the freshness and volume dashboards. A fixer proposes a patch and a verifier dry-runs it against the warehouse with explain; they loop until it holds or the attempt budget is spent. A reporter opens a pull request with the diagnosis and posts a short note to the data channel."
}