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/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..77792a3 100644 --- a/src/kairos/tool.py +++ b/src/kairos/tool.py @@ -18,8 +18,9 @@ from __future__ import annotations -import os +import contextlib import urllib.parse +from collections.abc import Callable from pathlib import Path from typing import Any @@ -33,7 +34,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 +46,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, ) @@ -56,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()} @@ -84,9 +92,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) @@ -120,11 +126,9 @@ 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): - 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 @@ -137,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 " " @@ -172,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: @@ -182,10 +186,10 @@ 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(): + 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) @@ -194,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: @@ -213,15 +217,19 @@ 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 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 +243,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, } @@ -251,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: @@ -265,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) @@ -293,9 +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: @@ -308,10 +313,10 @@ def kairos_trace( """ 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: @@ -333,7 +338,7 @@ def _run(): ), } ) - edges_out = [] + edges_out: list[dict[str, Any]] = [] for e in result.edges: edges_out.append( { @@ -359,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: @@ -370,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( @@ -406,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: @@ -418,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}") @@ -427,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), @@ -456,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: @@ -489,7 +494,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, @@ -502,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: @@ -534,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: @@ -558,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: @@ -584,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: @@ -616,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: @@ -642,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: @@ -661,7 +670,7 @@ def _run(): return _try(_run) -def kairos_status() -> dict: +def kairos_status() -> dict[str, Any]: """Check KAIROS workspace status. Returns: @@ -669,35 +678,24 @@ def kairos_status() -> dict: """ ctx = _ctx() - def _run(): + def _run() -> dict[str, Any]: + 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: + _ws_cfg: dict[str, Any] = {} + 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/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..6be641b 100644 --- a/src/kairos/tui/commands.py +++ b/src/kairos/tui/commands.py @@ -5,7 +5,11 @@ from __future__ import annotations +import contextlib +import json from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path _ALIASES: dict[str, str] = { "s": "search", @@ -98,3 +102,123 @@ 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] = [] + 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 + 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) + record = { + "timestamp": datetime.now(UTC).isoformat(), + "command": command, + "success": success, + } + 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) + with contextlib.suppress(OSError): + 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..84beaa3 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,12 +217,19 @@ 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, + ) + 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, ) @@ -298,8 +363,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 +458,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..767f323 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 @@ -25,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 = { @@ -50,11 +50,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..2786eda 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_viewhighlighted(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 @@ -209,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)] diff --git a/src/kairos/tui/widgets/header_line.py b/src/kairos/tui/widgets/header_line.py index 80ae2e0..da8eba8 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/src/kairos/tui/widgets/workspace_pane.py b/src/kairos/tui/widgets/workspace_pane.py index a0cf880..33f4b58 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 @@ -180,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) @@ -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: diff --git a/tests/tui/test_app.py b/tests/tui/test_app.py index 63e19f6..9b8369e 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)], @@ -366,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) diff --git a/tests/tui/test_command_history.py b/tests/tui/test_command_history.py new file mode 100644 index 0000000..6373b01 --- /dev/null +++ b/tests/tui/test_command_history.py @@ -0,0 +1,133 @@ +"""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_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") == "" + 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_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") + 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_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 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..7ea7ecc --- /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..632f8cd --- /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