feat(ui): add TUI/Web interface #667

Closed
freemo wants to merge 1 commits from feature/m7-post-tui into master
15 changed files with 1909 additions and 0 deletions
+172
View File
@@ -0,0 +1,172 @@
"""ASV benchmarks for UI data provider and web route overhead.
Measures the performance of:
- LocalUIDataProvider plan list retrieval
- LocalUIDataProvider plan detail retrieval
- Web UI route handler response times
"""
from __future__ import annotations
import importlib
import sys
from pathlib import Path
from unittest.mock import MagicMock
# Ensure the local *source* tree is importable even when ASV has an
# older build of the package installed.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
import cleveragents # noqa: E402
importlib.reload(cleveragents)
from cleveragents.ui.data_provider import ( # noqa: E402
DiffItem,
LocalUIDataProvider,
LogEntry,
PlanDetail,
PlanSummary,
SessionSummary,
ValidationResult,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _mock_plan(
plan_id: str = "plan-1",
name: str = "plan-alpha",
status: str = "draft",
) -> MagicMock:
p = MagicMock()
p.id = plan_id
p.name = name
p.status = status
p.created_at = "2025-01-01T00:00:00"
p.prompt = "Do something"
p.decisions = []
p.linked_resources = []
p.project_id = "proj-1"
return p
class _StubProvider:
"""Trivial UIDataProvider for benchmarks."""
def get_plans(self, project_id: str) -> list[PlanSummary]:
return [
PlanSummary(
plan_id=f"plan-{i}",
name=f"plan-{i}",
status="draft",
created_at="2025-01-01T00:00:00",
project_id=project_id,
)
for i in range(50)
]
def get_plan_detail(self, plan_id: str) -> PlanDetail | None:
return PlanDetail(
plan_id=plan_id,
name="plan-alpha",
status="draft",
created_at="2025-01-01T00:00:00",
project_id="proj-1",
prompt="Do something",
decisions=[
{"id": f"d{i}", "type": "approval", "status": "pending"}
for i in range(10)
],
linked_resources=[f"res-{i}" for i in range(5)],
)
def get_sessions(self) -> list[SessionSummary]:
return []
def get_validations(self, plan_id: str) -> list[ValidationResult]:
return []
def get_diffs(self, plan_id: str) -> list[DiffItem]:
return []
def get_logs(self, plan_id: str, level: str = "INFO") -> list[LogEntry]:
return []
_STUB = _StubProvider()
# ---------------------------------------------------------------------------
# Data provider benchmarks
# ---------------------------------------------------------------------------
class TimeDataProvider:
"""Benchmark LocalUIDataProvider backed by mock services."""
timeout = 30
def setup(self) -> None:
self._mock_svc = MagicMock()
plans = [_mock_plan(f"plan-{i}", f"plan-{i}") for i in range(50)]
self._mock_svc.list_plans.return_value = plans
detail = _mock_plan()
detail.decisions = [MagicMock(id="d1", type="a", status="p")]
detail.linked_resources = ["r1"]
detail.project_id = "proj-1"
self._mock_svc.get_plan.return_value = detail
self._provider = LocalUIDataProvider(plan_service=self._mock_svc)
def time_get_plans(self) -> None:
self._provider.get_plans("proj-1")
def time_get_plan_detail(self) -> None:
self._provider.get_plan_detail("plan-1")
def time_get_validations(self) -> None:
self._provider.get_validations("plan-1")
def time_get_diffs(self) -> None:
self._provider.get_diffs("plan-1")
def time_get_logs(self) -> None:
self._provider.get_logs("plan-1")
# ---------------------------------------------------------------------------
# Web route benchmarks
# ---------------------------------------------------------------------------
class TimeWebRoutes:
"""Benchmark FastAPI web UI route handler overhead."""
timeout = 30
def setup(self) -> None:
from fastapi.testclient import TestClient
from cleveragents.ui.web_app import create_web_app
app = create_web_app(provider=_STUB)
self._client = TestClient(app)
def time_list_plans(self) -> None:
self._client.get("/ui/plans")
def time_plan_detail(self) -> None:
self._client.get("/ui/plans/plan-1")
def time_sessions(self) -> None:
self._client.get("/ui/sessions")
def time_diffs(self) -> None:
self._client.get("/ui/diffs/plan-1")
def time_health(self) -> None:
self._client.get("/ui/health")
+352
View File
@@ -0,0 +1,352 @@
"""Step definitions for the TUI/Web UI interface feature."""
from __future__ import annotations
import os
from unittest.mock import MagicMock, patch
from behave import given, then, when
from behave.runner import Context
from cleveragents.ui.data_provider import (
DiffItem,
LocalUIDataProvider,
LogEntry,
PlanDetail,
PlanSummary,
SessionSummary,
ValidationResult,
)
# ---------------------------------------------------------------------------
# Fixtures: mock plan objects
# ---------------------------------------------------------------------------
_PLAN_ALPHA = MagicMock()
_PLAN_ALPHA.id = "plan-1"
_PLAN_ALPHA.name = "plan-alpha"
_PLAN_ALPHA.status = "draft"
_PLAN_ALPHA.created_at = "2025-01-01T00:00:00"
_PLAN_ALPHA.prompt = "Do something"
_PLAN_ALPHA.decisions = []
_PLAN_ALPHA.linked_resources = []
_PLAN_BETA = MagicMock()
_PLAN_BETA.id = "plan-2"
_PLAN_BETA.name = "plan-beta"
_PLAN_BETA.status = "built"
_PLAN_BETA.created_at = "2025-01-02T00:00:00"
_PLAN_BETA.prompt = "Do something else"
_DEC = MagicMock()
_DEC.id = "dec-1"
_DEC.type = "approval"
_DEC.status = "pending"
_PLAN_ALPHA_WITH_DEC = MagicMock()
_PLAN_ALPHA_WITH_DEC.id = "plan-1"
_PLAN_ALPHA_WITH_DEC.name = "plan-alpha"
_PLAN_ALPHA_WITH_DEC.status = "draft"
_PLAN_ALPHA_WITH_DEC.created_at = "2025-01-01T00:00:00"
_PLAN_ALPHA_WITH_DEC.prompt = "Do something"
_PLAN_ALPHA_WITH_DEC.decisions = [_DEC]
_PLAN_ALPHA_WITH_DEC.linked_resources = ["res-1"]
_PLAN_ALPHA_WITH_DEC.project_id = "proj-1"
# ---------------------------------------------------------------------------
# Mock data provider for web tests
# ---------------------------------------------------------------------------
class _MockProvider:
"""Trivial UIDataProvider for web route testing."""
def __init__(self, *, with_plans: bool = False) -> None:
self._with_plans = with_plans
def get_plans(self, project_id: str) -> list[PlanSummary]:
if not self._with_plans:
return []
return [
PlanSummary(
plan_id="plan-1",
name="plan-alpha",
status="draft",
created_at="2025-01-01T00:00:00",
project_id=project_id or "proj-1",
),
]
def get_plan_detail(self, plan_id: str) -> PlanDetail | None:
if not self._with_plans:
return None
if plan_id != "plan-1":
return None
return PlanDetail(
plan_id="plan-1",
name="plan-alpha",
status="draft",
created_at="2025-01-01T00:00:00",
project_id="proj-1",
prompt="Do something",
decisions=[{"id": "d1", "type": "approval", "status": "pending"}],
linked_resources=["res-1"],
)
def get_sessions(self) -> list[SessionSummary]:
return []
def get_validations(self, plan_id: str) -> list[ValidationResult]:
return []
def get_diffs(self, plan_id: str) -> list[DiffItem]:
return []
def get_logs(self, plan_id: str, level: str = "INFO") -> list[LogEntry]:
return []
# ---------------------------------------------------------------------------
# UIDataProvider steps
# ---------------------------------------------------------------------------
@given("a UIDataProvider backed by mock services")
def step_provider_empty(context: Context) -> None:
mock_svc = MagicMock()
mock_svc.list_plans.return_value = []
mock_svc.get_plan.return_value = None
context.ui_provider = LocalUIDataProvider(plan_service=mock_svc)
@given("a UIDataProvider backed by mock services with plans")
def step_provider_with_plans(context: Context) -> None:
mock_svc = MagicMock()
mock_svc.list_plans.return_value = [_PLAN_ALPHA, _PLAN_BETA]
mock_svc.get_plan.side_effect = lambda pid: (
_PLAN_ALPHA_WITH_DEC if pid == "plan-1" else None
)
context.ui_provider = LocalUIDataProvider(plan_service=mock_svc)
@when('I request UI plans for project "{project_id}"')
def step_request_ui_plans(context: Context, project_id: str) -> None:
context.ui_result_plans = context.ui_provider.get_plans(project_id)
@then("the UI plan list should be empty")
def step_ui_plans_empty(context: Context) -> None:
assert context.ui_result_plans == [], (
f"Expected empty, got {context.ui_result_plans}"
)
@then("the UI plan list should contain {count:d} plans")
def step_ui_plans_count(context: Context, count: int) -> None:
assert len(context.ui_result_plans) == count, (
f"Expected {count}, got {len(context.ui_result_plans)}"
)
@then('the first UI plan name should be "{name}"')
def step_first_ui_plan_name(context: Context, name: str) -> None:
assert context.ui_result_plans[0].name == name
@when('I request UI plan detail for "{plan_id}"')
def step_request_ui_detail(context: Context, plan_id: str) -> None:
context.ui_result_detail = context.ui_provider.get_plan_detail(plan_id)
@then('the UI plan detail should have name "{name}"')
def step_ui_detail_name(context: Context, name: str) -> None:
assert context.ui_result_detail is not None
assert context.ui_result_detail.name == name
@then("the UI plan detail should include decisions")
def step_ui_detail_decisions(context: Context) -> None:
assert context.ui_result_detail is not None
assert isinstance(context.ui_result_detail.decisions, list)
@then("the UI plan detail should be None")
def step_ui_detail_none(context: Context) -> None:
assert context.ui_result_detail is None
@when('I request UI validations for plan "{plan_id}"')
def step_request_ui_validations(context: Context, plan_id: str) -> None:
context.ui_result_validations = context.ui_provider.get_validations(plan_id)
@then("the UI validations list should be empty")
def step_ui_validations_empty(context: Context) -> None:
assert context.ui_result_validations == []
@when('I request UI diffs for plan "{plan_id}"')
def step_request_ui_diffs(context: Context, plan_id: str) -> None:
context.ui_result_diffs = context.ui_provider.get_diffs(plan_id)
@then("the UI diffs list should be empty")
def step_ui_diffs_empty(context: Context) -> None:
assert context.ui_result_diffs == []
@when('I request UI logs for plan "{plan_id}" with level "{level}"')
def step_request_ui_logs(context: Context, plan_id: str, level: str) -> None:
context.ui_result_logs = context.ui_provider.get_logs(plan_id, level)
@then("the UI logs list should be empty")
def step_ui_logs_empty(context: Context) -> None:
assert context.ui_result_logs == []
@when("I request UI sessions")
def step_request_ui_sessions(context: Context) -> None:
context.ui_result_sessions = context.ui_provider.get_sessions()
@then("the UI sessions list should be empty")
def step_ui_sessions_empty(context: Context) -> None:
assert context.ui_result_sessions == []
# ---------------------------------------------------------------------------
# Configuration steps
# ---------------------------------------------------------------------------
@given("default UI settings")
def step_default_ui_settings(context: Context) -> None:
for key in (
"CLEVERAGENTS_TUI_REFRESH_INTERVAL",
"CLEVERAGENTS_WEB_UI_ENABLED",
"CLEVERAGENTS_WEB_UI_PORT",
):
os.environ.pop(key, None)
from cleveragents.config.settings import Settings
context.ui_settings = Settings()
@given("UI settings with tui_refresh_interval set to {val:d}")
def step_custom_ui_refresh(context: Context, val: int) -> None:
os.environ["CLEVERAGENTS_TUI_REFRESH_INTERVAL"] = str(val)
from cleveragents.config.settings import Settings
context.ui_settings = Settings()
os.environ.pop("CLEVERAGENTS_TUI_REFRESH_INTERVAL", None)
@then("the TUI refresh interval should be {val:d}")
def step_check_tui_refresh(context: Context, val: int) -> None:
assert context.ui_settings.tui_refresh_interval == val
@then("web_ui_enabled should be false")
def step_web_ui_disabled(context: Context) -> None:
assert context.ui_settings.web_ui_enabled is False
@then("web_ui_port should be {port:d}")
def step_web_ui_port(context: Context, port: int) -> None:
assert context.ui_settings.web_ui_port == port
# ---------------------------------------------------------------------------
# Web route steps
# ---------------------------------------------------------------------------
@given("a web UI app backed by mock provider")
def step_web_ui_app_empty(context: Context) -> None:
from fastapi.testclient import TestClient
from cleveragents.ui.web_app import create_web_app
app = create_web_app(provider=_MockProvider())
context.web_ui_client = TestClient(app)
@given("a web UI app backed by mock provider with plans")
def step_web_ui_app_plans(context: Context) -> None:
from fastapi.testclient import TestClient
from cleveragents.ui.web_app import create_web_app
app = create_web_app(provider=_MockProvider(with_plans=True))
context.web_ui_client = TestClient(app)
@when('I send a GET to UI route "{path}"')
def step_get_ui_route(context: Context, path: str) -> None:
context.web_ui_response = context.web_ui_client.get(path)
@then("the UI response status should be {code:d}")
def step_ui_status_code(context: Context, code: int) -> None:
assert context.web_ui_response.status_code == code, (
f"Expected {code}, got {context.web_ui_response.status_code}"
)
@then('the UI response JSON should contain key "{key}"')
def step_ui_json_key(context: Context, key: str) -> None:
data = context.web_ui_response.json()
assert key in data, f"Key '{key}' not in {list(data.keys())}"
# ---------------------------------------------------------------------------
# CLI steps
# ---------------------------------------------------------------------------
@given("the CLI app is loaded for UI testing")
def step_cli_loaded_for_ui(context: Context) -> None:
from cleveragents.cli.main import ensure_cli_commands_registered
ensure_cli_commands_registered()
context.ui_cli_loaded = True
@then('the "{name}" command group should be available')
def step_command_group_available(context: Context, name: str) -> None:
from cleveragents.cli.commands.ui import app as ui_app
assert ui_app is not None
@given("a CLI runner with web UI disabled")
def step_cli_web_ui_disabled(context: Context) -> None:
from typer.testing import CliRunner
from cleveragents.cli.commands.ui import app as ui_app
context.ui_cli_runner = CliRunner()
context.ui_cli_app = ui_app
@when('I invoke the UI CLI with "{cmd}"')
def step_invoke_ui_cli(context: Context, cmd: str) -> None:
parts = cmd.split()
with patch.dict(os.environ, {"CLEVERAGENTS_WEB_UI_ENABLED": "false"}):
context.ui_cli_result = context.ui_cli_runner.invoke(context.ui_cli_app, parts)
@then("the UI CLI should exit with code {code:d}")
def step_ui_cli_exit_code(context: Context, code: int) -> None:
assert context.ui_cli_result.exit_code == code, (
f"Expected exit code {code}, got {context.ui_cli_result.exit_code}: "
f"{context.ui_cli_result.output}"
)
@then('the UI CLI output should mention "{text}"')
def step_ui_cli_output_contains(context: Context, text: str) -> None:
assert text.lower() in context.ui_cli_result.output.lower(), (
f"'{text}' not found in output: {context.ui_cli_result.output}"
)
View File
+122
View File
@@ -0,0 +1,122 @@
Feature: TUI and Web UI interface
As a developer using CleverAgents
I want a TUI dashboard and web UI
So that I can monitor plans, diffs, and validations visually
# --- UIDataProvider ---
Scenario: UIDataProvider returns empty plan list
Given a UIDataProvider backed by mock services
When I request UI plans for project "proj-1"
Then the UI plan list should be empty
Scenario: UIDataProvider returns plan list
Given a UIDataProvider backed by mock services with plans
When I request UI plans for project "proj-1"
Then the UI plan list should contain 2 plans
And the first UI plan name should be "plan-alpha"
Scenario: UIDataProvider returns plan detail
Given a UIDataProvider backed by mock services with plans
When I request UI plan detail for "plan-1"
Then the UI plan detail should have name "plan-alpha"
And the UI plan detail should include decisions
Scenario: UIDataProvider returns None for missing plan detail
Given a UIDataProvider backed by mock services
When I request UI plan detail for "nonexistent"
Then the UI plan detail should be None
Scenario: UIDataProvider returns validations
Given a UIDataProvider backed by mock services
When I request UI validations for plan "plan-1"
Then the UI validations list should be empty
Scenario: UIDataProvider returns diffs
Given a UIDataProvider backed by mock services
When I request UI diffs for plan "plan-1"
Then the UI diffs list should be empty
Scenario: UIDataProvider returns logs
Given a UIDataProvider backed by mock services
When I request UI logs for plan "plan-1" with level "INFO"
Then the UI logs list should be empty
Scenario: UIDataProvider returns sessions
Given a UIDataProvider backed by mock services
When I request UI sessions
Then the UI sessions list should be empty
# --- Configuration ---
Scenario: Default TUI refresh interval
Given default UI settings
Then the TUI refresh interval should be 5
Scenario: Custom TUI refresh interval
Given UI settings with tui_refresh_interval set to 10
Then the TUI refresh interval should be 10
Scenario: Web UI disabled by default
Given default UI settings
Then web_ui_enabled should be false
Scenario: Web UI port default
Given default UI settings
Then web_ui_port should be 8080
# --- Web route responses ---
Scenario: Web UI plan list returns JSON
Given a web UI app backed by mock provider
When I send a GET to UI route "/ui/plans"
Then the UI response status should be 200
And the UI response JSON should contain key "plans"
Scenario: Web UI plan detail returns JSON
Given a web UI app backed by mock provider with plans
When I send a GET to UI route "/ui/plans/plan-1"
Then the UI response status should be 200
And the UI response JSON should contain key "plan_id"
Scenario: Web UI plan detail 404 for missing plan
Given a web UI app backed by mock provider
When I send a GET to UI route "/ui/plans/nonexistent"
Then the UI response status should be 404
And the UI response JSON should contain key "error"
Scenario: Web UI sessions returns JSON
Given a web UI app backed by mock provider
When I send a GET to UI route "/ui/sessions"
Then the UI response status should be 200
And the UI response JSON should contain key "sessions"
Scenario: Web UI diffs returns JSON
Given a web UI app backed by mock provider
When I send a GET to UI route "/ui/diffs/plan-1"
Then the UI response status should be 200
And the UI response JSON should contain key "diffs"
Scenario: Web UI validations returns JSON
Given a web UI app backed by mock provider
When I send a GET to UI route "/ui/validations/plan-1"
Then the UI response status should be 200
And the UI response JSON should contain key "validations"
Scenario: Web UI health endpoint
Given a web UI app backed by mock provider
When I send a GET to UI route "/ui/health"
Then the UI response status should be 200
And the UI response JSON should contain key "status"
# --- CLI commands ---
Scenario: CLI ui tui command is registered
Given the CLI app is loaded for UI testing
Then the "ui" command group should be available
Scenario: CLI ui web rejects when disabled
Given a CLI runner with web UI disabled
When I invoke the UI CLI with "web"
Then the UI CLI should exit with code 1
And the UI CLI output should mention "disabled"
+2
View File
@@ -46,6 +46,8 @@ dependencies = [
"RestrictedPython>=7.0", # Secure sandbox for user-supplied code
"jsonschema>=4.20.0", # JSON Schema validation for tool inputs/outputs
"tomlkit>=0.13.0", # TOML writing with comment preservation for config CLI
"textual>=0.50.0", # TUI framework for interactive terminal UI (#341)
"fastapi>=0.115.0", # Lightweight web UI stub for read-only JSON routes (#341)
]
[project.optional-dependencies]
+274
View File
@@ -0,0 +1,274 @@
"""Helper script for tui_interface.robot smoke tests.
Each subcommand is a self-contained check that prints a sentinel on success.
"""
from __future__ import annotations
import os
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch
# Ensure local source tree is importable
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
from cleveragents.ui.data_provider import ( # noqa: E402
DiffItem,
LocalUIDataProvider,
LogEntry,
PlanDetail,
PlanSummary,
SessionSummary,
ValidationResult,
)
# ---------------------------------------------------------------------------
# Mock plan objects
# ---------------------------------------------------------------------------
def _mock_plan(
plan_id: str = "plan-1",
name: str = "plan-alpha",
status: str = "draft",
) -> MagicMock:
p = MagicMock()
p.id = plan_id
p.name = name
p.status = status
p.created_at = "2025-01-01T00:00:00"
p.prompt = "Do something"
p.decisions = []
p.linked_resources = []
p.project_id = "proj-1"
return p
# ---------------------------------------------------------------------------
# Provider tests
# ---------------------------------------------------------------------------
def _provider_empty_plans() -> None:
mock_svc = MagicMock()
mock_svc.list_plans.return_value = []
provider = LocalUIDataProvider(plan_service=mock_svc)
plans = provider.get_plans("proj-1")
assert plans == [], f"Expected empty, got {plans}"
print("tui-provider-empty-plans-ok")
def _provider_plans() -> None:
mock_svc = MagicMock()
mock_svc.list_plans.return_value = [_mock_plan(), _mock_plan("plan-2", "plan-beta")]
provider = LocalUIDataProvider(plan_service=mock_svc)
plans = provider.get_plans("proj-1")
assert len(plans) == 2, f"Expected 2, got {len(plans)}"
assert plans[0].name == "plan-alpha"
print("tui-provider-plans-ok")
def _provider_detail() -> None:
plan = _mock_plan()
dec = MagicMock()
dec.id = "dec-1"
dec.type = "approval"
dec.status = "pending"
plan.decisions = [dec]
plan.linked_resources = ["res-1"]
plan.project_id = "proj-1"
mock_svc = MagicMock()
mock_svc.get_plan.return_value = plan
provider = LocalUIDataProvider(plan_service=mock_svc)
detail = provider.get_plan_detail("plan-1")
assert detail is not None
assert detail.name == "plan-alpha"
assert len(detail.decisions) == 1
print("tui-provider-detail-ok")
# ---------------------------------------------------------------------------
# Web route tests
# ---------------------------------------------------------------------------
class _MockProvider:
"""Trivial UIDataProvider for web route testing."""
def __init__(self, *, with_plans: bool = False) -> None:
self._with_plans = with_plans
def get_plans(self, project_id: str) -> list[PlanSummary]:
if not self._with_plans:
return []
return [
PlanSummary(
plan_id="plan-1",
name="plan-alpha",
status="draft",
created_at="2025-01-01T00:00:00",
project_id="proj-1",
),
]
def get_plan_detail(self, plan_id: str) -> PlanDetail | None:
if not self._with_plans or plan_id != "plan-1":
return None
return PlanDetail(
plan_id="plan-1",
name="plan-alpha",
status="draft",
created_at="2025-01-01T00:00:00",
project_id="proj-1",
prompt="Do something",
decisions=[{"id": "d1", "type": "approval", "status": "pending"}],
linked_resources=["res-1"],
)
def get_sessions(self) -> list[SessionSummary]:
return []
def get_validations(self, plan_id: str) -> list[ValidationResult]:
return []
def get_diffs(self, plan_id: str) -> list[DiffItem]:
return []
def get_logs(self, plan_id: str, level: str = "INFO") -> list[LogEntry]:
return []
def _web_plans() -> None:
from fastapi.testclient import TestClient
from cleveragents.ui.web_app import create_web_app
app = create_web_app(provider=_MockProvider())
client = TestClient(app)
resp = client.get("/ui/plans")
assert resp.status_code == 200
assert "plans" in resp.json()
print("tui-web-plans-ok")
def _web_plan_detail() -> None:
from fastapi.testclient import TestClient
from cleveragents.ui.web_app import create_web_app
app = create_web_app(provider=_MockProvider(with_plans=True))
client = TestClient(app)
resp = client.get("/ui/plans/plan-1")
assert resp.status_code == 200
assert "plan_id" in resp.json()
print("tui-web-plan-detail-ok")
def _web_sessions() -> None:
from fastapi.testclient import TestClient
from cleveragents.ui.web_app import create_web_app
app = create_web_app(provider=_MockProvider())
client = TestClient(app)
resp = client.get("/ui/sessions")
assert resp.status_code == 200
assert "sessions" in resp.json()
print("tui-web-sessions-ok")
def _web_diffs() -> None:
from fastapi.testclient import TestClient
from cleveragents.ui.web_app import create_web_app
app = create_web_app(provider=_MockProvider())
client = TestClient(app)
resp = client.get("/ui/diffs/plan-1")
assert resp.status_code == 200
assert "diffs" in resp.json()
print("tui-web-diffs-ok")
def _web_health() -> None:
from fastapi.testclient import TestClient
from cleveragents.ui.web_app import create_web_app
app = create_web_app(provider=_MockProvider())
client = TestClient(app)
resp = client.get("/ui/health")
assert resp.status_code == 200
assert resp.json()["status"] == "ok"
print("tui-web-health-ok")
# ---------------------------------------------------------------------------
# Settings tests
# ---------------------------------------------------------------------------
def _settings_defaults() -> None:
for key in (
"CLEVERAGENTS_TUI_REFRESH_INTERVAL",
"CLEVERAGENTS_WEB_UI_ENABLED",
"CLEVERAGENTS_WEB_UI_PORT",
):
os.environ.pop(key, None)
from cleveragents.config.settings import Settings
s = Settings()
assert s.tui_refresh_interval == 5
assert s.web_ui_enabled is False
assert s.web_ui_port == 8080
print("tui-settings-defaults-ok")
# ---------------------------------------------------------------------------
# CLI tests
# ---------------------------------------------------------------------------
def _cli_web_disabled() -> None:
from typer.testing import CliRunner
from cleveragents.cli.commands.ui import app as ui_app
runner = CliRunner()
with patch.dict(os.environ, {"CLEVERAGENTS_WEB_UI_ENABLED": "false"}):
result = runner.invoke(ui_app, ["web"])
assert result.exit_code == 1
assert "disabled" in result.output.lower()
print("tui-cli-web-disabled-ok")
# ---------------------------------------------------------------------------
# Dispatcher
# ---------------------------------------------------------------------------
_CMDS = {
"provider-empty-plans": _provider_empty_plans,
"provider-plans": _provider_plans,
"provider-detail": _provider_detail,
"web-plans": _web_plans,
"web-plan-detail": _web_plan_detail,
"web-sessions": _web_sessions,
"web-diffs": _web_diffs,
"web-health": _web_health,
"settings-defaults": _settings_defaults,
"cli-web-disabled": _cli_web_disabled,
}
def main() -> None:
if len(sys.argv) < 2 or sys.argv[1] not in _CMDS:
print(f"Usage: {sys.argv[0]} <{'|'.join(_CMDS)}>", file=sys.stderr)
sys.exit(2)
_CMDS[sys.argv[1]]()
if __name__ == "__main__":
main()
+71
View File
@@ -0,0 +1,71 @@
*** Settings ***
Documentation End-to-end smoke tests for TUI/Web UI interface
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_tui_interface.py
*** Test Cases ***
UIDataProvider Returns Empty Plans
[Documentation] Verify UIDataProvider returns empty plan list
${result}= Run Process ${PYTHON} ${HELPER} provider-empty-plans cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} tui-provider-empty-plans-ok
UIDataProvider Returns Plans
[Documentation] Verify UIDataProvider returns populated plan list
${result}= Run Process ${PYTHON} ${HELPER} provider-plans cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} tui-provider-plans-ok
UIDataProvider Returns Plan Detail
[Documentation] Verify UIDataProvider returns plan detail
${result}= Run Process ${PYTHON} ${HELPER} provider-detail cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} tui-provider-detail-ok
Web UI Plans Route
[Documentation] Verify /ui/plans returns JSON
${result}= Run Process ${PYTHON} ${HELPER} web-plans cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} tui-web-plans-ok
Web UI Plan Detail Route
[Documentation] Verify /ui/plans/<id> returns JSON
${result}= Run Process ${PYTHON} ${HELPER} web-plan-detail cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} tui-web-plan-detail-ok
Web UI Sessions Route
[Documentation] Verify /ui/sessions returns JSON
${result}= Run Process ${PYTHON} ${HELPER} web-sessions cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} tui-web-sessions-ok
Web UI Diffs Route
[Documentation] Verify /ui/diffs/<plan_id> returns JSON
${result}= Run Process ${PYTHON} ${HELPER} web-diffs cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} tui-web-diffs-ok
Web UI Health Route
[Documentation] Verify /ui/health returns ok
${result}= Run Process ${PYTHON} ${HELPER} web-health cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} tui-web-health-ok
Settings Defaults
[Documentation] Verify default TUI/Web settings
${result}= Run Process ${PYTHON} ${HELPER} settings-defaults cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} tui-settings-defaults-ok
CLI UI Web Disabled
[Documentation] Verify CLI rejects web UI when disabled
${result}= Run Process ${PYTHON} ${HELPER} cli-web-disabled cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} tui-cli-web-disabled-ok
+79
View File
@@ -0,0 +1,79 @@
"""UI commands for launching TUI and Web interfaces.
``agents ui tui`` — Launch the Textual-based TUI dashboard.
``agents ui web`` — Start the read-only local web UI server.
Based on Forgejo #341.
"""
from __future__ import annotations
from typing import Annotated
import typer
app: typer.Typer = typer.Typer(
name="ui",
help="Launch the TUI dashboard or web UI server.",
)
@app.command()
def tui(
project_id: Annotated[
str,
typer.Option("--project-id", "-p", help="Project ID to display"),
] = "",
refresh: Annotated[
int | None,
typer.Option(
"--refresh",
"-r",
help="Auto-refresh interval in seconds (overrides settings)",
),
] = None,
) -> None:
"""Launch the Textual TUI dashboard."""
from cleveragents.config.settings import get_settings
from cleveragents.ui.data_provider import LocalUIDataProvider
from cleveragents.ui.tui_app import CleverAgentsTUI
settings = get_settings()
interval = refresh if refresh is not None else settings.tui_refresh_interval
provider = LocalUIDataProvider()
tui_app = CleverAgentsTUI(
provider=provider,
project_id=project_id,
refresh_interval=interval,
)
tui_app.run()
@app.command()
def web(
host: Annotated[
str,
typer.Option("--host", "-h", help="Bind address"),
] = "127.0.0.1",
port: Annotated[
int | None,
typer.Option("--port", help="Listen port (overrides settings)"),
] = None,
) -> None:
"""Start the read-only web UI server."""
from cleveragents.config.settings import get_settings
from cleveragents.ui.data_provider import LocalUIDataProvider
from cleveragents.ui.web_app import run_web_ui
settings = get_settings()
if not settings.web_ui_enabled:
typer.echo(
"Web UI is disabled. Set CLEVERAGENTS_WEB_UI_ENABLED=true "
"or update your configuration."
)
raise typer.Exit(code=1)
resolved_port = port if port is not None else settings.web_ui_port
provider = LocalUIDataProvider()
typer.echo(f"Starting web UI on {host}:{resolved_port}")
run_web_ui(provider=provider, host=host, port=resolved_port)
+7
View File
@@ -97,6 +97,7 @@ def _register_subcommands() -> None:
from cleveragents.cli.commands.auto_debug import app as auto_debug_app
from cleveragents.cli.commands.repl import _repl_app
from cleveragents.cli.commands.server import app as server_app
from cleveragents.cli.commands.ui import app as ui_app
except Exception as exc: # pragma: no cover
import traceback
@@ -194,6 +195,11 @@ def _register_subcommands() -> None:
name="server",
help="Server connection management (stub)",
)
app.add_typer(
ui_app,
name="ui",
help="Launch TUI dashboard or web UI server",
)
_subcommands_registered = True
@@ -611,6 +617,7 @@ def main(args: list[str] | None = None) -> int:
"invariant", # Invariant constraint management
"repl", # Interactive REPL
"server", # Server connection management
"ui", # TUI dashboard / web UI (#341)
"tell", # Shortcut for plan tell
"build", # Shortcut for plan build
"apply", # Shortcut for plan apply
+20
View File
@@ -435,6 +435,26 @@ class Settings(BaseSettings):
description="Completed job retention in seconds before cleanup.",
)
# TUI / Web UI configuration (#341)
tui_refresh_interval: int = Field(
default=5,
ge=1,
validation_alias=AliasChoices("CLEVERAGENTS_TUI_REFRESH_INTERVAL"),
description="Auto-refresh interval in seconds for the TUI dashboard.",
)
web_ui_enabled: bool = Field(
default=False,
validation_alias=AliasChoices("CLEVERAGENTS_WEB_UI_ENABLED"),
description="Enable the read-only local web UI server.",
)
web_ui_port: int = Field(
default=8080,
ge=1,
le=65535,
validation_alias=AliasChoices("CLEVERAGENTS_WEB_UI_PORT"),
description="Port for the read-only local web UI server.",
)
# Mock providers flag (M4 - provider fixes)
mock_providers: bool = Field(
default=False,
+29
View File
@@ -0,0 +1,29 @@
"""UI package for CleverAgents TUI and Web interfaces.
Provides a data-provider protocol that abstracts service access for both
the Textual-based TUI and the read-only FastAPI web stub.
Based on Forgejo #341.
"""
from cleveragents.ui.data_provider import (
DiffItem,
LocalUIDataProvider,
LogEntry,
PlanDetail,
PlanSummary,
SessionSummary,
UIDataProvider,
ValidationResult,
)
__all__ = [
"DiffItem",
"LocalUIDataProvider",
"LogEntry",
"PlanDetail",
"PlanSummary",
"SessionSummary",
"UIDataProvider",
"ValidationResult",
]
+294
View File
@@ -0,0 +1,294 @@
"""UI data-provider protocol and local implementation.
Defines a ``UIDataProvider`` protocol consumed by both the Textual TUI and
the FastAPI web stub. ``LocalUIDataProvider`` backs the protocol using
services resolved from the DI container.
Based on Forgejo #341.
"""
from __future__ import annotations
import contextlib
from dataclasses import dataclass, field
from typing import Protocol, runtime_checkable
# ---------------------------------------------------------------------------
# Data transfer objects
# ---------------------------------------------------------------------------
@dataclass(frozen=True, slots=True)
class PlanSummary:
"""Lightweight plan listing entry."""
plan_id: str
name: str
status: str
created_at: str
project_id: str
@dataclass(frozen=True, slots=True)
class PlanDetail:
"""Full plan detail with decision tree information."""
plan_id: str
name: str
status: str
created_at: str
project_id: str
prompt: str
decisions: list[dict[str, str]]
linked_resources: list[str]
@dataclass(frozen=True, slots=True)
class SessionSummary:
"""Session listing entry."""
session_id: str
actor_name: str
message_count: int
created_at: str
updated_at: str
@dataclass(frozen=True, slots=True)
class ValidationResult:
"""Single validation outcome."""
name: str
passed: bool
message: str
@dataclass(frozen=True, slots=True)
class DiffItem:
"""Changeset diff entry."""
file_path: str
operation: str
diff_text: str
@dataclass(frozen=True, slots=True)
class LogEntry:
"""Structured log entry."""
timestamp: str
level: str
message: str
source: str = ""
# ---------------------------------------------------------------------------
# Protocol
# ---------------------------------------------------------------------------
@runtime_checkable
class UIDataProvider(Protocol):
"""Read-only data provider consumed by UI layers."""
def get_plans(self, project_id: str) -> list[PlanSummary]:
"""Return plan summaries for a project."""
...
def get_plan_detail(self, plan_id: str) -> PlanDetail | None:
"""Return full plan detail or ``None`` if not found."""
...
def get_sessions(self) -> list[SessionSummary]:
"""Return all sessions."""
...
def get_validations(self, plan_id: str) -> list[ValidationResult]:
"""Return validation results for a plan."""
...
def get_diffs(self, plan_id: str) -> list[DiffItem]:
"""Return changeset diffs for a plan."""
...
def get_logs(self, plan_id: str, level: str = "INFO") -> list[LogEntry]:
"""Return log entries for a plan filtered by *level*."""
...
# ---------------------------------------------------------------------------
# Local implementation backed by DI container services
# ---------------------------------------------------------------------------
@dataclass(slots=True)
class _ServiceHolder:
"""Lazy container accessor to avoid import-time DI resolution."""
_plan_service: object | None = field(default=None, repr=False)
_project_service: object | None = field(default=None, repr=False)
def _resolve(self) -> None:
if self._plan_service is not None:
return
from cleveragents.application.container import get_container
container = get_container()
self._plan_service = container.plan_service()
self._project_service = container.project_service()
@property
def plan_service(self) -> object:
self._resolve()
assert self._plan_service is not None
return self._plan_service
@property
def project_service(self) -> object:
self._resolve()
assert self._project_service is not None
return self._project_service
class LocalUIDataProvider:
"""``UIDataProvider`` backed by local DI container services.
Each method gracefully returns empty data when the underlying service
raises (the UI should never crash due to a backend hiccup).
"""
def __init__(
self,
*,
plan_service: object | None = None,
project_service: object | None = None,
) -> None:
self._holder = _ServiceHolder(
_plan_service=plan_service,
_project_service=project_service,
)
# -- plans ---------------------------------------------------------------
def get_plans(self, project_id: str) -> list[PlanSummary]:
"""Return plan summaries for *project_id*."""
try:
svc = self._holder.plan_service
list_fn = getattr(svc, "list_plans", None)
if list_fn is None:
return []
# PlanService.list_plans expects a Project object; we build a
# lightweight stub with just the id set.
from pathlib import Path as _Path
from cleveragents.domain.models.core.project_legacy import Project
int_id: int | None = None
with contextlib.suppress(ValueError, TypeError):
int_id = int(project_id)
project = Project(name="stub", path=_Path.cwd())
project.id = int_id
raw_plans: list[object] = list_fn(project)
return [self._to_summary(p, project_id) for p in raw_plans]
except Exception:
return []
def get_plan_detail(self, plan_id: str) -> PlanDetail | None:
"""Return plan detail for *plan_id*."""
try:
svc = self._holder.plan_service
get_fn = getattr(svc, "get_plan", None)
if get_fn is None:
return None
raw = get_fn(plan_id)
if raw is None:
return None
return self._to_detail(raw)
except Exception:
return None
# -- sessions ------------------------------------------------------------
def get_sessions(self) -> list[SessionSummary]:
"""Return all sessions (empty list when service unavailable)."""
try:
from cleveragents.domain.models.core.session import SessionService
svc = self._holder.plan_service
session_svc_attr = getattr(svc, "session_service", None)
if session_svc_attr is None:
return []
if not isinstance(session_svc_attr, SessionService):
return []
sessions = session_svc_attr.list()
return [
SessionSummary(
session_id=getattr(s, "session_id", ""),
actor_name=getattr(s, "actor_name", "") or "",
message_count=len(getattr(s, "messages", [])),
created_at=str(getattr(s, "created_at", "")),
updated_at=str(getattr(s, "updated_at", "")),
)
for s in sessions
]
except Exception:
return []
# -- validations ---------------------------------------------------------
def get_validations(self, plan_id: str) -> list[ValidationResult]:
"""Return validation results (stub - returns empty)."""
_ = plan_id
return []
# -- diffs ---------------------------------------------------------------
def get_diffs(self, plan_id: str) -> list[DiffItem]:
"""Return changeset diffs (stub - returns empty)."""
_ = plan_id
return []
# -- logs ----------------------------------------------------------------
def get_logs(self, plan_id: str, level: str = "INFO") -> list[LogEntry]:
"""Return log entries (stub - returns empty)."""
_ = plan_id
_ = level
return []
# -- internal helpers ----------------------------------------------------
@staticmethod
def _to_summary(plan: object, project_id: str) -> PlanSummary:
return PlanSummary(
plan_id=str(getattr(plan, "id", "") or ""),
name=str(getattr(plan, "name", "") or ""),
status=str(getattr(plan, "status", "") or ""),
created_at=str(getattr(plan, "created_at", "") or ""),
project_id=project_id,
)
@staticmethod
def _to_detail(plan: object) -> PlanDetail:
decisions_raw = getattr(plan, "decisions", []) or []
decisions: list[dict[str, str]] = []
for dec in decisions_raw:
decisions.append(
{
"id": str(getattr(dec, "id", "")),
"type": str(getattr(dec, "type", "")),
"status": str(getattr(dec, "status", "")),
}
)
resources_raw = getattr(plan, "linked_resources", []) or []
linked: list[str] = [str(r) for r in resources_raw]
return PlanDetail(
plan_id=str(getattr(plan, "id", "") or ""),
name=str(getattr(plan, "name", "") or ""),
status=str(getattr(plan, "status", "") or ""),
created_at=str(getattr(plan, "created_at", "") or ""),
project_id=str(getattr(plan, "project_id", "") or ""),
prompt=str(getattr(plan, "prompt", "") or ""),
decisions=decisions,
linked_resources=linked,
)
+321
View File
@@ -0,0 +1,321 @@
"""Minimal Textual-based TUI for CleverAgents.
Provides a dashboard with plan list, plan detail, diff viewer, and
validation summary panes. Navigation uses keyboard shortcuts; data
refreshes on a configurable interval.
Based on Forgejo #341.
"""
from __future__ import annotations
import contextlib
from typing import Any, ClassVar
from textual.app import App, ComposeResult
from textual.binding import Binding, BindingType
from textual.containers import Horizontal, Vertical
from textual.reactive import reactive
from textual.widgets import DataTable, Footer, Header, Static
from cleveragents.ui.data_provider import (
DiffItem,
LocalUIDataProvider,
PlanDetail,
PlanSummary,
UIDataProvider,
ValidationResult,
)
# ---------------------------------------------------------------------------
# Helper widgets
# ---------------------------------------------------------------------------
class PlanListPane(Static):
"""Left-side pane showing a table of plans."""
DEFAULT_CSS = """
PlanListPane {
width: 1fr;
height: 1fr;
border: solid $primary;
}
"""
def __init__(
self,
plans: list[PlanSummary] | None = None,
**kwargs: Any,
) -> None:
super().__init__(**kwargs)
self._plans: list[PlanSummary] = plans or []
def compose(self) -> ComposeResult:
table: DataTable[str] = DataTable()
table.add_columns("ID", "Name", "Status", "Created")
for plan in self._plans:
table.add_row(
plan.plan_id[:12],
plan.name,
plan.status,
plan.created_at[:19],
)
yield table
def refresh_data(self, plans: list[PlanSummary]) -> None:
"""Replace the table contents with fresh data."""
self._plans = plans
with contextlib.suppress(Exception):
table = self.query_one(DataTable)
table.clear()
for plan in plans:
table.add_row(
plan.plan_id[:12],
plan.name,
plan.status,
plan.created_at[:19],
)
class PlanDetailPane(Static):
"""Right-side pane showing plan detail, decisions, and resources."""
DEFAULT_CSS = """
PlanDetailPane {
width: 1fr;
height: 1fr;
border: solid $secondary;
}
"""
def __init__(
self,
detail: PlanDetail | None = None,
**kwargs: Any,
) -> None:
super().__init__(**kwargs)
self._detail = detail
def compose(self) -> ComposeResult:
yield Static(self._render_text(), id="detail-text")
def _render_text(self) -> str:
if self._detail is None:
return "[Plan Detail]\nNo plan selected."
det = self._detail
lines = [
f"[Plan Detail] {det.name}",
f"Status: {det.status}",
f"ID: {det.plan_id}",
f"Created: {det.created_at}",
f"Prompt: {det.prompt[:200]}",
"",
"Decisions:",
]
for dec in det.decisions:
lines.append(
f" - {dec.get('id', '?')} "
f"({dec.get('type', '?')}) "
f"[{dec.get('status', '?')}]"
)
if det.linked_resources:
lines.append("")
lines.append("Linked Resources:")
for res in det.linked_resources:
lines.append(f" - {res}")
return "\n".join(lines)
def refresh_data(self, detail: PlanDetail | None) -> None:
"""Update the displayed detail."""
self._detail = detail
with contextlib.suppress(Exception):
widget = self.query_one("#detail-text", Static)
widget.update(self._render_text())
class DiffViewerPane(Static):
"""Pane showing changeset diffs for the selected plan."""
DEFAULT_CSS = """
DiffViewerPane {
width: 1fr;
height: 1fr;
border: solid $accent;
}
"""
def __init__(
self,
diffs: list[DiffItem] | None = None,
**kwargs: Any,
) -> None:
super().__init__(**kwargs)
self._diffs: list[DiffItem] = diffs or []
def compose(self) -> ComposeResult:
yield Static(self._render_text(), id="diff-text")
def _render_text(self) -> str:
if not self._diffs:
return "[Diffs]\nNo diffs available."
lines = ["[Diffs]"]
for diff in self._diffs:
lines.append(f"--- {diff.file_path} ({diff.operation})")
lines.append(diff.diff_text)
lines.append("")
return "\n".join(lines)
def refresh_data(self, diffs: list[DiffItem]) -> None:
"""Update displayed diffs."""
self._diffs = diffs
with contextlib.suppress(Exception):
widget = self.query_one("#diff-text", Static)
widget.update(self._render_text())
class ValidationPane(Static):
"""Pane showing validation results pass/fail."""
DEFAULT_CSS = """
ValidationPane {
width: 1fr;
height: 1fr;
border: solid $warning;
}
"""
def __init__(
self,
validations: list[ValidationResult] | None = None,
**kwargs: Any,
) -> None:
super().__init__(**kwargs)
self._validations: list[ValidationResult] = validations or []
def compose(self) -> ComposeResult:
yield Static(self._render_text(), id="validation-text")
def _render_text(self) -> str:
if not self._validations:
return "[Validations]\nNo validation results."
lines = ["[Validations]"]
for val in self._validations:
icon = "PASS" if val.passed else "FAIL"
lines.append(f" [{icon}] {val.name}: {val.message}")
return "\n".join(lines)
def refresh_data(self, validations: list[ValidationResult]) -> None:
"""Update displayed validations."""
self._validations = validations
with contextlib.suppress(Exception):
widget = self.query_one("#validation-text", Static)
widget.update(self._render_text())
# ---------------------------------------------------------------------------
# Main TUI application
# ---------------------------------------------------------------------------
class CleverAgentsTUI(App[None]):
"""Minimal TUI dashboard for CleverAgents."""
TITLE = "CleverAgents Dashboard"
CSS_PATH = None # inline CSS only
BINDINGS: ClassVar[list[BindingType]] = [
Binding("q", "quit", "Quit"),
Binding("r", "refresh", "Refresh"),
Binding("1", "focus_plans", "Plans"),
Binding("2", "focus_detail", "Detail"),
Binding("3", "focus_diffs", "Diffs"),
Binding("4", "focus_validations", "Validations"),
]
DEFAULT_CSS = """
Screen {
layout: vertical;
}
#top-row {
height: 1fr;
}
#bottom-row {
height: 1fr;
}
"""
project_id: reactive[str] = reactive("")
def __init__(
self,
provider: UIDataProvider | None = None,
project_id: str = "",
refresh_interval: int = 5,
**kwargs: Any,
) -> None:
super().__init__(**kwargs)
self._provider: UIDataProvider = provider or LocalUIDataProvider()
self.project_id = project_id
self._refresh_interval = max(1, refresh_interval)
def compose(self) -> ComposeResult:
yield Header()
with Vertical():
with Horizontal(id="top-row"):
yield PlanListPane(id="plan-list")
yield PlanDetailPane(id="plan-detail")
with Horizontal(id="bottom-row"):
yield DiffViewerPane(id="diff-viewer")
yield ValidationPane(id="validation-pane")
yield Footer()
def on_mount(self) -> None:
"""Start auto-refresh timer on mount."""
self.action_refresh()
self.set_interval(self._refresh_interval, self.action_refresh)
def action_refresh(self) -> None:
"""Refresh all panes from the data provider."""
plans = self._provider.get_plans(self.project_id)
with contextlib.suppress(Exception):
pane = self.query_one("#plan-list", PlanListPane)
pane.refresh_data(plans)
detail: PlanDetail | None = None
if plans:
detail = self._provider.get_plan_detail(plans[0].plan_id)
with contextlib.suppress(Exception):
det_pane = self.query_one("#plan-detail", PlanDetailPane)
det_pane.refresh_data(detail)
plan_id = plans[0].plan_id if plans else ""
diffs = self._provider.get_diffs(plan_id)
with contextlib.suppress(Exception):
diff_pane = self.query_one("#diff-viewer", DiffViewerPane)
diff_pane.refresh_data(diffs)
validations = self._provider.get_validations(plan_id)
with contextlib.suppress(Exception):
val_pane = self.query_one("#validation-pane", ValidationPane)
val_pane.refresh_data(validations)
def action_focus_plans(self) -> None:
"""Focus the plan list pane."""
with contextlib.suppress(Exception):
self.query_one("#plan-list").focus()
def action_focus_detail(self) -> None:
"""Focus the plan detail pane."""
with contextlib.suppress(Exception):
self.query_one("#plan-detail").focus()
def action_focus_diffs(self) -> None:
"""Focus the diff viewer pane."""
with contextlib.suppress(Exception):
self.query_one("#diff-viewer").focus()
def action_focus_validations(self) -> None:
"""Focus the validation pane."""
with contextlib.suppress(Exception):
self.query_one("#validation-pane").focus()
+134
View File
@@ -0,0 +1,134 @@
"""Read-only web UI stub serving plan data as JSON.
Exposes local-only endpoints consumed by browsers or API clients.
All routes are read-only no mutations are accepted.
Based on Forgejo #341.
"""
from __future__ import annotations
from dataclasses import asdict
from typing import Any
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from cleveragents.ui.data_provider import (
LocalUIDataProvider,
UIDataProvider,
)
# ---------------------------------------------------------------------------
# App factory
# ---------------------------------------------------------------------------
def create_web_app(
provider: UIDataProvider | None = None,
) -> FastAPI:
"""Build the read-only FastAPI application.
Parameters
----------
provider:
Data provider implementation. Defaults to
``LocalUIDataProvider`` backed by the DI container.
Returns
-------
FastAPI
Configured application with ``/ui/*`` routes.
"""
resolved: UIDataProvider = provider or LocalUIDataProvider()
app = FastAPI(
title="CleverAgents Web UI",
description="Read-only local dashboard for CleverAgents",
version="0.1.0",
)
# -- routes --------------------------------------------------------------
@app.get("/ui/plans")
def list_plans(project_id: str = "") -> JSONResponse:
"""Return a JSON array of plan summaries."""
plans = resolved.get_plans(project_id)
return JSONResponse(
content={"plans": [asdict(p) for p in plans]},
)
@app.get("/ui/plans/{plan_id}")
def plan_detail(plan_id: str) -> JSONResponse:
"""Return plan detail as JSON."""
detail = resolved.get_plan_detail(plan_id)
if detail is None:
return JSONResponse(
content={"error": "Plan not found"},
status_code=404,
)
return JSONResponse(content=asdict(detail))
@app.get("/ui/sessions")
def list_sessions() -> JSONResponse:
"""Return a JSON array of sessions."""
sessions = resolved.get_sessions()
return JSONResponse(
content={"sessions": [asdict(s) for s in sessions]},
)
@app.get("/ui/diffs/{plan_id}")
def plan_diffs(plan_id: str) -> JSONResponse:
"""Return changeset diffs for a plan."""
diffs = resolved.get_diffs(plan_id)
return JSONResponse(
content={"diffs": [asdict(d) for d in diffs]},
)
@app.get("/ui/validations/{plan_id}")
def plan_validations(plan_id: str) -> JSONResponse:
"""Return validation results for a plan."""
results = resolved.get_validations(plan_id)
return JSONResponse(
content={
"validations": [asdict(v) for v in results],
},
)
@app.get("/ui/health")
def health() -> dict[str, str]:
"""Simple health check."""
return {"status": "ok"}
return app
# ---------------------------------------------------------------------------
# Convenience runner
# ---------------------------------------------------------------------------
def run_web_ui(
provider: UIDataProvider | None = None,
host: str = "127.0.0.1",
port: int = 8080,
) -> None:
"""Start the web UI server (blocking).
Parameters
----------
provider:
Optional custom data provider.
host:
Bind address. Defaults to ``127.0.0.1`` (local only).
port:
Listen port.
"""
import uvicorn
app = create_web_app(provider)
config: dict[str, Any] = {
"host": host,
"port": port,
"log_level": "info",
}
uvicorn.run(app, **config)
+32
View File
@@ -957,3 +957,35 @@ depth_resolved_count # noqa: B018, F821
dropped_by_overage_guard # noqa: B018, F821
budget_utilization # noqa: B018, F821
fusion_input_count # noqa: B018, F821
# TUI / Web UI (#341)
UIDataProvider # noqa: B018, F821
LocalUIDataProvider # noqa: B018, F821
PlanSummary # noqa: B018, F821
PlanDetail # noqa: B018, F821
SessionSummary # noqa: B018, F821
ValidationResult # noqa: B018, F821
DiffItem # noqa: B018, F821
LogEntry # noqa: B018, F821
CleverAgentsTUI # noqa: B018, F821
PlanListPane # noqa: B018, F821
PlanDetailPane # noqa: B018, F821
DiffViewerPane # noqa: B018, F821
ValidationPane # noqa: B018, F821
create_web_app # noqa: B018, F821
run_web_ui # noqa: B018, F821
tui_refresh_interval # noqa: B018, F821
web_ui_enabled # noqa: B018, F821
web_ui_port # noqa: B018, F821
action_focus_plans # noqa: B018, F821
action_focus_detail # noqa: B018, F821
action_focus_diffs # noqa: B018, F821
action_focus_validations # noqa: B018, F821
action_refresh # noqa: B018, F821
refresh_data # noqa: B018, F821
list_plans # noqa: B018, F821
plan_detail # noqa: B018, F821
list_sessions # noqa: B018, F821
plan_diffs # noqa: B018, F821
plan_validations # noqa: B018, F821
health # noqa: B018, F821