feat(triggers): read-mostly API surface (aggregator + delete + bulk)

The frontend list view + bulk-delete needs a small HTTP surface over
the in-flow trigger model. This commit adds it in three slim layers:

NEW services/triggers/queries.py
  list_triggers_for_user(session, user_id) → list[TriggerInstance]
    Walks user-owned flows whose data is non-null, scans for
    CronTrigger nodes via discovery, joins each with its most-recent
    trigger_job rows. Output dataclass carries (flow_id, flow_name,
    component_id, cron_expression, timezone, max_attempts,
    next_fire_at, last_finished_status, last_finished_at) — enough
    for the table without further round-trips.
  O(F) flows + O(F) job batches. Volume bound by hand-edits in the
  canvas; query count traded for simplicity.

NEW services/triggers/removal.py
  remove_cron_trigger_node(flow_data, component_id)
  remove_all_cron_trigger_nodes(flow_data)
    Pure dict mutations — no DB. Return new flow_data plus a
    boolean / list of removed ids. Both prune edges referencing the
    removed nodes so the resulting graph is still valid.
  Routes invoke these, write the result back via the standard
  _patch_flow helper, which fires the lifecycle hook that cancels
  the queued trigger_job rows for the removed components. No
  manual cascade in the API layer.

REWRITTEN api/v1/triggers.py
  GET    /api/v1/triggers
            list[TriggerInstance] for the current user
  DELETE /api/v1/triggers/{flow_id}/{component_id}
            strip one node + 204; 404 when not found
  DELETE /api/v1/triggers
            bulk strip every CronTrigger node from every owned flow;
            returns {flows_updated, components_removed}
  GET    /api/v1/triggers/{flow_id}/{component_id}/jobs
            recent trigger_job rows for one component; optional
            ?status_filter, default limit 50

  No POST / PATCH / PUT — the canvas is the only editing surface.
  Routes are thin: ownership check + helper call + response shape.

NEW tests/unit/services/triggers/test_queries.py (6 tests)
  empty user, multi-flow aggregation, user isolation,
  next-fire computation, last-terminal computation, flows
  without data.

NEW tests/unit/services/triggers/test_removal.py (10 tests)
  single-remove: missing flow_data, missing component, id collision
  with non-CronTrigger node, multi-node selectivity, edge pruning,
  input immutability. Bulk-remove: empty case, multi-strip with
  edge pruning, missing edges, malformed nodes list.

53/53 tests pass, ruff clean. The /triggers route is now ready for
the frontend list view in Etapa G.

src/backend/base/langflow/api/v1/triggers.py

@@ -1,11 +1,218 @@
 """Read-mostly HTTP surface for the in-flow trigger feature.
 
-Filled in a later commit. For now this is just the router shell so
-the API package keeps importing cleanly while the schema refactor
-takes effect — the trigger configuration moved into ``flow.data`` and
-the old CRUD endpoints no longer apply.
+Creation lives in the flow canvas: a user drops a CronTrigger
+component, fills in its inputs, saves the flow. This module surfaces
+the resulting state for a management UI:
+
+    GET    /api/v1/triggers
+           One row per CronTrigger component across the current user's
+           flows. Combines live config (read from ``flow.data``) with
+           the most recent ``trigger_job`` rows so the list view can
+           render "next fire" and "last run" columns.
+
+    DELETE /api/v1/triggers/{flow_id}/{component_id}
+           Strip a single CronTrigger node from a flow. The lifecycle
+           hook on the subsequent save cancels its queued job rows.
+           Returns 404 when the flow or the component cannot be
+           found for the current user.
+
+    DELETE /api/v1/triggers
+           Bulk strip: remove every CronTrigger node from every flow
+           the current user owns. Idempotent — re-running on a
+           cleaned-up set is a no-op and still returns 200.
+
+    GET    /api/v1/triggers/{flow_id}/{component_id}/jobs
+           Recent ``trigger_job`` rows for one component, descending
+           ``scheduled_at``. Used by the per-trigger history drawer.
+
+There are no POST / PATCH / PUT endpoints by design: the canvas is the
+single editing surface for trigger configuration.
 """
 
