tui

Terminal User Interface for Snakemake workflow monitoring (Textual).

This package provides a Textual-based TUI for monitoring Snakemake workflows. The main entry point is SnakeseeApp, which is also exported as WorkflowMonitorTUI for backward compatibility with existing callers.

Module structure: - app.py: SnakeseeApp (the main Textual application) and bindings/reactives - data_source.py: WorkflowDataSource (pure-data layer: polling, estimator, event/log readers, filter/sort helpers, log tail, tool-progress cache) - renderables.py: Rich renderables (header, progress, summary, help, easter egg) - tables.py: DataTable row builders and sort helpers - screens.py: Modal screens (HelpScreen, EasterEggScreen, JobLogScreen) - accessibility.py: Visual encoding configs for colorblind users

Classes

AccessibilityConfig dataclass

Visual encoding configuration for progress bar rendering.

Controls which characters are used for each segment of the progress bar and whether the legend is always displayed.

Attributes:

Name	Type	Description
succeeded	BarStyle	Style for completed/succeeded jobs.
failed	BarStyle	Style for failed jobs.
running	BarStyle	Style for currently running jobs.
remaining	BarStyle	Style for remaining/pending jobs.
incomplete	BarStyle	Style for incomplete jobs (workflow interrupted).
show_legend	bool	If True, always show the legend (not just on failure).

Source code in snakesee/tui/accessibility.py

@dataclass(frozen=True, slots=True)
class AccessibilityConfig:
    """Visual encoding configuration for progress bar rendering.

    Controls which characters are used for each segment of the progress bar
    and whether the legend is always displayed.

    Attributes:
        succeeded: Style for completed/succeeded jobs.
        failed: Style for failed jobs.
        running: Style for currently running jobs.
        remaining: Style for remaining/pending jobs.
        incomplete: Style for incomplete jobs (workflow interrupted).
        show_legend: If True, always show the legend (not just on failure).
    """

    succeeded: BarStyle
    failed: BarStyle
    running: BarStyle
    remaining: BarStyle
    incomplete: BarStyle
    show_legend: bool

LayoutMode

Bases: Enum

Available TUI layout modes.

Source code in snakesee/tui/app.py

class LayoutMode(Enum):
    """Available TUI layout modes."""

    FULL = "full"
    COMPACT = "compact"
    MINIMAL = "minimal"

SnakeseeApp

Bases: App[None]

Textual application for monitoring Snakemake workflows.

Source code in snakesee/tui/app.py

class SnakeseeApp(App[None]):
    """Textual application for monitoring Snakemake workflows."""

    CSS_PATH = "app.tcss"
    BINDINGS: ClassVar[list[BindingType]] = [
        Binding("q,ctrl+c", "quit", "Quit"),
        Binding("tab", "cycle_layout", "Layout", priority=True),
        Binding("s", "cycle_sort_forward", "Sort →", show=False),
        Binding("S", "cycle_sort_back", "Sort ←", show=False),
        Binding("1", "sort_column(0)", show=False),
        Binding("2", "sort_column(1)", show=False),
        Binding("3", "sort_column(2)", show=False),
        Binding("4", "sort_column(3)", show=False),
        Binding("question_mark", "show_help", "Help"),
        Binding("f", "easter_pending", show=False),
        Binding("g", "easter_complete", show=False),
        Binding("slash", "open_filter", "Filter"),
        Binding("n", "next_match", show=False),
        Binding("N", "prev_match", show=False),
        Binding("escape", "clear_filter", "Clear filter"),
        Binding("p", "toggle_pause", "Pause"),
        Binding("e", "toggle_estimation", "Estimation"),
        Binding("w", "toggle_wildcard", "Wildcard"),
        Binding("a", "toggle_accessibility", "Accessibility"),
        Binding("r", "force_refresh", "Refresh"),
        Binding("ctrl+r", "hard_refresh", show=False),
        Binding("plus,equal", "rate_inc(0.5)", show=False),
        Binding("minus", "rate_dec(0.5)", show=False),
        Binding("greater_than_sign,full_stop", "rate_inc(5.0)", show=False),
        Binding("less_than_sign,comma", "rate_dec(5.0)", show=False),
        Binding("0", "rate_reset", show=False),
        Binding("G", "rate_min", show=False),
        Binding("left_square_bracket", "log_older(1)", show=False),
        Binding("right_square_bracket", "log_newer(1)", show=False),
        Binding("left_curly_bracket", "log_older(5)", show=False),
        Binding("right_curly_bracket", "log_newer(5)", show=False),
    ]

    paused: reactive[bool] = reactive(False)
    layout_mode: reactive[LayoutMode] = reactive(LayoutMode.FULL)
    sort_table: reactive[SortTable | None] = reactive(None, init=False)
    sort_column: reactive[int] = reactive(0, init=False)
    sort_ascending: reactive[bool] = reactive(True, init=False)
    filter_text: reactive[str | None] = reactive(None, init=False)
    accessibility_mode: reactive[bool] = reactive(False, init=False)
    refresh_rate: reactive[float] = reactive(DEFAULT_REFRESH_RATE, init=False)
    current_log_index: reactive[int] = reactive(0, init=False)

    _easter_timer: Timer | None = None
    _easter_pending: bool = False
    _refresh_timer: Timer | None = None
    _last_poll: tuple[WorkflowProgress, TimeEstimate | None] | None = None

    @property
    def last_poll(self) -> tuple[WorkflowProgress, TimeEstimate | None] | None:
        """The most recent (progress, estimate) snapshot taken by the refresh cycle.

        Read-only accessor for external tooling (e.g. the docs screenshot
        generator) so it does not have to reach into the private attribute.
        Returns None until the first refresh has polled the data source.
        """
        return self._last_poll

    def action_cycle_layout(self) -> None:
        """Cycle to the next layout mode."""
        modes = list(LayoutMode)
        idx = modes.index(self.layout_mode)
        self.layout_mode = modes[(idx + 1) % len(modes)]

    def watch_layout_mode(self, old: LayoutMode, new: LayoutMode) -> None:
        """Swap the CSS class on the root when the layout mode changes."""
        for mode in LayoutMode:
            self.remove_class(f"-{mode.value}")
        self.add_class(f"-{new.value}")

    def action_cycle_sort_forward(self) -> None:
        """Cycle the sort target one step forward (None → running → … → stats → None)."""
        self._cycle_sort(direction=1)

    def action_cycle_sort_back(self) -> None:
        """Cycle the sort target one step backward (None → stats → … → running → None)."""
        self._cycle_sort(direction=-1)

    def _cycle_sort(self, direction: int) -> None:
        """Advance the sort target by ``direction`` steps and refresh once.

        With no per-attribute watchers on the sort reactives, plain assignment is
        just storage; we explicitly call ``_refresh_panels`` once after all three
        settle so a single keystroke triggers exactly one redraw.
        """
        i = _SORT_CYCLE.index(self.sort_table)
        self.sort_table = _SORT_CYCLE[(i + direction) % len(_SORT_CYCLE)]
        self.sort_column = 0
        self.sort_ascending = True
        self._refresh_panels(ignore_pause=True)

    def action_sort_column(self, col: int) -> None:
        """Set the sort column for the active sort target, or toggle direction if same column.

        Columns are 0-indexed.  Each table enforces its own maximum:
        running and stats support columns 0-3; completions 0-2; pending 0-1.

        Args:
            col: Zero-based column index to sort by.
        """
        if self.sort_table is None:
            return
        if col >= _SORT_MAX_COLS[self.sort_table]:
            return
        if col == self.sort_column:
            self.sort_ascending = not self.sort_ascending
        else:
            self.sort_column = col
            self.sort_ascending = True
        self._refresh_panels(ignore_pause=True)

    def action_show_help(self) -> None:
        """Push the modal HelpScreen overlay."""
        self.push_screen(HelpScreen())

    def action_easter_pending(self) -> None:
        """Start (or restart) the 2-second window for completing the f-then-g easter egg."""
        self._easter_pending = True
        if self._easter_timer is not None:
            self._easter_timer.stop()
        self._easter_timer = self.set_timer(2.0, self._clear_easter)

    def _clear_easter(self) -> None:
        """Reset the easter-egg pending state when the 2-second window elapses."""
        self._easter_pending = False
        self._easter_timer = None

    def action_easter_complete(self) -> None:
        """Push the EasterEggScreen if the f-then-g chord finished within the window."""
        if self._easter_pending:
            self._easter_pending = False
            if self._easter_timer is not None:
                self._easter_timer.stop()
                self._easter_timer = None
            self.push_screen(EasterEggScreen())

    def action_open_filter(self) -> None:
        """Reveal the filter Input widget and focus it for keyboard entry."""
        f = self.query_one("#filter", Input)
        f.can_focus = True
        f.add_class("-active")
        f.focus()

    def on_input_submitted(self, event: Input.Submitted) -> None:
        """Apply the submitted filter text and hide the Input again."""
        if event.input.id != "filter":
            return
        self.filter_text = event.value or None
        event.input.remove_class("-active")
        event.input.value = ""
        self.set_focus(None)
        event.input.can_focus = False

    def action_clear_filter(self) -> None:
        """Hide the filter Input if focused, otherwise clear the filter and return to latest log."""
        focused = self.focused
        if isinstance(focused, Input) and focused.id == "filter":
            focused.remove_class("-active")
            focused.value = ""
            self.set_focus(None)
            focused.can_focus = False
            return
        self.filter_text = None
        self.current_log_index = 0

    def watch_filter_text(self, old: str | None, new: str | None) -> None:
        """Re-populate all panels when the filter text changes.

        Filter changes only come from user keystrokes, so the redraw bypasses
        the pause guard.
        """
        self._refresh_panels(ignore_pause=True)

    def action_next_match(self) -> None:
        """Move the cursor to the next row whose Rule column matches the filter."""
        self._jump_match(direction=1)

    def action_prev_match(self) -> None:
        """Move the cursor to the previous row whose Rule column matches the filter."""
        self._jump_match(direction=-1)

    def action_toggle_pause(self) -> None:
        """Toggle the paused state of the monitor and repaint immediately.

        The repaint bypasses the pause guard so the header's PAUSED indicator
        updates right away instead of waiting for the next interval tick.
        """
        self.paused = not self.paused
        self._refresh_panels(ignore_pause=True)

    def action_toggle_estimation(self) -> None:
        """Toggle time estimation and re-initialize the estimator in a worker thread."""
        self._data.use_estimation = not self._data.use_estimation
        self.run_worker(self._reinit_estimator, thread=True, exclusive=True)

    def action_toggle_wildcard(self) -> None:
        """Toggle wildcard conditioning and re-initialize the estimator in a worker thread."""
        self._data.use_wildcard_conditioning = not self._data.use_wildcard_conditioning
        self.run_worker(self._reinit_estimator, thread=True, exclusive=True)

    def action_toggle_accessibility(self) -> None:
        """Toggle accessibility mode, updating the accessibility config and refreshing panels.

        Toggling off restores the constructor-supplied config (which may be a
        custom override), not necessarily ``DEFAULT_CONFIG``.
        """
        self.accessibility_mode = not self.accessibility_mode
        self._accessibility_config = (
            ACCESSIBLE_CONFIG if self.accessibility_mode else self._base_accessibility_config
        )
        self._refresh_panels(ignore_pause=True)

    def action_force_refresh(self) -> None:
        """Force an immediate panel refresh, even when paused."""
        self._refresh_panels(ignore_pause=True)

    def action_hard_refresh(self) -> None:
        """Re-initialize the estimator in a worker thread and refresh panels."""
        self.run_worker(self._reinit_estimator, thread=True, exclusive=True)

    def watch_refresh_rate(self, old: float, new: float) -> None:
        """Restart the polling timer when the refresh rate changes."""
        if self._refresh_timer is not None:
            self._refresh_timer.stop()
        self._refresh_timer = self.set_interval(new, self._refresh_panels)
        self._data.refresh_rate = new
        self._data.update_cache_ttl()

    def action_rate_inc(self, delta: float) -> None:
        """Increase the refresh rate by ``delta`` seconds, clamped to MAX_REFRESH_RATE."""
        self.refresh_rate = min(MAX_REFRESH_RATE, self.refresh_rate + delta)

    def action_rate_dec(self, delta: float) -> None:
        """Decrease the refresh rate by ``delta`` seconds, clamped to MIN_REFRESH_RATE."""
        self.refresh_rate = max(MIN_REFRESH_RATE, self.refresh_rate - delta)

    def action_rate_reset(self) -> None:
        """Reset the refresh rate to the default value."""
        self.refresh_rate = DEFAULT_REFRESH_RATE

    def action_rate_min(self) -> None:
        """Set the refresh rate to the minimum value."""
        self.refresh_rate = MIN_REFRESH_RATE

    def watch_current_log_index(self, old: int, new: int) -> None:
        """Sync the log index to the data source and refresh panels when it changes.

        Log navigation only comes from user keystrokes, so the redraw bypasses
        the pause guard.
        """
        self._data.current_log_index = new
        self._refresh_panels(ignore_pause=True)

    def action_log_older(self, step: int) -> None:
        """Navigate to an older log file by ``step`` entries.

        Args:
            step: Number of log entries to step backward (toward older logs).
        """
        self._data.refresh_log_list()
        max_idx = self._data.available_log_count - 1
        if max_idx < 0:
            self.current_log_index = 0
            return
        self.current_log_index = min(max_idx, self.current_log_index + step)

    def action_log_newer(self, step: int) -> None:
        """Navigate to a newer log file by ``step`` entries.

        Args:
            step: Number of log entries to step forward (toward newer logs).
        """
        self.current_log_index = max(0, self.current_log_index - step)

    def _reinit_estimator(self) -> None:
        """Re-initialize the estimator (runs on a worker thread) then refresh panels.

        Loads silently: the Textual UI is already on screen, so the startup Rich
        progress spinner would corrupt the display.

        The refresh bypasses the pause guard: every path here is an explicit user
        action (Ctrl+R, estimation/wildcard toggles), so the result should render
        even while auto-refresh is paused.
        """
        self._data.init_estimator(show_progress=False)
        self.call_from_thread(lambda: self._refresh_panels(ignore_pause=True))

    def _sort_table_name(self) -> SortTableName | None:
        """Return the current sort target as the Literal alias the data source expects."""
        return self.sort_table.value if self.sort_table is not None else None

    def _jump_match(self, direction: int) -> None:
        """Jump the focused (or running) DataTable's cursor to the next/prev match.

        The filter matches against the table's Rule column, whose index varies by
        table (see ``_RULE_COLUMN_BY_TABLE``). Tables with no Rule column (e.g.
        ``#incomplete``) are skipped so a global n/N press can neither raise
        ``IndexError`` nor match against the wrong column.

        Args:
            direction: 1 to step forward, -1 to step backward.
        """
        if not self.filter_text:
            return
        focused = self.focused
        table = focused if isinstance(focused, DataTable) else self.query_one("#running", DataTable)
        rule_col = _RULE_COLUMN_BY_TABLE.get(table.id or "")
        if rule_col is None:
            return
        n = table.row_count
        if n == 0:
            return
        needle = self.filter_text.lower()
        start = (table.cursor_row + direction) % n
        i = start
        for _ in range(n):
            row = table.get_row_at(i)
            if rule_col < len(row) and needle in str(row[rule_col]).lower():
                table.move_cursor(row=i)
                return
            i = (i + direction) % n

    def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None:
        """Open the JobLogScreen for the selected job in the running/completions tables."""
        table_id = event.data_table.id
        if table_id not in {SortTable.RUNNING.value, SortTable.COMPLETIONS.value}:
            return
        # Reuse the latest poll snapshot — populated by _refresh_panels at most
        # `refresh_rate` seconds ago, which is also the data the table was rendered from.
        if self._last_poll is None:
            self._last_poll = self._data.poll_state()
        progress, _ = self._last_poll
        if table_id == SortTable.RUNNING.value:
            jobs = self._data.get_running_jobs_list(
                progress,
                filter_text=self.filter_text,
                sort_table=self._sort_table_name(),
                sort_column=self.sort_column,
                sort_ascending=self.sort_ascending,
            )
        else:  # completions
            jobs, _ = self._data.get_completions_list(
                progress,
                filter_text=self.filter_text,
                sort_table=self._sort_table_name(),
                sort_column=self.sort_column,
                sort_ascending=self.sort_ascending,
            )
        if event.cursor_row >= len(jobs):
            return
        job = jobs[event.cursor_row]

        # Remote jobs (e.g. AWS Batch) may have no local log file but still carry
        # an external id + console/CloudWatch links worth showing.
        from snakesee.tui.renderables import make_remote_job_info

        header_lines = make_remote_job_info(job)

        log_path = job.log_file
        if log_path is None:
            if not header_lines:
                return
            # No local log, but remote info is available — show just that.
            self.push_screen(JobLogScreen(None, [], header_lines=header_lines))
            return
        lines = self._data.read_log_tail(log_path, max_lines=500)
        self.push_screen(JobLogScreen(log_path, lines, header_lines=header_lines))

    def __init__(
        self,
        workflow_dir: Path,
        refresh_rate: float = DEFAULT_REFRESH_RATE,
        use_estimation: bool = True,
        profile_path: Path | None = None,
        use_wildcard_conditioning: bool = True,
        weighting_strategy: WeightingStrategy = "index",
        half_life_logs: int = 10,
        half_life_days: float = 7.0,
        accessibility_config: AccessibilityConfig | None = None,
    ) -> None:
        """Initialize the SnakeseeApp.

        Args:
            workflow_dir: Path to workflow directory containing ``.snakemake/``.
            refresh_rate: Refresh interval in seconds.
            use_estimation: Whether to enable time estimation.
            profile_path: Optional path to a timing profile for bootstrapping estimates.
            use_wildcard_conditioning: Whether to enable wildcard-conditioned estimates.
            weighting_strategy: Strategy for weighting historical data ("index" or "time").
            half_life_logs: Half-life in run count for index-based weighting.
            half_life_days: Half-life in days for time-based weighting.
            accessibility_config: Optional accessibility configuration override.
        """
        super().__init__()
        self._data = WorkflowDataSource(
            workflow_dir=workflow_dir,
            refresh_rate=refresh_rate,
            use_estimation=use_estimation,
            profile_path=profile_path,
            use_wildcard_conditioning=use_wildcard_conditioning,
            weighting_strategy=weighting_strategy,
            half_life_logs=half_life_logs,
            half_life_days=half_life_days,
        )
        # Keep the constructor-supplied config around so toggling accessibility
        # off restores it rather than falling back to DEFAULT_CONFIG.
        self._base_accessibility_config = accessibility_config or DEFAULT_CONFIG
        self._accessibility_config = self._base_accessibility_config
        # Resolve the workflow path once; the header truncates it per frame, so the
        # per-render cost stays a string slice rather than a filesystem resolve().
        self._resolved_workflow_dir = str(workflow_dir.resolve())
        # Seed the reactive without firing watch_refresh_rate. A plain assignment would
        # invoke the watcher (whenever refresh_rate differs from the reactive default),
        # which calls set_interval() before run() has started the event loop, raising
        # "RuntimeError: no running event loop". on_mount starts the timer once the loop
        # is running. The ignore works around Textual's stubs typing class-level reactive
        # access as the value type (float) rather than Reactive[float].
        self.set_reactive(SnakeseeApp.refresh_rate, refresh_rate)  # type: ignore[arg-type]
        # Whether the (lazily-added) Cost columns have been added to the
        # completions / stats tables. Once True they stay — the column and flag
        # must always move together to avoid an add_row cell-count mismatch.
        self._completions_cost_col: bool = False
        self._stats_cost_col: bool = False

    def compose(self) -> ComposeResult:
        """Compose the widget tree (header / progress / six tables / summary / footer)."""
        yield Static(id="header")
        yield Static(id="progress")
        with Container(id="body"):
            with Horizontal(id="left"):
                yield DataTable(id="running")
                yield DataTable(id="completions")
            with Horizontal(id="right"):
                yield DataTable(id="pending")
                yield DataTable(id="failed")
                yield DataTable(id="incomplete")
                yield DataTable(id="stats")
        yield Static(id="summary")
        filter_input = Input(placeholder="filter rules…", id="filter")
        filter_input.can_focus = False
        yield filter_input
        yield Footer()

    def on_mount(self) -> None:
        """Configure tables, populate panels, and start the refresh timer."""
        running = self.query_one("#running", DataTable)
        running.add_columns("#", "Rule", "Thr", "Started", "Elapsed", "Progress", "ETA")
        running.cursor_type = "row"
        completions = self.query_one("#completions", DataTable)
        completions.add_columns("#", "Rule", "Thr", "Duration", "Completed")
        completions.cursor_type = "row"
        self.query_one("#pending", DataTable).add_columns("Rule", "Est. Count")
        self.query_one("#failed", DataTable).add_columns("#", "Rule", "Job ID")
        self.query_one("#incomplete", DataTable).add_columns("Output File")
        self.query_one("#stats", DataTable).add_columns("Rule", "Thr", "Count", "Avg", "Std Dev")
        self._refresh_panels()
        # Start the polling timer now that the event loop is running. __init__ seeds
        # refresh_rate via set_reactive (no watcher), so this is the only timer created.
        self._refresh_timer = self.set_interval(self.refresh_rate, self._refresh_panels)
        self.add_class(f"-{self.layout_mode.value}")

    def _refresh_panels(self, ignore_pause: bool = False) -> None:
        """Poll the data source and update header/progress/summary/tables widgets.

        Args:
            ignore_pause: When True, refresh even if ``paused`` is set. Used by every
                explicit user-triggered redraw (force refresh, pause toggle, sort,
                filter, log nav, accessibility, estimator re-init) — pausing gates
                only the interval timer's automatic polling.
        """
        if self.paused and not ignore_pause:
            return
        progress, estimate = self._data.poll_state()
        self._last_poll = (progress, estimate)
        self.query_one("#header", Static).update(
            make_header(
                progress,
                self._resolved_workflow_dir,
                self.paused,
                self._data.event_reader,
                max_path_len=max(20, self.size.width - 80),
            )
        )
        self.query_one("#progress", Static).update(
            make_progress_panel(
                progress,
                estimate,
                self._data.use_estimation,
                self._accessibility_config,
            )
        )
        self.query_one("#summary", Static).update(make_summary_footer(progress))

        self._populate_running(progress)
        self._populate_completions(progress)
        self._populate_pending(progress)
        self._populate_failed(progress)
        self._populate_incomplete(progress)
        self._populate_stats()

    def _populate_running(self, progress: WorkflowProgress) -> None:
        """Populate the running-jobs table from the current workflow progress."""
        table = self.query_one("#running", DataTable)
        table.clear()
        jobs = self._data.get_running_jobs_list(
            progress,
            filter_text=self.filter_text,
            sort_table=self._sort_table_name(),
            sort_column=self.sort_column,
            sort_ascending=self.sort_ascending,
        )
        rows = running_rows(self._data.build_running_job_data(jobs))
        for idx, row in enumerate(rows):
            job = row.job
            elapsed_str = (
                format_duration(row.elapsed_seconds) if row.elapsed_seconds is not None else "?"
            )
            remaining_str = (
                f"~{format_duration(row.remaining_seconds)}"
                if row.remaining_seconds is not None
                else "?"
            )
            started_str = "?"
            if job.start_time is not None:
                started_str = datetime.fromtimestamp(job.start_time).strftime("%H:%M:%S")

            progress_str = "-"
            if row.tool_progress is not None:
                if row.tool_progress.percent_complete is not None:
                    progress_str = row.tool_progress.percent_str
                else:
                    progress_str = f"{row.tool_progress.items_processed:,} {row.tool_progress.unit}"

            threads_str = str(job.threads) if job.threads is not None else "-"
            job_id_str = str(job.job_id) if job.job_id else str(idx + 1)
            table.add_row(
                job_id_str,
                job.rule,
                threads_str,
                started_str,
                elapsed_str,
                progress_str,
                remaining_str,
            )

    def _populate_completions(self, progress: WorkflowProgress) -> None:
        """Populate the recent-completions table from the current workflow progress."""
        table = self.query_one("#completions", DataTable)
        table.clear()
        jobs, failed_job_ids = self._data.get_completions_list(
            progress,
            filter_text=self.filter_text,
            sort_table=self._sort_table_name(),
            sort_column=self.sort_column,
            sort_ascending=self.sort_ascending,
        )
        # Add a Cost column the first time any estimated cost is available, so
        # runs without cost estimation never get a blank column.
        if progress.total_cost_estimate is not None and not self._completions_cost_col:
            table.add_column("Cost", key="cost")
            self._completions_cost_col = True

        rows = completion_rows(jobs, failed_job_ids)
        for idx, row in enumerate(rows):
            job = row.job
            duration_str = format_duration(job.duration) if job.duration is not None else "?"
            threads_str = str(job.threads) if job.threads is not None else "-"
            completed_str = "?"
            if job.end_time is not None:
                completed_str = datetime.fromtimestamp(job.end_time).strftime("%H:%M:%S")
            job_id_str = str(job.job_id) if job.job_id else str(idx + 1)
            cells = [job_id_str, job.rule, threads_str, duration_str, completed_str]
            if self._completions_cost_col:
                cells.append(
                    format_cost(job.cost_estimate) if job.cost_estimate is not None else "-"
                )
            table.add_row(*cells)

    def _populate_pending(self, progress: WorkflowProgress) -> None:
        """Populate the pending-jobs table using inferred per-rule pending counts."""
        table = self.query_one("#pending", DataTable)
        table.clear()
        pending_rules = self._data.get_inferred_pending_rules(progress)
        if not pending_rules:
            return
        rows = pending_rows(pending_rules)
        if self.sort_table == SortTable.PENDING:
            rows = sort_rows(rows, self.sort_column, self.sort_ascending)
        for row in rows:
            table.add_row(row.rule, str(row.job_count))

    def _populate_failed(self, progress: WorkflowProgress) -> None:
        """Populate the failed-jobs table from ``progress.failed_jobs_list``."""
        table = self.query_one("#failed", DataTable)
        table.clear()
        rows = failed_rows(progress)
        for idx, row in enumerate(rows):
            job = row.job
            job_id_str = job.job_id if job.job_id else "-"
            table.add_row(str(idx + 1), job.rule, job_id_str)

    def _populate_incomplete(self, progress: WorkflowProgress) -> None:
        """Populate the incomplete-jobs table from ``progress.incomplete_jobs_list``."""
        table = self.query_one("#incomplete", DataTable)
        table.clear()
        for row in incomplete_rows(progress):
            table.add_row(row.display_path)

    def _populate_stats(self) -> None:
        """Populate the rule-statistics table from the data source's filtered stats."""
        table = self.query_one("#stats", DataTable)
        table.clear()
        if not self._data.use_estimation:
            return
        stats_list = self._data.get_filtered_stats()
        if not stats_list:
            return
        # Default ordering: most-frequently-run rules first.
        stats_list = sorted(stats_list, key=lambda s: s.count, reverse=True)
        rows = stats_rows(stats_list, self._data.thread_stats_dict())
        if self.sort_table == SortTable.STATS:
            rows = sort_stats_rows(rows, self.sort_column, self.sort_ascending)

        # Per-rule estimated cost: add a Cost column once any cost data exists.
        # Sourced from the live registry (the stats panel is itself registry-backed,
        # so this stays consistent with the rest of the stats frame).
        cost_by_rule = self._data.cost_by_rule()
        if cost_by_rule and not self._stats_cost_col:
            table.add_column("Cost", key="cost")
            self._stats_cost_col = True

        for row in rows:
            cells = [
                row.rule_display,
                row.threads,
                str(row.stats.count),
                format_duration(row.stats.mean_duration),
                format_duration(row.stats.std_dev) if row.stats.std_dev > 0 else "-",
            ]
            if self._stats_cost_col:
                rule_cost = cost_by_rule.get(row.stats.rule)
                cells.append(format_cost(rule_cost) if rule_cost is not None else "-")
            table.add_row(*cells)

Attributes

last_poll property

last_poll: tuple[WorkflowProgress, TimeEstimate | None] | None

The most recent (progress, estimate) snapshot taken by the refresh cycle.

Read-only accessor for external tooling (e.g. the docs screenshot generator) so it does not have to reach into the private attribute. Returns None until the first refresh has polled the data source.

Methods:

__init__

__init__(workflow_dir: Path, refresh_rate: float = DEFAULT_REFRESH_RATE, use_estimation: bool = True, profile_path: Path | None = None, use_wildcard_conditioning: bool = True, weighting_strategy: WeightingStrategy = 'index', half_life_logs: int = 10, half_life_days: float = 7.0, accessibility_config: AccessibilityConfig | None = None) -> None

Initialize the SnakeseeApp.

Parameters:

Name	Type	Description	Default
workflow_dir	Path	Path to workflow directory containing .snakemake/.	required
refresh_rate	float	Refresh interval in seconds.	DEFAULT_REFRESH_RATE
use_estimation	bool	Whether to enable time estimation.	True
profile_path	Path | None	Optional path to a timing profile for bootstrapping estimates.	None
use_wildcard_conditioning	bool	Whether to enable wildcard-conditioned estimates.	True
weighting_strategy	WeightingStrategy	Strategy for weighting historical data ("index" or "time").	'index'
half_life_logs	int	Half-life in run count for index-based weighting.	10
half_life_days	float	Half-life in days for time-based weighting.	7.0
accessibility_config	AccessibilityConfig | None	Optional accessibility configuration override.	None

Source code in snakesee/tui/app.py

