Flaky Test Triage Custom
blueprint · 4 agents · 3 MCP servers
1087★ stars
152forks
4.7rating
What it does
A scheduled workflow. A collector reads recent CI runs and clusters tests that both pass and fail on the same commit. A classifier reads the failure signatures and the prior baseline on disk to decide whether each cluster is a genuine flake or a real regression that only looks intermittent. The flow then branches: flakes are quarantined and tracked with an owner, and real regressions are escalated to the team before they spread. Point it at your CI and run it nightly.
The cast · 4 agents
Collectorcollectorfast
Read recent CI runs in the window. Cluster tests that both passed and failed on the same commit, or failed non-deterministically across retries. Return the clusters with their failure messages and frequency.
code.list_runs
Escalatorescalatorfast
For clusters judged to be real regressions, post to the team channel with the test, the suspect commit, and when it started failing. Keep it short and actionable.
chat.post_message
Classifierclassifierreasoning
Read the prior baseline on disk. For each cluster, decide flake or real regression: a test that started failing after a specific commit and never recovered is a regression, not a flake. Return an object with, per cluster, a 'regression' boolean and the reasoning. Be conservative: when unsure, treat it as a possible regression and escalate.
fs.read_file
Quarantinerquarantinerfast
For confirmed flakes, file a tracked issue per test with the failure evidence and a suggested owner, and mark them for quarantine so they stop blocking merges. Do not touch tests judged to be regressions.
code.create_issue
Flow
→ Collector → clusters
→ Classifier → verdict
⑂ branch
verdict.regression == true
→ Escalator → report
Interface
Inputs
windowstring
How far back in CI history to sweep, e.g. '200 runs' or '7d'.
Outputs
reportstring
What was quarantined and what was escalated this run.
Depends on · 3 MCP servers
io.github.github/github-mcpnot in the registry yetcode
io.github.modelcontextprotocol/filesystemnot in the registry yetfs
io.github.slack/slack-mcpnot in the registry yetchat⤓ Export runnable code
# Generated by @socketcat/compiler for target: langgraph
# blueprint: com.socketcat/flaky-test-triage 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):
window: object
report: object
clusters: object
verdict: object
def node_AgentInvoke_0(state):
return _rt.run_agent("collector", state, {"window":"window"}, "clusters")
def node_AgentInvoke_1(state):
return _rt.run_agent("classifier", state, {"clusters":"clusters"}, "verdict")
def node_Branch_2(state):
return {}
def node_Join_3(state):
return {}
def node_AgentInvoke_4(state):
return _rt.run_agent("escalator", state, {"verdict":"verdict"}, "report")
def node_AgentInvoke_5(state):
return _rt.run_agent("quarantiner", state, {"verdict":"verdict"}, "report")
def route_Branch_2(state):
if _rt.cond("verdict.regression == true", state): return "AgentInvoke_4"
return "AgentInvoke_5"
def build():
b = StateGraph(State)
b.add_node("AgentInvoke_0", node_AgentInvoke_0)
b.add_node("AgentInvoke_1", node_AgentInvoke_1)
b.add_node("Branch_2", node_Branch_2)
b.add_node("Join_3", node_Join_3)
b.add_node("AgentInvoke_4", node_AgentInvoke_4)
b.add_node("AgentInvoke_5", node_AgentInvoke_5)
b.add_edge(START, "AgentInvoke_0")
b.add_edge("AgentInvoke_0", "AgentInvoke_1")
b.add_edge("AgentInvoke_1", "Branch_2")
b.add_conditional_edges("Branch_2", route_Branch_2, {"AgentInvoke_4": "AgentInvoke_4", "AgentInvoke_5": "AgentInvoke_5"})
b.add_edge("Join_3", END)
b.add_edge("AgentInvoke_4", "Join_3")
b.add_edge("AgentInvoke_5", "Join_3")
return b.compile()
INPUTS = ["window"]
OUTPUTS = ["report"]
if __name__ == "__main__":
_rt.main(build, INPUTS, OUTPUTS)
▸blueprint.json (the portable format)
{
"id": "com.socketcat/flaky-test-triage",
"flow": {
"type": "sequence",
"blocks": [
{
"in": {
"window": "window"
},
"out": "clusters",
"use": "collector",
"type": "agent"
},
{
"in": {
"clusters": "clusters"
},
"out": "verdict",
"use": "classifier",
"type": "agent"
},
{
"else": [
{
"in": {
"verdict": "verdict"
},
"out": "report",
"use": "quarantiner",
"type": "agent"
}
],
"type": "branch",
"cases": [
{
"when": "verdict.regression == true",
"blocks": [
{
"in": {
"verdict": "verdict"
},
"out": "report",
"use": "escalator",
"type": "agent"
}
]
}
]
}
]
},
"tags": [
"testing",
"ci",
"scheduled",
"reliability"
],
"title": "Flaky Test Triage",
"agents": {
"collector": {
"model": {
"hint": "fast"
},
"title": "Collector",
"tools": [
"code.list_runs"
],
"instructions": "Read recent CI runs in the window. Cluster tests that both passed and failed on the same commit, or failed non-deterministically across retries. Return the clusters with their failure messages and frequency."
},
"escalator": {
"model": {
"hint": "fast"
},
"title": "Escalator",
"tools": [
"chat.post_message"
],
"instructions": "For clusters judged to be real regressions, post to the team channel with the test, the suspect commit, and when it started failing. Keep it short and actionable."
},
"classifier": {
"model": {
"hint": "reasoning"
},
"title": "Classifier",
"tools": [
"fs.read_file"
],
"output": {
"type": "object",
"required": [
"regression"
],
"properties": {
"byCluster": {
"type": "array"
},
"regression": {
"type": "boolean"
}
}
},
"instructions": "Read the prior baseline on disk. For each cluster, decide flake or real regression: a test that started failing after a specific commit and never recovered is a regression, not a flake. Return an object with, per cluster, a 'regression' boolean and the reasoning. Be conservative: when unsure, treat it as a possible regression and escalate."
},
"quarantiner": {
"model": {
"hint": "fast"
},
"title": "Quarantiner",
"tools": [
"code.create_issue"
],
"instructions": "For confirmed flakes, file a tracked issue per test with the failure evidence and a suggested owner, and mark them for quarantine so they stop blocking merges. Do not touch tests judged to be regressions."
}
},
"estate": {
"verdict": {
"type": "object",
"description": "The classifier's call per cluster, including a 'regression' flag."
},
"clusters": {
"type": "array",
"description": "Tests that both passed and failed on the same commit in the window."
}
},
"$schema": "socketcat.dev/blueprint/v0",
"authors": [
{
"url": "https://socketcat.com",
"name": "SocketCat"
}
],
"license": "MIT",
"servers": [
{
"ref": "io.github.github/github-mcp",
"alias": "code"
},
{
"ref": "io.github.modelcontextprotocol/filesystem",
"alias": "fs"
},
{
"ref": "io.github.slack/slack-mcp",
"alias": "chat"
}
],
"summary": "Finds intermittently-failing tests in CI history, decides flake vs. real regression, and quarantines or escalates accordingly.",
"targets": [
"langgraph",
"*"
],
"version": "1.0.0",
"interface": {
"inputs": {
"window": {
"type": "string",
"description": "How far back in CI history to sweep, e.g. '200 runs' or '7d'."
}
},
"outputs": {
"report": {
"type": "string",
"description": "What was quarantined and what was escalated this run."
}
}
},
"extensions": {
"dev.langgraph": {
"checkpointer": "memory"
}
},
"description": "A scheduled workflow. A collector reads recent CI runs and clusters tests that both pass and fail on the same commit. A classifier reads the failure signatures and the prior baseline on disk to decide whether each cluster is a genuine flake or a real regression that only looks intermittent. The flow then branches: flakes are quarantined and tracked with an owner, and real regressions are escalated to the team before they spread. Point it at your CI and run it nightly."
}More in Coding & QA
Browse all →E2E Test Author
com.socketcat/e2e-test-author
LangGraph
Turns a user story into browser tests: plans the scenarios, writes Playwright code, runs it against the real page, and self-heals until it passes for the right reason.
★ 15203 servers
⑂ 214