Merge remote-tracking branch 'origin/master' into feature/m6-cli-polish

# Conflicts:
#	src/cleveragents/cli/commands/action.py

CHANGELOG.md

@@ -804,6 +804,18 @@
   `OperationalMetricKey` values with `MetricEntry` / `MetricCollector` for plan-level
   metrics. `TraceService` provides recording, querying, metric computation, and
   optional LangSmith forwarding when `LANGCHAIN_TRACING_V2=true`. (#500)
+- Polished CLI help text, progress indicators, and error messages across core command files.
+  Extracted shared output renderers into `src/cleveragents/cli/renderers.py` with
+  `render_detail`, `render_list`, `render_error`, `render_success`, `render_warning`, and
+  `render_empty` functions parameterised by `ColumnSpec`/`FieldSpec` dataclasses, ensuring
+  structurally-similar outputs are identical by construction. Migrated an initial subset of
+  commands to the shared renderers and standardised `--format` (rich/plain/json/yaml) behaviour
+  for those commands with stable column names and ordering for list views. `plain` format
+  enforces ASCII-only output via `_ascii_safe()` in `formatting.py` for log pipeline
+  compatibility. Added unified error envelope (`error.code`/`error.message`/`error.details`) for
+  JSON/YAML error outputs. Includes CLI output contract documentation
+  (`docs/reference/cli_output.md`), 14 Behave scenarios in `features/cli_output_formats.feature`,
+  2 Robot Framework test cases, and 11 ASV benchmarks in `benchmarks/cli_render_bench.py`. (#210)
 - Added `SafetyProfile` domain model with configurable safety constraints (allowed skill
   categories, sandbox/checkpoint requirements, human-approval flag, cost/retry limits) and
   integrated it into the `Action` model via `from_config`/`as_cli_dict`. Persistence backed

benchmarks/cli_render_bench.py

@@ -0,0 +1,189 @@
+"""ASV benchmarks for CLI output rendering overhead (issue #210).
+
+Measures render_detail, render_list, render_error, and render_success
+across all six output formats.
+"""
+
+from __future__ import annotations
+
+import importlib
+import io
+import sys
+from pathlib import Path
+
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+import cleveragents  # noqa: E402
+
+importlib.reload(cleveragents)
+
+from rich.console import Console  # noqa: E402
+
+import cleveragents.cli.renderers as _renderers_mod  # noqa: E402
+from cleveragents.cli.renderers import (  # noqa: E402
+    ColumnSpec,
+    render_detail,
+    render_error,
+    render_list,
+    render_success,
+)
+
+# ---------------------------------------------------------------------------
+# Shared sample data
+# ---------------------------------------------------------------------------
+
+_DETAIL_DATA = {
+    "name": "bench-item",
+    "count": 42,
+    "active": True,
+    "description": "A benchmark test item for measuring renderer overhead",
+}
+
+_LIST_COLUMNS = [
+    ColumnSpec(key="name", label="Name"),
+    ColumnSpec(key="value", label="Value"),
+    ColumnSpec(key="status", label="Status"),
+]
+
+_LIST_ROWS_10 = [
+    {
+        "name": f"item-{i}",
+        "value": i * 10,
+        "status": "active" if i % 2 == 0 else "inactive",
+    }
+    for i in range(10)
+]
+
+
+# ---------------------------------------------------------------------------
+# Console buffer helpers
+# ---------------------------------------------------------------------------
+
+
+def _install_buffer_consoles() -> None:
+    """Replace module-level consoles with buffer-backed instances."""
+    _renderers_mod._console = Console(
+        file=io.StringIO(), no_color=True, highlight=False
+    )
+    _renderers_mod._err_console = Console(
+        file=io.StringIO(), no_color=True, highlight=False
+    )
+
+
+def _reset_buffers() -> None:
+    """Seek to start and truncate the console buffers between iterations."""
+    for attr in ("_console", "_err_console"):
+        console = getattr(_renderers_mod, attr)
+        if console is not None and hasattr(console, "file"):
+            console.file.seek(0)
+            console.file.truncate()
+
+
+# ---------------------------------------------------------------------------
+# RenderDetailSuite
+# ---------------------------------------------------------------------------
+
+
+class RenderDetailSuite:
+    """Benchmark render_detail across formats."""
+
+    def setup(self) -> None:
+        _install_buffer_consoles()
+
+    def teardown(self) -> None:
+        _renderers_mod._console = None
+        _renderers_mod._err_console = None
+
+    def time_render_detail_json(self) -> None:
+        _reset_buffers()
+        render_detail(_DETAIL_DATA, fmt="json")
+
+    def time_render_detail_yaml(self) -> None:
+        _reset_buffers()
+        render_detail(_DETAIL_DATA, fmt="yaml")
+
+    def time_render_detail_plain(self) -> None:
+        _reset_buffers()
+        render_detail(_DETAIL_DATA, fmt="plain")
+
+    def time_render_detail_rich(self) -> None:
+        _reset_buffers()
+        render_detail(_DETAIL_DATA, fmt="rich")
+
+
+# ---------------------------------------------------------------------------
+# RenderListSuite
+# ---------------------------------------------------------------------------
+
+
+class RenderListSuite:
+    """Benchmark render_list across formats."""
+
+    def setup(self) -> None:
+        _install_buffer_consoles()
+
+    def teardown(self) -> None:
+        _renderers_mod._console = None
+        _renderers_mod._err_console = None
+
+    def time_render_list_json_10_rows(self) -> None:
+        _reset_buffers()
+        render_list(_LIST_ROWS_10, columns=_LIST_COLUMNS, fmt="json")
+
+    def time_render_list_plain_10_rows(self) -> None:
+        _reset_buffers()
+        render_list(_LIST_ROWS_10, columns=_LIST_COLUMNS, fmt="plain")
+
+    def time_render_list_rich_10_rows(self) -> None:
+        _reset_buffers()
+        render_list(_LIST_ROWS_10, columns=_LIST_COLUMNS, fmt="rich")
+
+
+# ---------------------------------------------------------------------------
+# RenderErrorSuite
+# ---------------------------------------------------------------------------
+
+
+class RenderErrorSuite:
+    """Benchmark render_error across formats."""
+
+    def setup(self) -> None:
+        _install_buffer_consoles()
+
+    def teardown(self) -> None:
+        _renderers_mod._console = None
+        _renderers_mod._err_console = None
+
+    def time_render_error_rich(self) -> None:
+        _reset_buffers()
+        render_error("BenchError", "something went wrong", fmt="rich")
+
+    def time_render_error_json(self) -> None:
+        _reset_buffers()
+        render_error("BenchError", "something went wrong", fmt="json")
+
+
+# ---------------------------------------------------------------------------
+# RenderSuccessSuite
+# ---------------------------------------------------------------------------
+
+
+class RenderSuccessSuite:
+    """Benchmark render_success across formats."""
+
+    def setup(self) -> None:
+        _install_buffer_consoles()
+
+    def teardown(self) -> None:
+        _renderers_mod._console = None
+        _renderers_mod._err_console = None
+
+    def time_render_success_rich(self) -> None:
+        _reset_buffers()
+        render_success("Benchmark complete", fmt="rich")
+
+    def time_render_success_plain(self) -> None:
+        _reset_buffers()
+        render_success("Benchmark complete", fmt="plain")

docs/reference/cli_output.md

@@ -0,0 +1,107 @@
+# CLI Output Contract
+
+## Overview
+
+All CleverAgents CLI commands use a shared output rendering layer that
+guarantees structural consistency across commands and formats.
+
+## Supported Formats
+
+| Format  | Description                               | ASCII-only |
+|---------|-------------------------------------------|------------|
+| `rich`  | Full Rich terminal markup with colours    | No         |
+| `color` | ANSI colour codes, no cursor movement     | No         |
+| `table` | ASCII box-drawing table layout            | Yes        |
+| `plain` | Key-value pairs, no markup                | Yes        |
+| `json`  | Indented JSON                             | Yes        |
+| `yaml`  | YAML document                             | Yes        |
+
+## Output Groups
+
+### Detail View (`render_detail`)
+
+Used by `show` and `create` commands. Renders a key-value panel.
+
+### List View (`render_list`)
+
+Used by all `list` commands. Renders a table with stable column order.
+Columns are defined once per command in a `list[ColumnSpec]` and reused
+for all formats, guaranteeing identical field names and ordering.
+
+### Error Messages (`render_error`)
+
+All errors use `render_error(label, message, recovery=...)`.
+
+For `json` / `yaml` formats, errors are wrapped in a unified envelope:
+
+```json
+{
+  "error": {
+    "code": "NotFound",
+    "message": "Resource not found",
+    "recovery": "Run 'agents resource add' first"
+  }
+}
+```
+
+### Success Messages (`render_success`)
+
+Green checkmark for rich, `OK: ...` for plain.
+
+### Warning Messages (`render_warning`)
+
+Yellow text for rich, `WARNING: ...` for plain.
+
+### Empty Results (`render_empty`)
+
+Consistent "No <entity> found." with optional recovery hint.
+
+## Stable Field Names
+
+JSON and YAML outputs use the same field names as the domain model's
+`_*_spec_dict()` helpers.  Column ordering in list views matches the
+`ColumnSpec` declarations.
+
+## JSON/YAML Error Envelope
+
+All error output in JSON/YAML format uses:
+
+```json
+{
+  "error": {
+    "code": "<label>",
+    "message": "<description>",
+    "details": {},
+    "recovery": "<hint>"
+  }
+}
+```
+
+## ASCII Compatibility
+
+The `plain` format guarantees ASCII-only output for log pipeline
+compatibility. Non-ASCII characters are replaced with `?`.
+
+## Migration Status
+
+All command modules now route output through the shared `renderers.py`
+layer via `get_console()` instead of creating module-level `Console()`
+instances.
+
+| Module                | Shared Renderers | `get_console()` | Notes                                    |
+|-----------------------|------------------|-----------------|------------------------------------------|
+| `action.py`           | Full             | Yes             | list/show/archive                        |
+| `actor.py`            | Full             | Yes             | list/show                                |
+| `config.py`           | Partial          | Yes             | render_error/render_empty; Rich panels   |
+| `plan.py`             | Partial          | Yes (+ local)   | status/errors/lifecycle; streaming uses local Console for Live/Progress |
+| `project.py`          | Partial          | Yes             | render_error/render_empty/render_success; Rich panels |
+| `project_context.py`  | Partial          | Yes             | render_error/render_success; Rich panels |
+| `resource.py`         | Partial          | Yes             | render_error/render_empty/render_success; Rich panels |
+| `session.py`          | Partial          | Yes             | render_error/render_empty/render_success/render_warning |
+| `skill.py`            | Partial          | Yes             | render_error/render_empty/render_success; Rich panels |
+| `tool.py`             | Full             | Yes             | render_list/render_error/render_success/render_warning/render_empty |
+| `automation_profile.py` | None           | No              | Legacy; see issue #813                   |
+| `context.py`          | None             | No              | Legacy; see issue #813                   |
+| `lsp.py`              | None             | No              | Legacy; see issue #813                   |
+| `invariant.py`        | None             | No              | Legacy; see issue #813                   |
+| `validation.py`       | None             | No              | Legacy; see issue #813                   |

features/action_cli_edge_cases_coverage.feature

@@ -11,7 +11,7 @@ Feature: Action CLI edge cases coverage
   Scenario: Print action with no arguments shows none indicator
     Given an action edge case action with no arguments
     When I call action edge case print action
-    Then the action edge case output should contain "(none)"
+    Then the action edge case output should contain "[]"
 
   Scenario: Print action with arguments shows formatted list
     Given an action edge case action with multiple arguments

features/cli_output_formats.feature

@@ -111,3 +111,168 @@ Feature: CLI output formats parity
   Scenario: Serialize value handles enums and nested dicts
     When I call serialize_value with enum and nested data
     Then the serialized result should have string enum values
+
+  # -- Renderer consolidation tests (issue #210) --
+
+  Scenario: render_detail produces valid JSON for non-rich format
+    When I call render_detail with a sample dict and format json
+    Then the captured output should be valid JSON
+    And the captured JSON should contain field "name"
+
+  Scenario: render_detail produces valid YAML for non-rich format
+    When I call render_detail with a sample dict and format yaml
+    Then the captured output should be valid YAML
+    And the captured YAML should contain field "name"
+
+  Scenario: render_detail produces plain key-value for plain format
+    When I call render_detail with a sample dict and format plain
+    Then the captured plain output should contain "name:"
+
+  Scenario: render_list produces valid JSON with stable column order
+    When I call render_list with sample rows columns and format json
+    Then the captured output should be valid JSON list
+    And each JSON item should have keys in column order
+
+  Scenario: render_list produces valid YAML with stable column order
+    When I call render_list with sample rows columns and format yaml
+    Then the captured output should be valid YAML list
+
+  Scenario: render_list produces rich table for rich format
+    When I call render_list with sample rows columns and format rich
+    Then the captured output should contain a table header
+
+  Scenario: render_error writes to stderr with label and message
+    When I call render_error with label "NotFound" and message "missing item"
+    Then the stderr should contain "NotFound"
+    And the stderr should contain "missing item"
+
+  Scenario: render_error produces JSON envelope for json format
+    When I call render_error with label "NotFound" and message "missing" and format json
+    Then the stderr should be valid JSON
+    And the JSON should contain nested key "error.code"
+
+  Scenario: render_success produces green checkmark for rich format
+    When I call render_success with message "Done" and format rich
+    Then the captured output should contain "Done"
+
+  Scenario: render_success produces ASCII OK for plain format
+    When I call render_success with message "Done" and format plain
+    Then the captured output should contain "OK: Done"
+
+  Scenario: render_empty produces yellow message for rich format
+    When I call render_empty with entity "actions" and format rich
+    Then the captured output should contain "No actions found"
+
+  Scenario: render_empty produces JSON empty array for json format
+    When I call render_empty with entity "actions" and format json
+    Then the captured output should be valid JSON
+    And the JSON should be an empty list
+
+  Scenario: render_warning produces yellow message for rich format
+    When I call render_warning with message "Check config" and format rich
+    Then the captured output should contain "Check config"
+
+  Scenario: render_warning produces JSON for json format
+    When I call render_warning with message "Check config" and format json
+    Then the captured output should be valid JSON
+    And the captured JSON should contain field "status"
+
+  Scenario: render_warning produces YAML for yaml format
+    When I call render_warning with message "Check config" and format yaml
+    Then the captured output should be valid YAML
+    And the captured YAML should contain field "status"
+
+  Scenario: render_warning produces ASCII for plain format
+    When I call render_warning with message "Check config" and format plain
+    Then the captured plain output should contain "WARNING: Check config"
+
+  Scenario: render_success produces JSON for json format without data
+    When I call render_success with message "Done" and format json
+    Then the captured output should be valid JSON
+    And the captured JSON should contain field "status"
+
+  Scenario: Color format produces plain-like output for format_output
+    When I call format_output with a dict and format color
+    Then the format result should contain plain key-value pairs
+
+  Scenario: Plain format output is ASCII-only
+    When I call render_detail with unicode data and format plain
+    Then the captured plain output should be pure ASCII
+
+  Scenario: FORMAT_HELP constant is consistent
+    Then the FORMAT_HELP string should mention all six formats
+
+  # -- PR #787 regression tests (P2-5, P2-6, P2-7) --
+
+  Scenario: render_detail FieldSpec accessor returning zero displays "0" not blank
+    When I call render_detail with a FieldSpec accessor returning 0 and format rich
+    Then the captured output should contain "0"
+
+  Scenario: render_detail FieldSpec accessor returning False displays "False"
+    When I call render_detail with a FieldSpec accessor returning False and format rich
+    Then the captured output should contain "False"
+
+  Scenario: render_error plain format contains label and message without Rich markup
+    When I call render_error with label "Test Error" and message "Something failed" and format plain
+    Then the stderr should contain "Test Error"
+    And the stderr should contain "Something failed"
+    And the stderr should not contain Rich markup
+
+  Scenario: render_detail with FieldSpec fields and rich format produces output
+    When I call render_detail with FieldSpec fields and format rich
+    Then the captured output should contain "Item Detail"
+    And the captured output should contain "test-item"
+
+  # -- PR #787 additional coverage (T1-T9, M10-M12) --
+
+  Scenario: render_error with recovery and details produces JSON envelope with details
+    When I call render_error with details and recovery and format json
+    Then the stderr should be valid JSON
+    And the JSON should contain nested key "error.details"
+    And the JSON should contain nested key "error.recovery"
+
+  Scenario: render_error with empty details emits empty details object
+    When I call render_error with no details and format json
+    Then the stderr should be valid JSON
+    And the JSON should contain nested key "error.details"
+
+  Scenario: render_list with plain format outputs key-value pairs
+    When I call render_list with sample rows columns and format plain
+    Then the captured plain output should contain "name:"
+
+  Scenario: render_list with table format outputs table
+    When I call render_list with sample rows columns and format table
+    Then the captured plain output should contain "name"
+
+  Scenario: render_empty with plain format shows message
+    When I call render_empty with entity "items" and format plain
+    Then the captured plain output should contain "No items found"
+
+  Scenario: render_empty with yaml format shows empty list
+    When I call render_empty with entity "items" and format yaml
+    Then the captured output should be valid YAML
+
+  Scenario: render_empty with table format shows message
+    When I call render_empty with entity "items" and format table
+    Then the captured plain output should contain "No items found"
+
+  Scenario: render_success with yaml format produces structured output
+    When I call render_success with message "Done" and format yaml
+    Then the captured output should be valid YAML
+    And the captured YAML should contain field "status"
+
+  Scenario: render_detail with None accessor value shows empty string
+    When I call render_detail with a FieldSpec accessor returning None and format rich
+    Then the captured output should contain "Name"
+
+  Scenario: render_list with empty dict value shows empty cell
+    When I call render_list with None cell values and format rich
+    Then the captured output should contain "Results"
+
+  Scenario: render_warning with color format shows warning text
+    When I call render_warning with message "Heads up" and format color
+    Then the captured output should contain "Heads up"
+
+  Scenario: render_detail with color format produces output
+    When I call render_detail with a sample dict and format color
+    Then the captured plain output should contain "name:"

features/lsp_cli_new_coverage.feature

@@ -71,7 +71,7 @@ Feature: LSP CLI commands coverage
     And the LSP server has been registered via CLI
     When I run lsp CLI remove with name "local/test-lsp" and --yes
     Then the lsp CLI command should succeed
-    And the lsp CLI output should contain "LSP Server Removed"
+    And the lsp CLI output should contain "LSP server removed"
 
   @lsp_cli_remove @error_handling
   Scenario: Remove an unknown LSP server returns not found error
@@ -97,7 +97,7 @@ Feature: LSP CLI commands coverage
     When I run lsp CLI list
     Then the lsp CLI command should succeed
     And the lsp CLI output should contain "LSP Servers"
-    And the lsp CLI output should contain "listed"
+    And the lsp CLI output should contain "2 total"
 
   @lsp_cli_list
   Scenario: List LSP servers when no servers are registered

features/skill_cli.feature

@@ -164,7 +164,7 @@ Feature: Skill CLI commands
     When I run skill CLI remove "local/file-reader" with --yes
     Then the skill CLI remove should succeed
     And the skill CLI output should contain "Skill Removed"
-    And the skill CLI output should contain "✓ OK"
+    And the skill CLI output should contain "Skill removed"
 
   Scenario: Remove nonexistent skill fails
     When I run skill CLI remove "local/nonexistent" with --yes

features/skill_cli_coverage_r2.feature

@@ -168,7 +168,7 @@ Feature: Skill CLI branch coverage round 2
     When r2skill- I invoke remove "local/to-remove" with --yes in rich format
     Then r2skill- the CLI exit code should be 0
     And r2skill- the output should contain "Skill Removed"
-    And r2skill- the output should contain "OK"
+    And r2skill- the output should contain "Skill removed"
 
   # -------------------------------------------------------------------
   # remove — non-rich format (line 564 true)

features/steps/action_cli_edge_cases_coverage_steps.py

@@ -149,8 +149,11 @@ def step_call_edge_print_action(context: Context) -> None:
     buf = StringIO()
     test_console = Console(file=buf, force_terminal=False, width=200)
 
-    # Temporarily replace the module-level console so _print_action writes to buf
-    with patch("cleveragents.cli.commands.action.console", test_console):
+    # Temporarily replace the renderers console so _print_action writes to buf
+    with (
+        patch("cleveragents.cli.renderers._console", test_console),
+        patch("cleveragents.cli.renderers._err_console", test_console),
+    ):
         _print_action(context.edge_action)
 
     context.print_output = buf.getvalue()

features/steps/action_cli_spec_alignment_steps.py

@@ -106,7 +106,7 @@ def step_spec_service(context: Context) -> None:
         context._cleanup_handlers = []
     context._cleanup_handlers.append(context.service_patcher.stop)
 
-    # Patch the module-level Rich Console so it never injects ANSI escape
+    # Patch the renderers Console so it never injects ANSI escape
     # codes.  Without this, Rich detects a colour terminal and embeds
     # escape sequences in the CLI output, causing plain-text assertions
     # (e.g. ``"Action Created" in output``) to fail on real terminals
@@ -115,10 +115,15 @@ def step_spec_service(context: Context) -> None:
 
     _plain_console = _Console(no_color=True, highlight=False, width=300)
     context.console_patcher = patch(
-        "cleveragents.cli.commands.action.console", _plain_console
+        "cleveragents.cli.renderers._console", _plain_console
     )
     context.console_patcher.start()
     context._cleanup_handlers.append(context.console_patcher.stop)
+    context.err_console_patcher = patch(
+        "cleveragents.cli.renderers._err_console", _plain_console
+    )
+    context.err_console_patcher.start()
+    context._cleanup_handlers.append(context.err_console_patcher.stop)
 
 
 # ---------------------------------------------------------------------------

features/steps/actor_cli_coverage_boost_steps.py

@@ -211,7 +211,6 @@ def step_set_default_aborted(context):
 
 @when("I print that actor in json format")
 def step_print_actor_json(context):
-    import contextlib
     from io import StringIO
 
     from rich.console import Console
@@ -220,8 +219,8 @@ def step_print_actor_json(context):
     test_console = Console(file=buf, force_terminal=False)
 
     with (
-        patch("cleveragents.cli.commands.actor.console", test_console),
-        contextlib.redirect_stdout(buf),
+        patch("cleveragents.cli.renderers._console", test_console),
+        patch("cleveragents.cli.renderers._err_console", test_console),
     ):
         _print_actor(context.test_actor, title="Test", fmt="json")
 

features/steps/cli_output_formats_steps.py

@@ -93,7 +93,7 @@ def step_mocked_lifecycle_for_formats(context: Context) -> None:
     context.action_patcher.start()
     context.plan_patcher.start()
 
-    # Patch module-level Rich Console objects so they never inject ANSI
+    # Patch the shared renderer consoles so they never inject ANSI
     # escape codes into structured output (JSON/YAML/plain).  Without this,
     # console.print(json_string) adds syntax-highlighting escapes when Rich
     # detects a real terminal, which makes json.loads() / yaml.safe_load()
@@ -101,12 +101,21 @@ def step_mocked_lifecycle_for_formats(context: Context) -> None:
     # headless container where Console auto-disables colour.
     from rich.console import Console as _Console
 
+    import cleveragents.cli.renderers as _renderers_mod
+
     _plain_console = _Console(no_color=True, highlight=False)
-    context.action_console_patcher = patch.object(
-        _action_mod, "console", _plain_console
+    # Patch both the renderers module consoles and any remaining
+    # module-level consoles in plan.py (which still has one for complex
+    # streaming/Live displays).
+    context.renderer_console_patcher = patch.object(
+        _renderers_mod, "_console", _plain_console
+    )
+    context.renderer_err_console_patcher = patch.object(
+        _renderers_mod, "_err_console", _plain_console
     )
     context.plan_console_patcher = patch.object(_plan_mod, "console", _plain_console)
-    context.action_console_patcher.start()
+    context.renderer_console_patcher.start()
+    context.renderer_err_console_patcher.start()
     context.plan_console_patcher.start()
 
     # Patch _notify_facade so that the A2A facade bootstrap does not
@@ -122,7 +131,8 @@ def step_mocked_lifecycle_for_formats(context: Context) -> None:
         context._cleanup_handlers = []
     context._cleanup_handlers.append(context.action_patcher.stop)
     context._cleanup_handlers.append(context.plan_patcher.stop)
-    context._cleanup_handlers.append(context.action_console_patcher.stop)
+    context._cleanup_handlers.append(context.renderer_console_patcher.stop)
+    context._cleanup_handlers.append(context.renderer_err_console_patcher.stop)
     context._cleanup_handlers.append(context.plan_console_patcher.stop)
     context._cleanup_handlers.append(context.plan_facade_patcher.stop)
 
@@ -410,6 +420,16 @@ def step_call_format_rich(context: Context) -> None:
     context.format_result = _capture_format_output(data, "rich")
 
 
+@when("I call format_output with a dict and format color")
+def step_call_format_color(context: Context) -> None:
+    data = {
+        "name": "test",
+        "nested": {"a": 1, "b": 2},
+        "items": ["x", "y"],
+    }
+    context.format_result = format_output(data, "color")
+
+
 @when("I call format_output with a list and format plain")
 def step_call_format_list_plain(context: Context) -> None:
     data = [
@@ -470,3 +490,483 @@ def step_serialized_enum_values(context: Context) -> None:
     assert isinstance(context.serialized["nested"], dict)
     assert context.serialized["items"][0] == "archived"
     assert isinstance(context.serialized["ts"], str)
+
+
+# -----------------------------------------------------------------------
+# Renderer consolidation steps (issue #210)
+# -----------------------------------------------------------------------
+
+import io  # noqa: E402
+
+from rich.console import Console  # noqa: E402
+
+import cleveragents.cli.renderers as _renderers_mod  # noqa: E402
+from cleveragents.cli.renderers import (  # noqa: E402
+    FORMAT_HELP,
+    ColumnSpec,
+    FieldSpec,
+    render_detail,
+    render_empty,
+    render_error,
+    render_list,
+    render_success,
+    render_warning,
+)
+
+_SAMPLE_DETAIL = {"name": "test-item", "count": 42, "active": True}
+
+_SAMPLE_ROWS = [
+    {"name": "alpha", "value": 1},
+    {"name": "bravo", "value": 2},
+]
+
+_SAMPLE_COLUMNS = [
+    ColumnSpec(key="name", label="Name"),
+    ColumnSpec(key="value", label="Value"),
+]
+
+
+def _capture_renderer(func, *args, **kwargs):
+    """Call a renderer function while capturing stdout console output."""
+    buf = io.StringIO()
+    plain_console = Console(file=buf, no_color=True, highlight=False)
+    original = _renderers_mod._console
+    _renderers_mod._console = plain_console
+    try:
+        func(*args, **kwargs)
+    finally:
+        _renderers_mod._console = original
+    return buf.getvalue()
+
+
+def _capture_renderer_stderr(func, *args, **kwargs):
+    """Call a renderer function while capturing stderr console output."""
+    buf = io.StringIO()
+    plain_console = Console(file=buf, no_color=True, highlight=False)
+    original = _renderers_mod._err_console
+    _renderers_mod._err_console = plain_console
+    try:
+        func(*args, **kwargs)
+    finally:
+        _renderers_mod._err_console = original
+    return buf.getvalue()
+
+
+# ------- When steps: render_detail -------
+
+
+@when("I call render_detail with a sample dict and format json")
+def step_render_detail_json(context: Context) -> None:
+    context.captured_output = _capture_renderer(
+        render_detail, _SAMPLE_DETAIL, fmt="json"
+    )
+
+
+@when("I call render_detail with a sample dict and format yaml")
+def step_render_detail_yaml(context: Context) -> None:
+    context.captured_output = _capture_renderer(
+        render_detail, _SAMPLE_DETAIL, fmt="yaml"
+    )
+
+
+@when("I call render_detail with a sample dict and format plain")
+def step_render_detail_plain(context: Context) -> None:
+    context.captured_output = _capture_renderer(
+        render_detail, _SAMPLE_DETAIL, fmt="plain"
+    )
+
+
+@when("I call render_detail with unicode data and format plain")
+def step_render_detail_unicode_plain(context: Context) -> None:
+    data = {"name": "caf\u00e9-\u2603", "desc": "\u2714 done"}
+    context.captured_output = _capture_renderer(render_detail, data, fmt="plain")
+
+
+# ------- When steps: render_list -------
+
+
+@when("I call render_list with sample rows columns and format json")
+def step_render_list_json(context: Context) -> None:
+    context.captured_output = _capture_renderer(
+        render_list, _SAMPLE_ROWS, columns=_SAMPLE_COLUMNS, fmt="json"
+    )
+
+
+@when("I call render_list with sample rows columns and format yaml")
+def step_render_list_yaml(context: Context) -> None:
+    context.captured_output = _capture_renderer(
+        render_list, _SAMPLE_ROWS, columns=_SAMPLE_COLUMNS, fmt="yaml"
+    )
+
+
+@when("I call render_list with sample rows columns and format rich")
+def step_render_list_rich(context: Context) -> None:
+    context.captured_output = _capture_renderer(
+        render_list, _SAMPLE_ROWS, columns=_SAMPLE_COLUMNS, fmt="rich"
+    )
+
+
+# ------- When steps: render_error -------
+
+
+@when('I call render_error with label "{label}" and message "{message}"')
+def step_render_error_rich(context: Context, label: str, message: str) -> None:
+    context.captured_stderr = _capture_renderer_stderr(
+        render_error, label, message, fmt="rich"
+    )
+
+
+@when(
+    'I call render_error with label "{label}" and message "{message}" and format json'
+)
+def step_render_error_json(context: Context, label: str, message: str) -> None:
+    context.captured_stderr = _capture_renderer_stderr(
+        render_error, label, message, fmt="json"
+    )
+
+
+# ------- When steps: render_success -------
+
+
+@when('I call render_success with message "{message}" and format rich')
+def step_render_success_rich(context: Context, message: str) -> None:
+    context.captured_output = _capture_renderer(render_success, message, fmt="rich")
+
+
+@when('I call render_success with message "{message}" and format plain')
+def step_render_success_plain(context: Context, message: str) -> None:
+    context.captured_output = _capture_renderer(render_success, message, fmt="plain")
+
+
+@when('I call render_success with message "{message}" and format json')
+def step_render_success_json(context: Context, message: str) -> None:
+    context.captured_output = _capture_renderer(render_success, message, fmt="json")
+
+
+# ------- When steps: render_warning -------
+
+
+@when('I call render_warning with message "{message}" and format rich')
+def step_render_warning_rich(context: Context, message: str) -> None:
+    context.captured_output = _capture_renderer(render_warning, message, fmt="rich")
+
+
+@when('I call render_warning with message "{message}" and format json')
+def step_render_warning_json(context: Context, message: str) -> None:
+    context.captured_output = _capture_renderer(render_warning, message, fmt="json")
+
+
+@when('I call render_warning with message "{message}" and format yaml')
+def step_render_warning_yaml(context: Context, message: str) -> None:
+    context.captured_output = _capture_renderer(render_warning, message, fmt="yaml")
+
+
+@when('I call render_warning with message "{message}" and format plain')
+def step_render_warning_plain(context: Context, message: str) -> None:
+    context.captured_output = _capture_renderer(render_warning, message, fmt="plain")
+
+
+# ------- When steps: render_empty -------
+
+
+@when('I call render_empty with entity "{entity}" and format rich')
+def step_render_empty_rich(context: Context, entity: str) -> None:
+    context.captured_output = _capture_renderer(render_empty, entity, fmt="rich")
+
+
+@when('I call render_empty with entity "{entity}" and format json')
+def step_render_empty_json(context: Context, entity: str) -> None:
+    context.captured_output = _capture_renderer(render_empty, entity, fmt="json")
+
+
+# ------- Then steps: renderer output assertions -------
+
+
+@then("the captured output should be valid JSON")
+def step_captured_valid_json(context: Context) -> None:
+    parsed = json.loads(context.captured_output.strip())
+    context.captured_parsed = parsed
+    assert parsed is not None
+
+
+@then("the captured output should be valid YAML")
+def step_captured_valid_yaml(context: Context) -> None:
+    parsed = yaml.safe_load(context.captured_output.strip())
+    context.captured_parsed = parsed
+    assert parsed is not None
+
+
+@then('the captured plain output should contain "{text}"')
+def step_captured_plain_contains(context: Context, text: str) -> None:
+    assert text in context.captured_output, (
+        f"'{text}' not in captured output: {context.captured_output[:300]}"
+    )
+
+
+@then('the captured JSON should contain field "{key}"')
+def step_captured_json_contains_field(context: Context, key: str) -> None:
+    parsed = json.loads(context.captured_output.strip())
+    if isinstance(parsed, list):
+        assert key in parsed[0], f"Key '{key}' not in {list(parsed[0].keys())}"
+    else:
+        assert key in parsed, f"Key '{key}' not in {list(parsed.keys())}"
+
+
+@then('the captured YAML should contain field "{key}"')
+def step_captured_yaml_contains_field(context: Context, key: str) -> None:
+    parsed = yaml.safe_load(context.captured_output.strip())
+    if isinstance(parsed, list):
+        assert key in parsed[0], f"Key '{key}' not in {list(parsed[0].keys())}"
+    else:
+        assert key in parsed, f"Key '{key}' not in {list(parsed.keys())}"
+
+
+@then("the captured output should be valid JSON list")
+def step_captured_valid_json_list(context: Context) -> None:
+    parsed = json.loads(context.captured_output.strip())
+    context.captured_parsed = parsed
+    assert isinstance(parsed, list), f"Expected list, got {type(parsed)}"
+
+
+@then("each JSON item should have keys in column order")
+def step_json_keys_column_order(context: Context) -> None:
+    expected_order = [c.key for c in _SAMPLE_COLUMNS]
+    for item in context.captured_parsed:
+        keys = list(item.keys())
+        assert keys == expected_order, f"Expected {expected_order}, got {keys}"
+
+
+@then("the captured output should be valid YAML list")
+def step_captured_valid_yaml_list(context: Context) -> None:
+    parsed = yaml.safe_load(context.captured_output.strip())
+    context.captured_parsed = parsed
+    assert isinstance(parsed, list), f"Expected list, got {type(parsed)}"
+
+
+@then("the captured output should contain a table header")
+def step_captured_contains_table_header(context: Context) -> None:
+    # Rich tables include column headers; check for one of our column labels
+    assert "Name" in context.captured_output, (
+        f"'Name' not in captured output: {context.captured_output[:300]}"
+    )
+
+
+@then('the stderr should contain "{text}"')
+def step_stderr_contains(context: Context, text: str) -> None:
+    assert text in context.captured_stderr, (
+        f"'{text}' not in stderr: {context.captured_stderr[:300]}"
+    )
+
+
+@then("the stderr should be valid JSON")
+def step_stderr_valid_json(context: Context) -> None:
+    parsed = json.loads(context.captured_stderr.strip())
+    context.captured_parsed = parsed
+    assert parsed is not None
+
+
+@then('the JSON should contain nested key "{dotted_key}"')
+def step_json_nested_key(context: Context, dotted_key: str) -> None:
+    parts = dotted_key.split(".")
+    obj = context.captured_parsed
+    for part in parts:
+        assert isinstance(obj, dict), f"Expected dict at '{part}', got {type(obj)}"
+        assert part in obj, f"Key '{part}' not in {list(obj.keys())}"
+        obj = obj[part]
+
+
+@then('the captured output should contain "{text}"')
+def step_captured_output_contains(context: Context, text: str) -> None:
+    assert text in context.captured_output, (
+        f"'{text}' not in captured output: {context.captured_output[:300]}"
+    )
+
+
+@then("the JSON should be an empty list")
+def step_json_empty_list(context: Context) -> None:
+    assert context.captured_parsed == [], (
+        f"Expected empty list, got {context.captured_parsed}"
+    )
+
+
+@then("the captured plain output should be pure ASCII")
+def step_captured_plain_ascii(context: Context) -> None:
+    output = context.captured_output
+    for i, ch in enumerate(output):
+        assert ord(ch) < 128, (
+            f"Non-ASCII char U+{ord(ch):04X} at position {i}: ...{output[max(0, i - 5) : i + 5]}..."
+        )
+
+
+@then("the FORMAT_HELP string should mention all six formats")
+def step_format_help_mentions_formats(context: Context) -> None:
+    for fmt_name in ("json", "yaml", "plain", "table", "rich", "color"):
+        assert fmt_name in FORMAT_HELP, (
+            f"'{fmt_name}' not found in FORMAT_HELP: {FORMAT_HELP}"
+        )
+
+
+# -----------------------------------------------------------------------
+# PR #787 regression tests (P2-5, P2-6, P2-7)
+# -----------------------------------------------------------------------
+
+
+# ------- P2-5: FieldSpec accessor None/falsy coercion -------
+
+
+@when("I call render_detail with a FieldSpec accessor returning 0 and format rich")
+def step_render_detail_accessor_zero(context: Context) -> None:
+    data = {"name": "test-item", "metric": 0}
+    # The accessor intentionally returns int 0 to verify that
+    # render_detail coerces falsy-but-not-None values via str().
+    fields = [
+        FieldSpec(key="name", label="Name"),
+        FieldSpec(key="metric", label="Metric", accessor=lambda d: d.get("metric")),  # type: ignore[arg-type]
+    ]
+    context.captured_output = _capture_renderer(
+        render_detail, data, title="Zero Test", fields=fields, fmt="rich"
+    )
+
+
+@when("I call render_detail with a FieldSpec accessor returning False and format rich")
+def step_render_detail_accessor_false(context: Context) -> None:
+    data = {"name": "test-item", "flag": False}
+    fields = [
+        FieldSpec(key="name", label="Name"),
+        FieldSpec(key="flag", label="Flag", accessor=lambda d: d.get("flag")),  # type: ignore[arg-type]
+    ]
+    context.captured_output = _capture_renderer(
+        render_detail, data, title="Falsy Field Test", fields=fields, fmt="rich"
+    )
+
+
+# ------- P2-6: render_error plain format without Rich markup -------
+
+
+@when(
+    'I call render_error with label "{label}" and message "{message}" and format plain'
+)
+def step_render_error_plain(context: Context, label: str, message: str) -> None:
+    context.captured_stderr = _capture_renderer_stderr(
+        render_error, label, message, fmt="plain"
+    )
+
+
+@then("the stderr should not contain Rich markup")
+def step_stderr_no_rich_markup(context: Context) -> None:
+    for tag in ("[red]", "[/red]", "[bold]", "[/bold]", "[dim]", "[/dim]"):
+        assert tag not in context.captured_stderr, (
+            f"Rich markup '{tag}' found in stderr: {context.captured_stderr[:300]}"
+        )
+
+
+# ------- P2-7: render_detail with FieldSpec and rich format -------
+
+
+@when("I call render_detail with FieldSpec fields and format rich")
+def step_render_detail_fieldspec_rich(context: Context) -> None:
+    data = {"name": "test-item", "count": 42, "active": True}
+    fields = [
+        FieldSpec(key="name", label="Name"),
+        FieldSpec(key="count", label="Count"),
+        FieldSpec(key="active", label="Active"),
+    ]
+    context.captured_output = _capture_renderer(
+        render_detail, data, title="Item Detail", fields=fields, fmt="rich"
+    )
+
+
+# -----------------------------------------------------------------------
+# PR #787 additional coverage (T1-T9, M10-M12)
+# -----------------------------------------------------------------------
+
+
+@when("I call render_error with details and recovery and format json")
+def step_render_error_details_json(context: Context) -> None:
+    context.captured_stderr = _capture_renderer_stderr(
+        render_error,
+        "TestError",
+        "Something went wrong",
+        recovery="Try again later",
+        details={"field": "value"},
+        fmt="json",
+    )
+
+
+@when("I call render_error with no details and format json")
+def step_render_error_no_details_json(context: Context) -> None:
+    context.captured_stderr = _capture_renderer_stderr(
+        render_error,
+        "TestError",
+        "Something went wrong",
+        fmt="json",
+    )
+
+
+@when("I call render_list with sample rows columns and format plain")
+def step_render_list_plain(context: Context) -> None:
+    context.captured_output = _capture_renderer(
+        render_list, _SAMPLE_ROWS, columns=_SAMPLE_COLUMNS, fmt="plain"
+    )
+
+
+@when("I call render_list with sample rows columns and format table")
+def step_render_list_table(context: Context) -> None:
+    context.captured_output = _capture_renderer(
+        render_list, _SAMPLE_ROWS, columns=_SAMPLE_COLUMNS, fmt="table"
+    )
+
+
+@when('I call render_empty with entity "{entity}" and format plain')
+def step_render_empty_plain(context: Context, entity: str) -> None:
+    context.captured_output = _capture_renderer(render_empty, entity, fmt="plain")
+
+
+@when('I call render_empty with entity "{entity}" and format yaml')
+def step_render_empty_yaml(context: Context, entity: str) -> None:
+    context.captured_output = _capture_renderer(render_empty, entity, fmt="yaml")
+
+
+@when('I call render_empty with entity "{entity}" and format table')
+def step_render_empty_table(context: Context, entity: str) -> None:
+    context.captured_output = _capture_renderer(render_empty, entity, fmt="table")
+
+
+@when('I call render_success with message "{message}" and format yaml')
+def step_render_success_yaml(context: Context, message: str) -> None:
+    context.captured_output = _capture_renderer(render_success, message, fmt="yaml")
+
+
+@when("I call render_detail with a FieldSpec accessor returning None and format rich")
+def step_render_detail_accessor_none(context: Context) -> None:
+    data = {"name": "test-item", "metric": None}
+    fields = [
+        FieldSpec(key="name", label="Name"),
+        FieldSpec(key="metric", label="Metric", accessor=lambda d: d.get("metric")),  # type: ignore[arg-type]
+    ]
+    context.captured_output = _capture_renderer(
+        render_detail, data, title="None Test", fields=fields, fmt="rich"
+    )
+
+
+@when("I call render_list with None cell values and format rich")
+def step_render_list_none_cells(context: Context) -> None:
+    rows = [
+        {"name": "alpha", "value": None},
+        {"name": None, "value": 2},
+    ]
+    context.captured_output = _capture_renderer(
+        render_list, rows, columns=_SAMPLE_COLUMNS, title="Results", fmt="rich"
+    )
+
+
+@when('I call render_warning with message "{message}" and format color')
+def step_render_warning_color(context: Context, message: str) -> None:
+    context.captured_output = _capture_renderer(render_warning, message, fmt="color")
+
+
+@when("I call render_detail with a sample dict and format color")
+def step_render_detail_color(context: Context) -> None:
+    context.captured_output = _capture_renderer(
+        render_detail, _SAMPLE_DETAIL, fmt="color"
+    )

features/steps/config_cli_safety_net_coverage_steps.py

@@ -705,8 +705,10 @@ def step_sn_patch_console(context: Context) -> None:
     context._sn_test_console = Console(
         file=context._sn_console_buf, no_color=True, width=200
     )
+    import cleveragents.cli.renderers as _renderers_mod
+
     context._sn_console_patch = patch.object(
-        config_mod, "console", context._sn_test_console
+        _renderers_mod, "_console", context._sn_test_console
     )
     context._sn_console_patch.start()
     context._cleanup_handlers.append(context._sn_console_patch.stop)

features/steps/context_cli_wiring_steps.py

@@ -160,11 +160,11 @@ def _run_wired(context: Any, func: Any, *args: Any, **kwargs: Any) -> None:
             return_value=mock_cont,
         ),
         patch(
-            "cleveragents.cli.commands.project_context.console",
+            "cleveragents.cli.renderers._console",
             test_console,
         ),
         patch(
-            "cleveragents.cli.commands.project_context.err_console",
+            "cleveragents.cli.renderers._err_console",
             test_console,
         ),
         patch(

features/steps/coverage_boost_extra_steps.py

@@ -278,10 +278,12 @@ def _resource_capture_output(func: Any, *args: Any, **kwargs: Any) -> tuple[str,
     buf = StringIO()
     console = Console(file=buf, width=200, no_color=True)
 
-    import cleveragents.cli.commands.resource as resource_mod
+    import cleveragents.cli.renderers as _renderers_mod
 
-    orig_console = resource_mod.console
-    resource_mod.console = console
+    orig_r_console = _renderers_mod._console
+    orig_r_err = _renderers_mod._err_console
+    _renderers_mod._console = console
+    _renderers_mod._err_console = console
 
     failed = False
     try:
@@ -292,7 +294,8 @@ def _resource_capture_output(func: Any, *args: Any, **kwargs: Any) -> tuple[str,
     except Exception:
         failed = True
     finally:
-        resource_mod.console = orig_console
+        _renderers_mod._console = orig_r_console
+        _renderers_mod._err_console = orig_r_err
 
     return buf.getvalue(), failed
 

features/steps/coverage_security_template_boost_steps.py

@@ -557,9 +557,11 @@ def step_sess_create_err(context: Context) -> None:
 
 @when('I invoke session list with format "{fmt}"')
 def step_sess_list_fmt(context: Context, fmt: str) -> None:
+    import cleveragents.cli.renderers as _renderers_mod
     from cleveragents.cli.commands import session as session_mod
 
     buf = StringIO()
+    _capture_print = lambda x="", **kw: buf.write(str(x) + "\n")  # noqa: E731
     with (
         patch.object(
             session_mod,
@@ -567,9 +569,14 @@ def step_sess_list_fmt(context: Context, fmt: str) -> None:
             return_value=context.cov_mock_sess_svc,
         ),
         patch.object(
-            session_mod.console,
+            _renderers_mod.get_console(),
             "print",
-            side_effect=lambda x="", **kw: buf.write(str(x) + "\n"),
+            side_effect=_capture_print,
+        ),
+        patch.object(
+            _renderers_mod._get_err_console(),
+            "print",
+            side_effect=_capture_print,
         ),
     ):
         session_mod.list_sessions(fmt=fmt)
@@ -597,9 +604,11 @@ def step_sess_delete_no_confirm(context: Context) -> None:
 
 @when("I invoke session delete with yes flag")
 def step_sess_delete_yes(context: Context) -> None:
+    import cleveragents.cli.renderers as _renderers_mod
     from cleveragents.cli.commands import session as session_mod
 
     buf_lines: list[str] = []
+    _capture_print = lambda x="", **kw: buf_lines.append(str(x))  # noqa: E731
     with (
         patch.object(
             session_mod,
@@ -607,9 +616,14 @@ def step_sess_delete_yes(context: Context) -> None:
             return_value=context.cov_mock_sess_svc,
         ),
         patch.object(
-            session_mod.console,
+            _renderers_mod.get_console(),
             "print",
-            side_effect=lambda x="", **kw: buf_lines.append(str(x)),
+            side_effect=_capture_print,
+        ),
+        patch.object(
+            _renderers_mod._get_err_console(),
+            "print",
+            side_effect=_capture_print,
         ),
     ):
         session_mod.delete(

features/steps/error_recovery_coverage_boost_steps.py

@@ -192,7 +192,6 @@ def step_errcov_lifecycle_error(context: Context) -> None:
 
 @when('I invoke errcov plan errors in "{fmt}" format')
 def step_errcov_invoke_errors(context: Context, fmt: str) -> None:
-    import contextlib
 
     from cleveragents.cli.commands import plan as plan_mod
 
@@ -208,7 +207,8 @@ def step_errcov_invoke_errors(context: Context, fmt: str) -> None:
             return_value=svc,
         ),
         patch.object(plan_mod, "console", capture_console),
-        contextlib.redirect_stdout(buf),
+        patch("cleveragents.cli.renderers._console", capture_console),
+        patch("cleveragents.cli.renderers._err_console", capture_console),
     ):
         plan_mod.plan_errors(plan_id=_PLAN_ID, fmt=fmt)
     context.errcov_output = buf.getvalue()
@@ -227,6 +227,8 @@ def step_errcov_invoke_errors_abort(context: Context) -> None:
             return_value=context.errcov_lifecycle_svc,
         ),
         patch.object(plan_mod, "console", capture_console),
+        patch("cleveragents.cli.renderers._console", capture_console),
+        patch("cleveragents.cli.renderers._err_console", capture_console),
         contextlib.suppress(SystemExit, Exception),
     ):
         plan_mod.plan_errors(plan_id=_PLAN_ID, fmt="rich")
@@ -270,7 +272,8 @@ def step_errcov_invoke_diff(context: Context, fmt: str) -> None:
             plan_mod, "_get_apply_service", return_value=context.errcov_apply_svc
         ),
         patch.object(plan_mod, "console", capture_console),
-        contextlib.redirect_stdout(buf),
+        patch("cleveragents.cli.renderers._console", capture_console),
+        patch("cleveragents.cli.renderers._err_console", capture_console),
     ):
         plan_mod.plan_diff(plan_id=_PLAN_ID, fmt=fmt)
     context.errcov_output = buf.getvalue()
@@ -287,6 +290,8 @@ def step_errcov_invoke_diff_abort(context: Context) -> None:
             plan_mod, "_get_apply_service", return_value=context.errcov_apply_svc
         ),
         patch.object(plan_mod, "console", capture_console),
