kin: KIN-UI-015-backend_dev

core/models.py

@@ -266,12 +266,28 @@ def get_task(conn: sqlite3.Connection, id: str) -> dict | None:
     return _row_to_dict(row)
 
 
+VALID_TASK_SORT_FIELDS = frozenset({
+    "updated_at", "created_at", "priority", "status", "title", "id",
+})
+
+
 def list_tasks(
     conn: sqlite3.Connection,
     project_id: str | None = None,
     status: str | None = None,
+    limit: int | None = None,
+    sort: str = "updated_at",
+    sort_dir: str = "desc",
 ) -> list[dict]:
-    """List tasks with optional project/status filters."""
+    """List tasks with optional project/status filters, limit, and sort.
+
+    sort: column name (validated against VALID_TASK_SORT_FIELDS, default 'updated_at')
+    sort_dir: 'asc' or 'desc' (default 'desc')
+    """
+    # Validate sort field to prevent SQL injection
+    sort_col = sort if sort in VALID_TASK_SORT_FIELDS else "updated_at"
+    sort_direction = "DESC" if sort_dir.lower() != "asc" else "ASC"
+
     query = "SELECT * FROM tasks WHERE 1=1"
     params: list = []
     if project_id:
@@ -280,7 +296,10 @@ def list_tasks(
     if status:
         query += " AND status = ?"
         params.append(status)
-    query += " ORDER BY priority, created_at"
+    query += f" ORDER BY {sort_col} {sort_direction}"
+    if limit is not None:
+        query += " LIMIT ?"
+        params.append(limit)
     return _rows_to_list(conn.execute(query, params).fetchall())
 
 

web/api.py

@@ -654,6 +654,28 @@ def start_project_phase(project_id: str):
 # Tasks
 # ---------------------------------------------------------------------------
 
+@app.get("/api/tasks")
+def list_tasks(
+    status: str | None = Query(default=None),
+    limit: int = Query(default=20, ge=1, le=500),
+    sort: str = Query(default="updated_at"),
+    project_id: str | None = Query(default=None),
+):
+    """List tasks with optional filters. sort defaults to updated_at desc."""
+    from core.models import VALID_TASK_SORT_FIELDS
+    conn = get_conn()
+    tasks = models.list_tasks(
+        conn,
+        project_id=project_id,
+        status=status,
+        limit=limit,
+        sort=sort if sort in VALID_TASK_SORT_FIELDS else "updated_at",
+        sort_dir="desc",
+    )
+    conn.close()
+    return tasks
+
+
 @app.get("/api/tasks/{task_id}")
 def get_task(task_id: str):
     conn = get_conn()