From e52bfdfd23b841d7fa476e35b2f3f5a3113408a1 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 19:27:16 +0000 Subject: [PATCH 1/8] feat(tui): command history, explorer/evidence navigation, header stats Phase 1 of the v0.2 improvement plan (UX polish), plus integration tests: - Command history: persisted to .kairos/.tui_history (JSONL), cycled with up/down, prefix autocomplete via a ghost-text Suggester, a hint line below the command input, and `:history --clear`. - Explorer pane: item-count title, scroll-position indicators, a line-number gutter every 5th row, search/trace-term highlighting, and Ctrl+G "go to item". - Evidence pane: real keyboard scrolling (up/down/Page Up/Page Down/ Home/End) by wrapping its content in a VerticalScroll instead of a bare Static, which Textual never treats as scrollable. - Header/status line: active well, workspace stats (artifact count, size, well count), last-command runtime, and green/red status coloring plus a transient "running" indicator while a command is in flight. - Five Pilot-driven integration workflows covering ingest/search/show/ note, well switching mid-session, history surviving a restart, explorer navigation + evidence scrolling, and error recovery. Phases 2-6 (pagination, bookmarks, export, diff, themes, layouts, benchmarks, docs) are left for follow-up work. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01JoRVBGw6Ea7wd8d1ZDFKAx --- src/kairos/tui/app.py | 27 ++- src/kairos/tui/commands.py | 114 ++++++++++++ src/kairos/tui/controller.py | 93 ++++++++-- src/kairos/tui/screens/goto_line.py | 36 ++++ src/kairos/tui/screens/main.py | 1 + src/kairos/tui/state.py | 13 ++ src/kairos/tui/styles/kairos.tcss | 33 ++++ src/kairos/tui/widgets/command_line.py | 116 +++++++++++- src/kairos/tui/widgets/evidence_pane.py | 43 ++++- src/kairos/tui/widgets/explorer_pane.py | 85 ++++++++- src/kairos/tui/widgets/header_line.py | 28 ++- src/kairos/tui/widgets/status_line.py | 14 +- tests/tui/test_app.py | 48 +++++ tests/tui/test_command_history.py | 106 +++++++++++ tests/tui/test_evidence_pane_scrolling.py | 99 +++++++++++ tests/tui/test_explorer_navigation.py | 114 ++++++++++++ tests/tui/test_header_updates.py | 92 ++++++++++ tests/tui/test_workflows.py | 204 ++++++++++++++++++++++ 18 files changed, 1235 insertions(+), 31 deletions(-) create mode 100644 src/kairos/tui/screens/goto_line.py create mode 100644 tests/tui/test_command_history.py create mode 100644 tests/tui/test_evidence_pane_scrolling.py create mode 100644 tests/tui/test_explorer_navigation.py create mode 100644 tests/tui/test_header_updates.py create mode 100644 tests/tui/test_workflows.py diff --git a/src/kairos/tui/app.py b/src/kairos/tui/app.py index 00a5dde..053f204 100644 --- a/src/kairos/tui/app.py +++ b/src/kairos/tui/app.py @@ -7,6 +7,7 @@ from __future__ import annotations +import contextlib import dataclasses from pathlib import Path @@ -18,7 +19,9 @@ from kairos.services.context import RuntimeContext from kairos.tui import controller +from kairos.tui.commands import load_history from kairos.tui.screens.fuzzy_finder import FuzzyFinderScreen +from kairos.tui.screens.goto_line import GotoLineScreen from kairos.tui.screens.help import HelpScreen from kairos.tui.screens.main import MainScreen from kairos.tui.screens.tutorial import TutorialScreen @@ -26,6 +29,7 @@ from kairos.tui.state import Selection, TuiState from kairos.tui.widgets.evidence_pane import EvidencePane, citation_text, excerpt_text from kairos.tui.widgets.explorer_pane import ExplorerPane +from kairos.tui.widgets.status_line import StatusLine from kairos.tui.widgets.workspace_pane import WorkspacePane _STYLES_PATH = Path(__file__).parent / "styles" / "kairos.tcss" @@ -41,6 +45,7 @@ class KairosApp(App[None]): BINDINGS = [ Binding("ctrl+p", "open_fuzzy_finder", "Find"), + Binding("ctrl+g", "goto_line", "Go to item"), Binding("ctrl+r", "history_search", "History"), Binding("tab", "cycle_focus(false)", "Cycle pane", show=False), Binding("shift+tab", "cycle_focus(true)", "Cycle pane (reverse)", show=False), @@ -58,7 +63,10 @@ def __init__(self, runtime_ctx: RuntimeContext) -> None: super().__init__() self.runtime_ctx = runtime_ctx self._request_id = 0 - self.state = TuiState(workspace_path=runtime_ctx.workspace.root) + persisted_history = tuple(r.command for r in load_history(runtime_ctx.workspace.root)) + self.state = TuiState( + workspace_path=runtime_ctx.workspace.root, command_history=persisted_history + ) self.layout_mode = "wide" def on_mount(self) -> None: @@ -102,6 +110,8 @@ def _apply_layout_mode(self) -> None: def run_command(self, text: str) -> None: self._request_id += 1 + with contextlib.suppress(NoMatches): # fired before MainScreen mounted + self.query_one(StatusLine).show_running(text) self._dispatch_worker(text, self._request_id) @work(thread=True, exclusive=True, group="dispatch") @@ -150,6 +160,21 @@ def handle_result(item: object) -> None: self.push_screen(FuzzyFinderScreen(self.runtime_ctx), handle_result) + def action_goto_line(self) -> None: + if isinstance(self.focused, Input): + return + explorer = self.query_one(ExplorerPane) + count = len(explorer.children) + if count == 0: + return + + def handle_result(item_number: int | None) -> None: + if item_number is not None: + explorer.index = item_number - 1 + explorer.focus() + + self.push_screen(GotoLineScreen(count), handle_result) + def action_start_search(self) -> None: command_line = self.query_one("#command-line", Input) command_line.value = ":search " diff --git a/src/kairos/tui/commands.py b/src/kairos/tui/commands.py index c91aa88..eabc1df 100644 --- a/src/kairos/tui/commands.py +++ b/src/kairos/tui/commands.py @@ -5,7 +5,10 @@ from __future__ import annotations +import json from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path _ALIASES: dict[str, str] = { "s": "search", @@ -98,3 +101,114 @@ def _closest(name: str) -> str | None: if known.startswith(name) or name.startswith(known): return known return None + + +# ── Command-line hints & autocomplete ───────────────────────────────────── +# Names exported for the command line widget's `Suggester` (ghost-text +# completion) — kept as the same frozenset the parser validates against so +# the two never drift. +KNOWN_COMMAND_NAMES = _KNOWN_COMMANDS + +_COMMAND_HINTS: dict[str, str] = { + "home": "dashboard of workspace stats and recent activity", + "artifacts": "list ingested artifacts, optionally filtered by kind", + "search": "full-text search over ingested content", + "show": "show one artifact's structured detail and spans", + "trace": "trace an entity or term through its typed relations", + "well": "list/use/clear/show coherence wells (scoped views)", + "config": "show a Kconfig symbol's definition and provenance", + "logs": "search parsed log lines", + "doctor": "run workspace health checks (read-only)", + "history": "this session's command history (--clear to wipe it)", + "help": "open the help overlay", + "note": "add or list notes on an artifact/span", + "ingest": "ingest a file or directory into the workspace", + "tutorial": "open the guided tutorial overlay", + "refresh": "re-run the last successful command", + "quit": "quit the TUI", +} + + +def hint_text(partial: str) -> str: + """One-line hint for whatever command name is currently being typed at + the command line, shown just below it. Returns ``""`` when there's + nothing useful to show — not yet typing a command, or the input is + still just ``:`` — since an unrecognized command is already covered by + the parser's own error message once submitted. + """ + stripped = partial.strip() + if not stripped.startswith(":"): + return "" + body = stripped[1:] + typed_name = body.split()[0] if body else "" + if not typed_name: + return "" + name = _ALIASES.get(typed_name, typed_name) + if name in _KNOWN_COMMANDS: + return f":{name} — {_COMMAND_HINTS.get(name, '')}" + matches = sorted(n for n in _KNOWN_COMMANDS if n.startswith(name)) + if not matches: + return "" + if len(matches) == 1: + return f":{matches[0]} — {_COMMAND_HINTS.get(matches[0], '')}" + return "possible: " + ", ".join(f":{m}" for m in matches) + + +# ── Command history persistence ─────────────────────────────────────────── +# Append-only JSONL at .kairos/.tui_history, one record per submitted +# command line. Corrupt lines are skipped rather than sinking the whole +# file — this is a convenience log, not a source of truth. + +HISTORY_FILENAME = ".tui_history" + + +@dataclass(frozen=True, slots=True) +class HistoryRecord: + timestamp: datetime + command: str + success: bool + + +def history_file_path(workspace_root: Path) -> Path: + return workspace_root / ".kairos" / HISTORY_FILENAME + + +def load_history(workspace_root: Path) -> list[HistoryRecord]: + path = history_file_path(workspace_root) + if not path.exists(): + return [] + records: list[HistoryRecord] = [] + for line in path.read_text(encoding="utf-8").splitlines(): + line = line.strip() + if not line: + continue + try: + data = json.loads(line) + records.append( + HistoryRecord( + timestamp=datetime.fromisoformat(data["timestamp"]), + command=data["command"], + success=bool(data["success"]), + ) + ) + except (json.JSONDecodeError, KeyError, ValueError): + continue + return records + + +def append_history(workspace_root: Path, command: str, *, success: bool) -> None: + path = history_file_path(workspace_root) + path.parent.mkdir(parents=True, exist_ok=True) + record = { + "timestamp": datetime.now(UTC).isoformat(), + "command": command, + "success": success, + } + with path.open("a", encoding="utf-8") as f: + f.write(json.dumps(record) + "\n") + + +def clear_history(workspace_root: Path) -> None: + path = history_file_path(workspace_root) + if path.exists(): + path.write_text("", encoding="utf-8") diff --git a/src/kairos/tui/controller.py b/src/kairos/tui/controller.py index abf2932..e1b8fc0 100644 --- a/src/kairos/tui/controller.py +++ b/src/kairos/tui/controller.py @@ -7,6 +7,7 @@ from __future__ import annotations import dataclasses +import time from datetime import UTC, datetime from kairos.domain.errors import KairosError @@ -32,7 +33,13 @@ from kairos.services.show import show as show_service from kairos.services.trace import trace as trace_service from kairos.services.wells import list_all_wells, show_well -from kairos.tui.commands import Command, CommandParseError, parse +from kairos.tui.commands import ( + Command, + CommandParseError, + append_history, + clear_history, + parse, +) from kairos.tui.state import ActivityEntry, Mode, Selection, TuiState _MODE_BY_COMMAND: dict[str, Mode] = { @@ -56,28 +63,78 @@ def dispatch_text(runtime_ctx: RuntimeContext, state: TuiState, text: str) -> Tu raises. Parse errors and service errors both land as a "error" activity entry with a status-line message, per the spec's "actionable errors, no traceback" requirement. + + Every submitted line (successful or not) is timed and folded into + ``command_history`` for the command line's ↑/↓ cycling, except + ``:history --clear`` which wipes history instead of adding to it. """ + started = time.monotonic() + result = _dispatch_text(runtime_ctx, state, text) + elapsed_ms = int((time.monotonic() - started) * 1000) + label = _display_label(text) + return dataclasses.replace(result, last_command_label=label, last_command_ms=elapsed_ms) + + +def _display_label(text: str) -> str: + body = text.strip().lstrip(":").strip() + return body.split()[0] if body else text.strip() + + +def _dispatch_text(runtime_ctx: RuntimeContext, state: TuiState, text: str) -> TuiState: try: command = parse(text) except CommandParseError as exc: - return _record(state, mode=state.mode, command=text, status="error", summary=str(exc)) + new_state = _record(state, mode=state.mode, command=text, status="error", summary=str(exc)) + return _track_history(runtime_ctx, new_state, text, success=False) + + if command.name == "history" and "--clear" in command.args: + clear_history(runtime_ctx.workspace.root) + cleared = dataclasses.replace(state, command_history=()) + return _record( + cleared, + mode="history", + command=text, + status="success", + summary="command history cleared", + ) if command.name == "quit": - return _record(state, mode=state.mode, command=text, status="success", summary="quit") + new_state = _record(state, mode=state.mode, command=text, status="success", summary="quit") + return _track_history(runtime_ctx, new_state, text, success=True) if command.name == "refresh": last = next((e for e in reversed(state.activity) if e.status == "success"), None) if last is None: - return _record( + new_state = _record( state, mode=state.mode, command=text, status="error", summary="Nothing to refresh." ) - return dispatch_text(runtime_ctx, state, last.command) + return _track_history(runtime_ctx, new_state, text, success=False) + try: + new_state = _dispatch(runtime_ctx, state, parse(last.command)) + return _track_history(runtime_ctx, new_state, text, success=True) + except (CommandParseError, KairosError) as exc: + new_state = _record( + state, mode=state.mode, command=text, status="error", summary=str(exc) + ) + return _track_history(runtime_ctx, new_state, text, success=False) try: - return _dispatch(runtime_ctx, state, command) + new_state = _dispatch(runtime_ctx, state, command) + return _track_history(runtime_ctx, new_state, text, success=True) except KairosError as exc: mode = _MODE_BY_COMMAND.get(command.name, state.mode) - return _record(state, mode=mode, command=text, status="error", summary=str(exc)) + new_state = _record(state, mode=mode, command=text, status="error", summary=str(exc)) + return _track_history(runtime_ctx, new_state, text, success=False) + + +def _track_history( + runtime_ctx: RuntimeContext, state: TuiState, text: str, *, success: bool +) -> TuiState: + stripped = text.strip() + if not stripped: + return state + append_history(runtime_ctx.workspace.root, stripped, success=success) + return dataclasses.replace(state, command_history=(*state.command_history, stripped)) def _dispatch(runtime_ctx: RuntimeContext, state: TuiState, command: Command) -> TuiState: @@ -132,6 +189,7 @@ def _home(runtime_ctx: RuntimeContext, state: TuiState, command: Command) -> Tui total_relations = session.scalar(select(func.count(RelationRow.id))) or 0 total_spans = session.scalar(select(func.count(SourceSpanRow.id))) or 0 total_wells = session.scalar(select(func.count(CoherenceWellRow.id))) or 0 + total_size = session.scalar(select(func.sum(ArtifactRow.size_bytes))) or 0 # Breakdown by kind kind_rows = session.execute( @@ -159,8 +217,14 @@ def _home(runtime_ctx: RuntimeContext, state: TuiState, command: Command) -> Tui workspace_name=runtime_ctx.workspace.root.name, recent_activity=events[:5], ) - return _record( + updated_state = dataclasses.replace( state, + artifact_count=total_artifacts, + workspace_size_bytes=total_size, + well_count=total_wells, + ) + return _record( + updated_state, mode="home", command=command.raw, status="success", @@ -298,8 +362,9 @@ def _well(runtime_ctx: RuntimeContext, state: TuiState, command: Command) -> Tui sub = command.args[0] if command.args else "list" if sub == "list": wells = list_all_wells(runtime_ctx) + updated_state = dataclasses.replace(state, well_count=len(wells)) return _record( - state, + updated_state, mode="well", command=command.raw, status="success", @@ -392,13 +457,19 @@ def _ingest(runtime_ctx: RuntimeContext, state: TuiState, command: Command) -> T if diag_count: summary_parts.append(f"{diag_count} diagnostic(s)") - return _record( + artifacts = list_artifacts_service(runtime_ctx) + updated_state = dataclasses.replace( state, + artifact_count=len(artifacts), + workspace_size_bytes=sum(a.size_bytes for a in artifacts), + ) + return _record( + updated_state, mode="artifacts", command=command.raw, status="success", summary=", ".join(summary_parts), - last_result=list_artifacts_service(runtime_ctx), + last_result=artifacts, ) diff --git a/src/kairos/tui/screens/goto_line.py b/src/kairos/tui/screens/goto_line.py new file mode 100644 index 0000000..0286430 --- /dev/null +++ b/src/kairos/tui/screens/goto_line.py @@ -0,0 +1,36 @@ +"""Explorer's "go to item" prompt (Ctrl+G): jump straight to row N without +scrolling through a long result set by hand. +""" + +from __future__ import annotations + +from textual.app import ComposeResult +from textual.containers import Vertical +from textual.screen import ModalScreen +from textual.widgets import Input, Static + + +class GotoLineScreen(ModalScreen[int | None]): + BINDINGS = [("escape", "cancel", "Close")] + + def __init__(self, max_index: int) -> None: + super().__init__() + self._max_index = max_index + + def compose(self) -> ComposeResult: + with Vertical(id="goto-line-container"): + yield Static(f"Go to item # (1–{self._max_index}), Escape to cancel") + yield Input(placeholder="item number", id="goto-line-input") + + def on_mount(self) -> None: + self.query_one("#goto-line-input", Input).focus() + + def on_input_submitted(self, event: Input.Submitted) -> None: + text = event.value.strip() + if text.isdigit() and 1 <= int(text) <= self._max_index: + self.dismiss(int(text)) + else: + self.dismiss(None) + + def action_cancel(self) -> None: + self.dismiss(None) diff --git a/src/kairos/tui/screens/main.py b/src/kairos/tui/screens/main.py index 3bf85d6..0637e3c 100644 --- a/src/kairos/tui/screens/main.py +++ b/src/kairos/tui/screens/main.py @@ -46,6 +46,7 @@ def compose(self) -> ComposeResult: yield Static("\u25cf Evidence", id="evidence-title", classes="pane-title") yield EvidencePane(id="evidence-pane") yield CommandLine() + yield Static("", id="command-hint") yield StatusLine() def on_mount(self) -> None: diff --git a/src/kairos/tui/state.py b/src/kairos/tui/state.py index c7f199e..53b8bdb 100644 --- a/src/kairos/tui/state.py +++ b/src/kairos/tui/state.py @@ -109,6 +109,19 @@ class TuiState: status_message: str | None = None last_result: ModeResult = None focus: FocusTarget = "command_line" + # Every line submitted at the command line, oldest first — backs the + # command line's ↑/↓ cycling. Persisted separately to + # .kairos/.tui_history (see kairos.tui.commands) so it survives restarts. + command_history: tuple[str, ...] = () + # Set by the controller after every dispatch, for the status/header line. + last_command_label: str | None = None + last_command_ms: int | None = None + # Sticky workspace stats for the header — refreshed by the handlers that + # can actually change them (:home, :ingest, :well) rather than on every + # dispatch, so a plain :search doesn't pay for a recount. + artifact_count: int = 0 + workspace_size_bytes: int = 0 + well_count: int = 0 def as_list_of[T](value: object, item_type: type[T]) -> list[T] | None: diff --git a/src/kairos/tui/styles/kairos.tcss b/src/kairos/tui/styles/kairos.tcss index 90add02..f5c8e25 100644 --- a/src/kairos/tui/styles/kairos.tcss +++ b/src/kairos/tui/styles/kairos.tcss @@ -29,6 +29,8 @@ Screen { #explorer-pane { width: 30%; border-right: thick #21262d; + border-top: solid #21262d; + border-bottom: solid #21262d; background: #0d1117; scrollbar-color: #30363d; scrollbar-color-hover: #58a6ff; @@ -90,6 +92,12 @@ ExplorerPane > ListItem.-highlight { scrollbar-color-hover: #58a6ff; } +#evidence-content { + width: 1fr; + background: transparent; + color: #c9d1d9; +} + /* ── Responsive layout ─────────────────────────────────────────────── */ #panes.mode-medium #evidence-container { @@ -118,6 +126,17 @@ ExplorerPane > ListItem.-highlight { border: tall #58a6ff; } +/* ── Command hint ──────────────────────────────────────────────────── */ + +#command-hint { + dock: bottom; + height: 1; + background: #0d1117; + color: #8b949e; + padding: 0 1; + text-style: italic; +} + /* ── Status line ───────────────────────────────────────────────────── */ #status-line { @@ -219,6 +238,20 @@ FuzzyFinderScreen { border-left: thick #58a6ff; } +/* ── Goto-line overlay ─────────────────────────────────────────────── */ + +GotoLineScreen { + align: center middle; +} + +#goto-line-container { + background: #161b22; + border: heavy #58a6ff; + width: 50; + height: 5; + padding: 1 2; +} + /* ── Tab bar ───────────────────────────────────────────────────────── */ #tab-bar { diff --git a/src/kairos/tui/widgets/command_line.py b/src/kairos/tui/widgets/command_line.py index 60d45fa..a060a13 100644 --- a/src/kairos/tui/widgets/command_line.py +++ b/src/kairos/tui/widgets/command_line.py @@ -1,21 +1,133 @@ from __future__ import annotations -from textual.widgets import Input +from typing import cast -_PROMPT = "\u2b22" +from textual.css.query import NoMatches +from textual.suggester import Suggester +from textual.widgets import Input, Static + +from kairos.tui.commands import KNOWN_COMMAND_NAMES, hint_text + +_PROMPT = "⬢" + + +class _CommandSuggester(Suggester): + """Ghost-text completion for the command *name* only (e.g. ``:sear`` -> + ``:search ``) — never guesses at arguments, and stays silent once a + space has been typed since the rest is free-form. + """ + + def __init__(self) -> None: + super().__init__(case_sensitive=False, use_cache=False) + + async def get_suggestion(self, value: str) -> str | None: + if not value.startswith(":") or " " in value: + return None + prefix = value[1:] + if not prefix: + return None + matches = sorted(name for name in KNOWN_COMMAND_NAMES if name.startswith(prefix)) + if len(matches) == 1 and matches[0] != prefix: + return f":{matches[0]} " + return None + + +class _CommandHistoryCursor: + """Ephemeral ↑/↓ cursor over ``TuiState.command_history``. The history + list itself lives in state (and on disk); this only tracks *where in + it* the user has scrolled, plus the in-progress line they were typing + before they started cycling, so pressing Down back past the newest + entry restores it — same as a shell history search. + """ + + def __init__(self) -> None: + self._index: int | None = None + self._draft: str = "" + + def reset(self) -> None: + self._index = None + self._draft = "" + + def prev(self, entries: tuple[str, ...], current_value: str) -> str | None: + if not entries: + return None + if self._index is None: + self._draft = current_value + self._index = len(entries) - 1 + elif self._index > 0: + self._index -= 1 + else: + return None + return entries[self._index] + + def next(self, entries: tuple[str, ...]) -> str | None: + if self._index is None: + return None + if self._index < len(entries) - 1: + self._index += 1 + return entries[self._index] + self._index = None + return self._draft class CommandLine(Input): + BINDINGS = [ + ("up", "history_prev", "Previous command"), + ("down", "history_next", "Next command"), + ] + def __init__(self) -> None: super().__init__( placeholder=f"{_PROMPT} :search ? for help ^P to find", id="command-line", + suggester=_CommandSuggester(), ) + self._history = _CommandHistoryCursor() + # Set right before a history nav action assigns `self.value`, so the + # `Input.Changed` message that assignment posts (asynchronously — + # see textual's Input._watch_value) can be recognized as an echo of + # our own edit rather than the user typing, and skip resetting the + # cursor it just moved. + self._last_history_value: str | None = None + + def on_input_changed(self, event: Input.Changed) -> None: + event.stop() + if event.value == self._last_history_value: + self._last_history_value = None + else: + self._history.reset() + self._update_hint(event.value) def on_input_submitted(self, event: Input.Submitted) -> None: event.stop() text = event.value.strip() + self._history.reset() + self._update_hint("") if not text: return self.value = "" self.app.run_command(text) # type: ignore[attr-defined] + + def _history_entries(self) -> tuple[str, ...]: + return cast("tuple[str, ...]", self.app.state.command_history) # type: ignore[attr-defined] + + def action_history_prev(self) -> None: + value = self._history.prev(self._history_entries(), self.value) + if value is not None: + self._last_history_value = value + self.value = value + self.action_end() + + def action_history_next(self) -> None: + value = self._history.next(self._history_entries()) + if value is not None: + self._last_history_value = value + self.value = value + self.action_end() + + def _update_hint(self, value: str) -> None: + try: + hint = self.screen.query_one("#command-hint", Static) + except NoMatches: + return + hint.update(hint_text(value)) diff --git a/src/kairos/tui/widgets/evidence_pane.py b/src/kairos/tui/widgets/evidence_pane.py index 3708833..550d8f5 100644 --- a/src/kairos/tui/widgets/evidence_pane.py +++ b/src/kairos/tui/widgets/evidence_pane.py @@ -8,13 +8,14 @@ from __future__ import annotations from rich.markup import escape +from textual.app import ComposeResult +from textual.binding import Binding +from textual.containers import VerticalScroll from textual.widgets import Static from kairos.cli.citation import provenance_lines -from kairos.schemas.activity import ActivityEvent from kairos.schemas.artifact import ArtifactDetail, ArtifactSummary from kairos.schemas.config import ConfigSymbolResult -from kairos.schemas.dashboard import DashboardResult from kairos.schemas.doctor import DoctorReport from kairos.schemas.logs import LogHit from kairos.schemas.note import NoteResult @@ -50,11 +51,45 @@ def _artifact_summary_lines(a: ArtifactSummary) -> str: ) -class EvidencePane(Static): +class EvidencePane(VerticalScroll): + """A ``VerticalScroll`` wrapping a single inner ``Static`` — not a bare + ``Static`` directly, because a leaf widget with no children is never + ``is_scrollable`` in Textual (see ``Widget.is_scrollable``), so its + built-in ``action_scroll_*``/``action_page_*`` bindings would silently + no-op regardless of content overflow. Wrapping in a real scrollable + container makes ↑/↓ and Page Up/Down actually move the viewport. + ``renderable`` is proxied through so callers (and existing tests) that + read the pane's text can keep treating it like a ``Static``. + """ + can_focus = True + # `action_scroll_up`/`_down`/`_home`/`_end` and `action_page_up`/`_down` + # are built into `Widget` already (see textual.widget) — this only wires + # the keys, since a plain container doesn't bind them by default the way + # ListView binds arrow keys for its own highlight-driven scrolling. + BINDINGS = [ + Binding("up", "scroll_up", "Scroll up", show=False), + Binding("down", "scroll_down", "Scroll down", show=False), + Binding("pageup", "page_up", "Page up", show=False), + Binding("pagedown", "page_down", "Page down", show=False), + Binding("home", "scroll_home", "Top", show=False), + Binding("end", "scroll_end", "Bottom", show=False), + ] + + def compose(self) -> ComposeResult: + yield Static("", id="evidence-content") + + @property + def renderable(self) -> object: + return self.query_one("#evidence-content", Static).renderable + def refresh_from_state(self, state: TuiState) -> None: - self.update(_render(state)) + self.query_one("#evidence-content", Static).update(_render(state)) + # Deferred so the scroll reset happens after the layout pass that + # recomputes virtual_size for the new content — resetting before + # that pass runs against a stale size and gets overridden. + self.call_after_refresh(self.scroll_home, animate=False) def _render(state: TuiState) -> str: diff --git a/src/kairos/tui/widgets/explorer_pane.py b/src/kairos/tui/widgets/explorer_pane.py index b6064f0..449d91b 100644 --- a/src/kairos/tui/widgets/explorer_pane.py +++ b/src/kairos/tui/widgets/explorer_pane.py @@ -37,6 +37,26 @@ _LAYER_TAG = {"raw": "RAW", "extracted": "EXTRACTED", "derived": "DERIVED", "user": "USER"} +_TITLE_BY_MODE = { + "home": "Home", + "artifacts": "Artifacts", + "search": "Search", + "show": "Detail", + "trace": "Trace", + "well": "Wells", + "config": "Config", + "logs": "Logs", + "doctor": "Doctor", + "history": "History", + "help": "Help", + "notes": "Notes", +} + +# Every Nth row gets a printed line number in its gutter, so a long list +# gives you a sense of position (and a number to hand to Ctrl+G) without +# numbering — and cluttering — every single line. +_GUTTER_EVERY = 5 + @dataclass(frozen=True, slots=True) class _Row: @@ -46,11 +66,35 @@ class _Row: target_id: str | None +def _highlighted(text: str, term: str | None) -> str: + """Escape ``text`` for Rich markup, wrapping case-insensitive matches of + ``term`` in a reverse-video span. Escaping happens per-chunk so the + highlight markup itself is never escaped away. + """ + if not term: + return escape(text) + lower_text, lower_term = text.lower(), term.lower() + if lower_term not in lower_text: + return escape(text) + chunks: list[str] = [] + i = 0 + while i < len(text): + idx = lower_text.find(lower_term, i) + if idx == -1: + chunks.append(escape(text[i:])) + break + chunks.append(escape(text[i:idx])) + chunks.append(f"[reverse]{escape(text[idx : idx + len(term)])}[/reverse]") + i = idx + len(term) + return "".join(chunks) + + class ExplorerItem(ListItem): - def __init__(self, row: _Row) -> None: - text = escape(row.label) + def __init__(self, row: _Row, index: int, query_term: str | None = None) -> None: + gutter = f"{index + 1:>4} " if (index + 1) % _GUTTER_EVERY == 0 else " " + text = f"{gutter}{_highlighted(row.label, query_term)}" if row.sublabel: - text += f"\n[dim]{escape(row.sublabel)}[/dim]" + text += f"\n [dim]{escape(row.sublabel)}[/dim]" super().__init__(Static(text)) self.kind: SelectionKind | None = row.kind self.target_id: str | None = row.target_id @@ -60,10 +104,13 @@ class ExplorerPane(ListView): def refresh_from_state(self, state: TuiState) -> None: self.clear() rows = _rows_for(state) - for row in rows: - self.append(ExplorerItem(row)) + query_term = _query_term(state) + for index, row in enumerate(rows): + self.append(ExplorerItem(row, index, query_term)) if rows: self.index = 0 + self.border_title = f"{_TITLE_BY_MODE.get(state.mode, state.mode.title())} ({len(rows)})" + self.call_after_refresh(self._update_scroll_indicators) def selected_reference(self) -> tuple[SelectionKind, str] | None: item = self.highlighted_child @@ -71,6 +118,34 @@ def selected_reference(self) -> tuple[SelectionKind, str] | None: return item.kind, item.target_id return None + def on_list_view_highlighted(self, event: ListView.Highlighted) -> None: + self.call_after_refresh(self._update_scroll_indicators) + + def _update_scroll_indicators(self) -> None: + max_scroll = self.max_scroll_y + if max_scroll <= 0: + self.border_subtitle = "" + return + can_scroll_up = self.scroll_y > 0.5 + can_scroll_down = self.scroll_y < max_scroll - 0.5 + if can_scroll_up and can_scroll_down: + self.border_subtitle = "▲▼ more" + elif can_scroll_up: + self.border_subtitle = "▲ top" + elif can_scroll_down: + self.border_subtitle = "▼ more below" + else: + self.border_subtitle = "" + + +def _query_term(state: TuiState) -> str | None: + result = state.last_result + if isinstance(result, SearchResult): + return result.query + if isinstance(result, TraceResult): + return result.query + return None + def _rows_for(state: TuiState) -> list[_Row]: result = state.last_result diff --git a/src/kairos/tui/widgets/header_line.py b/src/kairos/tui/widgets/header_line.py index 80ae2e0..ff3ae3d 100644 --- a/src/kairos/tui/widgets/header_line.py +++ b/src/kairos/tui/widgets/header_line.py @@ -5,9 +5,18 @@ from kairos.tui.state import TuiState -_GLYPH = "\u2b22" -_WELL_GLYPH = "\u25c8" -_OFFLINE_GLYPH = "\u25cf" +_GLYPH = "⬢" +_WELL_GLYPH = "◈" +_OFFLINE_GLYPH = "●" + + +def _format_size(num_bytes: int) -> str: + size = float(num_bytes) + for unit in ("B", "KB", "MB", "GB"): + if size < 1024 or unit == "GB": + return f"{size:.0f}{unit}" if unit == "B" else f"{size:.1f}{unit}" + size /= 1024 + return f"{size:.1f}GB" class HeaderLine(Static): @@ -17,7 +26,16 @@ def __init__(self) -> None: def refresh_from_state(self, state: TuiState) -> None: workspace_name = escape(state.workspace_path.name) well = escape(state.active_well) if state.active_well else "none" + stats = ( + f"{state.artifact_count} artifact(s) · " + f"{_format_size(state.workspace_size_bytes)} · " + f"{state.well_count} well(s)" + ) + runtime = "" + if state.last_command_label is not None and state.last_command_ms is not None: + runtime = f" │ {state.last_command_label} took {state.last_command_ms}ms" self.update( - f" {_GLYPH} KAIROS \u2502 ws: {workspace_name} " - f"\u2502 {_WELL_GLYPH} well: {well} \u2502 {_OFFLINE_GLYPH} LOCAL" + f" {_GLYPH} KAIROS │ ws: {workspace_name} " + f"│ {_WELL_GLYPH} well: {well} │ {stats}" + f"{runtime} │ {_OFFLINE_GLYPH} LOCAL" ) diff --git a/src/kairos/tui/widgets/status_line.py b/src/kairos/tui/widgets/status_line.py index c0114c6..c70f056 100644 --- a/src/kairos/tui/widgets/status_line.py +++ b/src/kairos/tui/widgets/status_line.py @@ -18,14 +18,22 @@ class StatusLine(Static): def __init__(self) -> None: super().__init__(_LEGEND, id="status-line") + def show_running(self, command_text: str) -> None: + """Transient yellow "still running" indicator, shown the instant a + command is dispatched \u2014 before its worker thread has returned \u2014 + and overwritten by the next ``refresh_from_state`` once it does. + """ + message = escape(command_text.strip()) + self.update(f"[yellow]\u25cc running: {message}...[/yellow] {_SEP}{_LEGEND}") + def refresh_from_state(self, state: TuiState) -> None: if state.status_message: message = escape(state.status_message) if state.status == "error": prefix = "[red]\u2717[/red] " - self.update(f"{prefix}{message} {_SEP}{_LEGEND}") + self.update(f"{prefix}[red]{message}[/red] {_SEP}{_LEGEND}") else: - prefix = "[dim]\u2713[/dim] " - self.update(f"{prefix}[dim]{message}[/dim] {_SEP}{_LEGEND}") + prefix = "[green]\u2713[/green] " + self.update(f"{prefix}[green]{message}[/green] {_SEP}{_LEGEND}") else: self.update(_LEGEND) diff --git a/tests/tui/test_app.py b/tests/tui/test_app.py index 63e19f6..ec3f717 100644 --- a/tests/tui/test_app.py +++ b/tests/tui/test_app.py @@ -18,6 +18,8 @@ pytest.importorskip("textual") pytest.importorskip("pytest_asyncio") +from textual.widgets import Static + from kairos.schemas.artifact import ArtifactDetail, ArtifactSummary from kairos.schemas.config import ConfigSymbolResult from kairos.schemas.doctor import DoctorReport @@ -337,6 +339,52 @@ async def test_history_records_success_and_failure(runtime_ctx: RuntimeContext) assert "error" in statuses +@pytest.mark.asyncio +async def test_command_line_up_down_cycles_history(runtime_ctx: RuntimeContext) -> None: + from kairos.tui.widgets.command_line import CommandLine + + app = KairosApp(runtime_ctx) + async with app.run_test(size=WIDE) as pilot: + await _type_command(pilot, ":artifacts") + await _type_command(pilot, ":doctor") + + command_line = app.query_one(CommandLine) + command_line.focus() + await pilot.pause() + + await pilot.press("up") + await pilot.pause() + assert command_line.value == ":doctor" + + await pilot.press("up") + await pilot.pause() + assert command_line.value == ":artifacts" + + await pilot.press("down") + await pilot.pause() + assert command_line.value == ":doctor" + + await pilot.press("down") + await pilot.pause() + assert command_line.value == "" + + +@pytest.mark.asyncio +async def test_command_line_shows_hint_for_partial_command(runtime_ctx: RuntimeContext) -> None: + from kairos.tui.widgets.command_line import CommandLine + + app = KairosApp(runtime_ctx) + async with app.run_test(size=WIDE) as pilot: + command_line = app.query_one(CommandLine) + command_line.focus() + await pilot.pause() + await pilot.press(*":sear") + await pilot.pause() + + hint = str(app.query_one("#command-hint", Static).renderable) + assert ":search" in hint + + @pytest.mark.parametrize( "width,expect_explorer,expect_evidence", [(140, True, True), (90, True, False), (60, False, False)], diff --git a/tests/tui/test_command_history.py b/tests/tui/test_command_history.py new file mode 100644 index 0000000..ca406ee --- /dev/null +++ b/tests/tui/test_command_history.py @@ -0,0 +1,106 @@ +"""Command history: JSONL persistence roundtrip, hint text, and the +controller-level wiring that feeds the command line's ↑/↓ cycling. +""" + +from __future__ import annotations + +from pathlib import Path + +from kairos.services.context import RuntimeContext +from kairos.tui.commands import ( + append_history, + clear_history, + hint_text, + history_file_path, + load_history, +) +from kairos.tui.controller import dispatch_text +from kairos.tui.state import TuiState + + +def test_history_roundtrip_add_and_retrieve(tmp_path: Path) -> None: + assert load_history(tmp_path) == [] + + append_history(tmp_path, ":search widget", success=True) + append_history(tmp_path, ":bogus", success=False) + + records = load_history(tmp_path) + assert [r.command for r in records] == [":search widget", ":bogus"] + assert [r.success for r in records] == [True, False] + assert history_file_path(tmp_path).exists() + + +def test_history_clear_empties_the_file(tmp_path: Path) -> None: + append_history(tmp_path, ":search widget", success=True) + assert load_history(tmp_path) + + clear_history(tmp_path) + assert load_history(tmp_path) == [] + + +def test_history_load_skips_corrupt_lines(tmp_path: Path) -> None: + path = history_file_path(tmp_path) + path.parent.mkdir(parents=True) + path.write_text('not json\n{"timestamp": "bad", "command": "x", "success": true}\n') + append_history(tmp_path, ":search widget", success=True) + + records = load_history(tmp_path) + assert [r.command for r in records] == [":search widget"] + + +def test_history_clear_on_missing_file_is_a_no_op(tmp_path: Path) -> None: + clear_history(tmp_path) # must not raise + assert load_history(tmp_path) == [] + + +def test_hint_text_for_partial_and_full_commands() -> None: + assert hint_text("") == "" + assert hint_text("hello") == "" + assert hint_text(":") == "" + assert hint_text(":sear").startswith(":search —") + assert hint_text(":search").startswith(":search —") + assert hint_text(":search foo").startswith(":search —") + + +def test_hint_text_for_ambiguous_prefix_lists_candidates() -> None: + hint = hint_text(":h") + assert hint.startswith("possible:") + assert ":home" in hint + assert ":history" in hint + assert ":help" in hint + + +def test_hint_text_for_unknown_command_is_empty() -> None: + assert hint_text(":zzz") == "" + + +def test_dispatch_text_appends_to_state_and_disk_history(runtime_ctx: RuntimeContext) -> None: + state = TuiState(workspace_path=runtime_ctx.workspace.root) + state = dispatch_text(runtime_ctx, state, ":artifacts") + state = dispatch_text(runtime_ctx, state, ":bogus") + + assert state.command_history == (":artifacts", ":bogus") + records = load_history(runtime_ctx.workspace.root) + assert [r.command for r in records] == [":artifacts", ":bogus"] + assert [r.success for r in records] == [True, False] + + +def test_history_clear_command_wipes_state_and_disk(runtime_ctx: RuntimeContext) -> None: + state = TuiState(workspace_path=runtime_ctx.workspace.root) + state = dispatch_text(runtime_ctx, state, ":artifacts") + assert state.command_history + + state = dispatch_text(runtime_ctx, state, ":history --clear") + assert state.command_history == () + assert load_history(runtime_ctx.workspace.root) == [] + assert state.status == "idle" + assert state.activity[-1].status == "success" + assert "cleared" in (state.status_message or "") + + +def test_dispatch_text_records_command_runtime(runtime_ctx: RuntimeContext) -> None: + state = TuiState(workspace_path=runtime_ctx.workspace.root) + state = dispatch_text(runtime_ctx, state, ":artifacts") + assert state.last_command_label == "artifacts" + assert state.last_command_ms is not None + assert state.last_command_ms >= 0 diff --git a/tests/tui/test_evidence_pane_scrolling.py b/tests/tui/test_evidence_pane_scrolling.py new file mode 100644 index 0000000..074996d --- /dev/null +++ b/tests/tui/test_evidence_pane_scrolling.py @@ -0,0 +1,99 @@ +"""Evidence pane keyboard scrolling: ↑/↓ and Page Up/Down move through a +long citation excerpt when the pane has focus. +""" + +from __future__ import annotations + +import pytest + +pytest.importorskip("textual") +pytest.importorskip("pytest_asyncio") + +from kairos.services.context import RuntimeContext +from kairos.tui.app import KairosApp +from kairos.tui.widgets.evidence_pane import EvidencePane +from kairos.tui.widgets.explorer_pane import ExplorerPane + +WIDE = (140, 20) + + +async def _type_command(pilot: object, text: str) -> None: + await pilot.click("#command-line") # type: ignore[attr-defined] + await pilot.press(*text) # type: ignore[attr-defined] + await pilot.press("enter") # type: ignore[attr-defined] + await pilot.pause() # type: ignore[attr-defined] + + +async def _select_first_result(pilot: object, app: KairosApp) -> None: + explorer = app.query_one(ExplorerPane) + explorer.focus() + await pilot.pause() # type: ignore[attr-defined] + await pilot.press("enter") # type: ignore[attr-defined] + await pilot.pause() # type: ignore[attr-defined] + + +@pytest.mark.asyncio +async def test_evidence_pane_scrolls_down_and_up_when_focused(runtime_ctx: RuntimeContext) -> None: + app = KairosApp(runtime_ctx) + async with app.run_test(size=WIDE) as pilot: + await _type_command(pilot, ":artifacts") + await _select_first_result(pilot, app) + + evidence = app.query_one(EvidencePane) + evidence.focus() + await pilot.pause() + assert evidence.scroll_y == 0 + + await pilot.press("down") + await pilot.press("down") + await pilot.pause() + assert evidence.scroll_y >= 0 # never goes negative; content may be short + + await pilot.press("up") + await pilot.pause() + + +@pytest.mark.asyncio +async def test_evidence_pane_page_down_and_page_up_do_not_crash( + runtime_ctx: RuntimeContext, +) -> None: + app = KairosApp(runtime_ctx) + async with app.run_test(size=WIDE) as pilot: + await _type_command(pilot, ":artifacts") + await _select_first_result(pilot, app) + + evidence = app.query_one(EvidencePane) + evidence.focus() + await pilot.pause() + + await pilot.press("pagedown") + await pilot.pause() + await pilot.press("pageup") + await pilot.pause() + await pilot.press("end") + await pilot.pause() + await pilot.press("home") + await pilot.pause() + assert evidence.scroll_y == 0 + + +@pytest.mark.asyncio +async def test_evidence_pane_resets_scroll_on_new_selection(runtime_ctx: RuntimeContext) -> None: + app = KairosApp(runtime_ctx) + async with app.run_test(size=WIDE) as pilot: + await _type_command(pilot, ":artifacts") + await _select_first_result(pilot, app) + + evidence = app.query_one(EvidencePane) + evidence.scroll_y = 5 + await pilot.pause() + + explorer = app.query_one(ExplorerPane) + explorer.focus() + await pilot.pause() + await pilot.press("down") + await pilot.pause() + await pilot.press("enter") + await pilot.pause() + + assert evidence.scroll_y == 0 diff --git a/tests/tui/test_explorer_navigation.py b/tests/tui/test_explorer_navigation.py new file mode 100644 index 0000000..7c22098 --- /dev/null +++ b/tests/tui/test_explorer_navigation.py @@ -0,0 +1,114 @@ +"""Explorer pane polish: item-count title, scroll indicators, line-number +gutter, search-term highlighting, and Ctrl+G "go to item". +""" + +from __future__ import annotations + +import pytest + +pytest.importorskip("textual") +pytest.importorskip("pytest_asyncio") + +from kairos.services.context import RuntimeContext +from kairos.tui.app import KairosApp +from kairos.tui.widgets.explorer_pane import ExplorerPane, _highlighted + +WIDE = (140, 40) + + +async def _type_command(pilot: object, text: str) -> None: + await pilot.click("#command-line") # type: ignore[attr-defined] + await pilot.press(*text) # type: ignore[attr-defined] + await pilot.press("enter") # type: ignore[attr-defined] + await pilot.pause() # type: ignore[attr-defined] + + +def test_highlighted_wraps_case_insensitive_match() -> None: + result = _highlighted("The Widget Manual", "widget") + assert "[reverse]Widget[/reverse]" in result + + +def test_highlighted_escapes_when_no_match() -> None: + assert _highlighted("plain text", "nope") == "plain text" + + +def test_highlighted_handles_empty_term() -> None: + assert _highlighted("plain text", None) == "plain text" + assert _highlighted("plain text", "") == "plain text" + + +@pytest.mark.asyncio +async def test_explorer_border_title_shows_mode_and_count(runtime_ctx: RuntimeContext) -> None: + app = KairosApp(runtime_ctx) + async with app.run_test(size=WIDE) as pilot: + await _type_command(pilot, ":artifacts") + explorer = app.query_one(ExplorerPane) + assert explorer.border_title == "Artifacts (7)" + + +@pytest.mark.asyncio +async def test_explorer_rows_carry_line_number_gutter_every_fifth( + runtime_ctx: RuntimeContext, +) -> None: + app = KairosApp(runtime_ctx) + async with app.run_test(size=WIDE) as pilot: + await _type_command(pilot, ":artifacts") + explorer = app.query_one(ExplorerPane) + rows = list(explorer.children) + assert len(rows) == 7 + fifth_row_text = str(rows[4].children[0].renderable) # type: ignore[attr-defined] + assert fifth_row_text.startswith(" 5 ") + + +@pytest.mark.asyncio +async def test_trace_nodes_highlight_the_query_term(runtime_ctx: RuntimeContext) -> None: + # :trace's node labels are the entity/span text itself (unlike :search's + # rows, which only show the source path) — so this is where the query + # term actually shows up in the Explorer list for highlighting to matter. + app = KairosApp(runtime_ctx) + async with app.run_test(size=WIDE) as pilot: + await _type_command(pilot, ":trace widget") + explorer = app.query_one(ExplorerPane) + rows = list(explorer.children) + assert rows + any_highlighted = any( + "[reverse]" in str(row.children[0].renderable) # type: ignore[attr-defined] + for row in rows + ) + assert any_highlighted + + +@pytest.mark.asyncio +async def test_ctrl_g_jumps_to_requested_item(runtime_ctx: RuntimeContext) -> None: + app = KairosApp(runtime_ctx) + async with app.run_test(size=WIDE) as pilot: + await _type_command(pilot, ":artifacts") + explorer = app.query_one(ExplorerPane) + explorer.focus() + await pilot.pause() + + await pilot.press("ctrl+g") + await pilot.pause() + await pilot.press("3") + await pilot.press("enter") + await pilot.pause() + + assert explorer.index == 2 + + +@pytest.mark.asyncio +async def test_ctrl_g_cancel_leaves_index_unchanged(runtime_ctx: RuntimeContext) -> None: + app = KairosApp(runtime_ctx) + async with app.run_test(size=WIDE) as pilot: + await _type_command(pilot, ":artifacts") + explorer = app.query_one(ExplorerPane) + explorer.focus() + await pilot.pause() + original_index = explorer.index + + await pilot.press("ctrl+g") + await pilot.pause() + await pilot.press("escape") + await pilot.pause() + + assert explorer.index == original_index diff --git a/tests/tui/test_header_updates.py b/tests/tui/test_header_updates.py new file mode 100644 index 0000000..7012198 --- /dev/null +++ b/tests/tui/test_header_updates.py @@ -0,0 +1,92 @@ +"""Header/status line: active well, workspace stats, and last-command +runtime shown in the header; color-coded success/error in the status line. +""" + +from __future__ import annotations + +import pytest + +pytest.importorskip("textual") +pytest.importorskip("pytest_asyncio") + +from kairos.services.context import RuntimeContext +from kairos.tui.app import KairosApp +from kairos.tui.widgets.header_line import HeaderLine, _format_size +from kairos.tui.widgets.status_line import StatusLine + +WIDE = (160, 40) + + +async def _type_command(pilot: object, text: str) -> None: + await pilot.click("#command-line") # type: ignore[attr-defined] + await pilot.press(*text) # type: ignore[attr-defined] + await pilot.press("enter") # type: ignore[attr-defined] + await pilot.pause() # type: ignore[attr-defined] + + +def test_format_size_units() -> None: + assert _format_size(0) == "0B" + assert _format_size(512) == "512B" + assert _format_size(2048) == "2.0KB" + assert _format_size(5 * 1024 * 1024) == "5.0MB" + + +@pytest.mark.asyncio +async def test_header_shows_workspace_stats_and_runtime(runtime_ctx: RuntimeContext) -> None: + app = KairosApp(runtime_ctx) + async with app.run_test(size=WIDE) as pilot: + await _type_command(pilot, ":home") + header_text = str(app.query_one(HeaderLine).renderable) + assert "7 artifact(s)" in header_text + assert "well(s)" in header_text + assert "home took" in header_text + assert "ms" in header_text + + +@pytest.mark.asyncio +async def test_header_reflects_active_well(runtime_ctx: RuntimeContext) -> None: + from kairos.services.wells import create_well + + create_well(runtime_ctx, "docs", "just docs") + app = KairosApp(runtime_ctx) + async with app.run_test(size=WIDE) as pilot: + await _type_command(pilot, ":well use docs") + header_text = str(app.query_one(HeaderLine).renderable) + assert "well: docs" in header_text + + +@pytest.mark.asyncio +async def test_status_line_colors_success_and_error(runtime_ctx: RuntimeContext) -> None: + app = KairosApp(runtime_ctx) + async with app.run_test(size=WIDE) as pilot: + await _type_command(pilot, ":artifacts") + # Static's `.update()` stores a Rich renderable; render it to plain + # segments to check markup was actually applied, not just present as + # literal text. + status = app.query_one(StatusLine) + from rich.console import Console + + console = Console(record=True, width=120, force_terminal=True, color_system="standard") + console.print(status.renderable) + rendered = console.export_text(styles=True) + assert "\x1b[32m" in rendered # green + + await _type_command(pilot, ":bogus") + console = Console(record=True, width=120, force_terminal=True, color_system="standard") + console.print(status.renderable) + rendered = console.export_text(styles=True) + assert "\x1b[31m" in rendered # red + + +@pytest.mark.asyncio +async def test_status_line_shows_running_before_dispatch_completes( + runtime_ctx: RuntimeContext, +) -> None: + app = KairosApp(runtime_ctx) + async with app.run_test(size=WIDE) as pilot: + await pilot.pause() + status = app.query_one(StatusLine) + status.show_running(":doctor") + text = str(status.renderable) + assert "running" in text + assert ":doctor" in text diff --git a/tests/tui/test_workflows.py b/tests/tui/test_workflows.py new file mode 100644 index 0000000..17406db --- /dev/null +++ b/tests/tui/test_workflows.py @@ -0,0 +1,204 @@ +"""End-to-end TUI workflows driven through Textual's headless Pilot — +multi-step operator sessions rather than single-command unit checks, per +the v0.2 improvement plan's Phase 5.1 (integration coverage). +""" + +from __future__ import annotations + +import pytest + +pytest.importorskip("textual") +pytest.importorskip("pytest_asyncio") + +from kairos.schemas.artifact import ArtifactDetail, ArtifactSummary +from kairos.schemas.note import NoteResult +from kairos.schemas.search import SearchResult +from kairos.services.context import RuntimeContext +from kairos.tui.app import KairosApp +from kairos.tui.commands import load_history +from kairos.tui.state import as_list_of +from kairos.tui.widgets.evidence_pane import EvidencePane +from kairos.tui.widgets.explorer_pane import ExplorerPane +from kairos.tui.widgets.header_line import HeaderLine +from kairos.tui.widgets.status_line import StatusLine + +WIDE = (140, 30) + + +async def _type_command(pilot: object, text: str) -> None: + await pilot.click("#command-line") # type: ignore[attr-defined] + await pilot.press(*text) # type: ignore[attr-defined] + await pilot.press("enter") # type: ignore[attr-defined] + await pilot.pause() # type: ignore[attr-defined] + + +@pytest.mark.asyncio +async def test_workflow_search_select_show_note(runtime_ctx: RuntimeContext) -> None: + """search a term -> inspect a hit's full citation -> open its artifact + in detail -> attach a note -> confirm it round-trips via :note list. + """ + app = KairosApp(runtime_ctx) + async with app.run_test(size=WIDE) as pilot: + await _type_command(pilot, ":search widget") + assert isinstance(app.state.last_result, SearchResult) + artifact_id = app.state.last_result.hits[0].provenance.artifact_id + + explorer = app.query_one(ExplorerPane) + explorer.focus() + await pilot.pause() + await pilot.press("enter") + await pilot.pause() + evidence_text = str(app.query_one(EvidencePane).renderable) + assert "artifact_id:" in evidence_text + + await _type_command(pilot, f":show {artifact_id}") + assert isinstance(app.state.last_result, ArtifactDetail) + assert app.state.last_result.artifact.id == artifact_id + + await _type_command(pilot, f":note add {artifact_id} looks correct") + await _type_command(pilot, f":note list {artifact_id}") + notes = as_list_of(app.state.last_result, NoteResult) + assert notes is not None + assert notes[-1].body == "looks correct" + + # The whole session shows up as command history, in order (after the + # app's own startup commands — auto-ingest and the initial :home). + assert app.state.command_history[-4:] == ( + ":search widget", + f":show {artifact_id}", + f":note add {artifact_id} looks correct", + f":note list {artifact_id}", + ) + + +@pytest.mark.asyncio +async def test_workflow_well_switch_narrows_results_mid_session( + runtime_ctx: RuntimeContext, +) -> None: + """Search unscoped, create a well scoped to one artifact, switch into + it, and confirm both the results and the header reflect the new scope + without restarting the app. + """ + from kairos.services.wells import add_member, create_well + + app = KairosApp(runtime_ctx) + async with app.run_test(size=WIDE) as pilot: + await _type_command(pilot, ":artifacts markdown") + artifacts = as_list_of(app.state.last_result, ArtifactSummary) + assert artifacts is not None + md_id = artifacts[0].id + create_well(runtime_ctx, "scoped", "only markdown") + add_member(runtime_ctx, "scoped", md_id) + + await _type_command(pilot, ":search widget") + assert isinstance(app.state.last_result, SearchResult) + unscoped_count = len(app.state.last_result.hits) + + await _type_command(pilot, ":well use scoped") + assert "well: scoped" in str(app.query_one(HeaderLine).renderable) + + await _type_command(pilot, ":search widget") + assert isinstance(app.state.last_result, SearchResult) + scoped_count = len(app.state.last_result.hits) + assert scoped_count < unscoped_count + + await _type_command(pilot, ":well clear") + assert "well: none" in str(app.query_one(HeaderLine).renderable) + + +@pytest.mark.asyncio +async def test_workflow_command_history_survives_a_restart( + runtime_ctx: RuntimeContext, +) -> None: + """Commands typed in one session are still on disk (and cycle-able via + ↑) after the app is closed and a fresh KairosApp is opened on the same + workspace — the whole point of persisting to .kairos/.tui_history. + """ + first = KairosApp(runtime_ctx) + async with first.run_test(size=WIDE) as pilot: + await _type_command(pilot, ":artifacts") + await _type_command(pilot, ":search widget") + + on_disk = [r.command for r in load_history(runtime_ctx.workspace.root)] + assert ":artifacts" in on_disk + assert ":search widget" in on_disk + + second = KairosApp(runtime_ctx) + async with second.run_test(size=WIDE) as pilot: + await pilot.pause() + assert ":search widget" in second.state.command_history + + from kairos.tui.widgets.command_line import CommandLine + + command_line = second.query_one(CommandLine) + command_line.focus() + await pilot.pause() + # The second session's own startup (:home, and possibly a re-run of + # the auto-ingest) sits most-recent in history; walk back past it to + # reach the first session's last command. + for _ in range(len(second.state.command_history)): + await pilot.press("up") + await pilot.pause() + if command_line.value == ":search widget": + break + assert command_line.value == ":search widget" + + +@pytest.mark.asyncio +async def test_workflow_explorer_goto_line_then_inspect_evidence( + runtime_ctx: RuntimeContext, +) -> None: + """Jump straight to a specific row with Ctrl+G, select it, then scroll + its citation in the Evidence pane with the keyboard. + """ + app = KairosApp(runtime_ctx) + async with app.run_test(size=WIDE) as pilot: + await _type_command(pilot, ":artifacts") + explorer = app.query_one(ExplorerPane) + explorer.focus() + await pilot.pause() + + await pilot.press("ctrl+g") + await pilot.pause() + await pilot.press("4") + await pilot.press("enter") + await pilot.pause() + assert explorer.index == 3 + + await pilot.press("enter") # select the jumped-to row + await pilot.pause() + evidence = app.query_one(EvidencePane) + assert "id:" in str(evidence.renderable) + + evidence.focus() + await pilot.pause() + await pilot.press("down") + await pilot.press("pagedown") + await pilot.press("home") + await pilot.pause() + assert evidence.scroll_y == 0 + + +@pytest.mark.asyncio +async def test_workflow_error_recovery_then_success(runtime_ctx: RuntimeContext) -> None: + """A typo'd command shows an actionable error and colors the status + line red; the corrected command then succeeds, turns the status line + green, and both attempts land in history in order. + """ + app = KairosApp(runtime_ctx) + async with app.run_test(size=WIDE) as pilot: + await _type_command(pilot, ":serach widget") + assert app.state.status == "error" + status_text = str(app.query_one(StatusLine).renderable) + assert "Unknown command" in status_text + assert "Traceback" not in status_text + + await _type_command(pilot, ":search widget") + assert app.state.status == "idle" + assert isinstance(app.state.last_result, SearchResult) + + assert app.state.command_history[-2:] == (":serach widget", ":search widget") + records = load_history(runtime_ctx.workspace.root) + by_command = {r.command: r.success for r in records} + assert by_command[":serach widget"] is False + assert by_command[":search widget"] is True From 9c83353f206cb86c897ccaf1e1740d1938c88315 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 19:30:31 +0000 Subject: [PATCH 2/8] style(tui): apply ruff format to evidence/explorer panes Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01JoRVBGw6Ea7wd8d1ZDFKAx --- src/kairos/tui/widgets/evidence_pane.py | 3 +-- src/kairos/tui/widgets/explorer_pane.py | 13 ++++++++----- 2 files changed, 9 insertions(+), 7 deletions(-) diff --git a/src/kairos/tui/widgets/evidence_pane.py b/src/kairos/tui/widgets/evidence_pane.py index 550d8f5..767f323 100644 --- a/src/kairos/tui/widgets/evidence_pane.py +++ b/src/kairos/tui/widgets/evidence_pane.py @@ -26,8 +26,7 @@ from kairos.tui.state import TuiState, as_list_of _NOT_SIMILARITY_NOTICE = ( - "\u25c6 This is an explicit deterministic relation.\n" - " It is not a semantic similarity claim." + "\u25c6 This is an explicit deterministic relation.\n It is not a semantic similarity claim." ) _LAYER_GLYPH = { diff --git a/src/kairos/tui/widgets/explorer_pane.py b/src/kairos/tui/widgets/explorer_pane.py index 449d91b..70192f5 100644 --- a/src/kairos/tui/widgets/explorer_pane.py +++ b/src/kairos/tui/widgets/explorer_pane.py @@ -284,9 +284,12 @@ def _dashboard_rows(d: DashboardResult) -> list[_Row]: rows.append(_Row(f" ▸ {bk.kind}", sub, "artifact", f"dashboard:kind:{bk.kind}")) # Recent activity for ev in d.recent_activity: - rows.append(_Row( - f" ▸ {ev.event_type}", - ev.occurred_at.isoformat(timespec="minutes"), - None, None, - )) + rows.append( + _Row( + f" ▸ {ev.event_type}", + ev.occurred_at.isoformat(timespec="minutes"), + None, + None, + ) + ) return rows or [_Row("○ Empty workspace — try :ingest .", "", None, None)] From 500654aafcddc2f85a8ff7cbf42d739fa8a19d50 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 19:39:05 +0000 Subject: [PATCH 3/8] fix: clear pre-existing ruff format/lint drift blocking CI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit main was already failing CI's ruff-format and ruff-check steps (ruff 0.16.2 flags formatting/import-order/style issues in demo.py, tool.py, and workspace_pane.py that predate this branch). Fixing these mechanically here — no behavior changes — so this PR's CI can go green; none of it is TUI feature work. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01JoRVBGw6Ea7wd8d1ZDFKAx --- src/kairos/cli/commands/demo.py | 25 +++++---- src/kairos/tool.py | 67 +++++++++++------------- src/kairos/tui/controller.py | 3 +- src/kairos/tui/widgets/workspace_pane.py | 16 ++---- 4 files changed, 54 insertions(+), 57 deletions(-) diff --git a/src/kairos/cli/commands/demo.py b/src/kairos/cli/commands/demo.py index a8e2e2c..443e04f 100644 --- a/src/kairos/cli/commands/demo.py +++ b/src/kairos/cli/commands/demo.py @@ -19,12 +19,12 @@ from kairos.cli.errors import cli_command from kairos.services.context import RuntimeContext +from kairos.services.doctor import run_doctor from kairos.services.ingest import ingest from kairos.services.search import search -from kairos.services.trace import trace from kairos.services.show import show -from kairos.services.wells import create_well, add_member, show_well -from kairos.services.doctor import run_doctor +from kairos.services.trace import trace +from kairos.services.wells import add_member, create_well, show_well demo_console = Console() @@ -75,7 +75,8 @@ def run() -> None: demo_console.print( Panel( "[dim]A temporary workspace will be created and destroyed.\n" - "Every result shown is real — sourced from the test fixtures shipped with KAIROS.[/dim]", + "Every result shown is real — sourced from the test fixtures shipped " + "with KAIROS.[/dim]", width=72, ) ) @@ -83,8 +84,8 @@ def run() -> None: try: # -- init ----------------------------------------------------------- _heading("1. init — create a workspace") - from kairos.infrastructure.filesystem.workspace import init_workspace from kairos.infrastructure.database.migrate import upgrade_to_head + from kairos.infrastructure.filesystem.workspace import init_workspace workspace = init_workspace(workspace_path, name="demo-workspace") upgrade_to_head(workspace.db_path) @@ -115,7 +116,7 @@ def run() -> None: ingest(ctx, python_dir, recursive=True) _ok(f"Python repo: {python_dir.name}/ (AST nodes → imports → classes)") - _info(f"All files parsed by structure, not chunked by byte count.") + _info("All files parsed by structure, not chunked by byte count.") # -- artifacts ------------------------------------------------------- _heading("3. artifacts — what's in the workspace") @@ -124,7 +125,11 @@ def run() -> None: all_artifacts = list_artifacts(ctx) table = Table("kind", "source_path", "parser", "status") for a in all_artifacts: - status = "[green]ok[/green]" if a.parse_status == "ok" else f"[yellow]{a.parse_status}[/yellow]" + status = ( + "[green]ok[/green]" + if a.parse_status == "ok" + else f"[yellow]{a.parse_status}[/yellow]" + ) table.add_row(a.kind, escape(a.source_path), a.parser_name, status) demo_console.print(table) _info(f"{len(all_artifacts)} artifacts ingested.") @@ -142,7 +147,8 @@ def run() -> None: escape(h.snippet[:80]), ) demo_console.print(st) - _ok(f"{search_result.hits[0].provenance.locator_str} — exact locator, extracted by parser") + locator_str = search_result.hits[0].provenance.locator_str + _ok(f"{locator_str} — exact locator, extracted by parser") else: _info("(no hits for 'widget' — fixtures may vary)") @@ -198,7 +204,8 @@ def run() -> None: "All parsing is structure-aware (AST, headings, JSON paths, Kconfig symbols,\n" "log sessions). All results carry provenance: artifact id, exact locator,\n" "parser name, parser version, provenance layer.\n\n" - "[dim]Temporary workspace has been removed. Nothing was written to your sources.[/dim]", + "[dim]Temporary workspace has been removed. " + "Nothing was written to your sources.[/dim]", width=72, ) ) diff --git a/src/kairos/tool.py b/src/kairos/tool.py index 4bf15c5..2b47a65 100644 --- a/src/kairos/tool.py +++ b/src/kairos/tool.py @@ -18,7 +18,7 @@ from __future__ import annotations -import os +import contextlib import urllib.parse from pathlib import Path from typing import Any @@ -33,7 +33,6 @@ from kairos.infrastructure.database.engine import session_scope from kairos.infrastructure.database.repositories import ( get_artifact, - get_span, list_spans_for_artifact, ) from kairos.schemas.provenance import ProvenanceEnvelope @@ -46,9 +45,17 @@ from kairos.services.trace import trace as _trace from kairos.services.wells import ( add_member as _well_add, +) +from kairos.services.wells import ( create_well as _well_create, +) +from kairos.services.wells import ( list_all_wells as _list_wells, +) +from kairos.services.wells import ( remove_member as _well_remove, +) +from kairos.services.wells import ( show_well as _well_show, ) @@ -84,9 +91,7 @@ def _try(fn, **default: Any) -> dict: return {"status": "error", "error": f"{type(e).__name__}: {e}"} -def _source_link_for_envelope( - envelope: ProvenanceEnvelope, workspace_root: Path -) -> str | None: +def _source_link_for_envelope(envelope: ProvenanceEnvelope, workspace_root: Path) -> str | None: """Build a clickable source link from a provenance envelope.""" locator = envelope.locator source_path = Path(envelope.source_path) @@ -122,9 +127,7 @@ def _read_bytes_around_locator( context_lines: int = 3, ) -> dict | None: """Read source bytes around a locator and return a snippet dict.""" - if isinstance(locator, LineRangeLocator): - start, end = locator.start_line, locator.end_line - elif isinstance(locator, RepoFileLinesLocator): + if isinstance(locator, (LineRangeLocator, RepoFileLinesLocator)): start, end = locator.start_line, locator.end_line else: return None @@ -182,8 +185,8 @@ def kairos_init(path: str | None = None, name: str | None = None) -> dict: Returns: dict with workspace path on success. """ - from kairos.infrastructure.filesystem.workspace import init_workspace from kairos.infrastructure.database.migrate import upgrade_to_head + from kairos.infrastructure.filesystem.workspace import init_workspace def _run(): root = Path(path).resolve() if path else Path.cwd().resolve() @@ -221,7 +224,11 @@ def _run(): source_link = None for a in artifacts: if a.id == o.artifact.id: - abs_path = Path.cwd() / a.source_path if not Path(a.source_path).is_absolute() else Path(a.source_path) + abs_path = ( + Path.cwd() / a.source_path + if not Path(a.source_path).is_absolute() + else Path(a.source_path) + ) source_link = abs_path.resolve().as_uri() break outcomes.append( @@ -235,8 +242,7 @@ def _run(): "relation_count": o.relation_count, "already_ingested": o.already_ingested, "diagnostics": [ - {"message": d.message, "severity": d.severity} - for d in o.diagnostics + {"message": d.message, "severity": d.severity} for d in o.diagnostics ], "source_link": source_link, } @@ -293,9 +299,7 @@ def _run(): return _try(_run) -def kairos_trace( - term: str, depth: int = 2, well: str | None = None -) -> dict: +def kairos_trace(term: str, depth: int = 2, well: str | None = None) -> dict: """Bidirectional entity trace with provenance on every edge. Args: @@ -489,7 +493,11 @@ def _run(): locator = locator_from_json(spans[0].locator_json) file_uri = abs_path.as_uri() - source_link = _make_link(file_uri, locator.start_line, locator.end_line) if isinstance(locator, (LineRangeLocator, RepoFileLinesLocator)) else file_uri + source_link = ( + _make_link(file_uri, locator.start_line, locator.end_line) + if isinstance(locator, (LineRangeLocator, RepoFileLinesLocator)) + else file_uri + ) return { "artifact_id": artifact_id, @@ -670,34 +678,23 @@ def kairos_status() -> dict: ctx = _ctx() def _run(): + import json as _json + from sqlalchemy import text as _text - import json as _json from kairos.infrastructure.database.engine import fts5_is_available # read config for name/schema_version _ws_cfg = {} - try: + with contextlib.suppress(Exception): _ws_cfg = _json.loads(ctx.workspace.config_path.read_text(encoding="utf-8")) - except Exception: - pass with session_scope(ctx.session_factory) as session: - artifacts = session.execute( - _text("SELECT COUNT(*) FROM artifacts") - ).scalar() or 0 - entities = session.execute( - _text("SELECT COUNT(*) FROM entities") - ).scalar() or 0 - relations = session.execute( - _text("SELECT COUNT(*) FROM relations") - ).scalar() or 0 - spans = session.execute( - _text("SELECT COUNT(*) FROM source_spans") - ).scalar() or 0 - wells = session.execute( - _text("SELECT COUNT(*) FROM coherence_wells") - ).scalar() or 0 + artifacts = session.execute(_text("SELECT COUNT(*) FROM artifacts")).scalar() or 0 + entities = session.execute(_text("SELECT COUNT(*) FROM entities")).scalar() or 0 + relations = session.execute(_text("SELECT COUNT(*) FROM relations")).scalar() or 0 + spans = session.execute(_text("SELECT COUNT(*) FROM source_spans")).scalar() or 0 + wells = session.execute(_text("SELECT COUNT(*) FROM coherence_wells")).scalar() or 0 return { "workspace": { diff --git a/src/kairos/tui/controller.py b/src/kairos/tui/controller.py index e1b8fc0..84beaa3 100644 --- a/src/kairos/tui/controller.py +++ b/src/kairos/tui/controller.py @@ -223,12 +223,13 @@ def _home(runtime_ctx: RuntimeContext, state: TuiState, command: Command) -> Tui workspace_size_bytes=total_size, well_count=total_wells, ) + summary = f"{total_artifacts} artifacts, {total_entities} entities, {total_relations} relations" return _record( updated_state, mode="home", command=command.raw, status="success", - summary=f"{total_artifacts} artifacts, {total_entities} entities, {total_relations} relations", + summary=summary, last_result=dashboard, ) diff --git a/src/kairos/tui/widgets/workspace_pane.py b/src/kairos/tui/widgets/workspace_pane.py index a0cf880..a1ae874 100644 --- a/src/kairos/tui/widgets/workspace_pane.py +++ b/src/kairos/tui/widgets/workspace_pane.py @@ -95,14 +95,10 @@ def _render_result(state: TuiState) -> object: text = Text() text.append(f"\u25c6 trace: {result.query}\n", style="bold cyan") for edge in result.edges: - text.append( - f" {edge.subject_id[:8]} ", style="dim" - ) + text.append(f" {edge.subject_id[:8]} ", style="dim") text.append(f"\u2500\u2500{edge.predicate}\u2500\u2500> ", style="yellow") text.append(f"{edge.object_id[:8]}\n", style="dim") - text.append( - f" ({edge.layer}, rule={edge.derivation_rule or 'n/a'})\n", style="dim" - ) + text.append(f" ({edge.layer}, rule={edge.derivation_rule or 'n/a'})\n", style="dim") if not result.edges: text.append(" (no explicit relations found)\n", style="dim italic") return text @@ -203,9 +199,7 @@ def _render_dashboard(d: DashboardResult) -> object: # Breakdown by kind if d.artifacts_by_kind: - breakdown = Table( - title="Artifact breakdown", show_lines=False, padding=(0, 2), box=None - ) + breakdown = Table(title="Artifact breakdown", show_lines=False, padding=(0, 2), box=None) breakdown.add_column("kind", style="cyan") breakdown.add_column("count", justify="right") breakdown.add_column("ok", justify="right") @@ -225,9 +219,7 @@ def _render_dashboard(d: DashboardResult) -> object: # Recent activity if d.recent_activity: - events = Table( - title="Recent activity", show_lines=False, padding=(0, 2), box=None - ) + events = Table(title="Recent activity", show_lines=False, padding=(0, 2), box=None) events.add_column("time", style="dim") events.add_column("event", style="cyan") for ev in d.recent_activity: From 239604da1f547232d9d5cc616907dfed142773d5 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 19:47:41 +0000 Subject: [PATCH 4/8] fix(ci): install TUI extras before pyright, not after MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pyrightconfig's include covers all of src/ and tests/ unconditionally, but the workflow installed the optional `tui`/`tui-test` extras only after the Pyright step ran. Every file under src/kairos/tui and tests/tui was therefore unable to resolve `textual`/`textual.*` imports, cascading into 800+ reportUnknown* errors regardless of whether the code itself was correct — confirmed this reproduces on unmodified main under the same conditions, predating this branch. Moved "Install TUI extras" to right before Pyright, and moved the base "Pytest" step (which intentionally runs *without* the tui extra, to prove no accidental hard dependency on it) to run before that install so its behavior is unchanged. "Pytest (TUI)" already ran after the extras were installed and is unaffected. Also renamed two now-tested helpers (`_highlighted` -> `highlighted` in explorer_pane.py, `_format_size` -> `format_size` in header_line.py) that pyright flagged as reportPrivateUsage once it could actually see them, since my new unit tests import them directly. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01JoRVBGw6Ea7wd8d1ZDFKAx --- .github/workflows/ci.yml | 12 +++++++++--- src/kairos/tui/widgets/explorer_pane.py | 6 +++--- src/kairos/tui/widgets/header_line.py | 4 ++-- tests/tui/test_explorer_navigation.py | 10 +++++----- tests/tui/test_header_updates.py | 10 +++++----- 5 files changed, 24 insertions(+), 18 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 57e5988..4b1c89b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -25,15 +25,21 @@ jobs: - name: Ruff lint run: ruff check src tests - - name: Pyright - run: pyright - - name: Pytest run: pytest -q - name: Install TUI extras run: pip install -e ".[tui,tui-test]" + # Runs after the TUI extras are installed (unlike the base Pytest step + # above) so pyright can actually resolve `textual`/`textual.*` imports + # in src/kairos/tui and tests/tui — pyrightconfig's `include` covers + # both unconditionally, so without this ordering every TUI-touching + # file fails with hundreds of reportUnknown* cascade errors regardless + # of whether the code itself is otherwise correct. + - name: Pyright + run: pyright + - name: Pytest (TUI) run: pytest -q tests/tui diff --git a/src/kairos/tui/widgets/explorer_pane.py b/src/kairos/tui/widgets/explorer_pane.py index 70192f5..2786eda 100644 --- a/src/kairos/tui/widgets/explorer_pane.py +++ b/src/kairos/tui/widgets/explorer_pane.py @@ -66,7 +66,7 @@ class _Row: target_id: str | None -def _highlighted(text: str, term: str | None) -> str: +def highlighted(text: str, term: str | None) -> str: """Escape ``text`` for Rich markup, wrapping case-insensitive matches of ``term`` in a reverse-video span. Escaping happens per-chunk so the highlight markup itself is never escaped away. @@ -92,7 +92,7 @@ def _highlighted(text: str, term: str | None) -> str: class ExplorerItem(ListItem): def __init__(self, row: _Row, index: int, query_term: str | None = None) -> None: gutter = f"{index + 1:>4} " if (index + 1) % _GUTTER_EVERY == 0 else " " - text = f"{gutter}{_highlighted(row.label, query_term)}" + text = f"{gutter}{highlighted(row.label, query_term)}" if row.sublabel: text += f"\n [dim]{escape(row.sublabel)}[/dim]" super().__init__(Static(text)) @@ -118,7 +118,7 @@ def selected_reference(self) -> tuple[SelectionKind, str] | None: return item.kind, item.target_id return None - def on_list_view_highlighted(self, event: ListView.Highlighted) -> None: + def on_list_viewhighlighted(self, event: ListView.Highlighted) -> None: self.call_after_refresh(self._update_scroll_indicators) def _update_scroll_indicators(self) -> None: diff --git a/src/kairos/tui/widgets/header_line.py b/src/kairos/tui/widgets/header_line.py index ff3ae3d..da8eba8 100644 --- a/src/kairos/tui/widgets/header_line.py +++ b/src/kairos/tui/widgets/header_line.py @@ -10,7 +10,7 @@ _OFFLINE_GLYPH = "●" -def _format_size(num_bytes: int) -> str: +def format_size(num_bytes: int) -> str: size = float(num_bytes) for unit in ("B", "KB", "MB", "GB"): if size < 1024 or unit == "GB": @@ -28,7 +28,7 @@ def refresh_from_state(self, state: TuiState) -> None: well = escape(state.active_well) if state.active_well else "none" stats = ( f"{state.artifact_count} artifact(s) · " - f"{_format_size(state.workspace_size_bytes)} · " + f"{format_size(state.workspace_size_bytes)} · " f"{state.well_count} well(s)" ) runtime = "" diff --git a/tests/tui/test_explorer_navigation.py b/tests/tui/test_explorer_navigation.py index 7c22098..7ea7ecc 100644 --- a/tests/tui/test_explorer_navigation.py +++ b/tests/tui/test_explorer_navigation.py @@ -11,7 +11,7 @@ from kairos.services.context import RuntimeContext from kairos.tui.app import KairosApp -from kairos.tui.widgets.explorer_pane import ExplorerPane, _highlighted +from kairos.tui.widgets.explorer_pane import ExplorerPane, highlighted WIDE = (140, 40) @@ -24,17 +24,17 @@ async def _type_command(pilot: object, text: str) -> None: def test_highlighted_wraps_case_insensitive_match() -> None: - result = _highlighted("The Widget Manual", "widget") + result = highlighted("The Widget Manual", "widget") assert "[reverse]Widget[/reverse]" in result def test_highlighted_escapes_when_no_match() -> None: - assert _highlighted("plain text", "nope") == "plain text" + assert highlighted("plain text", "nope") == "plain text" def test_highlighted_handles_empty_term() -> None: - assert _highlighted("plain text", None) == "plain text" - assert _highlighted("plain text", "") == "plain text" + assert highlighted("plain text", None) == "plain text" + assert highlighted("plain text", "") == "plain text" @pytest.mark.asyncio diff --git a/tests/tui/test_header_updates.py b/tests/tui/test_header_updates.py index 7012198..632f8cd 100644 --- a/tests/tui/test_header_updates.py +++ b/tests/tui/test_header_updates.py @@ -11,7 +11,7 @@ from kairos.services.context import RuntimeContext from kairos.tui.app import KairosApp -from kairos.tui.widgets.header_line import HeaderLine, _format_size +from kairos.tui.widgets.header_line import HeaderLine, format_size from kairos.tui.widgets.status_line import StatusLine WIDE = (160, 40) @@ -25,10 +25,10 @@ async def _type_command(pilot: object, text: str) -> None: def test_format_size_units() -> None: - assert _format_size(0) == "0B" - assert _format_size(512) == "512B" - assert _format_size(2048) == "2.0KB" - assert _format_size(5 * 1024 * 1024) == "5.0MB" + assert format_size(0) == "0B" + assert format_size(512) == "512B" + assert format_size(2048) == "2.0KB" + assert format_size(5 * 1024 * 1024) == "5.0MB" @pytest.mark.asyncio From c41391d014551289b80e3b2729e88a30e826a593 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 19:49:59 +0000 Subject: [PATCH 5/8] fix(tests): update :home test for the dashboard result shape MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit test_home_lists_recent_activity predates the dashboard-home-screen feature (commit 8efe062) and still expected :home's last_result to be a bare list[ActivityEvent] — it's been a DashboardResult ever since, which also carries recent_activity. This was failing on main already; fixing it here so this branch's CI can actually reach the TUI-specific checks instead of dying on an unrelated pre-existing test bug. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01JoRVBGw6Ea7wd8d1ZDFKAx --- tests/tui/test_controller.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/tests/tui/test_controller.py b/tests/tui/test_controller.py index afcc924..cc06520 100644 --- a/tests/tui/test_controller.py +++ b/tests/tui/test_controller.py @@ -8,9 +8,9 @@ import dataclasses -from kairos.schemas.activity import ActivityEvent from kairos.schemas.artifact import ArtifactDetail, ArtifactSummary from kairos.schemas.config import ConfigSymbolResult +from kairos.schemas.dashboard import DashboardResult from kairos.schemas.doctor import DoctorReport from kairos.schemas.logs import LogHit from kairos.schemas.note import NoteResult @@ -172,8 +172,10 @@ def test_note_add_and_list_are_the_only_mutations(runtime_ctx: RuntimeContext) - assert notes[0].body == "looks good" -def test_home_lists_recent_activity(runtime_ctx: RuntimeContext) -> None: +def test_home_shows_dashboard_with_recent_activity(runtime_ctx: RuntimeContext) -> None: state = dispatch_text(runtime_ctx, _fresh_state(runtime_ctx), ":artifacts") state = dispatch_text(runtime_ctx, state, ":home") assert state.mode == "home" - assert as_list_of(state.last_result, ActivityEvent) is not None + assert isinstance(state.last_result, DashboardResult) + assert state.last_result.total_artifacts == 7 + assert state.last_result.recent_activity # :artifacts just ran, so activity isn't empty From 7b72273120acb3132d1b1c13bd561f8d9ff833e9 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 21:00:45 +0000 Subject: [PATCH 6/8] fix(tui): make history persistence fail closed, stream file reads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses Copilot review feedback on PR #3: - append_history/clear_history now suppress OSError instead of propagating it. History is a convenience feature — a read-only workspace or other filesystem error must not break command dispatch, which calls append_history after every single command. The in-memory TuiState.command_history update (what actually backs the command line's up/down cycling) is unaffected either way. - load_history now iterates the history file line-by-line instead of read_text().splitlines(), avoiding holding the whole file as both a raw string and a list of lines at once for a file that's append-only and can grow without bound. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01JoRVBGw6Ea7wd8d1ZDFKAx --- src/kairos/tui/commands.py | 48 +++++++++++++++++++------------ tests/tui/test_command_history.py | 27 +++++++++++++++++ 2 files changed, 56 insertions(+), 19 deletions(-) diff --git a/src/kairos/tui/commands.py b/src/kairos/tui/commands.py index eabc1df..6be641b 100644 --- a/src/kairos/tui/commands.py +++ b/src/kairos/tui/commands.py @@ -5,6 +5,7 @@ from __future__ import annotations +import contextlib import json from dataclasses import dataclass from datetime import UTC, datetime @@ -178,37 +179,46 @@ def load_history(workspace_root: Path) -> list[HistoryRecord]: if not path.exists(): return [] records: list[HistoryRecord] = [] - for line in path.read_text(encoding="utf-8").splitlines(): - line = line.strip() - if not line: - continue - try: - data = json.loads(line) - records.append( - HistoryRecord( - timestamp=datetime.fromisoformat(data["timestamp"]), - command=data["command"], - success=bool(data["success"]), + with path.open("r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + continue + try: + data = json.loads(line) + records.append( + HistoryRecord( + timestamp=datetime.fromisoformat(data["timestamp"]), + command=data["command"], + success=bool(data["success"]), + ) ) - ) - except (json.JSONDecodeError, KeyError, ValueError): - continue + except (json.JSONDecodeError, KeyError, ValueError): + continue return records def append_history(workspace_root: Path, command: str, *, success: bool) -> None: + """Best-effort: history is a convenience, not a source of truth, so a + read-only workspace or other filesystem error here must never break + command dispatch (which calls this after *every* command). The caller's + in-memory ``TuiState.command_history`` update happens independently and + always succeeds regardless of whether this persists to disk. + """ path = history_file_path(workspace_root) - path.parent.mkdir(parents=True, exist_ok=True) record = { "timestamp": datetime.now(UTC).isoformat(), "command": command, "success": success, } - with path.open("a", encoding="utf-8") as f: - f.write(json.dumps(record) + "\n") + with contextlib.suppress(OSError): + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("a", encoding="utf-8") as f: + f.write(json.dumps(record) + "\n") def clear_history(workspace_root: Path) -> None: path = history_file_path(workspace_root) - if path.exists(): - path.write_text("", encoding="utf-8") + with contextlib.suppress(OSError): + if path.exists(): + path.write_text("", encoding="utf-8") diff --git a/tests/tui/test_command_history.py b/tests/tui/test_command_history.py index ca406ee..6373b01 100644 --- a/tests/tui/test_command_history.py +++ b/tests/tui/test_command_history.py @@ -53,6 +53,22 @@ def test_history_clear_on_missing_file_is_a_no_op(tmp_path: Path) -> None: assert load_history(tmp_path) == [] +def test_append_history_never_raises_on_filesystem_error(tmp_path: Path) -> None: + # A file sits where the .kairos directory needs to go, so mkdir fails — + # history is a convenience feature and must fail closed, not take down + # command dispatch (which calls this after every command). + (tmp_path / ".kairos").write_text("not a directory") + append_history(tmp_path, ":search widget", success=True) # must not raise + assert load_history(tmp_path) == [] + + +def test_clear_history_never_raises_on_filesystem_error(tmp_path: Path) -> None: + history_dir = tmp_path / ".kairos" + history_dir.mkdir() + (history_dir / ".tui_history").mkdir() # a directory, not a file + clear_history(tmp_path) # must not raise + + def test_hint_text_for_partial_and_full_commands() -> None: assert hint_text("") == "" assert hint_text("hello") == "" @@ -85,6 +101,17 @@ def test_dispatch_text_appends_to_state_and_disk_history(runtime_ctx: RuntimeCon assert [r.success for r in records] == [True, False] +def test_dispatch_text_survives_history_write_failure(runtime_ctx: RuntimeContext) -> None: + # A directory sits where the history file needs to go, so the append + # write fails — dispatch (and the in-memory history it drives the + # command line's ↑/↓ from) must not be affected. + (runtime_ctx.workspace.root / ".kairos" / ".tui_history").mkdir() + state = TuiState(workspace_path=runtime_ctx.workspace.root) + state = dispatch_text(runtime_ctx, state, ":artifacts") + assert state.mode == "artifacts" + assert state.command_history == (":artifacts",) + + def test_history_clear_command_wipes_state_and_disk(runtime_ctx: RuntimeContext) -> None: state = TuiState(workspace_path=runtime_ctx.workspace.root) state = dispatch_text(runtime_ctx, state, ":artifacts") From c9b1883190429684b5ceecf9708e830745d00f27 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 8 Aug 2026 02:10:41 +0000 Subject: [PATCH 7/8] fix: clear remaining pre-existing pyright strict-mode debt MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Types out kairos/tool.py's dict returns (dict -> dict[str, Any], plus explicit return-type annotations and list[dict[str, Any]] locals on the inner _run() closures) and workspace_pane.py's dashboard renderable list (list[object] -> list[RenderableType]), clearing the ~88 pyright errors flagged in PR #3's review discussion. No behavior changes, except one real bug this surfaced: kairos_init() read a nonexistent Workspace.name attribute (Workspace only has root/kairos_dir/db_path/ etc., see infrastructure/filesystem/workspace.py) — would have raised AttributeError on every call. Fixed to mirror init_workspace()'s own `name or root.name` fallback. Untested/unused code path, so no regression risk, but worth calling out. pyright now reports 0 errors repo-wide. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01JoRVBGw6Ea7wd8d1ZDFKAx --- src/kairos/tool.py | 79 ++++++++++++------------ src/kairos/tui/widgets/workspace_pane.py | 4 +- 2 files changed, 42 insertions(+), 41 deletions(-) diff --git a/src/kairos/tool.py b/src/kairos/tool.py index 2b47a65..77792a3 100644 --- a/src/kairos/tool.py +++ b/src/kairos/tool.py @@ -20,6 +20,7 @@ import contextlib import urllib.parse +from collections.abc import Callable from pathlib import Path from typing import Any @@ -63,25 +64,25 @@ # helpers # --------------------------------------------------------------------------- -_CTX: RuntimeContext | None = None +_ctx_singleton: RuntimeContext | None = None def _ctx() -> RuntimeContext: - global _CTX - if _CTX is None: + global _ctx_singleton + if _ctx_singleton is None: try: - _CTX = RuntimeContext.open(Path.cwd()) + _ctx_singleton = RuntimeContext.open(Path.cwd()) except Exception as exc: raise KairosError(f"No KAIROS workspace found: {exc}") from exc - return _CTX + return _ctx_singleton def _reset_ctx() -> None: - global _CTX - _CTX = None + global _ctx_singleton + _ctx_singleton = None -def _try(fn, **default: Any) -> dict: +def _try(fn: Callable[[], dict[str, Any]], **default: Any) -> dict[str, Any]: """Wrap a KAIROS service call into a status dict.""" try: return {"status": "ok", **fn()} @@ -125,7 +126,7 @@ def _read_bytes_around_locator( abs_path: Path, locator: Locator, context_lines: int = 3, -) -> dict | None: +) -> dict[str, Any] | None: """Read source bytes around a locator and return a snippet dict.""" if isinstance(locator, (LineRangeLocator, RepoFileLinesLocator)): start, end = locator.start_line, locator.end_line @@ -140,7 +141,7 @@ def _read_bytes_around_locator( ctx_start = max(0, start - context_lines - 1) ctx_end = min(len(lines), end + context_lines) - snippet_lines = [] + snippet_lines: list[str] = [] for i in range(ctx_start, ctx_end): line_no = i + 1 marker = " >" if start - 1 <= i < end else " " @@ -175,7 +176,7 @@ def _resolve_artifact_path( # --------------------------------------------------------------------------- -def kairos_init(path: str | None = None, name: str | None = None) -> dict: +def kairos_init(path: str | None = None, name: str | None = None) -> dict[str, Any]: """Initialise a KAIROS workspace (like ``kairos init``). Args: @@ -188,7 +189,7 @@ def kairos_init(path: str | None = None, name: str | None = None) -> dict: from kairos.infrastructure.database.migrate import upgrade_to_head from kairos.infrastructure.filesystem.workspace import init_workspace - def _run(): + def _run() -> dict[str, Any]: root = Path(path).resolve() if path else Path.cwd().resolve() workspace = init_workspace(root, name=name) upgrade_to_head(workspace.db_path) @@ -197,13 +198,13 @@ def _run(): return { "workspace_path": str(root), "db_path": str(workspace.db_path), - "name": workspace.name, + "name": name or root.name, } return _try(_run) -def kairos_ingest(path: str = ".", recursive: bool = True) -> dict: +def kairos_ingest(path: str = ".", recursive: bool = True) -> dict[str, Any]: """Ingest files into the workspace. Args: @@ -216,9 +217,9 @@ def kairos_ingest(path: str = ".", recursive: bool = True) -> dict: """ ctx = _ctx() - def _run(): + def _run() -> dict[str, Any]: report = _ingest(ctx, Path(path), recursive=recursive) - outcomes = [] + outcomes: list[dict[str, Any]] = [] for o in report.outcomes: artifacts = _list_artifacts(ctx) source_link = None @@ -257,7 +258,7 @@ def _run(): return _try(_run) -def kairos_search(query: str, limit: int = 20, well: str | None = None) -> dict: +def kairos_search(query: str, limit: int = 20, well: str | None = None) -> dict[str, Any]: """Full-text search with provenance envelopes. Args: @@ -271,9 +272,9 @@ def kairos_search(query: str, limit: int = 20, well: str | None = None) -> dict: """ ctx = _ctx() - def _run(): + def _run() -> dict[str, Any]: result = _search(ctx, query, well=well) - hits = [] + hits: list[dict[str, Any]] = [] ws_root = ctx.workspace.root for h in result.hits[:limit]: source_link = _source_link_for_envelope(h.provenance, ws_root) @@ -299,7 +300,7 @@ def _run(): return _try(_run) -def kairos_trace(term: str, depth: int = 2, well: str | None = None) -> dict: +def kairos_trace(term: str, depth: int = 2, well: str | None = None) -> dict[str, Any]: """Bidirectional entity trace with provenance on every edge. Args: @@ -312,10 +313,10 @@ def kairos_trace(term: str, depth: int = 2, well: str | None = None) -> dict: """ ctx = _ctx() - def _run(): + def _run() -> dict[str, Any]: result: TraceResult = _trace(ctx, term, depth=depth, well=well) ws_root = ctx.workspace.root - nodes_out = [] + nodes_out: list[dict[str, Any]] = [] for n in result.nodes: source_link = None if n.provenance is not None: @@ -337,7 +338,7 @@ def _run(): ), } ) - edges_out = [] + edges_out: list[dict[str, Any]] = [] for e in result.edges: edges_out.append( { @@ -363,7 +364,7 @@ def _run(): return _try(_run) -def kairos_show(artifact_id: str) -> dict: +def kairos_show(artifact_id: str) -> dict[str, Any]: """Show full artifact detail with all spans and provenance. Args: @@ -374,10 +375,10 @@ def kairos_show(artifact_id: str) -> dict: """ ctx = _ctx() - def _run(): + def _run() -> dict[str, Any]: detail = _show(ctx, artifact_id) ws_root = ctx.workspace.root - spans = [] + spans: list[dict[str, Any]] = [] for s in detail.spans: source_link = _source_link_for_envelope(s.provenance, ws_root) spans.append( @@ -410,7 +411,7 @@ def _run(): def kairos_source_content( artifact_id: str, context_lines: int = 3, -) -> dict: +) -> dict[str, Any]: """Read actual source bytes around each locatable span. Args: @@ -422,7 +423,7 @@ def kairos_source_content( """ ctx = _ctx() - def _run(): + def _run() -> dict[str, Any]: resolved = _resolve_artifact_path(artifact_id) if resolved is None: raise KairosError(f"Artifact not found: {artifact_id}") @@ -431,7 +432,7 @@ def _run(): with session_scope(ctx.session_factory) as session: span_rows = list_spans_for_artifact(session, artifact_id) - payload = { + payload: dict[str, Any] = { "artifact_id": artifact_id, "source_path": str(rel_path), "file_path": str(abs_path), @@ -460,7 +461,7 @@ def _run(): return _try(_run) -def kairos_source_link(artifact_id: str, locator_str: str | None = None) -> dict: +def kairos_source_link(artifact_id: str, locator_str: str | None = None) -> dict[str, Any]: """Resolve an artifact + optional locator to clickable source links. Args: @@ -510,7 +511,7 @@ def _run(): return _try(_run) -def kairos_artifacts(kind: str | None = None) -> dict: +def kairos_artifacts(kind: str | None = None) -> dict[str, Any]: """List artifacts in the workspace. Args: @@ -542,7 +543,7 @@ def _run(): return _try(_run) -def kairos_well_create(name: str, purpose: str = "") -> dict: +def kairos_well_create(name: str, purpose: str = "") -> dict[str, Any]: """Create a coherence well to scope a working set. Args: @@ -566,7 +567,7 @@ def _run(): return _try(_run) -def kairos_well_add(well_name: str, target_id: str, note: str | None = None) -> dict: +def kairos_well_add(well_name: str, target_id: str, note: str | None = None) -> dict[str, Any]: """Add an artifact or span to a coherence well. Args: @@ -592,7 +593,7 @@ def _run(): return _try(_run) -def kairos_well_show(well_name: str) -> dict: +def kairos_well_show(well_name: str) -> dict[str, Any]: """Show a coherence well's contents. Args: @@ -624,7 +625,7 @@ def _run(): return _try(_run) -def kairos_well_list() -> dict: +def kairos_well_list() -> dict[str, Any]: """List all coherence wells. Returns: @@ -650,7 +651,7 @@ def _run(): return _try(_run) -def kairos_well_remove(well_name: str, member_id: str) -> dict: +def kairos_well_remove(well_name: str, member_id: str) -> dict[str, Any]: """Remove a member from a coherence well. Args: @@ -669,7 +670,7 @@ def _run(): return _try(_run) -def kairos_status() -> dict: +def kairos_status() -> dict[str, Any]: """Check KAIROS workspace status. Returns: @@ -677,7 +678,7 @@ def kairos_status() -> dict: """ ctx = _ctx() - def _run(): + def _run() -> dict[str, Any]: import json as _json from sqlalchemy import text as _text @@ -685,7 +686,7 @@ def _run(): from kairos.infrastructure.database.engine import fts5_is_available # read config for name/schema_version - _ws_cfg = {} + _ws_cfg: dict[str, Any] = {} with contextlib.suppress(Exception): _ws_cfg = _json.loads(ctx.workspace.config_path.read_text(encoding="utf-8")) diff --git a/src/kairos/tui/widgets/workspace_pane.py b/src/kairos/tui/widgets/workspace_pane.py index a1ae874..33f4b58 100644 --- a/src/kairos/tui/widgets/workspace_pane.py +++ b/src/kairos/tui/widgets/workspace_pane.py @@ -176,10 +176,10 @@ def _render_result(state: TuiState) -> object: def _render_dashboard(d: DashboardResult) -> object: - from rich.console import Group + from rich.console import Group, RenderableType from rich.table import Table - items: list[object] = [] + items: list[RenderableType] = [] # Metrics row metrics = Table(show_header=False, show_lines=False, padding=(0, 3), box=None) From 6a7ed25e9431f3096878ad6b135f3ed3acc3c9a7 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 8 Aug 2026 02:20:22 +0000 Subject: [PATCH 8/8] fix(tests): stop test_tui_makes_no_network_access from false-failing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The test blanket-blocked socket.socket() for any address family. But asyncio's own SelectorEventLoop uses an AF_UNIX socketpair internally for self-pipe wakeups — this is local IPC, not network access — and CPython can defer that loop's __del__/cleanup to run during this test's own teardown window (while the monkeypatch is still active), tripping the assertion for something the test was never meant to catch. Confirmed via CI logs: the blocked call was literally socket.socket(AF_UNIX, SOCK_STREAM, 0, ...). Narrowed the guard to only block AF_INET/AF_INET6 socket() calls (create_connection/getaddrinfo stay fully blocked, since those are unconditionally about real network access). Verified stable across repeated runs, both in isolation and as part of the full suite. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01JoRVBGw6Ea7wd8d1ZDFKAx --- tests/tui/test_app.py | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/tests/tui/test_app.py b/tests/tui/test_app.py index ec3f717..9b8369e 100644 --- a/tests/tui/test_app.py +++ b/tests/tui/test_app.py @@ -414,7 +414,22 @@ async def test_tui_makes_no_network_access( def _blocked(*_args: object, **_kwargs: object) -> None: raise AssertionError("KAIROS TUI attempted network access.") - monkeypatch.setattr(socket, "socket", _blocked) + real_socket = socket.socket + + def _guarded_socket( + family: int = socket.AF_INET, + type: int = socket.SOCK_STREAM, + *args: object, + **kwargs: object, + ) -> socket.socket: + # AF_UNIX/AF_UNIX-family sockets are local IPC — asyncio's own event + # loop uses one internally for self-pipe wakeups, unrelated to any + # actual network access. Only block real network address families. + if family in (socket.AF_INET, socket.AF_INET6): + _blocked() + return real_socket(family, type, *args, **kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(socket, "socket", _guarded_socket) monkeypatch.setattr(socket, "create_connection", _blocked) monkeypatch.setattr(socket, "getaddrinfo", _blocked)