Compare commits

...

2 Commits

Author SHA1 Message Date
freemo 3e93525f4d fix(ui): address TUI/Web review findings
CI / benchmark-publish (pull_request) Has been skipped
CI / build (pull_request) Successful in 25s
CI / lint (pull_request) Successful in 3m25s
CI / security (pull_request) Successful in 3m58s
CI / quality (pull_request) Successful in 4m15s
CI / typecheck (pull_request) Successful in 5m9s
CI / coverage (pull_request) Has started running
CI / benchmark-regression (pull_request) Has started running
CI / integration_tests (pull_request) Successful in 7m45s
CI / unit_tests (pull_request) Successful in 8m0s
CI / docker (pull_request) Successful in 1m21s
CI / e2e_tests (pull_request) Successful in 10m22s
CI / status-check (pull_request) Has been cancelled
- Add uvicorn dependency to pyproject.toml (verified already present)
- Add network binding warning for non-localhost hosts
- Narrow 12 contextlib.suppress(Exception) to specific NoMatches with debug logging
- Change -h flag to -H for --host to avoid --help conflict
- Improve type safety in _ServiceHolder with explicit method calls
- Fix fragile project ID conversion with proper error handling
- Add TODO for CORS middleware
2026-03-25 01:17:49 +00:00
freemo e7df15d453 feat(ui): add TUI/Web interface
Implement UI data-provider interface backed by local services.
Add Textual-based TUI with plan list, plan detail, diff viewer,
and validation summary panes. Add Web UI stub with local-only
read-only routes. Configure auto-refresh and manual keybinds.

