Skip to content

naqsha.tui

Workbench TUI (optional dependency)

naqsha.tui

NAQSHA Workbench TUI (optional [tui] extra).

WorkbenchApp

Bases: App[None]

Workbench TUI: core stream, budgets, trace tree, flame, memory, and patch review.

Source code in src/naqsha/tui/app.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
class WorkbenchApp(App[None]):
    """Workbench TUI: core stream, budgets, trace tree, flame, memory, and patch review."""

    CSS_PATH = "workbench.tcss"
    BINDINGS = [
        Binding("q", "quit", "Quit", tooltip="Exit the Workbench"),
        Binding("escape", "quit", "Quit", tooltip="Exit the Workbench"),
    ]

    def __init__(self, *, runtime: CoreRuntime, query: str) -> None:
        super().__init__()
        self._runtime = runtime
        self._query = query
        self._last_result: RunResult | None = None
        bus = runtime.config.event_bus
        if bus is None:
            msg = "WorkbenchApp requires RuntimeConfig.event_bus"
            raise ValueError(msg)
        self._bus = bus
        self._patch_sink = RuntimeBusReflectionSink(bus)
        self._workspace = runtime.config.workspace_path.expanduser().resolve()
        self._reflection_parent = self._workspace / ".naqsha" / "reflection-workspaces"

    @property
    def last_result(self) -> RunResult | None:
        return self._last_result

    def compose(self) -> ComposeResult:
        yield Header(show_clock=True)
        with Vertical(id="workbench-body"):
            with Horizontal(id="main-row"):
                yield ChatPanel(id="chat")
                with Vertical(id="side"):
                    yield BudgetPanel(id="budget")
                    yield SpanTreePanel(id="spans")
            with Horizontal(id="analytics-row"):
                yield FlamePanel(id="flame")
                yield MemoryBrowserPanel(id="memory", workspace_path=self._workspace)
                yield PatchReviewPanel(
                    id="patch",
                    team_workspace=self._workspace,
                    patch_workspace_parent=self._reflection_parent,
                    patch_event_sink=self._patch_sink,
                )
        yield Footer()

    def on_mount(self) -> None:
        self.title = "NAQSHA Workbench"
        q = self._query
        self.sub_title = q if len(q) <= 100 else q[:97] + "…"
        # Distinctive built-in theme; keeps panels readable with $surface / $boost.
        self.theme = "tokyo-night"
        chat = self.query_one(ChatPanel)
        chat.border_title = "Orchestrator"
        self.query_one(BudgetPanel).border_title = "Budget"
        self.query_one(SpanTreePanel).border_title = "Trace tree"
        self.query_one(FlamePanel).border_title = "Time & tokens"
        self.query_one(MemoryBrowserPanel).border_title = "Memory"
        self.query_one(PatchReviewPanel).border_title = "Reflection"
        self._bus.subscribe(self._on_bus_event)
        self.run_tui_agent()

    def _on_bus_event(self, event: RuntimeEvent) -> None:
        self.call_from_thread(self._dispatch_event, event)

    def _dispatch_event(self, event: RuntimeEvent) -> None:
        self.query_one(ChatPanel).consume_event(event)
        self.query_one(BudgetPanel).consume_event(event)
        self.query_one(SpanTreePanel).consume_event(event)
        self.query_one(FlamePanel).consume_event(event)
        self.query_one(MemoryBrowserPanel).consume_event(event)
        self.query_one(PatchReviewPanel).consume_event(event)

    @work(thread=True, exclusive=True)
    def run_tui_agent(self) -> None:
        self._last_result = self._runtime.run(self._query)
        self.call_from_thread(self._run_finished)

    def _run_finished(self) -> None:
        if self._last_result and self._last_result.failed:
            self.notify("Run failed", severity="error", timeout=6)
        else:
            self.notify("Run finished — press q to quit", timeout=6)

CSS_PATH class-attribute instance-attribute 

CSS_PATH = 'workbench.tcss'

BINDINGS class-attribute instance-attribute 

BINDINGS = [
    Binding(
        "q", "quit", "Quit", tooltip="Exit the Workbench"
    ),
    Binding(
        "escape",
        "quit",
        "Quit",
        tooltip="Exit the Workbench",
    ),
]

last_result property 

last_result: RunResult | None

compose 

compose() -> ComposeResult
Source code in src/naqsha/tui/app.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def compose(self) -> ComposeResult:
    yield Header(show_clock=True)
    with Vertical(id="workbench-body"):
        with Horizontal(id="main-row"):
            yield ChatPanel(id="chat")
            with Vertical(id="side"):
                yield BudgetPanel(id="budget")
                yield SpanTreePanel(id="spans")
        with Horizontal(id="analytics-row"):
            yield FlamePanel(id="flame")
            yield MemoryBrowserPanel(id="memory", workspace_path=self._workspace)
            yield PatchReviewPanel(
                id="patch",
                team_workspace=self._workspace,
                patch_workspace_parent=self._reflection_parent,
                patch_event_sink=self._patch_sink,
            )
    yield Footer()

on_mount 

on_mount() -> None
Source code in src/naqsha/tui/app.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def on_mount(self) -> None:
    self.title = "NAQSHA Workbench"
    q = self._query
    self.sub_title = q if len(q) <= 100 else q[:97] + "…"
    # Distinctive built-in theme; keeps panels readable with $surface / $boost.
    self.theme = "tokyo-night"
    chat = self.query_one(ChatPanel)
    chat.border_title = "Orchestrator"
    self.query_one(BudgetPanel).border_title = "Budget"
    self.query_one(SpanTreePanel).border_title = "Trace tree"
    self.query_one(FlamePanel).border_title = "Time & tokens"
    self.query_one(MemoryBrowserPanel).border_title = "Memory"
    self.query_one(PatchReviewPanel).border_title = "Reflection"
    self._bus.subscribe(self._on_bus_event)
    self.run_tui_agent()

run_tui_agent 

run_tui_agent() -> None
Source code in src/naqsha/tui/app.py 
95
96
97
98
@work(thread=True, exclusive=True)
def run_tui_agent(self) -> None:
    self._last_result = self._runtime.run(self._query)
    self.call_from_thread(self._run_finished)

build_workbench_app 

build_workbench_app(
    *, runtime: CoreRuntime, query: str
) -> WorkbenchApp

Construct a Workbench TUI bound to the given runtime and query.

Source code in src/naqsha/tui/app.py 
129
130
131
132
def build_workbench_app(*, runtime: CoreRuntime, query: str) -> WorkbenchApp:
    """Construct a Workbench TUI bound to the given runtime and query."""

    return WorkbenchApp(runtime=runtime, query=query)

cli_should_use_tui 

cli_should_use_tui() -> bool

Interactive terminal check (headless/CI safe).

Source code in src/naqsha/tui/app.py 
118
119
120
121
122
123
124
125
126
def cli_should_use_tui() -> bool:
    """Interactive terminal check (headless/CI safe)."""

    import os
    import sys

    if os.environ.get("NAQSHA_NO_TUI", "").lower() in ("1", "true", "yes"):
        return False
    return bool(sys.stdin.isatty() and sys.stdout.isatty())

tui_available 

tui_available() -> bool

True when optional [tui] / dev Textual dependencies are installed.

Source code in src/naqsha/tui/app.py 
107
108
109
110
111
112
113
114
115
def tui_available() -> bool:
    """True when optional ``[tui]`` / dev Textual dependencies are installed."""

    try:
        import textual  # noqa: F401
    except ImportError:
        return False
    else:
        return True