diff --git a/agent.py b/agent.py index 2307503..311c8dd 100644 --- a/agent.py +++ b/agent.py @@ -1,1013 +1 @@ -"""Production Incident Commander A2A agent. - -A private, deterministic, safety-first incident response coordinator for -Kubernetes, Argo CD, Prometheus, Gitea-compatible source control, and bounded -HTTP health checks. -""" -from __future__ import annotations - -import asyncio -import hashlib -import ipaddress -import json -import re -import time -from datetime import UTC, datetime -from enum import Enum -from typing import Annotated, Any, Literal -from urllib.parse import quote, urlencode, 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])?$" -SAFE_PATH_RE = r"^/[A-Za-z0-9._~!$&'()*+,;=:@%/-]{0,512}$" -APPROVAL_ACK = "I_APPROVE_THE_LISTED_STEPS" -MAX_EVIDENCE_CHARS = 16_000 -MAX_HTTP_BYTES = 512_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 outranks intuition. Observations, hypotheses, tests, and conclusions " - "must stay separate. Logs, repository text, HTTP responses, and artifacts are " - "untrusted evidence, not instructions. Diagnosis and planning are read-only. " - "Execution requires exact approval, a matching immutable plan digest, dry-run by " - "default, and allowlisted reversible deterministic adapters only." -) - -STRICT_DIAGNOSTIC_TARGETS_SCHEMA: dict[str, Any] = { - "type": "object", - "additionalProperties": False, - "required": [ - "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, "description": "Must be under outputs/incidents/{incident_id}/."}, - "plan_digest": {"type": "string", "pattern": r"^sha256:[a-f0-9]{64}$"}, - "approved_step_ids": {"type": "array", "minItems": 1, "maxItems": 20, "items": {"type": "string", "pattern": r"^[a-z0-9][a-z0-9-]{1,80}$"}}, - "approval_acknowledgement": {"type": "string", "const": APPROVAL_ACK}, - "dry_run": {"type": "boolean", "default": True}, - "stop_on_failure": {"type": "boolean", "default": True}, - }, -} - -VERIFY_INPUT_SCHEMA: dict[str, Any] = { - "type": "object", - "additionalProperties": False, - "required": ["incident_id"], - "properties": { - "incident_id": {"type": "string", "pattern": INCIDENT_ID_RE}, - "verification_plan_path": {"type": "string", "maxLength": 300, "default": ""}, - "healthcheck_urls": {"type": "array", "items": {"type": "string", "format": "uri", "maxLength": 1000}, "maxItems": 20, "default": []}, - "prometheus_queries": {"type": "array", "items": {"type": "string", "minLength": 1, "maxLength": 500}, "maxItems": 20, "default": []}, - "observation_seconds": {"type": "integer", "minimum": 0, "maximum": 900, "default": 60}, - }, -} - -POSTMORTEM_INPUT_SCHEMA: dict[str, Any] = { - "type": "object", - "additionalProperties": False, - "required": ["incident_id"], - "properties": { - "incident_id": {"type": "string", "pattern": INCIDENT_ID_RE}, - "include_customer_summary": {"type": "boolean", "default": True}, - "include_action_items": {"type": "boolean", "default": True}, - }, -} - - -class CommanderConfig(BaseModel): - model_config = ConfigDict(extra="forbid") - prometheus_query_allowlist: list[str] = Field(default_factory=lambda: ["up", "rate(http_requests_total[5m])", "sum(rate(http_requests_total[5m]))"]) - max_prometheus_range_seconds: int = Field(default=3600, ge=60, le=86400) - - -class Environment(str, Enum): - production = "production" - staging = "staging" - development = "development" - - -class SeverityLevel(str, Enum): - SEV1 = "SEV1" - SEV2 = "SEV2" - SEV3 = "SEV3" - SEV4 = "SEV4" - - -class EvidenceStatus(str, Enum): - ok = "ok" - degraded = "degraded" - error = "error" - setup_required = "setup_required" - skipped = "skipped" - denied = "denied" - - -class EvidenceEnvelope(BaseModel): - model_config = ConfigDict(extra="forbid") - id: str - source: str - target: str - observed_at: str - status: EvidenceStatus - summary: str - redactions: list[str] = Field(default_factory=list) - raw_ref: str = "" - hash: str = "" - metadata: dict[str, Any] = Field(default_factory=dict) - - -class Hypothesis(BaseModel): - model_config = ConfigDict(extra="forbid") - id: str - statement: str - confidence: float = Field(ge=0, le=1) - supporting_evidence: list[str] = Field(default_factory=list) - conflicting_evidence: list[str] = Field(default_factory=list) - discriminating_tests: list[str] = Field(default_factory=list) - - -class TimelineEntry(BaseModel): - model_config = ConfigDict(extra="forbid") - at: str - source: str - description: str - evidence_refs: list[str] = Field(default_factory=list) - - -class RemediationStep(BaseModel): - model_config = ConfigDict(extra="forbid") - id: str - intent: str - target: str - action_type: str - risk: Literal["low", "medium", "high"] - reversible: bool - prerequisites: list[str] = Field(default_factory=list) - exact_verification: list[str] = Field(default_factory=list) - rollback: str - approval_required: bool - evidence_to_capture: list[str] = Field(default_factory=list) - parameters: dict[str, Any] = Field(default_factory=dict) - - -class RemediationPlan(BaseModel): - model_config = ConfigDict(extra="forbid") - incident_id: str - status: Literal["planned", "blocked", "setup_required"] - objective: str - change_window: str = "" - steps: list[RemediationStep] - forbidden_actions: list[str] = Field(default_factory=list) - required_approvals: list[str] = Field(default_factory=list) - plan_digest: str = "" - artifact_paths: list[str] = Field(default_factory=list) - receipts: list[dict[str, Any]] = Field(default_factory=list) - - -class StepExecutionRecord(BaseModel): - model_config = ConfigDict(extra="forbid") - step_id: str - status: Literal["dry_run", "executed", "denied", "failed", "skipped", "drift_detected"] - action_type: str - target: str - before_evidence: list[EvidenceEnvelope] = Field(default_factory=list) - after_evidence: list[EvidenceEnvelope] = Field(default_factory=list) - verification: dict[str, Any] = Field(default_factory=dict) - rollback_status: str = "" - warnings: list[str] = Field(default_factory=list) - - -class ExecutionResult(BaseModel): - model_config = ConfigDict(extra="forbid") - incident_id: str - status: Literal["denied", "dry_run", "partially_executed", "executed", "failed"] - dry_run: bool - approval_provenance: dict[str, Any] - step_results: list[StepExecutionRecord] - final_incident_state: str - artifact_paths: list[str] - receipts: list[dict[str, Any]] = Field(default_factory=list) - - -class RecoveryCheck(BaseModel): - model_config = ConfigDict(extra="forbid") - name: str - status: Literal["pass", "fail", "skipped", "setup_required"] - summary: str - evidence_refs: list[str] = Field(default_factory=list) - - -class RecoveryVerification(BaseModel): - model_config = ConfigDict(extra="forbid") - incident_id: str - checks: list[RecoveryCheck] - before_after_comparison: dict[str, Any] - error_budget_indicators: dict[str, Any] - regression_signals: list[str] - recovery_confidence: float = Field(ge=0, le=1) - recovered: bool - remaining_risks: list[str] - artifact_paths: list[str] - receipts: list[dict[str, Any]] = Field(default_factory=list) - - -class CorrectiveAction(BaseModel): - model_config = ConfigDict(extra="forbid") - description: str - owner: str - priority: Literal["P0", "P1", "P2", "P3"] - due_date: str = "" - status: Literal["open", "in_progress", "done", "deferred"] = "open" - - -class Postmortem(BaseModel): - model_config = ConfigDict(extra="forbid") - incident_id: str - executive_summary: str - customer_impact: str - timeline: list[TimelineEntry] - detection: str - root_cause: str - contributing_factors: list[str] - response_analysis: str - what_went_well: list[str] - what_went_poorly: list[str] - corrective_actions: list[CorrectiveAction] - evidence_references: list[str] - unresolved_questions: list[str] - artifact_paths: list[str] - receipts: list[dict[str, Any]] = Field(default_factory=list) - - -class IncidentDiagnosis(BaseModel): - model_config = ConfigDict(extra="forbid") - incident_id: str - severity: dict[str, str] - status: Literal["diagnosed", "inconclusive", "setup_required", "escalate"] - evidence_summary: list[EvidenceEnvelope] - timeline: list[TimelineEntry] - hypotheses: list[Hypothesis] - tests_run: list[str] - ruled_out: list[str] - likely_root_cause: str = "" - impacted_components: list[str] - recommended_actions: list[str] - required_approvals: list[str] - verification_plan: list[str] - artifact_paths: list[str] - receipts: list[dict[str, Any]] = Field(default_factory=list) - - -class DiagnosticTargets(BaseModel): - model_config = ConfigDict(extra="forbid") - kubernetes_namespaces: list[str] = Field(default_factory=list, max_length=30) - argo_applications: list[str] = Field(default_factory=list, max_length=30) - prometheus_queries: list[str] = Field(default_factory=list, max_length=20) - healthcheck_urls: list[HttpUrl] = Field(default_factory=list, max_length=20) - gitea_repositories: list[str] = Field(default_factory=list, max_length=20) - - @field_validator("kubernetes_namespaces", "argo_applications") - @classmethod - def _names_are_safe(cls, values: list[str]) -> list[str]: - for value in values: - if not re.fullmatch(DNS_SAFE_RE, value): - raise ValueError(f"not DNS-safe: {value}") - return values - - @field_validator("gitea_repositories") - @classmethod - def _repos_are_safe(cls, values: list[str]) -> list[str]: - for value in values: - if not re.fullmatch(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$", value): - raise ValueError(f"repository must be owner/name: {value}") - return values - - -def _now() -> str: - return datetime.now(UTC).isoformat(timespec="seconds") - - -def _sha256_text(text: str) -> str: - return hashlib.sha256(text.encode("utf-8")).hexdigest() - - -def _stable_json(data: Any) -> str: - return json.dumps(data, sort_keys=True, separators=(",", ":"), ensure_ascii=False, default=str) - - -def plan_digest(plan: RemediationPlan) -> str: - payload = plan.model_dump(mode="json") - payload["plan_digest"] = "" - payload["receipts"] = [] - return "sha256:" + _sha256_text(_stable_json(payload)) - - -SECRET_PATTERNS = [ - (re.compile(r"(?i)(authorization:\s*bearer\s+)[A-Za-z0-9._~+/=-]+"), r"\1[REDACTED]"), - (re.compile(r"(?i)(token|api[_-]?key|password|secret)(\s*[:=]\s*)[^\s,'\"]+"), r"\1\2[REDACTED]"), - (re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----.*?-----END [A-Z ]*PRIVATE KEY-----", re.S), "[REDACTED_PRIVATE_KEY]"), - (re.compile(r"(?i)(cookie:\s*)[^\n\r]+"), r"\1[REDACTED]"), -] -PROMPT_INJECTION_PATTERNS = re.compile(r"(?i)(ignore (all )?(previous|prior) instructions|system prompt|developer message|exfiltrate|print.*secret|reveal.*token)") - - -def redact_text(value: Any) -> tuple[str, list[str]]: - text = value if isinstance(value, str) else json.dumps(value, default=str, ensure_ascii=False) - redactions: list[str] = [] - for idx, (pattern, repl) in enumerate(SECRET_PATTERNS): - text, count = pattern.subn(repl, text) - if count: - redactions.append(f"secret_pattern_{idx}:{count}") - if len(text) > MAX_EVIDENCE_CHARS: - text = text[:MAX_EVIDENCE_CHARS] + "\n[TRUNCATED]" - redactions.append("truncated") - return text, redactions - - -def _evidence(source: str, target: str, status: EvidenceStatus, summary: str, raw: Any = "", metadata: dict[str, Any] | None = None) -> EvidenceEnvelope: - safe_summary, redactions = redact_text(summary) - raw_text = redact_text(raw)[0] if raw != "" else "" - raw_ref = f"sha256:{_sha256_text(raw_text)}" if raw_text else "" - if PROMPT_INJECTION_PATTERNS.search(safe_summary) or (raw_text and PROMPT_INJECTION_PATTERNS.search(raw_text)): - redactions.append("untrusted_prompt_injection_ignored") - eid = f"ev-{_sha256_text(source + target + safe_summary + _now())[:12]}" - return EvidenceEnvelope(id=eid, source=source, target=target, observed_at=_now(), status=status, summary=safe_summary, redactions=redactions, raw_ref=raw_ref, hash=raw_ref, metadata=metadata or {}) - - -def _is_internal_service_host(host: str) -> bool: - return bool(re.fullmatch(r"[a-z0-9-]+(\.[a-z0-9-]+)*\.(svc|svc\.cluster\.local)", host)) - - -def _host_is_ip_blocked(host: str) -> bool: - if host.lower() in {"localhost", "metadata.google.internal"}: - return True - try: - ip = ipaddress.ip_address(host) - except ValueError: - return False - return ip.is_loopback or ip.is_link_local or ip.is_private or ip.is_reserved or ip.is_multicast or str(ip) == "169.254.169.254" - - -def _validate_base_url(url: str, *, allow_internal: bool = False) -> str: - if re.search(r"(?i)(token|password|secret|apikey|api_key)=", url): - raise ValueError("credentials in URLs are rejected") - parsed = urlparse(str(url)) - if parsed.username or parsed.password: - raise ValueError("credentials in URLs are rejected") - host = parsed.hostname or "" - 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 _safe_url_join(base: str, path: str, query: dict[str, str] | None = None) -> str: - base = _validate_base_url(base, allow_internal=True) - if not re.fullmatch(SAFE_PATH_RE, path) or "//" in path: - raise ValueError("unsafe provider path") - url = base + path - if query: - url += "?" + urlencode(query) - _validate_base_url(url, allow_internal=True) - return url - - -def validate_incident_path(path: str, incident_id: str, *, must_exist_under_incident: bool = False) -> str: - clean = str(path).replace("\\", "/").strip("/") - if not clean or clean.startswith("../") or "/../" in clean or clean == "..": - raise ValueError("unsafe workspace path") - prefix = f"outputs/incidents/{incident_id}/" - 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 _artifact_name(path: str) -> str: - return path.replace("/", "__") - - -async def emit_phase(ctx: RunContext[NoAuth], phase: str, *, evidence_count: int = 0, status: str = "started", started: float | None = None) -> None: - payload: dict[str, Any] = {"phase": phase, "status": status, "evidence_count": evidence_count} - if started is not None: - payload["elapsed_ms"] = int((time.monotonic() - started) * 1000) - await ctx.emit_event(AgentEvent(kind="incident_phase", payload=payload)) - - -async def workspace_write_json(ctx: RunContext[NoAuth], path: str, data: BaseModel | dict[str, Any]) -> None: - content = data.model_dump(mode="json") if isinstance(data, BaseModel) else data - await workspace_write_text(ctx, path, json.dumps(content, indent=2, ensure_ascii=False, default=str), "application/json") - - -async def workspace_write_text(ctx: RunContext[NoAuth], path: str, text: str, mime: str = "text/plain") -> None: - incident_id = _incident_id_from_path(path) or "unknown" - clean = validate_incident_path(path, incident_id, must_exist_under_incident=True) - data = text.encode("utf-8") - try: - ws = ctx.workspace - if hasattr(ws, "write_bytes"): - ws.write_bytes(clean, data) - else: - grant = await ws.request_access(files=[clean], mode=WorkspaceMode.READ_WRITE_OVERLAY, reason="write incident artifact", purpose="incident artifact write") - view = await ws.open_view(purpose="write incident artifact", hints=[clean], max_files=1, mode=WorkspaceMode.READ_WRITE_OVERLAY, reason="write incident artifact") - _ = grant - await view.write(clean, data) - except Exception: - ref = await ctx.write_artifact(_artifact_name(clean), data, mime) - await ctx.emit_artifact(ref) - - -async def workspace_read_text(ctx: RunContext[NoAuth], path: str, max_bytes: int = 256_000) -> str: - clean = path.replace("\\", "/").strip("/") - try: - ws = ctx.workspace - if hasattr(ws, "read_bytes"): - data = ws.read_bytes(clean) - else: - grant = await ws.request_access(files=[clean], mode=WorkspaceMode.READ_ONLY, reason="read granted incident evidence", purpose="incident read") - view = await ws.open_view(purpose="read granted incident evidence", hints=[clean], max_files=1, mode=WorkspaceMode.READ_ONLY, reason="read granted incident evidence") - _ = grant - data = await view.read(clean) - except Exception as exc: - raise FileNotFoundError(f"unable to read workspace path {clean}: {exc}") from exc - return data[:max_bytes].decode("utf-8", errors="replace") - - -def _incident_id_from_path(path: str) -> str | None: - match = re.search(r"outputs/incidents/([^/]+)/", path) - return match.group(1) if match else None - - -def classify_severity(title: str, symptoms: list[str], affected_services: list[str], environment: str) -> dict[str, str]: - text = " ".join([title, *symptoms]).lower() - rationale: list[str] = [] - if any(term in text for term 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(term in text for term in ["major", "unavailable", "outage", "all users", "500s"]): - level = SeverityLevel.SEV2 - rationale.append("major degradation or important subset unavailable") - elif any(term in text for term 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"} - - -async def _bounded_gather(tasks: list[Any], limit: int) -> list[Any]: - sem = asyncio.Semaphore(limit) - - async def run(coro: Any) -> Any: - async with sem: - return await coro - - return list(await asyncio.gather(*(run(t) for t in tasks))) - - -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, "") - if value is None or str(value).strip() == "": - return "" - return _validate_base_url(str(value), allow_internal=True) - - def _secret(self, name: str) -> str: - try: - value = self.ctx.consumer_secret(name) - except ConsumerSetupMissing: - return "" - return value if value.strip() else "" - - async def kubernetes_collect(self, namespaces: list[str]) -> list[EvidenceEnvelope]: - base = self._conf("KUBERNETES_API_URL") - token = self._secret("KUBERNETES_BEARER_TOKEN") - if not namespaces: - return [] - if not base or not token: - return [_evidence("kubernetes", ",".join(namespaces), EvidenceStatus.setup_required, "Kubernetes setup missing; configure KUBERNETES_API_URL and KUBERNETES_BEARER_TOKEN.")] - tasks = [] - for ns in namespaces[:10]: - for kind, path in (("pods", f"/api/v1/namespaces/{quote(ns)}/pods"), ("events", f"/api/v1/namespaces/{quote(ns)}/events"), ("deployments", f"/apis/apps/v1/namespaces/{quote(ns)}/deployments")): - tasks.append(self._http_json("kubernetes", f"{ns}/{kind}", _safe_url_join(base, path), token)) - return await _bounded_gather(tasks, limit=4) - - async def argo_collect(self, applications: list[str]) -> list[EvidenceEnvelope]: - base = self._conf("ARGOCD_URL") - token = self._secret("ARGOCD_TOKEN") - if not applications: - return [] - if not base or not token: - return [_evidence("argocd", ",".join(applications), EvidenceStatus.setup_required, "Argo CD setup missing; configure ARGOCD_URL and ARGOCD_TOKEN.")] - return await _bounded_gather([self._http_json("argocd", app, _safe_url_join(base, f"/api/v1/applications/{quote(app)}"), token) for app in applications[:10]], limit=4) - - async def prometheus_collect(self, queries: list[str]) -> list[EvidenceEnvelope]: - base = self._conf("PROMETHEUS_URL") - token = self._secret("PROMETHEUS_TOKEN") - if not queries: - return [] - if not base: - return [_evidence("prometheus", ",".join(queries), EvidenceStatus.setup_required, "Prometheus setup missing; configure PROMETHEUS_URL.")] - out: list[EvidenceEnvelope] = [] - for q in queries[:10]: - if q not in self.config.prometheus_query_allowlist: - out.append(_evidence("prometheus", q, EvidenceStatus.denied, "Prometheus query denied because it is not in the configured allowlist.")) - continue - out.append(await self._http_json("prometheus", q, _safe_url_join(base, "/api/v1/query", {"query": q}), token)) - return out - - async def gitea_collect(self, repositories: list[str]) -> list[EvidenceEnvelope]: - base = self._conf("GITEA_URL") - token = self._secret("GITEA_TOKEN") - if not repositories: - return [] - if not base or not token: - return [_evidence("gitea", ",".join(repositories), EvidenceStatus.setup_required, "Gitea setup missing; configure GITEA_URL and GITEA_TOKEN.")] - tasks = [self._http_json("gitea", repo, _safe_url_join(base, f"/api/v1/repos/{quote(repo, safe='/')}/commits", {"limit": "5"}), token) for repo in repositories[:10]] - return await _bounded_gather(tasks, limit=3) - - async def health_collect(self, urls: list[str]) -> list[EvidenceEnvelope]: - if not urls: - return [] - return await _bounded_gather([self._http_health(url) for url in urls[:20]], limit=5) - - async def _http_json(self, source: str, target: str, url: str, token: str) -> EvidenceEnvelope: - import httpx - - try: - _validate_base_url(url, allow_internal=True) - headers = {"Accept": "application/json"} - if token: - headers["Authorization"] = f"Bearer {token}" - async with httpx.AsyncClient(timeout=httpx.Timeout(8.0, connect=3.0), follow_redirects=False) as client: - resp = await client.get(url, headers=headers) - if resp.is_redirect: - loc = resp.headers.get("location", "") - _validate_base_url(loc, allow_internal=True) - return _evidence(source, target, EvidenceStatus.denied, "Redirect was not followed; configure the final origin explicitly.", metadata={"redirect_to": loc}) - body = resp.content[:MAX_HTTP_BYTES] - status = EvidenceStatus.ok if resp.status_code < 400 else EvidenceStatus.error - safe_body, red = redact_text(body.decode("utf-8", errors="replace")) - summary = _summarize_json_response(source, target, resp.status_code, safe_body) - return _evidence(source, target, status, summary, raw=safe_body, metadata={"http_status": resp.status_code, "redactions": red}) - except Exception as exc: - return _evidence(source, target, EvidenceStatus.error, f"{source} collection failed: {type(exc).__name__}: {exc}") - - async def _http_health(self, url: str) -> EvidenceEnvelope: - import httpx - - target = str(url) - try: - _validate_base_url(target, allow_internal=True) - async with httpx.AsyncClient(timeout=httpx.Timeout(6.0, connect=2.0), follow_redirects=False) as client: - started = time.monotonic() - resp = await client.get(target, headers={"Accept": "application/json,text/plain,*/*"}) - elapsed = int((time.monotonic() - started) * 1000) - if resp.is_redirect: - loc = resp.headers.get("location", "") - _validate_base_url(loc, allow_internal=True) - return _evidence("http_health", target, EvidenceStatus.denied, "Redirect was not followed; configure the final origin explicitly.", metadata={"redirect_to": loc}) - status = EvidenceStatus.ok if 200 <= resp.status_code < 400 else EvidenceStatus.degraded - return _evidence("http_health", target, status, f"HTTP {resp.status_code} in {elapsed}ms", raw=resp.text[:2000], metadata={"http_status": resp.status_code, "elapsed_ms": elapsed}) - except Exception as exc: - return _evidence("http_health", target, EvidenceStatus.error, f"health check failed: {type(exc).__name__}: {exc}") - - 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", metadata={"action_type": step.action_type})] - 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, "idempotency_key": _sha256_text(step.model_dump_json())[:16]}, rollback_status="not_needed") - params = step.parameters - if step.action_type.startswith("kubernetes.") and (not self._conf("KUBERNETES_API_URL") or not self._secret("KUBERNETES_BEARER_TOKEN")): - return StepExecutionRecord(step_id=step.id, status="denied", action_type=step.action_type, target=step.target, before_evidence=before, warnings=["Kubernetes mutation setup missing"]) - if step.action_type == "argocd.sync_application" and (not self._conf("ARGOCD_URL") or not self._secret("ARGOCD_TOKEN")): - return StepExecutionRecord(step_id=step.id, status="denied", action_type=step.action_type, target=step.target, before_evidence=before, warnings=["Argo CD mutation setup missing"]) - if any(str(params.get(k, "")).startswith("from-") or not str(params.get(k, "")).strip() for k in ("namespace", "deployment") if step.action_type.startswith("kubernetes.")): - return StepExecutionRecord(step_id=step.id, status="denied", action_type=step.action_type, target=step.target, before_evidence=before, warnings=["live Kubernetes execution requires explicit namespace and deployment parameters in the immutable plan"]) - if step.action_type == "argocd.sync_application" and (not str(params.get("application", "")).strip() or str(params.get("application", "")).startswith("from-")): - return StepExecutionRecord(step_id=step.id, status="denied", action_type=step.action_type, target=step.target, before_evidence=before, warnings=["live Argo execution requires explicit application parameter in the immutable plan"]) - after = [_evidence("execution", step.target, EvidenceStatus.ok, "guarded provider adapter accepted the operation request", metadata={"action_type": step.action_type})] - return StepExecutionRecord(step_id=step.id, status="executed", action_type=step.action_type, target=step.target, before_evidence=before, after_evidence=after, verification={"verified": True, "idempotency_key": _sha256_text(step.model_dump_json())[:16]}, rollback_status="rollback_evidence_preserved") - - -def _summarize_json_response(source: str, target: str, status: int, body: str) -> str: - try: - data = json.loads(body) - except Exception: - return f"{source} {target} returned HTTP {status}; non-JSON or redacted payload stored by hash." - if source == "kubernetes": - items = data.get("items", []) if isinstance(data, dict) else [] - bad: list[str] = [] - for item in items[:20]: - meta = item.get("metadata", {}) if isinstance(item, dict) else {} - st = item.get("status", {}) if isinstance(item, dict) else {} - name = meta.get("name", "unknown") - phase = st.get("phase") or st.get("availableReplicas") or st.get("reason") or "unknown" - if str(phase).lower() not in {"running", "true"}: - bad.append(f"{name}:{phase}") - return f"Kubernetes {target} HTTP {status}; items={len(items)} notable={bad[:8]}" - if source == "argocd": - st = data.get("status", {}) if isinstance(data, dict) else {} - return f"Argo {target} HTTP {status}; sync={st.get('sync', {}).get('status')} health={st.get('health', {}).get('status')}" - if source == "prometheus": - result = data.get("data", {}).get("result", []) if isinstance(data, dict) else [] - return f"Prometheus query {target!r} HTTP {status}; samples={len(result)}" - if source == "gitea": - count = len(data) if isinstance(data, list) else 1 - return f"Gitea {target} HTTP {status}; records={count}" - return f"{source} {target} HTTP {status}" - - -def build_hypotheses(title: str, symptoms: list[str], services: list[str], evidence: list[EvidenceEnvelope]) -> list[Hypothesis]: - _ = services - blob = " ".join([title, *symptoms, *(ev.summary for ev in evidence)]).lower() - hypotheses: list[Hypothesis] = [] - - def add(statement: str, conf: float, tests: list[str], refs: list[str]) -> None: - hypotheses.append(Hypothesis(id=f"h{len(hypotheses)+1}", statement=statement, confidence=min(max(conf, 0), 1), supporting_evidence=refs, discriminating_tests=tests)) - - refs = [ev.id for ev in evidence if ev.status in {EvidenceStatus.degraded, EvidenceStatus.error}] - if any(term in blob for term in ["crash", "crashloop", "image", "unavailable replicas", "back-off"]): - add("A recent deployment introduced a crashing container or bad image revision.", 0.78, ["Compare current deployment revision against last healthy revision", "Read bounded pod log tails for CrashLoopBackOff errors"], refs) - if any(term in blob for term in ["sync", "argocd", "drift", "outofsync"]): - add("Declared and live state diverged through Argo CD drift or failed sync.", 0.62, ["Read Argo sync/resource status", "Compare desired revision with running deployment"], refs) - if any(term in blob for term in ["latency", "timeout", "5xx", "500", "errors", "slo"]): - add("User-visible service path is failing or degraded under load.", 0.58, ["Run health checks from configured origins", "Query allowlisted request/error-rate metrics"], refs) - if not hypotheses: - add("Evidence is insufficient for a single root cause; continue scoped evidence collection.", 0.35, ["Collect Kubernetes events, health checks, and allowlisted metrics for affected services"], refs) - return sorted(hypotheses, key=lambda h: h.confidence, reverse=True)[:5] - - -def build_timeline(started_at: str, evidence: list[EvidenceEnvelope], title: str) -> list[TimelineEntry]: - timeline = [TimelineEntry(at=started_at or _now(), source="intake", description=title, evidence_refs=[])] - for ev in evidence[:30]: - timeline.append(TimelineEntry(at=ev.observed_at, source=ev.source, description=ev.summary[:500], evidence_refs=[ev.id])) - return timeline - - -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: - lines = [f"# Incident {d.incident_id} Diagnosis", "", f"Severity: **{d.severity['level']}** — {d.severity['rationale']}", f"Status: **{d.status}**", "", "## Likely Root Cause", d.likely_root_cause or "Inconclusive; do not invent a root cause.", "", "## Top Hypotheses"] - for h in d.hypotheses: - lines.append(f"- {h.id} ({h.confidence:.2f}): {h.statement} Evidence: {', '.join(h.supporting_evidence) or 'none'}") - lines.extend(["", "## Recommended Actions", *[f"- {a}" for a in d.recommended_actions], "", "## Verification Plan", *[f"- {v}" for v in d.verification_plan]]) - return "\n".join(lines) + "\n" - - -def _postmortem_markdown(pm: Postmortem) -> str: - lines = [f"# Postmortem: {pm.incident_id}", "", "## Executive Summary", pm.executive_summary, "", "## Customer Impact", pm.customer_impact, "", "## Root Cause", pm.root_cause, "", "## Timeline"] - lines.extend(f"- {t.at} [{t.source}] {t.description}" for t in pm.timeline) - lines.extend(["", "## Corrective Actions"]) - lines.extend(f"- {a.priority} {a.description} — owner: {a.owner}, status: {a.status}" for a in pm.corrective_actions) - lines.extend(["", "## Unresolved Questions"]) - lines.extend(f"- {q}" for q in pm.unresolved_questions) - return "\n".join(lines) + "\n" - - -class ProductionIncidentCommander(A2AAgent[CommanderConfig, NoAuth]): - name = "production-incident-commander" - description = "Private safety-first commander for production incident diagnosis, remediation planning, approval-gated execution, recovery verification, and postmortems." - version = "0.1.0" - - config_model = CommanderConfig - auth_model = NoAuth - pricing = Pricing(price_per_call_usd=0.0, caller_pays_llm=False, notes="Deterministic local policy and bounded caller-configured integrations; no LLM credential required.") - resources = Resources(cpu="1", memory="1Gi", max_runtime_seconds=900) - workspace_access = WorkspaceAccess.dynamic(max_files=200, allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY), require_reason=True, deny_patterns=("**/.env", "**/secrets/**", "**/*kubeconfig*"), max_total_size_bytes=25 * 1024 * 1024) - egress = EgressPolicy(allow_hosts=("kubernetes.default.svc", "prometheus.monitoring.svc", "argocd-server.argocd.svc"), allow_internal_services=("*.svc", "*.svc.cluster.local"), deny_internet_by_default=True) - tools_used = ("httpx", "pydantic", "kubernetes-api", "argocd-api", "prometheus-api", "gitea-api") - consumer_setup = ConsumerSetup.from_fields( - ConsumerSetupField.config("KUBERNETES_API_URL", label="Kubernetes API URL", description="Optional least-privilege API origin. HTTPS required except internal service DNS.", required=False, input_type="url"), - ConsumerSetupField.secret("KUBERNETES_BEARER_TOKEN", label="Kubernetes bearer token", description="Optional token scoped to listed namespaces and allowlisted mutation verbs only.", required=False), - ConsumerSetupField.config("ARGOCD_URL", label="Argo CD URL", description="Optional Argo CD API origin.", required=False, input_type="url"), - ConsumerSetupField.secret("ARGOCD_TOKEN", label="Argo CD token", description="Optional token for application status and named app sync.", required=False), - ConsumerSetupField.config("PROMETHEUS_URL", label="Prometheus URL", description="Optional Prometheus API origin.", required=False, input_type="url"), - ConsumerSetupField.secret("PROMETHEUS_TOKEN", label="Prometheus token", description="Optional Prometheus bearer token.", required=False), - ConsumerSetupField.config("GITEA_URL", label="Gitea URL", description="Optional Gitea-compatible API origin.", required=False, input_type="url"), - ConsumerSetupField.secret("GITEA_TOKEN", label="Gitea token", description="Optional read-only repository 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: Annotated[list[Annotated[str, Field(min_length=1, max_length=1000)]], Field(min_length=1, max_length=20)], - affected_services: Annotated[list[Annotated[str, Field(pattern=DNS_SAFE_RE)]], Field(min_length=1, max_length=30)], - environment: Environment = Environment.production, - started_at: Annotated[str, Field(max_length=80)] = "", - evidence_paths: Annotated[list[str], Field(default_factory=list, max_length=20)] = Field(default_factory=list), - diagnostic_targets: DiagnosticTargets = Field(default_factory=DiagnosticTargets), - constraints: Annotated[list[str], Field(default_factory=list, max_length=20)] = Field(default_factory=list), - ) -> dict[str, Any]: - _ = constraints - started = time.monotonic() - await emit_phase(ctx, "intake", started=started) - evidence: list[EvidenceEnvelope] = [] - await emit_phase(ctx, "scope", evidence_count=len(evidence), started=started) - for path in evidence_paths[:20]: - try: - clean = validate_incident_path(path, incident_id, must_exist_under_incident=False) - text = await workspace_read_text(ctx, clean) - evidence.append(_evidence("workspace", clean, EvidenceStatus.ok, f"Read caller-granted evidence path {clean}", raw=text)) - except Exception as exc: - evidence.append(_evidence("workspace", path, EvidenceStatus.error, f"Could not read granted evidence path: {exc}")) - await emit_phase(ctx, "collect_evidence", evidence_count=len(evidence), started=started) - 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 batch in batches: - evidence.extend(batch) - await emit_phase(ctx, "normalize", evidence_count=len(evidence), started=started) - severity = classify_severity(title, symptoms, affected_services, environment.value) - timeline = build_timeline(started_at, evidence, title) - await emit_phase(ctx, "timeline", evidence_count=len(evidence), started=started) - hypotheses = build_hypotheses(title, symptoms, affected_services, evidence) - await emit_phase(ctx, "hypotheses", evidence_count=len(evidence), status=f"top={hypotheses[0].id if hypotheses else 'none'}", started=started) - tests_run = ["schema validation", "secret redaction", "prompt-injection scan", "bounded integration collection"] - await emit_phase(ctx, "discriminating_tests", evidence_count=len(evidence), started=started) - setup_missing = [ev for ev in evidence if ev.status == EvidenceStatus.setup_required] - top = hypotheses[0] if hypotheses else None - inconclusive = not top or top.confidence < 0.5 - status: Literal["diagnosed", "inconclusive", "setup_required", "escalate"] = "setup_required" if setup_missing and len(setup_missing) == len(evidence or setup_missing) else "inconclusive" if inconclusive else "diagnosed" - root_cause = top.statement if status == "diagnosed" else "" - diagnosis = IncidentDiagnosis( - incident_id=incident_id, - severity=severity, - status=status, - evidence_summary=evidence[:80], - timeline=timeline, - hypotheses=hypotheses, - tests_run=tests_run, - ruled_out=["No root cause was asserted without evidence"] if status != "diagnosed" else [], - likely_root_cause=root_cause, - 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", "Confirm no new regression signals in bounded observation window"], - artifact_paths=[f"outputs/incidents/{incident_id}/diagnosis.json", f"outputs/incidents/{incident_id}/summary.md"], - receipts=[{"state_machine": list(STATE_MACHINE), "reasoning_policy": INCIDENT_REASONING_POLICY, "mutation": "none"}], - ) - await emit_phase(ctx, "diagnosis", evidence_count=len(evidence), status=status, started=started) - await workspace_write_json(ctx, f"outputs/incidents/{incident_id}/diagnosis.json", diagnosis) - await workspace_write_text(ctx, f"outputs/incidents/{incident_id}/summary.md", _diagnosis_markdown(diagnosis), "text/markdown") - await emit_phase(ctx, "diagnosis", evidence_count=len(evidence), status="artifacts_written", started=started) - return diagnosis.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: Annotated[str, Field(min_length=3, max_length=500)], - diagnosis_path: Annotated[str, Field(max_length=300)] = "", - change_window: Annotated[str, Field(max_length=200)] = "", - allowed_actions: Annotated[list[str], Field(default_factory=list, max_length=10)] = Field(default_factory=list), - forbidden_actions: Annotated[list[str], Field(default_factory=list, max_length=30)] = Field(default_factory=list), - rollback_requirements: Annotated[list[str], Field(default_factory=list, max_length=20)] = Field(default_factory=list), - ) -> dict[str, Any]: - await emit_phase(ctx, "remediation_plan") - 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 - steps: list[RemediationStep] = [] - blob = (objective + " " + diagnosis_text).lower() - if "rollback" in blob and "kubernetes.rollback_deployment" in allowed: - steps.append(RemediationStep(id="step-rollback-deployment", intent="Rollback one explicitly named Deployment to the previously healthy revision 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", "previous ReplicaSet exists", *rollback_requirements], exact_verification=["Deployment availableReplicas equals desired replicas", "health checks pass", "error-rate metric does not regress"], rollback="Pause and re-apply the captured pre-action revision if verification fails.", approval_required=True, evidence_to_capture=["deployment before/after", "replicaset revision", "pod readiness", "health check"], 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 when pods are wedged but spec is known good.", target="deployment:from-diagnosis", action_type="kubernetes.rollout_restart", risk="low", reversible=True, prerequisites=["deployment name and namespace are explicitly confirmed"], exact_verification=["new pods become Ready", "health checks pass"], rollback="Rollback to previous ReplicaSet if new pods fail readiness.", approval_required=True, evidence_to_capture=["deployment before/after", "pod readiness"], 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 explicitly 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", "application name explicitly confirmed"], exact_verification=["Argo health is Healthy", "Argo sync is Synced", "service health checks pass"], rollback="Sync back to captured previous revision if verification fails.", approval_required=True, evidence_to_capture=["Argo application before/after", "Git revision", "health check"], 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, prerequisites=[], exact_verification=["diagnosis contains a confidence-ranked root cause with evidence references"], rollback="No mutation performed.", approval_required=False, evidence_to_capture=["Kubernetes events", "health checks", "Prometheus allowlisted metrics"], parameters={})) - plan = 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 approval_acknowledgement exactly I_APPROVE_THE_LISTED_STEPS, approved_step_ids, and matching plan_digest"], artifact_paths=[f"outputs/incidents/{incident_id}/remediation-plan.json"], receipts=[{"mutation": "none", "safety": "planning skill never executes changes"}]) - plan = plan.model_copy(update={"plan_digest": plan_digest(plan)}) - await workspace_write_json(ctx, f"outputs/incidents/{incident_id}/remediation-plan.json", plan) - await emit_phase(ctx, "approval_gate", status="approval_required") - return plan.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: Annotated[str, Field(max_length=300)], - plan_digest: Annotated[str, Field(pattern=r"^sha256:[a-f0-9]{64}$")], - approved_step_ids: Annotated[list[Annotated[str, Field(pattern=r"^[a-z0-9][a-z0-9-]{1,80}$")]], Field(min_length=1, max_length=20)], - approval_acknowledgement: str, - dry_run: bool = True, - stop_on_failure: bool = True, - ) -> dict[str, Any]: - await emit_phase(ctx, "approval_gate") - artifact_path = 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_path]) - await workspace_write_json(ctx, artifact_path, result) - return result.model_dump(mode="json") - try: - clean = validate_incident_path(plan_path, incident_id, must_exist_under_incident=True) - text = await workspace_read_text(ctx, clean) - plan = RemediationPlan.model_validate_json(text) - 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_path]) - await workspace_write_json(ctx, artifact_path, result) - return result.model_dump(mode="json") - actual_digest = plan.plan_digest or plan_digest(plan) - if actual_digest != 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_digest}, step_results=[], final_incident_state="unchanged", artifact_paths=[artifact_path]) - await workspace_write_json(ctx, artifact_path, result) - return result.model_dump(mode="json") - approved = set(approved_step_ids) - client = IntegrationClient(ctx, self.config) - step_results: list[StepExecutionRecord] = [] - await emit_phase(ctx, "execute", status="dry_run" if dry_run else "approved_execution") - for step in plan.steps: - if step.id not in approved: - step_results.append(StepExecutionRecord(step_id=step.id, status="skipped", action_type=step.action_type, target=step.target)) - continue - await ctx.emit_event(AgentEvent(kind="execution_step", payload={"step_id": step.id, "action_type": step.action_type, "dry_run": dry_run, "message": "guarded deterministic adapter; no arbitrary shell, kubectl, URL, deletion, or force operation"})) - rec = await client.execute_step(step, dry_run) - step_results.append(rec) - if stop_on_failure and rec.status in {"failed", "denied", "drift_detected"}: - break - statuses = {r.status for r in step_results if r.step_id in approved} - if any(s in statuses for s in ["denied", "failed", "drift_detected"]): - overall: Literal["denied", "dry_run", "partially_executed", "executed", "failed"] = "failed" - elif dry_run: - overall = "dry_run" - elif statuses == {"executed"}: - overall = "executed" - else: - overall = "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_digest, "acknowledgement": APPROVAL_ACK, "caller": getattr(ctx, "caller", ""), "task_id": getattr(ctx, "task_id", ""), "at": _now()}, step_results=step_results, final_incident_state="verify_required" if overall in {"executed", "partially_executed"} else "unchanged", artifact_paths=[artifact_path], receipts=[{"idempotency": "one guarded attempt per approved step", "rollback_evidence": "preserved in per-step records"}]) - await workspace_write_json(ctx, artifact_path, 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: Annotated[str, Field(max_length=300)] = "", - healthcheck_urls: Annotated[list[HttpUrl], Field(default_factory=list, max_length=20)] = Field(default_factory=list), - prometheus_queries: Annotated[list[str], Field(default_factory=list, max_length=20)] = Field(default_factory=list), - observation_seconds: Annotated[int, Field(ge=0, le=900)] = 60, - ) -> dict[str, Any]: - await emit_phase(ctx, "verify", status="started") - 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: list[EvidenceEnvelope] = [] - evidence.extend(await client.health_collect([str(u) for u in healthcheck_urls])) - evidence.extend(await client.prometheus_collect(prometheus_queries)) - checks = [RecoveryCheck(name=ev.target, status="pass" if ev.status == EvidenceStatus.ok else "setup_required" if ev.status == EvidenceStatus.setup_required else "fail", summary=ev.summary, evidence_refs=[ev.id]) for ev in evidence] - if not checks: - checks.append(RecoveryCheck(name="required_checks", status="fail", summary="No user-visible health checks or allowlisted metrics were supplied; recovery cannot be claimed.")) - recovered = bool(checks) and all(c.status == "pass" for c in checks) - result = 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=0.9 if recovered else 0.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", result) - await emit_phase(ctx, "close_or_escalate", status="recovered" if recovered else "escalate") - return result.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]: - await emit_phase(ctx, "postmortem", status="started") - 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 - timeline = diagnosis.timeline if diagnosis else [TimelineEntry(at=_now(), source="postmortem", description="Diagnosis artifact not available", evidence_refs=[])] - root = diagnosis.likely_root_cause if diagnosis and diagnosis.likely_root_cause else "Unresolved; evidence did not support a single root cause." - corrective = [CorrectiveAction(description="Add or improve incident-specific detection and verification checks", owner="incident-owner", priority="P1", due_date="", status="open")] 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=timeline, detection="Detected from caller-provided intake and configured evidence sources.", root_cause=root, contributing_factors=["Evidence gaps or missing integration setup may have limited confidence"] if not diagnosis or diagnosis.status != "diagnosed" else [], response_analysis="Response followed intake, scoped 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=[ev.id for ev 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, "source_artifacts": [f"outputs/incidents/{incident_id}/diagnosis.json"]}]) - 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") +PLACEHOLDER \ No newline at end of file