feat(cli): implement full output rendering framework (#812)
CI / build (push) Successful in 16s
CI / lint (push) Successful in 18s
CI / quality (push) Successful in 28s
CI / typecheck (push) Successful in 1m0s
CI / security (push) Successful in 1m0s
CI / benchmark-regression (push) Has been skipped
CI / unit_tests (push) Successful in 3m14s
CI / integration_tests (push) Successful in 3m38s
CI / docker (push) Successful in 55s
CI / e2e_tests (push) Successful in 5m15s
CI / coverage (push) Successful in 7m8s
CI / benchmark-publish (push) Successful in 20m34s

## Summary

Implement the 6 missing element handle types (Tree, Text, Code, Diff, Separator, ActionHint) and ensure all 6 materialization strategies (rich, color, table, plain, json, yaml) support all 10 element types. This completes the output rendering framework per the specification (§25417-27276).

Closes #550

## Changes

### Source (restructured into smaller modules)

#### `handles/` package (was `handles.py` — 1062 lines → 5 files, all ≤500 lines)
- **`_models.py`**: All Pydantic data models, constants (MAX_TREE_DEPTH, MAX_ELEMENTS_PER_SESSION, MAX_TABLE_ROWS), exceptions (ElementClosedError), event classes, ElementSnapshot union. P2-1: `DiffLine.type` renamed to `line_type` with backward-compatible alias.
- **`_base.py`**: Generic `ElementHandle[E]` base class with thread-safe `element_copy()` (lock-protected).
- **`_panel_table.py`**: PanelHandle, TableHandle, StatusHandle.
- **`_concrete.py`**: ProgressHandle (P1-1: validates `total` non-negative), TreeHandle, TextHandle (P3-2: close() delegates to super()), CodeHandle, DiffHandle, SeparatorHandle, ActionHintHandle. P3-8: `increment(delta=0)` now rejected.
- **`__init__.py`**: Re-exports all public symbols.

#### `_renderers.py` — plain renderers, sanitization, and shared helpers
- Plain render functions for all 10 element types.
- Terminal escape sanitization (`strip_terminal_escapes`).
- P2-3: `_sort_table_rows` now consults `ColumnDef.col_type` for numeric sorting.
- P2-6: `compute_column_widths()` shared helper extracted (DRY fix).

#### `_color_renderers.py` — ANSI colour renderers
- Colour-coded render functions for all 10 element types.
- P2-2: TextBlock now gets color treatment (`_render_text_color`) instead of falling through to plain.

#### `_boxdraw.py` — box-drawing renderers
- P2-5: Upgraded from ASCII `+-|` to Unicode box-drawing `╭─╮│╰╯` with rounded corners per spec §26821.

#### `_ids.py` — ID generation helpers
- P3-1: Session and handle IDs now use separate counters, making IDs monotonic within their namespace.

#### `materializers.py` — strategy protocol and 6 concrete strategies
- P1-2: `_snapshot_to_dict` now includes `timing` field in JSON/YAML output per spec §27022.
- P2-4: `_column_def_to_dict` always includes all fields unconditionally for stable JSON schemas.

#### `selection.py` — materializer selection with fallback
- P1-4: `NO_COLOR` environment variable now respected (https://no-color.org/). When set, all visual formats fall back to plain. Precedence: explicit flag > NO_COLOR > terminal capability fallback.

#### `session.py`
- P1-1: `session.progress()` factory validates `total >= 0`.
- P1-2: `snapshot()` includes `timing` when available.

#### `__init__.py` — package docstring
- P1-3: SD-29 corrected to reflect actual Table → Color → Plain fallback chain.
- SD-14 marked as implemented (NO_COLOR support added).

### Spec Deviations (Documented)
28 deliberate deviations documented in `__init__.py` module docstring (SD-1 through SD-29, with SD-14 now implemented). SD-29 corrected.

### Tests (updated)
- **+23 new BDD scenarios** covering: P1-1 total validation, P1-4 NO_COLOR, P2-1 DiffLine.line_type alias, P2-2 text color, P2-3 numeric sorting, P2-4 ColumnDef serialization, P2-5 Unicode box-drawing, P3-3 add_rows limit, P3-5 10-thread stress test, P3-6 summary truncation, P3-7 explicit format for color/table, P3-8 zero delta.
- **Robot tests**: Updated box-drawing assertion for Unicode chars.

## Verification

| Check | Result |
|-------|--------|
| Pyright | 0 errors, 1 pre-existing warning |
| Ruff lint | All passed |
| Unit tests | 393 features, 11,344 scenarios, 0 failures |
| Integration tests | All passed |
| E2E tests | All passed |
| Coverage | 97% overall (threshold: 97%) |

## Review Fixes Applied (Luis Review #2412)

| ID | Severity | Fix |
|----|----------|-----|
| P1-1 | High | `set_progress()` and `session.progress()` validate `total >= 0` |
| P1-2 | High | `_snapshot_to_dict` includes `timing` field |
| P1-3 | High | SD-29 documentation corrected |
| P1-4 | High | `NO_COLOR` env var respected |
| P2-1 | Medium | `DiffLine.type` → `line_type` with alias |
| P2-2 | Medium | TextBlock gets color treatment |
| P2-3 | Medium | Numeric column sorting |
| P2-4 | Medium | ColumnDef always serializes all fields |
| P2-5 | Medium | Unicode box-drawing characters |
| P2-6 | Medium | Shared `compute_column_widths()` helper |
| P3-1 | Low | Separate ID counters |
| P3-2 | Low | TextHandle.close() delegates to super() |
| P3-3 | Low | add_rows batch limit test |
| P3-5 | Low | 10-thread stress test |
| P3-6 | Low | Summary truncation test |
| P3-7 | Low | Explicit format tests for color/table |
| P3-8 | Low | Zero delta rejected |

### Deferred Items
| ID | Reason |
|----|--------|
| P3-9 | snapshot() lock scope — acceptable correctness trade-off |
| P3-10 | CLEVERAGENTS_FORMAT env var — documented as SD-15, requires CLI framework changes |

Reviewed-on: #812
Reviewed-by: Jeffrey Phillips Freeman <jeffrey.freeman@cleverthis.com>
Co-authored-by: Rui Hu <rui.hu@cleverthis.com>
Co-committed-by: Rui Hu <rui.hu@cleverthis.com>
This commit was merged in pull request #812.
This commit is contained in:
2026-03-19 08:46:46 +00:00
committed by Forgejo
parent 3837327564
commit 2434253c1a
21 changed files with 6081 additions and 1064 deletions
+197
View File
@@ -20,6 +20,7 @@ import cleveragents # noqa: E402
importlib.reload(cleveragents)
from cleveragents.cli.output.handles import DiffLine # noqa: E402
from cleveragents.cli.output.materializers import ( # noqa: E402
ColorMaterializer,
JsonMaterializer,
@@ -112,3 +113,199 @@ class SessionLifecycleSuite:
t.close()
s = session.status("ok")
s.close()
class TreeThroughputSuite:
"""Benchmark tree-building throughput with deep/wide trees."""
params: list[str] = ["plain", "json"] # noqa: RUF012
param_names: list[str] = ["format"] # noqa: RUF012
def time_wide_tree(self, fmt: str) -> None:
strategy = _STRATEGY_MAP[fmt]()
with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
tree = session.tree("Root")
for i in range(100):
tree.add_child("Root", f"child-{i}")
tree.close()
strategy.get_output()
def time_deep_tree(self, fmt: str) -> None:
strategy = _STRATEGY_MAP[fmt]()
with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
tree = session.tree("Root")
path = "Root"
for i in range(50):
label = f"level-{i}"
path = tree.add_child(path, label)
tree.close()
strategy.get_output()
class DiffThroughputSuite:
"""Benchmark diff rendering with multi-hunk diffs."""
params: list[str] = ["plain", "color", "json"] # noqa: RUF012
param_names: list[str] = ["format"] # noqa: RUF012
def time_multi_hunk_diff(self, fmt: str) -> None:
strategy = _STRATEGY_MAP[fmt]()
with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
diff = session.diff(file_a="old.py", file_b="new.py")
for h in range(20):
lines = []
for j in range(10):
lines.append(DiffLine(type="context", content=f"ctx-{j}"))
lines.append(DiffLine(type="add", content=f"add-{j}"))
lines.append(DiffLine(type="remove", content=f"rem-{j}"))
diff.add_hunk(f"@@ -{h * 30},{30} +{h * 30},{30} @@", lines)
diff.set_stats(200, 200)
diff.close()
strategy.get_output()
class MixedElementSuite:
"""Benchmark session throughput with all 10 element types."""
params: list[str] = ["plain", "json"] # noqa: RUF012
param_names: list[str] = ["format"] # noqa: RUF012
def time_all_element_types(self, fmt: str) -> None:
strategy = _STRATEGY_MAP[fmt]()
with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
p = session.panel("P")
p.set_entry("k", "v")
p.close()
t = session.table(columns=["A"])
t.add_row({"A": "1"})
t.close()
s = session.status("ok")
s.close()
pr = session.progress("go", total=10)
pr.set_progress(5)
pr.close()
tr = session.tree("Root")
tr.add_child("Root", "leaf")
tr.close()
tx = session.text("hello world")
tx.close()
co = session.code("x = 1", language="python")
co.close()
d = session.diff(file_a="a", file_b="b")
d.add_hunk("@@", [DiffLine(type="add", content="new")])
d.close()
session.separator("line")
session.action_hint(["cmd"])
strategy.get_output()
class TextAppendSuite:
"""Benchmark TextHandle.append() throughput (N10: detect O(N²) regressions)."""
params: list[str] = ["plain", "json"] # noqa: RUF012
param_names: list[str] = ["format"] # noqa: RUF012
def time_text_append_1000(self, fmt: str) -> None:
strategy = _STRATEGY_MAP[fmt]()
with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
tx = session.text("")
for i in range(1000):
tx.append(f"Line {i}\n")
tx.close()
strategy.get_output()
def time_text_append_5000(self, fmt: str) -> None:
strategy = _STRATEGY_MAP[fmt]()
with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
tx = session.text("")
for i in range(5000):
tx.append(f"Line {i}\n")
tx.close()
strategy.get_output()
class LargeTableSuite:
"""Benchmark table rendering at 10K+ rows (N10)."""
params: list[str] = ["plain", "json"] # noqa: RUF012
param_names: list[str] = ["format"] # noqa: RUF012
def time_table_10k_rows(self, fmt: str) -> None:
strategy = _STRATEGY_MAP[fmt]()
with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
table = session.table(columns=["ID", "Name", "Value"])
for i in range(10_000):
table.add_row({"ID": str(i), "Name": f"item-{i}", "Value": str(i * 10)})
table.close()
strategy.get_output()
class SnapshotCostSuite:
"""Benchmark snapshot() cost (N10)."""
params: list[str] = ["plain", "json"] # noqa: RUF012
param_names: list[str] = ["format"] # noqa: RUF012
def time_snapshot_with_many_elements(self, fmt: str) -> None:
strategy = _STRATEGY_MAP[fmt]()
with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
for i in range(50):
p = session.panel(f"Panel-{i}")
p.set_entry("k", "v")
p.close()
for _ in range(20):
session.snapshot()
def time_snapshot_with_large_table(self, fmt: str) -> None:
strategy = _STRATEGY_MAP[fmt]()
with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
table = session.table(columns=["A", "B"])
for i in range(5000):
table.add_row({"A": str(i), "B": str(i)})
table.close()
for _ in range(10):
session.snapshot()
class DeepTreeLookupSuite:
"""Benchmark deep tree path lookups (N10)."""
params: list[str] = ["plain", "json"] # noqa: RUF012
param_names: list[str] = ["format"] # noqa: RUF012
def time_deep_tree_add_and_lookup(self, fmt: str) -> None:
strategy = _STRATEGY_MAP[fmt]()
with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
tree = session.tree("Root")
path = "Root"
for i in range(50):
label = f"level-{i}"
path = tree.add_child(path, label)
# Now do lookups at various depths
tree.add_child("Root", "wide-child-1")
tree.add_child("Root", "wide-child-2")
tree.close()
strategy.get_output()
class ElementUpdatedOverheadSuite:
"""Benchmark ElementUpdated event dispatch overhead (N10)."""
params: list[str] = ["plain", "json"] # noqa: RUF012
param_names: list[str] = ["format"] # noqa: RUF012
def time_panel_many_updates(self, fmt: str) -> None:
strategy = _STRATEGY_MAP[fmt]()
with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
panel = session.panel("Update Panel")
for i in range(500):
panel.set_entry(f"key-{i % 50}", f"value-{i}")
panel.close()
def time_progress_many_ticks(self, fmt: str) -> None:
strategy = _STRATEGY_MAP[fmt]()
with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
prog = session.progress("Ticking", total=5000)
for _ in range(5000):
prog.tick()
prog.close()
@@ -108,15 +108,16 @@ Feature: Materializers coverage boost
Then the strategy output is empty
# ---------------------------------------------------------------
# Scenario 10: TableMaterializer renders non-table element via plain
# Targets line 327 (_render_element_table delegates to _render_element_plain)
# Scenario 10: TableMaterializer renders non-table element via color
# Targets _render_element_table delegates to render_element_color
# (R2-C2 fix #12: table format now uses color for non-table elements)
# ---------------------------------------------------------------
Scenario: Table materializer renders status message via plain fallback
Scenario: Table materializer renders status message via color fallback
Given a TableMaterializer buffer strategy
And an OutputSession wired to that table buffer strategy
When I create and close a status element with message "deploy complete" and level "ok"
Then the table strategy output contains "[OK] deploy complete"
Then the table strategy output contains "deploy complete"
# ---------------------------------------------------------------
# Scenario 11: Session end with empty buffers (no remaining)
File diff suppressed because it is too large Load Diff
@@ -18,6 +18,12 @@ import re
from behave import given, then, when
from behave.runner import Context
from cleveragents.cli.output._color_renderers import (
render_element_color as _render_element_color,
)
from cleveragents.cli.output._renderers import (
render_element_plain as _render_element_plain,
)
from cleveragents.cli.output.handles import (
ElementClosed,
Panel,
@@ -32,8 +38,6 @@ from cleveragents.cli.output.materializers import (
_AccumulateStrategy,
_BaseBufferStrategy,
_element_to_dict,
_render_element_color,
_render_element_plain,
)
from cleveragents.cli.output.session import OutputSession, StructuredOutput
File diff suppressed because it is too large Load Diff
+191 -2
View File
@@ -15,6 +15,7 @@ if _SRC not in sys.path:
sys.path.insert(0, _SRC)
from cleveragents.cli.formatting import format_output # noqa: E402
from cleveragents.cli.output.handles import DiffLine # noqa: E402
from cleveragents.cli.output.materializers import ( # noqa: E402
ColorMaterializer,
JsonMaterializer,
@@ -78,14 +79,16 @@ def test_yaml_output() -> None:
def test_table_boxdraw() -> None:
"""Table format uses ASCII box-drawing characters."""
"""Table format uses Unicode box-drawing characters (P2-5 upgrade)."""
strategy = TableMaterializer()
with OutputSession(format="table", command="test", strategy=strategy) as session:
table = session.table(columns=["Name", "Status"])
table.add_row({"Name": "svc", "Status": "up"})
table.close()
output = strategy.get_output()
assert "|" in output or "+" in output, "Table output should contain box chars"
assert "" in output or "" in output, (
"Table output should contain Unicode box chars"
)
assert "svc" in output
print("output-rendering-table-boxdraw-ok")
@@ -161,6 +164,186 @@ def test_selection_fallback() -> None:
print("output-rendering-selection-fallback-ok")
def test_tree_rendering() -> None:
"""Tree element renders across all formats."""
for fmt, cls in [("plain", PlainMaterializer), ("json", JsonMaterializer)]:
strategy = cls()
with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
tree = session.tree("Root")
tree.add_child("Root", "src")
tree.add_child("Root", "tests")
tree.add_child("Root/src", "main.py")
tree.close()
output = strategy.get_output()
assert "Root" in output, f"Tree root not in {fmt} output"
assert "src" in output, f"Tree child 'src' not in {fmt} output"
print("output-rendering-tree-ok")
def test_code_rendering() -> None:
"""Code element renders with line numbers."""
strategy = PlainMaterializer()
with OutputSession(format="plain", command="bench", strategy=strategy) as session:
code = session.code("x = 1\ny = 2", language="python", line_numbers=True)
code.close()
output = strategy.get_output()
assert "1 |" in output, "Code should have line numbers"
assert "x = 1" in output
print("output-rendering-code-ok")
def test_diff_rendering() -> None:
"""Diff element renders unified diff format."""
strategy = PlainMaterializer()
with OutputSession(format="plain", command="bench", strategy=strategy) as session:
diff = session.diff(file_a="old.py", file_b="new.py")
diff.add_hunk(
"@@ -1,3 +1,4 @@",
[
DiffLine(type="context", content="import os"),
DiffLine(type="remove", content="x = 1"),
DiffLine(type="add", content="x = 2"),
DiffLine(type="add", content="y = 3"),
],
)
diff.set_stats(2, 1)
diff.close()
output = strategy.get_output()
assert "--- old.py" in output
assert "+++ new.py" in output
assert "+x = 2" in output
assert "-x = 1" in output
assert "2 insertion(s)" in output
print("output-rendering-diff-ok")
def test_text_separator_hint() -> None:
"""Text, separator, and action hint render correctly."""
strategy = PlainMaterializer()
with OutputSession(format="plain", command="bench", strategy=strategy) as session:
text = session.text("Hello World", indent=2)
text.close()
session.separator("line")
session.action_hint(["agents list", "agents deploy"], "Try these:")
output = strategy.get_output()
assert "Hello World" in output
assert "----" in output
assert "agents list" in output
print("output-rendering-text-sep-hint-ok")
def test_json_all_elements() -> None:
"""JSON format serializes all 10 element types."""
strategy = JsonMaterializer()
with OutputSession(format="json", command="all", strategy=strategy) as session:
# Panel
p = session.panel("P")
p.set_entry("k", "v")
p.close()
# Table
t = session.table(columns=["A"])
t.add_row({"A": "1"})
t.close()
# Status
s = session.status("ok")
s.close()
# Progress
pr = session.progress("go", total=10)
pr.set_progress(5, 10)
pr.close()
# Tree
tr = session.tree("Root")
tr.add_child("Root", "leaf")
tr.close()
# Text
tx = session.text("hello")
tx.close()
# Code
co = session.code("x=1", language="py")
co.close()
# Diff
d = session.diff(file_a="a", file_b="b")
d.add_hunk("@@", [DiffLine(type="add", content="new")])
d.close()
# Separator
session.separator("line")
# ActionHint
session.action_hint(["cmd"])
output = strategy.get_output()
data = json.loads(output)
types = [e["type"] for e in data["elements"]]
expected = [
"panel",
"table",
"status",
"progress",
"tree",
"text",
"code",
"diff",
"separator",
"action_hint",
]
assert types == expected, f"Expected {expected}, got {types}"
print("output-rendering-json-all-ok")
def test_yaml_all_elements() -> None:
"""YAML format serializes all 10 element types (M5 fix — Luis review)."""
strategy = YamlMaterializer()
with OutputSession(format="yaml", command="all", strategy=strategy) as session:
# Panel
p = session.panel("P")
p.set_entry("k", "v")
p.close()
# Table
t = session.table(columns=["A"])
t.add_row({"A": "1"})
t.close()
# Status
s = session.status("ok")
s.close()
# Progress
pr = session.progress("go", total=10)
pr.set_progress(5, 10)
pr.close()
# Tree
tr = session.tree("Root")
tr.add_child("Root", "leaf")
tr.close()
# Text
tx = session.text("hello")
tx.close()
# Code
co = session.code("x=1", language="py")
co.close()
# Diff
d = session.diff(file_a="a", file_b="b")
d.add_hunk("@@", [DiffLine(type="add", content="new")])
d.close()
# Separator
session.separator("line")
# ActionHint
session.action_hint(["cmd"])
output = strategy.get_output()
data = yaml.safe_load(output)
types = [e["type"] for e in data["elements"]]
expected = [
"panel",
"table",
"status",
"progress",
"tree",
"text",
"code",
"diff",
"separator",
"action_hint",
]
assert types == expected, f"Expected {expected}, got {types}"
print("output-rendering-yaml-all-ok")
_DISPATCH = {
"plain-panel": test_plain_panel,
"json-output": test_json_output,
@@ -170,6 +353,12 @@ _DISPATCH = {
"json-error": test_json_error,
"backward-compat": test_backward_compat,
"selection-fallback": test_selection_fallback,
"tree": test_tree_rendering,
"code": test_code_rendering,
"diff": test_diff_rendering,
"text-sep-hint": test_text_separator_hint,
"json-all": test_json_all_elements,
"yaml-all": test_yaml_all_elements,
}
+57 -9
View File
@@ -10,7 +10,7 @@ ${HELPER} ${CURDIR}/helper_output_rendering.py
*** Test Cases ***
Plain Format Renders Panel Without ANSI Codes
[Documentation] Verify plain format produces ASCII-only panel output
${result}= Run Process ${PYTHON} ${HELPER} plain-panel cwd=${WORKSPACE}
${result}= Run Process ${PYTHON} ${HELPER} plain-panel cwd=${WORKSPACE} timeout=120s on_timeout=kill
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
@@ -18,7 +18,7 @@ Plain Format Renders Panel Without ANSI Codes
JSON Format Produces Valid JSON With Elements
[Documentation] Verify JSON format serialises elements as valid JSON
${result}= Run Process ${PYTHON} ${HELPER} json-output cwd=${WORKSPACE}
${result}= Run Process ${PYTHON} ${HELPER} json-output cwd=${WORKSPACE} timeout=120s on_timeout=kill
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
@@ -26,15 +26,15 @@ JSON Format Produces Valid JSON With Elements
YAML Format Produces Valid YAML With Elements
[Documentation] Verify YAML format serialises elements as valid YAML
${result}= Run Process ${PYTHON} ${HELPER} yaml-output cwd=${WORKSPACE}
${result}= Run Process ${PYTHON} ${HELPER} yaml-output cwd=${WORKSPACE} timeout=120s on_timeout=kill
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} output-rendering-yaml-output-ok
Table Format Renders Box Drawing Characters
[Documentation] Verify table format uses ASCII box-drawing for tables
${result}= Run Process ${PYTHON} ${HELPER} table-boxdraw cwd=${WORKSPACE}
[Documentation] Verify table format uses Unicode box-drawing for tables (P2-5)
${result}= Run Process ${PYTHON} ${HELPER} table-boxdraw cwd=${WORKSPACE} timeout=120s on_timeout=kill
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
@@ -42,7 +42,7 @@ Table Format Renders Box Drawing Characters
Color Format Includes ANSI Codes
[Documentation] Verify color format includes ANSI escape codes
${result}= Run Process ${PYTHON} ${HELPER} color-ansi cwd=${WORKSPACE}
${result}= Run Process ${PYTHON} ${HELPER} color-ansi cwd=${WORKSPACE} timeout=120s on_timeout=kill
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
@@ -50,7 +50,7 @@ Color Format Includes ANSI Codes
Error Envelope In JSON Mode
[Documentation] Verify error envelope structure in JSON mode
${result}= Run Process ${PYTHON} ${HELPER} json-error cwd=${WORKSPACE}
${result}= Run Process ${PYTHON} ${HELPER} json-error cwd=${WORKSPACE} timeout=120s on_timeout=kill
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
@@ -58,7 +58,7 @@ Error Envelope In JSON Mode
Format Output Backward Compatibility
[Documentation] Verify format_output still works for all formats
${result}= Run Process ${PYTHON} ${HELPER} backward-compat cwd=${WORKSPACE}
${result}= Run Process ${PYTHON} ${HELPER} backward-compat cwd=${WORKSPACE} timeout=120s on_timeout=kill
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
@@ -66,8 +66,56 @@ Format Output Backward Compatibility
Materializer Selection Fallback
[Documentation] Verify materializer selection with fallback logic
${result}= Run Process ${PYTHON} ${HELPER} selection-fallback cwd=${WORKSPACE}
${result}= Run Process ${PYTHON} ${HELPER} selection-fallback cwd=${WORKSPACE} timeout=120s on_timeout=kill
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} output-rendering-selection-fallback-ok
Tree Element Renders Across Formats
[Documentation] Verify tree element renders in plain and JSON formats
${result}= Run Process ${PYTHON} ${HELPER} tree cwd=${WORKSPACE} timeout=120s on_timeout=kill
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} output-rendering-tree-ok
Code Element Renders With Line Numbers
[Documentation] Verify code block renders with line numbers in plain format
${result}= Run Process ${PYTHON} ${HELPER} code cwd=${WORKSPACE} timeout=120s on_timeout=kill
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} output-rendering-code-ok
Diff Element Renders Unified Diff
[Documentation] Verify diff block renders unified diff format
${result}= Run Process ${PYTHON} ${HELPER} diff cwd=${WORKSPACE} timeout=120s on_timeout=kill
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} output-rendering-diff-ok
Text Separator And Action Hint Render
[Documentation] Verify text, separator, and action hint elements render
${result}= Run Process ${PYTHON} ${HELPER} text-sep-hint cwd=${WORKSPACE} timeout=120s on_timeout=kill
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} output-rendering-text-sep-hint-ok
JSON Serializes All Ten Element Types
[Documentation] Verify JSON format serializes all 10 element types
${result}= Run Process ${PYTHON} ${HELPER} json-all cwd=${WORKSPACE} timeout=120s on_timeout=kill
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} output-rendering-json-all-ok
YAML Serializes All Ten Element Types
[Documentation] Verify YAML format serializes all 10 element types (M5 fix)
${result}= Run Process ${PYTHON} ${HELPER} yaml-all cwd=${WORKSPACE} timeout=120s on_timeout=kill
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} output-rendering-yaml-all-ok
+183 -1
View File
@@ -4,6 +4,139 @@ This package implements the Output Rendering Framework described in the
specification. It decouples command output data from visual presentation
through a reactive session/handle/materialiser pipeline.
Spec Deviations
---------------
The following deliberate deviations from the specification (v3_spec.md
§Output Rendering Framework, lines 25417-27276) are documented here.
Each deviation was made for simplicity or because the feature is not
yet needed in the current milestone.
**SD-1 (ElementRenderer / RendererRegistry collapsed)**
The spec defines separate ``ElementRenderer`` protocol and
``RendererRegistry`` class. This implementation collapses rendering
into the strategy classes directly, which is simpler and sufficient
for the six built-in formats.
**SD-2 (RichMaterializer delegates to ColorMaterializer)**
The ``rich`` strategy does not use the Rich library for in-place
terminal updates. It delegates to the colour renderer, maintaining
forward compatibility. Full Rich integration is deferred to M8.
**SD-3 (No per-element priority filtering)**
Element ``priority`` and ``collapse_hint`` fields are stored but not
used for filtering or collapsing during rendering. The values are
preserved in JSON/YAML serialisation for future consumers.
**SD-4 (No real-time streaming in accumulate strategies)**
The ``json`` and ``yaml`` strategies accumulate all output and
serialise on session close. They do not stream partial results.
**SD-5 (No OutputTheme / style customisation)**
The spec mentions ``OutputTheme`` for user-customisable colours.
This implementation uses hardcoded ANSI codes.
**SD-6 (No RendererConfig / column width negotiation)**
Terminal width detection and column width negotiation are not
implemented. Plain and colour renderers use a fixed 80-column
assumption for text wrapping.
**SD-7 (Simplified Rich format)**
The ``rich`` format uses the same colour renderer as ``color``
instead of Rich library widgets (Live, Table, Tree, etc.).
**SD-8 (No format_output_streaming)**
The spec defines a streaming variant of format_output. Only the
batch ``format_output`` and ``format_output_session`` helpers exist.
**SD-9 (No element handle pooling)**
The spec mentions handle pooling for high-throughput scenarios.
Handles are created fresh each time.
**SD-10 (No incremental JSON/YAML)**
JSON/YAML strategies set ``supports_incremental_updates = False``.
Incremental update events are only used by future Rich strategy.
**SD-11 (Auto-close for Separator/ActionHint)**
The spec does not specify auto-close semantics for immutable
element types. This implementation auto-closes Separator and
ActionHint handles immediately after creation for convenience.
**SD-12 (MAX_ELEMENTS_PER_SESSION soft limit)**
The spec does not define an element count limit. A soft limit of
10,000 is enforced as a DoS guard.
**SD-13 (JSON/YAML envelope structure)**
The spec defines ``status``/``data`` wrapper, ``messages``, and
``timing`` fields in the serialised envelope. This implementation
serialises a flat ``StructuredOutput`` dict with ``command``,
``session_id``, ``exit_code``, ``elements``, and ``metadata``.
**SD-14 (NO_COLOR env var IMPLEMENTED)**
``NO_COLOR`` (https://no-color.org/) is now respected: when set,
all visual formats fall back to plain. Implemented per P1-4.
**SD-15 (CLEVERAGENTS_FORMAT env var not checked)**
The spec defines ``CLEVERAGENTS_FORMAT`` for overriding the default
format. This implementation only accepts ``--format`` CLI flag.
**SD-16 (TerminalCapabilities partial)**
The spec defines 11 capability fields. This implementation exposes
4: ``is_tty``, ``supports_ansi``, ``supports_cursor``, ``term``.
**SD-17 (ElementEvent field differences)**
``ElementEvent.element_kind`` is used instead of spec's
``element_type``. Timestamps use ``float`` (monotonic) instead of
``datetime``. ``session_id`` is not included in events.
**SD-18 (SessionEnd missing snapshot field)**
The spec's ``SessionEnd`` event includes a full session snapshot.
This implementation only carries ``exit_code``.
**SD-19 (MaterializationStrategy.bind() absent)**
The spec defines a ``bind()`` method for strategy configuration.
This implementation configures strategies via constructor arguments.
**SD-20 (on_session_begin missing stream parameter)**
The spec's ``on_session_begin`` accepts a ``stream: IO`` parameter.
This implementation passes the ``OutputSession`` reference instead.
**SD-21 (ColumnDef.type renamed to col_type)**
``ColumnDef.type`` was renamed to ``col_type`` to avoid shadowing
the Python builtin ``type()``.
**SD-22 (Sequential IDs instead of ULIDs)**
Session and handle IDs use sequential counters (``ses-000001``,
``hdl-000002``) instead of the spec's ULID format.
**SD-23 (No OutputSession.open() classmethod)**
The spec defines an ``OutputSession.open()`` class method. This
implementation uses ``OutputSession()`` constructor and context
manager protocol.
**SD-24 (No "closing" session state)**
The spec defines a ``closing`` intermediate state. This
implementation transitions directly from ``open`` to ``closed``.
**SD-25 (created_at is float, not datetime)**
``OutputSession.created_at`` uses ``time.time()`` (float epoch)
instead of the spec's ``datetime`` type.
**SD-26 (TreeHandle.add_child parent_path: str not str|None)**
The spec allows ``parent_path: str | None`` where ``None`` means
the root. This implementation uses empty string for the root.
**SD-27 (ActionHint.description: str="" not str|None)**
``ActionHint.description`` defaults to ``""`` instead of ``None``.
**SD-28 (CodeBlock.highlight_lines: list[int]=[] not list[int]|None)**
``CodeBlock.highlight_lines`` defaults to an empty list instead of
``None``.
**SD-29 (Table format fallback includes color tier)**
The ``table`` format fallback chain is ``Table Color Plain``
(matching the spec). Previously documented incorrectly as skipping
the color tier; corrected per P1-3 (Luis review).
Quick start::
from cleveragents.cli.output import OutputSession
@@ -14,20 +147,45 @@ Quick start::
panel.set_entry("Status", "active")
"""
from cleveragents.cli.output._renderers import strip_terminal_escapes
from cleveragents.cli.output.handles import (
MAX_ELEMENTS_PER_SESSION,
MAX_TABLE_ROWS,
MAX_TREE_DEPTH,
ActionHint,
ActionHintHandle,
CodeBlock,
CodeHandle,
ColumnDef,
DiffBlock,
DiffHandle,
DiffHunk,
DiffLine,
ElementClosed,
ElementClosedError,
ElementCreated,
ElementEvent,
ElementHandle,
ElementSnapshot,
ElementUpdated,
Panel,
PanelEntry,
PanelHandle,
ProgressHandle,
ProgressIndicator,
ProgressStep,
Separator,
SeparatorHandle,
SessionEnd,
StatusHandle,
StatusMessage,
Table,
TableHandle,
TextBlock,
TextHandle,
Tree,
TreeHandle,
TreeNode,
)
from cleveragents.cli.output.materializers import (
ColorMaterializer,
@@ -46,11 +204,26 @@ from cleveragents.cli.output.selection import (
from cleveragents.cli.output.session import OutputSession, StructuredOutput
__all__ = [
"MAX_ELEMENTS_PER_SESSION",
"MAX_TABLE_ROWS",
"MAX_TREE_DEPTH",
"ActionHint",
"ActionHintHandle",
"CodeBlock",
"CodeHandle",
"ColorMaterializer",
# Element snapshots
"ColumnDef",
"DiffBlock",
"DiffHandle",
"DiffHunk",
"DiffLine",
"ElementClosed",
"ElementClosedError",
"ElementCreated",
"ElementEvent",
"ElementHandle",
"ElementSnapshot",
"ElementUpdated",
"JsonMaterializer",
"MaterializationStrategy",
"OutputSession",
@@ -62,6 +235,9 @@ __all__ = [
"ProgressIndicator",
"ProgressStep",
"RichMaterializer",
"Separator",
"SeparatorHandle",
"SessionEnd",
"StatusHandle",
"StatusMessage",
"StructuredOutput",
@@ -69,7 +245,13 @@ __all__ = [
"TableHandle",
"TableMaterializer",
"TerminalCapabilities",
"TextBlock",
"TextHandle",
"Tree",
"TreeHandle",
"TreeNode",
"YamlMaterializer",
"detect_terminal_capabilities",
"select_materializer",
"strip_terminal_escapes",
]
+100
View File
@@ -0,0 +1,100 @@
"""Box-drawing (``table`` format) renderers for the output framework.
Unicode box-drawing renderers for the ``table`` materialisation strategy.
P2-5 fix (Luis review): upgraded from ASCII ``+-|`` to Unicode
box-drawing characters ```` with rounded corners, matching
the specification (§26821).
Extracted from ``_renderers.py`` for file-length compliance (R2-C2 fix).
"""
from __future__ import annotations
from cleveragents.cli.output._renderers import (
_sanitize,
_sort_table_rows,
compute_column_widths,
)
from cleveragents.cli.output.handles import (
ElementSnapshot,
Table,
)
# ---------------------------------------------------------------------------
# Box-drawing characters
# ---------------------------------------------------------------------------
_TOP_LEFT = ""
_TOP_RIGHT = ""
_BOTTOM_LEFT = ""
_BOTTOM_RIGHT = ""
_HORIZONTAL = ""
_VERTICAL = ""
_T_DOWN = ""
_T_UP = ""
_T_RIGHT = ""
_T_LEFT = ""
_CROSS = ""
def _render_table_boxdraw(table: Table) -> str:
"""Render a table using ASCII box-drawing (tabulate style).
R2-C2 fix: summary row is truncated to ``pad_width`` to prevent
the row from exceeding the table border width.
"""
if not table.columns:
return "(empty table)"
col_names = [_sanitize(c.name) for c in table.columns]
rows = _sort_table_rows(table)
str_rows: list[list[str]] = []
for row in rows:
str_rows.append([_sanitize(str(row.get(c.name, ""))) for c in table.columns])
widths = compute_column_widths(col_names, str_rows)
def make_border(left: str, mid: str, right: str) -> str:
return left + mid.join(_HORIZONTAL * (w + 2) for w in widths) + right
def make_row(cells: list[str]) -> str:
padded = [f" {c.ljust(w)} " for c, w in zip(cells, widths, strict=False)]
return _VERTICAL + _VERTICAL.join(padded) + _VERTICAL
lines: list[str] = []
if table.title:
lines.append(_sanitize(table.title))
lines.append(make_border(_TOP_LEFT, _T_DOWN, _TOP_RIGHT))
lines.append(make_row(col_names))
lines.append(make_border(_T_RIGHT, _CROSS, _T_LEFT))
for row in str_rows:
lines.append(make_row(row))
if table.summary:
lines.append(make_border(_T_RIGHT, _CROSS, _T_LEFT))
summary_str = " ".join(
f"{_sanitize(str(k))}: {_sanitize(str(v))}"
for k, v in table.summary.items()
)
inner_width = sum(w + 2 for w in widths) + (len(widths) - 1)
pad_width = inner_width - 1
# Truncate summary to fit within the table border (R2-C2 fix #11).
if len(summary_str) > pad_width:
summary_str = summary_str[:pad_width]
lines.append(f"{_VERTICAL} {summary_str.ljust(pad_width)}{_VERTICAL}")
lines.append(make_border(_BOTTOM_LEFT, _T_UP, _BOTTOM_RIGHT))
return "\n".join(lines)
def render_element_table(element: ElementSnapshot) -> str:
"""Dispatch for the ``table`` format (box-drawing for tables,
colour for colour-capable elements, colour fallback, plain fallback).
R2-C2 fix #12: colour-capable elements (Tree, Code, Diff) use the
colour renderer; remaining elements also try colour before plain.
"""
# Lazy import to avoid circular dependency through __init__.py.
from cleveragents.cli.output._color_renderers import render_element_color
if isinstance(element, Table):
return _render_table_boxdraw(element)
return render_element_color(element)
@@ -0,0 +1,262 @@
"""ANSI colour renderers for the output rendering framework.
Provides colour-coded rendering using ANSI escape sequences for all 10
element types. Extracted from ``_renderers.py`` for file-length
compliance (R2-C2 fix).
"""
from __future__ import annotations
from cleveragents.cli.output._renderers import (
_render_code_plain,
_render_progress_plain,
_render_tree_plain,
_sanitize,
_sort_table_rows,
compute_column_widths,
)
from cleveragents.cli.output.handles import (
ActionHint,
CodeBlock,
DiffBlock,
ElementSnapshot,
Panel,
ProgressIndicator,
Separator,
StatusMessage,
Table,
TextBlock,
Tree,
)
# ---------------------------------------------------------------------------
# ANSI colour codes
# ---------------------------------------------------------------------------
_BOLD = "\033[1m"
_CYAN = "\033[36m"
_GREEN = "\033[32m"
_YELLOW = "\033[33m"
_RED = "\033[31m"
_RESET = "\033[0m"
_DIM = "\033[2m"
# ---------------------------------------------------------------------------
# Color renderers
# ---------------------------------------------------------------------------
def _render_panel_color(panel: Panel) -> str:
"""Render a panel with ANSI colour codes."""
title = _sanitize(panel.title)
lines: list[str] = [f"{_BOLD}{_CYAN}{title}{_RESET}"]
for entry in panel.entries:
key = _sanitize(entry.key)
value = _sanitize(entry.value)
value_str = value
if entry.style_hint == "success":
value_str = f"{_GREEN}{value}{_RESET}"
elif entry.style_hint == "warning":
value_str = f"{_YELLOW}{value}{_RESET}"
elif entry.style_hint == "error":
value_str = f"{_RED}{value}{_RESET}"
lines.append(f" {_BOLD}{key}{_RESET}: {value_str}")
return "\n".join(lines)
def _render_table_color(table: Table) -> str:
"""Render a table with ANSI colour codes."""
if not table.columns:
return f"{_DIM}(empty table){_RESET}"
col_names = [_sanitize(c.name) for c in table.columns]
rows = _sort_table_rows(table)
str_rows: list[list[str]] = []
for row in rows:
str_rows.append([_sanitize(str(row.get(c.name, ""))) for c in table.columns])
widths = compute_column_widths(col_names, str_rows)
header = " ".join(
f"{_BOLD}{_CYAN}{n.ljust(w)}{_RESET}"
for n, w in zip(col_names, widths, strict=False)
)
sep = " ".join("-" * w for w in widths)
lines: list[str] = []
if table.title:
lines.append(f"{_BOLD}{_sanitize(table.title)}{_RESET}")
lines.append(header)
lines.append(sep)
for row in str_rows:
lines.append(" ".join(c.ljust(w) for c, w in zip(row, widths, strict=False)))
if table.summary:
lines.append(sep)
summary_parts: list[str] = []
for key, val in table.summary.items():
summary_parts.append(
f"{_BOLD}{_sanitize(str(key))}{_RESET}: {_sanitize(str(val))}"
)
lines.append(" ".join(summary_parts))
return "\n".join(lines)
def _render_status_color(status: StatusMessage) -> str:
"""Render a status message with ANSI colour codes."""
colour_map = {"ok": _GREEN, "warn": _YELLOW, "error": _RED, "info": _CYAN}
prefix_map = {
"ok": "[OK]",
"warn": "[WARN]",
"error": "[ERROR]",
"info": "[INFO]",
}
colour = colour_map.get(status.level, "")
prefix = prefix_map.get(status.level, f"[{status.level.upper()}]")
msg = _sanitize(status.message)
line = f"{colour}{_BOLD}{prefix}{_RESET} {msg}"
if status.detail:
line += f"\n {_DIM}{_sanitize(status.detail)}{_RESET}"
return line
def _render_progress_color(progress: ProgressIndicator) -> str:
"""Render a progress indicator with ANSI colour codes."""
plain = _render_progress_plain(progress)
label = _sanitize(progress.label)
lines = plain.split("\n", 1)
first_line = lines[0]
coloured_label = f"{_BOLD}{_CYAN}{label}{_RESET}"
if first_line.startswith(label):
first_line = coloured_label + first_line[len(label) :]
rest = f"\n{lines[1]}" if len(lines) > 1 else ""
return first_line + rest
def _render_tree_color(tree: Tree) -> str:
"""Render a tree with ANSI colour codes."""
plain = _render_tree_plain(tree)
root_label = _sanitize(tree.root.label)
coloured_label = f"{_BOLD}{_CYAN}{root_label}{_RESET}"
lines = plain.split("\n", 1)
first_line = lines[0]
idx = first_line.find(root_label)
if idx >= 0:
first_line = (
first_line[:idx] + coloured_label + first_line[idx + len(root_label) :]
)
rest = f"\n{lines[1]}" if len(lines) > 1 else ""
return first_line + rest
def _render_code_color(code: CodeBlock) -> str:
"""Render a code block with ANSI colour codes."""
lines: list[str] = []
if code.language:
lines.append(f"{_DIM}[{_sanitize(code.language)}]{_RESET}")
lines.append(_render_code_plain(code))
return "\n".join(lines)
def _render_diff_color(diff: DiffBlock) -> str:
"""Render a diff block with ANSI colour codes."""
lines: list[str] = []
if diff.file_a or diff.file_b:
lines.append(f"{_BOLD}--- {_sanitize(diff.file_a or '/dev/null')}{_RESET}")
lines.append(f"{_BOLD}+++ {_sanitize(diff.file_b or '/dev/null')}{_RESET}")
for hunk in diff.hunks:
lines.append(f"{_CYAN}{_sanitize(hunk.header)}{_RESET}")
for dline in hunk.lines:
content = _sanitize(dline.content)
if dline.line_type == "add":
lines.append(f"{_GREEN}+{content}{_RESET}")
elif dline.line_type == "remove":
lines.append(f"{_RED}-{content}{_RESET}")
else:
lines.append(f" {content}")
if diff.stats:
ins = diff.stats.get("insertions", 0)
dels = diff.stats.get("deletions", 0)
lines.append(
f"{_GREEN}{ins} insertion(s){_RESET}, {_RED}{dels} deletion(s){_RESET}"
)
return "\n".join(lines)
def _render_separator_color(sep: Separator) -> str:
"""Render a separator with ANSI colour codes."""
if sep.style == "blank":
return ""
line_char = "=" if sep.style == "double" else "-"
return f"{_DIM}{line_char * 40}{_RESET}"
def _render_action_hint_color(hint: ActionHint) -> str:
"""Render an action hint with ANSI colour codes."""
lines: list[str] = []
if hint.description:
lines.append(f"{_DIM}{_sanitize(hint.description)}{_RESET}")
if hint.commands:
lines.append(f"{_BOLD}Next steps:{_RESET}")
for cmd in hint.commands:
lines.append(f" $ {_CYAN}{_sanitize(cmd)}{_RESET}")
return "\n".join(lines)
def _render_text_color(text: TextBlock) -> str:
"""Render a text block with ANSI colour codes.
P2-2 fix (Luis review): TextBlock now gets colour treatment instead
of falling through to the plain renderer. The first line gets bold
formatting and indent markers use dim styling.
"""
content = _sanitize(text.content)
indent_str = " " * text.indent
if text.wrap:
import textwrap
width = max(80 - text.indent, 20)
wrapped_lines: list[str] = []
for line in content.split("\n"):
if line:
wrapped_lines.append(
textwrap.fill(
line,
width=width,
initial_indent=f"{_DIM}{indent_str}{_RESET}",
subsequent_indent=f"{_DIM}{indent_str}{_RESET}",
)
)
else:
wrapped_lines.append("")
return "\n".join(wrapped_lines)
return "\n".join(
f"{_DIM}{indent_str}{_RESET}{line}" for line in content.split("\n")
)
def render_element_color(element: ElementSnapshot) -> str:
"""Dispatch to the correct colour renderer."""
if isinstance(element, Panel):
return _render_panel_color(element)
if isinstance(element, Table):
return _render_table_color(element)
if isinstance(element, StatusMessage):
return _render_status_color(element)
if isinstance(element, ProgressIndicator):
return _render_progress_color(element)
if isinstance(element, Tree):
return _render_tree_color(element)
if isinstance(element, TextBlock):
return _render_text_color(element)
if isinstance(element, CodeBlock):
return _render_code_color(element)
if isinstance(element, DiffBlock):
return _render_diff_color(element)
if isinstance(element, Separator):
return _render_separator_color(element)
if isinstance(element, ActionHint):
return _render_action_hint_color(element)
return _sanitize(str(element)) # pragma: no cover
+44
View File
@@ -0,0 +1,44 @@
"""Session and handle ID generation helpers.
Provides thread-safe sequential ID generation for OutputSession and
ElementHandle instances. Extracted from ``session.py`` for file-length
compliance (R2-C2 fix).
P3-1 fix (Luis review): session and handle IDs now use separate
counters so that IDs are monotonic within their own namespace.
"""
from __future__ import annotations
import threading
_SESSION_LOCK = threading.Lock()
_SESSION_COUNTER = 0
_HANDLE_LOCK = threading.Lock()
_HANDLE_COUNTER = 0
def generate_session_id() -> str:
"""Generate a unique session identifier."""
global _SESSION_COUNTER
with _SESSION_LOCK:
_SESSION_COUNTER += 1
return f"ses-{_SESSION_COUNTER:06d}"
def generate_handle_id() -> str:
"""Generate a unique handle identifier."""
global _HANDLE_COUNTER
with _HANDLE_LOCK:
_HANDLE_COUNTER += 1
return f"hdl-{_HANDLE_COUNTER:06d}"
def _reset_counters() -> None:
"""Reset both counters to zero (test helper only)."""
global _SESSION_COUNTER, _HANDLE_COUNTER
with _SESSION_LOCK:
_SESSION_COUNTER = 0
with _HANDLE_LOCK:
_HANDLE_COUNTER = 0
+420
View File
@@ -0,0 +1,420 @@
"""Plain-text render functions for the output rendering framework.
Plain renderers for all 10 element types and the terminal escape
sanitiser. Each renderer sanitizes user-supplied strings via
``_sanitize`` to prevent terminal escape injection (CWE-150, CWE-116).
Colour renderers live in ``_color_renderers.py``; box-drawing renderers
live in ``_boxdraw.py``. Both were extracted from this file for
file-length compliance (R2-C2 fix).
"""
from __future__ import annotations
import re
import textwrap
from typing import Any
from cleveragents.cli.output.handles import (
MAX_TREE_DEPTH,
ActionHint,
CodeBlock,
DiffBlock,
ElementSnapshot,
Panel,
ProgressIndicator,
Separator,
StatusMessage,
Table,
TextBlock,
Tree,
TreeNode,
)
# ---------------------------------------------------------------------------
# Terminal escape sanitization (C2 — CWE-150, CWE-116)
# ---------------------------------------------------------------------------
# Matches ANSI/VT escape sequences including:
# - 7-bit CSI (ESC [ params final) with private-mode parameter bytes
# - 8-bit CSI (0x9B params final)
# - OSC terminated by BEL or ST (ESC \)
# - Character set designation (ESC ( letter)
# - Miscellaneous single-char ESC commands
_TERMINAL_ESCAPE_RE = re.compile(
r"(?:"
r"\x1b\[[\x20-\x3f]*[\x40-\x7e]" # 7-bit CSI: ESC [ params final
r"|\x9b[\x20-\x3f]*[\x40-\x7e]" # 8-bit CSI: 0x9b params final
r"|\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)" # OSC: ESC ] ... (BEL | ST)
r"|\x1b\([A-Za-z]" # Charset designation: ESC ( letter
r"|\x1b[><=]" # Misc single-char: ESC> ESC< ESC=
r")"
)
# Control characters and 8-bit C1 control characters (0x80-0x9F),
# excluding tab (0x09), newline (0x0A), carriage return (0x0D).
_CONTROL_CHAR_RE = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]")
def strip_terminal_escapes(text: str) -> str:
"""Remove ANSI/VT escape sequences and dangerous control characters.
This prevents user-supplied strings from injecting terminal commands
such as title changes, cursor movement, or fake styling.
Applied to ALL user-supplied text at the rendering boundary in every
materializer (plain, color, table, rich). JSON/YAML serializers
output structured data and do not need terminal sanitization.
"""
text = _TERMINAL_ESCAPE_RE.sub("", text)
text = _CONTROL_CHAR_RE.sub("", text)
return text
# Readable shorthand used throughout the render functions.
_sanitize = strip_terminal_escapes
# ---------------------------------------------------------------------------
# Shared helpers
# ---------------------------------------------------------------------------
def _numeric_sort_key(val: Any) -> float:
"""Parse a numeric value for sorting; non-numeric values sort last."""
try:
return float(val)
except (ValueError, TypeError):
return float("inf")
def _sort_table_rows(table: Table) -> list[dict[str, Any]]:
"""Return rows sorted by sort_key if set, otherwise in insertion order.
N7 fix: sort_key/sort_descending were stored but never applied.
L2 fix (Luis review): always return a new list so callers never
accidentally mutate the internal ``table.rows`` reference.
P2-3 fix (Luis review): consults ``ColumnDef.col_type`` columns
with ``col_type="number"`` sort numerically instead of
lexicographically (``"9" < "10"`` instead of ``"9" > "10"``).
"""
if not table.sort_key:
return list(table.rows)
key = table.sort_key
# P2-3: resolve column type for sort-key-aware comparison.
col_type = "string"
for col in table.columns:
if col.name == key:
col_type = col.col_type
break
try:
if col_type == "number":
return sorted(
table.rows,
key=lambda r: _numeric_sort_key(r.get(key, "")),
reverse=table.sort_descending,
)
return sorted(
table.rows,
key=lambda r: str(r.get(key, "")),
reverse=table.sort_descending,
)
except TypeError:
return list(table.rows) # pragma: no cover
def compute_column_widths(col_names: list[str], str_rows: list[list[str]]) -> list[int]:
"""Compute the display width for each column.
P2-6 fix (Luis review): extracted from three renderers (plain,
color, boxdraw) to a single shared helper, eliminating duplication.
"""
widths = [len(c) for c in col_names]
for row in str_rows:
for i, cell in enumerate(row):
if i < len(widths):
widths[i] = max(widths[i], len(cell))
return widths
# ---------------------------------------------------------------------------
# Plain renderers
# ---------------------------------------------------------------------------
def _render_panel_plain(panel: Panel) -> str:
"""Render a panel as plain ASCII text."""
lines: list[str] = [_sanitize(panel.title)]
for entry in panel.entries:
lines.append(f" {_sanitize(entry.key)}: {_sanitize(entry.value)}")
return "\n".join(lines)
def _render_table_plain(table: Table) -> str:
"""Render a table as plain ASCII text with aligned columns."""
if not table.columns:
return "(empty table)"
col_names = [_sanitize(c.name) for c in table.columns]
rows = _sort_table_rows(table)
str_rows: list[list[str]] = []
for row in rows:
str_rows.append([_sanitize(str(row.get(c.name, ""))) for c in table.columns])
widths = compute_column_widths(col_names, str_rows)
header = " ".join(n.ljust(w) for n, w in zip(col_names, widths, strict=False))
sep = " ".join("-" * w for w in widths)
lines: list[str] = []
if table.title:
lines.append(_sanitize(table.title))
lines.append(header)
lines.append(sep)
for row in str_rows:
lines.append(" ".join(c.ljust(w) for c, w in zip(row, widths, strict=False)))
if table.summary:
lines.append(sep)
summary_parts: list[str] = []
for key, val in table.summary.items():
summary_parts.append(f"{_sanitize(str(key))}: {_sanitize(str(val))}")
lines.append(" ".join(summary_parts))
return "\n".join(lines)
def _render_status_plain(status: StatusMessage) -> str:
"""Render a status message as plain ASCII."""
prefix_map = {
"ok": "[OK]",
"warn": "[WARN]",
"error": "[ERROR]",
"info": "[INFO]",
}
prefix = prefix_map.get(status.level, f"[{status.level.upper()}]")
line = f"{prefix} {_sanitize(status.message)}"
if status.detail:
line += f"\n {_sanitize(status.detail)}"
return line
def _render_progress_plain(progress: ProgressIndicator) -> str:
"""Render a progress indicator as plain ASCII."""
label = _sanitize(progress.label)
lines: list[str] = []
if progress.indeterminate:
lines.append(f"{label}: ...")
elif progress.total is not None and progress.current is not None:
pct = int(progress.current * 100 / progress.total) if progress.total > 0 else 0
lines.append(f"{label}: {progress.current}/{progress.total} ({pct}%)")
elif progress.current is not None:
lines.append(f"{label}: {progress.current}")
else:
lines.append(f"{label}: pending")
if progress.steps:
for step in progress.steps:
marker_map = {
"done": "[x]",
"active": "[>]",
"pending": "[ ]",
"error": "[!]",
"skipped": "[-]",
}
marker = marker_map.get(step.status, "[ ]")
lines.append(f" {marker} {_sanitize(step.label)}")
return "\n".join(lines)
def _render_tree_plain(tree: Tree) -> str:
"""Render a tree as plain ASCII text with ``+--`` and ``|`` guides.
Uses an output accumulator list (N9 fix) and a depth guard (F4 fix)
to prevent O(N^2) list copies and unbounded recursion.
"""
max_render_depth = MAX_TREE_DEPTH
def _walk(
node: TreeNode,
prefix: str,
is_last: bool,
depth: int,
output: list[str],
) -> None:
connector = "+-- " if depth > 0 else ""
output.append(f"{prefix}{connector}{_sanitize(node.label)}")
if depth >= max_render_depth:
if node.children:
child_prefix = prefix + (" " if is_last else "| ")
count = len(node.children)
output.append(
f"{child_prefix}... ({count} more "
f"{'children' if count != 1 else 'child'})"
)
return
if tree.max_depth_hint is not None and depth >= tree.max_depth_hint:
if node.children:
child_prefix = prefix + (" " if is_last else "| ")
count = len(node.children)
output.append(
f"{child_prefix}... ({count} more "
f"{'children' if count != 1 else 'child'})"
)
return
for i, child in enumerate(node.children):
last = i == len(node.children) - 1
child_prefix = prefix + (" " if is_last else "| ") if depth > 0 else ""
# L3 fix (Luis review): skip collapsed subtrees.
if child.collapsed:
output.append(f"{child_prefix}+-- {_sanitize(child.label)} [collapsed]")
continue
_walk(child, child_prefix, last, depth + 1, output)
if not tree.show_guides:
# Flat indented rendering without guide characters.
effective_max = max_render_depth
if tree.max_depth_hint is not None:
effective_max = min(effective_max, tree.max_depth_hint)
def _flat(node: TreeNode, indent: int, depth: int, output: list[str]) -> None:
output.append(" " * indent + _sanitize(node.label))
if depth >= effective_max:
if node.children:
count = len(node.children)
output.append(
" " * (indent + 2)
+ f"... ({count} more "
+ f"{'children' if count != 1 else 'child'})"
)
return
for child in node.children:
# L3 fix: skip collapsed subtrees in flat mode too.
if child.collapsed:
output.append(
" " * (indent + 2) + _sanitize(child.label) + " [collapsed]"
)
continue
_flat(child, indent + 2, depth + 1, output)
out: list[str] = []
_flat(tree.root, 0, 0, out)
return "\n".join(out)
out_lines: list[str] = []
_walk(tree.root, "", True, 0, out_lines)
return "\n".join(out_lines)
def _render_text_plain(text: TextBlock) -> str:
"""Render a text block as plain ASCII.
When ``wrap`` is True, applies word-wrapping at 80 columns minus
indent (N6 fix). Splits on explicit newlines first so that
``textwrap.fill`` does not collapse them (R2-C2 fix #9).
When False, preserves content verbatim with indent.
"""
indent_str = " " * text.indent
content = _sanitize(text.content)
if text.wrap:
width = max(80 - text.indent, 20)
# Wrap each line independently to preserve explicit newlines.
wrapped_lines: list[str] = []
for line in content.split("\n"):
if line:
wrapped_lines.append(
textwrap.fill(
line,
width=width,
initial_indent=indent_str,
subsequent_indent=indent_str,
)
)
else:
wrapped_lines.append("")
return "\n".join(wrapped_lines)
return "\n".join(f"{indent_str}{line}" for line in content.split("\n"))
def _render_code_plain(code: CodeBlock) -> str:
"""Render a code block as plain ASCII with optional line numbers."""
content = _sanitize(code.content)
lines = content.split("\n")
if code.line_numbers:
width = len(str(len(lines)))
numbered: list[str] = []
for i, line in enumerate(lines, 1):
numbered.append(f"{i:>{width}} | {line}")
return "\n".join(numbered)
return content
def _render_diff_plain(diff: DiffBlock) -> str:
"""Render a diff block as plain unified diff."""
lines: list[str] = []
if diff.file_a or diff.file_b:
lines.append(f"--- {_sanitize(diff.file_a or '/dev/null')}")
lines.append(f"+++ {_sanitize(diff.file_b or '/dev/null')}")
for hunk in diff.hunks:
lines.append(_sanitize(hunk.header))
for dline in hunk.lines:
prefix_map = {"add": "+", "remove": "-", "context": " "}
prefix = prefix_map.get(dline.line_type, " ")
lines.append(f"{prefix}{_sanitize(dline.content)}")
if diff.stats:
ins = diff.stats.get("insertions", 0)
dels = diff.stats.get("deletions", 0)
lines.append(f"{ins} insertion(s), {dels} deletion(s)")
return "\n".join(lines)
def _render_separator_plain(sep: Separator) -> str:
"""Render a separator as plain ASCII."""
if sep.style == "blank":
return ""
if sep.style == "double":
return "=" * 40
return "-" * 40
def _render_action_hint_plain(hint: ActionHint) -> str:
"""Render an action hint as plain ASCII."""
lines: list[str] = []
if hint.description:
lines.append(_sanitize(hint.description))
if hint.commands:
lines.append("Next steps:")
for cmd in hint.commands:
lines.append(f" $ {_sanitize(cmd)}")
return "\n".join(lines)
def render_element_plain(element: ElementSnapshot) -> str:
"""Dispatch to the correct plain renderer.
All individual renderers apply sanitization to user strings internally.
The fallback path also sanitizes.
"""
if isinstance(element, Panel):
return _render_panel_plain(element)
if isinstance(element, Table):
return _render_table_plain(element)
if isinstance(element, StatusMessage):
return _render_status_plain(element)
if isinstance(element, ProgressIndicator):
return _render_progress_plain(element)
if isinstance(element, Tree):
return _render_tree_plain(element)
if isinstance(element, TextBlock):
return _render_text_plain(element)
if isinstance(element, CodeBlock):
return _render_code_plain(element)
if isinstance(element, DiffBlock):
return _render_diff_plain(element)
if isinstance(element, Separator):
return _render_separator_plain(element)
if isinstance(element, ActionHint):
return _render_action_hint_plain(element)
return _sanitize(str(element)) # pragma: no cover
-591
View File
@@ -1,591 +0,0 @@
"""Element handle classes for the output rendering framework.
Each handle type wraps a specific element kind and provides typed write
methods. Handles are format-agnostic -- producer code writes to handles
without knowledge of the active materialisation strategy.
Based on the Output Rendering Framework specification (v3_spec.md §Output
Rendering Framework / Element Handles).
"""
from __future__ import annotations
import threading
import time
from typing import TYPE_CHECKING, Any
from pydantic import BaseModel, Field
if TYPE_CHECKING:
from cleveragents.cli.output.session import OutputSession
# ---------------------------------------------------------------------------
# Exceptions
# ---------------------------------------------------------------------------
class ElementClosedError(RuntimeError):
"""Raised when a write method is called on a closed handle."""
# ---------------------------------------------------------------------------
# Element snapshot data classes
# ---------------------------------------------------------------------------
class PanelEntry(BaseModel):
"""A single key-value pair within a Panel."""
key: str
value: str
style_hint: str | None = None
icon: str | None = None
class Panel(BaseModel):
"""A titled group of key-value pairs."""
title: str
entries: list[PanelEntry] = Field(default_factory=list)
border_style: str = "rounded"
priority: str = "normal"
collapse_hint: str = "auto"
metadata: dict[str, Any] = Field(default_factory=dict)
element_type: str = "panel"
class ColumnDef(BaseModel):
"""Schema for a single table column."""
name: str
col_type: str = "string"
alignment: str = "left"
width_hint: int | None = None
sortable: bool = False
style_hint: str | None = None
class Table(BaseModel):
"""A tabular data set with typed columns."""
title: str | None
columns: list[ColumnDef] = Field(default_factory=list)
rows: list[dict[str, Any]] = Field(default_factory=list)
summary: dict[str, Any] | None = None
max_rows_hint: int | None = None
sort_key: str | None = None
priority: str = "normal"
collapse_hint: str = "auto"
metadata: dict[str, Any] = Field(default_factory=dict)
element_type: str = "table"
class StatusMessage(BaseModel):
"""A status line (success, warning, error, info)."""
message: str
level: str = "info"
detail: str | None = None
priority: str = "normal"
metadata: dict[str, Any] = Field(default_factory=dict)
element_type: str = "status"
class ProgressStep(BaseModel):
"""A named step within a progress indicator."""
label: str
status: str = "pending"
class ProgressIndicator(BaseModel):
"""A progress bar or spinner for long-running operations."""
label: str
current: int | None = None
total: int | None = None
indeterminate: bool = False
steps: list[ProgressStep] | None = None
priority: str = "normal"
metadata: dict[str, Any] = Field(default_factory=dict)
element_type: str = "progress"
# Union of all element snapshot types used by handles.
ElementSnapshot = Panel | Table | StatusMessage | ProgressIndicator
# ---------------------------------------------------------------------------
# Element events
# ---------------------------------------------------------------------------
class ElementEvent(BaseModel):
"""Base class for all events emitted by element handles."""
event_type: str
handle_id: str
element_kind: str
timestamp: float = Field(default_factory=time.monotonic)
class ElementCreated(ElementEvent):
"""Emitted when a new element handle is created."""
declaration_index: int = 0
initial_state: ElementSnapshot | None = None
class ElementUpdated(ElementEvent):
"""Emitted when data is written to an element handle."""
update_type: str = ""
delta: dict[str, Any] = Field(default_factory=dict)
element_snapshot: ElementSnapshot | None = None
class ElementClosed(ElementEvent):
"""Emitted when an element handle is closed."""
final_state: ElementSnapshot | None = None
class SessionEnd(ElementEvent):
"""Emitted when the session itself is closed."""
exit_code: int = 0
# ---------------------------------------------------------------------------
# Base ElementHandle
# ---------------------------------------------------------------------------
class ElementHandle:
"""Base class for all element handles.
An element handle is a write-only view of an output element. Producers
use handles to incrementally build element content without knowledge of
the active format or materialisation strategy.
Thread Safety
-------------
Individual handle methods are thread-safe. Concurrent writes to the
*same* handle are serialised via an internal lock.
Lifecycle
---------
``handle = session.panel(...)`` -- Created by session factory
``handle.set_entry(...)`` -- Write operations (zero or more)
``handle.close()`` -- Finalise (required unless auto-closed)
"""
def __init__(
self,
handle_id: str,
element_type: str,
declaration_index: int,
session: OutputSession,
element: ElementSnapshot,
) -> None:
if not handle_id:
raise ValueError("handle_id cannot be empty")
if not element_type:
raise ValueError("element_type cannot be empty")
if declaration_index < 0:
raise ValueError("declaration_index must be non-negative")
if session is None:
raise ValueError("session cannot be None")
if element is None:
raise ValueError("element cannot be None")
self.handle_id = handle_id
self.element_type = element_type
self.declaration_index = declaration_index
self._session: OutputSession = session
self._element: ElementSnapshot = element
self._state: str = "open"
self._lock: threading.Lock = threading.Lock()
def _require_open(self) -> None:
"""Raise if handle is closed."""
if self._state != "open":
raise ElementClosedError(
f"Handle {self.handle_id!r} ({self.element_type}) is closed"
)
def _emit_update(self, update_type: str, delta: dict[str, Any]) -> None:
"""Emit an ElementUpdated event to the session."""
event = ElementUpdated(
event_type="updated",
handle_id=self.handle_id,
element_kind=self.element_type,
update_type=update_type,
delta=delta,
element_snapshot=self._element,
)
self._session._dispatch_event(event)
@property
def is_open(self) -> bool:
"""Whether this handle is still accepting writes."""
return self._state == "open"
@property
def element(self) -> ElementSnapshot:
"""The accumulated element state (read-only snapshot)."""
return self._element
def close(self) -> None:
"""Close this handle, signalling that no more data will be written."""
with self._lock:
if self._state == "open":
self._state = "closed"
event = ElementClosed(
event_type="closed",
handle_id=self.handle_id,
element_kind=self.element_type,
final_state=self._element,
)
self._session._dispatch_event(event)
def __enter__(self) -> ElementHandle:
return self
def __exit__(
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: Any,
) -> None:
if self.is_open:
self.close()
# ---------------------------------------------------------------------------
# Concrete handle subclasses
# ---------------------------------------------------------------------------
class PanelHandle(ElementHandle):
"""Handle for building a Panel element incrementally."""
def __init__(
self,
handle_id: str,
declaration_index: int,
session: OutputSession,
element: Panel,
) -> None:
super().__init__(handle_id, "panel", declaration_index, session, element)
def set_entry(
self,
key: str,
value: str,
*,
style_hint: str | None = None,
icon: str | None = None,
) -> None:
"""Set or update a key-value entry in the panel."""
if not key:
raise ValueError("key cannot be empty")
with self._lock:
self._require_open()
panel: Panel = self._element # type: ignore[assignment]
for entry in panel.entries:
if entry.key == key:
entry.value = value
entry.style_hint = style_hint
entry.icon = icon
break
else:
panel.entries.append(
PanelEntry(key=key, value=value, style_hint=style_hint, icon=icon)
)
self._emit_update("entry_set", {"key": key, "value": value})
def set_entries(
self,
entries: dict[str, str],
*,
style_hints: dict[str, str] | None = None,
) -> None:
"""Set multiple entries at once (batch update)."""
if entries is None:
raise ValueError("entries cannot be None")
with self._lock:
self._require_open()
hints = style_hints or {}
panel: Panel = self._element # type: ignore[assignment]
for key, value in entries.items():
hint = hints.get(key)
found = False
for entry in panel.entries:
if entry.key == key:
entry.value = value
entry.style_hint = hint
found = True
break
if not found:
panel.entries.append(
PanelEntry(key=key, value=value, style_hint=hint)
)
self._emit_update("entries_set", {"entries": entries})
def remove_entry(self, key: str) -> None:
"""Remove an entry by key."""
if not key:
raise ValueError("key cannot be empty")
with self._lock:
self._require_open()
panel: Panel = self._element # type: ignore[assignment]
panel.entries = [e for e in panel.entries if e.key != key]
self._emit_update("entry_removed", {"key": key})
# Convenience alias used in the task description
def add_line(self, key: str, value: str) -> None:
"""Convenience alias for ``set_entry``."""
self.set_entry(key, value)
class TableHandle(ElementHandle):
"""Handle for building a Table element incrementally."""
def __init__(
self,
handle_id: str,
declaration_index: int,
session: OutputSession,
element: Table,
) -> None:
super().__init__(handle_id, "table", declaration_index, session, element)
def add_row(self, row: dict[str, Any] | list[Any]) -> None:
"""Append a single row to the table.
Accepts either a dict mapping column names to values, or a list
of positional values matching the column order.
"""
if row is None:
raise ValueError("row cannot be None")
with self._lock:
self._require_open()
tbl: Table = self._element # type: ignore[assignment]
if isinstance(row, list):
row_dict: dict[str, Any] = {}
for idx, val in enumerate(row):
if idx < len(tbl.columns):
row_dict[tbl.columns[idx].name] = val
tbl.rows.append(row_dict)
else:
tbl.rows.append(dict(row))
self._emit_update("row_added", {"row": row})
def add_rows(self, rows: list[dict[str, Any] | list[Any]]) -> None:
"""Append multiple rows in a batch."""
if rows is None:
raise ValueError("rows cannot be None")
with self._lock:
self._require_open()
tbl: Table = self._element # type: ignore[assignment]
for row in rows:
if isinstance(row, list):
row_dict: dict[str, Any] = {}
for idx, val in enumerate(row):
if idx < len(tbl.columns):
row_dict[tbl.columns[idx].name] = val
tbl.rows.append(row_dict)
else:
tbl.rows.append(dict(row))
self._emit_update("rows_added", {"count": len(rows)})
def set_summary(self, summary: dict[str, Any]) -> None:
"""Set or update the summary/aggregation row."""
if summary is None:
raise ValueError("summary cannot be None")
with self._lock:
self._require_open()
tbl: Table = self._element # type: ignore[assignment]
tbl.summary = summary
self._emit_update("summary_set", {"summary": summary})
def set_sort_key(self, column: str, *, descending: bool = False) -> None:
"""Change the sort key."""
if not column:
raise ValueError("column cannot be empty")
with self._lock:
self._require_open()
tbl: Table = self._element # type: ignore[assignment]
tbl.sort_key = column
self._emit_update(
"sort_key_changed",
{"column": column, "descending": descending},
)
class StatusHandle(ElementHandle):
"""Handle for a status message.
Status handles are typically created and immediately closed
(fire-and-forget). They can be kept open for messages that may be
revised.
"""
def __init__(
self,
handle_id: str,
declaration_index: int,
session: OutputSession,
element: StatusMessage,
) -> None:
super().__init__(handle_id, "status", declaration_index, session, element)
def set_message(self, message: str) -> None:
"""Update the status message text."""
if not message:
raise ValueError("message cannot be empty")
with self._lock:
self._require_open()
sm: StatusMessage = self._element # type: ignore[assignment]
sm.message = message
self._emit_update("message_changed", {"message": message})
def set_level(self, level: str) -> None:
"""Change the status level."""
if not level:
raise ValueError("level cannot be empty")
valid_levels = {"ok", "warn", "error", "info"}
if level not in valid_levels:
raise ValueError(f"level must be one of {valid_levels!r}, got {level!r}")
with self._lock:
self._require_open()
sm: StatusMessage = self._element # type: ignore[assignment]
sm.level = level
self._emit_update("level_changed", {"level": level})
def set_detail(self, detail: str | None) -> None:
"""Set or clear the detail text."""
with self._lock:
self._require_open()
sm: StatusMessage = self._element # type: ignore[assignment]
sm.detail = detail
self._emit_update("detail_changed", {"detail": detail})
class ProgressHandle(ElementHandle):
"""Handle for updating a ProgressIndicator element.
Progress handles are unique in that they receive many rapid updates.
The materialisation strategy may throttle update events to avoid
overwhelming the terminal (e.g. limiting redraws to 10/sec).
"""
# Minimum interval between emitted update events (100 ms → 10 FPS).
_MIN_UPDATE_INTERVAL: float = 0.1
def __init__(
self,
handle_id: str,
declaration_index: int,
session: OutputSession,
element: ProgressIndicator,
) -> None:
super().__init__(handle_id, "progress", declaration_index, session, element)
self._last_emit_time: float = 0.0
def _throttled_emit(self, update_type: str, delta: dict[str, Any]) -> None:
"""Emit an update event, throttled to _MIN_UPDATE_INTERVAL."""
now = time.monotonic()
if (now - self._last_emit_time) >= self._MIN_UPDATE_INTERVAL:
self._last_emit_time = now
self._emit_update(update_type, delta)
def set_progress(self, current: int, total: int | None = None) -> None:
"""Update the progress counter."""
if current < 0:
raise ValueError("current must be non-negative")
with self._lock:
self._require_open()
pi: ProgressIndicator = self._element # type: ignore[assignment]
pi.current = current
if total is not None:
pi.total = total
self._throttled_emit(
"progress_changed",
{"current": current, "total": pi.total},
)
def set_label(self, label: str) -> None:
"""Update the progress label text."""
if not label:
raise ValueError("label cannot be empty")
with self._lock:
self._require_open()
pi: ProgressIndicator = self._element # type: ignore[assignment]
pi.label = label
self._emit_update("label_changed", {"label": label})
def tick(self) -> None:
"""Increment progress by 1."""
with self._lock:
self._require_open()
pi: ProgressIndicator = self._element # type: ignore[assignment]
if pi.current is None:
pi.current = 1
else:
pi.current += 1
self._throttled_emit(
"progress_changed",
{"current": pi.current, "total": pi.total},
)
def increment(self, delta: int = 1) -> None:
"""Increment progress by *delta*."""
if delta < 0:
raise ValueError("delta must be non-negative")
with self._lock:
self._require_open()
pi: ProgressIndicator = self._element # type: ignore[assignment]
if pi.current is None:
pi.current = delta
else:
pi.current += delta
self._throttled_emit(
"progress_changed",
{"current": pi.current, "total": pi.total},
)
def set_indeterminate(self) -> None:
"""Switch to indeterminate (spinner) mode."""
with self._lock:
self._require_open()
pi: ProgressIndicator = self._element # type: ignore[assignment]
pi.indeterminate = True
self._emit_update("indeterminate", {})
def set_step_status(self, step_label: str, status: str) -> None:
"""Update the status of a named step."""
if not step_label:
raise ValueError("step_label cannot be empty")
valid_statuses = {"pending", "active", "done", "error", "skipped"}
if status not in valid_statuses:
raise ValueError(
f"status must be one of {valid_statuses!r}, got {status!r}"
)
with self._lock:
self._require_open()
pi: ProgressIndicator = self._element # type: ignore[assignment]
if pi.steps is None:
pi.steps = []
for step in pi.steps:
if step.label == step_label:
step.status = status
break
else:
pi.steps.append(ProgressStep(label=step_label, status=status))
self._emit_update(
"step_status_changed",
{"step_label": step_label, "status": status},
)
@@ -0,0 +1,93 @@
"""Element handle classes for the output rendering framework.
Each handle type wraps a specific element kind and provides typed write
methods. Handles are format-agnostic producer code writes to handles
without knowledge of the active materialisation strategy.
Based on the Output Rendering Framework specification (v3_spec.md §Output
Rendering Framework / Element Handles).
"""
from cleveragents.cli.output.handles._base import ElementHandle
from cleveragents.cli.output.handles._concrete import (
ActionHintHandle,
CodeHandle,
DiffHandle,
ProgressHandle,
SeparatorHandle,
TextHandle,
TreeHandle,
)
from cleveragents.cli.output.handles._models import (
MAX_ELEMENTS_PER_SESSION,
MAX_TABLE_ROWS,
MAX_TREE_DEPTH,
ActionHint,
CodeBlock,
ColumnDef,
DiffBlock,
DiffHunk,
DiffLine,
ElementClosed,
ElementClosedError,
ElementCreated,
ElementEvent,
ElementSnapshot,
ElementUpdated,
Panel,
PanelEntry,
ProgressIndicator,
ProgressStep,
Separator,
SessionEnd,
StatusMessage,
Table,
TextBlock,
Tree,
TreeNode,
)
from cleveragents.cli.output.handles._panel_table import (
PanelHandle,
StatusHandle,
TableHandle,
)
__all__ = [
"MAX_ELEMENTS_PER_SESSION",
"MAX_TABLE_ROWS",
"MAX_TREE_DEPTH",
"ActionHint",
"ActionHintHandle",
"CodeBlock",
"CodeHandle",
"ColumnDef",
"DiffBlock",
"DiffHandle",
"DiffHunk",
"DiffLine",
"ElementClosed",
"ElementClosedError",
"ElementCreated",
"ElementEvent",
"ElementHandle",
"ElementSnapshot",
"ElementUpdated",
"Panel",
"PanelEntry",
"PanelHandle",
"ProgressHandle",
"ProgressIndicator",
"ProgressStep",
"Separator",
"SeparatorHandle",
"SessionEnd",
"StatusHandle",
"StatusMessage",
"Table",
"TableHandle",
"TextBlock",
"TextHandle",
"Tree",
"TreeHandle",
"TreeNode",
]
@@ -0,0 +1,139 @@
"""Base ElementHandle class for the output rendering framework.
Provides the generic handle type that all concrete handles inherit from.
"""
from __future__ import annotations
import threading
from typing import TYPE_CHECKING, Any
from cleveragents.cli.output.handles._models import (
ElementClosed,
ElementClosedError,
ElementSnapshot,
ElementUpdated,
)
if TYPE_CHECKING:
from cleveragents.cli.output.session import OutputSession
class ElementHandle[E: ElementSnapshot]:
"""Base class for all element handles.
An element handle is a write-only view of an output element. Producers
use handles to incrementally build element content without knowledge of
the active format or materialisation strategy.
Thread Safety
-------------
Individual handle methods are thread-safe. Concurrent writes to the
*same* handle are serialised via an internal lock.
Lifecycle
---------
``handle = session.panel(...)`` -- Created by session factory
``handle.set_entry(...)`` -- Write operations (zero or more)
``handle.close()`` -- Finalise (required unless auto-closed)
"""
def __init__(
self,
handle_id: str,
element_type: str,
declaration_index: int,
session: OutputSession,
element: E,
) -> None:
if not handle_id:
raise ValueError("handle_id cannot be empty")
if not element_type:
raise ValueError("element_type cannot be empty")
if declaration_index < 0:
raise ValueError("declaration_index must be non-negative")
if session is None:
raise ValueError("session cannot be None")
if element is None:
raise ValueError("element cannot be None")
self.handle_id = handle_id
self.element_type = element_type
self.declaration_index = declaration_index
self._session: OutputSession = session
self._element: E = element
self._state: str = "open"
self._lock: threading.Lock = threading.Lock()
def _require_open(self) -> None:
"""Raise if handle is closed."""
if self._state != "open":
raise ElementClosedError(
f"Handle {self.handle_id!r} ({self.element_type}) is closed"
)
def _emit_update(self, update_type: str, delta: dict[str, Any]) -> None:
"""Emit an ElementUpdated event to the session.
N3 optimisation: skip Pydantic object construction entirely when
the active strategy does not consume incremental updates.
"""
if not self._session._strategy.supports_incremental_updates:
return
event = ElementUpdated(
event_type="updated",
handle_id=self.handle_id,
element_kind=self.element_type,
update_type=update_type,
delta=delta,
element_snapshot=self._element, # type-safe via type parameter E
)
self._session._dispatch_event(event)
@property
def is_open(self) -> bool:
"""Whether this handle is still accepting writes."""
return self._state == "open"
@property
def element(self) -> E:
"""The accumulated element state (read-only snapshot).
Returns the internal model reference for performance. Use
``element_copy()`` if a detached deep copy is required (L3).
"""
return self._element
def element_copy(self) -> E:
"""Return a deep copy of the accumulated element state.
Acquires the handle lock to prevent reading mutable state
while a concurrent write is in progress.
"""
with self._lock:
return self._element.model_copy(deep=True)
def close(self) -> None:
"""Close this handle, signalling that no more data will be written."""
with self._lock:
if self._state == "open":
self._state = "closed"
event = ElementClosed(
event_type="closed",
handle_id=self.handle_id,
element_kind=self.element_type,
final_state=self._element, # type-safe via type parameter E
)
self._session._dispatch_event(event)
def __enter__(self) -> ElementHandle[E]:
return self
def __exit__(
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: Any,
) -> None:
if self.is_open:
self.close()
@@ -0,0 +1,475 @@
"""Progress, Tree, Text, Code, Diff, Separator, and ActionHint handles.
These handles have more complex state management than the simpler panel,
table, and status handles.
"""
from __future__ import annotations
import time
from typing import TYPE_CHECKING, Any
from cleveragents.cli.output.handles._base import ElementHandle
from cleveragents.cli.output.handles._models import (
MAX_TREE_DEPTH,
ActionHint,
CodeBlock,
DiffBlock,
DiffHunk,
DiffLine,
ProgressIndicator,
ProgressStep,
Separator,
TextBlock,
Tree,
TreeNode,
)
if TYPE_CHECKING:
from cleveragents.cli.output.session import OutputSession
# ---------------------------------------------------------------------------
# ProgressHandle
# ---------------------------------------------------------------------------
class ProgressHandle(ElementHandle[ProgressIndicator]):
"""Handle for updating a ProgressIndicator element.
Progress handles are unique in that they receive many rapid updates.
The materialisation strategy may throttle update events to avoid
overwhelming the terminal (e.g. limiting redraws to 10/sec).
"""
_MIN_UPDATE_INTERVAL: float = 0.1
def __init__(
self,
handle_id: str,
declaration_index: int,
session: OutputSession,
element: ProgressIndicator,
) -> None:
super().__init__(handle_id, "progress", declaration_index, session, element)
self._last_emit_time: float = 0.0
def _throttled_emit(self, update_type: str, delta: dict[str, Any]) -> None:
"""Emit an update event, throttled to _MIN_UPDATE_INTERVAL."""
now = time.monotonic()
if (now - self._last_emit_time) >= self._MIN_UPDATE_INTERVAL:
self._last_emit_time = now
self._emit_update(update_type, delta)
def set_progress(self, current: int, total: int | None = None) -> None:
"""Update the progress counter.
P1-1 fix (Luis review): validates *total* is non-negative when
provided, preventing nonsensical percentages in renderers.
"""
if current < 0:
raise ValueError("current must be non-negative")
if total is not None and total < 0:
raise ValueError("total must be non-negative")
with self._lock:
self._require_open()
self._element.current = current
if total is not None:
self._element.total = total
self._throttled_emit(
"progress_changed",
{"current": current, "total": self._element.total},
)
def set_label(self, label: str) -> None:
"""Update the progress label text."""
if not label:
raise ValueError("label cannot be empty")
with self._lock:
self._require_open()
self._element.label = label
self._emit_update("label_changed", {"label": label})
def tick(self) -> None:
"""Increment progress by 1."""
with self._lock:
self._require_open()
if self._element.current is None:
self._element.current = 1
else:
self._element.current += 1
self._throttled_emit(
"progress_changed",
{"current": self._element.current, "total": self._element.total},
)
def increment(self, delta: int = 1) -> None:
"""Increment progress by *delta*.
P3-8 fix (Luis review): rejects zero delta as a likely caller
error (zero-delta makes no progress).
"""
if delta <= 0:
raise ValueError("delta must be positive")
with self._lock:
self._require_open()
if self._element.current is None:
self._element.current = delta
else:
self._element.current += delta
self._throttled_emit(
"progress_changed",
{"current": self._element.current, "total": self._element.total},
)
def set_indeterminate(self) -> None:
"""Switch to indeterminate (spinner) mode."""
with self._lock:
self._require_open()
self._element.indeterminate = True
self._emit_update("indeterminate", {})
def set_step_status(self, step_label: str, status: str) -> None:
"""Update the status of a named step."""
if not step_label:
raise ValueError("step_label cannot be empty")
valid_statuses = {"pending", "active", "done", "error", "skipped"}
if status not in valid_statuses:
raise ValueError(
f"status must be one of {valid_statuses!r}, got {status!r}"
)
with self._lock:
self._require_open()
if self._element.steps is None:
self._element.steps = []
for step in self._element.steps:
if step.label == step_label:
step.status = status
break
else:
self._element.steps.append(
ProgressStep(label=step_label, status=status)
)
self._emit_update(
"step_status_changed",
{"step_label": step_label, "status": status},
)
# ---------------------------------------------------------------------------
# TreeHandle
# ---------------------------------------------------------------------------
class TreeHandle(ElementHandle[Tree]):
"""Handle for building a Tree element incrementally.
The tree is accessed via path strings using ``/`` as separator,
starting from the root node's label.
"""
def __init__(
self,
handle_id: str,
declaration_index: int,
session: OutputSession,
element: Tree,
) -> None:
super().__init__(handle_id, "tree", declaration_index, session, element)
self._path_cache: dict[str, TreeNode] = {
element.root.label: element.root,
"": element.root,
}
def _find_node(self, path: str) -> TreeNode | None:
"""Locate a node by its ``/``-separated path using the path cache."""
cached = self._path_cache.get(path)
if cached is not None:
return cached
parts = [p for p in path.split("/") if p]
if not parts:
return self._element.root
if parts[0] != self._element.root.label:
return None
node = self._element.root
for part in parts[1:]:
found = False
for child in node.children:
if child.label == part:
node = child
found = True
break
if not found:
return None
self._path_cache[path] = node
return node
@staticmethod
def _node_depth(path: str) -> int:
"""Return the nesting depth implied by a ``/``-separated path."""
parts = [p for p in path.split("/") if p]
return len(parts)
def add_child(
self,
parent_path: str,
label: str,
*,
style_hint: str | None = None,
collapsed: bool = False,
metadata: dict[str, Any] | None = None,
) -> str:
"""Add a child node under *parent_path* and return the full path."""
if not label:
raise ValueError("label cannot be empty")
if "/" in label:
raise ValueError(f"Tree node label cannot contain '/': {label!r}")
with self._lock:
self._require_open()
parent = self._find_node(parent_path)
if parent is None:
raise ValueError(f"Parent path not found: {parent_path!r}")
normalized_path = parent_path if parent_path else self._element.root.label
child_depth = self._node_depth(normalized_path) + 1
if child_depth > MAX_TREE_DEPTH:
raise ValueError(f"Maximum tree depth ({MAX_TREE_DEPTH}) exceeded")
child = TreeNode(
label=label,
style_hint=style_hint,
collapsed=collapsed,
metadata=metadata or {},
)
parent.children.append(child)
if parent_path:
full_path = f"{parent_path}/{label}"
else:
full_path = f"{self._element.root.label}/{label}"
self._path_cache[full_path] = child
self._emit_update("child_added", {"parent": parent_path, "label": label})
return full_path
def set_node_style(self, path: str, style_hint: str) -> None:
"""Update the style hint on an existing node."""
if not path:
raise ValueError("path cannot be empty")
with self._lock:
self._require_open()
node = self._find_node(path)
if node is None:
raise ValueError(f"Node path not found: {path!r}")
node.style_hint = style_hint
self._emit_update("node_style_changed", {"path": path})
# ---------------------------------------------------------------------------
# TextHandle
# ---------------------------------------------------------------------------
class TextHandle(ElementHandle[TextBlock]):
"""Handle for a free-form text block.
Text content is buffered in a ``list[str]`` and joined lazily to
avoid O(N^2) string concatenation on repeated ``append()`` calls
(F2 fix).
"""
def __init__(
self,
handle_id: str,
declaration_index: int,
session: OutputSession,
element: TextBlock,
) -> None:
super().__init__(handle_id, "text", declaration_index, session, element)
self._text_parts: list[str] = [element.content] if element.content else []
def _flush_buffer(self) -> None:
"""Synchronise the element's content from the internal buffer."""
self._element.content = "".join(self._text_parts)
@property
def element(self) -> TextBlock:
"""The accumulated element state with buffered text flushed."""
with self._lock:
self._flush_buffer()
return self._element
def element_copy(self) -> TextBlock:
"""Return a deep copy with buffered text flushed."""
with self._lock:
self._flush_buffer()
return self._element.model_copy(deep=True)
def _emit_update(self, update_type: str, delta: dict[str, Any]) -> None:
"""Flush the text buffer before emitting a snapshot.
Short-circuits before flushing when the strategy does not consume
incremental updates, preserving the lazy-buffer O(1) amortised
cost of ``append()`` (R2-C2 fix).
"""
if not self._session._strategy.supports_incremental_updates:
return
self._flush_buffer()
super()._emit_update(update_type, delta)
def append(self, text: str) -> None:
"""Append text to the current content."""
if text is None:
raise ValueError("text cannot be None")
with self._lock:
self._require_open()
self._text_parts.append(text)
self._emit_update("text_appended", {"text": text})
def set_content(self, content: str) -> None:
"""Replace the entire content."""
if content is None:
raise ValueError("content cannot be None")
with self._lock:
self._require_open()
self._text_parts = [content]
self._element.content = content
self._emit_update("content_set", {"content": content})
def close(self) -> None:
"""Close the text handle after flushing buffered content.
P3-2 fix (Luis review): flush the buffer then delegate to the
base class ``close()`` instead of reimplementing the lifecycle
logic. This ensures TextHandle tracks any future base-class
changes (logging, metrics, etc.).
"""
with self._lock:
if self._state == "open":
self._flush_buffer()
super().close()
# ---------------------------------------------------------------------------
# CodeHandle
# ---------------------------------------------------------------------------
class CodeHandle(ElementHandle[CodeBlock]):
"""Handle for a source-code block with syntax highlighting hints."""
def __init__(
self,
handle_id: str,
declaration_index: int,
session: OutputSession,
element: CodeBlock,
) -> None:
super().__init__(handle_id, "code", declaration_index, session, element)
def set_content(self, content: str) -> None:
"""Replace the code content."""
if content is None:
raise ValueError("content cannot be None")
with self._lock:
self._require_open()
self._element.content = content
self._emit_update("content_set", {"content": content})
def set_language(self, language: str) -> None:
"""Set or change the language hint."""
if not language:
raise ValueError("language cannot be empty")
with self._lock:
self._require_open()
self._element.language = language
self._emit_update("language_set", {"language": language})
def set_highlight_lines(self, lines: list[int]) -> None:
"""Set lines to highlight."""
if lines is None:
raise ValueError("lines cannot be None")
with self._lock:
self._require_open()
self._element.highlight_lines = list(lines)
self._emit_update("highlight_lines_set", {"lines": lines})
# ---------------------------------------------------------------------------
# DiffHandle
# ---------------------------------------------------------------------------
class DiffHandle(ElementHandle[DiffBlock]):
"""Handle for a unified-diff display element."""
def __init__(
self,
handle_id: str,
declaration_index: int,
session: OutputSession,
element: DiffBlock,
) -> None:
super().__init__(handle_id, "diff", declaration_index, session, element)
def add_hunk(self, header: str, lines: list[DiffLine]) -> None:
"""Add a diff hunk."""
if not header:
raise ValueError("header cannot be empty")
if lines is None:
raise ValueError("lines cannot be None")
with self._lock:
self._require_open()
hunk = DiffHunk(header=header, lines=list(lines))
self._element.hunks.append(hunk)
self._emit_update("hunk_added", {"header": header})
def set_stats(
self,
insertions: int,
deletions: int,
**extra: Any,
) -> None:
"""Set diff statistics."""
if insertions < 0:
raise ValueError("insertions must be non-negative")
if deletions < 0:
raise ValueError("deletions must be non-negative")
with self._lock:
self._require_open()
self._element.stats = {
"insertions": insertions,
"deletions": deletions,
**extra,
}
self._emit_update(
"stats_set",
{"insertions": insertions, "deletions": deletions},
)
# ---------------------------------------------------------------------------
# SeparatorHandle / ActionHintHandle
# ---------------------------------------------------------------------------
class SeparatorHandle(ElementHandle[Separator]):
"""Handle for a visual separator between elements (auto-closed)."""
def __init__(
self,
handle_id: str,
declaration_index: int,
session: OutputSession,
element: Separator,
) -> None:
super().__init__(handle_id, "separator", declaration_index, session, element)
class ActionHintHandle(ElementHandle[ActionHint]):
"""Handle for suggested next-step CLI commands (auto-closed)."""
def __init__(
self,
handle_id: str,
declaration_index: int,
session: OutputSession,
element: ActionHint,
) -> None:
super().__init__(handle_id, "action_hint", declaration_index, session, element)
@@ -0,0 +1,297 @@
"""Element snapshot data models for the output rendering framework.
Each model represents the accumulated state of an element at a point in
time. Handles mutate these models incrementally; strategies consume them
for rendering.
Based on the Output Rendering Framework specification (v3_spec.md §Output
Rendering Framework / Element Handles).
"""
from __future__ import annotations
import time
from typing import Any
from pydantic import BaseModel, ConfigDict, Field, field_validator
# ---------------------------------------------------------------------------
# Resource limits (configurable soft caps — M7, M8)
# ---------------------------------------------------------------------------
MAX_TREE_DEPTH: int = 64
"""Maximum nesting depth for tree elements (DoS guard)."""
MAX_ELEMENTS_PER_SESSION: int = 10_000
"""Soft limit on the number of element handles per session."""
MAX_TABLE_ROWS: int = 100_000
"""Soft limit on the number of rows in a single table."""
# ---------------------------------------------------------------------------
# Exceptions
# ---------------------------------------------------------------------------
class ElementClosedError(RuntimeError):
"""Raised when a write method is called on a closed handle."""
# ---------------------------------------------------------------------------
# Element snapshot data classes
# ---------------------------------------------------------------------------
class PanelEntry(BaseModel):
"""A single key-value pair within a Panel."""
key: str
value: str
style_hint: str | None = None
icon: str | None = None
class Panel(BaseModel):
"""A titled group of key-value pairs."""
title: str
entries: list[PanelEntry] = Field(default_factory=list)
border_style: str = "rounded"
priority: str = "normal"
collapse_hint: str = "auto"
metadata: dict[str, Any] = Field(default_factory=dict)
element_type: str = "panel"
class ColumnDef(BaseModel):
"""Schema for a single table column."""
name: str
col_type: str = "string"
alignment: str = "left"
width_hint: int | None = None
sortable: bool = False
style_hint: str | None = None
class Table(BaseModel):
"""A tabular data set with typed columns."""
title: str | None
columns: list[ColumnDef] = Field(default_factory=list)
rows: list[dict[str, Any]] = Field(default_factory=list)
summary: dict[str, Any] | None = None
max_rows_hint: int | None = None
sort_key: str | None = None
sort_descending: bool = False
priority: str = "normal"
collapse_hint: str = "auto"
metadata: dict[str, Any] = Field(default_factory=dict)
element_type: str = "table"
class StatusMessage(BaseModel):
"""A status line (success, warning, error, info)."""
message: str
level: str = "info"
detail: str | None = None
priority: str = "normal"
metadata: dict[str, Any] = Field(default_factory=dict)
element_type: str = "status"
class ProgressStep(BaseModel):
"""A named step within a progress indicator."""
label: str
status: str = "pending"
class ProgressIndicator(BaseModel):
"""A progress bar or spinner for long-running operations."""
label: str
current: int | None = None
total: int | None = None
indeterminate: bool = False
steps: list[ProgressStep] | None = None
priority: str = "normal"
metadata: dict[str, Any] = Field(default_factory=dict)
element_type: str = "progress"
class TreeNode(BaseModel):
"""A single node in a hierarchical tree structure."""
label: str
style_hint: str | None = None
children: list[TreeNode] = Field(default_factory=list)
collapsed: bool = False
metadata: dict[str, Any] = Field(default_factory=dict)
class Tree(BaseModel):
"""A hierarchical tree display element."""
root: TreeNode
max_depth_hint: int | None = None
show_guides: bool = True
priority: str = "normal"
collapse_hint: str = "auto"
metadata: dict[str, Any] = Field(default_factory=dict)
element_type: str = "tree"
class TextBlock(BaseModel):
"""A free-form text block element."""
content: str = ""
wrap: bool = True
indent: int = 0
priority: str = "normal"
metadata: dict[str, Any] = Field(default_factory=dict)
element_type: str = "text"
@field_validator("indent")
@classmethod
def _validate_indent(cls, v: int) -> int:
if v < 0:
raise ValueError(f"indent must be non-negative, got {v}")
return v
class CodeBlock(BaseModel):
"""A source-code block with optional syntax highlighting hints."""
content: str = ""
language: str | None = None
line_numbers: bool = False
highlight_lines: list[int] = Field(default_factory=list)
priority: str = "normal"
metadata: dict[str, Any] = Field(default_factory=dict)
element_type: str = "code"
class DiffLine(BaseModel):
"""A single line within a unified diff hunk.
P2-1 fix (Luis review): renamed ``type`` ``line_type`` to avoid
shadowing the Python builtin ``type()``, consistent with the SD-21
rationale for ``ColumnDef.col_type``. The ``alias="type"`` preserves
backward-compatible construction via ``DiffLine(type="add", ...)``.
"""
model_config = ConfigDict(populate_by_name=True)
line_type: str = Field(alias="type") # "context", "add", "remove"
content: str
line_number_old: int | None = None
line_number_new: int | None = None
@field_validator("line_type")
@classmethod
def _validate_line_type(cls, v: str) -> str:
valid = {"context", "add", "remove"}
if v not in valid:
raise ValueError(f"DiffLine.line_type must be one of {valid!r}, got {v!r}")
return v
class DiffHunk(BaseModel):
"""A hunk within a unified diff."""
header: str
lines: list[DiffLine] = Field(default_factory=list)
class DiffBlock(BaseModel):
"""A unified-diff display element."""
hunks: list[DiffHunk] = Field(default_factory=list)
file_a: str | None = None
file_b: str | None = None
stats: dict[str, Any] | None = None
priority: str = "normal"
metadata: dict[str, Any] = Field(default_factory=dict)
element_type: str = "diff"
class Separator(BaseModel):
"""A visual separator between elements."""
style: str = "line" # "line", "blank", "double"
element_type: str = "separator"
@field_validator("style")
@classmethod
def _validate_style(cls, v: str) -> str:
valid = {"line", "blank", "double"}
if v not in valid:
raise ValueError(f"Separator.style must be one of {valid!r}, got {v!r}")
return v
class ActionHint(BaseModel):
"""Suggested next-step CLI commands."""
commands: list[str] = Field(default_factory=list)
description: str = ""
element_type: str = "action_hint"
# Union of all element snapshot types used by handles.
ElementSnapshot = (
Panel
| Table
| StatusMessage
| ProgressIndicator
| Tree
| TextBlock
| CodeBlock
| DiffBlock
| Separator
| ActionHint
)
# ---------------------------------------------------------------------------
# Element events
# ---------------------------------------------------------------------------
class ElementEvent(BaseModel):
"""Base class for all events emitted by element handles."""
event_type: str
handle_id: str
element_kind: str
timestamp: float = Field(default_factory=time.monotonic)
class ElementCreated(ElementEvent):
"""Emitted when a new element handle is created."""
declaration_index: int = 0
initial_state: ElementSnapshot | None = None
class ElementUpdated(ElementEvent):
"""Emitted when data is written to an element handle."""
update_type: str = ""
delta: dict[str, Any] = Field(default_factory=dict)
element_snapshot: ElementSnapshot | None = None
class ElementClosed(ElementEvent):
"""Emitted when an element handle is closed."""
final_state: ElementSnapshot | None = None
class SessionEnd(ElementEvent):
"""Emitted when the session itself is closed."""
exit_code: int = 0
@@ -0,0 +1,210 @@
"""Panel, Table, and Status handle implementations."""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from cleveragents.cli.output.handles._base import ElementHandle
from cleveragents.cli.output.handles._models import (
MAX_TABLE_ROWS,
Panel,
PanelEntry,
StatusMessage,
Table,
)
if TYPE_CHECKING:
from cleveragents.cli.output.session import OutputSession
class PanelHandle(ElementHandle[Panel]):
"""Handle for building a Panel element incrementally."""
def __init__(
self,
handle_id: str,
declaration_index: int,
session: OutputSession,
element: Panel,
) -> None:
super().__init__(handle_id, "panel", declaration_index, session, element)
def set_entry(
self,
key: str,
value: str,
*,
style_hint: str | None = None,
icon: str | None = None,
) -> None:
"""Set or update a key-value entry in the panel."""
if not key:
raise ValueError("key cannot be empty")
with self._lock:
self._require_open()
for entry in self._element.entries:
if entry.key == key:
entry.value = value
entry.style_hint = style_hint
entry.icon = icon
break
else:
self._element.entries.append(
PanelEntry(key=key, value=value, style_hint=style_hint, icon=icon)
)
self._emit_update("entry_set", {"key": key, "value": value})
def set_entries(
self,
entries: dict[str, str],
*,
style_hints: dict[str, str] | None = None,
) -> None:
"""Set multiple entries at once (batch update)."""
if entries is None:
raise ValueError("entries cannot be None")
with self._lock:
self._require_open()
hints = style_hints or {}
for key, value in entries.items():
hint = hints.get(key)
found = False
for entry in self._element.entries:
if entry.key == key:
entry.value = value
entry.style_hint = hint
found = True
break
if not found:
self._element.entries.append(
PanelEntry(key=key, value=value, style_hint=hint)
)
self._emit_update("entries_set", {"entries": entries})
def remove_entry(self, key: str) -> None:
"""Remove an entry by key."""
if not key:
raise ValueError("key cannot be empty")
with self._lock:
self._require_open()
self._element.entries = [e for e in self._element.entries if e.key != key]
self._emit_update("entry_removed", {"key": key})
def add_line(self, key: str, value: str) -> None:
"""Convenience alias for ``set_entry``."""
self.set_entry(key, value)
class TableHandle(ElementHandle[Table]):
"""Handle for building a Table element incrementally."""
def __init__(
self,
handle_id: str,
declaration_index: int,
session: OutputSession,
element: Table,
) -> None:
super().__init__(handle_id, "table", declaration_index, session, element)
def add_row(self, row: dict[str, Any] | list[Any]) -> None:
"""Append a single row to the table."""
if row is None:
raise ValueError("row cannot be None")
with self._lock:
self._require_open()
if len(self._element.rows) >= MAX_TABLE_ROWS:
raise ValueError(f"Maximum table rows ({MAX_TABLE_ROWS}) exceeded")
if isinstance(row, list):
row_dict: dict[str, Any] = {}
for idx, val in enumerate(row):
if idx < len(self._element.columns):
row_dict[self._element.columns[idx].name] = val
self._element.rows.append(row_dict)
else:
self._element.rows.append(dict(row))
self._emit_update("row_added", {"row": row})
def add_rows(self, rows: list[dict[str, Any] | list[Any]]) -> None:
"""Append multiple rows in a batch."""
if rows is None:
raise ValueError("rows cannot be None")
with self._lock:
self._require_open()
if len(self._element.rows) + len(rows) > MAX_TABLE_ROWS:
raise ValueError(
f"Maximum table rows ({MAX_TABLE_ROWS}) would be exceeded"
)
for row in rows:
if isinstance(row, list):
row_dict: dict[str, Any] = {}
for idx, val in enumerate(row):
if idx < len(self._element.columns):
row_dict[self._element.columns[idx].name] = val
self._element.rows.append(row_dict)
else:
self._element.rows.append(dict(row))
self._emit_update("rows_added", {"count": len(rows)})
def set_summary(self, summary: dict[str, Any]) -> None:
"""Set or update the summary/aggregation row."""
if summary is None:
raise ValueError("summary cannot be None")
with self._lock:
self._require_open()
self._element.summary = summary
self._emit_update("summary_set", {"summary": summary})
def set_sort_key(self, column: str, *, descending: bool = False) -> None:
"""Change the sort key and direction."""
if not column:
raise ValueError("column cannot be empty")
with self._lock:
self._require_open()
self._element.sort_key = column
self._element.sort_descending = descending
self._emit_update(
"sort_key_changed",
{"column": column, "descending": descending},
)
class StatusHandle(ElementHandle[StatusMessage]):
"""Handle for a status message."""
def __init__(
self,
handle_id: str,
declaration_index: int,
session: OutputSession,
element: StatusMessage,
) -> None:
super().__init__(handle_id, "status", declaration_index, session, element)
def set_message(self, message: str) -> None:
"""Update the status message text."""
if not message:
raise ValueError("message cannot be empty")
with self._lock:
self._require_open()
self._element.message = message
self._emit_update("message_changed", {"message": message})
def set_level(self, level: str) -> None:
"""Change the status level."""
if not level:
raise ValueError("level cannot be empty")
valid_levels = {"ok", "warn", "error", "info"}
if level not in valid_levels:
raise ValueError(f"level must be one of {valid_levels!r}, got {level!r}")
with self._lock:
self._require_open()
self._element.level = level
self._emit_update("level_changed", {"level": level})
def set_detail(self, detail: str | None) -> None:
"""Set or clear the detail text."""
with self._lock:
self._require_open()
self._element.detail = detail
self._emit_update("detail_changed", {"detail": detail})
+220 -337
View File
@@ -17,26 +17,56 @@ Rendering Framework / Materialization Strategies).
from __future__ import annotations
import json
import threading
from io import StringIO
from typing import TYPE_CHECKING, Any, Protocol
import yaml
from cleveragents.cli.output._boxdraw import render_element_table
from cleveragents.cli.output._color_renderers import render_element_color
from cleveragents.cli.output._renderers import (
render_element_plain,
strip_terminal_escapes,
)
from cleveragents.cli.output.handles import (
MAX_TREE_DEPTH,
ActionHint,
CodeBlock,
ColumnDef,
DiffBlock,
DiffLine,
ElementClosed,
ElementCreated,
ElementSnapshot,
ElementUpdated,
Panel,
PanelEntry,
ProgressIndicator,
Separator,
SessionEnd,
StatusMessage,
Table,
TextBlock,
Tree,
TreeNode,
)
if TYPE_CHECKING:
from cleveragents.cli.output.session import OutputSession, StructuredOutput
# Re-export for backward compatibility
__all__ = [
"ColorMaterializer",
"JsonMaterializer",
"MaterializationStrategy",
"PlainMaterializer",
"RichMaterializer",
"TableMaterializer",
"YamlMaterializer",
"strip_terminal_escapes",
]
# ---------------------------------------------------------------------------
# MaterializationStrategy protocol
@@ -47,6 +77,7 @@ class MaterializationStrategy(Protocol):
"""Interface for format-driven output materialisation."""
strategy_name: str
supports_incremental_updates: bool
def on_session_begin(self, session: OutputSession) -> None: ... # pragma: no cover
@@ -59,280 +90,50 @@ class MaterializationStrategy(Protocol):
def on_session_end(self, event: SessionEnd) -> None: ... # pragma: no cover
# ---------------------------------------------------------------------------
# Rendering helpers (shared across visual strategies)
# ---------------------------------------------------------------------------
def _render_panel_plain(panel: Panel) -> str:
"""Render a panel as plain ASCII text."""
lines: list[str] = [panel.title]
for entry in panel.entries:
lines.append(f" {entry.key}: {entry.value}")
return "\n".join(lines)
def _render_table_plain(table: Table) -> str:
"""Render a table as plain ASCII text with aligned columns."""
if not table.columns:
return "(empty table)"
col_names = [c.name for c in table.columns]
# Build rows as list of string lists
str_rows: list[list[str]] = []
for row in table.rows:
str_rows.append([str(row.get(c, "")) for c in col_names])
# Compute column widths
widths = [len(c) for c in col_names]
for row in str_rows:
for i, cell in enumerate(row):
if i < len(widths):
widths[i] = max(widths[i], len(cell))
# Format header
header = " ".join(n.ljust(w) for n, w in zip(col_names, widths, strict=False))
sep = " ".join("-" * w for w in widths)
lines: list[str] = []
if table.title:
lines.append(table.title)
lines.append(header)
lines.append(sep)
for row in str_rows:
lines.append(" ".join(c.ljust(w) for c, w in zip(row, widths, strict=False)))
if table.summary:
lines.append(sep)
summary_parts: list[str] = []
for key, val in table.summary.items():
summary_parts.append(f"{key}: {val}")
lines.append(" ".join(summary_parts))
return "\n".join(lines)
def _render_status_plain(status: StatusMessage) -> str:
"""Render a status message as plain ASCII."""
prefix_map = {
"ok": "[OK]",
"warn": "[WARN]",
"error": "[ERROR]",
"info": "[INFO]",
}
prefix = prefix_map.get(status.level, f"[{status.level.upper()}]")
line = f"{prefix} {status.message}"
if status.detail:
line += f"\n {status.detail}"
return line
def _render_progress_plain(progress: ProgressIndicator) -> str:
"""Render a progress indicator as plain ASCII."""
lines: list[str] = []
if progress.indeterminate:
lines.append(f"{progress.label}: ...")
elif progress.total is not None and progress.current is not None:
pct = int(progress.current * 100 / progress.total) if progress.total > 0 else 0
lines.append(f"{progress.label}: {progress.current}/{progress.total} ({pct}%)")
elif progress.current is not None:
lines.append(f"{progress.label}: {progress.current}")
else:
lines.append(f"{progress.label}: pending")
if progress.steps:
for step in progress.steps:
marker_map = {
"done": "[x]",
"active": "[>]",
"pending": "[ ]",
"error": "[!]",
"skipped": "[-]",
}
marker = marker_map.get(step.status, "[ ]")
lines.append(f" {marker} {step.label}")
return "\n".join(lines)
def _render_element_plain(element: Any) -> str:
"""Dispatch to the correct plain renderer."""
if isinstance(element, Panel):
return _render_panel_plain(element)
if isinstance(element, Table):
return _render_table_plain(element)
if isinstance(element, StatusMessage):
return _render_status_plain(element)
if isinstance(element, ProgressIndicator):
return _render_progress_plain(element)
return str(element) # pragma: no cover
# ANSI colour codes
_BOLD = "\033[1m"
_CYAN = "\033[36m"
_GREEN = "\033[32m"
_YELLOW = "\033[33m"
_RED = "\033[31m"
_RESET = "\033[0m"
_DIM = "\033[2m"
def _render_panel_color(panel: Panel) -> str:
"""Render a panel with ANSI colour codes."""
lines: list[str] = [f"{_BOLD}{_CYAN}{panel.title}{_RESET}"]
for entry in panel.entries:
value_str = entry.value
if entry.style_hint == "success":
value_str = f"{_GREEN}{entry.value}{_RESET}"
elif entry.style_hint == "warning":
value_str = f"{_YELLOW}{entry.value}{_RESET}"
elif entry.style_hint == "error":
value_str = f"{_RED}{entry.value}{_RESET}"
lines.append(f" {_BOLD}{entry.key}{_RESET}: {value_str}")
return "\n".join(lines)
def _render_table_color(table: Table) -> str:
"""Render a table with ANSI colour codes."""
if not table.columns:
return f"{_DIM}(empty table){_RESET}"
col_names = [c.name for c in table.columns]
str_rows: list[list[str]] = []
for row in table.rows:
str_rows.append([str(row.get(c, "")) for c in col_names])
widths = [len(c) for c in col_names]
for row in str_rows:
for i, cell in enumerate(row):
if i < len(widths):
widths[i] = max(widths[i], len(cell))
header = " ".join(
f"{_BOLD}{_CYAN}{n.ljust(w)}{_RESET}"
for n, w in zip(col_names, widths, strict=False)
)
sep = " ".join("-" * w for w in widths)
lines: list[str] = []
if table.title:
lines.append(f"{_BOLD}{table.title}{_RESET}")
lines.append(header)
lines.append(sep)
for row in str_rows:
lines.append(" ".join(c.ljust(w) for c, w in zip(row, widths, strict=False)))
return "\n".join(lines)
def _render_status_color(status: StatusMessage) -> str:
"""Render a status message with ANSI colour codes."""
colour_map = {
"ok": _GREEN,
"warn": _YELLOW,
"error": _RED,
"info": _CYAN,
}
prefix_map = {
"ok": "[OK]",
"warn": "[WARN]",
"error": "[ERROR]",
"info": "[INFO]",
}
colour = colour_map.get(status.level, "")
prefix = prefix_map.get(status.level, f"[{status.level.upper()}]")
line = f"{colour}{_BOLD}{prefix}{_RESET} {status.message}"
if status.detail:
line += f"\n {_DIM}{status.detail}{_RESET}"
return line
def _render_progress_color(progress: ProgressIndicator) -> str:
"""Render a progress indicator with ANSI colour codes."""
plain = _render_progress_plain(progress)
# Apply colour to the label
return plain.replace(progress.label, f"{_BOLD}{_CYAN}{progress.label}{_RESET}", 1)
def _render_element_color(element: Any) -> str:
"""Dispatch to the correct colour renderer."""
if isinstance(element, Panel):
return _render_panel_color(element)
if isinstance(element, Table):
return _render_table_color(element)
if isinstance(element, StatusMessage):
return _render_status_color(element)
if isinstance(element, ProgressIndicator):
return _render_progress_color(element)
return str(element) # pragma: no cover
# Box-drawing characters for the ``table`` format
_TOP_LEFT = "+"
_TOP_RIGHT = "+"
_BOTTOM_LEFT = "+"
_BOTTOM_RIGHT = "+"
_HORIZONTAL = "-"
_VERTICAL = "|"
_T_DOWN = "+"
_T_UP = "+"
_T_RIGHT = "+"
_T_LEFT = "+"
_CROSS = "+"
def _render_table_boxdraw(table: Table) -> str:
"""Render a table using ASCII box-drawing (tabulate style)."""
if not table.columns:
return "(empty table)"
col_names = [c.name for c in table.columns]
str_rows: list[list[str]] = []
for row in table.rows:
str_rows.append([str(row.get(c, "")) for c in col_names])
widths = [len(c) for c in col_names]
for row in str_rows:
for i, cell in enumerate(row):
if i < len(widths):
widths[i] = max(widths[i], len(cell))
def make_border(left: str, mid: str, right: str) -> str:
return left + mid.join(_HORIZONTAL * (w + 2) for w in widths) + right
def make_row(cells: list[str]) -> str:
padded = [f" {c.ljust(w)} " for c, w in zip(cells, widths, strict=False)]
return _VERTICAL + _VERTICAL.join(padded) + _VERTICAL
lines: list[str] = []
if table.title:
lines.append(table.title)
lines.append(make_border(_TOP_LEFT, _T_DOWN, _TOP_RIGHT))
lines.append(make_row(col_names))
lines.append(make_border(_T_RIGHT, _CROSS, _T_LEFT))
for row in str_rows:
lines.append(make_row(row))
lines.append(make_border(_BOTTOM_LEFT, _T_UP, _BOTTOM_RIGHT))
return "\n".join(lines)
def _render_element_table(element: Any) -> str:
"""Dispatch for the ``table`` format (box-drawing for tables,
plain for everything else)."""
if isinstance(element, Table):
return _render_table_boxdraw(element)
return _render_element_plain(element)
# ---------------------------------------------------------------------------
# JSON / YAML serialisation helpers
# ---------------------------------------------------------------------------
def _element_to_dict(element: Any) -> dict[str, Any]:
def _tree_node_to_dict(node: TreeNode, _depth: int = 0) -> dict[str, Any]:
"""Convert a TreeNode to a plain dict.
Includes a depth guard (F4 fix) to prevent RecursionError on
pathological trees that bypass the handle's depth enforcement.
L3 fix (Luis review): includes the ``collapsed`` flag in the output.
"""
result: dict[str, Any] = {"label": node.label}
if node.style_hint:
result["style_hint"] = node.style_hint
if node.collapsed:
result["collapsed"] = True
if node.children:
if _depth < MAX_TREE_DEPTH:
result["children"] = [
_tree_node_to_dict(c, _depth + 1) for c in node.children
]
else:
result["truncated"] = True
if node.metadata:
result["metadata"] = node.metadata
return result
def _diff_line_to_dict(line: DiffLine) -> dict[str, Any]:
"""Convert a DiffLine to a plain dict.
Serialises ``line_type`` under the key ``"type"`` for backward
compatibility with the JSON/YAML schema (P2-1 alias).
"""
result: dict[str, Any] = {"type": line.line_type, "content": line.content}
if line.line_number_old is not None:
result["line_number_old"] = line.line_number_old
if line.line_number_new is not None:
result["line_number_new"] = line.line_number_new
return result
def _element_to_dict(element: ElementSnapshot) -> dict[str, Any]:
"""Convert an element snapshot to a plain dict for serialisation."""
if isinstance(element, Panel):
return {
@@ -340,14 +141,23 @@ def _element_to_dict(element: Any) -> dict[str, Any]:
"title": element.title,
"entries": [_panel_entry_to_dict(e) for e in element.entries],
"border_style": element.border_style,
"priority": element.priority,
"collapse_hint": element.collapse_hint,
"metadata": element.metadata,
}
if isinstance(element, Table):
return {
"type": "table",
"title": element.title,
"columns": [c.name for c in element.columns],
"columns": [_column_def_to_dict(c) for c in element.columns],
"rows": element.rows,
"summary": element.summary,
"sort_key": element.sort_key,
"sort_descending": element.sort_descending,
"max_rows_hint": element.max_rows_hint,
"priority": element.priority,
"collapse_hint": element.collapse_hint,
"metadata": element.metadata,
}
if isinstance(element, StatusMessage):
return {
@@ -355,6 +165,8 @@ def _element_to_dict(element: Any) -> dict[str, Any]:
"level": element.level,
"message": element.message,
"detail": element.detail,
"priority": element.priority,
"metadata": element.metadata,
}
if isinstance(element, ProgressIndicator):
result: dict[str, Any] = {
@@ -363,15 +175,91 @@ def _element_to_dict(element: Any) -> dict[str, Any]:
"current": element.current,
"total": element.total,
"indeterminate": element.indeterminate,
"priority": element.priority,
"metadata": element.metadata,
}
if element.steps:
result["steps"] = [
{"label": s.label, "status": s.status} for s in element.steps
]
return result
if isinstance(element, Tree):
return {
"type": "tree",
"root": _tree_node_to_dict(element.root),
"max_depth_hint": element.max_depth_hint,
"show_guides": element.show_guides,
"priority": element.priority,
"collapse_hint": element.collapse_hint,
"metadata": element.metadata,
}
if isinstance(element, TextBlock):
return {
"type": "text",
"content": element.content,
"wrap": element.wrap,
"indent": element.indent,
"priority": element.priority,
"metadata": element.metadata,
}
if isinstance(element, CodeBlock):
result_code: dict[str, Any] = {
"type": "code",
"content": element.content,
"language": element.language,
"line_numbers": element.line_numbers,
"priority": element.priority,
"metadata": element.metadata,
}
if element.highlight_lines:
result_code["highlight_lines"] = element.highlight_lines
return result_code
if isinstance(element, DiffBlock):
result_diff: dict[str, Any] = {
"type": "diff",
"file_a": element.file_a,
"file_b": element.file_b,
"hunks": [
{
"header": h.header,
"lines": [_diff_line_to_dict(ln) for ln in h.lines],
}
for h in element.hunks
],
"priority": element.priority,
"metadata": element.metadata,
}
if element.stats:
result_diff["stats"] = element.stats
return result_diff
if isinstance(element, Separator):
return {"type": "separator", "style": element.style}
if isinstance(element, ActionHint):
return {
"type": "action_hint",
"commands": element.commands,
"description": element.description,
}
return {"type": "unknown"} # pragma: no cover
def _column_def_to_dict(col: ColumnDef) -> dict[str, Any]:
"""Convert a ColumnDef to a plain dict preserving all metadata.
P2-4 fix (Luis review): always include all fields unconditionally
so that machine consumers get a stable set of keys per ColumnDef,
matching the spec's unconditional field listing.
"""
return {
"name": col.name,
"col_type": col.col_type,
"alignment": col.alignment,
"width_hint": col.width_hint,
"sortable": col.sortable,
"style_hint": col.style_hint,
}
def _panel_entry_to_dict(entry: PanelEntry) -> dict[str, Any]:
"""Convert a PanelEntry to a plain dict."""
result: dict[str, Any] = {"key": entry.key, "value": entry.value}
@@ -383,26 +271,28 @@ def _panel_entry_to_dict(entry: PanelEntry) -> dict[str, Any]:
def _snapshot_to_dict(snapshot: StructuredOutput) -> dict[str, Any]:
"""Convert a StructuredOutput to a serialisable dict."""
return {
"""Convert a StructuredOutput to a serialisable dict.
P1-2 fix (Luis review): includes the ``timing`` field when present,
matching the spec's JSON envelope structure (§27022).
"""
result: dict[str, Any] = {
"command": snapshot.command,
"session_id": snapshot.session_id,
"exit_code": snapshot.exit_code,
"elements": [_element_to_dict(e) for e in snapshot.elements],
"metadata": snapshot.metadata,
}
if snapshot.timing is not None:
result["timing"] = snapshot.timing
return result
def _error_envelope(
code: str, message: str, details: dict[str, Any] | None = None
) -> dict[str, Any]:
"""Build a unified error envelope for JSON/YAML output."""
envelope: dict[str, Any] = {
"error": {
"code": code,
"message": message,
}
}
envelope: dict[str, Any] = {"error": {"code": code, "message": message}}
if details:
envelope["error"]["details"] = details
return envelope
@@ -414,9 +304,13 @@ def _error_envelope(
class _BaseBufferStrategy:
"""Base for strategies that buffer output and flush on element close."""
"""Base for strategies that buffer output and flush on element close.
Thread-safe: all buffer mutations are serialised via ``_buf_lock`` (M9).
"""
strategy_name: str = "base_buffer"
supports_incremental_updates: bool = False
def __init__(self) -> None:
self._session: OutputSession | None = None
@@ -425,34 +319,40 @@ class _BaseBufferStrategy:
self._closed_indices: set[int] = set()
self._next_render_index: int = 0
self._index_map: dict[str, int] = {} # handle_id -> declaration_index
self._buf_lock: threading.Lock = threading.Lock()
def on_session_begin(self, session: OutputSession) -> None:
self._session = session
def on_element_created(self, event: ElementCreated) -> None:
self._index_map[event.handle_id] = event.declaration_index
with self._buf_lock:
self._index_map[event.handle_id] = event.declaration_index
def on_element_updated(self, event: ElementUpdated) -> None:
pass # Buffer strategies accumulate via the handle's element state
def _render_element(self, element: Any) -> str:
def _render_element(self, element: ElementSnapshot) -> str:
"""Override in subclasses for format-specific rendering."""
return _render_element_plain(element) # pragma: no cover
return render_element_plain(element) # pragma: no cover
def on_element_closed(self, event: ElementClosed) -> None:
idx = self._index_map.get(event.handle_id)
if idx is None:
# Render outside the lock (may be expensive).
if event.final_state is None:
return # pragma: no cover
rendered = self._render_element(event.final_state)
self._closed_indices.add(idx)
self._buffers[idx] = rendered
# Flush all consecutive closed elements starting from _next_render_index
self._flush_ready()
with self._buf_lock:
idx = self._index_map.get(event.handle_id)
if idx is None:
return # pragma: no cover
self._closed_indices.add(idx)
self._buffers[idx] = rendered
self._flush_ready()
def _flush_ready(self) -> None:
"""Flush buffered content in declaration order."""
"""Flush buffered content in declaration order.
Caller must hold ``_buf_lock``.
"""
while self._next_render_index in self._closed_indices:
content = self._buffers.pop(self._next_render_index, "")
if content:
@@ -461,54 +361,45 @@ class _BaseBufferStrategy:
self._next_render_index += 1
def on_session_end(self, event: SessionEnd) -> None:
# Force-flush any remaining buffered content
for idx in sorted(self._buffers.keys()):
content = self._buffers[idx]
if content:
self._stream.write(content)
self._stream.write("\n\n")
self._buffers.clear()
with self._buf_lock:
for idx in sorted(self._buffers.keys()):
content = self._buffers[idx]
if content:
self._stream.write(content)
self._stream.write("\n\n")
self._buffers.clear()
def get_output(self) -> str:
"""Return all accumulated output."""
return self._stream.getvalue().rstrip("\n")
with self._buf_lock:
return self._stream.getvalue().rstrip("\n")
class PlainMaterializer(_BaseBufferStrategy):
"""Materialisation strategy for the ``plain`` format.
Sequential buffer-flush with ASCII-only output (no ANSI, no Unicode
box characters).
"""
"""Materialisation strategy for the ``plain`` format."""
strategy_name: str = "plain"
def _render_element(self, element: Any) -> str:
return _render_element_plain(element)
def _render_element(self, element: ElementSnapshot) -> str:
return render_element_plain(element)
class ColorMaterializer(_BaseBufferStrategy):
"""Materialisation strategy for the ``color`` format.
ANSI colour codes without cursor movement (scrolling output).
"""
"""Materialisation strategy for the ``color`` format."""
strategy_name: str = "color"
def _render_element(self, element: Any) -> str:
return _render_element_color(element)
def _render_element(self, element: ElementSnapshot) -> str:
return render_element_color(element)
class TableMaterializer(_BaseBufferStrategy):
"""Materialisation strategy for the ``table`` format.
Tabulate-style ASCII box-drawing for tables; plain for other elements.
"""
"""Materialisation strategy for the ``table`` format."""
strategy_name: str = "table"
def _render_element(self, element: Any) -> str:
return _render_element_table(element)
def _render_element(self, element: ElementSnapshot) -> str:
return render_element_table(element)
class RichMaterializer(_BaseBufferStrategy):
@@ -516,20 +407,21 @@ class RichMaterializer(_BaseBufferStrategy):
In a full implementation this would use the Rich library for in-place
terminal updates. For this initial implementation we delegate to the
colour renderer for the rendered output, maintaining forward
compatibility with the session/handle architecture.
colour renderer, maintaining forward compatibility with the
session/handle architecture.
"""
strategy_name: str = "rich"
def _render_element(self, element: Any) -> str:
return _render_element_color(element)
def _render_element(self, element: ElementSnapshot) -> str:
return render_element_color(element)
class _AccumulateStrategy:
"""Base for strategies that accumulate all output and serialise on close."""
strategy_name: str = "accumulate"
supports_incremental_updates: bool = False
def __init__(self) -> None:
self._session: OutputSession | None = None
@@ -541,10 +433,10 @@ class _AccumulateStrategy:
pass
def on_element_updated(self, event: ElementUpdated) -> None:
pass
return # accumulate strategies ignore incremental updates
def on_element_closed(self, event: ElementClosed) -> None:
pass
return # accumulate strategies ignore element close events
def _serialise(self, snapshot: StructuredOutput) -> str:
"""Override in subclasses."""
@@ -558,7 +450,6 @@ class _AccumulateStrategy:
if self._session is None:
return ""
snapshot = self._session.snapshot()
snapshot.exit_code = 0
return self._serialise(snapshot)
def get_error_output(
@@ -577,11 +468,7 @@ class _AccumulateStrategy:
class JsonMaterializer(_AccumulateStrategy):
"""Materialisation strategy for the ``json`` format.
Accumulates all element data and serialises as a single JSON document
when the session closes.
"""
"""Materialisation strategy for the ``json`` format."""
strategy_name: str = "json"
@@ -593,16 +480,12 @@ class JsonMaterializer(_AccumulateStrategy):
class YamlMaterializer(_AccumulateStrategy):
"""Materialisation strategy for the ``yaml`` format.
Accumulates all element data and dumps as a single YAML document
when the session closes.
"""
"""Materialisation strategy for the ``yaml`` format."""
strategy_name: str = "yaml"
def _serialise(self, snapshot: StructuredOutput) -> str:
return yaml.dump(
return yaml.safe_dump(
_snapshot_to_dict(snapshot),
default_flow_style=False,
sort_keys=False,
@@ -610,7 +493,7 @@ class YamlMaterializer(_AccumulateStrategy):
).rstrip("\n")
def _serialise_dict(self, data: dict[str, Any]) -> str:
return yaml.dump(
return yaml.safe_dump(
data,
default_flow_style=False,
sort_keys=False,
+34 -4
View File
@@ -2,7 +2,7 @@
Determines the correct materialisation strategy for the resolved output
format, applying automatic fallbacks when terminal capabilities are
insufficient (e.g. Rich Color Plain when stdout is not a TTY).
insufficient (e.g. Rich Table Color Plain).
Based on the Output Rendering Framework specification (v3_spec.md §Output
Rendering Framework / Format Resolution).
@@ -49,7 +49,17 @@ def detect_terminal_capabilities() -> TerminalCapabilities:
is_tty = hasattr(sys.stdout, "isatty") and sys.stdout.isatty()
term = os.environ.get("TERM", "")
supports_ansi = is_tty and term.lower() != "dumb"
supports_cursor = supports_ansi
# Cursor movement requires a terminal emulator that supports cursor
# addressing beyond basic ANSI colour codes. We check for known
# cursor-capable $TERM values. This allows the state
# ``supports_ansi=True, supports_cursor=False`` (e.g. a colour-only
# pipe), making the Color fallback tier reachable in the
# Rich → Table → Color → Plain chain (spec §27177, N15 fix).
_cursor_terms = {"xterm", "screen", "tmux", "rxvt", "alacritty", "kitty"}
term_lower = term.lower()
supports_cursor = supports_ansi and any(
term_lower.startswith(ct) for ct in _cursor_terms
)
return TerminalCapabilities(
is_tty=is_tty,
supports_ansi=supports_ansi,
@@ -107,14 +117,32 @@ def select_materializer(
caps = capabilities if capabilities is not None else detect_terminal_capabilities()
# When user explicitly sets a format, respect it without fallback
# When user explicitly sets a format, respect it without fallback.
# Explicit selection takes precedence over NO_COLOR.
if explicit:
return _create_strategy(fmt)
# Automatic fallback: Rich → Color → Plain
# P1-4 fix (Luis review): respect NO_COLOR env var per
# https://no-color.org/ and spec §26774. When set, all visual
# formats (rich, color, table) fall back to plain. This replaces
# SD-14 (previously documented as unimplemented).
# Precedence: explicit flag > NO_COLOR > terminal capability fallback.
# Only checked when capabilities are auto-detected (not when
# explicitly passed by callers simulating a specific environment).
if capabilities is None and os.environ.get("NO_COLOR") is not None:
return PlainMaterializer()
# Automatic fallback: Rich → Table → Color → Plain (N15 — spec §27177)
# Note: The Color tier (supports_ansi without supports_cursor) is
# narrowly reachable only when $TERM is a non-cursor-capable value
# but the output is still a TTY. In practice, most TTYs have
# cursor-capable $TERM values, so the Rich → Table path dominates.
# This is documented as SD-16 in __init__.py (R2-C2 fix #8).
if fmt == "rich":
if caps.supports_cursor:
return RichMaterializer()
if caps.is_tty:
return TableMaterializer()
if caps.supports_ansi:
return ColorMaterializer()
return PlainMaterializer()
@@ -127,6 +155,8 @@ def select_materializer(
if fmt == "table":
if caps.is_tty:
return TableMaterializer()
if caps.supports_ansi:
return ColorMaterializer()
return PlainMaterializer()
# plain — always works
+211 -59
View File
@@ -13,28 +13,46 @@ from __future__ import annotations
import threading
import time
from collections import OrderedDict
from collections.abc import Callable
from typing import TYPE_CHECKING, Any
from pydantic import BaseModel, Field
from cleveragents.cli.output._ids import generate_handle_id, generate_session_id
from cleveragents.cli.output.handles import (
MAX_ELEMENTS_PER_SESSION,
ActionHint,
ActionHintHandle,
CodeBlock,
CodeHandle,
ColumnDef,
DiffBlock,
DiffHandle,
ElementClosed,
ElementCreated,
ElementEvent,
ElementHandle,
ElementSnapshot,
ElementUpdated,
Panel,
PanelHandle,
ProgressHandle,
ProgressIndicator,
ProgressStep,
Separator,
SeparatorHandle,
SessionEnd,
StatusHandle,
StatusMessage,
Table,
TableHandle,
TextBlock,
TextHandle,
Tree,
TreeHandle,
TreeNode,
)
from cleveragents.cli.output.selection import select_materializer
if TYPE_CHECKING:
from cleveragents.cli.output.materializers import MaterializationStrategy
@@ -95,20 +113,20 @@ class OutputSession:
strategy: MaterializationStrategy | None = None,
metadata: dict[str, Any] | None = None,
) -> None:
from cleveragents.cli.output.selection import select_materializer
self.command = command
self.session_id = _generate_session_id()
self.session_id = generate_session_id()
self.created_at = time.time()
self._format = format
self._strategy: MaterializationStrategy = (
strategy if strategy is not None else select_materializer(format)
)
self._handles: OrderedDict[str, ElementHandle] = OrderedDict()
self._handles: OrderedDict[str, ElementHandle[ElementSnapshot]] = OrderedDict()
self._next_index: int = 0
self._state: str = "open"
self._lock: threading.Lock = threading.Lock()
self._metadata: dict[str, Any] = metadata or {}
self._exit_code: int = 0
self._timing: dict[str, Any] | None = None
self._strategy.on_session_begin(self)
# --- Element handle factories ---
@@ -178,6 +196,9 @@ class OutputSession:
"""Create a status message handle."""
if not message:
raise ValueError("message cannot be empty")
valid_levels = {"ok", "warn", "error", "info"}
if level not in valid_levels:
raise ValueError(f"level must be one of {valid_levels!r}, got {level!r}")
element = StatusMessage(
message=message,
level=level,
@@ -197,9 +218,15 @@ class OutputSession:
priority: str = "normal",
metadata: dict[str, Any] | None = None,
) -> ProgressHandle:
"""Create a progress indicator handle."""
"""Create a progress indicator handle.
P1-1 fix (Luis review): validates *total* is non-negative at
the factory level, consistent with ``set_progress()``.
"""
if not label:
raise ValueError("label cannot be empty")
if total is not None and total < 0:
raise ValueError("total must be non-negative")
step_objs: list[ProgressStep] | None = None
if steps is not None:
step_objs = [ProgressStep(label=s) for s in steps]
@@ -214,39 +241,162 @@ class OutputSession:
)
return self._register_handle(ProgressHandle, element, "progress")
def tree(
self,
root_label: str,
*,
root_style: str | None = None,
max_depth_hint: int | None = None,
show_guides: bool = True,
priority: str = "normal",
collapse_hint: str = "auto",
metadata: dict[str, Any] | None = None,
) -> TreeHandle:
"""Create a tree element handle for hierarchical data."""
if not root_label:
raise ValueError("root_label cannot be empty")
root_node = TreeNode(label=root_label, style_hint=root_style)
element = Tree(
root=root_node,
max_depth_hint=max_depth_hint,
show_guides=show_guides,
priority=priority,
collapse_hint=collapse_hint,
metadata=metadata or {},
)
return self._register_handle(TreeHandle, element, "tree")
def text(
self,
content: str = "",
*,
wrap: bool = True,
indent: int = 0,
priority: str = "normal",
metadata: dict[str, Any] | None = None,
) -> TextHandle:
"""Create a text block handle for free-form text."""
element = TextBlock(
content=content,
wrap=wrap,
indent=indent,
priority=priority,
metadata=metadata or {},
)
return self._register_handle(TextHandle, element, "text")
def code(
self,
content: str = "",
*,
language: str | None = None,
line_numbers: bool = False,
highlight_lines: list[int] | None = None,
priority: str = "normal",
metadata: dict[str, Any] | None = None,
) -> CodeHandle:
"""Create a code block handle for source code display."""
element = CodeBlock(
content=content,
language=language,
line_numbers=line_numbers,
highlight_lines=highlight_lines or [],
priority=priority,
metadata=metadata or {},
)
return self._register_handle(CodeHandle, element, "code")
def diff(
self,
*,
file_a: str | None = None,
file_b: str | None = None,
priority: str = "normal",
metadata: dict[str, Any] | None = None,
) -> DiffHandle:
"""Create a diff block handle for unified-diff display."""
element = DiffBlock(
file_a=file_a,
file_b=file_b,
priority=priority,
metadata=metadata or {},
)
return self._register_handle(DiffHandle, element, "diff")
def separator(self, style: str = "line") -> SeparatorHandle:
"""Create a separator element (auto-closed on creation)."""
valid_styles = {"line", "blank", "double"}
if style not in valid_styles:
raise ValueError(f"style must be one of {valid_styles!r}, got {style!r}")
element = Separator(style=style)
handle: SeparatorHandle = self._register_handle(
SeparatorHandle, element, "separator"
)
handle.close()
return handle
def action_hint(
self,
commands: list[str],
description: str = "",
) -> ActionHintHandle:
"""Create an action-hint element (auto-closed on creation)."""
if not commands:
raise ValueError("commands cannot be empty")
element = ActionHint(commands=list(commands), description=description)
handle: ActionHintHandle = self._register_handle(
ActionHintHandle, element, "action_hint"
)
handle.close()
return handle
# --- Internal registration ---
def _register_handle(
def _register_handle[H: ElementHandle[Any]](
self,
handle_cls: type,
handle_cls: Callable[..., H],
element: ElementSnapshot,
element_kind: str,
) -> Any:
"""Create a handle, register it, and emit a created event."""
) -> H:
"""Create a handle, register it, and emit a created event.
Uses a single lock scope to prevent TOCTOU races between state
check, index allocation, handle registration, and event dispatch.
The ``ElementCreated`` event is dispatched inside the lock to
prevent a race where ``session.close()`` could force-close the
handle before the created event reaches the strategy (N1 fix).
Enforces ``MAX_ELEMENTS_PER_SESSION`` soft limit (F3 fix).
"""
with self._lock:
if self._state != "open":
raise RuntimeError("Cannot create handles on a closed session")
handle_id = _generate_handle_id()
if len(self._handles) >= MAX_ELEMENTS_PER_SESSION:
limit = MAX_ELEMENTS_PER_SESSION
raise ValueError(f"Maximum elements per session ({limit}) exceeded")
handle_id = generate_handle_id()
idx = self._next_index
self._next_index += 1
handle = handle_cls(
handle_id=handle_id,
declaration_index=idx,
session=self,
element=element,
)
with self._lock:
handle: H = handle_cls(
handle_id=handle_id,
declaration_index=idx,
session=self,
element=element,
)
self._handles[handle_id] = handle
event = ElementCreated(
event_type="created",
handle_id=handle_id,
element_kind=element_kind,
declaration_index=idx,
initial_state=element,
)
self._dispatch_event(event)
# Dispatch inside the lock to ensure ElementCreated arrives
# before any ElementClosed from session.close() (N1 fix).
event = ElementCreated(
event_type="created",
handle_id=handle_id,
element_kind=element_kind,
declaration_index=idx,
initial_state=element,
)
self._dispatch_event(event)
return handle
# --- Event dispatch ---
@@ -255,32 +405,57 @@ class OutputSession:
"""Route an event to the active strategy."""
if isinstance(event, ElementCreated):
self._strategy.on_element_created(event)
elif isinstance(event, ElementUpdated):
self._strategy.on_element_updated(event)
elif isinstance(event, ElementClosed):
self._strategy.on_element_closed(event)
elif isinstance(event, SessionEnd):
self._strategy.on_session_end(event)
else:
self._strategy.on_element_updated(event) # type: ignore[arg-type]
# --- Session operations ---
def snapshot(self) -> StructuredOutput:
"""Return a static snapshot of all elements accumulated so far."""
"""Return a static snapshot of all elements accumulated so far.
F6 fix: uses ``element_copy()`` for deep copies to prevent
mutation of live session state.
N2 fix: includes ``exit_code`` from the session.
P1-2 fix (Luis review): includes ``timing`` when available so
that accumulate strategies (JSON/YAML) emit the timing field.
"""
with self._lock:
elements = [h.element for h in self._handles.values()]
return StructuredOutput(
elements = [h.element_copy() for h in self._handles.values()]
snap = StructuredOutput(
command=self.command,
session_id=self.session_id,
elements=list(elements),
exit_code=self._exit_code,
metadata=dict(self._metadata),
)
if self._timing is not None:
snap.timing = self._timing
return snap
def close(self, *, exit_code: int = 0) -> StructuredOutput:
"""Close the session and finalise all output."""
with self._lock:
if self._state == "closed":
return self.snapshot()
# Already closed — return snapshot with cached timing
# (R2-C2 fix: previously timing was None on double-close).
elements = [h.element_copy() for h in self._handles.values()]
snap = StructuredOutput(
command=self.command,
session_id=self.session_id,
elements=list(elements),
exit_code=self._exit_code,
metadata=dict(self._metadata),
)
snap.timing = self._timing
return snap
self._state = "closed"
self._exit_code = exit_code
# Force-close any open handles
for handle in list(self._handles.values()):
@@ -288,12 +463,13 @@ class OutputSession:
handle.close()
snap = self.snapshot()
snap.exit_code = exit_code
snap.timing = {
now = time.time()
self._timing = {
"start_time": self.created_at,
"end_time": time.time(),
"duration": time.time() - self.created_at,
"end_time": now,
"duration": now - self.created_at,
}
snap.timing = self._timing
event = SessionEnd(
event_type="session_end",
@@ -318,27 +494,3 @@ class OutputSession:
if self._state != "closed":
code = 1 if exc_type is not None else 0
self.close(exit_code=code)
# ---------------------------------------------------------------------------
# ID generation helpers
# ---------------------------------------------------------------------------
_COUNTER_LOCK = threading.Lock()
_COUNTER = 0
def _generate_session_id() -> str:
"""Generate a unique session identifier."""
global _COUNTER
with _COUNTER_LOCK:
_COUNTER += 1
return f"ses-{_COUNTER:06d}"
def _generate_handle_id() -> str:
"""Generate a unique handle identifier."""
global _COUNTER
with _COUNTER_LOCK:
_COUNTER += 1
return f"hdl-{_COUNTER:06d}"