bobseares-2/production-incident-commander
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

a2a-source-edit: write agent.py

Browse Source
This commit is contained in:
a2a-cloud
2026-07-13 11:07:51 +00:00
parent 40c5b4d653
commit 8e3fa83e88
1 changed files with 22 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
agent.py
View File

@@ -155,13 +155,14 @@ def classify_severity(title: str, symptoms: list[str], affected_services: list[s
else: level = SeverityLevel.SEV4; rationale.append("minor issue or investigation with limited evidence")
if environment != "production" and level == SeverityLevel.SEV1: level = SeverityLevel.SEV2; rationale.append("non-production environment caps automatic severity at SEV2")
return {"level": level.value, "rationale": "; ".join(rationale), "decrease_requires_operator_acknowledgement": "true"}
def _make_hypothesis(statement: str, confidence: float, supporting_evidence: list[str], tests: list[str], ordinal: int) -> Hypothesis:
return Hypothesis(id=f"h{ordinal}", statement=statement, confidence=confidence, supporting_evidence=supporting_evidence, discriminating_tests=tests)
def build_hypotheses(title: str, symptoms: list[str], services: list[str], evidence: list[EvidenceEnvelope]) -> list[Hypothesis]:
blob = " ".join([title, *symptoms, *(e.summary for e in evidence)]).lower(); refs = [e.id for e in evidence if e.status in {EvidenceStatus.error, EvidenceStatus.degraded}]; out: list[Hypothesis] = []
def add(s: str, c: float, tests: list[str]) -> None: out.append(Hypothesis(id=f"h{len(out)+1}", statement=s, confidence=c, supporting_evidence=refs, discriminating_tests=tests))
if any(t in blob for t in ["crash", "crashloop", "image"]): add("A recent deployment introduced a crashing container or bad image revision.", .78, ["Compare deployment revision", "Read bounded pod log tails"])
if any(t in blob for t in ["sync", "argocd", "drift"]): add("Declared and live state diverged through Argo CD drift or failed sync.", .62, ["Read Argo sync status"])
if any(t in blob for t in ["500", "5xx", "latency", "timeout", "errors"]): add("User-visible service path is failing or degraded under load.", .58, ["Run health checks", "Query allowlisted metrics"])
if not out: add("Evidence is insufficient for a single root cause; continue scoped evidence collection.", .35, ["Collect Kubernetes, health, and metric evidence"])
if any(t in blob for t in ["crash", "crashloop", "image"]): out.append(_make_hypothesis("A recent deployment introduced a crashing container or bad image revision.", .78, refs, ["Compare deployment revision", "Read bounded pod log tails"], len(out) + 1))
if any(t in blob for t in ["sync", "argocd", "drift"]): out.append(_make_hypothesis("Declared and live state diverged through Argo CD drift or failed sync.", .62, refs, ["Read Argo sync status"], len(out) + 1))
if any(t in blob for t in ["500", "5xx", "latency", "timeout", "errors"]): out.append(_make_hypothesis("User-visible service path is failing or degraded under load.", .58, refs, ["Run health checks", "Query allowlisted metrics"], len(out) + 1))
if not out: out.append(_make_hypothesis("Evidence is insufficient for a single root cause; continue scoped evidence collection.", .35, refs, ["Collect Kubernetes, health, and metric evidence"], 1))
return sorted(out, key=lambda h: h.confidence, reverse=True)
def build_timeline(started_at: str, evidence: list[EvidenceEnvelope], title: str) -> list[TimelineEntry]: return [TimelineEntry(at=started_at or _now(), source="intake", description=title)] + [TimelineEntry(at=e.observed_at, source=e.source, description=e.summary[:500], evidence_refs=[e.id]) for e in evidence[:30]]
def recommend_actions(h: list[Hypothesis]) -> list[str]: return ["Preserve evidence dossier before any mutation", "Run verification against user-visible health checks and metrics"] + (["Prepare a rollback plan for the named Deployment after operator approval"] if any("crashing container" in x.statement for x in h) else [])
@@ -191,7 +192,7 @@ class IntegrationClient:
return StepExecutionRecord(step_id=step.id, status="denied", action_type=step.action_type, target=step.target, before_evidence=before, warnings=["live execution requires explicit provider setup and exact plan parameters"])
class ProductionIncidentCommander(A2AAgent[CommanderConfig, NoAuth]):
name = "production-incident-commander"; description = "Private safety-first commander for production incident diagnosis, remediation planning, approval-gated execution, recovery verification, and postmortems."; version = "0.1.0"
name = "production-incident-commander"; description = "Private safety-first commander for production incident diagnosis, remediation planning, approval-gated execution, recovery verification, and postmortems."; version = "0.1.1"
config_model = CommanderConfig; auth_model = NoAuth
pricing = Pricing(price_per_call_usd=0.0, caller_pays_llm=False, notes="Deterministic local policy and bounded caller-configured integrations; no LLM credential required.")
resources = Resources(cpu="1", memory="1Gi", max_runtime_seconds=900)
@@ -227,19 +228,29 @@ class ProductionIncidentCommander(A2AAgent[CommanderConfig, NoAuth]):
@a2a.tool(description="Execute only explicitly approved allowlisted reversible remediation steps. Defaults to dry run and fails closed.", input_schema=EXECUTE_INPUT_SCHEMA, timeout_seconds=900, cost_class="incident-execution", grant_mode="read_write_overlay", grant_write_prefixes=("outputs/incidents/",), grant_allow_patterns=("outputs/incidents/**",))
async def execute_remediation(self, ctx: RunContext[NoAuth], incident_id: Annotated[str, Field(pattern=INCIDENT_ID_RE)], plan_path: str, plan_digest: str, approved_step_ids: list[str], approval_acknowledgement: str, dry_run: bool = True, stop_on_failure: bool = True) -> dict[str, Any]:
artifact = f"outputs/incidents/{incident_id}/execution.json"
async def ret(r: ExecutionResult) -> dict[str, Any]: await workspace_write_json(ctx, artifact, r); return r.model_dump(mode="json")
if approval_acknowledgement != APPROVAL_ACK: return await ret(ExecutionResult(incident_id=incident_id, status="denied", dry_run=dry_run, approval_provenance={"accepted": False, "reason": "approval acknowledgement mismatch"}, step_results=[], final_incident_state="unchanged", artifact_paths=[artifact]))
if approval_acknowledgement != APPROVAL_ACK:
result = ExecutionResult(incident_id=incident_id, status="denied", dry_run=dry_run, approval_provenance={"accepted": False, "reason": "approval acknowledgement mismatch"}, step_results=[], final_incident_state="unchanged", artifact_paths=[artifact])
await workspace_write_json(ctx, artifact, result)
return result.model_dump(mode="json")
try: plan = RemediationPlan.model_validate_json(await workspace_read_text(ctx, validate_incident_path(plan_path, incident_id, must_exist_under_incident=True)))
except (ValidationError, ValueError, FileNotFoundError) as exc: return await ret(ExecutionResult(incident_id=incident_id, status="denied", dry_run=dry_run, approval_provenance={"accepted": False, "reason": f"plan could not be read or validated: {type(exc).__name__}"}, step_results=[], final_incident_state="unchanged", artifact_paths=[artifact]))
except (ValidationError, ValueError, FileNotFoundError) as exc:
result = ExecutionResult(incident_id=incident_id, status="denied", dry_run=dry_run, approval_provenance={"accepted": False, "reason": f"plan could not be read or validated: {type(exc).__name__}"}, step_results=[], final_incident_state="unchanged", artifact_paths=[artifact])
await workspace_write_json(ctx, artifact, result)
return result.model_dump(mode="json")
actual = plan.plan_digest or globals()["plan_digest"](plan)
if actual != plan_digest: return await ret(ExecutionResult(incident_id=incident_id, status="denied", dry_run=dry_run, approval_provenance={"accepted": False, "reason": "plan digest mismatch", "expected": plan_digest, "actual": actual}, step_results=[], final_incident_state="unchanged", artifact_paths=[artifact]))
if actual != plan_digest:
result = ExecutionResult(incident_id=incident_id, status="denied", dry_run=dry_run, approval_provenance={"accepted": False, "reason": "plan digest mismatch", "expected": plan_digest, "actual": actual}, step_results=[], final_incident_state="unchanged", artifact_paths=[artifact])
await workspace_write_json(ctx, artifact, result)
return result.model_dump(mode="json")
approved = set(approved_step_ids); results: list[StepExecutionRecord] = []; client = IntegrationClient(ctx, self.config)
for step in plan.steps:
if step.id not in approved: results.append(StepExecutionRecord(step_id=step.id, status="skipped", action_type=step.action_type, target=step.target)); continue
rec = await client.execute_step(step, dry_run); results.append(rec)
if stop_on_failure and rec.status in {"failed", "denied", "drift_detected"}: break
statuses = {r.status for r in results if r.step_id in approved}; overall: Literal["denied", "dry_run", "partially_executed", "executed", "failed"] = "failed" if any(s in statuses for s in {"denied", "failed", "drift_detected"}) else "dry_run" if dry_run else "executed" if statuses == {"executed"} else "partially_executed"
return await ret(ExecutionResult(incident_id=incident_id, status=overall, dry_run=dry_run, approval_provenance={"accepted": True, "approved_step_ids": sorted(approved), "plan_digest": actual, "acknowledgement": APPROVAL_ACK, "at": _now()}, step_results=results, final_incident_state="verify_required" if overall in {"executed", "partially_executed"} else "unchanged", artifact_paths=[artifact], receipts=[{"idempotency": "one guarded attempt per approved step"}]))
result = ExecutionResult(incident_id=incident_id, status=overall, dry_run=dry_run, approval_provenance={"accepted": True, "approved_step_ids": sorted(approved), "plan_digest": actual, "acknowledgement": APPROVAL_ACK, "at": _now()}, step_results=results, final_incident_state="verify_required" if overall in {"executed", "partially_executed"} else "unchanged", artifact_paths=[artifact], receipts=[{"idempotency": "one guarded attempt per approved step"}])
await workspace_write_json(ctx, artifact, result)
return result.model_dump(mode="json")
@a2a.tool(description="Verify recovery using user-visible checks and metrics. Never marks recovered unless required checks pass.", input_schema=VERIFY_INPUT_SCHEMA, timeout_seconds=900, cost_class="incident-verification", grant_mode="read_write_overlay", grant_write_prefixes=("outputs/incidents/",), grant_allow_patterns=("outputs/incidents/**",))
async def verify_recovery(self, ctx: RunContext[NoAuth], incident_id: Annotated[str, Field(pattern=INCIDENT_ID_RE)], verification_plan_path: str = "", healthcheck_urls: list[HttpUrl] = [], prometheus_queries: list[str] = [], observation_seconds: int = 60) -> dict[str, Any]: