diff --git a/CHANGELOG.md b/CHANGELOG.md
index b3cd2bfd4..55341d793 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -13,6 +13,7 @@ Changed `wf10_batch.robot` to be less likely to create files, and
 - **fix(cli/plan): plan correct JSON output envelope fix and BDD test coverage** (#8584 / PR #8662): Restructured `agents plan correct --format json` output to nest correction fields under `data.correction` (e.g., `data.correction.mode`) and populate the spec-required CLI envelope with `command="plan correct"`, `status`, `exit_code`, `timing`, and `messages` fields. Added three BDD scenarios in `features/tdd_plan_correct_json_output.feature` validating the envelope structure for both revert and append modes.
 - **fix(cli): add --url flag to resource add for git resource type** (#6322): Added support for the `--url` flag on `agents resource add git` command, allowing users to specify a remote URL for git resources. The flag is validated to only apply to git resource types. Includes Behave BDD tests in `features/resource_cli_git_url_flag.feature` and Robot Framework integration tests verifying correct URL validation and CLI behavior.
 - **Session create JSON envelope** (#6441): Fixed `agents session create --format json` returning a flat `data` dict instead of the spec-required nested structure with `data.session`, `data.settings`, and `data.actor_details` sub-objects. The `command` field is now populated correctly. Extended JSON envelope coverage to `agents session list`, `show`, `delete --format json`, `export --output-format json`, and `import --format json` so all session commands emit a structured `messages[].text` field (`"0 sessions listed"`, `"Session details loaded"`, `"Session deleted"`, `"Export completed"`, `"Import completed"`).
+- **feat(tui): conversation content pruning** (#6350): Added `ConversationStream` to the TUI layer implementing hysteresis-based line-count pruning. When the rendered conversation exceeds `trigger_line_count` (`prune_low_mark + prune_excess`, defaults 1 500 + 1 000 = 2 500 lines), the oldest non-protected blocks are removed until total lines falls back to `prune_low_mark`. A styled pruning note is inserted at the head of the visible conversation. Pruning thresholds are configurable via `~/.config/cleveragents/tui-settings.json` (`ui.prune_low_mark`, `ui.prune_excess`). Includes Behave BDD tests, Robot Framework integration tests, and ASV performance benchmarks.
 - **fix(resources): remove unsupported executable resource type and fix resource list columns** (#3077 / PR #3248): Removed `executable` from `LSP_RESOURCE_TYPES` and `BUILTIN_TYPE_NAMES` (the specification defines no such built-in type). Updated `agents resource list` CLI table columns from `[ID, Name, Type, Status, Kind, Location, Description]` to the spec-required `[Name, ID, Type, Phys/Virt, Children, Projects]`. Deleted orphaned `examples/resource-types/executable.yaml`. Lifecycle state for container resources is now displayed as a note below the resource table.
 - **fix(cli): add Read-Only and Writes columns to tool list output** (#1476): Rewrote
   `list_tools()` in `src/cleveragents/cli/commands/tool.py` to render exactly the 5
diff --git a/benchmarks/conversation_stream_bench.py b/benchmarks/conversation_stream_bench.py
new file mode 100644
index 000000000..3e4867a93
--- /dev/null
+++ b/benchmarks/conversation_stream_bench.py
@@ -0,0 +1,99 @@
+"""ASV benchmarks for ConversationStream pruning performance.
+
+Measures the performance of:
+- ConversationStream.add_block() with no pruning (baseline)
+- ConversationStream.add_block() under heavy load (pruning fires on every block)
+- ConversationStream.render() on a pruned stream
+- ConversationStream.clear() followed by repopulation
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+try:
+    from cleveragents.tui.conversation import ConversationSettings, ConversationStream
+except ModuleNotFoundError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+    from cleveragents.tui.conversation import ConversationSettings, ConversationStream
+
+
+def _make_stream(
+    prune_low_mark: int = 1_500,
+    prune_excess: int = 1_000,
+    preserve_recent_lines: int = 500,
+) -> ConversationStream:
+    settings = ConversationSettings(
+        prune_low_mark=prune_low_mark,
+        prune_excess=prune_excess,
+        preserve_recent_lines=preserve_recent_lines,
+    )
+    return ConversationStream(settings=settings)
+
+
+_BLOCK_10_LINES = "\n".join(f"bench line {i}" for i in range(10))
+_BLOCK_50_LINES = "\n".join(f"bench line {i}" for i in range(50))
+
+
+class ConversationStreamAddBlockSuite:
+    """Benchmark ConversationStream.add_block under varying load."""
+
+    def setup(self):
+        # Pre-built stream near (but not at) the default trigger threshold.
+        self.stream_near_threshold = _make_stream()
+        for _i in range(240):
+            self.stream_near_threshold.add_block(_BLOCK_10_LINES, block_type="message")
+        # (240 * 10 = 2400 lines, trigger = 2500 — still below threshold)
+
+    def time_add_block_no_prune(self):
+        """Baseline: add a single block to an empty stream (no pruning)."""
+        stream = _make_stream()
+        stream.add_block(_BLOCK_10_LINES, block_type="message")
+
+    def time_add_block_triggers_prune(self):
+        """Add one block past the threshold so pruning fires once."""
+        stream = _make_stream()
+        for _i in range(241):
+            stream.add_block(_BLOCK_10_LINES, block_type="message")
+
+    def time_add_block_heavy_load(self):
+        """Add 500 blocks to a tight-threshold stream (pruning fires repeatedly)."""
+        stream = _make_stream(prune_low_mark=100, prune_excess=50)
+        for _ in range(500):
+            stream.add_block(_BLOCK_10_LINES, block_type="message")
+
+    def time_add_block_large_blocks(self):
+        """Add 50-line blocks; fewer blocks needed to trigger pruning."""
+        stream = _make_stream(prune_low_mark=100, prune_excess=50)
+        for _ in range(100):
+            stream.add_block(_BLOCK_50_LINES, block_type="message")
+
+
+class ConversationStreamRenderSuite:
+    """Benchmark ConversationStream.render() after pruning."""
+
+    def setup(self):
+        self.stream = _make_stream(prune_low_mark=100, prune_excess=50)
+        for _i in range(200):
+            self.stream.add_block(_BLOCK_10_LINES, block_type="message")
+
+    def time_render_pruned_stream(self):
+        """Benchmark render() on a stream that has been pruned."""
+        self.stream.render()
+
+
+class ConversationStreamClearSuite:
+    """Benchmark ConversationStream.clear() and repopulation."""
+
+    def setup(self):
+        self.block = _BLOCK_10_LINES
+
+    def time_clear_and_repopulate(self):
+        """Benchmark clear() followed by adding 50 blocks."""
+        stream = _make_stream()
+        for _ in range(50):
+            stream.add_block(self.block, block_type="message")
+        stream.clear()
+        for _ in range(50):
+            stream.add_block(self.block, block_type="message")
diff --git a/features/steps/tui_conversation_module_coverage_steps.py b/features/steps/tui_conversation_module_coverage_steps.py
new file mode 100644
index 000000000..43374daf1
--- /dev/null
+++ b/features/steps/tui_conversation_module_coverage_steps.py
@@ -0,0 +1,308 @@
+"""Step definitions for direct coverage of cleveragents.tui.conversation."""
+
+from __future__ import annotations
+
+import json
+import tempfile
+from pathlib import Path
+
+from behave import then, when
+
+from cleveragents.tui.conversation import (
+    ConversationBlock,
+    ConversationSettings,
+    ConversationStream,
+    load_conversation_settings,
+)
+
+
+def _unescape(value: str) -> str:
+    return value.encode("utf-8").decode("unicode_escape")
+
+
+@when("I build default ConversationSettings")
+def step_build_default_settings(context):
+    context._conv_settings = ConversationSettings()
+
+
+@when(
+    "I build ConversationSettings with prune_low_mark {low:d} "
+    "prune_excess {excess:d} preserve_recent_lines {preserve:d}"
+)
+def step_build_explicit_settings(context, low, excess, preserve):
+    context._conv_settings = ConversationSettings(
+        prune_low_mark=low,
+        prune_excess=excess,
+        preserve_recent_lines=preserve,
+    )
+
+
+@then("conversation settings prune_low_mark should equal {expected:d}")
+def step_assert_settings_low(context, expected):
+    assert context._conv_settings.prune_low_mark == expected, (
+        f"expected prune_low_mark={expected}, "
+        f"got {context._conv_settings.prune_low_mark}"
+    )
+
+
+@then("conversation settings prune_excess should equal {expected:d}")
+def step_assert_settings_excess(context, expected):
+    assert context._conv_settings.prune_excess == expected, (
+        f"expected prune_excess={expected}, got {context._conv_settings.prune_excess}"
+    )
+
+
+@then("conversation settings preserve_recent_lines should equal {expected:d}")
+def step_assert_settings_preserve(context, expected):
+    assert context._conv_settings.preserve_recent_lines == expected, (
+        f"expected preserve_recent_lines={expected}, "
+        f"got {context._conv_settings.preserve_recent_lines}"
+    )
+
+
+@then("conversation settings trigger_line_count should equal {expected:d}")
+def step_assert_trigger(context, expected):
+    assert context._conv_settings.trigger_line_count == expected, (
+        f"expected trigger_line_count={expected}, "
+        f"got {context._conv_settings.trigger_line_count}"
+    )
+
+
+@then('a conversation block with text "{text}" should have line count {count:d}')
+def step_assert_block_lines(context, text, count):
+    decoded = _unescape(text)
+    block = ConversationBlock(text=decoded)
+    assert block.line_count() == count, (
+        f"expected line_count={count} for {decoded!r}, got {block.line_count()}"
+    )
+
+
+@then("a conversation block with empty text should have line count {count:d}")
+def step_assert_empty_block_lines(context, count):
+    block = ConversationBlock(text="")
+    assert block.line_count() == count, (
+        f"expected empty block line_count={count}, got {block.line_count()}"
+    )
+
+
+def _make_stream(settings: ConversationSettings | None = None) -> ConversationStream:
+    return ConversationStream(settings=settings)
+
+
+@when("a fresh ConversationStream with default settings")
+def step_fresh_stream_default(context):
+    context._conv_stream = _make_stream()
+
+
+# Behave does not support a "Given" decorator import alias mismatch; use when too.
+from behave import given  # noqa: E402
+
+
+@given("a fresh ConversationStream with default settings")
+def step_given_fresh_stream_default(context):
+    context._conv_stream = _make_stream()
+
+
+@given(
+    "a fresh ConversationStream with low_mark {low:d} excess {excess:d} "
+    "preserve_recent_lines {preserve:d}"
+)
+def step_given_fresh_stream_custom(context, low, excess, preserve):
+    context._conv_stream = _make_stream(
+        ConversationSettings(
+            prune_low_mark=low,
+            prune_excess=excess,
+            preserve_recent_lines=preserve,
+        )
+    )
+
+
+@when('I bootstrap the stream with welcome text "{text}"')
+def step_bootstrap_stream(context, text):
+    context._conv_stream.bootstrap(welcome_text=text)
+
+
+@when("I clear the stream")
+def step_clear_stream(context):
+    context._conv_stream.clear()
+
+
+@when("I extend the stream with {count:d} plain message blocks")
+def step_extend_stream(context, count):
+    blocks = [
+        ConversationBlock(text=f"extend-block-{idx}") for idx in range(1, count + 1)
+    ]
+    context._conv_stream.extend(blocks)
+
+
+@when('I add a plain block "{text}" to the stream')
+def step_add_plain_block(context, text):
+    context._conv_stream.add_block(text)
+
+
+@when("I add an empty plain block to the stream")
+def step_add_empty_plain_block(context):
+    context._conv_stream.add_block("")
+
+
+@when('I add a markup block "{text}" to the stream')
+def step_add_markup_block(context, text):
+    context._conv_stream.add_block(text, markup=True)
+
+
+@when("I deliver {count:d} plain blocks of {lines:d} lines each through add_block")
+def step_deliver_blocks(context, count, lines):
+    for idx in range(1, count + 1):
+        payload = "\n".join(f"plain-{idx}-line-{line}" for line in range(1, lines + 1))
+        context._conv_stream.add_block(payload)
+
+
+@then("the stream should have exactly {count:d} block")
+@then("the stream should have exactly {count:d} blocks")
+def step_assert_block_count(context, count):
+    assert len(context._conv_stream.blocks) == count, (
+        f"expected {count} blocks, got {len(context._conv_stream.blocks)}"
+    )
+
+
+@then("the stream total_lines should be zero")
+def step_assert_zero_lines(context):
+    assert context._conv_stream.total_lines == 0, (
+        f"expected total_lines=0, got {context._conv_stream.total_lines}"
+    )
+
+
+@then("the stream total_lines should be greater than zero")
+def step_assert_positive_lines(context):
+    assert context._conv_stream.total_lines > 0, (
+        f"expected positive total_lines, got {context._conv_stream.total_lines}"
+    )
+
+
+@then("the stream total_lines should be less than or equal to {limit:d}")
+def step_assert_lines_bound(context, limit):
+    assert context._conv_stream.total_lines <= limit, (
+        f"expected total_lines<={limit}, got {context._conv_stream.total_lines}"
+    )
+
+
+def _rendered_line_count(stream: ConversationStream) -> int:
+    rendered = stream.render()
+    if not rendered:
+        return 0
+    return rendered.count("\n") + 1
+
+
+@then("the stream total_lines should equal the rendered line count")
+def step_assert_total_matches_rendered_lines(context):
+    rendered_line_count = _rendered_line_count(context._conv_stream)
+    assert context._conv_stream.total_lines == rendered_line_count, (
+        f"expected total_lines to match rendered line count {rendered_line_count}, "
+        f"got {context._conv_stream.total_lines}"
+    )
+
+
+@then("the stream rendered line count should be less than or equal to {limit:d}")
+def step_assert_rendered_line_count_bound(context, limit):
+    rendered_line_count = _rendered_line_count(context._conv_stream)
+    assert rendered_line_count <= limit, (
+        f"expected rendered line count <= {limit}, got {rendered_line_count}"
+    )
+
+
+@then('the stream rendered text should contain "{text}"')
+def step_assert_render_contains(context, text):
+    rendered = context._conv_stream.render()
+    assert text in rendered, f"expected {text!r} in render, got {rendered!r}"
+
+
+@then('the stream rendered text should not contain "{text}"')
+def step_assert_render_not_contains(context, text):
+    decoded = _unescape(text)
+    rendered = context._conv_stream.render()
+    assert decoded not in rendered, (
+        f"unexpected {decoded!r} in render, got {rendered!r}"
+    )
+
+
+@then('the stream rendered text should equal "{text}"')
+def step_assert_render_equals(context, text):
+    rendered = context._conv_stream.render()
+    assert rendered == text, f"expected render=={text!r}, got {rendered!r}"
+
+
+@then("the stream rendered plain segment should be backslash escaped")
+def step_assert_plain_escaped(context):
+    rendered = context._conv_stream.render()
+    assert "\\[red]plain\\[/]" in rendered, (
+        f"expected escaped plain markup in render, got {rendered!r}"
+    )
+
+
+@then("the stream should have exactly {count:d} note block")
+@then("the stream should have exactly {count:d} note blocks")
+def step_assert_note_count(context, count):
+    notes = [b for b in context._conv_stream.blocks if b.block_type == "note"]
+    assert len(notes) == count, (
+        f"expected {count} note blocks, got {len(notes)}: {notes!r}"
+    )
+
+
+def _temp_dir(context) -> Path:
+    if not hasattr(context, "_conv_tmp"):
+        context._conv_tmp = tempfile.mkdtemp(prefix="conv-settings-")
+    return Path(context._conv_tmp)
+
+
+@when("I load conversation settings from a missing config path")
+def step_load_missing(context):
+    missing = _temp_dir(context) / "absent.json"
+    context._loaded_settings = load_conversation_settings(missing)
+
+
+@when("I load conversation settings from a malformed JSON config")
+def step_load_malformed(context):
+    path = _temp_dir(context) / "malformed.json"
+    path.write_text("{not valid json", encoding="utf-8")
+    context._loaded_settings = load_conversation_settings(path)
+
+
+@when(
+    "I load conversation settings from a config with low_mark {low:d} excess {excess:d}"
+)
+def step_load_valid(context, low, excess):
+    path = _temp_dir(context) / "valid.json"
+    payload = {"ui": {"prune_low_mark": low, "prune_excess": excess}}
+    path.write_text(json.dumps(payload), encoding="utf-8")
+    context._loaded_settings = load_conversation_settings(path)
+
+
+@when("I load conversation settings from a config with non-numeric values")
+def step_load_bad_numeric(context):
+    path = _temp_dir(context) / "non_numeric.json"
+    payload = {"ui": {"prune_low_mark": "not-a-number", "prune_excess": {}}}
+    path.write_text(json.dumps(payload), encoding="utf-8")
+    context._loaded_settings = load_conversation_settings(path)
+
+
+@when("I load conversation settings from a list-valued JSON config")
+def step_load_list_root(context):
+    path = _temp_dir(context) / "list.json"
+    path.write_text(json.dumps(["this", "is", "a", "list"]), encoding="utf-8")
+    context._loaded_settings = load_conversation_settings(path)
+
+
+@then("loaded settings prune_low_mark should equal {expected:d}")
+def step_assert_loaded_low(context, expected):
+    assert context._loaded_settings.prune_low_mark == expected, (
+        f"expected loaded prune_low_mark={expected}, "
+        f"got {context._loaded_settings.prune_low_mark}"
+    )
+
+
+@then("loaded settings prune_excess should equal {expected:d}")
+def step_assert_loaded_excess(context, expected):
+    assert context._loaded_settings.prune_excess == expected, (
+        f"expected loaded prune_excess={expected}, "
+        f"got {context._loaded_settings.prune_excess}"
+    )
diff --git a/features/steps/tui_conversation_pruning_steps.py b/features/steps/tui_conversation_pruning_steps.py
new file mode 100644
index 000000000..243d1b2ba
--- /dev/null
+++ b/features/steps/tui_conversation_pruning_steps.py
@@ -0,0 +1,31 @@
+"""Step definitions for TUI conversation pruning behaviours."""
+
+from __future__ import annotations
+
+from behave import then, when
+
+
+@when("I deliver {count:d} conversation messages of {line_count:d} lines each")
+def step_deliver_messages(context, count, line_count):
+    app = context._tui_app
+    for index in range(1, count + 1):
+        lines = [f"message-{index} line-{line}" for line in range(1, line_count + 1)]
+        payload = "\n".join(lines)
+        app._append_conversation_block(payload, block_type="message")
+
+
+@then('the conversation widget should not contain "{text}"')
+def step_conversation_not_contains(context, text):
+    mock_static = context._tui_mock_static
+    conversation = context._tui_app.query_one("#conversation", mock_static)
+    assert text not in conversation._text, (
+        f"Unexpected '{text}' in conversation: {conversation._text!r}"
+    )
+
+
+@then("the session transcript should contain {count:d} entries")
+def step_transcript_entry_count(context, count):
+    transcript = context._tui_app._session.transcript
+    assert len(transcript) == count, (
+        f"Expected {count} transcript entries, found {len(transcript)}"
+    )
diff --git a/features/tui_conversation_module_coverage.feature b/features/tui_conversation_module_coverage.feature
new file mode 100644
index 000000000..b02f5166c
--- /dev/null
+++ b/features/tui_conversation_module_coverage.feature
@@ -0,0 +1,141 @@
+Feature: TUI conversation module coverage
+  Directly exercises the cleveragents.tui.conversation module to cover
+  pruning hysteresis, settings clamping, persistence loading, and edge
+  cases that the high-level TUI scenarios do not reach.
+
+  @tdd_issue @tdd_issue_6350
+  Scenario: Default ConversationSettings derive trigger_line_count from low_mark and excess
+    When I build default ConversationSettings
+    Then conversation settings prune_low_mark should equal 1500
+    And conversation settings prune_excess should equal 1000
+    And conversation settings trigger_line_count should equal 2500
+
+  @tdd_issue @tdd_issue_6350
+  Scenario: ConversationSettings clamps prune_low_mark below the minimum
+    When I build ConversationSettings with prune_low_mark 10 prune_excess 200 preserve_recent_lines 50
+    Then conversation settings prune_low_mark should equal 100
+    And conversation settings prune_excess should equal 200
+    And conversation settings preserve_recent_lines should equal 50
+
+  @tdd_issue @tdd_issue_6350
+  Scenario: ConversationSettings clamps prune_low_mark above the maximum
+    When I build ConversationSettings with prune_low_mark 99999 prune_excess 200 preserve_recent_lines 50
+    Then conversation settings prune_low_mark should equal 10000
+
+  @tdd_issue @tdd_issue_6350
+  Scenario: ConversationSettings clamps prune_excess below the minimum
+    When I build ConversationSettings with prune_low_mark 500 prune_excess 1 preserve_recent_lines 50
+    Then conversation settings prune_excess should equal 10
+
+  @tdd_issue @tdd_issue_6350
+  Scenario: ConversationSettings clamps prune_excess above the maximum
+    When I build ConversationSettings with prune_low_mark 500 prune_excess 99999 preserve_recent_lines 50
+    Then conversation settings prune_excess should equal 5000
+
+  @tdd_issue @tdd_issue_6350
+  Scenario: ConversationSettings clamps preserve_recent_lines below zero
+    When I build ConversationSettings with prune_low_mark 500 prune_excess 100 preserve_recent_lines -25
+    Then conversation settings preserve_recent_lines should equal 0
+
+  @tdd_issue @tdd_issue_6350
+  Scenario: ConversationBlock line_count counts trailing newlines and empty text
+    Then a conversation block with text "alpha" should have line count 1
+    And a conversation block with text "alpha\nbeta" should have line count 2
+    And a conversation block with empty text should have line count 0
+
+  @tdd_issue @tdd_issue_6350
+  Scenario: ConversationStream bootstrap installs a protected welcome block
+    Given a fresh ConversationStream with default settings
+    When I bootstrap the stream with welcome text "Hello world"
+    Then the stream should have exactly 1 block
+    And the stream total_lines should be greater than zero
+    And the stream rendered text should contain "Hello world"
+
+  @tdd_issue @tdd_issue_6350
+  Scenario: ConversationStream clear empties blocks and resets total_lines
+    Given a fresh ConversationStream with default settings
+    When I bootstrap the stream with welcome text "Hello"
+    And I clear the stream
+    Then the stream should have exactly 0 blocks
+    And the stream total_lines should be zero
+
+  @tdd_issue @tdd_issue_6350
+  Scenario: ConversationStream extend appends multiple blocks
+    Given a fresh ConversationStream with default settings
+    When I extend the stream with 3 plain message blocks
+    Then the stream should have exactly 3 blocks
+    And the stream rendered text should contain "extend-block-2"
+
+  @tdd_issue @tdd_issue_6350
+  Scenario: ConversationStream renders markup blocks verbatim and escapes plain blocks
+    Given a fresh ConversationStream with default settings
+    When I add a plain block "[red]plain[/]" to the stream
+    And I add a markup block "[red]styled[/]" to the stream
+    Then the stream rendered text should contain "[red]styled[/]"
+    And the stream rendered plain segment should be backslash escaped
+
+  @tdd_issue @tdd_issue_6350
+  Scenario: ConversationStream skips empty-text blocks during render
+    Given a fresh ConversationStream with default settings
+    When I add an empty plain block to the stream
+    And I add a plain block "after-empty" to the stream
+    Then the stream rendered text should equal "after-empty"
+
+  @tdd_issue @tdd_issue_6350
+  Scenario: ConversationStream pruning preserves protected blocks
+    Given a fresh ConversationStream with low_mark 200 excess 50 preserve_recent_lines 100
+    When I bootstrap the stream with welcome text "Protected welcome"
+    And I deliver 4 plain blocks of 200 lines each through add_block
+    Then the stream rendered text should contain "Protected welcome"
+    And the stream total_lines should be less than or equal to 350
+
+  @tdd_issue @tdd_issue_6350
+  Scenario: ConversationStream pruning counts rendered separators between blocks
+    Given a fresh ConversationStream with low_mark 100 excess 50 preserve_recent_lines 0
+    When I deliver 100 plain blocks of 1 lines each through add_block
+    Then the stream should have exactly 1 note block
+    And the stream rendered text should not contain "plain-1-line-1"
+    And the stream total_lines should equal the rendered line count
+    And the stream rendered line count should be less than or equal to 150
+
+  @tdd_issue @tdd_issue_6350
+  Scenario: ConversationStream pruning updates the note in place rather than re-inserting
+    Given a fresh ConversationStream with low_mark 200 excess 50 preserve_recent_lines 100
+    When I deliver 8 plain blocks of 200 lines each through add_block
+    Then the stream should have exactly 1 note block
+
+  @tdd_issue @tdd_issue_6350
+  Scenario: ConversationStream pruning with preserve_recent_lines zero allows full prune
+    Given a fresh ConversationStream with low_mark 200 excess 50 preserve_recent_lines 0
+    When I deliver 6 plain blocks of 200 lines each through add_block
+    Then the stream total_lines should be less than or equal to 250
+
+  @tdd_issue @tdd_issue_6350
+  Scenario: load_conversation_settings returns defaults when the config file is absent
+    When I load conversation settings from a missing config path
+    Then loaded settings prune_low_mark should equal 1500
+    And loaded settings prune_excess should equal 1000
+
+  @tdd_issue @tdd_issue_6350
+  Scenario: load_conversation_settings returns defaults when the config file is malformed JSON
+    When I load conversation settings from a malformed JSON config
+    Then loaded settings prune_low_mark should equal 1500
+    And loaded settings prune_excess should equal 1000
+
+  @tdd_issue @tdd_issue_6350
+  Scenario: load_conversation_settings honours explicit ui values
+    When I load conversation settings from a config with low_mark 800 excess 400
+    Then loaded settings prune_low_mark should equal 800
+    And loaded settings prune_excess should equal 400
+
+  @tdd_issue @tdd_issue_6350
+  Scenario: load_conversation_settings returns defaults when ui values are non-numeric
+    When I load conversation settings from a config with non-numeric values
+    Then loaded settings prune_low_mark should equal 1500
+    And loaded settings prune_excess should equal 1000
+
+  @tdd_issue @tdd_issue_6350
+  Scenario: load_conversation_settings returns defaults when payload is not a dict
+    When I load conversation settings from a list-valued JSON config
+    Then loaded settings prune_low_mark should equal 1500
+    And loaded settings prune_excess should equal 1000
diff --git a/features/tui_conversation_pruning.feature b/features/tui_conversation_pruning.feature
new file mode 100644
index 000000000..915244241
--- /dev/null
+++ b/features/tui_conversation_pruning.feature
@@ -0,0 +1,16 @@
+Feature: TUI conversation content pruning
+  Exercises the line-count based pruning behaviour for the main conversation stream.
+
+  Background:
+    Given the TUI app module is imported with mocked Textual
+
+  @tdd_issue @tdd_issue_6350
+  Scenario: Conversation exceeding the pruning threshold removes oldest blocks and inserts a note
+    Given a mock command router and persona state
+    When I instantiate the Textual TUI app
+    And I call on_mount on the app
+    And I deliver 6 conversation messages of 600 lines each
+    Then the conversation widget should contain "(Earlier messages pruned — see session history for full conversation)"
+    And the conversation widget should contain "message-6 line-4"
+    And the conversation widget should not contain "message-1 line-1"
+    And the session transcript should contain 6 entries
diff --git a/robot/helper_tui_conversation_pruning.py b/robot/helper_tui_conversation_pruning.py
new file mode 100644
index 000000000..427ed418d
--- /dev/null
+++ b/robot/helper_tui_conversation_pruning.py
@@ -0,0 +1,188 @@
+"""Helper script for tui_conversation_pruning.robot integration tests.
+
+Each subcommand is a self-contained check that prints a sentinel on success.
+Tests ConversationStream pruning behaviour directly — no Textual event loop required.
+"""
+
+# ruff: noqa: E402
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+from cleveragents.tui.conversation import (
+    DEFAULT_PRUNE_EXCESS,
+    DEFAULT_PRUNE_LOW_MARK,
+    ConversationSettings,
+    ConversationStream,
+    load_conversation_settings,
+)
+
+
+def _make_stream(
+    prune_low_mark: int = 100,
+    prune_excess: int = 50,
+    preserve_recent_lines: int = 10,
+) -> ConversationStream:
+    settings = ConversationSettings(
+        prune_low_mark=prune_low_mark,
+        prune_excess=prune_excess,
+        preserve_recent_lines=preserve_recent_lines,
+    )
+    return ConversationStream(settings=settings)
+
+
+def _multiline_block(n_lines: int, label: str = "line") -> str:
+    return "\n".join(f"{label} {i}" for i in range(n_lines))
+
+
+def _rendered_line_count(stream: ConversationStream) -> int:
+    rendered = stream.render()
+    if not rendered:
+        return 0
+    return rendered.count("\n") + 1
+
+
+# ---------------------------------------------------------------------------
+# Subcommands
+# ---------------------------------------------------------------------------
+
+
+def cmd_prune_trigger() -> None:
+    """Verify pruning fires and total_lines returns to <= prune_low_mark."""
+    stream = _make_stream(prune_low_mark=100, prune_excess=50)
+    # trigger_line_count = 150; push enough rendered lines to trigger pruning twice,
+    # ensuring total_lines ends at prune_low_mark or below.
+    for i in range(24):
+        stream.add_block(_multiline_block(10, f"msg{i}"), block_type="message")
+
+    assert stream.total_lines <= 100, (
+        f"Expected total_lines <= prune_low_mark (100), got {stream.total_lines}"
+    )
+    print("tui-prune-trigger-ok")
+
+
+def cmd_separator_aware_prune() -> None:
+    """Verify rendered inter-block separators count toward pruning thresholds."""
+    stream = _make_stream(
+        prune_low_mark=100,
+        prune_excess=50,
+        preserve_recent_lines=0,
+    )
+
+    for i in range(100):
+        stream.add_block(f"one-line-{i}", block_type="message")
+
+    rendered = stream.render()
+    assert stream.total_lines == _rendered_line_count(stream), (
+        "Expected total_lines to match rendered line count, "
+        f"got total_lines={stream.total_lines}, rendered={_rendered_line_count(stream)}"
+    )
+    assert stream.total_lines <= 150, (
+        "Expected rendered lines to stay within trigger threshold, "
+        f"got {stream.total_lines}"
+    )
+    assert "one-line-0" not in rendered, "Expected oldest one-line block to be pruned"
+    assert any(block.block_type == "note" for block in stream.blocks), (
+        "Expected pruning note after separator-aware pruning"
+    )
+    print("tui-separator-aware-prune-ok")
+
+
+def cmd_prune_note_inserted() -> None:
+    """Verify a note block is at index 0 with the expected text after pruning."""
+    stream = _make_stream(prune_low_mark=100, prune_excess=50)
+    for i in range(21):
+        stream.add_block(_multiline_block(10, f"msg{i}"), block_type="message")
+
+    blocks = stream.blocks
+    assert len(blocks) > 0, "Expected at least one block after pruning"
+    first = blocks[0]
+    assert first.block_type == "note", (
+        f"Expected first block to be 'note', got {first.block_type!r}"
+    )
+    assert "(Earlier messages pruned" in first.text, (
+        f"Expected pruned-note text in first block, got: {first.text!r}"
+    )
+    print("tui-prune-note-inserted-ok")
+
+
+def cmd_prune_protected() -> None:
+    """Verify protected blocks survive pruning."""
+    stream = _make_stream(prune_low_mark=100, prune_excess=50)
+    protected_text = "protected welcome block"
+    stream.add_block(protected_text, block_type="welcome", protected=True)
+
+    for i in range(20):
+        stream.add_block(_multiline_block(10, f"msg{i}"), block_type="message")
+
+    texts = [b.text for b in stream.blocks]
+    assert protected_text in texts, (
+        f"Protected block was removed during pruning. Block types: "
+        f"{[b.block_type for b in stream.blocks]}"
+    )
+    print("tui-prune-protected-ok")
+
+
+def cmd_clear_resets_state() -> None:
+    """Verify clear() resets blocks, total_lines, and the note-active state."""
+    stream = _make_stream(prune_low_mark=100, prune_excess=50)
+    for i in range(20):
+        stream.add_block(_multiline_block(10, f"msg{i}"), block_type="message")
+
+    assert len(stream.blocks) > 0, "Expected blocks before clear"
+
+    stream.clear()
+
+    assert len(stream.blocks) == 0, (
+        f"Expected 0 blocks after clear, got {len(stream.blocks)}"
+    )
+    assert stream.total_lines == 0, (
+        f"Expected total_lines == 0 after clear, got {stream.total_lines}"
+    )
+    # After clear, a single small block should not trigger pruning
+    stream.add_block("hello", block_type="message")
+    assert len(stream.blocks) == 1, (
+        f"Expected 1 block after clear + add, got {len(stream.blocks)}"
+    )
+    print("tui-clear-resets-state-ok")
+
+
+def cmd_settings_defaults() -> None:
+    """Verify load_conversation_settings returns defaults when config file is absent."""
+    import tempfile
+
+    absent_path = Path(tempfile.mkdtemp()) / "no-such-file.json"
+    settings = load_conversation_settings(config_path=absent_path)
+
+    assert settings.prune_low_mark == DEFAULT_PRUNE_LOW_MARK, (
+        f"prune_low_mark: {settings.prune_low_mark} != {DEFAULT_PRUNE_LOW_MARK}"
+    )
+    assert settings.prune_excess == DEFAULT_PRUNE_EXCESS, (
+        f"Expected prune_excess={DEFAULT_PRUNE_EXCESS}, got {settings.prune_excess}"
+    )
+    print("tui-settings-defaults-ok")
+
+
+# ---------------------------------------------------------------------------
+# Dispatch
+# ---------------------------------------------------------------------------
+
+COMMANDS = {
+    "prune-trigger": cmd_prune_trigger,
+    "separator-aware-prune": cmd_separator_aware_prune,
+    "prune-note-inserted": cmd_prune_note_inserted,
+    "prune-protected": cmd_prune_protected,
+    "clear-resets-state": cmd_clear_resets_state,
+    "settings-defaults": cmd_settings_defaults,
+}
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2 or sys.argv[1] not in COMMANDS:
+        print(f"Usage: {sys.argv[0]} [{' | '.join(COMMANDS)}]", file=sys.stderr)
+        sys.exit(1)
+    COMMANDS[sys.argv[1]]()
diff --git a/robot/tui_conversation_pruning.robot b/robot/tui_conversation_pruning.robot
new file mode 100644
index 000000000..27fd35f1d
--- /dev/null
+++ b/robot/tui_conversation_pruning.robot
@@ -0,0 +1,80 @@
+*** Settings ***
+Documentation    Integration tests for TUI conversation content pruning.
+...
+...              Verifies that:
+...              - ConversationStream enforces prune_low_mark / prune_excess thresholds
+...              - Oldest non-protected blocks are removed when the threshold is exceeded
+...              - A pruning note is inserted at position 0 after pruning
+...              - Protected blocks survive pruning
+...              - clear() resets all state
+...              - load_conversation_settings returns defaults when config file absent
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment With Database Isolation
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER}    ${CURDIR}/helper_tui_conversation_pruning.py
+
+*** Test Cases ***
+ConversationStream Prunes Oldest Blocks When Threshold Exceeded
+    [Documentation]    Verify pruning fires when total_lines exceeds trigger_line_count
+    ...                and total_lines is brought back down to prune_low_mark.
+    [Tags]    tui_conversation_pruning    tdd_issue    tdd_issue_6350
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    prune-trigger
+    ...    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    tui-prune-trigger-ok
+
+ConversationStream Counts Rendered Separators When Pruning
+    [Documentation]    Verify many one-line blocks prune based on rendered blank
+    ...                separators, not only block-local line counts.
+    [Tags]    tui_conversation_pruning    tdd_issue    tdd_issue_6350
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    separator-aware-prune
+    ...    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    tui-separator-aware-prune-ok
+
+ConversationStream Inserts Note Block After Pruning
+    [Documentation]    Verify a pruned-note block is inserted at index 0 after pruning.
+    [Tags]    tui_conversation_pruning    tdd_issue    tdd_issue_6350
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    prune-note-inserted
+    ...    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    tui-prune-note-inserted-ok
+
+ConversationStream Respects Protected Blocks During Pruning
+    [Documentation]    Verify protected blocks are never removed during pruning.
+    [Tags]    tui_conversation_pruning    tdd_issue    tdd_issue_6350
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    prune-protected
+    ...    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    tui-prune-protected-ok
+
+ConversationStream Clear Resets All State
+    [Documentation]    Verify clear() resets blocks, total_lines, and note state.
+    [Tags]    tui_conversation_pruning    tdd_issue    tdd_issue_6350
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    clear-resets-state
+    ...    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    tui-clear-resets-state-ok
+
+load_conversation_settings Returns Defaults When File Absent
+    [Documentation]    Verify load_conversation_settings returns ConversationSettings()
+    ...                defaults when the config file does not exist.
+    [Tags]    tui_conversation_pruning    tdd_issue    tdd_issue_6350
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    settings-defaults
+    ...    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    tui-settings-defaults-ok
diff --git a/src/cleveragents/tui/app.py b/src/cleveragents/tui/app.py
index 2b98f60a4..7b0cf0227 100644
--- a/src/cleveragents/tui/app.py
+++ b/src/cleveragents/tui/app.py
@@ -18,6 +18,7 @@ from cleveragents.domain.models.core.session import (
     SessionActorNotConfiguredError,
     SessionNotFoundError,
 )
+from cleveragents.tui.conversation import ConversationStream, load_conversation_settings
 from cleveragents.tui.first_run import create_default_persona_for_actor, is_first_run
 from cleveragents.tui.input.modes import InputMode, InputModeRouter
 from cleveragents.tui.input.reference_parser import suggestions
@@ -182,6 +183,7 @@ except Exception:  # pragma: no cover
 
 def textual_available() -> bool:
     """Return whether Textual import succeeded."""
+
     return _TEXTUAL_AVAILABLE
 
 
@@ -307,11 +309,15 @@ if _TEXTUAL_AVAILABLE:
                 if self._shell_warn_enabled
                 else None
             )
+            self._conversation_stream = ConversationStream(
+                settings=load_conversation_settings()
+            )
+            self._conversation_stream.bootstrap(welcome_text="CleverAgents TUI")
 
         def compose(self) -> Any:
             yield _Header(show_clock=True)
             with _Vertical(id="main-column"):
-                yield _Static("CleverAgents TUI", id="conversation")
+                yield _Static(self._conversation_stream.render(), id="conversation")
                 yield HelpPanelOverlay(id="help-panel")
                 yield ReferencePickerOverlay(id="reference-picker")
                 yield SlashCommandOverlay(id="slash-overlay")
@@ -328,6 +334,7 @@ if _TEXTUAL_AVAILABLE:
             # ensure_default() and would create a persona, masking the check.
             first_run = is_first_run(self._persona_state.registry)
             self._refresh_persona_bar()
+            self._update_conversation_widget()
             help_panel = self.query_one("#help-panel", HelpPanelOverlay)
             help_panel.hide()
             ref_picker = self.query_one("#reference-picker", ReferencePickerOverlay)
@@ -351,6 +358,7 @@ if _TEXTUAL_AVAILABLE:
 
         def _complete_first_run(self, actor: str) -> None:
             """Persist the chosen actor as the default persona and refresh the bar."""
+
             create_default_persona_for_actor(self._persona_state.registry, actor)
             self._refresh_persona_bar()
             # Refocus prompt after first-run overlay closes.
@@ -503,18 +511,24 @@ if _TEXTUAL_AVAILABLE:
             )
 
             if result.mode == InputMode.COMMAND:
-                conversation.update(result.command_result or "")
+                self._append_conversation_block(
+                    result.command_result or "", block_type="command"
+                )
                 self._refresh_persona_bar()
                 return
             if result.mode == InputMode.SHELL:
                 shell = result.shell_result
                 if shell is None:
-                    conversation.update("(no shell output)")
+                    self._append_conversation_block(
+                        "(no shell output)", block_type="shell"
+                    )
                     return
                 output = (
                     shell.stdout.strip() or shell.stderr.strip() or "(empty output)"
                 )
-                conversation.update(_escape(f"$ {shell.command}\n{output}"))
+                self._append_conversation_block(
+                    f"$ {shell.command}\n{output}", block_type="shell"
+                )
                 return
 
             expanded = result.expanded_text
@@ -526,14 +540,17 @@ if _TEXTUAL_AVAILABLE:
                     ref_picker.set_suggestions(query, suggestions(query))
 
             if self._facade is None:
-                # Facade not wired yet — preview only (graceful degradation)
-                conversation.update(_escape(expanded))
+                # Facade not wired yet — preview the expanded text only
+                # (graceful degradation).
+                self._append_conversation_block(expanded, block_type="message")
                 return
 
-            # Pre-escape the entry on storage so _render_transcript only joins
-            # (avoids O(N) re-escaping of the full history on every render).
-            self._session.transcript.append(_escape(f"You: {expanded}"))
-            self._render_transcript(conversation, thinking=True)
+            # Append the user message to the conversation stream so pruning
+            # applies, then render with a transient "Thinking..." indicator.
+            self._append_conversation_block(f"You: {expanded}", block_type="message")
+            conversation.update(
+                self._conversation_stream.render() + "\n\n⏳ Thinking..."
+            )
 
             # Run blocking LLM call off the main Textual thread so the
             # event loop stays responsive during the API round-trip.
@@ -553,13 +570,12 @@ if _TEXTUAL_AVAILABLE:
             facade = self._facade
             session_id = self._session.session_id
             actor = self._persona_state.active_persona(session_id).actor or None
-            transcript = self._session.transcript
 
             # Capture the current generation so the callback can detect if a
             # newer dispatch has superseded this one (exclusive=True cancels the
             # in-flight worker but its done_callback still fires — without this
-            # guard the cancelled callback would overwrite transcript[-1] with
-            # stale data from the older request).
+            # guard the cancelled callback would render stale data over the
+            # current state).
             self._dispatch_gen += 1
             current_gen = self._dispatch_gen
 
@@ -567,7 +583,7 @@ if _TEXTUAL_AVAILABLE:
                 return _run_llm_dispatch(facade, session_id, expanded, actor)
 
             def _on_llm_done(worker: Any) -> None:
-                """Accumulate outcome into transcript and re-render."""
+                """Append the assistant outcome to the stream and re-render."""
                 # Discard callbacks from cancelled workers.
                 # worker.is_cancelled is a public bool property on Textual's
                 # Worker — no WorkerState import needed (Textual is optional).
@@ -578,16 +594,19 @@ if _TEXTUAL_AVAILABLE:
                     return
 
                 outcome = _format_worker_outcome(worker.result, worker.error)
-                if outcome is not None and transcript:
-                    # Pre-escape and store; replace the placeholder entry.
-                    # outcome is "You: {msg}\n\nAssistant: {reply}" on success
-                    # or an error string — always prefix with user message so
-                    # the exchange is self-contained in the transcript.
-                    if outcome.startswith("You: "):
-                        transcript[-1] = _escape(outcome)
-                    else:
-                        transcript[-1] = _escape(f"You: {expanded}\n\n{outcome}")
-                self._render_transcript(conversation, thinking=False)
+                if outcome is None:
+                    # No outcome — just clear the thinking indicator.
+                    self._update_conversation_widget()
+                    return
+                # outcome is "You: {msg}\n\nAssistant: {reply}" on success
+                # or an error string. The user message is already in the
+                # stream, so append only the assistant portion (or error).
+                if outcome.startswith("You: "):
+                    parts = outcome.split("\n\n", 1)
+                    assistant_text = parts[1] if len(parts) == 2 else outcome
+                else:
+                    assistant_text = outcome
+                self._append_conversation_block(assistant_text, block_type="message")
 
             worker = self.run_worker(
                 _dispatch_llm, thread=True, exclusive=True, name="llm-dispatch"
@@ -638,6 +657,19 @@ if _TEXTUAL_AVAILABLE:
             # Default to disallowing dangerous commands unless explicitly enabled.
             return raw.lower() in {"1", "true", "yes", "on"}
 
+        def _append_conversation_block(
+            self, text: str, *, block_type: str = "message"
+        ) -> None:
+            self._session.transcript.append(text)
+            self._conversation_stream.add_block(text, block_type=block_type)
+            self._update_conversation_widget()
+
+        def _update_conversation_widget(self) -> None:
+            conversation = self._conversation or self.query_one(
+                "#conversation", _Static
+            )
+            conversation.update(self._conversation_stream.render())
+
     _ResolvedTuiApp = _TextualCleverAgentsTuiApp
 
 CleverAgentsTuiApp = _ResolvedTuiApp
diff --git a/src/cleveragents/tui/cleveragents.tcss b/src/cleveragents/tui/cleveragents.tcss
index be098ed9d..0ed5e150c 100644
--- a/src/cleveragents/tui/cleveragents.tcss
+++ b/src/cleveragents/tui/cleveragents.tcss
@@ -80,3 +80,8 @@ Screen {
   color: $text-primary;
   background: $foreground 5%;
 }
+
+.pruned-note {
+  color: $accent;
+  text-style: italic;
+}
diff --git a/src/cleveragents/tui/conversation.py b/src/cleveragents/tui/conversation.py
new file mode 100644
index 000000000..2008a2290
--- /dev/null
+++ b/src/cleveragents/tui/conversation.py
@@ -0,0 +1,290 @@
+"""Conversation stream utilities for the CleverAgents TUI."""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Iterable
+from dataclasses import dataclass
+from pathlib import Path
+
+from rich.markup import escape
+
+DEFAULT_PRUNE_LOW_MARK = 1_500
+DEFAULT_PRUNE_EXCESS = 1_000
+DEFAULT_PRESERVE_RECENT_LINES = 500
+PRUNED_NOTE_TEXT = (
+    "(Earlier messages pruned — see session history for full conversation)"
+)
+
+
+def _normalise_positive(value: int, *, minimum: int, maximum: int) -> int:
+    """Clamp *value* to ``[minimum, maximum]`` while keeping defaults predictable."""
+
+    if value < minimum:
+        return minimum
+    if value > maximum:
+        return maximum
+    return value
+
+
+@dataclass(slots=True)
+class ConversationSettings:
+    """Runtime configuration controlling conversation pruning behaviour.
+
+    Two thresholds implement hysteresis to prevent thrashing at the boundary:
+    ``prune_excess`` is added to ``prune_low_mark`` to form
+    ``trigger_line_count`` (the level at which pruning fires). Pruning then
+    removes blocks until the count falls back to ``prune_low_mark``, so a
+    stream hovering near the boundary is not pruned on every single added
+    block.
+    """
+
+    prune_low_mark: int = DEFAULT_PRUNE_LOW_MARK
+    prune_excess: int = DEFAULT_PRUNE_EXCESS
+    preserve_recent_lines: int = DEFAULT_PRESERVE_RECENT_LINES
+
+    def __post_init__(self) -> None:
+        self.prune_low_mark = _normalise_positive(
+            self.prune_low_mark, minimum=100, maximum=10_000
+        )
+        self.prune_excess = _normalise_positive(
+            self.prune_excess, minimum=10, maximum=5_000
+        )
+        self.preserve_recent_lines = max(0, self.preserve_recent_lines)
+
+    @property
+    def trigger_line_count(self) -> int:
+        """Threshold that activates pruning."""
+
+        return self.prune_low_mark + self.prune_excess
+
+
+@dataclass(slots=True)
+class ConversationBlock:
+    """Single conversation block rendered in the TUI."""
+
+    text: str
+    block_type: str = "message"
+    protected: bool = False
+    markup: bool = False
+
+    def line_count(self) -> int:
+        """Return the number of lines the block occupies when rendered."""
+
+        if not self.text:
+            return 0
+        return max(1, self.text.count("\n") + 1)
+
+
+class ConversationStream:
+    """In-memory representation of the live conversation display."""
+
+    def __init__(self, settings: ConversationSettings | None = None) -> None:
+        self._settings = settings or ConversationSettings()
+        self._blocks: list[ConversationBlock] = []
+        self._total_lines = 0
+        self._visible_blocks = 0
+        # Invariant: when True, _blocks[0] is a ConversationBlock with
+        # block_type == "note". Maintained by _insert_pruned_note().
+        self._note_active = False
+
+    @property
+    def blocks(self) -> tuple[ConversationBlock, ...]:
+        """Expose the current blocks (read-only) for inspection and testing."""
+
+        return tuple(self._blocks)
+
+    @property
+    def total_lines(self) -> int:
+        """Return the running line count for the rendered conversation."""
+
+        return self._total_lines
+
+    def bootstrap(self, *, welcome_text: str) -> None:
+        """Initialise the stream with the welcome block."""
+
+        self._blocks = [
+            ConversationBlock(text=welcome_text, block_type="welcome", protected=True)
+        ]
+        self._visible_blocks = 1 if welcome_text else 0
+        self._total_lines = self._blocks[0].line_count()
+        self._note_active = False
+
+    def clear(self) -> None:
+        """Remove all blocks from the conversation (used by /clear)."""
+
+        self._blocks.clear()
+        self._total_lines = 0
+        self._visible_blocks = 0
+        self._note_active = False
+
+    def extend(self, blocks: Iterable[ConversationBlock]) -> None:
+        """Append multiple blocks to the stream and enforce pruning."""
+
+        for block in blocks:
+            self.add_block(
+                block.text,
+                block_type=block.block_type,
+                protected=block.protected,
+                markup=block.markup,
+            )
+
+    def add_block(
+        self,
+        text: str,
+        *,
+        block_type: str = "message",
+        protected: bool = False,
+        markup: bool = False,
+    ) -> None:
+        """Append a new block to the stream and prune if required."""
+
+        block = ConversationBlock(
+            text=text,
+            block_type=block_type,
+            protected=protected,
+            markup=markup,
+        )
+        self._blocks.append(block)
+        self._add_block_lines(block)
+        self._prune_if_needed()
+
+    def render(self) -> str:
+        """Return the current conversation text for the widget."""
+
+        rendered: list[str] = []
+        for block in self._blocks:
+            if not block.text:
+                continue
+            if block.markup:
+                rendered.append(block.text)
+            else:
+                rendered.append(escape(block.text))
+        return "\n\n".join(rendered)
+
+    def _prune_if_needed(self) -> None:
+        """Prune the oldest non-protected blocks when the threshold is exceeded."""
+
+        if self._total_lines <= self._settings.trigger_line_count:
+            return
+
+        pruned = self._prune_oldest_blocks_to_low_mark()
+
+        if pruned:
+            self._insert_pruned_note()
+            self._prune_oldest_blocks_to_low_mark()
+
+    def _prune_oldest_blocks_to_low_mark(self) -> bool:
+        """Remove oldest eligible blocks until rendered lines reach the low mark."""
+
+        preserve_start = self._determine_preserve_start()
+        pruned = False
+        index = 0
+        while (
+            index < preserve_start and self._total_lines > self._settings.prune_low_mark
+        ):
+            block = self._blocks[index]
+            if block.protected:
+                index += 1
+                continue
+            self._remove_block_lines(block)
+            self._blocks.pop(index)
+            preserve_start -= 1
+            pruned = True
+        return pruned
+
+    def _determine_preserve_start(self) -> int:
+        """Return the index from which blocks must be preserved."""
+
+        if self._settings.preserve_recent_lines <= 0:
+            return len(self._blocks)
+
+        preserved_lines = 0
+        preserved_visible_blocks = 0
+        index = len(self._blocks) - 1
+        while index >= 0 and preserved_lines < self._settings.preserve_recent_lines:
+            block = self._blocks[index]
+            if block.text:
+                preserved_lines += block.line_count()
+                if preserved_visible_blocks:
+                    preserved_lines += 1
+                preserved_visible_blocks += 1
+            index -= 1
+        return max(0, index + 1)
+
+    def _insert_pruned_note(self) -> None:
+        """Ensure a pruning note exists at the top of the conversation."""
+
+        note_markup = f"[pruned-note]{escape(PRUNED_NOTE_TEXT)}[/]"
+
+        if self._note_active and self._blocks:
+            first = self._blocks[0]
+            if first.block_type == "note":
+                previous_line_count = first.line_count()
+                first.text = note_markup
+                first.markup = True
+                self._total_lines += first.line_count() - previous_line_count
+                return
+
+        note_block = ConversationBlock(
+            text=note_markup,
+            block_type="note",
+            protected=True,
+            markup=True,
+        )
+        self._blocks.insert(0, note_block)
+        self._add_block_lines(note_block)
+        self._note_active = True
+
+    def _add_block_lines(self, block: ConversationBlock) -> None:
+        """Add a block's rendered line contribution to the running total."""
+
+        if not block.text:
+            return
+
+        self._total_lines += block.line_count()
+        if self._visible_blocks:
+            self._total_lines += 1
+        self._visible_blocks += 1
+
+    def _remove_block_lines(self, block: ConversationBlock) -> None:
+        """Remove a block's rendered line contribution from the running total."""
+
+        if not block.text:
+            return
+
+        self._total_lines -= block.line_count()
+        if self._visible_blocks > 1:
+            self._total_lines -= 1
+        self._visible_blocks -= 1
+
+
+def load_conversation_settings(config_path: Path | None = None) -> ConversationSettings:
+    """Load pruning settings from the TUI settings file if available.
+
+    Reads ``ui.prune_low_mark`` and ``ui.prune_excess`` from
+    ``~/.config/cleveragents/tui-settings.json`` (or *config_path* when
+    supplied). Returns ``ConversationSettings()`` defaults on any read or
+    parse failure so the TUI always starts in a valid state.
+    """
+    path = config_path or Path.home() / ".config" / "cleveragents" / "tui-settings.json"
+    if not path.exists():
+        return ConversationSettings()
+
+    try:
+        data = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return ConversationSettings()
+
+    ui_config = data.get("ui", {}) if isinstance(data, dict) else {}
+    low_mark = ui_config.get("prune_low_mark", DEFAULT_PRUNE_LOW_MARK)
+    excess = ui_config.get("prune_excess", DEFAULT_PRUNE_EXCESS)
+
+    try:
+        return ConversationSettings(
+            prune_low_mark=int(low_mark),
+            prune_excess=int(excess),
+            preserve_recent_lines=DEFAULT_PRESERVE_RECENT_LINES,
+        )
+    except (ValueError, TypeError):
+        return ConversationSettings()