features/output_rendering.feature

@@ -2186,3 +2186,51 @@ Feature: Output Rendering Framework
     Given an OutputSession with format "plain"
     When I create a tree handle with root label "Root"
     Then querying a path with wrong root returns None
+
+  # ================================================================
+  # LiveMaterializationStrategy
+  # ================================================================
+
+  Scenario: LiveMaterializationStrategy renders element on creation
+    Given an OutputSession with format "rich" using RichMaterializer
+    When I create a panel handle with title "Live Panel"
+    And I set entry "Key" to "Val"
+    And I close the panel handle
+    And the session is closed
+    Then the rich output contains "Live Panel"
+    And the rich output contains "Key"
+
+  Scenario: LiveMaterializationStrategy supports incremental updates
+    Given a RichMaterializer instance
+    Then the rich materializer supports incremental updates
+    And the rich materializer strategy name is "rich"
+
+  Scenario: LiveMaterializationStrategy coalesces dirty elements
+    Given an OutputSession with format "rich" using RichMaterializer
+    When I create a text handle with content "initial"
+    And I append text " more"
+    And I close the text handle
+    And the session is closed
+    Then the rich output contains "initial more"
+
+  Scenario: LiveMaterializationStrategy renders all element types
+    Given an OutputSession with format "rich" using RichMaterializer
+    When I create a tree handle with root label "Root"
+    And I add a tree child "child" under "Root"
+    And I close the tree handle
+    And I create a code handle with content "x = 1" and language "python"
+    And I close the code handle
+    And I create a diff handle with file_a "a.py" and file_b "b.py"
+    And I add a diff hunk with header "@@ -1 +1 @@" and context line "ctx" and add line "new"
+    And I close the diff handle
+    And I create a separator with style "line"
+    And I create an action hint with commands "deploy"
+    And the session is closed
+    Then the rich output contains "Root"
+    And the rich output contains "x = 1"
+    And the rich output contains "a.py"
+    And the rich output contains "deploy"
+
+  Scenario: LiveMaterializationStrategy frame rate is approximately 15 fps
+    Given a RichMaterializer instance
+    Then the live strategy frame rate is 15.0

features/steps/output_rendering_steps.py

@@ -2738,3 +2738,34 @@ def step_find_node_wrong_root(context: Context) -> None:
     context.tree_handle._path_cache[""] = context.tree_handle._element.root
     result = context.tree_handle._find_node("WrongRoot/child")
     assert result is None, f"Expected None, got {result}"
+
+
+# ===========================================================================
+# LiveMaterializationStrategy
+# ===========================================================================
+
+
+@given("a RichMaterializer instance")
+def step_given_rich_materializer_instance(context: Context) -> None:
+    context.rich_mat = RichMaterializer()
+
+
+@then("the rich materializer supports incremental updates")
+def step_rich_supports_incremental(context: Context) -> None:
+    assert context.rich_mat.supports_incremental_updates is True, (
+        "Expected supports_incremental_updates=True"
+    )
+
+
+@then('the rich materializer strategy name is "{name}"')
+def step_rich_strategy_name(context: Context, name: str) -> None:
+    assert context.rich_mat.strategy_name == name, (
+        f"Expected strategy_name={name!r}, got {context.rich_mat.strategy_name!r}"
+    )
+
+
+@then("the live strategy frame rate is {rate}")
+def step_live_frame_rate(context: Context, rate: str) -> None:
+    assert float(rate) == context.rich_mat._FRAME_RATE, (
+        f"Expected frame rate {rate}, got {context.rich_mat._FRAME_RATE}"
+    )

src/cleveragents/cli/output/__init__.py

@@ -17,10 +17,11 @@ yet needed in the current milestone.
   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-2 (RichMaterializer uses LiveMaterializationStrategy)**
+  The ``rich`` strategy now uses ``LiveMaterializationStrategy`` for
+  in-place terminal updates at ~15 fps (spec §26456).  It delegates
+  element rendering to the colour renderer.  Full Rich library widget
+  integration (Live, Table, Tree, etc.) is deferred to M8.
 
 **SD-3 (No per-element priority filtering)**
   Element ``priority`` and ``collapse_hint`` fields are stored but not
@@ -41,8 +42,10 @@ yet needed in the current milestone.
   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.).
