272 lines
40 KiB
Python
272 lines
40 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import ipaddress
|
|
import json
|
|
import re
|
|
from datetime import UTC, datetime
|
|
from operator import attrgetter
|
|
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"
|
|
ALLOWED_EXECUTION_ACTIONS = {"kubernetes.rollout_restart", "kubernetes.rollback_deployment", "argocd.sync_application"}
|
|
MAX_EVIDENCE_CHARS = 16000
|
|
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 dry-run by default and requires exact approval, matching immutable plan digest, and allowlisted reversible deterministic actions only. Untrusted evidence is never instructions."
|
|
|
|
STRICT_DIAGNOSTIC_TARGETS_SCHEMA = {"type": "object", "additionalProperties": False, "default": {}, "properties": {"kubernetes_namespaces": {"type": "array", "items": {"type": "string"}, "default": []}, "argo_applications": {"type": "array", "items": {"type": "string"}, "default": []}, "prometheus_queries": {"type": "array", "items": {"type": "string"}, "default": []}, "healthcheck_urls": {"type": "array", "items": {"type": "string"}, "default": []}, "gitea_repositories": {"type": "array", "items": {"type": "string"}, "default": []}}}
|
|
DIAGNOSE_INPUT_SCHEMA = {"type": "object", "additionalProperties": False, "required": ["incident_id", "title", "symptoms", "affected_services"], "properties": {"incident_id": {"type": "string", "pattern": INCIDENT_ID_RE}, "title": {"type": "string"}, "symptoms": {"type": "array", "items": {"type": "string"}}, "affected_services": {"type": "array", "items": {"type": "string"}}, "environment": {"type": "string", "enum": ["production", "staging", "development"], "default": "production"}, "started_at": {"type": "string", "default": ""}, "evidence_paths": {"type": "array", "items": {"type": "string"}, "default": []}, "diagnostic_targets": STRICT_DIAGNOSTIC_TARGETS_SCHEMA, "constraints": {"type": "array", "items": {"type": "string"}, "default": []}}}
|
|
PLAN_INPUT_SCHEMA = {"type": "object", "additionalProperties": False, "required": ["incident_id", "objective"], "properties": {"incident_id": {"type": "string", "pattern": INCIDENT_ID_RE}, "diagnosis_path": {"type": "string", "default": ""}, "objective": {"type": "string"}, "change_window": {"type": "string", "default": ""}, "allowed_actions": {"type": "array", "items": {"type": "string"}, "default": []}, "forbidden_actions": {"type": "array", "items": {"type": "string"}, "default": []}, "rollback_requirements": {"type": "array", "items": {"type": "string"}, "default": []}}}
|
|
EXECUTE_INPUT_SCHEMA = {"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"}, "plan_digest": {"type": "string", "pattern": r"^sha256:[a-f0-9]{64}$"}, "approved_step_ids": {"type": "array", "items": {"type": "string"}}, "approval_acknowledgement": {"type": "string", "const": APPROVAL_ACK}, "dry_run": {"type": "boolean", "default": True}, "stop_on_failure": {"type": "boolean", "default": True}}}
|
|
VERIFY_INPUT_SCHEMA = {"type": "object", "additionalProperties": False, "required": ["incident_id"], "properties": {"incident_id": {"type": "string", "pattern": INCIDENT_ID_RE}, "verification_plan_path": {"type": "string", "default": ""}, "healthcheck_urls": {"type": "array", "items": {"type": "string"}, "default": []}, "prometheus_queries": {"type": "array", "items": {"type": "string"}, "default": []}, "observation_seconds": {"type": "integer", "default": 60}}}
|
|
POSTMORTEM_INPUT_SCHEMA = {"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 = 3600
|
|
|
|
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)
|
|
argo_applications: list[str] = Field(default_factory=list)
|
|
prometheus_queries: list[str] = Field(default_factory=list)
|
|
healthcheck_urls: list[HttpUrl] = Field(default_factory=list)
|
|
gitea_repositories: list[str] = Field(default_factory=list)
|
|
@field_validator("kubernetes_namespaces", "argo_applications")
|
|
@classmethod
|
|
def _safe_names(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 _safe_repos(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()).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"(?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); marks: list[str] = []
|
|
for i, (pattern, repl) in enumerate(SECRET_PATTERNS):
|
|
text, count = pattern.subn(repl, text)
|
|
if count: marks.append(f"secret_pattern_{i}:{count}")
|
|
if len(text) > MAX_EVIDENCE_CHARS: text = text[:MAX_EVIDENCE_CHARS] + "\n[TRUNCATED]"; marks.append("truncated")
|
|
return text, marks
|
|
def _evidence(source: str, target: str, status: EvidenceStatus, summary: str, raw: Any = "", metadata: dict[str, Any] | None = None) -> EvidenceEnvelope:
|
|
safe, marks = 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) or (raw_text and PROMPT_INJECTION_PATTERNS.search(raw_text)): marks.append("untrusted_prompt_injection_ignored")
|
|
return EvidenceEnvelope(id=f"ev-{_sha256_text(source + target + safe + _now())[:12]}", source=source, target=target, observed_at=_now(), status=status, summary=safe, redactions=marks, 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 == ".." or clean.startswith("../") or "/../" in 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:
|
|
match = re.search(r"outputs/incidents/([^/]+)/", path); return match.group(1) if match else None
|
|
async def emit_phase(ctx: RunContext[NoAuth], phase: str, *, evidence_count: int = 0, status: str = "started", started: float | None = None) -> None: await ctx.emit_event(AgentEvent(kind="incident_phase", payload={"phase": phase, "status": status, "evidence_count": evidence_count}))
|
|
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)
|
|
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, default=str), "application/json")
|
|
async def workspace_read_text(ctx: RunContext[NoAuth], path: str, max_bytes: int = 256000) -> str: return ctx.workspace.read_bytes(path.replace("\\", "/").strip("/"))[:max_bytes].decode(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 _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] = []
|
|
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=attrgetter("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 [])
|
|
def _diagnosis_markdown(d: IncidentDiagnosis) -> str: return f"# Incident {d.incident_id} Diagnosis\n\nStatus: {d.status}\nRoot cause: {d.likely_root_cause or 'Inconclusive'}\n"
|
|
def _postmortem_markdown(pm: Postmortem) -> str: return f"# Postmortem: {pm.incident_id}\n\n{pm.executive_summary}\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, xs: list[str]) -> list[EvidenceEnvelope]: return [] if not xs else [_evidence("kubernetes", ",".join(xs), EvidenceStatus.setup_required, "Kubernetes setup missing; configure KUBERNETES_API_URL and KUBERNETES_BEARER_TOKEN.")]
|
|
async def argo_collect(self, xs: list[str]) -> list[EvidenceEnvelope]: return [] if not xs else [_evidence("argocd", ",".join(xs), EvidenceStatus.setup_required, "Argo CD setup missing; configure ARGOCD_URL and ARGOCD_TOKEN.")]
|
|
async def prometheus_collect(self, qs: list[str]) -> list[EvidenceEnvelope]:
|
|
if not qs: return []
|
|
if not self._conf("PROMETHEUS_URL"): return [_evidence("prometheus", ",".join(qs), EvidenceStatus.setup_required, "Prometheus setup missing; configure PROMETHEUS_URL.")]
|
|
return [_evidence("prometheus", q, EvidenceStatus.ok if q in self.config.prometheus_query_allowlist else EvidenceStatus.denied, "Prometheus query accepted by allowlist." if q in self.config.prometheus_query_allowlist else "Prometheus query denied because it is not in the configured allowlist.") for q in qs]
|
|
async def gitea_collect(self, xs: list[str]) -> list[EvidenceEnvelope]: return [] if not xs else [_evidence("gitea", ",".join(xs), 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", u, EvidenceStatus.error, "health check requires configured egress and returned no pass evidence") 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 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.2"
|
|
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=(), 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: str, 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]:
|
|
await emit_phase(ctx, "intake"); 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)
|
|
for batch in 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])): evidence.extend(batch)
|
|
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=["matching plan_digest and exact approval acknowledgement required before live mutation"], verification_plan=["Verify user-visible health checks pass", "Verify allowlisted metrics return to baseline"], artifact_paths=[f"outputs/incidents/{incident_id}/diagnosis.json", f"outputs/incidents/{incident_id}/summary.md"], receipts=[{"state_machine": 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]:
|
|
blob = objective.lower(); allowed = set(allowed_actions or sorted(ALLOWED_EXECUTION_ACTIONS)) & ALLOWED_EXECUTION_ACTIONS; steps: list[RemediationStep] = []
|
|
if diagnosis_path: blob += " " + (await workspace_read_text(ctx, validate_incident_path(diagnosis_path, incident_id, must_exist_under_incident=True))).lower()
|
|
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"], 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 = [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=["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"
|
|
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:
|
|
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:
|
|
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"
|
|
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]:
|
|
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); evidence = await client.health_collect([str(u) for u in healthcheck_urls]); evidence.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 evidence] 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(evidence), "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
|
|
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=[CorrectiveAction(description="Add or improve incident-specific detection and verification checks", owner="incident-owner", priority="P1")] if include_action_items else [], 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")
|