cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

feat(cli): extend Diagnostic Dashboard with 5 health categories (index, plans, cost, performance) #791

Closed
brent.edwards wants to merge 2 commits from feature/m6-diagnostic-dashboard-health-categories into master
pull from: feature/m6-diagnostic-dashboard-health-categories
merge into: cleveragents:master
Conversation 17 Commits 2 Files Changed 26
+3003 -309
26 changed files with 3003 additions and 309 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+4
View File
@@ -2,6 +2,9 @@
## Unreleased
- Extended Diagnostic Dashboard with 5 health categories: index health,
active plan overview, cost summary with budget utilization, and
performance metrics. Split system.py into 4 modules. (#580)
- Added TDD bug-capture E2E tests for bug #1028 — ACMS indexing pipeline not
wired into CLI. Four Robot Framework E2E tests prove ContextTierService starts
empty on every CLI invocation. Tests use ``@tdd_expected_fail`` until the bug
@@ -102,6 +105,7 @@
- Added tool-level execution environment preferences with NONE, REQUIRED,
PREFERRED, and SPECIFIC modes. ToolRunner routes tool execution based on
preference mode with caller-override precedence. (#879)
- Added TDD bug-capture tests for #969 — `plan correct` expects `decision_id`
but M3 acceptance test passes `plan_id`. Behave BDD scenarios (revert and
append modes) and Robot Framework integration tests verify that
docs/reference/diagnostics_checks.md
+121
View File
@@ -94,6 +94,127 @@ Checks that the `git` binary is available on `PATH`.
| ok | `git --version` succeeds | None required |
| error | `git` is not found or times out | Install git for version control integration |
### Stale locks
Checks for stale concurrency locks in the lock database.
| Status | Condition | Remediation |
|--------|-----------|-------------|
| ok | No stale locks found | None required |
| warn | Stale locks detected | Investigate orphaned processes; restart the service |
### Async workers
Checks async worker configuration and health.
| Status | Condition | Remediation |
|--------|-----------|-------------|
| ok | Async disabled or properly configured | None required |
| warn | Unable to check configuration | Review settings |
### Error Pattern DB
Checks the Error Pattern Database for recorded patterns.
| Status | Condition | Remediation |
|--------|-----------|-------------|
| ok | Patterns accessible (or empty) | None required |
---
## Category 2: Index health
### Text index
Reports the number of documents in the text index backend.
| Status | Condition | Remediation |
|--------|-----------|-------------|
| ok | Backend accessible, reports document count | None required |
| warn | Backend unavailable (N/A) | Check container wiring or index backend configuration |
### Vector index
Reports the embedding count and dimensionality of the vector index.
| Status | Condition | Remediation |
|--------|-----------|-------------|
| ok | Backend accessible, reports embedding count and dimensions | None required |
| warn | Backend unavailable (N/A) | Check container wiring or index backend configuration |
### Graph store
Reports the number of triples stored in the graph backend.
| Status | Condition | Remediation |
|--------|-----------|-------------|
| ok | Backend accessible, reports triple count | None required |
| warn | Backend unavailable (N/A) | Check container wiring or graph backend configuration |
---
## Category 3: Active plan overview
### Active plans
Summary of running, queued, errored, and total plans.
| Status | Condition | Remediation |
|--------|-----------|-------------|
| ok | Plan service accessible | None required |
| warn | Plan service unavailable (N/A) | Check DI container wiring |
When plans are actively running, additional per-plan checks are emitted
showing the plan name, current phase, and subplan count.
---
## Category 4: Cost summary
### Cost summary
Aggregated cost across all tracked sessions with budget utilisation.
| Status | Condition | Remediation |
|--------|-----------|-------------|
| ok | Cost service accessible (idle or with data) | None required |
| warn | Cost service unavailable (N/A) | Check DI container wiring |
When provider-level cost breakdowns are available, an additional
**Provider costs** check is emitted.
---
## Category 5: Performance summary
### Performance: plan duration
Average plan duration computed from completed plans.
| Status | Condition | Remediation |
|--------|-----------|-------------|
| ok | Metric available or N/A (no completed plans) | None required |
### Performance: tool call latency
Latency percentiles (p50/p95/p99) computed from LLM trace data.
| Status | Condition | Remediation |
|--------|-----------|-------------|
| ok | Metric available or N/A (no trace data) | None required |
### Performance: context build time
Average context build time. Currently reports N/A (metric pipeline
not yet wired).
| Status | Condition | Remediation |
|--------|-----------|-------------|
| ok | Metric available or N/A | None required |
When the performance services are entirely unavailable, a single
**Performance summary** check with N/A status is emitted instead.
---
## Summary counts
features/diagnostic_dashboard_extended.feature
+187
View File
@@ -0,0 +1,187 @@
Feature: Extended Diagnostic Dashboard with 5 health categories
As a user of CleverAgents
I want the diagnostics command to cover all 5 spec health categories
So that I get a complete health overview (system, index, plans, cost, performance)
# -- Index health (category 2) --
Scenario: Diagnostics include text index health
When I run the extended diagnostics command with format "json"
Then the extended diagnostics checks should include "Text index"
Scenario: Diagnostics include vector index health
When I run the extended diagnostics command with format "json"
Then the extended diagnostics checks should include "Vector index"
Scenario: Diagnostics include graph store health
When I run the extended diagnostics command with format "json"
Then the extended diagnostics checks should include "Graph store"
Scenario: Text index shows document count
When I run the extended diagnostics with mock text index having 5 documents
Then the extended diagnostics check "Text index" should have details containing "5 documents"
Scenario: Vector index shows embedding count and dimensionality
When I run the extended diagnostics with mock vector index having 3 embeddings of dimension 128
Then the extended diagnostics check "Vector index" should have details containing "3 embeddings"
And the extended diagnostics check "Vector index" should have details containing "dimensionality=128"
Scenario: Graph store shows triple count
When I run the extended diagnostics with mock graph index having 10 triples
Then the extended diagnostics check "Graph store" should have details containing "10 triples"
Scenario: Index health degrades gracefully when backends unavailable
When I run the extended diagnostics with unavailable index backends
Then the extended diagnostics check "Text index" should have details containing "N/A"
And the extended diagnostics check "Vector index" should have details containing "N/A"
And the extended diagnostics check "Graph store" should have details containing "N/A"
# -- Active plan overview (category 3) --
Scenario: Diagnostics include active plans overview
When I run the extended diagnostics command with format "json"
Then the extended diagnostics checks should include "Active plans"
Scenario: Active plans shows running and queued counts
When I run the extended diagnostics with 2 running and 3 queued plans
Then the extended diagnostics check "Active plans" should have details containing "2 running"
And the extended diagnostics check "Active plans" should have details containing "3 queued"
Scenario: Active plans shows per-plan details for running plans
When I run the extended diagnostics with 1 running plan named "refactor"
Then the extended diagnostics checks should include "Plan: refactor"
Scenario: Active plans degrades gracefully when service unavailable
When I run the extended diagnostics with unavailable plan service
Then the extended diagnostics check "Active plans" should have details containing "N/A"
# -- Cost summary (category 4) --
Scenario: Diagnostics include cost summary
When I run the extended diagnostics command with format "json"
Then the extended diagnostics checks should include "Cost summary"
Scenario: Cost summary shows total cost and session count
When I run the extended diagnostics with session cost of 1.5 across 2 sessions
Then the extended diagnostics check "Cost summary" should have details containing "$1.50"
And the extended diagnostics check "Cost summary" should have details containing "2 sessions"
Scenario: Cost summary shows idle state when no sessions
When I run the extended diagnostics with no active sessions
Then the extended diagnostics check "Cost summary" should have details containing "no active sessions"
Scenario: Cost summary degrades gracefully when service unavailable
When I run the extended diagnostics with unavailable cost service
Then the extended diagnostics check "Cost summary" should have details containing "N/A"
# -- Performance summary (category 5) --
Scenario: Diagnostics include performance plan duration
When I run the extended diagnostics command with format "json"
Then the extended diagnostics checks should include "Performance: plan duration"
Scenario: Diagnostics include performance tool call latency
When I run the extended diagnostics command with format "json"
Then the extended diagnostics checks should include "Performance: tool call latency"
Scenario: Diagnostics include performance context build time
When I run the extended diagnostics command with format "json"
Then the extended diagnostics checks should include "Performance: context build time"
Scenario: Performance shows N/A when no data
When I run the extended diagnostics with no plans
Then the extended diagnostics check "Performance: plan duration" should have details containing "N/A"
And the extended diagnostics check "Performance: tool call latency" should have details containing "N/A"
Scenario: Performance degrades gracefully when service unavailable
When I run the extended diagnostics with unavailable performance services
Then the extended diagnostics check "Performance summary" should have details containing "N/A"
# -- Cost budget utilisation edge cases (M7) --
Scenario: Cost summary shows budget utilisation percentage for budgeted sessions
When I run the extended diagnostics with budgeted sessions costing 0.5 of 1.0 max
Then the extended diagnostics check "Cost summary" should have details containing "budget utilisation=50.0%"
Scenario: Cost summary shows N/A utilisation when no sessions have budgets
When I run the extended diagnostics with session cost of 1.5 across 2 sessions
Then the extended diagnostics check "Cost summary" should have details containing "budget utilisation=N/A"
# -- Context build time edge case (M7) --
Scenario: Performance context build time reports not yet implemented
When I run the extended diagnostics with no plans
Then the extended diagnostics check "Performance: context build time" should have details containing "not yet implemented"
# -- All 5 categories together --
Scenario: Diagnostics JSON output contains all 5 health categories
When I run the extended diagnostics command with format "json"
Then the extended diagnostics checks should include "Config file"
And the extended diagnostics checks should include "Text index"
And the extended diagnostics checks should include "Active plans"
And the extended diagnostics checks should include "Cost summary"
And the extended diagnostics checks should include "Performance: plan duration"
Scenario: Diagnostics plain format includes extended categories
When I run the extended diagnostics command with format "plain"
Then the extended diagnostics output should contain "Text index"
And the extended diagnostics output should contain "Active plans"
And the extended diagnostics output should contain "Cost summary"
And the extended diagnostics output should contain "Performance: plan duration"
Scenario: Diagnostics rich format renders all categories
When I run the extended diagnostics command with format "rich"
Then the extended diagnostics output should contain "Text index"
And the extended diagnostics output should contain "Active plans"
And the extended diagnostics output should contain "Cost summary"
# -- Performance summary success path (P2-1) --
Scenario: Performance summary reports computed duration and latency
When I run the extended diagnostics with completed plans and trace data
Then the extended diagnostics check "Performance: plan duration" should have details containing "avg="
And the extended diagnostics check "Performance: tool call latency" should have details containing "p50="
# -- Error patterns success path (P2-2) --
Scenario: Error Pattern DB reports pattern statistics
When I run the extended diagnostics with 5 error patterns and 42 occurrences
Then the extended diagnostics check "Error Pattern DB" should have details containing "5 patterns"
And the extended diagnostics check "Error Pattern DB" should have details containing "42 total occurrences"
# -- Completed plan durations (M7) --
Scenario: Performance reports completed plan durations
When I run the extended diagnostics with completed plans and trace data
Then the extended diagnostics check "Performance: plan duration" should have details containing "avg="
And the extended diagnostics check "Performance: plan duration" should have details containing "completed plans"
# -- Tool-call latency percentiles (M7) --
Scenario: Performance reports tool-call latency percentiles
When I run the extended diagnostics with completed plans and trace data
Then the extended diagnostics check "Performance: tool call latency" should have details containing "p50="
And the extended diagnostics check "Performance: tool call latency" should have details containing "p95="
And the extended diagnostics check "Performance: tool call latency" should have details containing "p99="
# -- Errored plan count (M7) --
Scenario: Active plans shows errored plan count
When I run the extended diagnostics with 1 running and 0 queued plans and 2 errored
Then the extended diagnostics check "Active plans" should have details containing "2 errored"
# -- Container-returns-None path (M7) --
Scenario: Extended checks degrade when container returns None
When I run the extended diagnostics with container returning None
Then the extended diagnostics check "Text index" should have details containing "N/A"
And the extended diagnostics check "Active plans" should have details containing "N/A"
And the extended diagnostics check "Cost summary" should have details containing "N/A"
And the extended diagnostics check "Performance summary" should have details containing "N/A"
# -- active_sessions property snapshot (P2-3) --
Scenario: Cost budget service provides session snapshot
When I run the extended diagnostics with session snapshot having 3 sessions
Then the extended diagnostics check "Cost summary" should have details containing "3 sessions"
features/mocks/diagnostic_dashboard_mocks.py
+136
View File
@@ -0,0 +1,136 @@
"""Shared mock factories for diagnostic-dashboard Behave scenarios.
Extracted from ``features/steps/diagnostic_dashboard_extended_steps.py``
(C2) so that mocks live in ``features/mocks/`` per CONTRIBUTING.md.
"""
from __future__ import annotations
from datetime import datetime, timedelta
from unittest.mock import MagicMock
def mock_text_backend(doc_count: int) -> MagicMock:
"""Create a mock text index backend with the given document count."""
backend = MagicMock()
backend.document_count = doc_count
return backend
def mock_vector_backend(
emb_count: int,
dim: int = 128,
) -> MagicMock:
"""Create a mock vector index backend."""
backend = MagicMock()
backend.embedding_count = emb_count
# Simulate _embeddings dict for dimensionality inference
if emb_count > 0:
backend._embeddings = {
("proj", f"doc_{i}"): ([0.0] * dim, {}) for i in range(emb_count)
}
else:
backend._embeddings = {}
return backend
def mock_graph_backend(triple_count: int) -> MagicMock:
"""Create a mock graph index backend."""
backend = MagicMock()
backend.triple_count.return_value = triple_count
return backend
def mock_container(
*,
text_backend: MagicMock | None = None,
vector_backend: MagicMock | None = None,
graph_backend: MagicMock | None = None,
plan_lifecycle: MagicMock | None = None,
cost_service: MagicMock | None = None,
trace_service: MagicMock | None = None,
raise_on_missing: bool = False,
) -> MagicMock:
"""Build a mock container with specified services."""
container = MagicMock()
if text_backend is not None:
container.index_text_backend.return_value = text_backend
elif raise_on_missing:
container.index_text_backend.side_effect = RuntimeError("unavailable")
if vector_backend is not None:
container.index_vector_backend.return_value = vector_backend
elif raise_on_missing:
container.index_vector_backend.side_effect = RuntimeError(
"unavailable",
)
if graph_backend is not None:
container.index_graph_backend.return_value = graph_backend
elif raise_on_missing:
container.index_graph_backend.side_effect = RuntimeError(
"unavailable",
)
if plan_lifecycle is not None:
container.plan_lifecycle_service.return_value = plan_lifecycle
elif raise_on_missing:
container.plan_lifecycle_service.side_effect = RuntimeError(
"unavailable",
)
if cost_service is not None:
container.cost_budget_service.return_value = cost_service
elif raise_on_missing:
container.cost_budget_service.side_effect = RuntimeError(
"unavailable",
)
if trace_service is not None:
container.trace_service.return_value = trace_service
elif raise_on_missing:
container.trace_service.side_effect = RuntimeError("unavailable")
return container
def make_plan(
*,
name: str = "test-plan",
processing_state: str = "processing",
phase: str = "execute",
plan_id: str = "01HTEST",
subplan_count: int = 0,
) -> MagicMock:
"""Create a mock Plan object."""
plan = MagicMock()
plan.processing_state = MagicMock()
plan.processing_state.__eq__ = lambda self, other: (
str(other) == processing_state or other.value == processing_state
if hasattr(other, "value")
else str(other) == processing_state
)
# Allow comparison against ProcessingState enum values
plan.processing_state.value = processing_state
plan.phase = MagicMock()
plan.phase.value = phase
plan.namespaced_name = MagicMock()
plan.namespaced_name.name = name
plan.identity = MagicMock()
plan.identity.plan_id = plan_id
plan.subplan_statuses = [MagicMock()] * subplan_count
ts = MagicMock()
ts.created_at = datetime.now() - timedelta(hours=1)
ts.updated_at = datetime.now()
ts.strategize_started_at = None
ts.execute_completed_at = None
ts.applied_at = None
plan.timestamps = ts
return plan
features/steps/cli_core_steps.py
+4 -1
View File
@@ -45,7 +45,10 @@ def _format_data(data: dict[str, Any], fmt: str) -> str:
buf = StringIO()
console = Console(file=buf, width=200, no_color=True)
with patch("cleveragents.cli.main.get_console", return_value=console):
with patch(
"cleveragents.cli.commands.system_rendering.get_console",
return_value=console,
):
if data.get("checks") is not None:
from cleveragents.cli.commands.system import render_diagnostics_rich
features/steps/coverage_security_template_boost_steps.py
+3 -3
View File
@@ -307,7 +307,7 @@ def step_sys_check_data_dir(context: Context) -> None:
from cleveragents.cli.commands.system import _check_data_dir
ms = _mock_settings(data_dir=context.cov_data_dir)
with patch("cleveragents.config.settings.get_settings", return_value=ms):
with patch("cleveragents.cli.commands.system.get_settings", return_value=ms):
context.cov_health_result = _check_data_dir()
@@ -331,7 +331,7 @@ def step_sys_check_perms(context: Context) -> None:
data_dir = getattr(context, "cov_data_dir", Path(context.cov_tmpdir))
ms = _mock_settings(data_dir=data_dir)
with patch("cleveragents.config.settings.get_settings", return_value=ms):
with patch("cleveragents.cli.commands.system.get_settings", return_value=ms):
context.cov_health_result = _check_file_permissions()
@@ -346,7 +346,7 @@ def step_sys_build_info(context: Context) -> None:
database_url=f"sqlite:///{db_path}",
log_dir=log_dir,
)
with patch("cleveragents.config.settings.get_settings", return_value=ms):
with patch("cleveragents.cli.commands.system.get_settings", return_value=ms):
context.cov_info_data = build_info_data()
features/steps/diagnostic_dashboard_extended_steps.py
+500
View File
@@ -0,0 +1,500 @@
"""Step definitions for extended diagnostic dashboard (5 health categories).
Covers features/diagnostic_dashboard_extended.feature.
"""
from __future__ import annotations
from datetime import datetime, timedelta
from io import StringIO
from typing import Any
from unittest.mock import MagicMock, patch
from behave import then, when
from behave.runner import Context
from rich.console import Console
from cleveragents.cli.commands.system import build_diagnostics_data
from cleveragents.cli.commands.system_health import (
_check_active_plans,
_check_cost_summary,
_check_error_patterns,
_check_index_health,
_check_performance_summary,
)
from cleveragents.cli.commands.system_rendering import render_diagnostics_rich
from cleveragents.cli.formatting import format_output
from cleveragents.domain.models.core.plan import ProcessingState
from features.mocks.diagnostic_dashboard_mocks import (
make_plan,
mock_container,
mock_graph_backend,
mock_text_backend,
mock_vector_backend,
)
_PERF_CHECK_NAMES = {
"Performance: plan duration",
"Performance: tool call latency",
"Performance: context build time",
"Performance summary",
}
_CONTAINER_PATCH = (
"cleveragents.cli.commands.system_health.get_container_if_initialized"
)
def _run_diagnostics(context: Context) -> dict[str, Any]:
"""Run build_diagnostics_data and cache on context."""
context.ext_diag_data = build_diagnostics_data()
return context.ext_diag_data
def _format_data(data: dict[str, Any], fmt: str) -> str:
"""Format diagnostics data and return rendered string."""
if fmt == "rich":
buf = StringIO()
console = Console(file=buf, width=200, no_color=True)
with patch(
"cleveragents.cli.commands.system_rendering.get_console",
return_value=console,
):
render_diagnostics_rich(data)
return buf.getvalue()
return format_output(data, fmt)
def _find_check(context: Context, name: str) -> dict[str, Any]:
"""Find a check by name in the diagnostics data."""
for check in context.ext_diag_data.get("checks", []):
if check["name"] == name:
return check
names = [c["name"] for c in context.ext_diag_data.get("checks", [])]
raise AssertionError(f"Check '{name}' not found. Available: {names}")
def _splice_checks(
data: dict[str, Any],
results: list[dict[str, Any]],
replace_names: set[str],
) -> None:
"""Replace matching checks in *data* with *results* in-place."""
data["checks"] = [
c for c in data["checks"] if c["name"] not in replace_names
] + results
def _with_container(container: MagicMock):
"""Shorthand context-manager to patch get_container_if_initialized."""
return patch(_CONTAINER_PATCH, return_value=container)
def _run_check_and_splice(
context: Context,
check_fn,
replace_names: set[str],
container: MagicMock | None = None,
) -> None:
"""Run *check_fn* inside a container patch, splice into fresh diag data."""
if container is not None:
with _with_container(container):
results = check_fn()
else:
results = check_fn()
data = build_diagnostics_data()
_splice_checks(
data, results if isinstance(results, list) else [results], replace_names
)
context.ext_diag_data = data
_PLAN_CHECK_NAMES = {"Active plans"}
def _plan_replace_names(data: dict[str, Any]) -> set[str]:
"""Compute the set of check names to replace for plan-related checks."""
return _PLAN_CHECK_NAMES | {
c["name"] for c in data["checks"] if c["name"].startswith("Plan: ")
}
def make_plans_list(
running: int = 0,
queued: int = 0,
errored: int = 0,
) -> list[MagicMock]:
"""Build a list of mock plans with the given state counts."""
plans: list[MagicMock] = []
for i in range(running):
p = make_plan(
name=f"run-{i}", processing_state=ProcessingState.PROCESSING.value
)
p.processing_state = ProcessingState.PROCESSING
plans.append(p)
for i in range(queued):
p = make_plan(name=f"queue-{i}", processing_state=ProcessingState.QUEUED.value)
p.processing_state = ProcessingState.QUEUED
plans.append(p)
for i in range(errored):
p = make_plan(name=f"err-{i}", processing_state=ProcessingState.ERRORED.value)
p.processing_state = ProcessingState.ERRORED
plans.append(p)
return plans
def _run_active_plans_check(
context: Context,
plans: list[MagicMock],
) -> None:
"""Run _check_active_plans with given plan list and splice results."""
lifecycle = MagicMock()
lifecycle.list_plans.return_value = plans
container = mock_container(plan_lifecycle=lifecycle)
with _with_container(container):
results = _check_active_plans()
data = build_diagnostics_data()
_splice_checks(data, results, _plan_replace_names(data))
context.ext_diag_data = data
@when('I run the extended diagnostics command with format "{fmt}"')
def step_run_extended_diagnostics_fmt(context: Context, fmt: str) -> None:
"""Run diagnostics and store data + formatted output."""
data = _run_diagnostics(context)
context.ext_diag_output = _format_data(data, fmt)
@when(
"I run the extended diagnostics with mock text index having {count:d} documents",
)
def step_run_diag_mock_text(context: Context, count: int) -> None:
"""Run diagnostics with a mocked text index backend."""
text = mock_text_backend(count)
container = mock_container(text_backend=text)
with _with_container(container):
results = _check_index_health()
data = build_diagnostics_data()
for i, check in enumerate(data["checks"]):
if check["name"] == "Text index":
data["checks"][i] = results[0]
break
else:
data["checks"].extend(results[:1])
context.ext_diag_data = data
@when(
"I run the extended diagnostics with mock vector index"
" having {count:d} embeddings of dimension {dim:d}",
)
def step_run_diag_mock_vector(context: Context, count: int, dim: int) -> None:
"""Run diagnostics with a mocked vector index backend."""
vector = mock_vector_backend(count, dim)
container = mock_container(vector_backend=vector)
mock_settings = MagicMock()
mock_settings.vector_embeddings_dimension = dim
with (
_with_container(container),
patch(
"cleveragents.cli.commands.system_health.get_settings",
return_value=mock_settings,
),
):
results = _check_index_health()
data = build_diagnostics_data()
for i, check in enumerate(data["checks"]):
if check["name"] == "Vector index":
data["checks"][i] = results[1]
break
else:
data["checks"].extend(results[1:2])
context.ext_diag_data = data
@when("I run the extended diagnostics with mock graph index having {count:d} triples")
def step_run_diag_mock_graph(context: Context, count: int) -> None:
"""Run diagnostics with a mocked graph index backend."""
graph = mock_graph_backend(count)
container = mock_container(graph_backend=graph)
with _with_container(container):
results = _check_index_health()
data = build_diagnostics_data()
for i, check in enumerate(data["checks"]):
if check["name"] == "Graph store":
data["checks"][i] = results[2]
break
else:
data["checks"].extend(results[2:3])
context.ext_diag_data = data
@when("I run the extended diagnostics with unavailable index backends")
def step_run_diag_unavailable_index(context: Context) -> None:
"""Run diagnostics with all index backends raising errors."""
container = mock_container(raise_on_missing=True)
_run_check_and_splice(
context,
_check_index_health,
{"Text index", "Vector index", "Graph store"},
container,
)
@when(
"I run the extended diagnostics with"
" {running:d} running and {queued:d} queued plans",
)
def step_run_diag_mock_plans(context: Context, running: int, queued: int) -> None:
"""Run diagnostics with mocked running and queued plans."""
_run_active_plans_check(context, make_plans_list(running=running, queued=queued))
@when('I run the extended diagnostics with 1 running plan named "{name}"')
def step_run_diag_mock_named_plan(context: Context, name: str) -> None:
"""Run diagnostics with one running plan of a given name."""
plan = make_plan(name=name)
plan.processing_state = ProcessingState.PROCESSING
_run_active_plans_check(context, [plan])
@when(
"I run the extended diagnostics with"
" {running:d} running and {queued:d} queued plans and {errored:d} errored",
)
def step_run_diag_mock_plans_errored(
context: Context,
running: int,
queued: int,
errored: int,
) -> None:
"""Run diagnostics with mocked running, queued, and errored plans."""
_run_active_plans_check(
context,
make_plans_list(running=running, queued=queued, errored=errored),
)
@when("I run the extended diagnostics with unavailable plan service")
def step_run_diag_unavailable_plans(context: Context) -> None:
"""Run diagnostics with plan service raising an error."""
_run_check_and_splice(
context,
_check_active_plans,
{"Active plans"},
mock_container(raise_on_missing=True),
)
@when("I run the extended diagnostics with container returning None")
def step_run_diag_container_none(context: Context) -> None:
"""Run diagnostics with container returning None for all services."""
with patch(_CONTAINER_PATCH, return_value=None):
idx = _check_index_health()
plans = _check_active_plans()
cost = _check_cost_summary()
perf = _check_performance_summary()
data = build_diagnostics_data()
_all_ext_names = {
"Text index",
"Vector index",
"Graph store",
"Active plans",
"Cost summary",
"Performance summary",
"Performance: plan duration",
"Performance: tool call latency",
"Performance: context build time",
}
_splice_checks(data, idx + plans + cost + perf, _all_ext_names)
context.ext_diag_data = data
@when(
"I run the extended diagnostics with session cost of"
" {cost:g} across {count:d} sessions",
)
def step_run_diag_mock_cost(
context: Context,
cost: float,
count: int,
) -> None:
"""Run diagnostics with mocked session cost data."""
sessions: dict[str, MagicMock] = {}
per_session = cost / count if count > 0 else 0
for i in range(count):
budget = MagicMock()
budget.total_cost = per_session
budget.max_cost_usd = None
sessions[f"session-{i}"] = budget
cost_service = MagicMock()
cost_service.active_sessions = sessions
_run_check_and_splice(
context,
_check_cost_summary,
{"Cost summary"},
mock_container(cost_service=cost_service),
)
@when(
"I run the extended diagnostics with budgeted sessions"
" costing {cost:g} of {max_cost:g} max",
)
def step_run_diag_mock_budgeted_cost(
context: Context,
cost: float,
max_cost: float,
) -> None:
"""Run diagnostics with sessions that have budget caps configured."""
budget = MagicMock()
budget.total_cost = cost
budget.max_cost_usd = max_cost
cost_service = MagicMock()
cost_service.active_sessions = {"budgeted-session": budget}
_run_check_and_splice(
context,
_check_cost_summary,
{"Cost summary"},
mock_container(cost_service=cost_service),
)
@when("I run the extended diagnostics with no active sessions")
def step_run_diag_no_sessions(context: Context) -> None:
cost_service = MagicMock()
cost_service.active_sessions = {}
_run_check_and_splice(
context,
_check_cost_summary,
{"Cost summary"},
mock_container(cost_service=cost_service),
)
@when("I run the extended diagnostics with unavailable cost service")
def step_run_diag_unavailable_cost(context: Context) -> None:
_run_check_and_splice(
context,
_check_cost_summary,
{"Cost summary"},
mock_container(raise_on_missing=True),
)
@when("I run the extended diagnostics with no plans")
def step_run_diag_no_plans(context: Context) -> None:
"""Run diagnostics with no plans for performance summary."""
lifecycle = MagicMock()
lifecycle.list_plans.return_value = []
_run_check_and_splice(
context,
_check_performance_summary,
_PERF_CHECK_NAMES,
mock_container(plan_lifecycle=lifecycle, trace_service=MagicMock()),
)
@when("I run the extended diagnostics with unavailable performance services")
def step_run_diag_unavailable_perf(context: Context) -> None:
_run_check_and_splice(
context,
_check_performance_summary,
_PERF_CHECK_NAMES,
mock_container(raise_on_missing=True),
)
@then('the extended diagnostics checks should include "{name}"')
def step_ext_diag_checks_include(context: Context, name: str) -> None:
"""Assert a check with the given name exists."""
checks = context.ext_diag_data.get("checks", [])
names = [c["name"] for c in checks]
assert name in names, f"Check '{name}' not found. Available: {names}"
@then(
'the extended diagnostics check "{name}" should have details containing "{text}"',
)
def step_ext_diag_check_details(
context: Context,
name: str,
text: str,
) -> None:
"""Assert the details field of a named check contains text."""
check = _find_check(context, name)
details = check.get("details", "")
assert text in details, f"Expected '{text}' in details of '{name}': {details}"
@then('the extended diagnostics output should contain "{text}"')
def step_ext_diag_output_contains(context: Context, text: str) -> None:
"""Assert the formatted output contains the given text."""
output = context.ext_diag_output
assert text in output, f"Expected '{text}' in diagnostics output:\n{output}"
@when("I run the extended diagnostics with completed plans and trace data")
def step_run_diag_perf_success(context: Context) -> None:
"""Run diagnostics with completed plans producing real duration & latency."""
plan = make_plan(name="done", processing_state=ProcessingState.COMPLETE.value)
plan.processing_state = ProcessingState.COMPLETE
now = datetime.now()
plan.timestamps.strategize_started_at = now - timedelta(seconds=120)
plan.timestamps.execute_completed_at = now
lifecycle = MagicMock()
lifecycle.list_plans.return_value = [plan]
trace = MagicMock()
trace.latency_ms = 50.0
trace_svc = MagicMock()
trace_svc.get_traces.return_value = [trace]
_run_check_and_splice(
context,
_check_performance_summary,
_PERF_CHECK_NAMES,
mock_container(plan_lifecycle=lifecycle, trace_service=trace_svc),
)
@when(
"I run the extended diagnostics with"
" {patterns:d} error patterns and {occurrences:d} occurrences",
)
def step_run_diag_error_patterns_success(
context: Context,
patterns: int,
occurrences: int,
) -> None:
"""Run diagnostics with error pattern service returning statistics."""
mock_svc = MagicMock()
mock_svc.get_statistics.return_value = {
"total_patterns": patterns,
"total_occurrences": occurrences,
}
mock_container = MagicMock()
mock_container.error_pattern_service.return_value = mock_svc
with patch(_CONTAINER_PATCH, return_value=mock_container):
result = _check_error_patterns()
data = build_diagnostics_data()
_splice_checks(data, [result], {"Error Pattern DB"})
context.ext_diag_data = data
@when(
"I run the extended diagnostics with session snapshot having {count:d} sessions",
)
def step_run_diag_session_snapshot(context: Context, count: int) -> None:
"""Run diagnostics verifying active_sessions returns a proper snapshot."""
sessions: dict[str, MagicMock] = {}
for i in range(count):
b = MagicMock()
b.total_cost, b.max_cost_usd = 0.25, None
sessions[f"snap-{i}"] = b
cost_service = MagicMock()
cost_service.active_sessions = sessions
_run_check_and_splice(
context,
_check_cost_summary,
{"Cost summary"},
mock_container(cost_service=cost_service),
)
features/steps/system_cli_coverage_boost_steps.py
+15 -12
View File
@@ -71,7 +71,7 @@ def step_perms_writable_only(context: Context) -> None:
return True
with (
patch("cleveragents.config.settings.get_settings", return_value=ms),
patch("cleveragents.cli.commands.system.get_settings", return_value=ms),
patch("cleveragents.cli.commands.system.os.access", side_effect=mock_access),
):
context.cov_result = _check_file_permissions()
@@ -94,16 +94,16 @@ def step_stale_locks_zero(context: Context) -> None:
with (
patch(
"cleveragents.cli.commands.system.get_database_url",
"cleveragents.cli.commands.system_health.get_database_url",
return_value="sqlite:///test.db",
),
patch("cleveragents.cli.commands.system.create_engine"),
patch("cleveragents.cli.commands.system_health.create_engine"),
patch(
"cleveragents.cli.commands.system.sa_inspect",
"cleveragents.cli.commands.system_health.sa_inspect",
return_value=mock_inspector,
),
patch(
"cleveragents.cli.commands.system.LockService",
"cleveragents.cli.commands.system_health.LockService",
return_value=mock_lock_service,
),
):
@@ -127,16 +127,16 @@ def step_stale_locks_nonzero(context: Context, count: int) -> None:
with (
patch(
"cleveragents.cli.commands.system.get_database_url",
"cleveragents.cli.commands.system_health.get_database_url",
return_value="sqlite:///test.db",
),
patch("cleveragents.cli.commands.system.create_engine"),
patch("cleveragents.cli.commands.system_health.create_engine"),
patch(
"cleveragents.cli.commands.system.sa_inspect",
"cleveragents.cli.commands.system_health.sa_inspect",
return_value=mock_inspector,
),
patch(
"cleveragents.cli.commands.system.LockService",
"cleveragents.cli.commands.system_health.LockService",
return_value=mock_lock_service,
),
):
@@ -153,7 +153,7 @@ def step_stale_locks_exception(context: Context) -> None:
from cleveragents.cli.commands.system import _check_stale_locks
with patch(
"cleveragents.cli.commands.system.get_database_url",
"cleveragents.cli.commands.system_health.get_database_url",
side_effect=RuntimeError("connection refused"),
):
context.cov_result = _check_stale_locks()
@@ -176,7 +176,10 @@ def step_async_enabled(context: Context, max_w: int, poll: int) -> None:
async_poll_interval=poll,
)
with patch("cleveragents.config.settings.get_settings", return_value=ms):
with patch(
"cleveragents.cli.commands.system_health.get_settings",
return_value=ms,
):
context.cov_result = _check_async_worker_health()
@@ -190,7 +193,7 @@ def step_async_exception(context: Context) -> None:
from cleveragents.cli.commands.system import _check_async_worker_health
with patch(
"cleveragents.config.settings.get_settings",
"cleveragents.cli.commands.system_health.get_settings",
side_effect=RuntimeError("settings unavailable"),
):
context.cov_result = _check_async_worker_health()
features/steps/system_cli_uncovered_branches_steps.py
+9 -9
View File
@@ -77,7 +77,7 @@ def step_branch_info_db_exists(context: Context) -> None:
storage_path=Path(tmpdir),
)
with patch("cleveragents.config.settings.get_settings", return_value=ms):
with patch("cleveragents.cli.commands.system.get_settings", return_value=ms):
context.branch_result = build_info_data()
@@ -111,7 +111,7 @@ def step_branch_info_db_exception(context: Context) -> None:
return original_stat(self, *args, **kwargs)
with (
patch("cleveragents.config.settings.get_settings", return_value=ms),
patch("cleveragents.cli.commands.system.get_settings", return_value=ms),
patch.object(Path, "stat", patched_stat),
):
context.branch_result = build_info_data()
@@ -137,7 +137,7 @@ def step_branch_info_log_exists(context: Context) -> None:
storage_path=Path(tmpdir),
)
with patch("cleveragents.config.settings.get_settings", return_value=ms):
with patch("cleveragents.cli.commands.system.get_settings", return_value=ms):
context.branch_result = build_info_data()
@@ -186,7 +186,7 @@ def step_branch_data_dir_not_writable(context: Context) -> None:
return not (str(p) == str(data_dir) and mode == os.W_OK)
with (
patch("cleveragents.config.settings.get_settings", return_value=ms),
patch("cleveragents.cli.commands.system.get_settings", return_value=ms),
patch("cleveragents.cli.commands.system.os.access", side_effect=mock_access),
):
context.branch_result = _check_data_dir()
@@ -203,7 +203,7 @@ def step_branch_data_dir_missing(context: Context) -> None:
ms = _make_mock_settings(data_dir=Path("/tmp/nonexistent_dir_xyz_99999"))
with patch("cleveragents.config.settings.get_settings", return_value=ms):
with patch("cleveragents.cli.commands.system.get_settings", return_value=ms):
context.branch_result = _check_data_dir()
@@ -226,7 +226,7 @@ def step_branch_db_not_writable(context: Context) -> None:
return not (str(p) == str(db_file) and mode == os.W_OK)
with (
patch("cleveragents.config.settings.get_settings", return_value=ms),
patch("cleveragents.cli.commands.system.get_settings", return_value=ms),
patch("cleveragents.cli.commands.system.os.access", side_effect=mock_access),
):
context.branch_result = _check_database()
@@ -319,7 +319,7 @@ def step_branch_perms_missing(context: Context) -> None:
ms = _make_mock_settings(data_dir=Path("/tmp/nonexistent_dir_xyz_99999"))
with patch("cleveragents.config.settings.get_settings", return_value=ms):
with patch("cleveragents.cli.commands.system.get_settings", return_value=ms):
context.branch_result = _check_file_permissions()
@@ -345,7 +345,7 @@ def step_branch_perms_read_no_write(context: Context) -> None:
return True
with (
patch("cleveragents.config.settings.get_settings", return_value=ms),
patch("cleveragents.cli.commands.system.get_settings", return_value=ms),
patch("cleveragents.cli.commands.system.os.access", side_effect=mock_access),
):
context.branch_result = _check_file_permissions()
@@ -368,7 +368,7 @@ def step_branch_perms_no_access(context: Context) -> None:
return str(p) != str(data_dir)
with (
patch("cleveragents.config.settings.get_settings", return_value=ms),
patch("cleveragents.cli.commands.system.get_settings", return_value=ms),
patch("cleveragents.cli.commands.system.os.access", side_effect=mock_access),
):
context.branch_result = _check_file_permissions()
features/steps/system_coverage_boost_steps.py
+7 -7
View File
@@ -70,7 +70,7 @@ def step_info_no_db(context: Context) -> None:
log_dir=Path(tmpdir) / "logs",
)
with patch("cleveragents.config.settings.get_settings", return_value=ms):
with patch("cleveragents.cli.commands.system.get_settings", return_value=ms):
context.result_data = build_info_data()
@@ -89,7 +89,7 @@ def step_info_no_log_dir(context: Context) -> None:
log_dir=Path(tmpdir) / "nonexistent_logs_dir",
)
with patch("cleveragents.config.settings.get_settings", return_value=ms):
with patch("cleveragents.cli.commands.system.get_settings", return_value=ms):
context.result_data = build_info_data()
@@ -104,7 +104,7 @@ def step_check_data_dir_nonexistent(context: Context) -> None:
ms = _mock_settings(data_dir=Path("/nonexistent/data/dir/xyz99"))
with patch("cleveragents.config.settings.get_settings", return_value=ms):
with patch("cleveragents.cli.commands.system.get_settings", return_value=ms):
context.result_data = _check_data_dir()
@@ -121,7 +121,7 @@ def step_check_db_postgres(context: Context) -> None:
ms = _mock_settings(database_url="postgresql://localhost/test")
with patch("cleveragents.config.settings.get_settings", return_value=ms):
with patch("cleveragents.cli.commands.system.get_settings", return_value=ms):
context.result_data = _check_database()
@@ -139,7 +139,7 @@ def step_check_db_nonwritable(context: Context) -> None:
ms = _mock_settings(database_url="sqlite:////nonexistent/parent/dir/test.db")
with patch("cleveragents.config.settings.get_settings", return_value=ms):
with patch("cleveragents.cli.commands.system.get_settings", return_value=ms):
context.result_data = _check_database()
@@ -156,7 +156,7 @@ def step_check_perms_nonexistent(context: Context) -> None:
ms = _mock_settings(data_dir=Path("/nonexistent/data/dir/xyz99"))
with patch("cleveragents.config.settings.get_settings", return_value=ms):
with patch("cleveragents.cli.commands.system.get_settings", return_value=ms):
context.result_data = _check_file_permissions()
@@ -176,7 +176,7 @@ def step_check_perms_readonly(context: Context) -> None:
return original_access(p, mode)
with (
patch("cleveragents.config.settings.get_settings", return_value=ms),
patch("cleveragents.cli.commands.system.get_settings", return_value=ms),
patch("cleveragents.cli.commands.system.os.access", side_effect=mock_access),
):
context.result_data = _check_file_permissions()
features/steps/system_diagnostics_coverage_steps.py
+18 -11
View File
@@ -84,7 +84,7 @@ def step_sysdiag_perms_no_read(context: Context) -> None:
return mode == _os.W_OK
with (
patch("cleveragents.config.settings.get_settings", return_value=ms),
patch("cleveragents.cli.commands.system.get_settings", return_value=ms),
patch("cleveragents.cli.commands.system.os.access", side_effect=fake_access),
):
context.sysdiag_result = _check_file_permissions()
@@ -97,7 +97,7 @@ def step_sysdiag_perms_no_access(context: Context) -> None:
ms = _sysdiag_mock_settings(data_dir=context.sysdiag_data_dir)
with (
patch("cleveragents.config.settings.get_settings", return_value=ms),
patch("cleveragents.cli.commands.system.get_settings", return_value=ms),
patch("cleveragents.cli.commands.system.os.access", return_value=False),
):
context.sysdiag_result = _check_file_permissions()
@@ -121,19 +121,19 @@ def step_sysdiag_stale_locks_count(context: Context, count: int) -> None:
with (
patch(
"cleveragents.cli.commands.system.get_database_url",
"cleveragents.cli.commands.system_health.get_database_url",
return_value="sqlite:///test.db",
),
patch(
"cleveragents.cli.commands.system.create_engine",
"cleveragents.cli.commands.system_health.create_engine",
return_value=mock_engine,
),
patch(
"cleveragents.cli.commands.system.sa_inspect",
"cleveragents.cli.commands.system_health.sa_inspect",
return_value=mock_inspector,
),
patch(
"cleveragents.cli.commands.system.LockService",
"cleveragents.cli.commands.system_health.LockService",
return_value=mock_lock_service,
),
):
@@ -145,7 +145,7 @@ def step_sysdiag_stale_locks_error(context: Context) -> None:
from cleveragents.cli.commands.system import _check_stale_locks
with patch(
"cleveragents.cli.commands.system.get_database_url",
"cleveragents.cli.commands.system_health.get_database_url",
side_effect=RuntimeError("database unreachable"),
):
context.sysdiag_result = _check_stale_locks()
@@ -165,7 +165,10 @@ def step_sysdiag_async_enabled(context: Context) -> None:
async_max_workers=8,
async_poll_interval=2.5,
)
with patch("cleveragents.config.settings.get_settings", return_value=ms):
with patch(
"cleveragents.cli.commands.system_health.get_settings",
return_value=ms,
):
context.sysdiag_result = _check_async_worker_health()
@@ -174,7 +177,7 @@ def step_sysdiag_async_error(context: Context) -> None:
from cleveragents.cli.commands.system import _check_async_worker_health
with patch(
"cleveragents.config.settings.get_settings",
"cleveragents.cli.commands.system_health.get_settings",
side_effect=RuntimeError("settings unavailable"),
):
context.sysdiag_result = _check_async_worker_health()
@@ -189,9 +192,13 @@ def step_sysdiag_async_error(context: Context) -> None:
def step_sysdiag_error_patterns_error(context: Context) -> None:
from cleveragents.cli.commands.system import _check_error_patterns
mock_container = MagicMock()
mock_container.error_pattern_service.side_effect = RuntimeError(
"service init failed"
)
with patch(
"cleveragents.application.services.error_pattern_service.ErrorPatternService",
side_effect=RuntimeError("service init failed"),
"cleveragents.cli.commands.system_health.get_container_if_initialized",
return_value=mock_container,
):
context.sysdiag_result = _check_error_patterns()
features/system_diagnostics_coverage.feature
+2 -2
View File
@@ -70,5 +70,5 @@ Feature: System diagnostics uncovered code paths
Scenario: Error pattern check when service initialization fails
Given a sysdiag test environment
When I run the error patterns check with a service error
Then the sysdiag check status should be "ok"
And the sysdiag check details should be "empty (no patterns recorded)"
Then the sysdiag check status should be "warn"
And the sysdiag check details should be "unable to check"
robot/diagnostic_dashboard_extended.robot
+47
View File
@@ -0,0 +1,47 @@
*** Settings ***
Documentation Integration tests for the extended diagnostic dashboard (5 health categories)
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_diagnostic_dashboard_extended.py
*** Test Cases ***
Diagnostics JSON Contains All Five Categories
[Documentation] Verify that diagnostics JSON includes all 5 health categories
${result}= Run Process ${PYTHON} ${HELPER} all-categories cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} diag-ext-all-categories-ok
Diagnostics Index Health Present
[Documentation] Verify index health checks appear in diagnostics
${result}= Run Process ${PYTHON} ${HELPER} index-health cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} diag-ext-index-health-ok
Diagnostics Active Plans Present
[Documentation] Verify active plans overview appears in diagnostics
${result}= Run Process ${PYTHON} ${HELPER} active-plans cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} diag-ext-active-plans-ok
Diagnostics Cost Summary Present
[Documentation] Verify cost summary appears in diagnostics
${result}= Run Process ${PYTHON} ${HELPER} cost-summary cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} diag-ext-cost-summary-ok
Diagnostics Performance Summary Present
[Documentation] Verify performance summary appears in diagnostics
${result}= Run Process ${PYTHON} ${HELPER} performance-summary cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} diag-ext-performance-summary-ok
Diagnostics Plain Format Contains Extended Categories
[Documentation] Verify plain format output includes extended categories
${result}= Run Process ${PYTHON} ${HELPER} plain-format cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} diag-ext-plain-format-ok
robot/helper_diagnostic_dashboard_extended.py
+185
View File
@@ -0,0 +1,185 @@
"""Robot helper for extended diagnostic dashboard integration tests.
Usage::
python helper_diagnostic_dashboard_extended.py <test-case>
Each test case exercises ``build_diagnostics_data()`` and checks that the
extended health categories (index, plans, cost, performance) are present.
"""
from __future__ import annotations
import sys
from typing import Any
def _build_data() -> dict[str, Any]:
"""Build diagnostics data dict.
Ensures the DI container is initialised first (as the real CLI would do)
so that extended health checks can query live services.
"""
from cleveragents.application.container import get_container
from cleveragents.cli.commands.system import build_diagnostics_data
# The real CLI boots the container before reaching the diagnostics
# command. Replicate that here so the check functions see an
# initialised container.
get_container()
return build_diagnostics_data()
def _check_names(data: dict[str, Any]) -> list[str]:
"""Extract check names from diagnostics data."""
return [c["name"] for c in data.get("checks", [])]
def _assert_structure(data: dict[str, Any]) -> None:
"""Assert the top-level diagnostics structure is well-formed."""
for key in ("checks", "summary", "recommendations", "has_errors", "has_warnings"):
if key not in data:
print(f"FAIL: Missing top-level key '{key}'", file=sys.stderr)
sys.exit(1)
for check in data["checks"]:
for field in ("name", "status", "details"):
if field not in check:
print(
f"FAIL: Check missing field '{field}': {check}",
file=sys.stderr,
)
sys.exit(1)
if check["status"] not in ("ok", "warn", "error"):
print(
f"FAIL: Invalid status '{check['status']}' in {check['name']}",
file=sys.stderr,
)
sys.exit(1)
def test_all_categories() -> None:
"""Verify all 5 categories are present."""
data = _build_data()
_assert_structure(data)
names = _check_names(data)
required = [
"Config file", # Category 1: System health
"Text index", # Category 2: Index health
"Active plans", # Category 3: Active plan overview
"Cost summary", # Category 4: Cost summary
"Performance: plan duration", # Category 5: Performance
]
missing = [r for r in required if r not in names]
if missing:
print(f"FAIL: Missing checks: {missing}", file=sys.stderr)
print(f"Available: {names}", file=sys.stderr)
sys.exit(1)
print("diag-ext-all-categories-ok")
def test_index_health() -> None:
"""Verify index health checks are present."""
data = _build_data()
_assert_structure(data)
names = _check_names(data)
required = ["Text index", "Vector index", "Graph store"]
missing = [r for r in required if r not in names]
if missing:
print(f"FAIL: Missing index checks: {missing}", file=sys.stderr)
sys.exit(1)
print("diag-ext-index-health-ok")
def test_active_plans() -> None:
"""Verify active plans overview is present."""
data = _build_data()
_assert_structure(data)
names = _check_names(data)
if "Active plans" not in names:
print("FAIL: 'Active plans' check missing", file=sys.stderr)
sys.exit(1)
print("diag-ext-active-plans-ok")
def test_cost_summary() -> None:
"""Verify cost summary is present."""
data = _build_data()
_assert_structure(data)
names = _check_names(data)
if "Cost summary" not in names:
print("FAIL: 'Cost summary' check missing", file=sys.stderr)
sys.exit(1)
print("diag-ext-cost-summary-ok")
def test_performance_summary() -> None:
"""Verify performance summary checks are present."""
data = _build_data()
_assert_structure(data)
names = _check_names(data)
required = [
"Performance: plan duration",
"Performance: tool call latency",
"Performance: context build time",
]
missing = [r for r in required if r not in names]
if missing:
print(
f"FAIL: Missing performance checks: {missing}",
file=sys.stderr,
)
sys.exit(1)
print("diag-ext-performance-summary-ok")
def test_plain_format() -> None:
"""Verify plain format output includes extended categories."""
data = _build_data()
from cleveragents.cli.formatting import format_output
output = format_output(data, "plain")
required_texts = [
"Text index",
"Active plans",
"Cost summary",
"Performance: plan duration",
]
missing = [t for t in required_texts if t not in output]
if missing:
print(
f"FAIL: Missing in plain output: {missing}",
file=sys.stderr,
)
sys.exit(1)
print("diag-ext-plain-format-ok")
_TESTS = {
"all-categories": test_all_categories,
"index-health": test_index_health,
"active-plans": test_active_plans,
"cost-summary": test_cost_summary,
"performance-summary": test_performance_summary,
"plain-format": test_plain_format,
}
def main() -> None:
"""Entry point."""
if len(sys.argv) < 2 or sys.argv[1] not in _TESTS:
print(
f"Usage: {sys.argv[0]} <{'|'.join(_TESTS)}>",
file=sys.stderr,
)
sys.exit(2)
_TESTS[sys.argv[1]]()
if __name__ == "__main__":
main()
robot/helper_e2e_common.py
+3 -3
View File
@@ -136,14 +136,14 @@ def write_yaml(content: str) -> str:
return path
def init_bare_git_repo() -> str:
"""Create a bare git repository with an initial commit.
def init_test_git_repo() -> str:
"""Create a non-bare git repository with an initial commit on the ``main`` branch.
Returns the path to the repository.
"""
repo_dir = tempfile.mkdtemp(prefix="e2e_git_")
cmds = [
["git", "init"],
["git", "init", "-b", "main"],
["git", "config", "user.email", "test@example.com"],
["git", "config", "user.name", "Test"],
]
robot/helper_m1_e2e_verification.py
+6 -6
View File
@@ -42,7 +42,7 @@ if _ROBOT not in sys.path:
from helper_e2e_common import ( # noqa: E402
cleanup_workspace,
init_bare_git_repo,
init_test_git_repo,
is_expected_provider_unavailable,
run_cli,
setup_workspace,
@@ -193,7 +193,7 @@ def action_create_from_yaml() -> None:
def resource_register_git_checkout() -> None:
"""Register a git-checkout resource via real CLI subprocess."""
workspace = setup_workspace(prefix="m1_resource_")
repo_dir = init_bare_git_repo()
repo_dir = init_test_git_repo()
try:
result = run_cli(
"resource",
@@ -237,7 +237,7 @@ def resource_register_git_checkout() -> None:
def project_create_and_link() -> None:
"""Create a project and link a resource via real CLI subprocess."""
workspace = setup_workspace(prefix="m1_project_")
repo_dir = init_bare_git_repo()
repo_dir = init_test_git_repo()
try:
# First register a resource (dependency for --resource flag)
r1 = run_cli(
@@ -307,7 +307,7 @@ def plan_full_lifecycle() -> None:
(``changeset-invocations``), not this CLI lifecycle test.
"""
workspace = setup_workspace(prefix="m1_plan_")
repo_dir = init_bare_git_repo()
repo_dir = init_test_git_repo()
yaml_path = write_yaml(_VALID_ACTION_YAML)
try:
# Setup: register resource + create project
@@ -536,7 +536,7 @@ def sandbox_isolation_check() -> None:
)
from cleveragents.infrastructure.sandbox.protocol import SandboxStatus
repo_dir = init_bare_git_repo()
repo_dir = init_test_git_repo()
try:
# Create sandbox
sandbox = GitWorktreeSandbox(
@@ -596,7 +596,7 @@ def post_apply_commit_check() -> None:
GitWorktreeSandbox,
)
repo_dir = init_bare_git_repo()
repo_dir = init_test_git_repo()
try:
sandbox = GitWorktreeSandbox(
resource_id="res-commit-check",
robot/helper_wf01_hello_world.py
+418
View File
@@ -0,0 +1,418 @@
"""Robot helper for Workflow Example 1 — Hello World, fix a single bug.
Exercises the ``manual`` automation profile through standalone subcommand
tests: init, resource registration, project creation, validation
registration and attachment, action creation, and post-apply commit
verification.
Each subcommand is fully independent: it provisions and tears down its own
workspace, so tests may be run in any order or subset.
Each subcommand prints a sentinel on success.
Exit code 0 = pass, 1 = failure.
"""
from __future__ import annotations
import os
import shutil
import subprocess
import sys
from collections.abc import Callable
from pathlib import Path
from typing import NoReturn
# Ensure the src directory is on the import path.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
# Ensure robot/ is on the import path for helper_e2e_common.
_ROBOT = str(Path(__file__).resolve().parent)
if _ROBOT not in sys.path:
sys.path.insert(0, _ROBOT)
from helper_e2e_common import ( # noqa: E402
cleanup_workspace,
init_test_git_repo,
run_cli,
setup_workspace,
write_yaml,
)
from cleveragents.infrastructure.sandbox.git_worktree import ( # noqa: E402
GitWorktreeSandbox,
)
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
_ACTION_YAML = """\
name: local/hello-world-fix
description: Fix a single bug in the hello-world project
strategy_actor: openai/gpt-4
execution_actor: openai/gpt-4
definition_of_done: The bug is fixed, tests pass, and the change is committed
automation_profile: manual
arguments:
- name: bug_description
type: string
required: true
description: Description of the bug to fix
"""
_VALIDATION_YAML = """\
name: local/unit-tests
description: Run unit tests before apply
source: custom
mode: required
code: |
def run(inputs):
return {"passed": True}
"""
def _fail(msg: str) -> NoReturn:
"""Print failure message and exit."""
print(f"FAIL: {msg}", file=sys.stderr)
raise SystemExit(1)
# ---------------------------------------------------------------------------
# Shared workspace context
# ---------------------------------------------------------------------------
class _WorkflowCtx:
"""Holds shared state across workflow steps."""
def __init__(self) -> None:
self.workspace: str = ""
self.repo_dir: str = ""
self.yaml_path: str = ""
self.plan_id: str = ""
def setup(self) -> None:
self.workspace = setup_workspace(prefix="wf01_")
self.repo_dir = init_test_git_repo()
self.yaml_path = write_yaml(_ACTION_YAML)
def teardown(self) -> None:
if self.yaml_path and os.path.exists(self.yaml_path):
os.unlink(self.yaml_path)
if self.repo_dir:
shutil.rmtree(self.repo_dir, ignore_errors=True)
if self.workspace:
cleanup_workspace(self.workspace)
# ---------------------------------------------------------------------------
# Subcommand: init
# ---------------------------------------------------------------------------
def wf01_init() -> None:
"""Verify agents init via real CLI subprocess."""
ctx = _WorkflowCtx()
ctx.setup()
try:
result = run_cli(
"init",
"--yes",
"--path",
ctx.workspace,
workspace=ctx.workspace,
)
if result.returncode != 0:
_fail(f"init rc={result.returncode}\n{result.stderr}")
print("wf01-init-ok")
finally:
ctx.teardown()
# ---------------------------------------------------------------------------
# Subcommand: resource-register
# ---------------------------------------------------------------------------
def wf01_resource_register() -> None:
"""Register a git-checkout resource via real CLI subprocess."""
ctx = _WorkflowCtx()
ctx.setup()
try:
result = run_cli(
"resource",
"add",
"git-checkout",
"local/hello-repo",
"--path",
ctx.repo_dir,
"--branch",
"main",
"--format",
"plain",
workspace=ctx.workspace,
)
if result.returncode != 0:
_fail(f"resource add rc={result.returncode}\n{result.stderr}")
if "hello-repo" not in result.stdout.lower():
_fail(f"resource name not in output:\n{result.stdout}")
if "error" in result.stdout.lower() or "not found" in result.stdout.lower():
_fail(f"resource registration appears to have failed:\n{result.stdout}")
print("wf01-resource-register-ok")
finally:
ctx.teardown()
# ---------------------------------------------------------------------------
# Subcommand: project-create
# ---------------------------------------------------------------------------
def wf01_project_create() -> None:
"""Create project, link resource, add invariant."""
ctx = _WorkflowCtx()
ctx.setup()
try:
# Register resource first
r1 = run_cli(
"resource",
"add",
"git-checkout",
"local/hello-repo",
"--path",
ctx.repo_dir,
"--branch",
"main",
workspace=ctx.workspace,
)
if r1.returncode != 0:
_fail(f"resource add: {r1.stderr}")
# Create project with linked resource and invariant
r2 = run_cli(
"project",
"create",
"local/hello-project",
"--description",
"Hello World project",
"--resource",
"local/hello-repo",
"--invariant",
"All tests must pass before apply",
"--format",
"plain",
workspace=ctx.workspace,
)
if r2.returncode != 0:
_fail(f"project create rc={r2.returncode}\n{r2.stderr}")
if "hello-project" not in r2.stdout.lower():
_fail(f"project name not in output:\n{r2.stdout}")
print("wf01-project-create-ok")
finally:
ctx.teardown()
# ---------------------------------------------------------------------------
# Subcommand: validation-register (C2)
# ---------------------------------------------------------------------------
def wf01_validation_register() -> None:
"""Register a validation and attach it to the project (spec §1d-e)."""
ctx = _WorkflowCtx()
ctx.setup()
val_path = write_yaml(_VALIDATION_YAML)
try:
# Setup: resource + project
r_res = run_cli(
"resource",
"add",
"git-checkout",
"local/hello-repo",
"--path",
ctx.repo_dir,
"--branch",
"main",
workspace=ctx.workspace,
)
if r_res.returncode != 0:
_fail(f"resource add: {r_res.stderr}")
r_proj = run_cli(
"project",
"create",
"local/hello-project",
"--resource",
"local/hello-repo",
workspace=ctx.workspace,
)
if r_proj.returncode != 0:
_fail(f"project create: {r_proj.stderr}")
# Validation add
r_val = run_cli(
"validation",
"add",
"--config",
val_path,
"--format",
"plain",
workspace=ctx.workspace,
)
if r_val.returncode != 0:
_fail(f"validation add rc={r_val.returncode}\n{r_val.stderr}")
if "unit-tests" not in r_val.stdout.lower():
_fail(f"validation name not in output:\n{r_val.stdout}")
# Validation attach to project
r_attach = run_cli(
"validation",
"attach",
"--project",
"local/hello-project",
"local/hello-repo",
"local/unit-tests",
"--format",
"plain",
workspace=ctx.workspace,
)
if r_attach.returncode != 0:
_fail(f"validation attach rc={r_attach.returncode}\n{r_attach.stderr}")
print("wf01-validation-register-ok")
finally:
if val_path and os.path.exists(val_path):
os.unlink(val_path)
ctx.teardown()
# ---------------------------------------------------------------------------
# Subcommand: action-create
# ---------------------------------------------------------------------------
def wf01_action_create() -> None:
"""Create an action from YAML config."""
ctx = _WorkflowCtx()
ctx.setup()
try:
result = run_cli(
"action",
"create",
"--config",
ctx.yaml_path,
workspace=ctx.workspace,
)
if result.returncode != 0:
_fail(f"action create rc={result.returncode}\n{result.stderr}")
# Verify action is retrievable
show = run_cli(
"action",
"show",
"local/hello-world-fix",
"--format",
"plain",
workspace=ctx.workspace,
)
if show.returncode != 0:
_fail(f"action show rc={show.returncode}\n{show.stderr}")
if "hello-world-fix" not in show.stdout:
_fail(f"action not found:\n{show.stdout}")
print("wf01-action-create-ok")
finally:
ctx.teardown()
# ---------------------------------------------------------------------------
# Subcommand: post-apply-commit (sandbox-level check)
# ---------------------------------------------------------------------------
def wf01_post_apply_commit() -> None:
"""Verify sandbox commit creates a git commit in the target repo.
Exercises the GitWorktreeSandbox directly (same as M1 verification)
to confirm the fundamental apply mechanism works.
"""
repo_dir = init_test_git_repo()
sandbox = None
try:
sandbox = GitWorktreeSandbox(
resource_id="res-wf01",
original_path=repo_dir,
)
plan_id = "01HWFTEST000000000000000WF"
sandbox.create(plan_id=plan_id)
# Write a "bug fix" file in the sandbox
fix_file = sandbox.get_path("src/bugfix.py")
os.makedirs(os.path.dirname(fix_file), exist_ok=True)
with open(fix_file, "w") as f:
f.write(
"# Bug fix for hello-world issue\n"
"def fixed_function():\n"
" return True\n"
)
# Commit the sandbox changes
result = sandbox.commit("fix: resolve hello-world bug")
if not result.success:
_fail(f"sandbox commit failed: {result.error}")
if result.commit_ref is None:
_fail("no commit ref after sandbox commit")
# Verify commit in original repo's git log
log = subprocess.run(
["git", "log", "--oneline", "-5"],
cwd=repo_dir,
capture_output=True,
text=True,
check=True,
)
if "hello-world bug" not in log.stdout:
_fail(f"commit not found in git log:\n{log.stdout}")
# Verify file on disk in original
if not os.path.exists(os.path.join(repo_dir, "src", "bugfix.py")):
_fail("bugfix.py not present in original repo after apply")
print("wf01-post-apply-commit-ok")
finally:
if sandbox is not None:
sandbox.cleanup()
shutil.rmtree(repo_dir, ignore_errors=True)
# ---------------------------------------------------------------------------
# Main dispatcher
# ---------------------------------------------------------------------------
_COMMANDS: dict[str, Callable[[], None]] = {
"init": wf01_init,
"resource-register": wf01_resource_register,
"project-create": wf01_project_create,
"validation-register": wf01_validation_register,
"action-create": wf01_action_create,
"post-apply-commit": wf01_post_apply_commit,
}
def main() -> int:
"""Entry point called by Robot Framework ``Run Process``."""
if len(sys.argv) < 2:
print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>")
return 1
command = sys.argv[1]
handler = _COMMANDS.get(command)
if handler is None:
print(f"Unknown command: {command}")
return 1
handler()
return 0
if __name__ == "__main__":
sys.exit(main())
robot/helper_wf01_plan_tests.py
+502
View File
@@ -0,0 +1,502 @@
"""Robot helper for WF01 plan lifecycle integration tests.
Extracted from ``helper_wf01_hello_world.py`` (M5) to keep each module
under the 500-line limit mandated by CONTRIBUTING.md.
Contains: plan-lifecycle, plan-state-transitions, tree-explain-output,
diff-output tests.
Each subcommand is fully independent: it provisions and tears down its own
workspace, so tests may be run in any order or subset.
Each subcommand prints a sentinel on success.
Exit code 0 = pass, 1 = failure.
"""
from __future__ import annotations
import json
import os
import re
import shutil
import sys
from collections.abc import Callable
from pathlib import Path
from typing import NoReturn
# Ensure the src directory is on the import path.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
# Ensure robot/ is on the import path for helper_e2e_common.
_ROBOT = str(Path(__file__).resolve().parent)
if _ROBOT not in sys.path:
sys.path.insert(0, _ROBOT)
from helper_e2e_common import ( # noqa: E402
cleanup_workspace,
init_test_git_repo,
run_cli,
setup_workspace,
write_yaml,
)
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
_ACTION_YAML = """\
name: local/hello-world-fix
description: Fix a single bug in the hello-world project
strategy_actor: openai/gpt-4
execution_actor: openai/gpt-4
definition_of_done: The bug is fixed, tests pass, and the change is committed
automation_profile: manual
arguments:
- name: bug_description
type: string
required: true
description: Description of the bug to fix
"""
_ULID_RE = re.compile(r"\b([0-9A-HJKMNP-TV-Z]{26})\b")
def _fail(msg: str) -> NoReturn:
"""Print failure message and exit."""
print(f"FAIL: {msg}", file=sys.stderr)
raise SystemExit(1)
def _extract_plan_id(output: str) -> str | None:
"""Extract a ULID plan_id from plain CLI output."""
match = _ULID_RE.search(output)
return match.group(1) if match else None
# ---------------------------------------------------------------------------
# Shared workspace context
# ---------------------------------------------------------------------------
class _WorkflowCtx:
"""Holds shared state across workflow steps."""
def __init__(self) -> None:
self.workspace: str = ""
self.repo_dir: str = ""
self.yaml_path: str = ""
self.plan_id: str = ""
def setup(self) -> None:
self.workspace = setup_workspace(prefix="wf01_")
self.repo_dir = init_test_git_repo()
self.yaml_path = write_yaml(_ACTION_YAML)
def teardown(self) -> None:
if self.yaml_path and os.path.exists(self.yaml_path):
os.unlink(self.yaml_path)
if self.repo_dir:
shutil.rmtree(self.repo_dir, ignore_errors=True)
if self.workspace:
cleanup_workspace(self.workspace)
def setup_with_plan(self) -> str:
"""Run shared resource+project+action+plan-use setup, return plan_id.
Checks return codes on every setup command (M2 fix).
Passes --arg for required action arguments (M1 fix).
"""
r_res = run_cli(
"resource",
"add",
"git-checkout",
"local/hello-repo",
"--path",
self.repo_dir,
"--branch",
"main",
workspace=self.workspace,
)
if r_res.returncode != 0:
_fail(f"resource add: {r_res.stderr}")
r_proj = run_cli(
"project",
"create",
"local/hello-project",
"--resource",
"local/hello-repo",
workspace=self.workspace,
)
if r_proj.returncode != 0:
_fail(f"project create: {r_proj.stderr}")
r_act = run_cli(
"action",
"create",
"--config",
self.yaml_path,
workspace=self.workspace,
)
if r_act.returncode != 0:
_fail(f"action create: {r_act.stderr}")
r_use = run_cli(
"plan",
"use",
"local/hello-world-fix",
"local/hello-project",
"--automation-profile",
"manual",
"--arg",
"bug_description=Fix the hello-world greeting function",
"--format",
"plain",
workspace=self.workspace,
)
if r_use.returncode != 0:
_fail(f"plan use rc={r_use.returncode}\n{r_use.stderr}")
plan_id = _extract_plan_id(r_use.stdout)
if not plan_id:
_fail(f"could not extract plan_id:\n{r_use.stdout}")
self.plan_id = plan_id
return plan_id
# ---------------------------------------------------------------------------
# Subcommand: plan-lifecycle (the full manual workflow)
# ---------------------------------------------------------------------------
def wf01_plan_lifecycle() -> None:
"""Full plan lifecycle: use -> execute -> tree -> diff -> apply.
With mocked AI the plan lands in strategize/queued (or
strategize/complete depending on the mock), so execute, tree, diff,
and apply may return graceful "not ready" messages. The test
verifies no crashes (no Traceback / INTERNAL errors) and correct
CLI wiring.
"""
ctx = _WorkflowCtx()
ctx.setup()
try:
plan_id = ctx.setup_with_plan()
# --- Plan Status (verify plan exists) ---
r_status = run_cli(
"plan",
"status",
plan_id,
"--format",
"plain",
workspace=ctx.workspace,
)
if r_status.returncode != 0:
_fail(f"plan status rc={r_status.returncode}\n{r_status.stderr}")
if plan_id not in r_status.stdout:
_fail(f"plan_id not in status output:\n{r_status.stdout}")
if "strategize" not in r_status.stdout.lower():
_fail(f"'strategize' not in status:\n{r_status.stdout}")
# --- Plan Execute ---
r_exec = run_cli(
"plan",
"execute",
plan_id,
"--format",
"plain",
workspace=ctx.workspace,
)
combined_exec = r_exec.stdout + r_exec.stderr
if "INTERNAL" in combined_exec or "Traceback" in combined_exec:
_fail(f"plan execute crashed:\n{combined_exec}")
if r_exec.returncode > 1:
_fail(f"plan execute unexpected rc={r_exec.returncode}\n{combined_exec}")
# --- Plan Tree ---
r_tree = run_cli(
"plan",
"tree",
plan_id,
"--format",
"plain",
workspace=ctx.workspace,
)
combined_tree = r_tree.stdout + r_tree.stderr
if "INTERNAL" in combined_tree or "Traceback" in combined_tree:
_fail(f"plan tree crashed:\n{combined_tree}")
if r_tree.returncode > 1:
_fail(f"plan tree unexpected rc={r_tree.returncode}\n{combined_tree}")
# --- Plan Diff ---
r_diff = run_cli(
"plan",
"diff",
plan_id,
"--format",
"plain",
workspace=ctx.workspace,
)
combined_diff = r_diff.stdout + r_diff.stderr
if "INTERNAL" in combined_diff or "Traceback" in combined_diff:
_fail(f"plan diff crashed:\n{combined_diff}")
if r_diff.returncode > 1:
_fail(f"plan diff unexpected rc={r_diff.returncode}\n{combined_diff}")
# --- Plan Lifecycle-Apply ---
r_apply = run_cli(
"plan",
"lifecycle-apply",
plan_id,
"--format",
"plain",
workspace=ctx.workspace,
)
combined_apply = r_apply.stdout + r_apply.stderr
if "INTERNAL" in combined_apply or "Traceback" in combined_apply:
_fail(f"plan apply crashed:\n{combined_apply}")
if r_apply.returncode > 1:
_fail(f"plan apply unexpected rc={r_apply.returncode}\n{combined_apply}")
print("wf01-plan-lifecycle-ok")
finally:
ctx.teardown()
# ---------------------------------------------------------------------------
# Subcommand: plan-state-transitions
# ---------------------------------------------------------------------------
def wf01_plan_state_transitions() -> None:
"""Verify plan state transitions through the lifecycle.
Creates a plan, inspects its state, runs execute to attempt a
transition, and re-checks status to verify the phase changed
(or remained consistent if mock AI prevents progression).
"""
ctx = _WorkflowCtx()
ctx.setup()
try:
plan_id = ctx.setup_with_plan()
# Check initial state — should be in strategize phase
r_pre = run_cli(
"plan",
"status",
plan_id,
"--format",
"plain",
workspace=ctx.workspace,
)
if r_pre.returncode != 0:
_fail(f"plan status (pre-execute): {r_pre.stderr}")
if "strategize" not in r_pre.stdout.lower():
_fail(f"expected 'strategize' in status:\n{r_pre.stdout}")
if "queued" not in r_pre.stdout.lower():
_fail(f"expected 'queued' in status:\n{r_pre.stdout}")
# Execute to attempt state transition
r_exec = run_cli(
"plan",
"execute",
plan_id,
"--format",
"plain",
workspace=ctx.workspace,
)
combined_exec = r_exec.stdout + r_exec.stderr
if "INTERNAL" in combined_exec or "Traceback" in combined_exec:
_fail(f"plan execute crashed:\n{combined_exec}")
# Re-check status after execute attempt. Under mock AI the plan
# may still be in strategize (if mock doesn't complete strategy)
# or may have transitioned to execute phase. Either is valid;
# the key assertion is that the status command still works and
# the plan_id is present.
r_post = run_cli(
"plan",
"status",
plan_id,
"--format",
"plain",
workspace=ctx.workspace,
)
if r_post.returncode != 0:
_fail(f"plan status (post-execute): {r_post.stderr}")
if plan_id not in r_post.stdout:
_fail(f"plan_id not in post-execute status:\n{r_post.stdout}")
# Log state transition for diagnostic visibility
pre_state = (
r_pre.stdout.strip().split("\n")[-1] if r_pre.stdout.strip() else "unknown"
)
post_state = (
r_post.stdout.strip().split("\n")[-1]
if r_post.stdout.strip()
else "unknown"
)
print(
f" state transition: pre={pre_state!r} -> post={post_state!r}",
file=sys.stderr,
)
print("wf01-plan-state-transitions-ok")
finally:
ctx.teardown()
# ---------------------------------------------------------------------------
# Subcommand: tree-explain-output (C1 fix: now includes plan explain)
# ---------------------------------------------------------------------------
def wf01_tree_explain_output() -> None:
"""Verify plan tree and plan explain produce structured output.
Uses JSON format for tree to verify the output is parseable.
Invokes plan explain with a dummy decision_id — under mock AI no
decisions exist, so the test verifies graceful error handling
rather than content.
"""
ctx = _WorkflowCtx()
ctx.setup()
try:
plan_id = ctx.setup_with_plan()
# Plan tree — verify it runs without crash
r_tree = run_cli(
"plan",
"tree",
plan_id,
"--format",
"json",
workspace=ctx.workspace,
)
combined_tree = r_tree.stdout + r_tree.stderr
if "Traceback" in combined_tree:
_fail(f"plan tree crashed:\n{combined_tree}")
if r_tree.returncode == 0 and r_tree.stdout.strip():
# Under mock AI no decisions exist, so the CLI may print a
# plain-text "No decisions found" message instead of JSON.
# Only validate JSON when the output actually looks like a
# JSON document (starts with '{' or '[' on the first
# non-debug line).
_first_content = ""
for _line in r_tree.stdout.strip().splitlines():
_stripped = _line.strip()
# Skip structlog debug lines (e.g. "2026-... [debug ...]")
if re.match(r"^\d{4}-\d{2}-\d{2}\s", _stripped):
continue
_first_content = _stripped
break
if _first_content.startswith(("{", "[")):
try:
json.loads(_first_content)
except json.JSONDecodeError:
_fail(f"plan tree JSON not parseable:\n{r_tree.stdout}")
# Plan explain — verify it runs without crash (C1 fix).
# Under mock AI no real decisions exist, so we use a synthetic
# decision ULID. The command should return a user-facing error
# message (e.g. "decision not found") rather than a traceback.
r_explain = run_cli(
"plan",
"explain",
"01HWFTEST0000000000DECDE0",
"--format",
"plain",
workspace=ctx.workspace,
)
combined_explain = r_explain.stdout + r_explain.stderr
if "Traceback" in combined_explain:
_fail(f"plan explain crashed:\n{combined_explain}")
print("wf01-tree-explain-output-ok")
finally:
ctx.teardown()
# ---------------------------------------------------------------------------
# Subcommand: diff-output (M3 fix: verify return code or message)
# ---------------------------------------------------------------------------
def wf01_diff_output() -> None:
"""Verify plan diff runs without crashing.
With mocked AI the plan doesn't reach Execute phase, so diff
returns a graceful message rather than actual changeset content.
Asserts on either a specific "not ready" indicator or a non-zero
return code, in addition to absence of tracebacks (M3 fix).
"""
ctx = _WorkflowCtx()
ctx.setup()
try:
plan_id = ctx.setup_with_plan()
# Plan diff — verify no internal errors
r_diff = run_cli(
"plan",
"diff",
plan_id,
"--format",
"plain",
workspace=ctx.workspace,
)
combined = r_diff.stdout + r_diff.stderr
lower_combined = combined.lower()
if "internal" in lower_combined or "traceback" in lower_combined:
_fail(f"plan diff crashed:\n{combined}")
if r_diff.returncode > 1:
_fail(f"plan diff unexpected rc={r_diff.returncode}\n{combined}")
# Under mock AI the plan hasn't executed, so diff should either
# return rc != 0 or contain a message indicating no changes are
# available (e.g. "no changes", "not ready", "no diff").
# A silent success with empty output is also acceptable since
# there are genuinely no changes to diff. The key invariant is
# that the command does not produce an unhandled exception.
if r_diff.returncode == 0 and r_diff.stdout.strip():
# If there IS output on success, it should not contain
# unhandled error indicators.
lower_out = combined.lower()
if "traceback" in lower_out or "internal" in lower_out:
_fail(f"plan diff contains error markers:\n{combined}")
print("wf01-diff-output-ok")
finally:
ctx.teardown()
# ---------------------------------------------------------------------------
# Main dispatcher
# ---------------------------------------------------------------------------
_COMMANDS: dict[str, Callable[[], None]] = {
"plan-lifecycle": wf01_plan_lifecycle,
"plan-state-transitions": wf01_plan_state_transitions,
"tree-explain-output": wf01_tree_explain_output,
"diff-output": wf01_diff_output,
}
def main() -> int:
"""Entry point called by Robot Framework ``Run Process``."""
if len(sys.argv) < 2:
print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>")
return 1
command = sys.argv[1]
handler = _COMMANDS.get(command)
if handler is None:
print(f"Unknown command: {command}")
return 1
handler()
return 0
if __name__ == "__main__":
sys.exit(main())
robot/wf01_hello_world.robot
+93
View File
@@ -0,0 +1,93 @@
*** Settings ***
Documentation Workflow Example 1 — Hello World: fix a single bug using the manual automation profile.
... Exercises the complete manual-profile workflow: init, resource registration,
... project creation with invariant, validation registration and attachment,
... action creation, plan use/execute/tree/explain/diff/apply.
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_wf01_hello_world.py
${PLAN_HELPER} ${CURDIR}/helper_wf01_plan_tests.py
*** Test Cases ***
WF01 Init Workspace
[Documentation] Verify agents init creates a workspace configuration
${result}= Run Process ${PYTHON} ${HELPER} init cwd=${WORKSPACE} timeout=60s
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} wf01-init-ok
WF01 Register Git Checkout Resource
[Documentation] Register a git-checkout resource for the hello-world project
${result}= Run Process ${PYTHON} ${HELPER} resource-register cwd=${WORKSPACE} timeout=60s
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} wf01-resource-register-ok
WF01 Create Project With Invariant
[Documentation] Create a project, link a resource, and register an invariant
${result}= Run Process ${PYTHON} ${HELPER} project-create cwd=${WORKSPACE} timeout=60s
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} wf01-project-create-ok
WF01 Register And Attach Validation
[Documentation] Register a validation via YAML and attach it to the project (spec §1d-e)
${result}= Run Process ${PYTHON} ${HELPER} validation-register cwd=${WORKSPACE} timeout=60s
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} wf01-validation-register-ok
WF01 Create Action From YAML
[Documentation] Create a hello-world fix action from YAML config with manual profile
${result}= Run Process ${PYTHON} ${HELPER} action-create cwd=${WORKSPACE} timeout=60s
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} wf01-action-create-ok
WF01 Full Plan Lifecycle Manual Profile
[Documentation] Full plan lifecycle: use -> execute -> tree -> diff -> apply (manual profile)
${result}= Run Process ${PYTHON} ${PLAN_HELPER} plan-lifecycle cwd=${WORKSPACE} timeout=120s
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} wf01-plan-lifecycle-ok
WF01 Plan State Transitions
[Documentation] Verify plan state transitions through the manual-profile lifecycle
${result}= Run Process ${PYTHON} ${PLAN_HELPER} plan-state-transitions cwd=${WORKSPACE} timeout=120s
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} wf01-plan-state-transitions-ok
WF01 Tree And Explain Output Structure
[Documentation] Verify plan tree and plan explain produce structured output
${result}= Run Process ${PYTHON} ${PLAN_HELPER} tree-explain-output cwd=${WORKSPACE} timeout=120s
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} wf01-tree-explain-output-ok
WF01 Diff Output
[Documentation] Verify plan diff runs without internal errors
${result}= Run Process ${PYTHON} ${PLAN_HELPER} diff-output cwd=${WORKSPACE} timeout=120s
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} wf01-diff-output-ok
WF01 Post Apply Commit Exists
[Documentation] Verify post-apply commit exists in the target repository
${result}= Run Process ${PYTHON} ${HELPER} post-apply-commit cwd=${WORKSPACE} timeout=60s
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} wf01-post-apply-commit-ok
src/cleveragents/application/container.py
+11
View File
@@ -832,6 +832,17 @@ def get_container() -> Container:
return _container
def get_container_if_initialized() -> Container | None:
"""Return the global container **only** if already initialised.
Unlike :func:`get_container`, this will never create a new
:class:`Container` instance. It is intended for *optional* diagnostic
code that should degrade gracefully when the application has not yet
booted.
"""
return _container
def reset_container() -> None:
"""Reset the global container.
src/cleveragents/application/services/cost_budget_service.py
+10
View File
@@ -118,6 +118,16 @@ class CostBudgetService:
with self._lock:
return self._sessions.get(session_id)
@property
def active_sessions(self) -> dict[str, SessionCostBudget]:
"""Return a snapshot of all active session budgets.
Returns a deep copy — callers may read values without
interference from concurrent ``record_cost()`` mutations.
"""
with self._lock:
return {k: v.model_copy(deep=True) for k, v in self._sessions.items()}
def remove_session(self, session_id: str) -> None:
"""Remove all budget state for a session."""
with self._lock:
src/cleveragents/cli/commands/system.py
+67 -254
View File
@@ -14,29 +14,60 @@ import subprocess
import sys
import time
from datetime import UTC, datetime
from enum import StrEnum
from importlib.metadata import version as _pkg_version
from pathlib import Path
from typing import Any
from sqlalchemy import create_engine
from sqlalchemy import inspect as sa_inspect
import structlog
from cleveragents import __version__
from cleveragents.application.container import get_database_url
from cleveragents.application.services.lock_service import LockService
from cleveragents.application.container import get_container_if_initialized
from cleveragents.cli.commands.server import resolve_server_mode
from cleveragents.cli.commands.system_health import (
_check_active_plans as _check_active_plans,
)
from cleveragents.cli.commands.system_health import (
_check_async_worker_health as _check_async_worker_health,
)
from cleveragents.cli.commands.system_health import (
_check_cost_summary as _check_cost_summary,
)
from cleveragents.cli.commands.system_health import (
_check_error_patterns as _check_error_patterns,
)
from cleveragents.cli.commands.system_health import (
_check_index_health as _check_index_health,
)
from cleveragents.cli.commands.system_health import (
_check_performance_summary as _check_performance_summary,
)
from cleveragents.cli.commands.system_health import (
_check_stale_locks as _check_stale_locks,
)
from cleveragents.cli.commands.system_rendering import (
render_diagnostics_rich as render_diagnostics_rich,
)
from cleveragents.cli.commands.system_rendering import (
render_info_rich as render_info_rich,
)
from cleveragents.cli.commands.system_rendering import (
render_version_rich as render_version_rich,
)
from cleveragents.cli.commands.system_types import CheckStatus as CheckStatus
from cleveragents.config.settings import get_settings
from cleveragents.shared.redaction import mask_database_url
# ---------------------------------------------------------------------------
# Diagnostic check status
# ---------------------------------------------------------------------------
class CheckStatus(StrEnum):
"""Status of a single diagnostic check."""
OK = "ok"
WARN = "warn"
ERROR = "error"
_logger = structlog.get_logger(__name__)
__all__ = [
"CheckStatus",
"build_diagnostics_data",
"build_info_data",
"build_version_data",
"render_diagnostics_rich",
"render_info_rich",
"render_version_rich",
]
# ---------------------------------------------------------------------------
# Data builders (format-agnostic)
@@ -62,10 +93,9 @@ def _git_sha() -> str:
def _dep_version(package: str) -> str:
"""Return the installed version of *package*, or 'not installed'."""
try:
from importlib.metadata import version as pkg_version
return pkg_version(package)
return _pkg_version(package)
except Exception:
_logger.debug("dep_version_lookup_failed", package=package, exc_info=True)
return "not installed"
@@ -90,7 +120,6 @@ def build_version_data() -> dict[str, Any]:
def build_info_data() -> dict[str, Any]:
"""Assemble structured data for the ``info`` command."""
from cleveragents.config.settings import get_settings
settings = get_settings()
@@ -107,14 +136,19 @@ def build_info_data() -> dict[str, Any]:
# Storage sizes (best effort)
storage: dict[str, str] = {}
try:
db_path_str = db_url.replace("sqlite:///", "")
db_path = Path(db_path_str)
if db_path.exists():
# Handle both sqlite:/// and sqlite+aiosqlite:/// URL schemes.
if db_url.startswith("sqlite"):
db_path_str = db_url.split("///", 1)[-1] if "///" in db_url else ""
else:
db_path_str = ""
db_path = Path(db_path_str) if db_path_str else None
if db_path is not None and db_path.exists():
size_mb = db_path.stat().st_size / (1024 * 1024)
storage["db_size"] = f"{size_mb:.1f} MB"
else:
storage["db_size"] = "0 MB"
except Exception:
_logger.debug("db_size_check_failed", exc_info=True)
storage["db_size"] = "unknown"
log_dir = settings.log_dir
@@ -124,15 +158,13 @@ def build_info_data() -> dict[str, Any]:
else:
storage["logs"] = "0 MB"
from cleveragents.cli.commands.server import resolve_server_mode
server_mode = resolve_server_mode()
return {
"version": __version__,
"data_dir": str(data_dir),
"config_path": str(config_path),
"database": db_url,
"database": mask_database_url(db_url),
"server_mode": server_mode,
"platform": f"{platform.system()} {platform.release()} ({platform.machine()})",
"automation": settings.default_automation_profile,
@@ -167,7 +199,6 @@ def _check_config_file() -> dict[str, Any]:
def _check_data_dir() -> dict[str, Any]:
"""Check that the data directory exists and is writable."""
from cleveragents.config.settings import get_settings
settings = get_settings()
data_dir = settings.data_dir
@@ -190,13 +221,12 @@ def _check_data_dir() -> dict[str, Any]:
def _check_database() -> dict[str, Any]:
"""Check database connectivity."""
from cleveragents.config.settings import get_settings
settings = get_settings()
db_url = settings.database_url
if db_url.startswith("sqlite"):
db_path_str = db_url.replace("sqlite:///", "")
db_path_str = db_url.split("///", 1)[-1] if "///" in db_url else ""
db_path = Path(db_path_str)
if db_path.exists():
writable = os.access(db_path, os.W_OK)
@@ -228,7 +258,6 @@ def _check_database() -> dict[str, Any]:
def _check_providers() -> list[dict[str, Any]]:
"""Check provider API key configuration."""
from cleveragents.config.settings import get_settings
settings = get_settings()
results: list[dict[str, Any]] = []
@@ -316,7 +345,6 @@ def _check_git() -> dict[str, Any]:
def _check_file_permissions() -> dict[str, Any]:
"""Check file permissions on the data directory."""
from cleveragents.config.settings import get_settings
settings = get_settings()
data_dir = settings.data_dir
@@ -348,101 +376,6 @@ def _check_file_permissions() -> dict[str, Any]:
}
def _check_stale_locks() -> dict[str, Any]:
"""Check for stale (expired) concurrency locks."""
try:
from sqlalchemy.orm import Session, sessionmaker
db_url = get_database_url()
engine = create_engine(db_url, echo=False)
# Only check if the locks table exists
inspector = sa_inspect(engine)
if "locks" not in inspector.get_table_names():
return {
"name": "Stale locks",
"status": CheckStatus.OK,
"details": "locks table not yet created",
}
factory: sessionmaker[Session] = sessionmaker(
bind=engine, expire_on_commit=False
)
svc = LockService(session_factory=factory)
count = svc.count_stale_locks()
if count == 0:
return {
"name": "Stale locks",
"status": CheckStatus.OK,
"details": "0 stale locks",
}
return {
"name": "Stale locks",
"status": CheckStatus.WARN,
"details": f"{count} stale lock(s) found",
"recommendation": "Run lock cleanup or restart the service",
}
except Exception:
return {
"name": "Stale locks",
"status": CheckStatus.WARN,
"details": "unable to check",
}
def _check_async_worker_health() -> dict[str, Any]:
"""Check async worker configuration and health."""
try:
from cleveragents.config.settings import get_settings
settings = get_settings()
enabled = settings.async_enabled
if not enabled:
return {
"name": "Async workers",
"status": CheckStatus.OK,
"details": "disabled (async.enabled=false)",
}
return {
"name": "Async workers",
"status": CheckStatus.OK,
"details": (
f"enabled, max_workers={settings.async_max_workers}, "
f"poll_interval={settings.async_poll_interval}s"
),
}
except Exception:
return {
"name": "Async workers",
"status": CheckStatus.WARN,
"details": "unable to check",
}
def _check_error_patterns() -> dict[str, Any]:
"""Check the Error Pattern Database status."""
try:
from cleveragents.application.services.error_pattern_service import (
ErrorPatternService,
)
service = ErrorPatternService()
stats = service.get_statistics()
total = stats["total_patterns"]
return {
"name": "Error Pattern DB",
"status": CheckStatus.OK,
"details": (
f"{total} patterns, {stats['total_occurrences']} total occurrences"
),
}
except Exception:
return {
"name": "Error Pattern DB",
"status": CheckStatus.OK,
"details": "empty (no patterns recorded)",
}
def build_diagnostics_data() -> dict[str, Any]:
"""Run all diagnostic checks and return structured results."""
@@ -461,6 +394,14 @@ def build_diagnostics_data() -> dict[str, Any]:
checks.append(_check_async_worker_health())
checks.append(_check_error_patterns())
# Extended health categories (spec: Diagnostic Dashboard §2-5).
# Single container lookup shared across all extended checks (m3 DRY).
_container = get_container_if_initialized()
checks.extend(_check_index_health(_container))
checks.extend(_check_active_plans(_container))
checks.extend(_check_cost_summary(_container))
checks.extend(_check_performance_summary(_container))
elapsed = time.monotonic() - start
total = len(checks)
@@ -487,131 +428,3 @@ def build_diagnostics_data() -> dict[str, Any]:
"has_errors": error_count > 0,
"has_warnings": warn_count > 0,
}
# ---------------------------------------------------------------------------
# Rich rendering helpers
# ---------------------------------------------------------------------------
def render_version_rich(data: dict[str, Any]) -> None:
"""Print version information using Rich panels."""
from rich.panel import Panel
from cleveragents.cli.main import get_console
console = get_console()
# CLI Version panel
version_lines = [
"[bold]CleverAgents CLI[/bold]",
f"[blue]Version:[/blue] {data['version']}",
f"[blue]Channel:[/blue] {data['channel']}",
f"[blue]Python:[/blue] {data['python']}",
]
console.print(Panel("\n".join(version_lines), title="CLI Version", expand=False))
# Build panel
build_lines = [
f"[green]Build Date:[/green] {data['build_date']}",
f"[magenta]Commit:[/magenta] {data['commit']}",
f"[blue]Schema:[/blue] {data['schema']}",
f"[blue]Platform:[/blue] {data['platform']}",
]
console.print(Panel("\n".join(build_lines), title="Build", expand=False))
# Dependencies panel
deps = data.get("dependencies", {})
dep_lines = [f"[blue]{k}:[/blue] {v}" for k, v in deps.items()]
if dep_lines:
console.print(Panel("\n".join(dep_lines), title="Dependencies", expand=False))
console.print("[green]OK[/green] Version reported")
def render_info_rich(data: dict[str, Any]) -> None:
"""Print info using Rich panels."""
from rich.panel import Panel
from cleveragents.cli.main import get_console
console = get_console()
env_lines = [
f"[blue]Data Dir:[/blue] {data['data_dir']}",
f"[blue]Config:[/blue] {data['config_path']}",
f"[green]Database:[/green] {data['database']}",
f"[yellow]Server Mode:[/yellow] {data['server_mode']}",
f"[blue]Platform:[/blue] {data['platform']}",
]
console.print(Panel("\n".join(env_lines), title="Environment", expand=False))
runtime_lines = [
f"[magenta]Automation:[/magenta] {data['automation']}",
f"[blue]Providers:[/blue] {data['providers_configured']} configured",
f"[blue]Debug Mode:[/blue] {data['debug_mode']}",
]
console.print(Panel("\n".join(runtime_lines), title="Runtime", expand=False))
storage = data.get("storage", {})
if storage:
storage_lines = [f"[blue]{k}:[/blue] {v}" for k, v in storage.items()]
console.print(Panel("\n".join(storage_lines), title="Storage", expand=False))
console.print("[green]OK[/green] Environment details ready")
def render_diagnostics_rich(data: dict[str, Any]) -> None:
"""Print diagnostics using Rich panels and table."""
from rich.panel import Panel
from rich.table import Table
from cleveragents.cli.main import get_console
console = get_console()
# Checks table
table = Table(title="Checks", show_header=True, expand=False)
table.add_column("Check", style="cyan")
table.add_column("Status")
table.add_column("Details")
for check in data["checks"]:
status = check["status"]
if status == CheckStatus.OK:
status_text = "[green]OK[/green]"
elif status == CheckStatus.WARN:
status_text = "[yellow]WARN[/yellow]"
else:
status_text = "[red]ERROR[/red]"
table.add_row(check["name"], status_text, check.get("details", ""))
console.print(table)
# Summary panel
summary = data["summary"]
summary_lines = [
f"[blue]Checks:[/blue] {summary['total']} total",
f"[yellow]Warnings:[/yellow] {summary['warnings']}",
f"[red]Errors:[/red] {summary['errors']}",
f"[green]Duration:[/green] {summary['duration_s']}s",
]
console.print(Panel("\n".join(summary_lines), title="Summary", expand=False))
# Recommendations
recs = data.get("recommendations", [])
if recs:
rec_lines = [f"- {r}" for r in recs]
console.print(
Panel("\n".join(rec_lines), title="Recommendations", expand=False)
)
# Final status line
if data["has_errors"]:
console.print(f"[red]ERROR[/red] {summary['errors']} errors must be resolved")
elif data["has_warnings"]:
console.print(
f"[yellow]WARN[/yellow] {summary['warnings']} warnings require attention"
)
else:
console.print("[green]OK[/green] All checks passed")
src/cleveragents/cli/commands/system_health.py
+498
View File
@@ -0,0 +1,498 @@
"""Health-check functions for the diagnostic dashboard (§2-5)."""
from __future__ import annotations
import statistics
from datetime import UTC, datetime
from typing import Any
import structlog
from rich.markup import escape as rich_escape
from sqlalchemy import create_engine
from sqlalchemy import inspect as sa_inspect
from sqlalchemy.orm import Session, sessionmaker
from cleveragents.application.container import (
get_container_if_initialized,
get_database_url,
)
from cleveragents.application.services.lock_service import LockService
from cleveragents.cli.commands.system_types import CheckStatus
from cleveragents.config.settings import get_settings
from cleveragents.domain.models.core.plan import ProcessingState
__all__ = [
"_check_active_plans",
"_check_async_worker_health",
"_check_cost_summary",
"_check_error_patterns",
"_check_index_health",
"_check_performance_summary",
"_check_stale_locks",
]
_log = structlog.get_logger(__name__)
_MAX_LATENCY_SAMPLES = 10_000
_MAX_PLANS_FOR_TRACES = 50
# ---------------------------------------------------------------------------
# Index health (§2)
# ---------------------------------------------------------------------------
def _check_index_health(container: Any = None) -> list[dict[str, Any]]:
"""Check index health: text, vector, and graph backends."""
if container is None:
container = get_container_if_initialized()
if container is None:
_na = {"status": CheckStatus.WARN, "details": "N/A (container not initialised)"}
return [
{"name": "Text index", **_na},
{"name": "Vector index", **_na},
{"name": "Graph store", **_na},
]
results: list[dict[str, Any]] = []
# --- Text index ---
try:
text_backend = container.index_text_backend()
doc_count = text_backend.document_count
results.append(
{
"name": "Text index",
"status": CheckStatus.OK,
"details": f"{doc_count} documents indexed",
}
)
except Exception:
_log.debug("text_index_check_failed", exc_info=True)
results.append(
{
"name": "Text index",
"status": CheckStatus.WARN,
"details": "N/A (backend unavailable)",
}
)
# --- Vector index ---
try:
vector_backend = container.index_vector_backend()
emb_count = vector_backend.embedding_count
# Read configured dimensionality from settings instead of
# accessing private backend internals.
try:
dim = str(get_settings().vector_embeddings_dimension)
except Exception:
dim = "N/A"
results.append(
{
"name": "Vector index",
"status": CheckStatus.OK,
"details": f"{emb_count} embeddings, dimensionality={dim}",
}
)
except Exception:
_log.debug("vector_index_check_failed", exc_info=True)
results.append(
{
"name": "Vector index",
"status": CheckStatus.WARN,
"details": "N/A (backend unavailable)",
}
)
# --- Graph store ---
try:
graph_backend = container.index_graph_backend()
triple_ct = graph_backend.triple_count()
results.append(
{
"name": "Graph store",
"status": CheckStatus.OK,
"details": f"{triple_ct} triples stored",
}
)
except Exception:
_log.debug("graph_store_check_failed", exc_info=True)
results.append(
{
"name": "Graph store",
"status": CheckStatus.WARN,
"details": "N/A (backend unavailable)",
}
)
return results
# ---------------------------------------------------------------------------
# Active plans (§3)
# ---------------------------------------------------------------------------
def _check_active_plans(container: Any = None) -> list[dict[str, Any]]:
"""Check active plan overview: running, queued, and resource utilisation."""
try:
if container is None:
container = get_container_if_initialized()
if container is None:
raise RuntimeError("container not initialised")
service = container.plan_lifecycle_service()
plans = service.list_plans()
running = [p for p in plans if p.processing_state == ProcessingState.PROCESSING]
queued = [p for p in plans if p.processing_state == ProcessingState.QUEUED]
errored = [p for p in plans if p.processing_state == ProcessingState.ERRORED]
total = len(plans)
status = CheckStatus.WARN if errored else CheckStatus.OK
results: list[dict[str, Any]] = []
results.append(
{
"name": "Active plans",
"status": status,
"details": (
f"{len(running)} running, {len(queued)} queued, "
f"{len(errored)} errored, {total} total"
),
}
)
# Per-running-plan resource utilisation
# TODO(M6): Include token counts per plan once the domain model
# exposes cumulative token usage (currently not tracked at plan
# level — only at session/trace granularity).
for plan in running:
# Guard against None: phase/namespaced_name are Optional on the
# domain model because they may be unset for newly-created plans
# that haven't entered the strategize phase yet.
phase = plan.phase.value if plan.phase else "unknown"
plan_name = plan.namespaced_name.name if plan.namespaced_name else "unnamed"
subplan_ct = len(plan.subplan_statuses)
results.append(
{
"name": f"Plan: {rich_escape(plan_name)}",
"status": CheckStatus.OK,
"details": f"phase={phase}, subplans={subplan_ct}",
}
)
return results
except Exception:
_log.debug("active_plans_check_failed", exc_info=True)
return [
{
"name": "Active plans",
"status": CheckStatus.WARN,
"details": "N/A (plan service unavailable)",
}
]
# ---------------------------------------------------------------------------
# Cost summary (§4)
# ---------------------------------------------------------------------------
def _check_cost_summary(container: Any = None) -> list[dict[str, Any]]:
"""Check cost summary: session costs, budget utilisation."""
try:
if container is None:
container = get_container_if_initialized()
if container is None:
raise RuntimeError("container not initialised")
cost_service = container.cost_budget_service()
# Use the public accessor for session budget data.
sessions = cost_service.active_sessions
if not sessions:
return [
{
"name": "Cost summary",
"status": CheckStatus.OK,
"details": "no active sessions (cost tracking idle)",
}
]
total_cost = sum(b.total_cost for b in sessions.values())
session_count = len(sessions)
# Budget utilisation from sessions with budgets configured.
# Use cost of only budgeted sessions as numerator so the ratio
# is meaningful (not total_cost which includes unbounded sessions).
# TODO(#580): Per-provider cost breakdown is a known gap.
# SessionCostBudget does not yet track per-provider spend;
# implement a public API on the cost model when provider-level
# attribution is available.
budget_sessions = [
b
for b in sessions.values()
if b.max_cost_usd is not None and b.max_cost_usd > 0
]
utilisation_pct = "N/A"
if budget_sessions:
budget_cost = sum(b.total_cost for b in budget_sessions)
total_budget = sum(
b.max_cost_usd for b in budget_sessions if b.max_cost_usd is not None
)
if total_budget > 0:
utilisation_pct = f"{(budget_cost / total_budget * 100):.1f}%"
results: list[dict[str, Any]] = []
results.append(
{
"name": "Cost summary",
"status": CheckStatus.OK,
"details": (
f"${total_cost:.2f} across {session_count} sessions, "
f"budget utilisation={utilisation_pct}"
),
}
)
return results
except Exception:
_log.debug("cost_summary_check_failed", exc_info=True)
return [
{
"name": "Cost summary",
"status": CheckStatus.WARN,
"details": "N/A (cost service unavailable)",
}
]
# ---------------------------------------------------------------------------
# Performance summary (§5)
# ---------------------------------------------------------------------------
def _check_performance_summary(container: Any = None) -> list[dict[str, Any]]:
"""Check performance: duration, latency percentiles, context times."""
try:
if container is None:
container = get_container_if_initialized()
if container is None:
raise RuntimeError("container not initialised")
lifecycle = container.plan_lifecycle_service()
trace_service = container.trace_service()
plans = lifecycle.list_plans()
completed_plans = [
p
for p in plans
if p.processing_state in (ProcessingState.COMPLETE, ProcessingState.APPLIED)
]
# Average plan duration from completed plans with timestamps.
def _to_naive_utc(dt: datetime | None) -> datetime:
"""Normalize to naive UTC for safe subtraction."""
if dt is None:
raise TypeError("Expected datetime, got None")
if dt.tzinfo is not None:
return dt.astimezone(UTC).replace(tzinfo=None)
return dt
durations: list[float] = []
for plan in completed_plans:
ts = plan.timestamps
start_t = ts.strategize_started_at or ts.created_at
end_t = ts.applied_at or ts.execute_completed_at or ts.updated_at
_pid = str(getattr(plan.identity, "plan_id", "unknown"))
try:
delta = (_to_naive_utc(end_t) - _to_naive_utc(start_t)).total_seconds()
except Exception:
_log.debug("duration_calc_failed", plan_id=_pid)
continue
if delta > 0:
durations.append(delta)
else:
_log.debug("negative_duration_skipped", plan_id=_pid, delta_s=delta)
avg_duration = "N/A"
if durations:
avg = sum(durations) / len(durations)
avg_duration = f"{avg:.1f}s"
# Collect latency data from a sample of recent plans to avoid
# N+1 queries across all plans (M2 fix).
all_latencies: list[float] = []
# Prefix slice: relies on list_plans() returning roughly recent-first
# ordering. A random sample would be more representative but adds
# complexity for a diagnostic display.
sampled_plans = plans[:_MAX_PLANS_FOR_TRACES]
for plan in sampled_plans:
if len(all_latencies) >= _MAX_LATENCY_SAMPLES:
break
plan_id = plan.identity.plan_id if plan.identity else None
if plan_id:
try:
traces = trace_service.get_traces(plan_id)
for t in traces:
all_latencies.append(t.latency_ms)
if len(all_latencies) >= _MAX_LATENCY_SAMPLES:
break
except Exception:
_log.debug(
"trace_fetch_failed",
plan_id=plan_id,
)
latency_str = "N/A"
if len(all_latencies) == 1:
val = all_latencies[0]
latency_str = f"p50={val:.0f}ms, p95={val:.0f}ms, p99={val:.0f}ms"
elif len(all_latencies) >= 2:
q = statistics.quantiles(all_latencies, n=100)
p50, p95, p99 = q[49], q[94], q[98]
latency_str = f"p50={p50:.0f}ms, p95={p95:.0f}ms, p99={p99:.0f}ms"
# Context build time: not yet wired to a metrics source.
# TODO(#813): wire to ACMS context assembly latency metrics
# once the metrics pipeline is implemented.
context_str = "Context build time tracking not yet implemented"
results: list[dict[str, Any]] = []
results.append(
{
"name": "Performance: plan duration",
"status": CheckStatus.OK,
"details": (
f"avg={avg_duration} "
f"(from {len(durations)}/{len(completed_plans)} completed plans)"
),
}
)
results.append(
{
"name": "Performance: tool call latency",
"status": CheckStatus.OK,
"details": (latency_str if all_latencies else "N/A (no trace data)"),
}
)
results.append(
{
"name": "Performance: context build time",
"status": CheckStatus.WARN,
"details": context_str,
}
)
return results
except Exception:
_log.debug("performance_summary_failed", exc_info=True)
return [
{
"name": "Performance summary",
"status": CheckStatus.WARN,
"details": "N/A (performance data unavailable)",
}
]
# ---------------------------------------------------------------------------
# Infrastructure checks (also extracted from system.py for line budget)
# ---------------------------------------------------------------------------
def _check_stale_locks() -> dict[str, Any]:
"""Check for stale (expired) concurrency locks."""
try:
db_url = get_database_url()
engine = create_engine(db_url, echo=False)
# Only check if the locks table exists
inspector = sa_inspect(engine)
if "locks" not in inspector.get_table_names():
return {
"name": "Stale locks",
"status": CheckStatus.OK,
"details": "locks table not yet created",
}
factory: sessionmaker[Session] = sessionmaker(
bind=engine, expire_on_commit=False
)
svc = LockService(session_factory=factory)
count = svc.count_stale_locks()
if count == 0:
return {
"name": "Stale locks",
"status": CheckStatus.OK,
"details": "0 stale locks",
}
return {
"name": "Stale locks",
"status": CheckStatus.WARN,
"details": f"{count} stale lock(s) found",
"recommendation": "Run lock cleanup or restart the service",
}
except Exception:
_log.debug("stale_locks_check_failed", exc_info=True)
return {
"name": "Stale locks",
"status": CheckStatus.WARN,
"details": "unable to check",
}
def _check_async_worker_health() -> dict[str, Any]:
"""Check async worker configuration and health."""
try:
settings = get_settings()
enabled = settings.async_enabled
if not enabled:
return {
"name": "Async workers",
"status": CheckStatus.OK,
"details": "disabled (async.enabled=false)",
}
return {
"name": "Async workers",
"status": CheckStatus.OK,
"details": (
f"enabled, max_workers={settings.async_max_workers}, "
f"poll_interval={settings.async_poll_interval}s"
),
}
except Exception:
_log.debug("async_worker_check_failed", exc_info=True)
return {
"name": "Async workers",
"status": CheckStatus.WARN,
"details": "unable to check",
}
def _check_error_patterns() -> dict[str, Any]:
"""Check the Error Pattern Database status."""
try:
container = get_container_if_initialized()
if container is None:
return {
"name": "Error Pattern DB",
"status": CheckStatus.WARN,
"details": "N/A (container not initialised)",
}
service = container.error_pattern_service()
stats = service.get_statistics()
total = stats["total_patterns"]
return {
"name": "Error Pattern DB",
"status": CheckStatus.OK,
"details": (
f"{total} patterns, {stats['total_occurrences']} total occurrences"
),
}
except Exception:
_log.debug("error_pattern_check_failed", exc_info=True)
return {
"name": "Error Pattern DB",
"status": CheckStatus.WARN,
"details": "unable to check",
}
src/cleveragents/cli/commands/system_rendering.py
+127
View File
@@ -0,0 +1,127 @@
"""Rich rendering helpers for system commands (version, info, diagnostics).
Extracted from ``system.py`` to keep each module under the 500-line limit
mandated by CONTRIBUTING.md.
"""
from __future__ import annotations
from typing import Any
from rich.panel import Panel
from rich.table import Table
from cleveragents.cli.commands.system_types import CheckStatus
from cleveragents.cli.main import get_console
__all__ = ["render_diagnostics_rich", "render_info_rich", "render_version_rich"]
def render_version_rich(data: dict[str, Any]) -> None:
"""Print version information using Rich panels."""
console = get_console()
# CLI Version panel
version_lines = [
"[bold]CleverAgents CLI[/bold]",
f"[blue]Version:[/blue] {data['version']}",
f"[blue]Channel:[/blue] {data['channel']}",
f"[blue]Python:[/blue] {data['python']}",
]
console.print(Panel("\n".join(version_lines), title="CLI Version", expand=False))
# Build panel
build_lines = [
f"[green]Build Date:[/green] {data['build_date']}",
f"[magenta]Commit:[/magenta] {data['commit']}",
f"[blue]Schema:[/blue] {data['schema']}",
f"[blue]Platform:[/blue] {data['platform']}",
]
console.print(Panel("\n".join(build_lines), title="Build", expand=False))
# Dependencies panel
deps = data.get("dependencies", {})
dep_lines = [f"[blue]{k}:[/blue] {v}" for k, v in deps.items()]
if dep_lines:
console.print(Panel("\n".join(dep_lines), title="Dependencies", expand=False))
console.print("[green]OK[/green] Version reported")
def render_info_rich(data: dict[str, Any]) -> None:
"""Print info using Rich panels."""
console = get_console()
env_lines = [
f"[blue]Data Dir:[/blue] {data['data_dir']}",
f"[blue]Config:[/blue] {data['config_path']}",
f"[green]Database:[/green] {data['database']}",
f"[yellow]Server Mode:[/yellow] {data['server_mode']}",
f"[blue]Platform:[/blue] {data['platform']}",
]
console.print(Panel("\n".join(env_lines), title="Environment", expand=False))
runtime_lines = [
f"[magenta]Automation:[/magenta] {data['automation']}",
f"[blue]Providers:[/blue] {data['providers_configured']} configured",
f"[blue]Debug Mode:[/blue] {data['debug_mode']}",
]
console.print(Panel("\n".join(runtime_lines), title="Runtime", expand=False))
storage = data.get("storage", {})
if storage:
storage_lines = [f"[blue]{k}:[/blue] {v}" for k, v in storage.items()]
console.print(Panel("\n".join(storage_lines), title="Storage", expand=False))
console.print("[green]OK[/green] Environment details ready")
def render_diagnostics_rich(data: dict[str, Any]) -> None:
"""Print diagnostics using Rich panels and table."""
console = get_console()
# Checks table
table = Table(title="Checks", show_header=True, expand=False)
table.add_column("Check", style="cyan")
table.add_column("Status")
table.add_column("Details")
for check in data["checks"]:
status = CheckStatus(check["status"])
if status == CheckStatus.OK:
status_text = "[green]OK[/green]"
elif status == CheckStatus.WARN:
status_text = "[yellow]WARN[/yellow]"
else:
status_text = "[red]ERROR[/red]"
table.add_row(check["name"], status_text, check.get("details", ""))
console.print(table)
# Summary panel
summary = data["summary"]
summary_lines = [
f"[blue]Checks:[/blue] {summary['total']} total",
f"[yellow]Warnings:[/yellow] {summary['warnings']}",
f"[red]Errors:[/red] {summary['errors']}",
f"[green]Duration:[/green] {summary['duration_s']}s",
]
console.print(Panel("\n".join(summary_lines), title="Summary", expand=False))
# Recommendations
recs = data.get("recommendations", [])
if recs:
rec_lines = [f"- {r}" for r in recs]
console.print(
Panel("\n".join(rec_lines), title="Recommendations", expand=False)
)
# Final status line
if data["has_errors"]:
console.print(f"[red]ERROR[/red] {summary['errors']} errors must be resolved")
elif data["has_warnings"]:
console.print(
f"[yellow]WARN[/yellow] {summary['warnings']} warnings require attention"
)
else:
console.print("[green]OK[/green] All checks passed")
src/cleveragents/cli/commands/system_types.py
+29
View File
@@ -0,0 +1,29 @@
"""Shared types for system command modules.
Provides ``CheckStatus`` used by ``system.py``, ``system_health.py``, and
``system_rendering.py`` without circular imports.
"""
from __future__ import annotations
from enum import StrEnum
from typing import NotRequired, TypedDict
__all__ = ["CheckResult", "CheckStatus"]
class CheckStatus(StrEnum):
"""Status of a single diagnostic check."""
OK = "ok"
WARN = "warn"
ERROR = "error"
class CheckResult(TypedDict):
"""A single diagnostic check result."""
name: str
status: CheckStatus
details: str
recommendation: NotRequired[str]
src/cleveragents/shared/redaction.py
+1 -1
View File
@@ -203,7 +203,7 @@ def mask_database_url(url: str) -> str:
The URL with any embedded password masked.
"""
if not url:
return url
return url or ""
if url.startswith("sqlite"):
return url
# Pattern: scheme://user:password@host...