-from fastapi import APIRouter
+from __future__ import annotations
+
+from dataclasses import asdict
+from datetime import datetime
+from typing import TYPE_CHECKING
+from uuid import UUID
+
+from fastapi import APIRouter, HTTPException, status
+from sqlmodel import select
+
+from langflow.api.utils import CurrentActiveUser, DbSession
+from langflow.services.database.models.flow.model import Flow
+from langflow.services.database.models.jobs.model import JobStatus
+from langflow.services.database.models.triggers.crud import list_jobs_for_flow
+from langflow.services.database.models.triggers.model import TriggerJobRead
+from langflow.services.triggers.queries import (
+    TriggerInstance,
+    list_triggers_for_user,
+)
+from langflow.services.triggers.removal import (
+    remove_all_cron_trigger_nodes,
+    remove_cron_trigger_node,
+)
+
+if TYPE_CHECKING:
+    from sqlmodel.ext.asyncio.session import AsyncSession
+
 
 router = APIRouter(prefix="/triggers", tags=["Triggers"])
+
+
+# --------------------------------------------------------------------------- #
+#  Response schemas (kept minimal & inline — no separate Pydantic dance)
+# --------------------------------------------------------------------------- #
+
+
+def _serialize_instance(instance: TriggerInstance) -> dict:
+    """``dataclasses.asdict`` plus JSON-friendly types for datetimes/UUIDs."""
+    raw = asdict(instance)
+    return {
+        **raw,
+        "flow_id": str(instance.flow_id),
+        "next_fire_at": _iso(instance.next_fire_at),
+        "last_finished_at": _iso(instance.last_finished_at),
+        "last_finished_status": (
+            instance.last_finished_status.value if instance.last_finished_status else None
+        ),
+    }
+
+
+def _iso(value: datetime | None) -> str | None:
+    return value.isoformat() if value else None
+
+
+# --------------------------------------------------------------------------- #
+#  Helpers shared by the routes
+# --------------------------------------------------------------------------- #
+
+
+async def _get_owned_flow(
+    session: AsyncSession,
+    flow_id: UUID,
+    user_id: UUID,
+) -> Flow:
+    """Fetch a flow restricted to ``user_id``; 404 on miss.
+
+    Centralises the ownership check so every route enforces it the
+    same way and the response payloads do not need to redo the
+    cross-user → 404 mapping inline.
+    """
+    statement = select(Flow).where(Flow.id == flow_id, Flow.user_id == user_id)
+    flow = (await session.exec(statement)).first()
+    if flow is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Flow not found")
+    return flow
+
+
+async def _persist_flow_data_change(session: AsyncSession, flow: Flow) -> None:
+    """Save the in-memory flow.data mutation back through the same hooks the API uses.
+
+    Importing the helper inside the function avoids a hard module-load
+    coupling between this route and ``flows_helpers`` (which itself
+    imports from the triggers services package).
+    """
+    from langflow.api.v1.flows_helpers import _patch_flow
+    from langflow.services.database.models.flow.model import FlowUpdate
+
+    update = FlowUpdate(data=flow.data)
+    from langflow.services.deps import get_storage_service
+
+    storage_service = get_storage_service()
+    await _patch_flow(
+        session=session,
+        db_flow=flow,
+        flow=update,
+        user_id=flow.user_id,
+        storage_service=storage_service,
+    )
+
+
+# --------------------------------------------------------------------------- #
+#  Routes
+# --------------------------------------------------------------------------- #
+
+
+@router.get("", response_model=list[dict])
+async def list_triggers(
+    *,
+    session: DbSession,
+    current_user: CurrentActiveUser,
+) -> list[dict]:
+    """All CronTrigger components the current user has across all flows."""
+    instances = await list_triggers_for_user(session, current_user.id)
+    return [_serialize_instance(i) for i in instances]
+
+
+@router.delete("/{flow_id}/{component_id}", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_trigger(
+    *,
+    session: DbSession,
+    current_user: CurrentActiveUser,
+    flow_id: UUID,
+    component_id: str,
+) -> None:
+    """Strip one CronTrigger node from a flow.
+
+    The post-save lifecycle hook downgrades the trigger's queued jobs
+    to ``cancelled``, so no manual cascade is needed here.
+    """
+    flow = await _get_owned_flow(session, flow_id, current_user.id)
+    new_data, was_removed = remove_cron_trigger_node(flow.data, component_id)
+    if not was_removed:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Trigger component not found in this flow",
+        )
+    flow.data = new_data
+    await _persist_flow_data_change(session, flow)
+
+
+@router.delete("", response_model=dict, status_code=status.HTTP_200_OK)
+async def delete_all_triggers(
+    *,
+    session: DbSession,
+    current_user: CurrentActiveUser,
+) -> dict:
+    """Remove every CronTrigger component from every flow the user owns.
+
+    Returns ``{"flows_updated": int, "components_removed": int}`` so the
+    UI can show a meaningful confirmation toast.
+    """
+    statement = select(Flow).where(Flow.user_id == current_user.id)
+    flows = list((await session.exec(statement)).all())
+
+    flows_updated = 0
+    components_removed = 0
+    for flow in flows:
+        new_data, removed_ids = remove_all_cron_trigger_nodes(flow.data)
+        if not removed_ids:
+            continue
+        flow.data = new_data
+        await _persist_flow_data_change(session, flow)
+        flows_updated += 1
+        components_removed += len(removed_ids)
+
+    return {"flows_updated": flows_updated, "components_removed": components_removed}
+
+
+@router.get("/{flow_id}/{component_id}/jobs", response_model=list[TriggerJobRead])
+async def list_trigger_jobs(
+    *,
+    session: DbSession,
+    current_user: CurrentActiveUser,
+    flow_id: UUID,
+    component_id: str,
+    status_filter: JobStatus | None = None,
+    limit: int = 50,
+):
+    """Recent trigger_jobs for one component on one flow."""
+    await _get_owned_flow(session, flow_id, current_user.id)
+    return await list_jobs_for_flow(
+        session,
+        flow_id,
+        component_id=component_id,
+        status=status_filter,
+        limit=limit,
+    )

