Reduce Media Studio payloads and redundant reads

apps/api/app/graph/normalization.py

@@ -102,7 +102,7 @@ def materialize_workflow_defaults(
     recipe_catalog_items: list[dict] | None = None,
     preset_catalog_items: list[dict] | None = None,
 ) -> GraphWorkflow:
-    definitions = definitions_by_type or registry.definitions_by_type()
+    definitions = definitions_by_type if definitions_by_type is not None else registry.definitions_by_type()
     all_recipe_catalog = recipe_catalog_items if recipe_catalog_items is not None else prompt_recipe_catalog(status="all")
     recipe_lookup = _recipe_by_id(all_recipe_catalog)
     all_preset_catalog = preset_catalog_items if preset_catalog_items is not None else media_preset_catalog(status="all")

apps/api/app/graph/routes.py

@@ -41,6 +41,7 @@
     GraphWorkflow,
     GraphWorkflowListResponse,
     GraphWorkflowRecord,
+    GraphWorkflowSummaryListResponse,
 )
 from .validator import validate_workflow
 
@@ -213,6 +214,15 @@ def list_workflows() -> GraphWorkflowListResponse:
     )
 
 
+@router.get("/workflows/summary", response_model=GraphWorkflowSummaryListResponse)
+def list_workflow_summaries(
+    limit: int = Query(default=60, ge=1, le=60),
+    offset: int = Query(default=0, ge=0),
+    q: Optional[str] = Query(default=None),
+) -> GraphWorkflowSummaryListResponse:
+    return GraphWorkflowSummaryListResponse(**store.list_graph_workflow_summaries(limit=limit, offset=offset, q=q))
+
+
 @router.post("/workflows", response_model=GraphWorkflowRecord)
 def create_workflow(payload: GraphWorkflow) -> GraphWorkflowRecord:
     workflow = materialize_workflow_defaults(payload)
@@ -430,9 +440,17 @@ def recover_run(run_id: str) -> GraphRun:
 
 @router.get("/templates", response_model=GraphTemplateListResponse)
 def list_templates() -> GraphTemplateListResponse:
+    definitions_by_type = registry.definitions_by_type()
+    recipe_catalog_items = prompt_recipe_catalog(status="all")
+    preset_catalog_items = media_preset_catalog(status="all")
     items: List[GraphTemplateRecord] = []
     for item in store.list_graph_templates():
-        workflow = materialize_workflow_defaults(GraphWorkflow(**(item.get("workflow_json") or {})))
+        workflow = materialize_workflow_defaults(
+            GraphWorkflow(**(item.get("workflow_json") or {})),
+            definitions_by_type=definitions_by_type,
+            recipe_catalog_items=recipe_catalog_items,
+            preset_catalog_items=preset_catalog_items,
+        )
         items.append(GraphTemplateRecord(**{**item, "workflow_json": workflow.model_dump(mode="json")}))
     return GraphTemplateListResponse(items=items)
 

apps/api/app/graph/schemas.py

@@ -83,17 +83,28 @@ class GraphWorkflow(BaseModel):
     metadata: Dict[str, Any] = Field(default_factory=dict)
 
 
-class GraphWorkflowRecord(BaseModel):
+class GraphWorkflowSummary(BaseModel):
     workflow_id: str
     name: str
     description: Optional[str] = None
     status: str = "active"
     schema_version: int = 1
-    workflow_json: Dict[str, Any] = Field(default_factory=dict)
     created_at: Optional[str] = None
     updated_at: Optional[str] = None
 
 
+class GraphWorkflowSummaryListResponse(BaseModel):
+    items: List[GraphWorkflowSummary] = Field(default_factory=list)
+    total: int
+    limit: int
+    offset: int
+    next_offset: Optional[int] = None
+
+
+class GraphWorkflowRecord(GraphWorkflowSummary):
+    workflow_json: Dict[str, Any] = Field(default_factory=dict)
+
+
 class GraphTemplate(BaseModel):
     template_id: Optional[str] = None
     name: str

apps/api/app/main.py

@@ -6,7 +6,7 @@
 import logging
 from pathlib import Path
 import tempfile
-from typing import List, Optional
+from typing import List, Literal, Optional
 
 from fastapi import FastAPI, File, HTTPException, Query, Request, UploadFile
 from fastapi.responses import FileResponse, JSONResponse
@@ -52,6 +52,8 @@
     ProjectUpsertRequest,
     PresetListResponse,
     PresetRecord,
+    PresetSummaryRecord,
+    PresetSummaryListResponse,
     PresetUpsertRequest,
     PricingResponse,
     PricingEstimateResponse,
@@ -390,6 +392,23 @@ def search_presets(
     )
 
 
+@app.get("/media/presets/summary", response_model=List[PresetSummaryRecord])
+def list_preset_summaries():
+    return [PresetSummaryRecord(**item) for item in store.list_presets(summary=True)]
+
+
+@app.get("/media/presets/search/summary", response_model=PresetSummaryListResponse)
+def search_preset_summaries(
+    limit: int = Query(default=60, ge=1, le=100),
+    offset: int = Query(default=0, ge=0),
+    q: Optional[str] = Query(default=None),
+    category: Optional[str] = Query(default=None),
+    status: str = Query(default="active"),
+):
+    page = store.list_presets_page(limit=limit, offset=offset, q=q, category=category, status=status, summary=True)
+    return PresetSummaryListResponse(**{**page, "items": [PresetSummaryRecord(**item) for item in page["items"]]})
+
+
 @app.get("/media/presets/{preset_id}", response_model=PresetRecord)
 def get_preset(preset_id: str):
     record = store.get_preset(preset_id)
@@ -1006,23 +1025,30 @@ def list_assets(
     project_id: Optional[str] = Query(default=None),
     compact: bool = False,
     q: Optional[str] = Query(default=None),
+    cursor_mode: Literal["timestamp", "asset_id"] = "timestamp",
 ):
-    rows = store.list_assets(
-        limit=limit + 1,
-        cursor=cursor,
-        favorites_only=favorites,
-        media_type=media_type,
-        model_key=model_key,
-        status=status,
-        preset_key=preset_key,
-        project_id=project_id,
-        compact=compact,
-        q=q,
-    )
+    try:
+        rows = store.list_assets(
+            limit=limit + 1,
+            cursor=cursor,
+            favorites_only=favorites,
+            media_type=media_type,
+            model_key=model_key,
+            status=status,
+            preset_key=preset_key,
+            project_id=project_id,
+            compact=compact,
+            q=q,
+            cursor_mode=cursor_mode,
+        )
+    except ValueError as exc:
+        raise _bad_request(str(exc)) from exc
     has_more = len(rows) > limit
     rows = rows[:limit]
     items = [AssetRecord(**item) for item in rows]
-    next_cursor = items[-1].created_at if has_more and items else None
+    next_cursor = (
+        items[-1].asset_id if cursor_mode == "asset_id" else items[-1].created_at
+    ) if has_more and items else None
     return AssetListResponse(items=items, next_cursor=next_cursor)
 
 

apps/api/app/schemas.py

@@ -296,6 +296,41 @@ class PresetListResponse(BaseModel):
     next_offset: Optional[int] = None
 
 
