feat: create a /jobs api to return queue and history jobs

Author: ric-yu

execution.py

@@ -1124,7 +1124,8 @@ class ExecutionStatus(NamedTuple):
         messages: List[str]
 
     def task_done(self, item_id, history_result,
-                  status: Optional['PromptQueue.ExecutionStatus'], process_item=None):
+                  status: Optional['PromptQueue.ExecutionStatus'], process_item=None,
+                  execution_time: Optional[float] = None):
         with self.mutex:
             prompt = self.currently_running.pop(item_id)
             if len(self.history) > MAXIMUM_HISTORY_SIZE:
@@ -1141,6 +1142,7 @@ def task_done(self, item_id, history_result,
                 "prompt": prompt,
                 "outputs": {},
                 'status': status_dict,
+                'execution_time': execution_time,
             }
             self.history[prompt[1]].update(history_result)
             self.server.queue_updated()
@@ -1217,6 +1219,202 @@ def delete_history_item(self, id_to_delete):
         with self.mutex:
             self.history.pop(id_to_delete, None)
 
+    def get_job(self, prompt_id):
+        """Get a single job by prompt_id from history or queue."""
+        with self.mutex:
+            # Check history first (most likely location)
+            if prompt_id in self.history:
+                return self._normalize_history_item(prompt_id, self.history[prompt_id], include_outputs=True)
+
+            # Check currently running
+            for item in self.currently_running.values():
+                if item[1] == prompt_id:
+                    return self._normalize_queue_item(item, 'in_progress')
+
+            # Check pending queue
+            for item in self.queue:
+                if item[1] == prompt_id:
+                    return self._normalize_queue_item(item, 'pending')
+
+            return None
+
+    def get_all_jobs(self, status_filter=None, sort_by="created_at", sort_order="desc", limit=None, offset=0):
+        """
+        Get all jobs (running, pending, completed) with filtering and sorting.
+
+        Args:
+            status_filter: list of statuses to include ['pending', 'in_progress', 'completed', 'error']
+            sort_by: field to sort by ('created_at', 'execution_time')
+            sort_order: 'asc' or 'desc'
+            limit: maximum number of items to return
+            offset: number of items to skip
+
+        Returns:
+            tuple: (jobs_list, total_count)
+        """
+        with self.mutex:
+            jobs = []
+
+            if status_filter is None:
+                status_filter = ['pending', 'in_progress', 'completed', 'error']
+
+            if 'in_progress' in status_filter:
+                for item in self.currently_running.values():
+                    jobs.append(self._normalize_queue_item(item, 'in_progress'))
+
+            if 'pending' in status_filter:
+                for item in self.queue:
+                    jobs.append(self._normalize_queue_item(item, 'pending'))
+
+            include_completed = 'completed' in status_filter
+            include_error = 'error' in status_filter
+            if include_completed or include_error:
+                for prompt_id, history_item in self.history.items():
+                    is_error = history_item.get('status', {}).get('status_str') == 'error'
+                    if (is_error and include_error) or (not is_error and include_completed):
+                        jobs.append(self._normalize_history_item(prompt_id, history_item))
+
+            jobs = self._apply_sorting(jobs, sort_by, sort_order)
+
+            total_count = len(jobs)
+
+            if offset > 0:
+                jobs = jobs[offset:]
+            if limit is not None:
+                jobs = jobs[:limit]
+
+            return (jobs, total_count)
+
+    def _normalize_queue_item(self, item, status):
+        """Convert queue item tuple to unified job dict."""
+        number, prompt_id, prompt, extra_data, outputs_to_execute = item[:5]
+        create_time = extra_data.get('create_time')
+
+        return {
+            'id': prompt_id,
+            'status': status,
+            'create_time': create_time,
+            'execution_time': None,
+            'error_message': None,
+            'outputs_count': 0,
+            'preview_output': None,
+            'workflow_id': None,
+        }
+
+    def _normalize_history_item(self, prompt_id, history_item, include_outputs=False):
+        """Convert history item dict to unified job dict."""
+        prompt_tuple = history_item['prompt']
+        number, _, prompt, extra_data, outputs_to_execute = prompt_tuple[:5]
+        create_time = extra_data.get('create_time')
+
+        # Determine status from history status
+        status_info = history_item.get('status', {})
+        if status_info:
+            status = 'completed' if status_info.get('status_str') == 'success' else 'error'
+        else:
+            status = 'completed'
+
+        outputs = history_item.get('outputs', {})
+
+        outputs_count, preview_output = self._get_outputs_summary(outputs)
+
+        # Get error message from status if error
+        error_message = None
+        if status == 'error' and status_info:
+            messages = status_info.get('messages', [])
+            if messages:
+                error_message = messages[0] if isinstance(messages[0], str) else str(messages[0])
+
+        execution_time = history_item.get('execution_time')
+
+        job = {
+            'id': prompt_id,
+            'status': status,
+            'create_time': create_time,
+            'execution_time': execution_time,
+            'error_message': error_message,
+            'outputs_count': outputs_count,
+            'preview_output': preview_output,
+            'workflow_id': None,
+        }
+
+        if include_outputs:
+            job['outputs'] = outputs
+            job['prompt'] = prompt
+            job['extra_data'] = extra_data
+            job['outputs_to_execute'] = outputs_to_execute
+
+        return job
+
+    def _get_outputs_summary(self, outputs):
+        """
+        Count outputs and find preview in a single pass.
+        Returns (outputs_count, preview_output).
+
+        Preview priority (matching frontend):
+        1. type="output" with previewable media
+        2. Any previewable media
+        """
+        count = 0
+        preview_output = None
+        fallback_preview = None
+
+        for node_id, node_outputs in outputs.items():
+            for media_type, items in node_outputs.items():
+                if media_type == 'animated' or not isinstance(items, list):
+                    continue
+                for item in items:
+                    count += 1
+
+                    # Skip if we already have the best preview (type=output)
+                    if preview_output is not None:
+                        continue
+
+                    filename = item.get('filename', '').lower()
+                    fmt = item.get('format', '')
+
+                    # Check if previewable (image/video/audio/3D) - matching frontend logic
+                    is_previewable = (
+                        media_type == 'images' or
+                        media_type == 'video' or
+                        media_type == 'audio' or
+                        filename.endswith(('.png', '.jpg', '.jpeg', '.gif', '.webp', '.bmp')) or  # images
+                        filename.endswith(('.mp4', '.webm', '.mov', '.avi')) or  # video
+                        filename.endswith(('.mp3', '.wav', '.ogg', '.flac')) or  # audio
+                        filename.endswith(('.obj', '.fbx', '.gltf', '.glb')) or  # 3D
+                        (fmt and (fmt.startswith('video/') or fmt.startswith('audio/')))
+                    )
+
+                    if not is_previewable:
+                        continue
+
+                    enriched = {
+                        **item,
+                        'nodeId': node_id,
+                        'mediaType': media_type
+                    }
+
+                    if item.get('type') == 'output':
+                        preview_output = enriched
+                    elif fallback_preview is None:
+                        fallback_preview = enriched
+
+        return count, preview_output or fallback_preview
+
+    def _apply_sorting(self, jobs, sort_by, sort_order):
+        """Sort jobs list by specified field and order."""
+        reverse = (sort_order == 'desc')
+
+        if sort_by == 'execution_time':
+            def get_sort_key(job):
+                return job.get('execution_time') or 0
+        else:
+            # Default to create_time
+            def get_sort_key(job):
+                return job.get('create_time') or 0
+
+        return sorted(jobs, key=get_sort_key, reverse=reverse)
+
     def set_flag(self, name, data):
         with self.mutex:
             self.flags[name] = data

main.py

@@ -202,19 +202,21 @@ def prompt_worker(q, server_instance):
             e.execute(item[2], prompt_id, extra_data, item[4])
             need_gc = True
 
+            current_time = time.perf_counter()
+            execution_time = current_time - execution_start_time
+
             remove_sensitive = lambda prompt: prompt[:5] + prompt[6:]
             q.task_done(item_id,
                         e.history_result,
                         status=execution.PromptQueue.ExecutionStatus(
                             status_str='success' if e.success else 'error',
                             completed=e.success,
-                            messages=e.status_messages), process_item=remove_sensitive)
+                            messages=e.status_messages),
+                        process_item=remove_sensitive,
+                        execution_time=execution_time)
             if server_instance.client_id is not None:
                 server_instance.send_sync("executing", {"node": None, "prompt_id": prompt_id}, server_instance.client_id)
 
-            current_time = time.perf_counter()
-            execution_time = current_time - execution_start_time
-
             # Log Time in a more readable way after 10 minutes
             if execution_time > 600:
                 execution_time = time.strftime("%H:%M:%S", time.gmtime(execution_time))

server.py

@@ -688,6 +688,96 @@ async def get_object_info_node(request):
                 out[node_class] = node_info(node_class)
             return web.json_response(out)
 
+        @routes.get("/api/jobs")
+        async def get_jobs(request):
+            """List all jobs with filtering, sorting, and pagination."""
+            query = request.rel_url.query
+
+            status_param = query.get("status", None)
+            status_filter = None
+            if status_param:
+                status_filter = [s.strip() for s in status_param.split(',') if s.strip()]
+                valid_statuses = {'pending', 'in_progress', 'completed', 'error'}
+                status_filter = [s for s in status_filter if s in valid_statuses]
+                if not status_filter:
+                    status_filter = None
+
+            sort_by = query.get('sort', 'created_at')
+            if sort_by != 'created_at':
+                return web.json_response(
+                    {"error": "sort must be 'created_at'"},
+                    status=400
+                )
+
+            sort_order = query.get('order', 'desc')
+            if sort_order not in {'asc', 'desc'}:
+                return web.json_response(
+                    {"error": "order must be 'asc' or 'desc'"},
+                    status=400
+                )
+
+            limit = None
+            if 'limit' in query:
+                try:
+                    limit = int(query.get('limit'))
+                    if limit <= 0 or limit > 500:
+                        return web.json_response(
+                            {"error": "limit must be between 1 and 500"},
+                            status=400
+                        )
+                except (ValueError, TypeError):
+                    return web.json_response(
+                        {"error": "limit must be an integer"},
+                        status=400
+                    )
+
+            offset = 0
+            if 'offset' in query:
+                try:
+                    offset = int(query.get('offset'))
+                    if offset < 0:
+                        offset = 0
+                except (ValueError, TypeError):
+                    return web.json_response(
+                        {"error": "offset must be an integer"},
+                        status=400
+                    )
+
+            jobs, total = self.prompt_queue.get_all_jobs(
+                status_filter=status_filter,
+                sort_by=sort_by,
+                sort_order=sort_order,
+                limit=limit,
+                offset=offset
+            )
+
+            has_more = (offset + len(jobs)) < total
+
+            return web.json_response({
+                'jobs': jobs,
+                'total': total,
+                'has_more': has_more
+            })
+
+        @routes.get("/api/jobs/{job_id}")
+        async def get_job(request):
+            """Get a single job by ID."""
+            job_id = request.match_info.get("job_id", None)
+            if not job_id:
+                return web.json_response(
+                    {"error": "job_id is required"},
+                    status=400
+                )
+
+            job = self.prompt_queue.get_job(job_id)
+            if job is None:
+                return web.json_response(
+                    {"error": "Job not found"},
+                    status=404
+                )
+
+            return web.json_response(job)
+
         @routes.get("/history")
         async def get_history(request):
             max_items = request.rel_url.query.get("max_items", None)