| """End-to-end validation: Cortex.0 driving the real OpenHands agent loop.""" |
|
|
| from __future__ import annotations |
|
|
| import os |
| import sys |
| import tempfile |
|
|
| sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) |
|
|
| from tests.fake_server import start |
|
|
|
|
| def main() -> int: |
| from cortex0.fusion import Cortex0Agent |
| from cortex0.settings import Cortex0Settings |
|
|
| server, base_url = start() |
| print(f"[1] Cortex.0 endpoint up at {base_url}") |
|
|
| settings = Cortex0Settings.load() |
| settings.base_url = base_url |
| settings.timeout = 60.0 |
|
|
| |
| from cortex0.health import probe_server |
|
|
| probe = probe_server(base_url, model="cortex0") |
| assert probe.reachable, f"not reachable: {probe.detail}" |
| assert probe.chat_ok, f"chat failed: {probe.detail}" |
| assert probe.tool_calling_ok, f"no tool call: {probe.detail}" |
| print(f"[2] probe ok (latency {probe.latency_s}s, models {probe.models})") |
|
|
| |
| agent = Cortex0Agent(settings=settings, verify=True) |
| summary = agent.summary() |
| assert summary["tool_calling_ok"], summary |
| print(f"[3] agent built with tools {summary['tools']}") |
|
|
| |
| with tempfile.TemporaryDirectory() as workdir: |
| conversation = agent.run("Say hello.", workspace=workdir) |
| events = list(conversation.state.events) |
| kinds = [type(e).__name__ for e in events] |
| print(f"[4] conversation ran: {len(events)} events") |
| for k in sorted(set(kinds)): |
| print(f" {k}: {kinds.count(k)}") |
|
|
| assert any("Action" in k for k in kinds), "no action was executed" |
| assert any("Observation" in k for k in kinds), "no observation returned" |
| assert any("Message" in k for k in kinds), "no message produced" |
| print("[5] action-perception loop confirmed") |
|
|
| server.shutdown() |
| print("\nRESULT: fusion validated end to end") |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| sys.exit(main()) |
|
|
