feat: queue items pagination (#1585)

* **New Features**
* Added paginated queue items browsing with filtering options (running
or queued).
  * Added log download capability for DAG runs, steps, and sub-DAG runs.

* **Improvements**
* Queue display now shows summary statistics with item counts and
utilization metrics.
* Enhanced queue interface with pagination controls and improved data
presentation.

Author: yottahmd

api/v2/api.gen.go

@@ -2688,8 +2688,8 @@ paths:
 
   /queues:
     get:
-      summary: "List all execution queues with active DAG-runs"
-      description: "Retrieves all queues showing both running and queued DAG-runs, organized by queue/process group"
+      summary: "List all execution queues with summary statistics"
+      description: "Returns queue list with running/queued counts. Use /queues/{name}/items for paginated item details."
       operationId: "listQueues"
       tags:
         - "queues"
@@ -2709,6 +2709,51 @@ paths:
               schema:
                 $ref: "#/components/schemas/Error"
 
+  /queues/{name}/items:
+    get:
+      summary: "Get paginated items for a specific queue"
+      description: "Returns paginated list of running or queued DAG-runs for the specified queue"
+      operationId: "listQueueItems"
+      tags:
+        - "queues"
+      parameters:
+        - $ref: "#/components/parameters/RemoteNode"
+        - $ref: "#/components/parameters/Page"
+        - $ref: "#/components/parameters/PerPage"
+        - name: name
+          in: path
+          description: "Queue name"
+          required: true
+          schema:
+            type: string
+        - name: type
+          in: query
+          description: "Item type to fetch"
+          required: false
+          schema:
+            type: string
+            enum: ["running", "queued"]
+            default: "queued"
+      responses:
+        "200":
+          description: "A successful response"
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/QueueItemsResponse"
+        "404":
+          description: "Queue not found"
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/Error"
+        default:
+          description: "Generic error response"
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/Error"
+
   /services/resources/history:
     get:
       summary: "Get resource usage history"
@@ -4691,7 +4736,7 @@ components:
 
     Queue:
       type: object
-      description: "A queue/process group with its active DAG-runs"
+      description: "A queue/process group with summary statistics"
       properties:
         name:
           type: string
@@ -4704,21 +4749,40 @@ components:
           type: integer
           description: "Maximum number of concurrent runs allowed. For 'global' queues, this is the configured maxConcurrency. For 'dag-based' queues, this is the DAG's maxActiveRuns (default 1)"
           minimum: 1
+        runningCount:
+          type: integer
+          description: "Number of currently running DAG-runs"
+          minimum: 0
+        queuedCount:
+          type: integer
+          description: "Number of queued DAG-runs waiting to execute"
+          minimum: 0
         running:
           type: array
-          description: "List of currently running DAG-runs"
-          items:
-            $ref: "#/components/schemas/DAGRunSummary"
-        queued:
-          type: array
-          description: "List of DAG-runs waiting to execute"
+          description: "List of currently running DAG-runs (bounded by maxConcurrency)"
           items:
             $ref: "#/components/schemas/DAGRunSummary"
       required:
         - name
         - type
+        - runningCount
+        - queuedCount
         - running
-        - queued
+
+    QueueItemsResponse:
+      type: object
+      description: "Paginated queue items response"
+      properties:
+        items:
+          type: array
+          description: "List of DAG-run summaries"
+          items:
+            $ref: "#/components/schemas/DAGRunSummary"
+        pagination:
+          $ref: "#/components/schemas/Pagination"
+      required:
+        - items
+        - pagination
 
     QueuesSummary:
       type: object

api/v2/api.yaml

@@ -25,6 +25,8 @@ var (
 )
 
 // Collector implements prometheus.Collector interface
+var _ prometheus.Collector = (*Collector)(nil)
+
 type Collector struct {
 	startTime       time.Time
 	version         string

internal/cmn/telemetry/collector.go

@@ -23,6 +23,8 @@ type mockDAGStore struct {
 	mock.Mock
 }
 
+var _ exec.DAGStore = (*mockDAGStore)(nil)
+
 func (m *mockDAGStore) Create(ctx context.Context, fileName string, spec []byte) error {
 	args := m.Called(ctx, fileName, spec)
 	return args.Error(0)
@@ -103,6 +105,8 @@ type mockDAGRunStore struct {
 	mock.Mock
 }
 
+var _ exec.DAGRunStore = (*mockDAGRunStore)(nil)
+
 // RemoveDAGRun implements models.DAGRunStore.
 func (m *mockDAGRunStore) RemoveDAGRun(_ context.Context, _ exec.DAGRunRef) error {
 	panic("unimplemented")
@@ -174,12 +178,12 @@ func (m *mockDAGRunStore) RenameDAGRuns(ctx context.Context, oldName, newName st
 	return args.Error(0)
 }
 
-var _ exec.QueueStore = (*mockQueueStore)(nil)
-
 type mockQueueStore struct {
 	mock.Mock
 }
 
+var _ exec.QueueStore = (*mockQueueStore)(nil)
+
 // QueueWatcher implements execution.QueueStore.
 func (m *mockQueueStore) QueueWatcher(_ context.Context) exec.QueueWatcher {
 	panic("unimplemented")
@@ -223,6 +227,11 @@ func (m *mockQueueStore) List(ctx context.Context, name string) ([]exec.QueuedIt
 	return args.Get(0).([]exec.QueuedItemData), args.Error(1)
 }
 
+func (m *mockQueueStore) ListPaginated(ctx context.Context, name string, pg exec.Paginator) (exec.PaginatedResult[exec.QueuedItemData], error) {
+	args := m.Called(ctx, name, pg)
+	return args.Get(0).(exec.PaginatedResult[exec.QueuedItemData]), args.Error(1)
+}
+
 func (m *mockQueueStore) All(ctx context.Context) ([]exec.QueuedItemData, error) {
 	args := m.Called(ctx)
 	if args.Get(0) == nil {
@@ -237,6 +246,8 @@ type mockServiceRegistry struct {
 	mock.Mock
 }
 
+var _ exec.ServiceRegistry = (*mockServiceRegistry)(nil)
+
 func (m *mockServiceRegistry) Register(ctx context.Context, serviceName exec.ServiceName, hostInfo exec.HostInfo) error {
 	args := m.Called(ctx, serviceName, hostInfo)
 	return args.Error(0)

internal/cmn/telemetry/collector_test.go

@@ -24,6 +24,8 @@ type QueueStore interface {
 	Len(ctx context.Context, name string) (int, error)
 	// List returns all items in the queue with the given name
 	List(ctx context.Context, name string) ([]QueuedItemData, error)
+	// ListPaginated returns paginated items for a specific queue
+	ListPaginated(ctx context.Context, name string, pg Paginator) (PaginatedResult[QueuedItemData], error)
 	// All returns all items in the queue
 	All(ctx context.Context) ([]QueuedItemData, error)
 	// ListByDAGName returns all items that has a specific DAG name

internal/core/exec/queue.go

@@ -32,6 +32,8 @@ func init() {
 }
 
 // Provider implements the llm.Provider interface for Anthropic Claude.
+var _ llm.Provider = (*Provider)(nil)
+
 type Provider struct {
 	config     llm.Config
 	httpClient *llm.HTTPClient

internal/llm/providers/anthropic/anthropic.go

@@ -25,6 +25,8 @@ func init() {
 }
 
 // Provider implements the llm.Provider interface for Google Gemini.
+var _ llm.Provider = (*Provider)(nil)
+
 type Provider struct {
 	config     llm.Config
 	httpClient *llm.HTTPClient

internal/llm/providers/gemini/gemini.go

@@ -25,6 +25,8 @@ func init() {
 }
 
 // Provider implements the llm.Provider interface for local OpenAI-compatible servers.
+var _ llm.Provider = (*Provider)(nil)
+
 type Provider struct {
 	config     llm.Config
 	httpClient *llm.HTTPClient

internal/llm/providers/local/local.go

@@ -24,6 +24,8 @@ func init() {
 }
 
 // Provider implements the llm.Provider interface for OpenAI.
+var _ llm.Provider = (*Provider)(nil)
+
 type Provider struct {
 	config     llm.Config
 	httpClient *llm.HTTPClient

internal/llm/providers/openai/openai.go

@@ -25,6 +25,8 @@ func init() {
 }
 
 // Provider implements the llm.Provider interface for OpenRouter.
+var _ llm.Provider = (*Provider)(nil)
+
 type Provider struct {
 	config     llm.Config
 	httpClient *llm.HTTPClient