def __init__(
    self,
    workflow_dir: Path,
    refresh_rate: float = DEFAULT_REFRESH_RATE,
    use_estimation: bool = True,
    profile_path: Path | None = None,
    use_wildcard_conditioning: bool = True,
    weighting_strategy: WeightingStrategy = "index",
    half_life_logs: int = 10,
    half_life_days: float = 7.0,
    accessibility_config: AccessibilityConfig | None = None,
) -> None:
    """Initialize the SnakeseeApp.

    Args:
        workflow_dir: Path to workflow directory containing ``.snakemake/``.
        refresh_rate: Refresh interval in seconds.
        use_estimation: Whether to enable time estimation.
        profile_path: Optional path to a timing profile for bootstrapping estimates.
        use_wildcard_conditioning: Whether to enable wildcard-conditioned estimates.
        weighting_strategy: Strategy for weighting historical data ("index" or "time").
        half_life_logs: Half-life in run count for index-based weighting.
        half_life_days: Half-life in days for time-based weighting.
        accessibility_config: Optional accessibility configuration override.
    """
    super().__init__()
    self._data = WorkflowDataSource(
        workflow_dir=workflow_dir,
        refresh_rate=refresh_rate,
        use_estimation=use_estimation,
        profile_path=profile_path,
        use_wildcard_conditioning=use_wildcard_conditioning,
        weighting_strategy=weighting_strategy,
        half_life_logs=half_life_logs,
        half_life_days=half_life_days,
    )
    # Keep the constructor-supplied config around so toggling accessibility
    # off restores it rather than falling back to DEFAULT_CONFIG.
    self._base_accessibility_config = accessibility_config or DEFAULT_CONFIG
    self._accessibility_config = self._base_accessibility_config
    # Resolve the workflow path once; the header truncates it per frame, so the
    # per-render cost stays a string slice rather than a filesystem resolve().
    self._resolved_workflow_dir = str(workflow_dir.resolve())
    # Seed the reactive without firing watch_refresh_rate. A plain assignment would
    # invoke the watcher (whenever refresh_rate differs from the reactive default),
    # which calls set_interval() before run() has started the event loop, raising
    # "RuntimeError: no running event loop". on_mount starts the timer once the loop
    # is running. The ignore works around Textual's stubs typing class-level reactive
    # access as the value type (float) rather than Reactive[float].
    self.set_reactive(SnakeseeApp.refresh_rate, refresh_rate)  # type: ignore[arg-type]
    # Whether the (lazily-added) Cost columns have been added to the
    # completions / stats tables. Once True they stay — the column and flag
    # must always move together to avoid an add_row cell-count mismatch.
    self._completions_cost_col: bool = False
    self._stats_cost_col: bool = False

action_clear_filter

action_clear_filter() -> None

Hide the filter Input if focused, otherwise clear the filter and return to latest log.

Source code in snakesee/tui/app.py

def action_clear_filter(self) -> None:
    """Hide the filter Input if focused, otherwise clear the filter and return to latest log."""
    focused = self.focused
    if isinstance(focused, Input) and focused.id == "filter":
        focused.remove_class("-active")
        focused.value = ""
        self.set_focus(None)
        focused.can_focus = False
        return
    self.filter_text = None
    self.current_log_index = 0

action_cycle_layout

action_cycle_layout() -> None

Cycle to the next layout mode.

Source code in snakesee/tui/app.py

def action_cycle_layout(self) -> None:
    """Cycle to the next layout mode."""
    modes = list(LayoutMode)
    idx = modes.index(self.layout_mode)
    self.layout_mode = modes[(idx + 1) % len(modes)]

action_cycle_sort_back

action_cycle_sort_back() -> None

Cycle the sort target one step backward (None → stats → … → running → None).

Source code in snakesee/tui/app.py

def action_cycle_sort_back(self) -> None:
    """Cycle the sort target one step backward (None → stats → … → running → None)."""
    self._cycle_sort(direction=-1)

action_cycle_sort_forward

action_cycle_sort_forward() -> None

Cycle the sort target one step forward (None → running → … → stats → None).

Source code in snakesee/tui/app.py

def action_cycle_sort_forward(self) -> None:
    """Cycle the sort target one step forward (None → running → … → stats → None)."""
    self._cycle_sort(direction=1)

action_easter_complete

action_easter_complete() -> None

Push the EasterEggScreen if the f-then-g chord finished within the window.

Source code in snakesee/tui/app.py

def action_easter_complete(self) -> None:
    """Push the EasterEggScreen if the f-then-g chord finished within the window."""
    if self._easter_pending:
        self._easter_pending = False
        if self._easter_timer is not None:
            self._easter_timer.stop()
            self._easter_timer = None
        self.push_screen(EasterEggScreen())

action_easter_pending

action_easter_pending() -> None

Start (or restart) the 2-second window for completing the f-then-g easter egg.

Source code in snakesee/tui/app.py

def action_easter_pending(self) -> None:
    """Start (or restart) the 2-second window for completing the f-then-g easter egg."""
    self._easter_pending = True
    if self._easter_timer is not None:
        self._easter_timer.stop()
    self._easter_timer = self.set_timer(2.0, self._clear_easter)

action_force_refresh

action_force_refresh() -> None

Force an immediate panel refresh, even when paused.

Source code in snakesee/tui/app.py

def action_force_refresh(self) -> None:
    """Force an immediate panel refresh, even when paused."""
    self._refresh_panels(ignore_pause=True)

action_hard_refresh

action_hard_refresh() -> None

Re-initialize the estimator in a worker thread and refresh panels.

Source code in snakesee/tui/app.py

def action_hard_refresh(self) -> None:
    """Re-initialize the estimator in a worker thread and refresh panels."""
    self.run_worker(self._reinit_estimator, thread=True, exclusive=True)

action_log_newer

action_log_newer(step: int) -> None

Navigate to a newer log file by step entries.

Parameters:

Name	Type	Description	Default
step	int	Number of log entries to step forward (toward newer logs).	required

Source code in snakesee/tui/app.py

def action_log_newer(self, step: int) -> None:
    """Navigate to a newer log file by ``step`` entries.

    Args:
        step: Number of log entries to step forward (toward newer logs).
    """
    self.current_log_index = max(0, self.current_log_index - step)

action_log_older

action_log_older(step: int) -> None

Navigate to an older log file by step entries.

Parameters:

Name	Type	Description	Default
step	int	Number of log entries to step backward (toward older logs).	required

Source code in snakesee/tui/app.py

def action_log_older(self, step: int) -> None:
    """Navigate to an older log file by ``step`` entries.

    Args:
        step: Number of log entries to step backward (toward older logs).
    """
    self._data.refresh_log_list()
    max_idx = self._data.available_log_count - 1
    if max_idx < 0:
        self.current_log_index = 0
        return
    self.current_log_index = min(max_idx, self.current_log_index + step)

action_next_match

action_next_match() -> None

Move the cursor to the next row whose Rule column matches the filter.

Source code in snakesee/tui/app.py

def action_next_match(self) -> None:
    """Move the cursor to the next row whose Rule column matches the filter."""
    self._jump_match(direction=1)

action_open_filter

action_open_filter() -> None

Reveal the filter Input widget and focus it for keyboard entry.

Source code in snakesee/tui/app.py

def action_open_filter(self) -> None:
    """Reveal the filter Input widget and focus it for keyboard entry."""
    f = self.query_one("#filter", Input)
    f.can_focus = True
    f.add_class("-active")
    f.focus()

action_prev_match

action_prev_match() -> None

Move the cursor to the previous row whose Rule column matches the filter.

Source code in snakesee/tui/app.py

def action_prev_match(self) -> None:
    """Move the cursor to the previous row whose Rule column matches the filter."""
    self._jump_match(direction=-1)

action_rate_dec

action_rate_dec(delta: float) -> None

Decrease the refresh rate by delta seconds, clamped to MIN_REFRESH_RATE.

Source code in snakesee/tui/app.py

def action_rate_dec(self, delta: float) -> None:
    """Decrease the refresh rate by ``delta`` seconds, clamped to MIN_REFRESH_RATE."""
    self.refresh_rate = max(MIN_REFRESH_RATE, self.refresh_rate - delta)

action_rate_inc

action_rate_inc(delta: float) -> None

Increase the refresh rate by delta seconds, clamped to MAX_REFRESH_RATE.

Source code in snakesee/tui/app.py

def action_rate_inc(self, delta: float) -> None:
    """Increase the refresh rate by ``delta`` seconds, clamped to MAX_REFRESH_RATE."""
    self.refresh_rate = min(MAX_REFRESH_RATE, self.refresh_rate + delta)

action_rate_min

action_rate_min() -> None

Set the refresh rate to the minimum value.

Source code in snakesee/tui/app.py

def action_rate_min(self) -> None:
    """Set the refresh rate to the minimum value."""
    self.refresh_rate = MIN_REFRESH_RATE

action_rate_reset

action_rate_reset() -> None

Reset the refresh rate to the default value.

Source code in snakesee/tui/app.py

def action_rate_reset(self) -> None:
    """Reset the refresh rate to the default value."""
    self.refresh_rate = DEFAULT_REFRESH_RATE

action_show_help

action_show_help() -> None

Push the modal HelpScreen overlay.

Source code in snakesee/tui/app.py

def action_show_help(self) -> None:
    """Push the modal HelpScreen overlay."""
    self.push_screen(HelpScreen())

action_sort_column

action_sort_column(col: int) -> None

Set the sort column for the active sort target, or toggle direction if same column.

Columns are 0-indexed. Each table enforces its own maximum: running and stats support columns 0-3; completions 0-2; pending 0-1.

Parameters:

Name	Type	Description	Default
col	int	Zero-based column index to sort by.	required

Source code in snakesee/tui/app.py

def action_sort_column(self, col: int) -> None:
    """Set the sort column for the active sort target, or toggle direction if same column.

    Columns are 0-indexed.  Each table enforces its own maximum:
    running and stats support columns 0-3; completions 0-2; pending 0-1.

    Args:
        col: Zero-based column index to sort by.
    """
    if self.sort_table is None:
        return
    if col >= _SORT_MAX_COLS[self.sort_table]:
        return
    if col == self.sort_column:
        self.sort_ascending = not self.sort_ascending
    else:
        self.sort_column = col
        self.sort_ascending = True
    self._refresh_panels(ignore_pause=True)

action_toggle_accessibility

action_toggle_accessibility() -> None

Toggle accessibility mode, updating the accessibility config and refreshing panels.

Toggling off restores the constructor-supplied config (which may be a custom override), not necessarily DEFAULT_CONFIG.

Source code in snakesee/tui/app.py

def action_toggle_accessibility(self) -> None:
    """Toggle accessibility mode, updating the accessibility config and refreshing panels.

    Toggling off restores the constructor-supplied config (which may be a
    custom override), not necessarily ``DEFAULT_CONFIG``.
    """
    self.accessibility_mode = not self.accessibility_mode
    self._accessibility_config = (
        ACCESSIBLE_CONFIG if self.accessibility_mode else self._base_accessibility_config
    )
    self._refresh_panels(ignore_pause=True)

action_toggle_estimation

action_toggle_estimation() -> None

Toggle time estimation and re-initialize the estimator in a worker thread.

Source code in snakesee/tui/app.py

def action_toggle_estimation(self) -> None:
    """Toggle time estimation and re-initialize the estimator in a worker thread."""
    self._data.use_estimation = not self._data.use_estimation
    self.run_worker(self._reinit_estimator, thread=True, exclusive=True)

action_toggle_pause

action_toggle_pause() -> None

Toggle the paused state of the monitor and repaint immediately.

The repaint bypasses the pause guard so the header's PAUSED indicator updates right away instead of waiting for the next interval tick.

Source code in snakesee/tui/app.py

def action_toggle_pause(self) -> None:
    """Toggle the paused state of the monitor and repaint immediately.

    The repaint bypasses the pause guard so the header's PAUSED indicator
    updates right away instead of waiting for the next interval tick.
    """
    self.paused = not self.paused
    self._refresh_panels(ignore_pause=True)

action_toggle_wildcard

action_toggle_wildcard() -> None

Toggle wildcard conditioning and re-initialize the estimator in a worker thread.

Source code in snakesee/tui/app.py

def action_toggle_wildcard(self) -> None:
    """Toggle wildcard conditioning and re-initialize the estimator in a worker thread."""
    self._data.use_wildcard_conditioning = not self._data.use_wildcard_conditioning
    self.run_worker(self._reinit_estimator, thread=True, exclusive=True)

compose

compose() -> ComposeResult

Compose the widget tree (header / progress / six tables / summary / footer).

Source code in snakesee/tui/app.py

def compose(self) -> ComposeResult:
    """Compose the widget tree (header / progress / six tables / summary / footer)."""
    yield Static(id="header")
    yield Static(id="progress")
    with Container(id="body"):
        with Horizontal(id="left"):
            yield DataTable(id="running")
            yield DataTable(id="completions")
        with Horizontal(id="right"):
            yield DataTable(id="pending")
            yield DataTable(id="failed")
            yield DataTable(id="incomplete")
            yield DataTable(id="stats")
    yield Static(id="summary")
    filter_input = Input(placeholder="filter rules…", id="filter")
    filter_input.can_focus = False
    yield filter_input
    yield Footer()

on_data_table_row_selected

on_data_table_row_selected(event: RowSelected) -> None

Open the JobLogScreen for the selected job in the running/completions tables.

Source code in snakesee/tui/app.py

def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None:
    """Open the JobLogScreen for the selected job in the running/completions tables."""
    table_id = event.data_table.id
    if table_id not in {SortTable.RUNNING.value, SortTable.COMPLETIONS.value}:
        return
    # Reuse the latest poll snapshot — populated by _refresh_panels at most
    # `refresh_rate` seconds ago, which is also the data the table was rendered from.
    if self._last_poll is None:
        self._last_poll = self._data.poll_state()
    progress, _ = self._last_poll
    if table_id == SortTable.RUNNING.value:
        jobs = self._data.get_running_jobs_list(
            progress,
            filter_text=self.filter_text,
            sort_table=self._sort_table_name(),
            sort_column=self.sort_column,
            sort_ascending=self.sort_ascending,
        )
    else:  # completions
        jobs, _ = self._data.get_completions_list(
            progress,
            filter_text=self.filter_text,
            sort_table=self._sort_table_name(),
            sort_column=self.sort_column,
            sort_ascending=self.sort_ascending,
        )
    if event.cursor_row >= len(jobs):
        return
    job = jobs[event.cursor_row]

    # Remote jobs (e.g. AWS Batch) may have no local log file but still carry
    # an external id + console/CloudWatch links worth showing.
    from snakesee.tui.renderables import make_remote_job_info

    header_lines = make_remote_job_info(job)

    log_path = job.log_file
    if log_path is None:
        if not header_lines:
            return
        # No local log, but remote info is available — show just that.
        self.push_screen(JobLogScreen(None, [], header_lines=header_lines))
        return
    lines = self._data.read_log_tail(log_path, max_lines=500)
    self.push_screen(JobLogScreen(log_path, lines, header_lines=header_lines))

on_input_submitted

on_input_submitted(event: Submitted) -> None

Apply the submitted filter text and hide the Input again.

Source code in snakesee/tui/app.py

def on_input_submitted(self, event: Input.Submitted) -> None:
    """Apply the submitted filter text and hide the Input again."""
    if event.input.id != "filter":
        return
    self.filter_text = event.value or None
    event.input.remove_class("-active")
    event.input.value = ""
    self.set_focus(None)
    event.input.can_focus = False

on_mount

on_mount() -> None

Configure tables, populate panels, and start the refresh timer.

Source code in snakesee/tui/app.py

def on_mount(self) -> None:
    """Configure tables, populate panels, and start the refresh timer."""
    running = self.query_one("#running", DataTable)
    running.add_columns("#", "Rule", "Thr", "Started", "Elapsed", "Progress", "ETA")
    running.cursor_type = "row"
    completions = self.query_one("#completions", DataTable)
    completions.add_columns("#", "Rule", "Thr", "Duration", "Completed")
    completions.cursor_type = "row"
    self.query_one("#pending", DataTable).add_columns("Rule", "Est. Count")
    self.query_one("#failed", DataTable).add_columns("#", "Rule", "Job ID")
    self.query_one("#incomplete", DataTable).add_columns("Output File")
    self.query_one("#stats", DataTable).add_columns("Rule", "Thr", "Count", "Avg", "Std Dev")
    self._refresh_panels()
    # Start the polling timer now that the event loop is running. __init__ seeds
    # refresh_rate via set_reactive (no watcher), so this is the only timer created.
    self._refresh_timer = self.set_interval(self.refresh_rate, self._refresh_panels)
    self.add_class(f"-{self.layout_mode.value}")

watch_current_log_index

watch_current_log_index(old: int, new: int) -> None

Sync the log index to the data source and refresh panels when it changes.

Log navigation only comes from user keystrokes, so the redraw bypasses the pause guard.

Source code in snakesee/tui/app.py

def watch_current_log_index(self, old: int, new: int) -> None:
    """Sync the log index to the data source and refresh panels when it changes.

    Log navigation only comes from user keystrokes, so the redraw bypasses
    the pause guard.
    """
    self._data.current_log_index = new
    self._refresh_panels(ignore_pause=True)

watch_filter_text

watch_filter_text(old: str | None, new: str | None) -> None

Re-populate all panels when the filter text changes.

Filter changes only come from user keystrokes, so the redraw bypasses the pause guard.

Source code in snakesee/tui/app.py

def watch_filter_text(self, old: str | None, new: str | None) -> None:
    """Re-populate all panels when the filter text changes.

    Filter changes only come from user keystrokes, so the redraw bypasses
    the pause guard.
    """
    self._refresh_panels(ignore_pause=True)

watch_layout_mode

watch_layout_mode(old: LayoutMode, new: LayoutMode) -> None

Swap the CSS class on the root when the layout mode changes.

Source code in snakesee/tui/app.py

def watch_layout_mode(self, old: LayoutMode, new: LayoutMode) -> None:
    """Swap the CSS class on the root when the layout mode changes."""
    for mode in LayoutMode:
        self.remove_class(f"-{mode.value}")
    self.add_class(f"-{new.value}")

watch_refresh_rate

watch_refresh_rate(old: float, new: float) -> None

Restart the polling timer when the refresh rate changes.

Source code in snakesee/tui/app.py

def watch_refresh_rate(self, old: float, new: float) -> None:
    """Restart the polling timer when the refresh rate changes."""
    if self._refresh_timer is not None:
        self._refresh_timer.stop()
    self._refresh_timer = self.set_interval(new, self._refresh_panels)
    self._data.refresh_rate = new
    self._data.update_cache_ttl()

Modules

accessibility

Accessibility configuration for colorblind-friendly rendering.

Provides alternative visual encodings so that progress bar status can be distinguished without relying on color perception alone.

Classes

AccessibilityConfig dataclass

Visual encoding configuration for progress bar rendering.

Controls which characters are used for each segment of the progress bar and whether the legend is always displayed.

Attributes:

Name	Type	Description
succeeded	BarStyle	Style for completed/succeeded jobs.
failed	BarStyle	Style for failed jobs.
running	BarStyle	Style for currently running jobs.
remaining	BarStyle	Style for remaining/pending jobs.
incomplete	BarStyle	Style for incomplete jobs (workflow interrupted).
show_legend	bool	If True, always show the legend (not just on failure).

Source code in snakesee/tui/accessibility.py

@dataclass(frozen=True, slots=True)
class AccessibilityConfig:
    """Visual encoding configuration for progress bar rendering.

    Controls which characters are used for each segment of the progress bar
    and whether the legend is always displayed.

    Attributes:
        succeeded: Style for completed/succeeded jobs.
        failed: Style for failed jobs.
        running: Style for currently running jobs.
        remaining: Style for remaining/pending jobs.
        incomplete: Style for incomplete jobs (workflow interrupted).
        show_legend: If True, always show the legend (not just on failure).
    """

    succeeded: BarStyle
    failed: BarStyle
    running: BarStyle
    remaining: BarStyle
    incomplete: BarStyle
    show_legend: bool

BarStyle dataclass

Character and label for a single progress bar segment.

Attributes:

Name	Type	Description
char	str	The character used to fill the segment.
label	str	Human-readable label for the legend.

Source code in snakesee/tui/accessibility.py

@dataclass(frozen=True, slots=True)
class BarStyle:
    """Character and label for a single progress bar segment.

    Attributes:
        char: The character used to fill the segment.
        label: Human-readable label for the legend.
    """

    char: str
    label: str

app

Textual App for snakesee TUI.

Classes

LayoutMode

Bases: Enum

Available TUI layout modes.

Source code in snakesee/tui/app.py

class LayoutMode(Enum):
    """Available TUI layout modes."""

    FULL = "full"
    COMPACT = "compact"
    MINIMAL = "minimal"

SnakeseeApp

Bases: App[None]

Textual application for monitoring Snakemake workflows.

Source code in snakesee/tui/app.py

