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

feature/m8-tui-personas #976

Merged
aditya merged 3 commits from feature/m8-tui-personas into master 2026-03-18 10:21:05 +00:00
Conversation 10 Commits 3 Files Changed 41
+3427 -89
41 changed files with 3427 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+8
View File
@@ -48,6 +48,14 @@
outcomes while tracebacks/unexpected internal failures still fail.
- Runtime enforcement of `response_format` in provider invocation remains
planned and tracked via TODO comments in runtime code. (#650)
- Added interactive TUI persona and input-mode support with a dedicated
`agents tui` entry point and Textual app scaffolding. Personas are now
managed as local YAML configs with per-session binding/state, and input
handling supports Normal (`@` references), Command (`/` slash commands),
and Shell (`!` passthrough) flows with picker/overlay components and fuzzy
matching primitives. Includes Behave feature coverage for TUI persona/input
behavior, Robot TUI smoke validation, and an ASV fuzzy-reference benchmark.
(#695)
- Added four CLI-based integration test cases to M5 E2E verification suite
for v3.4.0 milestone acceptance criteria validation. Tests exercise
`project create`, `resource add git-checkout`, `project link-resource`, and
asv/benchmarks/tui_reference_fuzzy_bench.py
+26
View File
@@ -0,0 +1,26 @@
"""ASV benchmark for TUI fuzzy reference ranking."""
from __future__ import annotations
from typing import ClassVar
from cleveragents.tui.search.fuzzy import rank_candidates
class TuiReferenceFuzzyBench:
"""Bench weighted fuzzy ranking on a medium candidate set."""
params: ClassVar[list[list[object]]] = [
[1_000, 10_000],
["module_42/file_42", "zzzzzz-no-match-query"],
]
param_names: ClassVar[list[str]] = ["candidate_count", "query"]
def setup(self, candidate_count: int) -> None:
self.values = [
f"src/module_{idx}/file_{idx}.py" for idx in range(candidate_count)
]
def time_rank_candidates(self, candidate_count: int, query: str) -> None:
del candidate_count
rank_candidates(query, self.values, limit=20)
features/mocks/fake_repl_input.py
+22
View File
@@ -0,0 +1,22 @@
"""Shared fake input helper for REPL Behave scenarios."""
from __future__ import annotations
class FakeReplInput:
"""Feed predefined lines and then terminate input with EOFError."""
def __init__(self, lines: list[str], *, interrupt_token: str | None = None) -> None:
self._lines = list(lines)
self._index = 0
self._interrupt_token = interrupt_token
def __call__(self, prompt: str = "") -> str:
del prompt
if self._index >= len(self._lines):
raise EOFError
line = self._lines[self._index]
self._index += 1
if self._interrupt_token is not None and line == self._interrupt_token:
raise KeyboardInterrupt
return line
features/repl_input_modes.feature
+220
View File
@@ -0,0 +1,220 @@
Feature: REPL input modes and persona controls
The interactive REPL supports slash commands, shell passthrough, and
inline @ reference expansion.
Scenario: Slash persona commands create and activate personas
Given a temporary REPL config directory
When I run the REPL with input lines
| line |
| /persona create dev --actor local/mock-default |
| /persona list |
| /persona set dev |
| /persona list |
Then the REPL mode output should contain "Created persona: dev"
And the REPL mode output should contain "Active persona: dev"
And the REPL mode output should contain "dev -> local/mock-default"
Scenario: Shell mode runs host shell commands
Given a temporary REPL config directory
When I run the REPL with input lines
| line |
| !echo hello |
Then the REPL mode output should contain "hello"
Scenario: Shell mode blocks dangerous commands without confirmation
Given a temporary REPL config directory
When I run the REPL with input lines
| line |
| !rm -rf / |
| n |
Then the REPL mode output should contain "Blocked dangerous shell command."
Scenario: @ references are expanded with canonical category prefixes
Given a deterministic REPL reference catalog fixture
When I expand REPL references for "info @README.md"
Then the expanded REPL line should contain "@file:README.md"
Scenario: Slash persona shorthand switches active persona
Given a temporary REPL config directory
When I run the REPL with input lines
| line |
| /persona create reviewer --actor local/mock-default |
| /persona reviewer |
Then the REPL mode output should contain "Active persona: reviewer"
Scenario: CLI persona model rejects ANSI/control characters in names
When I validate CLI persona name "bad\\x1bname"
Then CLI persona validation should fail with "path/control"
Scenario: CLI persona model preserves color field
When I validate CLI persona payload with color "red"
Then CLI persona color should be "red"
Scenario: Persona export import and delete workflow
Given a temporary REPL config directory
And a temporary persona export path
When I run the REPL with input lines
| line |
| /persona create dev --actor local/mock-default |
| /persona export dev {export_path} |
| /persona delete dev |
| /persona import {export_path} |
| /persona set dev |
Then the REPL mode output should contain "Exported persona:"
And the REPL mode output should contain "Deleted persona: dev"
And the REPL mode output should contain "Imported persona: dev"
And the REPL mode output should contain "Active persona: dev"
Scenario: Persona create rejects path traversal names
Given a temporary REPL config directory
When I run the REPL with input lines
| line |
| /persona create ../../etc/cron.d/evil --actor local/mock-default |
Then the REPL mode output should contain "name must not contain path/control separators"
Scenario: Persona export rejects absolute path targets
Given a temporary REPL config directory
And an absolute persona export path
When I run the REPL with input lines
| line |
| /persona create dev --actor local/mock-default |
| /persona export dev {export_path} |
Then the REPL mode output should contain "Export path must be relative to current working directory"
Scenario: Persona export rejects parent directory escape
Given a temporary REPL config directory
When I run the REPL with input lines
| line |
| /persona create dev --actor local/mock-default |
| /persona export dev ../outside.yaml |
Then the REPL mode output should contain "Export path must stay within working directory"
Scenario: Persona import rejects absolute path targets
Given a temporary REPL config directory
And an absolute persona import file path
When I run the REPL with input lines
| line |
| /persona import {absolute_import_path} |
Then the REPL mode output should contain "Import path must be relative to current working directory"
Scenario: Persona binding is independent per REPL session
Given a temporary REPL config directory
When I run the REPL with input lines
| line |
| /persona create dev --actor local/mock-default |
| /persona create reviewer --actor local/mock-default |
| /session new work |
| /session switch work |
| /persona set dev |
| /session switch default |
| /persona set reviewer |
| /session list |
Then the REPL mode output should contain "work (persona=dev)"
And the REPL mode output should contain "default (persona=reviewer)"
Scenario: Deleting persona resets all sessions using it
Given a temporary REPL config directory
When I run the REPL with input lines
| line |
| /persona create dev --actor local/mock-default |
| /session new work |
| /session switch work |
| /persona set dev |
| /session switch default |
| /persona set dev |
| /persona delete dev |
| /session list |
Then the REPL mode output should contain "work (persona=default)"
And the REPL mode output should contain "default (persona=default)"
Scenario: Slash persona commands surface validation and usage errors
Given a temporary REPL config directory
And an invalid persona import file path
When I run the REPL with input lines
| line |
| /persona set missing |
| /persona create dev |
| /persona delete default |
| /persona import {invalid_import_path} |
Then the REPL mode output should contain "Persona not found: missing"
And the REPL mode output should contain "Usage: /persona create <name> --actor <namespace/name>"
And the REPL mode output should contain "Cannot delete default persona."
And the REPL mode output should contain "Invalid persona YAML."
Scenario: Persona create parser accepts optional fields and validates cycle-order
When I parse persona create tokens "/persona create dev --actor local/mock-default --description fast --icon robot --greeting hi --cycle-order 7"
Then parsed persona name should be "dev"
And parsed persona actor should be "local/mock-default"
And parsed persona cycle order should be 7
When I parse persona create tokens "/persona create dev --actor local/mock-default --cycle-order nope"
Then persona create parse should fail with "--cycle-order must be an integer"
Scenario: Shell helper handles empty, disabled, and timeout paths
When I run REPL shell helper with empty command
Then REPL shell helper exit code should be 2
And REPL shell helper output should contain "Shell mode requires a command after '!'."
When I run REPL shell helper with command "echo hi" and shell disabled
Then REPL shell helper exit code should be 2
And REPL shell helper output should contain "Shell mode is disabled by configuration."
When I run REPL shell helper with command "echo hi" and subprocess timeout
Then REPL shell helper exit code should be 124
And REPL shell helper output should contain "Shell command timed out after 30s."
Scenario: Slash command parser reports errors and session usage guidance
Given slash command test sessions
When I handle slash text that raises parser error
Then slash handler exit code should be 2
And slash handler output should contain "Parse error:"
When I handle empty slash text
Then slash handler exit code should be 2
And slash handler output should contain "Empty slash command."
When I handle slash text "session new"
Then slash handler exit code should be 2
And slash handler output should contain "Usage:"
Scenario: Reference helpers cover catalog, fuzzy, and suggestion branches
Given a rich REPL reference fixture
When I build REPL reference catalog from fixture
Then REPL catalog should include fixture actor tool skill and env entries
And REPL catalog should include non-relative walk file path
When I evaluate REPL fuzzy and suggestion helpers
Then REPL helper evaluation should pass
Scenario: Slash helper covers picker and unknown command branches
Given slash command test sessions
And slash command test registry with personas
When I handle picker slash text "persona pick" using input "x"
Then slash handler exit code should be 2
And slash handler output should contain "Invalid selection."
When I handle picker slash text "persona pick" using input "9"
Then slash handler exit code should be 2
And slash handler output should contain "Selection out of range."
When I handle picker slash text "persona pick reviewer" using input "1"
Then slash handler exit code should be 0
And slash handler output should contain "Active persona: reviewer"
When I handle slash text "persona pick zzz"
Then slash handler exit code should be 1
And slash handler output should contain "No personas matched 'zzz'."
When I handle slash text "persona import"
Then slash handler exit code should be 2
And slash handler output should contain "Usage: /persona import <path>"
When I handle slash text "session new default"
Then slash handler exit code should be 1
And slash handler output should contain "Session already exists: default"
When I handle slash text "session switch missing"
Then slash handler exit code should be 1
And slash handler output should contain "Session not found: missing"
When I handle slash text "session boom"
Then slash handler exit code should be 2
And slash handler output should contain "Unknown session command:"
When I handle slash text "persona boom"
Then slash handler exit code should be 1
And slash handler output should contain "Persona not found: boom"
When I handle slash text "foo bar"
Then slash handler exit code should be 2
And slash handler output should contain "Unknown slash command:"
Scenario: Shell helper prints stderr and returns subprocess code
When I run REPL shell helper with command "echo hi" and subprocess stderr "boom" code 7
Then REPL shell helper exit code should be 7
And REPL shell helper output should contain "boom"
features/steps/m1_sourcecode_smoke_steps.py
+16 -25
View File
@@ -10,6 +10,7 @@ import contextlib
import json
from datetime import datetime
from pathlib import Path
from typing import Any
from unittest.mock import MagicMock, patch
import yaml
@@ -35,24 +36,6 @@ _FIXTURES_DIR = Path(__file__).resolve().parents[1] / "fixtures" / "m1"
_PLAN_ULID = "01M1SM0KE00000000000000001"
def _register_patcher_cleanup(context: Context, patcher: object) -> None:
"""Register a patcher's stop() with Behave scenario cleanup."""
stop = getattr(patcher, "stop", None)
if not callable(stop):
return
def _safe_stop() -> None:
with contextlib.suppress(RuntimeError):
stop()
if hasattr(context, "add_cleanup"):
context.add_cleanup(_safe_stop)
return
if not hasattr(context, "_cleanup_handlers"):
context._cleanup_handlers = []
context._cleanup_handlers.append(_safe_stop)
def _make_m1_plan(
*,
name: str = "local/m1-smoke-plan",
@@ -105,6 +88,18 @@ def _make_m1_action(
)
def _register_patcher_cleanup(context: Context, patcher: Any) -> None:
"""Start a patcher and ensure it is always stopped."""
patcher.start()
context.add_cleanup(_safe_stop_patcher, patcher)
def _safe_stop_patcher(patcher: Any) -> None:
"""Best-effort patcher stop for repeated/failed scenarios."""
with contextlib.suppress(RuntimeError):
patcher.stop()
# ---------------------------------------------------------------------------
# Background
# ---------------------------------------------------------------------------
@@ -153,16 +148,14 @@ def step_m1_smoke_mock_service(context: Context) -> None:
context.link_patcher = patch(
"cleveragents.cli.commands.project._get_resource_link_repo",
)
context.plan_patcher.start()
_register_patcher_cleanup(context, context.plan_patcher)
context.action_patcher.start()
_register_patcher_cleanup(context, context.action_patcher)
context.mock_project_repo = context.project_patcher.start()
_register_patcher_cleanup(context, context.project_patcher)
context.add_cleanup(_safe_stop_patcher, context.project_patcher)
context.mock_resource_svc = context.resource_patcher.start()
_register_patcher_cleanup(context, context.resource_patcher)
context.add_cleanup(_safe_stop_patcher, context.resource_patcher)
context.mock_link_repo = context.link_patcher.start()
_register_patcher_cleanup(context, context.link_patcher)
context.add_cleanup(_safe_stop_patcher, context.link_patcher)
context.last_result = None
context.captured_plan_id = None
@@ -476,7 +469,6 @@ def step_m1_plan_in_strategize(context: Context) -> None:
"cleveragents.cli.commands.plan._get_plan_executor",
return_value=MagicMock(),
)
context._m1_executor_patcher.start()
_register_patcher_cleanup(context, context._m1_executor_patcher)
@@ -528,7 +520,6 @@ def step_m1_plan_with_changeset(context: Context) -> None:
"cleveragents.cli.commands.plan._get_apply_service",
return_value=mock_apply_svc,
)
context.apply_patcher.start()
_register_patcher_cleanup(context, context.apply_patcher)
features/steps/m4_correction_subplan_smoke_steps.py
+12 -21
View File
@@ -41,24 +41,6 @@ _PLAN_ULID = "01M4SM0KE00000000000000001"
_DECISION_ULID = "01M4DEC00000000000000000001"
def _register_patcher_cleanup(context: Context, patcher: object) -> None:
"""Register a patcher's stop() with Behave scenario cleanup."""
stop = getattr(patcher, "stop", None)
if not callable(stop):
return
def _safe_stop() -> None:
with contextlib.suppress(RuntimeError):
stop()
if hasattr(context, "add_cleanup"):
context.add_cleanup(_safe_stop)
return
if not hasattr(context, "_cleanup_handlers"):
context._cleanup_handlers = []
context._cleanup_handlers.append(_safe_stop)
def _make_m4_plan(
*,
phase: PlanPhase = PlanPhase.EXECUTE,
@@ -96,6 +78,18 @@ def _make_m4_plan(
)
def _register_patcher_cleanup(context: Context, patcher: object) -> None:
"""Start a patcher and always stop it via Behave cleanup."""
patcher.start()
context.add_cleanup(_safe_stop_patcher, patcher)
def _safe_stop_patcher(patcher: object) -> None:
"""Stop patcher safely even if it was already stopped."""
with contextlib.suppress(RuntimeError):
patcher.stop()
# ---------------------------------------------------------------------------
# Background
# ---------------------------------------------------------------------------
@@ -117,7 +111,6 @@ def step_m4_smoke_mock_service(context: Context) -> None:
"cleveragents.cli.commands.plan._get_lifecycle_service",
return_value=context.m4_mock_service,
)
context.m4_plan_patcher.start()
_register_patcher_cleanup(context, context.m4_plan_patcher)
context.m4_last_result = None
@@ -249,7 +242,6 @@ def step_m4_plan_with_decision_tree(context: Context) -> None:
"cleveragents.application.services.correction_service.CorrectionService",
return_value=mock_correction_svc,
)
context.m4_correction_patcher.start()
_register_patcher_cleanup(context, context.m4_correction_patcher)
context.m4_mock_correction_service = mock_correction_svc
@@ -263,7 +255,6 @@ def step_m4_plan_with_decision_tree(context: Context) -> None:
"cleveragents.application.container.get_container",
return_value=mock_container,
)
context.m4_container_patcher.start()
_register_patcher_cleanup(context, context.m4_container_patcher)
features/steps/plan_lifecycle_cli_steps.py
+3 -1
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import importlib
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
@@ -113,11 +114,12 @@ def step_call_lifecycle_service_helper(context) -> None:
settings = MagicMock()
service = PlanLifecycleService(settings=settings)
container = SimpleNamespace(plan_lifecycle_service=MagicMock(return_value=service))
reloaded_plan_module = importlib.reload(plan_module)
with patch(
"cleveragents.application.container.get_container", return_value=container
):
context.helper_settings = settings
context.helper_service = plan_module._get_lifecycle_service()
context.helper_service = reloaded_plan_module._get_lifecycle_service()
@then("the lifecycle service helper should return a plan lifecycle service")
features/steps/repl_input_modes_steps.py
+539
View File
@@ -0,0 +1,539 @@
"""Behave steps for REPL input mode behavior."""
from __future__ import annotations
import os
import shutil
import subprocess
import tempfile
from io import StringIO
from pathlib import Path
from typing import Any
from unittest.mock import patch
from behave import given, then, when
from behave.runner import Context
from pydantic import ValidationError
from rich.console import Console
from cleveragents.cli.commands.repl import (
_best_fuzzy_match,
_build_reference_catalog,
_expand_references,
_find_reference_candidates,
_handle_slash_command,
_parse_persona_create,
_reference_suggestions,
_ReplSessionState,
_run_shell_command,
run_repl,
)
from cleveragents.cli.persona import PersonaConfig
from features.mocks.fake_repl_input import FakeReplInput
def _run_repl_capture(lines: list[str], config_dir: Path) -> tuple[int, str]:
from io import StringIO
from rich.console import Console
output = StringIO()
console = Console(file=output, no_color=True, width=180)
fake_input = FakeReplInput(lines)
with (
patch("cleveragents.cli.commands.repl._console", console),
patch("builtins.input", fake_input),
patch.dict(os.environ, {"CLEVERAGENTS_CONFIG_DIR": str(config_dir)}),
):
code = run_repl(no_history=True)
return code, output.getvalue()
@given("a temporary REPL config directory")
def step_temp_repl_config(context: Context) -> None:
config_dir = Path(tempfile.mkdtemp())
context.repl_mode_config_dir = config_dir
context.add_cleanup(lambda: shutil.rmtree(str(config_dir), ignore_errors=True))
@given("a deterministic REPL reference catalog fixture")
def step_repl_reference_fixture(context: Context) -> None:
fixture_dir = Path(tempfile.mkdtemp())
(fixture_dir / "README.md").write_text("# fixture\n", encoding="utf-8")
context.repl_reference_fixture_dir = fixture_dir
context.add_cleanup(lambda: shutil.rmtree(str(fixture_dir), ignore_errors=True))
@given("a rich REPL reference fixture")
def step_rich_repl_reference_fixture(context: Context) -> None:
fixture_dir = Path(tempfile.mkdtemp())
(fixture_dir / "README.md").write_text("# fixture\n", encoding="utf-8")
(fixture_dir / "examples" / "actors").mkdir(parents=True, exist_ok=True)
(fixture_dir / "examples" / "tools").mkdir(parents=True, exist_ok=True)
(fixture_dir / "examples" / "skills").mkdir(parents=True, exist_ok=True)
(fixture_dir / "examples" / "actors" / "review.yaml").write_text(
"name: review\n", encoding="utf-8"
)
(fixture_dir / "examples" / "tools" / "search.yml").write_text(
"name: search\n", encoding="utf-8"
)
(fixture_dir / "examples" / "skills" / "guide").mkdir(parents=True, exist_ok=True)
(fixture_dir / "examples" / "skills" / "skill.yml").write_text(
"name: skill\n", encoding="utf-8"
)
context.repl_reference_fixture_dir = fixture_dir
context.add_cleanup(lambda: shutil.rmtree(str(fixture_dir), ignore_errors=True))
@given("a temporary persona export path")
def step_temp_persona_export_path(context: Context) -> None:
export_path = Path("tmp-persona-export.yaml")
context.repl_mode_export_path = export_path
context.add_cleanup(lambda: export_path.unlink(missing_ok=True))
@given("an absolute persona export path")
def step_absolute_persona_export_path(context: Context) -> None:
export_path = Path(tempfile.mkdtemp()) / "persona-export.yaml"
context.repl_mode_export_path = export_path
context.add_cleanup(
lambda: shutil.rmtree(str(export_path.parent), ignore_errors=True)
)
@given("an invalid persona import file path")
def step_invalid_persona_import_file(context: Context) -> None:
invalid_path = Path("tmp-invalid-persona.yaml")
invalid_path.write_text("- bad\n- yaml\n", encoding="utf-8")
context.repl_mode_invalid_import_path = invalid_path
context.add_cleanup(lambda: invalid_path.unlink(missing_ok=True))
@given("an absolute persona import file path")
def step_absolute_persona_import_path(context: Context) -> None:
temp_dir = Path(tempfile.mkdtemp())
input_path = temp_dir / "persona-import.yaml"
input_path.write_text("name: x\nactor: local/mock-default\n", encoding="utf-8")
context.repl_mode_absolute_import_path = input_path
context.add_cleanup(lambda: shutil.rmtree(str(temp_dir), ignore_errors=True))
@when("I run the REPL with input lines")
def step_run_repl_lines(context: Context) -> None:
replacements: dict[str, str] = {}
export_path = getattr(context, "repl_mode_export_path", None)
if export_path is not None:
replacements["export_path"] = str(export_path)
invalid_import_path = getattr(context, "repl_mode_invalid_import_path", None)
if invalid_import_path is not None:
replacements["invalid_import_path"] = str(invalid_import_path)
absolute_import_path = getattr(context, "repl_mode_absolute_import_path", None)
if absolute_import_path is not None:
replacements["absolute_import_path"] = str(absolute_import_path)
lines = []
for row in context.table:
line = row["line"]
if replacements:
line = line.format(**replacements)
lines.append(line)
code, output = _run_repl_capture(lines, context.repl_mode_config_dir)
context.repl_mode_code = code
context.repl_mode_output = output
@then('the REPL mode output should contain "{text}"')
def step_output_contains(context: Context, text: str) -> None:
assert text in context.repl_mode_output, (
f"Expected {text!r} in output, got:\n{context.repl_mode_output}"
)
@when('I expand REPL references for "{line}"')
def step_expand_references(context: Context, line: str) -> None:
fixture_dir = getattr(context, "repl_reference_fixture_dir", None)
if fixture_dir is None:
context.expanded_repl_line = _expand_references(line)
return
with patch("cleveragents.cli.commands.repl.Path.cwd", return_value=fixture_dir):
context.expanded_repl_line = _expand_references(line)
@then('the expanded REPL line should contain "{text}"')
def step_expanded_contains(context: Context, text: str) -> None:
assert text in context.expanded_repl_line, (
f"Expected {text!r} in expanded line: {context.expanded_repl_line}"
)
@when('I validate CLI persona name "{name}"')
def step_validate_cli_persona_name(context: Context, name: str) -> None:
decoded = name.encode("utf-8").decode("unicode_escape")
context.repl_cli_persona_error = None
try:
PersonaConfig(name=decoded, actor="local/mock-default")
except ValidationError as exc:
context.repl_cli_persona_error = exc
@then('CLI persona validation should fail with "{message}"')
def step_cli_persona_validation_fails(context: Context, message: str) -> None:
assert context.repl_cli_persona_error is not None
assert message in str(context.repl_cli_persona_error)
@when('I validate CLI persona payload with color "{color}"')
def step_validate_cli_persona_payload_with_color(context: Context, color: str) -> None:
context.repl_cli_persona_payload = PersonaConfig.model_validate(
{"name": "colorful", "actor": "local/mock-default", "color": color}
)
@then('CLI persona color should be "{color}"')
def step_cli_persona_color(context: Context, color: str) -> None:
assert context.repl_cli_persona_payload.color == color
@when('I parse persona create tokens "{command}"')
def step_parse_persona_create_tokens(context: Context, command: str) -> None:
context.repl_parse_error = None
context.repl_parsed_persona = None
text = command[1:] if command.startswith("/") else command
tokens = text.split()
try:
context.repl_parsed_persona = _parse_persona_create(tokens)
except Exception as exc:
context.repl_parse_error = exc
@then('parsed persona name should be "{name}"')
def step_parsed_persona_name(context: Context, name: str) -> None:
assert context.repl_parsed_persona is not None
assert context.repl_parsed_persona.name == name
@then('parsed persona actor should be "{actor}"')
def step_parsed_persona_actor(context: Context, actor: str) -> None:
assert context.repl_parsed_persona is not None
assert context.repl_parsed_persona.actor == actor
@then("parsed persona cycle order should be {cycle:d}")
def step_parsed_persona_cycle(context: Context, cycle: int) -> None:
assert context.repl_parsed_persona is not None
assert context.repl_parsed_persona.cycle_order == cycle
@then('persona create parse should fail with "{message}"')
def step_parse_persona_create_fails(context: Context, message: str) -> None:
assert context.repl_parse_error is not None
assert message in str(context.repl_parse_error)
def _run_shell_helper_with_capture(command: str) -> tuple[int, str]:
output = StringIO()
console = Console(file=output, no_color=True, width=180)
with patch("cleveragents.cli.commands.repl._console", console):
code = _run_shell_command(command)
return code, output.getvalue()
@when('I run REPL shell helper with command "{command}"')
def step_run_shell_helper(context: Context, command: str) -> None:
code, output = _run_shell_helper_with_capture(command)
context.repl_shell_helper_code = code
context.repl_shell_helper_output = output
@when("I run REPL shell helper with empty command")
def step_run_shell_helper_empty(context: Context) -> None:
code, output = _run_shell_helper_with_capture("")
context.repl_shell_helper_code = code
context.repl_shell_helper_output = output
@when('I run REPL shell helper with command "{command}" and shell disabled')
def step_run_shell_helper_disabled(context: Context, command: str) -> None:
with patch.dict(os.environ, {"CLEVERAGENTS_DISABLE_SHELL_MODE": "true"}):
code, output = _run_shell_helper_with_capture(command)
context.repl_shell_helper_code = code
context.repl_shell_helper_output = output
@when('I run REPL shell helper with command "{command}" and subprocess timeout')
def step_run_shell_helper_timeout(context: Context, command: str) -> None:
output = StringIO()
console = Console(file=output, no_color=True, width=180)
with (
patch("cleveragents.cli.commands.repl._console", console),
patch(
"cleveragents.cli.commands.repl.subprocess.run",
side_effect=subprocess.TimeoutExpired(command, 30),
),
):
context.repl_shell_helper_code = _run_shell_command(command)
context.repl_shell_helper_output = output.getvalue()
@when(
'I run REPL shell helper with command "{command}" and subprocess stderr "{stderr}" code {code:d}'
)
def step_run_shell_helper_stderr(
context: Context, command: str, stderr: str, code: int
) -> None:
output = StringIO()
console = Console(file=output, no_color=True, width=180)
proc = subprocess.CompletedProcess(
args=command, returncode=code, stdout="", stderr=stderr
)
with (
patch("cleveragents.cli.commands.repl._console", console),
patch("cleveragents.cli.commands.repl.subprocess.run", return_value=proc),
):
context.repl_shell_helper_code = _run_shell_command(command)
context.repl_shell_helper_output = output.getvalue()
@then("REPL shell helper exit code should be {code:d}")
def step_shell_helper_exit_code(context: Context, code: int) -> None:
assert context.repl_shell_helper_code == code
@then('REPL shell helper output should contain "{text}"')
def step_shell_helper_output_contains(context: Context, text: str) -> None:
assert text in context.repl_shell_helper_output
class _SlashRegistryStub:
def __init__(self) -> None:
self.personas: dict[str, PersonaConfig] = {}
self.active_name = "default"
def list_personas(self) -> list[PersonaConfig]:
return list(self.personas.values())
def get_persona(self, _name: str) -> PersonaConfig | None:
return self.personas.get(_name)
def set_active_persona_name(self, name: str) -> None:
self.active_name = name
def save_persona(self, persona: PersonaConfig) -> None:
self.personas[persona.name] = persona
def delete_persona(self, name: str) -> bool:
return self.personas.pop(name, None) is not None
def resolve_import_path(self, path: Path) -> Path:
return path
@given("slash command test sessions")
def step_slash_test_sessions(context: Context) -> None:
context.slash_registry = _SlashRegistryStub()
context.slash_sessions = {
"default": _ReplSessionState(name="default", active_persona="default")
}
context.slash_current_session = "default"
def _handle_slash_with_capture(
slash_text: str,
*,
registry: Any,
sessions: Any,
current_session: str,
input_value: str | None = None,
) -> tuple[int, str, str]:
output = StringIO()
console = Console(file=output, no_color=True, width=180)
with patch("cleveragents.cli.commands.repl._console", console):
if input_value is None:
code, new_session = _handle_slash_command(
slash_text,
registry=registry,
sessions=sessions,
current_session=current_session,
)
else:
with patch("builtins.input", return_value=input_value):
code, new_session = _handle_slash_command(
slash_text,
registry=registry,
sessions=sessions,
current_session=current_session,
)
return code, new_session, output.getvalue()
@when('I handle slash text with parser error "{slash_text}"')
def step_handle_slash_parser_error(context: Context, slash_text: str) -> None:
code, new_session, output = _handle_slash_with_capture(
slash_text,
registry=context.slash_registry,
sessions=context.slash_sessions,
current_session=context.slash_current_session,
)
context.slash_handler_code = code
context.slash_handler_session = new_session
context.slash_handler_output = output
@when("I handle slash text that raises parser error")
def step_handle_slash_parser_error_forced(context: Context) -> None:
output = StringIO()
console = Console(file=output, no_color=True, width=180)
with (
patch("cleveragents.cli.commands.repl._console", console),
patch(
"cleveragents.cli.commands.repl.shlex.split",
side_effect=ValueError("forced parser failure"),
),
):
code, new_session = _handle_slash_command(
"persona test",
registry=context.slash_registry,
sessions=context.slash_sessions,
current_session=context.slash_current_session,
)
context.slash_handler_code = code
context.slash_handler_session = new_session
context.slash_handler_output = output.getvalue()
@when('I handle slash text "{slash_text}"')
def step_handle_slash_text(context: Context, slash_text: str) -> None:
code, new_session, output = _handle_slash_with_capture(
slash_text,
registry=context.slash_registry,
sessions=context.slash_sessions,
current_session=context.slash_current_session,
)
context.slash_handler_code = code
context.slash_handler_session = new_session
context.slash_handler_output = output
@when('I handle picker slash text "{slash_text}" using input "{input_value}"')
def step_handle_slash_text_with_input(
context: Context, slash_text: str, input_value: str
) -> None:
code, new_session, output = _handle_slash_with_capture(
slash_text,
registry=context.slash_registry,
sessions=context.slash_sessions,
current_session=context.slash_current_session,
input_value=input_value,
)
context.slash_handler_code = code
context.slash_handler_session = new_session
context.slash_handler_output = output
@when("I handle empty slash text")
def step_handle_empty_slash_text(context: Context) -> None:
code, new_session, output = _handle_slash_with_capture(
"",
registry=context.slash_registry,
sessions=context.slash_sessions,
current_session=context.slash_current_session,
)
context.slash_handler_code = code
context.slash_handler_session = new_session
context.slash_handler_output = output
@then("slash handler exit code should be {code:d}")
def step_slash_handler_code(context: Context, code: int) -> None:
assert context.slash_handler_code == code
@then('slash handler output should contain "{text}"')
def step_slash_handler_output_contains(context: Context, text: str) -> None:
assert text in context.slash_handler_output
@given("slash command test registry with personas")
def step_slash_registry_with_personas(context: Context) -> None:
context.slash_registry.save_persona(
PersonaConfig(name="dev", actor="local/mock-default")
)
context.slash_registry.save_persona(
PersonaConfig(name="reviewer", actor="local/mock-default")
)
@when("I build REPL reference catalog from fixture")
def step_build_repl_reference_catalog_fixture(context: Context) -> None:
context.repl_reference_catalog = {}
outside_root = Path("/tmp/repl-outside-root")
def fake_walk(_cwd: Path, followlinks: bool = False):
yield (
str(context.repl_reference_fixture_dir),
["examples", ".git"],
["README.md"],
)
yield str(outside_root), [], ["outside.txt"]
with (
patch(
"cleveragents.cli.commands.repl.Path.cwd",
return_value=context.repl_reference_fixture_dir,
),
patch("cleveragents.cli.commands.repl.os.walk", side_effect=fake_walk),
patch.dict(
os.environ,
{"CLEVERAGENTS_PROJECT": "local/proj", "CLEVERAGENTS_PLAN": "local/plan"},
clear=False,
),
):
context.repl_reference_catalog = _build_reference_catalog()
@then("REPL catalog should include fixture actor tool skill and env entries")
def step_catalog_includes_fixture_entries(context: Context) -> None:
catalog = context.repl_reference_catalog
assert "local/review" in catalog["actor"]
assert "local/search" in catalog["tool"]
assert "local/guide" in catalog["skill"]
assert "local/skill" in catalog["skill"]
assert catalog["project"] == ["local/proj"]
assert catalog["plan"] == ["local/plan"]
@then("REPL catalog should include non-relative walk file path")
def step_catalog_has_non_relative_path(context: Context) -> None:
assert any(
value.endswith("/outside.txt")
for value in context.repl_reference_catalog["file"]
)
@when("I evaluate REPL fuzzy and suggestion helpers")
def step_eval_repl_fuzzy_helpers(context: Context) -> None:
catalog = {
"file": ["README.md", "docs/guide.md"],
"actor": ["local/reviewer"],
"tool": ["local/searcher"],
"skill": ["local/guide"],
"project": ["local/proj"],
"plan": ["local/plan"],
}
context.repl_helper_checks = [
_best_fuzzy_match("", ["x"]) is None,
_best_fuzzy_match("REA", ["README.md"]) == "README.md",
_best_fuzzy_match("guide", ["docs/guide.md"]) == "docs/guide.md",
_best_fuzzy_match("reviwer", ["local/reviewer"]) == "local/reviewer",
_find_reference_candidates("", ["a", "b"], limit=1) == ["a"],
_expand_references(
"echo \\@literal @file:README @zzz", catalog=catalog
).startswith("echo @literal @file:"),
len(_reference_suggestions("inspect @file:READ", catalog=catalog)) > 0,
len(_reference_suggestions("inspect @rev", catalog=catalog)) > 0,
_reference_suggestions("inspect \\@file:README", catalog=catalog) == [],
]
@then("REPL helper evaluation should pass")
def step_repl_helper_eval_pass(context: Context) -> None:
assert all(context.repl_helper_checks)
features/steps/repl_steps.py
+2 -18
View File
@@ -14,29 +14,13 @@ from cleveragents.cli.commands.repl import (
_get_prompt_context,
run_repl,
)
from features.mocks.fake_repl_input import FakeReplInput
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
class _FakeInput:
"""Simulate user input for the REPL by feeding lines then raising EOFError."""
def __init__(self, lines: list[str]) -> None:
self._lines = list(lines)
self._index = 0
def __call__(self, prompt: str = "") -> str:
if self._index >= len(self._lines):
raise EOFError
line = self._lines[self._index]
self._index += 1
if line == "__KEYBOARD_INTERRUPT__":
raise KeyboardInterrupt
return line
def _run_repl_with_input(
lines: list[str],
*,
@@ -55,7 +39,7 @@ def _run_repl_with_input(
if history_path is not None:
kw["history_path"] = history_path
fake = _FakeInput(lines)
fake = FakeReplInput(lines, interrupt_token="__KEYBOARD_INTERRUPT__")
with (
patch("cleveragents.cli.commands.repl._console", console),
patch("builtins.input", fake),
features/steps/tui_input_modes_steps.py
+145
View File
@@ -0,0 +1,145 @@
"""Behave steps for TUI input mode router."""
from __future__ import annotations
import os
import shutil
import tempfile
from pathlib import Path
from unittest.mock import patch
from behave import given, then, when
from behave.runner import Context
from cleveragents.tui import app as tui_app_module
from cleveragents.tui.input import reference_parser
from cleveragents.tui.input.modes import InputModeRouter
from cleveragents.tui.input.reference_parser import parse_references, suggestions
from cleveragents.tui.input.shell_exec import looks_dangerous
@when('I parse TUI references for "{line}"')
def step_parse_references(context: Context, line: str) -> None:
context.tui_ref_result = parse_references(line)
@then('the parsed line should include "{token}"')
def step_parsed_contains(context: Context, token: str) -> None:
assert token in context.tui_ref_result.expanded_line
@when('I route TUI input "{line}"')
def step_route_input(context: Context, line: str) -> None:
router = InputModeRouter(command_handler=lambda raw: f"ok:/{raw}")
context.tui_mode_result = router.process(line)
@then('the TUI mode should be "{mode}"')
def step_mode_equals(context: Context, mode: str) -> None:
assert context.tui_mode_result.mode.value == mode
@then('the TUI command result should be "{value}"')
def step_command_result(context: Context, value: str) -> None:
assert context.tui_mode_result.command_result == value
@then('the TUI shell stdout should contain "{value}"')
def step_shell_stdout(context: Context, value: str) -> None:
shell = context.tui_mode_result.shell_result
assert shell is not None
assert value in shell.stdout
@then('the TUI shell stderr should contain "{value}"')
def step_shell_stderr(context: Context, value: str) -> None:
shell = context.tui_mode_result.shell_result
assert shell is not None
assert value in shell.stderr
@when('I check shell danger detection for "{command}"')
def step_check_shell_danger(context: Context, command: str) -> None:
context.shell_danger_result = looks_dangerous(command)
@then('shell danger detection result should be "{value}"')
def step_shell_danger_result(context: Context, value: str) -> None:
expected = value.strip().lower() == "true"
assert context.shell_danger_result is expected
@given("a deterministic TUI reference fixture")
def step_deterministic_tui_reference_fixture(context: Context) -> None:
fixture_dir = Path(tempfile.mkdtemp())
(fixture_dir / "README.md").write_text("# fixture\n", encoding="utf-8")
(fixture_dir / ".git").mkdir(exist_ok=True)
(fixture_dir / ".git" / "secret.txt").write_text("secret\n", encoding="utf-8")
context.tui_reference_fixture_dir = fixture_dir
context.add_cleanup(lambda: shutil.rmtree(str(fixture_dir), ignore_errors=True))
@when("I parse and suggest TUI references using fixture")
def step_parse_and_suggest_with_fixture(context: Context) -> None:
context.tui_walk_count = 0
context.tui_fixture_suggestions = []
reference_parser._catalog_cache["cwd"] = None
reference_parser._catalog_cache["created_at"] = 0.0
reference_parser._catalog_cache["catalog"] = None
real_walk = os.walk
def counting_walk(*args: object, **kwargs: object):
context.tui_walk_count += 1
return real_walk(*args, **kwargs)
with (
patch(
"cleveragents.tui.input.reference_parser.Path.cwd",
return_value=context.tui_reference_fixture_dir,
),
patch(
"cleveragents.tui.input.reference_parser.os.walk", side_effect=counting_walk
),
):
parse_references("inspect @README.md")
context.tui_fixture_suggestions = suggestions("secret", category="resource")
@then('TUI walk count should be "{count}"')
def step_tui_walk_count(context: Context, count: str) -> None:
assert context.tui_walk_count == int(count)
@then('TUI suggestions should not include "{value}"')
def step_tui_suggestions_not_include(context: Context, value: str) -> None:
assert all(
value not in suggestion for suggestion in context.tui_fixture_suggestions
)
@when("I check TUI app textual availability flag")
def step_check_tui_textual_availability(context: Context) -> None:
context.tui_textual_available = tui_app_module.textual_available()
@then("TUI textual availability should be boolean")
def step_tui_textual_is_bool(context: Context) -> None:
assert isinstance(context.tui_textual_available, bool)
@when("I run fallback TUI app")
def step_run_fallback_tui_app(context: Context) -> None:
context.tui_fallback_error = None
fallback = tui_app_module._FallbackCleverAgentsTuiApp(
command_router=object(), persona_state=object()
)
try:
fallback.run()
except RuntimeError as exc:
context.tui_fallback_error = exc
@then('fallback TUI app should fail with "{message}"')
def step_fallback_tui_fails(context: Context, message: str) -> None:
assert context.tui_fallback_error is not None
assert message in str(context.tui_fallback_error)
features/steps/tui_persona_system_steps.py
+215
View File
@@ -0,0 +1,215 @@
"""Behave steps for TUI persona system."""
from __future__ import annotations
import shutil
import tempfile
import threading
from pathlib import Path
from behave import given, then, when
from behave.runner import Context
from cleveragents.tui.persona.registry import PersonaRegistry
from cleveragents.tui.persona.schema import Persona, PersonaPreset
from cleveragents.tui.persona.state import PersonaState
def _registry_for_temp_dir(path: Path) -> PersonaRegistry:
return PersonaRegistry(config_dir=path)
@given("a temporary TUI persona registry")
def step_temp_registry(context: Context) -> None:
temp_dir = Path(tempfile.mkdtemp())
context.tui_persona_dir = temp_dir
context.tui_registry = _registry_for_temp_dir(temp_dir)
context.add_cleanup(lambda: shutil.rmtree(str(temp_dir), ignore_errors=True))
@when('I build a TUI persona named "{name}" with actor "{actor}"')
def step_build_persona(context: Context, name: str, actor: str) -> None:
context.tui_persona = Persona(name=name, actor=actor)
@then("the persona should include exactly one default preset")
def step_default_preset(context: Context) -> None:
defaults = [
item for item in context.tui_persona.argument_presets if item.name == "default"
]
assert len(defaults) == 1
@when('I save a TUI persona named "{name}" with actor "{actor}"')
def step_save_persona(context: Context, name: str, actor: str) -> None:
persona = Persona(name=name, actor=actor)
context.tui_registry.save(persona)
@given('I save a TUI persona named "{name}" with actor "{actor}"')
def step_given_save_persona(context: Context, name: str, actor: str) -> None:
step_save_persona(context, name, actor)
@given(
'I save TUI persona "{name}" with actor "{actor}" and extra preset "{preset_name}"'
)
def step_save_persona_with_extra_preset(
context: Context, name: str, actor: str, preset_name: str
) -> None:
persona = Persona(
name=name,
actor=actor,
argument_presets=[
PersonaPreset(name="default", display="default", overrides={}),
PersonaPreset(
name=preset_name, display=preset_name, overrides={"mode": "x"}
),
],
)
context.tui_registry.save(persona)
@given(
'I save a TUI persona named "{name}" with actor "{actor}" and cycle order {cycle:d}'
)
def step_save_persona_cycle(
context: Context, name: str, actor: str, cycle: int
) -> None:
persona = Persona(name=name, actor=actor, cycle_order=cycle)
context.tui_registry.save(persona)
@when(
'I try to save a TUI persona named "{name}" with actor "{actor}" and cycle order {cycle:d}'
)
def step_save_persona_cycle_fail(
context: Context, name: str, actor: str, cycle: int
) -> None:
context.tui_cycle_error = None
try:
context.tui_registry.save(Persona(name=name, actor=actor, cycle_order=cycle))
except Exception as exc:
context.tui_cycle_error = exc
@then('loading persona "{name}" should succeed')
def step_loading_persona(context: Context, name: str) -> None:
context.tui_loaded_persona = context.tui_registry.get(name)
assert context.tui_loaded_persona is not None
@then('the loaded persona actor should be "{actor}"')
def step_loaded_actor(context: Context, actor: str) -> None:
assert context.tui_loaded_persona.actor == actor
@then("saving the second persona should fail with cycle order error")
def step_cycle_error(context: Context) -> None:
assert context.tui_cycle_error is not None
assert "cycle_order" in str(context.tui_cycle_error)
@when('I set active persona to "{persona_name}" for session "{session_id}"')
def step_set_active_persona(
context: Context, persona_name: str, session_id: str
) -> None:
if not hasattr(context, "tui_state"):
context.tui_state = PersonaState(registry=context.tui_registry)
context.tui_state.set_active_persona(session_id, persona_name)
@then('active persona for session "{session_id}" should be "{persona_name}"')
def step_active_persona(context: Context, session_id: str, persona_name: str) -> None:
persona = context.tui_state.active_persona(session_id)
assert persona.name == persona_name
@when('I cycle persona preset for session "{session_id}"')
def step_cycle_preset(context: Context, session_id: str) -> None:
if not hasattr(context, "tui_state"):
context.tui_state = PersonaState(registry=context.tui_registry)
context.tui_state.cycle_preset(session_id)
@then('current preset for session "{session_id}" should be "{preset_name}"')
def step_current_preset(context: Context, session_id: str, preset_name: str) -> None:
assert context.tui_state.current_preset(session_id) == preset_name
@given('I create malformed TUI persona file "{name}"')
def step_create_malformed_persona_file(context: Context, name: str) -> None:
malformed = context.tui_registry.personas_dir / f"{name}.yaml"
context.tui_registry.ensure_dirs()
malformed.write_text("name: broken\nactor:\n- not-a-string\n", encoding="utf-8")
@when("I list TUI personas")
def step_list_tui_personas(context: Context) -> None:
context.tui_list_error = None
context.tui_persona_names = []
try:
personas = context.tui_registry.list_personas()
context.tui_persona_names = [persona.name for persona in personas]
except Exception as exc:
context.tui_list_error = exc
@then('TUI persona list should include "{name}"')
def step_tui_persona_list_includes(context: Context, name: str) -> None:
assert context.tui_list_error is None
assert name in context.tui_persona_names
@when('I export TUI persona "{name}" to "{output_path}"')
def step_export_tui_persona(context: Context, name: str, output_path: str) -> None:
context.tui_export_error = None
try:
context.tui_registry.export_persona(name, Path(output_path))
except Exception as exc:
context.tui_export_error = exc
@then('TUI export should fail with "{message}"')
def step_tui_export_fails(context: Context, message: str) -> None:
assert context.tui_export_error is not None
assert message in str(context.tui_export_error)
@when('I import TUI persona from "{input_path}"')
def step_import_tui_persona(context: Context, input_path: str) -> None:
context.tui_import_error = None
try:
context.tui_registry.import_persona(Path(input_path))
except Exception as exc:
context.tui_import_error = exc
@then('TUI import should fail with "{message}"')
def step_tui_import_fails(context: Context, message: str) -> None:
assert context.tui_import_error is not None
assert message in str(context.tui_import_error)
@when("I concurrently set TUI last persona names")
def step_concurrent_set_last_persona_names(context: Context) -> None:
context.tui_concurrency_errors = []
def _worker(persona_name: str) -> None:
try:
context.tui_registry.set_last_persona(persona_name)
except Exception as exc: # pragma: no cover - assertion handled after join
context.tui_concurrency_errors.append(exc)
threads: list[threading.Thread] = []
for row in context.table:
thread = threading.Thread(target=_worker, args=(row["name"],))
threads.append(thread)
thread.start()
for thread in threads:
thread.join()
@then("TUI concurrent update should complete without errors")
def step_tui_concurrent_update_no_errors(context: Context) -> None:
assert context.tui_concurrency_errors == []
features/tui_input_modes.feature
+39
View File
@@ -0,0 +1,39 @@
Feature: TUI input modes
TUI prompt supports normal reference mode, slash command mode, and shell mode.
Scenario: Normal mode expands @ references
When I parse TUI references for "inspect @README.md"
Then the parsed line should include "@resource:README.md"
Scenario: Command mode dispatches slash command handler
When I route TUI input "/help"
Then the TUI mode should be "command"
And the TUI command result should be "ok:/help"
Scenario: Shell mode runs host command
When I route TUI input "!echo hello"
Then the TUI mode should be "shell"
And the TUI shell stdout should contain "hello"
Scenario: Shell mode blocks dangerous command by default
When I route TUI input "!rm -rf /"
Then the TUI mode should be "shell"
And the TUI shell stderr should contain "blocked dangerous shell command"
Scenario: Dangerous shell pattern detection helper
When I check shell danger detection for "rm -rf /tmp"
Then shell danger detection result should be "true"
When I check shell danger detection for "echo hello"
Then shell danger detection result should be "false"
Scenario: Reference catalog ignores .git and caches scans
Given a deterministic TUI reference fixture
When I parse and suggest TUI references using fixture
Then TUI walk count should be "1"
And TUI suggestions should not include ".git/secret.txt"
Scenario: TUI app fallback path reports missing textual dependency
When I check TUI app textual availability flag
Then TUI textual availability should be boolean
When I run fallback TUI app
Then fallback TUI app should fail with "Textual dependency missing."
features/tui_persona_system.feature
+63
View File
@@ -0,0 +1,63 @@
Feature: TUI persona system
Personas are TUI-only YAML records with per-session binding.
Scenario: Persona schema auto-creates default preset
When I build a TUI persona named "dev" with actor "local/mock-default"
Then the persona should include exactly one default preset
Scenario: Persona registry saves and loads YAML records
Given a temporary TUI persona registry
When I save a TUI persona named "reviewer" with actor "local/mock-default"
Then loading persona "reviewer" should succeed
And the loaded persona actor should be "local/mock-default"
Scenario: Persona registry validates duplicate cycle order
Given a temporary TUI persona registry
And I save a TUI persona named "a" with actor "local/mock-default" and cycle order 1
When I try to save a TUI persona named "b" with actor "local/mock-default" and cycle order 1
Then saving the second persona should fail with cycle order error
Scenario: Persona state is bound per session
Given a temporary TUI persona registry
And I save a TUI persona named "dev" with actor "local/mock-default"
And I save a TUI persona named "reviewer" with actor "local/mock-default"
When I set active persona to "dev" for session "s1"
And I set active persona to "reviewer" for session "s2"
Then active persona for session "s1" should be "dev"
And active persona for session "s2" should be "reviewer"
Scenario: Persona state cycles through presets per session
Given a temporary TUI persona registry
And I save TUI persona "cycler" with actor "local/mock-default" and extra preset "focused"
When I set active persona to "cycler" for session "s1"
And I cycle persona preset for session "s1"
Then current preset for session "s1" should be "focused"
When I cycle persona preset for session "s1"
Then current preset for session "s1" should be "default"
Scenario: Malformed persona YAML does not break registry operations
Given a temporary TUI persona registry
And I save a TUI persona named "good" with actor "local/mock-default"
And I create malformed TUI persona file "broken"
When I list TUI personas
Then TUI persona list should include "good"
When I save a TUI persona named "stillworks" with actor "local/mock-default"
Then loading persona "stillworks" should succeed
Scenario: TUI registry rejects unsafe export and import paths
Given a temporary TUI persona registry
And I save a TUI persona named "safe" with actor "local/mock-default"
When I export TUI persona "safe" to "../escape.yaml"
Then TUI export should fail with "stay within working directory"
When I import TUI persona from "../escape.yaml"
Then TUI import should fail with "stay within working directory"
Scenario: TUI registry state writes tolerate concurrent updates
Given a temporary TUI persona registry
When I concurrently set TUI last persona names
| name |
| a |
| b |
| c |
| d |
Then TUI concurrent update should complete without errors
pyproject.toml
+3
View File
@@ -50,6 +50,9 @@ dependencies = [
]
[project.optional-dependencies]
tui = [
"textual>=1.0.0,<2.0.0",
]
dev = [
# Code formatting and linting
"ruff>=0.15.0,<0.16.0",
robot/tui_smoke.robot
+42
View File
@@ -0,0 +1,42 @@
*** Settings ***
Library Process
Library String
*** Test Cases ***
TUI Headless Startup Check Returns JSON
${result}= Run Process ${PYTHON} -m cleveragents tui --headless shell=False stderr=STDOUT env:CLEVERAGENTS_DATABASE_URL=sqlite:///:memory:
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} textual_available
Should Contain ${result.stdout} default_persona
TUI Headless Works When Shell Disabled
${result}= Run Process ${PYTHON} -m cleveragents tui --headless shell=False stderr=STDOUT env:CLEVERAGENTS_DATABASE_URL=sqlite:///:memory: env:CLEVERAGENTS_DISABLE_SHELL_MODE=1
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} textual_available
Should Contain ${result.stdout} default_persona
TUI Input Mode Router And Prompt Widget Behavior
${script}= Catenate SEPARATOR=\n
... from cleveragents.tui.input.modes import InputModeRouter
... from cleveragents.tui.widgets.prompt import PromptInput
... prompt = PromptInput()
... prompt.value = "hello"
... payload = prompt.consume_text()
... assert payload.text == "hello"
... assert prompt.value == ""
... router = InputModeRouter(command_handler=lambda raw: f"ok:{raw}")
... command_result = router.process("/help")
... assert command_result.mode.value == "command"
... assert command_result.command_result == "ok:help"
... normal_result = router.process("inspect @README.md")
... assert normal_result.mode.value == "normal"
... print("router-widget-ok")
${result}= Run Process ${PYTHON} -c ${script} shell=False stderr=STDOUT
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} router-widget-ok
TUI Headless Includes Router Help Payload
${result}= Run Process ${PYTHON} -m cleveragents tui --headless shell=False stderr=STDOUT env:CLEVERAGENTS_DATABASE_URL=sqlite:///:memory:
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} "help"
Should Contain ${result.stdout} /persona
src/cleveragents/application/container.py
+32
View File
@@ -4,6 +4,7 @@ Based on ADR-003 (Dependency Injection Framework).
Uses dependency-injector for managing service instances.
"""
from dataclasses import dataclass
from pathlib import Path
import structlog
@@ -95,10 +96,27 @@ from cleveragents.providers.registry import ProviderRegistry, get_provider_regis
from cleveragents.reactive.route_bridge import RouteBridge
from cleveragents.reactive.stream_router import ReactiveStreamRouter
from cleveragents.shared.redaction import redact_value
from cleveragents.tui.persona.registry import PersonaRegistry
from cleveragents.tui.persona.state import PersonaState
_logger = structlog.get_logger(__name__)
@dataclass(slots=True)
class TuiServiceAdapters:
"""Lightweight adapter bundle for TUI-facing read operations.
The TUI is presentation-layer only; this keeps integration explicit
without introducing TUI concepts into domain models.
"""
actor_service: ActorService
session_service: PersistentSessionService
project_service: ProjectService
resource_registry_service: ResourceRegistryService
plan_service: PlanService
def get_ai_provider(
settings: Settings | None = None,
provider_registry: ProviderRegistry | None = None,
@@ -513,6 +531,20 @@ class Container(containers.DeclarativeContainer):
event_bus=event_bus,
)
# TUI adapter bundle (presentation-layer only composition)
tui_adapters = providers.Factory(
TuiServiceAdapters,
actor_service=actor_service,
session_service=session_service,
project_service=project_service,
resource_registry_service=resource_registry_service,
plan_service=plan_service,
)
# TUI persona wiring through DI container
persona_registry = providers.Factory(PersonaRegistry)
persona_state = providers.Factory(PersonaState, registry=persona_registry)
# Metrics Emitter - structured metric emission (Forgejo #579)
metrics_emitter = providers.Singleton(
MetricsEmitter.from_settings,
src/cleveragents/application/services/resource_file_watcher.py
+71 -22
View File
@@ -29,6 +29,7 @@ from watchdog.events import (
FileSystemEventHandler,
)
from watchdog.observers import Observer # type: ignore[import-untyped]
from watchdog.observers.polling import PollingObserver # type: ignore[import-untyped]
from cleveragents.infrastructure.events.models import DomainEvent
from cleveragents.infrastructure.events.protocol import EventBus
@@ -100,6 +101,7 @@ class ResourceFileWatcher:
# resolved path (str) -> pending debounce Timer
self._pending_timers: dict[str, threading.Timer] = {}
self._running = False
self._using_polling = False
_logger.debug(
"file_watcher.initialized",
@@ -157,11 +159,28 @@ class ResourceFileWatcher:
and parent_str not in self._dir_watches
):
handler = _ResourceChangeHandler(self)
watch_handle = self._observer.schedule(
handler,
parent_str,
recursive=False,
)
try:
watch_handle = self._observer.schedule(
handler,
parent_str,
recursive=False,
)
except OSError as exc:
if not self._using_polling:
_logger.warning(
"file_watcher.watch_schedule_fallback",
error_type=type(exc).__name__,
error_message=str(exc),
fallback="polling_observer",
)
self._switch_to_polling_observer_locked()
watch_handle = self._observer.schedule( # type: ignore[union-attr]
handler,
parent_str,
recursive=False,
)
else:
raise
self._dir_watches[parent_str] = watch_handle
_logger.debug(
@@ -229,23 +248,20 @@ class ResourceFileWatcher:
if self._running:
msg = "File watcher is already running"
raise RuntimeError(msg)
observer = Observer()
observer.daemon = True
handler = _ResourceChangeHandler(self)
seen_dirs: set[str] = set()
for path_str in self._watched_paths:
parent_str = str(Path(path_str).parent)
if parent_str not in seen_dirs:
watch_handle = observer.schedule(
handler,
parent_str,
recursive=False,
)
self._dir_watches[parent_str] = watch_handle
seen_dirs.add(parent_str)
observer.start()
observer = self._build_observer(use_polling=False)
try:
observer.start()
self._using_polling = False
except OSError as exc:
_logger.warning(
"file_watcher.inotify_unavailable",
error_type=type(exc).__name__,
error_message=str(exc),
fallback="polling_observer",
)
observer = self._build_observer(use_polling=True)
observer.start()
self._using_polling = True
self._observer = observer
self._running = True
@@ -255,6 +271,38 @@ class ResourceFileWatcher:
watched_dirs=len(self._dir_watches),
)
def _build_observer(self, *, use_polling: bool) -> Observer: # type: ignore[valid-type]
observer: Observer = ( # type: ignore[valid-type]
PollingObserver() if use_polling else Observer()
)
observer.daemon = True
self._dir_watches.clear()
handler = _ResourceChangeHandler(self)
seen_dirs: set[str] = set()
for path_str in self._watched_paths:
parent_str = str(Path(path_str).parent)
if parent_str in seen_dirs:
continue
watch_handle = observer.schedule(
handler,
parent_str,
recursive=False,
)
self._dir_watches[parent_str] = watch_handle
seen_dirs.add(parent_str)
return observer
def _switch_to_polling_observer_locked(self) -> None:
"""Swap current observer to polling backend while lock is held."""
old_observer = self._observer
if old_observer is not None:
old_observer.stop()
old_observer.join(timeout=5.0)
polling_observer = self._build_observer(use_polling=True)
polling_observer.start()
self._observer = polling_observer
self._using_polling = True
def stop(self) -> None:
"""Stop the watchdog observer and cancel pending timers."""
observer: Observer | None = None # type: ignore[valid-type]
@@ -268,6 +316,7 @@ class ResourceFileWatcher:
self._observer = None
self._dir_watches.clear()
self._running = False
self._using_polling = False
# Join outside lock to avoid deadlock with timer callbacks.
if observer is not None:
src/cleveragents/cli/commands/plan.py
+11
View File
@@ -930,6 +930,11 @@ def _lifecycle_apply_with_id(plan_id: str, fmt: str = "rich") -> None:
except PlanNotReadyError as e:
console.print(f"[red]Plan not ready:[/red] {e}")
raise typer.Abort() from e
except ValueError as e:
# Provider-resolution failures (e.g. missing API key/config) should be
# reported as a controlled CLI error instead of bubbling to a 500.
console.print(f"[red]Execution Error:[/red] {e}")
raise typer.Abort() from e
except CleverAgentsError as e:
console.print(f"[red]Error:[/red] {e.message}")
raise typer.Abort() from e
@@ -1898,6 +1903,12 @@ def execute_plan(
except PlanNotReadyError as e:
console.print(f"[red]Plan not ready:[/red] {e}")
raise typer.Abort() from e
except ValueError as e:
# Provider/config resolution errors can occur during strategize/execute
# when required model credentials are unavailable in the environment.
# Surface as a controlled CLI abort instead of an INTERNAL 500.
console.print(f"[red]Execution Error:[/red] {e}")
raise typer.Abort() from e
except CleverAgentsError as e:
console.print(f"[red]Error:[/red] {e.message}")
raise typer.Abort() from e
src/cleveragents/cli/commands/repl.py
+552 -2
View File
1
 @@ -7,17 +7,25 @@ prompt context showing the active project/plan.
from __future__ import annotations
import difflib
import os
import readline
import shlex
import subprocess
import sys
import time
from collections.abc import Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import Annotated
import typer
import yaml
from rich.console import Console
from cleveragents.cli.persona import PersonaConfig, PersonaRegistry
from cleveragents.tui.input.shell_exec import looks_dangerous
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
@@ -26,6 +34,11 @@ _DEFAULT_HISTORY_DIR = Path.home() / ".cleveragents"
_DEFAULT_HISTORY_PATH = _DEFAULT_HISTORY_DIR / "history"
_MULTILINE_CONTINUATION = "\\"
_LAST_COMMAND_TOKEN = "!!"
_MAX_REFERENCE_FILES = 300
_SHELL_TIMEOUT_SECONDS = 30
_IGNORED_DIRS = {".git", ".venv", "node_modules", "__pycache__", ".mypy_cache"}
_REFERENCE_CACHE_TTL_SECONDS = 5.0
_reference_cache: dict[str, object] = {"cwd": None, "created_at": 0.0, "catalog": None}
# The subset of first-position CLI commands exposed inside the REPL.
# Mirrors the ``valid_cmds`` list in ``main.py`` minus the flags.
@@ -55,9 +68,23 @@ _REPL_COMMANDS: list[str] = [
"apply",
"context-load",
"context-add",
"tui",
]
_BUILTIN_COMMANDS: list[str] = [":help", ":exit", ":quit"]
_SLASH_COMMANDS: list[str] = [
"/persona",
"/persona list",
"/persona set",
"/persona create",
"/persona pick",
"/persona delete",
"/persona export",
"/persona import",
"/session list",
"/session new",
"/session switch",
]
_console = Console()
@@ -66,6 +93,18 @@ _console = Console()
# ---------------------------------------------------------------------------
@dataclass(slots=True)
class _ReplSessionState:
name: str
active_persona: str
def _compose_prompt(base_prompt: str, session_state: _ReplSessionState) -> str:
"""Compose prompt with session/persona status."""
core = base_prompt.removesuffix("> ").rstrip()
return f"{core} [{session_state.name}:{session_state.active_persona}]> "
def _get_prompt_context() -> str:
"""Return a prompt string reflecting the active project/plan.
@@ -104,7 +143,7 @@ def _save_history(history_path: Path) -> None:
def _setup_completer() -> None:
"""Install a tab-completer for known commands and built-ins."""
all_tokens = sorted(set(_REPL_COMMANDS + _BUILTIN_COMMANDS))
all_tokens = sorted(set(_REPL_COMMANDS + _BUILTIN_COMMANDS + _SLASH_COMMANDS))
def completer(text: str, state: int) -> str | None:
matches = [t for t in all_tokens if t.startswith(text)]
@@ -141,6 +180,16 @@ def _print_repl_help() -> None:
_console.print(" :exit / :quit Exit the REPL")
_console.print(" !! Repeat the last command")
_console.print("")
_console.print("[underline]Input modes[/underline]")
_console.print(" @reference Inline reference expansion and fuzzy matching")
_console.print(" /command Slash command mode")
_console.print(" !shell cmd Shell command passthrough")
_console.print(" Full-screen UI: run [bold]agents tui[/bold]")
_console.print("")
_console.print("[underline]Slash commands[/underline]")
for cmd in _SLASH_COMMANDS:
_console.print(f" {cmd}")
_console.print("")
_console.print("[underline]CLI commands[/underline]")
for cmd in _REPL_COMMANDS:
_console.print(f" {cmd}")
@@ -148,6 +197,472 @@ def _print_repl_help() -> None:
_console.print("Use [bold]\\\\[/bold] at end of line for multi-line input.")
def _build_reference_catalog() -> dict[str, list[str]]:
"""Build lightweight catalog for @ reference expansion."""
cwd = Path.cwd()
now = time.time()
cached_cwd = _reference_cache.get("cwd")
cached_time = _reference_cache.get("created_at")
cached_catalog = _reference_cache.get("catalog")
if (
isinstance(cached_cwd, Path)
and cached_cwd == cwd
and isinstance(cached_time, float)
and now - cached_time < _REFERENCE_CACHE_TTL_SECONDS
and isinstance(cached_catalog, dict)
):
return cached_catalog
files: list[str] = []
for root, dirs, filenames in os.walk(cwd, followlinks=False):
dirs[:] = [name for name in dirs if name not in _IGNORED_DIRS]
root_path = Path(root)
for filename in filenames:
if len(files) >= _MAX_REFERENCE_FILES:
break
path = root_path / filename
try:
files.append(str(path.relative_to(cwd).as_posix()))
except ValueError:
files.append(path.as_posix())
if len(files) >= _MAX_REFERENCE_FILES:
break
actors: set[str] = set()
actor_examples = cwd / "examples" / "actors"
if actor_examples.is_dir():
for file in actor_examples.glob("*.y*ml"):
actors.add(f"local/{file.stem}")
tools: set[str] = set()
tool_examples = cwd / "examples" / "tools"
if tool_examples.is_dir():
for file in tool_examples.glob("*.y*ml"):
tools.add(f"local/{file.stem}")
skills: set[str] = set()
skill_examples = cwd / "examples" / "skills"
if skill_examples.is_dir():
for item in skill_examples.iterdir():
if item.is_dir():
skills.add(f"local/{item.name}")
elif item.suffix in {".yaml", ".yml"}:
skills.add(f"local/{item.stem}")
project_env = os.environ.get("CLEVERAGENTS_PROJECT", "")
plan_env = os.environ.get("CLEVERAGENTS_PLAN", "")
catalog = {
"file": sorted(files),
"actor": sorted(actors),
"tool": sorted(tools),
"skill": sorted(skills),
"project": [project_env] if project_env else [],
"plan": [plan_env] if plan_env else [],
}
_reference_cache["cwd"] = cwd
_reference_cache["created_at"] = now
_reference_cache["catalog"] = catalog
return catalog
def _best_fuzzy_match(query: str, options: list[str]) -> str | None:
"""Return the best match using prefix/substring/fuzzy ordering."""
if not query or not options:
return None
for option in options:
if option.startswith(query):
return option
for option in options:
if query in option:
return option
matches = difflib.get_close_matches(query, options, n=1, cutoff=0.4)
if matches:
return matches[0]
return None
def _find_reference_candidates(
query: str, options: list[str], limit: int = 5
) -> list[str]:
"""Return top fuzzy candidates for preview/display."""
if not query:
return options[:limit]
prefix = [value for value in options if value.startswith(query)]
contains = [value for value in options if query in value and value not in prefix]
fuzzy = [
value
for value in difflib.get_close_matches(query, options, n=limit, cutoff=0.4)
if value not in prefix and value not in contains
]
return (prefix + contains + fuzzy)[:limit]
def _expand_references(
line: str, *, catalog: dict[str, list[str]] | None = None
) -> str:
"""Expand inline @ tokens to canonical references when possible."""
active_catalog = catalog or _build_reference_catalog()
tokens = line.split()
expanded_tokens: list[str] = []
for token in tokens:
if token.startswith("\\@"):
expanded_tokens.append(token[1:])
continue
if not token.startswith("@") or len(token) == 1:
expanded_tokens.append(token)
continue
raw = token[1:]
category = ""
query = raw
if ":" in raw:
maybe_category, rest = raw.split(":", 1)
if maybe_category in active_catalog:
category = maybe_category
query = rest
if category:
candidate = _best_fuzzy_match(query, active_catalog.get(category, []))
if candidate:
expanded_tokens.append(f"@{category}:{candidate}")
continue
expanded_tokens.append(token)
continue
best_category = ""
best_value = ""
for candidate_category, values in active_catalog.items():
candidate = _best_fuzzy_match(query, values)
if candidate:
best_category = candidate_category
best_value = candidate
break
if best_category:
expanded_tokens.append(f"@{best_category}:{best_value}")
else:
expanded_tokens.append(token)
return " ".join(expanded_tokens)
brent.edwards commented 2026-03-16 20:29:30 +00:00
Outdated
Review
Copy Link
Copy Source

