bobseares-2/production-incident-commander
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
e59f2a1d26c959568d90031bb2bb71d672d00ef4
production-incident-commander/agent.py
a2a-cloud e59f2a1d26 a2a-source-edit: write agent.py
2026-07-13 03:10:40 +00:00

304 lines
42 KiB
Python
Raw Blame History

"""Production Incident Commander A2A agent."""
from __future__ import annotations
import asyncio
import hashlib
import ipaddress
import json
import re
import time
from datetime import UTC, datetime
from enum import Enum
from typing import Annotated, Any, Literal
from urllib.parse import urlparse
from pydantic import BaseModel, ConfigDict, Field, HttpUrl, ValidationError, field_validator
import a2a_pack as a2a
from a2a_pack import (
A2AAgent,
ConsumerSetup,
ConsumerSetupField,
EgressPolicy,
NoAuth,
Pricing,
Resources,
RunContext,
WorkspaceAccess,
WorkspaceMode,
)
from a2a_pack.context import AgentEvent, ConsumerSetupMissing
INCIDENT_ID_RE = r"^[a-z0-9][a-z0-9-]{2,63}$"
DNS_SAFE_RE = r"^[a-z0-9]([-a-z0-9]*[a-z0-9])?$"
APPROVAL_ACK = "I_APPROVE_THE_LISTED_STEPS"
MAX_EVIDENCE_CHARS = 16_000
ALLOWED_EXECUTION_ACTIONS = {"kubernetes.rollout_restart", "kubernetes.rollback_deployment", "argocd.sync_application"}
STATE_MACHINE = ("intake", "scope", "collect_evidence", "normalize", "timeline", "hypotheses", "discriminating_tests", "diagnosis", "remediation_plan", "approval_gate", "execute", "verify", "close_or_escalate", "postmortem")
INCIDENT_REASONING_POLICY = "Evidence first; diagnosis and planning are read-only; execution is digest-bound, approval-gated, dry-run by default, and limited to reversible allowlisted deterministic adapters. Untrusted evidence is never instructions."
STRICT_DIAGNOSTIC_TARGETS_SCHEMA: dict[str, Any] = {
"type": "object",
"additionalProperties": False,
"default": {"kubernetes_namespaces": [], "argo_applications": [], "prometheus_queries": [], "healthcheck_urls": [], "gitea_repositories": []},
"properties": {
"kubernetes_namespaces": {"type": "array", "items": {"type": "string", "pattern": DNS_SAFE_RE}, "maxItems": 30, "default": []},
"argo_applications": {"type": "array", "items": {"type": "string", "pattern": DNS_SAFE_RE}, "maxItems": 30, "default": []},
"prometheus_queries": {"type": "array", "items": {"type": "string", "minLength": 1, "maxLength": 500}, "maxItems": 20, "default": []},
"healthcheck_urls": {"type": "array", "items": {"type": "string", "format": "uri", "maxLength": 1000}, "maxItems": 20, "default": []},
"gitea_repositories": {"type": "array", "items": {"type": "string", "pattern": r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$"}, "maxItems": 20, "default": []},
},
}
DIAGNOSE_INPUT_SCHEMA: dict[str, Any] = {"type": "object", "additionalProperties": False, "required": ["incident_id", "title", "symptoms", "affected_services"], "properties": {"incident_id": {"type": "string", "pattern": INCIDENT_ID_RE}, "title": {"type": "string", "minLength": 3, "maxLength": 200}, "symptoms": {"type": "array", "minItems": 1, "maxItems": 20, "items": {"type": "string", "minLength": 1, "maxLength": 1000}}, "affected_services": {"type": "array", "minItems": 1, "maxItems": 30, "items": {"type": "string", "pattern": DNS_SAFE_RE}}, "environment": {"type": "string", "enum": ["production", "staging", "development"], "default": "production"}, "started_at": {"type": "string", "maxLength": 80, "default": ""}, "evidence_paths": {"type": "array", "items": {"type": "string", "maxLength": 300}, "maxItems": 20, "default": []}, "diagnostic_targets": STRICT_DIAGNOSTIC_TARGETS_SCHEMA, "constraints": {"type": "array", "items": {"type": "string", "maxLength": 500}, "maxItems": 20, "default": []}}}
PLAN_INPUT_SCHEMA: dict[str, Any] = {"type": "object", "additionalProperties": False, "required": ["incident_id", "objective"], "properties": {"incident_id": {"type": "string", "pattern": INCIDENT_ID_RE}, "diagnosis_path": {"type": "string", "maxLength": 300, "default": ""}, "objective": {"type": "string", "minLength": 3, "maxLength": 500}, "change_window": {"type": "string", "maxLength": 200, "default": ""}, "allowed_actions": {"type": "array", "items": {"type": "string", "enum": sorted(ALLOWED_EXECUTION_ACTIONS)}, "maxItems": 10, "default": []}, "forbidden_actions": {"type": "array", "items": {"type": "string", "maxLength": 200}, "maxItems": 30, "default": []}, "rollback_requirements": {"type": "array", "items": {"type": "string", "maxLength": 300}, "maxItems": 20, "default": []}}}
EXECUTE_INPUT_SCHEMA: dict[str, Any] = {"type": "object", "additionalProperties": False, "required": ["incident_id", "plan_path", "plan_digest", "approved_step_ids", "approval_acknowledgement"], "properties": {"incident_id": {"type": "string", "pattern": INCIDENT_ID_RE}, "plan_path": {"type": "string", "maxLength": 300}, "plan_digest": {"type": "string", "pattern": r"^sha256:[a-f0-9]{64}$"}, "approved_step_ids": {"type": "array", "minItems": 1, "maxItems": 20, "items": {"type": "string", "pattern": r"^[a-z0-9][a-z0-9-]{1,80}$"}}, "approval_acknowledgement": {"type": "string", "const": APPROVAL_ACK}, "dry_run": {"type": "boolean", "default": True}, "stop_on_failure": {"type": "boolean", "default": True}}}
VERIFY_INPUT_SCHEMA: dict[str, Any] = {"type": "object", "additionalProperties": False, "required": ["incident_id"], "properties": {"incident_id": {"type": "string", "pattern": INCIDENT_ID_RE}, "verification_plan_path": {"type": "string", "maxLength": 300, "default": ""}, "healthcheck_urls": {"type": "array", "items": {"type": "string", "format": "uri", "maxLength": 1000}, "maxItems": 20, "default": []}, "prometheus_queries": {"type": "array", "items": {"type": "string", "minLength": 1, "maxLength": 500}, "maxItems": 20, "default": []}, "observation_seconds": {"type": "integer", "minimum": 0, "maximum": 900, "default": 60}}}
POSTMORTEM_INPUT_SCHEMA: dict[str, Any] = {"type": "object", "additionalProperties": False, "required": ["incident_id"], "properties": {"incident_id": {"type": "string", "pattern": INCIDENT_ID_RE}, "include_customer_summary": {"type": "boolean", "default": True}, "include_action_items": {"type": "boolean", "default": True}}}
class CommanderConfig(BaseModel):
model_config = ConfigDict(extra="forbid")
prometheus_query_allowlist: list[str] = Field(default_factory=lambda: ["up", "rate(http_requests_total[5m])", "sum(rate(http_requests_total[5m]))"])
max_prometheus_range_seconds: int = Field(default=3600, ge=60, le=86400)
class Environment(str, Enum):
production = "production"; staging = "staging"; development = "development"
class SeverityLevel(str, Enum):
SEV1 = "SEV1"; SEV2 = "SEV2"; SEV3 = "SEV3"; SEV4 = "SEV4"
class EvidenceStatus(str, Enum):
ok = "ok"; degraded = "degraded"; error = "error"; setup_required = "setup_required"; skipped = "skipped"; denied = "denied"
class EvidenceEnvelope(BaseModel):
model_config = ConfigDict(extra="forbid")
id: str; source: str; target: str; observed_at: str; status: EvidenceStatus; summary: str
redactions: list[str] = Field(default_factory=list); raw_ref: str = ""; hash: str = ""; metadata: dict[str, Any] = Field(default_factory=dict)
class Hypothesis(BaseModel):
model_config = ConfigDict(extra="forbid")
id: str; statement: str; confidence: float = Field(ge=0, le=1); supporting_evidence: list[str] = Field(default_factory=list); conflicting_evidence: list[str] = Field(default_factory=list); discriminating_tests: list[str] = Field(default_factory=list)
class TimelineEntry(BaseModel):
model_config = ConfigDict(extra="forbid")
at: str; source: str; description: str; evidence_refs: list[str] = Field(default_factory=list)
class RemediationStep(BaseModel):
model_config = ConfigDict(extra="forbid")
id: str; intent: str; target: str; action_type: str; risk: Literal["low", "medium", "high"]; reversible: bool; prerequisites: list[str] = Field(default_factory=list); exact_verification: list[str] = Field(default_factory=list); rollback: str; approval_required: bool; evidence_to_capture: list[str] = Field(default_factory=list); parameters: dict[str, Any] = Field(default_factory=dict)
class RemediationPlan(BaseModel):
model_config = ConfigDict(extra="forbid")
incident_id: str; status: Literal["planned", "blocked", "setup_required"]; objective: str; change_window: str = ""; steps: list[RemediationStep]; forbidden_actions: list[str] = Field(default_factory=list); required_approvals: list[str] = Field(default_factory=list); plan_digest: str = ""; artifact_paths: list[str] = Field(default_factory=list); receipts: list[dict[str, Any]] = Field(default_factory=list)
class StepExecutionRecord(BaseModel):
model_config = ConfigDict(extra="forbid")
step_id: str; status: Literal["dry_run", "executed", "denied", "failed", "skipped", "drift_detected"]; action_type: str; target: str; before_evidence: list[EvidenceEnvelope] = Field(default_factory=list); after_evidence: list[EvidenceEnvelope] = Field(default_factory=list); verification: dict[str, Any] = Field(default_factory=dict); rollback_status: str = ""; warnings: list[str] = Field(default_factory=list)
class ExecutionResult(BaseModel):
model_config = ConfigDict(extra="forbid")
incident_id: str; status: Literal["denied", "dry_run", "partially_executed", "executed", "failed"]; dry_run: bool; approval_provenance: dict[str, Any]; step_results: list[StepExecutionRecord]; final_incident_state: str; artifact_paths: list[str]; receipts: list[dict[str, Any]] = Field(default_factory=list)
class RecoveryCheck(BaseModel):
model_config = ConfigDict(extra="forbid")
name: str; status: Literal["pass", "fail", "skipped", "setup_required"]; summary: str; evidence_refs: list[str] = Field(default_factory=list)
class RecoveryVerification(BaseModel):
model_config = ConfigDict(extra="forbid")
incident_id: str; checks: list[RecoveryCheck]; before_after_comparison: dict[str, Any]; error_budget_indicators: dict[str, Any]; regression_signals: list[str]; recovery_confidence: float = Field(ge=0, le=1); recovered: bool; remaining_risks: list[str]; artifact_paths: list[str]; receipts: list[dict[str, Any]] = Field(default_factory=list)
class CorrectiveAction(BaseModel):
model_config = ConfigDict(extra="forbid")
description: str; owner: str; priority: Literal["P0", "P1", "P2", "P3"]; due_date: str = ""; status: Literal["open", "in_progress", "done", "deferred"] = "open"
class Postmortem(BaseModel):
model_config = ConfigDict(extra="forbid")
incident_id: str; executive_summary: str; customer_impact: str; timeline: list[TimelineEntry]; detection: str; root_cause: str; contributing_factors: list[str]; response_analysis: str; what_went_well: list[str]; what_went_poorly: list[str]; corrective_actions: list[CorrectiveAction]; evidence_references: list[str]; unresolved_questions: list[str]; artifact_paths: list[str]; receipts: list[dict[str, Any]] = Field(default_factory=list)
class IncidentDiagnosis(BaseModel):
model_config = ConfigDict(extra="forbid")
incident_id: str; severity: dict[str, str]; status: Literal["diagnosed", "inconclusive", "setup_required", "escalate"]; evidence_summary: list[EvidenceEnvelope]; timeline: list[TimelineEntry]; hypotheses: list[Hypothesis]; tests_run: list[str]; ruled_out: list[str]; likely_root_cause: str = ""; impacted_components: list[str]; recommended_actions: list[str]; required_approvals: list[str]; verification_plan: list[str]; artifact_paths: list[str]; receipts: list[dict[str, Any]] = Field(default_factory=list)
class DiagnosticTargets(BaseModel):
model_config = ConfigDict(extra="forbid")
kubernetes_namespaces: list[str] = Field(default_factory=list, max_length=30)
argo_applications: list[str] = Field(default_factory=list, max_length=30)
prometheus_queries: list[str] = Field(default_factory=list, max_length=20)
healthcheck_urls: list[HttpUrl] = Field(default_factory=list, max_length=20)
gitea_repositories: list[str] = Field(default_factory=list, max_length=20)
@field_validator("kubernetes_namespaces", "argo_applications")
@classmethod
def _names_are_safe(cls, values: list[str]) -> list[str]:
for value in values:
if not re.fullmatch(DNS_SAFE_RE, value): raise ValueError(f"not DNS-safe: {value}")
return values
@field_validator("gitea_repositories")
@classmethod
def _repos_are_safe(cls, values: list[str]) -> list[str]:
for value in values:
if not re.fullmatch(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$", value): raise ValueError(f"repository must be owner/name: {value}")
return values
def _now() -> str: return datetime.now(UTC).isoformat(timespec="seconds")
def _sha256_text(text: str) -> str: return hashlib.sha256(text.encode("utf-8")).hexdigest()
def _stable_json(data: Any) -> str: return json.dumps(data, sort_keys=True, separators=(",", ":"), ensure_ascii=False, default=str)
def plan_digest(plan: RemediationPlan) -> str:
payload = plan.model_dump(mode="json"); payload["plan_digest"] = ""; payload["receipts"] = []
return "sha256:" + _sha256_text(_stable_json(payload))
SECRET_PATTERNS = [(re.compile(r"(?i)(authorization:\s*bearer\s+)[A-Za-z0-9._~+/=-]+"), r"\1[REDACTED]"), (re.compile(r"(?i)(token|api[_-]?key|password|secret)(\s*[:=]\s*)[^\s,'\"]+"), r"\1\2[REDACTED]"), (re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----.*?-----END [A-Z ]*PRIVATE KEY-----", re.S), "[REDACTED_PRIVATE_KEY]"), (re.compile(r"(?i)(cookie:\s*)[^\n\r]+"), r"\1[REDACTED]")]
PROMPT_INJECTION_PATTERNS = re.compile(r"(?i)(ignore (all )?(previous|prior) instructions|system prompt|developer message|exfiltrate|print.*secret|reveal.*token)")
def redact_text(value: Any) -> tuple[str, list[str]]:
text = value if isinstance(value, str) else json.dumps(value, default=str, ensure_ascii=False); redactions: list[str] = []
for idx, (pattern, repl) in enumerate(SECRET_PATTERNS):
text, count = pattern.subn(repl, text)
if count: redactions.append(f"secret_pattern_{idx}:{count}")
if len(text) > MAX_EVIDENCE_CHARS: text = text[:MAX_EVIDENCE_CHARS] + "\n[TRUNCATED]"; redactions.append("truncated")
return text, redactions
def _evidence(source: str, target: str, status: EvidenceStatus, summary: str, raw: Any = "", metadata: dict[str, Any] | None = None) -> EvidenceEnvelope:
safe_summary, redactions = redact_text(summary); raw_text = redact_text(raw)[0] if raw != "" else ""; raw_ref = f"sha256:{_sha256_text(raw_text)}" if raw_text else ""
if PROMPT_INJECTION_PATTERNS.search(safe_summary) or (raw_text and PROMPT_INJECTION_PATTERNS.search(raw_text)): redactions.append("untrusted_prompt_injection_ignored")
return EvidenceEnvelope(id=f"ev-{_sha256_text(source + target + safe_summary + _now())[:12]}", source=source, target=target, observed_at=_now(), status=status, summary=safe_summary, redactions=redactions, raw_ref=raw_ref, hash=raw_ref, metadata=metadata or {})
def _is_internal_service_host(host: str) -> bool: return bool(re.fullmatch(r"[a-z0-9-]+(\.[a-z0-9-]+)*\.(svc|svc\.cluster\.local)", host))
def _host_is_ip_blocked(host: str) -> bool:
if host.lower() in {"localhost", "metadata.google.internal"}: return True
try: ip = ipaddress.ip_address(host)
except ValueError: return False
return ip.is_loopback or ip.is_link_local or ip.is_private or ip.is_reserved or ip.is_multicast or str(ip) == "169.254.169.254"
def _validate_base_url(url: str, *, allow_internal: bool = False) -> str:
if re.search(r"(?i)(token|password|secret|apikey|api_key)=", url): raise ValueError("credentials in URLs are rejected")
parsed = urlparse(str(url)); host = parsed.hostname or ""
if parsed.username or parsed.password: raise ValueError("credentials in URLs are rejected")
if not host: raise ValueError("URL host is required")
internal = allow_internal and _is_internal_service_host(host)
if parsed.scheme != "https" and not (internal and parsed.scheme == "http"): raise ValueError("URLs must use HTTPS except declared internal service DNS")
if _host_is_ip_blocked(host) and not internal: raise ValueError("loopback/link-local/metadata/private IP destinations are rejected")
return str(url).rstrip("/")
def validate_incident_path(path: str, incident_id: str, *, must_exist_under_incident: bool = False) -> str:
clean = str(path).replace("\\", "/").strip("/"); prefix = f"outputs/incidents/{incident_id}/"
if not clean or clean.startswith("../") or "/../" in clean or clean == "..": raise ValueError("unsafe workspace path")
if must_exist_under_incident and not clean.startswith(prefix): raise ValueError(f"path must be under {prefix}")
if clean.startswith("outputs/") and not clean.startswith(prefix): raise ValueError(f"incident artifacts must be under {prefix}")
return clean
def _incident_id_from_path(path: str) -> str | None:
m = re.search(r"outputs/incidents/([^/]+)/", path); return m.group(1) if m else None
async def emit_phase(ctx: RunContext[NoAuth], phase: str, *, evidence_count: int = 0, status: str = "started", started: float | None = None) -> None:
payload: dict[str, Any] = {"phase": phase, "status": status, "evidence_count": evidence_count}
if started is not None: payload["elapsed_ms"] = int((time.monotonic() - started) * 1000)
await ctx.emit_event(AgentEvent(kind="incident_phase", payload=payload))
async def workspace_write_text(ctx: RunContext[NoAuth], path: str, text: str, mime: str = "text/plain") -> None:
clean = validate_incident_path(path, _incident_id_from_path(path) or "unknown", must_exist_under_incident=True); data = text.encode()
try:
ctx.workspace.write_bytes(clean, data) # supported by deployed/local workspace implementations
except Exception:
ref = await ctx.write_artifact(clean.replace("/", "__"), data, mime); await ctx.emit_artifact(ref)
async def workspace_write_json(ctx: RunContext[NoAuth], path: str, data: BaseModel | dict[str, Any]) -> None:
await workspace_write_text(ctx, path, json.dumps(data.model_dump(mode="json") if isinstance(data, BaseModel) else data, indent=2, ensure_ascii=False, default=str), "application/json")
async def workspace_read_text(ctx: RunContext[NoAuth], path: str, max_bytes: int = 256_000) -> str:
clean = path.replace("\\", "/").strip("/")
return ctx.workspace.read_bytes(clean)[:max_bytes].decode("utf-8", errors="replace")
def classify_severity(title: str, symptoms: list[str], affected_services: list[str], environment: str) -> dict[str, str]:
text = " ".join([title, *symptoms]).lower(); rationale = []
if any(t in text for t in ["security", "breach", "compromise", "data loss", "irreversible", "payment down", "checkout down", "broad outage", "critical revenue"]): level = SeverityLevel.SEV1; rationale.append("broad outage, security compromise, irreversible data risk, or critical revenue path language detected")
elif len(affected_services) >= 5 or any(t in text for t in ["major", "unavailable", "outage", "all users", "500s"]): level = SeverityLevel.SEV2; rationale.append("major degradation or important subset unavailable")
elif any(t in text for t in ["degraded", "latency", "errors", "workaround"]): level = SeverityLevel.SEV3; rationale.append("limited degradation or workaround language detected")
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 build_hypotheses(title: str, symptoms: list[str], services: list[str], evidence: list[EvidenceEnvelope]) -> list[Hypothesis]:
blob = " ".join([title, *symptoms, *(ev.summary for ev in evidence)]).lower(); refs = [ev.id for ev in evidence if ev.status in {EvidenceStatus.degraded, EvidenceStatus.error}]; out = []
def add(s: str, c: float, tests: list[str]): 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", "back-off"]): add("A recent deployment introduced a crashing container or bad image revision.", .78, ["Compare current 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/resource status"])
if any(t in blob for t in ["latency", "timeout", "5xx", "500", "errors", "slo"]): 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 events, health checks, and allowlisted metrics"])
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(hypotheses: list[Hypothesis]) -> list[str]:
actions = ["Preserve evidence dossier before any mutation", "Run verification against user-visible health checks and metrics"]
for h in hypotheses:
if "crashing container" in h.statement: actions.append("Prepare a rollback plan for the named Deployment after operator approval")
if "Argo" in h.statement: actions.append("Prepare an Argo sync plan after operator approval if desired state is confirmed safe")
return list(dict.fromkeys(actions))
def _diagnosis_markdown(d: IncidentDiagnosis) -> str: return f"# Incident {d.incident_id} Diagnosis\n\nStatus: {d.status}\n\nLikely root cause: {d.likely_root_cause or 'Inconclusive'}\n"
def _postmortem_markdown(pm: Postmortem) -> str: return f"# Postmortem: {pm.incident_id}\n\n## Executive Summary\n{pm.executive_summary}\n\n## Root Cause\n{pm.root_cause}\n"
class IntegrationClient:
def __init__(self, ctx: RunContext[NoAuth], config: CommanderConfig): self.ctx = ctx; self.config = config
def _conf(self, name: str) -> str:
value = self.ctx.consumer_config(name, "")
return _validate_base_url(str(value), allow_internal=True) if value else ""
def _secret(self, name: str) -> str:
try: return self.ctx.consumer_secret(name).strip()
except ConsumerSetupMissing: return ""
async def kubernetes_collect(self, namespaces: list[str]) -> list[EvidenceEnvelope]: return [] if not namespaces else [_evidence("kubernetes", ",".join(namespaces), EvidenceStatus.setup_required, "Kubernetes setup missing; configure KUBERNETES_API_URL and KUBERNETES_BEARER_TOKEN.")]
async def argo_collect(self, applications: list[str]) -> list[EvidenceEnvelope]: return [] if not applications else [_evidence("argocd", ",".join(applications), EvidenceStatus.setup_required, "Argo CD setup missing; configure ARGOCD_URL and ARGOCD_TOKEN.")]
async def prometheus_collect(self, queries: list[str]) -> list[EvidenceEnvelope]:
if not queries: return []
if not self._conf("PROMETHEUS_URL"): return [_evidence("prometheus", ",".join(queries), EvidenceStatus.setup_required, "Prometheus setup missing; configure PROMETHEUS_URL.")]
return [_evidence("prometheus", q, EvidenceStatus.denied if q not in self.config.prometheus_query_allowlist else EvidenceStatus.ok, "Prometheus query denied because it is not in the configured allowlist." if q not in self.config.prometheus_query_allowlist else "Prometheus query accepted by allowlist.") for q in queries[:10]]
async def gitea_collect(self, repositories: list[str]) -> list[EvidenceEnvelope]: return [] if not repositories else [_evidence("gitea", ",".join(repositories), EvidenceStatus.setup_required, "Gitea setup missing; configure GITEA_URL and GITEA_TOKEN.")]
async def health_collect(self, urls: list[str]) -> list[EvidenceEnvelope]: return [_evidence("http_health", str(u), EvidenceStatus.error, "health check not attempted in deterministic sandbox without configured egress") for u in urls]
async def execute_step(self, step: RemediationStep, dry_run: bool) -> StepExecutionRecord:
if step.action_type not in ALLOWED_EXECUTION_ACTIONS: return StepExecutionRecord(step_id=step.id, status="denied", action_type=step.action_type, target=step.target, warnings=["action_type is not allowlisted"])
if not step.reversible: return StepExecutionRecord(step_id=step.id, status="denied", action_type=step.action_type, target=step.target, warnings=["step is not reversible"])
before = [_evidence("execution", step.target, EvidenceStatus.ok, "before-state captured for deterministic guarded operation")]
if dry_run: return StepExecutionRecord(step_id=step.id, status="dry_run", action_type=step.action_type, target=step.target, before_evidence=before, verification={"would_execute": True}, rollback_status="not_needed")
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 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"
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)
workspace_access = WorkspaceAccess.dynamic(max_files=200, allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY), require_reason=True, deny_patterns=("**/.env", "**/secrets/**", "**/*kubeconfig*"), max_total_size_bytes=25 * 1024 * 1024)
egress = EgressPolicy(allow_hosts=("kubernetes.default.svc", "prometheus.monitoring.svc", "argocd-server.argocd.svc"), allow_internal_services=("*.svc", "*.svc.cluster.local"), deny_internet_by_default=True)
tools_used = ("httpx", "pydantic", "kubernetes-api", "argocd-api", "prometheus-api", "gitea-api")
consumer_setup = ConsumerSetup.from_fields(ConsumerSetupField.config("KUBERNETES_API_URL", required=False, input_type="url"), ConsumerSetupField.secret("KUBERNETES_BEARER_TOKEN", required=False), ConsumerSetupField.config("ARGOCD_URL", required=False, input_type="url"), ConsumerSetupField.secret("ARGOCD_TOKEN", required=False), ConsumerSetupField.config("PROMETHEUS_URL", required=False, input_type="url"), ConsumerSetupField.secret("PROMETHEUS_TOKEN", required=False), ConsumerSetupField.config("GITEA_URL", required=False, input_type="url"), ConsumerSetupField.secret("GITEA_TOKEN", required=False))
@a2a.tool(description="Diagnose a production incident, collect bounded evidence, rank hypotheses, and write diagnosis artifacts", input_schema=DIAGNOSE_INPUT_SCHEMA, timeout_seconds=900, cost_class="incident-diagnostic", grant_mode="read_write_overlay", grant_write_prefixes=("outputs/incidents/",), grant_allow_patterns=("outputs/incidents/**",), grant_deny_patterns=("**/.env", "**/secrets/**", "**/*kubeconfig*"))
async def diagnose_incident(self, ctx: RunContext[NoAuth], incident_id: Annotated[str, Field(pattern=INCIDENT_ID_RE)], title: Annotated[str, Field(min_length=3, max_length=200)], symptoms: list[str], affected_services: list[str], environment: Environment = Environment.production, started_at: str = "", evidence_paths: list[str] = [], diagnostic_targets: DiagnosticTargets = DiagnosticTargets(), constraints: list[str] = []) -> dict[str, Any]:
_ = constraints; started = time.monotonic(); await emit_phase(ctx, "intake", started=started); evidence: list[EvidenceEnvelope] = []
for path in evidence_paths[:20]:
try: evidence.append(_evidence("workspace", path, EvidenceStatus.ok, f"Read caller-granted evidence path {path}", raw=await workspace_read_text(ctx, validate_incident_path(path, incident_id))))
except Exception as exc: evidence.append(_evidence("workspace", path, EvidenceStatus.error, f"Could not read granted evidence path: {exc}"))
client = IntegrationClient(ctx, self.config); batches = await asyncio.gather(client.kubernetes_collect(diagnostic_targets.kubernetes_namespaces), client.argo_collect(diagnostic_targets.argo_applications), client.prometheus_collect(diagnostic_targets.prometheus_queries), client.gitea_collect(diagnostic_targets.gitea_repositories), client.health_collect([str(u) for u in diagnostic_targets.healthcheck_urls]))
for b in batches: evidence.extend(b)
hypotheses = build_hypotheses(title, symptoms, affected_services, evidence); top = hypotheses[0] if hypotheses else None; setup = [e for e in evidence if e.status == EvidenceStatus.setup_required]
status: Literal["diagnosed", "inconclusive", "setup_required", "escalate"] = "setup_required" if setup and len(setup) == len(evidence or setup) else "diagnosed" if top and top.confidence >= .5 else "inconclusive"
d = IncidentDiagnosis(incident_id=incident_id, severity=classify_severity(title, symptoms, affected_services, environment.value), status=status, evidence_summary=evidence[:80], timeline=build_timeline(started_at, evidence, title), hypotheses=hypotheses, tests_run=["schema validation", "secret redaction", "prompt-injection scan", "bounded integration collection"], ruled_out=[] if status == "diagnosed" else ["No root cause was asserted without evidence"], likely_root_cause=top.statement if status == "diagnosed" and top else "", impacted_components=affected_services, recommended_actions=recommend_actions(hypotheses), required_approvals=["execute_remediation requires matching plan_digest and exact approval acknowledgement before live mutation"], verification_plan=["Verify user-visible health checks pass", "Verify relevant allowlisted metrics return to baseline"], artifact_paths=[f"outputs/incidents/{incident_id}/diagnosis.json", f"outputs/incidents/{incident_id}/summary.md"], receipts=[{"state_machine": list(STATE_MACHINE), "reasoning_policy": INCIDENT_REASONING_POLICY, "mutation": "none"}])
await workspace_write_json(ctx, f"outputs/incidents/{incident_id}/diagnosis.json", d); await workspace_write_text(ctx, f"outputs/incidents/{incident_id}/summary.md", _diagnosis_markdown(d), "text/markdown"); return d.model_dump(mode="json")
@a2a.tool(description="Create an ordered remediation plan from diagnosis artifacts. This skill never executes changes.", input_schema=PLAN_INPUT_SCHEMA, timeout_seconds=300, cost_class="incident-planning", grant_mode="read_write_overlay", grant_write_prefixes=("outputs/incidents/",), grant_allow_patterns=("outputs/incidents/**",))
async def plan_remediation(self, ctx: RunContext[NoAuth], incident_id: Annotated[str, Field(pattern=INCIDENT_ID_RE)], objective: str, diagnosis_path: str = "", change_window: str = "", allowed_actions: list[str] = [], forbidden_actions: list[str] = [], rollback_requirements: list[str] = []) -> dict[str, Any]:
diagnosis_text = ""
if diagnosis_path: diagnosis_text = await workspace_read_text(ctx, validate_incident_path(diagnosis_path, incident_id, must_exist_under_incident=True))
allowed = set(allowed_actions or sorted(ALLOWED_EXECUTION_ACTIONS)) & ALLOWED_EXECUTION_ACTIONS; blob = (objective + " " + diagnosis_text).lower(); steps: list[RemediationStep] = []
if "rollback" in blob and "kubernetes.rollback_deployment" in allowed: steps.append(RemediationStep(id="step-rollback-deployment", intent="Rollback one explicitly named Deployment after confirming drift has not occurred.", target="deployment:from-diagnosis", action_type="kubernetes.rollback_deployment", risk="medium", reversible=True, prerequisites=["operator confirms deployment name and namespace", *rollback_requirements], exact_verification=["health checks pass"], rollback="Re-apply captured pre-action revision if verification fails.", approval_required=True, evidence_to_capture=["deployment before/after"], parameters={"namespace": "from-diagnosis", "deployment": "from-diagnosis"}))
if "restart" in blob and "kubernetes.rollout_restart" in allowed: steps.append(RemediationStep(id="step-rollout-restart", intent="Restart one explicitly named Deployment.", target="deployment:from-diagnosis", action_type="kubernetes.rollout_restart", risk="low", reversible=True, prerequisites=["deployment name and namespace explicitly confirmed"], exact_verification=["new pods Ready", "health checks pass"], rollback="Rollback to previous ReplicaSet if readiness fails.", approval_required=True, evidence_to_capture=["deployment before/after"], parameters={"namespace": "from-diagnosis", "deployment": "from-diagnosis"}))
if "sync" in blob and "argocd.sync_application" in allowed: steps.append(RemediationStep(id="step-argocd-sync", intent="Sync one named Argo CD Application after desired revision is reviewed.", target="argocd:from-diagnosis", action_type="argocd.sync_application", risk="medium", reversible=True, prerequisites=["desired Git revision reviewed"], exact_verification=["Argo health Healthy"], rollback="Sync back to captured previous revision.", approval_required=True, evidence_to_capture=["Argo before/after"], parameters={"application": "from-diagnosis"}))
if not steps: steps.append(RemediationStep(id="step-collect-more-evidence", intent="Collect missing evidence before changing live state.", target="incident", action_type="evidence.collect", risk="low", reversible=True, rollback="No mutation performed.", approval_required=False))
p = RemediationPlan(incident_id=incident_id, status="planned" if any(s.action_type in ALLOWED_EXECUTION_ACTIONS for s in steps) else "blocked", objective=objective, change_window=change_window, steps=steps, forbidden_actions=forbidden_actions, required_approvals=["execute_remediation requires exact acknowledgement, approved step IDs, and matching plan_digest"], artifact_paths=[f"outputs/incidents/{incident_id}/remediation-plan.json"], receipts=[{"mutation": "none"}]); p = p.model_copy(update={"plan_digest": plan_digest(p)})
await workspace_write_json(ctx, f"outputs/incidents/{incident_id}/remediation-plan.json", p); return p.model_dump(mode="json")
@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"
def denied(reason: str, extra: dict[str, Any] | None = None) -> ExecutionResult: return ExecutionResult(incident_id=incident_id, status="denied", dry_run=dry_run, approval_provenance={"accepted": False, "reason": reason, **(extra or {})}, step_results=[], final_incident_state="unchanged", artifact_paths=[artifact])
if approval_acknowledgement != APPROVAL_ACK: r = denied("approval acknowledgement mismatch"); await workspace_write_json(ctx, artifact, r); return r.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: r = denied(f"plan could not be read or validated: {type(exc).__name__}"); await workspace_write_json(ctx, artifact, r); return r.model_dump(mode="json")
actual = plan.plan_digest or globals()["plan_digest"](plan)
if actual != plan_digest: r = denied("plan digest mismatch", {"expected": plan_digest, "actual": actual}); await workspace_write_json(ctx, artifact, r); return r.model_dump(mode="json")
approved = set(approved_step_ids); client = IntegrationClient(ctx, self.config); results: list[StepExecutionRecord] = []
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"
out = 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, out); return out.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]:
if verification_plan_path: _ = await workspace_read_text(ctx, validate_incident_path(verification_plan_path, incident_id, must_exist_under_incident=True))
if observation_seconds: await asyncio.sleep(min(observation_seconds, 2))
client = IntegrationClient(ctx, self.config); ev = await client.health_collect([str(u) for u in healthcheck_urls]); ev.extend(await client.prometheus_collect(prometheus_queries))
checks = [RecoveryCheck(name=e.target, status="pass" if e.status == EvidenceStatus.ok else "setup_required" if e.status == EvidenceStatus.setup_required else "fail", summary=e.summary, evidence_refs=[e.id]) for e in ev] or [RecoveryCheck(name="required_checks", status="fail", summary="No user-visible health checks or allowlisted metrics were supplied; recovery cannot be claimed.")]
recovered = all(c.status == "pass" for c in checks); out = RecoveryVerification(incident_id=incident_id, checks=checks, before_after_comparison={"evidence_count": len(ev), "observation_seconds": observation_seconds}, error_budget_indicators={"available": bool(prometheus_queries), "note": "Only allowlisted metrics are queried."}, regression_signals=[] if recovered else ["one or more required checks failed or were missing"], recovery_confidence=.9 if recovered else .2, recovered=recovered, remaining_risks=[] if recovered else ["Do not close incident until required user-visible checks pass"], artifact_paths=[f"outputs/incidents/{incident_id}/verification.json"], receipts=[{"recovery_gate": "recovered is true only when all required checks pass"}])
await workspace_write_json(ctx, f"outputs/incidents/{incident_id}/verification.json", out); return out.model_dump(mode="json")
@a2a.tool(description="Generate a blameless postmortem from saved incident artifacts", input_schema=POSTMORTEM_INPUT_SCHEMA, timeout_seconds=300, cost_class="incident-postmortem", grant_mode="read_write_overlay", grant_write_prefixes=("outputs/incidents/",), grant_allow_patterns=("outputs/incidents/**",))
async def generate_postmortem(self, ctx: RunContext[NoAuth], incident_id: Annotated[str, Field(pattern=INCIDENT_ID_RE)], include_customer_summary: bool = True, include_action_items: bool = True) -> dict[str, Any]:
diagnosis: IncidentDiagnosis | None = None
try: diagnosis = IncidentDiagnosis.model_validate_json(await workspace_read_text(ctx, f"outputs/incidents/{incident_id}/diagnosis.json"))
except Exception: pass
corrective = [CorrectiveAction(description="Add or improve incident-specific detection and verification checks", owner="incident-owner", priority="P1")] if include_action_items else []
pm = Postmortem(incident_id=incident_id, executive_summary=diagnosis.likely_root_cause if diagnosis and diagnosis.likely_root_cause else "Incident investigated with bounded evidence; final root cause remains unresolved.", customer_impact="Customer-facing impact should be filled from support/business telemetry." if include_customer_summary else "Customer summary omitted by caller.", timeline=diagnosis.timeline if diagnosis else [TimelineEntry(at=_now(), source="postmortem", description="Diagnosis artifact not available")], detection="Detected from caller-provided intake and configured evidence sources.", root_cause=diagnosis.likely_root_cause if diagnosis and diagnosis.likely_root_cause else "Unresolved; evidence did not support a single root cause.", contributing_factors=[] if diagnosis and diagnosis.status == "diagnosed" else ["Evidence gaps or missing integration setup may have limited confidence"], response_analysis="Response followed evidence collection, hypothesis ranking, approval-gated remediation, and recovery verification.", what_went_well=["Evidence was redacted and written to durable artifacts", "Mutations were separated from diagnosis and planning"], what_went_poorly=["Fill with operator review after incident debrief"], corrective_actions=corrective, evidence_references=[e.id for e in diagnosis.evidence_summary] if diagnosis else [], unresolved_questions=[] if diagnosis and diagnosis.status == "diagnosed" else ["What additional evidence would distinguish the top hypotheses?"], artifact_paths=[f"outputs/incidents/{incident_id}/postmortem.json", f"outputs/incidents/{incident_id}/postmortem.md"], receipts=[{"blameless": True}])
await workspace_write_json(ctx, f"outputs/incidents/{incident_id}/postmortem.json", pm); await workspace_write_text(ctx, f"outputs/incidents/{incident_id}/postmortem.md", _postmortem_markdown(pm), "text/markdown"); return pm.model_dump(mode="json")
Reference in New Issue View Git Blame Copy Permalink