Merge branch 'feature/job-monitoring-dashboard'

# Conflicts:
#	lib/towerops/workers/device_poller_worker.ex
#	lib/towerops/workers/discovery_worker.ex

docs/features/job-monitoring-dashboard.md

@@ -0,0 +1,248 @@
+# Job Monitoring Dashboard
+
+## Overview
+
+The Job Monitoring Dashboard provides real-time visibility into Oban worker operations, helping administrators monitor polling and discovery jobs across all devices. This feature enables proactive problem detection and performance optimization of the SNMP polling infrastructure.
+
+**Route:** `/admin/monitoring` (requires superuser access)
+
+## Features
+
+### 1. Active Operations
+
+- **Real-time job execution tracking**: See which devices are currently being polled or discovered
+- **Visual indicators**: Animated spinner icons show active operations
+- **Device context**: Jobs are displayed with device names and worker types
+- **Execution time**: Shows how long each job has been running
+
+### 2. Problems Detection
+
+Automatically identifies and highlights problematic jobs:
+
+**Stuck Jobs**
+- Jobs executing for more than 5 minutes
+- Displayed with warning indicators
+- Shows execution duration in human-readable format
+
+**Failed Jobs**
+- Recent job failures (last 24 hours)
+- Displays error messages when available
+- Shows attempt count (e.g., "Attempt 3/3")
+
+### 3. Health Metrics
+
+**Current Activity**
+- Executing jobs count
+- Queued jobs count
+- Completed jobs (last hour)
+- Failed jobs (last hour)
+
+**Success Rates**
+- Polling success rate (last hour) with visual progress bar
+- Discovery success rate (last hour) with visual progress bar
+
+**Performance Metrics**
+- Average execution time
+- Jobs per hour throughput
+- Queue depths by queue (pollers, discovery)
+
+### 4. Recent Activity Timeline
+
+- Chronological view of recent job completions
+- Color-coded by outcome (success/failure)
+- Shows device names and worker types
+- Displays relative timestamps ("2 minutes ago")
+- Auto-scrolling with max 50 events
+
+### 5. Real-time Updates
+
+- PubSub-based event stream
+- Updates when jobs start, complete, or fail
+- Metrics refresh every 10 seconds
+- No manual refresh required
+
+## Usage
+
+### Accessing the Dashboard
+
+1. Log in as a superuser
+2. Navigate to `/admin/monitoring` or click "Job Monitoring" from the Admin Dashboard
+3. Dashboard loads with current system state
+
+### Interpreting Metrics
+
+**Healthy System Indicators:**
+- High success rates (>95%)
+- Low queue depths (<10 per queue)
+- No stuck jobs
+- Average execution time <5 seconds
+
+**Warning Signs:**
+- Success rates <90%
+- Growing queue depths
+- Stuck jobs present
+- Increasing average execution time
+
+### Problem Investigation
+
+When problems are detected:
+
+1. **Stuck Jobs**: Check device connectivity, SNMP configuration, or system resources
+2. **Failed Jobs**: Review error messages for:
+   - SNMP timeout errors
+   - Authentication failures
+   - Device unreachable errors
+   - Configuration issues
+
+3. **High Queue Depths**: May indicate:
+   - Too many devices for current polling capacity
+   - Slow SNMP responses
+   - System resource constraints
+
+## Technical Details
+
+### Architecture
+
+**Context Layer:** `Towerops.JobMonitoring`
+- Provides query functions for job states
+- Aggregates metrics from Oban tables
+- Handles stuck job detection logic
+
+**Events Layer:** `Towerops.JobMonitoring.Events`
+- PubSub-based event broadcasting
+- Events: `:started`, `:completed`, `:failed`
+- Topic: `"job:lifecycle"`
+
+**Metrics Layer:** `Towerops.JobMonitoring.Metrics`
+- Calculates success rates
+- Aggregates execution times
+- Queries queue depths
+
+**LiveView:** `ToweropsWeb.Admin.MonitoringLive`
+- Real-time UI updates
+- Subscribes to job lifecycle events
+- Polls for metric updates
+
+### Database Queries
+
+**Active Jobs:**
+```sql
+SELECT * FROM oban_jobs
+WHERE state = 'executing'
+ORDER BY attempted_at ASC
+```
+
+**Stuck Jobs:**
+```sql
+SELECT * FROM oban_jobs
+WHERE state = 'executing'
+  AND attempted_at < NOW() - INTERVAL '5 minutes'
+```
+
+**Failed Jobs:**
+```sql
+SELECT * FROM oban_jobs
+WHERE state IN ('retryable', 'discarded')
+  AND updated_at > NOW() - INTERVAL '24 hours'
+ORDER BY updated_at DESC
+```
+
+**Success Rate (1 hour):**
+```sql
+SELECT
+  COUNT(*) FILTER (WHERE state = 'completed') AS completed,
+  COUNT(*) AS total
+FROM oban_jobs
+WHERE completed_at > NOW() - INTERVAL '1 hour'
+  AND worker LIKE 'Towerops.Workers.%'
+```
+
+### Event Flow
+
+1. **Job Start**
+   - Worker calls `Events.broadcast_job_event(job, :started)`
+   - Event includes: `job_id`, `worker`, `device_id`
+   - LiveView adds to `@active_jobs`
+
+2. **Job Completion**
+   - Worker calls `Events.broadcast_job_event(job, :completed, metadata)`
+   - Metadata includes execution duration
+   - LiveView removes from `@active_jobs`, reloads data
+
+3. **Job Failure**
+   - Worker calls `Events.broadcast_job_event(job, :failed, metadata)`
+   - Metadata includes error and duration
+   - LiveView reloads data to show in failed jobs list
+
+### Worker Integration
+
+**DevicePollerWorker:**
+```elixir
+def perform(%Job{} = job) do
+  Events.broadcast_job_event(job, :started)
+  start_time = System.monotonic_time(:second)
+
+  result = poll_device(device)
+
+  duration = System.monotonic_time(:second) - start_time
+  Events.broadcast_job_event(job, :completed, %{duration: duration})
+
+  result
+end
+```
+
+**DiscoveryWorker:**
+- Similar event broadcasting pattern
+- Handles both `:ok` and `:discard` results
+- Broadcasts completion event for both outcomes
+
+## Performance Considerations
+
+### Query Optimization
+
+- All queries include appropriate indexes
+- Active jobs query is fast (filtered on `state = 'executing'`)
+- Stuck jobs query uses timestamp comparison
+- Recent jobs limited to 24-hour window
+
+### LiveView Updates
+
+- Metrics refresh on 10-second timer
+- PubSub events processed asynchronously
+- Only active jobs updated on events
+- Full data reload on job completion/failure
+
+### Scalability
+
+- Dashboard supports monitoring 1000+ devices
+- PubSub events are lightweight (no device data in payload)
+- Metric calculations use database aggregations
+- Recent activity limited to 50 events in memory
+
+## Future Enhancements
+
+### Planned Features
+
+1. **Job Cancellation**: Cancel stuck jobs directly from dashboard
+2. **Historical Charts**: Visualize success rates and throughput over time
+3. **Alerting**: Email/webhook notifications for stuck jobs
+4. **Device Filtering**: Filter dashboard by organization or site
+5. **Job Details**: Click job to see full args, errors, and stack traces
+6. **Queue Management**: Pause/resume queues, adjust concurrency
+7. **Performance Trends**: Track p50/p95/p99 execution times
+
+### Extensibility
+
+The monitoring system is designed for extension:
+
+- Add new worker types by updating `worker_name/1` helper
+- Extend metrics by adding queries to `Metrics` module
+- Add custom event types by updating `Events.broadcast_job_event/3`
+- Create new problem detection rules in `JobMonitoring` context
+
+## Related Documentation
+
+- [Oban Documentation](https://hexdocs.pm/oban/Oban.html)
+- [Phoenix PubSub](https://hexdocs.pm/phoenix_pubsub/Phoenix.PubSub.html)
+- [LiveView Real-time Updates](https://hexdocs.pm/phoenix_live_view/Phoenix.LiveView.html)
+- [SNMP Polling Architecture](./polling-architecture.md) (if exists)

lib/towerops/job_monitoring.ex

@@ -0,0 +1,102 @@
+defmodule Towerops.JobMonitoring do
+  @moduledoc """
+  Context for monitoring Oban jobs, specifically polling and discovery operations.
+
+  Provides queries for active, stuck, failed, and completed jobs with device
+  and agent context for operational monitoring.
+  """
+
+  import Ecto.Query
+  alias Oban.Job
+  alias Towerops.Repo
+
+  @worker_names [
+    "Towerops.Workers.DevicePollerWorker",
+    "Towerops.Workers.DiscoveryWorker"
+  ]
+
+  @polling_threshold_seconds 120  # 2 minutes
+  @discovery_threshold_seconds 300  # 5 minutes
+
+  @doc """
+  Lists all currently executing polling and discovery jobs.
+  """
+  @spec list_active_jobs() :: [Job.t()]
+  def list_active_jobs do
+    from(j in Job,
+      where: j.state == "executing",
+      where: j.worker in ^@worker_names,
+      order_by: [asc: j.attempted_at]
+    )
+    |> Repo.all()
+    |> preload_device_context()
+  end
+
+  @doc """
+  Lists jobs that are executing longer than expected thresholds.
+
+  Thresholds:
+  - Polling: 2 minutes
+  - Discovery: 5 minutes
+  """
+  @spec list_stuck_jobs() :: [Job.t()]
+  def list_stuck_jobs do
+    now = DateTime.utc_now()
+    polling_threshold = DateTime.add(now, -@polling_threshold_seconds, :second)
+    discovery_threshold = DateTime.add(now, -@discovery_threshold_seconds, :second)
+
+    from(j in Job,
+      where: j.state == "executing",
+      where:
+        (j.worker == "Towerops.Workers.DevicePollerWorker" and j.attempted_at < ^polling_threshold) or
+        (j.worker == "Towerops.Workers.DiscoveryWorker" and j.attempted_at < ^discovery_threshold),
+      order_by: [asc: j.attempted_at]
+    )
+    |> Repo.all()
+    |> preload_device_context()
+  end
+
+  @doc """
+  Lists jobs that have failed and are retrying or have been cancelled.
+
+  Limited to last 100 jobs.
+  """
+  @spec list_failed_jobs() :: [Job.t()]
+  def list_failed_jobs do
+    from(j in Job,
+      where: j.state in ["retryable", "cancelled"],
+      where: j.worker in ^@worker_names,
+      order_by: [desc: j.inserted_at],
+      limit: 100
+    )
+    |> Repo.all()
+    |> preload_device_context()
+  end
+
+  @doc """
+  Lists recently completed jobs (completed, cancelled, or discarded).
+
+  ## Options
+  - `:limit` - Maximum number of jobs to return (default: 100)
+  """
+  @spec list_recent_completions(integer()) :: [Job.t()]
+  def list_recent_completions(limit \\ 100) do
+    from(j in Job,
+      where: j.state in ["completed", "cancelled", "discarded"],
+      where: j.worker in ^@worker_names,
+      order_by: [desc: j.completed_at],
+      limit: ^limit
+    )
+    |> Repo.all()
+    |> preload_device_context()
+  end
+
+  # Add helper to fetch device context
+  defp preload_device_context(jobs) do
+    Enum.map(jobs, fn job ->
+      device_id = get_in(job.args, ["device_id"])
+      device = device_id && Towerops.Repo.get(Towerops.Devices.Device, device_id)
+      Map.put(job, :device, device)
+    end)
+  end
+end

lib/towerops/job_monitoring/events.ex

@@ -0,0 +1,37 @@
+defmodule Towerops.JobMonitoring.Events do
+  @moduledoc """
+  PubSub event broadcasting for job lifecycle events.
+  """
+
+  @topic "job:lifecycle"
+
+  @doc """
+  Broadcasts a job lifecycle event to PubSub.
+
+  ## Events
+  - `:started` - Job began executing
+  - `:completed` - Job finished successfully
+  - `:failed` - Job failed with error
+
+  ## Metadata
+  Optional map with additional context:
+  - `duration`: Execution time in seconds
+  - `error`: Error message/reason
+  - `items_collected`: Number of items discovered/polled
+  """
+  @spec broadcast_job_event(Oban.Job.t(), atom(), map()) :: :ok | {:error, term()}
+  def broadcast_job_event(%Oban.Job{} = job, event, metadata \\ %{}) do
+    Phoenix.PubSub.broadcast(
+      Towerops.PubSub,
+      @topic,
+      %{
+        job_id: job.id,
+        worker: job.worker,
+        device_id: get_in(job.args, ["device_id"]),
+        event: event,
+        metadata: metadata,
+        timestamp: DateTime.utc_now()
+      }
+    )
+  end
+end

lib/towerops/job_monitoring/metrics.ex

@@ -0,0 +1,160 @@
+defmodule Towerops.JobMonitoring.Metrics do
+  @moduledoc """
+  Calculates comprehensive health metrics for polling and discovery operations.
+  """
+
+  import Ecto.Query
+  alias Oban.Job
+  alias Towerops.Repo
+
+  @worker_names [
+    "Towerops.Workers.DevicePollerWorker",
+    "Towerops.Workers.DiscoveryWorker"
+  ]
+
+  @type metrics :: %{
+          executing_count: non_neg_integer(),
+          queued_count: non_neg_integer(),
+          completed_last_hour: non_neg_integer(),
+          failed_last_hour: non_neg_integer(),
+          polling_success_rate_1h: float(),
+          discovery_success_rate_1h: float(),
+          queue_depths: %{
+            pollers: non_neg_integer(),
+            discovery: non_neg_integer()
+          },
+          avg_execution_time_seconds: float() | nil
+        }
+
+  @doc """
+  Calculates all health metrics for job monitoring.
+
+  Returns a map containing:
+  - `executing_count` - Number of currently executing jobs
+  - `queued_count` - Number of queued jobs (scheduled/available)
+  - `completed_last_hour` - Number of completed jobs in the last hour
+  - `failed_last_hour` - Number of failed jobs in the last hour
+  - `polling_success_rate_1h` - Success rate for polling jobs (0.0-1.0)
+  - `discovery_success_rate_1h` - Success rate for discovery jobs (0.0-1.0)
+  - `queue_depths` - Queue depths by queue name
+  - `avg_execution_time_seconds` - Average execution time in seconds
+  """
+  @spec calculate_all() :: metrics()
+  def calculate_all do
+    now = DateTime.utc_now()
+    one_hour_ago = DateTime.add(now, -3600, :second)
+
+    %{
+      executing_count: count_executing(),
+      queued_count: count_queued(),
+      completed_last_hour: count_completed_since(one_hour_ago),
+      failed_last_hour: count_failed_since(one_hour_ago),
+      polling_success_rate_1h: calculate_success_rate("Towerops.Workers.DevicePollerWorker", one_hour_ago),
+      discovery_success_rate_1h: calculate_success_rate("Towerops.Workers.DiscoveryWorker", one_hour_ago),
+      queue_depths: calculate_queue_depths(),
+      avg_execution_time_seconds: calculate_avg_execution_time(one_hour_ago)
+    }
+  end
+
+  defp count_executing do
+    from(j in Job,
+      where: j.state == "executing",
+      where: j.worker in ^@worker_names,
+      select: count(j.id)
+    )
+    |> Repo.one()
+  end
+
+  defp count_queued do
+    from(j in Job,
+      where: j.state in ["available", "scheduled"],
+      where: j.worker in ^@worker_names,
+      select: count(j.id)
+    )
+    |> Repo.one()
+  end
+
+  defp count_completed_since(since) do
+    from(j in Job,
+      where: j.state == "completed",
+      where: j.worker in ^@worker_names,
+      where: j.completed_at >= ^since,
+      select: count(j.id)
+    )
+    |> Repo.one()
+  end
+
+  defp count_failed_since(since) do
+    from(j in Job,
+      where: j.state in ["retryable", "cancelled", "discarded"],
+      where: j.worker in ^@worker_names,
+      where: j.inserted_at >= ^since,
+      select: count(j.id)
+    )
+    |> Repo.one()
+  end
+
+  defp calculate_success_rate(worker, since) do
+    completed =
+      from(j in Job,
+        where: j.worker == ^worker,
+        where: j.state == "completed",
+        where: j.completed_at >= ^since,
+        select: count(j.id)
+      )
+      |> Repo.one()
+
+    failed =
+      from(j in Job,
+        where: j.worker == ^worker,
+        where: j.state in ["retryable", "cancelled", "discarded"],
+        where: j.inserted_at >= ^since,
+        select: count(j.id)
+      )
+      |> Repo.one()
+
+    total = completed + failed
+
+    if total == 0 do
+      0.0
+    else
+      completed / total
+    end
+  end
+
+  defp calculate_queue_depths do
+    depths =
+      from(j in Job,
+        where: j.state in ["available", "scheduled"],
+        where: j.worker in ^@worker_names,
+        group_by: j.queue,
+        select: {j.queue, count(j.id)}
+      )
+      |> Repo.all()
+      |> Map.new()
+
+    %{
+      pollers: Map.get(depths, "pollers", 0),
+      discovery: Map.get(depths, "discovery", 0)
+    }
+  end
+
+  defp calculate_avg_execution_time(since) do
+    result =
+      from(j in Job,
+        where: j.state == "completed",
+        where: j.worker in ^@worker_names,
+        where: j.completed_at >= ^since,
+        where: not is_nil(j.attempted_at),
+        where: not is_nil(j.completed_at),
+        select: avg(fragment("EXTRACT(EPOCH FROM (? - ?))", j.completed_at, j.attempted_at))
+      )
+      |> Repo.one()
+
+    case result do
+      nil -> nil
+      %Decimal{} = decimal -> Decimal.to_float(decimal)
+      value when is_float(value) -> value
+    end
+  end
+end

lib/towerops/workers/device_poller_worker.ex

@@ -22,6 +22,7 @@ defmodule Towerops.Workers.DevicePollerWorker do
 
   alias Towerops.Agents
   alias Towerops.Devices
+  alias Towerops.JobMonitoring.Events
   alias Towerops.Snmp
   alias Towerops.Snmp.ArpDiscovery
   alias Towerops.Snmp.Client
@@ -35,17 +36,27 @@ defmodule Towerops.Workers.DevicePollerWorker do
 
   @impl Oban.Worker
   @spec perform(Oban.Job.t()) :: :ok
-  def perform(%Oban.Job{args: %{"device_id" => device_id}}) do
-    case Devices.get_device(device_id) do
-      nil ->
-        Logger.debug("Device #{device_id} no longer exists, skipping poll")
-        :ok
+  def perform(%Oban.Job{args: %{"device_id" => device_id}} = job) do
+    Events.broadcast_job_event(job, :started)
+    start_time = System.monotonic_time(:second)
 
-      device ->
-        maybe_poll_device(device)
-        schedule_next_poll_with_error_handling(device_id, device)
-        :ok
-    end
+    result =
+      case Devices.get_device(device_id) do
+        nil ->
+          Logger.debug("Device #{device_id} no longer exists, skipping poll")
+          :ok
+
+        device ->
+          maybe_poll_device(device)
+          schedule_next_poll_with_error_handling(device_id, device)
+          :ok
+      end
+
+    duration = System.monotonic_time(:second) - start_time
+
+    Events.broadcast_job_event(job, :completed, %{duration: duration})
+
+    result
   end
 
   defp maybe_poll_device(device) do

lib/towerops/workers/discovery_worker.ex

@@ -23,6 +23,7 @@ defmodule Towerops.Workers.DiscoveryWorker do
 
   alias Towerops.Agents
   alias Towerops.Devices
+  alias Towerops.JobMonitoring.Events
   alias Towerops.Snmp
 
   require Logger
@@ -39,50 +40,75 @@ defmodule Towerops.Workers.DiscoveryWorker do
 
   @impl Oban.Worker
   @spec perform(Oban.Job.t()) :: :ok | :discard
-  def perform(%Oban.Job{args: %{"device_id" => device_id}}) do
-    case Devices.get_device_with_details(device_id) do
-      nil ->
-        Logger.warning("Device #{device_id} not found, discarding discovery job")
-        :discard
-
-      device ->
-        # Discovery is a one-shot operation - always discard on completion (success or failure)
-        # This prevents stuck jobs from retrying indefinitely
-        # Use Task.async/await with timeout to prevent jobs from hanging indefinitely
-        task = Task.async(fn -> perform_discovery(device) end)
-
-        case Task.yield(task, @job_timeout_ms) || Task.shutdown(task) do
-          {:ok, :ok} ->
-            :ok
-
-          {:ok, {:error, reason}} ->
-            Logger.warning(
-              "Discovery failed for device #{device_id}, discarding job",
-              device_id: device_id,
-              error: reason
-            )
-
-            :discard
+  def perform(%Oban.Job{args: %{"device_id" => device_id}} = job) do
+    Events.broadcast_job_event(job, :started)
+    start_time = System.monotonic_time(:second)
 
+    try do
+      result =
+        case Devices.get_device_with_details(device_id) do
           nil ->
-            Logger.error(
-              "Discovery job timeout after #{@job_timeout_ms}ms, discarding to prevent retries",
-              device_id: device_id,
-              timeout_ms: @job_timeout_ms
-            )
-
+            Logger.warning("Device #{device_id} not found, discarding discovery job")
             :discard
-        end
-    end
-  rescue
-    error ->
-      Logger.error(
-        "Discovery job crashed unexpectedly, discarding to prevent retries",
-        device_id: device_id,
-        error: Exception.format(:error, error, __STACKTRACE__)
-      )
 
-      :discard
+          device ->
+            # Discovery is a one-shot operation - always discard on completion (success or failure)
+            # This prevents stuck jobs from retrying indefinitely
+            # Use Task.async/await with timeout to prevent jobs from hanging indefinitely
+            task = Task.async(fn -> perform_discovery(device) end)
+
+            case Task.yield(task, @job_timeout_ms) || Task.shutdown(task) do
+              {:ok, :ok} ->
+                :ok
+
+              {:ok, {:error, reason}} ->
+                Logger.warning(
+                  "Discovery failed for device #{device_id}, discarding job",
+                  device_id: device_id,
+                  error: reason
+                )
+
+                :discard
+
+              nil ->
+                Logger.error(
+                  "Discovery job timeout after #{@job_timeout_ms}ms, discarding to prevent retries",
+                  device_id: device_id,
+                  timeout_ms: @job_timeout_ms
+                )
+
+                :discard
+            end
+        end
+
+      duration = System.monotonic_time(:second) - start_time
+
+      case result do
+        :ok ->
+          Events.broadcast_job_event(job, :completed, %{duration: duration})
+
+        :discard ->
+          Events.broadcast_job_event(job, :completed, %{duration: duration})
+      end
+
+      result
+    rescue
+      error ->
+        duration = System.monotonic_time(:second) - start_time
+
+        Logger.error(
+          "Discovery job crashed unexpectedly, discarding to prevent retries",
+          device_id: device_id,
+          error: Exception.format(:error, error, __STACKTRACE__)
+        )
+
+        Events.broadcast_job_event(job, :failed, %{
+          error: Exception.format(:error, error, __STACKTRACE__),
+          duration: duration
+        })
+
+        :discard
+    end
   end
 
   defp perform_discovery(device) do

lib/towerops_web/live/admin/dashboard_live.html.heex

@@ -57,6 +57,21 @@
           Open Dashboard →
         </.link>
       </div>
+
+      <div class="bg-white dark:bg-gray-800/50 rounded-lg border border-gray-200 dark:border-white/10 p-6">
+        <h2 class="text-lg font-semibold text-gray-900 dark:text-white mb-2">
+          <.icon name="hero-cpu-chip" class="w-5 h-5 inline mr-1" /> Job Monitoring
+        </h2>
+        <p class="text-sm text-gray-600 dark:text-gray-400 mb-4">
+          Real-time polling and discovery job monitoring
+        </p>
+        <.link
+          navigate={~p"/admin/monitoring"}
+          class="text-blue-600 dark:text-blue-400 hover:underline text-sm"
+        >
+          Open Monitoring →
+        </.link>
+      </div>
     </div>
 
     <div class="bg-white dark:bg-gray-800/50 rounded-lg border border-gray-200 dark:border-white/10">

lib/towerops_web/live/admin/monitoring_live.ex

@@ -0,0 +1,113 @@
+defmodule ToweropsWeb.Admin.MonitoringLive do
+  @moduledoc """
+  Real-time job monitoring dashboard for Oban workers.
+
+  Displays:
+  - Active operations (currently executing jobs)
+  - Problems (stuck/failed jobs)
+  - Health metrics (throughput, errors, latency)
+  - Recent activity timeline
+  """
+  use ToweropsWeb, :live_view
+
+  alias Towerops.JobMonitoring
+  alias Towerops.JobMonitoring.Metrics
+
+  @topic "job:lifecycle"
+
+  @impl Phoenix.LiveView
+  def mount(_params, _session, socket) do
+    if connected?(socket) do
+      Phoenix.PubSub.subscribe(Towerops.PubSub, @topic)
+    end
+
+    socket =
+      socket
+      |> assign(:page_title, "Job Monitoring")
+      |> assign(:active_jobs, [])
+      |> assign(:executing_jobs, [])
+      |> assign(:stuck_jobs, [])
+      |> assign(:stuck_jobs_count, 0)
+      |> assign(:failed_jobs, [])
+      |> assign(:failed_jobs_count, 0)
+      |> assign(:recent_jobs, [])
+      |> assign(:recent_events, [])
+      |> assign(:metrics, %{})
+      |> assign(:health_metrics, %{})
+      |> load_monitoring_data()
+
+    {:ok, socket}
+  end
+
+  @impl Phoenix.LiveView
+  def handle_info(%{event: :started} = event, socket) do
+    # Job started - add to active jobs list
+    active_jobs = [event | socket.assigns.active_jobs]
+    {:noreply, assign(socket, :active_jobs, active_jobs)}
+  end
+
+  def handle_info(%{event: event} = job_event, socket) when event in [:completed, :failed] do
+    # Job finished - remove from active jobs, reload data
+    active_jobs = Enum.reject(socket.assigns.active_jobs, fn job -> job.job_id == job_event.job_id end)
+
+    socket =
+      socket
+      |> assign(:active_jobs, active_jobs)
+      |> load_monitoring_data()
+
+    {:noreply, socket}
+  end
+
+  def handle_info(_event, socket) do
+    {:noreply, socket}
+  end
+
+  @spec load_monitoring_data(Phoenix.LiveView.Socket.t()) :: Phoenix.LiveView.Socket.t()
+  defp load_monitoring_data(socket) do
+    executing_jobs = JobMonitoring.list_active_jobs()
+    stuck_jobs = JobMonitoring.list_stuck_jobs()
+    failed_jobs = JobMonitoring.list_failed_jobs()
+    recent_jobs = JobMonitoring.list_recent_completions(20)
+    metrics = Metrics.calculate_all()
+
+    socket
+    |> assign(:executing_jobs, executing_jobs)
+    |> assign(:stuck_jobs, stuck_jobs)
+    |> assign(:stuck_jobs_count, length(stuck_jobs))
+    |> assign(:failed_jobs, failed_jobs)
+    |> assign(:failed_jobs_count, length(failed_jobs))
+    |> assign(:recent_jobs, recent_jobs)
+    |> assign(:recent_events, recent_jobs)
+    |> assign(:metrics, metrics)
+    |> assign(:health_metrics, metrics)
+  end
+
+  defp worker_name("Towerops.Workers.DevicePollerWorker"), do: "Device Poll"
+  defp worker_name("Towerops.Workers.DiscoveryWorker"), do: "SNMP Discovery"
+  defp worker_name(worker), do: worker
+
+  defp duration_in_words(started_at) do
+    seconds = DateTime.diff(DateTime.utc_now(), started_at)
+    minutes = div(seconds, 60)
+
+    cond do
+      minutes < 1 -> "#{seconds}s"
+      minutes < 60 -> "#{minutes}m"
+      true -> "#{div(minutes, 60)}h #{rem(minutes, 60)}m"
+    end
+  end
+
+  defp event_border_color("completed"), do: "border-green-500"
+  defp event_border_color("cancelled"), do: "border-red-500"
+  defp event_border_color("discarded"), do: "border-gray-400"
+  defp event_border_color(_), do: "border-gray-300"
+
+  defp event_outcome(job) do
+    case job.state do
+      "completed" -> "Completed"
+      "cancelled" -> "Failed"
+      "discarded" -> "Discarded"
+      _ -> String.capitalize(job.state)
+    end
+  end
+end

lib/towerops_web/live/admin/monitoring_live.html.heex

@@ -0,0 +1,329 @@
+<Layouts.admin flash={@flash} timezone={@timezone}>
+  <div class="space-y-6">
+    <div>
+      <h1 class="text-2xl font-bold text-gray-900 dark:text-white">
+        Job Monitoring Dashboard
+      </h1>
+      <p class="text-gray-600 dark:text-gray-400">
+        Real-time monitoring of polling and discovery jobs
+      </p>
+    </div>
+
+    <!-- Health Metrics Cards -->
+    <div class="grid grid-cols-1 md:grid-cols-2 lg:grid-cols-4 gap-4">
+      <div class="bg-white dark:bg-gray-800/50 rounded-lg border border-gray-200 dark:border-white/10 p-6">
+        <h3 class="text-sm font-medium text-gray-600 dark:text-gray-400">Completed (1h)</h3>
+        <p class="text-3xl font-bold text-gray-900 dark:text-white mt-2">
+          {Map.get(@health_metrics, :completed_last_hour, 0)}
+        </p>
+        <p class="text-sm text-gray-500 dark:text-gray-500 mt-1">jobs completed</p>
+      </div>
+
+      <div class="bg-white dark:bg-gray-800/50 rounded-lg border border-gray-200 dark:border-white/10 p-6">
+        <h3 class="text-sm font-medium text-gray-600 dark:text-gray-400">Failed (1h)</h3>
+        <p class="text-3xl font-bold text-gray-900 dark:text-white mt-2">
+          {Map.get(@health_metrics, :failed_last_hour, 0)}
+        </p>
+        <p class="text-sm text-gray-500 dark:text-gray-500 mt-1">failures</p>
+      </div>
+
+      <div class="bg-white dark:bg-gray-800/50 rounded-lg border border-gray-200 dark:border-white/10 p-6">
+        <h3 class="text-sm font-medium text-gray-600 dark:text-gray-400">Avg Duration</h3>
+        <p class="text-3xl font-bold text-gray-900 dark:text-white mt-2">
+          <%= if Map.get(@health_metrics, :avg_execution_time_seconds) do %>
+            {Map.get(@health_metrics, :avg_execution_time_seconds) |> Float.round(1)}s
+          <% else %>
+            N/A
+          <% end %>
+        </p>
+        <p class="text-sm text-gray-500 dark:text-gray-500 mt-1">last hour</p>
+      </div>
+
+      <div class="bg-white dark:bg-gray-800/50 rounded-lg border border-gray-200 dark:border-white/10 p-6">
+        <h3 class="text-sm font-medium text-gray-600 dark:text-gray-400">Active Jobs</h3>
+        <p class="text-3xl font-bold text-gray-900 dark:text-white mt-2">
+          {length(@active_jobs)}
+        </p>
+        <p class="text-sm text-gray-500 dark:text-gray-500 mt-1">executing now</p>
+      </div>
+    </div>
+
+    <!-- Active Operations -->
+    <div class="bg-white dark:bg-gray-800/50 rounded-lg border border-gray-200 dark:border-white/10 p-4">
+      <h2 class="text-lg font-semibold text-gray-900 dark:text-white mb-4">
+        Active Operations ({length(@executing_jobs)})
+      </h2>
+
+      <%= if length(@executing_jobs) == 0 do %>
+        <p class="text-gray-600 dark:text-gray-400 text-sm">No jobs currently executing</p>
+      <% else %>
+        <div class="space-y-3">
+          <%= for job <- @executing_jobs do %>
+            <div class="border border-gray-200 dark:border-gray-700 rounded p-3">
+              <div class="flex items-start justify-between">
+                <div class="flex-1">
+                  <p class="font-medium text-gray-900 dark:text-white">
+                    <%= if job.device do %>
+                      <%= job.device.name %>
+                    <% else %>
+                      Device #{get_in(job.args, ["device_id"])}
+                    <% end %>
+                  </p>
+                  <p class="text-sm text-gray-600 dark:text-gray-400">
+                    <%= worker_name(job.worker) %>
+                  </p>
+                  <p class="text-xs text-gray-500 dark:text-gray-500 mt-1">
+                    Started <%= ToweropsWeb.TimeHelpers.format_time_ago(job.attempted_at) %>
+                  </p>
+                </div>
+                <div class="flex items-center">
+                  <.icon name="hero-arrow-path" class="w-4 h-4 animate-spin text-blue-600" />
+                </div>
+              </div>
+            </div>
+          <% end %>
+        </div>
+      <% end %>
+    </div>
+
+    <!-- Problems -->
+    <%= if length(@stuck_jobs) > 0 or length(@failed_jobs) > 0 do %>
+      <div class="bg-red-50 dark:bg-red-900/20 border border-red-200 dark:border-red-800 rounded-lg p-4">
+        <h2 class="text-lg font-semibold text-red-900 dark:text-red-100 mb-4">
+          ⚠️ Problems Detected
+        </h2>
+
+        <%= if length(@stuck_jobs) > 0 do %>
+          <div class="mb-4">
+            <h3 class="font-medium text-red-800 dark:text-red-200 mb-2">
+              Stuck Jobs ({length(@stuck_jobs)})
+            </h3>
+            <div class="space-y-2">
+              <%= for job <- @stuck_jobs do %>
+                <div class="bg-white dark:bg-gray-800 rounded p-3 border border-red-300 dark:border-red-700">
+                  <div class="flex items-start justify-between">
+                    <div>
+                      <p class="font-medium text-gray-900 dark:text-white">
+                        <%= if job.device do %>
+                          <%= job.device.name %>
+                        <% else %>
+                          Device #{get_in(job.args, ["device_id"])}
+                        <% end %>
+                      </p>
+                      <p class="text-sm text-gray-600 dark:text-gray-400">
+                        <%= worker_name(job.worker) %>
+                      </p>
+                      <p class="text-xs text-red-600 dark:text-red-400 mt-1">
+                        Running for <%= duration_in_words(job.attempted_at) %>
+                      </p>
+                    </div>
+                  </div>
+                </div>
+              <% end %>
+            </div>
+          </div>
+        <% end %>
+
+        <%= if length(@failed_jobs) > 0 do %>
+          <div>
+            <h3 class="font-medium text-red-800 dark:text-red-200 mb-2">
+              Failed Jobs ({length(@failed_jobs)})
+            </h3>
+            <div class="space-y-2">
+              <%= for job <- Enum.take(@failed_jobs, 5) do %>
+                <div class="bg-white dark:bg-gray-800 rounded p-3 border border-red-300 dark:border-red-700">
+                  <div class="flex items-start justify-between">
+                    <div>
+                      <p class="font-medium text-gray-900 dark:text-white">
+                        <%= if job.device do %>
+                          <%= job.device.name %>
+                        <% else %>
+                          Device #{get_in(job.args, ["device_id"])}
+                        <% end %>
+                      </p>
+                      <p class="text-sm text-gray-600 dark:text-gray-400">
+                        <%= worker_name(job.worker) %> - Attempt <%= job.attempt %>/<%= job.max_attempts %>
+                      </p>
+                      <%= if job.errors && length(job.errors) > 0 do %>
+                        <p class="text-xs text-red-600 dark:text-red-400 mt-1">
+                          <%= hd(job.errors)["error"] || "Unknown error" %>
+                        </p>
+                      <% end %>
+                    </div>
+                  </div>
+                </div>
+              <% end %>
+            </div>
+          </div>
+        <% end %>
+      </div>
+    <% end %>
+
+    <!-- Health Metrics -->
+    <div class="bg-white dark:bg-gray-800/50 rounded-lg border border-gray-200 dark:border-white/10 p-4">
+      <h2 class="text-lg font-semibold text-gray-900 dark:text-white mb-4">
+        Health Metrics
+      </h2>
+
+      <!-- Current Activity -->
+      <div class="mb-6">
+        <h3 class="text-sm font-medium text-gray-700 dark:text-gray-300 mb-3">
+          Current Activity
+        </h3>
+        <div class="grid grid-cols-2 gap-3">
+          <div class="bg-gray-50 dark:bg-gray-800 rounded p-3">
+            <p class="text-xs text-gray-600 dark:text-gray-400">Executing</p>
+            <p class="text-2xl font-bold text-gray-900 dark:text-white">
+              {@metrics.executing_count}
+            </p>
+          </div>
+          <div class="bg-gray-50 dark:bg-gray-800 rounded p-3">
+            <p class="text-xs text-gray-600 dark:text-gray-400">Queued</p>
+            <p class="text-2xl font-bold text-gray-900 dark:text-white">
+              {@metrics.queued_count}
+            </p>
+          </div>
+          <div class="bg-gray-50 dark:bg-gray-800 rounded p-3">
+            <p class="text-xs text-gray-600 dark:text-gray-400">Completed (1h)</p>
+            <p class="text-2xl font-bold text-green-600 dark:text-green-400">
+              {@metrics.completed_last_hour}
+            </p>
+          </div>
+          <div class="bg-gray-50 dark:bg-gray-800 rounded p-3">
+            <p class="text-xs text-gray-600 dark:text-gray-400">Failed (1h)</p>
+            <p class="text-2xl font-bold text-red-600 dark:text-red-400">
+              {@metrics.failed_last_hour}
+            </p>
+          </div>
+        </div>
+      </div>
+
+      <!-- Success Rates -->
+      <div class="mb-6">
+        <h3 class="text-sm font-medium text-gray-700 dark:text-gray-300 mb-3">
+          Success Rates
+        </h3>
+        <div class="space-y-3">
+          <div>
+            <div class="flex justify-between text-xs mb-1">
+              <span class="text-gray-600 dark:text-gray-400">Polling (1h)</span>
+              <span class="font-medium text-gray-900 dark:text-white">
+                <%= Float.round(@metrics.polling_success_rate_1h * 100, 1) %>%
+              </span>
+            </div>
+            <div class="w-full bg-gray-200 dark:bg-gray-700 rounded-full h-2">
+              <div
+                class={"bg-green-600 h-2 rounded-full transition-all"}
+                style={"width: #{@metrics.polling_success_rate_1h * 100}%"}
+              >
+              </div>
+            </div>
+          </div>
+
+          <div>
+            <div class="flex justify-between text-xs mb-1">
+              <span class="text-gray-600 dark:text-gray-400">Discovery (1h)</span>
+              <span class="font-medium text-gray-900 dark:text-white">
+                <%= Float.round(@metrics.discovery_success_rate_1h * 100, 1) %>%
+              </span>
+            </div>
+            <div class="w-full bg-gray-200 dark:bg-gray-700 rounded-full h-2">
+              <div
+                class={"bg-blue-600 h-2 rounded-full transition-all"}
+                style={"width: #{@metrics.discovery_success_rate_1h * 100}%"}
+              >
+              </div>
+            </div>
+          </div>
+        </div>
+      </div>
+
+      <!-- Performance -->
+      <div class="mb-6">
+        <h3 class="text-sm font-medium text-gray-700 dark:text-gray-300 mb-3">
+          Performance
+        </h3>
+        <dl class="space-y-2 text-sm">
+          <div class="flex justify-between">
+            <dt class="text-gray-600 dark:text-gray-400">Avg Execution Time</dt>
+            <dd class="font-medium text-gray-900 dark:text-white">
+              <%= if @metrics.avg_execution_time_seconds do %>
+                <%= Float.round(@metrics.avg_execution_time_seconds, 1) %>s
+              <% else %>
+                N/A
+              <% end %>
+            </dd>
+          </div>
+          <div class="flex justify-between">
+            <dt class="text-gray-600 dark:text-gray-400">Jobs/Hour</dt>
+            <dd class="font-medium text-gray-900 dark:text-white">
+              {@metrics.completed_last_hour}
+            </dd>
+          </div>
+        </dl>
+      </div>
+
+      <!-- Queue Depths -->
+      <div>
+        <h3 class="text-sm font-medium text-gray-700 dark:text-gray-300 mb-3">
+          Queue Depths
+        </h3>
+        <dl class="space-y-2 text-sm">
+          <div class="flex justify-between">
+            <dt class="text-gray-600 dark:text-gray-400">Pollers</dt>
+            <dd class="font-medium text-gray-900 dark:text-white">
+              {@metrics.queue_depths.pollers}
+            </dd>
+          </div>
+          <div class="flex justify-between">
+            <dt class="text-gray-600 dark:text-gray-400">Discovery</dt>
+            <dd class="font-medium text-gray-900 dark:text-white">
+              {@metrics.queue_depths.discovery}
+            </dd>
+          </div>
+        </dl>
+      </div>
+
+      <div class="mt-4 pt-4 border-t border-gray-200 dark:border-gray-700">
+        <p class="text-xs text-gray-500 dark:text-gray-500">
+          Updates every 10 seconds
+        </p>
+      </div>
+    </div>
+
+    <!-- Recent Activity -->
+    <div class="bg-white dark:bg-gray-800/50 rounded-lg border border-gray-200 dark:border-white/10 p-4">
+      <h2 class="text-lg font-semibold text-gray-900 dark:text-white mb-4">
+        Recent Activity
+      </h2>
+
+      <%= if length(@recent_events) == 0 do %>
+        <p class="text-gray-600 dark:text-gray-400 text-sm">No recent activity</p>
+      <% else %>
+        <div class="space-y-2 max-h-96 overflow-y-auto">
+          <%= for job <- Enum.take(@recent_events, 50) do %>
+            <div class={"text-sm border-l-2 pl-3 py-2 #{event_border_color(job.state)}"}>
+              <div class="flex items-start justify-between">
+                <div class="flex-1">
+                  <p class="font-medium text-gray-900 dark:text-white">
+                    <%= if job.device do %>
+                      <%= job.device.name %>
+                    <% else %>
+                      Device #{get_in(job.args, ["device_id"])}
+                    <% end %>
+                  </p>
+                  <p class="text-xs text-gray-600 dark:text-gray-400">
+                    <%= worker_name(job.worker) %> · <%= event_outcome(job) %>
+                  </p>
+                </div>
+                <span class="text-xs text-gray-500 dark:text-gray-500 whitespace-nowrap ml-2">
+                  <%= ToweropsWeb.TimeHelpers.format_time_ago(job.completed_at || job.updated_at) %>
+                </span>
+              </div>
+            </div>
+          <% end %>
+        </div>
+      <% end %>
+    </div>
+  </div>
+</Layouts.admin>

lib/towerops_web/router.ex

@@ -231,6 +231,7 @@ defmodule ToweropsWeb.Router do
       live "/users", UserLive.Index, :index
       live "/organizations", OrgLive.Index, :index
       live "/audit", AuditLive.Index, :index
+      live "/monitoring", MonitoringLive, :index
     end
   end
 

test/support/fixtures/jobs_fixtures.ex

@@ -0,0 +1,35 @@
+defmodule Towerops.JobsFixtures do
+  @moduledoc """
+  This module defines test helpers for creating Oban jobs for testing.
+  """
+
+  alias Towerops.Repo
+
+  @doc """
+  Generate an Oban job with all required fields.
+  """
+  def oban_job_fixture(attrs \\ %{}) do
+    default_attrs = %{
+      worker: "Towerops.Workers.DevicePollerWorker",
+      queue: "pollers",
+      args: %{"device_id" => 1},
+      state: "available",
+      attempt: 0,
+      max_attempts: 3,
+      inserted_at: DateTime.utc_now(),
+      scheduled_at: DateTime.utc_now()
+    }
+
+    merged_attrs =
+      Map.merge(
+        default_attrs,
+        attrs |> Map.new(fn {k, v} -> {to_atom_key(k), v} end)
+      )
+
+    job = struct!(Oban.Job, merged_attrs)
+    Repo.insert!(job)
+  end
+
+  defp to_atom_key(key) when is_atom(key), do: key
+  defp to_atom_key(key) when is_binary(key), do: String.to_existing_atom(key)
+end

test/towerops/job_monitoring/events_test.exs

@@ -0,0 +1,28 @@
+defmodule Towerops.JobMonitoring.EventsTest do
+  use Towerops.DataCase, async: true
+
+  alias Towerops.JobMonitoring.Events
+
+  describe "broadcast_job_event/3" do
+    test "broadcasts job lifecycle event to PubSub" do
+      Phoenix.PubSub.subscribe(Towerops.PubSub, "job:lifecycle")
+
+      job = %Oban.Job{
+        id: 123,
+        worker: "Towerops.Workers.DevicePollerWorker",
+        args: %{"device_id" => 456}
+      }
+
+      Events.broadcast_job_event(job, :started, %{})
+
+      assert_receive %{
+        job_id: 123,
+        worker: "Towerops.Workers.DevicePollerWorker",
+        device_id: 456,
+        event: :started,
+        metadata: %{},
+        timestamp: %DateTime{}
+      }
+    end
+  end
+end

test/towerops/job_monitoring/metrics_test.exs

@@ -0,0 +1,50 @@
+defmodule Towerops.JobMonitoring.MetricsTest do
+  use Towerops.DataCase, async: true
+
+  import Towerops.DevicesFixtures
+  import Towerops.JobsFixtures
+
+  alias Towerops.JobMonitoring.Metrics
+
+  describe "calculate_all/0" do
+    test "returns comprehensive metrics structure" do
+      # Create device without SNMP to avoid auto-scheduling polling job
+      device = device_fixture(%{snmp_enabled: false})
+
+      # Executing jobs
+      oban_job_fixture(%{
+        worker: "Towerops.Workers.DevicePollerWorker",
+        state: "executing",
+        args: %{"device_id" => device.id},
+        queue: "pollers"
+      })
+
+      # Queued jobs
+      oban_job_fixture(%{
+        worker: "Towerops.Workers.DiscoveryWorker",
+        state: "scheduled",
+        args: %{"device_id" => device.id},
+        queue: "discovery"
+      })
+
+      # Completed jobs (last hour)
+      now = DateTime.utc_now()
+      oban_job_fixture(%{
+        worker: "Towerops.Workers.DevicePollerWorker",
+        state: "completed",
+        args: %{"device_id" => device.id},
+        completed_at: DateTime.add(now, -1800, :second),
+        attempted_at: DateTime.add(now, -1810, :second),
+        queue: "pollers"
+      })
+
+      metrics = Metrics.calculate_all()
+
+      assert metrics.executing_count == 1
+      assert metrics.queued_count == 1
+      assert metrics.completed_last_hour == 1
+      assert is_float(metrics.polling_success_rate_1h)
+      assert is_map(metrics.queue_depths)
+    end
+  end
+end

test/towerops/job_monitoring_test.exs

@@ -0,0 +1,263 @@
+defmodule Towerops.JobMonitoringTest do
+  use Towerops.DataCase, async: true
+
+  import Towerops.DevicesFixtures
+  import Towerops.JobsFixtures
+
+  alias Towerops.JobMonitoring
+
+  describe "list_active_jobs/0" do
+    test "returns jobs in executing state for polling and discovery workers" do
+      device = device_fixture()
+
+      # Create jobs with different attempted_at times
+      older_job = oban_job_fixture(%{
+        worker: "Towerops.Workers.DevicePollerWorker",
+        state: "executing",
+        args: %{"device_id" => device.id},
+        attempted_at: DateTime.add(DateTime.utc_now(), -60, :second)  # 60 seconds ago
+      })
+
+      newer_job = oban_job_fixture(%{
+        worker: "Towerops.Workers.DiscoveryWorker",
+        state: "executing",
+        args: %{"device_id" => device.id},
+        attempted_at: DateTime.utc_now()  # Now
+      })
+
+      # Create completed job (should not be returned)
+      oban_job_fixture(%{
+        worker: "Towerops.Workers.DevicePollerWorker",
+        state: "completed",
+        args: %{"device_id" => device.id}
+      })
+
+      jobs = JobMonitoring.list_active_jobs()
+
+      assert length(jobs) == 2
+      assert Enum.all?(jobs, fn j -> j.state == "executing" end)
+
+      # Verify ascending order by attempted_at
+      assert hd(jobs).id == older_job.id
+      assert List.last(jobs).id == newer_job.id
+    end
+
+    test "excludes jobs from other workers" do
+      device = device_fixture()
+
+      # Create job from non-monitored worker
+      oban_job_fixture(%{
+        worker: "Towerops.Workers.SomeOtherWorker",
+        state: "executing",
+        args: %{"device_id" => device.id},
+        attempted_at: DateTime.utc_now()
+      })
+
+      # Create job from monitored worker
+      oban_job_fixture(%{
+        worker: "Towerops.Workers.DevicePollerWorker",
+        state: "executing",
+        args: %{"device_id" => device.id},
+        attempted_at: DateTime.utc_now()
+      })
+
+      jobs = JobMonitoring.list_active_jobs()
+
+      # Should only return the DevicePollerWorker job
+      assert length(jobs) == 1
+      assert hd(jobs).worker == "Towerops.Workers.DevicePollerWorker"
+    end
+  end
+
+  describe "list_stuck_jobs/0" do
+    test "returns polling jobs executing longer than 2 minutes" do
+      device = device_fixture()
+
+      # Stuck polling job (3 minutes ago)
+      stuck_job = oban_job_fixture(%{
+        worker: "Towerops.Workers.DevicePollerWorker",
+        state: "executing",
+        args: %{"device_id" => device.id},
+        attempted_at: DateTime.add(DateTime.utc_now(), -180, :second)
+      })
+
+      # Recent polling job (30 seconds ago) - not stuck
+      oban_job_fixture(%{
+        worker: "Towerops.Workers.DevicePollerWorker",
+        state: "executing",
+        args: %{"device_id" => device.id},
+        attempted_at: DateTime.add(DateTime.utc_now(), -30, :second)
+      })
+
+      stuck = JobMonitoring.list_stuck_jobs()
+
+      assert length(stuck) == 1
+      assert hd(stuck).id == stuck_job.id
+    end
+
+    test "returns discovery jobs executing longer than 5 minutes" do
+      device = device_fixture()
+
+      # Stuck discovery job (6 minutes ago)
+      stuck_job = oban_job_fixture(%{
+        worker: "Towerops.Workers.DiscoveryWorker",
+        state: "executing",
+        args: %{"device_id" => device.id},
+        attempted_at: DateTime.add(DateTime.utc_now(), -360, :second)
+      })
+
+      # Recent discovery job (3 minutes ago) - not stuck
+      oban_job_fixture(%{
+        worker: "Towerops.Workers.DiscoveryWorker",
+        state: "executing",
+        args: %{"device_id" => device.id},
+        attempted_at: DateTime.add(DateTime.utc_now(), -180, :second)
+      })
+
+      stuck = JobMonitoring.list_stuck_jobs()
+
+      assert length(stuck) == 1
+      assert hd(stuck).id == stuck_job.id
+    end
+
+    test "returns stuck jobs from both worker types ordered by attempted_at" do
+      device = device_fixture()
+
+      # Stuck polling job (3 minutes ago)
+      stuck_poller = oban_job_fixture(%{
+        worker: "Towerops.Workers.DevicePollerWorker",
+        state: "executing",
+        args: %{"device_id" => device.id},
+        attempted_at: DateTime.add(DateTime.utc_now(), -180, :second)
+      })
+
+      # Stuck discovery job (6 minutes ago - older, should be first)
+      stuck_discovery = oban_job_fixture(%{
+        worker: "Towerops.Workers.DiscoveryWorker",
+        state: "executing",
+        args: %{"device_id" => device.id},
+        attempted_at: DateTime.add(DateTime.utc_now(), -360, :second)
+      })
+
+      # Recent jobs that should not be included
+      oban_job_fixture(%{
+        worker: "Towerops.Workers.DevicePollerWorker",
+        state: "executing",
+        args: %{"device_id" => device.id},
+        attempted_at: DateTime.add(DateTime.utc_now(), -30, :second)
+      })
+
+      stuck = JobMonitoring.list_stuck_jobs()
+
+      assert length(stuck) == 2
+      # Verify oldest first (discovery job at -360s)
+      assert Enum.at(stuck, 0).id == stuck_discovery.id
+      assert Enum.at(stuck, 1).id == stuck_poller.id
+    end
+
+    test "excludes jobs just under the threshold boundary" do
+      device = device_fixture()
+
+      # Polling job at 119 seconds - should NOT be stuck (under 120s threshold)
+      oban_job_fixture(%{
+        worker: "Towerops.Workers.DevicePollerWorker",
+        state: "executing",
+        args: %{"device_id" => device.id},
+        attempted_at: DateTime.add(DateTime.utc_now(), -119, :second)
+      })
+
+      # Discovery job at 299 seconds - should NOT be stuck (under 300s threshold)
+      oban_job_fixture(%{
+        worker: "Towerops.Workers.DiscoveryWorker",
+        state: "executing",
+        args: %{"device_id" => device.id},
+        attempted_at: DateTime.add(DateTime.utc_now(), -299, :second)
+      })
+
+      assert JobMonitoring.list_stuck_jobs() == []
+    end
+
+    test "includes jobs just over the threshold" do
+      device = device_fixture()
+
+      # Polling job at 125 seconds - should be stuck (over 120s threshold)
+      stuck_poller = oban_job_fixture(%{
+        worker: "Towerops.Workers.DevicePollerWorker",
+        state: "executing",
+        args: %{"device_id" => device.id},
+        attempted_at: DateTime.add(DateTime.utc_now(), -125, :second)
+      })
+
+      # Discovery job at 305 seconds - should be stuck (over 300s threshold)
+      stuck_discovery = oban_job_fixture(%{
+        worker: "Towerops.Workers.DiscoveryWorker",
+        state: "executing",
+        args: %{"device_id" => device.id},
+        attempted_at: DateTime.add(DateTime.utc_now(), -305, :second)
+      })
+
+      stuck = JobMonitoring.list_stuck_jobs()
+
+      assert length(stuck) == 2
+      assert stuck |> Enum.map(& &1.id) |> Enum.sort() == [stuck_discovery.id, stuck_poller.id] |> Enum.sort()
+    end
+  end
+
+  describe "list_failed_jobs/0" do
+    test "returns retryable and cancelled jobs" do
+      device = device_fixture()
+
+      oban_job_fixture(%{
+        worker: "Towerops.Workers.DevicePollerWorker",
+        state: "retryable",
+        args: %{"device_id" => device.id}
+      })
+
+      oban_job_fixture(%{
+        worker: "Towerops.Workers.DiscoveryWorker",
+        state: "cancelled",
+        args: %{"device_id" => device.id}
+      })
+
+      # Should not include completed
+      oban_job_fixture(%{
+        worker: "Towerops.Workers.DevicePollerWorker",
+        state: "completed",
+        args: %{"device_id" => device.id}
+      })
+
+      failed = JobMonitoring.list_failed_jobs()
+
+      assert length(failed) == 2
+      assert Enum.all?(failed, fn j -> j.state in ["retryable", "cancelled"] end)
+    end
+  end
+
+  describe "list_recent_completions/1" do
+    test "returns completed, cancelled, and discarded jobs up to limit" do
+      device = device_fixture()
+
+      # Create 5 completed jobs
+      for _ <- 1..5 do
+        oban_job_fixture(%{
+          worker: "Towerops.Workers.DevicePollerWorker",
+          state: "completed",
+          args: %{"device_id" => device.id},
+          completed_at: DateTime.utc_now()
+        })
+      end
+
+      # Should not include executing
+      oban_job_fixture(%{
+        worker: "Towerops.Workers.DevicePollerWorker",
+        state: "executing",
+        args: %{"device_id" => device.id}
+      })
+
+      recent = JobMonitoring.list_recent_completions(3)
+
+      assert length(recent) == 3
+      assert Enum.all?(recent, fn j -> j.state == "completed" end)
+    end
+  end
+end

test/towerops_web/live/admin/monitoring_live_test.exs

@@ -0,0 +1,81 @@
+defmodule ToweropsWeb.Admin.MonitoringLiveTest do
+  use ToweropsWeb.ConnCase
+
+  import Phoenix.LiveViewTest
+  import Towerops.DevicesFixtures
+  import Towerops.JobsFixtures
+
+  describe "MonitoringLive" do
+    setup [:register_and_log_in_superuser]
+
+    test "renders monitoring dashboard", %{conn: conn} do
+      {:ok, _view, html} = live(conn, ~p"/admin/monitoring")
+
+      assert html =~ "Job Monitoring Dashboard"
+      assert html =~ "Active Operations"
+      assert html =~ "Problems"
+      assert html =~ "Health Metrics"
+    end
+
+    test "subscribes to job lifecycle events", %{conn: conn} do
+      {:ok, view, _html} = live(conn, ~p"/admin/monitoring")
+
+      # Verify PubSub subscription by checking it's alive
+      assert Process.alive?(view.pid)
+    end
+
+    test "displays active operations section", %{conn: conn} do
+      {:ok, _view, html} = live(conn, ~p"/admin/monitoring")
+
+      assert html =~ "Active Operations"
+      assert html =~ "No jobs currently executing"
+    end
+  end
+
+  describe "real-time updates" do
+    setup [:register_and_log_in_superuser]
+
+    test "displays executing jobs from database", %{conn: conn} do
+      device = device_fixture()
+
+      _job =
+        oban_job_fixture(%{
+          worker: "Towerops.Workers.DevicePollerWorker",
+          state: "executing",
+          args: %{"device_id" => device.id},
+          attempted_at: DateTime.utc_now()
+        })
+
+      {:ok, _view, html} = live(conn, ~p"/admin/monitoring")
+
+      # Should show the device name in executing jobs
+      assert html =~ device.name
+      assert html =~ "Active Operations (1)"
+    end
+  end
+
+  describe "metrics display" do
+    setup [:register_and_log_in_superuser]
+
+    test "shows current activity metrics", %{conn: conn} do
+      device = device_fixture()
+
+      oban_job_fixture(%{
+        worker: "Towerops.Workers.DevicePollerWorker",
+        state: "executing",
+        args: %{"device_id" => device.id}
+      })
+
+      {:ok, _view, html} = live(conn, ~p"/admin/monitoring")
+
+      assert html =~ "Executing"
+      assert html =~ "1"
+    end
+  end
+
+  defp register_and_log_in_superuser(%{conn: conn}) do
+    user = Towerops.AccountsFixtures.user_fixture()
+    user = Towerops.Repo.update!(Ecto.Changeset.change(user, is_superuser: true))
+    %{conn: log_in_user(conn, user), user: user}
+  end
+end