P0: Shell passthrough with shell=True and no timeout, no danger check, no sandboxing.

This is the REPL's version of the same issue in shell_exec.py. At minimum add timeout= and call looks_dangerous() (or the shared equivalent) before execution.

**P0: Shell passthrough with `shell=True` and no timeout, no danger check, no sandboxing.** This is the REPL's version of the same issue in `shell_exec.py`. At minimum add `timeout=` and call `looks_dangerous()` (or the shared equivalent) before execution.
def _reference_suggestions(
line: str, *, catalog: dict[str, list[str]] | None = None
) -> list[str]:
"""Return lightweight suggestion hints for unresolved @ tokens."""
active_catalog = catalog or _build_reference_catalog()
for token in line.split():
if not token.startswith("@") or token.startswith("\\@"):
continue
raw = token[1:]
if not raw:
continue
if ":" in raw:
category, query = raw.split(":", 1)
if category in active_catalog:
candidates = _find_reference_candidates(
query, active_catalog[category], limit=5
)
return [f"@{category}:{value}" for value in candidates]
else:
merged: list[str] = []
for category, values in active_catalog.items():
for value in _find_reference_candidates(raw, values, limit=2):
merged.append(f"@{category}:{value}")
return merged[:5]
return []
def _run_shell_command(command_text: str) -> int:
"""Execute a shell command with minimal safeguards.
Threat model:
- Shell mode is intended for trusted local development workflows.
- It is not a sandbox and must never be exposed as a remote execution surface.
- Dangerous patterns require explicit user confirmation before execution.
"""
if not command_text.strip():
_console.print("[yellow]Shell mode requires a command after '!'.[/yellow]")
return 2
if os.environ.get("CLEVERAGENTS_DISABLE_SHELL_MODE", "").strip() in {"1", "true"}:
_console.print("[yellow]Shell mode is disabled by configuration.[/yellow]")
return 2
if looks_dangerous(command_text):
answer = input("Dangerous command detected. Continue? [y/N]: ").strip().lower()
if answer not in {"y", "yes"}:
_console.print("[yellow]Blocked dangerous shell command.[/yellow]")
return 1
try:
proc = subprocess.run(
command_text,
shell=True,
text=True,
capture_output=True,
timeout=_SHELL_TIMEOUT_SECONDS,
) # nosec B602
except subprocess.TimeoutExpired:
_console.print(
f"[red]Shell command timed out after {_SHELL_TIMEOUT_SECONDS}s.[/red]"
)
return 124
if proc.stdout:
_console.print(proc.stdout.rstrip("\n"))
if proc.stderr:
_console.print(f"[red]{proc.stderr.rstrip()}[/red]")
return proc.returncode
def _print_persona_list(
registry: PersonaRegistry,
sessions: dict[str, _ReplSessionState],
current_session: str,
) -> None:
personas = registry.list_personas()
if not personas:
_console.print("[yellow]No personas found.[/yellow]")
return
active = sessions[current_session].active_persona
_console.print("[bold]Personas[/bold]")
for persona in personas:
marker = "*" if persona.name == active else " "
_console.print(f" {marker} {persona.name} -> {persona.actor}")
def _parse_persona_create(tokens: list[str]) -> PersonaConfig:
if len(tokens) < 4:
raise ValueError("Usage: /persona create <name> --actor <namespace/name>")
name = tokens[2]
actor = ""
description = ""
icon = ""
greeting = ""
cycle_order = 0
idx = 3
while idx < len(tokens):
current = tokens[idx]
if current == "--actor" and idx + 1 < len(tokens):
actor = tokens[idx + 1]
idx += 2
continue
if current == "--description" and idx + 1 < len(tokens):
description = tokens[idx + 1]
idx += 2
continue
if current == "--icon" and idx + 1 < len(tokens):
icon = tokens[idx + 1]
idx += 2
continue
if current == "--greeting" and idx + 1 < len(tokens):
greeting = tokens[idx + 1]
idx += 2
continue
if current == "--cycle-order" and idx + 1 < len(tokens):
try:
cycle_order = int(tokens[idx + 1])
except ValueError as exc:
raise ValueError("--cycle-order must be an integer") from exc
idx += 2
continue
idx += 1
if not actor:
raise ValueError("Missing --actor for /persona create")
return PersonaConfig(
name=name,
actor=actor,
description=description,
icon=icon,
greeting=greeting,
cycle_order=cycle_order,
)
def _handle_slash_command(
slash_text: str,
*,
registry: PersonaRegistry,
sessions: dict[str, _ReplSessionState],
current_session: str,
) -> tuple[int, str]:
"""Handle slash commands and return (exit_code, current_session)."""
try:
tokens = shlex.split(slash_text)
except ValueError as exc:
_console.print(f"[red]Parse error:[/red] {exc}")
return 2, current_session
if not tokens:
_console.print("[yellow]Empty slash command.[/yellow]")
return 2, current_session
if tokens[0] == "persona":
subcommands = {"list", "set", "create", "pick", "delete", "export", "import"}
if len(tokens) == 2 and tokens[1] not in subcommands:
tokens = ["persona", "set", tokens[1]]
if len(tokens) == 1 or (len(tokens) > 1 and tokens[1] == "list"):
_print_persona_list(registry, sessions, current_session)
return 0, current_session
if tokens[1] == "set":
if len(tokens) < 3:
_console.print("[red]Usage:[/red] /persona set <name>")
return 2, current_session
target = tokens[2]
persona = registry.get_persona(target)
if persona is None:
_console.print(f"[red]Persona not found:[/red] {target}")
return 1, current_session
sessions[current_session].active_persona = persona.name
registry.set_active_persona_name(persona.name)
_console.print(f"[green]Active persona:[/green] {persona.name}")
return 0, current_session
if tokens[1] == "create":
try:
persona = _parse_persona_create(tokens)
except ValueError as exc:
_console.print(f"[red]{exc}[/red]")
return 2, current_session
registry.save_persona(persona)
_console.print(f"[green]Created persona:[/green] {persona.name}")
return 0, current_session
if tokens[1] == "pick":
personas = registry.list_personas()
if not personas:
_console.print("[yellow]No personas available.[/yellow]")
return 1, current_session
if len(tokens) > 2:
query = tokens[2]
names = [persona.name for persona in personas]
narrowed = _find_reference_candidates(query, names, limit=len(names))
personas = [persona for persona in personas if persona.name in narrowed]
if not personas:
_console.print(f"[yellow]No personas matched '{query}'.[/yellow]")
return 1, current_session
_console.print("[bold]Persona picker[/bold]")
for idx, persona in enumerate(personas, start=1):
_console.print(f" {idx}. {persona.name} ({persona.actor})")
selection = input("Select persona number: ").strip()
try:
selected_idx = int(selection)
except ValueError:
_console.print("[red]Invalid selection.[/red]")
return 2, current_session
if selected_idx < 1 or selected_idx > len(personas):
_console.print("[red]Selection out of range.[/red]")
return 2, current_session
selected = personas[selected_idx - 1]
sessions[current_session].active_persona = selected.name
registry.set_active_persona_name(selected.name)
_console.print(f"[green]Active persona:[/green] {selected.name}")
return 0, current_session
if tokens[1] == "delete":
if len(tokens) < 3:
_console.print("[red]Usage:[/red] /persona delete <name>")
return 2, current_session
target = tokens[2]
if target == "default":
_console.print("[red]Cannot delete default persona.[/red]")
return 1, current_session
deleted = registry.delete_persona(target)
if not deleted:
_console.print(f"[red]Persona not found:[/red] {target}")
return 1, current_session
reset_any = False
for state in sessions.values():
if state.active_persona == target:
state.active_persona = "default"
reset_any = True
if reset_any:
registry.set_active_persona_name("default")
_console.print(f"[green]Deleted persona:[/green] {target}")
return 0, current_session
if tokens[1] == "export":
if len(tokens) < 3:
_console.print("[red]Usage:[/red] /persona export <name> [path]")
return 2, current_session
target = tokens[2]
persona = registry.get_persona(target)
if persona is None:
_console.print(f"[red]Persona not found:[/red] {target}")
return 1, current_session
output_path = Path(tokens[3]) if len(tokens) > 3 else Path(f"{target}.yaml")
try:
safe_output_path = registry.resolve_export_path(output_path)
except ValueError as exc:
_console.print(f"[red]{exc}[/red]")
return 1, current_session
payload = persona.model_dump(mode="json", exclude_none=True)
safe_output_path.write_text(
yaml.safe_dump(payload, sort_keys=False, default_flow_style=False),
encoding="utf-8",
)
_console.print(f"[green]Exported persona:[/green] {safe_output_path}")
return 0, current_session
if tokens[1] == "import":
if len(tokens) < 3:
_console.print("[red]Usage:[/red] /persona import <path>")
return 2, current_session
input_path = Path(tokens[2])
try:
safe_input_path = registry.resolve_import_path(input_path)
except ValueError as exc:
_console.print(f"[red]{exc}[/red]")
return 1, current_session
if not safe_input_path.exists():
_console.print(f"[red]File not found:[/red] {safe_input_path}")
return 1, current_session
raw = yaml.safe_load(safe_input_path.read_text(encoding="utf-8")) or {}
if not isinstance(raw, dict):
_console.print("[red]Invalid persona YAML.[/red]")
return 1, current_session
try:
persona = PersonaConfig.model_validate(raw)
except Exception as exc:
_console.print(f"[red]Persona validation failed:[/red] {exc}")
return 1, current_session
registry.save_persona(persona)
_console.print(f"[green]Imported persona:[/green] {persona.name}")
return 0, current_session
_console.print(f"[red]Unknown persona command:[/red] {' '.join(tokens)}")
return 2, current_session
if tokens[0] == "session":
if len(tokens) == 1 or (len(tokens) > 1 and tokens[1] == "list"):
_console.print("[bold]REPL Sessions[/bold]")
for name, state in sessions.items():
marker = "*" if name == current_session else " "
_console.print(f" {marker} {name} (persona={state.active_persona})")
return 0, current_session
if tokens[1] == "new":
if len(tokens) < 3:
_console.print("[red]Usage:[/red] /session new <name>")
return 2, current_session
new_name = tokens[2]
if new_name in sessions:
_console.print(f"[red]Session already exists:[/red] {new_name}")
return 1, current_session
sessions[new_name] = _ReplSessionState(
name=new_name,
active_persona=sessions[current_session].active_persona,
)
_console.print(f"[green]Created session:[/green] {new_name}")
return 0, current_session
if tokens[1] == "switch":
if len(tokens) < 3:
_console.print("[red]Usage:[/red] /session switch <name>")
return 2, current_session
target_session = tokens[2]
if target_session not in sessions:
_console.print(f"[red]Session not found:[/red] {target_session}")
return 1, current_session
_console.print(f"[green]Switched session:[/green] {target_session}")
return 0, target_session
_console.print(f"[red]Unknown session command:[/red] {' '.join(tokens)}")
return 2, current_session
_console.print(f"[red]Unknown slash command:[/red] /{' '.join(tokens)}")
return 2, current_session
def dispatch_command(argv: Sequence[str]) -> int:
"""Invoke the main Typer CLI with the given argument vector.
@@ -198,6 +713,13 @@ def run_repl(
if use_history:
_setup_history(history_path)
_setup_completer()
persona_registry = PersonaRegistry()
default_persona = persona_registry.ensure_default_persona()
active_persona = persona_registry.get_active_persona_name() or default_persona.name
sessions: dict[str, _ReplSessionState] = {
"default": _ReplSessionState(name="default", active_persona=active_persona)
}
current_session = "default"
_console.print(
"[bold green]CleverAgents REPL[/bold green] — "
@@ -209,7 +731,7 @@ def run_repl(
exit_code = 0
while True:
prompt = _get_prompt_context()
prompt = _compose_prompt(_get_prompt_context(), sessions[current_session])
try:
line = input(prompt)
except EOFError:
@@ -244,6 +766,34 @@ def run_repl(
_print_repl_help()
continue
# Shell mode
if line.startswith("!"):
last_command = line
exit_code = _run_shell_command(line[1:].strip())
continue
# Slash command mode
if line.startswith("/"):
last_command = line
exit_code, current_session = _handle_slash_command(
line[1:],
registry=persona_registry,
sessions=sessions,
current_session=current_session,
)
continue
# Normal mode @ references
reference_catalog = _build_reference_catalog() if "@" in line else None
expanded_line = _expand_references(line, catalog=reference_catalog)
if expanded_line != line:
_console.print(f"[dim]{expanded_line}[/dim]")
elif "@" in line:
hints = _reference_suggestions(line, catalog=reference_catalog)
if hints:
_console.print("[dim]Reference suggestions:[/dim] " + ", ".join(hints))
line = expanded_line
# Tokenise and dispatch
try:
argv = shlex.split(line)
src/cleveragents/cli/commands/tui.py
+26
View File
@@ -0,0 +1,26 @@
"""CLI entrypoint for the Textual TUI."""
from __future__ import annotations
from typing import Annotated
import typer
app = typer.Typer(help="Launch the Textual TUI.")
@app.callback(invoke_without_command=True)
def tui_callback(
headless: Annotated[
bool,
typer.Option(
"--headless",
help="Run a one-shot headless startup check instead of full UI loop.",
),
] = False,
) -> None:
"""Launch the CleverAgents TUI."""
# Import lazily so non-TUI commands avoid Textual startup cost.
from cleveragents.tui.commands import run_tui
raise typer.Exit(run_tui(headless=headless))
src/cleveragents/cli/main.py
+8
View File
@@ -92,6 +92,7 @@ def _register_subcommands() -> None:
session,
skill,
tool,
tui,
validation,
)
from cleveragents.cli.commands.auto_debug import app as auto_debug_app
@@ -189,6 +190,11 @@ def _register_subcommands() -> None:
name="repl",
help="Start an interactive REPL session",
)
app.add_typer(
tui.app,
name="tui",
help="Launch the Textual TUI",
)
app.add_typer(
server_app,
name="server",
@@ -226,6 +232,7 @@ def _print_basic_help() -> None:
typer.echo(" apply Apply plan changes")
typer.echo(" auto-debug Auto-debug operations")
typer.echo(" repl Interactive REPL session")
typer.echo(" tui Textual terminal UI")
typer.echo(" version Show version")
typer.echo("")
typer.echo("Actors: set a default with 'agents actor set-default <name>'.")
@@ -610,6 +617,7 @@ def main(args: list[str] | None = None) -> int:
"automation-profile", # Automation profile management
"invariant", # Invariant constraint management
"repl", # Interactive REPL
"tui", # Textual TUI
"server", # Server connection management
"tell", # Shortcut for plan tell
"build", # Shortcut for plan build
src/cleveragents/cli/persona.py
+51
View File
@@ -0,0 +1,51 @@
"""CLI compatibility layer for canonical TUI persona modules."""
from __future__ import annotations
from pathlib import Path
from cleveragents.tui.persona.registry import PersonaRegistry as _TuiPersonaRegistry
from cleveragents.tui.persona.schema import Persona as PersonaConfig
class PersonaRegistry(_TuiPersonaRegistry):
"""Backwards-compatible API for CLI callers.
Canonical persona schema/registry live in `cleveragents.tui.persona.*`.
This class preserves old method names used by REPL and CLI command code.
"""
@property
def state_file(self) -> Path:
return self.state_path
@property
def personas_lock_file(self) -> Path:
return self.personas_lock_path
@property
def state_lock_file(self) -> Path:
return self.state_lock_path
def list_personas(self) -> list[PersonaConfig]:
return super().list_personas()
def get_persona(self, name: str) -> PersonaConfig | None:
return super().get(name)
def save_persona(self, persona: PersonaConfig) -> Path:
return super().save(persona)
def delete_persona(self, name: str) -> bool:
return super().delete(name)
def get_active_persona_name(self) -> str | None:
return super().get_last_persona()
def set_active_persona_name(self, name: str) -> None:
super().set_last_persona(name)
def ensure_default_persona(
self, actor: str = "local/mock-default"
) -> PersonaConfig:
return super().ensure_default(actor=actor)
src/cleveragents/tui/__init__.py
+18
View File
@@ -0,0 +1,18 @@
"""Textual TUI package for CleverAgents."""
from __future__ import annotations
from cleveragents.tui.app import CleverAgentsTuiApp
def run_tui(*, headless: bool = False) -> int:
"""Run TUI command entrypoint lazily to avoid import cycles."""
from cleveragents.tui.commands import run_tui as _run_tui
return _run_tui(headless=headless)
__all__ = [
"CleverAgentsTuiApp",
"run_tui",
]
src/cleveragents/tui/app.py
+189
View File
@@ -0,0 +1,189 @@
"""Textual TUI application shell."""
from __future__ import annotations
import importlib
import os
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, ClassVar, Protocol
from cleveragents.tui.input.modes import InputMode, InputModeRouter
from cleveragents.tui.input.reference_parser import suggestions
from cleveragents.tui.persona.state import PersonaState
from cleveragents.tui.widgets.persona_bar import PersonaBar
from cleveragents.tui.widgets.prompt import PromptInput
from cleveragents.tui.widgets.reference_picker import ReferencePickerOverlay
from cleveragents.tui.widgets.slash_command_overlay import SlashCommandOverlay
if TYPE_CHECKING:
InputSubmittedEvent = Any
else:
InputSubmittedEvent = Any
_TEXTUAL_AVAILABLE = False
_TextualApp: type[Any] = object
_Vertical: type[Any] = object
_Header: type[Any] = object
_Footer: type[Any] = object
_Static: type[Any] = object
try: # pragma: no branch - import gate for optional dependency
_textual_app = importlib.import_module("textual.app")
_textual_containers = importlib.import_module("textual.containers")
_textual_widgets = importlib.import_module("textual.widgets")
_TextualApp = _textual_app.App
_Vertical = _textual_containers.Vertical
_Header = _textual_widgets.Header
_Footer = _textual_widgets.Footer
_Static = _textual_widgets.Static
_TEXTUAL_AVAILABLE = True
except Exception: # pragma: no cover
_TEXTUAL_AVAILABLE = False
def textual_available() -> bool:
"""Return whether Textual import succeeded."""
return _TEXTUAL_AVAILABLE
@dataclass(slots=True)
class SessionView:
"""Minimal per-session TUI view model."""
session_id: str
transcript: list[str]
class _CommandRouter(Protocol):
"""Router protocol used by the TUI input-mode command handler."""
def handle(self, raw: str, *, session_id: str) -> str:
"""Dispatch a slash command and return a rendered response."""
...
class _FallbackCleverAgentsTuiApp: # pragma: no cover
"""Fallback app that raises actionable dependency error."""
def __init__(
self, *, command_router: _CommandRouter, persona_state: PersonaState
) -> None:
del command_router, persona_state
def run(self) -> None:
raise RuntimeError(
"Textual dependency missing. Install with: pip install 'cleveragents[tui]'"
)
_ResolvedTuiApp: type[Any] = _FallbackCleverAgentsTuiApp
if _TEXTUAL_AVAILABLE:
class _TextualCleverAgentsTuiApp(_TextualApp):
"""Main TUI app."""
CSS_PATH: ClassVar[str] = "cleveragents.tcss"
BINDINGS: ClassVar[list[tuple[str, str, str]]] = [
("ctrl+q", "quit", "Quit"),
("f1", "help", "Help"),
("ctrl+t", "cycle_preset", "Cycle Preset"),
]
def __init__(
self,
*,
command_router: _CommandRouter,
persona_state: PersonaState,
) -> None:
brent.edwards commented 2026-03-16 20:29:30 +00:00
Outdated
Review
Copy Link
Copy Source

P2: compose() return type should be ComposeResult, not Any.

Use a TYPE_CHECKING conditional import:

if TYPE_CHECKING:
    from textual.app import ComposeResult

Then annotate def compose(self) -> ComposeResult:.

**P2: `compose()` return type should be `ComposeResult`, not `Any`.** Use a `TYPE_CHECKING` conditional import: ```python if TYPE_CHECKING: from textual.app import ComposeResult ``` Then annotate `def compose(self) -> ComposeResult:`.
super().__init__()
self._command_router = command_router
self._persona_state = persona_state
self._session = SessionView(session_id="default", transcript=[])
def compose(self) -> Any:
yield _Header(show_clock=True)
with _Vertical(id="main-column"):
yield _Static("CleverAgents TUI", id="conversation")
yield ReferencePickerOverlay(id="reference-picker")
yield SlashCommandOverlay(id="slash-overlay")
yield PromptInput(
placeholder="Type message, /command, or !shell ...", id="prompt"
)
yield PersonaBar(id="persona-bar")
yield _Footer()
def on_mount(self) -> None:
self._refresh_persona_bar()
ref_picker = self.query_one("#reference-picker", ReferencePickerOverlay)
ref_picker.set_suggestions("", [])
slash = self.query_one("#slash-overlay", SlashCommandOverlay)
slash.set_commands(
"", ["persona list", "persona set", "session show", "help"]
)
def action_help(self) -> None:
conversation = self.query_one("#conversation", _Static)
conversation.update("Hotkeys: F1 help, Ctrl+Q quit, Ctrl+T cycle preset")
def action_cycle_preset(self) -> None:
self._persona_state.cycle_preset(self._session.session_id)
self._refresh_persona_bar()
def _refresh_persona_bar(self) -> None:
persona = self._persona_state.active_persona(self._session.session_id)
preset = self._persona_state.current_preset(self._session.session_id)
scope_count = len(persona.scoped_projects) + len(persona.scoped_plans)
scope_text = f"{scope_count} scope refs"
bar = self.query_one("#persona-bar", PersonaBar)
bar.set_content(
persona_name=persona.name,
actor_name=persona.actor,
preset_name=preset,
scope_text=scope_text,
)
def on_input_submitted(self, event: InputSubmittedEvent) -> None:
del event
prompt = self.query_one("#prompt", PromptInput)
payload = prompt.consume_text()
text = payload.text.strip()
if not text:
return
mode_router = InputModeRouter(
command_handler=lambda raw: self._command_router.handle(
raw, session_id=self._session.session_id
),
shell_confirm=lambda _cmd: (
os.environ.get("CLEVERAGENTS_ALLOW_DANGEROUS_SHELL", "").strip()
in {"1", "true"}
),
)
result = mode_router.process(text)
conversation = self.query_one("#conversation", _Static)
if result.mode == InputMode.COMMAND:
conversation.update(result.command_result or "")
self._refresh_persona_bar()
return
if result.mode == InputMode.SHELL:
shell = result.shell_result
if shell is None:
conversation.update("(no shell output)")
return
output = (
shell.stdout.strip() or shell.stderr.strip() or "(empty output)"
)
conversation.update(f"$ {shell.command}\n{output}")
return
preview = result.expanded_text
if "@" in text:
ref_picker = self.query_one("#reference-picker", ReferencePickerOverlay)
ref_picker.set_suggestions(
text, suggestions(text.replace("@", "").strip())
)
conversation.update(preview)
_ResolvedTuiApp = _TextualCleverAgentsTuiApp
CleverAgentsTuiApp = _ResolvedTuiApp
src/cleveragents/tui/cleveragents.tcss
+44
View File
@@ -0,0 +1,44 @@
Screen {
layout: vertical;
}
#main-column {
layout: vertical;
height: 1fr;
}
#conversation {
height: 1fr;
padding: 1 2;
border: heavy $primary;
background: $panel;
}
#reference-picker {
height: auto;
max-height: 8;
padding: 0 1;
border: round $secondary;
color: $text-muted;
}
#slash-overlay {
height: auto;
max-height: 8;
padding: 0 1;
border: round $accent;
color: $text;
}
#prompt {
height: auto;
border: round $primary;
margin: 1 0 0 0;
}
#persona-bar {
height: auto;
padding: 0 1;
color: $text-primary;
background: $foreground 5%;
}
src/cleveragents/tui/commands.py
+72
View File
@@ -0,0 +1,72 @@
"""TUI command routing and CLI launch integration."""
from __future__ import annotations
import json
from dataclasses import dataclass
from cleveragents.application.container import get_container
from cleveragents.tui.app import CleverAgentsTuiApp, textual_available
from cleveragents.tui.persona.registry import PersonaRegistry
from cleveragents.tui.persona.state import PersonaState
@dataclass(slots=True)
class TuiCommandRouter:
"""Slash command router used by the TUI prompt."""
persona_registry: PersonaRegistry
persona_state: PersonaState
def handle(self, raw: str, *, session_id: str) -> str:
tokens = raw.strip().split()
if not tokens:
return "Empty command"
if tokens[0] == "persona":
return self._persona_command(tokens[1:], session_id=session_id)
if tokens[0] == "session":
return self._session_command(tokens[1:], session_id=session_id)
if tokens[0] == "help":
return "Commands: /persona, /session, /help"
return f"Unknown command: /{raw}"
def _persona_command(self, tokens: list[str], *, session_id: str) -> str:
if not tokens or tokens[0] == "list":
names = [persona.name for persona in self.persona_registry.list_personas()]
return ", ".join(names) if names else "No personas"
if tokens[0] == "set":
if len(tokens) < 2:
return "Usage: /persona set <name>"
persona = self.persona_state.set_active_persona(session_id, tokens[1])
return f"Active persona: {persona.name}"
return f"Unknown persona command: {' '.join(tokens)}"
@staticmethod
def _session_command(tokens: list[str], *, session_id: str) -> str:
if not tokens or tokens[0] == "show":
return f"Current session: {session_id}"
return f"Unknown session command: {' '.join(tokens)}"
def run_tui(*, headless: bool = False) -> int:
"""Run the Textual TUI app or a headless startup check."""
container = get_container()
registry = container.persona_registry()
state = container.persona_state(registry=registry)
router = TuiCommandRouter(persona_registry=registry, persona_state=state)
brent.edwards commented 2026-03-16 20:29:30 +00:00
Outdated
Review
Copy Link
Copy Source

P1: Persona registry and state are instantiated directly, bypassing the DI container (ADR-003).

registry = PersonaRegistry()
state = PersonaState(registry=registry)

These should be registered as providers in Container and resolved via container.persona_registry() etc., consistent with all other services.

**P1: Persona registry and state are instantiated directly, bypassing the DI container (ADR-003).** ```python registry = PersonaRegistry() state = PersonaState(registry=registry) ``` These should be registered as providers in `Container` and resolved via `container.persona_registry()` etc., consistent with all other services.
if headless:
payload = {
"textual_available": textual_available(),
"persona_count": len(registry.list_personas())
if registry.personas_dir.exists()
else 0,
"default_persona": state.active_name("default"),
"help": router.handle("help", session_id="default"),
}
print(json.dumps(payload, indent=2))
return 0
app = CleverAgentsTuiApp(command_router=router, persona_state=state)
app.run()
return 0
src/cleveragents/tui/input/__init__.py
+22
View File
@@ -0,0 +1,22 @@
"""Input handling for TUI."""
from cleveragents.tui.input.modes import InputMode, InputModeRouter, ModeResult
from cleveragents.tui.input.reference_parser import (
ReferenceMatch,
ReferenceParseResult,
parse_references,
suggestions,
)
from cleveragents.tui.input.shell_exec import ShellResult, run_shell_command
__all__ = [
"InputMode",
"InputModeRouter",
"ModeResult",
"ReferenceMatch",
"ReferenceParseResult",
"ShellResult",
"parse_references",
"run_shell_command",
"suggestions",
]
src/cleveragents/tui/input/modes.py
+92
View File
@@ -0,0 +1,92 @@
"""Input mode router for TUI prompt handling."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from enum import StrEnum
from cleveragents.tui.input.reference_parser import (
ReferenceParseResult,
parse_references,
)
from cleveragents.tui.input.shell_exec import ShellResult, run_shell_command
class InputMode(StrEnum):
"""Prompt input mode."""
NORMAL = "normal"
COMMAND = "command"
SHELL = "shell"
@dataclass(slots=True, frozen=True)
class ModeResult:
"""Mode-router output."""
mode: InputMode
expanded_text: str
references: list[str]
shell_result: ShellResult | None
command_result: str | None
class InputModeRouter:
"""Dispatch user input to normal/command/shell handlers."""
def __init__(
self,
command_handler: Callable[[str], str],
*,
shell_confirm: Callable[[str], bool] | None = None,
shell_timeout_seconds: int = 30,
) -> None:
self._command_handler = command_handler
self._shell_confirm = shell_confirm
self._shell_timeout_seconds = shell_timeout_seconds
@staticmethod
def detect_mode(text: str) -> InputMode:
stripped = text.lstrip()
if stripped.startswith("/"):
return InputMode.COMMAND
if stripped.startswith("!"):
return InputMode.SHELL
return InputMode.NORMAL
def process(self, text: str) -> ModeResult:
mode = self.detect_mode(text)
if mode == InputMode.COMMAND:
cmd = text.lstrip()[1:]
result = self._command_handler(cmd)
return ModeResult(
mode=mode,
expanded_text=text,
references=[],
shell_result=None,
command_result=result,
)
if mode == InputMode.SHELL:
command = text.lstrip()[1:].strip()
shell_result = run_shell_command(
command,
confirm_dangerous=self._shell_confirm,
timeout_seconds=self._shell_timeout_seconds,
)
return ModeResult(
mode=mode,
expanded_text=text,
references=[],
shell_result=shell_result,
command_result=None,
)
parse_result: ReferenceParseResult = parse_references(text)
return ModeResult(
mode=mode,
expanded_text=parse_result.expanded_line,
references=[match.canonical for match in parse_result.matches],
shell_result=None,
command_result=None,
)
src/cleveragents/tui/input/reference_parser.py
+183
View File
@@ -0,0 +1,183 @@
"""Reference parsing and canonical expansion for @ tokens."""
from __future__ import annotations
import os
import time
from dataclasses import dataclass
from pathlib import Path
from cleveragents.tui.search.fuzzy import rank_candidates
_REFERENCE_TYPES = ("project", "plan", "resource", "actor", "tool", "skill")
_IGNORED_DIRS = {".git", ".venv", "node_modules", "__pycache__", ".mypy_cache"}
_CATALOG_LIMIT = 400
_CATALOG_CACHE_TTL_SECONDS = 5.0
_catalog_cache: dict[str, object] = {"cwd": None, "created_at": 0.0, "catalog": None}
@dataclass(slots=True, frozen=True)
class ReferenceMatch:
"""Represents one resolved reference token."""
token: str
canonical: str
category: str
@dataclass(slots=True)
class ReferenceParseResult:
"""Result of parsing a line for references."""
expanded_line: str
matches: list[ReferenceMatch]
unresolved_tokens: list[str]
def _catalog() -> dict[str, list[str]]:
brent.edwards commented 2026-03-16 20:29:30 +00:00
Outdated
Review
Copy Link
Copy Source