class SnakeseeApp(App[None]):
    """Textual application for monitoring Snakemake workflows."""

    CSS_PATH = "app.tcss"
    BINDINGS: ClassVar[list[BindingType]] = [
        Binding("q,ctrl+c", "quit", "Quit"),
        Binding("tab", "cycle_layout", "Layout", priority=True),
        Binding("s", "cycle_sort_forward", "Sort →", show=False),
        Binding("S", "cycle_sort_back", "Sort ←", show=False),
        Binding("1", "sort_column(0)", show=False),
        Binding("2", "sort_column(1)", show=False),
        Binding("3", "sort_column(2)", show=False),
        Binding("4", "sort_column(3)", show=False),
        Binding("question_mark", "show_help", "Help"),
        Binding("f", "easter_pending", show=False),
        Binding("g", "easter_complete", show=False),
        Binding("slash", "open_filter", "Filter"),
        Binding("n", "next_match", show=False),
        Binding("N", "prev_match", show=False),
        Binding("escape", "clear_filter", "Clear filter"),
        Binding("p", "toggle_pause", "Pause"),
        Binding("e", "toggle_estimation", "Estimation"),
        Binding("w", "toggle_wildcard", "Wildcard"),
        Binding("a", "toggle_accessibility", "Accessibility"),
        Binding("r", "force_refresh", "Refresh"),
        Binding("ctrl+r", "hard_refresh", show=False),
        Binding("plus,equal", "rate_inc(0.5)", show=False),
        Binding("minus", "rate_dec(0.5)", show=False),
        Binding("greater_than_sign,full_stop", "rate_inc(5.0)", show=False),
        Binding("less_than_sign,comma", "rate_dec(5.0)", show=False),
        Binding("0", "rate_reset", show=False),
        Binding("G", "rate_min", show=False),
        Binding("left_square_bracket", "log_older(1)", show=False),
        Binding("right_square_bracket", "log_newer(1)", show=False),
        Binding("left_curly_bracket", "log_older(5)", show=False),
        Binding("right_curly_bracket", "log_newer(5)", show=False),
    ]

    paused: reactive[bool] = reactive(False)
    layout_mode: reactive[LayoutMode] = reactive(LayoutMode.FULL)
    sort_table: reactive[SortTable | None] = reactive(None, init=False)
    sort_column: reactive[int] = reactive(0, init=False)
    sort_ascending: reactive[bool] = reactive(True, init=False)
    filter_text: reactive[str | None] = reactive(None, init=False)
    accessibility_mode: reactive[bool] = reactive(False, init=False)
    refresh_rate: reactive[float] = reactive(DEFAULT_REFRESH_RATE, init=False)
    current_log_index: reactive[int] = reactive(0, init=False)

    _easter_timer: Timer | None = None
    _easter_pending: bool = False
    _refresh_timer: Timer | None = None
    _last_poll: tuple[WorkflowProgress, TimeEstimate | None] | None = None

    @property
    def last_poll(self) -> tuple[WorkflowProgress, TimeEstimate | None] | None:
        """The most recent (progress, estimate) snapshot taken by the refresh cycle.

        Read-only accessor for external tooling (e.g. the docs screenshot
        generator) so it does not have to reach into the private attribute.
        Returns None until the first refresh has polled the data source.
        """
        return self._last_poll

    def action_cycle_layout(self) -> None:
        """Cycle to the next layout mode."""
        modes = list(LayoutMode)
        idx = modes.index(self.layout_mode)
        self.layout_mode = modes[(idx + 1) % len(modes)]

    def watch_layout_mode(self, old: LayoutMode, new: LayoutMode) -> None:
        """Swap the CSS class on the root when the layout mode changes."""
        for mode in LayoutMode:
            self.remove_class(f"-{mode.value}")
        self.add_class(f"-{new.value}")

    def action_cycle_sort_forward(self) -> None:
        """Cycle the sort target one step forward (None → running → … → stats → None)."""
        self._cycle_sort(direction=1)

    def action_cycle_sort_back(self) -> None:
        """Cycle the sort target one step backward (None → stats → … → running → None)."""
        self._cycle_sort(direction=-1)

    def _cycle_sort(self, direction: int) -> None:
        """Advance the sort target by ``direction`` steps and refresh once.

        With no per-attribute watchers on the sort reactives, plain assignment is
        just storage; we explicitly call ``_refresh_panels`` once after all three
        settle so a single keystroke triggers exactly one redraw.
        """
        i = _SORT_CYCLE.index(self.sort_table)
        self.sort_table = _SORT_CYCLE[(i + direction) % len(_SORT_CYCLE)]
        self.sort_column = 0
        self.sort_ascending = True
        self._refresh_panels(ignore_pause=True)

    def action_sort_column(self, col: int) -> None:
        """Set the sort column for the active sort target, or toggle direction if same column.

        Columns are 0-indexed.  Each table enforces its own maximum:
        running and stats support columns 0-3; completions 0-2; pending 0-1.

        Args:
            col: Zero-based column index to sort by.
        """
        if self.sort_table is None:
            return
        if col >= _SORT_MAX_COLS[self.sort_table]:
            return
        if col == self.sort_column:
            self.sort_ascending = not self.sort_ascending
        else:
            self.sort_column = col
            self.sort_ascending = True
        self._refresh_panels(ignore_pause=True)

    def action_show_help(self) -> None:
        """Push the modal HelpScreen overlay."""
        self.push_screen(HelpScreen())

    def action_easter_pending(self) -> None:
        """Start (or restart) the 2-second window for completing the f-then-g easter egg."""
        self._easter_pending = True
        if self._easter_timer is not None:
            self._easter_timer.stop()
        self._easter_timer = self.set_timer(2.0, self._clear_easter)

    def _clear_easter(self) -> None:
        """Reset the easter-egg pending state when the 2-second window elapses."""
        self._easter_pending = False
        self._easter_timer = None

    def action_easter_complete(self) -> None:
        """Push the EasterEggScreen if the f-then-g chord finished within the window."""
        if self._easter_pending:
            self._easter_pending = False
            if self._easter_timer is not None:
                self._easter_timer.stop()
                self._easter_timer = None
            self.push_screen(EasterEggScreen())

    def action_open_filter(self) -> None:
        """Reveal the filter Input widget and focus it for keyboard entry."""
        f = self.query_one("#filter", Input)
        f.can_focus = True
        f.add_class("-active")
        f.focus()

    def on_input_submitted(self, event: Input.Submitted) -> None:
        """Apply the submitted filter text and hide the Input again."""
        if event.input.id != "filter":
            return
        self.filter_text = event.value or None
        event.input.remove_class("-active")
        event.input.value = ""
        self.set_focus(None)
        event.input.can_focus = False

    def action_clear_filter(self) -> None:
        """Hide the filter Input if focused, otherwise clear the filter and return to latest log."""
        focused = self.focused
        if isinstance(focused, Input) and focused.id == "filter":
            focused.remove_class("-active")
            focused.value = ""
            self.set_focus(None)
            focused.can_focus = False
            return
        self.filter_text = None
        self.current_log_index = 0

    def watch_filter_text(self, old: str | None, new: str | None) -> None:
        """Re-populate all panels when the filter text changes.

        Filter changes only come from user keystrokes, so the redraw bypasses
        the pause guard.
        """
        self._refresh_panels(ignore_pause=True)

    def action_next_match(self) -> None:
        """Move the cursor to the next row whose Rule column matches the filter."""
        self._jump_match(direction=1)

    def action_prev_match(self) -> None:
        """Move the cursor to the previous row whose Rule column matches the filter."""
        self._jump_match(direction=-1)

    def action_toggle_pause(self) -> None:
        """Toggle the paused state of the monitor and repaint immediately.

        The repaint bypasses the pause guard so the header's PAUSED indicator
        updates right away instead of waiting for the next interval tick.
        """
        self.paused = not self.paused
        self._refresh_panels(ignore_pause=True)

    def action_toggle_estimation(self) -> None:
        """Toggle time estimation and re-initialize the estimator in a worker thread."""
        self._data.use_estimation = not self._data.use_estimation
        self.run_worker(self._reinit_estimator, thread=True, exclusive=True)

    def action_toggle_wildcard(self) -> None:
        """Toggle wildcard conditioning and re-initialize the estimator in a worker thread."""
        self._data.use_wildcard_conditioning = not self._data.use_wildcard_conditioning
        self.run_worker(self._reinit_estimator, thread=True, exclusive=True)

    def action_toggle_accessibility(self) -> None:
        """Toggle accessibility mode, updating the accessibility config and refreshing panels.

        Toggling off restores the constructor-supplied config (which may be a
        custom override), not necessarily ``DEFAULT_CONFIG``.
        """
        self.accessibility_mode = not self.accessibility_mode
        self._accessibility_config = (
            ACCESSIBLE_CONFIG if self.accessibility_mode else self._base_accessibility_config
        )
        self._refresh_panels(ignore_pause=True)

    def action_force_refresh(self) -> None:
        """Force an immediate panel refresh, even when paused."""
        self._refresh_panels(ignore_pause=True)

    def action_hard_refresh(self) -> None:
        """Re-initialize the estimator in a worker thread and refresh panels."""
        self.run_worker(self._reinit_estimator, thread=True, exclusive=True)

    def watch_refresh_rate(self, old: float, new: float) -> None:
        """Restart the polling timer when the refresh rate changes."""
        if self._refresh_timer is not None:
            self._refresh_timer.stop()
        self._refresh_timer = self.set_interval(new, self._refresh_panels)
        self._data.refresh_rate = new
        self._data.update_cache_ttl()

    def action_rate_inc(self, delta: float) -> None:
        """Increase the refresh rate by ``delta`` seconds, clamped to MAX_REFRESH_RATE."""
        self.refresh_rate = min(MAX_REFRESH_RATE, self.refresh_rate + delta)

    def action_rate_dec(self, delta: float) -> None:
        """Decrease the refresh rate by ``delta`` seconds, clamped to MIN_REFRESH_RATE."""
        self.refresh_rate = max(MIN_REFRESH_RATE, self.refresh_rate - delta)

    def action_rate_reset(self) -> None:
        """Reset the refresh rate to the default value."""
        self.refresh_rate = DEFAULT_REFRESH_RATE

    def action_rate_min(self) -> None:
        """Set the refresh rate to the minimum value."""
        self.refresh_rate = MIN_REFRESH_RATE

    def watch_current_log_index(self, old: int, new: int) -> None:
        """Sync the log index to the data source and refresh panels when it changes.

        Log navigation only comes from user keystrokes, so the redraw bypasses
        the pause guard.
        """
        self._data.current_log_index = new
        self._refresh_panels(ignore_pause=True)

    def action_log_older(self, step: int) -> None:
        """Navigate to an older log file by ``step`` entries.

        Args:
            step: Number of log entries to step backward (toward older logs).
        """
        self._data.refresh_log_list()
        max_idx = self._data.available_log_count - 1
        if max_idx < 0:
            self.current_log_index = 0
            return
        self.current_log_index = min(max_idx, self.current_log_index + step)

    def action_log_newer(self, step: int) -> None:
        """Navigate to a newer log file by ``step`` entries.

        Args:
            step: Number of log entries to step forward (toward newer logs).
        """
        self.current_log_index = max(0, self.current_log_index - step)

    def _reinit_estimator(self) -> None:
        """Re-initialize the estimator (runs on a worker thread) then refresh panels.

        Loads silently: the Textual UI is already on screen, so the startup Rich
        progress spinner would corrupt the display.

        The refresh bypasses the pause guard: every path here is an explicit user
        action (Ctrl+R, estimation/wildcard toggles), so the result should render
        even while auto-refresh is paused.
        """
        self._data.init_estimator(show_progress=False)
        self.call_from_thread(lambda: self._refresh_panels(ignore_pause=True))

    def _sort_table_name(self) -> SortTableName | None:
        """Return the current sort target as the Literal alias the data source expects."""
        return self.sort_table.value if self.sort_table is not None else None

    def _jump_match(self, direction: int) -> None:
        """Jump the focused (or running) DataTable's cursor to the next/prev match.

        The filter matches against the table's Rule column, whose index varies by
        table (see ``_RULE_COLUMN_BY_TABLE``). Tables with no Rule column (e.g.
        ``#incomplete``) are skipped so a global n/N press can neither raise
        ``IndexError`` nor match against the wrong column.

        Args:
            direction: 1 to step forward, -1 to step backward.
        """
        if not self.filter_text:
            return
        focused = self.focused
        table = focused if isinstance(focused, DataTable) else self.query_one("#running", DataTable)
        rule_col = _RULE_COLUMN_BY_TABLE.get(table.id or "")
        if rule_col is None:
            return
        n = table.row_count
        if n == 0:
            return
        needle = self.filter_text.lower()
        start = (table.cursor_row + direction) % n
        i = start
        for _ in range(n):
            row = table.get_row_at(i)
            if rule_col < len(row) and needle in str(row[rule_col]).lower():
                table.move_cursor(row=i)
                return
            i = (i + direction) % n

    def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None:
        """Open the JobLogScreen for the selected job in the running/completions tables."""
        table_id = event.data_table.id
        if table_id not in {SortTable.RUNNING.value, SortTable.COMPLETIONS.value}:
            return
        # Reuse the latest poll snapshot — populated by _refresh_panels at most
        # `refresh_rate` seconds ago, which is also the data the table was rendered from.
        if self._last_poll is None:
            self._last_poll = self._data.poll_state()
        progress, _ = self._last_poll
        if table_id == SortTable.RUNNING.value:
            jobs = self._data.get_running_jobs_list(
                progress,
                filter_text=self.filter_text,
                sort_table=self._sort_table_name(),
                sort_column=self.sort_column,
                sort_ascending=self.sort_ascending,
            )
        else:  # completions
            jobs, _ = self._data.get_completions_list(
                progress,
                filter_text=self.filter_text,
                sort_table=self._sort_table_name(),
                sort_column=self.sort_column,
                sort_ascending=self.sort_ascending,
            )
        if event.cursor_row >= len(jobs):
            return
        job = jobs[event.cursor_row]

        # Remote jobs (e.g. AWS Batch) may have no local log file but still carry
        # an external id + console/CloudWatch links worth showing.
        from snakesee.tui.renderables import make_remote_job_info

        header_lines = make_remote_job_info(job)

        log_path = job.log_file
        if log_path is None:
            if not header_lines:
                return
            # No local log, but remote info is available — show just that.
            self.push_screen(JobLogScreen(None, [], header_lines=header_lines))
            return
        lines = self._data.read_log_tail(log_path, max_lines=500)
        self.push_screen(JobLogScreen(log_path, lines, header_lines=header_lines))

    def __init__(
        self,
        workflow_dir: Path,
        refresh_rate: float = DEFAULT_REFRESH_RATE,
        use_estimation: bool = True,
        profile_path: Path | None = None,
        use_wildcard_conditioning: bool = True,
        weighting_strategy: WeightingStrategy = "index",
        half_life_logs: int = 10,
        half_life_days: float = 7.0,
        accessibility_config: AccessibilityConfig | None = None,
    ) -> None:
        """Initialize the SnakeseeApp.

        Args:
            workflow_dir: Path to workflow directory containing ``.snakemake/``.
            refresh_rate: Refresh interval in seconds.
            use_estimation: Whether to enable time estimation.
            profile_path: Optional path to a timing profile for bootstrapping estimates.
            use_wildcard_conditioning: Whether to enable wildcard-conditioned estimates.
            weighting_strategy: Strategy for weighting historical data ("index" or "time").
            half_life_logs: Half-life in run count for index-based weighting.
            half_life_days: Half-life in days for time-based weighting.
            accessibility_config: Optional accessibility configuration override.
        """
        super().__init__()
        self._data = WorkflowDataSource(
            workflow_dir=workflow_dir,
            refresh_rate=refresh_rate,
            use_estimation=use_estimation,
            profile_path=profile_path,
            use_wildcard_conditioning=use_wildcard_conditioning,
            weighting_strategy=weighting_strategy,
            half_life_logs=half_life_logs,
            half_life_days=half_life_days,
        )
        # Keep the constructor-supplied config around so toggling accessibility
        # off restores it rather than falling back to DEFAULT_CONFIG.
        self._base_accessibility_config = accessibility_config or DEFAULT_CONFIG
        self._accessibility_config = self._base_accessibility_config
        # Resolve the workflow path once; the header truncates it per frame, so the
        # per-render cost stays a string slice rather than a filesystem resolve().
        self._resolved_workflow_dir = str(workflow_dir.resolve())
        # Seed the reactive without firing watch_refresh_rate. A plain assignment would
        # invoke the watcher (whenever refresh_rate differs from the reactive default),
        # which calls set_interval() before run() has started the event loop, raising
        # "RuntimeError: no running event loop". on_mount starts the timer once the loop
        # is running. The ignore works around Textual's stubs typing class-level reactive
        # access as the value type (float) rather than Reactive[float].
        self.set_reactive(SnakeseeApp.refresh_rate, refresh_rate)  # type: ignore[arg-type]
        # Whether the (lazily-added) Cost columns have been added to the
        # completions / stats tables. Once True they stay — the column and flag
        # must always move together to avoid an add_row cell-count mismatch.
        self._completions_cost_col: bool = False
        self._stats_cost_col: bool = False

    def compose(self) -> ComposeResult:
        """Compose the widget tree (header / progress / six tables / summary / footer)."""
        yield Static(id="header")
        yield Static(id="progress")
        with Container(id="body"):
            with Horizontal(id="left"):
                yield DataTable(id="running")
                yield DataTable(id="completions")
            with Horizontal(id="right"):
                yield DataTable(id="pending")
                yield DataTable(id="failed")
                yield DataTable(id="incomplete")
                yield DataTable(id="stats")
        yield Static(id="summary")
        filter_input = Input(placeholder="filter rules…", id="filter")
        filter_input.can_focus = False
        yield filter_input
        yield Footer()

    def on_mount(self) -> None:
        """Configure tables, populate panels, and start the refresh timer."""
        running = self.query_one("#running", DataTable)
        running.add_columns("#", "Rule", "Thr", "Started", "Elapsed", "Progress", "ETA")
        running.cursor_type = "row"
        completions = self.query_one("#completions", DataTable)
        completions.add_columns("#", "Rule", "Thr", "Duration", "Completed")
        completions.cursor_type = "row"
        self.query_one("#pending", DataTable).add_columns("Rule", "Est. Count")
        self.query_one("#failed", DataTable).add_columns("#", "Rule", "Job ID")
        self.query_one("#incomplete", DataTable).add_columns("Output File")
        self.query_one("#stats", DataTable).add_columns("Rule", "Thr", "Count", "Avg", "Std Dev")
        self._refresh_panels()
        # Start the polling timer now that the event loop is running. __init__ seeds
        # refresh_rate via set_reactive (no watcher), so this is the only timer created.
        self._refresh_timer = self.set_interval(self.refresh_rate, self._refresh_panels)
        self.add_class(f"-{self.layout_mode.value}")

    def _refresh_panels(self, ignore_pause: bool = False) -> None:
        """Poll the data source and update header/progress/summary/tables widgets.

        Args:
            ignore_pause: When True, refresh even if ``paused`` is set. Used by every
                explicit user-triggered redraw (force refresh, pause toggle, sort,
                filter, log nav, accessibility, estimator re-init) — pausing gates
                only the interval timer's automatic polling.
        """
        if self.paused and not ignore_pause:
            return
        progress, estimate = self._data.poll_state()
        self._last_poll = (progress, estimate)
        self.query_one("#header", Static).update(
            make_header(
                progress,
                self._resolved_workflow_dir,
                self.paused,
                self._data.event_reader,
                max_path_len=max(20, self.size.width - 80),
            )
        )
        self.query_one("#progress", Static).update(
            make_progress_panel(
                progress,
                estimate,
                self._data.use_estimation,
                self._accessibility_config,
            )
        )
        self.query_one("#summary", Static).update(make_summary_footer(progress))

        self._populate_running(progress)
        self._populate_completions(progress)
        self._populate_pending(progress)
        self._populate_failed(progress)
        self._populate_incomplete(progress)
        self._populate_stats()

    def _populate_running(self, progress: WorkflowProgress) -> None:
        """Populate the running-jobs table from the current workflow progress."""
        table = self.query_one("#running", DataTable)
        table.clear()
        jobs = self._data.get_running_jobs_list(
            progress,
            filter_text=self.filter_text,
            sort_table=self._sort_table_name(),
            sort_column=self.sort_column,
            sort_ascending=self.sort_ascending,
        )
        rows = running_rows(self._data.build_running_job_data(jobs))
        for idx, row in enumerate(rows):
            job = row.job
            elapsed_str = (
                format_duration(row.elapsed_seconds) if row.elapsed_seconds is not None else "?"
            )
            remaining_str = (
                f"~{format_duration(row.remaining_seconds)}"
                if row.remaining_seconds is not None
                else "?"
            )
            started_str = "?"
            if job.start_time is not None:
                started_str = datetime.fromtimestamp(job.start_time).strftime("%H:%M:%S")

            progress_str = "-"
            if row.tool_progress is not None:
                if row.tool_progress.percent_complete is not None:
                    progress_str = row.tool_progress.percent_str
                else:
                    progress_str = f"{row.tool_progress.items_processed:,} {row.tool_progress.unit}"

            threads_str = str(job.threads) if job.threads is not None else "-"
            job_id_str = str(job.job_id) if job.job_id else str(idx + 1)
            table.add_row(
                job_id_str,
                job.rule,
                threads_str,
                started_str,
                elapsed_str,
                progress_str,
                remaining_str,
            )

    def _populate_completions(self, progress: WorkflowProgress) -> None:
        """Populate the recent-completions table from the current workflow progress."""
        table = self.query_one("#completions", DataTable)
        table.clear()
        jobs, failed_job_ids = self._data.get_completions_list(
            progress,
            filter_text=self.filter_text,
            sort_table=self._sort_table_name(),
            sort_column=self.sort_column,
            sort_ascending=self.sort_ascending,
        )
        # Add a Cost column the first time any estimated cost is available, so
        # runs without cost estimation never get a blank column.
        if progress.total_cost_estimate is not None and not self._completions_cost_col:
            table.add_column("Cost", key="cost")
            self._completions_cost_col = True

        rows = completion_rows(jobs, failed_job_ids)
        for idx, row in enumerate(rows):
            job = row.job
            duration_str = format_duration(job.duration) if job.duration is not None else "?"
            threads_str = str(job.threads) if job.threads is not None else "-"
            completed_str = "?"
            if job.end_time is not None:
                completed_str = datetime.fromtimestamp(job.end_time).strftime("%H:%M:%S")
            job_id_str = str(job.job_id) if job.job_id else str(idx + 1)
            cells = [job_id_str, job.rule, threads_str, duration_str, completed_str]
            if self._completions_cost_col:
                cells.append(
                    format_cost(job.cost_estimate) if job.cost_estimate is not None else "-"
                )
            table.add_row(*cells)

    def _populate_pending(self, progress: WorkflowProgress) -> None:
        """Populate the pending-jobs table using inferred per-rule pending counts."""
        table = self.query_one("#pending", DataTable)
        table.clear()
        pending_rules = self._data.get_inferred_pending_rules(progress)
        if not pending_rules:
            return
        rows = pending_rows(pending_rules)
        if self.sort_table == SortTable.PENDING:
            rows = sort_rows(rows, self.sort_column, self.sort_ascending)
        for row in rows:
            table.add_row(row.rule, str(row.job_count))

    def _populate_failed(self, progress: WorkflowProgress) -> None:
        """Populate the failed-jobs table from ``progress.failed_jobs_list``."""
        table = self.query_one("#failed", DataTable)
        table.clear()
        rows = failed_rows(progress)
        for idx, row in enumerate(rows):
            job = row.job
            job_id_str = job.job_id if job.job_id else "-"
            table.add_row(str(idx + 1), job.rule, job_id_str)

    def _populate_incomplete(self, progress: WorkflowProgress) -> None:
        """Populate the incomplete-jobs table from ``progress.incomplete_jobs_list``."""
        table = self.query_one("#incomplete", DataTable)
        table.clear()
        for row in incomplete_rows(progress):
            table.add_row(row.display_path)

    def _populate_stats(self) -> None:
        """Populate the rule-statistics table from the data source's filtered stats."""
        table = self.query_one("#stats", DataTable)
        table.clear()
        if not self._data.use_estimation:
            return
        stats_list = self._data.get_filtered_stats()
        if not stats_list:
            return
        # Default ordering: most-frequently-run rules first.
        stats_list = sorted(stats_list, key=lambda s: s.count, reverse=True)
        rows = stats_rows(stats_list, self._data.thread_stats_dict())
        if self.sort_table == SortTable.STATS:
            rows = sort_stats_rows(rows, self.sort_column, self.sort_ascending)

        # Per-rule estimated cost: add a Cost column once any cost data exists.
        # Sourced from the live registry (the stats panel is itself registry-backed,
        # so this stays consistent with the rest of the stats frame).
        cost_by_rule = self._data.cost_by_rule()
        if cost_by_rule and not self._stats_cost_col:
            table.add_column("Cost", key="cost")
            self._stats_cost_col = True

        for row in rows:
            cells = [
                row.rule_display,
                row.threads,
                str(row.stats.count),
                format_duration(row.stats.mean_duration),
                format_duration(row.stats.std_dev) if row.stats.std_dev > 0 else "-",
            ]
            if self._stats_cost_col:
                rule_cost = cost_by_rule.get(row.stats.rule)
                cells.append(format_cost(rule_cost) if rule_cost is not None else "-")
            table.add_row(*cells)

Attributes

last_poll property

last_poll: tuple[WorkflowProgress, TimeEstimate | None] | None

The most recent (progress, estimate) snapshot taken by the refresh cycle.

Read-only accessor for external tooling (e.g. the docs screenshot generator) so it does not have to reach into the private attribute. Returns None until the first refresh has polled the data source.

Methods:

__init__

__init__(workflow_dir: Path, refresh_rate: float = DEFAULT_REFRESH_RATE, use_estimation: bool = True, profile_path: Path | None = None, use_wildcard_conditioning: bool = True, weighting_strategy: WeightingStrategy = 'index', half_life_logs: int = 10, half_life_days: float = 7.0, accessibility_config: AccessibilityConfig | None = None) -> None

Initialize the SnakeseeApp.

Parameters:

Name	Type	Description	Default
workflow_dir	Path	Path to workflow directory containing .snakemake/.	required
refresh_rate	float	Refresh interval in seconds.	DEFAULT_REFRESH_RATE
use_estimation	bool	Whether to enable time estimation.	True
profile_path	Path | None	Optional path to a timing profile for bootstrapping estimates.	None
use_wildcard_conditioning	bool	Whether to enable wildcard-conditioned estimates.	True
weighting_strategy	WeightingStrategy	Strategy for weighting historical data ("index" or "time").	'index'
half_life_logs	int	Half-life in run count for index-based weighting.	10
half_life_days	float	Half-life in days for time-based weighting.	7.0
accessibility_config	AccessibilityConfig | None	Optional accessibility configuration override.	None

Source code in snakesee/tui/app.py

def __init__(
    self,
    workflow_dir: Path,
    refresh_rate: float = DEFAULT_REFRESH_RATE,
    use_estimation: bool = True,
    profile_path: Path | None = None,
    use_wildcard_conditioning: bool = True,
    weighting_strategy: WeightingStrategy = "index",
    half_life_logs: int = 10,
    half_life_days: float = 7.0,
    accessibility_config: AccessibilityConfig | None = None,
) -> None:
    """Initialize the SnakeseeApp.

    Args:
        workflow_dir: Path to workflow directory containing ``.snakemake/``.
        refresh_rate: Refresh interval in seconds.
        use_estimation: Whether to enable time estimation.
        profile_path: Optional path to a timing profile for bootstrapping estimates.
        use_wildcard_conditioning: Whether to enable wildcard-conditioned estimates.
        weighting_strategy: Strategy for weighting historical data ("index" or "time").
        half_life_logs: Half-life in run count for index-based weighting.
        half_life_days: Half-life in days for time-based weighting.
        accessibility_config: Optional accessibility configuration override.
    """
    super().__init__()
    self._data = WorkflowDataSource(
        workflow_dir=workflow_dir,
        refresh_rate=refresh_rate,
        use_estimation=use_estimation,
        profile_path=profile_path,
        use_wildcard_conditioning=use_wildcard_conditioning,
        weighting_strategy=weighting_strategy,
        half_life_logs=half_life_logs,
        half_life_days=half_life_days,
    )
    # Keep the constructor-supplied config around so toggling accessibility
    # off restores it rather than falling back to DEFAULT_CONFIG.
    self._base_accessibility_config = accessibility_config or DEFAULT_CONFIG
    self._accessibility_config = self._base_accessibility_config
    # Resolve the workflow path once; the header truncates it per frame, so the
    # per-render cost stays a string slice rather than a filesystem resolve().
    self._resolved_workflow_dir = str(workflow_dir.resolve())
    # Seed the reactive without firing watch_refresh_rate. A plain assignment would
    # invoke the watcher (whenever refresh_rate differs from the reactive default),
    # which calls set_interval() before run() has started the event loop, raising
    # "RuntimeError: no running event loop". on_mount starts the timer once the loop
    # is running. The ignore works around Textual's stubs typing class-level reactive
    # access as the value type (float) rather than Reactive[float].
    self.set_reactive(SnakeseeApp.refresh_rate, refresh_rate)  # type: ignore[arg-type]
    # Whether the (lazily-added) Cost columns have been added to the
    # completions / stats tables. Once True they stay — the column and flag
    # must always move together to avoid an add_row cell-count mismatch.
    self._completions_cost_col: bool = False
    self._stats_cost_col: bool = False

action_clear_filter

action_clear_filter() -> None

Hide the filter Input if focused, otherwise clear the filter and return to latest log.

Source code in snakesee/tui/app.py

def action_clear_filter(self) -> None:
    """Hide the filter Input if focused, otherwise clear the filter and return to latest log."""
    focused = self.focused
    if isinstance(focused, Input) and focused.id == "filter":
        focused.remove_class("-active")
        focused.value = ""
        self.set_focus(None)
        focused.can_focus = False
        return
    self.filter_text = None
    self.current_log_index = 0

action_cycle_layout

action_cycle_layout() -> None

Cycle to the next layout mode.

Source code in snakesee/tui/app.py

def action_cycle_layout(self) -> None:
    """Cycle to the next layout mode."""
    modes = list(LayoutMode)
    idx = modes.index(self.layout_mode)
    self.layout_mode = modes[(idx + 1) % len(modes)]

action_cycle_sort_back

action_cycle_sort_back() -> None

Cycle the sort target one step backward (None → stats → … → running → None).

Source code in snakesee/tui/app.py

def action_cycle_sort_back(self) -> None:
    """Cycle the sort target one step backward (None → stats → … → running → None)."""
    self._cycle_sort(direction=-1)

action_cycle_sort_forward

action_cycle_sort_forward() -> None

Cycle the sort target one step forward (None → running → … → stats → None).

Source code in snakesee/tui/app.py

def action_cycle_sort_forward(self) -> None:
    """Cycle the sort target one step forward (None → running → … → stats → None)."""
    self._cycle_sort(direction=1)

action_easter_complete

action_easter_complete() -> None

Push the EasterEggScreen if the f-then-g chord finished within the window.

Source code in snakesee/tui/app.py

def action_easter_complete(self) -> None:
    """Push the EasterEggScreen if the f-then-g chord finished within the window."""
    if self._easter_pending:
        self._easter_pending = False
        if self._easter_timer is not None:
            self._easter_timer.stop()
            self._easter_timer = None
        self.push_screen(EasterEggScreen())

action_easter_pending

action_easter_pending() -> None

Start (or restart) the 2-second window for completing the f-then-g easter egg.

Source code in snakesee/tui/app.py

def action_easter_pending(self) -> None:
    """Start (or restart) the 2-second window for completing the f-then-g easter egg."""
    self._easter_pending = True
    if self._easter_timer is not None:
        self._easter_timer.stop()
    self._easter_timer = self.set_timer(2.0, self._clear_easter)

action_force_refresh

action_force_refresh() -> None

Force an immediate panel refresh, even when paused.

Source code in snakesee/tui/app.py

def action_force_refresh(self) -> None:
    """Force an immediate panel refresh, even when paused."""
    self._refresh_panels(ignore_pause=True)

action_hard_refresh

action_hard_refresh() -> None

Re-initialize the estimator in a worker thread and refresh panels.

Source code in snakesee/tui/app.py

def action_hard_refresh(self) -> None:
    """Re-initialize the estimator in a worker thread and refresh panels."""
    self.run_worker(self._reinit_estimator, thread=True, exclusive=True)

action_log_newer

action_log_newer(step: int) -> None

Navigate to a newer log file by step entries.

Parameters:

Name	Type	Description	Default
step	int	Number of log entries to step forward (toward newer logs).	required

Source code in snakesee/tui/app.py

def action_log_newer(self, step: int) -> None:
    """Navigate to a newer log file by ``step`` entries.

    Args:
        step: Number of log entries to step forward (toward newer logs).
    """
    self.current_log_index = max(0, self.current_log_index - step)

action_log_older

action_log_older(step: int) -> None

Navigate to an older log file by step entries.

Parameters:

Name	Type	Description	Default
step	int	Number of log entries to step backward (toward older logs).	required

Source code in snakesee/tui/app.py

def action_log_older(self, step: int) -> None:
    """Navigate to an older log file by ``step`` entries.

    Args:
        step: Number of log entries to step backward (toward older logs).
    """
    self._data.refresh_log_list()
    max_idx = self._data.available_log_count - 1
    if max_idx < 0:
        self.current_log_index = 0
        return
    self.current_log_index = min(max_idx, self.current_log_index + step)

action_next_match

action_next_match() -> None

Move the cursor to the next row whose Rule column matches the filter.

Source code in snakesee/tui/app.py

def action_next_match(self) -> None:
    """Move the cursor to the next row whose Rule column matches the filter."""
    self._jump_match(direction=1)

action_open_filter

action_open_filter() -> None

Reveal the filter Input widget and focus it for keyboard entry.

Source code in snakesee/tui/app.py

def action_open_filter(self) -> None:
    """Reveal the filter Input widget and focus it for keyboard entry."""
    f = self.query_one("#filter", Input)
    f.can_focus = True
    f.add_class("-active")
    f.focus()

action_prev_match

action_prev_match() -> None

Move the cursor to the previous row whose Rule column matches the filter.

Source code in snakesee/tui/app.py

def action_prev_match(self) -> None:
    """Move the cursor to the previous row whose Rule column matches the filter."""
    self._jump_match(direction=-1)

action_rate_dec

action_rate_dec(delta: float) -> None

Decrease the refresh rate by delta seconds, clamped to MIN_REFRESH_RATE.

Source code in snakesee/tui/app.py

def action_rate_dec(self, delta: float) -> None:
    """Decrease the refresh rate by ``delta`` seconds, clamped to MIN_REFRESH_RATE."""
    self.refresh_rate = max(MIN_REFRESH_RATE, self.refresh_rate - delta)

action_rate_inc

action_rate_inc(delta: float) -> None

Increase the refresh rate by delta seconds, clamped to MAX_REFRESH_RATE.

Source code in snakesee/tui/app.py

def action_rate_inc(self, delta: float) -> None:
    """Increase the refresh rate by ``delta`` seconds, clamped to MAX_REFRESH_RATE."""
    self.refresh_rate = min(MAX_REFRESH_RATE, self.refresh_rate + delta)

action_rate_min

action_rate_min() -> None

Set the refresh rate to the minimum value.

Source code in snakesee/tui/app.py

def action_rate_min(self) -> None:
    """Set the refresh rate to the minimum value."""
    self.refresh_rate = MIN_REFRESH_RATE

action_rate_reset

action_rate_reset() -> None

Reset the refresh rate to the default value.

Source code in snakesee/tui/app.py

def action_rate_reset(self) -> None:
    """Reset the refresh rate to the default value."""
    self.refresh_rate = DEFAULT_REFRESH_RATE

action_show_help

action_show_help() -> None

Push the modal HelpScreen overlay.

Source code in snakesee/tui/app.py

def action_show_help(self) -> None:
    """Push the modal HelpScreen overlay."""
    self.push_screen(HelpScreen())

action_sort_column

action_sort_column(col: int) -> None

Set the sort column for the active sort target, or toggle direction if same column.

Columns are 0-indexed. Each table enforces its own maximum: running and stats support columns 0-3; completions 0-2; pending 0-1.

Parameters:

Name	Type	Description	Default
col	int	Zero-based column index to sort by.	required

Source code in snakesee/tui/app.py

def action_sort_column(self, col: int) -> None:
    """Set the sort column for the active sort target, or toggle direction if same column.

    Columns are 0-indexed.  Each table enforces its own maximum:
    running and stats support columns 0-3; completions 0-2; pending 0-1.

    Args:
        col: Zero-based column index to sort by.
    """
    if self.sort_table is None:
        return
    if col >= _SORT_MAX_COLS[self.sort_table]:
        return
    if col == self.sort_column:
        self.sort_ascending = not self.sort_ascending
    else:
        self.sort_column = col
        self.sort_ascending = True
    self._refresh_panels(ignore_pause=True)

action_toggle_accessibility

action_toggle_accessibility() -> None

Toggle accessibility mode, updating the accessibility config and refreshing panels.

Toggling off restores the constructor-supplied config (which may be a custom override), not necessarily DEFAULT_CONFIG.

Source code in snakesee/tui/app.py

def action_toggle_accessibility(self) -> None:
    """Toggle accessibility mode, updating the accessibility config and refreshing panels.

    Toggling off restores the constructor-supplied config (which may be a
    custom override), not necessarily ``DEFAULT_CONFIG``.
    """
    self.accessibility_mode = not self.accessibility_mode
    self._accessibility_config = (
        ACCESSIBLE_CONFIG if self.accessibility_mode else self._base_accessibility_config
    )
    self._refresh_panels(ignore_pause=True)

action_toggle_estimation

action_toggle_estimation() -> None

Toggle time estimation and re-initialize the estimator in a worker thread.

Source code in snakesee/tui/app.py

def action_toggle_estimation(self) -> None:
    """Toggle time estimation and re-initialize the estimator in a worker thread."""
    self._data.use_estimation = not self._data.use_estimation
    self.run_worker(self._reinit_estimator, thread=True, exclusive=True)

action_toggle_pause

action_toggle_pause() -> None

Toggle the paused state of the monitor and repaint immediately.

The repaint bypasses the pause guard so the header's PAUSED indicator updates right away instead of waiting for the next interval tick.

Source code in snakesee/tui/app.py

def action_toggle_pause(self) -> None:
    """Toggle the paused state of the monitor and repaint immediately.

    The repaint bypasses the pause guard so the header's PAUSED indicator
    updates right away instead of waiting for the next interval tick.
    """
    self.paused = not self.paused
    self._refresh_panels(ignore_pause=True)

action_toggle_wildcard

action_toggle_wildcard() -> None

Toggle wildcard conditioning and re-initialize the estimator in a worker thread.

Source code in snakesee/tui/app.py

def action_toggle_wildcard(self) -> None:
    """Toggle wildcard conditioning and re-initialize the estimator in a worker thread."""
    self._data.use_wildcard_conditioning = not self._data.use_wildcard_conditioning
    self.run_worker(self._reinit_estimator, thread=True, exclusive=True)

compose

compose() -> ComposeResult

Compose the widget tree (header / progress / six tables / summary / footer).

Source code in snakesee/tui/app.py

def compose(self) -> ComposeResult:
    """Compose the widget tree (header / progress / six tables / summary / footer)."""
    yield Static(id="header")
    yield Static(id="progress")
    with Container(id="body"):
        with Horizontal(id="left"):
            yield DataTable(id="running")
            yield DataTable(id="completions")
        with Horizontal(id="right"):
            yield DataTable(id="pending")
            yield DataTable(id="failed")
            yield DataTable(id="incomplete")
            yield DataTable(id="stats")
    yield Static(id="summary")
    filter_input = Input(placeholder="filter rules…", id="filter")
    filter_input.can_focus = False
    yield filter_input
    yield Footer()

on_data_table_row_selected

on_data_table_row_selected(event: RowSelected) -> None

Open the JobLogScreen for the selected job in the running/completions tables.

Source code in snakesee/tui/app.py

