30 lines
1.2 KiB
Python
30 lines
1.2 KiB
Python
from __future__ import annotations
|
|
|
|
from a2a_pack import A2AAgent, NoAuth, RunContext, skill
|
|
|
|
|
|
class SelfHealingDemo(A2AAgent):
|
|
name = "self-healing-demo-v1"
|
|
description = "A live demonstration of bounded, auditable source self-repair."
|
|
version = "0.1.0"
|
|
|
|
@skill(description="Return the current self-healing demo state.")
|
|
async def status(self, ctx: RunContext[NoAuth]) -> dict[str, object]:
|
|
return {
|
|
"ok": True,
|
|
"agent": self.name,
|
|
"message": "The agent is online. Trigger the controlled failure once to exercise repair.",
|
|
}
|
|
|
|
@skill(description="Run the demo workflow, including its controlled production failure.")
|
|
async def run(
|
|
self,
|
|
ctx: RunContext[NoAuth],
|
|
scenario: str = "normal",
|
|
) -> dict[str, object]:
|
|
if scenario == "controlled-failure":
|
|
# Intentional smoke fixture: the regression test specifies the safe
|
|
# result. The production self-healing worker must repair this source.
|
|
raise RuntimeError("SELF_HEAL_SMOKE: controlled invoice total handler crashed")
|
|
return {"ok": True, "scenario": scenario, "result": "healthy"}
|