src/backend/base/langflow/services/triggers/queries.py

@@ -0,0 +1,167 @@
+"""Read-only aggregator queries for the triggers HTTP surface.
+
+Walks every flow the current user owns, surfaces the CronTrigger
+components present in their saved JSON, and joins each component
+with its most recent ``trigger_job`` row(s). The HTTP layer hands
+the result straight to the frontend list view.
+
+Kept separate from the route handlers so the query logic can be
+exercised in unit tests without standing up a FastAPI client.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from sqlmodel import col, select
+
+from langflow.services.database.models.flow.model import Flow
+from langflow.services.database.models.jobs.model import JobStatus
+from langflow.services.database.models.triggers.model import TriggerJob
+from langflow.services.triggers.discovery import (
+    CronTriggerConfig,
+    find_cron_trigger_configs,
+)
+
+if TYPE_CHECKING:
+    from datetime import datetime
+    from uuid import UUID
+
+    from sqlmodel.ext.asyncio.session import AsyncSession
+
+
+@dataclass(frozen=True)
+class TriggerInstance:
+    """One CronTrigger component instance surfaced to the API.
+
+    Combines the component config (live, from ``flow.data``) with two
+    cached views into the work queue:
+
+    * ``next_fire_at`` — when the worker will pick up the next queued
+      job for this component, if any.
+    * ``last_finished_status`` / ``last_finished_at`` — the most
+      recent terminal job (completed / failed / cancelled / timed_out)
+      for this component, for the "last run" column in the UI.
+
+    All fields are derived; nothing here is persisted in this shape.
+    """
+
+    flow_id: UUID
+    flow_name: str
+    component_id: str
+    cron_expression: str
+    timezone: str
+    max_attempts: int
+    next_fire_at: datetime | None
+    last_finished_status: JobStatus | None
+    last_finished_at: datetime | None
+
+
+_TERMINAL_STATES: tuple[JobStatus, ...] = (
+    JobStatus.COMPLETED,
+    JobStatus.FAILED,
+    JobStatus.CANCELLED,
+    JobStatus.TIMED_OUT,
+)
+
+
+async def _user_flows_with_data(session: AsyncSession, user_id: UUID) -> list[Flow]:
+    """Return user-owned flows that have ``flow.data`` populated.
+
+    Filters NULL/empty data server-side so the Python iteration only
+    sees rows worth inspecting.
+    """
+    statement = select(Flow).where(
+        Flow.user_id == user_id,
+        col(Flow.data).is_not(None),
+    )
+    result = await session.exec(statement)
+    return list(result.all())
+
+
+async def _jobs_for_components(
+    session: AsyncSession,
+    flow_id: UUID,
+    component_ids: list[str],
+) -> dict[str, list[TriggerJob]]:
+    """Group existing trigger_job rows by component_id for one flow.
+
+    Returns ``{component_id: [TriggerJob, ...]}`` ordered with the most
+    recently-scheduled jobs first so the aggregator can grab "next
+    queued" and "last finished" with simple list comprehensions.
+    """
+    if not component_ids:
+        return {}
+    statement = (
+        select(TriggerJob)
+        .where(
+            TriggerJob.flow_id == flow_id,
+            col(TriggerJob.component_id).in_(component_ids),
+        )
+        .order_by(col(TriggerJob.scheduled_at).desc())
+    )
+    result = await session.exec(statement)
+    by_component: dict[str, list[TriggerJob]] = {cid: [] for cid in component_ids}
+    for job in result.all():
+        by_component.setdefault(job.component_id, []).append(job)
+    return by_component
+
+
+def _summarise(
+    flow: Flow,
+    config: CronTriggerConfig,
+    jobs: list[TriggerJob],
+) -> TriggerInstance:
+    """Combine one component's config with its job history into a TriggerInstance."""
+    next_fire = next(
+        (j.scheduled_at for j in reversed(jobs) if j.status == JobStatus.QUEUED),
+        None,
+    )
+    last_terminal = next(
+        (j for j in jobs if j.status in _TERMINAL_STATES),
+        None,
+    )
+    return TriggerInstance(
+        flow_id=flow.id,
+        flow_name=flow.name,
+        component_id=config.component_id,
+        cron_expression=config.cron_expression,
+        timezone=config.timezone,
+        max_attempts=config.max_attempts,
+        next_fire_at=next_fire,
+        last_finished_status=last_terminal.status if last_terminal else None,
+        last_finished_at=last_terminal.finished_at if last_terminal else None,
+    )
+
+
+async def list_triggers_for_user(
+    session: AsyncSession,
+    user_id: UUID,
+) -> list[TriggerInstance]:
+    """Surface every CronTrigger component owned by ``user_id``.
+
+    O(F) flows + O(F) job batches — one query per flow that has
+    triggers. The volume of triggers a single user creates is bounded
+    by hand-editing the canvas, so we trade query count for simplicity.
+    """
+    flows = await _user_flows_with_data(session, user_id)
+
+    instances: list[TriggerInstance] = []
+    for flow in flows:
+        configs = find_cron_trigger_configs(flow.data)
+        if not configs:
+            continue
+        jobs_by_component = await _jobs_for_components(
+            session,
+            flow.id,
+            [c.component_id for c in configs],
+        )
+        instances.extend(
+            _summarise(flow, config, jobs_by_component.get(config.component_id, []))
+            for config in configs
+        )
+
+    # Stable order: flow name first, then component id within the flow.
+    instances.sort(key=lambda i: (i.flow_name.lower(), i.component_id))
+    return instances