def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None:
    """Open the JobLogScreen for the selected job in the running/completions tables."""
    table_id = event.data_table.id
    if table_id not in {SortTable.RUNNING.value, SortTable.COMPLETIONS.value}:
        return
    # Reuse the latest poll snapshot — populated by _refresh_panels at most
    # `refresh_rate` seconds ago, which is also the data the table was rendered from.
    if self._last_poll is None:
        self._last_poll = self._data.poll_state()
    progress, _ = self._last_poll
    if table_id == SortTable.RUNNING.value:
        jobs = self._data.get_running_jobs_list(
            progress,
            filter_text=self.filter_text,
            sort_table=self._sort_table_name(),
            sort_column=self.sort_column,
            sort_ascending=self.sort_ascending,
        )
    else:  # completions
        jobs, _ = self._data.get_completions_list(
            progress,
            filter_text=self.filter_text,
            sort_table=self._sort_table_name(),
            sort_column=self.sort_column,
            sort_ascending=self.sort_ascending,
        )
    if event.cursor_row >= len(jobs):
        return
    job = jobs[event.cursor_row]

    # Remote jobs (e.g. AWS Batch) may have no local log file but still carry
    # an external id + console/CloudWatch links worth showing.
    from snakesee.tui.renderables import make_remote_job_info

    header_lines = make_remote_job_info(job)

    log_path = job.log_file
    if log_path is None:
        if not header_lines:
            return
        # No local log, but remote info is available — show just that.
        self.push_screen(JobLogScreen(None, [], header_lines=header_lines))
        return
    lines = self._data.read_log_tail(log_path, max_lines=500)
    self.push_screen(JobLogScreen(log_path, lines, header_lines=header_lines))

on_input_submitted

on_input_submitted(event: Submitted) -> None

Apply the submitted filter text and hide the Input again.

Source code in snakesee/tui/app.py

def on_input_submitted(self, event: Input.Submitted) -> None:
    """Apply the submitted filter text and hide the Input again."""
    if event.input.id != "filter":
        return
    self.filter_text = event.value or None
    event.input.remove_class("-active")
    event.input.value = ""
    self.set_focus(None)
    event.input.can_focus = False

on_mount

on_mount() -> None

Configure tables, populate panels, and start the refresh timer.

Source code in snakesee/tui/app.py

def on_mount(self) -> None:
    """Configure tables, populate panels, and start the refresh timer."""
    running = self.query_one("#running", DataTable)
    running.add_columns("#", "Rule", "Thr", "Started", "Elapsed", "Progress", "ETA")
    running.cursor_type = "row"
    completions = self.query_one("#completions", DataTable)
    completions.add_columns("#", "Rule", "Thr", "Duration", "Completed")
    completions.cursor_type = "row"
    self.query_one("#pending", DataTable).add_columns("Rule", "Est. Count")
    self.query_one("#failed", DataTable).add_columns("#", "Rule", "Job ID")
    self.query_one("#incomplete", DataTable).add_columns("Output File")
    self.query_one("#stats", DataTable).add_columns("Rule", "Thr", "Count", "Avg", "Std Dev")
    self._refresh_panels()
    # Start the polling timer now that the event loop is running. __init__ seeds
    # refresh_rate via set_reactive (no watcher), so this is the only timer created.
    self._refresh_timer = self.set_interval(self.refresh_rate, self._refresh_panels)
    self.add_class(f"-{self.layout_mode.value}")

watch_current_log_index

watch_current_log_index(old: int, new: int) -> None

Sync the log index to the data source and refresh panels when it changes.

Log navigation only comes from user keystrokes, so the redraw bypasses the pause guard.

Source code in snakesee/tui/app.py

def watch_current_log_index(self, old: int, new: int) -> None:
    """Sync the log index to the data source and refresh panels when it changes.

    Log navigation only comes from user keystrokes, so the redraw bypasses
    the pause guard.
    """
    self._data.current_log_index = new
    self._refresh_panels(ignore_pause=True)

watch_filter_text

watch_filter_text(old: str | None, new: str | None) -> None

Re-populate all panels when the filter text changes.

Filter changes only come from user keystrokes, so the redraw bypasses the pause guard.

Source code in snakesee/tui/app.py

def watch_filter_text(self, old: str | None, new: str | None) -> None:
    """Re-populate all panels when the filter text changes.

    Filter changes only come from user keystrokes, so the redraw bypasses
    the pause guard.
    """
    self._refresh_panels(ignore_pause=True)

watch_layout_mode

watch_layout_mode(old: LayoutMode, new: LayoutMode) -> None

Swap the CSS class on the root when the layout mode changes.

Source code in snakesee/tui/app.py

def watch_layout_mode(self, old: LayoutMode, new: LayoutMode) -> None:
    """Swap the CSS class on the root when the layout mode changes."""
    for mode in LayoutMode:
        self.remove_class(f"-{mode.value}")
    self.add_class(f"-{new.value}")

watch_refresh_rate

watch_refresh_rate(old: float, new: float) -> None

Restart the polling timer when the refresh rate changes.

Source code in snakesee/tui/app.py

def watch_refresh_rate(self, old: float, new: float) -> None:
    """Restart the polling timer when the refresh rate changes."""
    if self._refresh_timer is not None:
        self._refresh_timer.stop()
    self._refresh_timer = self.set_interval(new, self._refresh_panels)
    self._data.refresh_rate = new
    self._data.update_cache_ttl()

SortTable

Bases: StrEnum

Sortable DataTable identifiers; values match the widget IDs in compose().

Source code in snakesee/tui/app.py

class SortTable(StrEnum):
    """Sortable DataTable identifiers; values match the widget IDs in compose()."""

    RUNNING = "running"
    COMPLETIONS = "completions"
    PENDING = "pending"
    STATS = "stats"

Functions:

data_source

Workflow data source: pure-data layer for the snakesee TUI.

This module contains :class:WorkflowDataSource, which owns the non-presentation state of the TUI: polling, estimator initialization, event/log readers, event handlers, filter/sort helpers, log tail caching, and tool-progress caching.

:class:snakesee.tui.app.SnakeseeApp composes a WorkflowDataSource and delegates data-layer calls to it.

Classes

WorkflowDataSource

Pure data layer for snakesee TUI.

Owns polling, estimator state, event/log readers, event handlers, filter/sort helpers, log tail caching, and tool-progress caching. Rendering and input handling live in :class:snakesee.tui.app.SnakeseeApp.

Source code in snakesee/tui/data_source.py

