Index

panels

TUI panels — individual display components for the monitor layout.

Classes

DetailPanel

DetailPanel(*, name=None, id=None, classes=None)

Bases: VerticalScroll

Renders details for the selected item in a scrollable container.

Content varies based on selected item type: - Process: strace summary, open FDs, full command, environment - Completed sheet: validation results, stdout tail, retry history - Anomaly: description, affected resources, historical context

Source code in src/marianne/tui/panels/detail.py

def __init__(
    self,
    *,
    name: str | None = None,
    id: str | None = None,
    classes: str | None = None,
) -> None:
    super().__init__(name=name, id=id, classes=classes)
    self._content: Static | None = None

Functions

compose

compose()

Build the scrollable content area.

Source code in src/marianne/tui/panels/detail.py

def compose(self) -> Any:
    """Build the scrollable content area."""
    self._content = Static("", id="detail-content")
    yield self._content

show_empty

show_empty()

Show the default empty state.

Source code in src/marianne/tui/panels/detail.py

def show_empty(self) -> None:
    """Show the default empty state."""
    self._set_content(
        "[dim]Select a process/event to see: full strace, logs, "
        "validation details, resource history, or learning correlations[/]"
    )

show_process

show_process(proc)

Show detailed process information.

Source code in src/marianne/tui/panels/detail.py

def show_process(self, proc: ProcessMetric) -> None:
    """Show detailed process information."""
    lines: list[str] = []

    # Header
    lines.append(f"[bold]Process Detail: PID {proc.pid}[/]")
    lines.append("")

    # Command
    cmd_display = proc.command if proc.command else "[dim]unknown[/]"
    lines.append(f"  Command:  {cmd_display}")
    lines.append(f"  State:    {proc.state}")
    lines.append(
        f"  CPU:      {proc.cpu_percent:.1f}%    "
        f"Memory: {_format_bytes_mb(proc.rss_mb)} RSS / {_format_bytes_mb(proc.vms_mb)} VMS"
    )
    lines.append(f"  Threads:  {proc.threads}    Open FDs: {proc.open_fds}")

    if proc.job_id:
        sheet_str = f"  Sheet: S{proc.sheet_num}" if proc.sheet_num is not None else ""
        lines.append(f"  Job:      {proc.job_id}{sheet_str}")

    # Syscall summary
    if proc.syscall_counts:
        lines.append("")
        lines.append("  [bold]Syscall Summary:[/]")
        sorted_sc = sorted(
            proc.syscall_counts.items(), key=lambda x: x[1], reverse=True
        )[:10]
        for sc_name, count in sorted_sc:
            time_pct = proc.syscall_time_pct.get(sc_name, 0.0)
            lines.append(f"    {sc_name:<16s} count={count:>8,}  time={time_pct:.1f}%")
    elif proc.syscall_time_pct:
        lines.append("")
        lines.append("  [bold]Syscall Time:[/]")
        sorted_time = sorted(
            proc.syscall_time_pct.items(), key=lambda x: x[1], reverse=True
        )[:10]
        for sc_name, pct in sorted_time:
            lines.append(f"    {sc_name:<16s} {pct:.1f}%")

    self._set_content("\n".join(lines))

show_anomaly

show_anomaly(anomaly)

Show detailed anomaly information.

Source code in src/marianne/tui/panels/detail.py

def show_anomaly(self, anomaly: Anomaly) -> None:
    """Show detailed anomaly information."""
    lines: list[str] = []

    sev_color = {
        "low": "green",
        "medium": "yellow",
        "high": "bold yellow",
        "critical": "bold red",
    }.get(anomaly.severity.value, "white")

    lines.append(f"[bold]Anomaly: {anomaly.anomaly_type.value}[/]")
    lines.append("")
    lines.append(f"  Severity:  [{sev_color}]{anomaly.severity.value.upper()}[/]")
    lines.append(f"  Value:     {anomaly.metric_value:.1f}")
    lines.append(f"  Threshold: {anomaly.threshold:.1f}")

    if anomaly.pid is not None:
        lines.append(f"  PID:       {anomaly.pid}")
    if anomaly.job_id:
        sheet_str = f"  Sheet: S{anomaly.sheet_num}" if anomaly.sheet_num is not None else ""
        lines.append(f"  Job:       {anomaly.job_id}{sheet_str}")

    if anomaly.description:
        lines.append("")
        lines.append(f"  {anomaly.description}")

    self._set_content("\n".join(lines))

show_file_activity

show_file_activity(events)

Show recent file activity from observer events.

Parameters:

Name	Type	Description	Default
events	list[dict[str, Any]]	Observer events filtered to observer.file_* types.	required

Source code in src/marianne/tui/panels/detail.py

def show_file_activity(self, events: list[dict[str, Any]]) -> None:
    """Show recent file activity from observer events.

    Args:
        events: Observer events filtered to ``observer.file_*`` types.
    """
    if not events:
        self._set_content("[dim]No file activity[/]")
        return

    lines: list[str] = []
    lines.append("[bold]File Activity[/]")
    lines.append("")

    # Show most recent events (already newest-first from recorder)
    for evt in events[:20]:
        evt_name = evt.get("event", "")
        data = evt.get("data") or {}
        path = data.get("path", "unknown")
        ts = evt.get("timestamp", 0)
        ts_str = datetime.fromtimestamp(ts).strftime("%H:%M:%S") if ts else "??:??:??"

        if "created" in evt_name:
            action = "[green]+[/]"
        elif "deleted" in evt_name:
            action = "[red]-[/]"
        else:
            action = "[yellow]~[/]"

        lines.append(f"  {ts_str}  {action} {path}")

    self._set_content("\n".join(lines))

show_item

show_item(item)

Show details for a generic selected item.

The item dict should have a 'type' key ('process', 'anomaly', 'job').

Source code in src/marianne/tui/panels/detail.py

def show_item(self, item: dict[str, Any] | None) -> None:
    """Show details for a generic selected item.

    The item dict should have a 'type' key ('process', 'anomaly', 'job').
    """
    if item is None:
        self.show_empty()
        return

    item_type = item.get("type")
    if item_type == "process" and "process" in item:
        self.show_process(item["process"])
    elif item_type == "anomaly" and "anomaly" in item:
        self.show_anomaly(item["anomaly"])
    elif item_type == "job":
        job_id = item.get("job_id", "unknown")
        procs = item.get("processes", [])
        total_cpu = sum(p.cpu_percent for p in procs)
        total_mem = sum(p.rss_mb for p in procs)
        lines = [
            f"[bold]Job: {job_id}[/]",
            "",
            f"  Processes: {len(procs)}",
            f"  Total CPU: {total_cpu:.1f}%",
            f"  Total MEM: {_format_bytes_mb(total_mem)}",
        ]
        # Append file activity if observer events are present
        file_events = item.get("observer_file_events", [])
        if file_events:
            lines.append("")
            lines.append("[bold]File Activity[/]")
            for evt in file_events[:10]:
                evt_name = evt.get("event", "")
                data = evt.get("data") or {}
                path = data.get("path", "unknown")
                ts = evt.get("timestamp", 0)
                ts_str = datetime.fromtimestamp(ts).strftime("%H:%M:%S") if ts else "??:??:??"
                if "created" in evt_name:
                    action = "[green]+[/]"
                elif "deleted" in evt_name:
                    action = "[red]-[/]"
                else:
                    action = "[yellow]~[/]"
                lines.append(f"  {ts_str}  {action} {path}")
        self._set_content("\n".join(lines))
    else:
        self.show_empty()

HeaderPanel

HeaderPanel(*, name=None, id=None, classes=None)

Bases: Static

Renders the system summary header with Rich markup.

Layout matches the htop style: Conductor: UP 2h15m CPU [|||| ] 12.0% Mem [|||||||| ] 45.0% Pressure: LOW Jobs: 2 Sheets: 3 active

Source code in src/marianne/tui/panels/header.py

def __init__(
    self,
    *,
    name: str | None = None,
    id: str | None = None,
    classes: str | None = None,
) -> None:
    super().__init__(name=name, id=id, classes=classes)
    self._snapshot: SystemSnapshot | None = None
    self._conductor_up: bool = False
    self._uptime_seconds: float = 0.0
    self._anomaly_count: int = 0

Functions

update_data

update_data(snapshot, conductor_up=False, uptime_seconds=0.0, anomaly_count=0)

Update the header with new snapshot data and refresh display.

Source code in src/marianne/tui/panels/header.py

def update_data(
    self,
    snapshot: SystemSnapshot | None,
    conductor_up: bool = False,
    uptime_seconds: float = 0.0,
    anomaly_count: int = 0,
) -> None:
    """Update the header with new snapshot data and refresh display."""
    self._snapshot = snapshot
    self._conductor_up = conductor_up
    self._uptime_seconds = uptime_seconds
    self._anomaly_count = anomaly_count
    self._render_header()

JobsPanel

JobsPanel(*, name=None, id=None, classes=None)

Bases: VerticalScroll

Renders the job tree with per-process metrics in a scrollable container.

Jobs are displayed as collapsible tree nodes. When there are many jobs, they start collapsed; with few jobs, they start expanded.

Source code in src/marianne/tui/panels/jobs.py