src/backend/base/langflow/services/triggers/removal.py

@@ -0,0 +1,122 @@
+"""Strip CronTrigger nodes out of ``flow.data``.
+
+Used by the bulk and single-trigger DELETE endpoints. The function is
+pure on the input dict — it returns a new ``flow.data`` plus the list
+of removed component ids. The route handler writes the new data back
+to the flow row and then calls the standard lifecycle hook, which
+takes care of cancelling the now-orphaned queued ``trigger_job`` rows.
+
+Keeping the mutation pure (no DB access) makes the deletion logic
+trivially unit-testable and keeps the API handlers free of dict
+gymnastics.
+"""
+
+from __future__ import annotations
+
+from copy import deepcopy
+from typing import Any
+
+from langflow.services.triggers.discovery import CRON_TRIGGER_TYPE
+
+
+def remove_cron_trigger_node(
+    flow_data: dict[str, Any] | None,
+    component_id: str,
+) -> tuple[dict[str, Any], bool]:
+    """Return ``(new_flow_data, was_removed)`` after dropping a single node.
+
+    Also prunes any edge whose source or target was the removed node,
+    so saving the result through the standard flow update path leaves
+    a valid graph for downstream consumers. The caller decides what
+    to do with ``was_removed`` (typically: 404 when False).
+    """
+    if not flow_data:
+        return flow_data or {"nodes": [], "edges": []}, False
+
+    new_data = deepcopy(flow_data)
+    nodes = new_data.get("nodes")
+    if not isinstance(nodes, list):
+        return new_data, False
+
+    survivors = [n for n in nodes if not _matches(n, component_id)]
+    was_removed = len(survivors) != len(nodes)
+    new_data["nodes"] = survivors
+
+    if was_removed:
+        new_data["edges"] = _prune_edges(new_data.get("edges"), {component_id})
+
+    return new_data, was_removed
+
+
+def remove_all_cron_trigger_nodes(
+    flow_data: dict[str, Any] | None,
+) -> tuple[dict[str, Any], list[str]]:
+    """Strip every CronTrigger node from ``flow_data`` in one pass.
+
+    Returns ``(new_flow_data, removed_component_ids)``. ``removed_*``
+    is the list of node ids that disappeared, in the order they
+    appeared in the original graph — useful for bulk-delete responses
+    that want to surface what was actually removed.
+    """
+    if not flow_data:
+        return flow_data or {"nodes": [], "edges": []}, []
+
+    new_data = deepcopy(flow_data)
+    nodes = new_data.get("nodes")
+    if not isinstance(nodes, list):
+        return new_data, []
+
+    removed_ids: list[str] = []
+    survivors: list[dict[str, Any]] = []
+    for node in nodes:
+        if _is_cron_trigger_node(node):
+            node_id = node.get("id") if isinstance(node, dict) else None
+            if isinstance(node_id, str):
+                removed_ids.append(node_id)
+            continue
+        survivors.append(node)
+
+    new_data["nodes"] = survivors
+    if removed_ids:
+        new_data["edges"] = _prune_edges(new_data.get("edges"), set(removed_ids))
+    return new_data, removed_ids
+
+
+# --------------------------------------------------------------------------- #
+#  internal helpers
+# --------------------------------------------------------------------------- #
+
+
+def _matches(node: object, component_id: str) -> bool:
+    """Return True when ``node`` is the CronTrigger with the given id."""
+    return (
+        isinstance(node, dict)
+        and node.get("id") == component_id
+        and _is_cron_trigger_node(node)
+    )
+
+
+def _is_cron_trigger_node(node: object) -> bool:
+    """Same matcher used by ``discovery.find_cron_trigger_nodes``.
+
+    Guarded inline so this module stays import-free from heavy
+    discovery state — just the constant.
+    """
+    if not isinstance(node, dict):
+        return False
+    data = node.get("data")
+    return isinstance(data, dict) and data.get("type") == CRON_TRIGGER_TYPE
+
+
+def _prune_edges(edges: Any, removed_ids: set[str]) -> list[Any]:
+    """Drop any edge that references a removed node id."""
+    if not isinstance(edges, list):
+        return []
+    return [
+        edge
+        for edge in edges
+        if not (
+            isinstance(edge, dict)
+            and (edge.get("source") in removed_ids or edge.get("target") in removed_ids)
+        )
+    ]

src/backend/tests/unit/services/triggers/test_queries.py

@@ -0,0 +1,190 @@
+"""Unit tests for the read aggregator (services.triggers.queries)."""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+from uuid import uuid4
+
+import pytest
+from langflow.services.database.models.flow.model import Flow
+from langflow.services.database.models.jobs.model import JobStatus
+from langflow.services.database.models.triggers import TriggerJob
+from langflow.services.database.models.user.model import User
+from langflow.services.triggers.queries import list_triggers_for_user
+
+
+def _cron_node(component_id: str, *, cron: str = "*/5 * * * *", tz: str = "UTC") -> dict:
+    return {
+        "id": component_id,
+        "type": "genericNode",
+        "data": {
+            "type": "CronTrigger",
+            "node": {
+                "template": {
+                    "cron_expression": {"value": cron},
+                    "timezone": {"value": tz},
+                    "max_attempts": {"value": 3},
+                },
+            },
+        },
+    }
+
+
+async def _seed_user(session, *, suffix: str = "") -> User:
+    user = User(
+        id=uuid4(),
+        username=f"u-{uuid4().hex[:6]}{suffix}",
+        password="x",  # noqa: S106
+        is_active=True,
+    )
+    session.add(user)
+    await session.commit()
+    return user
+
+
+async def _seed_flow(session, *, user_id, name: str, nodes: list[dict]) -> Flow:
+    flow = Flow(id=uuid4(), name=name, data={"nodes": nodes, "edges": []}, user_id=user_id)
+    session.add(flow)
+    await session.commit()
+    return flow
+
+
+@pytest.mark.asyncio
+async def test_list_returns_empty_for_user_without_triggers(async_session):
+    user = await _seed_user(async_session)
+    instances = await list_triggers_for_user(async_session, user.id)
+    assert instances == []
+
+
+@pytest.mark.asyncio
+async def test_list_surfaces_one_instance_per_component(async_session):
+    user = await _seed_user(async_session)
+    await _seed_flow(
+        async_session,
+        user_id=user.id,
+        name="flow-a",
+        nodes=[
+            _cron_node("CronTrigger-a1", cron="*/5 * * * *"),
+            _cron_node("CronTrigger-a2", cron="0 9 * * *"),
+        ],
+    )
+    await _seed_flow(
+        async_session,
+        user_id=user.id,
+        name="flow-b",
+        nodes=[_cron_node("CronTrigger-b1", cron="0 0 * * *", tz="Europe/London")],
+    )
+
+    instances = await list_triggers_for_user(async_session, user.id)
+    by_id = {(i.flow_name, i.component_id): i for i in instances}
+    assert set(by_id.keys()) == {
+        ("flow-a", "CronTrigger-a1"),
+        ("flow-a", "CronTrigger-a2"),
+        ("flow-b", "CronTrigger-b1"),
+    }
+    london = by_id[("flow-b", "CronTrigger-b1")]
+    assert london.cron_expression == "0 0 * * *"
+    assert london.timezone == "Europe/London"
+
+
+@pytest.mark.asyncio
+async def test_list_isolates_users(async_session):
+    """A user must never see another user's triggers."""
+    alice = await _seed_user(async_session, suffix="a")
+    bob = await _seed_user(async_session, suffix="b")
+    await _seed_flow(async_session, user_id=alice.id, name="alice-flow", nodes=[_cron_node("CronTrigger-a")])
+    await _seed_flow(async_session, user_id=bob.id, name="bob-flow", nodes=[_cron_node("CronTrigger-b")])
+
+    alice_view = await list_triggers_for_user(async_session, alice.id)
+    bob_view = await list_triggers_for_user(async_session, bob.id)
+
+    assert [i.component_id for i in alice_view] == ["CronTrigger-a"]
+    assert [i.component_id for i in bob_view] == ["CronTrigger-b"]
+
+
+@pytest.mark.asyncio
+async def test_list_picks_next_fire_from_queued_job(async_session):
+    user = await _seed_user(async_session)
+    flow = await _seed_flow(
+        async_session,
+        user_id=user.id,
+        name="alpha",
+        nodes=[_cron_node("CronTrigger-x")],
+    )
+    next_fire = datetime.now(timezone.utc) + timedelta(minutes=4)
+    async_session.add(
+        TriggerJob(
+            id=uuid4(),
+            flow_id=flow.id,
+            component_id="CronTrigger-x",
+            status=JobStatus.QUEUED,
+            scheduled_at=next_fire,
+            attempt=1,
+            max_attempts=3,
+        ),
+    )
+    await async_session.commit()
+
+    instances = await list_triggers_for_user(async_session, user.id)
+    assert len(instances) == 1
+    # SQLAlchemy may strip the tz when storing into TIMESTAMP, so
+    # compare the wall-clock instant tolerantly.
+    assert instances[0].next_fire_at is not None
+    delta = abs((instances[0].next_fire_at.replace(tzinfo=None) - next_fire.replace(tzinfo=None)).total_seconds())
+    assert delta < 1
+
+
+@pytest.mark.asyncio
+async def test_list_picks_last_terminal_job(async_session):
+    user = await _seed_user(async_session)
+    flow = await _seed_flow(
+        async_session,
+        user_id=user.id,
+        name="alpha",
+        nodes=[_cron_node("CronTrigger-x")],
+    )
+    older = datetime.now(timezone.utc) - timedelta(hours=2)
+    newer = datetime.now(timezone.utc) - timedelta(minutes=2)
+    async_session.add(
+        TriggerJob(
+            id=uuid4(),
+            flow_id=flow.id,
+            component_id="CronTrigger-x",
+            status=JobStatus.COMPLETED,
+            scheduled_at=older,
+            started_at=older,
+            finished_at=older,
+            attempt=1,
+            max_attempts=3,
+        ),
+    )
+    async_session.add(
+        TriggerJob(
+            id=uuid4(),
+            flow_id=flow.id,
+            component_id="CronTrigger-x",
+            status=JobStatus.FAILED,
+            scheduled_at=newer,
+            started_at=newer,
+            finished_at=newer,
+            attempt=1,
+            max_attempts=3,
+            error="boom",
+        ),
+    )
+    await async_session.commit()
+
+    instances = await list_triggers_for_user(async_session, user.id)
+    assert len(instances) == 1
+    assert instances[0].last_finished_status == JobStatus.FAILED
+
+
+@pytest.mark.asyncio
+async def test_list_skips_flows_without_data(async_session):
+    user = await _seed_user(async_session)
+    # ``data=None`` flow — should be skipped by the server-side filter.
+    flow = Flow(id=uuid4(), name="empty", data=None, user_id=user.id)
+    async_session.add(flow)
+    await async_session.commit()
+    instances = await list_triggers_for_user(async_session, user.id)
+    assert instances == []

src/backend/tests/unit/services/triggers/test_removal.py

@@ -0,0 +1,131 @@
+"""Unit tests for the pure removal helpers (no DB, no I/O)."""
+
+from __future__ import annotations
+
+from copy import deepcopy
+
+from langflow.services.triggers.removal import (
+    remove_all_cron_trigger_nodes,
+    remove_cron_trigger_node,
+)
+
+
+def _cron_node(component_id: str) -> dict:
+    return {
+        "id": component_id,
+        "type": "genericNode",
+        "data": {"type": "CronTrigger", "node": {"template": {}}},
+    }
+
+
+def _chat_node(component_id: str = "ChatInput-c") -> dict:
+    return {
+        "id": component_id,
+        "type": "genericNode",
+        "data": {"type": "ChatInput", "node": {"template": {}}},
+    }
+
+
+def _flow(nodes: list[dict], edges: list[dict] | None = None) -> dict:
+    return {"nodes": nodes, "edges": edges or []}
+
+
+# --------------------------------------------------------------------------- #
+#  remove_cron_trigger_node (single)
+# --------------------------------------------------------------------------- #
+
+
+def test_remove_single_returns_false_for_none_flow_data():
+    result, removed = remove_cron_trigger_node(None, "CronTrigger-x")
+    assert removed is False
+    assert result == {"nodes": [], "edges": []}
+
+
+def test_remove_single_returns_false_when_component_not_present():
+    flow = _flow([_chat_node()])
+    result, removed = remove_cron_trigger_node(flow, "CronTrigger-x")
+    assert removed is False
+    assert result["nodes"] == flow["nodes"]
+
+
+def test_remove_single_does_not_match_a_non_cron_node_with_same_id():
+    """Safety: id collisions never reach non-CronTrigger nodes.
+
+    Even if another node accidentally has an id colliding with the target,
+    only nodes whose ``data.type`` is ``CronTrigger`` should be touched.
+    """
+    colliding = {"id": "CronTrigger-x", "type": "genericNode", "data": {"type": "ChatInput"}}
+    flow = _flow([colliding])
+    result, removed = remove_cron_trigger_node(flow, "CronTrigger-x")
+    assert removed is False
+    assert result["nodes"] == [colliding]
+
+
+def test_remove_single_drops_only_the_targeted_node():
+    flow = _flow([_chat_node(), _cron_node("CronTrigger-x"), _cron_node("CronTrigger-y")])
+    result, removed = remove_cron_trigger_node(flow, "CronTrigger-x")
+    assert removed is True
+    ids = [n["id"] for n in result["nodes"]]
+    assert ids == ["ChatInput-c", "CronTrigger-y"]
+
+
+def test_remove_single_prunes_edges_referencing_the_removed_node():
+    flow = _flow(
+        [_cron_node("CronTrigger-x"), _chat_node()],
+        edges=[
+            {"source": "CronTrigger-x", "target": "ChatInput-c"},
+            {"source": "ChatInput-c", "target": "OtherNode"},
+        ],
+    )
+    result, removed = remove_cron_trigger_node(flow, "CronTrigger-x")
+    assert removed is True
+    assert result["edges"] == [{"source": "ChatInput-c", "target": "OtherNode"}]
+
+
+def test_remove_single_does_not_mutate_the_input():
+    flow = _flow([_cron_node("CronTrigger-x")])
+    snapshot = deepcopy(flow)
+    remove_cron_trigger_node(flow, "CronTrigger-x")
+    assert flow == snapshot
+
+
+# --------------------------------------------------------------------------- #
+#  remove_all_cron_trigger_nodes (bulk)
+# --------------------------------------------------------------------------- #
+
+
+def test_remove_all_returns_empty_list_when_no_triggers_present():
+    flow = _flow([_chat_node()])
+    result, removed_ids = remove_all_cron_trigger_nodes(flow)
+    assert removed_ids == []
+    assert result["nodes"] == flow["nodes"]
+
+
+def test_remove_all_strips_every_cron_node():
+    flow = _flow(
+        [_chat_node(), _cron_node("CronTrigger-1"), _cron_node("CronTrigger-2")],
+        edges=[
+            {"source": "CronTrigger-1", "target": "ChatInput-c"},
+            {"source": "CronTrigger-2", "target": "ChatInput-c"},
+            {"source": "ChatInput-c", "target": "Other"},
+        ],
+    )
+    result, removed_ids = remove_all_cron_trigger_nodes(flow)
+    assert removed_ids == ["CronTrigger-1", "CronTrigger-2"]
+    assert [n["id"] for n in result["nodes"]] == ["ChatInput-c"]
+    assert result["edges"] == [{"source": "ChatInput-c", "target": "Other"}]
+
+
+def test_remove_all_handles_missing_edges_gracefully():
+    flow = {"nodes": [_cron_node("CronTrigger-x")]}
+    result, removed_ids = remove_all_cron_trigger_nodes(flow)
+    assert removed_ids == ["CronTrigger-x"]
+    assert result["nodes"] == []
+    assert result["edges"] == []
+
+
+def test_remove_all_returns_input_when_node_list_is_malformed():
+    flow = {"nodes": "not-a-list"}
+    result, removed_ids = remove_all_cron_trigger_nodes(flow)
+    assert removed_ids == []
+    assert result == {"nodes": "not-a-list"}