class WorkflowDataSource:
    """Pure data layer for snakesee TUI.

    Owns polling, estimator state, event/log readers, event handlers, filter/sort
    helpers, log tail caching, and tool-progress caching. Rendering and input
    handling live in :class:`snakesee.tui.app.SnakeseeApp`.
    """

    def __init__(
        self,
        workflow_dir: Path,
        refresh_rate: float = DEFAULT_REFRESH_RATE,
        use_estimation: bool = True,
        profile_path: Path | None = None,
        use_wildcard_conditioning: bool = True,
        weighting_strategy: WeightingStrategy = "index",
        half_life_logs: int = 10,
        half_life_days: float = 7.0,
    ) -> None:
        """Initialize the data source.

        Args:
            workflow_dir: Path to workflow directory containing ``.snakemake/``.
            refresh_rate: Refresh interval in seconds (used to size cache TTL).
            use_estimation: Whether to enable time estimation.
            profile_path: Optional path to a timing profile for bootstrapping estimates.
            use_wildcard_conditioning: Whether to enable wildcard-conditioned estimates.
            weighting_strategy: Strategy for weighting historical data ("index" or "time").
            half_life_logs: Half-life in run count for index-based weighting.
            half_life_days: Half-life in days for time-based weighting.
        """
        self.workflow_dir = workflow_dir
        self.refresh_rate = refresh_rate
        self.use_estimation = use_estimation
        self.profile_path = profile_path
        self.weighting_strategy = weighting_strategy
        self.half_life_logs = half_life_logs
        self.half_life_days = half_life_days

        self._use_wildcard_conditioning: bool = use_wildcard_conditioning

        self._estimator: TimeEstimator | None = None

        # Log file navigation
        self._available_logs: list[Path] = []
        self._current_log_index: int = 0  # 0 = most recent
        self._latest_log_path: Path | None = None  # Track latest log to detect new workflows
        self.refresh_log_list()

        # Cutoff time for historical view (updated in poll_state)
        self._cutoff_time: float | None = None

        # Cached log tail data
        self._cached_log_path: Path | None = None
        self._cached_log_lines: list[str] = []
        self._cached_log_mtime: float = 0

        # Tool progress cache (to avoid parsing job logs on every refresh).
        # Cache stores: (cached_time, file_mtime, progress) - invalidates if file changes.
        self._tool_progress_cache: dict[str, tuple[float, float, ToolProgress | None]] = {}
        # Adaptive TTL: scales with refresh rate to avoid cache outliving refresh cycles.
        self._tool_progress_cache_ttl: float = min(
            ADAPTIVE_CACHE_TTL_MULTIPLIER * refresh_rate, MAX_CACHE_TTL
        )

        # Event reader for real-time events from logger plugin
        self._event_reader: EventReader | None = None
        self._events_enabled: bool = True
        self.init_event_reader()

        # All scheduled jobs from log (for pending job estimation without logger plugin)
        self._all_scheduled_jobs: dict[str, JobInfo] = {}

        # Incremental log reader for efficient polling
        self._log_reader: IncrementalLogReader | None = None
        self.init_log_reader()

        # Validation: compare event-based state with parsed state
        self._event_accumulator: EventAccumulator | None = None
        self._validation_logger: ValidationLogger | None = None
        self.init_validation()

        # Centralized workflow state
        self._workflow_state: WorkflowState = WorkflowState.create(
            workflow_dir=workflow_dir,
        )

        self.init_estimator()

    # --------------------------------------------------- public properties
    @property
    def use_wildcard_conditioning(self) -> bool:
        """Whether wildcard-conditioned estimates are enabled."""
        return self._use_wildcard_conditioning

    @use_wildcard_conditioning.setter
    def use_wildcard_conditioning(self, value: bool) -> None:
        self._use_wildcard_conditioning = value

    @property
    def current_log_index(self) -> int:
        """Index into ``available_logs``; 0 = most recent."""
        return self._current_log_index

    @current_log_index.setter
    def current_log_index(self, value: int) -> None:
        self._current_log_index = value

    @property
    def available_log_count(self) -> int:
        """Number of historical log files currently discovered."""
        return len(self._available_logs)

    @property
    def event_reader(self) -> EventReader | None:
        """Event reader for the current workflow run, if any."""
        return self._event_reader

    def build_running_job_data(
        self, jobs: list[JobInfo]
    ) -> list[tuple[JobInfo, float | None, float | None, float | None, ToolProgress | None]]:
        """Build per-job tuples of (job, elapsed, remaining, start_time, tool_progress)."""
        return self._build_running_job_data(jobs)

    def thread_stats_dict(self) -> "dict[str, ThreadTimingStats]":
        """Return per-rule, per-thread timing statistics."""
        return self._workflow_state.rules.to_thread_stats_dict()

    def cost_by_rule(self) -> dict[str, float]:
        """Return summed estimated cost per rule (empty when no cost data)."""
        return self._workflow_state.jobs.cost_by_rule()

    # ------------------------------------------------------------------ logs
    def refresh_log_list(self) -> None:
        """Refresh the list of available log files."""
        log_dir = self.workflow_dir / ".snakemake" / "log"
        if log_dir.exists():
            # Sort by modification time, newest first
            logs = sorted(
                log_dir.glob("*.snakemake.log"),
                key=lambda p: p.stat().st_mtime,
                reverse=True,
            )
            self._available_logs = logs
        else:
            self._available_logs = []

        # Reset to most recent if current index is out of bounds
        if self._current_log_index >= len(self._available_logs):
            self._current_log_index = 0

        # Detect when a new workflow starts (new latest log)
        # and re-parse current_rules to filter pending jobs correctly
        new_latest = self._available_logs[0] if self._available_logs else None
        if new_latest != self._latest_log_path:
            self._latest_log_path = new_latest
            self._init_current_rules_from_log()

    def get_current_log(self) -> Path | None:
        """Get the currently selected log file."""
        if not self._available_logs:
            return None
        if self._current_log_index < len(self._available_logs):
            return self._available_logs[self._current_log_index]
        return self._available_logs[0] if self._available_logs else None

    # ----------------------------------------------------------- estimator
    def init_estimator(self, *, show_progress: bool = True) -> None:
        """Initialize or reinitialize the time estimator.

        At startup the load can take many seconds and the user needs feedback
        before the App's compose() returns, so a transient Rich progress spinner
        is rendered directly to the terminal. This is a pragmatic exception to the
        data source being otherwise rendering-agnostic.

        At runtime (re-init triggered by a key press while the Textual UI is
        live), rendering Rich output would corrupt the display, so callers pass
        ``show_progress=False`` to load silently.

        Args:
            show_progress: Render a transient Rich progress spinner during load.
                Set False when the Textual UI is already running.
        """
        self._workflow_state.rules.clear()
        self._workflow_state.jobs.clear()

        if not self.use_estimation:
            self._estimator = None
            return

        self._estimator = TimeEstimator(
            use_wildcard_conditioning=self._use_wildcard_conditioning,
            weighting_strategy=self.weighting_strategy,
            half_life_logs=self.half_life_logs,
            half_life_days=self.half_life_days,
            rule_registry=self._workflow_state.rules,
        )

        metadata_dir = self.workflow_dir / ".snakemake" / "metadata"
        has_metadata_fs = metadata_dir.exists()
        has_profile = self.profile_path is not None and self.profile_path.exists()

        # Check if there's anything to load (worth showing progress)
        paths = WorkflowPaths(self.workflow_dir)
        has_metadata_db = paths.has_metadata_db
        has_metadata = has_metadata_fs or has_metadata_db
        log_paths = paths.find_all_logs()

        # Skip loading entirely if there's nothing to load
        if not has_metadata and not has_profile and not log_paths:
            return

        # Render a real Rich progress spinner only at startup; load silently when
        # the Textual UI is already on screen (a Rich render would corrupt it).
        progress_cm: AbstractContextManager[Any]
        if show_progress:
            from rich.console import Console
            from rich.progress import BarColumn
            from rich.progress import MofNCompleteColumn
            from rich.progress import Progress
            from rich.progress import SpinnerColumn
            from rich.progress import TextColumn

            progress_cm = Progress(
                SpinnerColumn(),
                TextColumn("[progress.description]{task.description}"),
                BarColumn(),
                MofNCompleteColumn(),
                console=Console(),
                transient=True,
            )
        else:
            progress_cm = nullcontext(_NULL_PROGRESS)

        with progress_cm as progress:
            # Load from profile first if available
            if has_profile:
                task = progress.add_task("Loading profile...", total=1)
                try:
                    from snakesee.profile import load_profile

                    assert self.profile_path is not None
                    profile = load_profile(self.profile_path)
                    self._estimator.rule_stats = profile.to_rule_stats()
                except (OSError, ValueError) as e:
                    # Log failure and fall back to metadata only
                    logger.debug("Failed to load profile %s: %s", self.profile_path, e)
                progress.update(task, completed=1)

            # Load metadata via persistence backend (supports both FS and DB)
            from snakesee.persistence import detect_backend

            backend = detect_backend(self.workflow_dir)

            if has_metadata:
                # Determine progress bar total from FS file count (DB has no cheap count)
                if has_metadata_fs and not has_metadata_db:
                    metadata_files = list(metadata_dir.rglob("*"))
                    metadata_files = [f for f in metadata_files if f.is_file()]
                    file_count = len(metadata_files)
                else:
                    file_count = 0

                task = progress.add_task(
                    "Loading metadata...", total=file_count if file_count > 0 else None
                )

                def metadata_cb(current: int, _total: int) -> None:
                    progress.update(task, completed=current)

                self._estimator.load_from_backend(backend, progress_callback=metadata_cb)

            # Load historical timing from events file (complements metadata)
            events_file = get_event_file_path(self.workflow_dir)
            if events_file.exists():
                task = progress.add_task("Loading events...", total=1)
                self._estimator.load_from_events(events_file)
                progress.update(task, completed=1)

            # Initialize thread stats from log parsing
            if log_paths:
                task = progress.add_task("Analyzing thread usage...", total=len(log_paths))
                self._init_thread_stats_from_log(
                    log_paths=log_paths,
                    progress_callback=lambda current, _total: progress.update(
                        task, completed=current
                    ),
                )

            # Parse current rules (fast, no progress needed)
            self._init_current_rules_from_log()

    def _init_thread_stats_from_log(
        self,
        log_paths: list[Path] | None = None,
        progress_callback: "ProgressCallback | None" = None,
    ) -> None:
        """Initialize thread stats from all log files (metadata doesn't have threads).

        Populates the centralized RuleRegistry with thread-specific timing data.

        Args:
            log_paths: Optional list of log paths to process (avoids re-discovering).
            progress_callback: Optional callback(current, total) for progress reporting.
        """
        from snakesee.parser import parse_completed_jobs_from_log

        if log_paths is None:
            paths = WorkflowPaths(self.workflow_dir)
            log_paths = paths.find_all_logs()

        if not log_paths:
            return

        total = len(log_paths)
        for i, log_path in enumerate(log_paths):
            if progress_callback is not None:
                progress_callback(i + 1, total)

            for job in parse_completed_jobs_from_log(log_path):
                if job.threads is None or job.duration is None:
                    continue
                # Record to centralized RuleRegistry (includes thread info)
                self._workflow_state.rules.record_completion(
                    rule=job.rule,
                    duration=job.duration,
                    timestamp=job.end_time or 0.0,
                    threads=job.threads,
                    wildcards=dict(job.wildcards) if job.wildcards else None,
                    input_size=job.input_size,
                )

    def _init_current_rules_from_log(self) -> None:
        """Parse current rules, job counts, cores, and all scheduled jobs from the latest log.

        Always resets the inferred-run state first so that stale fields from a previous
        run can't survive into the next workflow when a new log appears before snakemake
        has emitted the corresponding job_stats / scheduled_jobs blocks.
        """
        from snakesee.parser import parse_all_jobs_from_log
        from snakesee.parser import parse_cores_from_log
        from snakesee.parser import parse_job_stats_counts_from_log
        from snakesee.parser import parse_job_stats_from_log

        if self._estimator is None:
            return

        # Clear previous-run inference state before re-parsing.
        self._estimator.current_rules = None
        self._estimator.expected_job_counts = None
        self._all_scheduled_jobs = {}

        paths = WorkflowPaths(self.workflow_dir)
        log_path = paths.find_latest_log()
        if log_path is None:
            return

        # Parse rule names for filtering
        current_rules = parse_job_stats_from_log(log_path)
        if current_rules:
            self._estimator.current_rules = current_rules

        # Parse job counts for accurate pending job inference
        job_counts = parse_job_stats_counts_from_log(log_path)
        if job_counts:
            self._estimator.expected_job_counts = job_counts

        # Parse "Provided cores: N" for definitive parallelism info
        cores = parse_cores_from_log(log_path)
        if cores is not None:
            self._estimator.set_provided_cores(cores)

        # Parse all scheduled jobs with wildcards for pending job estimation
        all_jobs = parse_all_jobs_from_log(log_path)
        if all_jobs:
            self._all_scheduled_jobs = {job.job_id: job for job in all_jobs if job.job_id}

    # ------------------------------------------------------- events / readers
    def init_event_reader(self) -> None:
        """Initialize the event reader if event file exists and is current.

        The events file is validated against the current log file's start time
        to ensure we don't use stale events from a previous workflow run.
        """
        if not self._events_enabled:
            return

        event_file = get_event_file_path(self.workflow_dir)
        if not event_file.exists():
            self._event_reader = None
            return

        # Get the current log file's start time for validation. Use the first
        # timestamp recorded in the log (the workflow-start time) rather than the
        # file's mtime: mtime is the last-append time and drifts forward as the
        # run writes, so after a minute of activity it would make the current
        # run's own event file look "stale". The first log timestamp is fixed for
        # the life of the run.
        paths = WorkflowPaths(self.workflow_dir)
        log_path = paths.find_latest_log()
        log_start_time = _parse_log_start_time(log_path) if log_path is not None else None

        # Validate the events file is for the current workflow run
        if _is_event_file_current(event_file, log_start_time):
            self._event_reader = EventReader(event_file)
            logger.debug("Events file is current, using for monitoring")
        else:
            self._event_reader = None
            logger.info(
                "Ignoring stale events file %s (from a previous workflow run)",
                event_file,
            )

    def init_log_reader(self) -> None:
        """Initialize the incremental log reader.

        Creates a reader for the current log file, enabling efficient
        incremental parsing instead of re-reading the entire file on each poll.
        """
        paths = WorkflowPaths(self.workflow_dir)
        log_path = paths.find_latest_log()
        if log_path is not None:
            self._log_reader = IncrementalLogReader(log_path)
        else:
            # Create with a placeholder path; will be updated when log appears
            self._log_reader = IncrementalLogReader(paths.log_dir / "placeholder.snakemake.log")

    def init_validation(self) -> None:
        """Initialize validation if event file exists.

        Validation is automatically enabled when the logger plugin's event
        file is detected, allowing comparison between event-based and
        parsed state to find bugs in either approach.
        """
        # Close existing validation logger to prevent file handle leaks
        if self._validation_logger is not None:
            self._validation_logger.close()
            self._validation_logger = None

        event_file = get_event_file_path(self.workflow_dir)
        if event_file.exists():
            self._event_accumulator = EventAccumulator()
            self._validation_logger = ValidationLogger(self.workflow_dir)
            self._validation_logger.log_session_start()

    def validate_state(self, events: list[SnakeseeEvent], parsed: WorkflowProgress) -> None:
        """Compare event-based state with parsed state and log discrepancies.

        Args:
            events: New events to process.
            parsed: Current parsed workflow progress.
        """
        # Initialize validation if not yet done (event file may have appeared)
        if self._event_accumulator is None:
            self.init_validation()

        if self._event_accumulator is None or self._validation_logger is None:
            return

        # Accumulate new events
        self._event_accumulator.process_events(events)

        # Only compare if we have meaningful state from events
        if not self._event_accumulator.workflow_started:
            return

        # Compare states and log discrepancies
        discrepancies = compare_states(self._event_accumulator, parsed)

        if discrepancies:
            self._validation_logger.log_discrepancies(discrepancies)

        # Log summary periodically (every comparison for now)
        self._validation_logger.log_summary(self._event_accumulator, parsed)

    def read_new_events(self) -> list[SnakeseeEvent]:
        """Read new events from the event file if available.

        Returns:
            List of new events, or empty list if no events or event reading disabled.
        """
        if not self._events_enabled or self._event_reader is None:
            # Try to initialize if event file now exists (with validation)
            if self._events_enabled and self._event_reader is None:
                self.init_event_reader()

            if self._event_reader is None:
                return []

        return self._event_reader.read_new_events()

    # ------------------------------------------------------ event handlers
    def _handle_job_submitted_event(
        self,
        event: SnakeseeEvent,
        running_jobs: list[JobInfo],
    ) -> None:
        """Handle JOB_SUBMITTED event - track pending job with wildcards."""
        from snakesee.state.job_registry import Job
        from snakesee.state.job_registry import JobStatus

        if event.job_id is None:
            return
        job_id_str = str(event.job_id)

        # Create or update job in registry with SUBMITTED status
        existing_job = self._workflow_state.jobs.get_by_job_id(job_id_str)
        if existing_job is None:
            # Create new job with SUBMITTED status
            new_job = Job(
                key=job_id_str,
                rule=event.rule_name or "unknown",
                status=JobStatus.SUBMITTED,
                job_id=job_id_str,
                wildcards=dict(event.wildcards) if event.wildcards else {},
                threads=event.threads,
            )
            self._workflow_state.jobs.add(new_job)
        else:
            # Update existing job with submitted info
            if event.wildcards:
                existing_job.wildcards = dict(event.wildcards)
            if event.threads is not None:
                existing_job.threads = event.threads

        # Also store threads for backward compatibility
        if event.threads is not None:
            self._workflow_state.jobs.store_threads(job_id_str, event.threads)

        # Update running_jobs list if this job is already running
        registry_job = self._workflow_state.jobs.get_by_job_id(job_id_str)
        threads = event.threads or (registry_job.threads if registry_job else None)
        for i, job in enumerate(running_jobs):
            if job.job_id == job_id_str:
                running_jobs[i] = JobInfo(
                    rule=job.rule,
                    job_id=job.job_id,
                    start_time=job.start_time,
                    end_time=job.end_time,
                    output_file=job.output_file,
                    wildcards=event.wildcards_dict or job.wildcards,
                    input_size=job.input_size,
                    threads=threads,
                )
                break

    def _handle_job_queued_event(self, event: SnakeseeEvent) -> None:
        """Handle JOB_QUEUED event - mark a remote job as queued (awaiting a node).

        The registry already transitioned the job to QUEUED via ``apply_event``;
        here we just ensure a job record exists with the rule name and remote
        fields so it appears in the queued list. Queued jobs are deliberately not
        added to ``running_jobs`` — that is the whole point of the distinction.
        """
        from snakesee.state.job_registry import Job
        from snakesee.state.job_registry import JobStatus

        if event.job_id is None:
            return
        job_id_str = str(event.job_id)

        existing = self._workflow_state.jobs.get_by_job_id(job_id_str)
        if existing is None:
            new_job = Job(
                key=job_id_str,
                rule=event.rule_name or "unknown",
                status=JobStatus.QUEUED,
                job_id=job_id_str,
                wildcards=dict(event.wildcards) if event.wildcards else {},
                threads=event.threads,
                external_jobid=event.external_jobid,
                executor=event.executor,
                region=event.region,
                log_stream=event.log_stream,
                queued_at=event.queued_at if event.queued_at is not None else event.timestamp,
            )
            self._workflow_state.jobs.add(new_job)

    def _handle_job_started_event(
        self,
        event: SnakeseeEvent,
        running_jobs: list[JobInfo],
    ) -> None:
        """Handle JOB_STARTED event - transition from SUBMITTED to RUNNING."""
        from snakesee.state.job_registry import JobStatus

        if event.job_id is None:
            return
        job_id_str = str(event.job_id)

        # For a remote job the executor reports the true execution start; prefer it
        # over the event emission time so elapsed/duration/queue_wait exclude queue
        # wait. Local jobs have no started_at and fall back to the event timestamp.
        start = event.started_at if event.started_at is not None else event.timestamp

        # Transition job from SUBMITTED to RUNNING
        registry_job = self._workflow_state.jobs.get_by_job_id(job_id_str)
        if registry_job is not None:
            registry_job.start_time = start
            self._workflow_state.jobs.set_status(registry_job, JobStatus.RUNNING)

        threads = event.threads or (registry_job.threads if registry_job else None)
        for i, job in enumerate(running_jobs):
            if job.job_id == job_id_str:
                running_jobs[i] = JobInfo(
                    rule=job.rule,
                    job_id=job.job_id,
                    start_time=start,
                    end_time=job.end_time,
                    output_file=job.output_file,
                    wildcards=event.wildcards_dict or job.wildcards,
                    input_size=job.input_size,
                    threads=threads or job.threads,
                )
                break
        else:
            # Job wasn't already in the running list (e.g. the event arrived
            # before the next log parse saw it) - append a fresh entry so it
            # shows up as running immediately rather than after the next re-parse.
            rule = event.rule_name or (registry_job.rule if registry_job else "unknown")
            running_jobs.append(
                JobInfo(
                    rule=rule,
                    job_id=job_id_str,
                    start_time=start,
                    wildcards=event.wildcards_dict,
                    threads=threads,
                )
            )

    def _handle_job_finished_event(
        self,
        event: SnakeseeEvent,
        running_jobs: list[JobInfo],
        completions: list[JobInfo],
    ) -> None:
        """Handle JOB_FINISHED event - transition to COMPLETED.

        Mutates ``running_jobs`` to drop the finished job and ``completions`` to
        either patch the existing entry or append a new one. The previous version
        only patched ``completions``, which could leave the job in ``running_jobs``
        until the next log re-parse and miss it from completions entirely if the
        log parser hadn't seen the completion line yet.
        """
        from snakesee.state.job_registry import JobStatus

        if event.job_id is None or event.duration is None:
            return
        job_id_str = str(event.job_id)
        registry_job = self._workflow_state.jobs.get_by_job_id(job_id_str)

        # Transition job to COMPLETED
        if registry_job is not None:
            registry_job.end_time = event.timestamp
            self._workflow_state.jobs.set_status(registry_job, JobStatus.COMPLETED)

        threads = event.threads or (registry_job.threads if registry_job else None)

        # Drop the finished job from the running list immediately so the UI
        # doesn't keep showing it as running until the next log re-parse.
        running_jobs[:] = [job for job in running_jobs if job.job_id != job_id_str]

        for i, job in enumerate(completions):
            if job.job_id == job_id_str:
                completions[i] = JobInfo(
                    rule=job.rule,
                    job_id=job.job_id,
                    start_time=event.timestamp - event.duration,
                    end_time=event.timestamp,
                    output_file=job.output_file,
                    wildcards=job.wildcards,
                    input_size=job.input_size,
                    threads=threads or job.threads,
                )
                return

        # Job wasn't already in completions (e.g. event arrived before the next
        # log parse) - append a fresh entry so it shows up in recent completions.
        rule = event.rule_name or (registry_job.rule if registry_job else "unknown")
        completions.append(
            JobInfo(
                rule=rule,
                job_id=job_id_str,
                start_time=event.timestamp - event.duration,
                end_time=event.timestamp,
                wildcards=event.wildcards_dict,
                threads=threads,
            )
        )

    def _record_job_stats_from_event(self, event: SnakeseeEvent) -> None:
        """Record job stats to RuleRegistry from a JOB_FINISHED event.

        Uses JobRegistry to track which jobs have had stats recorded
        to avoid duplicates across poll cycles.
        """
        if event.job_id is None or event.duration is None or event.rule_name is None:
            return

        # Check if we've already recorded stats for this job
        job_key = str(event.job_id)
        job = self._workflow_state.jobs.get(job_key)
        if job is not None and job.stats_recorded:
            return

        # Get threads from event or JobRegistry
        threads = event.threads or (job.threads if job else None)

        # Record to RuleRegistry
        self._workflow_state.rules.record_completion(
            rule=event.rule_name,
            duration=event.duration,
            timestamp=event.timestamp,
            threads=threads,
            wildcards=event.wildcards_dict,
        )

        # For remote jobs, record the queue wait separately from execution time.
        # The registry job carries queued_at/start_time/queue (the event may not).
        if job is not None and job.queue_wait is not None:
            self._workflow_state.rules.record_queue_wait(
                event.rule_name, job.queue_wait, queue=job.queue
            )

        # Mark as recorded
        if job is not None:
            job.stats_recorded = True

    def _handle_job_error_event(
        self,
        event: SnakeseeEvent,
        failed_list: list[JobInfo],
    ) -> int:
        """Handle JOB_ERROR event - track failed job. Returns new failed count."""
        if event.job_id is None:
            return len(failed_list)
        job_id_str = str(event.job_id)
        if not any(j.job_id == job_id_str for j in failed_list):
            failed_list.append(
                JobInfo(
                    rule=event.rule_name or "unknown",
                    job_id=job_id_str,
                    start_time=event.timestamp - event.duration if event.duration else None,
                    end_time=event.timestamp,
                    wildcards=event.wildcards_dict,
                    threads=event.threads,
                )
            )
        return len(failed_list)

    def _compute_pending_jobs_from_scheduled(
        self,
        running_jobs: Sequence[JobInfo],
        completions: Sequence[JobInfo],
        failed_jobs: Sequence[JobInfo] | None = None,
    ) -> list[JobInfo]:
        """Compute pending jobs by subtracting running/completed/failed from all scheduled.

        When the snakesee logger plugin isn't available, we fall back to parsing
        all scheduled jobs from the snakemake log. This method computes which of
        those scheduled jobs are still pending (not yet running, completed, or failed).

        Args:
            running_jobs: Currently running jobs.
            completions: Completed jobs.
            failed_jobs: Failed jobs (to exclude from pending).

        Returns:
            List of pending jobs with their wildcards and threads.
        """
        if not self._all_scheduled_jobs:
            return []
        running_ids = {job.job_id for job in running_jobs if job.job_id}
        completed_ids = {job.job_id for job in completions if job.job_id}
        failed_ids = {job.job_id for job in (failed_jobs or []) if job.job_id}
        excluded_ids = running_ids | completed_ids | failed_ids
        return [
            job for job_id, job in self._all_scheduled_jobs.items() if job_id not in excluded_ids
        ]

    def apply_events_to_progress(
        self, progress: WorkflowProgress, events: list[SnakeseeEvent]
    ) -> WorkflowProgress:
        """Apply event updates to enhance progress accuracy.

        Events from the logger plugin provide more accurate timing and
        status information than log parsing. For remote executors this also
        populates ``queued_jobs_list`` (jobs awaiting a node) and keeps those
        jobs out of ``running_jobs``.

        Args:
            progress: The current workflow progress from parsing.
            events: New events from the logger plugin.

        Returns:
            Updated WorkflowProgress with event data applied.
        """
        # Track updates from events
        new_total = progress.total_jobs
        new_completed = progress.completed_jobs
        new_running_jobs = list(progress.running_jobs)
        new_completions = list(progress.recent_completions)

        # Process events FIRST to update registry state
        for event in events:
            # Route event through centralized JobRegistry (Phase 10)
            self._workflow_state.jobs.apply_event(event)

            if event.event_type == EventType.PROGRESS:
                if event.total_jobs is not None:
                    new_total = event.total_jobs
                    self._workflow_state.total_jobs = event.total_jobs
                if event.completed_jobs is not None:
                    new_completed = event.completed_jobs
            elif event.event_type == EventType.JOB_SUBMITTED:
                self._handle_job_submitted_event(event, new_running_jobs)
            elif event.event_type == EventType.JOB_QUEUED:
                self._handle_job_queued_event(event)
            elif event.event_type == EventType.JOB_STARTED:
                self._handle_job_started_event(event, new_running_jobs)
            elif event.event_type == EventType.JOB_FINISHED:
                self._handle_job_finished_event(event, new_running_jobs, new_completions)
                # Record stats to RuleRegistry for newly completed jobs
                self._record_job_stats_from_event(event)
            # JOB_ERROR is handled by apply_event above (updates registry)

        # AFTER events are processed, merge failed jobs from registry with log-parsed
        # Registry is source of truth (events), log parsing may miss some failures
        registry_failed = self._workflow_state.jobs.failed_job_infos()
        registry_failed_ids = {job.job_id for job in registry_failed if job.job_id}
        # Start with registry failed (authoritative), add any log-parsed that registry missed
        new_failed_list = list(registry_failed)
        for job in progress.failed_jobs_list:
            if job.job_id and job.job_id not in registry_failed_ids:
                new_failed_list.append(job)
        new_failed = len(new_failed_list)

        # Filter running jobs to exclude failed jobs (a job can't be both running and failed)
        failed_job_ids = {job.job_id for job in new_failed_list if job.job_id}
        new_running_jobs = [job for job in new_running_jobs if job.job_id not in failed_job_ids]

        # Apply stored threads/wildcards to running jobs that may have lost them
        # (log-parsed jobs may not have threads if the line order varies)
        for i, job in enumerate(new_running_jobs):
            if job.job_id and (job.threads is None or job.wildcards is None):
                registry_job = self._workflow_state.jobs.get_by_job_id(job.job_id)
                stored_threads = registry_job.threads if registry_job else None
                if job.threads is None and stored_threads is not None:
                    new_running_jobs[i] = JobInfo(
                        rule=job.rule,
                        job_id=job.job_id,
                        start_time=job.start_time,
                        end_time=job.end_time,
                        output_file=job.output_file,
                        wildcards=job.wildcards,
                        input_size=job.input_size,
                        threads=stored_threads,
                        log_file=job.log_file,
                    )

        # Get pending jobs from the registry (jobs submitted but not yet started)
        pending_jobs_list = self._workflow_state.jobs.submitted_job_infos()

        # Fallback: if no pending jobs from events, compute from log-based scheduled jobs
        if not pending_jobs_list:
            # Registry is the single source of truth (populated from events or log parsing)
            all_completed = self._workflow_state.jobs.completed_job_infos()
            pending_jobs_list = self._compute_pending_jobs_from_scheduled(
                new_running_jobs, all_completed, new_failed_list
            )

        # Remote jobs that are queued (submitted to the executor, awaiting a node)
        # are tracked separately so they don't masquerade as RUNNING. A job the log
        # parser thinks is "running" but the registry knows is QUEUED is filtered
        # out of the running list here.
        queued_jobs_list = self._workflow_state.jobs.queued_job_infos()
        queued_ids = {job.job_id for job in queued_jobs_list if job.job_id}
        if queued_ids:
            new_running_jobs = [j for j in new_running_jobs if j.job_id not in queued_ids]

        # Fill remote fields (external id, links, queue timing) onto running jobs
        # from the registry — log-parsed/started JobInfos don't carry them.
        new_running_jobs = [self._enrich_remote_fields(job) for job in new_running_jobs]

        # Return updated progress
        return WorkflowProgress(
            workflow_dir=progress.workflow_dir,
            status=progress.status,
            total_jobs=new_total,
            completed_jobs=new_completed,
            failed_jobs=new_failed,
            failed_jobs_list=new_failed_list,
            running_jobs=new_running_jobs,
            recent_completions=new_completions,
            pending_jobs_list=pending_jobs_list,
            queued_jobs_list=queued_jobs_list,
            start_time=progress.start_time,
            log_file=progress.log_file,
            total_cost_estimate=self._workflow_state.jobs.total_cost_estimate(),
        )

    def _enrich_remote_fields(self, job: JobInfo) -> JobInfo:
        """Return a copy of ``job`` with remote fields filled in from the registry.

        Log-parsed and event-constructed running JobInfos don't carry the external
        id / executor / region / log stream; the registry does. When the registry
        has them for this job and the JobInfo lacks them, merge them in so the
        running view can show the external id and links.
        """
        if job.job_id is None:
            return job
        registry_job = self._workflow_state.jobs.get_by_job_id(job.job_id)
        if registry_job is None or registry_job.external_jobid is None:
            return job
        from dataclasses import replace

        # Per-field merge that prefers values already on the JobInfo, so this never
        # clobbers job-specific data and also backfills any partially-missing fields.
        return replace(
            job,
            external_jobid=job.external_jobid or registry_job.external_jobid,
            executor=job.executor or registry_job.executor,
            region=job.region or registry_job.region,
            log_stream=job.log_stream or registry_job.log_stream,
            queue=job.queue or registry_job.queue,
            queued_at=job.queued_at if job.queued_at is not None else registry_job.queued_at,
            attempt=job.attempt if job.attempt is not None else registry_job.attempt,
            exit_code=job.exit_code if job.exit_code is not None else registry_job.exit_code,
            status_reason=job.status_reason or registry_job.status_reason,
            termination_category=job.termination_category or registry_job.termination_category,
            termination_source=job.termination_source or registry_job.termination_source,
            termination_confidence=(
                job.termination_confidence or registry_job.termination_confidence
            ),
            cost_estimate=(
                job.cost_estimate if job.cost_estimate is not None else registry_job.cost_estimate
            ),
        )

    def _update_rule_stats_from_completions(self, progress: WorkflowProgress) -> None:
        """Update rule_stats with newly completed jobs from registry.

        This handles log-parsed completions that don't go through the event path.
        Event-based completions are handled by _record_job_stats_from_event().
        Uses registry (not recent_completions) to ensure all completed jobs get stats recorded.
        """
        if self._estimator is None:
            return

        # Use registry completed jobs (single source of truth) instead of recent_completions
        # to ensure we don't miss any jobs
        for registry_job in self._workflow_state.jobs.completed():
            # Skip if already recorded (deduplication)
            if registry_job.stats_recorded:
                continue

            # Skip if we don't have a valid duration
            duration = registry_job.duration
            if duration is None:
                continue

            # Record stats to RuleRegistry
            self._workflow_state.rules.record_completion(
                rule=registry_job.rule,
                duration=duration,
                timestamp=registry_job.end_time or 0.0,
                threads=registry_job.threads,
                wildcards=dict(registry_job.wildcards) if registry_job.wildcards else None,
                input_size=registry_job.input_size,
            )

            # For remote jobs, record the queue wait separately from execution time.
            queue_wait = registry_job.queue_wait
            if queue_wait is not None:
                self._workflow_state.rules.record_queue_wait(
                    registry_job.rule, queue_wait, queue=registry_job.queue
                )

            # Mark as recorded for deduplication
            registry_job.stats_recorded = True

    # --------------------------------------------------- filter / sort helpers
    def filter_jobs(self, jobs: Sequence[JobInfo], filter_text: str | None) -> list[JobInfo]:
        """Filter jobs by rule name if filter is active.

        Args:
            jobs: Jobs to filter.
            filter_text: Text to filter by (case-insensitive substring match).

        Returns:
            Filtered list of jobs (all jobs if filter_text is empty).
        """
        if not filter_text:
            return list(jobs)

        return [j for j in jobs if filter_text.lower() in j.rule.lower()]

    def _get_job_id_column_width(self, jobs: list[JobInfo]) -> int:
        """Calculate column width needed for job IDs.

        Args:
            jobs: List of jobs to check for max job ID.

        Returns:
            Minimum column width needed to display all job IDs (minimum 2).
        """
        if not jobs:
            return 2
        max_id = 0
        for job in jobs:
            if job.job_id:
                try:
                    job_id_int = int(job.job_id)
                    max_id = max(max_id, job_id_int)
                except ValueError:
                    # Non-integer job ID, use string length
                    max_id = max(max_id, 10 ** (len(job.job_id) - 1))
        # Use row index as fallback if no job IDs
        if max_id == 0:
            max_id = len(jobs)
        return max(2, len(str(max_id)))

    def get_tool_progress(self, job: JobInfo) -> ToolProgress | None:
        """Get tool-specific progress for a running job.

        Results are cached for the TTL duration to avoid parsing job logs
        on every refresh cycle.

        Args:
            job: The running job to check.

        Returns:
            ToolProgress if parseable, None otherwise.
        """
        # Use job.log_file (parsed from snakemake log, keyed by job_id)
        if job.log_file is None:
            return None

        # Use job_id as cache key (unique per job run)
        cache_key = job.job_id if job.job_id else str(job.log_file)
        now = time.time()

        log_path = self.workflow_dir / job.log_file
        if not log_path.exists():
            self._tool_progress_cache[cache_key] = (now, 0.0, None)
            return None

        # Get current file mtime for cache invalidation
        try:
            current_mtime = log_path.stat().st_mtime
        except OSError:
            return None

        # Check cache validity - must be within TTL AND file unchanged
        if cache_key in self._tool_progress_cache:
            cached_time, cached_mtime, cached_progress = self._tool_progress_cache[cache_key]
            if now - cached_time < self._tool_progress_cache_ttl and cached_mtime >= current_mtime:
                return cached_progress

        # Parse and cache the result with current mtime.  Third-party plugins
        # may raise arbitrary exceptions; catch broadly so a bad plugin can
        # never crash the TUI.
        try:
            progress = parse_tool_progress(job.rule, log_path)
        except Exception:  # noqa: BLE001 - intentional broad catch to protect TUI from plugin errors
            logger.debug(
                "Failed to parse tool progress for %s: %s", job.rule, log_path, exc_info=True
            )
            progress = None
        self._tool_progress_cache[cache_key] = (now, current_mtime, progress)
        return progress

    def cleanup_tool_progress_cache(self) -> None:
        """Remove expired entries from tool progress cache.

        Should be called periodically to prevent unbounded memory growth
        in long-running workflows.
        """
        now = time.time()
        # Remove entries that have been stale for 10x the TTL
        max_age = self._tool_progress_cache_ttl * 10
        expired = [
            key
            for key, (cached_time, _, _) in self._tool_progress_cache.items()
            if now - cached_time > max_age
        ]
        for key in expired:
            del self._tool_progress_cache[key]

    def update_cache_ttl(self) -> None:
        """Update tool progress cache TTL based on current refresh rate.

        Called when refresh_rate changes to keep cache behavior in sync.
        """
        self._tool_progress_cache_ttl = min(
            ADAPTIVE_CACHE_TTL_MULTIPLIER * self.refresh_rate, MAX_CACHE_TTL
        )

    def _build_running_job_data(
        self, jobs: list[JobInfo]
    ) -> list[tuple[JobInfo, float | None, float | None, float | None, ToolProgress | None]]:
        """Build sortable data for running jobs."""
        job_data: list[
            tuple[JobInfo, float | None, float | None, float | None, ToolProgress | None]
        ] = []
        for job in jobs:
            elapsed = job.elapsed
            remaining: float | None = None
            tool_progress: ToolProgress | None = None

            if self._estimator is not None:
                # Use wildcard+thread-aware ETA when available
                expected, variance = self._estimator.get_estimate_for_job(
                    rule=job.rule,
                    wildcards=job.wildcards,
                    threads=job.threads,
                )
                if elapsed is None:
                    # No start time yet - use expected duration as remaining estimate
                    remaining = expected
                elif elapsed <= expected:
                    remaining = expected - elapsed
                else:
                    # Job running longer than expected - use variance to estimate
                    std_dev = math.sqrt(variance) if variance > 0 else expected * 0.5
                    if elapsed <= expected + 2 * std_dev:
                        # Within reasonable variance - assume nearly done
                        remaining = 0.0
                    else:
                        # Far outside expected range - estimate based on elapsed time
                        # Assume job is ~60% done (heuristic for long-running jobs)
                        # This gives a rough estimate rather than "unknown"
                        remaining = elapsed * 0.67  # ~40% more time expected

            # Try to get tool-specific progress
            tool_progress = self.get_tool_progress(job)

            # If we have tool progress with percentage, use it to improve ETA
            if tool_progress is not None and tool_progress.percent_complete is not None:
                if elapsed is not None and tool_progress.percent_complete > 0:
                    # Estimate remaining time based on progress
                    pct = tool_progress.percent_complete / 100.0
                    tool_remaining = elapsed * (1 - pct) / pct if pct > 0 else None
                    # Prefer tool-based estimate if available
                    if tool_remaining is not None:
                        remaining = tool_remaining

            job_data.append((job, elapsed, remaining, job.start_time, tool_progress))
        return job_data

    def _sort_running_job_data(
        self,
        job_data: list[
            tuple[JobInfo, float | None, float | None, float | None, ToolProgress | None]
        ],
        sort_column: int,
        sort_ascending: bool,
    ) -> list[tuple[JobInfo, float | None, float | None, float | None, ToolProgress | None]]:
        """Sort running job data based on the given sort settings."""
        if not job_data:
            return job_data
        sort_keys = {
            0: lambda x: x[0].rule.lower(),
            1: lambda x: x[3] or 0,
            2: lambda x: x[1] or 0,
            3: lambda x: x[2] if x[2] is not None else float("inf"),
        }
        key_fn = sort_keys.get(sort_column, sort_keys[0])
        return sorted(job_data, key=key_fn, reverse=not sort_ascending)

    def get_completions_sorted(
        self,
        progress: WorkflowProgress,
        *,
        filter_text: str | None,
        sort_table: SortTableName | None,
        sort_column: int,
        sort_ascending: bool,
        limit: int | None = None,
    ) -> tuple[list[JobInfo], set[int], list[str]]:
        """Get merged, filtered, and sorted completions + failed jobs.

        When ``limit`` is provided and no selection mode is active, uses
        :func:`heapq` for ``O(n * log(limit))`` top-N selection instead of a
        full ``O(n * log(n))`` sort.

        Args:
            progress: Current workflow progress.
            filter_text: Optional filter text (case-insensitive substring match on rule name).
            sort_table: Currently sorted table name (only "completions" triggers custom sort).
            sort_column: 0-indexed column to sort by when ``sort_table == "completions"``.
            sort_ascending: Sort direction when sorting.
            limit: If set, return at most this many items using heap selection.
                When None, returns the full sorted list.

        Returns:
            Tuple of (sorted jobs list, set of failed job ids, list of unique rule names
            for filter navigation).
        """
        failed_job_ids = {id(job) for job in progress.failed_jobs_list}

        # Determine effective sort key and direction
        is_sorting = sort_table == "completions"
        if is_sorting:
            sort_keys: dict[int, Any] = {
                0: lambda j: j.rule.lower(),
                1: lambda j: j.threads or 0,
                2: lambda j: j.duration or 0,
                3: lambda j: j.end_time or 0,
            }
            key_fn = sort_keys.get(sort_column, sort_keys[3])
            descending = not sort_ascending
        else:
            key_fn = lambda j: j.end_time or 0  # noqa: E731
            descending = True

        # Stream completions + failures through filter into heap selection to
        # avoid materializing the full merged list on the hot path.
        merged: Iterable[JobInfo] = itertools.chain(
            progress.recent_completions, progress.failed_jobs_list
        )
        filter_lower = filter_text.lower() if filter_text else ""
        if filter_lower:
            merged = (j for j in merged if filter_lower in j.rule.lower())

        # Use heap selection when we only need the top N items
        if limit is not None:
            if descending:
                jobs = heapq.nlargest(limit, merged, key=key_fn)
            else:
                jobs = heapq.nsmallest(limit, merged, key=key_fn)
        else:
            jobs = sorted(merged, key=key_fn, reverse=descending)

        # Compute filter matches for n/N navigation (preserve insertion order)
        filter_matches = list(dict.fromkeys(j.rule for j in jobs)) if filter_lower else []

        return jobs, failed_job_ids, filter_matches

    def get_completions_list(
        self,
        progress: WorkflowProgress,
        *,
        filter_text: str | None,
        sort_table: SortTableName | None,
        sort_column: int,
        sort_ascending: bool,
    ) -> tuple[list[JobInfo], set[int]]:
        """Get merged list of completed and failed jobs with same order as table.

        Applies the same filtering and sorting as ``_make_completions_table()`` to
        ensure the selected index matches between the table display and log panel.

        Returns:
            Tuple of (jobs_list, failed_job_ids_set).
        """
        # Always return the full sorted list (used for index-based selection)
        jobs, failed_job_ids, _ = self.get_completions_sorted(
            progress,
            filter_text=filter_text,
            sort_table=sort_table,
            sort_column=sort_column,
            sort_ascending=sort_ascending,
        )
        return jobs, failed_job_ids

    def get_running_jobs_list(
        self,
        progress: WorkflowProgress,
        *,
        filter_text: str | None,
        sort_table: SortTableName | None,
        sort_column: int,
        sort_ascending: bool,
    ) -> list[JobInfo]:
        """Get running jobs list with same order as table.

        Applies the same filtering and sorting as ``_make_running_table()`` to
        ensure the selected index matches between the table display and log panel.

        Returns:
            List of running jobs in display order.
        """
        jobs = self.filter_jobs(progress.running_jobs, filter_text)

        # Apply custom sorting if running table is being sorted
        if sort_table == "running" and jobs:
            # Build job data tuples for sorting
            job_data = self._build_running_job_data(jobs)
            job_data = self._sort_running_job_data(job_data, sort_column, sort_ascending)
            # Extract just the jobs from the sorted tuples
            jobs = [jd[0] for jd in job_data]

        return jobs

    # ---------------------------------------------------- pending / stats
    def get_inferred_pending_rules(self, progress: WorkflowProgress) -> dict[str, int] | None:
        """Get inferred pending rules from completions and historical data."""
        if not self._estimator:
            return None

        # Registry is the single source of truth (populated from events or log parsing)
        all_completed = self._workflow_state.jobs.completed_job_infos()

        # If we have expected job counts, we can infer pending even without completions
        if not all_completed and not self._estimator.expected_job_counts:
            return None

        # Count completed jobs by rule (using ALL completed, not just recent)
        completed_by_rule: dict[str, int] = {}
        for job in all_completed:
            completed_by_rule[job.rule] = completed_by_rule.get(job.rule, 0) + 1

        # Count running jobs by rule
        running_by_rule: dict[str, int] = {}
        for job in progress.running_jobs:
            running_by_rule[job.rule] = running_by_rule.get(job.rule, 0) + 1

        # Only augment with historical counts if we don't have expected_job_counts
        if not self._estimator.expected_job_counts and self._estimator.rule_stats:
            for rule, stats in self._estimator.rule_stats.items():
                if rule not in completed_by_rule:
                    completed_by_rule[rule] = stats.count

        return self._estimator._infer_pending_rules(
            completed_by_rule, progress.pending_jobs, self._estimator.current_rules, running_by_rule
        )

    def _parse_stats_from_logs(self, cutoff: float) -> dict[str, RuleTimingStats]:
        """Parse rule stats from log files created before the cutoff time."""
        from snakesee.parser import parse_completed_jobs_from_log

        stats_dict: dict[str, RuleTimingStats] = {}
        for log in self._available_logs:
            try:
                # Use st_mtime for cross-platform consistency (st_ctime is
                # inode-change time on POSIX, creation time on Windows).
                if log.stat().st_mtime >= cutoff:
                    continue
                for job in parse_completed_jobs_from_log(log):
                    if job.duration is not None:
                        if job.rule not in stats_dict:
                            stats_dict[job.rule] = RuleTimingStats(rule=job.rule)
                        stats_dict[job.rule].durations.append(job.duration)
            except OSError:
                continue
        return stats_dict

    def get_filtered_stats(self) -> list[RuleTimingStats]:
        """Get rule stats filtered by cutoff time if viewing historical log."""
        from snakesee.parser import parse_metadata_files

        if self._cutoff_time is None:
            # Latest log: use stats from estimator, filtered by current workflow rules
            if self._estimator and self._estimator.rule_stats:
                current_rules = self._estimator.current_rules
                if current_rules is not None:
                    return [
                        stats
                        for stats in self._estimator.rule_stats.values()
                        if stats.rule in current_rules
                    ]
                return list(self._estimator.rule_stats.values())
            return []

        # Historical log: rebuild stats from metadata, filtering by cutoff time
        metadata_dir = self.workflow_dir / ".snakemake" / "metadata"
        stats_dict: dict[str, RuleTimingStats] = {}
        for job in parse_metadata_files(metadata_dir):
            if job.duration is not None and job.end_time is not None:
                if job.end_time < self._cutoff_time:
                    if job.rule not in stats_dict:
                        stats_dict[job.rule] = RuleTimingStats(rule=job.rule)
                    stats_dict[job.rule].durations.append(job.duration)

        # If no metadata found, parse stats from log files up to the cutoff
        if not stats_dict:
            stats_dict = self._parse_stats_from_logs(self._cutoff_time)

        return list(stats_dict.values())

    # ------------------------------------------------------------- log tail
    def read_log_tail(self, log_path: Path, max_lines: int = 500) -> list[str]:
        """Read the last N lines of a log file efficiently.

        For large files, seeks near the end instead of reading the entire file.

        Args:
            log_path: Path to the log file.
            max_lines: Maximum number of lines to read.

        Returns:
            List of lines (most recent at end).
        """
        # Average bytes per line estimate for seeking
        BYTES_PER_LINE_ESTIMATE = 120

        try:
            # Check if cache is still valid
            stat = log_path.stat()
            mtime = stat.st_mtime
            file_size = stat.st_size
            if (
                self._cached_log_path == log_path
                and self._cached_log_mtime == mtime
                and self._cached_log_lines
            ):
                return self._cached_log_lines

            # For small files, just read the whole thing
            if file_size < BYTES_PER_LINE_ESTIMATE * max_lines * 2:
                content = log_path.read_text(errors="ignore")
                lines = content.splitlines()
            else:
                # For large files, seek near the end to avoid reading everything
                # Read extra bytes to ensure we get enough lines
                seek_bytes = BYTES_PER_LINE_ESTIMATE * max_lines * 2
                with open(log_path, "rb") as f:
                    # Seek to near the end
                    f.seek(max(0, file_size - seek_bytes))
                    # Read to end
                    content = f.read().decode("utf-8", errors="ignore")
                    lines = content.splitlines()
                    # Skip first line (likely partial from seek)
                    if lines and file_size > seek_bytes:
                        lines = lines[1:]

            # Take last max_lines
            result = lines[-max_lines:] if len(lines) > max_lines else lines

            # Update cache
            self._cached_log_path = log_path
            self._cached_log_mtime = mtime
            self._cached_log_lines = result

            return result
        except OSError:
            return ["[Error reading log file]"]

    # ----------------------------------------------------- cutoff / poll
    def get_cutoff_time(self) -> float | None:
        """Get the cutoff time for filtering (when the next log started)."""
        if self._current_log_index == 0:
            return None  # Latest log, no cutoff
        if self._current_log_index > 0 and len(self._available_logs) > 1:
            # Cutoff is the start of the next newer log. Use st_mtime (cross-platform)
            # rather than st_ctime, which is inode-change time on POSIX.
            next_log_index = self._current_log_index - 1
            if next_log_index >= 0:
                try:
                    return self._available_logs[next_log_index].stat().st_mtime
                except OSError:
                    pass
        return None

    def poll_state(self) -> tuple[WorkflowProgress, TimeEstimate | None]:
        """Poll the current workflow state and estimate.

        For the latest run (``current_log_index == 0``) we merge live events and
        the in-memory ``JobRegistry`` into the parsed log to enrich timing and
        catch jobs the parser missed. For historical runs we deliberately skip
        all of that — events and the registry describe the *current* run and
        would otherwise leak into older log views.

        Returns:
            Tuple of (workflow progress, optional time estimate).
        """
        # Refresh log list if viewing latest
        if self._current_log_index == 0:
            self.refresh_log_list()

        is_latest_view = self._current_log_index == 0

        # Get the selected log file and cutoff time for historical view
        log_file = self.get_current_log() if not is_latest_view else None
        self._cutoff_time = self.get_cutoff_time()

        # Live events / readers only apply to the latest run.
        events = self.read_new_events() if is_latest_view else []
        reader = self._log_reader if is_latest_view else None

        progress = parse_workflow_state(
            self.workflow_dir,
            log_file=log_file,
            cutoff_time=self._cutoff_time,
            log_reader=reader,
        )

        # Sync log reader's completed jobs to registry when events aren't available
        # This ensures the registry is the single source of truth regardless of
        # whether the snakesee logger plugin is being used
        if is_latest_view and not events and self._log_reader:
            for job in self._log_reader.completed_jobs:
                if job.job_id:
                    self._workflow_state.jobs.apply_job_info(job, key=job.job_id)

        # Validate: compare event-based state with parsed state (before applying)
        # This logs discrepancies to help find bugs in either approach
        if is_latest_view and events:
            self.validate_state(events, progress)

        # Apply events to enhance progress accuracy
        if is_latest_view and events:
            progress = self.apply_events_to_progress(progress, events)
        elif is_latest_view:
            # Even without new events, merge registry-tracked failed jobs into progress
            # This ensures failed jobs discovered via earlier events are not lost
            # when log re-parsing misses them
            registry_failed = self._workflow_state.jobs.failed_job_infos()
            if registry_failed:
                from dataclasses import replace

                registry_failed_ids = {job.job_id for job in registry_failed if job.job_id}
                merged_failed = list(registry_failed)
                for job in progress.failed_jobs_list:
                    if job.job_id and job.job_id not in registry_failed_ids:
                        merged_failed.append(job)
                progress = replace(
                    progress,
                    failed_jobs=len(merged_failed),
                    failed_jobs_list=merged_failed,
                )

        # Always populate pending_jobs_list from log-based scheduled jobs
        # (even when no new events, we need this for wildcard-conditioned ETA).
        # Historical views derive pending purely from the parsed progress so
        # the live registry can't bleed into a finished run.
        if is_latest_view and not progress.pending_jobs_list:
            from dataclasses import replace

            # Registry is the single source of truth (populated from events or log parsing)
            all_completed = self._workflow_state.jobs.completed_job_infos()
            pending_jobs_list = self._compute_pending_jobs_from_scheduled(
                progress.running_jobs, all_completed, progress.failed_jobs_list
            )
            if pending_jobs_list:
                progress = replace(progress, pending_jobs_list=pending_jobs_list)

        # Infer total_jobs from the Job stats table when the progress line hasn't
        # appeared yet (it only emerges after the first completion). This lets the
        # pending panel show correct counts immediately on a fresh run.
        #
        # Scoped to the latest/live run (log index 0): expected_job_counts is parsed
        # from the latest log, so inferring it onto a historical run's progress would
        # graft the live run's totals onto the wrong run.
        if (
            progress.total_jobs == 0
            and self._estimator is not None
            and self._current_log_index == 0
        ):
            if not self._estimator.expected_job_counts:
                self._init_current_rules_from_log()
            if self._estimator.expected_job_counts:
                from dataclasses import replace

                inferred_total = sum(self._estimator.expected_job_counts.values())
                if inferred_total > 0:
                    progress = replace(progress, total_jobs=inferred_total)

        # Update rule_stats with newly completed jobs (for Rule Statistics panel)
        self._update_rule_stats_from_completions(progress)

        estimate = None
        if self._estimator is not None:
            estimate = self._estimator.estimate_remaining(progress)

        # Periodically clean up stale cache entries
        self.cleanup_tool_progress_cache()

        return progress, estimate