P1: cwd.rglob("*") traverses .git/, .venv/, node_modules/, etc.

This is both a performance issue (can be very slow) and an information-disclosure concern (.env files, credentials, etc. show up as reference candidates). Add a skip-list for common ignored directories and filter during traversal.

**P1: `cwd.rglob("*")` traverses `.git/`, `.venv/`, `node_modules/`, etc.** This is both a performance issue (can be very slow) and an information-disclosure concern (`.env` files, credentials, etc. show up as reference candidates). Add a skip-list for common ignored directories and filter during traversal.
cwd = Path.cwd()
now = time.time()
cached_cwd = _catalog_cache.get("cwd")
cached_time = _catalog_cache.get("created_at")
cached_catalog = _catalog_cache.get("catalog")
if (
isinstance(cached_cwd, Path)
and cached_cwd == cwd
and isinstance(cached_time, float)
and now - cached_time < _CATALOG_CACHE_TTL_SECONDS
and isinstance(cached_catalog, dict)
):
return cached_catalog
files = []
for root, dirs, filenames in os.walk(cwd, followlinks=False):
dirs[:] = [name for name in dirs if name not in _IGNORED_DIRS]
root_path = Path(root)
for filename in filenames:
try:
files.append((root_path / filename).relative_to(cwd).as_posix())
except ValueError:
files.append((root_path / filename).as_posix())
if len(files) >= _CATALOG_LIMIT:
break
if len(files) >= _CATALOG_LIMIT:
break
catalog = {
"resource": sorted(files),
"project": [value for value in [cwd.name] if value],
"plan": [],
"actor": sorted(
f"local/{file.stem}"
for file in (cwd / "examples" / "actors").glob("*.y*ml")
)
if (cwd / "examples" / "actors").is_dir()
else [],
"tool": sorted(
f"local/{file.stem}" for file in (cwd / "examples" / "tools").glob("*.y*ml")
)
if (cwd / "examples" / "tools").is_dir()
else [],
"skill": sorted(
f"local/{entry.name}"
for entry in (cwd / "examples" / "skills").iterdir()
if entry.is_dir()
)
if (cwd / "examples" / "skills").is_dir()
else [],
}
_catalog_cache["cwd"] = cwd
_catalog_cache["created_at"] = now
_catalog_cache["catalog"] = catalog
return catalog
def _resolve(category: str, query: str, items: dict[str, list[str]]) -> str | None:
values = items.get(category, [])
ranked = rank_candidates(query, values, limit=1)
if not ranked:
return None
return ranked[0].value
def parse_references(line: str) -> ReferenceParseResult:
"""Expand @tokens and return canonical forms."""
items = _catalog()
parts = line.split()
expanded: list[str] = []
matches: list[ReferenceMatch] = []
unresolved: list[str] = []
for part in parts:
if part.startswith("\\@"):
expanded.append(part[1:])
continue
if not part.startswith("@") or part == "@":
expanded.append(part)
continue
raw = part[1:]
chosen_category = ""
query = raw
if ":" in raw:
category, rest = raw.split(":", 1)
if category in _REFERENCE_TYPES:
chosen_category = category
query = rest
if chosen_category:
resolved = _resolve(chosen_category, query, items)
if resolved:
canonical = f"@{chosen_category}:{resolved}"
expanded.append(canonical)
matches.append(
ReferenceMatch(
token=part, canonical=canonical, category=chosen_category
)
)
else:
expanded.append(part)
unresolved.append(part)
continue
resolved_category = ""
resolved_value = ""
for category in _REFERENCE_TYPES:
candidate = _resolve(category, raw, items)
if candidate:
resolved_category = category
resolved_value = candidate
break
if resolved_category:
canonical = f"@{resolved_category}:{resolved_value}"
expanded.append(canonical)
matches.append(
ReferenceMatch(
token=part, canonical=canonical, category=resolved_category
)
)
else:
expanded.append(part)
unresolved.append(part)
return ReferenceParseResult(
expanded_line=" ".join(expanded),
matches=matches,
unresolved_tokens=unresolved,
)
def suggestions(
query: str, *, category: str | None = None, limit: int = 8
) -> list[str]:
"""Return canonical suggestion strings for a query."""
items = _catalog()
target_categories = [category] if category else list(_REFERENCE_TYPES)
result: list[str] = []
for target in target_categories:
if target is None or target not in items:
continue
ranked = rank_candidates(query, items[target], limit=limit)
for candidate in ranked:
result.append(f"@{target}:{candidate.value}")
return result[:limit]
src/cleveragents/tui/input/shell_exec.py
+89
View File
@@ -0,0 +1,89 @@
"""Shell mode execution helpers for the TUI."""
from __future__ import annotations
import os
import subprocess
from collections.abc import Callable
from dataclasses import dataclass
@dataclass(slots=True, frozen=True)
class ShellResult:
"""Structured shell execution result."""
command: str
exit_code: int
stdout: str
stderr: str
brent.edwards commented 2026-03-16 20:29:30 +00:00
Outdated
Review
Copy Link
Copy Source

P0: looks_dangerous() is dead code — never called by any caller.

run_shell_command() below executes with shell=True and no guard. Neither modes.py:InputModeRouter.process() nor repl.py:_run_shell_command() call looks_dangerous() before execution.

At minimum, this function should be called and a confirmation prompted. Add a timeout= parameter to subprocess.run() as well (default 30s).

**P0: `looks_dangerous()` is dead code — never called by any caller.** `run_shell_command()` below executes with `shell=True` and no guard. Neither `modes.py:InputModeRouter.process()` nor `repl.py:_run_shell_command()` call `looks_dangerous()` before execution. At minimum, this function should be called and a confirmation prompted. Add a `timeout=` parameter to `subprocess.run()` as well (default 30s).
def looks_dangerous(command: str) -> bool:
"""Best-effort dangerous command detector."""
lowered = command.strip().lower()
patterns = (
"rm -rf /",
"git push --force",
"mkfs.",
"dd if=",
":(){:|:&};:",
)
return any(pattern in lowered for pattern in patterns)
def run_shell_command(
command: str,
*,
confirm_dangerous: Callable[[str], bool] | None = None,
timeout_seconds: int = 30,
brent.edwards commented 2026-03-16 20:29:30 +00:00
Outdated
Review
Copy Link
Copy Source

P0: No timeout on subprocess.run — user can hang the entire process.

!sleep 999999 or !cat /dev/urandom will block indefinitely. Add timeout=30 (or make it configurable) and handle subprocess.TimeoutExpired.

**P0: No timeout on subprocess.run — user can hang the entire process.** `!sleep 999999` or `!cat /dev/urandom` will block indefinitely. Add `timeout=30` (or make it configurable) and handle `subprocess.TimeoutExpired`.
) -> ShellResult:
"""Execute shell command with basic safeguards.
Threat model:
- Shell mode is a convenience feature for local development, not a sandbox.
- We block obviously dangerous command patterns unless explicitly confirmed.
- We apply a timeout to avoid hanging the UI event loop indefinitely.
"""
if not command.strip():
return ShellResult(
command=command, exit_code=2, stdout="", stderr="empty command"
)
if os.environ.get("CLEVERAGENTS_DISABLE_SHELL_MODE", "").strip() in {"1", "true"}:
return ShellResult(
command=command,
exit_code=2,
stdout="",
stderr="shell mode is disabled",
)
if looks_dangerous(command):
confirmed = False
if confirm_dangerous is not None:
confirmed = confirm_dangerous(command)
if not confirmed:
return ShellResult(
command=command,
exit_code=1,
stdout="",
stderr="blocked dangerous shell command",
)
try:
proc = subprocess.run(
command,
shell=True,
text=True,
capture_output=True,
timeout=timeout_seconds,
) # nosec B602
except subprocess.TimeoutExpired:
return ShellResult(
command=command,
exit_code=124,
stdout="",
stderr=f"command timed out after {timeout_seconds}s",
)
return ShellResult(
command=command,
exit_code=proc.returncode,
stdout=proc.stdout or "",
stderr=proc.stderr or "",
)
src/cleveragents/tui/persona/__init__.py
+12
View File
@@ -0,0 +1,12 @@
"""Persona support for TUI."""
from cleveragents.tui.persona.registry import PersonaRegistry
from cleveragents.tui.persona.schema import Persona, PersonaPreset
from cleveragents.tui.persona.state import PersonaState
__all__ = [
"Persona",
"PersonaPreset",
"PersonaRegistry",
"PersonaState",
]
src/cleveragents/tui/persona/registry.py
+213
View File
@@ -0,0 +1,213 @@
"""Persona registry backed by local YAML files."""
from __future__ import annotations
import fcntl
import logging
import os
import tempfile
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
import yaml
from pydantic import ValidationError
from cleveragents.tui.persona.schema import Persona
_logger = logging.getLogger(__name__)
def _default_config_dir() -> Path:
override = os.environ.get("CLEVERAGENTS_CONFIG_DIR", "").strip()
if override:
return Path(override)
return Path.home() / ".config" / "cleveragents"
@dataclass(slots=True)
class PersonaRegistry:
"""Load and persist TUI-only personas."""
config_dir: Path = field(default_factory=_default_config_dir)
@property
def personas_dir(self) -> Path:
return self.config_dir / "personas"
@property
def state_path(self) -> Path:
return self.config_dir / "tui-state.yaml"
@property
def personas_lock_path(self) -> Path:
return self.config_dir / ".personas.lock"
@property
def state_lock_path(self) -> Path:
return self.config_dir / ".state.lock"
def ensure_dirs(self) -> None:
self.personas_dir.mkdir(parents=True, exist_ok=True)
def _lock_file(self, lock_path: Path) -> Any:
lock_path.parent.mkdir(parents=True, exist_ok=True)
handle = lock_path.open("a+", encoding="utf-8")
fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
return handle
@staticmethod
def _atomic_write_yaml(path: Path, payload: dict[str, Any]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with tempfile.NamedTemporaryFile(
mode="w",
encoding="utf-8",
dir=str(path.parent),
prefix=f".{path.name}.",
suffix=".tmp",
delete=False,
) as handle:
yaml.safe_dump(payload, handle, sort_keys=False, default_flow_style=False)
temp_name = handle.name
Path(temp_name).replace(path)
def persona_path(self, name: str) -> Path:
Persona.validate_name_safe(name)
result = (self.personas_dir / f"{name}.yaml").resolve()
if not result.is_relative_to(self.personas_dir.resolve()):
raise ValueError(f"Invalid persona name: {name}")
return result
def resolve_export_path(self, output_path: Path) -> Path:
if output_path.is_absolute():
raise ValueError(
"Export path must be relative to current working directory"
)
base = Path.cwd().resolve()
resolved = (base / output_path).resolve()
if not resolved.is_relative_to(base):
raise ValueError("Export path must stay within working directory")
return resolved
def resolve_import_path(self, input_path: Path) -> Path:
if input_path.is_absolute():
raise ValueError(
"Import path must be relative to current working directory"
)
base = Path.cwd().resolve()
resolved = (base / input_path).resolve()
if not resolved.is_relative_to(base):
raise ValueError("Import path must stay within working directory")
return resolved
def list_personas(self) -> list[Persona]:
self.ensure_dirs()
personas: list[Persona] = []
for file in sorted(self.personas_dir.glob("*.y*ml")):
try:
raw = yaml.safe_load(file.read_text(encoding="utf-8")) or {}
if not isinstance(raw, dict):
continue
personas.append(Persona.model_validate(raw))
except (yaml.YAMLError, ValidationError) as exc:
_logger.warning("Skipping invalid persona file %s: %s", file, exc)
return personas
def get(self, name: str) -> Persona | None:
file = self.persona_path(name)
if not file.exists():
return None
raw = yaml.safe_load(file.read_text(encoding="utf-8")) or {}
if not isinstance(raw, dict):
return None
return Persona.model_validate(raw)
def _validate_cycle_order(self, persona: Persona) -> None:
if persona.cycle_order <= 0:
return
for existing in self.list_personas():
if existing.name == persona.name:
continue
if existing.cycle_order == persona.cycle_order:
raise ValueError("cycle_order must be unique when greater than zero")
def save(self, persona: Persona) -> Path:
self.ensure_dirs()
lock = self._lock_file(self.personas_lock_path)
try:
self._validate_cycle_order(persona)
file = self.persona_path(persona.name)
payload = persona.model_dump(mode="json", exclude_none=True)
self._atomic_write_yaml(file, payload)
return file
finally:
fcntl.flock(lock.fileno(), fcntl.LOCK_UN)
lock.close()
def delete(self, name: str) -> bool:
lock = self._lock_file(self.personas_lock_path)
try:
file = self.persona_path(name)
try:
file.unlink()
return True
except FileNotFoundError:
return False
finally:
fcntl.flock(lock.fileno(), fcntl.LOCK_UN)
lock.close()
def export_persona(self, name: str, output_path: Path) -> Path:
persona = self.get(name)
if persona is None:
raise ValueError(f"Persona not found: {name}")
payload = persona.model_dump(mode="json", exclude_none=True)
safe_output = self.resolve_export_path(output_path)
self._atomic_write_yaml(safe_output, payload)
return safe_output
def import_persona(self, input_path: Path) -> Persona:
safe_input = self.resolve_import_path(input_path)
raw = yaml.safe_load(safe_input.read_text(encoding="utf-8")) or {}
if not isinstance(raw, dict):
raise ValueError("Invalid persona file")
persona = Persona.model_validate(raw)
self.save(persona)
return persona
def load_state(self) -> dict[str, Any]:
if not self.state_path.exists():
return {}
raw = yaml.safe_load(self.state_path.read_text(encoding="utf-8")) or {}
if not isinstance(raw, dict):
return {}
return dict(raw)
def save_state(self, state: dict[str, Any]) -> None:
self.ensure_dirs()
self._atomic_write_yaml(self.state_path, state)
def get_last_persona(self) -> str | None:
state = self.load_state()
value = state.get("last_persona")
return value if isinstance(value, str) and value else None
def set_last_persona(self, name: str) -> None:
lock = self._lock_file(self.state_lock_path)
try:
state = self.load_state()
state["last_persona"] = name
self.save_state(state)
finally:
fcntl.flock(lock.fileno(), fcntl.LOCK_UN)
lock.close()
def ensure_default(self, actor: str = "local/mock-default") -> Persona:
existing = self.get("default")
if existing is not None:
return existing
persona = Persona(name="default", actor=actor, description="Default persona")
saved = Persona.model_validate(persona.model_dump())
self.save(saved)
self.set_last_persona(saved.name)
return saved
src/cleveragents/tui/persona/schema.py
+84
View File
@@ -0,0 +1,84 @@
"""TUI persona schema models."""
from __future__ import annotations
from typing import Any
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
class PersonaPreset(BaseModel):
"""Named argument override preset."""
name: str = Field(..., min_length=1)
display: str = Field(..., min_length=1)
overrides: dict[str, Any] = Field(default_factory=dict)
model_config = ConfigDict(str_strip_whitespace=True, validate_assignment=True)
class Persona(BaseModel):
"""Primary persona record persisted as YAML."""
brent.edwards commented 2026-03-16 20:29:30 +00:00
Outdated
Review
Copy Link
Copy Source

P1: This is a near-duplicate of cli/persona.py:PersonaConfig.

Two separate Pydantic persona models with subtly different validation logic (this one has model_validator, color field; the other has different default-preset injection). Pick one canonical location and re-export from the other to prevent drift.

**P1: This is a near-duplicate of `cli/persona.py:PersonaConfig`.** Two separate Pydantic persona models with subtly different validation logic (this one has `model_validator`, `color` field; the other has different default-preset injection). Pick one canonical location and re-export from the other to prevent drift.
name: str = Field(..., min_length=1)
brent.edwards commented 2026-03-16 20:29:30 +00:00
Outdated
Review
Copy Link
Copy Source

P1: No name validation against path-separator characters.

Same path-traversal issue as cli/persona.py. Add:

@field_validator("name")
@classmethod
def validate_name_safe(cls, value: str) -> str:
    if any(c in value for c in ('/', '\\', '..', '\x00')):
        raise ValueError("name must not contain path separators")
    return value
**P1: No name validation against path-separator characters.** Same path-traversal issue as `cli/persona.py`. Add: ```python @field_validator("name") @classmethod def validate_name_safe(cls, value: str) -> str: if any(c in value for c in ('/', '\\', '..', '\x00')): raise ValueError("name must not contain path separators") return value ```
description: str = Field(default="")
icon: str = Field(default="")
actor: str = Field(..., min_length=1)
color: str | None = None
base_arguments: dict[str, Any] = Field(default_factory=dict)
scoped_projects: list[str] = Field(default_factory=list)
scoped_plans: list[str] = Field(default_factory=list)
argument_presets: list[PersonaPreset] = Field(default_factory=list)
cycle_order: int = Field(default=0, ge=0)
greeting: str = Field(default="")
@field_validator("actor")
@classmethod
def validate_actor_ref(cls, value: str) -> str:
if "/" not in value:
raise ValueError("actor must be namespaced (namespace/name)")
return value
@field_validator("name")
@classmethod
def validate_name_safe(cls, value: str) -> str:
if (
"/" in value
or "\\" in value
or ".." in value
or "\x00" in value
or any(ord(char) < 32 or ord(char) == 127 for char in value)
):
raise ValueError("name must not contain path/control separators")
return value
@model_validator(mode="after")
def ensure_default_preset(self) -> Persona:
if not self.argument_presets:
self.argument_presets = [
PersonaPreset(name="default", display="default", overrides={})
]
return self
defaults = [
preset for preset in self.argument_presets if preset.name == "default"
]
if len(defaults) != 1:
raise ValueError("must contain exactly one default preset")
if defaults[0].overrides:
raise ValueError("default preset overrides must be empty")
return self
def effective_arguments(self, preset_name: str | None = None) -> dict[str, Any]:
"""Compute effective arguments with shallow merge semantics."""
effective = dict(self.base_arguments)
if not preset_name:
return effective
preset = next(
(item for item in self.argument_presets if item.name == preset_name), None
)
if preset is None:
return effective
effective.update(preset.overrides)
return effective
model_config = ConfigDict(str_strip_whitespace=True, validate_assignment=True)
src/cleveragents/tui/persona/state.py
+69
View File
@@ -0,0 +1,69 @@
"""In-memory TUI session-to-persona state management."""
from __future__ import annotations
from dataclasses import dataclass, field
from cleveragents.tui.persona.registry import PersonaRegistry
from cleveragents.tui.persona.schema import Persona
@dataclass(slots=True)
class PersonaState:
"""Tracks active persona per TUI session."""
registry: PersonaRegistry
active_by_session: dict[str, str] = field(default_factory=dict)
preset_by_session: dict[str, str] = field(default_factory=dict)
def _resolve_default_name(self) -> str:
default = self.registry.ensure_default()
return self.registry.get_last_persona() or default.name
def active_name(self, session_id: str) -> str:
if session_id not in self.active_by_session:
self.active_by_session[session_id] = self._resolve_default_name()
return self.active_by_session[session_id]
def active_persona(self, session_id: str) -> Persona:
name = self.active_name(session_id)
persona = self.registry.get(name)
if persona is None:
persona = self.registry.ensure_default()
self.active_by_session[session_id] = persona.name
return persona
def set_active_persona(self, session_id: str, persona_name: str) -> Persona:
persona = self.registry.get(persona_name)
if persona is None:
raise ValueError(f"Unknown persona: {persona_name}")
self.active_by_session[session_id] = persona.name
self.registry.set_last_persona(persona.name)
if session_id not in self.preset_by_session:
self.preset_by_session[session_id] = "default"
return persona
def current_preset(self, session_id: str) -> str:
if session_id not in self.preset_by_session:
self.preset_by_session[session_id] = "default"
return self.preset_by_session[session_id]
def cycle_preset(self, session_id: str) -> str:
persona = self.active_persona(session_id)
if not persona.argument_presets:
self.preset_by_session[session_id] = "default"
return "default"
names = [item.name for item in persona.argument_presets]
current = self.current_preset(session_id)
if current not in names:
self.preset_by_session[session_id] = names[0]
return names[0]
idx = names.index(current)
next_name = names[(idx + 1) % len(names)]
self.preset_by_session[session_id] = next_name
return next_name
def effective_arguments(self, session_id: str) -> dict[str, object]:
persona = self.active_persona(session_id)
preset = self.current_preset(session_id)
return persona.effective_arguments(preset)
src/cleveragents/tui/search/__init__.py
+9
View File
@@ -0,0 +1,9 @@
"""Search utilities for TUI pickers."""
from cleveragents.tui.search.fuzzy import FuzzyCandidate, rank_candidates, score_match
__all__ = [
"FuzzyCandidate",
"rank_candidates",
"score_match",
]
src/cleveragents/tui/search/fuzzy.py
+57
View File
@@ -0,0 +1,57 @@
"""Weighted fuzzy search helpers for TUI pickers."""
from __future__ import annotations
import difflib
from dataclasses import dataclass
@dataclass(slots=True, frozen=True)
class FuzzyCandidate:
"""Result candidate with score metadata."""
value: str
score: float
reason: str
def score_match(query: str, value: str) -> FuzzyCandidate | None:
"""Score a query against value using ADR-aligned weighting."""
if not query:
return FuzzyCandidate(value=value, score=0.0, reason="empty")
if value.startswith(query):
return FuzzyCandidate(value=value, score=1.0, reason="prefix")
if query in value:
return FuzzyCandidate(value=value, score=0.7, reason="substring")
query_parts = [part for part in query.split("/") if part]
if query_parts and all(part in value for part in query_parts):
return FuzzyCandidate(value=value, score=0.8, reason="path-component")
ratio = difflib.SequenceMatcher(a=query, b=value).ratio()
if ratio >= 0.4:
return FuzzyCandidate(value=value, score=0.4 * ratio, reason="fuzzy")
return None
def rank_candidates(
query: str,
values: list[str],
*,
recency: dict[str, int] | None = None,
limit: int = 20,
) -> list[FuzzyCandidate]:
"""Return ranked fuzzy candidates with deterministic tie-breaking."""
scored: list[FuzzyCandidate] = []
for value in values:
candidate = score_match(query, value)
if candidate is not None:
scored.append(candidate)
recency_map = recency or {}
scored.sort(
key=lambda item: (
-item.score,
-recency_map.get(item.value, 0),
item.value,
)
)
return scored[:limit]
src/cleveragents/tui/widgets/__init__.py
+14
View File
@@ -0,0 +1,14 @@
"""Widget collection for CleverAgents TUI."""
from cleveragents.tui.widgets.persona_bar import PersonaBar
from cleveragents.tui.widgets.prompt import PromptInput, PromptSubmitted
from cleveragents.tui.widgets.reference_picker import ReferencePickerOverlay
from cleveragents.tui.widgets.slash_command_overlay import SlashCommandOverlay
__all__ = [
"PersonaBar",
"PromptInput",
"PromptSubmitted",
"ReferencePickerOverlay",
"SlashCommandOverlay",
]
src/cleveragents/tui/widgets/persona_bar.py
+38
View File
@@ -0,0 +1,38 @@
"""Persona bar widget."""
from __future__ import annotations
import importlib
from typing import Any
def _load_static_base() -> type[Any]:
brent.edwards commented 2026-03-16 20:29:30 +00:00
Outdated
Review
Copy Link
Copy Source

P2: Textual import fallback pattern is duplicated across all 4 widget files.

Extract a shared _widget_bases.py with the _load_static_base() / _load_input_base() helpers and the fallback classes to eliminate the duplication.

**P2: Textual import fallback pattern is duplicated across all 4 widget files.** Extract a shared `_widget_bases.py` with the `_load_static_base()` / `_load_input_base()` helpers and the fallback classes to eliminate the duplication.
try:
return importlib.import_module("textual.widgets").Static
except Exception: # pragma: no cover - optional dependency
class _FallbackStatic:
def __init__(self, *args: object, **kwargs: object) -> None:
self._text = ""
def update(self, text: str) -> None:
self._text = text
return _FallbackStatic
_StaticBase = _load_static_base()
class PersonaBar(_StaticBase):
"""Displays active persona and preset metadata."""
def set_content(
self,
*,
persona_name: str,
actor_name: str,
preset_name: str,
scope_text: str,
) -> None:
self.update(f"{persona_name} | {actor_name} | {preset_name} | {scope_text}")
src/cleveragents/tui/widgets/prompt.py
+40
View File
@@ -0,0 +1,40 @@
"""Prompt widget and submitted message event."""
from __future__ import annotations
import importlib
from dataclasses import dataclass
from typing import Any
def _load_input_base() -> type[Any]:
try:
return importlib.import_module("textual.widgets").Input
except Exception: # pragma: no cover
class _FallbackInput:
value = ""
def __init__(self, *args: object, **kwargs: object) -> None:
self.value = ""
return _FallbackInput
_InputBase = _load_input_base()
@dataclass(slots=True, frozen=True)
class PromptSubmitted:
"""Submitted prompt payload."""
text: str
class PromptInput(_InputBase):
"""Input widget wrapper with helper methods."""
def consume_text(self) -> PromptSubmitted:
text = self.value
self.value = ""
return PromptSubmitted(text=text)
src/cleveragents/tui/widgets/reference_picker.py
+37
View File
@@ -0,0 +1,37 @@
"""Reference picker overlay widget."""
from __future__ import annotations
import importlib
from typing import Any
def _load_static_base() -> type[Any]:
try:
return importlib.import_module("textual.widgets").Static
except Exception: # pragma: no cover
class _FallbackStatic:
def __init__(self, *args: object, **kwargs: object) -> None:
self._text = ""
def update(self, text: str) -> None:
self._text = text
return _FallbackStatic
_StaticBase = _load_static_base()
class ReferencePickerOverlay(_StaticBase):
"""Simple overlay displaying reference suggestions."""
def set_suggestions(self, query: str, suggestions: list[str]) -> None:
if not suggestions:
self.update(f"@{query}\n(no matches)")
return
lines = [f"@{query}"]
for item in suggestions[:10]:
lines.append(f" {item}")
self.update("\n".join(lines))
src/cleveragents/tui/widgets/slash_command_overlay.py
+39
View File
@@ -0,0 +1,39 @@
"""Slash command overlay widget."""
from __future__ import annotations
import importlib
from typing import Any
def _load_static_base() -> type[Any]:
try:
return importlib.import_module("textual.widgets").Static
except Exception: # pragma: no cover
class _FallbackStatic:
def __init__(self, *args: object, **kwargs: object) -> None:
self._text = ""
def update(self, text: str) -> None:
self._text = text
return _FallbackStatic
_StaticBase = _load_static_base()
class SlashCommandOverlay(_StaticBase):
"""Renderable list of slash command candidates."""
def set_commands(self, query: str, commands: list[str]) -> None:
filtered = (
[item for item in commands if item.startswith(query)] if query else commands
)
lines = [f"/{query}"]
for command in filtered[:12]:
lines.append(f" /{command}")
if len(lines) == 1:
lines.append(" (no commands)")
self.update("\n".join(lines))