diff --git a/agent.py b/agent.py index bef24bb..c86195e 100644 --- a/agent.py +++ b/agent.py @@ -1,4 +1,3 @@ -"""Production Incident Commander A2A agent.""" from __future__ import annotations import asyncio @@ -6,7 +5,6 @@ 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 @@ -15,62 +13,36 @@ 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 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." +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: 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}}} +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 = Field(default=3600, ge=60, le=86400) + 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 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) + 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) @@ -104,45 +76,46 @@ class Postmortem(BaseModel): 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) + 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 _names_are_safe(cls, values: list[str]) -> list[str]: + 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 _repos_are_safe(cls, values: list[str]) -> list[str]: + 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("utf-8")).hexdigest() +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"-----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]")] +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); redactions: list[str] = [] - for idx, (pattern, repl) in enumerate(SECRET_PATTERNS): + 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: 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 + 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_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 {}) + 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 @@ -160,27 +133,20 @@ def _validate_base_url(url: str, *, allow_internal: bool = False) -> str: 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 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: - 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)) + 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) # supported by deployed/local workspace implementations + 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, 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") +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") @@ -190,46 +156,39 @@ def classify_severity(title: str, symptoms: list[str], affected_services: list[s 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"]) + 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"]) 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" +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 "" + 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 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 plan parameters"]) + 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" @@ -237,60 +196,58 @@ class ProductionIncidentCommander(A2AAgent[CommanderConfig, 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) + 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: 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] = [] + 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); 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) + 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=["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"}]) + 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]: - 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] = [] + 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", "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 "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.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)}) + 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" - 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") + 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])) 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") + 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])) 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] = [] + 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])) + 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" - 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") + 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"}])) @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"}]) + 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/**",)) @@ -298,6 +255,5 @@ class ProductionIncidentCommander(A2AAgent[CommanderConfig, NoAuth]): 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}]) + 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")