Attributes

available_log_count property

available_log_count: int

Number of historical log files currently discovered.

current_log_index property writable

current_log_index: int

Index into available_logs; 0 = most recent.

event_reader property

event_reader: EventReader | None

Event reader for the current workflow run, if any.

use_wildcard_conditioning property writable

use_wildcard_conditioning: bool

Whether wildcard-conditioned estimates are enabled.

Methods:

__init__

__init__(workflow_dir: Path, refresh_rate: float = DEFAULT_REFRESH_RATE, use_estimation: bool = True, profile_path: Path | None = None, use_wildcard_conditioning: bool = True, weighting_strategy: WeightingStrategy = 'index', half_life_logs: int = 10, half_life_days: float = 7.0) -> None

Initialize the data source.

Parameters:

Name	Type	Description	Default
workflow_dir	Path	Path to workflow directory containing .snakemake/.	required
refresh_rate	float	Refresh interval in seconds (used to size cache TTL).	DEFAULT_REFRESH_RATE
use_estimation	bool	Whether to enable time estimation.	True
profile_path	Path | None	Optional path to a timing profile for bootstrapping estimates.	None
use_wildcard_conditioning	bool	Whether to enable wildcard-conditioned estimates.	True
weighting_strategy	WeightingStrategy	Strategy for weighting historical data ("index" or "time").	'index'
half_life_logs	int	Half-life in run count for index-based weighting.	10
half_life_days	float	Half-life in days for time-based weighting.	7.0

Source code in snakesee/tui/data_source.py

def __init__(
    self,
    workflow_dir: Path,
    refresh_rate: float = DEFAULT_REFRESH_RATE,
    use_estimation: bool = True,
    profile_path: Path | None = None,
    use_wildcard_conditioning: bool = True,
    weighting_strategy: WeightingStrategy = "index",
    half_life_logs: int = 10,
    half_life_days: float = 7.0,
) -> None:
    """Initialize the data source.

    Args:
        workflow_dir: Path to workflow directory containing ``.snakemake/``.
        refresh_rate: Refresh interval in seconds (used to size cache TTL).
        use_estimation: Whether to enable time estimation.
        profile_path: Optional path to a timing profile for bootstrapping estimates.
        use_wildcard_conditioning: Whether to enable wildcard-conditioned estimates.
        weighting_strategy: Strategy for weighting historical data ("index" or "time").
        half_life_logs: Half-life in run count for index-based weighting.
        half_life_days: Half-life in days for time-based weighting.
    """
    self.workflow_dir = workflow_dir
    self.refresh_rate = refresh_rate
    self.use_estimation = use_estimation
    self.profile_path = profile_path
    self.weighting_strategy = weighting_strategy
    self.half_life_logs = half_life_logs
    self.half_life_days = half_life_days

    self._use_wildcard_conditioning: bool = use_wildcard_conditioning

    self._estimator: TimeEstimator | None = None

    # Log file navigation
    self._available_logs: list[Path] = []
    self._current_log_index: int = 0  # 0 = most recent
    self._latest_log_path: Path | None = None  # Track latest log to detect new workflows
    self.refresh_log_list()

    # Cutoff time for historical view (updated in poll_state)
    self._cutoff_time: float | None = None

    # Cached log tail data
    self._cached_log_path: Path | None = None
    self._cached_log_lines: list[str] = []
    self._cached_log_mtime: float = 0

    # Tool progress cache (to avoid parsing job logs on every refresh).
    # Cache stores: (cached_time, file_mtime, progress) - invalidates if file changes.
    self._tool_progress_cache: dict[str, tuple[float, float, ToolProgress | None]] = {}
    # Adaptive TTL: scales with refresh rate to avoid cache outliving refresh cycles.
    self._tool_progress_cache_ttl: float = min(
        ADAPTIVE_CACHE_TTL_MULTIPLIER * refresh_rate, MAX_CACHE_TTL
    )

    # Event reader for real-time events from logger plugin
    self._event_reader: EventReader | None = None
    self._events_enabled: bool = True
    self.init_event_reader()

    # All scheduled jobs from log (for pending job estimation without logger plugin)
    self._all_scheduled_jobs: dict[str, JobInfo] = {}

    # Incremental log reader for efficient polling
    self._log_reader: IncrementalLogReader | None = None
    self.init_log_reader()

    # Validation: compare event-based state with parsed state
    self._event_accumulator: EventAccumulator | None = None
    self._validation_logger: ValidationLogger | None = None
    self.init_validation()

    # Centralized workflow state
    self._workflow_state: WorkflowState = WorkflowState.create(
        workflow_dir=workflow_dir,
    )

    self.init_estimator()

apply_events_to_progress

apply_events_to_progress(progress: WorkflowProgress, events: list[SnakeseeEvent]) -> WorkflowProgress

Apply event updates to enhance progress accuracy.

Events from the logger plugin provide more accurate timing and status information than log parsing. For remote executors this also populates queued_jobs_list (jobs awaiting a node) and keeps those jobs out of running_jobs.

Parameters:

Name	Type	Description	Default
progress	WorkflowProgress	The current workflow progress from parsing.	required
events	list[SnakeseeEvent]	New events from the logger plugin.	required

Returns:

Type	Description
WorkflowProgress	Updated WorkflowProgress with event data applied.

Source code in snakesee/tui/data_source.py

def apply_events_to_progress(
    self, progress: WorkflowProgress, events: list[SnakeseeEvent]
) -> WorkflowProgress:
    """Apply event updates to enhance progress accuracy.

    Events from the logger plugin provide more accurate timing and
    status information than log parsing. For remote executors this also
    populates ``queued_jobs_list`` (jobs awaiting a node) and keeps those
    jobs out of ``running_jobs``.

    Args:
        progress: The current workflow progress from parsing.
        events: New events from the logger plugin.

    Returns:
        Updated WorkflowProgress with event data applied.
    """
    # Track updates from events
    new_total = progress.total_jobs
    new_completed = progress.completed_jobs
    new_running_jobs = list(progress.running_jobs)
    new_completions = list(progress.recent_completions)

    # Process events FIRST to update registry state
    for event in events:
        # Route event through centralized JobRegistry (Phase 10)
        self._workflow_state.jobs.apply_event(event)

        if event.event_type == EventType.PROGRESS:
            if event.total_jobs is not None:
                new_total = event.total_jobs
                self._workflow_state.total_jobs = event.total_jobs
            if event.completed_jobs is not None:
                new_completed = event.completed_jobs
        elif event.event_type == EventType.JOB_SUBMITTED:
            self._handle_job_submitted_event(event, new_running_jobs)
        elif event.event_type == EventType.JOB_QUEUED:
            self._handle_job_queued_event(event)
        elif event.event_type == EventType.JOB_STARTED:
            self._handle_job_started_event(event, new_running_jobs)
        elif event.event_type == EventType.JOB_FINISHED:
            self._handle_job_finished_event(event, new_running_jobs, new_completions)
            # Record stats to RuleRegistry for newly completed jobs
            self._record_job_stats_from_event(event)
        # JOB_ERROR is handled by apply_event above (updates registry)

    # AFTER events are processed, merge failed jobs from registry with log-parsed
    # Registry is source of truth (events), log parsing may miss some failures
    registry_failed = self._workflow_state.jobs.failed_job_infos()
    registry_failed_ids = {job.job_id for job in registry_failed if job.job_id}
    # Start with registry failed (authoritative), add any log-parsed that registry missed
    new_failed_list = list(registry_failed)
    for job in progress.failed_jobs_list:
        if job.job_id and job.job_id not in registry_failed_ids:
            new_failed_list.append(job)
    new_failed = len(new_failed_list)

    # Filter running jobs to exclude failed jobs (a job can't be both running and failed)
    failed_job_ids = {job.job_id for job in new_failed_list if job.job_id}
    new_running_jobs = [job for job in new_running_jobs if job.job_id not in failed_job_ids]

    # Apply stored threads/wildcards to running jobs that may have lost them
    # (log-parsed jobs may not have threads if the line order varies)
    for i, job in enumerate(new_running_jobs):
        if job.job_id and (job.threads is None or job.wildcards is None):
            registry_job = self._workflow_state.jobs.get_by_job_id(job.job_id)
            stored_threads = registry_job.threads if registry_job else None
            if job.threads is None and stored_threads is not None:
                new_running_jobs[i] = JobInfo(
                    rule=job.rule,
                    job_id=job.job_id,
                    start_time=job.start_time,
                    end_time=job.end_time,
                    output_file=job.output_file,
                    wildcards=job.wildcards,
                    input_size=job.input_size,
                    threads=stored_threads,
                    log_file=job.log_file,
                )

    # Get pending jobs from the registry (jobs submitted but not yet started)
    pending_jobs_list = self._workflow_state.jobs.submitted_job_infos()

    # Fallback: if no pending jobs from events, compute from log-based scheduled jobs
    if not pending_jobs_list:
        # Registry is the single source of truth (populated from events or log parsing)
        all_completed = self._workflow_state.jobs.completed_job_infos()
        pending_jobs_list = self._compute_pending_jobs_from_scheduled(
            new_running_jobs, all_completed, new_failed_list
        )

    # Remote jobs that are queued (submitted to the executor, awaiting a node)
    # are tracked separately so they don't masquerade as RUNNING. A job the log
    # parser thinks is "running" but the registry knows is QUEUED is filtered
    # out of the running list here.
    queued_jobs_list = self._workflow_state.jobs.queued_job_infos()
    queued_ids = {job.job_id for job in queued_jobs_list if job.job_id}
    if queued_ids:
        new_running_jobs = [j for j in new_running_jobs if j.job_id not in queued_ids]

    # Fill remote fields (external id, links, queue timing) onto running jobs
    # from the registry — log-parsed/started JobInfos don't carry them.
    new_running_jobs = [self._enrich_remote_fields(job) for job in new_running_jobs]

    # Return updated progress
    return WorkflowProgress(
        workflow_dir=progress.workflow_dir,
        status=progress.status,
        total_jobs=new_total,
        completed_jobs=new_completed,
        failed_jobs=new_failed,
        failed_jobs_list=new_failed_list,
        running_jobs=new_running_jobs,
        recent_completions=new_completions,
        pending_jobs_list=pending_jobs_list,
        queued_jobs_list=queued_jobs_list,
        start_time=progress.start_time,
        log_file=progress.log_file,
        total_cost_estimate=self._workflow_state.jobs.total_cost_estimate(),
    )

build_running_job_data

build_running_job_data(jobs: list[JobInfo]) -> list[tuple[JobInfo, float | None, float | None, float | None, ToolProgress | None]]

Build per-job tuples of (job, elapsed, remaining, start_time, tool_progress).

Source code in snakesee/tui/data_source.py

def build_running_job_data(
    self, jobs: list[JobInfo]
) -> list[tuple[JobInfo, float | None, float | None, float | None, ToolProgress | None]]:
    """Build per-job tuples of (job, elapsed, remaining, start_time, tool_progress)."""
    return self._build_running_job_data(jobs)

cleanup_tool_progress_cache

cleanup_tool_progress_cache() -> None

Remove expired entries from tool progress cache.

Should be called periodically to prevent unbounded memory growth in long-running workflows.

Source code in snakesee/tui/data_source.py

def cleanup_tool_progress_cache(self) -> None:
    """Remove expired entries from tool progress cache.

    Should be called periodically to prevent unbounded memory growth
    in long-running workflows.
    """
    now = time.time()
    # Remove entries that have been stale for 10x the TTL
    max_age = self._tool_progress_cache_ttl * 10
    expired = [
        key
        for key, (cached_time, _, _) in self._tool_progress_cache.items()
        if now - cached_time > max_age
    ]
    for key in expired:
        del self._tool_progress_cache[key]

cost_by_rule

cost_by_rule() -> dict[str, float]

Return summed estimated cost per rule (empty when no cost data).

Source code in snakesee/tui/data_source.py

def cost_by_rule(self) -> dict[str, float]:
    """Return summed estimated cost per rule (empty when no cost data)."""
    return self._workflow_state.jobs.cost_by_rule()

filter_jobs

filter_jobs(jobs: Sequence[JobInfo], filter_text: str | None) -> list[JobInfo]

Filter jobs by rule name if filter is active.

Parameters:

Name	Type	Description	Default
jobs	Sequence[JobInfo]	Jobs to filter.	required
filter_text	str | None	Text to filter by (case-insensitive substring match).	required

Returns:

Type	Description
list[JobInfo]	Filtered list of jobs (all jobs if filter_text is empty).

Source code in snakesee/tui/data_source.py

def filter_jobs(self, jobs: Sequence[JobInfo], filter_text: str | None) -> list[JobInfo]:
    """Filter jobs by rule name if filter is active.

    Args:
        jobs: Jobs to filter.
        filter_text: Text to filter by (case-insensitive substring match).

    Returns:
        Filtered list of jobs (all jobs if filter_text is empty).
    """
    if not filter_text:
        return list(jobs)

    return [j for j in jobs if filter_text.lower() in j.rule.lower()]

get_completions_list

get_completions_list(progress: WorkflowProgress, *, filter_text: str | None, sort_table: SortTableName | None, sort_column: int, sort_ascending: bool) -> tuple[list[JobInfo], set[int]]

Get merged list of completed and failed jobs with same order as table.

Applies the same filtering and sorting as _make_completions_table() to ensure the selected index matches between the table display and log panel.

Returns:

Type	Description
tuple[list[JobInfo], set[int]]	Tuple of (jobs_list, failed_job_ids_set).

Source code in snakesee/tui/data_source.py

def get_completions_list(
    self,
    progress: WorkflowProgress,
    *,
    filter_text: str | None,
    sort_table: SortTableName | None,
    sort_column: int,
    sort_ascending: bool,
) -> tuple[list[JobInfo], set[int]]:
    """Get merged list of completed and failed jobs with same order as table.

    Applies the same filtering and sorting as ``_make_completions_table()`` to
    ensure the selected index matches between the table display and log panel.

    Returns:
        Tuple of (jobs_list, failed_job_ids_set).
    """
    # Always return the full sorted list (used for index-based selection)
    jobs, failed_job_ids, _ = self.get_completions_sorted(
        progress,
        filter_text=filter_text,
        sort_table=sort_table,
        sort_column=sort_column,
        sort_ascending=sort_ascending,
    )
    return jobs, failed_job_ids

get_completions_sorted

get_completions_sorted(progress: WorkflowProgress, *, filter_text: str | None, sort_table: SortTableName | None, sort_column: int, sort_ascending: bool, limit: int | None = None) -> tuple[list[JobInfo], set[int], list[str]]

Get merged, filtered, and sorted completions + failed jobs.

When limit is provided and no selection mode is active, uses :func:heapq for O(n * log(limit)) top-N selection instead of a full O(n * log(n)) sort.

Parameters:

Name	Type	Description	Default
progress	WorkflowProgress	Current workflow progress.	required
filter_text	str | None	Optional filter text (case-insensitive substring match on rule name).	required
sort_table	SortTableName | None	Currently sorted table name (only "completions" triggers custom sort).	required
sort_column	int	0-indexed column to sort by when sort_table == "completions".	required
sort_ascending	bool	Sort direction when sorting.	required
limit	int | None	If set, return at most this many items using heap selection. When None, returns the full sorted list.	None

Returns:

Type	Description
list[JobInfo]	Tuple of (sorted jobs list, set of failed job ids, list of unique rule names
set[int]	for filter navigation).

Source code in snakesee/tui/data_source.py

def get_completions_sorted(
    self,
    progress: WorkflowProgress,
    *,
    filter_text: str | None,
    sort_table: SortTableName | None,
    sort_column: int,
    sort_ascending: bool,
    limit: int | None = None,
) -> tuple[list[JobInfo], set[int], list[str]]:
    """Get merged, filtered, and sorted completions + failed jobs.

    When ``limit`` is provided and no selection mode is active, uses
    :func:`heapq` for ``O(n * log(limit))`` top-N selection instead of a
    full ``O(n * log(n))`` sort.

    Args:
        progress: Current workflow progress.
        filter_text: Optional filter text (case-insensitive substring match on rule name).
        sort_table: Currently sorted table name (only "completions" triggers custom sort).
        sort_column: 0-indexed column to sort by when ``sort_table == "completions"``.
        sort_ascending: Sort direction when sorting.
        limit: If set, return at most this many items using heap selection.
            When None, returns the full sorted list.

    Returns:
        Tuple of (sorted jobs list, set of failed job ids, list of unique rule names
        for filter navigation).
    """
    failed_job_ids = {id(job) for job in progress.failed_jobs_list}

    # Determine effective sort key and direction
    is_sorting = sort_table == "completions"
    if is_sorting:
        sort_keys: dict[int, Any] = {
            0: lambda j: j.rule.lower(),
            1: lambda j: j.threads or 0,
            2: lambda j: j.duration or 0,
            3: lambda j: j.end_time or 0,
        }
        key_fn = sort_keys.get(sort_column, sort_keys[3])
        descending = not sort_ascending
    else:
        key_fn = lambda j: j.end_time or 0  # noqa: E731
        descending = True

    # Stream completions + failures through filter into heap selection to
    # avoid materializing the full merged list on the hot path.
    merged: Iterable[JobInfo] = itertools.chain(
        progress.recent_completions, progress.failed_jobs_list
    )
    filter_lower = filter_text.lower() if filter_text else ""
    if filter_lower:
        merged = (j for j in merged if filter_lower in j.rule.lower())

    # Use heap selection when we only need the top N items
    if limit is not None:
        if descending:
            jobs = heapq.nlargest(limit, merged, key=key_fn)
        else:
            jobs = heapq.nsmallest(limit, merged, key=key_fn)
    else:
        jobs = sorted(merged, key=key_fn, reverse=descending)

    # Compute filter matches for n/N navigation (preserve insertion order)
    filter_matches = list(dict.fromkeys(j.rule for j in jobs)) if filter_lower else []

    return jobs, failed_job_ids, filter_matches

get_current_log

get_current_log() -> Path | None

Get the currently selected log file.

Source code in snakesee/tui/data_source.py

def get_current_log(self) -> Path | None:
    """Get the currently selected log file."""
    if not self._available_logs:
        return None
    if self._current_log_index < len(self._available_logs):
        return self._available_logs[self._current_log_index]
    return self._available_logs[0] if self._available_logs else None

get_cutoff_time

get_cutoff_time() -> float | None

Get the cutoff time for filtering (when the next log started).

Source code in snakesee/tui/data_source.py

def get_cutoff_time(self) -> float | None:
    """Get the cutoff time for filtering (when the next log started)."""
    if self._current_log_index == 0:
        return None  # Latest log, no cutoff
    if self._current_log_index > 0 and len(self._available_logs) > 1:
        # Cutoff is the start of the next newer log. Use st_mtime (cross-platform)
        # rather than st_ctime, which is inode-change time on POSIX.
        next_log_index = self._current_log_index - 1
        if next_log_index >= 0:
            try:
                return self._available_logs[next_log_index].stat().st_mtime
            except OSError:
                pass
    return None

get_filtered_stats

get_filtered_stats() -> list[RuleTimingStats]

Get rule stats filtered by cutoff time if viewing historical log.

Source code in snakesee/tui/data_source.py

def get_filtered_stats(self) -> list[RuleTimingStats]:
    """Get rule stats filtered by cutoff time if viewing historical log."""
    from snakesee.parser import parse_metadata_files

    if self._cutoff_time is None:
        # Latest log: use stats from estimator, filtered by current workflow rules
        if self._estimator and self._estimator.rule_stats:
            current_rules = self._estimator.current_rules
            if current_rules is not None:
                return [
                    stats
                    for stats in self._estimator.rule_stats.values()
                    if stats.rule in current_rules
                ]
            return list(self._estimator.rule_stats.values())
        return []

    # Historical log: rebuild stats from metadata, filtering by cutoff time
    metadata_dir = self.workflow_dir / ".snakemake" / "metadata"
    stats_dict: dict[str, RuleTimingStats] = {}
    for job in parse_metadata_files(metadata_dir):
        if job.duration is not None and job.end_time is not None:
            if job.end_time < self._cutoff_time:
                if job.rule not in stats_dict:
                    stats_dict[job.rule] = RuleTimingStats(rule=job.rule)
                stats_dict[job.rule].durations.append(job.duration)

    # If no metadata found, parse stats from log files up to the cutoff
    if not stats_dict:
        stats_dict = self._parse_stats_from_logs(self._cutoff_time)

    return list(stats_dict.values())

get_inferred_pending_rules

get_inferred_pending_rules(progress: WorkflowProgress) -> dict[str, int] | None

Get inferred pending rules from completions and historical data.

Source code in snakesee/tui/data_source.py

def get_inferred_pending_rules(self, progress: WorkflowProgress) -> dict[str, int] | None:
    """Get inferred pending rules from completions and historical data."""
    if not self._estimator:
        return None

    # Registry is the single source of truth (populated from events or log parsing)
    all_completed = self._workflow_state.jobs.completed_job_infos()

    # If we have expected job counts, we can infer pending even without completions
    if not all_completed and not self._estimator.expected_job_counts:
        return None

    # Count completed jobs by rule (using ALL completed, not just recent)
    completed_by_rule: dict[str, int] = {}
    for job in all_completed:
        completed_by_rule[job.rule] = completed_by_rule.get(job.rule, 0) + 1

    # Count running jobs by rule
    running_by_rule: dict[str, int] = {}
    for job in progress.running_jobs:
        running_by_rule[job.rule] = running_by_rule.get(job.rule, 0) + 1

    # Only augment with historical counts if we don't have expected_job_counts
    if not self._estimator.expected_job_counts and self._estimator.rule_stats:
        for rule, stats in self._estimator.rule_stats.items():
            if rule not in completed_by_rule:
                completed_by_rule[rule] = stats.count

    return self._estimator._infer_pending_rules(
        completed_by_rule, progress.pending_jobs, self._estimator.current_rules, running_by_rule
    )

get_running_jobs_list

get_running_jobs_list(progress: WorkflowProgress, *, filter_text: str | None, sort_table: SortTableName | None, sort_column: int, sort_ascending: bool) -> list[JobInfo]

Get running jobs list with same order as table.

Applies the same filtering and sorting as _make_running_table() to ensure the selected index matches between the table display and log panel.

Returns:

Type	Description
list[JobInfo]	List of running jobs in display order.

Source code in snakesee/tui/data_source.py

def get_running_jobs_list(
    self,
    progress: WorkflowProgress,
    *,
    filter_text: str | None,
    sort_table: SortTableName | None,
    sort_column: int,
    sort_ascending: bool,
) -> list[JobInfo]:
    """Get running jobs list with same order as table.

    Applies the same filtering and sorting as ``_make_running_table()`` to
    ensure the selected index matches between the table display and log panel.

    Returns:
        List of running jobs in display order.
    """
    jobs = self.filter_jobs(progress.running_jobs, filter_text)

    # Apply custom sorting if running table is being sorted
    if sort_table == "running" and jobs:
        # Build job data tuples for sorting
        job_data = self._build_running_job_data(jobs)
        job_data = self._sort_running_job_data(job_data, sort_column, sort_ascending)
        # Extract just the jobs from the sorted tuples
        jobs = [jd[0] for jd in job_data]

    return jobs

get_tool_progress

get_tool_progress(job: JobInfo) -> ToolProgress | None

Get tool-specific progress for a running job.

Results are cached for the TTL duration to avoid parsing job logs on every refresh cycle.

Parameters:

Name	Type	Description	Default
job	JobInfo	The running job to check.	required

Returns:

Type	Description
ToolProgress | None	ToolProgress if parseable, None otherwise.

Source code in snakesee/tui/data_source.py

def get_tool_progress(self, job: JobInfo) -> ToolProgress | None:
    """Get tool-specific progress for a running job.

    Results are cached for the TTL duration to avoid parsing job logs
    on every refresh cycle.

    Args:
        job: The running job to check.

    Returns:
        ToolProgress if parseable, None otherwise.
    """
    # Use job.log_file (parsed from snakemake log, keyed by job_id)
    if job.log_file is None:
        return None

    # Use job_id as cache key (unique per job run)
    cache_key = job.job_id if job.job_id else str(job.log_file)
    now = time.time()

    log_path = self.workflow_dir / job.log_file
    if not log_path.exists():
        self._tool_progress_cache[cache_key] = (now, 0.0, None)
        return None

    # Get current file mtime for cache invalidation
    try:
        current_mtime = log_path.stat().st_mtime
    except OSError:
        return None

    # Check cache validity - must be within TTL AND file unchanged
    if cache_key in self._tool_progress_cache:
        cached_time, cached_mtime, cached_progress = self._tool_progress_cache[cache_key]
        if now - cached_time < self._tool_progress_cache_ttl and cached_mtime >= current_mtime:
            return cached_progress

    # Parse and cache the result with current mtime.  Third-party plugins
    # may raise arbitrary exceptions; catch broadly so a bad plugin can
    # never crash the TUI.
    try:
        progress = parse_tool_progress(job.rule, log_path)
    except Exception:  # noqa: BLE001 - intentional broad catch to protect TUI from plugin errors
        logger.debug(
            "Failed to parse tool progress for %s: %s", job.rule, log_path, exc_info=True
        )
        progress = None
    self._tool_progress_cache[cache_key] = (now, current_mtime, progress)
    return progress

init_estimator

init_estimator(*, show_progress: bool = True) -> None

Initialize or reinitialize the time estimator.

At startup the load can take many seconds and the user needs feedback before the App's compose() returns, so a transient Rich progress spinner is rendered directly to the terminal. This is a pragmatic exception to the data source being otherwise rendering-agnostic.

At runtime (re-init triggered by a key press while the Textual UI is live), rendering Rich output would corrupt the display, so callers pass show_progress=False to load silently.

Parameters:

Name	Type	Description	Default
show_progress	bool	Render a transient Rich progress spinner during load. Set False when the Textual UI is already running.	True

Source code in snakesee/tui/data_source.py

def init_estimator(self, *, show_progress: bool = True) -> None:
    """Initialize or reinitialize the time estimator.

    At startup the load can take many seconds and the user needs feedback
    before the App's compose() returns, so a transient Rich progress spinner
    is rendered directly to the terminal. This is a pragmatic exception to the
    data source being otherwise rendering-agnostic.

    At runtime (re-init triggered by a key press while the Textual UI is
    live), rendering Rich output would corrupt the display, so callers pass
    ``show_progress=False`` to load silently.

    Args:
        show_progress: Render a transient Rich progress spinner during load.
            Set False when the Textual UI is already running.
    """
    self._workflow_state.rules.clear()
    self._workflow_state.jobs.clear()

    if not self.use_estimation:
        self._estimator = None
        return

    self._estimator = TimeEstimator(
        use_wildcard_conditioning=self._use_wildcard_conditioning,
        weighting_strategy=self.weighting_strategy,
        half_life_logs=self.half_life_logs,
        half_life_days=self.half_life_days,
        rule_registry=self._workflow_state.rules,
    )

    metadata_dir = self.workflow_dir / ".snakemake" / "metadata"
    has_metadata_fs = metadata_dir.exists()
    has_profile = self.profile_path is not None and self.profile_path.exists()

    # Check if there's anything to load (worth showing progress)
    paths = WorkflowPaths(self.workflow_dir)
    has_metadata_db = paths.has_metadata_db
    has_metadata = has_metadata_fs or has_metadata_db
    log_paths = paths.find_all_logs()

    # Skip loading entirely if there's nothing to load
    if not has_metadata and not has_profile and not log_paths:
        return

    # Render a real Rich progress spinner only at startup; load silently when
    # the Textual UI is already on screen (a Rich render would corrupt it).
    progress_cm: AbstractContextManager[Any]
    if show_progress:
        from rich.console import Console
        from rich.progress import BarColumn
        from rich.progress import MofNCompleteColumn
        from rich.progress import Progress
        from rich.progress import SpinnerColumn
        from rich.progress import TextColumn

        progress_cm = Progress(
            SpinnerColumn(),
            TextColumn("[progress.description]{task.description}"),
            BarColumn(),
            MofNCompleteColumn(),
            console=Console(),
            transient=True,
        )
    else:
        progress_cm = nullcontext(_NULL_PROGRESS)

    with progress_cm as progress:
        # Load from profile first if available
        if has_profile:
            task = progress.add_task("Loading profile...", total=1)
            try:
                from snakesee.profile import load_profile

                assert self.profile_path is not None
                profile = load_profile(self.profile_path)
                self._estimator.rule_stats = profile.to_rule_stats()
            except (OSError, ValueError) as e:
                # Log failure and fall back to metadata only
                logger.debug("Failed to load profile %s: %s", self.profile_path, e)
            progress.update(task, completed=1)

        # Load metadata via persistence backend (supports both FS and DB)
        from snakesee.persistence import detect_backend

        backend = detect_backend(self.workflow_dir)

        if has_metadata:
            # Determine progress bar total from FS file count (DB has no cheap count)
            if has_metadata_fs and not has_metadata_db:
                metadata_files = list(metadata_dir.rglob("*"))
                metadata_files = [f for f in metadata_files if f.is_file()]
                file_count = len(metadata_files)
            else:
                file_count = 0

            task = progress.add_task(
                "Loading metadata...", total=file_count if file_count > 0 else None
            )

            def metadata_cb(current: int, _total: int) -> None:
                progress.update(task, completed=current)

            self._estimator.load_from_backend(backend, progress_callback=metadata_cb)

        # Load historical timing from events file (complements metadata)
        events_file = get_event_file_path(self.workflow_dir)
        if events_file.exists():
            task = progress.add_task("Loading events...", total=1)
            self._estimator.load_from_events(events_file)
            progress.update(task, completed=1)

        # Initialize thread stats from log parsing
        if log_paths:
            task = progress.add_task("Analyzing thread usage...", total=len(log_paths))
            self._init_thread_stats_from_log(
                log_paths=log_paths,
                progress_callback=lambda current, _total: progress.update(
                    task, completed=current
                ),
            )

        # Parse current rules (fast, no progress needed)
        self._init_current_rules_from_log()

init_event_reader

init_event_reader() -> None

Initialize the event reader if event file exists and is current.

The events file is validated against the current log file's start time to ensure we don't use stale events from a previous workflow run.

Source code in snakesee/tui/data_source.py