+class PresetSummaryRecord(BaseModel):
+    preset_id: str
+    key: str
+    label: str
+    description: Optional[str] = None
+    category: str = "general"
+    status: str = "active"
+    model_key: Optional[str] = None
+    source_kind: str = "custom"
+    base_builtin_key: Optional[str] = None
+    applies_to_models: List[str] = Field(default_factory=list, validation_alias="applies_to_models_json")
+    applies_to_task_modes: List[str] = Field(default_factory=list, validation_alias="applies_to_task_modes_json")
+    applies_to_input_patterns: List[str] = Field(default_factory=list, validation_alias="applies_to_input_patterns_json")
+    requires_image: bool = False
+    requires_video: bool = False
+    requires_audio: bool = False
+    thumbnail_path: Optional[str] = None
+    thumbnail_url: Optional[str] = None
+    version: Optional[str] = None
+    priority: int = 100
+    created_at: Optional[str] = None
+    updated_at: Optional[str] = None
+    input_schema_count: int = 0
+    input_slots_count: int = 0
+    image_input_policy: str = "none"
+
+
+class PresetSummaryListResponse(BaseModel):
+    items: List[PresetSummaryRecord] = Field(default_factory=list)
+    total: int = 0
+    limit: int = 60
+    offset: int = 0
+    next_offset: Optional[int] = None
+
+
 class PromptRecipeVariable(BaseModel):
     key: str
     token: Optional[str] = None

apps/api/app/store.py

@@ -9,7 +9,7 @@
 from typing import Any, Dict, Iterable, List, Optional, Tuple
 
 from .db_backup import backup_database