ISSUES CLOSED: #341
2026-03-24 20:30:54 +00:00
22 changed files with 2818 additions and 23 deletions
+184
View File
@@ -0,0 +1,184 @@
"""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:
"""Prepare provider with mock plan service."""
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:
"""Benchmark plan list retrieval."""
self._provider.get_plans("proj-1")
def time_get_plan_detail(self) -> None:
"""Benchmark plan detail retrieval."""
self._provider.get_plan_detail("plan-1")
def time_get_validations(self) -> None:
"""Benchmark validation retrieval."""
self._provider.get_validations("plan-1")
def time_get_diffs(self) -> None:
"""Benchmark diff retrieval."""
self._provider.get_diffs("plan-1")
def time_get_logs(self) -> None:
"""Benchmark log retrieval."""
self._provider.get_logs("plan-1")
# ---------------------------------------------------------------------------
# Web route benchmarks
# ---------------------------------------------------------------------------
class TimeWebRoutes:
"""Benchmark FastAPI web UI route handler overhead."""
timeout = 30
def setup(self) -> None:
"""Prepare test client with stub provider."""
from fastapi.testclient import TestClient
from cleveragents.ui.web_app import create_web_app
web = create_web_app(provider=_STUB)
self._client = TestClient(web)
def time_list_plans(self) -> None:
"""Benchmark plans list route."""
self._client.get("/ui/plans")
def time_plan_detail(self) -> None:
"""Benchmark plan detail route."""
self._client.get("/ui/plans/plan-1")
def time_sessions(self) -> None:
"""Benchmark sessions route."""
self._client.get("/ui/sessions")
def time_diffs(self) -> None:
"""Benchmark diffs route."""
self._client.get("/ui/diffs/plan-1")
def time_health(self) -> None:
"""Benchmark health route."""
self._client.get("/ui/health")
+121
View File
@@ -0,0 +1,121 @@
# UI Guide
CleverAgents provides two interface options for monitoring plans, sessions,
diffs, and validation results: a **Textual-based TUI** for terminal use and a
**read-only Web UI** stub that serves JSON endpoints locally.
Both interfaces consume the same `UIDataProvider` protocol, so the data
displayed is identical regardless of which interface is used.
## TUI Dashboard
### Launching
```bash
agents ui tui
```
Optional flags:
| Flag | Description |
|------|-------------|
| `--project-id`, `-p` | Project ID to display (default: empty — show all) |
| `--refresh`, `-r` | Auto-refresh interval in seconds (overrides `tui_refresh_interval` setting) |
### Layout
The TUI is organized into four panes:
| Pane | Position | Content |
|------|----------|---------|
| **Plan List** | Top-left | Table of plans with ID, name, status, created date |
| **Plan Detail** | Top-right | Full detail of the selected plan including decisions and linked resources |
| **Diff Viewer** | Bottom-left | Changeset diffs for the selected plan |
| **Validation Summary** | Bottom-right | Pass/fail validation results |
### Keybindings
| Key | Action |
|-----|--------|
| `q` | Quit the TUI |
| `r` | Manual refresh all panes |
| `1` | Focus the plan list pane |
| `2` | Focus the plan detail pane |
| `3` | Focus the diff viewer pane |
| `4` | Focus the validation summary pane |
### Auto-Refresh
The TUI polls the data provider at a configurable interval. The default is
**5 seconds**. Override with:
- **CLI flag**: `--refresh 10`
- **Environment variable**: `CLEVERAGENTS_TUI_REFRESH_INTERVAL=10`
- **Configuration setting**: `tui_refresh_interval` in the settings file
## Web UI
### Enabling
The web UI is disabled by default. Enable it with:
```bash
export CLEVERAGENTS_WEB_UI_ENABLED=true
```
### Launching
```bash
agents ui web
```
Optional flags:
| Flag | Description |
|------|-------------|
| `--host`, `-h` | Bind address (default: `127.0.0.1` — local only) |
| `--port` | Listen port (overrides `web_ui_port` setting, default: `8080`) |
### Endpoints
All endpoints are **read-only** (GET only).
| Route | Description |
|-------|-------------|
| `GET /ui/plans?project_id=...` | JSON array of plan summaries |
| `GET /ui/plans/{plan_id}` | Plan detail (404 if not found) |
| `GET /ui/sessions` | JSON array of sessions |
| `GET /ui/diffs/{plan_id}` | Changeset diffs for a plan |
| `GET /ui/validations/{plan_id}` | Validation results for a plan |
| `GET /ui/health` | Health check (`{"status": "ok"}`) |
### Security
The web UI binds to `127.0.0.1` by default. **Do not expose it to the
network** — it is intended for local development only and has no
authentication or authorization.
## Configuration Reference
| Setting | Env Var | Default | Description |
|---------|---------|---------|-------------|
| `tui_refresh_interval` | `CLEVERAGENTS_TUI_REFRESH_INTERVAL` | `5` | Auto-refresh interval in seconds |
| `web_ui_enabled` | `CLEVERAGENTS_WEB_UI_ENABLED` | `false` | Enable the web UI server |
| `web_ui_port` | `CLEVERAGENTS_WEB_UI_PORT` | `8080` | Web UI listen port |
## Data Provider Architecture
Both the TUI and web UI consume the `UIDataProvider` protocol defined in
`cleveragents.ui.data_provider`. The default implementation,
`LocalUIDataProvider`, resolves services from the DI container.
The protocol exposes these methods:
- `get_plans(project_id)` — plan summaries
- `get_plan_detail(plan_id)` — full plan detail
- `get_sessions()` — session list
- `get_validations(plan_id)` — validation results
- `get_diffs(plan_id)` — changeset diffs
- `get_logs(plan_id, level)` — log entries
Custom providers can be injected for testing or alternative data sources.
+490
View File
@@ -0,0 +1,490 @@
"""Step definitions for TUI coverage-boost scenarios.
Exercises Textual widget classes, CleverAgentsTUI construction,
data-provider helpers, and CLI import paths to increase branch
and line coverage for the ui package.
"""
from __future__ import annotations
import asyncio
from unittest.mock import MagicMock
from behave import given, then, when
from behave.runner import Context
from cleveragents.ui.data_provider import (
DiffItem,
LocalUIDataProvider,
PlanDetail,
PlanSummary,
UIDataProvider,
ValidationResult,
)
from cleveragents.ui.tui_app import (
CleverAgentsTUI,
DiffViewerPane,
PlanDetailPane,
PlanListPane,
ValidationPane,
)
# ---------------------------------------------------------------------------
# Sample data fixtures
# ---------------------------------------------------------------------------
_SAMPLE_PLANS = [
PlanSummary(
plan_id="plan-1",
name="plan-alpha",
status="draft",
created_at="2025-01-01T00:00:00",
project_id="proj-1",
),
PlanSummary(
plan_id="plan-2",
name="plan-beta",
status="built",
created_at="2025-01-02T00:00:00",
project_id="proj-1",
),
]
_SAMPLE_DETAIL = PlanDetail(
plan_id="plan-1",
name="plan-alpha",
status="draft",
created_at="2025-01-01T00:00:00",
project_id="proj-1",
prompt="Do something useful",
decisions=[{"id": "d1", "type": "approval", "status": "pending"}],
linked_resources=["res-1"],
)
_SAMPLE_DIFFS = [
DiffItem(file_path="file_a.py", operation="modified", diff_text="+line1"),
]
_SAMPLE_VALIDATIONS = [
ValidationResult(name="lint", passed=True, message="ok"),
]
# ---------------------------------------------------------------------------
# PlanListPane steps
# ---------------------------------------------------------------------------
@given("a PlanListPane with no plans")
def step_plan_list_pane_empty(context: Context) -> None:
context.plan_list_pane = PlanListPane(plans=None)
@given("a PlanListPane with sample plans")
def step_plan_list_pane_sample(context: Context) -> None:
context.plan_list_pane = PlanListPane(plans=_SAMPLE_PLANS)
@then("the PlanListPane should be instantiated")
def step_plan_list_pane_ok(context: Context) -> None:
assert context.plan_list_pane is not None
@when("I call refresh_data on PlanListPane with sample plans")
def step_refresh_plan_list(context: Context) -> None:
context.plan_list_pane.refresh_data(_SAMPLE_PLANS)
@then("the PlanListPane internal plans list should have {n:d} entries")
def step_plan_list_count(context: Context, n: int) -> None:
assert len(context.plan_list_pane._plans) == n
# ---------------------------------------------------------------------------
# PlanDetailPane steps
# ---------------------------------------------------------------------------
@given("a PlanDetailPane with no detail")
def step_detail_pane_empty(context: Context) -> None:
context.detail_pane = PlanDetailPane(detail=None)
@given("a PlanDetailPane with sample detail")
def step_detail_pane_sample(context: Context) -> None:
context.detail_pane = PlanDetailPane(detail=_SAMPLE_DETAIL)
@then('the PlanDetailPane render text should contain "{text}"')
def step_detail_pane_render(context: Context, text: str) -> None:
rendered = context.detail_pane._render_text()
assert text in rendered, f"'{text}' not in render: {rendered}"
@when("I call refresh_data on PlanDetailPane with sample detail")
def step_refresh_detail(context: Context) -> None:
context.detail_pane.refresh_data(_SAMPLE_DETAIL)
@then("the PlanDetailPane internal detail should not be None")
def step_detail_not_none(context: Context) -> None:
assert context.detail_pane._detail is not None
# ---------------------------------------------------------------------------
# DiffViewerPane steps
# ---------------------------------------------------------------------------
@given("a DiffViewerPane with no diffs")
def step_diff_pane_empty(context: Context) -> None:
context.diff_pane = DiffViewerPane(diffs=None)
@given("a DiffViewerPane with sample diffs")
def step_diff_pane_sample(context: Context) -> None:
context.diff_pane = DiffViewerPane(diffs=_SAMPLE_DIFFS)
@then('the DiffViewerPane render text should contain "{text}"')
def step_diff_pane_render(context: Context, text: str) -> None:
rendered = context.diff_pane._render_text()
assert text in rendered, f"'{text}' not in render: {rendered}"
@when("I call refresh_data on DiffViewerPane with sample diffs")
def step_refresh_diffs(context: Context) -> None:
context.diff_pane.refresh_data(_SAMPLE_DIFFS)
@then("the DiffViewerPane internal diffs list should have {n:d} entry")
def step_diff_count(context: Context, n: int) -> None:
assert len(context.diff_pane._diffs) == n
# ---------------------------------------------------------------------------
# ValidationPane steps
# ---------------------------------------------------------------------------
@given("a ValidationPane with no validations")
def step_val_pane_empty(context: Context) -> None:
context.val_pane = ValidationPane(validations=None)
@given("a ValidationPane with sample validations")
def step_val_pane_sample(context: Context) -> None:
context.val_pane = ValidationPane(validations=_SAMPLE_VALIDATIONS)
@then('the ValidationPane render text should contain "{text}"')
def step_val_pane_render(context: Context, text: str) -> None:
rendered = context.val_pane._render_text()
assert text in rendered, f"'{text}' not in render: {rendered}"
@when("I call refresh_data on ValidationPane with sample validations")
def step_refresh_validations(context: Context) -> None:
context.val_pane.refresh_data(_SAMPLE_VALIDATIONS)
@then("the ValidationPane internal validations list should have {n:d} entry")
def step_val_count(context: Context, n: int) -> None:
assert len(context.val_pane._validations) == n
# ---------------------------------------------------------------------------
# CleverAgentsTUI steps
# ---------------------------------------------------------------------------
class _StubProvider:
"""Minimal provider for TUI construction tests."""
def get_plans(self, project_id: str) -> list[PlanSummary]:
return []
def get_plan_detail(self, plan_id: str) -> PlanDetail | None:
return None
def get_sessions(self):
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"):
return []
@given("a CleverAgentsTUI with default provider")
def step_tui_default(context: Context) -> None:
context.tui_app = CleverAgentsTUI(provider=_StubProvider())
@given('a CleverAgentsTUI with a mock provider and project "{pid}"')
def step_tui_custom(context: Context, pid: str) -> None:
context.tui_app = CleverAgentsTUI(
provider=_StubProvider(),
project_id=pid,
refresh_interval=2,
)
@then('the TUI app title should be "{title}"')
def step_tui_title(context: Context, title: str) -> None:
assert title == context.tui_app.TITLE
@then('the TUI app project_id should be "{pid}"')
def step_tui_project_id(context: Context, pid: str) -> None:
assert context.tui_app.project_id == pid
@then('the TUI BINDINGS should contain keys "{k1}" and "{k2}"')
def step_tui_bindings(context: Context, k1: str, k2: str) -> None:
keys = {b.key if hasattr(b, "key") else b[0] for b in context.tui_app.BINDINGS}
assert k1 in keys, f"'{k1}' not in bindings keys {keys}"
assert k2 in keys, f"'{k2}' not in bindings keys {keys}"
# ---------------------------------------------------------------------------
# UIDataProvider protocol steps
# ---------------------------------------------------------------------------
@given("the UIDataProvider protocol")
def step_protocol(context: Context) -> None:
context.ui_protocol = UIDataProvider
@then("a mock provider should satisfy the protocol check")
def step_protocol_check(context: Context) -> None:
assert isinstance(_StubProvider(), context.ui_protocol)
# ---------------------------------------------------------------------------
# LocalUIDataProvider helper steps
# ---------------------------------------------------------------------------
@given("a raw mock plan object")
def step_raw_mock_plan(context: Context) -> None:
p = MagicMock()
p.id = "p-1"
p.name = "mock-plan"
p.status = "draft"
p.created_at = "2025-06-01T00:00:00"
context.raw_plan = p
@given("a raw mock plan object with decisions")
def step_raw_mock_plan_dec(context: Context) -> None:
dec = MagicMock()
dec.id = "d-1"
dec.type = "approval"
dec.status = "pending"
p = MagicMock()
p.id = "p-1"
p.name = "mock-plan"
p.status = "draft"
p.created_at = "2025-06-01T00:00:00"
p.prompt = "test prompt"
p.decisions = [dec]
p.linked_resources = ["r-1"]
p.project_id = "proj-1"
context.raw_plan = p
@when('I call _to_summary with project_id "{pid}"')
def step_to_summary(context: Context, pid: str) -> None:
context.plan_summary = LocalUIDataProvider._to_summary(context.raw_plan, pid)
@then('the PlanSummary project_id should be "{pid}"')
def step_summary_pid(context: Context, pid: str) -> None:
assert context.plan_summary.project_id == pid
@when("I call _to_detail on the plan")
def step_to_detail(context: Context) -> None:
context.plan_detail_result = LocalUIDataProvider._to_detail(context.raw_plan)
@then("the PlanDetail should contain {n:d} decision")
def step_detail_decisions(context: Context, n: int) -> None:
assert len(context.plan_detail_result.decisions) == n
# ---------------------------------------------------------------------------
# run_web_ui / CLI import steps
# ---------------------------------------------------------------------------
@given("the run_web_ui function is imported")
def step_import_run_web_ui(context: Context) -> None:
from cleveragents.ui.web_app import run_web_ui
context.run_web_ui = run_web_ui
@then("run_web_ui should be callable")
def step_run_web_ui_callable(context: Context) -> None:
assert callable(context.run_web_ui)
@given("the CLI ui app is imported")
def step_import_cli_ui(context: Context) -> None:
from cleveragents.cli.commands.ui import app as ui_app
context.cli_ui_app = ui_app
@then('the ui app should have a "{cmd}" command')
def step_ui_app_cmd(context: Context, cmd: str) -> None:
# Typer stores registered commands/groups in .registered_groups and
# .registered_commands.
names: set[str] = set()
for info in getattr(context.cli_ui_app, "registered_commands", []):
names.add(info.name or info.callback.__name__)
for info in getattr(context.cli_ui_app, "registered_groups", []):
names.add(info.name or "")
assert cmd in names, f"'{cmd}' not in {names}"
# ---------------------------------------------------------------------------
# Headless TUI test steps (exercises compose / on_mount / action_refresh)
# ---------------------------------------------------------------------------
class _PopulatedProvider:
"""Provider returning real data for headless TUI tests."""
def get_plans(self, project_id: str) -> list[PlanSummary]:
return [
PlanSummary("p1", "alpha", "draft", "2025-01-01", "proj"),
]
def get_plan_detail(self, plan_id: str) -> PlanDetail | None:
return PlanDetail(
"p1",
"alpha",
"draft",
"2025-01-01",
"proj",
"prompt",
[{"id": "d1", "type": "t", "status": "s"}],
["r1"],
)
def get_sessions(self):
return []
def get_validations(self, plan_id: str) -> list[ValidationResult]:
return [ValidationResult("lint", True, "ok")]
def get_diffs(self, plan_id: str) -> list[DiffItem]:
return [DiffItem("a.py", "modified", "+x")]
def get_logs(self, plan_id: str, level: str = "INFO"):
return []
@given("a headless TUI app with a populated provider")
def step_headless_tui(context: Context) -> None:
context.headless_provider = _PopulatedProvider()
context.headless_tui = CleverAgentsTUI(
provider=context.headless_provider,
project_id="proj",
refresh_interval=1,
)
context.headless_ok = False
context.headless_dom_widgets: dict[str, bool] = {}
@when("I run the TUI headless test with refresh")
def step_run_headless_refresh(context: Context) -> None:
async def _run() -> None:
async with context.headless_tui.run_test(
headless=True, size=(120, 40)
) as pilot:
await pilot.pause()
await pilot.press("r")
await pilot.pause()
# Record which widgets exist
from cleveragents.ui.tui_app import (
DiffViewerPane as DVP,
)
from cleveragents.ui.tui_app import (
PlanDetailPane as PDP,
)
from cleveragents.ui.tui_app import (
PlanListPane as PLP,
)
from cleveragents.ui.tui_app import (
ValidationPane as VP,
)
for cls, key in [
(PLP, "PlanListPane"),
(PDP, "PlanDetailPane"),
(DVP, "DiffViewerPane"),
(VP, "ValidationPane"),
]:
try:
context.headless_tui.query_one(cls)
context.headless_dom_widgets[key] = True
except Exception:
context.headless_dom_widgets[key] = False
context.headless_ok = True
asyncio.run(_run())
@when("I press focus keys 1 2 3 4 in headless mode")
def step_headless_focus_keys(context: Context) -> None:
async def _run() -> None:
async with context.headless_tui.run_test(
headless=True, size=(120, 40)
) as pilot:
await pilot.pause()
for key in ("1", "2", "3", "4"):
await pilot.press(key)
await pilot.pause()
context.headless_ok = True
asyncio.run(_run())
@then("the headless TUI test should complete successfully")
def step_headless_ok(context: Context) -> None:
assert context.headless_ok, "Headless TUI test did not complete"
@then("the PlanListPane widget should exist in the DOM")
def step_dom_plan_list(context: Context) -> None:
assert context.headless_dom_widgets.get("PlanListPane"), "PlanListPane not found"
@then("the PlanDetailPane widget should exist in the DOM")
def step_dom_plan_detail(context: Context) -> None:
assert context.headless_dom_widgets.get("PlanDetailPane"), (
"PlanDetailPane not found"
)
@then("the DiffViewerPane widget should exist in the DOM")
def step_dom_diff_viewer(context: Context) -> None:
assert context.headless_dom_widgets.get("DiffViewerPane"), (
"DiffViewerPane not found"
)
@then("the ValidationPane widget should exist in the DOM")
def step_dom_validation(context: Context) -> None:
assert context.headless_dom_widgets.get("ValidationPane"), (
"ValidationPane not found"
)
+353
View File
@@ -0,0 +1,353 @@
"""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
web = create_web_app(provider=_MockProvider())
context.web_ui_client = TestClient(web)
@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
web = create_web_app(provider=_MockProvider(with_plans=True))
context.web_ui_client = TestClient(web)
@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
_ = name
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
+124
View File
@@ -0,0 +1,124 @@
Feature: TUI coverage boost
Coverage-boost scenarios exercising Textual widget classes,
the CLI tui subcommand path, data-provider protocol methods,
and the run_web_ui helper.
Scenario: PlanListPane renders empty table
Given a PlanListPane with no plans
Then the PlanListPane should be instantiated
Scenario: PlanListPane renders with plans
Given a PlanListPane with sample plans
Then the PlanListPane should be instantiated
Scenario: PlanListPane refresh_data replaces rows
Given a PlanListPane with no plans
When I call refresh_data on PlanListPane with sample plans
Then the PlanListPane internal plans list should have 2 entries
Scenario: PlanDetailPane renders no detail
Given a PlanDetailPane with no detail
Then the PlanDetailPane render text should contain "No plan selected"
Scenario: PlanDetailPane renders with detail
Given a PlanDetailPane with sample detail
Then the PlanDetailPane render text should contain "plan-alpha"
Scenario: PlanDetailPane refresh_data updates detail
Given a PlanDetailPane with no detail
When I call refresh_data on PlanDetailPane with sample detail
Then the PlanDetailPane internal detail should not be None
Scenario: DiffViewerPane renders empty
Given a DiffViewerPane with no diffs
Then the DiffViewerPane render text should contain "No diffs"
Scenario: DiffViewerPane renders with diffs
Given a DiffViewerPane with sample diffs
Then the DiffViewerPane render text should contain "file_a.py"
Scenario: DiffViewerPane refresh_data updates diffs
Given a DiffViewerPane with no diffs
When I call refresh_data on DiffViewerPane with sample diffs
Then the DiffViewerPane internal diffs list should have 1 entry
Scenario: ValidationPane renders empty
Given a ValidationPane with no validations
Then the ValidationPane render text should contain "No validation"
Scenario: ValidationPane renders with validations
Given a ValidationPane with sample validations
Then the ValidationPane render text should contain "PASS"
Scenario: ValidationPane refresh_data updates validations
Given a ValidationPane with no validations
When I call refresh_data on ValidationPane with sample validations
Then the ValidationPane internal validations list should have 1 entry
Scenario: CleverAgentsTUI can be constructed with defaults
Given a CleverAgentsTUI with default provider
Then the TUI app title should be "CleverAgents Dashboard"
Scenario: CleverAgentsTUI can be constructed with custom provider
Given a CleverAgentsTUI with a mock provider and project "proj-1"
Then the TUI app project_id should be "proj-1"
Scenario: CleverAgentsTUI bindings include quit and refresh
Given a CleverAgentsTUI with default provider
Then the TUI BINDINGS should contain keys "q" and "r"
Scenario: UIDataProvider protocol is runtime-checkable
Given the UIDataProvider protocol
Then a mock provider should satisfy the protocol check
Scenario: LocalUIDataProvider _to_summary helper
Given a raw mock plan object
When I call _to_summary with project_id "proj-x"
Then the PlanSummary project_id should be "proj-x"
Scenario: LocalUIDataProvider _to_detail helper
Given a raw mock plan object with decisions
When I call _to_detail on the plan
Then the PlanDetail should contain 1 decision
Scenario: run_web_ui is importable
Given the run_web_ui function is imported
Then run_web_ui should be callable
Scenario: CLI tui subcommand is importable
Given the CLI ui app is imported
Then the ui app should have a "tui" command
Scenario: CLI web subcommand is importable
Given the CLI ui app is imported
Then the ui app should have a "web" command
Scenario: CleverAgentsTUI compose and mount in headless mode
Given a headless TUI app with a populated provider
When I run the TUI headless test with refresh
Then the headless TUI test should complete successfully
Scenario: CleverAgentsTUI focus actions via keybindings
Given a headless TUI app with a populated provider
When I press focus keys 1 2 3 4 in headless mode
Then the headless TUI test should complete successfully
Scenario: PlanListPane compose yields DataTable
Given a headless TUI app with a populated provider
When I run the TUI headless test with refresh
Then the PlanListPane widget should exist in the DOM
Scenario: PlanDetailPane compose and render with detail
Given a headless TUI app with a populated provider
When I run the TUI headless test with refresh
Then the PlanDetailPane widget should exist in the DOM
Scenario: DiffViewerPane compose with diff content
Given a headless TUI app with a populated provider
When I run the TUI headless test with refresh
Then the DiffViewerPane widget should exist in the DOM
Scenario: ValidationPane compose with validation content
Given a headless TUI app with a populated provider
When I run the TUI headless test with refresh
Then the ValidationPane widget should exist in the DOM
+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 "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 "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"
+1 -1
View File
@@ -539,7 +539,7 @@ def serve_docs(session: nox.Session):
@nox.session(python=SUPPORTED_PYTHONS, reuse_venv=True, venv_backend="uv")
def build(session: nox.Session):
"""Build the wheel distribution."""
session.install("build")
session.install("pip", "build")
session.run("python", "-m", "build", "--wheel")
+2
View File
@@ -47,6 +47,8 @@ dependencies = [
"jsonschema>=4.20.0", # JSON Schema validation for tool inputs/outputs
"tomlkit>=0.13.0", # TOML writing with comment preservation for config CLI
"tenacity>=8.2.0", # Retry framework for service layer resilience
"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]
+281
View File
@@ -0,0 +1,281 @@
"""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 = "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("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("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 = "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="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="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
web = create_web_app(provider=_MockProvider())
client = TestClient(web)
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
web = create_web_app(provider=_MockProvider(with_plans=True))
client = TestClient(web)
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
web = create_web_app(provider=_MockProvider())
client = TestClient(web)
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
web = create_web_app(provider=_MockProvider())
client = TestClient(web)
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
web = create_web_app(provider=_MockProvider())
client = TestClient(web)
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:
"""Dispatch subcommand from argv."""
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
+4 -2
View File
@@ -1,6 +1,8 @@
*** Settings ***
Library Process
Library String
Documentation Smoke tests for TUI headless mode, input router, and prompt widget
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Test Cases ***
TUI Headless Startup Check Returns JSON
+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
@@ -101,6 +101,7 @@ def _register_subcommands() -> None:
from cleveragents.cli.commands.db import app as db_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
@@ -208,6 +209,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",
)
app.add_typer(
repo.app,
name="repo",
@@ -688,6 +694,7 @@ def main(args: list[str] | None = None) -> int:
"repl", # Interactive REPL
"tui", # Textual TUI
"server", # Server connection management
"ui", # TUI dashboard / web UI (#341)
"repo", # Repository indexing management
"tell", # Shortcut for plan tell
"build", # Shortcut for plan build
+20
View File
@@ -564,6 +564,26 @@ class Settings(BaseSettings):
)
return self
# 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,
@@ -259,14 +259,14 @@ class SandboxStrategyRegistry:
# ------------------------------------------------------------------
@staticmethod
def _validate_protocol(cls: type[Any]) -> None:
def _validate_protocol(klass: type[Any]) -> None:
"""Validate that a class satisfies :class:`SandboxStrategyProtocol`.
Uses structural subtyping: checks that all 9 required methods
are present as callable attributes on the class.
Args:
cls: The class to validate.
klass: The class to validate.
Raises:
ProtocolMismatchError: If the class is missing required methods.
@@ -286,12 +286,12 @@ class SandboxStrategyRegistry:
missing = [
method
for method in required_methods
if not callable(getattr(cls, method, None))
if not callable(getattr(klass, method, None))
]
if missing:
msg = (
f"Class '{cls.__name__}' does not satisfy "
f"Class '{klass.__name__}' does not satisfy "
f"SandboxStrategyProtocol. Missing methods: "
f"{', '.join(missing)}"
)
+85 -16
View File
@@ -7,21 +7,15 @@ from dataclasses import dataclass
from typing import Any
def _load_input_base() -> type[Any]:
def _get_textual_input() -> type[Any] | None:
"""Return the ``textual.widgets.Input`` class, or ``None``."""
try:
return importlib.import_module("textual.widgets").Input
except Exception: # pragma: no cover
class _FallbackInput:
value = ""
def __init__(self, *args: object, **kwargs: object) -> None:
self.value = ""
return _FallbackInput
return None
_InputBase = _load_input_base()
_TextualInput: type[Any] | None = _get_textual_input()
@dataclass(slots=True, frozen=True)
@@ -31,10 +25,85 @@ class PromptSubmitted:
text: str
class PromptInput(_InputBase):
"""Input widget wrapper with helper methods."""
def _has_active_app() -> bool:
"""Return True when a Textual App is mounted and accessible."""
try:
from textual._context import active_app
def consume_text(self) -> PromptSubmitted:
text = self.value
self.value = ""
return PromptSubmitted(text=text)
active_app.get()
except Exception:
return False
else:
return True
if _TextualInput is not None:
# Textual is installed - subclass the real Input widget.
class PromptInput(_TextualInput): # type: ignore[misc]
"""Input widget wrapper with helper methods.
When instantiated outside a running Textual application (e.g. in
unit/robot tests) the widget skips the Textual super().__init__
and stores ``value`` as a plain instance attribute so that
:meth:`consume_text` works without an active app context.
"""
# Marker so we know which init path was taken.
_headless: bool
def __init__(self, *args: Any, **kwargs: Any) -> None:
if _has_active_app():
super().__init__(*args, **kwargs)
self._headless = False
else:
# Bypass Textual's DOMNode init entirely - the reactive
# descriptors require a mounted app. Instead we store
# ``value`` in a private slot and expose it via a
# property override that shadows the class descriptor
# when ``_headless`` is True.
self._headless = True
self._plain_value: str = ""
# -- headless value access -----------------------------------------
# Python data-descriptor (``__get__`` + ``__set__``) priority means
# the inherited ``reactive`` descriptor will intercept normal
# attribute access even when a value lives in ``__dict__``. We
# therefore override the property on the *class* only when the
# instance is in headless mode, falling through to ``super()``
# when inside a real Textual app.
@property # type: ignore[override]
def value(self) -> str: # type: ignore[override]
if self._headless:
return self._plain_value
# Inside a Textual app - delegate to the reactive descriptor
# on the parent class.
return super().value # type: ignore[return-value]
@value.setter
def value(self, new: str) -> None:
if self._headless:
self._plain_value = new
else:
# Delegate to the reactive descriptor's __set__.
_TextualInput.value.__set__(self, new) # type: ignore[union-attr]
def consume_text(self) -> PromptSubmitted:
"""Return the current text and clear the input."""
text = self.value
self.value = ""
return PromptSubmitted(text=text)
else:
# Textual is not installed - lightweight fallback.
class PromptInput: # type: ignore[no-redef]
"""Fallback PromptInput when Textual is not installed."""
def __init__(self, *args: Any, **kwargs: Any) -> None:
self.value: str = ""
def consume_text(self) -> PromptSubmitted:
text = self.value
self.value = ""
return PromptSubmitted(text=text)
+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",
]
+316
View File
@@ -0,0 +1,316 @@
"""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 logging
from dataclasses import dataclass, field
from typing import Protocol, runtime_checkable
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# 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:
"""Lazily resolve and return the plan service."""
self._resolve()
assert self._plan_service is not None
return self._plan_service
@property
def project_service(self) -> object:
"""Lazily resolve and return the project service."""
self._resolve()
assert self._project_service is not None
return self._project_service
def list_plans(self, project: object) -> list[object]:
"""Call ``plan_service.list_plans(project)`` with type-safe dispatch."""
svc = self.plan_service
method = getattr(svc, "list_plans", None)
if method is None:
logger.debug("_ServiceHolder: plan_service has no list_plans method")
return []
result: list[object] = method(project)
return result
def get_plan(self, plan_id: str) -> object | None:
"""Call ``plan_service.get_plan(plan_id)`` with type-safe dispatch."""
svc = self.plan_service
method = getattr(svc, "get_plan", None)
if method is None:
logger.debug("_ServiceHolder: plan_service has no get_plan method")
return None
return method(plan_id)
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:
# 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
if project_id:
try:
int_id = int(project_id)
except (ValueError, TypeError):
logger.warning(
"get_plans: non-numeric project_id=%r, returning empty",
project_id,
)
return []
project = Project(name="stub", path=_Path.cwd())
project.id = int_id
raw_plans = self._holder.list_plans(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:
raw = self._holder.get_plan(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,
)
+350
View File
@@ -0,0 +1,350 @@
"""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 logging
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.css.query import NoMatches
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,
)
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# 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
try:
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],
)
except NoMatches:
logger.debug("PlanListPane: DataTable widget not found during refresh")
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
try:
widget = self.query_one("#detail-text", Static)
widget.update(self._render_text())
except NoMatches:
logger.debug("PlanDetailPane: #detail-text widget not found during refresh")
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
try:
widget = self.query_one("#diff-text", Static)
widget.update(self._render_text())
except NoMatches:
logger.debug("DiffViewerPane: #diff-text widget not found during refresh")
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
try:
widget = self.query_one("#validation-text", Static)
widget.update(self._render_text())
except NoMatches:
logger.debug(
"ValidationPane: #validation-text widget not found during refresh"
)
# ---------------------------------------------------------------------------
# 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)
try:
pane = self.query_one("#plan-list", PlanListPane)
pane.refresh_data(plans)
except NoMatches:
logger.debug("action_refresh: #plan-list pane not found")
detail: PlanDetail | None = None
if plans:
detail = self._provider.get_plan_detail(plans[0].plan_id)
try:
det_pane = self.query_one("#plan-detail", PlanDetailPane)
det_pane.refresh_data(detail)
except NoMatches:
logger.debug("action_refresh: #plan-detail pane not found")
plan_id = plans[0].plan_id if plans else ""
diffs = self._provider.get_diffs(plan_id)
try:
diff_pane = self.query_one("#diff-viewer", DiffViewerPane)
diff_pane.refresh_data(diffs)
except NoMatches:
logger.debug("action_refresh: #diff-viewer pane not found")
validations = self._provider.get_validations(plan_id)
try:
val_pane = self.query_one("#validation-pane", ValidationPane)
val_pane.refresh_data(validations)
except NoMatches:
logger.debug("action_refresh: #validation-pane pane not found")
def action_focus_plans(self) -> None:
"""Focus the plan list pane."""
try:
self.query_one("#plan-list").focus()
except NoMatches:
logger.debug("action_focus_plans: #plan-list not found")
def action_focus_detail(self) -> None:
"""Focus the plan detail pane."""
try:
self.query_one("#plan-detail").focus()
except NoMatches:
logger.debug("action_focus_detail: #plan-detail not found")
def action_focus_diffs(self) -> None:
"""Focus the diff viewer pane."""
try:
self.query_one("#diff-viewer").focus()
except NoMatches:
logger.debug("action_focus_diffs: #diff-viewer not found")
def action_focus_validations(self) -> None:
"""Focus the validation pane."""
try:
self.query_one("#validation-pane").focus()
except NoMatches:
logger.debug("action_focus_validations: #validation-pane not found")
+145
View File
@@ -0,0 +1,145 @@
"""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
import logging
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,
)
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# 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()
# TODO: Add CORSMiddleware if frontend makes cross-origin API calls
web = FastAPI(
title="CleverAgents Web UI",
description="Read-only local dashboard for CleverAgents",
version="0.1.0",
)
# -- routes --------------------------------------------------------------
@web.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]},
)
@web.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))
@web.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]},
)
@web.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]},
)
@web.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],
},
)
@web.get("/ui/health")
def health() -> dict[str, str]:
"""Simple health check."""
return {"status": "ok"}
return web
# ---------------------------------------------------------------------------
# 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
if host not in ("127.0.0.1", "localhost", "::1"):
logger.warning(
"web_ui.network_bind: Web UI bound to network "
"address without authentication (host=%s)",
host,
)
web = create_web_app(provider)
config: dict[str, Any] = {
"host": host,
"port": port,
"log_level": "info",
}
uvicorn.run(web, **config)
+30
View File
@@ -1190,3 +1190,33 @@ detect_directory_languages # noqa: B018, F821
get_servers_for_language # noqa: B018, F821
restart_server # noqa: B018, F821
_make_runtime_handler # 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
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