def init_event_reader(self) -> None:
    """Initialize the event reader if event file exists and is current.

    The events file is validated against the current log file's start time
    to ensure we don't use stale events from a previous workflow run.
    """
    if not self._events_enabled:
        return

    event_file = get_event_file_path(self.workflow_dir)
    if not event_file.exists():
        self._event_reader = None
        return

    # Get the current log file's start time for validation. Use the first
    # timestamp recorded in the log (the workflow-start time) rather than the
    # file's mtime: mtime is the last-append time and drifts forward as the
    # run writes, so after a minute of activity it would make the current
    # run's own event file look "stale". The first log timestamp is fixed for
    # the life of the run.
    paths = WorkflowPaths(self.workflow_dir)
    log_path = paths.find_latest_log()
    log_start_time = _parse_log_start_time(log_path) if log_path is not None else None

    # Validate the events file is for the current workflow run
    if _is_event_file_current(event_file, log_start_time):
        self._event_reader = EventReader(event_file)
        logger.debug("Events file is current, using for monitoring")
    else:
        self._event_reader = None
        logger.info(
            "Ignoring stale events file %s (from a previous workflow run)",
            event_file,
        )

init_log_reader

init_log_reader() -> None

Initialize the incremental log reader.

Creates a reader for the current log file, enabling efficient incremental parsing instead of re-reading the entire file on each poll.

Source code in snakesee/tui/data_source.py

def init_log_reader(self) -> None:
    """Initialize the incremental log reader.

    Creates a reader for the current log file, enabling efficient
    incremental parsing instead of re-reading the entire file on each poll.
    """
    paths = WorkflowPaths(self.workflow_dir)
    log_path = paths.find_latest_log()
    if log_path is not None:
        self._log_reader = IncrementalLogReader(log_path)
    else:
        # Create with a placeholder path; will be updated when log appears
        self._log_reader = IncrementalLogReader(paths.log_dir / "placeholder.snakemake.log")

init_validation

init_validation() -> None

Initialize validation if event file exists.

Validation is automatically enabled when the logger plugin's event file is detected, allowing comparison between event-based and parsed state to find bugs in either approach.

Source code in snakesee/tui/data_source.py

def init_validation(self) -> None:
    """Initialize validation if event file exists.

    Validation is automatically enabled when the logger plugin's event
    file is detected, allowing comparison between event-based and
    parsed state to find bugs in either approach.
    """
    # Close existing validation logger to prevent file handle leaks
    if self._validation_logger is not None:
        self._validation_logger.close()
        self._validation_logger = None

    event_file = get_event_file_path(self.workflow_dir)
    if event_file.exists():
        self._event_accumulator = EventAccumulator()
        self._validation_logger = ValidationLogger(self.workflow_dir)
        self._validation_logger.log_session_start()

poll_state

poll_state() -> tuple[WorkflowProgress, TimeEstimate | None]

Poll the current workflow state and estimate.

For the latest run (current_log_index == 0) we merge live events and the in-memory JobRegistry into the parsed log to enrich timing and catch jobs the parser missed. For historical runs we deliberately skip all of that — events and the registry describe the current run and would otherwise leak into older log views.

Returns:

Type	Description
tuple[WorkflowProgress, TimeEstimate | None]	Tuple of (workflow progress, optional time estimate).

Source code in snakesee/tui/data_source.py

def poll_state(self) -> tuple[WorkflowProgress, TimeEstimate | None]:
    """Poll the current workflow state and estimate.

    For the latest run (``current_log_index == 0``) we merge live events and
    the in-memory ``JobRegistry`` into the parsed log to enrich timing and
    catch jobs the parser missed. For historical runs we deliberately skip
    all of that — events and the registry describe the *current* run and
    would otherwise leak into older log views.

    Returns:
        Tuple of (workflow progress, optional time estimate).
    """
    # Refresh log list if viewing latest
    if self._current_log_index == 0:
        self.refresh_log_list()

    is_latest_view = self._current_log_index == 0

    # Get the selected log file and cutoff time for historical view
    log_file = self.get_current_log() if not is_latest_view else None
    self._cutoff_time = self.get_cutoff_time()

    # Live events / readers only apply to the latest run.
    events = self.read_new_events() if is_latest_view else []
    reader = self._log_reader if is_latest_view else None

    progress = parse_workflow_state(
        self.workflow_dir,
        log_file=log_file,
        cutoff_time=self._cutoff_time,
        log_reader=reader,
    )

    # Sync log reader's completed jobs to registry when events aren't available
    # This ensures the registry is the single source of truth regardless of
    # whether the snakesee logger plugin is being used
    if is_latest_view and not events and self._log_reader:
        for job in self._log_reader.completed_jobs:
            if job.job_id:
                self._workflow_state.jobs.apply_job_info(job, key=job.job_id)

    # Validate: compare event-based state with parsed state (before applying)
    # This logs discrepancies to help find bugs in either approach
    if is_latest_view and events:
        self.validate_state(events, progress)

    # Apply events to enhance progress accuracy
    if is_latest_view and events:
        progress = self.apply_events_to_progress(progress, events)
    elif is_latest_view:
        # Even without new events, merge registry-tracked failed jobs into progress
        # This ensures failed jobs discovered via earlier events are not lost
        # when log re-parsing misses them
        registry_failed = self._workflow_state.jobs.failed_job_infos()
        if registry_failed:
            from dataclasses import replace

            registry_failed_ids = {job.job_id for job in registry_failed if job.job_id}
            merged_failed = list(registry_failed)
            for job in progress.failed_jobs_list:
                if job.job_id and job.job_id not in registry_failed_ids:
                    merged_failed.append(job)
            progress = replace(
                progress,
                failed_jobs=len(merged_failed),
                failed_jobs_list=merged_failed,
            )

    # Always populate pending_jobs_list from log-based scheduled jobs
    # (even when no new events, we need this for wildcard-conditioned ETA).
    # Historical views derive pending purely from the parsed progress so
    # the live registry can't bleed into a finished run.
    if is_latest_view and not progress.pending_jobs_list:
        from dataclasses import replace

        # Registry is the single source of truth (populated from events or log parsing)
        all_completed = self._workflow_state.jobs.completed_job_infos()
        pending_jobs_list = self._compute_pending_jobs_from_scheduled(
            progress.running_jobs, all_completed, progress.failed_jobs_list
        )
        if pending_jobs_list:
            progress = replace(progress, pending_jobs_list=pending_jobs_list)

    # Infer total_jobs from the Job stats table when the progress line hasn't
    # appeared yet (it only emerges after the first completion). This lets the
    # pending panel show correct counts immediately on a fresh run.
    #
    # Scoped to the latest/live run (log index 0): expected_job_counts is parsed
    # from the latest log, so inferring it onto a historical run's progress would
    # graft the live run's totals onto the wrong run.
    if (
        progress.total_jobs == 0
        and self._estimator is not None
        and self._current_log_index == 0
    ):
        if not self._estimator.expected_job_counts:
            self._init_current_rules_from_log()
        if self._estimator.expected_job_counts:
            from dataclasses import replace

            inferred_total = sum(self._estimator.expected_job_counts.values())
            if inferred_total > 0:
                progress = replace(progress, total_jobs=inferred_total)

    # Update rule_stats with newly completed jobs (for Rule Statistics panel)
    self._update_rule_stats_from_completions(progress)

    estimate = None
    if self._estimator is not None:
        estimate = self._estimator.estimate_remaining(progress)

    # Periodically clean up stale cache entries
    self.cleanup_tool_progress_cache()

    return progress, estimate

read_log_tail

read_log_tail(log_path: Path, max_lines: int = 500) -> list[str]

Read the last N lines of a log file efficiently.

For large files, seeks near the end instead of reading the entire file.

Parameters:

Name	Type	Description	Default
log_path	Path	Path to the log file.	required
max_lines	int	Maximum number of lines to read.	500

Returns:

Type	Description
list[str]	List of lines (most recent at end).

Source code in snakesee/tui/data_source.py

def read_log_tail(self, log_path: Path, max_lines: int = 500) -> list[str]:
    """Read the last N lines of a log file efficiently.

    For large files, seeks near the end instead of reading the entire file.

    Args:
        log_path: Path to the log file.
        max_lines: Maximum number of lines to read.

    Returns:
        List of lines (most recent at end).
    """
    # Average bytes per line estimate for seeking
    BYTES_PER_LINE_ESTIMATE = 120

    try:
        # Check if cache is still valid
        stat = log_path.stat()
        mtime = stat.st_mtime
        file_size = stat.st_size
        if (
            self._cached_log_path == log_path
            and self._cached_log_mtime == mtime
            and self._cached_log_lines
        ):
            return self._cached_log_lines

        # For small files, just read the whole thing
        if file_size < BYTES_PER_LINE_ESTIMATE * max_lines * 2:
            content = log_path.read_text(errors="ignore")
            lines = content.splitlines()
        else:
            # For large files, seek near the end to avoid reading everything
            # Read extra bytes to ensure we get enough lines
            seek_bytes = BYTES_PER_LINE_ESTIMATE * max_lines * 2
            with open(log_path, "rb") as f:
                # Seek to near the end
                f.seek(max(0, file_size - seek_bytes))
                # Read to end
                content = f.read().decode("utf-8", errors="ignore")
                lines = content.splitlines()
                # Skip first line (likely partial from seek)
                if lines and file_size > seek_bytes:
                    lines = lines[1:]

        # Take last max_lines
        result = lines[-max_lines:] if len(lines) > max_lines else lines

        # Update cache
        self._cached_log_path = log_path
        self._cached_log_mtime = mtime
        self._cached_log_lines = result

        return result
    except OSError:
        return ["[Error reading log file]"]

read_new_events

read_new_events() -> list[SnakeseeEvent]

Read new events from the event file if available.

Returns:

Type	Description
list[SnakeseeEvent]	List of new events, or empty list if no events or event reading disabled.

Source code in snakesee/tui/data_source.py

def read_new_events(self) -> list[SnakeseeEvent]:
    """Read new events from the event file if available.

    Returns:
        List of new events, or empty list if no events or event reading disabled.
    """
    if not self._events_enabled or self._event_reader is None:
        # Try to initialize if event file now exists (with validation)
        if self._events_enabled and self._event_reader is None:
            self.init_event_reader()

        if self._event_reader is None:
            return []

    return self._event_reader.read_new_events()

refresh_log_list

refresh_log_list() -> None

Refresh the list of available log files.

Source code in snakesee/tui/data_source.py

def refresh_log_list(self) -> None:
    """Refresh the list of available log files."""
    log_dir = self.workflow_dir / ".snakemake" / "log"
    if log_dir.exists():
        # Sort by modification time, newest first
        logs = sorted(
            log_dir.glob("*.snakemake.log"),
            key=lambda p: p.stat().st_mtime,
            reverse=True,
        )
        self._available_logs = logs
    else:
        self._available_logs = []

    # Reset to most recent if current index is out of bounds
    if self._current_log_index >= len(self._available_logs):
        self._current_log_index = 0

    # Detect when a new workflow starts (new latest log)
    # and re-parse current_rules to filter pending jobs correctly
    new_latest = self._available_logs[0] if self._available_logs else None
    if new_latest != self._latest_log_path:
        self._latest_log_path = new_latest
        self._init_current_rules_from_log()

thread_stats_dict

thread_stats_dict() -> dict[str, ThreadTimingStats]

Return per-rule, per-thread timing statistics.

Source code in snakesee/tui/data_source.py

def thread_stats_dict(self) -> "dict[str, ThreadTimingStats]":
    """Return per-rule, per-thread timing statistics."""
    return self._workflow_state.rules.to_thread_stats_dict()

update_cache_ttl

update_cache_ttl() -> None

Update tool progress cache TTL based on current refresh rate.

Called when refresh_rate changes to keep cache behavior in sync.

Source code in snakesee/tui/data_source.py

def update_cache_ttl(self) -> None:
    """Update tool progress cache TTL based on current refresh rate.

    Called when refresh_rate changes to keep cache behavior in sync.
    """
    self._tool_progress_cache_ttl = min(
        ADAPTIVE_CACHE_TTL_MULTIPLIER * self.refresh_rate, MAX_CACHE_TTL
    )

validate_state

validate_state(events: list[SnakeseeEvent], parsed: WorkflowProgress) -> None

Compare event-based state with parsed state and log discrepancies.

Parameters:

Name	Type	Description	Default
events	list[SnakeseeEvent]	New events to process.	required
parsed	WorkflowProgress	Current parsed workflow progress.	required

Source code in snakesee/tui/data_source.py

def validate_state(self, events: list[SnakeseeEvent], parsed: WorkflowProgress) -> None:
    """Compare event-based state with parsed state and log discrepancies.

    Args:
        events: New events to process.
        parsed: Current parsed workflow progress.
    """
    # Initialize validation if not yet done (event file may have appeared)
    if self._event_accumulator is None:
        self.init_validation()

    if self._event_accumulator is None or self._validation_logger is None:
        return

    # Accumulate new events
    self._event_accumulator.process_events(events)

    # Only compare if we have meaningful state from events
    if not self._event_accumulator.workflow_started:
        return

    # Compare states and log discrepancies
    discrepancies = compare_states(self._event_accumulator, parsed)

    if discrepancies:
        self._validation_logger.log_discrepancies(discrepancies)

    # Log summary periodically (every comparison for now)
    self._validation_logger.log_summary(self._event_accumulator, parsed)

Functions:

renderables

Rich renderables (header, progress bar, progress panel, summary, help, easter egg).

Classes

Functions:

format_cost

format_cost(usd: float) -> str

Format a USD cost: 4 decimals under $1 (per-job costs are tiny), else 2.

Source code in snakesee/tui/renderables.py

def format_cost(usd: float) -> str:
    """Format a USD cost: 4 decimals under $1 (per-job costs are tiny), else 2."""
    return f"${usd:.4f}" if usd < 1 else f"${usd:,.2f}"

make_easter_egg

make_easter_egg(console_width: int = 80, console_height: int = 24) -> RenderableType

Create the Fulcrum Genomics easter egg renderable.

Renders the bundled logo (snakesee/assets/logo.png, rasterized from the upstream SVG with a flat dark background) into the terminal at the given size using rich-pixels' half-block characters. Falls back to a text logo if the image is missing or unreadable.

The image is downscaled directly to the target rich-pixels render size with bilinear resampling — bilinear blurs slightly more than LANCZOS, but the softer edges read better through 1px-wide half-block cells.

Parameters:

Name	Type	Description	Default
console_width	int	Width of the terminal console in characters.	80
console_height	int	Height of the terminal console in lines.	24

Returns:

Type	Description
RenderableType	A Rich renderable (Group of centered pixels + dismiss hint) displaying
RenderableType	the Fulcrum Genomics logo, or a text fallback.

Source code in snakesee/tui/renderables.py

def make_easter_egg(console_width: int = 80, console_height: int = 24) -> "RenderableType":
    """Create the Fulcrum Genomics easter egg renderable.

    Renders the bundled logo (``snakesee/assets/logo.png``, rasterized from the
    upstream SVG with a flat dark background) into the terminal at the given
    size using rich-pixels' half-block characters. Falls back to a text logo if
    the image is missing or unreadable.

    The image is downscaled directly to the target rich-pixels render size with
    bilinear resampling — bilinear blurs slightly more than LANCZOS, but the
    softer edges read better through 1px-wide half-block cells.

    Args:
        console_width: Width of the terminal console in characters.
        console_height: Height of the terminal console in lines.

    Returns:
        A Rich renderable (Group of centered pixels + dismiss hint) displaying
        the Fulcrum Genomics logo, or a text fallback.
    """
    from PIL import Image
    from rich.align import Align
    from rich.console import Group
    from rich_pixels import Pixels

    # Reserve one line at the bottom for the dismiss hint.
    image_height_chars = max(3, console_height - 2)
    # rich-pixels: 1 char = 1 px wide, 2 px tall (half-block).
    target_pixel_width = max(8, console_width)
    target_pixel_height = max(8, image_height_chars * 2)

    hint = Text("\n[ press any key to return ]", style=f"dim {FG_BLUE}", justify="center")

    if FG_LOGO_PATH.exists():
        try:
            source = Image.open(FG_LOGO_PATH).convert("RGB")

            # Pick the larger axis we can fully fit while preserving aspect ratio.
            img_ratio = source.width / source.height
            terminal_ratio = target_pixel_width / target_pixel_height
            if terminal_ratio > img_ratio:
                # Terminal is wider than image — height-bound.
                new_height = target_pixel_height
                new_width = int(new_height * img_ratio)
            else:
                # Terminal is taller than image — width-bound.
                new_width = target_pixel_width
                new_height = int(new_width / img_ratio)

            resized = source.resize((new_width, new_height), Image.Resampling.BILINEAR)
            pixels = Pixels.from_image(resized)
            return Group(Align.center(pixels, vertical="middle"), hint)
        except (OSError, ValueError, TypeError):
            pass  # Image missing or unreadable; fall through to text fallback.

    fallback = Text()
    fallback.append("\n")
    fallback.append("FULCRUM GENOMICS", style=f"bold {FG_BLUE}")
    fallback.append("\n")
    return Group(Align.center(fallback, vertical="middle"), hint)

make_header

make_header(progress: WorkflowProgress, workflow_path: str, paused: bool, event_reader: EventReader | None, max_path_len: int = 60) -> Panel

Create the header panel with workflow path and status.

Parameters:

Name	Type	Description	Default
progress	WorkflowProgress	Current workflow progress snapshot.	required
workflow_path	str	Resolved absolute path to the monitored workflow directory. Resolving once at the call site keeps the per-frame cost to a string truncation rather than a filesystem resolve().	required
paused	bool	Whether auto-refresh is currently paused.	required
event_reader	EventReader | None	Active event reader if using event-based monitoring, else None.	required
max_path_len	int	Maximum characters to spend on the path before middle-truncating, so a long path can't crowd out the status fields.	60

Returns:

Type	Description
Panel	A Rich Panel containing the header text.

Source code in snakesee/tui/renderables.py

def make_header(
    progress: WorkflowProgress,
    workflow_path: str,
    paused: bool,
    event_reader: EventReader | None,
    max_path_len: int = 60,
) -> Panel:
    """Create the header panel with workflow path and status.

    Args:
        progress: Current workflow progress snapshot.
        workflow_path: Resolved absolute path to the monitored workflow directory.
            Resolving once at the call site keeps the per-frame cost to a string
            truncation rather than a filesystem ``resolve()``.
        paused: Whether auto-refresh is currently paused.
        event_reader: Active event reader if using event-based monitoring, else None.
        max_path_len: Maximum characters to spend on the path before middle-truncating,
            so a long path can't crowd out the status fields.

    Returns:
        A Rich Panel containing the header text.
    """
    status_styles = {
        WorkflowStatus.RUNNING: "bold green",
        WorkflowStatus.COMPLETED: "bold blue",
        WorkflowStatus.FAILED: "bold red",
        WorkflowStatus.INCOMPLETE: "bold yellow",
        WorkflowStatus.UNKNOWN: "bold yellow",
    }
    style = status_styles.get(progress.status, "bold white")

    header_text = Text()
    header_text.append("FULCRUM GENOMICS", style=f"bold {FG_BLUE}")
    header_text.append(" │ ", style="dim")
    header_text.append("Snakemake Monitor", style="bold white")
    header_text.append("  │  ", style="dim")
    header_text.append(_truncate_path(workflow_path, max_path_len), style="dim")
    header_text.append("  │  Status: ")
    header_text.append(progress.status.value.upper(), style=style)

    if progress.elapsed_seconds is not None:
        header_text.append("  │  Elapsed: ")
        header_text.append(format_duration(progress.elapsed_seconds), style=FG_BLUE)

    # Remote executors can have jobs queued (awaiting a node) but not yet running;
    # surface that count so a "running" workflow waiting on the queue is honest.
    queued_count = len(progress.queued_jobs_list)
    if queued_count > 0:
        header_text.append("  │  Queued: ")
        header_text.append(str(queued_count), style="bold yellow")

    # Estimated workflow cost so far (remote executors with cost estimation on).
    if progress.total_cost_estimate is not None:
        header_text.append("  │  Cost: ")
        header_text.append(f"~{format_cost(progress.total_cost_estimate)}", style=FG_GREEN)
        header_text.append(" (est)", style="dim")

    if paused:
        header_text.append("  │  ")
        header_text.append("PAUSED", style="bold yellow")

    # Monitoring method indicator
    header_text.append("  │  ")
    if event_reader is not None:
        header_text.append("⚡ Events", style="bold green")
    else:
        header_text.append("📄 Parsing", style="bold blue")

    return Panel(header_text, style="white on grey23", border_style=FG_BLUE, height=3)

make_help

make_help() -> Panel

Create the help overlay panel.

Returns:

Type	Description
Panel	A Rich Panel containing the keyboard shortcut reference.

Source code in snakesee/tui/renderables.py

def make_help() -> Panel:
    """Create the help overlay panel.

    Returns:
        A Rich Panel containing the keyboard shortcut reference.
    """
    help_text = Table(show_header=False, box=None, padding=(0, 2))
    help_text.add_column("Key", style="bold cyan")
    help_text.add_column("Action")

    help_text.add_row("", "[bold]General[/bold]")
    help_text.add_row("q", "Quit")
    help_text.add_row("?", "Toggle this help")
    help_text.add_row("p", "Pause/resume auto-refresh")
    help_text.add_row("e", "Toggle time estimation")
    help_text.add_row("w", "Toggle wildcard conditioning")
    help_text.add_row("a", "Toggle colorblind-accessible mode")
    help_text.add_row("r", "Force refresh")
    help_text.add_row("Ctrl+r", "Hard refresh (reload historical data)")
    help_text.add_row("", "")
    help_text.add_row("", "[bold]Refresh Rate[/bold]")
    help_text.add_row("- / +", "Decrease/increase by 0.5s")
    help_text.add_row("< / >", "Decrease/increase by 5s")
    help_text.add_row("0", f"Reset to default ({DEFAULT_REFRESH_RATE}s)")
    help_text.add_row("G", f"Set to minimum ({MIN_REFRESH_RATE}s, fastest)")
    help_text.add_row("", "")
    help_text.add_row("", "[bold]Layout & Filter[/bold]")
    help_text.add_row("Tab", "Cycle layout (full/compact/minimal)")
    help_text.add_row("/", "Filter rules by name")
    help_text.add_row("n / N", "Next/previous filter match")
    help_text.add_row("Esc", "Clear filter, return to latest log")
    help_text.add_row("", "")
    help_text.add_row("", "[bold]Log Navigation[/bold]")
    help_text.add_row("[ / ]", "View older/newer log (1 step)")
    help_text.add_row("{ / }", "View older/newer log (5 steps)")
    help_text.add_row("", "")
    help_text.add_row("", "[bold]Table Sorting[/bold]")
    help_text.add_row("s / S", "Cycle sort table (forward/backward)")
    help_text.add_row("1-4", "Sort by column (press again to reverse)")
    help_text.add_row("", "")
    help_text.add_row("", "[bold]Table Navigation (Enter to start)[/bold]")
    help_text.add_row("j / k", "Move down/up one row")
    help_text.add_row("g / G", "Jump to first/last row")
    help_text.add_row("Ctrl+d/u", "Move down/up half page")
    help_text.add_row("Ctrl+f/b", "Move down/up full page")
    help_text.add_row("Tab / S-Tab", "Cycle all tables")
    help_text.add_row("h / l", "Switch to left/right column table")
    help_text.add_row("Enter", "View job log (running/completions only)")
    help_text.add_row("Esc", "Exit table navigation")
    help_text.add_row("", "")
    help_text.add_row("", "[bold]Log Viewing (Enter on job)[/bold]")
    help_text.add_row("j / k", "Scroll down/up one line")
    help_text.add_row("g / G", "Jump to start/end of log")
    help_text.add_row("Ctrl+d/u", "Scroll down/up half page")
    help_text.add_row("Ctrl+f/b", "Scroll down/up full page")
    help_text.add_row("Esc", "Return to table navigation")

    from snakesee import __version__

    return Panel(
        help_text,
        title="[bold]Keyboard Shortcuts[/bold]",
        subtitle=f"Press any key to close [dim]│ snakesee v{__version__}[/dim]",
        border_style="cyan",
    )

make_progress_bar

make_progress_bar(progress: WorkflowProgress, width: int, accessibility: AccessibilityConfig) -> Text

Create a colored progress bar showing succeeded/failed/in-flight/pending portions.

Parameters:

Name	Type	Description	Default
progress	WorkflowProgress	Current workflow progress snapshot.	required
width	int	Total character width of the bar.	required
accessibility	AccessibilityConfig	Visual encoding config controlling bar characters.	required

Returns:

Type	Description
Text	A Rich Text object representing the progress bar.

Source code in snakesee/tui/renderables.py

def make_progress_bar(
    progress: WorkflowProgress,
    width: int,
    accessibility: AccessibilityConfig,
) -> Text:
    """Create a colored progress bar showing succeeded/failed/in-flight/pending portions.

    Args:
        progress: Current workflow progress snapshot.
        width: Total character width of the bar.
        accessibility: Visual encoding config controlling bar characters.

    Returns:
        A Rich Text object representing the progress bar.
    """
    total = max(1, progress.total_jobs)
    succeeded = progress.completed_jobs
    failed = progress.failed_jobs
    in_flight, in_flight_style = _in_flight_segment(progress, accessibility)
    config = accessibility

    # Calculate widths for each segment. Clamp each to non-negative bounds within the
    # remaining width so a transient counter skew (counts briefly exceeding total) can
    # never produce a negative segment and under-render the bar.
    succeeded_width = min(width, max(0, int((succeeded / total) * width)))
    failed_width = min(width - succeeded_width, max(0, int((failed / total) * width)))
    in_flight_width = min(
        width - succeeded_width - failed_width, max(0, int((in_flight / total) * width))
    )
    pending_width = max(0, width - succeeded_width - failed_width - in_flight_width)

    # Build the bar with colored segments
    bar = Text()
    bar.append(config.succeeded.char * succeeded_width, style="green")
    bar.append(config.failed.char * failed_width, style="red")
    bar.append(in_flight_style.char * in_flight_width, style="yellow")
    bar.append(config.remaining.char * pending_width, style="dim")

    return bar

make_progress_panel

make_progress_panel(progress: WorkflowProgress, estimate: TimeEstimate | None, use_estimation: bool, accessibility: AccessibilityConfig, console_width: int = 80) -> Panel

Create the progress bar panel.

Parameters:

Name	Type	Description	Default
progress	WorkflowProgress	Current workflow progress snapshot.	required
estimate	TimeEstimate | None	Time estimate from the estimator, or None if unavailable.	required
use_estimation	bool	Whether time estimation is enabled.	required
accessibility	AccessibilityConfig	Visual encoding config for the progress bar.	required
console_width	int	Width of the terminal console in characters.	80

Returns:

Type	Description
Panel	A Rich Panel containing the progress bar, ETA, and legend.

Source code in snakesee/tui/renderables.py

def make_progress_panel(
    progress: WorkflowProgress,
    estimate: TimeEstimate | None,
    use_estimation: bool,
    accessibility: AccessibilityConfig,
    console_width: int = 80,
) -> Panel:
    """Create the progress bar panel.

    Args:
        progress: Current workflow progress snapshot.
        estimate: Time estimate from the estimator, or None if unavailable.
        use_estimation: Whether time estimation is enabled.
        accessibility: Visual encoding config for the progress bar.
        console_width: Width of the terminal console in characters.

    Returns:
        A Rich Panel containing the progress bar, ETA, and legend.
    """
    total = max(1, progress.total_jobs)
    completed = progress.completed_jobs + progress.failed_jobs
    percent = (completed / total) * 100

    # Calculate bar width based on console width
    # Reserve space for: "Progress " (9) + " XX.X% " (7) + "(XXX/XXX jobs)" (~15) + borders (~4)
    bar_width = max(20, console_width - 40)

    # Create colored progress bar
    progress_bar = make_progress_bar(progress, bar_width, accessibility)

    # Progress text line
    progress_line = Text()
    progress_line.append("Progress ", style=f"bold {FG_BLUE}")
    progress_line.append(progress_bar)
    progress_line.append(f" {percent:5.1f}% ", style="bold")
    progress_line.append(f"({completed}/{total} jobs)", style="dim")

    # ETA text - handle different workflow states
    eta_parts = []
    if progress.status == WorkflowStatus.FAILED:
        eta_parts.append("[bold red]FAILED[/bold red]")
        if progress.failed_jobs > 0:
            eta_parts.append(f"[dim]({progress.failed_jobs} job(s) failed)[/dim]")
    elif progress.status == WorkflowStatus.INCOMPLETE:
        eta_parts.append("[bold yellow]INCOMPLETE[/bold yellow]")
        if progress.incomplete_jobs_list:
            eta_parts.append(
                f"[dim]({len(progress.incomplete_jobs_list)} job(s) were in progress)[/dim]"
            )
    elif progress.status == WorkflowStatus.COMPLETED:
        eta_parts.append("[bold blue]Complete[/bold blue]")
    elif estimate is not None:
        eta_parts.append(f"ETA: {estimate.format_eta()}")

        if estimate.seconds_remaining < float("inf") and estimate.seconds_remaining > 0:
            # Use the injectable clock so tests can pin completion-time formatting.
            now = datetime.fromtimestamp(get_clock().now()).astimezone()
            completion_dt = now + timedelta(seconds=estimate.seconds_remaining)
            tz_name = completion_dt.strftime("%Z") or "local"
            # Include the date (and always the timezone) when the ETA crosses midnight,
            # so an overnight estimate isn't mistaken for one later today.
            if completion_dt.date() == now.date():
                completion_str = completion_dt.strftime("%H:%M:%S")
            else:
                completion_str = completion_dt.strftime("%Y-%m-%d %H:%M:%S")
            eta_parts.append(f"({completion_str} {tz_name})")

        # Show estimation method and inferred cores for transparency
        method_info = estimate.method
        if estimate.inferred_cores is not None and estimate.inferred_cores > 1:
            method_info += f" cores≈{estimate.inferred_cores:.0f}"
        eta_parts.append(f"[dim][{method_info}][/dim]")
    elif not use_estimation:
        eta_parts.append("[dim]ETA: disabled[/dim]")

    eta_text = Text.from_markup("  ".join(eta_parts)) if eta_parts else Text("")

    # Legend for the progress bar, showing every non-zero segment so the bar is
    # informative even before any job completes (and regardless of accessibility mode).
    config = accessibility
    legend = Text()
    legend_parts: list[tuple[str, str, str]] = []
    if progress.completed_jobs > 0:
        legend_parts.append(
            (config.succeeded.char, "green", f"{progress.completed_jobs} {config.succeeded.label}")
        )
    if progress.failed_jobs > 0:
        legend_parts.append(
            (config.failed.char, "red", f"{progress.failed_jobs} {config.failed.label}")
        )
    in_flight, in_flight_style = _in_flight_segment(progress, config)
    if in_flight > 0:
        legend_parts.append(
            (in_flight_style.char, "yellow", f"{in_flight} {in_flight_style.label}")
        )
    pending = progress.pending_jobs
    if pending > 0:
        legend_parts.append((config.remaining.char, "dim", f"{pending} {config.remaining.label}"))
    show_legend = bool(legend_parts)
    if show_legend:
        legend.append("  (", style="dim")
        for i, (symbol, style, label) in enumerate(legend_parts):
            if i > 0:
                legend.append("  ", style="dim")
            legend.append(symbol, style=style)
            legend.append(f"={label}", style="dim")
        legend.append(")", style="dim")

    # Border color based on status (use FG colors for normal states)
    border_colors = {
        WorkflowStatus.RUNNING: FG_BLUE,
        WorkflowStatus.COMPLETED: FG_GREEN,
        WorkflowStatus.FAILED: "red",
        WorkflowStatus.INCOMPLETE: "yellow",
        WorkflowStatus.UNKNOWN: "yellow",
    }
    border_style = border_colors.get(progress.status, FG_BLUE)

    # Combine progress line with legend if present
    if show_legend:
        full_progress = Text()
        full_progress.append(progress_line)
        full_progress.append(legend)
        return Panel(
            Group(full_progress, eta_text),
            title="Progress",
            border_style=border_style,
        )

    return Panel(
        Group(progress_line, eta_text),
        title="Progress",
        border_style=border_style,
    )