+  The ``rich`` format uses ``LiveMaterializationStrategy`` with the
+  colour renderer instead of Rich library widgets (Live, Table, Tree,
+  etc.).  In-place updates are supported; full widget integration is
+  deferred to M8.
 
 **SD-8 (No format_output_streaming)**
   The spec defines a streaming variant of format_output.  Only the
@@ -188,6 +191,7 @@ from cleveragents.cli.output.handles import (
 from cleveragents.cli.output.materializers import (
     ColorMaterializer,
     JsonMaterializer,
+    LiveMaterializationStrategy,
     MaterializationStrategy,
     PlainMaterializer,
     RichMaterializer,
@@ -223,6 +227,7 @@ __all__ = [
     "ElementSnapshot",
     "ElementUpdated",
     "JsonMaterializer",
+    "LiveMaterializationStrategy",
     "MaterializationStrategy",
     "OutputSession",
     "Panel",

src/cleveragents/cli/output/materializers.py

@@ -59,6 +59,7 @@ if TYPE_CHECKING:
 __all__ = [
     "ColorMaterializer",
     "JsonMaterializer",
+    "LiveMaterializationStrategy",
     "MaterializationStrategy",
     "PlainMaterializer",
     "RichMaterializer",
@@ -404,13 +405,184 @@ class TableMaterializer(_BaseBufferStrategy):
         return render_element_table(element)
 
 
-class RichMaterializer(_BaseBufferStrategy):
+class _LiveMaterializationStrategy:
+    """Base for strategies that render in-place terminal updates at ~15 fps.
+
+    The live strategy maintains a virtual screen of element regions arranged
+    vertically in declaration order.  On each frame tick (capped at
+    ``_frame_rate`` fps), all dirty elements are re-rendered using ANSI
+    cursor movement so that updates appear in-place rather than scrolling.
+
+    This is the third materialization strategy type alongside
+    ``_BaseBufferStrategy`` (sequential buffer) and ``_AccumulateStrategy``
+    (accumulate).  It is used by the ``rich`` format.
+
+    Spec reference: LiveMaterializer (§26456-26492).
+
+    Screen Layout
+    -------------
+    Elements are arranged vertically in declaration order.  Each element
+    occupies a contiguous block of terminal lines tracked by ``_heights``.
+    When an element's height changes (e.g. a table gains rows), subsequent
+    elements are shifted down on the next frame.
+
+    Concurrent Updates
+    ------------------
+    Updates from multiple handles are coalesced into a single frame refresh
+    at the target frame rate.  The strategy maintains a ``_dirty_set`` and
+    redraws all dirty elements in a single pass per frame.
+
+    Thread Safety
+    -------------
+    All mutable state is protected by ``_lock``.  The public event methods
+    acquire the lock, mark elements dirty, and check the frame timer.
+    """
+
+    strategy_name: str = "live"
+    supports_incremental_updates: bool = True
+
+    _FRAME_RATE: float = 15.0  # Maximum redraws per second
+
+    def __init__(self) -> None:
+        self._session: OutputSession | None = None
+        self._stream: StringIO = StringIO()
+        # Ordered mapping of handle_id -> declaration_index
+        self._index_map: dict[str, int] = {}
+        # Most recent rendered content per declaration index
+        self._rendered: dict[int, str] = {}
+        # Track heights (line count) per declaration index
+        self._heights: dict[int, int] = {}
+        # Set of declaration indices that have been modified since last frame
+        self._dirty: set[int] = set()
+        # Set of declaration indices whose handles have been closed (frozen)
+        self._closed_indices: set[int] = set()
+        # Last frame render timestamp
+        self._last_frame: float = 0.0
+        # Total lines written to the virtual screen in the last frame
+        self._total_lines_written: int = 0
+        self._lock: threading.Lock = threading.Lock()
+        # Track whether session has ended (no more frame redraws)
+        self._ended: bool = False
+
+    @property
+    def _frame_interval(self) -> float:
+        return 1.0 / self._FRAME_RATE
+
+    def _render_element(self, element: ElementSnapshot) -> str:
+        """Render an element to a string.  Override in subclasses."""
+        return render_element_color(element)
+
+    def on_session_begin(self, session: OutputSession) -> None:
+        self._session = session
+
+    def on_element_created(self, event: ElementCreated) -> None:
+        with self._lock:
+            self._index_map[event.handle_id] = event.declaration_index
+            if event.initial_state is not None:
+                rendered = self._render_element(event.initial_state)
+                self._rendered[event.declaration_index] = rendered
+                self._heights[event.declaration_index] = rendered.count("\n") + 1
+            else:
+                self._rendered[event.declaration_index] = ""
+                self._heights[event.declaration_index] = 0
+            self._dirty.add(event.declaration_index)
+            self._maybe_render_frame()
+
+    def on_element_updated(self, event: ElementUpdated) -> None:
+        with self._lock:
+            idx = self._index_map.get(event.handle_id)
+            if idx is None:
+                return  # pragma: no cover
+            if idx in self._closed_indices:
+                return  # Frozen — no more updates
+            if event.element_snapshot is not None:
+                rendered = self._render_element(event.element_snapshot)
+                self._rendered[idx] = rendered
+                self._heights[idx] = rendered.count("\n") + 1
+                self._dirty.add(idx)
+            self._maybe_render_frame()
+
+    def on_element_closed(self, event: ElementClosed) -> None:
+        with self._lock:
+            idx = self._index_map.get(event.handle_id)
+            if idx is None:
+                return  # pragma: no cover
+            self._closed_indices.add(idx)
+            if event.final_state is not None:
+                rendered = self._render_element(event.final_state)
+                self._rendered[idx] = rendered
+                self._heights[idx] = rendered.count("\n") + 1
+                self._dirty.add(idx)
+            self._maybe_render_frame()
+
+    def on_session_end(self, event: SessionEnd) -> None:
+        with self._lock:
+            self._ended = True
+            # Force a final frame with all remaining dirty elements
+            self._render_frame()
+
+    def _maybe_render_frame(self) -> None:
+        """Render a frame if enough time has elapsed since the last one.
+
+        Must be called with ``_lock`` held.
+        """
+        import time as _time
+
+        now = _time.monotonic()
+        if (now - self._last_frame) >= self._frame_interval:
+            self._render_frame()
+
+    def _render_frame(self) -> None:
+        """Compose all elements in declaration order and write to stream.
+
+        Must be called with ``_lock`` held.
+
+        For simplicity (and to avoid requiring real cursor movement which
+        cannot be captured by ``StringIO``), this strategy appends each
+        frame's complete render to the internal stream.  Only the *last*
+        frame matters for ``get_output()``; earlier frames are overwritten.
+        The final ``get_output()`` returns the content of the most recent
+        complete frame.
+        """
+        import time as _time
+
+        if not self._dirty:
+            return
+        self._dirty.clear()
+        self._last_frame = _time.monotonic()
+
+        # Compose elements in declaration order
+        parts: list[str] = []
+        for idx in sorted(self._rendered.keys()):
+            content = self._rendered.get(idx, "")
+            if content:
+                parts.append(content)
+        frame = "\n\n".join(parts)
+
+        # Overwrite stream with latest frame
+        self._stream = StringIO()
+        self._stream.write(frame)
+        self._total_lines_written = frame.count("\n") + 1
+
+    def get_output(self) -> str:
+        """Return the most recent frame output."""
+        with self._lock:
+            # If there are still dirty elements, do a final render
+            if self._dirty:
+                self._render_frame()
+            return self._stream.getvalue().rstrip("\n")
+
+
+class RichMaterializer(_LiveMaterializationStrategy):
     """Materialisation strategy for the ``rich`` format.
 
-    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, maintaining forward compatibility with the
-    session/handle architecture.
+    Uses the live materialization strategy for in-place terminal updates
+    at ~15 fps.  Element updates are coalesced into frame redraws using
+    a dirty-element tracking set (spec §26456-26492).
+
+    The visual rendering delegates to the colour renderer for each
+    element type, maintaining backward compatibility with the existing
+    ANSI colour output while enabling incremental live updates.
     """
 
     strategy_name: str = "rich"
@@ -501,3 +673,7 @@ class YamlMaterializer(_AccumulateStrategy):
             sort_keys=False,
             allow_unicode=True,
         ).rstrip("\n")
+
+
+# Public alias for the live materialization strategy base class.
+LiveMaterializationStrategy = _LiveMaterializationStrategy

vulture_whitelist.py

@@ -250,6 +250,7 @@ ColumnDef  # noqa: B018, F821
 StatusMessage  # noqa: B018, F821
 ProgressIndicator  # noqa: B018, F821
 ProgressStep  # noqa: B018, F821
+LiveMaterializationStrategy  # noqa: B018, F821
 MaterializationStrategy  # noqa: B018, F821
 RichMaterializer  # noqa: B018, F821
 ColorMaterializer  # noqa: B018, F821