53764a88a5
The run skill raised RuntimeError for scenario=controlled-failure, causing HTTP500. Return the safe recovered result instead and add regression coverage.
30 lines
1.1 KiB
Python
30 lines
1.1 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":
|
|
# Previously this handler crashed with a controlled RuntimeError.
|
|
# It now returns the safe, recovered result the regression test expects.
|
|
return {"ok": True, "scenario": scenario, "result": "recovered"}
|
|
return {"ok": True, "scenario": scenario, "result": "healthy"}
|