-from .store_preset_search import preset_search_sql, normalize_search_text
+from .store_preset_listing import list_presets, list_presets_page
 from .db import get_connection
 from .settings import settings
 from .store_support import (
@@ -59,6 +59,7 @@
     list_graph_runs_for_workflow,
     list_graph_templates,
     list_graph_workflows,
+    list_graph_workflow_summaries,
     mark_interrupted_graph_runs,
     update_graph_run,
     update_graph_run_node,
@@ -194,63 +195,6 @@ def get_model_queue_policy(model_key: str) -> Optional[Dict[str, Any]]:
     return _decode_row(row) if row else None
 
 
-def list_presets() -> List[Dict[str, Any]]:
-    with get_connection() as connection:
-        rows = connection.execute(
-            "SELECT * FROM media_presets WHERE status != 'archived' ORDER BY priority DESC, updated_at DESC, key ASC"
-        ).fetchall()
-    return [_decode_row(row) for row in rows]
-
-
-def list_presets_page(
-    *,
-    limit: int = 60,
-    offset: int = 0,
-    q: Optional[str] = None,
-    category: Optional[str] = None,
-    status: str = "active",
-) -> Dict[str, Any]:
-    safe_limit = max(1, min(int(limit or 60), 100))
-    safe_offset = max(0, int(offset or 0))
-    clauses: List[str] = []
-    params: List[Any] = []
-    normalized_status = (status or "active").strip().lower()
-    if normalized_status == "all":
-        clauses.append("1 = 1")
-    elif normalized_status == "archived":
-        clauses.append("status = ?")
-        params.append("archived")
-    else:
-        clauses.append("status != ?")
-        params.append("archived")
-    search_clause, search_params, search_order, order_params = preset_search_sql(q)
-    if search_clause:
-        clauses.append(search_clause)
-        params.extend(search_params)
-    normalized_category = (category or "").strip().lower()
-    if normalized_category and normalized_category != "all":
-        clauses.append("lower(COALESCE(category, 'general')) = ?")
-        params.append(normalized_category)
-    where_sql = " AND ".join(clauses)
-    with get_connection() as connection:
-        connection.create_function("preset_search_normalize", 1, normalize_search_text, deterministic=True)
-        total_row = connection.execute(f"SELECT COUNT(*) AS total FROM media_presets WHERE {where_sql}", params).fetchone()
-        rows = connection.execute(
-            f"""
-            SELECT *
-            FROM media_presets
-            WHERE {where_sql}
-            ORDER BY {search_order}priority DESC, updated_at DESC, key ASC
-            LIMIT ? OFFSET ?
-            """,
-            [*params, *order_params, safe_limit, safe_offset],
-        ).fetchall()
-    total = int(total_row["total"] if total_row else 0)
-    items = [_decode_row(row) for row in rows]
-    next_offset = safe_offset + len(items) if safe_offset + len(items) < total else None
-    return {"items": items, "total": total, "limit": safe_limit, "offset": safe_offset, "next_offset": next_offset}
-
-
 def get_preset(preset_id: str) -> Optional[Dict[str, Any]]:
     return _get_table("media_presets", "preset_id", preset_id)
 
@@ -629,12 +573,24 @@ def list_assets(
     project_id: Optional[str] = None,
     compact: bool = False,
     q: Optional[str] = None,
+    cursor_mode: str = "timestamp",
 ) -> List[Dict[str, Any]]:
     clauses = ["dismissed = 0", "hidden_from_dashboard = 0"]
     params: List[Any] = []
     if cursor:
-        clauses.append("created_at < ?")
-        params.append(cursor)
+        if cursor_mode == "asset_id":
+            with get_connection() as connection:
+                anchor = connection.execute(
+                    "SELECT created_at, rowid FROM media_assets WHERE asset_id = ?",
+                    (cursor,),
+                ).fetchone()
+            if not anchor:
+                raise ValueError("Invalid gallery continuation. Reload the gallery to restart.")
+            clauses.append("(created_at < ? OR (created_at = ? AND rowid > ?))")
+            params.extend([anchor["created_at"], anchor["created_at"], anchor["rowid"]])
+        else:
+            clauses.append("created_at < ?")
+            params.append(cursor)
     if favorites_only:
         clauses.append("favorited = 1")
     if media_type == "image":
@@ -683,7 +639,9 @@ def list_assets(
         if compact
         else "*"
     )
-    query = "SELECT %s FROM media_assets WHERE %s ORDER BY created_at DESC LIMIT ?" % (columns, " AND ".join(clauses))
+    # SQLite's existing timestamp sort retains rowid order inside ties.
+    order = "created_at DESC, rowid ASC" if cursor_mode == "asset_id" else "created_at DESC"
+    query = "SELECT %s FROM media_assets WHERE %s ORDER BY %s LIMIT ?" % (columns, " AND ".join(clauses), order)
     params.append(limit)
     with get_connection() as connection:
         rows = connection.execute(query, params).fetchall()

apps/api/app/store_graph.py

@@ -22,6 +22,29 @@ def list_graph_workflows() -> List[Dict[str, Any]]:
     return [_decode_row(row) for row in rows]
 
 
+def list_graph_workflow_summaries(*, limit: int = 60, offset: int = 0, q: Optional[str] = None) -> Dict[str, Any]:
+    limit = max(1, min(int(limit), 60))
+    offset = max(0, int(offset))
+    where = "status != 'archived'"
+    params: List[Any] = []
+    search = (q or "").strip().lower()
+    if search:
+        pattern = "%" + search.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + "%"
+        where += " AND (LOWER(name) LIKE ? ESCAPE '\\' OR LOWER(COALESCE(description, '')) LIKE ? ESCAPE '\\')"
+        params.extend([pattern, pattern])
+    with get_connection() as connection:
+        total = int(connection.execute(f"SELECT COUNT(*) AS count FROM graph_workflows WHERE {where}", params).fetchone()["count"])
+        rows = connection.execute(
+            f"SELECT workflow_id, name, description, status, schema_version, created_at, updated_at "
+            f"FROM graph_workflows WHERE {where} "
+            "ORDER BY updated_at DESC, name ASC, workflow_id ASC LIMIT ? OFFSET ?",
+            [*params, limit, offset],
+        ).fetchall()
+    items = [dict(row) for row in rows]
+    next_offset = offset + len(items) if items and offset + len(items) < total else None
+    return {"items": items, "total": total, "limit": limit, "offset": offset, "next_offset": next_offset}
+
+
 def get_graph_workflow(workflow_id: str) -> Optional[Dict[str, Any]]:
     return _get_table("graph_workflows", "workflow_id", workflow_id)