+        patch("cleveragents.cli.renderers._console", capture_console),
+        patch("cleveragents.cli.renderers._err_console", capture_console),
         contextlib.suppress(SystemExit, Exception),
     ):
         plan_mod.plan_diff(plan_id=_PLAN_ID, fmt="rich")
@@ -330,7 +335,8 @@ def step_errcov_invoke_artifacts(context: Context, fmt: str) -> None:
             plan_mod, "_get_apply_service", return_value=context.errcov_apply_svc
         ),
         patch.object(plan_mod, "console", capture_console),
-        contextlib.redirect_stdout(buf),
+        patch("cleveragents.cli.renderers._console", capture_console),
+        patch("cleveragents.cli.renderers._err_console", capture_console),
     ):
         plan_mod.plan_artifacts(plan_id=_PLAN_ID, fmt=fmt)
     context.errcov_output = buf.getvalue()
@@ -347,6 +353,8 @@ def step_errcov_invoke_artifacts_abort(context: Context) -> None:
             plan_mod, "_get_apply_service", return_value=context.errcov_apply_svc
         ),
         patch.object(plan_mod, "console", capture_console),
+        patch("cleveragents.cli.renderers._console", capture_console),
+        patch("cleveragents.cli.renderers._err_console", capture_console),
         contextlib.suppress(SystemExit, Exception),
     ):
         plan_mod.plan_artifacts(plan_id=_PLAN_ID, fmt="rich")

features/steps/project_cli_commands_steps.py

@@ -133,15 +133,15 @@ def _capture(context: Any, func: Any, *args: Any, **kwargs: Any) -> None:
 
     import typer
 
-    import cleveragents.cli.commands.project as project_mod
+    import cleveragents.cli.renderers as _renderers_mod
 
     buf = StringIO()
     fake_console = Console(file=buf, width=200, no_color=True)
 
-    orig_console = project_mod.console
-    orig_err = project_mod.err_console
-    project_mod.console = fake_console
-    project_mod.err_console = fake_console
+    orig_r_console = _renderers_mod._console
+    orig_r_err = _renderers_mod._err_console
+    _renderers_mod._console = fake_console
+    _renderers_mod._err_console = fake_console
 
     _patch_project_mod(context)
     failed = False
@@ -153,8 +153,8 @@ def _capture(context: Any, func: Any, *args: Any, **kwargs: Any) -> None:
     except Exception:
         failed = True
     finally:
-        project_mod.console = orig_console
-        project_mod.err_console = orig_err
+        _renderers_mod._console = orig_r_console
+        _renderers_mod._err_console = orig_r_err
         _unpatch_project_mod()
 
     context._cmd_output = buf.getvalue()

features/steps/project_context_cli_coverage_boost_steps.py

@@ -116,11 +116,11 @@ def _run(context: Any, func: Any, *args: Any, **kwargs: Any) -> None:
             return_value=mc,
         ),
         patch(
-            "cleveragents.cli.commands.project_context.console",
+            "cleveragents.cli.renderers._console",
             test_console,
         ),
         patch(
-            "cleveragents.cli.commands.project_context.err_console",
+            "cleveragents.cli.renderers._err_console",
             test_console,
         ),
         patch(

features/steps/project_context_cli_steps.py

@@ -160,11 +160,11 @@ def _run_with_container(context: Any, func: Any, *args: Any, **kwargs: Any) -> N
             return_value=mock_cont,
         ),
         patch(
-            "cleveragents.cli.commands.project_context.console",
+            "cleveragents.cli.renderers._console",
             test_console,
         ),
         patch(
-            "cleveragents.cli.commands.project_context.err_console",
+            "cleveragents.cli.renderers._err_console",
             test_console,
         ),
         patch(

features/steps/repositories_uncovered_lines_steps.py

@@ -1206,9 +1206,14 @@ def step_then_va_dup_propagate(context):
 )
 def step_given_ap_repo(context):
     engine, factory = _make_engine_and_session()
+    # Pin a single session for the lifetime of this scenario so that
+    # upsert (flush) and list_all (query) share the same connection,
+    # avoiding visibility issues with in-memory SQLite under heavy
+    # parallel load.
+    pinned = factory()
     context._ucl_ap_engine = engine
     context._ucl_ap_session_factory = factory
-    context._ucl_ap_repo = AutomationProfileRepository(factory)
+    context._ucl_ap_repo = AutomationProfileRepository(lambda: pinned)
 
 
 @when('a profile is fetched by name "{name}" for uncovered lines')

features/steps/resource_cli_coverage_boost_steps.py

@@ -30,7 +30,8 @@ from cleveragents.core.exceptions import ValidationError
 
 _PATCH_TARGET = "cleveragents.cli.commands.resource._get_registry_service"
 _PATCH_CONTAINER = "cleveragents.cli.commands.resource.get_container"
-_PATCH_CONSOLE = "cleveragents.cli.commands.resource.console"
+_PATCH_R_CONSOLE = "cleveragents.cli.renderers._console"
+_PATCH_R_ERR_CONSOLE = "cleveragents.cli.renderers._err_console"
 
 
 def _no_color_console():
