TUI: chat pane LEFT + post-response dim-spam consolidation (T5.14/T5.13) #2470

Merged
conductor merged 5 commits from feat/tui-pane-swap-chat-left into master 2026-10-06 03:27:02 +01:00
2 changed files with 185 additions and 73 deletions

No files matched your search

+70 -73
View File
@@ -110,6 +110,56 @@ try:
except ImportError:
BACKEND_AVAILABLE = False
def collect_post_response_signals(response: str, messages_count: int) -> list[str]:
"""Gather the operational signals that used to each print a dim
system message after every task/build response (the dim-spam source,
T5.13): budget pressure, degraded claims, growth delta, the periodic
self-improvement cycle.
PURE and headless — no channel, no app. The caller renders ONE
collapsed line (or none when nothing fired). Queued-message execution
and on-complete cron firing stay in the worker: they move real
messages, not chatter, and keep their own echoes.
Returns short human-readable fragments; empty list = silence.
"""
signals: list[str] = []
if not BACKEND_AVAILABLE:
return signals
try:
budget_data = provider_budget.get_status()
ratios = []
for provider in budget_data.get("providers", {}).values():
if provider.get("monthly_limit", 0) > 0:
ratios.append(provider.get("monthly_used", 0) / provider["monthly_limit"])
if provider.get("token_limit", 0) > 0:
ratios.append(provider.get("tokens_used", 0) / provider["token_limit"])
usage_percent = max(ratios, default=0) * 100
if usage_percent > 80:
signals.append(f"⚠ Budget: {usage_percent:.0f}% used")
except Exception:
pass
try:
claims_result = claim_audit.scan_text(response[:500])
if claims_result and claims_result.get("degraded"):
signals.append(f"⚠ Claims degraded: {len(claims_result['degraded'])}")
except Exception:
pass
try:
growth = growth_delta.check()
if growth and growth.get("delta"):
signals.append(f"📈 Growth: {growth['delta']}")
except Exception:
pass
try:
if messages_count % 10 == 0:
self_improve.run()
signals.append("🔄 Self-improvement cycle triggered")
except Exception:
pass
return signals
from textual.app import App, ComposeResult
from textual.containers import Horizontal, Vertical, VerticalScroll
from textual.widgets import Button, Header, Input, ProgressBar, RichLog, Static
@@ -1533,18 +1583,21 @@ class RuneApp(App):
def compose(self) -> ComposeResult:
yield Header()
with Horizontal(id="channels"):
yield ChannelView(
"task",
"[b]TASK[/b] — focused work",
classes="channel-view",
id="task-channel",
)
# T5.14 (Nick 2026-10-05): chat LEFT, task RIGHT — feels more
# natural (conversation first, focused work second). WorkPane
# stays third.
yield ChannelView(
"chat",
"[b]CHAT[/b] — free conversation",
classes="channel-view",
id="chat-channel",
)
yield ChannelView(
"task",
"[b]TASK[/b] — focused work",
classes="channel-view",
id="task-channel",
)
# P1-07 (#30): third collapsible column — the operational index.
yield WorkPane(id="work-pane")
with Horizontal(id="footer-bar"):
@@ -3737,69 +3790,13 @@ class RuneApp(App):
except Exception:
pass
# Auto-warn budget when >80% used
try:
if BACKEND_AVAILABLE:
budget_data = provider_budget.get_status()
ratios = []
for provider in budget_data.get("providers", {}).values():
if provider.get("monthly_limit", 0) > 0:
ratios.append(provider.get("monthly_used", 0) / provider["monthly_limit"])
if provider.get("token_limit", 0) > 0:
ratios.append(provider.get("tokens_used", 0) / provider["token_limit"])
usage_percent = max(ratios, default=0) * 100
if usage_percent > 80:
ch.add_message(
"system",
f"[dim]⚠ Budget: {usage_percent:.0f}% used[/dim]",
)
except Exception:
pass
# Auto-check claims after response
try:
if BACKEND_AVAILABLE:
claims_result = claim_audit.scan_text(response[:500])
if claims_result and claims_result.get("degraded"):
ch.add_message(
"system",
f"[dim]⚠ Claims degraded: {len(claims_result['degraded'])}[/dim]",
)
except Exception:
pass
# NOTE (Runtime Reachability audit 2026-09-10): the
# old hook_engine.fire("post_message", ...) call here
# was dead — hook_engine defines no fire(); the
# AttributeError was silently swallowed, so "hooks"
# never ran. Removed rather than left as a false-
# liveness surface. Real hook dispatch is unwired
# (microkernel.fire has zero callers) — see the
# reachability audit, P3.
# Auto-check growth delta
try:
if BACKEND_AVAILABLE:
growth = growth_delta.check()
if growth and growth.get("delta"):
ch.add_message(
"system",
f"[dim]📈 Growth: {growth['delta']}[/dim]",
)
except Exception:
pass
# Auto-improve every 10 messages
try:
if self.status_bar.messages_count % 10 == 0:
if BACKEND_AVAILABLE:
self_improve.run()
ch.add_message(
"system",
f"[dim]🔄 Self-improvement cycle triggered[/dim]",
)
except Exception:
pass
# T5.13 dim-spam consolidation: previously five
# separate dim system messages printed after every
# response. ONE collapsed line now — empty means
# silence, not a dash.
signals = collect_post_response_signals(response[:500], self.status_bar.messages_count)
if signals:
ch.add_message("system", "[dim]" + " · ".join(signals) + "[/dim]")
elif result.get("error") == "RESEARCH_REQUIRED":
# Nonterminal routing disposition, not failure (2026-09-10):
# the task hit the deep-research floor and was dispatched.
@@ -3980,13 +3977,13 @@ class RuneApp(App):
pass
def action_resize_left(self) -> None:
"""Shrink task channel, grow chat channel."""
self.task_width = max(20, self.task_width - 5)
"""Shrink the LEFT channel (chat, post-swap), grow the RIGHT (task)."""
self.task_width = min(80, self.task_width + 5)
self._apply_resize()
def action_resize_right(self) -> None:
"""Grow task channel, shrink chat channel."""
self.task_width = min(80, self.task_width + 5)
"""Grow the LEFT channel (chat), shrink the RIGHT (task)."""
self.task_width = max(20, self.task_width - 5)
self._apply_resize()
def _apply_resize(self):
@@ -0,0 +1,115 @@
"""T5.13/T5.14 frozen contracts: pane order and post-response silence.
The pane swap is structural (compose order); the dim-spam consolidation
is behavioral (one collapsed line instead of five dim messages).
"""
from __future__ import annotations
import re
import sys
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import Mock
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "src"))
from rune import rune_tui # noqa: E402 — the TUI lives in the package
def test_chat_channel_is_yielded_first_in_compose():
src = (ROOT / "src" / "rune" / "rune_tui.py").read_text()
body = re.search(
r'with Horizontal\(id="channels"\):(.*?)with Horizontal\(id="footer-bar"\)',
src,
re.DOTALL,
).group(1)
order = re.findall(r'ChannelView\(\s*"(\w+)"', body)
assert order == ["chat", "task"], f"compose yields {order}"
def test_resize_actions_follow_the_new_visual_positions():
"""After the swap LEFT=chat, RIGHT=task. resize_left shrinks the LEFT
(chat) — which GROWS the task share; resize_right is the reverse."""
assert "grow the RIGHT (task)" in rune_tui.RuneApp.action_resize_left.__doc__
assert "shrink the RIGHT (task)" in rune_tui.RuneApp.action_resize_right.__doc__
# Behavioral proof against an instance WITHOUT constructing the App
# (reactive initialization needs a running textual app; the docstring
# carries the contract, the arithmetic is what matters).
app = SimpleNamespace(
task_width=50,
action_resize_left=None,
action_resize_right=None,
_apply_resize=None,
)
app._apply_resize = lambda: None
# Bind the real methods onto the stand-in and call them.
rune_tui.RuneApp.action_resize_left(app)
assert app.task_width == 55
rune_tui.RuneApp.action_resize_right(app)
assert app.task_width == 50
def _isolate_collector(monkeypatch, **modules):
monkeypatch.setattr(rune_tui, "BACKEND_AVAILABLE", True)
for name, value in modules.items():
monkeypatch.setattr(rune_tui, name, value)
def test_collector_returns_all_signals_in_one_pass(monkeypatch):
_isolate_collector(
monkeypatch,
provider_budget=SimpleNamespace(
get_status=lambda: {
"providers": {"p": {"monthly_limit": 100, "monthly_used": 90, "token_limit": 0, "tokens_used": 0}}
}
),
claim_audit=SimpleNamespace(scan_text=lambda _t: {"degraded": ["a", "b"]}),
growth_delta=SimpleNamespace(check=lambda: {"delta": "focus: +1"}),
self_improve=SimpleNamespace(run=lambda: None),
)
signals = rune_tui.collect_post_response_signals("response text", messages_count=10)
assert signals == ["⚠ Budget: 90% used", "⚠ Claims degraded: 2", "📈 Growth: focus: +1", "🔄 Self-improvement cycle triggered"]
def test_collector_silence_when_nothing_fires(monkeypatch):
_isolate_collector(
monkeypatch,
provider_budget=Mock(get_status=lambda: {"providers": {}}),
claim_audit=Mock(scan_text=lambda _t: None),
growth_delta=Mock(check=lambda: None),
self_improve=Mock(run=lambda: None),
)
collector = rune_tui.collect_post_response_signals
app_signals = collector("plain", messages_count=7) # not a cycle multiple
assert app_signals == []
def test_collector_never_raises_on_backend_noise(monkeypatch):
def boom(*_a, **_k):
raise RuntimeError("backend gone")
_isolate_collector(
monkeypatch,
provider_budget=Mock(get_status=boom),
claim_audit=Mock(scan_text=boom),
growth_delta=Mock(check=boom),
self_improve=Mock(run=boom),
)
assert rune_tui.collect_post_response_signals("x", 10) == []
def test_self_improve_only_at_cycle_multiple(monkeypatch):
ran = []
_isolate_collector(
monkeypatch,
provider_budget=Mock(get_status=lambda: {"providers": {}}),
claim_audit=Mock(scan_text=lambda _t: None),
growth_delta=Mock(check=lambda: None),
self_improve=Mock(run=lambda: ran.append(1)),
)
rune_tui.collect_post_response_signals("x", 7)
assert ran == []
rune_tui.collect_post_response_signals("x", 10)
assert ran == [1]