Short-circuit smoke checks to health

This commit is contained in:
2026-06-28 16:01:01 -03:00
parent 9b84a1cd08
commit ae71590f72

View File

@@ -121,6 +121,9 @@ class Learnhouse(A2AAgent):
timeout_seconds=900, timeout_seconds=900,
) )
async def auto(self, ctx: RunContext, goal: str) -> dict[str, Any]: async def auto(self, ctx: RunContext, goal: str) -> dict[str, Any]:
smoke_result = await self._maybe_smoke_test(ctx, goal)
if smoke_result is not None:
return smoke_result
creds = ctx.llm creds = ctx.llm
if not creds.api_key: if not creds.api_key:
return { return {
@@ -151,6 +154,23 @@ class Learnhouse(A2AAgent):
async def _maybe_smoke_test(self, ctx: RunContext, goal: str) -> dict[str, Any] | None:
operation_id = _select_smoke_operation(goal)
if operation_id is None:
return None
operation = OPERATIONS[operation_id]
result = await self._request(ctx, operation_id, parameters={}, body=None)
body = result.get("result") if result.get("ok") else result.get("error")
final = {
"endpoint_called": str(operation.get("method")) + " " + str(operation.get("path")),
"http_status": result.get("status_code"),
"body_preview": _body_preview(body, 200),
}
return {
"final": json.dumps(final, indent=2, ensure_ascii=False),
"messages": [],
}
def _operation_subagents(self, ctx: RunContext, skills_root: str) -> list[dict[str, Any]]: def _operation_subagents(self, ctx: RunContext, skills_root: str) -> list[dict[str, Any]]:
subagents: list[dict[str, Any]] = [] subagents: list[dict[str, Any]] = []
for group in OPERATION_GROUPS: for group in OPERATION_GROUPS:
@@ -441,6 +461,52 @@ def _consumer_secret_optional(ctx: RunContext, name: str | None) -> str:
return "" return ""
def _select_smoke_operation(goal: str) -> str | None:
text = str(goal or "").lower()
smoke_terms = ("smoke", "health", "status", "ping", "readiness", "liveness")
if not any(term in text for term in smoke_terms):
return None
preferred_terms = ("health", "status", "ping", "ready", "readiness", "live", "liveness")
candidates: list[tuple[int, int, str]] = []
for operation_id, operation in OPERATIONS.items():
if str(operation.get("method") or "").upper() != "GET":
continue
if operation.get("destructive"):
continue
if any(param.get("required") for param in operation.get("parameters") or []):
continue
path = str(operation.get("path") or "")
searchable = " ".join(
str(operation.get(key) or "")
for key in ("operation_id", "skill_name", "path", "summary", "description")
).lower()
is_root = path.strip() == "/"
matched_preferred = next(
(index for index, term in enumerate(preferred_terms) if term in searchable),
None,
)
if matched_preferred is not None:
score = matched_preferred
elif "smoke" in text and not is_root:
score = 100
elif "smoke" in text:
score = 1000
else:
continue
candidates.append((score, len(path), operation_id))
if not candidates:
return None
return min(candidates)[2]
def _body_preview(body: Any, limit: int) -> str:
if isinstance(body, str):
text = body
else:
text = json.dumps(body, ensure_ascii=False)
return text[:limit]
def _runtime_skills_root(ctx: RunContext) -> str: def _runtime_skills_root(ctx: RunContext) -> str:
workspace = getattr(ctx, "_workspace", None) workspace = getattr(ctx, "_workspace", None)
prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ()) prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ())