fix: stabilize maintenance flow run/resume and procedural scrolling

backend/app/api/endpoints/maintenance_schedules.py

@@ -51,7 +51,9 @@ async def create_schedule(
 ):
     """Create a cron schedule for a maintenance flow. One per flow."""
     # Verify user's team owns the tree
-    await _get_tree_or_403(data.tree_id, current_user, db)
+    tree = await _get_tree_or_403(data.tree_id, current_user, db)
+    if tree.tree_type != "maintenance":
+        raise HTTPException(status_code=400, detail="Schedules are only supported for maintenance flows")
 
     # Check no existing schedule for this tree
     existing = await db.execute(

backend/app/api/endpoints/sessions.py

@@ -37,10 +37,13 @@ async def list_sessions(
     ticket_number: Optional[str] = Query(None, description="Search by ticket number (partial match)"),
     client_name: Optional[str] = Query(None, description="Search by client name (partial match)"),
     tree_name: Optional[str] = Query(None, description="Filter by tree name from snapshot"),
+    tree_id: Optional[UUID] = Query(None, description="Filter by tree ID"),
     started_after: Optional[datetime] = Query(None, description="Filter sessions started after this datetime"),
     started_before: Optional[datetime] = Query(None, description="Filter sessions started before this datetime"),
     completed_after: Optional[datetime] = Query(None, description="Filter sessions completed after this datetime"),
     completed_before: Optional[datetime] = Query(None, description="Filter sessions completed before this datetime"),
+    page: Optional[int] = Query(None, ge=1, description="1-based page number (frontend compatibility)"),
+    size: Optional[int] = Query(None, ge=1, le=100, description="Page size (frontend compatibility)"),
     skip: int = Query(0, ge=0),
     limit: int = Query(50, ge=1, le=100)
 ):
@@ -66,6 +69,10 @@ async def list_sessions(
     if tree_name:
         query = query.where(Session.tree_snapshot['name'].astext.ilike(f"%{tree_name}%"))
 
+    # Tree ID filter
+    if tree_id:
+        query = query.where(Session.tree_id == tree_id)
+
     # Date range filters
     if started_after:
         query = query.where(Session.started_at >= started_after)
@@ -76,8 +83,13 @@ async def list_sessions(
     if completed_before:
         query = query.where(Session.completed_at <= completed_before)
 
+    effective_limit = size if size is not None else limit
+    effective_skip = skip
+    if page is not None:
+        effective_skip = (page - 1) * effective_limit
+
     query = query.order_by(Session.started_at.desc())
-    query = query.offset(skip).limit(limit)
+    query = query.offset(effective_skip).limit(effective_limit)
 
     result = await db.execute(query)
     sessions = result.scalars().all()

backend/app/core/scheduler.py

@@ -47,6 +47,16 @@ async def _fire_maintenance_schedule(schedule_id: str) -> None:
                 logger.error(f"Tree {schedule.tree_id} not found for schedule {schedule_id}")
                 return
 
+            if tree.tree_type != "maintenance":
+                logger.warning(f"Skipping schedule {schedule_id}: tree {tree.id} is not a maintenance flow")
+                return
+
+            if not tree.is_active or tree.status == "draft":
+                logger.warning(
+                    f"Skipping schedule {schedule_id}: tree {tree.id} is inactive or draft"
+                )
+                return
+
             # Resolve targets
             targets: list[dict] = []
             if schedule.target_list_id:
@@ -61,7 +71,12 @@ async def _fire_maintenance_schedule(schedule_id: str) -> None:
                 targets = [{"label": "Unassigned"}]
 
             batch_id = uuid.uuid4()
-            tree_snapshot = tree.tree_structure
+            tree_snapshot = {
+                **tree.tree_structure,
+                "name": tree.name,
+                "description": tree.description,
+                "tree_type": tree.tree_type,
+            }
 
             sessions_to_add = []
             for target in targets:

backend/tests/test_maintenance_schedules.py

@@ -98,6 +98,23 @@ async def test_get_schedule_not_found(client: AsyncClient, auth_headers: dict):
     assert resp.status_code == 404
 
 
+@pytest.mark.asyncio
+async def test_cannot_create_schedule_for_non_maintenance_tree(
+    client: AsyncClient, auth_headers: dict, test_tree: dict
+):
+    """Schedules are restricted to maintenance flows."""
+    resp = await client.post(
+        "/api/v1/maintenance-schedules",
+        json={
+            "tree_id": test_tree["id"],
+            "cron_expression": "0 0 1 * *",
+            "timezone": "UTC",
+        },
+        headers=auth_headers,
+    )
+    assert resp.status_code == 400
+
+
 @pytest.mark.asyncio
 async def test_cannot_schedule_other_teams_tree(client: AsyncClient, auth_headers: dict, test_db):
     """User cannot create a schedule for a tree belonging to another team."""

backend/tests/test_sessions.py

@@ -882,6 +882,79 @@ class TestSessions:
         assert len(data) >= 1
         assert test_tree["name"] in data[0]["tree_snapshot"]["name"]
 
+    @pytest.mark.asyncio
+    async def test_filter_sessions_by_tree_id(
+        self, client: AsyncClient, auth_headers: dict, test_tree: dict
+    ):
+        """Test filtering sessions by tree_id."""
+        other_tree_response = await client.post(
+            "/api/v1/trees",
+            json={
+                "name": "Other Tree",
+                "description": "Second tree for filter test",
+                "category": test_tree["category"],
+                "tree_structure": test_tree["tree_structure"],
+            },
+            headers=auth_headers,
+        )
+        assert other_tree_response.status_code == 201
+        other_tree_id = other_tree_response.json()["id"]
+
+        await client.post(
+            "/api/v1/sessions",
+            json={"tree_id": test_tree["id"], "ticket_number": "TREE-A"},
+            headers=auth_headers,
+        )
+        await client.post(
+            "/api/v1/sessions",
+            json={"tree_id": other_tree_id, "ticket_number": "TREE-B"},
+            headers=auth_headers,
+        )
+
+        response = await client.get(
+            f"/api/v1/sessions?tree_id={test_tree['id']}",
+            headers=auth_headers,
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert len(data) >= 1
+        assert all(item["tree_id"] == test_tree["id"] for item in data)
+
+    @pytest.mark.asyncio
+    async def test_list_sessions_supports_size_and_page_params(
+        self, client: AsyncClient, auth_headers: dict, test_tree: dict
+    ):
+        """Test frontend-compatible page/size query params."""
+        await client.post(
+            "/api/v1/sessions",
+            json={"tree_id": test_tree["id"], "ticket_number": "P1"},
+            headers=auth_headers,
+        )
+        await client.post(
+            "/api/v1/sessions",
+            json={"tree_id": test_tree["id"], "ticket_number": "P2"},
+            headers=auth_headers,
+        )
+        await client.post(
+            "/api/v1/sessions",
+            json={"tree_id": test_tree["id"], "ticket_number": "P3"},
+            headers=auth_headers,
+        )
+
+        first_page = await client.get("/api/v1/sessions?size=2&page=1", headers=auth_headers)
+        assert first_page.status_code == 200
+        first_data = first_page.json()
+        assert len(first_data) == 2
+
+        second_page = await client.get("/api/v1/sessions?size=2&page=2", headers=auth_headers)
+        assert second_page.status_code == 200
+        second_data = second_page.json()
+        assert len(second_data) >= 1
+
+        first_ids = {item["id"] for item in first_data}
+        second_ids = {item["id"] for item in second_data}
+        assert first_ids.isdisjoint(second_ids)
+
     @pytest.mark.asyncio
     async def test_filter_sessions_by_started_date_range(
         self, client: AsyncClient, auth_headers: dict, test_tree: dict