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 02:51:16 +00:00
parent f99f3d07cb
commit bddec204c3
1 changed files with 2 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
agent.py
View File

@@ -793,7 +793,7 @@ class ProductionIncidentCommander(A2AAgent[CommanderConfig, NoAuth]):
overall = "partially_executed" overall = "partially_executed"
result = ExecutionResult(incident_id=incident_id, status=overall, dry_run=dry_run, approval_provenance={"accepted": True, "approved_step_ids": sorted(approved), "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=[f"outputs/incidents/{incident_id}/execution.json"], receipts=[{"idempotency": "one guarded attempt per approved step", "rollback_evidence": "preserved in per-step records"}]) result = ExecutionResult(incident_id=incident_id, status=overall, dry_run=dry_run, approval_provenance={"accepted": True, "approved_step_ids": sorted(approved), "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=[f"outputs/incidents/{incident_id}/execution.json"], receipts=[{"idempotency": "one guarded attempt per approved step", "rollback_evidence": "preserved in per-step records"}])
await workspace_write_json(ctx, f"outputs/incidents/{incident_id}/execution.json", result) await workspace_write_json(ctx, f"outputs/incidents/{incident_id}/execution.json", result)
return 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.", 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.", 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( async def verify_recovery(
@@ -822,7 +822,7 @@ class ProductionIncidentCommander(A2AAgent[CommanderConfig, NoAuth]):
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=confidence, 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"}]) 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=confidence, 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 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") await emit_phase(ctx, "close_or_escalate", status="recovered" if recovered else "escalate")
return result return result.model_dump(mode="json")
@a2a.tool(description="Generate a blameless postmortem from saved incident artifacts", 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", 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( async def generate_postmortem(