make_remote_job_info

make_remote_job_info(job: JobInfo) -> list[Text]

Build display lines describing a remote job's external identifier and links.

For a job that ran on a remote executor (e.g. AWS Batch), this surfaces the external job id and, when enough information is available, deep links to the AWS console and CloudWatch logs. It degrades gracefully: a bare job id with no region yields just the id line; a local job yields no lines at all.

Lines are Rich Text rather than str so styling (e.g. the dimmed termination-source parenthetical) survives the job-detail RichLog, which deliberately disables markup to avoid misrendering log content.

Parameters:

Name	Type	Description	Default
job	JobInfo	The job to describe.	required

Returns:

Type	Description
list[Text]	A list of text lines (empty if the job has no external identifier).

Source code in snakesee/tui/renderables.py

def make_remote_job_info(job: "JobInfo") -> list[Text]:
    """Build display lines describing a remote job's external identifier and links.

    For a job that ran on a remote executor (e.g. AWS Batch), this surfaces the
    external job id and, when enough information is available, deep links to the
    AWS console and CloudWatch logs. It degrades gracefully: a bare job id with
    no region yields just the id line; a local job yields no lines at all.

    Lines are Rich ``Text`` rather than ``str`` so styling (e.g. the dimmed
    termination-source parenthetical) survives the job-detail ``RichLog``,
    which deliberately disables markup to avoid misrendering log content.

    Args:
        job: The job to describe.

    Returns:
        A list of text lines (empty if the job has no external identifier).
    """
    if not job.external_jobid:
        return []

    from snakesee.remote_links import batch_console_url
    from snakesee.remote_links import cloudwatch_url

    label = job.executor or "remote"
    lines = [Text(f"{label} job: {job.external_jobid}")]

    if job.queue is not None:
        lines.append(Text(f"  queue:   {job.queue}"))

    # Queue wait is distinct from run time: it's how long the job waited for a node.
    queue_wait = job.queue_wait
    if queue_wait is not None:
        lines.append(Text(f"  queued for: {format_duration(queue_wait)}"))

    # Attempt > 1 means the job was retried/preempted; worth surfacing.
    if job.attempt is not None and job.attempt > 1:
        lines.append(Text(f"  attempt: {job.attempt}"))

    if job.exit_code is not None:
        lines.append(Text(f"  exit code: {job.exit_code}"))

    # Prefer the executor's structured termination classification (rendered with
    # confidence). Fall back to snakesee's own low-confidence string heuristic only
    # when no structured category arrived (e.g. an older executor).
    from snakesee.remote_termination import SOURCE_STATUS_REASON
    from snakesee.remote_termination import format_termination_marker
    from snakesee.remote_termination import format_termination_source

    marker = format_termination_marker(job.termination_category, job.termination_confidence)
    # Provenance only ever annotates a rendered marker: a source arriving with
    # no usable category (no marker) has nothing to attribute and is dropped.
    source = format_termination_source(job.termination_source) if marker is not None else None
    if job.termination_category is None and job.status_reason:
        from snakesee.remote_links import is_spot_interruption

        if is_spot_interruption(job.status_reason):
            marker = "possibly spot interrupted"
            # The reader-side heuristic inspects the same field as the
            # executor's status_reason source, so it carries the same label.
            source = format_termination_source(SOURCE_STATUS_REASON)
    if marker is not None:
        marker_line = Text(f"  {marker}")
        if source is not None:
            marker_line.append(f" ({source})", style="dim")
        lines.append(marker_line)

    if job.status_reason:
        lines.append(Text(f"  reason: {job.status_reason}"))

    if job.cost_estimate is not None:
        lines.append(Text(f"  est. cost: {format_cost(job.cost_estimate)}"))

    console = batch_console_url(job.external_jobid, region=job.region)
    if console is not None:
        lines.append(Text(f"  console: {console}"))

    logs = cloudwatch_url(job.log_stream, region=job.region)
    if logs is not None:
        lines.append(Text(f"  logs:    {logs}"))

    return lines

make_summary_footer

make_summary_footer(progress: WorkflowProgress) -> Panel

Create the job status summary as a one-line footer panel.

Parameters:

Name	Type	Description	Default
progress	WorkflowProgress	Current workflow progress snapshot.	required

Returns:

Type	Description
Panel	A Rich Panel containing the job status summary.

Source code in snakesee/tui/renderables.py

def make_summary_footer(progress: WorkflowProgress) -> Panel:
    """Create the job status summary as a one-line footer panel.

    Args:
        progress: Current workflow progress snapshot.

    Returns:
        A Rich Panel containing the job status summary.
    """
    succeeded = progress.completed_jobs
    failed = progress.failed_jobs
    running = len(progress.running_jobs)
    incomplete = len(progress.incomplete_jobs_list)
    pending = progress.pending_jobs

    summary = Text()
    summary.append("Jobs: ", style="dim")
    summary.append(f"{succeeded}", style="green")
    summary.append(" succeeded", style="dim")
    summary.append("  │  ", style="dim")
    summary.append(f"{failed}", style="red" if failed > 0 else "dim")
    summary.append(" failed", style="dim")
    summary.append("  │  ", style="dim")
    summary.append(f"{running}", style="cyan" if running > 0 else "dim")
    summary.append(" running", style="dim")
    # Show incomplete count if there are incomplete jobs
    if incomplete > 0:
        summary.append("  │  ", style="dim")
        summary.append(f"{incomplete}", style="yellow")
        summary.append(" incomplete", style="dim")
    summary.append("  │  ", style="dim")
    summary.append(f"{pending}", style="yellow" if pending > 0 else "dim")
    summary.append(" pending", style="dim")

    border_style = "red" if failed > 0 else FG_BLUE
    return Panel(summary, border_style=border_style, padding=(0, 1))

screens

Modal screens: HelpScreen, EasterEggScreen, JobLogScreen.

Classes

EasterEggScreen

Bases: ModalScreen[None]

Modal easter egg overlay; any of escape/space/enter/q dismisses it.

The Fulcrum logo is rendered to fill the terminal at mount time and re-rendered on resize so it tracks window changes.

Source code in snakesee/tui/screens.py

class EasterEggScreen(ModalScreen[None]):
    """Modal easter egg overlay; any of escape/space/enter/q dismisses it.

    The Fulcrum logo is rendered to fill the terminal at mount time and
    re-rendered on resize so it tracks window changes.
    """

    BINDINGS: ClassVar[list[BindingType]] = [Binding("escape,space,enter,q", "app.pop_screen")]
    DEFAULT_CSS = """
    EasterEggScreen {
        align: center middle;
    }
    EasterEggScreen #easter-content {
        width: 100%;
        height: 100%;
    }
    """

    def compose(self) -> ComposeResult:
        """Yield a Static that will be populated with the resized logo on mount."""
        yield Static(id="easter-content")

    def on_mount(self) -> None:
        """Render the logo sized to the current terminal."""
        self._render_logo()

    def on_resize(self) -> None:
        """Re-render the logo when the terminal size changes."""
        self._render_logo()

    def _render_logo(self) -> None:
        size = self.app.size
        self.query_one("#easter-content", Static).update(
            make_easter_egg(console_width=size.width, console_height=size.height)
        )

Methods:

compose

compose() -> ComposeResult

Yield a Static that will be populated with the resized logo on mount.

Source code in snakesee/tui/screens.py

def compose(self) -> ComposeResult:
    """Yield a Static that will be populated with the resized logo on mount."""
    yield Static(id="easter-content")

on_mount

on_mount() -> None

Render the logo sized to the current terminal.

Source code in snakesee/tui/screens.py

def on_mount(self) -> None:
    """Render the logo sized to the current terminal."""
    self._render_logo()

on_resize

on_resize() -> None

Re-render the logo when the terminal size changes.

Source code in snakesee/tui/screens.py

def on_resize(self) -> None:
    """Re-render the logo when the terminal size changes."""
    self._render_logo()

HelpScreen

Bases: ModalScreen[None]

Modal help overlay; any of escape/space/enter/q/? dismisses it.

Source code in snakesee/tui/screens.py

class HelpScreen(ModalScreen[None]):
    """Modal help overlay; any of escape/space/enter/q/? dismisses it."""

    BINDINGS: ClassVar[list[BindingType]] = [
        Binding("escape,space,enter,q,question_mark", "app.pop_screen")
    ]

    def compose(self) -> ComposeResult:
        """Yield a Static containing the rendered help panel."""
        yield Static(make_help(), id="help-content")

Methods:

compose

compose() -> ComposeResult

Yield a Static containing the rendered help panel.

Source code in snakesee/tui/screens.py

def compose(self) -> ComposeResult:
    """Yield a Static containing the rendered help panel."""
    yield Static(make_help(), id="help-content")

JobLogScreen

Bases: ModalScreen[None]

Modal log viewer for a single job; escape or q dismisses it.

The global toggle keys (pause/estimation/wildcard/accessibility/refresh) are re-bound here to the app's actions so they keep working while a log is open — app-level BINDINGS don't fire under a modal screen, so the keys would otherwise be inert in log-viewing mode.

Source code in snakesee/tui/screens.py

class JobLogScreen(ModalScreen[None]):
    """Modal log viewer for a single job; escape or q dismisses it.

    The global toggle keys (pause/estimation/wildcard/accessibility/refresh) are
    re-bound here to the app's actions so they keep working while a log is open —
    app-level BINDINGS don't fire under a modal screen, so the keys would otherwise
    be inert in log-viewing mode.
    """

    BINDINGS: ClassVar[list[BindingType]] = [
        Binding("escape,q", "app.pop_screen"),
        Binding("p", "app.toggle_pause", show=False),
        Binding("e", "app.toggle_estimation", show=False),
        Binding("w", "app.toggle_wildcard", show=False),
        Binding("a", "app.toggle_accessibility", show=False),
        Binding("r", "app.force_refresh", show=False),
        Binding("ctrl+r", "app.hard_refresh", show=False),
    ]

    def __init__(
        self,
        log_path: Path | None,
        lines: list[str],
        header_lines: list[Text] | None = None,
    ) -> None:
        """Initialize with the log path (shown as the border title) and tail lines.

        Args:
            log_path: Path to the job's log file, or None if unknown.
            lines: Tail lines (most recent at end) to render in the RichLog.
            header_lines: Optional styled lines rendered above the log (e.g. a
                remote job's external id and console/CloudWatch links). Rich
                ``Text`` rather than ``str`` so styles survive the markup-less
                RichLog. For a remote job with no local log file, these may be
                the only content.
        """
        super().__init__()
        self._log_path = log_path
        self._lines = lines
        self._header_lines = header_lines or []

    def compose(self) -> ComposeResult:
        """Yield a single RichLog widget that will be populated on mount."""
        log = RichLog(
            id="job-log",
            highlight=True,
            markup=False,
            wrap=False,
            auto_scroll=False,
        )
        if self._log_path is not None:
            log.border_title = str(self._log_path)
        elif self._header_lines:
            log.border_title = "remote job"
        else:
            log.border_title = "job log"
        yield log

    def on_mount(self) -> None:
        """Write the optional header and captured tail lines into the RichLog widget."""
        log = self.query_one("#job-log", RichLog)
        for header_line in self._header_lines:
            log.write(header_line)
        if self._header_lines and self._lines:
            log.write("")  # blank separator between the remote header and the log tail
        for tail_line in self._lines:
            log.write(tail_line)

Methods:

__init__

__init__(log_path: Path | None, lines: list[str], header_lines: list[Text] | None = None) -> None

Initialize with the log path (shown as the border title) and tail lines.

Parameters:

Name	Type	Description	Default
log_path	Path | None	Path to the job's log file, or None if unknown.	required
lines	list[str]	Tail lines (most recent at end) to render in the RichLog.	required
header_lines	list[Text] | None	Optional styled lines rendered above the log (e.g. a remote job's external id and console/CloudWatch links). Rich Text rather than str so styles survive the markup-less RichLog. For a remote job with no local log file, these may be the only content.	None

Source code in snakesee/tui/screens.py

def __init__(
    self,
    log_path: Path | None,
    lines: list[str],
    header_lines: list[Text] | None = None,
) -> None:
    """Initialize with the log path (shown as the border title) and tail lines.

    Args:
        log_path: Path to the job's log file, or None if unknown.
        lines: Tail lines (most recent at end) to render in the RichLog.
        header_lines: Optional styled lines rendered above the log (e.g. a
            remote job's external id and console/CloudWatch links). Rich
            ``Text`` rather than ``str`` so styles survive the markup-less
            RichLog. For a remote job with no local log file, these may be
            the only content.
    """
    super().__init__()
    self._log_path = log_path
    self._lines = lines
    self._header_lines = header_lines or []

compose

compose() -> ComposeResult

Yield a single RichLog widget that will be populated on mount.

Source code in snakesee/tui/screens.py

def compose(self) -> ComposeResult:
    """Yield a single RichLog widget that will be populated on mount."""
    log = RichLog(
        id="job-log",
        highlight=True,
        markup=False,
        wrap=False,
        auto_scroll=False,
    )
    if self._log_path is not None:
        log.border_title = str(self._log_path)
    elif self._header_lines:
        log.border_title = "remote job"
    else:
        log.border_title = "job log"
    yield log

on_mount

on_mount() -> None

Write the optional header and captured tail lines into the RichLog widget.

Source code in snakesee/tui/screens.py

def on_mount(self) -> None:
    """Write the optional header and captured tail lines into the RichLog widget."""
    log = self.query_one("#job-log", RichLog)
    for header_line in self._header_lines:
        log.write(header_line)
    if self._header_lines and self._lines:
        log.write("")  # blank separator between the remote header and the log tail
    for tail_line in self._lines:
        log.write(tail_line)

Functions:

tables

DataTable row builders and sort helpers.

Classes

CompletionRow

Bases: NamedTuple

One row of the recent-completions table.

Source code in snakesee/tui/tables.py

class CompletionRow(NamedTuple):
    """One row of the recent-completions table."""

    job: JobInfo
    is_failed: bool

FailedRow

Bases: NamedTuple

One row of the failed-jobs panel.

Source code in snakesee/tui/tables.py

class FailedRow(NamedTuple):
    """One row of the failed-jobs panel."""

    job: JobInfo

IncompleteRow

Bases: NamedTuple

One row of the incomplete-jobs panel.

Source code in snakesee/tui/tables.py

class IncompleteRow(NamedTuple):
    """One row of the incomplete-jobs panel."""

    job: JobInfo
    display_path: str

PendingRow

Bases: NamedTuple

One row of the pending-jobs table.

Source code in snakesee/tui/tables.py

class PendingRow(NamedTuple):
    """One row of the pending-jobs table."""

    rule: str
    job_count: int

RunningRow

Bases: NamedTuple

One row of the running-jobs table.

Fields mirror what _build_running_job_data returns for each job so that callers can convert the existing data-source tuples with zero additional computation.

Source code in snakesee/tui/tables.py

class RunningRow(NamedTuple):
    """One row of the running-jobs table.

    Fields mirror what ``_build_running_job_data`` returns for each job so that
    callers can convert the existing data-source tuples with zero additional
    computation.
    """

    job: JobInfo
    elapsed_seconds: float | None
    remaining_seconds: float | None
    start_time: float | None
    tool_progress: ToolProgress | None

StatsRow

Bases: NamedTuple

One row of the rule-statistics panel.

rule_display is the rule name for the first thread-count sub-row and an empty string for subsequent sub-rows (visual grouping). threads is the thread-count string shown in the Thr column ("-" when unknown). stats holds the per-thread (or aggregate) timing data.

Source code in snakesee/tui/tables.py

class StatsRow(NamedTuple):
    """One row of the rule-statistics panel.

    ``rule_display`` is the rule name for the first thread-count sub-row and
    an empty string for subsequent sub-rows (visual grouping).  ``threads``
    is the thread-count string shown in the Thr column (``"-"`` when unknown).
    ``stats`` holds the per-thread (or aggregate) timing data.
    """

    rule_display: str
    threads: str
    stats: RuleTimingStats

Functions:

completion_rows

completion_rows(jobs: list[JobInfo], failed_job_ids: set[int]) -> list[CompletionRow]

Build completion rows from a sorted list of completed/failed jobs.

Parameters:

Name	Type	Description	Default
jobs	list[JobInfo]	Ordered list of completed (and failed) jobs as returned by :meth:~snakesee.tui.data_source.WorkflowDataSource.get_completions_sorted.	required
failed_job_ids	set[int]	Set of id(job) values for jobs that failed.	required

Returns:

Type	Description
list[CompletionRow]	List of :class:CompletionRow with one entry per job.

Source code in snakesee/tui/tables.py

def completion_rows(
    jobs: list[JobInfo],
    failed_job_ids: set[int],
) -> list[CompletionRow]:
    """Build completion rows from a sorted list of completed/failed jobs.

    Args:
        jobs: Ordered list of completed (and failed) jobs as returned by
            :meth:`~snakesee.tui.data_source.WorkflowDataSource.get_completions_sorted`.
        failed_job_ids: Set of ``id(job)`` values for jobs that failed.

    Returns:
        List of :class:`CompletionRow` with one entry per job.
    """
    return [CompletionRow(job=job, is_failed=id(job) in failed_job_ids) for job in jobs]

failed_rows

failed_rows(progress: WorkflowProgress) -> list[FailedRow]

Build failed-job rows from workflow progress.

Parameters:

Name	Type	Description	Default
progress	WorkflowProgress	Current :class:~snakesee.models.WorkflowProgress.	required

Returns:

Type	Description
list[FailedRow]	List of :class:FailedRow in the order they appear in
list[FailedRow]	progress.failed_jobs_list.

Source code in snakesee/tui/tables.py

def failed_rows(progress: WorkflowProgress) -> list[FailedRow]:
    """Build failed-job rows from workflow progress.

    Args:
        progress: Current :class:`~snakesee.models.WorkflowProgress`.

    Returns:
        List of :class:`FailedRow` in the order they appear in
        ``progress.failed_jobs_list``.
    """
    return [FailedRow(job=job) for job in progress.failed_jobs_list]

incomplete_rows

incomplete_rows(progress: WorkflowProgress) -> list[IncompleteRow]

Build incomplete-job rows from workflow progress.

Each row's display_path is the output file path relative to the workflow directory when possible, the absolute path otherwise, or the string "unknown" when no output file is recorded.

Parameters:

Name	Type	Description	Default
progress	WorkflowProgress	Current :class:~snakesee.models.WorkflowProgress.	required

Returns:

Type	Description
list[IncompleteRow]	List of :class:IncompleteRow in the order they appear in
list[IncompleteRow]	progress.incomplete_jobs_list.

Source code in snakesee/tui/tables.py

def incomplete_rows(progress: WorkflowProgress) -> list[IncompleteRow]:
    """Build incomplete-job rows from workflow progress.

    Each row's ``display_path`` is the output file path relative to the
    workflow directory when possible, the absolute path otherwise, or the
    string ``"unknown"`` when no output file is recorded.

    Args:
        progress: Current :class:`~snakesee.models.WorkflowProgress`.

    Returns:
        List of :class:`IncompleteRow` in the order they appear in
        ``progress.incomplete_jobs_list``.
    """
    rows: list[IncompleteRow] = []
    for job in progress.incomplete_jobs_list:
        if job.output_file is not None:
            try:
                display_path = str(job.output_file.relative_to(progress.workflow_dir))
            except ValueError:
                display_path = str(job.output_file)
        else:
            display_path = "unknown"
        rows.append(IncompleteRow(job=job, display_path=display_path))
    return rows

pending_rows

pending_rows(pending_rules: dict[str, int]) -> list[PendingRow]

Build pending rows from the inferred pending-rule counts.

The default ordering (count descending) is applied here; callers may re-sort via :func:sort_rows if custom sorting is active.

Parameters:

Name	Type	Description	Default
pending_rules	dict[str, int]	Mapping from rule name to estimated pending count, as returned by :meth:~snakesee.tui.data_source.WorkflowDataSource.get_inferred_pending_rules.	required

Returns:

Type	Description
list[PendingRow]	List of :class:PendingRow sorted by count descending.

Source code in snakesee/tui/tables.py

def pending_rows(pending_rules: dict[str, int]) -> list[PendingRow]:
    """Build pending rows from the inferred pending-rule counts.

    The default ordering (count descending) is applied here; callers may
    re-sort via :func:`sort_rows` if custom sorting is active.

    Args:
        pending_rules: Mapping from rule name to estimated pending count, as
            returned by
            :meth:`~snakesee.tui.data_source.WorkflowDataSource.get_inferred_pending_rules`.

    Returns:
        List of :class:`PendingRow` sorted by count descending.
    """
    rows = [PendingRow(rule=rule, job_count=count) for rule, count in pending_rules.items()]
    rows.sort(key=lambda r: r.job_count, reverse=True)
    return rows

running_rows

running_rows(job_data: list[tuple[JobInfo, float | None, float | None, float | None, ToolProgress | None]]) -> list[RunningRow]

Convert raw running-job data tuples to typed :class:RunningRow objects.

The raw tuples are produced by :meth:~snakesee.tui.data_source.WorkflowDataSource._build_running_job_data, which already computes elapsed, remaining, start_time, and tool_progress. This function is a thin adapter so that consumers hold typed data rather than anonymous tuples.

Parameters:

Name	Type	Description	Default
job_data	list[tuple[JobInfo, float | None, float | None, float | None, ToolProgress | None]]	List of (job, elapsed, remaining, start_time, tool_progress) tuples as returned by WorkflowDataSource._build_running_job_data.	required

Returns:

Type	Description
list[RunningRow]	Equivalent list of :class:RunningRow named tuples.

Source code in snakesee/tui/tables.py

def running_rows(
    job_data: list[tuple[JobInfo, float | None, float | None, float | None, ToolProgress | None]],
) -> list[RunningRow]:
    """Convert raw running-job data tuples to typed :class:`RunningRow` objects.

    The raw tuples are produced by
    :meth:`~snakesee.tui.data_source.WorkflowDataSource._build_running_job_data`,
    which already computes elapsed, remaining, start_time, and tool_progress.
    This function is a thin adapter so that consumers hold typed data rather
    than anonymous tuples.

    Args:
        job_data: List of ``(job, elapsed, remaining, start_time, tool_progress)``
            tuples as returned by ``WorkflowDataSource._build_running_job_data``.

    Returns:
        Equivalent list of :class:`RunningRow` named tuples.
    """
    return [
        RunningRow(job, elapsed, remaining, start, tp)
        for job, elapsed, remaining, start, tp in job_data
    ]

sort_rows

sort_rows(rows: list[_R], column: int, ascending: bool) -> list[_R]

Sort rows by the value at column index, in-place, and return them.

Works on any :class:NamedTuple because named tuples support integer indexing. None values sort last regardless of direction.

Parameters:

Name	Type	Description	Default
rows	list[_R]	List of row tuples to sort.	required
column	int	Zero-based column index to use as the sort key.	required
ascending	bool	When True sort smallest-first; otherwise largest-first.	required

Returns:

Type	Description
list[_R]	The same list, sorted in-place and returned for convenience.

Source code in snakesee/tui/tables.py

def sort_rows(rows: list[_R], column: int, ascending: bool) -> list[_R]:
    """Sort *rows* by the value at *column* index, in-place, and return them.

    Works on any :class:`NamedTuple` because named tuples support integer
    indexing.  ``None`` values sort last regardless of direction.

    Args:
        rows: List of row tuples to sort.
        column: Zero-based column index to use as the sort key.
        ascending: When ``True`` sort smallest-first; otherwise largest-first.

    Returns:
        The same list, sorted in-place and returned for convenience.
    """

    # The leading bucket flag pushes None values to the end regardless of sort
    # direction. With ascending=True we put non-None in bucket 0 and None in
    # bucket 1; with descending we invert (since the final reverse=True flips
    # bucket order too) so None still ends up after non-None.
    none_bucket = 1 if ascending else 0
    value_bucket = 1 - none_bucket

    def _key(row: _R) -> tuple[int, object]:
        val = row[column]
        if val is None:
            return (none_bucket, "")
        if isinstance(val, str):
            return (value_bucket, val.lower())
        return (value_bucket, val)

    rows.sort(key=_key, reverse=not ascending)
    return rows

sort_stats_rows

sort_stats_rows(rows: list[StatsRow], column: int, ascending: bool) -> list[StatsRow]

Sort stats rows by a visible column index, mapped to the right StatsRow field.

Unlike :func:sort_rows, this does not index the tuple positionally — the stats table's visible columns don't line up with the StatsRow shape (see _STATS_SORT_KEYS). An unrecognized column leaves the rows untouched.

Parameters:

Name	Type	Description	Default
rows	list[StatsRow]	List of :class:StatsRow to sort.	required
column	int	Zero-based visible column index (0=Rule, 1=Thr, 2=Count, 3=Avg).	required
ascending	bool	When True sort smallest-first; otherwise largest-first.	required

Returns:

Type	Description
list[StatsRow]	A new sorted list, or the input list unchanged for an unknown column.

Source code in snakesee/tui/tables.py

def sort_stats_rows(rows: list[StatsRow], column: int, ascending: bool) -> list[StatsRow]:
    """Sort stats rows by a *visible* column index, mapped to the right StatsRow field.

    Unlike :func:`sort_rows`, this does not index the tuple positionally — the stats
    table's visible columns don't line up with the StatsRow shape (see
    ``_STATS_SORT_KEYS``). An unrecognized column leaves the rows untouched.

    Args:
        rows: List of :class:`StatsRow` to sort.
        column: Zero-based *visible* column index (0=Rule, 1=Thr, 2=Count, 3=Avg).
        ascending: When ``True`` sort smallest-first; otherwise largest-first.

    Returns:
        A new sorted list, or the input list unchanged for an unknown column.
    """
    key = _STATS_SORT_KEYS.get(column)
    if key is None:
        return rows
    return sorted(rows, key=key, reverse=not ascending)

stats_rows

stats_rows(stats_list: list[RuleTimingStats], thread_stats_dict: dict[str, ThreadTimingStats]) -> list[StatsRow]

Build flattened stats rows for the rule-statistics panel.

Rules with per-thread timing data are expanded into one sub-row per thread count, with the rule name shown only on the first sub-row. Rules without thread data get a single row with "-" in the Thr column.

The default ordering (count descending) should be applied by the caller before calling this function so that the thread expansion preserves the sorted order.

Parameters:

Name	Type	Description	Default
stats_list	list[RuleTimingStats]	Ordered list of :class:~snakesee.models.RuleTimingStats objects (already filtered and sorted by the caller).	required
thread_stats_dict	dict[str, ThreadTimingStats]	Mapping from rule name to :class:~snakesee.models.ThreadTimingStats as returned by :meth:~snakesee.state.rule_registry.RuleRegistry.to_thread_stats_dict.	required

Returns:

Type	Description
list[StatsRow]	Flat list of :class:StatsRow ready for rendering.

Source code in snakesee/tui/tables.py

def stats_rows(
    stats_list: list[RuleTimingStats],
    thread_stats_dict: dict[str, ThreadTimingStats],
) -> list[StatsRow]:
    """Build flattened stats rows for the rule-statistics panel.

    Rules with per-thread timing data are expanded into one sub-row per
    thread count, with the rule name shown only on the first sub-row.  Rules
    without thread data get a single row with ``"-"`` in the Thr column.

    The default ordering (count descending) should be applied by the caller
    *before* calling this function so that the thread expansion preserves the
    sorted order.

    Args:
        stats_list: Ordered list of :class:`~snakesee.models.RuleTimingStats`
            objects (already filtered and sorted by the caller).
        thread_stats_dict: Mapping from rule name to
            :class:`~snakesee.models.ThreadTimingStats` as returned by
            :meth:`~snakesee.state.rule_registry.RuleRegistry.to_thread_stats_dict`.

    Returns:
        Flat list of :class:`StatsRow` ready for rendering.
    """
    rows: list[StatsRow] = []
    for stats in stats_list:
        rule = stats.rule
        if rule in thread_stats_dict and thread_stats_dict[rule].stats_by_threads:
            rule_thread_stats = thread_stats_dict[rule]
            sorted_threads = sorted(rule_thread_stats.stats_by_threads.keys())
            for i, threads in enumerate(sorted_threads):
                ts = rule_thread_stats.stats_by_threads[threads]
                rule_display = rule if i == 0 else ""
                rows.append(StatsRow(rule_display=rule_display, threads=str(threads), stats=ts))
        else:
            rows.append(StatsRow(rule_display=rule, threads="-", stats=stats))
    return rows