@@ -146,7 +147,6 @@ def step_invoke_type_add_update(context: Context) -> None:
 
     with (
         patch(_PATCH_TARGET, return_value=context.rcb_mock_service),
-        patch(_PATCH_CONSOLE, _no_color_console()),
     ):
         context.rcb_result = runner.invoke(
             app,
@@ -178,7 +178,6 @@ def step_invoke_type_add_fnf(context: Context) -> None:
 
     with (
         patch(_PATCH_TARGET, return_value=context.rcb_mock_service),
-        patch(_PATCH_CONSOLE, _no_color_console()),
     ):
         context.rcb_result = runner.invoke(
             app,
@@ -205,7 +204,6 @@ def step_invoke_type_remove_no(context: Context) -> None:
     runner = CliRunner()
     with (
         patch(_PATCH_TARGET, return_value=context.rcb_mock_service),
-        patch(_PATCH_CONSOLE, _no_color_console()),
     ):
         context.rcb_result = runner.invoke(
             app,
@@ -239,7 +237,6 @@ def step_invoke_type_remove_not_found(context: Context) -> None:
     runner = CliRunner()
     with (
         patch(_PATCH_TARGET, return_value=context.rcb_mock_service),
-        patch(_PATCH_CONSOLE, _no_color_console()),
     ):
         context.rcb_result = runner.invoke(
             app,
@@ -270,7 +267,6 @@ def step_invoke_type_remove_validation_error(context: Context) -> None:
     runner = CliRunner()
     with (
         patch(_PATCH_TARGET, return_value=context.rcb_mock_service),
-        patch(_PATCH_CONSOLE, _no_color_console()),
     ):
         context.rcb_result = runner.invoke(
             app,
@@ -312,7 +308,6 @@ def step_invoke_resource_remove_edges(context: Context) -> None:
     runner = CliRunner()
     with (
         patch(_PATCH_TARGET, return_value=context.rcb_mock_service),
-        patch(_PATCH_CONSOLE, _no_color_console()),
     ):
         context.rcb_result = runner.invoke(
             app,
@@ -366,7 +361,6 @@ def step_invoke_resource_remove_exception(context: Context) -> None:
     runner = CliRunner()
     with (
         patch(_PATCH_TARGET, return_value=context.rcb_mock_service),
-        patch(_PATCH_CONSOLE, _no_color_console()),
     ):
         context.rcb_result = runner.invoke(
             app,

features/steps/resource_cli_steps.py

@@ -44,10 +44,12 @@ def _capture_output(func: Any, *args: Any, **kwargs: Any) -> tuple[str, bool]:
         file=buf, width=200, no_color=True, highlight=False, force_terminal=False
     )
 
-    import cleveragents.cli.commands.resource as resource_mod
+    import cleveragents.cli.renderers as _renderers_mod
 
-    orig_console = resource_mod.console
-    resource_mod.console = console
+    orig_r_console = _renderers_mod._console
+    orig_r_err = _renderers_mod._err_console
+    _renderers_mod._console = console
+    _renderers_mod._err_console = console
 
     failed = False
     try:
@@ -58,7 +60,8 @@ def _capture_output(func: Any, *args: Any, **kwargs: Any) -> tuple[str, bool]:
     except Exception:
         failed = True
     finally:
-        resource_mod.console = orig_console
+        _renderers_mod._console = orig_r_console
+        _renderers_mod._err_console = orig_r_err
 
     return buf.getvalue(), failed
 

features/steps/resource_cli_tree_steps.py

@@ -43,10 +43,12 @@ def _capture_tree_output(func: Any, *args: Any, **kwargs: Any) -> tuple[str, boo
     buf = StringIO()
     console = Console(file=buf, width=200, no_color=True)
 
-    import cleveragents.cli.commands.resource as resource_mod
+    import cleveragents.cli.renderers as _renderers_mod
 
-    orig_console = resource_mod.console
-    resource_mod.console = console
+    orig_r_console = _renderers_mod._console
+    orig_r_err = _renderers_mod._err_console
+    _renderers_mod._console = console
+    _renderers_mod._err_console = console
 
     failed = False
     try:
@@ -57,7 +59,8 @@ def _capture_tree_output(func: Any, *args: Any, **kwargs: Any) -> tuple[str, boo
     except Exception:
         failed = True
     finally:
-        resource_mod.console = orig_console
+        _renderers_mod._console = orig_r_console
+        _renderers_mod._err_console = orig_r_err
 
     return buf.getvalue(), failed
 

features/steps/resource_type_bootstrap_fs_steps.py

@@ -31,7 +31,7 @@ from cleveragents.infrastructure.database.models import Base
 # ── Helpers ──────────────────────────────────────────────────
 
 _PATCH_SERVICE = "cleveragents.cli.commands.resource._get_registry_service"
-_PATCH_CONSOLE = "cleveragents.cli.commands.resource.console"
+_PATCH_CONSOLE = "cleveragents.cli.renderers._console"
 
 
 def _make_service(context: Context, *, run_bootstrap: bool) -> ResourceRegistryService:

features/steps/resource_type_bootstrap_git_steps.py

@@ -31,7 +31,7 @@ from cleveragents.infrastructure.database.models import Base
 # ── Helpers ──────────────────────────────────────────────────
 
 _PATCH_SERVICE = "cleveragents.cli.commands.resource._get_registry_service"
-_PATCH_CONSOLE = "cleveragents.cli.commands.resource.console"
+_PATCH_CONSOLE = "cleveragents.cli.renderers._console"
 
 
 def _make_service(context: Context, *, run_bootstrap: bool) -> ResourceRegistryService:

features/steps/resource_type_inheritance_extra_steps.py

@@ -368,7 +368,7 @@ def step_resolve_handler_bad_ref(context: Any, ref: str) -> None:
 from unittest.mock import MagicMock, patch  # noqa: E402
 
 _PATCH_SVC = "cleveragents.cli.commands.resource._get_registry_service"
-_PATCH_CON = "cleveragents.cli.commands.resource.console"
+_PATCH_CON = "cleveragents.cli.renderers._console"
 
 
 def _plain_console():  # type: ignore[return]

features/steps/skill_cli_coverage_boost_steps.py

@@ -20,7 +20,6 @@ from behave import given, then, when
 from behave.runner import Context
 from typer.testing import CliRunner
 
-import cleveragents.cli.commands.skill as skill_mod
 from cleveragents.application.services.skill_service import SkillService
 from cleveragents.cli.commands.skill import (
     _get_skill_service,
@@ -187,10 +186,13 @@ def step_boost_refresh_bypass_first_guard(context: Context) -> None:
     """
     import typer
 
-    # Track what console.print receives
-    p_print = patch.object(skill_mod, "console")
-    mock_console = p_print.start()
-    context.boost_patches.append(p_print)
+    # Track what the renderer's error console receives
+    import cleveragents.cli.renderers as _renderers_mod
+
+    mock_err_console = MagicMock()
+    p_err = patch.object(_renderers_mod, "_err_console", mock_err_console)
+    p_err.start()
+    context.boost_patches.append(p_err)
 
     try:
         from cleveragents.cli.commands.skill import refresh as refresh_fn
@@ -199,8 +201,8 @@ def step_boost_refresh_bypass_first_guard(context: Context) -> None:
     except (typer.Abort, SystemExit):
         context.boost_guard_aborted = True
 
-    # Check if the error message was printed
-    for call_args in mock_console.print.call_args_list:
+    # Check if the error message was printed via render_error
+    for call_args in mock_err_console.print.call_args_list:
         if call_args and call_args[0]:
             msg = str(call_args[0][0])
             if "Must specify either" in msg:

features/steps/tool_cli_uncovered_branches_steps.py

@@ -28,10 +28,16 @@ class _FakeToolWithCliDict:
 
 def _capture_rich_output(tool, title="Tool") -> str:
     """Call _print_tool in rich format and capture console output."""
+    import cleveragents.cli.renderers as _renderers_mod
+
     buf = StringIO()
     test_console = Console(file=buf, force_terminal=False, width=120)
-    with patch("cleveragents.cli.commands.tool.console", test_console):
+    orig = _renderers_mod._console
+    _renderers_mod._console = test_console
+    try:
         _print_tool(tool, title=title, fmt="rich")
+    finally:
+        _renderers_mod._console = orig
     return buf.getvalue()
 
 

robot/cli_formats.robot

@@ -33,3 +33,15 @@ Plan Status Format Plain Outputs Key-Value Pairs
     ${result}=    Run Process    ${PYTHON}    ${HELPER}    plan-status-plain    cwd=${WORKSPACE}
     Should Be Equal As Integers    ${result.rc}    0
     Should Contain    ${result.stdout}    cli-formats-plan-status-plain-ok
+
+Renderer Detail JSON Produces Valid JSON
+    [Documentation]    Verify render_detail produces parseable JSON output
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    renderer-detail-json    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    cli-formats-renderer-detail-json-ok
+
+Renderer List Plain Uses ASCII Only
+    [Documentation]    Verify render_list plain output is ASCII-safe
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    renderer-list-plain-ascii    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    cli-formats-renderer-list-plain-ascii-ok

robot/helper_cli_formats.py

@@ -133,11 +133,55 @@ def plan_status_plain() -> None:
     print("cli-formats-plan-status-plain-ok")
 
 
+def renderer_detail_json() -> None:
+    import io
+
+    from rich.console import Console
+
+    import cleveragents.cli.renderers as _renderers_mod
+    from cleveragents.cli.renderers import render_detail
+
+    buf = io.StringIO()
+    _renderers_mod._console = Console(file=buf, no_color=True, highlight=False)
+    try:
+        render_detail({"name": "smoke-test", "value": 99}, fmt="json")
+    finally:
+        _renderers_mod._console = None
+    output = buf.getvalue().strip()
+    parsed = json.loads(output)
+    assert "name" in parsed, f"'name' not in {list(parsed.keys())}"
+    print("cli-formats-renderer-detail-json-ok")
+
+
+def renderer_list_plain_ascii() -> None:
+    import io
+
+    from rich.console import Console
+
+    import cleveragents.cli.renderers as _renderers_mod
+    from cleveragents.cli.renderers import ColumnSpec, render_list
+
+    cols = [ColumnSpec(key="name", label="Name"), ColumnSpec(key="val", label="Val")]
+    rows = [{"name": "caf\u00e9", "val": "\u2714 ok"}]
+    buf = io.StringIO()
+    _renderers_mod._console = Console(file=buf, no_color=True, highlight=False)
+    try:
+        render_list(rows, columns=cols, fmt="plain")
+    finally:
+        _renderers_mod._console = None
+    output = buf.getvalue()
+    for ch in output:
+        assert ord(ch) < 128, f"Non-ASCII char U+{ord(ch):04X} in plain output"
+    print("cli-formats-renderer-list-plain-ascii-ok")
+
+
 _COMMANDS = {
     "action-list-json": action_list_json,
     "action-show-yaml": action_show_yaml,
     "plan-list-json": plan_list_json,
     "plan-status-plain": plan_status_plain,
+    "renderer-detail-json": renderer_detail_json,
+    "renderer-list-plain-ascii": renderer_list_plain_ascii,
 }
 
 if __name__ == "__main__":

robot/helper_project_context_cli.py

@@ -141,11 +141,11 @@ def _run_cli_func(
             return_value=mock_cont,
         ),
         patch(
-            "cleveragents.cli.commands.project_context.console",
+            "cleveragents.cli.renderers._console",
             test_console,
         ),
         patch(
-            "cleveragents.cli.commands.project_context.err_console",
+            "cleveragents.cli.renderers._err_console",
             test_console,
         ),
         patch(

src/cleveragents/cli/commands/action.py

@@ -54,9 +54,6 @@ from typing import TYPE_CHECKING, Annotated
 
 import typer
 from pydantic import ValidationError as PydanticValidationError
-from rich.console import Console
-from rich.panel import Panel
-from rich.table import Table
 
 if TYPE_CHECKING:
     from cleveragents.application.services.plan_lifecycle_service import (
@@ -64,7 +61,20 @@ if TYPE_CHECKING:
     )
 
 from cleveragents.application.container import get_container
-from cleveragents.cli.formatting import OutputFormat, format_output
+from cleveragents.application.services.plan_lifecycle_service import (
+    PlanLifecycleService,
+)
+from cleveragents.cli.formatting import OutputFormat
+from cleveragents.cli.renderers import (
+    FORMAT_HELP,
+    ColumnSpec,
+    FieldSpec,
+    render_detail,
+    render_empty,
+    render_error,
+    render_list,
+    render_success,
+)
 from cleveragents.core.exceptions import (
     CleverAgentsError,
     NotFoundError,
@@ -76,10 +86,6 @@ from cleveragents.domain.models.core.action import Action
 app = typer.Typer(
     help="Manage actions (reusable plan templates) for the v3 plan lifecycle."
 )
-console = Console()
-
-# Reusable --format option description
-_FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich)"
 
 
 def _get_lifecycle_service() -> PlanLifecycleService:
@@ -132,75 +138,54 @@ def _action_spec_dict(action: Action) -> dict[str, object]:
     return result
 
 
+# -- Shared field / column specs for action outputs -----------------------
+
+_ACTION_DETAIL_FIELDS: list[FieldSpec] = [
+    FieldSpec(key="namespaced_name", label="Namespaced Name"),
+    FieldSpec(key="short_name", label="Short Name"),
+    FieldSpec(key="description", label="Description"),
+    FieldSpec(key="state", label="State"),
+    FieldSpec(key="strategy_actor", label="Strategy Actor"),
+    FieldSpec(key="execution_actor", label="Execution Actor"),
+    FieldSpec(key="estimation_actor", label="Estimation Actor"),
+    FieldSpec(key="invariant_actor", label="Invariant Actor"),
+    FieldSpec(key="reusable", label="Reusable"),
+    FieldSpec(key="read_only", label="Read Only"),
+    FieldSpec(
+        key="definition_of_done",
+        label="Definition of Done",
+        multiline=True,
+        truncate=120,
+    ),
+    FieldSpec(key="arguments", label="Arguments", multiline=True),
+    FieldSpec(key="invariants", label="Invariants", multiline=True),
+    FieldSpec(key="inputs_schema", label="Inputs Schema", multiline=True),
+    FieldSpec(key="automation_profile", label="Automation Profile"),
+    FieldSpec(key="created_at", label="Created"),
+]
+
+_ACTION_LIST_COLUMNS: list[ColumnSpec] = [
+    ColumnSpec(key="namespaced_name", label="Namespaced Name", style="cyan"),
+    ColumnSpec(key="short_name", label="Short Name", style="blue"),
+    ColumnSpec(key="state", label="State", style="yellow"),
+    ColumnSpec(key="strategy_actor", label="Strategy Actor", style="magenta"),
+    ColumnSpec(key="execution_actor", label="Execution Actor", style="magenta"),
+    ColumnSpec(
+        key="definition_of_done", label="Definition of Done", style="dim", truncate=40
+    ),
+    ColumnSpec(key="reusable", label="Reusable", justify="center"),
+    ColumnSpec(key="created_at", label="Created", style="green"),
+]
+
+
 def _print_action(
     action: Action,
     title: str = "Action",
     fmt: str = OutputFormat.RICH.value,
 ) -> None:
     """Print action details in the requested format."""
-    if fmt != OutputFormat.RICH.value:
-        data = _action_spec_dict(action)
-        console.print(format_output(data, fmt))
-        return
-
-    # Rich panel (original behaviour)
-    args_display = ""
-    if action.arguments:
-        args_lines: list[str] = []
-        for arg in action.arguments:
-            req = "required" if arg.requirement.value == "required" else "optional"
-            args_lines.append(f"  • {arg.name} ({arg.arg_type.value}, {req})")
-            if arg.description:
-                args_lines.append(f"    {arg.description}")
-        args_display = "\n".join(args_lines)
-    else:
-        args_display = "  (none)"
-
-    dod_summary = action.definition_of_done
-    if len(dod_summary) > 120:
-        dod_summary = dod_summary[:117] + "..."
-
-    details = (
-        f"[bold]Namespaced Name:[/bold] {action.namespaced_name}\n"
-        f"[bold]Short Name:[/bold] {action.namespaced_name.name}\n"
-        f"[bold]Description:[/bold] {action.description}\n"
-        f"[bold]State:[/bold] {action.state.value}\n"
-        f"[bold]Strategy Actor:[/bold] {action.strategy_actor}\n"
-        f"[bold]Execution Actor:[/bold] {action.execution_actor}\n"
-    )
-
-    # Optional actors
-    if action.estimation_actor:
-        details += f"[bold]Estimation Actor:[/bold] {action.estimation_actor}\n"
-    if action.invariant_actor:
-        details += f"[bold]Invariant Actor:[/bold] {action.invariant_actor}\n"
-
-    details += (
-        f"[bold]Reusable:[/bold] {'yes' if action.reusable else 'no'}\n"
-        f"[bold]Read Only:[/bold] {'yes' if action.read_only else 'no'}\n"
-        f"[bold]Definition of Done:[/bold]\n  {dod_summary}\n"
-        f"[bold]Arguments:[/bold]\n{args_display}\n"
-    )
-
-    # Invariants
-    if action.invariants:
-        inv_lines = "\n".join(f"  • {inv}" for inv in action.invariants)
-        details += f"[bold]Invariants:[/bold]\n{inv_lines}\n"
-
-    # Inputs schema
-    if action.inputs_schema:
-        import json
-
-        schema_str = json.dumps(action.inputs_schema, indent=2)
-        details += f"[bold]Inputs Schema:[/bold]\n  {schema_str}\n"
-
-    # Automation profile
-    if action.automation_profile:
-        details += f"[bold]Automation Profile:[/bold] {action.automation_profile}\n"
-
-    details += f"[bold]Created:[/bold] {action.created_at}"
-
-    console.print(Panel(details, title=title, expand=False))
+    data = _action_spec_dict(action)
+    render_detail(data, title=title, fields=_ACTION_DETAIL_FIELDS, fmt=fmt)
 
 
 @app.command()
@@ -214,7 +199,7 @@ def create(
             exists=False,
         ),
     ],
-    fmt: Annotated[str, typer.Option("--format", "-f", help=_FORMAT_HELP)] = "rich",
+    fmt: Annotated[str, typer.Option("--format", "-f", help=FORMAT_HELP)] = "rich",
 ) -> None:
     """Create a new action from a YAML configuration file.
 
@@ -255,19 +240,19 @@ def create(
         _print_action(action, title="Action Created", fmt=fmt)
 
     except FileNotFoundError as e:
-        console.print(f"[red]Config file error:[/red] {e}")
+        render_error("Config file error", str(e))
         raise typer.Abort() from e
     except PydanticValidationError as e:
-        console.print(f"[red]Schema validation error:[/red] {e}")
+        render_error("Schema validation error", str(e))
         raise typer.Abort() from e
     except ValueError as e:
-        console.print(f"[red]Config validation error:[/red] {e}")
+        render_error("Config validation error", str(e))
         raise typer.Abort() from e
     except ValidationError as e:
-        console.print(f"[red]Validation Error:[/red] {e.message}")
+        render_error("Validation Error", e.message)
         raise typer.Abort() from e
     except CleverAgentsError as e:
-        console.print(f"[red]Error:[/red] {e.message}")
+        render_error("Error", e.message)
         raise typer.Abort() from e
 
 
@@ -300,7 +285,7 @@ def list_actions(
         typer.Option(
             "--format",
             "-f",
-            help=_FORMAT_HELP,
+            help=FORMAT_HELP,
         ),
     ] = "rich",
 ) -> None:
@@ -326,9 +311,10 @@ def list_actions(
             try:
                 state_filter = ActionState(state.lower())
             except ValueError as exc:
-                console.print(
-                    f"[red]Invalid state:[/red] {state}. "
-                    "Valid values: available, archived"
+                render_error(
+                    "Invalid state",
+                    f"{state}. Valid values: available, archived",
+                    fmt=fmt,
                 )
                 raise typer.Abort() from exc
 
@@ -339,7 +325,7 @@ def list_actions(
             try:
                 pattern = re.compile(regex)
             except re.error as exc:
-                console.print(f"[red]Invalid regex pattern:[/red] {regex}")
+                render_error("Invalid regex pattern", regex, fmt=fmt)
                 raise typer.Abort() from exc
             actions = [
                 a
@@ -349,48 +335,18 @@ def list_actions(
             ]
 
         if not actions:
-            console.print("[yellow]No actions found.[/yellow]")
-            console.print("Create one with 'agents action create --config <file>'")
-            return
-
-        # Non-rich formats use the formatting helper
-        if fmt != OutputFormat.RICH.value:
-            data = [_action_spec_dict(a) for a in actions]
-            console.print(format_output(data, fmt))
-            return
-
-        # Display actions table (rich default)
-        table = Table(title=f"Actions ({len(actions)} total)")
-        table.add_column("Namespaced Name", style="cyan")
-        table.add_column("Short Name", style="blue")
-        table.add_column("State", style="yellow")
-        table.add_column("Strategy Actor", style="magenta")
-        table.add_column("Execution Actor", style="magenta")
-        table.add_column("Definition of Done", style="dim")
-        table.add_column("Reusable", justify="center")
-        table.add_column("Created", style="green")
-
-        for action in actions:
-            # Truncate definition of done for table display
-            dod = action.definition_of_done
-            if len(dod) > 40:
-                dod = dod[:37] + "..."
-
-            table.add_row(
-                str(action.namespaced_name),
-                action.namespaced_name.name,
-                action.state.value,
-                action.strategy_actor,
-                action.execution_actor,
-                dod,
-                "✓" if action.reusable else "",
-                str(action.created_at.strftime("%Y-%m-%d %H:%M")),
+            render_empty(
+                "actions",
+                recovery="Create one with 'agents action create --config <file>'",
+                fmt=fmt,
             )
+            return
 
-        console.print(table)
+        data = [_action_spec_dict(a) for a in actions]
+        render_list(data, columns=_ACTION_LIST_COLUMNS, title="Actions", fmt=fmt)
 
     except CleverAgentsError as e:
-        console.print(f"[red]Error:[/red] {e.message}")
+        render_error("Error", e.message, fmt=fmt)
         raise typer.Abort() from e
 
 
@@ -405,7 +361,7 @@ def show(
         typer.Option(
             "--format",
             "-f",
-            help=_FORMAT_HELP,
+            help=FORMAT_HELP,
         ),
     ] = "rich",
 ) -> None:
@@ -421,10 +377,10 @@ def show(
         _print_action(action, title="Action Details", fmt=fmt)
 
     except NotFoundError as e:
-        console.print(f"[red]Action not found:[/red] {name}")
+        render_error("Action not found", name, fmt=fmt)
         raise typer.Abort() from e
     except CleverAgentsError as e:
-        console.print(f"[red]Error:[/red] {e.message}")
+        render_error("Error", e.message, fmt=fmt)
         raise typer.Abort() from e
 
 
@@ -439,7 +395,7 @@ def archive(
         typer.Option(
             "--format",
             "-f",
-            help=_FORMAT_HELP,
+            help=FORMAT_HELP,
         ),
     ] = "rich",
 ) -> None:
@@ -453,17 +409,17 @@ def archive(
         action = service.get_action_by_name(name)
         action = service.archive_action(str(action.namespaced_name))
 
-        if fmt != OutputFormat.RICH.value:
-            data = _action_spec_dict(action)
-            data["archived"] = True
-            console.print(format_output(data, fmt))
-            return
-
-        console.print(f"[green]✓[/green] Action archived: {action.namespaced_name}")
+        data = _action_spec_dict(action)
+        data["archived"] = True
+        render_success(
+            f"Action archived: {action.namespaced_name}",
+            fmt=fmt,
+            data=data,
+        )
 
     except NotFoundError as e:
-        console.print(f"[red]Action not found:[/red] {name}")
+        render_error("Action not found", name, fmt=fmt)
         raise typer.Abort() from e
     except CleverAgentsError as e:
-        console.print(f"[red]Error:[/red] {e.message}")
+        render_error("Error", e.message, fmt=fmt)
         raise typer.Abort() from e

src/cleveragents/cli/commands/actor.py

@@ -6,14 +6,21 @@ from pathlib import Path
 from typing import Annotated, Any, cast
 
 import typer
-from rich.console import Console
-from rich.panel import Panel
-from rich.table import Table
 
 from cleveragents.actor.config import ActorConfiguration
 from cleveragents.actor.schema import actor_role_warnings
 from cleveragents.application.container import get_container
-from cleveragents.cli.formatting import OutputFormat, format_output
+from cleveragents.cli.renderers import (
+    FORMAT_HELP,
+    ColumnSpec,
+    FieldSpec,
+    render_detail,
+    render_empty,
+    render_error,
+    render_list,
+    render_success,
+    render_warning,
+)
 from cleveragents.core.exceptions import (
     BusinessRuleViolation,
     CleverAgentsError,
@@ -33,10 +40,6 @@ app = typer.Typer(
         "use --unsafe when configs are marked unsafe."
     )
 )
-console = Console()
-
-# Reusable --format option description
-_FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich)"
 
 
 @app.command()
@@ -324,34 +327,42 @@ def _actor_spec_dict(actor: Actor) -> dict[str, object]:
     return result
 
 
+_ACTOR_DETAIL_FIELDS: list[FieldSpec] = [
+    FieldSpec(key="name", label="Name"),
+    FieldSpec(key="provider", label="Provider"),
+    FieldSpec(key="model", label="Model"),
+    FieldSpec(key="unsafe", label="Unsafe"),
+    FieldSpec(key="is_default", label="Default"),
+    FieldSpec(key="is_built_in", label="Built-in"),
+    FieldSpec(key="config_hash", label="Config Hash"),
+    FieldSpec(key="updated_at", label="Updated"),
+]
+
+_ACTOR_LIST_COLUMNS: list[ColumnSpec] = [
+    ColumnSpec(key="name", label="Name", style="cyan"),
+    ColumnSpec(key="provider", label="Provider", style="magenta"),
+    ColumnSpec(key="model", label="Model", style="magenta"),
+    ColumnSpec(key="is_default", label="Default", justify="center"),
+    ColumnSpec(key="is_built_in", label="Built-in", justify="center"),
+    ColumnSpec(key="unsafe", label="Unsafe", justify="center"),
+    ColumnSpec(key="updated_at", label="Updated", style="green"),
+]
+
+
 def _print_actor(
     actor: Actor,
     title: str = "Actor",
-    fmt: str = OutputFormat.RICH.value,
+    fmt: str = "rich",
 ) -> None:
     """Print actor details in the requested format."""
-    if fmt != OutputFormat.RICH.value:
-        data = _actor_spec_dict(actor)
-        console.print(format_output(data, fmt))
-        return
-
-    details = (
-        f"[bold]Name:[/bold] {actor.name}\n"
-        f"[bold]Provider:[/bold] {actor.provider}\n"
-        f"[bold]Model:[/bold] {actor.model}\n"
-        f"[bold]Unsafe:[/bold] {'yes' if actor.unsafe else 'no'}\n"
-        f"[bold]Default:[/bold] {'yes' if actor.is_default else 'no'}\n"
-        f"[bold]Built-in:[/bold] {'yes' if actor.is_built_in else 'no'}\n"
-        f"[bold]Config Hash:[/bold] {actor.config_hash}\n"
-        f"[bold]Updated:[/bold] {actor.updated_at}"
-    )
-    console.print(Panel(details, title=title, expand=False))
+    data = _actor_spec_dict(actor)
+    render_detail(data, title=title, fields=_ACTOR_DETAIL_FIELDS, fmt=fmt)
 
 
 def _print_role_warnings(config_blob: dict[str, Any]) -> None:
     """Print non-fatal role compatibility warnings for actor configs."""
     for warning in actor_role_warnings(config_blob):
-        console.print(f"[yellow]Warning:[/yellow] {warning}")
+        render_warning(warning)
 
 
 @app.command()
@@ -381,7 +392,7 @@ def add(
     ] = None,
     fmt: Annotated[
         str,
-        typer.Option("--format", "-f", help=_FORMAT_HELP),
+        typer.Option("--format", "-f", help=FORMAT_HELP),
     ] = "rich",
 ) -> None:
     """Add a new actor configuration.
@@ -445,7 +456,7 @@ def add(
         title = "Actor updated" if update_existing else "Actor added"
         _print_actor(actor, title=title, fmt=fmt)
     except (ValidationError, BusinessRuleViolation) as exc:
-        console.print(f"[red]Error:[/red] {exc}")
+        render_error("Error", str(exc), fmt=fmt)
         raise typer.Abort() from exc
 
 
@@ -475,7 +486,7 @@ def update(
     ] = None,
     fmt: Annotated[
         str,
-        typer.Option("--format", "-f", help=_FORMAT_HELP),
+        typer.Option("--format", "-f", help=FORMAT_HELP),
     ] = "rich",
 ) -> None:
     """Update an existing actor."""
@@ -487,7 +498,7 @@ def update(
     try:
         current = registry.get_actor(name) if registry else service.get_actor(name)
     except NotFoundError as exc:
-        console.print(f"[red]Actor not found:[/red] {exc}")
+        render_error("Actor not found", str(exc), fmt=fmt)
         raise typer.Abort() from exc
 
     raw_config = _load_config(config)
@@ -568,7 +579,7 @@ def update(
             )
         _print_actor(actor, title="Actor updated", fmt=fmt)
     except (ValidationError, BusinessRuleViolation) as exc:
-        console.print(f"[red]Error:[/red] {exc}")
+        render_error("Error", str(exc), fmt=fmt)
         raise typer.Abort() from exc
 
 
@@ -585,9 +596,9 @@ def remove(name: Annotated[str, typer.Argument(help="Actor name to remove")]) ->
             registry.remove_actor(name)
         else:
             service.remove_actor(name)
-        console.print(f"[green]✓[/green] Removed actor: {name}")
+        render_success(f"Removed actor: {name}")
     except (ValidationError, BusinessRuleViolation, NotFoundError) as exc:
-        console.print(f"[red]Error:[/red] {exc}")
+        render_error("Error", str(exc))
         raise typer.Abort() from exc
 
 
@@ -595,7 +606,7 @@ def remove(name: Annotated[str, typer.Argument(help="Actor name to remove")]) ->
 def list_actors(
     fmt: Annotated[
         str,
-        typer.Option("--format", "-f", help=_FORMAT_HELP),
+        typer.Option("--format", "-f", help=FORMAT_HELP),
     ] = "rich",
 ) -> None:
     """List all actors.
@@ -612,36 +623,11 @@ def list_actors(
     service, registry = _get_services()
     actors = registry.list_actors() if registry else service.list_actors()
     if not actors:
-        console.print("[yellow]No actors configured.[/yellow]")
+        render_empty("actors", message="No actors configured.", fmt=fmt)
         return
 
-    # Non-rich formats use the formatting helper
-    if fmt != OutputFormat.RICH.value:
-        data = [_actor_spec_dict(a) for a in actors]
-        console.print(format_output(data, fmt))
-        return
-
-    table = Table(title=f"Actors ({len(actors)} total)")
-    table.add_column("Name", style="cyan")
-    table.add_column("Provider", style="magenta")
-    table.add_column("Model", style="magenta")
-    table.add_column("Default", justify="center")
-    table.add_column("Built-in", justify="center")
-    table.add_column("Unsafe", justify="center")
-    table.add_column("Updated", style="green")
-
-    for actor in actors:
-        table.add_row(
-            actor.name,
-            actor.provider,
-            actor.model,
-            "✓" if actor.is_default else "",
-            "✓" if actor.is_built_in else "",
-            "yes" if actor.unsafe else "no",
-            str(actor.updated_at),
-        )
-
-    console.print(table)
+    data = [_actor_spec_dict(a) for a in actors]
+    render_list(data, columns=_ACTOR_LIST_COLUMNS, title="Actors", fmt=fmt)
 
 
 @app.command()
@@ -649,7 +635,7 @@ def show(
     name: Annotated[str, typer.Argument(help="Actor name to show")],
     fmt: Annotated[
         str,
-        typer.Option("--format", "-f", help=_FORMAT_HELP),
+        typer.Option("--format", "-f", help=FORMAT_HELP),
     ] = "rich",
 ) -> None:
     """Show details for an actor.
@@ -667,7 +653,7 @@ def show(
         actor = registry.get_actor(name) if registry else service.get_actor(name)
         _print_actor(actor, title="Actor details", fmt=fmt)
     except (ValidationError, NotFoundError) as exc:
-        console.print(f"[red]Error:[/red] {exc}")
+        render_error("Error", str(exc), fmt=fmt)
         raise typer.Abort() from exc
 
 
@@ -686,5 +672,5 @@ def set_default(
         )
         _print_actor(actor, title="Default actor updated")
     except (ValidationError, NotFoundError, BusinessRuleViolation) as exc:
-        console.print(f"[red]Error:[/red] {exc}")
+        render_error("Error", str(exc))
         raise typer.Abort() from exc

src/cleveragents/cli/commands/invariant.py

@@ -40,20 +40,23 @@ import re
 from typing import Annotated
 
 import typer
-from rich.console import Console
-from rich.table import Table
 
 from cleveragents.application.services.invariant_service import InvariantService
-from cleveragents.cli.formatting import OutputFormat, format_output
+from cleveragents.cli.renderers import (
+    FORMAT_HELP,
+    ColumnSpec,
+    render_empty,
+    render_error,
+    render_list,
+    render_success,
+    render_warning,
+)
 from cleveragents.core.exceptions import CleverAgentsError, NotFoundError
 from cleveragents.domain.models.core.invariant import Invariant, InvariantScope
 
 app = typer.Typer(
     help="Manage invariant constraints for plan execution.",
 )
-console = Console()
-
-_FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich)"
 
 # Module-level service instance (in-memory, same lifetime as CLI process)
 _service: InvariantService | None = None
@@ -109,6 +112,15 @@ def _invariant_dict(inv: Invariant) -> dict[str, object]:
     }
 
 
+_INVARIANT_LIST_COLUMNS: list[ColumnSpec] = [
+    ColumnSpec(key="id", label="ID", style="cyan"),
+    ColumnSpec(key="scope", label="Scope", style="yellow"),
+    ColumnSpec(key="source_name", label="Source", style="magenta"),
+    ColumnSpec(key="text", label="Text", style="white"),
+    ColumnSpec(key="active", label="Active", justify="center"),
+]
+
+
 @app.command()
 def add(
     text: Annotated[str, typer.Argument(help="The invariant constraint text")],
@@ -120,7 +132,7 @@ def add(
     ] = None,
     plan: Annotated[str | None, typer.Option("--plan", help="Plan ID (ULID)")] = None,
     action: Annotated[str | None, typer.Option("--action", help="Action name")] = None,
-    fmt: Annotated[str, typer.Option("--format", "-f", help=_FORMAT_HELP)] = "rich",
+    fmt: Annotated[str, typer.Option("--format", "-f", help=FORMAT_HELP)] = "rich",
 ) -> None:
     """Add a new invariant constraint.
 
@@ -133,17 +145,18 @@ def add(
         service = _get_service()
         inv = service.add_invariant(text=text, scope=scope, source_name=source_name)
 
-        if fmt != OutputFormat.RICH.value:
-            console.print(format_output(_invariant_dict(inv), fmt))
-            return
-
-        console.print(f"[green]Invariant added:[/green] {inv.id}")
-        console.print(f"  Text:   {inv.text}")
-        console.print(f"  Scope:  {inv.scope.value}")
-        console.print(f"  Source: {inv.source_name}")
+        data = _invariant_dict(inv)
+        render_success(
+            f"Invariant added: {inv.id}\n"
+            f"  Text:   {inv.text}\n"
+            f"  Scope:  {inv.scope.value}\n"
+            f"  Source: {inv.source_name}",
+            fmt=fmt,
+            data=data,
+        )
 
     except CleverAgentsError as e:
-        console.print(f"[red]Error:[/red] {e.message}")
+        render_error("Error", e.message, fmt=fmt)
         raise typer.Abort() from e
 
 
@@ -168,7 +181,7 @@ def list_invariants(
         str | None,
         typer.Argument(help="Optional regex to filter invariant text"),
     ] = None,
-    fmt: Annotated[str, typer.Option("--format", "-f", help=_FORMAT_HELP)] = "rich",
+    fmt: Annotated[str, typer.Option("--format", "-f", help=FORMAT_HELP)] = "rich",
 ) -> None:
     """List invariants with optional filters.
 
@@ -207,39 +220,24 @@ def list_invariants(
             try:
                 pattern = re.compile(regex)
             except re.error as exc:
-                console.print(f"[red]Invalid regex:[/red] {regex}")
+                render_error("Invalid regex", regex, fmt=fmt)
                 raise typer.Abort() from exc
             invariants = [inv for inv in invariants if pattern.search(inv.text)]
 
         if not invariants:
-            console.print("[yellow]No invariants found.[/yellow]")
+            render_empty("invariants", fmt=fmt)
             return
 
-        if fmt != OutputFormat.RICH.value:
-            data = [_invariant_dict(inv) for inv in invariants]
-            console.print(format_output(data, fmt))
-            return
-
-        table = Table(title=f"Invariants ({len(invariants)} total)")
-        table.add_column("ID", style="cyan", max_width=26)
-        table.add_column("Scope", style="yellow")
-        table.add_column("Source", style="magenta")
-        table.add_column("Text", style="white")
-        table.add_column("Active", justify="center")
-
-        for inv in invariants:
-            table.add_row(
-                inv.id,
-                inv.scope.value,
-                inv.source_name,
-                inv.text,
-                "yes" if inv.active else "no",
-            )
-
-        console.print(table)
+        data = [_invariant_dict(inv) for inv in invariants]
+        render_list(
+            data,
+            columns=_INVARIANT_LIST_COLUMNS,
+            title="Invariants",
+            fmt=fmt,
+        )
 
     except CleverAgentsError as e:
-        console.print(f"[red]Error:[/red] {e.message}")
+        render_error("Error", e.message, fmt=fmt)
         raise typer.Abort() from e
 
 
@@ -247,7 +245,7 @@ def list_invariants(
 def remove(
     invariant_id: Annotated[str, typer.Argument(help="Invariant ULID to remove")],
     yes: Annotated[bool, typer.Option("--yes", "-y", help="Skip confirmation")] = False,
-    fmt: Annotated[str, typer.Option("--format", "-f", help=_FORMAT_HELP)] = "rich",
+    fmt: Annotated[str, typer.Option("--format", "-f", help=FORMAT_HELP)] = "rich",
 ) -> None:
     """Remove (soft-delete) an invariant by ID.
 
@@ -261,22 +259,22 @@ def remove(
                 f"Remove invariant {invariant_id}?", default=False
             )
             if not confirmed:
-                console.print("[yellow]Cancelled.[/yellow]")
+                render_warning("Cancelled.", fmt=fmt)
                 raise typer.Abort()
 
         service = _get_service()
         inv = service.remove_invariant(invariant_id)
 
-        if fmt != OutputFormat.RICH.value:
-            data = _invariant_dict(inv)
-            console.print(format_output(data, fmt))
-            return
-
-        console.print(f"[green]Invariant removed:[/green] {inv.id}")
+        data = _invariant_dict(inv)
+        render_success(
+            f"Invariant removed: {inv.id}",
+            fmt=fmt,
+            data=data,
+        )
 
     except NotFoundError as e:
-        console.print(f"[red]Invariant not found:[/red] {invariant_id}")
+        render_error("Invariant not found", invariant_id, fmt=fmt)
         raise typer.Abort() from e
     except CleverAgentsError as e:
-        console.print(f"[red]Error:[/red] {e.message}")
+        render_error("Error", e.message, fmt=fmt)
         raise typer.Abort() from e

src/cleveragents/cli/commands/lsp.py

@@ -55,18 +55,23 @@ import typer
 import yaml
 from pydantic import ValidationError as PydanticValidationError
 from rich.console import Console
-from rich.panel import Panel
-from rich.table import Table
 
-from cleveragents.cli.formatting import OutputFormat, format_output
+from cleveragents.cli.renderers import (
+    FORMAT_HELP,
+    ColumnSpec,
+    FieldSpec,
+    render_detail,
+    render_empty,
+    render_error,
+    render_list,
+    render_success,
+    render_warning,
+)
 from cleveragents.lsp.errors import LspServerNotFoundError
 from cleveragents.lsp.models import LspServerConfig
 from cleveragents.lsp.registry import LspRegistry
 
 app = typer.Typer(help="Manage LSP (Language Server Protocol) servers in the registry.")
-console = Console()
-
-_FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich)"
 
 # ---------------------------------------------------------------------------
 # Module-level registry singleton (in-memory; replaced by DI container
@@ -102,6 +107,59 @@ def _config_dict(config: LspServerConfig) -> dict[str, Any]:
     return data
 
 
+_LSP_DETAIL_FIELDS: list[FieldSpec] = [
+    FieldSpec(key="name", label="Name"),
+    FieldSpec(key="namespace", label="Namespace"),
+    FieldSpec(key="command", label="Command"),
+    FieldSpec(
+        key="args",
+        label="Args",
+        accessor=lambda d: " ".join(d.get("args", [])) or "(none)",
+    ),
+    FieldSpec(
+        key="languages",
+        label="Languages",
+        accessor=lambda d: ", ".join(d.get("languages", [])) or "(none)",
+    ),
+    FieldSpec(
+        key="capabilities",
+        label="Capabilities",
+        accessor=lambda d: (
+            ", ".join(
+                c if isinstance(c, str) else str(c) for c in d.get("capabilities", [])
+            )
+            or "(none)"
+        ),
+    ),
+    FieldSpec(
+        key="env",
+        label="Environment",
+        multiline=True,
+        accessor=lambda d: (
+            "\n".join(f"{k} = {v}" for k, v in sorted(d.get("env", {}).items()))
+            if d.get("env")
+            else ""
+        ),
+    ),
+]
+
+_LSP_LIST_COLUMNS: list[ColumnSpec] = [
+    ColumnSpec(key="name", label="Name", style="cyan"),
+    ColumnSpec(key="command", label="Command"),
+    ColumnSpec(
+        key="languages",
+        label="Languages",
+        accessor=lambda d: ", ".join(d.get("languages", [])) or "(none)",
+    ),
+    ColumnSpec(
+        key="capabilities",
+        label="Capabilities",
+        justify="right",
+        accessor=lambda d: str(len(d.get("capabilities", []))),
+    ),
+]
+
+
 # ---------------------------------------------------------------------------
 # CLI commands
 # ---------------------------------------------------------------------------
@@ -126,7 +184,7 @@ def add(
     ] = False,
     fmt: Annotated[
         str,
-        typer.Option("--format", "-f", help=_FORMAT_HELP),
+        typer.Option("--format", "-f", help=FORMAT_HELP),
     ] = "rich",
 ) -> None:
     """Register a new LSP server from a YAML configuration file.
@@ -142,12 +200,12 @@ def add(
         # error message instead of Typer's default traceback-style
         # ``FileNotFoundError``.
         if not config.exists():
-            console.print(f"[red]Config file error:[/red] {config} not found")
+            render_error("Config file error", f"{config} not found", fmt=fmt)
             raise typer.Abort()
 
         raw = yaml.safe_load(config.read_text(encoding="utf-8"))
         if not isinstance(raw, dict):
-            console.print("[red]Schema validation error:[/red] YAML must be a mapping")
+            render_error("Schema validation error", "YAML must be a mapping", fmt=fmt)
             raise typer.Abort()
 
         server_config = LspServerConfig(**raw)
@@ -158,34 +216,24 @@ def add(
 
         registry.register(server_config)
 
-        if fmt != OutputFormat.RICH.value:
-            data = _config_dict(server_config)
-            data["registered"] = True
-            typer.echo(format_output(data, fmt))
-            return
-
-        langs = ", ".join(server_config.languages) or "(none)"
-        caps = ", ".join(c.value for c in server_config.capabilities) or "(none)"
-        details = (
-            f"[bold]Name:[/bold] {server_config.name}\n"
-            f"[bold]Command:[/bold] {server_config.command}\n"
-            f"[bold]Languages:[/bold] {langs}\n"
-            f"[bold]Capabilities:[/bold] {caps}"
+        data = _config_dict(server_config)
+        data["registered"] = True
+        render_detail(
+            data, title="LSP Server Registered", fields=_LSP_DETAIL_FIELDS, fmt=fmt
         )
-        console.print(Panel(details, title="LSP Server Registered", expand=False))
-        console.print("[green]OK[/green] LSP server registered")
 
     except PydanticValidationError as exc:
-        console.print(f"[red]Schema validation error:[/red] {exc}")
+        render_error("Schema validation error", str(exc), fmt=fmt)
         raise typer.Abort() from exc
     except ValueError as exc:
         error_msg = str(exc)
-        console.print(f"[red]Error:[/red] {error_msg}")
+        recovery = None
         if "already registered" in error_msg:
-            console.print(
-                "\nTo overwrite, re-run with [cyan]--update[/cyan]:\n\n"
+            recovery = (
+                f"To overwrite, re-run with --update:\n"
                 f"  agents lsp add --config {config} --update"
             )
+        render_error("Error", error_msg, recovery=recovery, fmt=fmt)
         raise typer.Abort() from exc
 
 
@@ -201,7 +249,7 @@ def remove(
     ] = False,
     fmt: Annotated[
         str,
-        typer.Option("--format", "-f", help=_FORMAT_HELP),
+        typer.Option("--format", "-f", help=FORMAT_HELP),
     ] = "rich",
 ) -> None:
     """Remove a registered LSP server.
@@ -213,25 +261,19 @@ def remove(
     registry = _get_registry()
     config = registry.get(name)
     if config is None:
-        console.print(f"[red]LSP server not found:[/red] {name}")
+        render_error("LSP server not found", name, fmt=fmt)
         raise typer.Abort()
 
     if not yes:
         confirm = typer.confirm(f"Remove LSP server {name}?", default=False)
         if not confirm:
-            console.print("[yellow]Aborted.[/yellow]")
+            render_warning("Aborted.", fmt=fmt)
             raise typer.Abort()
 
     registry.remove(name)
 
-    if fmt != OutputFormat.RICH.value:
-        typer.echo(format_output({"name": name, "removed": True}, fmt))
-        return
-
-    console.print(
-        Panel(f"[bold]Name:[/bold] {name}", title="LSP Server Removed", expand=False)
-    )
-    console.print("[green]OK[/green] LSP server removed")
+    data = {"name": name, "removed": True}
+    render_success(f"LSP server removed: {name}", fmt=fmt, data=data)
 
 
 @app.command("list")
@@ -246,7 +288,7 @@ def list_servers(
     ] = None,
     fmt: Annotated[
         str,
-        typer.Option("--format", "-f", help=_FORMAT_HELP),
+        typer.Option("--format", "-f", help=FORMAT_HELP),
     ] = "rich",
 ) -> None:
     """List registered LSP servers.
@@ -261,31 +303,15 @@ def list_servers(
     servers = registry.list_servers(namespace=namespace, language=language)
 
     if not servers:
-        console.print("[yellow]No LSP servers found.[/yellow]")
-        console.print("Register one with 'agents lsp add --config <file>'")
-        return
-
-    if fmt != OutputFormat.RICH.value:
-        data = [_config_dict(s) for s in servers]
-        typer.echo(format_output(data, fmt))
-        return
-
-    table = Table(title=f"LSP Servers ({len(servers)} total)", show_header=True)
-    table.add_column("Name", style="cyan")
-    table.add_column("Command")
-    table.add_column("Languages")
-    table.add_column("Capabilities", justify="right")
-
-    for server in servers:
-        table.add_row(
-            server.name,
-            server.command,
-            ", ".join(server.languages) or "(none)",
-            str(len(server.capabilities)),
+        render_empty(
+            "LSP servers",
+            recovery="Register one with 'agents lsp add --config <file>'",
+            fmt=fmt,
         )
+        return
 
-    console.print(table)
-    console.print(f"[green]OK[/green] {len(servers)} LSP server(s) listed")
+    data = [_config_dict(s) for s in servers]
+    render_list(data, columns=_LSP_LIST_COLUMNS, title="LSP Servers", fmt=fmt)
 
 
 @app.command("show")
@@ -296,7 +322,7 @@ def show(
     ],
     fmt: Annotated[
         str,
-        typer.Option("--format", "-f", help=_FORMAT_HELP),
+        typer.Option("--format", "-f", help=FORMAT_HELP),
     ] = "rich",
 ) -> None:
     """Show full details for a registered LSP server.
@@ -309,31 +335,11 @@ def show(
     try:
         config = registry.get_or_raise(name)
     except LspServerNotFoundError as exc:
-        console.print(f"[red]LSP server not found:[/red] {name}")
+        render_error("LSP server not found", name, fmt=fmt)
         raise typer.Abort() from exc
 
-    if fmt != OutputFormat.RICH.value:
-        typer.echo(format_output(_config_dict(config), fmt))
-        return
-
-    details = (
-        f"[bold]Name:[/bold] {config.name}\n"
-        f"[bold]Namespace:[/bold] {config.namespace}\n"
-        f"[bold]Command:[/bold] {config.command}\n"
-        f"[bold]Args:[/bold] {' '.join(config.args) or '(none)'}\n"
-        f"[bold]Languages:[/bold] {', '.join(config.languages) or '(none)'}\n"
-        f"[bold]Capabilities:[/bold] "
-        f"{', '.join(c.value for c in config.capabilities) or '(none)'}"
-    )
-    console.print(Panel(details, title="LSP Server Details", expand=False))
-
-    if config.env:
-        env_lines: list[str] = []
-        for key, val in sorted(config.env.items()):
-            env_lines.append(f"[blue]{key}[/blue] = {val}")
-        console.print(Panel("\n".join(env_lines), title="Environment", expand=False))
-
-    console.print("[green]OK[/green] LSP server loaded")
+    data = _config_dict(config)
+    render_detail(data, title="LSP Server Details", fields=_LSP_DETAIL_FIELDS, fmt=fmt)
 
 
 _VALID_LOG_LEVELS = ("debug", "info", "warning", "error")

src/cleveragents/cli/commands/session.py

@@ -22,12 +22,19 @@ from pathlib import Path
 from typing import Annotated, Any, cast
 
 import typer
-from rich.console import Console
 from rich.panel import Panel
 from rich.table import Table
 
 from cleveragents.a2a.models import A2aRequest
 from cleveragents.cli.formatting import OutputFormat, format_output
+from cleveragents.cli.renderers import (
+    FORMAT_HELP,
+    get_console,
+    render_empty,
+    render_error,
+    render_success,
+    render_warning,
+)
 from cleveragents.core.exceptions import DatabaseError
 from cleveragents.domain.models.core.session import (
     MessageRole,
@@ -40,13 +47,9 @@ from cleveragents.domain.models.core.session import (
 
 # Create sub-app for session commands
 app = typer.Typer(help="Manage interactive sessions.")
-console = Console()
 
 _log = logging.getLogger(__name__)
 
-# Reusable --format option description
-_FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich)"
-
 # ---------------------------------------------------------------------------
 # Module-level service accessor (patchable in tests)
 # ---------------------------------------------------------------------------
@@ -152,7 +155,7 @@ def create(
     ] = None,
     fmt: Annotated[
         str,
-        typer.Option("--format", "-f", help=_FORMAT_HELP),
+        typer.Option("--format", "-f", help=FORMAT_HELP),
     ] = "rich",
 ) -> None:
     """Create a new interactive session.
@@ -191,15 +194,15 @@ def create(
             f"[bold]Namespace:[/bold] {session.namespace}\n"
             f"[bold]Created:[/bold] {session.created_at.strftime('%Y-%m-%d %H:%M')}"
         )
-        console.print(Panel(details, title="Session Created", expand=False))
-        console.print("[green]✓ OK[/green] Session created")
+        get_console().print(Panel(details, title="Session Created", expand=False))
+        render_success("Session created", fmt=fmt)
 
     except SessionNotFoundError as exc:
-        console.print(f"[red]Error:[/red] {exc}")
+        render_error("Error", str(exc), fmt=fmt)
         raise typer.Exit(1) from exc
     except DatabaseError as exc:
         _log.debug("session create failed", exc_info=True)
-        console.print(
+        get_console().print(
             f"[red]Error:[/red] Database unavailable: {exc}\n"
             "Hint: run 'agents init' to initialise the database."
         )
@@ -210,7 +213,7 @@ def create(
 def list_sessions(
     fmt: Annotated[
         str,
-        typer.Option("--format", "-f", help=_FORMAT_HELP),
+        typer.Option("--format", "-f", help=FORMAT_HELP),
     ] = "rich",
 ) -> None:
     """List all sessions.
@@ -227,20 +230,20 @@ def list_sessions(
         sessions = service.list()
     except DatabaseError as exc:
         _log.debug("session list failed", exc_info=True)
-        console.print(
-            f"[red]Error:[/red] Database unavailable: {exc}\n"
-            "Hint: run 'agents init' to initialise the database."
+        render_error(
+            "Database Error",
+            f"Database unavailable: {exc}",
+            recovery="Run 'agents init' to initialise the database.",
         )
         raise typer.Exit(1) from exc
 
     if not sessions:
-        # For machine-readable formats, always emit a structured empty list
-        # so that callers parsing JSON/YAML receive valid output.
-        if fmt not in (OutputFormat.RICH.value, OutputFormat.COLOR.value):
-            typer.echo(format_output({"sessions": [], "total": 0}, fmt))
-            return
-        console.print("[yellow]No sessions found.[/yellow]")
-        console.print("Create one with 'agents session create'")
+        render_empty(
+            "sessions",
+            recovery="Create one with 'agents session create'",
+            fmt=fmt,
+            data={"sessions": [], "total": 0},
+        )
         return
 
     data = _session_list_dict(sessions)
@@ -264,7 +267,7 @@ def list_sessions(
             s.updated_at.strftime("%Y-%m-%d %H:%M"),
         )
 
-    console.print(table)
+    get_console().print(table)
 
     # Summary
     total_msgs = sum(s.message_count for s in sessions)
@@ -272,7 +275,7 @@ def list_sessions(
         f"[yellow]Total Sessions:[/yellow] {len(sessions)}\n"
         f"[blue]Total Messages:[/blue] {total_msgs}"
     )
-    console.print(Panel(summary, title="Summary", expand=False))
+    get_console().print(Panel(summary, title="Summary", expand=False))
 
 
 @app.command()
@@ -283,7 +286,7 @@ def show(
     ],
     fmt: Annotated[
         str,
-        typer.Option("--format", "-f", help=_FORMAT_HELP),
+        typer.Option("--format", "-f", help=FORMAT_HELP),
     ] = "rich",
 ) -> None:
     """Show session details and recent messages.
@@ -312,7 +315,7 @@ def show(
             f"[bold]Created:[/bold] {session.created_at.strftime('%Y-%m-%d %H:%M')}\n"
             f"[bold]Updated:[/bold] {session.updated_at.strftime('%Y-%m-%d %H:%M')}"
         )
-        console.print(Panel(details, title="Session Details", expand=False))
+        get_console().print(Panel(details, title="Session Details", expand=False))
 
         # Recent messages
         if session.messages:
@@ -330,12 +333,12 @@ def show(
                     content,
                     msg.timestamp.strftime("%H:%M:%S"),
                 )
-            console.print(msg_table)
+            get_console().print(msg_table)
 
         # Linked plans
         if session.linked_plan_ids:
             plan_text = "\n".join(f"  • {pid}" for pid in session.linked_plan_ids)
-            console.print(Panel(plan_text, title="Linked Plans", expand=False))
+            get_console().print(Panel(plan_text, title="Linked Plans", expand=False))
 
         # Token usage
         tu = session.token_usage
@@ -344,7 +347,7 @@ def show(
             f"[blue]Output Tokens:[/blue] {tu.output_tokens}\n"
             f"[yellow]Estimated Cost:[/yellow] ${tu.estimated_cost:.4f}"
         )
-        console.print(Panel(usage_text, title="Token Usage", expand=False))
+        get_console().print(Panel(usage_text, title="Token Usage", expand=False))
 
         # Cost budget (per-session budget cap, #584)
         if session.cost_budget is not None:
@@ -366,14 +369,14 @@ def show(
                 f"[yellow]Utilization:[/yellow] {util_display}\n"
                 f"[green]Remaining:[/green] {remaining_display}"
             )
-            console.print(Panel(budget_text, title="Cost Budget", expand=False))
+            get_console().print(Panel(budget_text, title="Cost Budget", expand=False))
 
     except SessionNotFoundError as exc:
-        console.print(f"[red]Session not found:[/red] {session_id}")
+        render_error("Session not found", session_id, fmt=fmt)
         raise typer.Exit(1) from exc
     except DatabaseError as exc:
         _log.debug("session show failed", exc_info=True)
-        console.print(
+        get_console().print(
             f"[red]Error:[/red] Database unavailable: {exc}\n"
             "Hint: run 'agents init' to initialise the database."
         )
@@ -408,18 +411,18 @@ def delete(
         if not yes:
             confirm = typer.confirm(f"Delete session {session_id}?", default=False)
             if not confirm:
-                console.print("[yellow]Aborted.[/yellow]")
+                render_warning("Aborted.")
                 raise typer.Abort()
 
         service.delete(session_id)
-        console.print(f"[green]✓ OK[/green] Session {session_id} deleted")
+        render_success(f"Session {session_id} deleted")
 
     except SessionNotFoundError as exc:
-        console.print(f"[red]Session not found:[/red] {session_id}")
+        render_error("Session not found", session_id)
         raise typer.Exit(1) from exc
     except DatabaseError as exc:
         _log.debug("session delete failed", exc_info=True)
-        console.print(
+        get_console().print(
             f"[red]Error:[/red] Database unavailable: {exc}\n"
             "Hint: run 'agents init' to initialise the database."
         )
@@ -458,27 +461,27 @@ def export_session(
 
         if output is not None:
             if output.exists() and not force:
-                console.print(
-                    f"[red]File already exists:[/red] {output}\n"
-                    "Use --force to overwrite."
+                render_error(
+                    "File already exists",
+                    f"{output}\nUse --force to overwrite.",
                 )
                 raise typer.Exit(1)
             # Create parent directories if needed
             output.parent.mkdir(parents=True, exist_ok=True)
             output.write_text(json_str, encoding="utf-8")
-            console.print(f"[green]✓ OK[/green] Session exported to {output}")
+            render_success(f"Session exported to {output}")
         else:
             typer.echo(json_str)
 
     except SessionNotFoundError as exc:
-        console.print(f"[red]Session not found:[/red] {session_id}")
+        render_error("Session not found", session_id)
         raise typer.Exit(1) from exc
     except SessionExportError as exc:
-        console.print(f"[red]Export error:[/red] {exc}")
+        render_error("Export error", str(exc))
         raise typer.Exit(1) from exc
     except DatabaseError as exc:
         _log.debug("session export failed", exc_info=True)
-        console.print(
+        get_console().print(
             f"[red]Error:[/red] Database unavailable: {exc}\n"
             "Hint: run 'agents init' to initialise the database."
         )
@@ -501,14 +504,14 @@ def import_session(
         agents session import -i session.json
     """
     if not input_file.exists():
-        console.print(f"[red]File not found:[/red] {input_file}")
+        render_error("File not found", str(input_file))
         raise typer.Exit(1)
 
     try:
         raw = input_file.read_text(encoding="utf-8")
         data = json.loads(raw)
     except json.JSONDecodeError as exc:
-        console.print(f"[red]Invalid JSON:[/red] {exc}")
+        render_error("Invalid JSON", str(exc))
         raise typer.Exit(1) from exc
 
     try:
@@ -521,15 +524,15 @@ def import_session(
             f"[bold]Messages:[/bold] {session.message_count}\n"
             f"[bold]Namespace:[/bold] {session.namespace}"
         )
-        console.print(Panel(details, title="Session Imported", expand=False))
-        console.print("[green]✓ OK[/green] Session imported")
+        get_console().print(Panel(details, title="Session Imported", expand=False))
+        render_success("Session imported")
 
     except SessionImportError as exc:
-        console.print(f"[red]Import error:[/red] {exc}")
+        render_error("Import error", str(exc))
         raise typer.Exit(1) from exc
     except DatabaseError as exc:
         _log.debug("session import failed", exc_info=True)
-        console.print(
+        get_console().print(
             f"[red]Error:[/red] Database unavailable: {exc}\n"
             "Hint: run 'agents init' to initialise the database."
         )
@@ -596,15 +599,15 @@ def tell(
         else:
             from rich.markup import escape
 
-            console.print(f"[dim]user:[/dim] {escape(prompt)}")
-            console.print(f"[cyan]assistant:[/cyan] {escape(assistant_content)}")
+            get_console().print(f"[dim]user:[/dim] {escape(prompt)}")
+            get_console().print(f"[cyan]assistant:[/cyan] {escape(assistant_content)}")
 
     except SessionNotFoundError as exc:
-        console.print(f"[red]Session not found:[/red] {session_id}")
+        render_error("Session not found", session_id)
         raise typer.Exit(1) from exc
     except DatabaseError as exc:
         _log.debug("session tell failed", exc_info=True)
-        console.print(
+        get_console().print(
             f"[red]Error:[/red] Database unavailable: {exc}\n"
             "Hint: run 'agents init' to initialise the database."
         )

src/cleveragents/cli/commands/validation.py

@@ -51,10 +51,15 @@ from typing import Annotated, Any
 
 import typer
 import yaml
-from rich.console import Console
-from rich.panel import Panel
 
-from cleveragents.cli.formatting import OutputFormat, format_output
+from cleveragents.cli.renderers import (
+    FORMAT_HELP,
+    FieldSpec,
+    render_detail,
+    render_error,
+    render_success,
+    render_warning,
+)
 from cleveragents.core.exceptions import (
     CleverAgentsError,
     NotFoundError,
@@ -64,10 +69,6 @@ from cleveragents.domain.models.core.tool import Validation
 
 # Create sub-app for validation commands
 app = typer.Typer(help="Manage validations (pass/fail tools) and resource attachments.")
-console = Console()
-
-# Reusable --format option description
-_FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich)"
 
 
 def _get_tool_registry_service() -> Any:
@@ -143,37 +144,39 @@ def _attachment_dict(attachment: Any) -> dict[str, Any]:
     }
 
 
+_VALIDATION_DETAIL_FIELDS: list[FieldSpec] = [
+    FieldSpec(key="name", label="Name"),
+    FieldSpec(key="description", label="Description"),
+    FieldSpec(key="source", label="Source"),
+    FieldSpec(key="tool_type", label="Type"),
+    FieldSpec(key="mode", label="Mode"),
+    FieldSpec(
+        key="wraps",
+        label="Wraps",
+        accessor=lambda d: str(d.get("wraps", "")) if d.get("wraps") else "",
+    ),
+    FieldSpec(
+        key="transform",
+        label="Transform",
+        accessor=lambda d: str(d.get("transform", "")) if d.get("transform") else "",
+    ),
+]
+
+
 def _print_validation(
     tool: Any,
     title: str = "Validation",
-    fmt: str = OutputFormat.RICH.value,
+    fmt: str = "rich",
 ) -> None:
     """Print validation details in the requested format."""
     data = _validation_spec_dict(tool)
-    if fmt != OutputFormat.RICH.value:
-        console.print(format_output(data, fmt))
-        return
-
-    name = data.get("name", "")
-    desc = data.get("description", "")
-    source = data.get("source", "")
-    mode = data.get("mode", "required")
-
-    details = (
-        f"[bold]Name:[/bold] {name}\n"
-        f"[bold]Description:[/bold] {desc}\n"
-        f"[bold]Source:[/bold] {source}\n"
-        f"[bold]Mode:[/bold] {mode}"
-    )
-
-    wraps = data.get("wraps")
-    if wraps:
-        details += f"\n[bold]Wraps:[/bold] {wraps}"
-    transform = data.get("transform")
-    if transform:
-        details += f"\n[bold]Transform:[/bold] {transform}"
-
-    console.print(Panel(details, title=title, expand=False))
+    # Filter out fields with empty values for optional wraps/transform
+    fields = [
+        f
+        for f in _VALIDATION_DETAIL_FIELDS
+        if f.key not in ("wraps", "transform") or data.get(f.key)
+    ]
+    render_detail(data, title=title, fields=fields, fmt=fmt)
 
 
 @app.command("add")
@@ -193,7 +196,7 @@ def add(
     ] = False,
     fmt: Annotated[
         str,
-        typer.Option("--format", "-f", help=_FORMAT_HELP),
+        typer.Option("--format", "-f", help=FORMAT_HELP),
     ] = "rich",
 ) -> None:
     """Register a new validation from a YAML configuration file.
@@ -227,19 +230,19 @@ def add(
         _print_validation(registered, title="Validation Registered", fmt=fmt)
 
     except FileNotFoundError as exc:
-        console.print(f"[red]Config file error:[/red] {exc}")
+        render_error("Config file error", str(exc), fmt=fmt)
         raise typer.Abort() from exc
     except yaml.YAMLError as exc:
-        console.print(f"[red]YAML parse error:[/red] {exc}")
+        render_error("YAML parse error", str(exc), fmt=fmt)
         raise typer.Abort() from exc
     except ValueError as exc:
-        console.print(f"[red]Schema validation error:[/red] {exc}")
+        render_error("Schema validation error", str(exc), fmt=fmt)
         raise typer.Abort() from exc
     except ValidationError as exc:
-        console.print(f"[red]Validation Error:[/red] {exc.message}")
+        render_error("Validation Error", exc.message, fmt=fmt)
         raise typer.Abort() from exc
     except CleverAgentsError as exc:
-        console.print(f"[red]Error:[/red] {exc.message}")
+        render_error("Error", exc.message, fmt=fmt)
         raise typer.Abort() from exc
 
 
@@ -271,7 +274,7 @@ def attach(
     ] = "required",
     fmt: Annotated[
         str,
-        typer.Option("--format", "-f", help=_FORMAT_HELP),
+        typer.Option("--format", "-f", help=FORMAT_HELP),
     ] = "rich",
 ) -> None:
     """Attach a validation to a resource.
@@ -288,9 +291,10 @@ def attach(
             extra_args = {}
             for arg in args:
                 if "=" not in arg:
-                    console.print(
-                        f"[red]Invalid argument format:[/red] {arg} "
-                        "(expected key=value)"
+                    render_error(
+                        "Invalid argument format",
+                        f"{arg} (expected key=value)",
+                        fmt=fmt,
                     )
                     raise typer.Abort()
                 key, val = arg.split("=", 1)
@@ -307,25 +311,21 @@ def attach(
         )
 
         att_data = _attachment_dict(attachment)
-
-        if fmt != OutputFormat.RICH.value:
-            console.print(format_output(att_data, fmt))
-            return
-
         att_id = att_data.get("attachment_id", "")
-        console.print(
-            f"[green]Attached validation:[/green] {validation_name} -> "
-            f"{resource} (id: {att_id})"
+        render_success(
+            f"Attached validation: {validation_name} -> {resource} (id: {att_id})",
+            fmt=fmt,
+            data=att_data,
         )
 
     except NotFoundError as exc:
-        console.print(f"[red]Validation not found:[/red] {validation_name}")
+        render_error("Validation not found", validation_name, fmt=fmt)
         raise typer.Abort() from exc
     except ValidationError as exc:
-        console.print(f"[red]Validation Error:[/red] {exc.message}")
+        render_error("Validation Error", exc.message, fmt=fmt)
         raise typer.Abort() from exc
     except CleverAgentsError as exc:
-        console.print(f"[red]Error:[/red] {exc.message}")
+        render_error("Error", exc.message, fmt=fmt)
         raise typer.Abort() from exc
 
 
@@ -341,7 +341,7 @@ def detach(
     ] = False,
     fmt: Annotated[
         str,
-        typer.Option("--format", "-f", help=_FORMAT_HELP),
+        typer.Option("--format", "-f", help=FORMAT_HELP),
     ] = "rich",
 ) -> None:
     """Detach a validation from a resource.
@@ -354,23 +354,19 @@ def detach(
         if not yes:
             confirm = typer.confirm(f"Detach validation attachment '{attachment_id}'?")
             if not confirm:
-                console.print("[yellow]Aborted.[/yellow]")
+                render_warning("Aborted.", fmt=fmt)
                 raise typer.Abort()
 
         service = _get_tool_registry_service()
         removed = service.detach_validation(attachment_id)
 
         if not removed:
-            console.print(f"[red]Attachment not found:[/red] {attachment_id}")
+            render_error("Attachment not found", attachment_id, fmt=fmt)
             raise typer.Abort()
 
-        if fmt != OutputFormat.RICH.value:
-            data = {"attachment_id": attachment_id, "detached": True}
-            console.print(format_output(data, fmt))
-            return
-
-        console.print(f"[green]Detached validation:[/green] {attachment_id}")
+        data = {"attachment_id": attachment_id, "detached": True}
+        render_success(f"Detached validation: {attachment_id}", fmt=fmt, data=data)
 
     except CleverAgentsError as exc:
-        console.print(f"[red]Error:[/red] {exc.message}")
+        render_error("Error", exc.message, fmt=fmt)
         raise typer.Abort() from exc

src/cleveragents/cli/formatting.py

@@ -14,6 +14,8 @@ Based on v3_spec.md implementation plan Stage A4b / G5b.render.
 from __future__ import annotations
 
 import json
+import re
+import sys
 from datetime import datetime
 from enum import Enum, StrEnum
 from io import StringIO
@@ -32,6 +34,14 @@ from cleveragents.cli.output.materializers import (
     YamlMaterializer,
 )
 from cleveragents.cli.output.session import OutputSession
+from cleveragents.shared.redaction import redact_dict
+
+__all__ = [
+    "OutputFormat",
+    "ascii_safe",
+    "format_output",
+    "format_output_session",
+]
 
 
 class OutputFormat(StrEnum):
@@ -80,16 +90,33 @@ def _format_yaml(data: dict[str, Any] | list[dict[str, Any]]) -> str:
     ).rstrip("\n")
 
 
+def ascii_safe(text: str) -> str:
+    """Replace non-ASCII characters with ``?`` for log-pipeline safety.
+
+    Also strips ANSI escape sequences and control characters (except
+    standard whitespace \\t, \\n, \\r) to prevent terminal corruption
+    in plain-text log pipelines.
+    """
+    # Strip ANSI escape sequences (CSI sequences, OSC, etc.)
+    text = re.sub(r"\x1b\[[?0-9;]*[A-Za-z]|\x1b\][^\x07]*\x07", "", text)
+    # Remove control characters except tab, newline, carriage return
+    text = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]", "", text)
+    return text.encode("ascii", errors="replace").decode("ascii")
+
+
 def _format_plain(data: dict[str, Any] | list[dict[str, Any]]) -> str:
-    """Render data as plain key: value lines (no Rich markup)."""
+    """Render data as plain key: value lines (no Rich markup).
+
+    Output is ASCII-only for log-pipeline compatibility.
+    """
     if isinstance(data, list):
         parts: list[str] = []
         for idx, item in enumerate(data):
             if idx > 0:
                 parts.append("---")
             parts.append(_format_plain_dict(item))
-        return "\n".join(parts)
-    return _format_plain_dict(data)
+        return ascii_safe("\n".join(parts))
+    return ascii_safe(_format_plain_dict(data))
 
 
 def _format_plain_dict(data: dict[str, Any]) -> str:
@@ -113,6 +140,17 @@ def _format_plain_dict(data: dict[str, Any]) -> str:
     return "\n".join(lines)
 
 
+_table_console: Console | None = None
+
+
+def _get_table_console() -> Console:
+    """Return a reusable Console for table rendering."""
+    global _table_console
+    if _table_console is None:
+        _table_console = Console(file=StringIO(), width=200, no_color=True)
+    return _table_console
+
+
 def _format_table(data: dict[str, Any] | list[dict[str, Any]]) -> str:
     """Render data as an ASCII table (no Rich styling)."""
     rows: list[dict[str, Any]] = [data] if isinstance(data, dict) else data
@@ -143,7 +181,8 @@ def _format_table(data: dict[str, Any] | list[dict[str, Any]]) -> str:
         table.add_row(*cells)
 
     buf = StringIO()
-    console = Console(file=buf, width=200, no_color=True)
+    console = _get_table_console()
+    console.file = buf
     console.print(table)
     return buf.getvalue().rstrip("\n")
 
@@ -152,8 +191,6 @@ def _redact_data(
     data: dict[str, Any] | list[dict[str, Any]],
 ) -> dict[str, Any] | list[dict[str, Any]]:
     """Apply secrets redaction to output data before rendering."""
-    from cleveragents.shared.redaction import redact_dict
-
     if isinstance(data, list):
         return [redact_dict(item) if isinstance(item, dict) else item for item in data]
     return redact_dict(data)
@@ -188,8 +225,6 @@ def format_output(
         The rendered string.  For machine-readable formats the output
         is written directly to stdout and an empty string is returned.
     """
-    import sys
-
     safe_data = _redact_data(data)
     fmt = format_type.lower()
 

src/cleveragents/cli/render_specs.py

@@ -0,0 +1,96 @@
+"""Shared column and field specification types for CLI renderers.
+
+Extracted from ``renderers.py`` to keep that module under 500 lines.
+Consumers should import from here or from ``renderers`` (which
+re-exports everything).
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any
+
+from cleveragents.cli.formatting import OutputFormat
+
+if TYPE_CHECKING:
+    from rich.console import JustifyMethod
+
+__all__ = [
+    "FORMAT_HELP",
+    "ColumnSpec",
+    "FieldSpec",
+]
+
+# ---------------------------------------------------------------------------
+# Column / field specification types
+# ---------------------------------------------------------------------------
+
+
+@dataclass(frozen=True, slots=True)
+class ColumnSpec:
+    """Declares a column for ``render_list``.
+
+    Parameters
+    ----------
+    key:
+        The dict key in the data row (and the JSON/YAML field name).
+    label:
+        Human-readable column header shown in Rich/table/plain output.
+    style:
+        Rich style string for the column (e.g. ``"cyan"``).  Ignored
+        outside of ``rich`` format.
+    justify:
+        Column alignment (``"left"``, ``"center"``, ``"right"``).
+    truncate:
+        Maximum display width before truncation (``...``).
+        ``0`` means no truncation.
+    accessor:
+        Optional callable ``(row_dict) -> str`` for custom cell
+        rendering.  When ``None``, the value at ``row_dict[key]`` is
+        stringified directly.
+    """
+
+    key: str
+    label: str
+    style: str = ""
+    justify: JustifyMethod = "left"
+    truncate: int = 0
+    accessor: Callable[[dict[str, Any]], str] | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class FieldSpec:
+    """Declares a field for ``render_detail``.
+
+    Parameters
+    ----------
+    key:
+        The dict key (and JSON/YAML field name).
+    label:
+        Human-readable label for Rich/plain display.
+    multiline:
+        If ``True``, value is displayed on the next line (indented).
+    accessor:
+        Optional callable ``(data_dict) -> str`` for custom Rich
+        display.  When ``None``, ``str(data_dict[key])`` is used.
+    """
+
+    key: str
+    label: str
+    multiline: bool = False
+    truncate: int = 0
+    accessor: Callable[[dict[str, Any]], str] | None = None
+
+
+# ---------------------------------------------------------------------------
+# Shared format option
+# ---------------------------------------------------------------------------
+
+FORMAT_HELP: str = (
+    "Output format: json, yaml, plain, table, color, or rich (default: rich)"
+)
+"""Canonical ``--format`` help string.  Import this instead of defining
+a per-module ``_FORMAT_HELP``."""
+
+_VALID_FORMATS: set[str] = {fmt.value for fmt in OutputFormat}

src/cleveragents/cli/renderers.py

@@ -2,41 +2,86 @@
 
 Shared output functions for every CLI command: detail views, list views,
 errors, successes, warnings, and empty-result placeholders.  All respect
-the ``--format`` flag (rich/color/table/plain/json/yaml).
-
-Provides ``_get_console()`` and ``_get_err_console()`` for lazy-initialised
-shared Console instances, eliminating redundant module-level Console objects
-across command modules.
+the ``--format`` flag (rich/color/table/plain/json/yaml).  ``plain``
+output is strictly ASCII-only for log-pipeline compatibility.
 """
 
 from __future__ import annotations
 
+import json
+import logging
 from typing import Any
 
 from rich.console import Console
 from rich.markup import escape as _esc
+from rich.panel import Panel
+from rich.table import Table
 
 from cleveragents.cli.formatting import (
     OutputFormat,
     format_output,
 )
+from cleveragents.cli.formatting import (
+    ascii_safe as _format_ascii_safe,
+)
+from cleveragents.cli.render_specs import (
+    _VALID_FORMATS,
+    FORMAT_HELP,
+    ColumnSpec,
+    FieldSpec,
+)
+from cleveragents.shared.redaction import redact_dict
 
 __all__ = [
+    "FORMAT_HELP",
+    "ColumnSpec",
+    "FieldSpec",
     "_get_console",
-    "_get_err_console",
+    "get_console",
+    "render_detail",
     "render_empty",
     "render_error",
+    "render_list",
     "render_success",
     "render_warning",
 ]
 
 
+# Error envelope helper
+
+
+def _error_envelope(
+    code: str,
+    message: str,
+    recovery: str | None = None,
+    details: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Build a unified error envelope for JSON/YAML output.
+
+    Structure::
+
+        {
+          "error": {
+            "code": "<CODE>",
+            "message": "<message>",
+            "details": { ... },
+            "recovery": "<hint>"
+          }
+        }
+    """
+    inner: dict[str, Any] = {"code": code, "message": message}
+    inner["details"] = details if details else {}
+    if recovery:
+        inner["recovery"] = recovery
+    return {"error": inner}
+
+
 # Console helpers
 _console: Console | None = None
 _err_console: Console | None = None
 
 
-def _get_console() -> Console:
+def get_console() -> Console:
     """Return the shared stdout Rich console (created on first call)."""
     global _console
     if _console is None:
@@ -44,6 +89,10 @@ def _get_console() -> Console:
     return _console
 
 
+# Alias used by modules migrated via PR #1059.
+_get_console = get_console
+
+
 def _get_err_console() -> Console:
     """Return the shared stderr Rich console (created on first call)."""
     global _err_console
@@ -52,6 +101,212 @@ def _get_err_console() -> Console:
     return _err_console
 
 
+def _checkmark(value: bool, fmt: str) -> str:
+    """Return a check-mark or empty string appropriate for the format."""
+    if fmt == OutputFormat.RICH.value:
+        return "\u2713" if value else ""
+    # ASCII-safe for plain/table/color
+    return "yes" if value else "no"
+
+
+# Shared helpers
+_ACCESSOR_ERRORS = (KeyError, TypeError, AttributeError)
+"""Exception types caught when a ``ColumnSpec`` / ``FieldSpec`` accessor
+raises.  Kept narrow to avoid silently swallowing programming errors."""
+
+
+def _truncate_text(text: str, limit: int) -> str:
+    """Truncate *text* to *limit* chars without exceeding the limit."""
+    if limit <= 0 or len(text) <= limit:
+        return text
+    if limit < 4:
+        return text[:limit]
+    return text[: limit - 3] + "..."
+
+
+def _apply_accessor(
+    accessor: Any,
+    row: dict[str, Any],
+    key: str,
+) -> str:
+    """Invoke *accessor* on *row*, falling back to ``str(row[key])``."""
+    try:
+        result = accessor(row)
+    except _ACCESSOR_ERRORS:
+        logging.getLogger(__name__).debug(
+            "accessor %s raised",
+            key,
+            exc_info=True,
+        )
+        # Fall back to raw dict value instead of empty string
+        raw = row.get(key)
+        return str(raw) if raw is not None else ""
+    return str(result) if result is not None else ""
+
+
+# render_detail
+
+
+def render_detail(
+    data: dict[str, Any],
+    *,
+    title: str = "Details",
+    fields: list[FieldSpec] | None = None,
+    fmt: str = OutputFormat.RICH.value,
+    console: Console | None = None,
+) -> None:
+    """Render a key-value detail view.
+
+    Parameters
+    ----------
+    data:
+        Dict produced by the command's ``_*_spec_dict()`` helper.
+    title:
+        Panel title (rich) or section header (plain).
+    fields:
+        Ordered field specifications.  If ``None``, all keys in *data*
+        are rendered in insertion order.
+    fmt:
+        One of ``OutputFormat`` values.
+    """
+    fmt = fmt.lower()
+    if fmt not in _VALID_FORMATS:
+        raise ValueError(f"Unsupported format: {fmt!r}")
+    console = console or get_console()
+
+    if fmt != OutputFormat.RICH.value:
+        console.print(format_output(data, fmt))
+        return
+
+    # Rich panel -------------------------------------------------------
+    # Apply redaction consistently for rich format too (S3)
+    safe_data: dict[str, Any] = redact_dict(data)
+
+    if fields is None:
+        # Auto-generate from dict keys
+        lines: list[str] = []
+        for key, val in safe_data.items():
+            lines.append(f"[bold]{key}:[/bold] {_esc(str(val))}")
+        console.print(Panel("\n".join(lines), title=title, expand=False))
+        return
+
+    lines = []
+    for fs in fields:
+        val = safe_data.get(fs.key, "")
+        if fs.accessor:
+            display = _apply_accessor(fs.accessor, safe_data, fs.key)
+        else:
+            display = str(val)
+        if fs.truncate and len(display) > fs.truncate:
+            display = _truncate_text(display, fs.truncate)
+        # Skip multiline fields whose computed display is blank
+        if not display.strip() and fs.multiline:
+            continue
+        if fs.multiline:
+            lines.append(f"[bold]{fs.label}:[/bold]")
+            for sub_line in display.split("\n"):
+                lines.append(f"  {_esc(sub_line)}")
+        else:
+            lines.append(f"[bold]{fs.label}:[/bold] {_esc(display)}")
+    console.print(Panel("\n".join(lines), title=title, expand=False))
+
+
+# render_list
+
+
+def render_list(
+    rows: list[dict[str, Any]],
+    *,
+    columns: list[ColumnSpec],
+    title: str = "Results",
+    fmt: str = OutputFormat.RICH.value,
+    console: Console | None = None,
+) -> None:
+    """Render a tabular list view.
+
+    Column ordering is locked to the *columns* spec -- every invocation
+    with the same spec produces the same column order, styles, and
+    truncation rules.
+
+    Parameters
+    ----------
+    rows:
+        List of dicts (one per row).  Keys must match ``ColumnSpec.key``.
+    columns:
+        Ordered column specifications.
+    title:
+        Table title shown in rich format.
+    fmt:
+        Output format string.
+    """
+    fmt = fmt.lower()
+    if fmt not in _VALID_FORMATS:
+        raise ValueError(f"Unsupported format: {fmt!r}")
+    console = console or get_console()
+
+    if fmt != OutputFormat.RICH.value:
+        # Ensure stable field ordering -- re-order each dict to match
+        # the column spec so JSON/YAML/plain/table outputs have the same
+        # key order.  Invoke accessors so non-rich paths produce the
+        # same derived values as the rich table.
+        ordered: list[dict[str, Any]] = []
+        col_keys = [c.key for c in columns]
+        col_map = {c.key: c for c in columns}
+        for row in rows:
+            ordered_row: dict[str, Any] = {}
+            for key in col_keys:
+                col = col_map[key]
+                if col.accessor:
+                    ordered_row[key] = _apply_accessor(
+                        col.accessor,
+                        row,
+                        key,
+                    )
+                elif key in row:
+                    ordered_row[key] = row[key]
+            # Append any remaining keys not in columns (rare)
+            for key in row:
+                if key not in ordered_row:
+                    ordered_row[key] = row[key]
+            ordered.append(ordered_row)
+        console.print(format_output(ordered, fmt))
+        return
+
+    # Rich table -------------------------------------------------------
+    # Apply redaction consistently for rich format too (S3)
+    safe_rows: list[dict[str, Any]] = [
+        redact_dict(r) if isinstance(r, dict) else r for r in rows
+    ]
+
+    table = Table(title=f"{title} ({len(safe_rows)} total)")
+    for col in columns:
+        table.add_column(
+            col.label,
+            style=col.style,
+            justify=col.justify,
+        )
+
+    for row in safe_rows:
+        cells: list[str] = []
+        for col in columns:
+            if col.accessor:
+                cell = _apply_accessor(col.accessor, row, col.key)
+            else:
+                raw = row.get(col.key, "")
+                if isinstance(raw, bool):
+                    cell = _checkmark(raw, fmt)
+                elif isinstance(raw, (dict, list)):
+                    cell = json.dumps(raw, default=str)
+                else:
+                    cell = str(raw) if raw is not None else ""
+            if col.truncate and len(cell) > col.truncate:
+                cell = _truncate_text(cell, col.truncate)
+            cells.append(cell)
+        table.add_row(*cells)
+
+    console.print(table)
+
+
 # render_error
 
 
@@ -66,6 +321,11 @@ def render_error(
 ) -> None:
     """Render a uniform error message.
 
+    For ``json`` / ``yaml`` formats the error is wrapped in a unified
+    envelope::
+
+        {"error": {"code": "<label>", "message": "...", "recovery": "..."}}
+
     Parameters
     ----------
     label:
@@ -73,34 +333,30 @@ def render_error(
     message:
         Human-readable description of what went wrong.
     recovery:
-        Optional recovery hint.
+        Optional recovery hint (e.g. ``"Run 'agents init' first."``).
     details:
         Optional structured details dict for JSON/YAML.
     fmt:
         Output format string.
     """
+    fmt = fmt.lower()
+    if fmt not in _VALID_FORMATS:
+        raise ValueError(f"Unsupported format: {fmt!r}")
     console = console or _get_err_console()
 
     if fmt in (OutputFormat.JSON.value, OutputFormat.YAML.value):
-        envelope: dict[str, Any] = {
-            "error": {
-                "code": label,
-                "message": message,
-                "details": details or {},
-            }
-        }
-        if recovery:
-            envelope["error"]["recovery"] = recovery
+        envelope = _error_envelope(label, message, recovery=recovery, details=details)
         console.print(format_output(envelope, fmt))
         return
 
     if fmt in (OutputFormat.PLAIN.value, OutputFormat.TABLE.value):
-        console.print(f"ERROR: {label}: {message}")
+        console.print(_format_ascii_safe(f"ERROR: {label}: {message}"))
         if recovery:
-            console.print(recovery)
+            console.print(_format_ascii_safe(recovery))
         return
 
-    # Rich / Color
+    # Rich / Color — escape user-provided text to prevent
+    # accidental Rich markup injection (e.g. "[bold]" in error messages).
     console.print(f"[red]{_esc(label)}:[/red] {_esc(message)}")
     if recovery:
         console.print(f"[dim]{_esc(recovery)}[/dim]")
@@ -121,13 +377,17 @@ def render_success(
     Parameters
     ----------
     message:
-        Confirmation text.
+        Confirmation text (e.g. ``"Action archived: local/foo"``).
     fmt:
         Output format string.
     data:
-        Optional dict to render as JSON/YAML instead of the plain message.
+        Optional dict to render as JSON/YAML instead of the plain
+        message.  Ignored for non-structured formats.
     """
-    console = console or _get_console()
+    fmt = fmt.lower()
+    if fmt not in _VALID_FORMATS:
+        raise ValueError(f"Unsupported format: {fmt!r}")
+    console = console or get_console()
 
     if fmt in (OutputFormat.JSON.value, OutputFormat.YAML.value):
         payload = data if data is not None else {"status": "ok", "message": message}
@@ -135,7 +395,7 @@ def render_success(
         return
 
     if fmt == OutputFormat.PLAIN.value:
-        console.print(f"OK: {message}")
+        console.print(_format_ascii_safe(f"OK: {message}"))
         return
 
     console.print(f"[green]\u2713[/green] {_esc(message)}")
@@ -159,7 +419,10 @@ def render_warning(
     fmt:
         Output format string.
     """
-    console = console or _get_console()
+    fmt = fmt.lower()
+    if fmt not in _VALID_FORMATS:
+        raise ValueError(f"Unsupported format: {fmt!r}")
+    console = console or get_console()
 
     if fmt in (OutputFormat.JSON.value, OutputFormat.YAML.value):
         payload = {"status": "warning", "message": message}
@@ -167,7 +430,7 @@ def render_warning(
         return
 
     if fmt == OutputFormat.PLAIN.value:
-        console.print(f"WARNING: {message}")
+        console.print(_format_ascii_safe(f"WARNING: {message}"))
         return
 
     console.print(f"[yellow]{_esc(message)}[/yellow]")
@@ -183,6 +446,7 @@ def render_empty(
     recovery: str | None = None,
     fmt: str = OutputFormat.RICH.value,
     console: Console | None = None,
+    data: dict[str, Any] | None = None,
 ) -> None:
     """Render a "no items found" message with optional recovery hint.
 
@@ -196,18 +460,34 @@ def render_empty(
         Suggestion for what to do next.
     fmt:
         Output format string.
+    console:
+        Optional Rich console override (used by tests).
+    data:
+        Optional structured payload for machine-readable formats.
+        When provided, JSON/YAML/plain/table output uses *data*
+        instead of a bare empty list, so callers always receive a
+        typed object (e.g. ``{"sessions": [], "total": 0}``).
     """
-    console = console or _get_console()
+    fmt = fmt.lower()
+    if fmt not in _VALID_FORMATS:
+        raise ValueError(f"Unsupported format: {fmt!r}")
+    console = console or get_console()
     text = message or f"No {entity_type} found."
 
-    if fmt in (OutputFormat.JSON.value, OutputFormat.YAML.value):
-        console.print(format_output([], fmt))
-        return
-
-    if fmt in (OutputFormat.PLAIN.value, OutputFormat.TABLE.value):
-        console.print(text)
-        if recovery:
-            console.print(recovery)
+    if fmt in (
+        OutputFormat.JSON.value,
+        OutputFormat.YAML.value,
+        OutputFormat.PLAIN.value,
+        OutputFormat.TABLE.value,
+    ):
+        if data is not None:
+            console.print(format_output(data, fmt))
+        elif fmt in (OutputFormat.JSON.value, OutputFormat.YAML.value):
+            console.print(format_output([], fmt))
+        else:
+            console.print(_format_ascii_safe(text))
+            if recovery:
+                console.print(_format_ascii_safe(recovery))
         return
 
     console.print(f"[yellow]{_esc(text)}[/yellow]")