diff --git a/agent.py b/agent.py index d7d5afc..1115698 100644 --- a/agent.py +++ b/agent.py @@ -121,6 +121,9 @@ class Learnhouse(A2AAgent): timeout_seconds=900, ) 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 if not creds.api_key: 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]]: subagents: list[dict[str, Any]] = [] for group in OPERATION_GROUPS: @@ -441,6 +461,52 @@ def _consumer_secret_optional(ctx: RunContext, name: str | None) -> str: 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: workspace = getattr(ctx, "_workspace", None) prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ())