cjm-fasthtml-job-monitor 0.0.6

A FastHTML component library for monitoring plugin job execution with progress tracking, log tailing, resource usage, and cancellation via a tabbed modal with content overlay.

cjm-fasthtml-job-monitor

Install

pip install cjm_fasthtml_job_monitor

Project Structure

nbs/
├── components/ (6)
│   ├── tabs/ (3)
│   │   ├── logs_tab.ipynb       # Log tailing view with auto-scroll.
│   │   ├── progress_tab.ipynb   # Progress bar, stage message, elapsed time, and job status badge.
│   │   └── resources_tab.ipynb  # Worker CPU, RAM, and GPU resource usage display.
│   ├── modal.ipynb    # Tabbed modal with progress, logs, and resources tabs.
│   ├── overlay.ipynb  # Semi-transparent content overlay with loading spinner.
│   └── trigger.ipynb  # Trigger button and progress button for job monitor.
├── routes/ (1)
│   └── init.ipynb  # Route factory for job trigger, SSE progress streaming, and cancellation.
├── services/ (1)
│   └── monitor.ipynb  # Service for job execution monitoring with resource telemetry.
├── html_ids.ipynb  # Prefix-based HTML ID generator for job monitor DOM elements.
└── models.ipynb    # Data types for job monitor URLs, configuration, and resource snapshots.

Total: 10 notebooks across 3 directories

Module Dependencies

graph LR
    components_modal[components.modal<br/>Modal Component]
    components_overlay[components.overlay<br/>Overlay Components]
    components_tabs_logs_tab[components.tabs.logs_tab<br/>Logs Tab]
    components_tabs_progress_tab[components.tabs.progress_tab<br/>Progress Tab]
    components_tabs_resources_tab[components.tabs.resources_tab<br/>Resources Tab]
    components_trigger[components.trigger<br/>Trigger Components]
    html_ids[html_ids<br/>HTML IDs]
    models[models<br/>Models]
    routes_init[routes.init<br/>Route Factory]
    services_monitor[services.monitor<br/>Monitor Service]

    components_modal --> models
    components_modal --> html_ids
    components_modal --> components_tabs_logs_tab
    components_modal --> components_tabs_resources_tab
    components_modal --> components_tabs_progress_tab
    components_overlay --> html_ids
    components_overlay --> models
    components_tabs_logs_tab --> html_ids
    components_tabs_progress_tab --> html_ids
    components_tabs_resources_tab --> models
    components_trigger --> html_ids
    components_trigger --> models
    routes_init --> components_overlay
    routes_init --> html_ids
    routes_init --> components_modal
    routes_init --> models
    routes_init --> components_trigger
    routes_init --> services_monitor
    services_monitor --> models

19 cross-module dependencies detected

CLI Reference

No CLI commands found in this project.

Module Overview

Detailed documentation for each module in the project:

HTML IDs (html_ids.ipynb)

Prefix-based HTML ID generator for job monitor DOM elements.

Import

from cjm_fasthtml_job_monitor.html_ids import (
    JobMonitorHtmlIds
)

Classes

@dataclass
class JobMonitorHtmlIds:
    "Prefix-based HTML ID generator for job monitor DOM elements."
    
    prefix: str  # ID prefix for this job monitor instance
    
    def modal(self) -> str:  # The modal dialog element
            """Modal dialog element."""
            return f"{self.prefix}-modal"
    
        @property
        def modal_content(self) -> str:  # Static modal body (not replaced by SSE)
        "Modal dialog element."
    
    def modal_content(self) -> str:  # Static modal body (not replaced by SSE)
            """Modal body container (holds tabs + footer, rendered once)."""
            return f"{self.prefix}-modal-content"
    
        @property
        def sse_connection(self) -> str:  # Hidden div that carries SSE connection
        "Modal body container (holds tabs + footer, rendered once)."
    
    def sse_connection(self) -> str:  # Hidden div that carries SSE connection
            """Hidden SSE connection element (sse-connect, receives push updates)."""
            return f"{self.prefix}-sse"
    
        # --- Overlay ---
    
        @property
        def overlay(self) -> str:  # Semi-transparent content blocker
        "Hidden SSE connection element (sse-connect, receives push updates)."
    
    def overlay(self) -> str:  # Semi-transparent content blocker
            """Content overlay element."""
            return f"{self.prefix}-overlay"
    
        # --- Trigger slot ---
    
        @property
        def trigger_slot(self) -> str:  # Slot for trigger/progress button
        "Content overlay element."
    
    def trigger_slot(self) -> str:  # Slot for trigger/progress button
            """Trigger button slot (swaps between trigger and progress button)."""
            return f"{self.prefix}-trigger-slot"
    
        # --- Progress ---
    
        @property
        def progress_bar(self) -> str:  # Progress bar element
        "Trigger button slot (swaps between trigger and progress button)."
    
    def progress_bar(self) -> str:  # Progress bar element
            """Progress bar element."""
            return f"{self.prefix}-progress-bar"
    
        @property
        def elapsed(self) -> str:  # Elapsed time display
        "Progress bar element."
    
    def elapsed(self) -> str:  # Elapsed time display
            """Elapsed time display element."""
            return f"{self.prefix}-elapsed"
    
        # --- Tabs ---
    
        @property
        def tabs(self) -> str:  # Tab container
        "Elapsed time display element."
    
    def tabs(self) -> str:  # Tab container
            """Tab navigation container."""
            return f"{self.prefix}-tabs"
    
        @property
        def tab_progress(self) -> str:  # Progress tab inner content
        "Tab navigation container."
    
    def tab_progress(self) -> str:  # Progress tab inner content
            """Progress tab inner content (OOB target)."""
            return f"{self.prefix}-tab-progress"
    
        @property
        def tab_logs(self) -> str:  # Logs tab inner content
        "Progress tab inner content (OOB target)."
    
    def tab_logs(self) -> str:  # Logs tab inner content
            """Logs tab inner content (OOB target)."""
            return f"{self.prefix}-tab-logs"
    
        @property
        def tab_resources(self) -> str:  # Resources tab inner content
        "Logs tab inner content (OOB target)."
    
    def tab_resources(self) -> str:  # Resources tab inner content
            """Resources tab inner content (OOB target)."""
            return f"{self.prefix}-tab-resources"
    
        # --- Footer ---
    
        @property
        def modal_footer(self) -> str:  # Modal footer (cancel button area)
        "Resources tab inner content (OOB target)."
    
    def modal_footer(self) -> str:  # Modal footer (cancel button area)
            """Modal footer (OOB target for cancel button show/hide)."""
            return f"{self.prefix}-modal-footer"
    
        # --- Logs ---
    
        @property
        def log_container(self) -> str:  # Scrollable log display
        "Modal footer (OOB target for cancel button show/hide)."
    
    def log_container(self) -> str:  # Scrollable log display
            """Log tailing container."""
            return f"{self.prefix}-log-container"
    
        # --- Keyboard script slot ---
    
        @property
        def kb_script(self) -> str:  # OOB target for keyboard pause/resume scripts
        "Log tailing container."
    
    def kb_script(self) -> str:  # OOB target for keyboard pause/resume scripts
        "Hidden div for keyboard pause/resume script execution via OOB swap."

Route Factory (init.ipynb)

Route factory for job trigger, SSE progress streaming, and cancellation.

Import

from cjm_fasthtml_job_monitor.routes.init import (
    init_job_monitor_routes,
    check_inflight_job
)

Functions

def _get_job_data(service, job_id):
    """Extract job status fields from a Job object."""
    job = service.get_job(job_id)
    if not job
    "Extract job status fields from a Job object."

def _get_step_state(state_store, workflow_id, session_id, step_id):
    """Get step state dict from the state store."""
    state = state_store.get_state(workflow_id, session_id)
    return state.get("step_states", {}).get(step_id, {}), state


def _update_step_state(state_store, workflow_id, session_id, step_id, step_state, state)
    "Get step state dict from the state store."

def _update_step_state(state_store, workflow_id, session_id, step_id, step_state, state)
    "Write step state back to the state store."

def init_job_monitor_routes(
    monitor_service: JobMonitorService,           # Service instance
    plugin_name: str,                             # Target plugin for jobs
    state_store: SQLiteWorkflowStateStore,         # For persisting job_id
    workflow_id: str,                             # Workflow ID for state access
    step_id: str,                                 # Step ID for state access
    state_key: str,                               # Key in step state for sequence tracker
    prefix: str,                                  # URL prefix
    overlay_target_id: str,                       # ID of element to overlay
    kb_system_id: Optional[str] = None,           # Keyboard system ID to pause/resume
    on_complete: Optional[Callable] = None,       # async fn(results, request, sess) -> List[FT]
    on_cancel: Optional[Callable] = None,         # async fn(job, request, sess) -> List[FT]
    on_fail: Optional[Callable] = None,           # async fn(job, request, sess) -> List[FT]
    job_args_builder: Optional[Callable] = None,  # fn(state_store, workflow_id, session_id) -> List[(args, kwargs)]
    config: Optional[JobMonitorConfig] = None,    # UI config
    id_prefix: str = "jm",                        # HTML ID prefix
    icon_fn: Optional[Callable] = None,           # Icon renderer fn(name, **kwargs) -> FT
    restore_trigger_on_complete: bool = True,      # Restore trigger button after completion (False if on_complete handles it)
) -> Tuple[APIRouter, JobMonitorUrls, JobMonitorHtmlIds]:  # (router, urls, ids)
    """
    Initialize job monitor routes with SSE-based progress streaming.
    
    Supports job sequences: `job_args_builder` returns a list of (args, kwargs)
    tuples. Jobs are submitted sequentially, with aggregate progress tracking.
    Single-source is a list of length 1.
    
    `on_complete` receives a list of job result objects (one per source).
    
    When `restore_trigger_on_complete` is False, the trigger slot is not
    restored after completion — the consumer's `on_complete` callback is
    responsible for the post-completion UI in that slot.
    """

def check_inflight_job(
    monitor_service: JobMonitorService,       # Service instance
    plugin_name: str,                         # Target plugin
    state_store: SQLiteWorkflowStateStore,    # State store
    workflow_id: str,                         # Workflow ID
    session_id: str,                          # Session ID
    step_id: str,                             # Step ID
    state_key: str,                           # State key for sequence tracker
    config: JobMonitorConfig,                 # Display config
    ids: JobMonitorHtmlIds,                   # Element IDs
    urls: JobMonitorUrls,                     # Route URLs
    icon_fn: Optional[Callable] = None,       # Icon renderer
) -> Tuple[Optional[FT], Optional[FT], Optional[FT], bool]
    """
    Check for in-flight job sequence and return appropriate UI state.
    
    The state_key stores a sequence tracker dict with a "job_id" field.
    
    Returns:
        - Button element (trigger or progress button)
        - Overlay element (active overlay or empty placeholder)
        - Modal element (with SSE connection if running, or empty placeholder)
        - Whether a job is currently running
    """

Variables

_TERMINAL_STATUSES

Logs Tab (logs_tab.ipynb)

Log tailing view with auto-scroll.

Import

from cjm_fasthtml_job_monitor.components.tabs.logs_tab import (
    render_logs_tab
)

Functions

def render_logs_tab(
    ids: JobMonitorHtmlIds,  # Element IDs
    logs: str = '',          # Log text content
) -> FT:  # Logs tab content
    "Render logs tab with auto-scroll to bottom."

Modal Component (modal.ipynb)

Tabbed modal with progress, logs, and resources tabs.

Import

from cjm_fasthtml_job_monitor.components.modal import (
    render_sse_connection,
    render_tab_content_oob,
    render_footer_oob,
    render_sse_response,
    get_sse_headers,
    render_job_modal
)

Functions

def render_sse_connection(
    ids: JobMonitorHtmlIds,       # Element IDs
    urls: JobMonitorUrls,         # Route URLs (progress URL used for SSE endpoint)
    job_id: str,                  # Active job ID (passed as query param)
    is_active: bool = True,       # Whether SSE should be connected
) -> FT:  # Hidden SSE connection div
    """
    Render the hidden SSE connection element.
    
    When `is_active`, includes hx-ext=sse and sse-connect for real-time updates.
    When not active, renders as an inert hidden div (no connection).
    """

def render_tab_content_oob(
    ids: JobMonitorHtmlIds,                      # Element IDs
    status: str = 'pending',                     # Job status
    progress_value: float = 0.0,                 # 0.0 to 1.0
    status_message: str = '',                    # Stage message
    started_at: Optional[float] = None,          # Unix timestamp
    completed_at: Optional[float] = None,        # Unix timestamp
    logs: str = '',                              # Log text
    resources: Optional[ResourceSnapshot] = None, # Resource data
) -> tuple:  # (progress_div, logs_div, resources_div) with OOB attrs
    "Render the three tab inner content divs as OOB swap targets."

def render_footer_oob(
    ids: JobMonitorHtmlIds,    # Element IDs
    urls: JobMonitorUrls,      # Route URLs
    is_active: bool = True,    # Whether job is active (show cancel)
) -> FT:  # Footer div with OOB attr
    "Render the modal footer with cancel button as OOB swap."

def _make_tab_oob(ids, tab_id, render_fn, *args, **kwargs):
    """Render a single tab's inner content as OOB swap target."""
    div = Div(render_fn(*args, **kwargs), id=tab_id)
    div.attrs['hx-swap-oob'] = "true"
    return div


def render_sse_response(
    ids: JobMonitorHtmlIds,                      # Element IDs
    urls: JobMonitorUrls,                        # Route URLs
    status: str = 'pending',                     # Job status
    progress_value: float = 0.0,                 # 0.0 to 1.0
    status_message: str = '',                    # Stage message
    started_at: Optional[float] = None,          # Unix timestamp
    completed_at: Optional[float] = None,        # Unix timestamp
    logs: Optional[str] = None,                  # Log text (None = skip logs tab update)
    resources: Optional[ResourceSnapshot] = None, # Resource data (None = skip resources tab update)
    include_footer: bool = False,                # Include footer OOB (only needed on state transitions)
    extra_oob: Optional[List[FT]] = None,        # Additional OOB elements (cleanup, callbacks)
) -> FT:  # Div wrapping all OOB elements for sse_message()
    "Render a single tab's inner content as OOB swap target."

def render_sse_response(
    ids: JobMonitorHtmlIds,                      # Element IDs
    urls: JobMonitorUrls,                        # Route URLs
    status: str = 'pending',                     # Job status
    progress_value: float = 0.0,                 # 0.0 to 1.0
    status_message: str = '',                    # Stage message
    started_at: Optional[float] = None,          # Unix timestamp
    completed_at: Optional[float] = None,        # Unix timestamp
    logs: Optional[str] = None,                  # Log text (None = skip logs tab update)
    resources: Optional[ResourceSnapshot] = None, # Resource data (None = skip resources tab update)
    include_footer: bool = False,                # Include footer OOB (only needed on state transitions)
    extra_oob: Optional[List[FT]] = None,        # Additional OOB elements (cleanup, callbacks)
) -> FT:  # Div wrapping all OOB elements for sse_message()
    """
    Build the OOB update payload for an SSE push.
    
    Selectively includes only changed elements:
    - Progress tab: always included (status/progress/elapsed change frequently)
    - Logs tab: included only when `logs` is not None
    - Resources tab: included only when `resources` is not None
    - Footer: included only when `include_footer` is True (state transitions)
    """

def get_sse_headers() -> List[FT]
    """
    Return the HTMX SSE extension script + cleanup script for app headers.
    
    Consumers should add these to their FastHTML app's hdrs list.
    The SSE extension must load after the main HTMX script.
    """

def render_job_modal(
    config: JobMonitorConfig,                    # Display config
    ids: JobMonitorHtmlIds,                      # Element IDs
    urls: JobMonitorUrls,                        # Route URLs
    job_id: str = '',                            # Active job ID for SSE connection
    status: str = 'pending',                     # Job status
    progress_value: float = 0.0,                 # 0.0 to 1.0
    status_message: str = '',                    # Stage message
    started_at: Optional[float] = None,          # Unix timestamp
    completed_at: Optional[float] = None,        # Unix timestamp
    logs: str = '',                              # Log text
    resources: Optional[ResourceSnapshot] = None, # Resource data
    open_on_render: bool = False,                # Auto-open via JS
) -> FT:  # Dialog element
    """
    Render the full tabbed modal dialog.
    
    The tab structure (radio inputs + tab-content wrappers) is static.
    Each tab-content wrapper contains an inner div with a stable ID
    that gets OOB-swapped by the SSE stream. This prevents the
    selected tab from resetting on each update.
    
    Closable via: Escape key, X button (top-right), or clicking backdrop.
    """

Variables

_TERMINAL_STATUSES

Models (models.ipynb)

Data types for job monitor URLs, configuration, and resource snapshots.

Import

from cjm_fasthtml_job_monitor.models import (
    JobMonitorUrls,
    ResourceSnapshot,
    JobMonitorConfig
)

Classes

@dataclass
class JobMonitorUrls:
    "URL endpoints for the job monitor routes."
    
    trigger: str  # POST -- submit job
    progress: str  # GET -- poll progress
    cancel: str  # POST -- cancel job

@dataclass
class ResourceSnapshot:
    "Point-in-time resource usage for a worker."
    
    worker_pid: int  # Worker process ID
    cpu_percent: float  # CPU utilization %
    memory_rss_mb: float  # Process tree RSS in MB
    gpu_memory_mb: Optional[float]  # Per-process GPU memory in MB
    gpu_index: Optional[int]  # Which GPU device
    gpu_name: Optional[str]  # GPU device name
    gpu_total_mb: Optional[float]  # Total GPU memory in MB
    gpu_load_percent: Optional[float]  # GPU compute utilization %

@dataclass
class JobMonitorConfig:
    "Configuration for a job monitor instance."
    
    modal_title: str = 'Job Execution'  # Modal header title
    trigger_label: str = 'Run'  # Trigger button label
    trigger_icon: Optional[str]  # Lucide icon name for trigger button
    progress_label: str = 'View Progress'  # Progress button label (when modal closed)
    sse_interval_s: float = 0.5  # Server-side SSE push interval in seconds
    log_lines: int = 50  # Number of log lines to show
    overlay_z_index: int = 10  # Overlay z-index

Monitor Service (monitor.ipynb)

Service for job execution monitoring with resource telemetry.

Import

from cjm_fasthtml_job_monitor.services.monitor import (
    JobMonitorService
)

Functions

async def _submit_job(
    self,
    plugin_name: str,       # Target plugin
    *args,
    priority: int = 0,
    **kwargs
) -> str:  # job_id
    "Submit a job to the queue."

def _get_job(self, job_id: str) -> Optional[Job]:  # Job or None
    """Get job by ID."""
    return self._queue.get_job(job_id)

async def _cancel_job(self, job_id: str) -> bool:  # True if cancelled
    "Get job by ID."

async def _cancel_job(self, job_id: str) -> bool:  # True if cancelled
    "Cancel a job."

def _get_logs(
    self,
    plugin_name: str,              # Plugin whose logs to read
    lines: int = 50,               # Max lines to return
    current_session_only: bool = True,  # Filter to current session
) -> str:  # Log text
    "Get plugin logs, optionally filtered to current session."

def _filter_current_session(
    raw: str,        # Full log text
    max_lines: int,  # Max lines to return
) -> str:  # Filtered log text
    "Extract logs from the most recent session (after last '--- Starting' marker)."

def _get_resource_snapshot(
    self,
    plugin_name: str,  # Plugin whose worker to query
) -> Optional[ResourceSnapshot]:  # Snapshot or None
    "Get current resource usage for a plugin's worker."

def _enrich_gpu_stats(
    self,
    snapshot: ResourceSnapshot,  # Snapshot to enrich in place
) -> None
    "Add per-process GPU stats from system monitor plugin."

Classes

class JobMonitorService:
    def __init__(
        self,
        queue: JobQueue,                          # Job queue instance
        manager: PluginManager,                   # For worker stats, logs, sysmon
        sysmon_plugin_name: Optional[str] = None, # System monitor plugin name (e.g., 'cjm-system-monitor-nvidia')
    )
    "Service for job execution monitoring with resource telemetry."
    
    def __init__(
            self,
            queue: JobQueue,                          # Job queue instance
            manager: PluginManager,                   # For worker stats, logs, sysmon
            sysmon_plugin_name: Optional[str] = None, # System monitor plugin name (e.g., 'cjm-system-monitor-nvidia')
        )

Overlay Components (overlay.ipynb)

Semi-transparent content overlay with loading spinner.

Import

from cjm_fasthtml_job_monitor.components.overlay import (
    render_job_overlay,
    render_job_overlay_placeholder
)

Functions

def render_job_overlay(
    ids: JobMonitorHtmlIds,       # Element IDs
    config: JobMonitorConfig,     # Display config (for z-index)
) -> FT:  # Overlay div with centered spinner
    "Render semi-transparent overlay with loading spinner."

def render_job_overlay_placeholder(
    ids: JobMonitorHtmlIds,  # Element IDs
) -> FT:  # Empty div with overlay ID
    "Render empty placeholder (swapping this in removes the overlay)."

Progress Tab (progress_tab.ipynb)

Progress bar, stage message, elapsed time, and job status badge.

Import

from cjm_fasthtml_job_monitor.components.tabs.progress_tab import (
    render_progress_tab
)

Functions

def _render_status_badge(
    status: str,  # Job status string
) -> FT:  # Badge element
    "Render a colored badge for job status."

def _format_elapsed(
    started_at: Optional[float],   # Unix timestamp when job started
    completed_at: Optional[float], # Unix timestamp when job completed
) -> str:  # Formatted elapsed time string
    "Format elapsed time as M:SS."

def _elapsed_timer_script(
    ids: JobMonitorHtmlIds,      # Element IDs
    started_at: Optional[float], # Unix timestamp
) -> FT:  # Script element for client-side timer
    "Generate JS for client-side elapsed time updates."

def render_progress_tab(
    ids: JobMonitorHtmlIds,                # Element IDs
    status: str = 'pending',               # Job status
    progress_value: float = 0.0,           # 0.0 to 1.0
    status_message: str = '',              # Stage message
    started_at: Optional[float] = None,    # Unix timestamp
    completed_at: Optional[float] = None,  # Unix timestamp
) -> FT:  # Progress tab content
    "Render progress tab content."

Variables

_STATUS_BADGE_COLORS = {5 items}

Resources Tab (resources_tab.ipynb)

Worker CPU, RAM, and GPU resource usage display.

Import

from cjm_fasthtml_job_monitor.components.tabs.resources_tab import (
    render_resources_tab
)

Functions

def _render_stat_row(
    label: str,          # Stat label (e.g., 'CPU')
    value_text: str,     # Formatted value (e.g., '45.2%')
    bar_pct: Optional[int] = None,   # Progress bar percentage (0-100)
    bar_color: str = '',             # DaisyUI progress color class
) -> FT:  # Stat row element
    "Render a single stat row with label, value, and optional progress bar."

def render_resources_tab(
    resources: Optional[ResourceSnapshot] = None,  # Resource data
) -> FT:  # Resources tab content
    "Render resources tab content."

Trigger Components (trigger.ipynb)

Trigger button and progress button for job monitor.

Import

from cjm_fasthtml_job_monitor.components.trigger import (
    render_job_trigger,
    render_job_progress_button
)

Functions

def render_job_trigger(
    config: JobMonitorConfig,     # Display config
    ids: JobMonitorHtmlIds,       # Element IDs
    urls: JobMonitorUrls,         # Route URLs
    disabled: bool = False,       # Disable button
    icon_fn: Optional[callable] = None,  # Icon renderer fn(name, **kwargs) -> FT
) -> FT:  # Trigger button wrapped in slot div
    "Render the initial trigger button."

def render_job_progress_button(
    config: JobMonitorConfig,   # Display config
    ids: JobMonitorHtmlIds,     # Element IDs
) -> FT:  # Progress button wrapped in slot div
    "Render 'View Progress' button with spinner."