def __init__(
    self,
    *,
    name: str | None = None,
    id: str | None = None,
    classes: str | None = None,
) -> None:
    super().__init__(name=name, id=id, classes=classes)
    self._snapshot: SystemSnapshot | None = None
    self._selected_index: int = 0
    self._items: list[dict[str, Any]] = []
    self._tree: Tree[dict[str, Any]] | None = None
    self._empty_label: Static | None = None
    self._observer_file_events: list[dict[str, Any]] = []
    self._sort_key: str = "job_id"
    self._filter_query: str = ""
    self._fleet_data: list[dict[str, Any]] = []

Attributes

selected_item property

selected_item

Return the currently selected item, if any.

Functions

compose

compose()

Build the widget tree with a Tree for collapsible jobs.

Source code in src/marianne/tui/panels/jobs.py

def compose(self) -> Any:
    """Build the widget tree with a Tree for collapsible jobs."""
    self._empty_label = Static("[dim]No active jobs[/]", id="jobs-empty")
    yield self._empty_label
    tree: Tree[dict[str, Any]] = Tree("Jobs", id="jobs-tree")
    tree.show_root = False
    tree.guide_depth = 3
    self._tree = tree
    yield tree

select_next

select_next()

Move selection down in the tree.

Source code in src/marianne/tui/panels/jobs.py

def select_next(self) -> None:
    """Move selection down in the tree."""
    if self._tree is not None:
        self._tree.action_cursor_down()
    if self._items:
        self._selected_index = min(
            self._selected_index + 1, len(self._items) - 1
        )

select_prev

select_prev()

Move selection up in the tree.

Source code in src/marianne/tui/panels/jobs.py

def select_prev(self) -> None:
    """Move selection up in the tree."""
    if self._tree is not None:
        self._tree.action_cursor_up()
    if self._items:
        self._selected_index = max(self._selected_index - 1, 0)

update_data

update_data(snapshot, observer_file_events=None, fleet_data=None)

Update the panel with new snapshot data.

Parameters:

Name	Type	Description	Default
snapshot	SystemSnapshot | None	Current system snapshot with process metrics.	required
observer_file_events	list[dict[str, Any]] | None	Observer file events for job correlation.	None
fleet_data	list[dict[str, Any]] | None	Fleet status dicts from fleet manager. Each dict has fleet_id, name, members (list of {job_id, group, status}).	None

Source code in src/marianne/tui/panels/jobs.py

def update_data(
    self,
    snapshot: SystemSnapshot | None,
    observer_file_events: list[dict[str, Any]] | None = None,
    fleet_data: list[dict[str, Any]] | None = None,
) -> None:
    """Update the panel with new snapshot data.

    Args:
        snapshot: Current system snapshot with process metrics.
        observer_file_events: Observer file events for job correlation.
        fleet_data: Fleet status dicts from fleet manager. Each dict has
            fleet_id, name, members (list of {job_id, group, status}).
    """
    self._snapshot = snapshot
    if observer_file_events is not None:
        self._observer_file_events = observer_file_events
    if fleet_data is not None:
        self._fleet_data = fleet_data
    self._render_jobs()

TimelinePanel

TimelinePanel(*, name=None, id=None, classes=None)

Bases: RichLog

Renders a scrollable event timeline with color coding.

Uses RichLog for built-in scrolling and append-oriented display.

Events are color-coded by type: - blue: SPAWN - green: EXIT - yellow: SIGNAL - red: KILL/OOM/ANOMALY - magenta: LEARNING insights

Source code in src/marianne/tui/panels/timeline.py

def __init__(
    self,
    *,
    name: str | None = None,
    id: str | None = None,
    classes: str | None = None,
) -> None:
    super().__init__(
        name=name, id=id, classes=classes, wrap=True, max_lines=self.MAX_VISIBLE, markup=True
    )
    self._events: list[ProcessEvent] = []
    self._anomalies: list[Anomaly] = []
    self._learning_insights: list[dict[str, Any]] = []
    self._observer_events: list[dict[str, Any]] = []

Functions

update_data

update_data(events=None, anomalies=None, learning_insights=None, observer_events=None)

Update the timeline with new data and refresh display.

Source code in src/marianne/tui/panels/timeline.py

def update_data(
    self,
    events: list[ProcessEvent] | None = None,
    anomalies: list[Anomaly] | None = None,
    learning_insights: list[dict[str, Any]] | None = None,
    observer_events: list[dict[str, Any]] | None = None,
) -> None:
    """Update the timeline with new data and refresh display."""
    if events is not None:
        self._events = events
    if anomalies is not None:
        self._anomalies = anomalies
    if learning_insights is not None:
        self._learning_insights = learning_insights
    if observer_events is not None:
        self._observer_events = observer_events
    self._render_timeline()

add_event

add_event(event)

Add a single event to the timeline incrementally.

Source code in src/marianne/tui/panels/timeline.py

def add_event(self, event: dict[str, Any]) -> None:
    """Add a single event to the timeline incrementally."""
    line = self._format_event_line(event)
    if line:
        self.write(line)