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

a2a-source-edit: write agent.py

Browse Source
This commit is contained in:
a2a-cloud
2026-07-13 03:15:12 +00:00
parent 414a8d0f78
commit ef70865747
1 changed files with 81 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files

206
agent.py
View File

@@ -1,4 +1,3 @@
"""Production Incident Commander A2A agent."""
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
@@ -6,7 +5,6 @@ import hashlib
import ipaddress import ipaddress
import json import json
import re import re
import time
from datetime import UTC, datetime from datetime import UTC, datetime
from enum import Enum from enum import Enum
from typing import Annotated, Any, Literal 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 from pydantic import BaseModel, ConfigDict, Field, HttpUrl, ValidationError, field_validator
import a2a_pack as a2a import a2a_pack as a2a
from a2a_pack import ( from a2a_pack import A2AAgent, ConsumerSetup, ConsumerSetupField, EgressPolicy, NoAuth, Pricing, Resources, RunContext, WorkspaceAccess, WorkspaceMode
A2AAgent,
ConsumerSetup,
ConsumerSetupField,
EgressPolicy,
NoAuth,
Pricing,
Resources,
RunContext,
WorkspaceAccess,
WorkspaceMode,
)
from a2a_pack.context import AgentEvent, ConsumerSetupMissing from a2a_pack.context import AgentEvent, ConsumerSetupMissing
INCIDENT_ID_RE = r"^[a-z0-9][a-z0-9-]{2,63}$" 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])?$" DNS_SAFE_RE = r"^[a-z0-9]([-a-z0-9]*[a-z0-9])?$"
APPROVAL_ACK = "I_APPROVE_THE_LISTED_STEPS" APPROVAL_ACK = "I_APPROVE_THE_LISTED_STEPS"
MAX_EVIDENCE_CHARS = 16_000
ALLOWED_EXECUTION_ACTIONS = {"kubernetes.rollout_restart", "kubernetes.rollback_deployment", "argocd.sync_application"} 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") MAX_EVIDENCE_CHARS = 16000
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." 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] = { 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": []}}}
"type": "object", 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": []}}}
"additionalProperties": False, 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": []}}}
"default": {"kubernetes_namespaces": [], "argo_applications": [], "prometheus_queries": [], "healthcheck_urls": [], "gitea_repositories": []}, 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}}}
"properties": { 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}}}
"kubernetes_namespaces": {"type": "array", "items": {"type": "string", "pattern": DNS_SAFE_RE}, "maxItems": 30, "default": []}, 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}}}
"argo_applications": {"type": "array", "items": {"type": "string", "pattern": DNS_SAFE_RE}, "maxItems": 30, "default": []},
"prometheus_queries": {"type": "array", "items": {"type": "string", "minLength": 1, "maxLength": 500}, "maxItems": 20, "default": []},
"healthcheck_urls": {"type": "array", "items": {"type": "string", "format": "uri", "maxLength": 1000}, "maxItems": 20, "default": []},
"gitea_repositories": {"type": "array", "items": {"type": "string", "pattern": r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$"}, "maxItems": 20, "default": []},
},
}
DIAGNOSE_INPUT_SCHEMA: dict[str, Any] = {"type": "object", "additionalProperties": False, "required": ["incident_id", "title", "symptoms", "affected_services"], "properties": {"incident_id": {"type": "string", "pattern": INCIDENT_ID_RE}, "title": {"type": "string", "minLength": 3, "maxLength": 200}, "symptoms": {"type": "array", "minItems": 1, "maxItems": 20, "items": {"type": "string", "minLength": 1, "maxLength": 1000}}, "affected_services": {"type": "array", "minItems": 1, "maxItems": 30, "items": {"type": "string", "pattern": DNS_SAFE_RE}}, "environment": {"type": "string", "enum": ["production", "staging", "development"], "default": "production"}, "started_at": {"type": "string", "maxLength": 80, "default": ""}, "evidence_paths": {"type": "array", "items": {"type": "string", "maxLength": 300}, "maxItems": 20, "default": []}, "diagnostic_targets": STRICT_DIAGNOSTIC_TARGETS_SCHEMA, "constraints": {"type": "array", "items": {"type": "string", "maxLength": 500}, "maxItems": 20, "default": []}}}
PLAN_INPUT_SCHEMA: dict[str, Any] = {"type": "object", "additionalProperties": False, "required": ["incident_id", "objective"], "properties": {"incident_id": {"type": "string", "pattern": INCIDENT_ID_RE}, "diagnosis_path": {"type": "string", "maxLength": 300, "default": ""}, "objective": {"type": "string", "minLength": 3, "maxLength": 500}, "change_window": {"type": "string", "maxLength": 200, "default": ""}, "allowed_actions": {"type": "array", "items": {"type": "string", "enum": sorted(ALLOWED_EXECUTION_ACTIONS)}, "maxItems": 10, "default": []}, "forbidden_actions": {"type": "array", "items": {"type": "string", "maxLength": 200}, "maxItems": 30, "default": []}, "rollback_requirements": {"type": "array", "items": {"type": "string", "maxLength": 300}, "maxItems": 20, "default": []}}}
EXECUTE_INPUT_SCHEMA: dict[str, Any] = {"type": "object", "additionalProperties": False, "required": ["incident_id", "plan_path", "plan_digest", "approved_step_ids", "approval_acknowledgement"], "properties": {"incident_id": {"type": "string", "pattern": INCIDENT_ID_RE}, "plan_path": {"type": "string", "maxLength": 300}, "plan_digest": {"type": "string", "pattern": r"^sha256:[a-f0-9]{64}$"}, "approved_step_ids": {"type": "array", "minItems": 1, "maxItems": 20, "items": {"type": "string", "pattern": r"^[a-z0-9][a-z0-9-]{1,80}$"}}, "approval_acknowledgement": {"type": "string", "const": APPROVAL_ACK}, "dry_run": {"type": "boolean", "default": True}, "stop_on_failure": {"type": "boolean", "default": True}}}
VERIFY_INPUT_SCHEMA: dict[str, Any] = {"type": "object", "additionalProperties": False, "required": ["incident_id"], "properties": {"incident_id": {"type": "string", "pattern": INCIDENT_ID_RE}, "verification_plan_path": {"type": "string", "maxLength": 300, "default": ""}, "healthcheck_urls": {"type": "array", "items": {"type": "string", "format": "uri", "maxLength": 1000}, "maxItems": 20, "default": []}, "prometheus_queries": {"type": "array", "items": {"type": "string", "minLength": 1, "maxLength": 500}, "maxItems": 20, "default": []}, "observation_seconds": {"type": "integer", "minimum": 0, "maximum": 900, "default": 60}}}
POSTMORTEM_INPUT_SCHEMA: dict[str, Any] = {"type": "object", "additionalProperties": False, "required": ["incident_id"], "properties": {"incident_id": {"type": "string", "pattern": INCIDENT_ID_RE}, "include_customer_summary": {"type": "boolean", "default": True}, "include_action_items": {"type": "boolean", "default": True}}}
class CommanderConfig(BaseModel): class CommanderConfig(BaseModel):
model_config = ConfigDict(extra="forbid") 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]))"]) 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): class Environment(str, Enum): production = "production"; staging = "staging"; development = "development"
production = "production"; staging = "staging"; development = "development" class SeverityLevel(str, Enum): SEV1 = "SEV1"; SEV2 = "SEV2"; SEV3 = "SEV3"; SEV4 = "SEV4"
class SeverityLevel(str, Enum): class EvidenceStatus(str, Enum): ok = "ok"; degraded = "degraded"; error = "error"; setup_required = "setup_required"; skipped = "skipped"; denied = "denied"
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): class EvidenceEnvelope(BaseModel):
model_config = ConfigDict(extra="forbid") model_config = ConfigDict(extra="forbid")
id: str; source: str; target: str; observed_at: str; status: EvidenceStatus; summary: str 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)
redactions: list[str] = Field(default_factory=list); raw_ref: str = ""; hash: str = ""; metadata: dict[str, Any] = Field(default_factory=dict)
class Hypothesis(BaseModel): class Hypothesis(BaseModel):
model_config = ConfigDict(extra="forbid") 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) 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): class IncidentDiagnosis(BaseModel):
model_config = ConfigDict(extra="forbid") 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) 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): class DiagnosticTargets(BaseModel):
model_config = ConfigDict(extra="forbid") model_config = ConfigDict(extra="forbid")
kubernetes_namespaces: list[str] = Field(default_factory=list, max_length=30) kubernetes_namespaces: list[str] = Field(default_factory=list)
argo_applications: list[str] = Field(default_factory=list, max_length=30) argo_applications: list[str] = Field(default_factory=list)
prometheus_queries: list[str] = Field(default_factory=list, max_length=20) prometheus_queries: list[str] = Field(default_factory=list)
healthcheck_urls: list[HttpUrl] = Field(default_factory=list, max_length=20) healthcheck_urls: list[HttpUrl] = Field(default_factory=list)
gitea_repositories: list[str] = Field(default_factory=list, max_length=20) gitea_repositories: list[str] = Field(default_factory=list)
@field_validator("kubernetes_namespaces", "argo_applications") @field_validator("kubernetes_namespaces", "argo_applications")
@classmethod @classmethod
def _names_are_safe(cls, values: list[str]) -> list[str]: def _safe_names(cls, values: list[str]) -> list[str]:
for value in values: for value in values:
if not re.fullmatch(DNS_SAFE_RE, value): raise ValueError(f"not DNS-safe: {value}") if not re.fullmatch(DNS_SAFE_RE, value): raise ValueError(f"not DNS-safe: {value}")
return values return values
@field_validator("gitea_repositories") @field_validator("gitea_repositories")
@classmethod @classmethod
def _repos_are_safe(cls, values: list[str]) -> list[str]: def _safe_repos(cls, values: list[str]) -> list[str]:
for value in values: 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}") 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 return values
def _now() -> str: return datetime.now(UTC).isoformat(timespec="seconds") 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 _stable_json(data: Any) -> str: return json.dumps(data, sort_keys=True, separators=(",", ":"), ensure_ascii=False, default=str)
def plan_digest(plan: RemediationPlan) -> str: def plan_digest(plan: RemediationPlan) -> str:
payload = plan.model_dump(mode="json"); payload["plan_digest"] = ""; payload["receipts"] = [] payload = plan.model_dump(mode="json"); payload["plan_digest"] = ""; payload["receipts"] = []
return "sha256:" + _sha256_text(_stable_json(payload)) 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)") 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]]: 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] = [] text = value if isinstance(value, str) else json.dumps(value, default=str, ensure_ascii=False); marks: list[str] = []
for idx, (pattern, repl) in enumerate(SECRET_PATTERNS): for i, (pattern, repl) in enumerate(SECRET_PATTERNS):
text, count = pattern.subn(repl, text) text, count = pattern.subn(repl, text)
if count: redactions.append(f"secret_pattern_{idx}:{count}") if count: marks.append(f"secret_pattern_{i}:{count}")
if len(text) > MAX_EVIDENCE_CHARS: text = text[:MAX_EVIDENCE_CHARS] + "\n[TRUNCATED]"; redactions.append("truncated") if len(text) > MAX_EVIDENCE_CHARS: text = text[:MAX_EVIDENCE_CHARS] + "\n[TRUNCATED]"; marks.append("truncated")
return text, redactions return text, marks
def _evidence(source: str, target: str, status: EvidenceStatus, summary: str, raw: Any = "", metadata: dict[str, Any] | None = None) -> EvidenceEnvelope: 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 "" 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_summary) or (raw_text and PROMPT_INJECTION_PATTERNS.search(raw_text)): redactions.append("untrusted_prompt_injection_ignored") 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_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 {}) 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 _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: def _host_is_ip_blocked(host: str) -> bool:
if host.lower() in {"localhost", "metadata.google.internal"}: return True 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("/") return str(url).rstrip("/")
def validate_incident_path(path: str, incident_id: str, *, must_exist_under_incident: bool = False) -> str: 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}/" 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 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}") if clean.startswith("outputs/") and not clean.startswith(prefix): raise ValueError(f"incident artifacts must be under {prefix}")
return clean return clean
def _incident_id_from_path(path: str) -> str | None: def _incident_id_from_path(path: str) -> str | None:
m = re.search(r"outputs/incidents/([^/]+)/", path); return m.group(1) if m else None match = re.search(r"outputs/incidents/([^/]+)/", path); return match.group(1) if match else None
async def emit_phase(ctx: RunContext[NoAuth], phase: str, *, evidence_count: int = 0, status: str = "started", started: float | None = None) -> None: 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}))
payload: dict[str, Any] = {"phase": phase, "status": status, "evidence_count": evidence_count}
if started is not None: payload["elapsed_ms"] = int((time.monotonic() - started) * 1000)
await ctx.emit_event(AgentEvent(kind="incident_phase", payload=payload))
async def workspace_write_text(ctx: RunContext[NoAuth], path: str, text: str, mime: str = "text/plain") -> None: 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() clean = validate_incident_path(path, _incident_id_from_path(path) or "unknown", must_exist_under_incident=True); data = text.encode()
try: try: ctx.workspace.write_bytes(clean, data)
ctx.workspace.write_bytes(clean, data) # supported by deployed/local workspace implementations
except Exception: except Exception:
ref = await ctx.write_artifact(clean.replace("/", "__"), data, mime); await ctx.emit_artifact(ref) 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: 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")
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 = 256000) -> str: return ctx.workspace.read_bytes(path.replace("\\", "/").strip("/"))[:max_bytes].decode(errors="replace")
async def workspace_read_text(ctx: RunContext[NoAuth], path: str, max_bytes: int = 256_000) -> str:
clean = path.replace("\\", "/").strip("/")
return ctx.workspace.read_bytes(clean)[:max_bytes].decode("utf-8", errors="replace")
def classify_severity(title: str, symptoms: list[str], affected_services: list[str], environment: str) -> dict[str, str]: def classify_severity(title: str, symptoms: list[str], affected_services: list[str], environment: str) -> dict[str, str]:
text = " ".join([title, *symptoms]).lower(); rationale = [] 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") 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") 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"} 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]: 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 = [] 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]): out.append(Hypothesis(id=f"h{len(out)+1}", statement=s, confidence=c, supporting_evidence=refs, discriminating_tests=tests)) 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", "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 ["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/resource status"]) 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 ["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 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 events, health checks, and 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) return sorted(out, key=lambda h: h.confidence, reverse=True)
def build_timeline(started_at: str, evidence: list[EvidenceEnvelope], title: str) -> list[TimelineEntry]: 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]]
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 recommend_actions(hypotheses: list[Hypothesis]) -> list[str]: 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"
actions = ["Preserve evidence dossier before any mutation", "Run verification against user-visible health checks and metrics"] def _postmortem_markdown(pm: Postmortem) -> str: return f"# Postmortem: {pm.incident_id}\n\n{pm.executive_summary}\n"
for h in hypotheses:
if "crashing container" in h.statement: actions.append("Prepare a rollback plan for the named Deployment after operator approval")
if "Argo" in h.statement: actions.append("Prepare an Argo sync plan after operator approval if desired state is confirmed safe")
return list(dict.fromkeys(actions))
def _diagnosis_markdown(d: IncidentDiagnosis) -> str: return f"# Incident {d.incident_id} Diagnosis\n\nStatus: {d.status}\n\nLikely root cause: {d.likely_root_cause or 'Inconclusive'}\n"
def _postmortem_markdown(pm: Postmortem) -> str: return f"# Postmortem: {pm.incident_id}\n\n## Executive Summary\n{pm.executive_summary}\n\n## Root Cause\n{pm.root_cause}\n"
class IntegrationClient: class IntegrationClient:
def __init__(self, ctx: RunContext[NoAuth], config: CommanderConfig): self.ctx = ctx; self.config = config def __init__(self, ctx: RunContext[NoAuth], config: CommanderConfig): self.ctx = ctx; self.config = config
def _conf(self, name: str) -> str: def _conf(self, name: str) -> str:
value = self.ctx.consumer_config(name, "") value = self.ctx.consumer_config(name, ""); return _validate_base_url(str(value), allow_internal=True) if value else ""
return _validate_base_url(str(value), allow_internal=True) if value else ""
def _secret(self, name: str) -> str: def _secret(self, name: str) -> str:
try: return self.ctx.consumer_secret(name).strip() try: return self.ctx.consumer_secret(name).strip()
except ConsumerSetupMissing: return "" 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 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, 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 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, queries: list[str]) -> list[EvidenceEnvelope]: async def prometheus_collect(self, qs: list[str]) -> list[EvidenceEnvelope]:
if not queries: return [] if not qs: return []
if not self._conf("PROMETHEUS_URL"): return [_evidence("prometheus", ",".join(queries), EvidenceStatus.setup_required, "Prometheus setup missing; configure PROMETHEUS_URL.")] 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.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]] 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, 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 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", str(u), EvidenceStatus.error, "health check not attempted in deterministic sandbox without configured egress") for u in urls] 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: 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 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"]) 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")] 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") 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]): class ProductionIncidentCommander(A2AAgent[CommanderConfig, NoAuth]):
name = "production-incident-commander"; description = "Private safety-first commander for production incident diagnosis, remediation planning, approval-gated execution, recovery verification, and postmortems."; version = "0.1.0" name = "production-incident-commander"; description = "Private safety-first commander for production incident diagnosis, remediation planning, approval-gated execution, recovery verification, and postmortems."; version = "0.1.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.") 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) 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) 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") 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)) 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*")) @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]: 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]:
_ = constraints; started = time.monotonic(); await emit_phase(ctx, "intake", started=started); evidence: list[EvidenceEnvelope] = [] await emit_phase(ctx, "intake"); evidence: list[EvidenceEnvelope] = []
for path in evidence_paths[:20]: 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)))) 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}")) 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])) client = IntegrationClient(ctx, self.config)
for b in batches: evidence.extend(b) 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] 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" 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") 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/**",)) @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]: 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 = "" blob = objective.lower(); allowed = set(allowed_actions or sorted(ALLOWED_EXECUTION_ACTIONS)) & ALLOWED_EXECUTION_ACTIONS; steps: list[RemediationStep] = []
if diagnosis_path: diagnosis_text = await workspace_read_text(ctx, validate_incident_path(diagnosis_path, incident_id, must_exist_under_incident=True)) if diagnosis_path: blob += " " + (await workspace_read_text(ctx, validate_incident_path(diagnosis_path, incident_id, must_exist_under_incident=True))).lower()
allowed = set(allowed_actions or sorted(ALLOWED_EXECUTION_ACTIONS)) & ALLOWED_EXECUTION_ACTIONS; blob = (objective + " " + diagnosis_text).lower(); steps: list[RemediationStep] = []
if "rollback" in blob and "kubernetes.rollback_deployment" in allowed: steps.append(RemediationStep(id="step-rollback-deployment", intent="Rollback one explicitly named Deployment after confirming drift has not occurred.", target="deployment:from-diagnosis", action_type="kubernetes.rollback_deployment", risk="medium", reversible=True, prerequisites=["operator confirms deployment name and namespace", *rollback_requirements], exact_verification=["health checks pass"], rollback="Re-apply captured pre-action revision if verification fails.", approval_required=True, evidence_to_capture=["deployment before/after"], parameters={"namespace": "from-diagnosis", "deployment": "from-diagnosis"})) if "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 "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)) 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=["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)}) 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") 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/**",)) @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]: 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" 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]) 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: r = denied("approval acknowledgement mismatch"); 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))) 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) 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") 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); client = IntegrationClient(ctx, self.config); results: list[StepExecutionRecord] = [] approved = set(approved_step_ids); results: list[StepExecutionRecord] = []; client = IntegrationClient(ctx, self.config)
for step in plan.steps: 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 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) rec = await client.execute_step(step, dry_run); results.append(rec)
if stop_on_failure and rec.status in {"failed", "denied", "drift_detected"}: break 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" 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"}]) 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"}]))
await workspace_write_json(ctx, artifact, out); return out.model_dump(mode="json")
@a2a.tool(description="Verify recovery using user-visible checks and metrics. Never marks recovered unless required checks pass.", input_schema=VERIFY_INPUT_SCHEMA, timeout_seconds=900, cost_class="incident-verification", grant_mode="read_write_overlay", grant_write_prefixes=("outputs/incidents/",), grant_allow_patterns=("outputs/incidents/**",)) @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]: 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 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)) 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)) 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 ev] or [RecoveryCheck(name="required_checks", status="fail", summary="No user-visible health checks or allowlisted metrics were supplied; recovery cannot be claimed.")] 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(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"}]) 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") 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/**",)) @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 diagnosis: IncidentDiagnosis | None = None
try: diagnosis = IncidentDiagnosis.model_validate_json(await workspace_read_text(ctx, f"outputs/incidents/{incident_id}/diagnosis.json")) try: diagnosis = IncidentDiagnosis.model_validate_json(await workspace_read_text(ctx, f"outputs/incidents/{incident_id}/diagnosis.json"))
except Exception: pass 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=[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}])
pm = Postmortem(incident_id=incident_id, executive_summary=diagnosis.likely_root_cause if diagnosis and diagnosis.likely_root_cause else "Incident investigated with bounded evidence; final root cause remains unresolved.", customer_impact="Customer-facing impact should be filled from support/business telemetry." if include_customer_summary else "Customer summary omitted by caller.", timeline=diagnosis.timeline if diagnosis else [TimelineEntry(at=_now(), source="postmortem", description="Diagnosis artifact not available")], detection="Detected from caller-provided intake and configured evidence sources.", root_cause=diagnosis.likely_root_cause if diagnosis and diagnosis.likely_root_cause else "Unresolved; evidence did not support a single root cause.", contributing_factors=[] if diagnosis and diagnosis.status == "diagnosed" else ["Evidence gaps or missing integration setup may have limited confidence"], response_analysis="Response followed evidence collection, hypothesis ranking, approval-gated remediation, and recovery verification.", what_went_well=["Evidence was redacted and written to durable artifacts", "Mutations were separated from diagnosis and planning"], what_went_poorly=["Fill with operator review after incident debrief"], corrective_actions=corrective, evidence_references=[e.id for e in diagnosis.evidence_summary] if diagnosis else [], unresolved_questions=[] if diagnosis and diagnosis.status == "diagnosed" else ["What additional evidence would distinguish the top hypotheses?"], artifact_paths=[f"outputs/incidents/{incident_id}/postmortem.json", f"outputs/incidents/{incident_id}/postmortem.md"], receipts=[{"blameless": True}])
await workspace_write_json(ctx, f"outputs/incidents/{incident_id}/postmortem.json", pm); await workspace_write_text(ctx, f"outputs/incidents/{incident_id}/postmortem.md", _postmortem_markdown(pm), "text/markdown"); return pm.model_dump(mode="json") 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")